"""Durable at-most-once orchestration for Telegram physique check-in cards."""

from __future__ import annotations

import asyncio
import os
from dataclasses import dataclass, field, replace
from pathlib import Path

from pydantic import JsonValue, TypeAdapter
import pytest

from gateway.platforms.physique_checkin import PhysiqueCheckinBridge
from gateway.platforms.physique_checkin_config import PhysiqueCheckinConfig
from gateway.platforms.physique_checkin_bindings import (
    BindingStore,
    BindingStoreError,
    CursorIdentity,
    IngressIdentity,
    IngressKind,
    ProjectionPhase,
    TelegramProjection,
    WizardBinding,
)


NOW = 1_900_000_000
EXPIRES = NOW + 600
SESSION = "0123456789abcdef0123456789abcdef"
SOURCE = CursorIdentity(SESSION, "calories", 4)
TARGET = CursorIdentity(SESSION, "macros", 5)
_JSON_OBJECT = TypeAdapter(dict[str, JsonValue])


@dataclass(frozen=True, slots=True)
class _DomainResult:
    accepted: bool
    publication: object | None


@dataclass(slots=True)
class _Transition:
    ingress: IngressIdentity = field(
        default_factory=lambda: IngressIdentity(
            41, IngressKind.CALLBACK, 101, 7, -1007, 3,
        ),
    )
    source: CursorIdentity = SOURCE
    target: CursorIdentity = TARGET
    expires_at: int = EXPIRES
    accepted: bool = True
    error: Exception | None = None
    calls: int = 0

    def commit(self) -> _DomainResult:
        self.calls += 1
        if self.error is not None:
            raise self.error
        publication: object | None = (
            {"rendered": "transient customer card"} if self.accepted else None
        )
        return _DomainResult(self.accepted, publication)


@dataclass(frozen=True, slots=True)
class _Receipt:
    message_id: int


@dataclass(slots=True)
class _Transport:
    outcome: _Receipt | BaseException = field(
        default_factory=lambda: _Receipt(202),
    )
    preflight_error: Exception | None = None
    prepare_calls: int = 0
    send_calls: int = 0

    def prepare(self, publication: object) -> object:
        self.prepare_calls += 1
        if self.preflight_error is not None:
            raise self.preflight_error
        return publication

    async def send(self, prepared: object) -> _Receipt:
        _ = prepared
        self.send_calls += 1
        if isinstance(self.outcome, BaseException):
            raise self.outcome
        return self.outcome


@dataclass(frozen=True, slots=True)
class _BridgeResult:
    session_id: str = SESSION
    version: int = 4
    step: str = "calories"
    status: str = "advanced"


@dataclass(frozen=True, slots=True)
class _BridgeSession:
    flow: str = "nutrition_daily"
    step: str = "calories"
    answers: dict[str, str] = field(default_factory=dict)


class _BridgeStorage:
    def load(self, session_id: str) -> _BridgeSession:
        assert session_id == SESSION
        return _BridgeSession()


class _BridgeService:
    def __init__(self) -> None:
        self.answers: list[tuple[str, int, str, str | None]] = []
        self._storage: _BridgeStorage = _BridgeStorage()

    def start_nutrition(self, context: object, kst_day: str) -> _BridgeResult:
        _ = (context, kst_day)
        return _BridgeResult()

    def answer(
        self,
        context: object,
        session_id: str,
        version: int,
        action: str,
        value: str | None = None,
    ) -> _BridgeResult:
        _ = context
        self.answers.append((session_id, version, action, value))
        return _BridgeResult(version=5, step="macros")


def _store(tmp_path: Path) -> BindingStore:
    return BindingStore(tmp_path / "private" / "bindings.json")


def _phase(store: BindingStore) -> ProjectionPhase:
    return store.load_projections(NOW)[0].phase


def _stale_cursor(session_id: str) -> CursorIdentity | None:
    return CursorIdentity(session_id, "meals", 6)


