"""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

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"
_OBJECT = TypeAdapter(dict[str, JsonValue])


@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 = _OBJECT.validate_python(_simple_yaml_mapping(
            _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 _simple_yaml_mapping(payload: bytes) -> dict[str, JsonValue]:
    """Parse the mapping-only profile configuration surface without untyped YAML.

    The runtime configuration consumed here is intentionally restricted to
    indentation-based mappings and scalar values.  Unsupported YAML is denied
    instead of being interpreted by a permissive loader.
    """
    try:
        lines = payload.decode("utf-8").splitlines()
    except UnicodeDecodeError as error:
        raise MaintenancePreparationError("profile_config") from error
    root: dict[str, JsonValue] = {}
    stack: list[tuple[int, dict[str, JsonValue]]] = [(-1, root)]
    for raw in lines:
        line = raw.split("#", 1)[0].rstrip()
        if not line:
            continue
        indent = len(line) - len(line.lstrip(" "))
        if "\t" in line or ":" not in line:
            raise MaintenancePreparationError("profile_config")
        key, value = line.lstrip(" ").split(":", 1)
        if not key or key.strip() != key:
            raise MaintenancePreparationError("profile_config")
        while stack[-1][0] >= indent:
            _ = stack.pop()
        parent = stack[-1][1]
        value = value.strip()
        if not value:
            child: dict[str, JsonValue] = {}
            parent[key] = child
            stack.append((indent, child))
            continue
        parent[key] = _yaml_scalar(value)
    if not root:
        raise MaintenancePreparationError("profile_config")
    return root


def _yaml_scalar(value: str) -> JsonValue:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
        return value[1:-1]
    if value in {"true", "True"}:
        return True
    if value in {"false", "False"}:
        return False
    if value in {"null", "Null", "~"}:
        return None
    try:
        return int(value)
    except ValueError:
        return value


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)