"""Explicit, sanitized r71b maintenance package preparation."""

from __future__ import annotations

import hashlib
import stat
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Final, Protocol

import yaml
from yaml.constructor import ConstructorError
from yaml.events import Event
from yaml.nodes import Node
from pydantic import JsonValue, TypeAdapter, ValidationError
from typing_extensions import override

from gateway.platforms.nutrition_weekly_maintenance_contract import (
    Topic59MaintenanceContractError,
    Topic59MaintenanceScopeV1,
    canonical_document,
    maintenance_scope_digest,
)
from gateway.platforms.nutrition_weekly_operations_config import (
    WeeklyOperationsConfig,
    WeeklyOperationsConfigError,
    parse_weekly_operations_config,
)
from gateway.platforms.nutrition_weekly_operations_identifiers import (
    TOPIC59_PROJECTION_SCHEMA,
)
from gateway.platforms.nutrition_weekly_operations_ledger_history import (
    Topic59LedgerEntry,
)
from gateway.platforms.nutrition_weekly_operations_publication_contract import (
    Topic59LedgerState,
)
from scripts.nutricoach_v150_r71b_maintenance_transaction import (
    KNOWN_R70_CRON_ERROR,
    KNOWN_R70_CRON_JOB_ID,
    KNOWN_R70_CRON_JOB_NAME,
    KNOWN_R70_CRON_SCHEDULE,
    KNOWN_R70_RUNTIME_CANDIDATE_DIGEST,
)
from scripts.nutricoach_v150_r71b_task10_evidence import (
    CandidateTask10EvidenceCopy,
    Task10EvidenceError,
    candidate_task10_provenance,
    load_candidate_task10_evidence,
    task10_evidence_spec,
)
from scripts.nutricoach_v150_r71b_package import (
    PackageDerivationError,
    R71bPackageArtifacts,
    derive_r71b_package,
)

_ORACLE_SCHEMA: Final = "nutricoach-r71b-no-send-oracle-v1"
_SCOPE_PATH: Final = "maintenance/maintenance-scope.json"
_ORACLE_PATH: Final = "maintenance/no-send-oracle.json"
MAX_PROFILE_YAML_BYTES: Final[int] = 1_048_576
_OBJECT = TypeAdapter(dict[str, JsonValue])


class _YamlComposer(Protocol):
    @property
    def anchors(self) -> Mapping[str, Node]: ...

    def get_event(self) -> Event | None: ...

    def compose_node(self, parent: Node | None, index: int) -> Node | None: ...


class _YamlConstructor(Protocol):
    def construct_object(self, node: Node, deep: bool = False) -> JsonValue: ...


class _YamlMappingNode(Protocol):
    value: list[tuple[Node, Node]]


YamlKey = str | int | float | bool | bytes | None


class _StrictSafeLoader(yaml.SafeLoader):
    """Safe YAML loader that refuses duplicate keys and graph references."""

    @override
    def compose_document(self) -> Node | None:
        return _compose_document(self)

    @override
    def construct_document(self, node: Node) -> JsonValue:
        return _construct_document(self, node)



def _compose_document(loader: _YamlComposer) -> Node | None:
    _ = loader.get_event()
    node = loader.compose_node(None, 0)
    _ = loader.get_event()
    if loader.anchors:
        raise ConstructorError(None, None, "anchors and aliases are forbidden")
    return node



def _construct_document(loader: _YamlConstructor, node: Node) -> JsonValue:
    return loader.construct_object(node, deep=True)



def _construct_unique_mapping(
    loader: _YamlConstructor,
    node: _YamlMappingNode,
    deep: bool = False,
) -> dict[YamlKey, JsonValue]:
    mapping: dict[YamlKey, JsonValue] = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        if not isinstance(key, (str, int, float, bool, bytes)) and key is not None:
            raise ConstructorError(None, None, "unhashable mapping key")
        if key in mapping:
            raise ConstructorError(None, None, "duplicate mapping key")
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


_StrictSafeLoader.add_constructor(
    "tag:yaml.org,2002:map",
    _construct_unique_mapping,
)


@dataclass(frozen=True, slots=True)
class MaintenancePreparationError(Exception):
    """An explicit maintenance package input is incomplete or unsafe."""

    reason: str

    @override
    def __str__(self) -> str:
        return self.reason


