"""Private append-only persistence for weekly day-status rows."""

from __future__ import annotations

import os
from collections.abc import Callable
from dataclasses import dataclass
from typing import TypeVar

from checkin_cli.weekly_operations import (
    AppendResult,
    CustomerIdentityDigest,
    CustomerKey,
    DayState,
    RepairResult,
    WEEKLY_OPERATIONS_SCHEMA,
    ZERO_DIGEST,
    WeeklyOperationInput,
    WeeklyOperationRow,
    WeeklyOperationsConflict,
    WeeklyOperationsCorruption,
    WeeklyOperationsInputError,
    canonical_weekly_row,
    customer_identity_digest,
    customer_storage_digest,
    operation_logical_key,
    weekly_row_digest,
)
from checkin_cli.weekly_operations_authority import WeeklyOperationsAuthorityRoot
from checkin_cli.weekly_operations_history import validate_history_bytes, validate_weekly_rows
from checkin_cli.weekly_operations_repair import validate_bound_repair_lock, validate_bound_repair_payload
from checkin_cli.weekly_operations_repair_publish import publish_repaired_prefix
from checkin_cli.weekly_operations_fs import (
    FileIdentity,
    SidecarDirectory,
    locked_repair_sidecar,
    locked_sidecar,
    open_existing_file,
    open_or_create_file,
    verify_file,
)


_T = TypeVar("_T")


@dataclass(frozen=True, slots=True)
class _FileSnapshot:
    data: bytes
    identity: FileIdentity | None


@dataclass(frozen=True, slots=True)
class _LockedHistory:
    directory: SidecarDirectory
    rows: tuple[WeeklyOperationRow, ...]
    identity: FileIdentity | None


def _read_private(directory: SidecarDirectory) -> _FileSnapshot:
    try:
        opened = open_existing_file(directory.plans, directory.data_name, os.O_RDONLY)
    except FileNotFoundError:
        return _FileSnapshot(b"", None)
    try:
        chunks: list[bytes] = []
        while chunk := os.read(opened.descriptor, 65536):
            chunks.append(chunk)
        directory.verify()
        _ = verify_file(directory.plans, directory.data_name, os.fstat(opened.descriptor))
        return _FileSnapshot(b"".join(chunks), opened.identity)
    finally:
        os.close(opened.descriptor)


def _append_private(directory: SidecarDirectory, payload: bytes, expected: FileIdentity | None) -> None:
    directory.verify()
    opened = open_or_create_file(directory.plans, directory.data_name, os.O_WRONLY | os.O_APPEND)
    try:
        if (expected is None and not opened.created) or (expected is not None and opened.identity != expected):
            raise WeeklyOperationsCorruption("sidecar data was replaced before append")
        directory.verify()
        offset = 0
        while offset < len(payload):
            offset += os.write(opened.descriptor, payload[offset:])
        _ = verify_file(directory.plans, directory.data_name, os.fstat(opened.descriptor))
        os.fsync(opened.descriptor)
        if opened.created:
            os.fsync(directory.plans)
    finally:
        os.close(opened.descriptor)