def test_characterization_bridge_accepted_answer_advances_cas_and_binding_without_projection(
    tmp_path: Path,
) -> None:
    """Baseline: the bridge mutates domain/binding but has no publication journal."""
    config = PhysiqueCheckinConfig("7", "-1007", "3", 600, False, False)
    service = _BridgeService()
    path = tmp_path / "bindings.json"
    bridge = PhysiqueCheckinBridge(config, service=service, binding_path=path)
    launcher = bridge.open_launcher(
        "nutrition_daily", message_id="101", now_epoch=NOW,
    )
    assert launcher.callback_data is not None
    opened = bridge.handle_callback(
        launcher.callback_data, "7", "-1007", "3", "101", now_epoch=NOW,
    )
    assert opened.accepted

    reply = bridge.handle_text("2300", "7", "-1007", "3", now_epoch=NOW)

    assert reply is not None and reply.accepted
    assert service.answers == [(SESSION, 4, "value", "2300")]
    bindings, active_session_id = BindingStore(path).load("7", "-1007", "3", NOW)
    assert active_session_id == SESSION
    assert bindings[SESSION] == WizardBinding(
        SESSION, "7", "-1007", "3", "macros", 5, "101", EXPIRES, True,
    )
    assert BindingStore(path).load_projections(NOW) == ()


def test_typed_ingress_identity_rejects_malformed_and_content_values() -> None:
    with pytest.raises(ValueError):
        _ = IngressIdentity(-1, IngressKind.CALLBACK, 1, 1, -1, 0)
    assert IngressIdentity.from_dict({
        "update_id": 1,
        "kind": "raw callback",
        "message_id": 1,
        "actor_id": 1,
        "chat_id": -1,
        "topic_id": 0,
    }) is None


@pytest.mark.asyncio
async def test_happy_transition_is_delivered_with_exactly_one_transport_call(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    transition, transport, store = _Transition(), _Transport(), _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=NOW)

    assert result.disposition is StepperDisposition.DELIVERED
    assert result.poll_terminal and not result.ui_recovery_required
    assert transition.calls == transport.prepare_calls == transport.send_calls == 1
    projection = store.load_projections(NOW)[0]
    assert projection.phase is ProjectionPhase.DELIVERED
    assert projection.receipt_message_id == 202


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("initial_phase", "receipt"),
    (
        (ProjectionPhase.PREPARED, None),
        (ProjectionPhase.DOMAIN_COMMITTED, None),
        (ProjectionPhase.SEND_STARTED, None),
        (ProjectionPhase.DELIVERY_FAILED, None),
        (ProjectionPhase.DELIVERY_UNCERTAIN, None),
        (ProjectionPhase.DELIVERED, 202),
    ),
)
async def test_restart_from_every_phase_is_observation_only(
    tmp_path: Path, initial_phase: ProjectionPhase, receipt: int | None,
) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    store, transition, transport = _store(tmp_path), _Transition(), _Transport()
    projection = TelegramProjection(transition.ingress, SOURCE, TARGET, initial_phase, EXPIRES, receipt)
    # Install a legal history, because the store intentionally rejects phase jumps.
    prepared = replace(projection, phase=ProjectionPhase.PREPARED, receipt_message_id=None)
    _ = store.record_projection(prepared, now_epoch=NOW)
    if initial_phase is not ProjectionPhase.PREPARED:
        committed = replace(prepared, phase=ProjectionPhase.DOMAIN_COMMITTED)
        _ = store.record_projection(committed, now_epoch=NOW)
        if initial_phase not in {ProjectionPhase.DOMAIN_COMMITTED, ProjectionPhase.DELIVERY_FAILED, ProjectionPhase.DELIVERY_UNCERTAIN}:
            started = replace(committed, phase=ProjectionPhase.SEND_STARTED)
            _ = store.record_projection(started, now_epoch=NOW)
        _ = store.record_projection(projection, now_epoch=NOW)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=NOW)

    expected = StepperDisposition.DELIVERED if initial_phase is ProjectionPhase.DELIVERED else StepperDisposition.RECOVERY_REQUIRED
    assert result.disposition is expected
    assert result.poll_terminal
    assert transition.calls == transport.prepare_calls == transport.send_calls == 0


