"""Concrete installed-store executor and durable execution-root receipts."""

from __future__ import annotations

import hashlib
import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Final, assert_never

from checkin_cli.weekly_operations import (
    CustomerKey,
    WeeklyOperationsError,
    customer_identity_digest,
)
from checkin_cli.weekly_operations_authority import (
    AuthorityId,
    WeeklyOperationsAuthorityRoot,
    construct_authority_binding,
    open_authority_root,
)
from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME
from checkin_cli.weekly_operations_parent import acquire_parent_authority, read_authority_marker
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from pydantic import JsonValue, TypeAdapter, ValidationError

from .contract import SOURCE_ROW_DIGEST, PackageContract, canonical_source_frame
from .controller import provenance_payload
from .durable import publish_atomic
from .executor import canonical_reissue
from .fdio import identity, open_directory_walk, read_pinned_absolute
from .installed import InstalledStoreAdapter
from .store import source_row

_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)


class ProductionError(RuntimeError):
    """Concrete installed execution cannot satisfy its sealed binding."""


def open_installed_authority(path: Path) -> WeeklyOperationsAuthorityRoot:
    """Reconstruct an installed authority capability from its canonical marker."""
    parent = acquire_parent_authority(path)
    marker, marker_digest, marker_info = read_authority_marker(parent.descriptor)
    registry = os.stat(REGISTRY_NAME, dir_fd=parent.descriptor, follow_symlinks=False)
    binding = construct_authority_binding(
        AuthorityId(marker.authority_id),
        os.fstat(parent.descriptor),
        marker_info,
        marker_digest,
        registry,
        marker.registry_digest,
    )
    return open_authority_root(parent, binding)


def _customer_key(contract: PackageContract, path: Path) -> CustomerKey:
    matches = tuple(pin for pin in contract.file_pins if pin.path == path)
    if len(matches) != 1:
        raise ProductionError("customer key pin")
    payload = read_pinned_absolute(matches[0])
    raw = payload.decode("utf-8").strip()
    direct = CustomerKey(raw)
    if (
        raw
        and "\n" not in raw
        and customer_identity_digest(direct) == source_row().customer_identity_digest
    ):
        return direct
    try:
        document: JsonValue = _JSON.validate_json(payload)
    except ValidationError as error:
        raise ProductionError("customer registry bytes") from error
    found = tuple(_matching_customer_keys(document, source_row().customer_identity_digest))
    if len(found) != 1:
        raise ProductionError("customer registry identity")
    return found[0]


def _matching_customer_keys(value: JsonValue, expected: str) -> list[CustomerKey]:
    matches: list[CustomerKey] = []
    match value:
        case dict() as mapping:
            candidates = (
                *mapping.keys(),
                *(item for item in mapping.values() if isinstance(item, str)),
            )
            for raw in candidates:
                candidate = CustomerKey(raw)
                if customer_identity_digest(candidate) == expected:
                    matches.append(candidate)
            for child in mapping.values():
                matches.extend(_matching_customer_keys(child, expected))
        case list() as sequence:
            for child in sequence:
                matches.extend(_matching_customer_keys(child, expected))
        case str() | int() | float() | bool() | None:
            pass
        case unreachable:
            assert_never(unreachable)
    return matches


@dataclass(frozen=True, slots=True)
class InstalledRowExecutor:
    """Open both pinned roots and invoke real WeeklyOperationsStore read/append."""

    contract: PackageContract
    customer_key_file: Path

    def append(self, *, recovery: bool = False) -> None:
        """Perform the sole canonical reissue through installed stores."""
        _verify_mutable_target(self.contract)
        customer = _customer_key(self.contract, self.customer_key_file)
        source: WeeklyOperationsAuthorityRoot | None = None
        target: WeeklyOperationsAuthorityRoot | None = None
        try:
            source = open_installed_authority(self.contract.source_root.path)
            target = open_installed_authority(self.contract.target_root.path)
            source_store = WeeklyOperationsStore.for_authority(source, customer)
            target_store = WeeklyOperationsStore.for_authority(target, customer)
            result = canonical_reissue(
                InstalledStoreAdapter(source_store),
                InstalledStoreAdapter(target_store),
                recovery=recovery,
            )
            if result.row_digest != SOURCE_ROW_DIGEST:
                raise ProductionError("row digest")
        except WeeklyOperationsError as error:
            raise ProductionError("installed store") from error
        finally:
            if source is not None:
                source.close()
            if target is not None:
                target.close()


def _verify_mutable_target(contract: PackageContract) -> None:
    pins = tuple(
        pin
        for pin in contract.file_pins
        if pin.path.parent == contract.target_root.path
        and pin.path.name.endswith(".day-status-v1.jsonl")
    )
    if len(pins) != 1:
        raise ProductionError("target data pin")
    pin = pins[0]
    directory = open_directory_walk(pin.path.parent)
    try:
        descriptor = os.open(
            pin.path.name,
            os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_NONBLOCK,
            dir_fd=directory,
        )
        try:
            opened = os.fstat(descriptor)
            named = os.stat(pin.path.name, dir_fd=directory, follow_symlinks=False)
            expected = (pin.device, pin.inode, pin.uid, pin.gid, pin.mode, pin.nlink)
            if identity(opened) != expected or identity(named) != expected:
                raise ProductionError("target data inode")
            current = os.read(descriptor, len(canonical_source_frame()) + 1)
            if current not in {
                b"",
                canonical_source_frame(),
            } and not canonical_source_frame().startswith(current):
                raise ProductionError("target data divergence")
        finally:
            os.close(descriptor)
    finally:
        os.close(directory)


@dataclass(frozen=True, slots=True)
class DurableReceiptWriter:
    """Persist intent and authenticated cross-root provenance externally."""

    execution_root: Path
    contract: PackageContract

    def write_intent(self, baseline_digest: str) -> None:
        """Publish an idempotent append intent before authority mutation."""
        payload = self._canonical(
            {
                "baseline_digest": baseline_digest,
                "expected_frame_sha256": self.contract.expected_frame_sha256,
                "package_digest": self.contract.package_digest,
                "source_row_digest": self.contract.source_row_digest,
            }
        )
        self._publish_exact("append-intent.json", payload)

    def write_provenance(self) -> None:
        """Publish authenticated source-to-target provenance."""
        intent = (self.execution_root / "append-intent.json").read_bytes()
        payload = provenance_payload(self.contract, hashlib.sha256(intent).hexdigest())
        self._publish_exact("provenance-receipt.json", payload)

    def _publish_exact(self, name: str, payload: bytes) -> None:
        path = self.execution_root / name
        if path.exists():
            if path.read_bytes() != payload:
                raise ProductionError("receipt divergence")
            return
        publish_atomic(self.execution_root, name, payload, checkpoint=name)

    @staticmethod
    def _canonical(value: dict[str, str]) -> bytes:
        return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n"