from __future__ import annotations

import asyncio
import json
import logging

from app.ws_gateway.session_lifecycle import (
    RoomLifecycleState,
    register_room_lifecycle_handlers,
    watch_call_done,
)


class _FakeRoom:
    def __init__(self) -> None:
        self.handlers = {}

    def on(self, event_name: str):
        def _decorator(fn):
            self.handlers[event_name] = fn
            return fn

        return _decorator


class _FakeTimeline:
    def __init__(self) -> None:
        self.events: list[tuple[str, dict]] = []

    def emit(self, event: str, **fields) -> None:
        self.events.append((event, fields))


class _FakeParticipant:
    def __init__(self, identity: str) -> None:
        self.identity = identity


class _FakePacket:
    def __init__(self, topic: str, data) -> None:
        self.topic = topic
        self.data = data


class _FakeWebSocket:
    def __init__(self) -> None:
        self.sent: list[dict] = []
        self.closed = 0
        self.close_calls: list[dict] = []

    async def send_text(self, payload: str) -> None:
        self.sent.append(json.loads(payload))

    async def close(self, code: int = 1000, reason: str = "") -> None:
        self.closed += 1
        self.close_calls.append({"code": code, "reason": reason})


def test_register_room_lifecycle_handlers_updates_state_and_emits_timeline() -> None:
    room = _FakeRoom()
    timeline = _FakeTimeline()
    logger = logging.getLogger("test.session_lifecycle.register")
    call_done = asyncio.Event()
    agent_ready = asyncio.Event()
    state = RoomLifecycleState()

    register_room_lifecycle_handlers(
        room=room,
        logger=logger,
        timeline=timeline,
        room_name="room-dev-123",
        protocol="PRT-1",
        dispatch_started_at=100.0,
        agent_ready=agent_ready,
        call_done=call_done,
        state=state,
        is_agent=lambda participant: str(getattr(participant, "identity", "")).startswith("agent"),
        monotonic_fn=lambda: 100.123,
    )

    participant = _FakeParticipant("agent-001")
    room.handlers["participant_connected"](participant)
    room.handlers["data_received"](
        _FakePacket("agent.stage", b'{"stage":"DONE","reason":"completed"}')
    )
    room.handlers["participant_disconnected"](participant)

    assert agent_ready.is_set()
    assert call_done.is_set()
    assert state.agent_participant is None
    assert state.done_payload == {"stage": "DONE", "reason": "completed"}
    assert timeline.events == [
        ("agent_join", {"agent_identity": "agent-001", "dispatch_dt_ms": 123}),
        (
            "done_packet_received",
            {"reason": "completed", "payload": {"stage": "DONE", "reason": "completed"}},
        ),
        ("agent_leave", {"agent_identity": "agent-001"}),
    ]


def test_agent_disconnect_without_done_notifies_recovery_handler() -> None:
    async def _run() -> None:
        room = _FakeRoom()
        timeline = _FakeTimeline()
        logger = logging.getLogger("test.session_lifecycle.agent_disconnect")
        call_done = asyncio.Event()
        agent_ready = asyncio.Event()
        state = RoomLifecycleState()
        calls: list[dict] = []

        async def _on_agent_disconnect(agent_identity: str, generation: int) -> None:
            calls.append({"agent_identity": agent_identity, "generation": generation})

        register_room_lifecycle_handlers(
            room=room,
            logger=logger,
            timeline=timeline,
            room_name="room-dev-123",
            protocol="PRT-1",
            dispatch_started_at=100.0,
            agent_ready=agent_ready,
            call_done=call_done,
            state=state,
            is_agent=lambda participant: str(getattr(participant, "identity", "")).startswith("agent"),
            on_agent_disconnect=_on_agent_disconnect,
            monotonic_fn=lambda: 100.123,
        )

        participant = _FakeParticipant("agent-001")
        room.handlers["participant_connected"](participant)
        connected_generation = state.agent_generation
        room.handlers["participant_disconnected"](participant)
        await asyncio.sleep(0)

        assert state.agent_participant is None
        assert not state.agent_connected.is_set()
        assert state.agent_generation == connected_generation + 1
        assert calls == [
            {"agent_identity": "agent-001", "generation": connected_generation + 1}
        ]

    asyncio.run(_run())


def test_watch_call_done_sends_stop_and_closes_socket() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        ws = _FakeWebSocket()
        logger = logging.getLogger("test.session_lifecycle.done")
        call_done = asyncio.Event()
        state = RoomLifecycleState(done_payload={"reason": "finished"})
        call_done.set()

        await watch_call_done(
            call_done=call_done,
            state=state,
            ws=ws,
            logger=logger,
            room_name="room-dev-123",
            protocol="PRT-1",
            timeline=timeline,
        )

        assert ws.sent == [
            {
                "type": "stop",
                "data": {
                    "status": "stop_resolvido_e_finalizado",
                    "reason": "finished",
                    "phase": "in_session",
                },
            }
        ]
        assert ws.closed == 1
        assert ws.close_calls == [{"code": 1000, "reason": "call_done:finished"}]
        assert timeline.events == [
            ("call_done", {"reason": "finished"}),
            ("stop_sent", {"reason": "finished"}),
        ]

    asyncio.run(_run())