@dataclass(frozen=True, slots=True)
class MaintenanceWindow:
    """Explicit KST first-use window; there is deliberately no current-time default."""

    kst_day: date
    not_before: datetime
    expires_at: datetime


def parse_maintenance_window(
    kst_day: str,
    not_before: str,
    expires_at: str,
) -> MaintenanceWindow:
    """Parse the three mandatory KST CLI values without a current-time fallback."""
    try:
        window = MaintenanceWindow(
            date.fromisoformat(kst_day),
            datetime.fromisoformat(not_before),
            datetime.fromisoformat(expires_at),
        )
    except ValueError as error:
        raise MaintenancePreparationError("maintenance_window") from error
    if (
        window.not_before.utcoffset() != timedelta(hours=9)
        or window.expires_at.utcoffset() != timedelta(hours=9)
        or window.not_before.date() != window.kst_day
        or window.expires_at.date() != window.kst_day
        or window.not_before >= window.expires_at
    ):
        raise MaintenancePreparationError("maintenance_window")
    return window


@dataclass(frozen=True, slots=True)
class HistoricalSendingRow:
    """One sealed historical uncertain row that is never retried or rewritten."""

    card_slot: str
    entry_digest: str


@dataclass(frozen=True, slots=True)
class SanitizedTopic59Preflight:
    """Exact opaque current projection and unchanged-ledger evidence."""

    candidate_digest: str
    successor_config_digest: str
    route_digest: str
    customer_identity_digest: str
    card_slot: str
    publication_ledger_sha256: str
    historical_sending_rows: tuple[HistoricalSendingRow, ...]


@dataclass(frozen=True, slots=True)
class NoSendFacts:
    """The complete due-work oracle, expressed only as sealed counts."""

    non_topic59_due_send_count: int
    non_topic59_due_edit_count: int
    activation_notice_due_count: int
    owner_delivery_due_count: int
    target_topic59_projection_count: int


@dataclass(frozen=True, slots=True)
class AuthoritativeR71bPreflight:
    """Preflight facts reconstructed from profile and candidate artifacts only."""

    preflight: SanitizedTopic59Preflight
    facts: NoSendFacts
    evidence: dict[str, JsonValue]


@dataclass(frozen=True, slots=True)
class PreparedR71bMaintenance:
    """All deterministic derivation outputs retained before any package write."""

    scope: Topic59MaintenanceScopeV1
    oracle: dict[str, JsonValue]
    artifacts: R71bPackageArtifacts


def prepare_r71b_maintenance(
    preflight: SanitizedTopic59Preflight,
    window: MaintenanceWindow,
    facts: NoSendFacts,
    binding_seed: Mapping[str, JsonValue],
) -> PreparedR71bMaintenance:
    """Derive scope, complete no-send oracle, hold, authority, and both package digests."""
    _require_complete_oracle(preflight, facts)
    try:
        scope = Topic59MaintenanceScopeV1(
            candidate_digest=preflight.candidate_digest,
            config_digest=preflight.successor_config_digest,
            route_digest=preflight.route_digest,
            customer_identity_digest=preflight.customer_identity_digest,
            card_slot=preflight.card_slot,
            kst_day=window.kst_day,
            not_before=window.not_before,
            expires_at=window.expires_at,
        )
    except Topic59MaintenanceContractError as error:
        raise MaintenancePreparationError("maintenance_window") from error
    oracle = _oracle_document(preflight, scope, facts)
    scope_bytes = canonical_document(scope)
    oracle_bytes = _canonical_oracle(oracle)
    binding = dict(binding_seed)
    binding.update({
        "maintenance_scope": scope.model_dump(mode="json", by_alias=True),
        "maintenance_scope_path": _SCOPE_PATH,
        "maintenance_scope_file_sha256": hashlib.sha256(scope_bytes).hexdigest(),
        "maintenance_scope_digest": maintenance_scope_digest(scope),
        "no_send_oracle": oracle,
        "no_send_oracle_path": _ORACLE_PATH,
        "no_send_oracle_file_sha256": hashlib.sha256(oracle_bytes).hexdigest(),
    })
    try:
        artifacts = derive_r71b_package(binding, scope)
    except PackageDerivationError as error:
        raise MaintenancePreparationError("binding_inputs") from error
    return PreparedR71bMaintenance(scope, oracle, artifacts)


