from __future__ import annotations

import asyncio
import json
import logging
import time
from types import SimpleNamespace

from app.ws_gateway.session_audio import (
    enable_client_audio,
    notify_client_audio_enabled,
    stream_agent_audio_when_ready,
    watch_mock_stop_after_first_audio,
)
from app.ws_gateway.session_lifecycle import RoomLifecycleState


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 _FakeLocalParticipant:
    def __init__(self) -> None:
        self.published: list[dict] = []

    async def publish_data(self, payload: str, **kwargs) -> None:
        self.published.append({"payload": json.loads(payload), "kwargs": kwargs})


class _FakeRoom:
    def __init__(self) -> None:
        self.local_participant = _FakeLocalParticipant()


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


class _FakeWebSocket:
    def __init__(self) -> None:
        self.text_messages: list[dict] = []
        self.closed = False

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

    async def close(self) -> None:
        self.closed = True


def test_enable_client_audio_sets_gate_and_emits_timeline() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        logger = logging.getLogger("test.session_audio.enable")
        client_audio_enabled = asyncio.Event()

        await enable_client_audio(
            logger=logger,
            timeline=timeline,
            client_audio_enabled=client_audio_enabled,
            room_name="room-dev-123",
            protocol="PRT-1",
        )

        assert client_audio_enabled.is_set()
        assert timeline.events == [
            ("client_audio_enabled", {}),
        ]

    asyncio.run(_run())


def test_enable_client_audio_is_idempotent() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        logger = logging.getLogger("test.session_audio.enable.idempotent")
        client_audio_enabled = asyncio.Event()

        await enable_client_audio(
            logger=logger,
            timeline=timeline,
            client_audio_enabled=client_audio_enabled,
            room_name="room-dev-123",
            protocol="PRT-1",
        )
        await enable_client_audio(
            logger=logger,
            timeline=timeline,
            client_audio_enabled=client_audio_enabled,
            room_name="room-dev-123",
            protocol="PRT-1",
        )

        assert client_audio_enabled.is_set()
        assert timeline.events == [("client_audio_enabled", {})]

    asyncio.run(_run())


def test_notify_client_audio_enabled_publishes_bridge_control_message() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        logger = logging.getLogger("test.session_audio.notify")
        room = _FakeRoom()
        track_published = asyncio.Event()
        track_published.set()
        agent_ready = asyncio.Event()
        agent_ready.set()
        lifecycle = RoomLifecycleState(agent_participant=_FakeParticipant("agent-001"))

        await notify_client_audio_enabled(
            room=room,
            logger=logger,
            timeline=timeline,
            room_name="room-dev-123",
            protocol="PRT-1",
            track_published=track_published,
            agent_ready=agent_ready,
            lifecycle=lifecycle,
        )

        assert room.local_participant.published == [
            {
                "payload": {
                    "type": "client_audio_enabled",
                    "room": "room-dev-123",
                    "protocol": "PRT-1",
                },
                "kwargs": {
                    "reliable": True,
                    "destination_identities": ["agent-001"],
                    "topic": "bridge.control",
                },
            }
        ]
        assert timeline.events == [
            (
                "bridge_control_sent",
                {"control_type": "client_audio_enabled", "agent_identity": "agent-001"},
            )
        ]

    asyncio.run(_run())


def test_stream_agent_audio_when_ready_uses_lifecycle_participant() -> None:
    async def _run() -> None:
        agent_ready = asyncio.Event()
        agent_ready.set()
        participant = _FakeParticipant("agent-001")
        lifecycle = RoomLifecycleState(agent_participant=participant)
        calls: list[dict] = []

        async def _stream_agent_audio(agent_q, agent_participant, activity, timeline=None) -> None:
            calls.append(
                {
                    "agent_q": agent_q,
                    "agent_participant": agent_participant,
                    "activity": activity,
                    "timeline": timeline,
                }
            )

        queue: asyncio.Queue[bytes] = asyncio.Queue()
        activity = SimpleNamespace(last_client_in=0.0, last_agent_out=0.0)
        timeline = _FakeTimeline()

        await stream_agent_audio_when_ready(
            agent_ready=agent_ready,
            lifecycle=lifecycle,
            agent_q=queue,
            activity=activity,
            timeline=timeline,
            stream_agent_audio=_stream_agent_audio,
        )

        assert calls == [
            {
                "agent_q": queue,
                "agent_participant": participant,
                "activity": activity,
                "timeline": timeline,
            }
        ]

    asyncio.run(_run())


def test_stream_agent_audio_when_ready_repeats_for_reconnected_agent() -> None:
    async def _run() -> None:
        agent_ready = asyncio.Event()
        agent_ready.set()
        first = _FakeParticipant("agent-001")
        second = _FakeParticipant("agent-002")
        lifecycle = RoomLifecycleState(agent_participant=first)
        calls: list[str] = []
        second_call = asyncio.Event()

        async def _stream_agent_audio(agent_q, agent_participant, activity, timeline=None) -> None:
            calls.append(agent_participant.identity)
            if agent_participant is first:
                lifecycle.agent_participant = None
                lifecycle.agent_generation += 1
                lifecycle.agent_connected.clear()

                async def _reconnect() -> None:
                    await asyncio.sleep(0.01)
                    lifecycle.agent_participant = second
                    lifecycle.agent_generation += 1
                    lifecycle.agent_connected.set()

                asyncio.create_task(_reconnect())
                return

            second_call.set()
            await asyncio.Event().wait()

        queue: asyncio.Queue[bytes] = asyncio.Queue()
        activity = SimpleNamespace(last_client_in=0.0, last_agent_out=0.0)
        task = asyncio.create_task(
            stream_agent_audio_when_ready(
                agent_ready=agent_ready,
                lifecycle=lifecycle,
                agent_q=queue,
                activity=activity,
                timeline=None,
                stream_agent_audio=_stream_agent_audio,
                repeat=True,
            )
        )

        await asyncio.wait_for(second_call.wait(), timeout=1.0)
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)

        assert calls == ["agent-001", "agent-002"]

    asyncio.run(_run())


def test_watch_mock_stop_after_first_audio_sends_resolved_terminal_stop() -> None:
    async def _run() -> None:
        timeline = _FakeTimeline()
        logger = logging.getLogger("test.session_audio.mock_stop")
        ws = _FakeWebSocket()
        first_agent_audio_sent = asyncio.Event()
        activity = SimpleNamespace(last_agent_out=time.monotonic())
        agent_q: asyncio.Queue[bytes] = asyncio.Queue()

        task = asyncio.create_task(
            watch_mock_stop_after_first_audio(
                ws=ws,
                logger=logger,
                timeline=timeline,
                room_name="room-dev-123",
                protocol="PRT-1",
                first_agent_audio_sent=first_agent_audio_sent,
                activity=activity,
                agent_q=agent_q,
                silence_s=0.01,
                reason="stage_done",
            )
        )

        first_agent_audio_sent.set()
        await asyncio.sleep(0.03)
        await task

        assert ws.text_messages == [
            {
                "type": "stop",
                "data": {
                    "status": "stop_resolvido_e_finalizado",
                    "reason": "stage_done",
                    "phase": "in_session",
                },
            }
        ]
        assert ws.closed is True
        assert timeline.events == [
            ("mock_stop_after_first_audio_armed", {"silence_ms": 10, "reason": "stage_done"}),
            ("mock_stop_after_first_audio_sent", {"reason": "stage_done"}),
        ]

    asyncio.run(_run())