def test_agent_debug_packet_is_forwarded_when_enabled() -> None:
    async def _run() -> None:
        room = _FakeRoom()
        ws = _FakeWebSocket()
        register_room_lifecycle_handlers(
            room=room,
            logger=logging.getLogger("test.session_lifecycle.debug"),
            timeline=_FakeTimeline(),
            room_name="room-load-1",
            protocol="LOAD-1",
            dispatch_started_at=1.0,
            agent_ready=asyncio.Event(),
            call_done=asyncio.Event(),
            state=RoomLifecycleState(),
            is_agent=lambda _participant: False,
            ws=ws,
            debug_events_enabled=True,
        )
        event = {
            "type": "debug_event",
            "version": 1,
            "source": "agent",
            "event": "agent.speech.finished",
            "data": {"text": "Resposta", "stage": "ARGUMENTATION"},
        }
        room.handlers["data_received"](
            _FakePacket("agent.debug", json.dumps(event).encode("utf-8"))
        )
        await asyncio.sleep(0)

        assert ws.sent == [{**event, "stress_test": True}]

    asyncio.run(_run())


def test_agent_debug_packet_is_not_forwarded_when_disabled() -> None:
    async def _run() -> None:
        room = _FakeRoom()
        ws = _FakeWebSocket()
        register_room_lifecycle_handlers(
            room=room,
            logger=logging.getLogger("test.session_lifecycle.debug.disabled"),
            timeline=_FakeTimeline(),
            room_name="room-regular-1",
            protocol="REGULAR-1",
            dispatch_started_at=1.0,
            agent_ready=asyncio.Event(),
            call_done=asyncio.Event(),
            state=RoomLifecycleState(),
            is_agent=lambda _participant: False,
            ws=ws,
            debug_events_enabled=False,
        )
        event = {
            "type": "debug_event",
            "version": 1,
            "source": "agent",
            "event": "stt.completed",
            "data": {"duration_ms": 100},
        }
        room.handlers["data_received"](
            _FakePacket("agent.debug", json.dumps(event).encode("utf-8"))
        )
        await asyncio.sleep(0)

        assert ws.sent == []

    asyncio.run(_run())


def test_watch_call_done_maps_no_user_response_to_long_silence_status() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        ws = _FakeWebSocket()
        logger = logging.getLogger("test.session_lifecycle.done.long_silence")
        call_done = asyncio.Event()
        state = RoomLifecycleState(done_payload={"reason": "no_user_response"})
        call_done.set()

        await watch_call_done(
            call_done=call_done,
            state=state,
            ws=ws,
            logger=logger,
            room_name="room-dev-123",
            protocol="PRT-1",
            timeline=timeline,
        )

        assert ws.sent == [
            {
                "type": "stop",
                "data": {
                    "status": "stop_silencio_longo",
                    "reason": "no_user_response",
                    "phase": "in_session",
                },
            }
        ]
        assert ws.closed == 1

    asyncio.run(_run())


def test_watch_call_done_uses_default_status_when_reason_is_not_exact_match() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        ws = _FakeWebSocket()
        logger = logging.getLogger("test.session_lifecycle.done.default")
        call_done = asyncio.Event()
        state = RoomLifecycleState(done_payload={"reason": "finished"})
        call_done.set()

        await watch_call_done(
            call_done=call_done,
            state=state,
            ws=ws,
            logger=logger,
            room_name="room-dev-123",
            protocol="PRT-1",
            timeline=timeline,
        )

        assert ws.sent == [
            {
                "type": "stop",
                "data": {
                    "status": "stop_resolvido_e_finalizado",
                    "reason": "finished",
                    "phase": "in_session",
                },
            }
        ]
        assert ws.closed == 1

    asyncio.run(_run())


def test_watch_call_done_preserves_terminal_status_from_agent_payload() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        ws = _FakeWebSocket()
        logger = logging.getLogger("test.session_lifecycle.done.custom")
        call_done = asyncio.Event()
        state = RoomLifecycleState(
            done_payload={
                "stage": "DONE",
                "status": "stop_agent_backend_unavailable",
                "reason": "resource_unhealthy",
                "resource": "agent_backend",
                "failed_resources": ["agent_backend"],
                "phase": "in_session",
            }
        )
        call_done.set()

        await watch_call_done(
            call_done=call_done,
            state=state,
            ws=ws,
            logger=logger,
            room_name="room-dev-123",
            protocol="PRT-1",
            timeline=timeline,
        )

        assert ws.sent == [
            {
                "type": "stop",
                "data": {
                    "status": "stop_agent_backend_unavailable",
                    "reason": "resource_unhealthy",
                    "resource": "agent_backend",
                    "failed_resources": ["agent_backend"],
                    "phase": "in_session",
                },
            }
        ]
        assert ws.closed == 1

    asyncio.run(_run())
