#!/usr/bin/env python3
"""Zero-network manual oracle for the Telegram Todo 5 durable-stepper path."""

from __future__ import annotations

import asyncio
import json
import logging
import socket
import sys
from dataclasses import dataclass, field
from datetime import date, time
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import NoReturn, final, override


REPOSITORY = Path(__file__).resolve().parents[1]
PROFILE_PACKAGE = REPOSITORY / "dualcoach" / "profile"
for source_path in (REPOSITORY, PROFILE_PACKAGE):
    source_text = str(source_path)
    if source_text not in sys.path:
        sys.path.insert(0, source_text)

from checkin_cli.customer_coaching import (
    AiProcessingConsent,
    CustomerRegistry,
    CustomerRuntime,
    CustomerSchedule,
    CustomerSpec,
    PlanWeek,
    TelegramAddress,
    TwelveWeekPlan,
)
from checkin_cli.wizard import WizardService
from gateway.config import PlatformConfig
from gateway.platforms.base import MessageEvent
from gateway.platforms.nutrition_coaching import (
    CallbackInput,
    CustomerTransition,
    IncomingAddress,
    NutritionCoachingCoordinator,
    ResolvedCustomer,
)
from gateway.platforms.physique_checkin import (
    CallbackData,
    PhysiqueCheckinBridge,
    WizardReply,
)
from gateway.platforms.physique_checkin_bindings import ProjectionPhase
from gateway.platforms.physique_checkin_config import PhysiqueCheckinConfig
from gateway.platforms.telegram import TelegramAdapter
from gateway.platforms.telegram_polling_receipts import (
    TelegramIngressReceiptStore,
    TelegramPollingReceiptGate,
)


ADDRESS = IncomingAddress("7", "-1007", "3")


@final
@dataclass(frozen=True, slots=True)
class _Actor:
    id: int


@final
@dataclass(frozen=True, slots=True)
class _Chat:
    id: int
    type: str = "supergroup"
    is_forum: bool = True


@final
@dataclass(slots=True)
class _TextMessage:
    text: str
    message_id: int
    from_user: _Actor
    chat: _Chat
    message_thread_id: int

    async def reply_text(
        self,
        text: str,
        reply_markup: object | None = None,
    ) -> None:
        _ = (text, reply_markup)
        raise AssertionError("manual oracle must not emit customer messages")


@final
@dataclass(frozen=True, slots=True)
class _TextUpdate:
    update_id: object
    message: _TextMessage

    @property
    def effective_message(self) -> _TextMessage:
        return self.message


@final
@dataclass(slots=True)
class _CallbackMessage:
    message_id: int
    chat: _Chat
    message_thread_id: int

    @property
    def chat_id(self) -> int:
        return self.chat.id


@final
@dataclass(slots=True)
class _CallbackQuery:
    data: str
    from_user: _Actor
    message: _CallbackMessage
    ack_fails: bool = False
    answers: int = 0
    edits: int = 0

    async def answer(self, text: str | None = None) -> None:
        _ = text
        self.answers += 1
        if self.ack_fails:
            raise RuntimeError("ack_unavailable")

    async def edit_message_text(
        self,
        text: str,
        reply_markup: object | None = None,
    ) -> None:
        _ = (text, reply_markup)
        self.edits += 1


@final
@dataclass(frozen=True, slots=True)
class _CallbackUpdate:
    update_id: object
    callback_query: _CallbackQuery


@final
@dataclass(frozen=True, slots=True)
class _BotUser:
    id: int = 0


@final
@dataclass(frozen=True, slots=True)
class _ParentChat:
    id: int = 0


@final
@dataclass(frozen=True, slots=True)
class _BotChat:
    is_direct_messages: bool = False
    parent_chat: _ParentChat | None = None


@final
@dataclass(frozen=True, slots=True)
class _BotMember:
    can_manage_direct_messages: bool = False


@final
class _NoNetworkBot:
    async def get_me(self) -> _BotUser:
        return _BotUser()

    async def get_chat(self, chat_id: int) -> _BotChat:
        _ = chat_id
        return _BotChat()

    async def get_chat_member(self, chat_id: int, user_id: int) -> _BotMember:
        _ = (chat_id, user_id)
        return _BotMember()


@final
@dataclass(frozen=True, slots=True)
class _IngressContext:
    bot: _NoNetworkBot = field(default_factory=_NoNetworkBot)


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


