"""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 contextlib import contextmanager
from pathlib import Path
from typing import Callable, Iterator, Mapping, cast

from .task26_candidate_authority import (
    AuthorityAction,
    append_candidate_authority,
    verify_candidate_authority,
)

HEX = set("0123456789abcdef")


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 _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)
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("runtime authority document is invalid")
    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) or set(value) != {"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")
    snapshot = cast(dict[str, object], value)
    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


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")
        events = registry.get("events")
        if not isinstance(events, list) or not events:
            raise ValueError("runtime authority event chain is unavailable")
        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 = {
            **result,
            "source_id": source_id,
            "genesis_sha256": genesis,
            "event_count": len(events),
            "events": events,
            "rows": ledger.get("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) -> Iterator[dict[str, object]]:
        if not _hex(candidate) or stage not in {
            "activation",
            "service_activation",
            "capability_issue",
            "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)


CandidateAuthoritySource = FileCandidateAuthoritySource


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 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:
            if action not in {"qualify", "revoke"}:
                raise ValueError("runtime authority action is invalid")
            typed_action: AuthorityAction = (
                "qualify" if action == "qualify" else "revoke"
            )
            result = append_candidate_authority(
                root,
                candidate_digest=candidate_digest,
                action=typed_action,
                historical_pass_digest=historical_pass_digest,
                reason=reason,
            )
            registry_path = root / "candidate-authority/registry.json"
            ledger_path = root / "candidate-authority/qualification-ledger.json"
            registry = json.loads(registry_path.read_text())
            ledger = json.loads(ledger_path.read_text())
            genesis = registry.get("genesis_sha256") or registry["events"][0]["event_sha256"]
            for document, path in ((registry, registry_path), (ledger, ledger_path)):
                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"})
                path.write_bytes(canonical(document) + b"\n")
                path.chmod(0o600)
            return {**result, "source_id": source_id, "genesis_sha256": genesis}
        finally:
            fcntl.flock(lock.fileno(), fcntl.LOCK_UN)