"""Tests for per-connection transport implementations.

Covers WebSocketConnectionTransport and TwilioConnectionTransport using
a real websockets server/client pair (no mocks).
"""

from __future__ import annotations

import asyncio
import base64
import json

import pytest
import websockets

from easycat.audio_format import PCM16_MONO_8K, PCM16_MONO_16K, AudioChunk
from easycat.events import DTMF, EventBus, PlaybackMarkAck
from easycat.transports.twilio_media import (
    TwilioConnectionTransport,
    pcm16_to_mulaw,
)
from easycat.transports.websocket import WebSocketConnectionTransport

from ._webrtc_fakes import _UsesPytestTcpPortFactory
from .conftest import make_chunk

pytestmark = pytest.mark.integration_socket

# ── Helpers ────────────────────────────────────────────────────────


def _twilio_start_msg(stream_sid: str = "MZ123", call_sid: str = "CA456") -> str:
    return json.dumps(
        {
            "event": "start",
            "streamSid": stream_sid,
            "start": {"streamSid": stream_sid, "callSid": call_sid},
        }
    )


def _twilio_media_msg(mulaw_data: bytes, stream_sid: str = "MZ123") -> str:
    payload = base64.b64encode(mulaw_data).decode("ascii")
    return json.dumps(
        {
            "event": "media",
            "streamSid": stream_sid,
            "media": {"payload": payload},
        }
    )


def _twilio_mark_msg(name: str, stream_sid: str = "MZ123") -> str:
    return json.dumps({"event": "mark", "streamSid": stream_sid, "mark": {"name": name}})


def _twilio_dtmf_msg(digit: str, stream_sid: str = "MZ123") -> str:
    return json.dumps({"event": "dtmf", "streamSid": stream_sid, "dtmf": {"digit": digit}})


def _twilio_stop_msg(stream_sid: str = "MZ123") -> str:
    return json.dumps({"event": "stop", "streamSid": stream_sid})


class _DummyTwilioWebSocket:
    request = object()

    async def send(self, _message: str) -> None:
        return None

    async def close(self) -> None:
        return None


class _DummyWebSocket:
    def __init__(self) -> None:
        self.request = object()


# ── WebSocketConnectionTransport tests ────────────────────────────