def _read_open_file(descriptor: int) -> bytes:
    _ = os.lseek(descriptor, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    while chunk := os.read(descriptor, 65536):
        chunks.append(chunk)
    return b"".join(chunks)


class WeeklyOperationsStore:
    """Own one customer's hash-chained weekly-operations JSONL sidecar."""

    def __init__(self, authority: WeeklyOperationsAuthorityRoot, customer_key: CustomerKey) -> None:
        identity = customer_identity_digest(customer_key)
        authority.verify()
        if authority.repair_target_digest is not None and authority.repair_target_digest != customer_storage_digest(identity):
            raise WeeklyOperationsInputError("repair authority customer mismatch")
        self.authority: WeeklyOperationsAuthorityRoot = authority
        self.customer_identity_digest: CustomerIdentityDigest = identity

    @property
    def authority_binding_digest(self) -> str:
        """Expose the verified sidecar authority pin without its capability."""
        self.authority.verify()
        return self.authority.binding.binding_digest

    @classmethod
    def for_authority(cls, authority: WeeklyOperationsAuthorityRoot, customer_key: CustomerKey) -> WeeklyOperationsStore:
        return cls(authority, customer_key)

    def _require_normal_authority(self) -> None:
        if self.authority.repair_target_digest is not None:
            raise WeeklyOperationsInputError("repair-only authority")

    def append(self, operation: WeeklyOperationInput) -> AppendResult:
        """Append one valid transition or return its exact durable replay."""
        self._require_normal_authority()
        if operation.customer_identity_digest != self.customer_identity_digest:
            raise WeeklyOperationsInputError("wrong customer identity")
        with locked_sidecar(self.authority, self.customer_identity_digest, exclusive=True) as directory:
            return self._append_locked(operation, self._locked_history(directory))

    def transact(
        self,
        decide: Callable[[tuple[WeeklyOperationRow, ...]], tuple[WeeklyOperationInput | None, _T]],
    ) -> tuple[AppendResult | None, _T]:
        """Decide and append at most once under this customer's sidecar lock."""
        self._require_normal_authority()
        with locked_sidecar(self.authority, self.customer_identity_digest, exclusive=True) as directory:
            history = self._locked_history(directory)
            operation, result = decide(history.rows)
            if operation is None:
                return None, result
            if operation.customer_identity_digest != self.customer_identity_digest:
                raise WeeklyOperationsInputError("wrong customer identity")
            return self._append_locked(operation, history), result

    def _locked_history(self, directory: SidecarDirectory) -> _LockedHistory:
        rows, complete_end, size, identity = self._read_rows(directory, allow_torn_tail=False)
        if complete_end != size:
            raise WeeklyOperationsCorruption("sidecar has a torn tail; explicit repair required")
        self._validate_rows(rows)
        return _LockedHistory(directory, tuple(rows), identity)

    def _append_locked(
        self, operation: WeeklyOperationInput, history: _LockedHistory
    ) -> AppendResult:
        rows = list(history.rows)
        replay = next((row for row in rows if self._matches(row, operation)), None)
        if replay is not None:
            return AppendResult(row=replay, appended=False)
        day_rows = tuple(row for row in rows if row.kst_day == operation.kst_day)
        logical_key = self._logical_key(operation, day_rows)
        if any(row.logical_key == logical_key for row in rows):
            raise WeeklyOperationsConflict("duplicate logical key has conflicting content")
        source, reminder = operation.source, operation.reminder
        provisional = WeeklyOperationRow(
            schema_version=WEEKLY_OPERATIONS_SCHEMA,
            customer_identity_digest=operation.customer_identity_digest,
            kst_day=operation.kst_day,
            state=operation.state,
            canonical_sequence=operation.canonical.sequence,
            canonical_digest=operation.canonical.digest,
            source_event_id=None if source is None else source.event_id,
            source_event_digest=None if source is None else source.event_digest,
            reminder_reservation_id=None if reminder is None else reminder.reservation_id,
            reminder_audit_id=None if reminder is None else reminder.audit_id,
            predecessor_row_digest=rows[-1].row_digest if rows else ZERO_DIGEST,
            occurred_at_kst=operation.occurred_at,
            logical_key=logical_key,
            row_digest=ZERO_DIGEST,
        )
        row = provisional.model_copy(update={"row_digest": weekly_row_digest(provisional)})
        self._validate_rows([*rows, row], candidate=True)
        _append_private(
            history.directory,
            canonical_weekly_row(row, include_digest=True) + b"\n",
            history.identity,
        )
        return AppendResult(row=row, appended=True)

    def read(self) -> tuple[WeeklyOperationRow, ...]:
        """Read only a complete, valid sidecar under a shared lock."""
        self._require_normal_authority()
        with locked_sidecar(self.authority, self.customer_identity_digest, exclusive=False) as directory:
            rows, complete_end, size, _ = self._read_rows(directory, allow_torn_tail=False)
            if complete_end != size:
                raise WeeklyOperationsCorruption("sidecar has a torn tail; explicit repair required")
            self._validate_rows(rows)
            return tuple(rows)

    def repair_torn_tail(self) -> RepairResult:
        """Remove only the strictly bound positive torn fragment."""
        binding = self.authority.repair_binding
        if binding is None:
            raise WeeklyOperationsInputError("repair authority required")
        if binding.customer_identity_digest != self.customer_identity_digest:
            raise WeeklyOperationsInputError("repair authority customer mismatch")
        with locked_repair_sidecar(self.authority, self.customer_identity_digest, binding) as locked:
            current = _read_open_file(locked.data_descriptor)
            _ = validate_bound_repair_payload(binding, os.fstat(locked.data_descriptor), current)
            lock_bytes = _read_open_file(locked.lock_descriptor)
            validate_bound_repair_lock(binding, os.fstat(locked.lock_descriptor), lock_bytes)
            prefix = current[: binding.valid_prefix_offset]
            publish_repaired_prefix(locked.directory.plans, locked.directory.data_name, prefix, binding.data_identity)
            self.authority.verify_complete_locked()
            repaired_snapshot = _read_private(locked.directory)
            validated = validate_history_bytes(repaired_snapshot.data, binding.customer_storage_digest, expected_customer=self.customer_identity_digest)
            if validated.size != binding.valid_prefix_offset:
                raise WeeklyOperationsCorruption("repair result size disagrees")
            return RepairResult(len(validated.rows), binding.torn_tail_length)

    def _read_rows(self, directory: SidecarDirectory, *, allow_torn_tail: bool) -> tuple[list[WeeklyOperationRow], int, int, FileIdentity | None]:
        snapshot = _read_private(directory)
        validated = validate_history_bytes(
            snapshot.data,
            directory.customer_digest,
            expected_customer=self.customer_identity_digest,
            allow_torn_tail=allow_torn_tail,
        )
        return list(validated.rows), validated.complete_end, validated.size, snapshot.identity

    def _validate_rows(self, rows: list[WeeklyOperationRow], *, candidate: bool = False) -> None:
        validate_weekly_rows(rows, self.customer_identity_digest, candidate=candidate)

    @staticmethod
    def _matches(row: WeeklyOperationRow, operation: WeeklyOperationInput) -> bool:
        source, reminder = operation.source, operation.reminder
        return row.customer_identity_digest == operation.customer_identity_digest and row.kst_day == operation.kst_day and row.state is operation.state and row.canonical_sequence == operation.canonical.sequence and row.canonical_digest == operation.canonical.digest and row.source_event_id == (None if source is None else source.event_id) and row.source_event_digest == (None if source is None else source.event_digest) and row.reminder_reservation_id == (None if reminder is None else reminder.reservation_id) and row.reminder_audit_id == (None if reminder is None else reminder.audit_id) and row.occurred_at_kst == operation.occurred_at

    def _logical_key(self, operation: WeeklyOperationInput, day_rows: tuple[WeeklyOperationRow, ...]) -> str:
        if not day_rows:
            identity = "terminal"
        elif day_rows[-1].state is DayState.MISSED and operation.state is DayState.LATE_SUBMITTED:
            identity = "late"
        elif operation.source is not None:
            identity = f"source:{operation.source.event_id}"
        else:
            raise WeeklyOperationsConflict("same-state append requires source lineage")
        return operation_logical_key(self.customer_identity_digest, operation.kst_day, identity)