def collect_authoritative_r71b_preflight(
    candidate_root: Path,
    profile_root: Path,
    window: MaintenanceWindow,
) -> AuthoritativeR71bPreflight:
    """Derive the maintenance facts from current immutable candidate/profile files.

    This deliberately has no fact parameters.  A missing or ambiguous source is a
    denial, rather than an opportunity to substitute a caller-supplied digest or
    no-send count.
    """
    candidate = _load_object(candidate_root / "manifest.json", "candidate_manifest")
    candidate_digest = _required_digest(candidate, "candidate_digest", "candidate")
    config = _load_profile_config(profile_root / "config.yaml")
    ledger_path = profile_root / "data" / "weekly-operations-topic59.jsonl"
    ledger_bytes = _read_regular(ledger_path, "publication_ledger")
    rows = _load_topic59_rows(ledger_bytes)
    targets = [row for row in rows if row.kst_day == window.kst_day.isoformat()]
    if len(targets) != 1:
        raise MaintenancePreparationError("target_topic59_projection")
    target = targets[0]
    if (
        target.candidate_digest != candidate_digest
        or target.config_digest != config.digest
        or config.review_route is None
        or target.route_digest != _route_digest(config.review_route.key)
    ):
        raise MaintenancePreparationError("target_topic59_projection")
    historical = tuple(
        HistoricalSendingRow(row.card_slot, row.entry_digest)
        for row in rows
        if row.state is Topic59LedgerState.SENDING
    )
    facts = _collect_no_send_facts(profile_root, target.card_slot, window.kst_day)
    try:
        r70_evidence = load_candidate_task10_evidence(
            candidate_root,
            "inputs/r70-canonical-authority-drift.json",
            task10_evidence_spec("r70_error"),
        )
        observer = load_candidate_task10_evidence(
            candidate_root,
            "inputs/r71b-observer-final.json",
            task10_evidence_spec("observer"),
        )
        fixed_collector = load_candidate_task10_evidence(
            candidate_root,
            "inputs/r71b-fixed-collector.json",
            task10_evidence_spec("health_recovery"),
        )
    except Task10EvidenceError as error:
        raise MaintenancePreparationError("task10_evidence") from error
    preflight = SanitizedTopic59Preflight(
        candidate_digest=candidate_digest,
        successor_config_digest=config.digest,
        route_digest=target.route_digest,
        customer_identity_digest=target.customer_identity_digest,
        card_slot=target.card_slot,
        publication_ledger_sha256=hashlib.sha256(ledger_bytes).hexdigest(),
        historical_sending_rows=historical,
    )
    evidence = _OBJECT.validate_python({
        "r70_error_evidence_path": r70_evidence.source_path,
        "r70_error_evidence_sha256": r70_evidence.source_sha256,
        "observer_evidence": _candidate_evidence_document(observer),
        "fixed_collector_evidence": _candidate_evidence_document(fixed_collector),
        "task10_evidence_provenance": candidate_task10_provenance((
            r70_evidence,
            observer,
            fixed_collector,
        )),
        "publication_ledger_path": str(ledger_path),
        "publication_ledger_sha256": preflight.publication_ledger_sha256,
        "scheduled_delivery_ledger": _scheduled_delivery_summary(profile_root),
        "r70_cron": _r70_cron_proof(profile_root / "cron/jobs.json"),
    })
    return AuthoritativeR71bPreflight(preflight, facts, evidence)


def _load_profile_config(path: Path) -> WeeklyOperationsConfig:
    try:
        document = parse_profile_yaml(_read_regular(path, "profile_config"))
    except ValidationError as error:
        raise MaintenancePreparationError("profile_config") from error
    platforms = document.get("platforms")
    if not isinstance(platforms, dict):
        raise MaintenancePreparationError("profile_config")
    telegram = platforms.get("telegram")
    if not isinstance(telegram, dict):
        raise MaintenancePreparationError("profile_config")
    extra = telegram.get("extra")
    if not isinstance(extra, dict):
        raise MaintenancePreparationError("profile_config")
    try:
        return parse_weekly_operations_config(extra)
    except WeeklyOperationsConfigError as error:
        raise MaintenancePreparationError("profile_config") from error