class TestWebSocketConnectionTransport(_UsesPytestTcpPortFactory):
    def test_request_exposes_underlying_handshake(self) -> None:
        ws = _DummyWebSocket()
        transport = WebSocketConnectionTransport(ws)  # type: ignore[arg-type]

        assert transport.request is ws.request

    @pytest.mark.asyncio
    async def test_connect_disconnect(self):
        port = self._unused_port()
        transport_holder: list[WebSocketConnectionTransport] = []

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            transport_holder.append(t)
            await t.connect()
            assert t.is_connected
            # Keep alive until client disconnects.
            await ws.wait_closed()

        server = await websockets.serve(handler, "127.0.0.1", port)
        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            ready = await ws.recv()
            assert json.loads(ready)["type"] == "ready"
            # Transport should be connected.
            await asyncio.sleep(0.05)
            assert transport_holder[0].is_connected

        # Client disconnected — give receive loop time to notice.
        await asyncio.sleep(0.1)
        transport = transport_holder[0]
        await transport.disconnect()
        assert not transport.is_connected

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_send_receive_audio(self):
        port = self._unused_port()
        received: list[AudioChunk] = []

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            await t.connect()
            async for chunk in t.receive_audio():
                received.append(chunk)
                if len(received) >= 3:
                    break
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            for _ in range(3):
                await ws.send(bytes(320))
            await asyncio.sleep(0.1)

        server.close()
        await server.wait_closed()
        assert len(received) == 3

    @pytest.mark.asyncio
    async def test_server_sends_audio_to_client(self):
        port = self._unused_port()

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            await t.connect()
            chunk = make_chunk(640)
            await t.send_audio(chunk)
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            fmt_msg = await asyncio.wait_for(ws.recv(), timeout=2.0)  # audio_format
            assert json.loads(fmt_msg)["type"] == "audio_format"
            data = await asyncio.wait_for(ws.recv(), timeout=2.0)
            assert isinstance(data, bytes)
            assert len(data) == 640

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_clear_audio_sends_client_playback_reset(self):
        port = self._unused_port()
        transport_ready = asyncio.Event()
        transport_holder: list[WebSocketConnectionTransport] = []

        async def handler(ws):
            transport = WebSocketConnectionTransport(ws)
            transport_holder.append(transport)
            await transport.connect()
            transport_ready.set()
            await ws.wait_closed()
            await transport.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            await asyncio.wait_for(transport_ready.wait(), timeout=2.0)

            await transport_holder[0].clear_audio()

            message = await asyncio.wait_for(ws.recv(), timeout=2.0)
            assert json.loads(message) == {"type": "clear"}

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_control_message_config(self):
        port = self._unused_port()
        transport_holder: list[WebSocketConnectionTransport] = []

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            transport_holder.append(t)
            await t.connect()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            await ws.send(json.dumps({"type": "config", "sample_rate": 24000}))
            await asyncio.sleep(0.1)
            assert transport_holder[0]._audio_format.sample_rate == 24000

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_invalid_sample_rate_config_is_ignored(self):
        transport = WebSocketConnectionTransport(object())  # type: ignore[arg-type]

        for sample_rate in (True, False, 0, 1, 7999, -16000, 384001, 16000.0, "16000", None):
            transport._handle_control_message(
                json.dumps({"type": "config", "sample_rate": sample_rate})
            )
            assert transport._audio_format.sample_rate == 16000

        transport._handle_control_message(json.dumps({"type": "config", "sample_rate": 48000}))
        assert transport._audio_format.sample_rate == 48000

    @pytest.mark.asyncio
    async def test_non_object_control_frame_does_not_end_receive_loop(self):
        port = self._unused_port()
        transport_holder: list[WebSocketConnectionTransport] = []

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            transport_holder.append(t)
            await t.connect()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            await ws.send("[]")
            await asyncio.sleep(0.1)
            transport = transport_holder[0]
            assert transport._receive_task is not None
            assert not transport._receive_task.done()
            assert transport.is_connected

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_client_disconnect_ends_receive(self):
        port = self._unused_port()
        received: list[AudioChunk] = []
        collect_done = asyncio.Event()

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            await t.connect()
            async for chunk in t.receive_audio():
                received.append(chunk)
            collect_done.set()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            await ws.send(bytes(320))
            await asyncio.sleep(0.05)

        # Client disconnected; handler should finish.
        await asyncio.wait_for(collect_done.wait(), timeout=2.0)
        assert len(received) == 1

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_disconnect_idempotent(self):
        port = self._unused_port()

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            await t.connect()
            await t.disconnect()
            await t.disconnect()  # Should not raise.

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.recv()  # ready
            await asyncio.sleep(0.1)

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_connect_idempotent(self):
        port = self._unused_port()
        ready_count = 0

        async def handler(ws):
            t = WebSocketConnectionTransport(ws)
            await t.connect()
            await t.connect()  # Should be a no-op.
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            ready = await ws.recv()
            assert json.loads(ready)["type"] == "ready"
            ready_count += 1
            # Second connect is a no-op, so no second ready message.
            try:
                await asyncio.wait_for(ws.recv(), timeout=0.2)
                ready_count += 1
            except (TimeoutError, websockets.exceptions.ConnectionClosed):
                pass

        server.close()
        await server.wait_closed()
        assert ready_count == 1


# ── TwilioConnectionTransport tests ──────────────────────────────


