from __future__ import annotations

import io
import json
import socket
import struct
import tempfile
import unittest
import wave
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import patch

from agent_call.real_cell import (
    CellCallConfig,
    CellCallError,
    CellExecutionLedger,
    CellRoute,
    RealCellCall,
    RealCellWorker,
    RTPMedia,
    alaw_to_pcm16,
    load_cell_routes,
    voice_level,
)


class FakeBroker:
    def __init__(self, messages: list[dict]) -> None:
        self.messages = list(messages)
        self.acked: list[dict] = []
        self.requeued: list[dict] = []
        self.rejected: list[dict] = []
        self.published: list[dict] = []

    def declare_tenant(self, tenant_key: str) -> None:
        self.tenant_key = tenant_key

    def consume(self, _queue: str) -> dict | None:
        return self.messages.pop(0) if self.messages else None

    def ack(self, message: dict) -> None:
        self.acked.append(message)

    def requeue(self, message: dict) -> None:
        self.requeued.append(message)

    def reject(self, message: dict) -> None:
        self.rejected.append(message)

    def publish(
        self, _exchange: str, _route: str, body: dict, **_kwargs: object
    ) -> None:
        self.published.append(body)


def wav_bytes(pcm: bytes) -> bytes:
    output = io.BytesIO()
    with wave.open(output, "wb") as source:
        source.setnchannels(1)
        source.setsampwidth(2)
        source.setframerate(8000)
        source.writeframes(pcm)
    return output.getvalue()


