"""External, lock-linearized runtime authority for Task26 production operations."""

from __future__ import annotations

import asyncio
import fcntl
import hashlib
import json
import os
import stat
import threading
from collections.abc import Callable, Generator, Mapping
from contextlib import AbstractContextManager, contextmanager
from pathlib import Path
from typing import cast, final

from pydantic import JsonValue, TypeAdapter

from .task26_candidate_authority import (
    AuthorityAction,
    CandidateAuthorityPostimages,
    prepare_candidate_authority_postimages,
    publish_candidate_authority_postimages,
    verify_candidate_authority,
)

HEX = set("0123456789abcdef")
_TRANSITION_WAL = "runtime-authority-transition.json"
_OBJECT = TypeAdapter(dict[str, JsonValue])


def canonical(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode()


def digest(value: object) -> str:
    return hashlib.sha256(canonical(value)).hexdigest()


def _write_all(file_descriptor: int, payload: bytes) -> None:
    view = memoryview(payload)
    while view:
        try:
            written = os.write(file_descriptor, view)
        except InterruptedError:
            continue
        if written <= 0:
            raise OSError("runtime authority write made no progress")
        view = view[written:]


def _fsync_directory(path: Path) -> None:
    descriptor = os.open(
        path,
        os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW,
    )
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _atomic_private_json(path: Path, value: object) -> None:
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    descriptor = os.open(
        temporary,
        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW,
        0o600,
    )
    try:
        _write_all(descriptor, canonical(value) + b"\n")
        os.fchmod(descriptor, 0o600)
        os.fsync(descriptor)
    except BaseException:
        temporary.unlink(missing_ok=True)
        raise
    finally:
        os.close(descriptor)
    os.replace(temporary, path)
    _fsync_directory(path.parent)


def _hex(value: object) -> bool:
    return isinstance(value, str) and len(value) == 64 and set(value) <= HEX


def _private(path: Path, *, directory: bool = False) -> None:
    info = path.lstat()
    expected = stat.S_ISDIR if directory else stat.S_ISREG
    modes = {0o700, 0o500} if directory else {0o600, 0o400}
    if (
        stat.S_ISLNK(info.st_mode)
        or not expected(info.st_mode)
        or info.st_uid != os.geteuid()
        or (not directory and info.st_nlink != 1)
        or stat.S_IMODE(info.st_mode) not in modes
    ):
        raise ValueError("runtime authority path is not private")


def _read(path: Path) -> dict[str, object]:
    _private(path)
    try:
        value = _OBJECT.validate_json(path.read_bytes())
    except ValueError as exc:
        raise ValueError("runtime authority document is invalid") from exc
    return cast(dict[str, object], value)


def _snapshot(value: Mapping[str, object]) -> dict[str, object]:
    snapshot = {
        "schema": "task26-runtime-authority-snapshot-v1",
        "source_id": value["source_id"],
        "genesis_sha256": value["genesis_sha256"],
        "candidate_digest": value["current_qualified_candidate"],
        "registry_head_sha256": value["registry_head_sha256"],
        "ledger_head_sha256": value["ledger_head_sha256"],
        "event_count": value["event_count"],
    }
    snapshot["snapshot_sha256"] = digest(snapshot)
    return snapshot


def validate_snapshot(value: object) -> dict[str, object]:
    if not isinstance(value, dict):
        raise ValueError("runtime authority snapshot schema is invalid")
    snapshot = cast(dict[str, object], value)
    if set(snapshot) != {
        "schema",
        "source_id",
        "genesis_sha256",
        "candidate_digest",
        "registry_head_sha256",
        "ledger_head_sha256",
        "event_count",
        "snapshot_sha256",
    }:
        raise ValueError("runtime authority snapshot schema is invalid")
    unsigned = {key: item for key, item in snapshot.items() if key != "snapshot_sha256"}
    if (
        snapshot.get("schema") != "task26-runtime-authority-snapshot-v1"
        or any(
            not _hex(snapshot.get(field))
            for field in (
                "genesis_sha256",
                "candidate_digest",
                "registry_head_sha256",
                "ledger_head_sha256",
            )
        )
        or not isinstance(snapshot.get("source_id"), str)
        or type(snapshot.get("event_count")) is not int
        or cast(int, snapshot["event_count"]) < 1
        or snapshot.get("snapshot_sha256") != digest(unsigned)
    ):
        raise ValueError("runtime authority snapshot binding is invalid")
    return snapshot


@final
class FileCandidateAuthoritySource:
    def __init__(self, pin_path: Path, *, forbidden_roots: tuple[Path, ...]) -> None:
        pin = _read(pin_path.absolute())
        if set(pin) != {
            "schema",
            "authority_root",
            "source_id",
            "genesis_sha256",
            "registry_head_sha256",
            "ledger_head_sha256",
            "event_count",
            "pin_sha256",
        }:
            raise ValueError("runtime authority pin schema is invalid")
        unsigned = {key: item for key, item in pin.items() if key != "pin_sha256"}
        if (
            pin.get("schema") != "task26-authority-pin-v1"
            or pin.get("pin_sha256") != digest(unsigned)
            or not isinstance(pin.get("source_id"), str)
            or any(
                not _hex(pin.get(field))
                for field in (
                    "genesis_sha256",
                    "registry_head_sha256",
                    "ledger_head_sha256",
                )
            )
            or type(pin.get("event_count")) is not int
            or cast(int, pin["event_count"]) < 2
        ):
            raise ValueError("runtime authority pin binding is invalid")
        if not forbidden_roots:
            raise ValueError("runtime authority forbidden roots are required")
        root = Path(str(pin["authority_root"]))
        if not root.is_absolute():
            raise ValueError("runtime authority root must be absolute")
        resolved = root.resolve(strict=True)
        _private(resolved, directory=True)
        for forbidden in forbidden_roots:
            candidate = forbidden.resolve(strict=False)
            if resolved == candidate or resolved.is_relative_to(candidate):
                raise ValueError("runtime authority root is under a forbidden root")
        self.root = resolved
        self.pin = pin
        self.lock_path = resolved / "runtime-authority.lock"
        if not self.lock_path.exists():
            fd = os.open(
                self.lock_path,
                os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_NOFOLLOW,
                0o600,
            )
            os.close(fd)
        self.lock_path.chmod(0o600)
        _ = self._read_current(require_pin=True)

    def _read_current(self, *, require_pin: bool = False) -> dict[str, object]:
        result = verify_candidate_authority(self.root, None)
        registry = _read(self.root / "candidate-authority/registry.json")
        ledger = _read(self.root / "candidate-authority/qualification-ledger.json")
        raw_events = registry.get("events")
        raw_rows = ledger.get("rows")
        if (
            not isinstance(raw_events, list)
            or not raw_events
            or not isinstance(raw_rows, list)
        ):
            raise ValueError("runtime authority event chain is unavailable")
        events = cast(list[object], raw_events)
        rows = cast(list[object], raw_rows)
        source_id = registry.get("source_id")
        genesis = registry.get("genesis_sha256")
        if (
            ledger.get("source_id") != source_id
            or ledger.get("genesis_sha256") != genesis
        ):
            raise ValueError("paired runtime authority source differs")
        current: dict[str, object] = {
            **result,
            "source_id": source_id,
            "genesis_sha256": genesis,
            "event_count": len(events),
            "events": events,
            "rows": rows,
        }
        if source_id != self.pin["source_id"] or genesis != self.pin["genesis_sha256"]:
            raise ValueError("runtime authority source or genesis differs")
        if require_pin and (
            len(events) != self.pin["event_count"]
            or result["registry_head_sha256"] != self.pin["registry_head_sha256"]
            or result["ledger_head_sha256"] != self.pin["ledger_head_sha256"]
        ):
            raise ValueError("runtime authority pin is stale")
        return current

    @contextmanager
    def authorize(
        self,
        candidate: str,
        stage: str,
        predecessor: object | None = None,
    ) -> Generator[dict[str, object]]:
        if not _hex(candidate) or stage not in {
            "activation",
            "service_activation",
            "capability_issue",
            "transport",
            "update_processing",
            "background_generation",
            "background_recovery",
            "scheduled_transport",
        }:
            raise ValueError("runtime authority request is invalid")
        previous = validate_snapshot(predecessor) if predecessor is not None else None
        with self.lock_path.open("r+") as lock:
            fcntl.flock(lock.fileno(), fcntl.LOCK_SH)
            try:
                current = self._read_current(require_pin=previous is None)
                if current.get(
                    "current_qualified_candidate"
                ) != candidate or candidate in cast(
                    list[str], current.get("invalidated_candidate_digests", [])
                ):
                    raise ValueError("runtime candidate is not current or was revoked")
                if previous is not None:
                    if (
                        previous["source_id"] != current["source_id"]
                        or previous["genesis_sha256"] != current["genesis_sha256"]
                        or previous["candidate_digest"] != candidate
                        or cast(int, previous["event_count"])
                        > cast(int, current["event_count"])
                    ):
                        raise ValueError("runtime authority predecessor is stale")
                    index = cast(int, previous["event_count"]) - 1
                    events = cast(list[dict[str, object]], current["events"])
                    rows = cast(list[dict[str, object]], current["rows"])
                    if (
                        events[index].get("event_sha256")
                        != previous["registry_head_sha256"]
                        or rows[index].get("row_sha256")
                        != previous["ledger_head_sha256"]
                    ):
                        raise ValueError(
                            "runtime authority rollback or successor mismatch"
                        )
                yield _snapshot(current)
            finally:
                fcntl.flock(lock.fileno(), fcntl.LOCK_UN)

    async def acquire_operation_lease(
        self,
        candidate: str,
        stage: str,
        predecessor: object,
    ) -> AsyncAuthorityLease:
        """Acquire one verified shared lease without blocking the event loop."""
        manager = self.authorize(candidate, stage, predecessor)
        acquisition = asyncio.create_task(
            asyncio.to_thread(_enter_authority_lease, manager)
        )
        try:
            return await asyncio.shield(acquisition)
        except asyncio.CancelledError:
            try:
                lease = await acquisition
            except Exception:
                pass
            else:
                await lease.release()
            raise

    async def acquire_update_lease(
        self,
        candidate: str,
        predecessor: object,
    ) -> AsyncAuthorityLease:
        """Acquire and verify one update's shared authority lease off-loop."""
        return await self.acquire_operation_lease(
            candidate,
            "update_processing",
            predecessor,
        )


@final
class AsyncAuthorityLease:
    """A verified shared lock held for one complete asynchronous update."""

    def __init__(
        self,
        manager: AbstractContextManager[dict[str, object]],
        snapshot: dict[str, object],
    ) -> None:
        self._manager = manager
        self.snapshot = snapshot
        self._released = False

    async def release(self) -> None:
        if self._released:
            return
        self._released = True
        release = asyncio.create_task(
            asyncio.to_thread(self._manager.__exit__, None, None, None)
        )
        try:
            _ = await asyncio.shield(release)
        except asyncio.CancelledError:
            _ = await release
            raise


CandidateAuthoritySource = FileCandidateAuthoritySource


def _enter_authority_lease(
    manager: AbstractContextManager[dict[str, object]],
) -> AsyncAuthorityLease:
    return AsyncAuthorityLease(manager, manager.__enter__())


@final
class ExternalAuthorityWatcher:
    """One-resource, event-driven monitor for the paired external authority."""

    def __init__(
        self,
        source: FileCandidateAuthoritySource,
        candidate: str,
        predecessor: Mapping[str, object],
        loop: asyncio.AbstractEventLoop,
        on_affirm: Callable[[dict[str, object]], None],
        on_failure: Callable[[str], None],
    ) -> None:
        self.source = source
        self.candidate = candidate
        self.snapshot = validate_snapshot(dict(predecessor))
        self.loop = loop
        self.on_affirm = on_affirm
        self.on_failure = on_failure
        self.resources: dict[str, object] = {}
        self._armed = threading.Event()
        self._backend: object | None = None
        self._setup_error: BaseException | None = None
        self._thread = threading.Thread(
            target=self._run,
            name="task26-external-authority-watcher",
            daemon=False,
        )

    def arm(self) -> dict[str, object]:
        self._thread.start()
        if not self._armed.wait(timeout=2):
            self.request_close()
            self._thread.join(timeout=2)
            raise ValueError("runtime authority watcher arm timed out")
        if self._setup_error is not None:
            raise ValueError(
                "runtime authority watcher setup failed"
            ) from self._setup_error
        return dict(self.resources)

    def request_close(self) -> None:
        backend = self._backend
        request = getattr(backend, "request_close", None)
        if callable(request):
            _ = request()

    def close(self) -> None:
        self.request_close()
        if self._thread.is_alive():
            self._thread.join(timeout=2)
        if self._thread.is_alive():
            raise RuntimeError("runtime authority watcher did not stop")

    @property
    def is_alive(self) -> bool:
        return self._thread.is_alive()

    def _notify(self, callback: Callable[..., object], *args: object) -> bool:
        if self.loop.is_closed():
            return False
        try:
            _ = self.loop.call_soon_threadsafe(callback, *args)
        except RuntimeError:
            if self.loop.is_closed():
                return False
            raise
        return True

    def _run(self) -> None:
        from gateway.commit_observer import DnotifySignalfdDirectoryWatcher

        authority_directory = self.source.root / "candidate-authority"
        backend = DnotifySignalfdDirectoryWatcher(
            authority_directory,
            (
                authority_directory / "registry.json",
                authority_directory / "qualification-ledger.json",
            ),
        )
        self._backend = backend
        try:
            self.resources = backend.arm()
        except BaseException as exc:
            self._setup_error = exc
            _ = self._armed.set()
            return
        _ = self._armed.set()
        try:
            while backend.wait():
                try:
                    with self.source.authorize(
                        self.candidate,
                        "service_activation",
                        self.snapshot,
                    ) as current:
                        self.snapshot = current
                    if not self._notify(self.on_affirm, current):
                        return
                except BaseException as exc:
                    _ = self._notify(
                        self.on_failure,
                        f"{type(exc).__name__}: {exc}",
                    )
                    return
        except BaseException as exc:
            _ = self._notify(
                self.on_failure,
                f"watcher_failure:{type(exc).__name__}:{exc}",
            )
        finally:
            _ = backend.close_owner()


def _credential_path(name: str, environment_name: str) -> Path:
    explicit = os.environ.get(environment_name)
    if explicit:
        path = Path(explicit)
    else:
        credential_directory = os.environ.get("CREDENTIALS_DIRECTORY")
        if not credential_directory:
            raise ValueError(f"{name} credential is unavailable")
        path = Path(credential_directory) / name
    if not path.is_absolute():
        raise ValueError(f"{name} credential path must be absolute")
    return path


def load_task26_production_authority(
    *,
    profile_root: Path,
    package_root: Path,
    additional_forbidden_roots: tuple[Path, ...] = (),
) -> tuple[FileCandidateAuthoritySource, str]:
    """Resolve production authority only from env/systemd credential inputs."""
    pin_path = _credential_path("task26-authority-pin.json", "TASK26_AUTHORITY_PIN")
    candidate_path = _credential_path(
        "task26-candidate-digest", "TASK26_CANDIDATE_DIGEST_FILE"
    )
    _private(candidate_path)
    candidate = candidate_path.read_text(encoding="utf-8").strip()
    if not _hex(candidate):
        raise ValueError("Task26 candidate digest credential is invalid")
    configured_roots: list[Path] = [profile_root, package_root]
    for name in (
        "TASK26_CANDIDATE_ROOT",
        "TASK26_WHEELHOUSE_ROOT",
        "TASK26_RECEIPT_ROOT",
    ):
        value = os.environ.get(name)
        if value:
            path = Path(value)
            if not path.is_absolute():
                raise ValueError(f"{name} must be absolute")
            configured_roots.append(path)
    configured_roots.extend(additional_forbidden_roots)
    source = FileCandidateAuthoritySource(
        pin_path,
        forbidden_roots=tuple(configured_roots),
    )
    with source.authorize(candidate, "activation"):
        pass
    return source, candidate


def _file_sha256(path: Path) -> str | None:
    if not path.exists():
        return None
    _private(path)
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _bind_authority_source(
    postimages: CandidateAuthorityPostimages,
    *,
    source_id: str,
    existing_registry: Mapping[str, object] | None,
) -> CandidateAuthorityPostimages:
    registry = dict(postimages.registry)
    ledger = dict(postimages.ledger)
    existing_source = (
        existing_registry.get("source_id") if existing_registry is not None else None
    )
    existing_genesis = (
        existing_registry.get("genesis_sha256")
        if existing_registry is not None
        else None
    )
    if existing_source is not None and existing_source != source_id:
        raise ValueError("runtime authority source differs")
    events = registry.get("events")
    if not isinstance(events, list) or not events:
        raise ValueError("runtime authority event chain is unavailable")
    event_values = cast(list[object], events)
    first_value = event_values[0]
    if not isinstance(first_value, dict):
        raise ValueError("runtime authority genesis event is invalid")
    first_event = cast(dict[str, object], first_value)
    genesis = existing_genesis or first_event.get("event_sha256")
    if not _hex(genesis):
        raise ValueError("runtime authority genesis is invalid")
    for document in (registry, ledger):
        document["source_id"] = source_id
        document["genesis_sha256"] = genesis
        document["document_sha256"] = digest({
            key: value for key, value in document.items() if key != "document_sha256"
        })
    return CandidateAuthorityPostimages(
        registry=registry,
        ledger=ledger,
        result={
            **postimages.result,
            "source_id": source_id,
            "genesis_sha256": genesis,
        },
    )


def _transition_intent(
    root: Path,
    postimages: CandidateAuthorityPostimages,
) -> dict[str, object]:
    registry_path = root / "candidate-authority/registry.json"
    ledger_path = root / "candidate-authority/qualification-ledger.json"
    unsigned: dict[str, object] = {
        "schema": "task26-runtime-authority-transition-v1",
        "pre_registry_file_sha256": _file_sha256(registry_path),
        "pre_ledger_file_sha256": _file_sha256(ledger_path),
        "post_registry": postimages.registry,
        "post_ledger": postimages.ledger,
        "post_registry_file_sha256": hashlib.sha256(
            canonical(postimages.registry) + b"\n"
        ).hexdigest(),
        "post_ledger_file_sha256": hashlib.sha256(
            canonical(postimages.ledger) + b"\n"
        ).hexdigest(),
    }
    return {**unsigned, "intent_sha256": digest(unsigned)}


def _validated_transition_intent(path: Path) -> dict[str, object]:
    intent = _read(path)
    unsigned = {key: value for key, value in intent.items() if key != "intent_sha256"}
    if (
        intent.get("schema") != "task26-runtime-authority-transition-v1"
        or intent.get("intent_sha256") != digest(unsigned)
        or not isinstance(intent.get("post_registry"), dict)
        or not isinstance(intent.get("post_ledger"), dict)
    ):
        raise ValueError("runtime authority transition intent is invalid")
    for name in (
        "pre_registry_file_sha256",
        "pre_ledger_file_sha256",
        "post_registry_file_sha256",
        "post_ledger_file_sha256",
    ):
        value = intent.get(name)
        if value is not None and not _hex(value):
            raise ValueError("runtime authority transition digest is invalid")
    return intent


def _recover_transition_locked(root: Path) -> dict[str, object] | None:
    wal = root / _TRANSITION_WAL
    if not wal.exists():
        return None
    intent = _validated_transition_intent(wal)
    registry_path = root / "candidate-authority/registry.json"
    ledger_path = root / "candidate-authority/qualification-ledger.json"
    registry_digest = _file_sha256(registry_path)
    ledger_digest = _file_sha256(ledger_path)
    accepted_registry = {
        intent["pre_registry_file_sha256"],
        intent["post_registry_file_sha256"],
    }
    accepted_ledger = {
        intent["pre_ledger_file_sha256"],
        intent["post_ledger_file_sha256"],
    }
    if registry_digest not in accepted_registry or ledger_digest not in accepted_ledger:
        raise ValueError("runtime authority transition requires manual recovery")
    postimages = CandidateAuthorityPostimages(
        registry=cast(dict[str, object], intent["post_registry"]),
        ledger=cast(dict[str, object], intent["post_ledger"]),
        result={},
    )
    publish_candidate_authority_postimages(root, postimages)
    if (
        _file_sha256(registry_path) != intent["post_registry_file_sha256"]
        or _file_sha256(ledger_path) != intent["post_ledger_file_sha256"]
    ):
        raise ValueError("runtime authority transition recovery did not converge")
    result = verify_candidate_authority(root, None)
    wal.unlink()
    _fsync_directory(root)
    return result


def recover_external_authority(root: Path) -> dict[str, object] | None:
    """Converge one interrupted paired append under the external lock."""
    lock_path = root / "runtime-authority.lock"
    with lock_path.open("a+") as lock:
        lock_path.chmod(0o600)
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        try:
            return _recover_transition_locked(root)
        finally:
            fcntl.flock(lock.fileno(), fcntl.LOCK_UN)


def build_runtime_authority_pin(root: Path) -> dict[str, object]:
    """Derive the exact canonical pin for the current paired-chain heads."""
    current = verify_candidate_authority(root, None)
    registry = _read(root / "candidate-authority/registry.json")
    raw_events = registry.get("events")
    if not isinstance(raw_events, list):
        raise ValueError("runtime authority event chain is unavailable")
    events = cast(list[object], raw_events)
    unsigned: dict[str, object] = {
        "schema": "task26-authority-pin-v1",
        "authority_root": str(root.resolve(strict=True)),
        "source_id": registry["source_id"],
        "genesis_sha256": registry["genesis_sha256"],
        "registry_head_sha256": current["registry_head_sha256"],
        "ledger_head_sha256": current["ledger_head_sha256"],
        "event_count": len(events),
    }
    return {**unsigned, "pin_sha256": digest(unsigned)}


def publish_runtime_authority_pin(
    path: Path,
    pin: Mapping[str, object],
) -> None:
    """Durably publish one canonical runtime-authority pin."""
    validate = dict(pin)
    unsigned = {key: value for key, value in validate.items() if key != "pin_sha256"}
    if validate.get("schema") != "task26-authority-pin-v1" or validate.get(
        "pin_sha256"
    ) != digest(unsigned):
        raise ValueError("runtime authority pin binding is invalid")
    _atomic_private_json(path, validate)


def append_external_authority(
    root: Path,
    *,
    source_id: str,
    candidate_digest: str,
    action: str,
    historical_pass_digest: str,
    reason: str,
) -> dict[str, object]:
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    root.chmod(0o700)
    lock_path = root / "runtime-authority.lock"
    with lock_path.open("a+") as lock:
        lock_path.chmod(0o600)
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        try:
            _ = _recover_transition_locked(root)
            if action not in {"qualify", "revoke"}:
                raise ValueError("runtime authority action is invalid")
            typed_action: AuthorityAction = (
                "qualify" if action == "qualify" else "revoke"
            )
            registry_path = root / "candidate-authority/registry.json"
            existing_registry = _read(registry_path) if registry_path.exists() else None
            prepared = prepare_candidate_authority_postimages(
                root,
                candidate_digest=candidate_digest,
                action=typed_action,
                historical_pass_digest=historical_pass_digest,
                reason=reason,
            )
            postimages = _bind_authority_source(
                prepared,
                source_id=source_id,
                existing_registry=existing_registry,
            )
            intent = _transition_intent(root, postimages)
            _atomic_private_json(root / _TRANSITION_WAL, intent)
            publish_candidate_authority_postimages(root, postimages)
            verified = verify_candidate_authority(
                root,
                candidate_digest if typed_action == "qualify" else None,
            )
            if (
                verified["registry_head_sha256"]
                != postimages.result["registry_head_sha256"]
                or verified["ledger_head_sha256"]
                != postimages.result["ledger_head_sha256"]
            ):
                raise ValueError("runtime authority transition postimage differs")
            (root / _TRANSITION_WAL).unlink()
            _fsync_directory(root)
            return postimages.result
        finally:
            fcntl.flock(lock.fileno(), fcntl.LOCK_UN)