@final
class _Coordinator(NutritionCoachingCoordinator):
    _bridge: PhysiqueCheckinBridge
    _address: IncomingAddress
    _resolved: ResolvedCustomer
    text_calls: int
    callback_calls: int

    def __init__(
        self,
        bridge: PhysiqueCheckinBridge,
        address: IncomingAddress,
        customer: CustomerRuntime,
    ) -> None:
        customer.data_root.mkdir(parents=True, exist_ok=True)
        disabled_customer = CustomerRuntime(
            customer.spec.model_copy(update={"enabled": False}),
            customer.data_root,
        )
        super().__init__(
            customer.data_root.parent,
            CustomerRegistry(
                TelegramAddress(user_id="owner", chat_id="owner", topic_id="owner"),
                (disabled_customer,),
            ),
        )
        self._bridge = bridge
        self._address = address
        self._resolved = ResolvedCustomer(customer, bridge)
        self.text_calls = 0
        self.callback_calls = 0

    @override
    def resolve(self, address: IncomingAddress) -> ResolvedCustomer | None:
        return self._resolved if address == self._address else None

    @override
    def resolve_trainer(self, address: IncomingAddress) -> ResolvedCustomer | None:
        _ = address
        return None

    @override
    def trainer_private_active(
        self,
        address: IncomingAddress,
    ) -> ResolvedCustomer | None:
        _ = address
        return None

    @override
    def owns_space(self, chat_id: str, topic_id: str) -> bool:
        return (chat_id, topic_id) == (self._address.chat_id, self._address.topic_id)

    @override
    def handle_text(
        self,
        address: IncomingAddress,
        text: str,
    ) -> CustomerTransition:
        self.text_calls += 1
        reply = self._bridge.handle_text(
            text,
            owner_id=address.user_id,
            chat_id=address.chat_id,
            topic_id=address.topic_id,
        )
        if reply is None:
            reply = WizardReply(True, False, "no active check-in", None, None)
        return CustomerTransition(reply)

    @override
    def handle_callback(self, incoming: CallbackInput) -> CustomerTransition:
        self.callback_calls += 1
        reply = self._bridge.handle_callback(
            incoming.data,
            owner_id=incoming.address.user_id,
            chat_id=incoming.address.chat_id,
            topic_id=incoming.address.topic_id,
            message_id=incoming.message_id,
        )
        return CustomerTransition(reply)


@final
class _FakeBotAdapter(TelegramAdapter):
    _coordinator: NutritionCoachingCoordinator
    _sends: list[_SentPrompt]
    _send_outcomes: list[int | Exception]
    generic_events: int

    def __init__(self, coordinator: NutritionCoachingCoordinator) -> None:
        super().__init__(PlatformConfig(enabled=True, token="network-disabled", extra={}))
        self._coordinator = coordinator
        self._sends = []
        self._send_outcomes = []
        self.generic_events = 0

    @property
    def sends(self) -> tuple[_SentPrompt, ...]:
        return tuple(self._sends)

    def set_send_outcomes(self, outcomes: tuple[int | Exception, ...]) -> None:
        self._send_outcomes = list(outcomes)

    async def receive_text(self, update: _TextUpdate) -> None:
        await self._handle_text_message(update, _IngressContext())

    async def receive_callback(self, update: _CallbackUpdate) -> None:
        await self._handle_callback_query(update, _IngressContext())

    @override
    def _get_nutrition_coaching(self) -> NutritionCoachingCoordinator | None:
        return self._coordinator

    @override
    def _enqueue_text_event(self, event: MessageEvent) -> None:
        _ = event
        self.generic_events += 1

    @override
    async def _send_nutrition_topic(
        self,
        *,
        chat_id: object,
        topic_id: object,
        **kwargs: object,
    ) -> object:
        _ = (chat_id, topic_id)
        text = kwargs.get("text")
        if not isinstance(text, str) or not text:
            raise RuntimeError("fake transport requires prompt text")
        outcome = self._send_outcomes.pop(0) if self._send_outcomes else 200 + len(self._sends)
        if isinstance(outcome, Exception):
            raise outcome
        prompt = _SentPrompt(outcome)
        self._sends.append(prompt)
        return prompt


@final
class _SocketGuard:
    _original: object
    network_events: int

    def __init__(self) -> None:
        self._original = socket.socket
        self.network_events = 0

    def __enter__(self) -> _SocketGuard:
        setattr(socket, "socket", self._blocked_socket)
        return self

    def __exit__(self, exception_type: object, exception: object, traceback: object) -> bool:
        _ = (exception_type, exception, traceback)
        setattr(socket, "socket", self._original)
        return False

    def _blocked_socket(self, *args: object, **kwargs: object) -> NoReturn:
        _ = (args, kwargs)
        self.network_events += 1
        raise AssertionError("real socket use is forbidden in the manual oracle")


