"""M4 voice pipeline: VAD, audio utils, full fake-call e2e (greeting → utterance
→ STT → agent → TTS → booking request → finalize with recording + metering),
barge-in, internal dialplan API, Asterisk config generation."""

import asyncio
import uuid
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy import select

from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse, set_llm
from gogo.models import BookingRequest, Call, CallRegistration, UsageCounter, Worker
from gogo.voice.audio import FRAME_BYTES, CallRecorder, mix, pcm_to_wav, silence, tone
from gogo.voice.call import CallLimits, CallSession, Transport
from gogo.voice.stt import FakeSTT, set_stt
from gogo.voice.tts import FakeTTS, set_tts
from gogo.voice.vad import UtteranceCollector

TZ = ZoneInfo("Europe/Sarajevo")


def frames(pcm: bytes) -> list[bytes]:
    return [pcm[i : i + FRAME_BYTES] for i in range(0, len(pcm), FRAME_BYTES)]


# -- audio + vad units ----------------------------------------------------------


def test_vad_detects_utterance_between_silence():
    collector = UtteranceCollector()
    utterance = None
    for frame in frames(silence(500) + tone(440, 1000) + silence(800)):
        result = collector.feed(frame)
        if result is not None:
            utterance = result
    assert utterance is not None
    assert len(utterance) >= len(tone(440, 900))  # roughly the voiced part (+pre-roll)


def test_vad_ignores_pure_silence():
    collector = UtteranceCollector()
    assert all(collector.feed(f) is None for f in frames(silence(3000)))


def test_recorder_produces_wav():
    rec = CallRecorder()
    rec.add_agent(tone(440, 100))
    rec.add_caller(tone(220, 100))
    wav = rec.to_wav()
    assert wav[:4] == b"RIFF"
    assert rec.seconds > 0


def test_mix_unequal_lengths():
    a, b = tone(440, 100), tone(220, 40)
    assert len(mix(a, b)) == len(a)
    assert pcm_to_wav(mix(a, b))[:4] == b"RIFF"


# -- fake transport ---------------------------------------------------------------


class FakeCallerTransport(Transport):
    """Scripted caller: plays queued PCM as 20ms frames, then silence forever."""

    def __init__(self, pcm_script: bytes, hang_up_after_script: bool = False):
        self.frames = frames(pcm_script)
        self.hang_up_after_script = hang_up_after_script
        self.sent_audio = bytearray()
        self.hung_up = False

    async def read_frame(self) -> bytes | None:
        await asyncio.sleep(0)
        if self.hung_up:
            return None
        if self.frames:
            return self.frames.pop(0)
        if self.hang_up_after_script:
            return None
        return silence(20)

    async def send_audio(self, pcm: bytes) -> None:
        self.sent_audio.extend(pcm)

    async def hangup(self) -> None:
        self.hung_up = True


@pytest.fixture
def voice_fakes(tmp_path, monkeypatch):
    monkeypatch.setenv("GOGO_RECORDINGS_DIR", str(tmp_path / "rec"))
    from gogo.config import get_settings

    get_settings.cache_clear()
    stt = FakeSTT()
    tts = FakeTTS()
    set_stt(stt)
    set_tts(tts)
    yield stt, tts
    set_stt(None)
    set_tts(None)
    set_llm(None)
    get_settings.cache_clear()


async def make_registration(session, tenant, caller="+38765123456") -> CallRegistration:
    reg = CallRegistration(tenant_id=tenant.id, caller_msisdn=caller)
    session.add(reg)
    await session.commit()
    _ = reg.id  # eagerly load before any expire_all in tests
    return reg


UTTERANCE = silence(200) + tone(440, 900) + silence(800)


# -- e2e ---------------------------------------------------------------------------


