"""Globally sealed one-use ledger and durable rollback authority."""

from __future__ import annotations

import hashlib
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from types import TracebackType
from typing import Literal, Self, final

from pydantic import JsonValue, TypeAdapter

_ROWS = TypeAdapter(list[dict[str, JsonValue]])
_OBJECT = TypeAdapter(dict[str, JsonValue])


class AuthorityError(RuntimeError):
    """Fail-closed authority error."""


class WriteIo:
    """Hostile-write test seam."""

    def write(self, descriptor: int, payload: bytes) -> int:
        return os.write(descriptor, payload)


def write_all(descriptor: int, payload: bytes, io: WriteIo) -> None:
    """Write an exact payload despite short writes or EINTR."""
    offset = 0
    while offset < len(payload):
        try:
            written = io.write(descriptor, payload[offset:])
        except InterruptedError:
            continue
        if written <= 0:
            raise AuthorityError("zero_write")
        offset += written


def digest(payload: bytes) -> str:
    """Return one SHA-256 digest."""
    return hashlib.sha256(payload).hexdigest()


def atomic_write(path: Path, payload: bytes, mode: int = 0o600) -> None:
    """Durably replace one file."""
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    descriptor, temporary_text = tempfile.mkstemp(
        prefix=f".{path.name}.", dir=path.parent
    )
    temporary = Path(temporary_text)
    try:
        try:
            write_all(descriptor, payload, WriteIo())
            while True:
                try:
                    os.fsync(descriptor)
                    break
                except InterruptedError:
                    continue
        finally:
            os.close(descriptor)
        temporary.chmod(mode)
        os.replace(temporary, path)
        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
        if path.read_bytes() != payload:
            raise AuthorityError("write_readback")
    finally:
        temporary.unlink(missing_ok=True)


@dataclass(frozen=True, slots=True)
class SnapshotEntry:
    """One durable rollback entry."""

    live: Path
    saved: Path
    sha256: str
    mode: int


@dataclass(frozen=True, slots=True)
class Snapshot:
    """Complete post-stop rollback authority."""

    entries: tuple[SnapshotEntry, ...]
    manifest_digest: str


def capture(paths: tuple[Path, ...], root: Path) -> Snapshot:
    """Capture exact post-stop bytes under a sealed execution root."""
    snapshot_root = root / "post-stop-snapshots"
    snapshot_root.mkdir(parents=True, mode=0o700)
    entries: list[SnapshotEntry] = []
    rows: list[dict[str, str | int]] = []
    for ordinal, live in enumerate(paths):
        info = live.stat(follow_symlinks=False)
        if live.is_symlink() or not live.is_file() or info.st_nlink != 1:
            raise AuthorityError("unsafe_snapshot_source")
        payload = live.read_bytes()
        saved = snapshot_root / f"{ordinal:04d}.snapshot"
        atomic_write(saved, payload, 0o400)
        entry = SnapshotEntry(live, saved, digest(payload), info.st_mode & 0o7777)
        entries.append(entry)
        rows.append({
            "live": str(live),
            "mode": entry.mode,
            "ordinal": ordinal,
            "sha256": entry.sha256,
        })
    manifest = json.dumps(rows, sort_keys=True, separators=(",", ":")).encode() + b"\n"
    atomic_write(snapshot_root / "manifest.json", manifest, 0o400)
    return Snapshot(tuple(entries), f"sha256:{digest(manifest)}")


def load_snapshot(root: Path) -> Snapshot:
    """Rebind durable snapshot authority after abrupt process death."""
    snapshot_root = root / "post-stop-snapshots"
    manifest_path = snapshot_root / "manifest.json"
    rows = _ROWS.validate_json(manifest_path.read_bytes())
    entries: list[SnapshotEntry] = []
    for row in rows:
        live = row.get("live")
        mode = row.get("mode")
        ordinal = row.get("ordinal")
        sha256 = row.get("sha256")
        if (
            not isinstance(live, str)
            or not isinstance(mode, int)
            or isinstance(mode, bool)
            or not isinstance(ordinal, int)
            or isinstance(ordinal, bool)
            or not isinstance(sha256, str)
        ):
            raise AuthorityError("snapshot_manifest")
        entries.append(
            SnapshotEntry(
                Path(live),
                snapshot_root / f"{ordinal:04d}.snapshot",
                sha256,
                mode,
            )
        )
    manifest = manifest_path.read_bytes()
    return Snapshot(tuple(entries), f"sha256:{digest(manifest)}")