class TestTwilioConnectionTransport(_UsesPytestTcpPortFactory):
    def test_request_exposes_underlying_handshake(self) -> None:
        ws = _DummyTwilioWebSocket()
        transport = TwilioConnectionTransport(ws)  # type: ignore[arg-type]

        assert transport.request is ws.request

    def test_audio_contract_declares_distinct_tts_preference(self):
        transport = TwilioConnectionTransport(_DummyTwilioWebSocket())

        assert transport.audio_format == PCM16_MONO_16K
        assert transport.preferred_tts_output_format == PCM16_MONO_8K

    @pytest.mark.asyncio
    async def test_connect_disconnect(self):
        port = self._unused_port()
        transport_holder: list[TwilioConnectionTransport] = []

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            transport_holder.append(t)
            await t.connect()
            assert t.is_connected
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as _ws:
            await asyncio.sleep(0.05)
            assert transport_holder[0].is_connected

        await asyncio.sleep(0.1)
        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_receive_audio(self):
        port = self._unused_port()
        received: list[AudioChunk] = []

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            async for chunk in t.receive_audio():
                received.append(chunk)
                if len(received) >= 1:
                    break
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg())
            pcm_silence = bytes(320)
            mulaw_data = pcm16_to_mulaw(pcm_silence, source_rate=8000)
            await ws.send(_twilio_media_msg(mulaw_data))
            await asyncio.sleep(0.2)

        server.close()
        await server.wait_closed()
        assert len(received) == 1
        assert received[0].format.sample_rate == 16000

    @pytest.mark.asyncio
    async def test_send_audio(self):
        port = self._unused_port()
        send_ready = asyncio.Event()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            # Wait for the start message to be processed by the receive loop.
            await asyncio.sleep(0.1)
            chunk = make_chunk(640, sample_rate=16000)
            await t.send_audio(chunk)
            send_ready.set()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg("STREAM1"))
            await send_ready.wait()
            raw = await asyncio.wait_for(ws.recv(), timeout=2.0)
            msg = json.loads(raw)
            assert msg["event"] == "media"
            assert msg["streamSid"] == "STREAM1"
            payload = base64.b64decode(msg["media"]["payload"])
            assert len(payload) > 0

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_send_mark(self):
        port = self._unused_port()
        send_ready = asyncio.Event()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            # Wait for the receive loop to process the start message.
            await asyncio.sleep(0.1)
            name = await t.send_mark("test_mark")
            assert name == "test_mark"
            send_ready.set()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg("STREAM1"))
            await send_ready.wait()
            raw = await asyncio.wait_for(ws.recv(), timeout=2.0)
            msg = json.loads(raw)
            assert msg["event"] == "mark"
            assert msg["mark"]["name"] == "test_mark"

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_send_mark_no_stream_raises(self):
        """send_mark without an active stream reports that no mark was sent."""
        port = self._unused_port()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            with pytest.raises(RuntimeError, match="active stream"):
                await t.send_mark("x")
            with pytest.raises(RuntimeError, match="active stream"):
                await t.send_playback_mark("playback_x")
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as _ws:
            await asyncio.sleep(0.1)

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_send_playback_mark(self):
        port = self._unused_port()
        send_ready = asyncio.Event()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            await asyncio.sleep(0.1)
            name = await t.send_playback_mark("pb_mark")
            assert name == "pb_mark"
            send_ready.set()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg("STREAM1"))
            await send_ready.wait()
            raw = await asyncio.wait_for(ws.recv(), timeout=2.0)
            msg = json.loads(raw)
            assert msg["event"] == "mark"
            assert msg["mark"]["name"] == "pb_mark"

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_dtmf_emitted(self):
        port = self._unused_port()
        digits: list[str] = []

        async def handler(ws):
            event_bus = EventBus()
            event_bus.subscribe(DTMF, lambda e: digits.append(e.digit))
            t = TwilioConnectionTransport(ws, event_bus=event_bus)
            await t.connect()
            await ws.wait_closed()
            await asyncio.sleep(0.05)
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg())
            await ws.send(_twilio_dtmf_msg("5"))
            await ws.send(_twilio_dtmf_msg("#"))
            await asyncio.sleep(0.1)

        await asyncio.sleep(0.1)
        server.close()
        await server.wait_closed()
        assert digits == ["5", "#"]

    @pytest.mark.asyncio
    async def test_mark_ack_emitted(self):
        port = self._unused_port()
        marks: list[str] = []

        async def handler(ws):
            event_bus = EventBus()
            event_bus.subscribe(PlaybackMarkAck, lambda e: marks.append(e.mark_name))
            t = TwilioConnectionTransport(ws, event_bus=event_bus)
            await t.connect()
            await ws.wait_closed()
            await asyncio.sleep(0.05)
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg())
            await ws.send(_twilio_mark_msg("m1"))
            await ws.send(_twilio_mark_msg("m2"))
            await asyncio.sleep(0.1)

        await asyncio.sleep(0.1)
        server.close()
        await server.wait_closed()
        assert marks == ["m1", "m2"]

    @pytest.mark.asyncio
    async def test_control_events_ignore_wrong_stream_sid(self):
        event_bus = EventBus()
        digits: list[str] = []
        marks: list[str] = []
        event_bus.subscribe(DTMF, lambda e: digits.append(e.digit))
        event_bus.subscribe(PlaybackMarkAck, lambda e: marks.append(e.mark_name))
        transport = TwilioConnectionTransport(_DummyTwilioWebSocket(), event_bus=event_bus)

        await transport._handle_message(_twilio_start_msg("STREAM1", "CALL1"))
        await transport._handle_message(_twilio_dtmf_msg("5", stream_sid="WRONG"))
        await transport._handle_message(_twilio_mark_msg("mark_1", stream_sid="WRONG"))
        await transport._handle_message(_twilio_stop_msg(stream_sid="WRONG"))

        assert digits == []
        assert marks == []
        assert transport.stream_sid == "STREAM1"
        assert transport.call_sid == "CALL1"

        await transport._handle_message(_twilio_dtmf_msg("6", stream_sid="STREAM1"))
        await transport._handle_message(_twilio_mark_msg("mark_2", stream_sid="STREAM1"))
        await transport._handle_message(_twilio_stop_msg(stream_sid="STREAM1"))

        assert digits == ["6"]
        assert marks == ["mark_2"]
        assert transport.stream_sid is None
        assert transport.call_sid is None

    @pytest.mark.asyncio
    async def test_malformed_frames_do_not_end_receive_loop(self):
        port = self._unused_port()
        transport_holder: list[TwilioConnectionTransport] = []

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            transport_holder.append(t)
            await t.connect()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(b"\xff")
            await ws.send("[]")
            await ws.send(json.dumps({"event": "start", "start": []}))
            await asyncio.sleep(0.1)
            transport = transport_holder[0]
            assert transport._receive_task is not None
            assert not transport._receive_task.done()
            assert transport.is_connected
            assert transport.stream_sid is None

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_stop_message_ends_receive(self):
        port = self._unused_port()
        chunks: list[AudioChunk] = []
        collect_done = asyncio.Event()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            async for chunk in t.receive_audio():
                chunks.append(chunk)
            collect_done.set()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg())
            await ws.send(_twilio_stop_msg())
            await asyncio.sleep(0.1)

        await asyncio.wait_for(collect_done.wait(), timeout=2.0)
        assert len(chunks) == 0

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_disconnect_idempotent(self):
        port = self._unused_port()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            await t.disconnect()
            await t.disconnect()  # Should not raise.

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as _ws:
            await asyncio.sleep(0.1)

        server.close()
        await server.wait_closed()

    @pytest.mark.asyncio
    async def test_clear_audio(self):
        port = self._unused_port()
        send_ready = asyncio.Event()

        async def handler(ws):
            t = TwilioConnectionTransport(ws)
            await t.connect()
            await asyncio.sleep(0.1)
            await t.clear_audio()  # Should send clear event.
            send_ready.set()
            await ws.wait_closed()
            await t.disconnect()

        server = await websockets.serve(handler, "127.0.0.1", port)

        async with websockets.connect(f"ws://127.0.0.1:{port}") as ws:
            await ws.send(_twilio_start_msg("STREAM1"))
            await send_ready.wait()
            raw = await asyncio.wait_for(ws.recv(), timeout=2.0)
            msg = json.loads(raw)
            assert msg["event"] == "clear"
            assert msg["streamSid"] == "STREAM1"

        server.close()
        await server.wait_closed()