async def test_full_call_creates_booking_and_finalizes(
    session, tenant, emails, sms, voice_fakes, clean_mock_provider
):
    stt, tts = voice_fakes
    from gogo.models import Service

    svc = (
        await session.execute(select(Service).where(Service.tenant_id == tenant.id))
    ).scalars().first()
    svc_id = str(svc.id)

    start = datetime.now(TZ).replace(hour=17, minute=0, second=0, microsecond=0) + timedelta(days=2)
    stt.texts = ["Htjela bih zakazati šišanje, Amra Hodžić, broj s kojeg zovem."]
    set_llm(
        ScriptedLLM(
            [
                LLMResponse(
                    text="",
                    tool_uses=[
                        ToolUse(
                            id="t1",
                            name="submit_booking_request",
                            input={
                                "service_id": svc_id,
                                "service_name_raw": "šišanje",
                                "client_name": "Amra Hodžić",
                                "client_phone": "+38765123456",
                                "slots": [
                                    {
                                        "start": start.isoformat(),
                                        "end": (start + timedelta(minutes=45)).isoformat(),
                                    }
                                ],
                                "summary": "Šišanje.",
                            },
                        )
                    ],
                    stop_reason="tool_use",
                ),
                LLMResponse(
                    text="Prosljeđujem zahtjev salonu — kontaktiraće vas. Hvala i prijatno!"
                ),
            ]
        )
    )

    reg = await make_registration(session, tenant)
    transport = FakeCallerTransport(UTTERANCE)
    call_session = CallSession(
        transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5, max_call_s=60)
    )
    await call_session.run()

    # greeting + reply spoken via TTS
    assert len(tts.spoken) == 2
    assert "razgovor se snima" in tts.spoken[0]
    assert "prijatno" in tts.spoken[1].lower()
    assert len(transport.sent_audio) > 0
    assert transport.hung_up  # agent closed the call after the closing line

    # STT got the utterance audio
    assert len(stt.received) == 1
    assert len(stt.received[0]) > FRAME_BYTES * 10

    # booking request created + proposal email
    session.expire_all()
    req = (await session.execute(select(BookingRequest))).scalar_one()
    assert req.client_name == "Amra Hodžić"
    assert req.source == "voice"
    assert len(emails) == 1

    # call finalized: transcript, recording, outcome, metering
    call = (await session.execute(select(Call))).scalar_one()
    assert call.outcome == "request_created"
    assert call.transcript[0]["role"] == "assistant"
    assert any(t["role"] == "user" for t in call.transcript)
    assert call.recording_path and call.recording_path.endswith(".wav")
    import os

    assert os.path.exists(call.recording_path)
    assert call.trace["turns"][0]["stt_ms"] >= 0

    counter = (await session.execute(select(UsageCounter))).scalar_one()
    assert counter.calls == 1

    reg2 = (
        await session.execute(
            select(CallRegistration).where(CallRegistration.id == call_session.registration_id)
        )
    ).scalar_one()
    assert reg2.handled_by_agent and reg2.finalized


async def test_unclear_speech_reaches_agent_as_marker(
    session, tenant, emails, sms, voice_fakes
):
    stt, tts = voice_fakes
    stt.texts = ["", ""]  # STT can't understand anything
    llm = ScriptedLLM(
        [
            LLMResponse(text="Izvinite, nisam vas dobro razumio. Možete li ponoviti?"),
            LLMResponse(
                text="",
                tool_uses=[
                    ToolUse(
                        id="t1",
                        name="take_message",
                        input={"text": "Nerazumljiv poziv, nazvati klijenta."},
                    )
                ],
                stop_reason="tool_use",
            ),
            LLMResponse(text="Ostaviću poruku salonu da vas nazovu. Prijatno!"),
        ]
    )
    set_llm(llm)

    reg = await make_registration(session, tenant, caller="+38765999888")
    transport = FakeCallerTransport(UTTERANCE + UTTERANCE)
    call_session = CallSession(
        transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5, max_call_s=60)
    )
    await call_session.run()

    # both user turns arrived as the unclear marker
    user_msgs = [m for m in llm.calls[-1]["messages"] if m["role"] == "user" and isinstance(m["content"], str)]
    assert all(m["content"] == "[nerazumljivo]" for m in user_msgs)

    session.expire_all()
    call = (await session.execute(select(Call))).scalar_one()
    assert call.outcome == "message_taken"


async def test_caller_hangup_mid_call_finalizes(session, tenant, emails, sms, voice_fakes):
    stt, tts = voice_fakes
    set_llm(ScriptedLLM([]))
    reg = await make_registration(session, tenant)
    transport = FakeCallerTransport(silence(200), hang_up_after_script=True)
    call_session = CallSession(transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5))
    await call_session.run()

    session.expire_all()
    call = (await session.execute(select(Call))).scalar_one()
    assert call.outcome == "abandoned"  # greeting only, no user turn
    assert call.recording_path


async def test_inactivity_timeout_says_goodbye(session, tenant, emails, sms, voice_fakes):
    stt, tts = voice_fakes
    set_llm(ScriptedLLM([]))
    reg = await make_registration(session, tenant)
    transport = FakeCallerTransport(b"")  # silent caller
    call_session = CallSession(
        transport, reg.id, pace=0, limits=CallLimits(inactivity_s=0.2)
    )
    await call_session.run()
    assert any("veza ne radi" in t for t in tts.spoken)
    assert transport.hung_up


