"""Crash-safe, no-follow storage gate for the r71b Topic-59 hold."""

from __future__ import annotations

import fcntl
import hashlib
import json
import os
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Final, TypeVar

from pydantic import BaseModel, ValidationError

from .nutrition_weekly_maintenance_contract import (
    Topic59MaintenanceAuthorityV1,
    Topic59MaintenanceContractError,
    Topic59MaintenanceDecision,
    Topic59MaintenanceDecisionKind,
    Topic59MaintenanceFailure,
    Topic59MaintenanceHoldV1,
    Topic59MaintenanceSkipAuditV1,
    canonical_document,
)
from .nutrition_weekly_maintenance_io import (
    atomic_write_at as _atomic_write_at,
    fsync_directory as _fsync_directory,
    OwnedMaintenanceDirectory,
    open_file as _open_file,
    open_owned_directory as _open_owned_directory,
    parse_document as _parse_document,
    read_credential as _read_credential,
    read_regular_at as _read_regular_at,
)
from .nutrition_weekly_operations_models import Topic59Projection

_DocumentT = TypeVar("_DocumentT", bound=BaseModel)
_EXPECTED_NAMES: Final = frozenset({
    "hold.json", "hold.consuming.json", "skip-audit.json", "maintenance.lock",
})


@dataclass(frozen=True, slots=True)
class Topic59MaintenancePaths:
    profile_root: Path
    directory: Path


@dataclass(frozen=True, slots=True)
class Topic59MaintenanceGate:
    paths: Topic59MaintenancePaths

    @classmethod
    def for_profile(cls, profile_root: Path) -> Topic59MaintenanceGate:
        return cls(Topic59MaintenancePaths(
            profile_root, profile_root / "data" / "topic59-maintenance-r71b",
        ))

    def evaluate(
        self, projection: Topic59Projection, now: datetime,
    ) -> Topic59MaintenanceDecision:
        credential_root = os.environ.get("CREDENTIALS_DIRECTORY")
        if credential_root is None:
            if not os.path.lexists(self.paths.directory):
                return _allow()
            return _deny(Topic59MaintenanceFailure.AUTHORITY_MISSING_FOR_TARGET_DAY)
        try:
            authority = _read_credential(Path(credential_root))
        except FileNotFoundError:
            if not os.path.lexists(self.paths.directory):
                return _allow()
            return _deny(Topic59MaintenanceFailure.AUTHORITY_MISSING_FOR_TARGET_DAY)
        except (OSError, ValidationError, Topic59MaintenanceContractError):
            return _deny(Topic59MaintenanceFailure.AUTHORITY_CORRUPT)
        try:
            directory = _open_owned_directory(self.paths.profile_root)
        except FileNotFoundError:
            if authority.hold.kst_day == projection.kst_day:
                return _deny(Topic59MaintenanceFailure.HOLD_MISSING)
            return _allow()
        except OSError:
            return _deny(Topic59MaintenanceFailure.FILESYSTEM_CONTRACT)
        try:
            return self._evaluate_locked(directory, authority, projection, now)
        finally:
            os.close(directory.descriptor)

    def _evaluate_locked(
        self,
        directory: OwnedMaintenanceDirectory,
        authority: Topic59MaintenanceAuthorityV1,
        projection: Topic59Projection,
        now: datetime,
    ) -> Topic59MaintenanceDecision:
        try:
            if frozenset(os.listdir(directory.descriptor)) - _EXPECTED_NAMES:
                return _deny(Topic59MaintenanceFailure.FILESYSTEM_CONTRACT)
            lock_fd = _open_file(
                directory.descriptor, "maintenance.lock", 0o600, directory.uid, directory.gid,
            )
            try:
                fcntl.flock(lock_fd, fcntl.LOCK_EX)
                return _consume_locked(directory, authority, projection, now)
            finally:
                fcntl.flock(lock_fd, fcntl.LOCK_UN)
                os.close(lock_fd)
        except FileNotFoundError:
            return _deny(Topic59MaintenanceFailure.FILESYSTEM_CONTRACT)
        except (OSError, ValidationError, Topic59MaintenanceContractError):
            return _deny(Topic59MaintenanceFailure.FILESYSTEM_CONTRACT)


