"""Exact r71b credential/hold files and safe predecessor-cron rollback."""

from __future__ import annotations

import hashlib
import json
import os
import stat
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Protocol, final

from pydantic import JsonValue, TypeAdapter, ValidationError

from scripts.nutricoach_v150_r71b_maintenance_confirmation import (
    ArmedCronWatcher,
    MaintenanceConfirmation,
    MaintenanceConfirmationPaths,
    newly_written_output,
    snapshot_outputs,
    verify_maintenance_confirmation,
    verify_recovered_maintenance_confirmation,
)

from gateway.platforms.nutrition_weekly_maintenance_contract import (
    Topic59MaintenanceAuthorityV1,
    Topic59MaintenanceContractError,
    canonical_document,
)
from scripts.nutricoach_v150_r71b_maintenance_transaction import KNOWN_R70_CRON_JOB_ID
from scripts.nutricoach_v150_sealed_authority import atomic_write

_CREDENTIAL_NAME = "nutricoach-topic59-maintenance-r71b.json"
_MAINTENANCE_DIRECTORY = "topic59-maintenance-r71b"
_CONFIRMATION_SCHEMA = "nutricoach-r71b-maintenance-confirmation-v1"
_OBJECT = TypeAdapter(dict[str, JsonValue])


class MaintenanceFileError(RuntimeError):
    """An exact maintenance filesystem postimage cannot be established."""


class StoppedService(Protocol):
    """The maintenance write fence requires a synchronously stopped service."""

    @property
    def running(self) -> bool: ...


@dataclass(frozen=True, slots=True)
class R71bMaintenancePaths:
    """Bound paths which the r71b controller may create or restore."""

    profile: Path
    successor_runtime: Path
    cron_jobs: Path
    dropin: Path


@dataclass(frozen=True, slots=True)
class _CronSnapshot:
    """Exact stopped-state cron bytes retained solely for rollback."""

    payload: bytes
    mode: int