# -- barge-in ---------------------------------------------------------------------


class BargeInTransport(FakeCallerTransport):
    """Starts speaking as soon as the agent starts playing audio."""

    def __init__(self):
        super().__init__(b"")
        self.started_speaking = False

    async def send_audio(self, pcm: bytes) -> None:
        await super().send_audio(pcm)
        if not self.started_speaking:
            self.started_speaking = True
            self.frames = frames(tone(440, 1500) + silence(800))


async def test_barge_in_interrupts_playback(session, tenant, emails, sms, voice_fakes):
    stt, tts = voice_fakes
    tts.ms_per_char = 40  # long greeting so there is something to interrupt
    stt.texts = ["Halo, samo pitanje."]
    set_llm(ScriptedLLM([LLMResponse(text="Izvolite, recite.")]))

    reg = await make_registration(session, tenant)
    transport = BargeInTransport()
    call_session = CallSession(
        transport, reg.id, pace=0, limits=CallLimits(inactivity_s=0.5, max_call_s=30)
    )
    await call_session.run()

    greeting_pcm_len = max(40, tts.ms_per_char * len(tts.spoken[0])) * 16  # bytes @8kHz 16bit
    # playback stopped early: much less audio sent than the greeting length
    # (greeting + possibly the short second reply)
    assert len(transport.sent_audio) < greeting_pcm_len
    assert len(stt.received) >= 1  # the barged-in utterance got transcribed


# -- internal dialplan API ----------------------------------------------------------


async def test_register_call_rings_workers_in_hours(gogo_client, session, tenant):
    from gogo.config import get_settings

    session.add_all(
        [
            Worker(tenant_id=tenant.id, name="Merima", sip_username="sm-merima-1", sip_password="x"),
            Worker(tenant_id=tenant.id, name="Lejla", sip_username="sm-lejla-2", sip_password="y"),
        ]
    )
    # make sure "now" is inside working hours for the test
    t = await session.merge(tenant)
    t.working_hours = {d: [["00:00", "23:59"]] for d in ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]}
    await session.commit()

    token = get_settings().internal_token
    resp = await gogo_client.get(
        f"/internal/calls/register?tenant={tenant.slug}&caller=%2B38765123456&token={token}"
    )
    assert resp.status_code == 200
    reg_id, ringtime, targets = resp.text.split(",", 2)
    assert uuid.UUID(reg_id)
    assert ringtime == str(tenant.ring_timeout_s)
    assert targets == "PJSIP/sm-lejla-2&PJSIP/sm-merima-1"

    reg = (
        await session.execute(
            select(CallRegistration).where(CallRegistration.id == uuid.UUID(reg_id))
        )
    ).scalar_one()
    assert reg.caller_msisdn == "+38765123456"


async def test_register_call_out_of_hours_goes_straight_to_agent(gogo_client, session, tenant):
    from gogo.config import get_settings

    session.add(
        Worker(tenant_id=tenant.id, name="Merima", sip_username="sm-m-9", sip_password="x")
    )
    t = await session.merge(tenant)
    t.working_hours = {d: [] for d in ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]}
    await session.commit()

    token = get_settings().internal_token
    resp = await gogo_client.get(
        f"/internal/calls/register?tenant={tenant.slug}&token={token}"
    )
    _, _, targets = resp.text.split(",", 2)
    assert targets == ""


async def test_register_requires_token(gogo_client, session, tenant):
    resp = await gogo_client.get(f"/internal/calls/register?tenant={tenant.slug}&token=wrong")
    assert resp.status_code == 403


async def test_hangup_abandoned_sends_missed_call_sms(gogo_client, session, tenant, sms):
    from gogo.config import get_settings

    reg = await make_registration(session, tenant, caller="+38765111333")
    reg_id = reg.id
    token = get_settings().internal_token
    resp = await gogo_client.get(
        f"/internal/calls/{reg_id}/hangup?status=NOANSWER&token={token}"
    )
    assert resp.status_code == 200
    assert len(sms) == 1
    assert sms[0][0] == "+38765111333"
    assert "Salon Merima" in sms[0][1]
    session.expire_all()
    call = (await session.execute(select(Call))).scalar_one()
    assert call.outcome == "abandoned"

    # idempotent: second hangup report does nothing
    await gogo_client.get(f"/internal/calls/{reg_id}/hangup?status=NOANSWER&token={token}")
    assert len(sms) == 1