def _customer(root: Path) -> CustomerRuntime:
    weeks = tuple(
        PlanWeek(
            week=week,
            calories_kcal=2_300,
            protein_g=150,
            carbohydrate_g=280,
            fat_g=65,
            meal_structure=("meal",),
        )
        for week in range(1, 13)
    )
    spec = CustomerSpec(
        customer_key="customer-1",
        display_name="Customer",
        enabled=True,
        telegram=TelegramAddress(user_id="7", chat_id="-1007", topic_id="3"),
        schedule=CustomerSchedule(daily_time=time(8), weekly_weekday=0, monthly_day=1),
        ai_processing_consent=AiProcessingConsent(
            granted=True,
            recorded_on=date(2026, 1, 1),
            notice_version="privacy-v1",
        ),
        plan=TwelveWeekPlan(
            starts_on=date(2026, 1, 1),
            focus="nutrition_90_training_10",
            weeks=weeks,
        ),
    )
    return CustomerRuntime(spec, root / "customer")


def _bridge(root: Path) -> tuple[PhysiqueCheckinBridge, str]:
    bridge = PhysiqueCheckinBridge(
        PhysiqueCheckinConfig("7", "-1007", "3", 3_600, False, False),
        service=WizardService.for_standalone(root / "wizard"),
        binding_path=root / "bindings.json",
    )
    launcher = bridge.open_launcher("nutrition_daily", message_id="101")
    if launcher.callback_data is None:
        raise AssertionError("launcher callback missing")
    opened = bridge.handle_callback(
        launcher.callback_data,
        owner_id="7",
        chat_id="-1007",
        topic_id="3",
        message_id="101",
    )
    if not opened.accepted:
        raise AssertionError("launcher rejected")
    return bridge, launcher.callback_data


def _adapter(root: Path) -> tuple[_FakeBotAdapter, _Coordinator, PhysiqueCheckinBridge]:
    bridge, _launcher = _bridge(root)
    coordinator = _Coordinator(bridge, ADDRESS, _customer(root))
    return _FakeBotAdapter(coordinator), coordinator, bridge


def _text_update(update_id: object, text: str, *, message_id: int = 102) -> _TextUpdate:
    return _TextUpdate(
        update_id,
        _TextMessage(text, message_id, _Actor(7), _Chat(-1007), 3),
    )


def _callback_update(
    update_id: object,
    callback_data: str,
    *,
    actor_id: int = 7,
    topic_id: int = 3,
    message_id: int = 101,
    ack_fails: bool = False,
) -> _CallbackUpdate:
    return _CallbackUpdate(
        update_id,
        _CallbackQuery(
            callback_data,
            _Actor(actor_id),
            _CallbackMessage(message_id, _Chat(-1007), topic_id),
            ack_fails=ack_fails,
        ),
    )


def _active(bridge: PhysiqueCheckinBridge) -> tuple[str, int, str]:
    cursor = bridge.active_cursor_identity()
    if cursor is None:
        raise AssertionError("active cursor missing")
    snapshot = bridge.active_checkin_snapshot()
    if snapshot is None:
        raise AssertionError("active snapshot missing")
    flow = snapshot.get("flow")
    if not isinstance(flow, str):
        raise AssertionError("active flow missing")
    return cursor[0].step, cursor[0].version, flow


def _callback_for(bridge: PhysiqueCheckinBridge, action: str) -> str:
    prompt = bridge.active_prompt()
    if prompt is None:
        raise AssertionError("active prompt missing")
    for row in prompt.button_rows:
        for _label, callback_data in row:
            if callback_data.endswith(f":{action}"):
                return callback_data
    raise AssertionError("expected callback missing")


def _expect_no_advance(
    before: tuple[str, int, str],
    bridge: PhysiqueCheckinBridge,
    adapter: _FakeBotAdapter,
    sent_before: int,
) -> None:
    if _active(bridge) != before or len(adapter.sends) != sent_before:
        raise AssertionError("adversarial callback was not fail-closed")


