"""Durable human acknowledgement for owner-attention onboarding cards."""

from __future__ import annotations

import fcntl
import json
import os
import tempfile
from collections.abc import Mapping
from contextlib import contextmanager
from dataclasses import asdict
from pathlib import Path
from typing import Final, Iterator

from gateway.platforms.telegram_nutrition_onboarding_operator_ack_model import (
    GatewayOperatorAttentionAcknowledgement,
    build_acknowledgement,
    canonical,
    parse_acknowledgement,
    validate_attention_key,
)
from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import (
    GatewayPublicationReceipt,
)

_SCHEMA: Final = "telegram-nutrition-onboarding-operator-attention-v1"
_STATE_DIR: Final = Path("data/onboarding/telegram-publication-outbox-v1")


class GatewayOperatorAttentionStore:
    """Append-only acknowledgements distinct from provider send receipts."""

    def __init__(self, profile_root: Path | str) -> None:
        self.state_dir = Path(profile_root) / _STATE_DIR
        self.path = self.state_dir / "operator-attention.json"
        self.lock_path = self.state_dir / ".operator-attention.lock"
        self.key_path = self.state_dir / ".receipt-key"
        self._prepare()

    def acknowledge(
        self,
        *,
        session_id: str,
        customer_key: str,
        attention_identity: str,
        candidate_epoch: str,
        actor_user_id: int,
        authority: tuple[int, int, int],
        route: tuple[str, str],
        message_id: int,
        update_id: int,
        callback_data: str,
        publication: GatewayPublicationReceipt,
    ) -> tuple[GatewayOperatorAttentionAcknowledgement, bool]:
        candidate = build_acknowledgement(
            key=self._key(),
            session_id=session_id,
            customer_key=customer_key,
            attention_identity=attention_identity,
            candidate_epoch=candidate_epoch,
            actor_user_id=actor_user_id,
            authority=authority,
            route=route,
            message_id=message_id,
            update_id=update_id,
            callback_data=callback_data,
            publication=publication,
        )
        with self._locked():
            records = list(self._read())
            matches = [
                item
                for item in records
                if item.session_id == candidate.session_id
                and item.attention_identity == candidate.attention_identity
            ]
            if not matches:
                records.append(candidate)
                self._write(records)
                return candidate, True
            if len(matches) == 1:
                return matches[0], False
            raise ValueError("operator attention acknowledgement conflicts")

    def acknowledged(
        self,
        *,
        session_id: str,
        attention_identity: str,
    ) -> bool:
        validate_attention_key(session_id, attention_identity)
        with self._locked():
            return any(
                item.session_id == session_id
                and item.attention_identity == attention_identity
                for item in self._read()
            )

    def _prepare(self) -> None:
        if not self.state_dir.is_dir() or not self.key_path.is_file():
            raise ValueError("publication receipt authority is unavailable")
        for path in (self.path, self.lock_path):
            if not path.exists():
                descriptor = os.open(
                    path,
                    os.O_CREAT | os.O_EXCL | os.O_WRONLY,
                    0o600,
                )
                try:
                    if path == self.path:
                        os.write(
                            descriptor,
                            canonical({"schema": _SCHEMA, "records": []}) + b"\n",
                        )
                        os.fsync(descriptor)
                finally:
                    os.close(descriptor)
            if path.is_symlink() or path.stat().st_mode & 0o077:
                raise ValueError(
                    "operator attention ledger permissions are invalid"
                )

    def _read(self) -> tuple[GatewayOperatorAttentionAcknowledgement, ...]:
        value = json.loads(self.path.read_text(encoding="utf-8"))
        if not isinstance(value, Mapping) or value.get("schema") != _SCHEMA:
            raise ValueError("operator attention ledger integrity is invalid")
        raw = value.get("records")
        if not isinstance(raw, list):
            raise ValueError("operator attention ledger integrity is invalid")
        records = tuple(
            parse_acknowledgement(item, key=self._key()) for item in raw
        )
        keys = {
            (item.session_id, item.attention_identity)
            for item in records
        }
        if len(keys) != len(records):
            raise ValueError("operator attention acknowledgement is ambiguous")
        return records

    def _write(
        self,
        records: list[GatewayOperatorAttentionAcknowledgement],
    ) -> None:
        document = {
            "schema": _SCHEMA,
            "records": [asdict(item) for item in records],
        }
        descriptor, temporary = tempfile.mkstemp(dir=self.state_dir)
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(canonical(document) + b"\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, self.path)
            directory = os.open(
                self.state_dir,
                os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC,
            )
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        finally:
            if os.path.exists(temporary):
                os.unlink(temporary)

    def _key(self) -> bytes:
        key = self.key_path.read_bytes()
        if len(key) != 32:
            raise ValueError("publication receipt key is invalid")
        return key

    @contextmanager
    def _locked(self) -> Iterator[None]:
        with self.lock_path.open("r+b") as handle:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


__all__ = [
    "GatewayOperatorAttentionAcknowledgement",
    "GatewayOperatorAttentionStore",
]