@pytest.mark.asyncio
async def test_duplicate_before_domain_commit_never_commits_or_sends(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import TelegramPhysiqueCheckinStepper
    store, transition = _store(tmp_path), _Transition()
    _ = store.record_projection(TelegramProjection(transition.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES), now_epoch=NOW)
    transport = _Transport()

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=NOW)

    assert result.poll_terminal and result.ui_recovery_required
    assert transition.calls == transport.send_calls == 0


@pytest.mark.asyncio
async def test_crash_after_domain_commit_is_never_automatically_sent(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import TelegramPhysiqueCheckinStepper
    store, transition = _store(tmp_path), _Transition()
    prepared = TelegramProjection(transition.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES)
    _ = store.record_projection(prepared, now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.DOMAIN_COMMITTED), now_epoch=NOW)
    transport = _Transport()

    observed = TelegramPhysiqueCheckinStepper(store, transport).observe_session(SESSION, now_epoch=NOW)

    assert observed is not None and observed.ui_recovery_required
    assert transport.send_calls == 0


@pytest.mark.asyncio
async def test_preflight_failure_is_definite_and_calls_send_zero_times(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper, TransportPreflightError
    transport = _Transport(preflight_error=TransportPreflightError("local render rejected"))
    store = _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(_Transition(), now_epoch=NOW)

    assert result.disposition is StepperDisposition.DELIVERY_FAILED
    assert _phase(store) is ProjectionPhase.DELIVERY_FAILED
    assert transport.send_calls == 0


@pytest.mark.asyncio
async def test_definite_provider_rejection_is_failed_after_one_call(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import DeliveryRejected, StepperDisposition, TelegramPhysiqueCheckinStepper
    transport, store = _Transport(DeliveryRejected("provider rejected before acceptance")), _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(_Transition(), now_epoch=NOW)

    assert result.disposition is StepperDisposition.DELIVERY_FAILED
    assert _phase(store) is ProjectionPhase.DELIVERY_FAILED
    assert transport.send_calls == 1


@pytest.mark.asyncio
@pytest.mark.parametrize("error", [RuntimeError("socket lost after write"), asyncio.CancelledError()])
async def test_provider_started_without_receipt_is_delivery_uncertain(tmp_path: Path, error: BaseException) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    transport, store = _Transport(error), _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(_Transition(), now_epoch=NOW)

    assert result.disposition is StepperDisposition.DELIVERY_UNCERTAIN
    assert _phase(store) is ProjectionPhase.DELIVERY_UNCERTAIN
    assert result.poll_terminal and result.ui_recovery_required


@pytest.mark.asyncio
async def test_explicit_resume_is_ui_only_and_never_replays_transport(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import TelegramPhysiqueCheckinStepper
    store, transition, transport = _store(tmp_path), _Transition(), _Transport()
    prepared = TelegramProjection(transition.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES)
    _ = store.record_projection(prepared, now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.DOMAIN_COMMITTED), now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.DELIVERY_FAILED), now_epoch=NOW)

    result = TelegramPhysiqueCheckinStepper(store, transport).resume(41, now_epoch=NOW)

    assert result is not None and result.ui_recovery_required
    assert transport.send_calls == 0


@pytest.mark.asyncio
async def test_stale_cursor_fails_closed_before_domain_or_transport(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    transition, transport = _Transition(), _Transport()
    stepper = TelegramPhysiqueCheckinStepper(
        _store(tmp_path), transport, current_cursor=_stale_cursor,
    )

    result = await stepper.execute(transition, now_epoch=NOW)

    assert result.disposition is StepperDisposition.STALE
    assert transition.calls == transport.send_calls == 0


@pytest.mark.asyncio
async def test_typed_text_after_uncertainty_only_observes_session(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import TelegramPhysiqueCheckinStepper
    store, callback = _store(tmp_path), _Transition()
    prepared = TelegramProjection(callback.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES)
    _ = store.record_projection(prepared, now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.DOMAIN_COMMITTED), now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.SEND_STARTED), now_epoch=NOW)
    _ = store.record_projection(replace(prepared, phase=ProjectionPhase.DELIVERY_UNCERTAIN), now_epoch=NOW)
    text_identity = IngressIdentity(42, IngressKind.TEXT, 102, 7, -1007, 3)
    transport = _Transport()

    result = TelegramPhysiqueCheckinStepper(store, transport).observe_ingress(text_identity, SESSION, now_epoch=NOW)

    assert result is not None and result.ui_recovery_required and result.poll_terminal
    assert transport.send_calls == 0
    assert len(store.load_projections(NOW)) == 1


@pytest.mark.asyncio
async def test_terminal_save_failure_never_rolls_back_domain_or_replays(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    store, transition, transport = _store(tmp_path), _Transition(), _Transport()
    original = store.record_projection

    def fail_delivered(projection: TelegramProjection, *, now_epoch: int | None = None) -> bool:
        if projection.phase is ProjectionPhase.DELIVERED:
            raise BindingStoreError("post-send disk failure")
        return original(projection, now_epoch=now_epoch)

    monkeypatch.setattr(store, "record_projection", fail_delivered)
    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=NOW)

    assert result.disposition is StepperDisposition.DELIVERY_UNCERTAIN
    assert transition.calls == transport.send_calls == 1
    assert _phase(store) is ProjectionPhase.SEND_STARTED
    again = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=NOW)
    assert again.poll_terminal and transport.send_calls == 1 and transition.calls == 1