def restore(snapshot: Snapshot) -> None:
    """Restore every exact post-stop byte and mode."""
    failures: list[str] = []
    for entry in snapshot.entries:
        try:
            payload = entry.saved.read_bytes()
            if digest(payload) != entry.sha256:
                raise AuthorityError("snapshot_drift")
            atomic_write(entry.live, payload, entry.mode)
            info = entry.live.stat(follow_symlinks=False)
            if (
                entry.live.is_symlink()
                or not entry.live.is_file()
                or info.st_nlink != 1
                or info.st_mode & 0o7777 != entry.mode
                or digest(entry.live.read_bytes()) != entry.sha256
            ):
                raise AuthorityError("restore_identity")
        except (AuthorityError, OSError) as exc:
            failures.append(f"{entry.live}:{exc}")
    if failures:
        raise AuthorityError("restore:" + ",".join(failures))


@final
class GlobalLedger:
    """Reserve and consume authority across every possible execution root."""

    def __init__(self, root: Path, candidate: str) -> None:
        self.root = root
        self.candidate = candidate
        self.reserved = root / "authorization-reserved.json"
        self.consumed = root / "authorization-consumed.json"

    def reserve(self) -> None:
        """Create the sole global reservation before stop."""
        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
        if self.consumed.exists():
            raise AuthorityError("authorization_already_used")
        payload = (
            json.dumps(
                {"candidate_digest": self.candidate, "status": "RESERVED"},
                sort_keys=True,
                separators=(",", ":"),
            ).encode()
            + b"\n"
        )
        try:
            descriptor = os.open(
                self.reserved,
                os.O_CREAT | os.O_EXCL | os.O_WRONLY,
                0o400,
            )
        except FileExistsError as exc:
            raise AuthorityError("authorization_already_used") from exc
        try:
            write_all(descriptor, payload, WriteIo())
            while True:
                try:
                    os.fsync(descriptor)
                    break
                except InterruptedError:
                    continue
        finally:
            os.close(descriptor)
        directory = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
        if self.reserved.read_bytes() != payload:
            raise AuthorityError("reservation_readback")

    def consume(self, outcome: Literal["FAILED", "SUCCEEDED"]) -> None:
        """Irreversibly consume the one attempt."""
        if self.consumed.exists():
            return
        payload = (
            json.dumps(
                {
                    "candidate_digest": self.candidate,
                    "outcome": outcome,
                    "status": "CONSUMED",
                },
                sort_keys=True,
                separators=(",", ":"),
            ).encode()
            + b"\n"
        )
        atomic_write(self.consumed, payload, 0o400)
        self.reserved.unlink(missing_ok=True)

    def outcome(self) -> Literal["FAILED", "SUCCEEDED"] | None:
        """Return a durable consumed outcome without mutating authority."""
        if not self.consumed.is_file():
            return None
        value = _OBJECT.validate_json(self.consumed.read_bytes())
        outcome = value.get("outcome")
        if outcome == "FAILED":
            return "FAILED"
        if outcome == "SUCCEEDED":
            return "SUCCEEDED"
        raise AuthorityError("consumed_outcome")

    def reservation_pending(self) -> bool:
        """Return whether an unconsumed reservation exists."""
        return self.reserved.is_file() and not self.consumed.exists()

    def attempt(self) -> Attempt:
        """Open the consume-on-every-BaseException boundary."""
        return Attempt(self)


@final
class Attempt:
    """Consume after success or any BaseException without broad catches."""

    def __init__(self, ledger: GlobalLedger) -> None:
        self.ledger = ledger

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        error_type: type[BaseException] | None,
        error: BaseException | None,
        traceback: TracebackType | None,
    ) -> Literal[False]:
        del error_type, traceback
        try:
            self.ledger.consume("FAILED" if error is not None else "SUCCEEDED")
        except (AuthorityError, OSError) as consume_error:
            if error is None:
                raise
            error.add_note(f"authorization_consume:{consume_error}")
        return False