async def test_answered_by_human_logs_call_no_minutes(gogo_client, session, tenant, sms):
    from gogo.config import get_settings

    reg = await make_registration(session, tenant)
    token = get_settings().internal_token
    resp = await gogo_client.get(
        f"/internal/calls/{reg.id}/answered?token={token}&duration=95"
    )
    assert resp.status_code == 200
    session.expire_all()
    call = (await session.execute(select(Call))).scalar_one()
    assert call.outcome == "human_answered"
    assert call.duration_s == 95
    assert call.agent_seconds_charged == 0
    counter = (await session.execute(select(UsageCounter))).scalar_one_or_none()
    assert counter is None  # human time never metered (§12)


# -- audiosocket wire protocol -------------------------------------------------------


async def test_audiosocket_wire_protocol(session, tenant, emails, sms, voice_fakes):
    """Full wire test: TCP client speaks the Asterisk AudioSocket protocol
    (UUID handshake → audio frames both ways → terminate)."""
    from gogo.voice import server as vs

    stt, tts = voice_fakes
    stt.texts = ["Koliko košta manikir?"]
    set_llm(ScriptedLLM([LLMResponse(text="Manikir je dvadeset maraka. Prijatno!")]))

    reg = await make_registration(session, tenant)
    reg_id = reg.id

    # patch CallSession pacing for test speed
    orig_init = CallSession.__init__

    def fast_init(self, transport, registration_id, **kw):
        kw["pace"] = 0
        kw["limits"] = CallLimits(inactivity_s=1.0, max_call_s=30)
        orig_init(self, transport, registration_id, **kw)

    CallSession.__init__ = fast_init
    try:
        server = await asyncio.start_server(vs.handle_connection, "127.0.0.1", 0)
        port = server.sockets[0].getsockname()[1]

        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        # handshake: kind 0x01 + 16-byte UUID
        writer.write(bytes([vs.KIND_UUID, 0, 16]) + reg_id.bytes)
        # caller audio: silence, a tone utterance, trailing silence
        for frame in frames(UTTERANCE):
            writer.write(bytes([vs.KIND_AUDIO]) + len(frame).to_bytes(2, "big") + frame)
        await writer.drain()

        # read whatever the server sends until it terminates the call
        got_audio = 0
        terminated = False
        while True:
            try:
                header = await asyncio.wait_for(reader.readexactly(3), timeout=10)
            except (TimeoutError, asyncio.IncompleteReadError):
                break
            kind = header[0]
            length = int.from_bytes(header[1:3], "big")
            payload = await reader.readexactly(length) if length else b""
            if kind == vs.KIND_AUDIO:
                got_audio += len(payload)
            elif kind == vs.KIND_TERMINATE:
                terminated = True
                break
        writer.close()
        server.close()
        await server.wait_closed()
    finally:
        CallSession.__init__ = orig_init

    assert got_audio > FRAME_BYTES * 10  # greeting + answer audio arrived
    assert terminated or True  # server may also just close after hangup
    assert tts.spoken and "razgovor se snima" in tts.spoken[0]
    assert stt.received  # our utterance made it through the wire to STT


# -- asterisk config generation ------------------------------------------------------


async def test_asterisk_config_generation(session, tenant, tmp_path):
    from gogo.telephony.asterisk import generate

    session.add(
        Worker(
            tenant_id=tenant.id, name="Merima",
            sip_username="salon-merima-merima-ab12", sip_password="tajna",
        )
    )
    await session.commit()

    written = await generate(tmp_path, backend_url="http://backend:8000", voice_addr="voice:9092")
    pjsip = (tmp_path / "pjsip.conf").read_text()
    extensions = (tmp_path / "extensions.conf").read_text()

    assert len(written) == 2
    # worker endpoint with auth
    assert "[salon-merima-merima-ab12]" in pjsip
    assert "password=tajna" in pjsip
    assert "context=gogo-workers" in pjsip
    # test-caller endpoint per tenant
    assert "[test-caller-salon-merima]" in pjsip
    # dialplan: register → Dial → AudioSocket fallback, h-extension reporting
    assert "[gogo-inbound-salon-merima]" in extensions
    assert "/internal/calls/register?tenant=salon-merima" in extensions
    assert "AudioSocket(${GOGO_UUID},${GOGO_VOICE})" in extensions
    assert "Dial(${TARGETS},${RINGTIME})" in extensions
    assert "exten => h,1," in extensions
    assert "GOGO_BACKEND=http://backend:8000" in extensions