@final
class R71bMaintenanceFiles:
    """Mutable one-use file transaction; service state is checked at its boundary."""

    def __init__(
        self,
        paths: R71bMaintenancePaths,
        service: StoppedService,
        authority_bytes: bytes,
    ) -> None:
        self._paths: R71bMaintenancePaths = paths
        self._service: StoppedService = service
        self._authority_bytes: bytes = authority_bytes
        self._authority: Topic59MaintenanceAuthorityV1 = _parse_authority(authority_bytes)
        self._cron_snapshot: _CronSnapshot | None = None
        self._dropin_snapshot: _CronSnapshot | None = None
        self._output_before: dict[Path, str] | None = None
        self._credential_parent_created = False

    @property
    def authority(self) -> Topic59MaintenanceAuthorityV1:
        """Return the validated immutable authority bound to this transaction."""
        return self._authority

    @property
    def maintenance_root(self) -> Path:
        """Return the one exact profile-owned maintenance directory."""
        return self._paths.profile / "data" / _MAINTENANCE_DIRECTORY

    @property
    def hold_path(self) -> Path:
        """Return the active one-use hold path."""
        return self.maintenance_root / "hold.json"

    @property
    def consuming_path(self) -> Path:
        """Return the crash-recovery consuming hold path."""
        return self.maintenance_root / "hold.consuming.json"

    @property
    def audit_path(self) -> Path:
        """Return the terminal maintenance audit path."""
        return self.maintenance_root / "skip-audit.json"

    @property
    def confirmation_path(self) -> Path:
        """Return the durable receipt needed for confirmation-gated recovery."""
        return self.maintenance_root / "confirmation.json"

    @property
    def output_directory(self) -> Path:
        """Return the immutable dispatcher output directory for the exact r70 job."""
        return self._paths.cron_jobs.parent / "output" / KNOWN_R70_CRON_JOB_ID

    @property
    def credential_path(self) -> Path:
        """Return the exact successor systemd credential source path."""
        return (
            self._paths.successor_runtime.parent
            / "runtime-authority"
            / _CREDENTIAL_NAME
        )

    def arm(self) -> None:
        """Capture stopped cron bytes, then install exact credential and active hold."""
        if self._service.running:
            raise MaintenanceFileError("service_running")
        self._cron_snapshot = _snapshot_file(self._paths.cron_jobs)
        self._output_before = snapshot_outputs(self.output_directory)
        dropin_snapshot = _snapshot_file(self._paths.dropin)
        self._dropin_snapshot = dropin_snapshot
        if self.maintenance_root.exists() or self.maintenance_root.is_symlink():
            raise MaintenanceFileError("maintenance_root_occupied")
        if self.credential_path.exists() or self.credential_path.is_symlink():
            raise MaintenanceFileError("credential_occupied")
        credential_root = self.credential_path.parent
        self._credential_parent_created = not credential_root.exists()
        credential_root.mkdir(parents=True, mode=0o700)
        credential_root.chmod(0o700)
        self.maintenance_root.mkdir(parents=True, mode=0o700)
        self.maintenance_root.chmod(0o700)
        atomic_write(self.credential_path, self._authority_bytes, 0o400)
        _append_maintenance_credential(
            self._paths.dropin,
            dropin_snapshot,
            self.credential_path,
        )
        hold_bytes = canonical_document(self._authority.hold)
        atomic_write(self.hold_path, hold_bytes, 0o600)
        _require_file(self.credential_path, 0o400, self._authority_bytes)
        _require_file(self.hold_path, 0o600, hold_bytes)
        if (
            self.audit_path.exists()
            or self.consuming_path.exists()
            or self.confirmation_path.exists()
        ):
            raise MaintenanceFileError("audit_before_start")

    def confirm(
        self,
        watcher: ArmedCronWatcher,
        evidence: MaintenanceConfirmation,
    ) -> None:
        """Require the watched cron success, typed no-op output, audit, and zero calls."""
        if evidence.output_path != self.output_directory:
            raise MaintenanceFileError("cron_output_directory")
        confirmed = replace(
            evidence,
            output_path=newly_written_output(self.output_directory, self._output_before),
        )
        verify_maintenance_confirmation(
            MaintenanceConfirmationPaths(
                self._paths.cron_jobs,
                self.hold_path,
                self.consuming_path,
                self.audit_path,
            ),
            self._authority,
            watcher,
            confirmed,
        )
        self._write_confirmation_receipt(confirmed)

    def recover_confirmed(self, evidence: MaintenanceConfirmation) -> None:
        """Reverify only a completion recorded after the watched confirmation passed."""
        output = self._read_confirmation_receipt(evidence)
        verify_recovered_maintenance_confirmation(
            MaintenanceConfirmationPaths(
                self._paths.cron_jobs,
                self.hold_path,
                self.consuming_path,
                self.audit_path,
            ),
            self._authority,
            replace(evidence, output_path=output),
        )

    def finalize(self) -> None:
        """Retain the same-day credential required by authenticated repeat no-ops."""
        _require_file(self.credential_path, 0o400, self._authority_bytes)
        snapshot = _snapshot_file(self._paths.dropin)
        credential = (
            b"LoadCredential=" + _CREDENTIAL_NAME.encode() + b":"
            + str(self.credential_path).encode()
        )
        if sum(line == credential for line in snapshot.payload.splitlines()) != 1:
            raise MaintenanceFileError("maintenance_credential_postimage")

    def _write_confirmation_receipt(self, evidence: MaintenanceConfirmation) -> None:
        output = evidence.output_path
        if output.parent != self.output_directory:
            raise MaintenanceFileError("cron_output_directory")
        payload = {
            "schema": _CONFIRMATION_SCHEMA,
            "ledger_sha256": evidence.ledger_sha256,
            "output_name": output.name,
            "output_sha256": hashlib.sha256(output.read_bytes()).hexdigest(),
        }
        atomic_write(
            self.confirmation_path,
            json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + b"\n",
            0o400,
        )

    def _read_confirmation_receipt(self, evidence: MaintenanceConfirmation) -> Path:
        try:
            receipt = _OBJECT.validate_json(self.confirmation_path.read_bytes())
        except (OSError, ValidationError) as error:
            raise MaintenanceFileError("confirmation_receipt") from error
        name = receipt.get("output_name")
        if (
            receipt.get("schema") != _CONFIRMATION_SCHEMA
            or receipt.get("ledger_sha256") != evidence.ledger_sha256
            or not isinstance(name, str)
            or Path(name).name != name
        ):
            raise MaintenanceFileError("confirmation_receipt")
        output = self.output_directory / name
        try:
            output_sha256 = hashlib.sha256(output.read_bytes()).hexdigest()
        except OSError as error:
            raise MaintenanceFileError("confirmation_receipt") from error
        if receipt.get("output_sha256") != output_sha256:
            raise MaintenanceFileError("confirmation_receipt")
        return output

    def _pause_r70_scheduler(self) -> None:
        from scripts.nutricoach_v150_r71b_maintenance_transaction import (
            pause_r70_scheduler,
        )

        pause_r70_scheduler(self._paths.cron_jobs)

    def rollback(self) -> None:
        """Restore stopped cron bytes then pause its exact provider-capable job."""
        snapshot = self._cron_snapshot
        if snapshot is None:
            raise MaintenanceFileError("cron_snapshot_missing")
        atomic_write(self._paths.cron_jobs, snapshot.payload, snapshot.mode)
        self._pause_r70_scheduler()
        self.hold_path.unlink(missing_ok=True)
        self.consuming_path.unlink(missing_ok=True)
        self.confirmation_path.unlink(missing_ok=True)
        self.credential_path.unlink(missing_ok=True)
        if self._credential_parent_created:
            self.credential_path.parent.rmdir()
        dropin = self._dropin_snapshot
        if dropin is not None:
            atomic_write(self._paths.dropin, dropin.payload, dropin.mode)
        if not self.audit_path.exists():
            self.maintenance_root.rmdir()


