"""Durable, at-most-once publication boundary for Telegram check-in steps.

The journal is deliberately content-free.  Customer input and rendered cards live
only in the caller-owned transition/publication objects and are never serialized
here.  In particular, a process restart observes unfinished delivery but never
replays it automatically.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, replace
from enum import StrEnum
from typing import Protocol, final, runtime_checkable

from gateway.platforms.physique_checkin_bindings import (
    BindingStore,
    BindingStoreConflict,
    BindingStoreError,
    CursorIdentity,
    IngressIdentity,
    ProjectionPhase,
    TelegramProjection,
)


class DeliveryRejected(RuntimeError):
    """The provider definitively rejected a started request before delivery."""


class TransportPreflightError(RuntimeError):
    """A local request could not be prepared; provider I/O has not started."""


@runtime_checkable
class DomainCommitResult(Protocol):
    """Transient result of one compare-and-swap domain transition."""

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

    @property
    def publication(self) -> object | None: ...


@runtime_checkable
class CheckinTransition(Protocol):
    """Typed, value-free identity plus a caller-owned domain CAS operation."""

    ingress: IngressIdentity
    source: CursorIdentity
    target: CursorIdentity
    expires_at: int

    def commit(self) -> DomainCommitResult: ...


class CursorLoader(Protocol):
    """Reload the canonical value-free cursor for stale-ingress rejection."""

    def __call__(self, session_id: str) -> CursorIdentity | None: ...


@runtime_checkable
class DeliveryReceipt(Protocol):
    """The only provider value retained by the journal."""

    @property
    def message_id(self) -> int: ...


@runtime_checkable
class TelegramStepperTransport(Protocol):
    """Narrow transport seam; ``prepare`` must perform no provider I/O."""

    def prepare(self, publication: object) -> object: ...

    async def send(self, prepared: object) -> DeliveryReceipt: ...


class StepperDisposition(StrEnum):
    DELIVERED = "delivered"
    DELIVERY_FAILED = "delivery_failed"
    DELIVERY_UNCERTAIN = "delivery_uncertain"
    DOMAIN_REJECTED = "domain_rejected"
    DOMAIN_FAILURE = "domain_failure"
    RECOVERY_REQUIRED = "recovery_required"
    STALE = "stale"
    EXPIRED = "expired"
    STORE_FAILURE = "store_failure"


@dataclass(frozen=True, slots=True)
class StepperResult:
    """Content-free ingress decision suitable for polling and UI recovery."""

    disposition: StepperDisposition
    poll_terminal: bool
    ui_recovery_required: bool
    phase: ProjectionPhase | None = None
    receipt_message_id: int | None = None


@final
class TelegramPhysiqueCheckinStepper:
    """Coordinate one domain CAS and one non-replayed Telegram publication.

    Each store call owns its lock only for an atomic file operation.  The domain
    commit, transport preflight, and awaited provider call all happen outside
    that lock.
    """

    def __init__(
        self,
        store: BindingStore,
        transport: TelegramStepperTransport,
        *,
        current_cursor: CursorLoader | None = None,
    ) -> None:
        self._store = store
        self._transport = transport
        self._current_cursor = current_cursor

    async def execute(
        self,
        transition: CheckinTransition,
        *,
        now_epoch: int,
    ) -> StepperResult:
        """Execute a fresh ingress once, or observe its durable prior decision."""
        if not self._valid_transition(transition):
            return self._result(StepperDisposition.STORE_FAILURE)
        if now_epoch >= transition.expires_at:
            return self._result(StepperDisposition.EXPIRED)

        try:
            prior = self._projection_for_update(transition.ingress.update_id, now_epoch)
        except BindingStoreError:
            return self._result(StepperDisposition.STORE_FAILURE)
        if prior is not None:
            if (
                prior.ingress != transition.ingress
                or prior.source != transition.source
                or prior.target != transition.target
                or prior.expires_at != transition.expires_at
            ):
                return self._result(StepperDisposition.STORE_FAILURE)
            return self._observed(prior)

        current = self._load_current_cursor(transition.source.session_id)
        if current is not None and current != transition.source:
            return self._result(StepperDisposition.STALE)

        prepared = TelegramProjection(
            transition.ingress,
            transition.source,
            transition.target,
            ProjectionPhase.PREPARED,
            transition.expires_at,
        )
        try:
            inserted = self._store.record_projection(prepared, now_epoch=now_epoch)
        except (BindingStoreError, BindingStoreConflict, ValueError):
            return self._result(StepperDisposition.STORE_FAILURE)
        if not inserted:
            return self._observed(prepared)

        try:
            domain = transition.commit()
        except Exception:
            return self._result(
                StepperDisposition.DOMAIN_FAILURE,
                phase=ProjectionPhase.PREPARED,
                recover=True,
            )
        if type(domain.accepted) is not bool:
            return self._result(
                StepperDisposition.DOMAIN_FAILURE,
                phase=ProjectionPhase.PREPARED,
                recover=True,
            )
        if not domain.accepted or domain.publication is None:
            return self._result(
                StepperDisposition.DOMAIN_REJECTED,
                phase=ProjectionPhase.PREPARED,
                recover=True,
            )

        committed = replace(prepared, phase=ProjectionPhase.DOMAIN_COMMITTED)
        try:
            _ = self._store.record_projection(committed, now_epoch=now_epoch)
        except (BindingStoreError, BindingStoreConflict, ValueError):
            return self._result(
                StepperDisposition.RECOVERY_REQUIRED,
                phase=ProjectionPhase.PREPARED,
                recover=True,
            )

        try:
            transport_request = self._transport.prepare(domain.publication)
        except Exception:
            failed = replace(committed, phase=ProjectionPhase.DELIVERY_FAILED)
            return self._persist_terminal(
                failed,
                StepperDisposition.DELIVERY_FAILED,
                now_epoch,
                recover=True,
            )

        started = replace(committed, phase=ProjectionPhase.SEND_STARTED)
        try:
            _ = self._store.record_projection(started, now_epoch=now_epoch)
        except (BindingStoreError, BindingStoreConflict, ValueError):
            return self._result(
                StepperDisposition.RECOVERY_REQUIRED,
                phase=ProjectionPhase.DOMAIN_COMMITTED,
                recover=True,
            )

        try:
            receipt = await self._transport.send(transport_request)
        except DeliveryRejected:
            failed = replace(started, phase=ProjectionPhase.DELIVERY_FAILED)
            return self._persist_terminal(
                failed,
                StepperDisposition.DELIVERY_FAILED,
                now_epoch,
                recover=True,
            )
        except asyncio.CancelledError:
            uncertain = replace(started, phase=ProjectionPhase.DELIVERY_UNCERTAIN)
            return self._persist_terminal(
                uncertain,
                StepperDisposition.DELIVERY_UNCERTAIN,
                now_epoch,
                recover=True,
            )
        except Exception:
            uncertain = replace(started, phase=ProjectionPhase.DELIVERY_UNCERTAIN)
            return self._persist_terminal(
                uncertain,
                StepperDisposition.DELIVERY_UNCERTAIN,
                now_epoch,
                recover=True,
            )

        message_id = getattr(receipt, "message_id", None)
        if type(message_id) is not int or message_id <= 0:
            uncertain = replace(started, phase=ProjectionPhase.DELIVERY_UNCERTAIN)
            return self._persist_terminal(
                uncertain,
                StepperDisposition.DELIVERY_UNCERTAIN,
                now_epoch,
                recover=True,
            )
        delivered = replace(
            started,
            phase=ProjectionPhase.DELIVERED,
            receipt_message_id=message_id,
        )
        try:
            _ = self._store.record_projection(delivered, now_epoch=now_epoch)
        except (BindingStoreError, BindingStoreConflict, ValueError):
            # The provider receipt cannot safely be replayed or inferred later.
            return self._result(
                StepperDisposition.DELIVERY_UNCERTAIN,
                phase=ProjectionPhase.SEND_STARTED,
                recover=True,
            )
        return self._result(
            StepperDisposition.DELIVERED,
            phase=ProjectionPhase.DELIVERED,
            receipt=message_id,
        )

    def observe_session(
        self,
        session_id: str,
        *,
        now_epoch: int,
    ) -> StepperResult | None:
        """Observe the newest durable session row without causing publication."""
        try:
            rows = self._store.load_projections(now_epoch)
        except BindingStoreError:
            return self._result(StepperDisposition.STORE_FAILURE)
        matching = tuple(row for row in rows if row.session_id == session_id)
        return self._observed(matching[-1]) if matching else None

    def observe_ingress(
        self,
        ingress: IngressIdentity,
        session_id: str,
        *,
        now_epoch: int,
    ) -> StepperResult | None:
        """Observe UI recovery for typed callback/text ingress; never journal raw input."""
        if type(ingress) is not IngressIdentity:
            return self._result(StepperDisposition.STORE_FAILURE)
        return self.observe_session(session_id, now_epoch=now_epoch)

    def resume(self, update_id: int, *, now_epoch: int) -> StepperResult | None:
        """Explicitly resume the UI decision only; delivery is intentionally not retried."""
        try:
            projection = self._projection_for_update(update_id, now_epoch)
        except BindingStoreError:
            return self._result(StepperDisposition.STORE_FAILURE)
        return self._observed(projection) if projection is not None else None

    def _projection_for_update(
        self,
        update_id: int,
        now_epoch: int,
    ) -> TelegramProjection | None:
        return next(
            (
                item
                for item in self._store.load_projections(now_epoch)
                if item.ingress.update_id == update_id
            ),
            None,
        )

    def _load_current_cursor(self, session_id: str) -> CursorIdentity | None:
        loader = self._current_cursor
        if loader is None:
            return None
        try:
            value = loader(session_id)
        except Exception:
            return CursorIdentity(session_id, "launch", 0)
        return value if type(value) is CursorIdentity else CursorIdentity(session_id, "launch", 0)

    def _persist_terminal(
        self,
        projection: TelegramProjection,
        disposition: StepperDisposition,
        now_epoch: int,
        *,
        recover: bool,
    ) -> StepperResult:
        try:
            _ = self._store.record_projection(projection, now_epoch=now_epoch)
        except (BindingStoreError, BindingStoreConflict, ValueError):
            return self._result(
                StepperDisposition.DELIVERY_UNCERTAIN,
                phase=ProjectionPhase.SEND_STARTED,
                recover=True,
            )
        return self._result(
            disposition,
            phase=projection.phase,
            receipt=projection.receipt_message_id,
            recover=recover,
        )

    @staticmethod
    def _valid_transition(transition: object) -> bool:
        if not isinstance(transition, CheckinTransition):
            return False
        return (
            type(transition.ingress) is IngressIdentity
            and type(transition.source) is CursorIdentity
            and type(transition.target) is CursorIdentity
            and type(transition.expires_at) is int
            and transition.expires_at > 0
            and transition.source.session_id == transition.target.session_id
            and transition.target.version == transition.source.version + 1
        )

    @classmethod
    def _observed(cls, projection: TelegramProjection) -> StepperResult:
        if projection.phase is ProjectionPhase.DELIVERED:
            return cls._result(
                StepperDisposition.DELIVERED,
                phase=projection.phase,
                receipt=projection.receipt_message_id,
            )
        return cls._result(
            StepperDisposition.RECOVERY_REQUIRED,
            phase=projection.phase,
            recover=True,
        )

    @staticmethod
    def _result(
        disposition: StepperDisposition,
        *,
        phase: ProjectionPhase | None = None,
        receipt: int | None = None,
        recover: bool = False,
    ) -> StepperResult:
        return StepperResult(disposition, True, recover, phase, receipt)