async def _exercise() -> dict[str, bool | int | str]:
    payload: dict[str, bool | int | str] = {}

    def polling_blocked(update_id: int, reason: str) -> None:
        _ = (update_id, reason)
        raise AssertionError("polling blocked")

    with TemporaryDirectory() as temporary:
        root = Path(temporary)
        with _SocketGuard() as sockets:
            adapter, coordinator, bridge = _adapter(root / "primary")
            known = _text_update(501, "71.2")
            await adapter.receive_text(known)
            if _active(bridge)[:2] != ("calories", 1):
                raise AssertionError("known text did not advance once")
            known_sends1 = len(adapter.sends) == 1

            unknown = _callback_for(bridge, "u")
            unknown_update = _callback_update(
                502,
                unknown,
                message_id=adapter.sends[0].message_id,
                ack_fails=True,
            )
            await adapter.receive_callback(unknown_update)
            if unknown_update.callback_query.answers != 1 or unknown_update.callback_query.edits != 0:
                raise AssertionError("ack failure was not non-authoritative")
            if _active(bridge)[:2] != ("macros", 2):
                raise AssertionError("unknown callback did not advance once")
            unknown_sends1 = len(adapter.sends) == 2

            duplicate_before = _active(bridge)
            duplicate_sends = len(adapter.sends)
            await adapter.receive_text(known)
            duplicate_advances0 = _active(bridge) == duplicate_before and len(adapter.sends) == duplicate_sends

            current = bridge.active_cursor_identity()
            if current is None:
                raise AssertionError("missing active cursor for adversarial callbacks")
            callback = CallbackData(current[0].session_id, current[0].step, current[0].version, "u").encode()
            sent_before = len(adapter.sends)
            before = _active(bridge)
            await adapter.receive_callback(_callback_update(503, callback, actor_id=8, message_id=201))
            _expect_no_advance(before, bridge, adapter, sent_before)
            await adapter.receive_callback(_callback_update(504, callback, topic_id=4, message_id=201))
            _expect_no_advance(before, bridge, adapter, sent_before)
            await adapter.receive_callback(_callback_update(505, callback, message_id=999))
            _expect_no_advance(before, bridge, adapter, sent_before)
            wrong_version = CallbackData(
                current[0].session_id,
                current[0].step,
                current[0].version + 1,
                "u",
            ).encode()
            await adapter.receive_callback(_callback_update(506, wrong_version, message_id=201))
            _expect_no_advance(before, bridge, adapter, sent_before)
            leave_without_action = CallbackData(
                current[0].session_id,
                current[0].step,
                current[0].version,
                "a0",
            ).encode()
            await adapter.receive_callback(_callback_update(507, leave_without_action, message_id=201))
            _expect_no_advance(before, bridge, adapter, sent_before)

            resume_bridge, _launcher = _bridge(root / "resume")
            resumed = resume_bridge.open_launcher("nutrition_daily", message_id="802")
            if resumed.callback_data is None:
                raise AssertionError("explicit nutrition resume callback missing")
            resume_reply = resume_bridge.handle_callback(
                resumed.callback_data,
                owner_id="7",
                chat_id="-1007",
                topic_id="3",
                message_id="802",
            )
            if not resume_reply.accepted or _active(resume_bridge)[2] != "nutrition_daily":
                raise AssertionError("explicit resume did not target nutrition daily")

            uncertain_adapter, uncertain_coordinator, uncertain_bridge = _adapter(root / "uncertain")
            uncertain_adapter.set_send_outcomes((RuntimeError("write_outcome_unknown"),))
            uncertain = _text_update(701, "71.2")
            gate = TelegramPollingReceiptGate(
                TelegramIngressReceiptStore(root / "polling-receipts.json"),
                on_blocked=polling_blocked,
            )
            gate.captured((uncertain,))
            if await gate.begin(uncertain):
                raise AssertionError("fresh polling update was pre-receipted")
            await uncertain_adapter.receive_text(uncertain)
            gate.completed(uncertain)
            polling_unknown_terminal = await gate.permits_offset(702, running=True)
            projections = uncertain_bridge.binding_store.load_projections(0)
            if (
                len(projections) != 1
                or projections[0].phase is not ProjectionPhase.DELIVERY_UNCERTAIN
                or uncertain_coordinator.text_calls != 1
            ):
                raise AssertionError("uncertain send was not durably terminal")
            sends_before_recovery = len(uncertain_adapter.sends)
            await uncertain_adapter.receive_text(_text_update(702, "2300"))
            automatic_recovery_sends0 = len(uncertain_adapter.sends) == sends_before_recovery
            if _active(uncertain_bridge)[:2] != ("calories", 1):
                raise AssertionError("automatic uncertainty recovery advanced the domain")

            if coordinator.text_calls != 1 or coordinator.callback_calls != 4:
                raise AssertionError("manual oracle did not exercise expected ingress")
            if adapter.generic_events != 0 or uncertain_adapter.generic_events != 0:
                raise AssertionError("typed ingress reached generic batching")
            payload = {
                "status": "PASS",
                "known_sends1": known_sends1,
                "unknown_sends1": unknown_sends1,
                "duplicate_advances0": duplicate_advances0,
                "automatic_recovery_sends0": automatic_recovery_sends0,
                "polling_unknown_terminal": polling_unknown_terminal,
                "resume_target": "nutrition_daily",
                "network_events": sockets.network_events,
                "customer_messages": 0,
            }
            if not all(value is not False for value in payload.values()):
                raise AssertionError("manual oracle invariant failed")
        return payload


def main() -> None:
    logging.getLogger("gateway.platforms.telegram").disabled = True
    print(json.dumps(asyncio.run(_exercise()), separators=(",", ":"), sort_keys=True))


if __name__ == "__main__":
    main()