def parse_profile_yaml(payload: bytes) -> dict[str, JsonValue]:
    """Load profile YAML through a bounded, safe, strictly narrowed boundary."""
    if len(payload) > MAX_PROFILE_YAML_BYTES:
        raise MaintenancePreparationError("profile_config")
    try:
        text = payload.decode("utf-8")
    except UnicodeDecodeError as error:
        raise MaintenancePreparationError("profile_config") from error
    try:
        return _OBJECT.validate_python(
            yaml.load(text, Loader=_StrictSafeLoader), strict=True
        )
    except (yaml.YAMLError, ValidationError) as error:
        raise MaintenancePreparationError("profile_config") from error


_simple_yaml_mapping = parse_profile_yaml


def _load_topic59_rows(payload: bytes) -> tuple[Topic59LedgerEntry, ...]:
    try:
        return tuple(
            Topic59LedgerEntry.model_validate_json(line)
            for line in payload.splitlines()
            if line
        )
    except ValidationError as error:
        raise MaintenancePreparationError("publication_ledger") from error


def _collect_no_send_facts(
    profile_root: Path,
    target_slot: str,
    kst_day: date,
) -> NoSendFacts:
    cron = _load_object(profile_root / "cron/jobs.json", "cron_jobs")
    jobs = cron.get("jobs")
    if not isinstance(jobs, list) or not all(isinstance(job, dict) for job in jobs):
        raise MaintenancePreparationError("cron_jobs")
    non_target_enabled: list[dict[str, JsonValue]] = []
    for job in jobs:
        if not isinstance(job, dict):
            raise MaintenancePreparationError("cron_jobs")
        if job.get("id") != KNOWN_R70_CRON_JOB_ID and job.get("enabled") is True:
            non_target_enabled.append(job)
    scheduled = _scheduled_delivery_summary(profile_root)
    outstanding = scheduled["nonterminal_count"]
    if not isinstance(outstanding, int):
        raise MaintenancePreparationError("scheduled_delivery_ledger")
    # The only allowed first tick is the one exact Topic-59 projection.  Each
    # other enabled job or nonterminal scheduled delivery is treated as due.
    return NoSendFacts(
        non_topic59_due_send_count=len(non_target_enabled),
        non_topic59_due_edit_count=0,
        activation_notice_due_count=outstanding,
        owner_delivery_due_count=0,
        target_topic59_projection_count=1 if target_slot and kst_day else 0,
    )


def _scheduled_delivery_summary(profile_root: Path) -> dict[str, JsonValue]:
    path = profile_root / "data/scheduled-deliveries.jsonl"
    payload = _read_regular(path, "scheduled_delivery_ledger")
    rows = [_load_json_line(line, "scheduled_delivery_ledger") for line in payload.splitlines() if line]
    nonterminal = sum(
        1
        for row in rows
        if row.get("state") not in {"sent", "cancelled", "expired", "failed"}
    )
    return {
        "path": str(path),
        "sha256": hashlib.sha256(payload).hexdigest(),
        "row_count": len(rows),
        "nonterminal_count": nonterminal,
    }


def _r70_cron_proof(path: Path) -> dict[str, JsonValue]:
    document = _load_object(path, "cron_jobs")
    rows = document.get("jobs")
    if not isinstance(rows, list):
        raise MaintenancePreparationError("cron_jobs")
    matches = [row for row in rows if isinstance(row, dict) and row.get("id") == KNOWN_R70_CRON_JOB_ID]
    if len(matches) != 1:
        raise MaintenancePreparationError("r70_cron")
    row = matches[0]
    if (
        row.get("name") != KNOWN_R70_CRON_JOB_NAME
        or row.get("schedule_display") != KNOWN_R70_CRON_SCHEDULE
        or row.get("last_error") != KNOWN_R70_CRON_ERROR
    ):
        raise MaintenancePreparationError("r70_cron")
    return {
        "jobs_path": str(path),
        "jobs_sha256": hashlib.sha256(_read_regular(path, "cron_jobs")).hexdigest(),
        "job_id": KNOWN_R70_CRON_JOB_ID,
        "runtime_candidate_digest": KNOWN_R70_RUNTIME_CANDIDATE_DIGEST,
    }