class RealCellTests(unittest.TestCase):
    def _build_lifecycle_call(
        self, order: list[str], answer_error: CellCallError | None = None
    ) -> tuple[RealCellCall, Any]:
        config = CellCallConfig(
            "http://127.0.0.1:8088", "u", "p", event_timeout_s=0.1
        )
        engine = SimpleNamespace(
            config={
                "agent_version_id": "agent_v1",
                "conversation": {
                    "opening": "hello",
                    "max_turns": 1,
                    "allow_interrupt": False,
                },
            },
            history=[],
        )
        engine.speak = lambda _text, _turn_id: (
            order.append("opening") or {"audio": b""}
        )
        call = RealCellCall(config, cast(Any, engine), cast(Any, SimpleNamespace()))

        class FakeARI:
            def __init__(self) -> None:
                self.record_requests = 0

            def events(self, _app: str) -> Any:
                return SimpleNamespace(close=lambda: None)

            def request(
                self, method: str, resource: str, _params=None, **_kwargs: object
            ) -> dict[str, str] | dict:
                if method == "POST" and resource.endswith("/record"):
                    self.record_requests += 1
                    order.append("record")
                if resource == "channels/externalMedia":
                    return {"id": "external"}
                if resource == "channels":
                    return {"id": "target"}
                return {}

        fake_ari = FakeARI()
        call_any = cast(Any, call)
        call_any.ari = fake_ari
        call_any._start_event_reader = lambda: None
        call_any._try_set_external_media_peer = lambda: False
        call_any._add_when_ready = lambda _channel_id: True

        def wait_answer(_result: Any) -> None:
            order.append("answer")
            if answer_error is not None:
                raise answer_error

        call_any._wait_answer = wait_answer
        call_any._play = lambda _audio: order.append("play")
        call_any._capture_turn = lambda _timeout: (
            order.append("capture") or b""
        )
        call_any._finish_recording = lambda: None
        call_any._cleanup = lambda: None
        return call, fake_ari

    def test_recording_starts_after_answer_before_opening_tts(self) -> None:
        order: list[str] = []
        call, fake_ari = self._build_lifecycle_call(order)
        try:
            result = call.start_authorized_call("123")
        finally:
            if call.media is not None:
                call.media.close()
        self.assertTrue(result.connected)
        self.assertEqual(fake_ari.record_requests, 1)
        self.assertLess(order.index("answer"), order.index("record"))
        self.assertLess(order.index("record"), order.index("opening"))

    def test_answer_failure_does_not_start_recording(self) -> None:
        order: list[str] = []
        call, fake_ari = self._build_lifecycle_call(
            order, CellCallError("CALL_NOT_ANSWERED", "not answered")
        )
        try:
            result = call.start_authorized_call("123")
        finally:
            if call.media is not None:
                call.media.close()
        self.assertFalse(result.connected)
        self.assertEqual(fake_ari.record_requests, 0)
        self.assertNotIn("opening", order)

    def test_rtp_send_keeps_packet_clock_across_frames(self) -> None:
        media = RTPMedia("127.0.0.1", 0)
        receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        receiver.bind(("127.0.0.1", 0))
        media.peer = ("127.0.0.1", receiver.getsockname()[1])
        clock = [0.0]
        sleeps: list[float] = []

        def monotonic() -> float:
            return clock[0]

        def sleep(seconds: float) -> None:
            sleeps.append(seconds)
            clock[0] += seconds

        try:
            with (
                patch("agent_call.real_cell.time.monotonic", monotonic),
                patch("agent_call.real_cell.time.sleep", sleep),
            ):
                media.send_pcm16(bytes([0, 0]) * 160, 8000)
                media.send_pcm16(bytes([0, 0]) * 160, 8000)
            self.assertEqual(sleeps, [0.02])
            self.assertEqual(len(receiver.recvfrom(512)[0]), 172)
            self.assertEqual(len(receiver.recvfrom(512)[0]), 172)
        finally:
            media.close()
            receiver.close()

    def test_pcma_decode_and_rtp_payload(self) -> None:
        media = RTPMedia("127.0.0.1", 0)
        receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        receiver.bind(("127.0.0.1", 0))
        try:
            media.peer = ("127.0.0.1", receiver.getsockname()[1])
            sent = media.send_pcm16(b"\x00\x00" * 160, 8000)
            packet, _ = receiver.recvfrom(512)
            self.assertEqual(sent, 160)
            self.assertEqual(len(packet), 172)
            self.assertEqual(packet[1] & 0x7F, 8)
            self.assertEqual(packet[12:], b"\xd5" * 160)
            self.assertEqual(len(alaw_to_pcm16(packet[12:])), 320)
            self.assertEqual(voice_level(alaw_to_pcm16(packet[12:])), 8)
        finally:
            media.close()
            receiver.close()

    def test_rtp_payload_handles_extension_and_padding(self) -> None:
        payload = b"abc"
        packet = struct.pack("!BBHII", 0xB0, 8, 1, 2, 3)
        packet += struct.pack("!HH", 0, 1) + b"xxxx" + payload + b"\x00\x00\x03"
        self.assertEqual(RTPMedia._payload(packet), payload)
        self.assertIsNone(RTPMedia._payload(b"bad"))

    def test_nonblocking_receive_returns_no_packet(self) -> None:
        media = RTPMedia("127.0.0.1", 0)
        try:
            self.assertIsNone(media.receive(0.0))
        finally:
            media.close()

    def test_rtp_receive_filters_payload_peer_and_ssrc(self) -> None:
        media = RTPMedia("127.0.0.1", 0)
        sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sender.bind(("127.0.0.1", 0))
        media.peer = ("127.0.0.1", sender.getsockname()[1])

        def packet(payload_type: int, ssrc: int, payload: bytes) -> bytes:
            return struct.pack("!BBHII", 0x80, payload_type, 1, 0, ssrc) + payload

        try:
            sender.sendto(packet(0, 1, b"wrong-pt"), media.address)
            self.assertIsNone(media.receive(0.2))
            sender.sendto(packet(8, 1, b"voice"), media.address)
            self.assertEqual(media.receive(0.2), b"voice")
            sender.sendto(packet(8, 2, b"wrong-ssrc"), media.address)
            self.assertIsNone(media.receive(0.2))
        finally:
            sender.close()
            media.close()

    def test_external_media_peer_comes_from_ari(self) -> None:
        config = CellCallConfig("http://127.0.0.1:8088", "u", "p")
        call = cast(
            Any,
            RealCellCall(
                config,
                cast(Any, SimpleNamespace(config={})),
                cast(Any, SimpleNamespace()),
            ),
        )
        media = RTPMedia("127.0.0.1", 0)
        calls: list[tuple[str, str, dict | None]] = []

        class FakeARI:
            def request(self, method: str, resource: str, params=None, **_kwargs):
                calls.append((method, resource, params))
                if params and params.get("variable") == "UNICASTRTP_LOCAL_ADDRESS":
                    return {"value": "127.0.0.1"}
                return {"value": "12345"}

        call.media = media
        call.ari = FakeARI()
        try:
            call._set_external_media_peer("external-1")
            self.assertEqual(media.peer, ("127.0.0.1", 12345))
            self.assertEqual(len(calls), 2)
        finally:
            media.close()

    def test_add_when_ready_marks_only_after_success(self) -> None:
        config = CellCallConfig("http://127.0.0.1:8088", "u", "p")
        call = cast(
            Any,
            RealCellCall(
                config,
                cast(Any, SimpleNamespace(config={})),
                cast(Any, SimpleNamespace()),
            ),
        )

        class FakeARI:
            def __init__(self) -> None:
                self.attempts = 0

            def request(self, method: str, _resource: str, _params=None, **_kwargs):
                if method == "POST":
                    self.attempts += 1
                    if self.attempts == 1:
                        raise CellCallError("ARI_HTTP_422", "not in Stasis", True)
                return {}

        fake_ari = FakeARI()
        call.ari = fake_ari
        self.assertTrue(call._add_when_ready("target"))
        self.assertEqual(fake_ari.attempts, 2)
        self.assertIn("target", call._known_channels)

    def test_wait_answer_requires_bridge_membership(self) -> None:
        config = CellCallConfig("http://127.0.0.1:8088", "u", "p", event_timeout_s=1.0)
        call = cast(
            Any,
            RealCellCall(
                config,
                cast(Any, SimpleNamespace(config={})),
                cast(Any, SimpleNamespace()),
            ),
        )
        call.target_channel_id = "target"
        call.external_channel_id = "external"
        call.media = RTPMedia("127.0.0.1", 0)
        call.media.peer = ("127.0.0.1", 12345)
        bridge_members = [{"external"}, {"external", "target"}]

        class FakeARI:
            def request(self, method: str, resource: str, _params=None, **_kwargs):
                if method == "GET" and resource == f"bridges/{call.bridge_id}":
                    return {"channels": list(bridge_members.pop(0))}
                return {}

        call.ari = FakeARI()
        call.events.put(
            {"type": "ChannelStateChange", "channel": {"id": "target", "state": "Up"}}
        )
        call.events.put({"type": "StasisStart", "channel": {"id": "target"}})
        try:
            call._wait_answer(SimpleNamespace())
            self.assertEqual(bridge_members, [])
        finally:
            call.media.close()

    def test_finish_recording_stops_then_reads_stored_file(self) -> None:
        class FakeARI:
            def __init__(self) -> None:
                self.calls: list[tuple[str, str]] = []

            def request(self, method: str, resource: str, **_kwargs):
                self.calls.append((method, resource))
                return {} if method == "POST" else wav_bytes(b"\x00\x00")

        with tempfile.TemporaryDirectory() as directory:
            config = CellCallConfig(
                "http://127.0.0.1:8088", "u", "p", recording_dir=directory
            )
            call = cast(
                Any,
                RealCellCall(
                    config,
                    cast(Any, SimpleNamespace(config={})),
                    cast(Any, SimpleNamespace()),
                ),
            )
            fake_ari = FakeARI()
            call.ari = fake_ari
            path = call._finish_recording()
            self.assertIsNotNone(path)
            self.assertEqual(fake_ari.calls[0][0], "POST")
            self.assertIn("/stop", fake_ari.calls[0][1])
            self.assertNotIn(
                ("DELETE", fake_ari.calls[0][1].rsplit("/stop", 1)[0]), fake_ari.calls
            )

    def test_finish_recording_rejects_zero_frame_wav(self) -> None:
        class FakeARI:
            def request(self, method: str, resource: str, **_kwargs):
                return {} if method == "POST" else wav_bytes(b"")

        with tempfile.TemporaryDirectory() as directory:
            config = CellCallConfig(
                "http://127.0.0.1:8088", "u", "p", recording_dir=directory
            )
            call = cast(
                Any,
                RealCellCall(
                    config,
                    cast(Any, SimpleNamespace(config={})),
                    cast(Any, SimpleNamespace()),
                ),
            )
            call.ari = FakeARI()
            self.assertIsNone(call._finish_recording())
            self.assertFalse(
                (Path(directory) / f"{call.recording_name}.wav").exists()
            )

    def test_cell_config_rejects_non_pcma_bad_port_and_prefix(self) -> None:
        with self.assertRaises(CellCallError):
            CellCallConfig("http://127.0.0.1:8088", "u", "p", rtp_format="ulaw")
        with self.assertRaises(CellCallError):
            CellCallConfig("http://127.0.0.1:8088", "u", "p", rtp_bind_port=65536)
        with self.assertRaises(CellCallError):
            CellCallConfig("http://127.0.0.1:8088", "u", "p", dial_prefix="7089+")
        self.assertEqual(
            CellCallConfig(
                "http://127.0.0.1:8088", "u", "p", dial_prefix=""
            ).dial_prefix,
            "",
        )
        self.assertEqual(
            CellCallConfig(
                "http://127.0.0.1:8088", "u", "p", dial_prefix="mka755"
            ).dial_prefix,
            "mka755",
        )

    def test_route_map_binds_task_policy_to_trusted_trunk(self) -> None:
        routes = load_cell_routes(
            json.dumps(
                {
                    "route-a": {
                        "caller_profile_id": "caller-a",
                        "trunk_id": "provider-second",
                        "caller_id": "mbkq",
                        "dial_prefix": "",
                    }
                }
            )
        )
        self.assertEqual(routes["route-a"].trunk_id, "provider-second")
        self.assertEqual(routes["route-a"].caller_id, "mbkq")
        with self.assertRaises(CellCallError):
            load_cell_routes(
                json.dumps(
                    {
                        "route-a": {
                            "caller_profile_id": "caller-a",
                            "trunk_id": "provider-second",
                            "caller_id": "mbkq\nspoof",
                            "dial_prefix": "",
                        }
                    }
                )
            )

    def test_worker_selects_installed_route_without_payload_sip_values(self) -> None:
        command = {
            "body": {
                "command_type": "call.execute",
                "tenant_id": "tenant-demo",
                "tenant_key": "tenant-key",
                "payload": {
                    "execution_id": "exec-route-1",
                    "agent_version_id": "agent_v1",
                    "callee": "18625770806",
                    "route_policy_id": "route-a",
                    "caller_profile_id": "caller-a",
                },
            }
        }
        broker = FakeBroker([command])
        with tempfile.TemporaryDirectory() as directory:
            ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3")
            fake_result = SimpleNamespace(
                as_dict=lambda: {
                    "call_id": "call-route",
                    "status": "failed",
                    "turns": [],
                }
            )

            class FakeExecutor:
                def __init__(self) -> None:
                    self.engine = SimpleNamespace(
                        config={"agent_version_id": "agent_v1"}
                    )
                    self.calls: list[tuple[str, CellRoute | None]] = []

                def start_authorized_call(
                    self, callee: str, route: CellRoute | None = None
                ) -> SimpleNamespace:
                    self.calls.append((callee, route))
                    return fake_result

            fake_executor = FakeExecutor()
            worker = RealCellWorker(
                broker,
                "tenant-key",
                ledger,
                cast(RealCellCall, fake_executor),
                routes={
                    "route-a": CellRoute(
                        "route-a", "caller-a", "provider-second", "mbkq", ""
                    )
                },
            )
            event = worker.process_once()
        if event is None:
            self.fail("worker did not publish a call.finished event")
        self.assertEqual(fake_executor.calls[0][0], "18625770806")
        selected_route = fake_executor.calls[0][1]
        assert selected_route is not None
        self.assertEqual(selected_route.trunk_id, "provider-second")

    def test_ledger_marks_in_progress_as_in_doubt(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3")
            first = ledger.claim("exec-1", "15003164745")
            second = ledger.claim("exec-1", "15003164745")
            self.assertTrue(first["claimed"])
            self.assertTrue(second["in_doubt"])

    def test_worker_uses_queue_and_never_redials_terminal_execution(self) -> None:
        command = {
            "body": {
                "command_type": "call.execute",
                "tenant_id": "tenant-demo",
                "tenant_key": "tenant-key",
                "payload": {
                    "execution_id": "exec-1",
                    "agent_version_id": "agent_v1",
                    "callee": "15003164745",
                },
            }
        }
        broker = FakeBroker([command, command])
        with tempfile.TemporaryDirectory() as directory:
            ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3")
            fake_result = SimpleNamespace(
                as_dict=lambda: {
                    "call_id": "call-1",
                    "status": "failed",
                    "reason_code": "CUSTOMER_SILENT",
                    "recording_path": "/private/recording.wav",
                }
            )

            class FakeExecutor:
                def __init__(self) -> None:
                    self.engine = SimpleNamespace(
                        config={"agent_version_id": "agent_v1"}
                    )
                    self.calls: list[str] = []

                def start_authorized_call(self, callee: str) -> SimpleNamespace:
                    self.calls.append(callee)
                    return fake_result

            fake_executor = FakeExecutor()
            executor = cast(RealCellCall, fake_executor)
            worker = RealCellWorker(broker, "tenant-key", ledger, executor)
            first = worker.process_once()
            second = worker.process_once()
            if first is None or second is None:
                self.fail("worker did not publish a call.finished event")
            self.assertEqual(first["event_type"], "call.finished")
            self.assertEqual(second["event_type"], "call.finished")
            self.assertEqual(fake_executor.calls, ["15003164745"])
            self.assertEqual(len(broker.acked), 2)
            self.assertNotIn("recording_path", broker.published[0]["payload"])

    def test_worker_rejects_wrong_tenant_route(self) -> None:
        broker = FakeBroker(
            [
                {
                    "body": {
                        "command_type": "call.execute",
                        "tenant_id": "tenant-demo",
                        "tenant_key": "other-tenant",
                        "payload": {
                            "execution_id": "exec-1",
                            "agent_version_id": "agent_v1",
                            "callee": "15003164745",
                        },
                    }
                }
            ]
        )
        with tempfile.TemporaryDirectory() as directory:
            ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3")
            executor = cast(
                RealCellCall,
                SimpleNamespace(
                    engine=SimpleNamespace(config={"agent_version_id": "agent_v1"})
                ),
            )
            worker = RealCellWorker(broker, "tenant-key", ledger, executor)
            result = worker.process_once()
        if result is None:
            self.fail("worker returned no command result")
        self.assertEqual(result["reason_code"], "COMMAND_INVALID")
        self.assertEqual(len(broker.rejected), 1)


if __name__ == "__main__":
    unittest.main()