@pytest.mark.asyncio
async def test_rejected_domain_transition_is_terminal_without_transport(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    transport, store = _Transport(), _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(_Transition(accepted=False), now_epoch=NOW)

    assert result.disposition is StepperDisposition.DOMAIN_REJECTED
    assert _phase(store) is ProjectionPhase.PREPARED
    assert result.ui_recovery_required
    assert transport.prepare_calls == transport.send_calls == 0


@pytest.mark.asyncio
async def test_expired_transition_is_terminal_without_persistence_or_calls(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    transition, transport, store = _Transition(), _Transport(), _store(tmp_path)

    result = await TelegramPhysiqueCheckinStepper(store, transport).execute(transition, now_epoch=EXPIRES)

    assert result.disposition is StepperDisposition.EXPIRED
    assert result.poll_terminal and transition.calls == transport.send_calls == 0
    assert store.load_projections(NOW) == ()


@pytest.mark.asyncio
async def test_corrupt_or_conflicting_projection_fails_closed(tmp_path: Path) -> None:
    from gateway.platforms.telegram_physique_checkin_stepper import StepperDisposition, TelegramPhysiqueCheckinStepper
    path = tmp_path / "private" / "bindings.json"
    path.parent.mkdir(mode=0o700)
    _ = path.write_text('{"version":2,"active_session_id":null,"bindings":[],"projections":[', encoding="utf-8")
    os.chmod(path, 0o600)
    transport = _Transport()
    corrupt = await TelegramPhysiqueCheckinStepper(BindingStore(path), transport).execute(_Transition(), now_epoch=NOW)
    assert corrupt.disposition is StepperDisposition.STORE_FAILURE and transport.send_calls == 0

    clean = _store(tmp_path / "other")
    first = _Transition()
    _ = clean.record_projection(TelegramProjection(first.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES), now_epoch=NOW)
    conflict = _Transition()
    conflict.ingress = replace(conflict.ingress, kind=IngressKind.TEXT)
    failed = await TelegramPhysiqueCheckinStepper(clean, transport).execute(conflict, now_epoch=NOW)
    assert failed.disposition is StepperDisposition.STORE_FAILURE and transport.send_calls == 0


def test_projection_file_contains_no_customer_content_callback_or_hash(tmp_path: Path) -> None:
    store, transition = _store(tmp_path), _Transition()
    _ = store.record_projection(TelegramProjection(transition.ingress, SOURCE, TARGET, ProjectionPhase.PREPARED, EXPIRES), now_epoch=NOW)
    raw = (tmp_path / "private" / "bindings.json").read_text(encoding="utf-8")
    assert "transient customer card" not in raw
    assert "callback_data" not in raw
    assert "hash" not in raw
    payload = _JSON_OBJECT.validate_json(raw)
    projections = payload["projections"]
    assert isinstance(projections, list) and projections
    first = projections[0]
    assert isinstance(first, dict)
    assert first["phase"] == "prepared"