def _candidate_evidence_document(
    evidence: CandidateTask10EvidenceCopy,
) -> dict[str, JsonValue]:
    return {
        "path": evidence.source_path,
        "sha256": evidence.source_sha256,
        "schema": evidence.source_schema,
        "status": evidence.source_status,
    }


def _load_object(path: Path, label: str) -> dict[str, JsonValue]:
    try:
        return _OBJECT.validate_json(_read_regular(path, label))
    except ValidationError as error:
        raise MaintenancePreparationError(label) from error


def _load_json_line(line: bytes, label: str) -> dict[str, JsonValue]:
    try:
        return _OBJECT.validate_json(line)
    except ValidationError as error:
        raise MaintenancePreparationError(label) from error


def _read_regular(path: Path, label: str) -> bytes:
    try:
        info = path.stat(follow_symlinks=False)
        if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
            raise MaintenancePreparationError(label)
        return path.read_bytes()
    except OSError as error:
        raise MaintenancePreparationError(label) from error


def _required_digest(
    document: Mapping[str, JsonValue],
    key: str,
    label: str,
) -> str:
    value = document.get(key)
    if not isinstance(value, str) or not _is_digest(value):
        raise MaintenancePreparationError(label)
    return value


def _route_digest(route: tuple[str, str, str]) -> str:
    material = "\0".join((
        TOPIC59_PROJECTION_SCHEMA, "route", *route,
    )).encode("utf-8")
    return hashlib.sha256(material).hexdigest()


def _require_complete_oracle(
    preflight: SanitizedTopic59Preflight,
    facts: NoSendFacts,
) -> None:
    if (
        facts.non_topic59_due_send_count != 0
        or facts.non_topic59_due_edit_count != 0
        or facts.activation_notice_due_count != 0
        or facts.owner_delivery_due_count != 0
        or facts.target_topic59_projection_count != 1
        or len(preflight.historical_sending_rows) != 2
        or not all(_is_digest(value) for value in _preflight_digests(preflight))
        or not all(
            _is_digest(row.card_slot) and _is_digest(row.entry_digest)
            for row in preflight.historical_sending_rows
        )
    ):
        raise MaintenancePreparationError("no_send_oracle")


def _preflight_digests(preflight: SanitizedTopic59Preflight) -> tuple[str, ...]:
    return (
        preflight.candidate_digest,
        preflight.successor_config_digest,
        preflight.route_digest,
        preflight.customer_identity_digest,
        preflight.card_slot,
        preflight.publication_ledger_sha256,
    )


def _oracle_document(
    preflight: SanitizedTopic59Preflight,
    scope: Topic59MaintenanceScopeV1,
    facts: NoSendFacts,
) -> dict[str, JsonValue]:
    return {
        "schema": _ORACLE_SCHEMA,
        "target_candidate_digest": scope.candidate_digest,
        "target_config_digest": scope.config_digest,
        "target_route_digest": scope.route_digest,
        "target_customer_identity_digest": scope.customer_identity_digest,
        "target_card_slot": scope.card_slot,
        "target_kst_day": scope.kst_day.isoformat(),
        "non_topic59_due_send_count": facts.non_topic59_due_send_count,
        "non_topic59_due_edit_count": facts.non_topic59_due_edit_count,
        "activation_notice_due_count": facts.activation_notice_due_count,
        "owner_delivery_due_count": facts.owner_delivery_due_count,
        "target_topic59_projection_count": facts.target_topic59_projection_count,
        "historical_incident_slots": [
            {
                "card_slot": row.card_slot,
                "latest_entry_digest": row.entry_digest,
                "state": "sending",
                "message_id": None,
            }
            for row in preflight.historical_sending_rows
        ],
        "topic59_publication_ledger_sha256": preflight.publication_ledger_sha256,
    }


def _canonical_oracle(oracle: dict[str, JsonValue]) -> bytes:
    from gateway.platforms.nutrition_weekly_maintenance_contract import canonical_json

    return canonical_json(oracle) + b"\n"


def _is_digest(value: str) -> bool:
    return len(value) == 64 and all(character in "0123456789abcdef" for character in value)