def _consume_locked(
    directory: OwnedMaintenanceDirectory,
    authority: Topic59MaintenanceAuthorityV1,
    projection: Topic59Projection,
    now: datetime,
) -> Topic59MaintenanceDecision:
    directory_fd = directory.descriptor
    hold = authority.hold
    audit = _optional_document(directory, "skip-audit.json", Topic59MaintenanceSkipAuditV1)
    active = _optional_document(directory, "hold.json", Topic59MaintenanceHoldV1)
    consuming = _optional_document(directory, "hold.consuming.json", Topic59MaintenanceHoldV1)
    if audit is not None:
        if not _audit_matches(audit, authority, hold):
            return _deny(Topic59MaintenanceFailure.AUDIT_CORRUPT)
        if projection.kst_day != hold.kst_day:
            if active is None and consuming is None:
                return _allow()
            return _deny(Topic59MaintenanceFailure.HOLD_REPLAYED)
        if not _projection_matches(hold, projection):
            return _deny(Topic59MaintenanceFailure.BINDING_MISMATCH)
        if active is not None:
            return _deny(Topic59MaintenanceFailure.HOLD_REPLAYED)
        if consuming is not None:
            if consuming != hold:
                return _deny(Topic59MaintenanceFailure.HOLD_REPLAYED)
            os.unlink("hold.consuming.json", dir_fd=directory_fd)
            _fsync_directory(directory_fd)
        return _noop(canonical_document(audit))
    selected = consuming if consuming is not None else active
    if selected is None:
        if hold.kst_day == projection.kst_day:
            return _deny(Topic59MaintenanceFailure.HOLD_MISSING)
        return _allow()
    if selected != hold or not _projection_matches(hold, projection):
        return _deny(Topic59MaintenanceFailure.BINDING_MISMATCH)
    if now < hold.not_before:
        return _deny(Topic59MaintenanceFailure.HOLD_NOT_YET_VALID)
    if now >= hold.expires_at:
        return _deny(Topic59MaintenanceFailure.HOLD_EXPIRED)
    if active is not None:
        if consuming is not None:
            return _deny(Topic59MaintenanceFailure.HOLD_REPLAYED)
        os.rename(
            "hold.json", "hold.consuming.json",
            src_dir_fd=directory_fd, dst_dir_fd=directory_fd,
        )
        _fsync_directory(directory_fd)
    audit_bytes = _build_audit(authority, hold, now)
    _atomic_write_at(directory_fd, "skip-audit.json", audit_bytes)
    os.unlink("hold.consuming.json", dir_fd=directory_fd)
    _fsync_directory(directory_fd)
    return _noop(audit_bytes)


def _optional_document(
    directory: OwnedMaintenanceDirectory, name: str, model: type[_DocumentT],
) -> _DocumentT | None:
    try:
        raw = _read_regular_at(
            directory.descriptor, name, 0o600, directory.uid, directory.gid,
        )
    except FileNotFoundError:
        return None
    return _parse_document(raw, model)


def _projection_matches(hold: Topic59MaintenanceHoldV1, projection: Topic59Projection) -> bool:
    return (
        hold.candidate_digest == projection.candidate_digest
        and hold.config_digest == projection.config_digest
        and hold.route_digest == projection.route_digest
        and hold.customer_identity_digest == projection.customer_identity_digest
        and hold.card_slot == projection.card_slot
        and hold.kst_day == projection.kst_day
    )


def _audit_matches(
    audit: Topic59MaintenanceSkipAuditV1,
    authority: Topic59MaintenanceAuthorityV1,
    hold: Topic59MaintenanceHoldV1,
) -> bool:
    return (
        audit.hold_id == hold.hold_id and audit.hold_sha256 == authority.hold_sha256
        and audit.authority_id == authority.authority_id
        and audit.package_binding_digest == hold.package_binding_digest
        and audit.candidate_digest == hold.candidate_digest
        and audit.config_digest == hold.config_digest and audit.route_digest == hold.route_digest
        and audit.customer_identity_digest == hold.customer_identity_digest
        and audit.card_slot == hold.card_slot and audit.kst_day == hold.kst_day
        and hold.not_before <= audit.recorded_at < hold.expires_at
    )


def _build_audit(
    authority: Topic59MaintenanceAuthorityV1, hold: Topic59MaintenanceHoldV1, now: datetime,
) -> bytes:
    fields: dict[str, str | bool | int | None | date | datetime] = {
        "schema": "nutricoach-topic59-maintenance-skip-audit-v1",
        "hold_id": hold.hold_id, "hold_sha256": authority.hold_sha256,
        "authority_id": authority.authority_id, "package_binding_digest": hold.package_binding_digest,
        "candidate_digest": hold.candidate_digest, "config_digest": hold.config_digest,
        "route_digest": hold.route_digest,
        "customer_identity_digest": hold.customer_identity_digest,
        "card_slot": hold.card_slot, "kst_day": hold.kst_day.isoformat(),
        "recorded_at": now.isoformat(), "terminal_state": "maintenance_skipped",
        "disposition": "noop_maintenance", "sent": False, "delivered": False,
        "message_id": None, "provider_receipt": None, "provider_calls": 0, "network_calls": 0,
    }
    encoded = json.dumps(fields, sort_keys=True, separators=(",", ":")).encode()
    fields["row_digest"] = hashlib.sha256(encoded).hexdigest()
    fields["kst_day"] = hold.kst_day
    fields["recorded_at"] = now
    return canonical_document(Topic59MaintenanceSkipAuditV1.model_validate(fields))


def _allow() -> Topic59MaintenanceDecision:
    return Topic59MaintenanceDecision(kind=Topic59MaintenanceDecisionKind.ALLOW)


def _deny(reason: Topic59MaintenanceFailure) -> Topic59MaintenanceDecision:
    return Topic59MaintenanceDecision(kind=Topic59MaintenanceDecisionKind.DENY, reason=reason)


def _noop(raw: bytes) -> Topic59MaintenanceDecision:
    return Topic59MaintenanceDecision(
        kind=Topic59MaintenanceDecisionKind.NOOP_MAINTENANCE,
        audit_digest=hashlib.sha256(raw).hexdigest(),
    )