def _parse_authority(payload: bytes) -> Topic59MaintenanceAuthorityV1:
    try:
        authority = Topic59MaintenanceAuthorityV1.model_validate_json(payload)
    except (ValidationError, Topic59MaintenanceContractError) as error:
        raise MaintenanceFileError("authority") from error
    if canonical_document(authority) != payload:
        raise MaintenanceFileError("authority_canonical")
    if hashlib.sha256(canonical_document(authority.hold)).hexdigest() != authority.hold_sha256:
        raise MaintenanceFileError("hold_hash")
    return authority


def _snapshot_file(path: Path) -> _CronSnapshot:
    info = path.stat(follow_symlinks=False)
    if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
        raise MaintenanceFileError("cron_snapshot")
    return _CronSnapshot(path.read_bytes(), stat.S_IMODE(info.st_mode))


def _append_maintenance_credential(
    dropin: Path,
    snapshot: _CronSnapshot,
    credential: Path,
) -> None:
    name = _CREDENTIAL_NAME.encode()
    existing = snapshot.payload.splitlines()
    if any(
        line.startswith(b"LoadCredential=" + name + b":") for line in existing
    ):
        raise MaintenanceFileError("maintenance_credential_declared")
    payload = snapshot.payload.rstrip(b"\n") + b"\nLoadCredential=" + name + b":"
    payload += str(credential).encode() + b"\n"
    atomic_write(dropin, payload, snapshot.mode)


def _require_file(path: Path, mode: int, payload: bytes) -> None:
    info = path.stat(follow_symlinks=False)
    if (
        path.is_symlink()
        or not stat.S_ISREG(info.st_mode)
        or info.st_nlink != 1
        or stat.S_IMODE(info.st_mode) != mode
        or info.st_uid != os.geteuid()
        or info.st_gid != os.getegid()
        or path.read_bytes() != payload
    ):
        raise MaintenanceFileError("maintenance_postimage")