"""ElevenLabs Conversational AI (ConvAI) end-to-end voice provider.

Bridges a carrier media stream to a single ElevenLabs ConvAI WebSocket that
handles STT, LLM, and TTS in one hop. Used in :mod:`stream_handler` as the
``elevenlabs_convai`` provider mode.
"""

import asyncio
import base64
import json
import logging
from typing import Any

import httpx
import websockets

logger = logging.getLogger("getpatter")

ELEVENLABS_CONVAI_URL = "wss://api.elevenlabs.io/v1/convai/conversation"
ELEVENLABS_SIGNED_URL = (
    "https://api.elevenlabs.io/v1/convai/conversation/get-signed-url"
)

# Silence threshold: if no audio chunk arrives for this many ms after
# `agent_response`, treat the agent turn as finished and emit `response_done`.
_AGENT_SILENCE_MS = 500


class ElevenLabsConvAIAdapter:
    """Bridges Twilio/Telnyx media stream to ElevenLabs Conversational AI.

    Handles full conversation: STT + LLM + TTS in one WebSocket.
    Uses ElevenLabs premium voices.

    Telephony optimization
    ----------------------
    For real phone calls prefer the carrier-specific factories:

    * :meth:`for_twilio` — negotiates ``ulaw_8000`` natively for both
      directions. Twilio media streams are PCMU @ 8 kHz, so emitting and
      accepting μ-law directly skips two resamples (16 kHz → 8 kHz outbound
      and 8 kHz → 16 kHz inbound) plus the PCM ↔ μ-law transcode. Saves
      ~30–80 ms first-byte and meaningful per-frame CPU on every turn.
    * :meth:`for_telnyx` — negotiates ``ulaw_8000`` too. Telnyx
      bidirectional media is PCMU @ 8 kHz when ``streaming_start`` requests
      ``stream_bidirectional_codec=PCMU`` (our default), so the same μ-law
      passthrough applies.

    The bare constructor still defaults to PCM16 16 kHz (server defaults),
    which is the right choice for non-telephony embeddings.
    """

    def __init__(
        self,
        api_key: str,
        agent_id: str,
        voice_id: str = "EXAVITQu4vr4xnSDxMaL",
        model_id: str = "eleven_flash_v2_5",
        language: str = "it",
        first_message: str = "",
        output_audio_format: str | None = None,
        input_audio_format: str | None = None,
        use_signed_url: bool = False,
    ):
        if not agent_id:
            raise ValueError(
                "ElevenLabsConvAIAdapter requires a non-empty agent_id. "
                "Create an agent in the ElevenLabs Conversational AI dashboard "
                "and pass its id."
            )
        self.api_key = api_key
        self.agent_id = agent_id
        self.voice_id = voice_id
        self.model_id = model_id
        self.language = language
        self.first_message = first_message
        self.output_audio_format = output_audio_format
        self.input_audio_format = input_audio_format
        self.use_signed_url = use_signed_url
        # Populated from `conversation_initiation_metadata`.
        self.conversation_id: str | None = None
        self.agent_output_audio_format: str | None = None
        self.user_input_audio_format: str | None = None
        self._ws = None
        self._running = False
        # Async queue bridging the message-reader task to `receive_events`.
        self._events: asyncio.Queue[tuple[str, Any] | None] | None = None
        self._reader_task: asyncio.Task | None = None
        # Silence-tracking for synthetic `response_done` emission.
        self._silence_task: asyncio.Task | None = None
        self._agent_speaking = False
        # Session start time for ``patter.cost.realtime_minutes`` emission on close.
        import time as _time

        self._session_start_monotonic: float = _time.monotonic()

    def record_session_end(self) -> None:
        """Emit ``patter.cost.realtime_minutes`` for the elapsed session duration."""
        try:
            import time as _time

            from getpatter.observability.attributes import record_patter_attrs

            elapsed = _time.monotonic() - self._session_start_monotonic
            record_patter_attrs(
                {
                    "patter.cost.realtime_minutes": elapsed / 60.0,
                    "patter.realtime.provider": "elevenlabs_convai",
                }
            )
        except Exception:  # pragma: no cover — defense in depth
            logger.debug("record_session_end failed", exc_info=True)

    def __repr__(self) -> str:
        return f"ElevenLabsConvAIAdapter(agent_id={self.agent_id!r}, model_id={self.model_id!r})"

    # ------------------------------------------------------------------
    # Telephony factories
    # ------------------------------------------------------------------

    @classmethod
    def for_twilio(
        cls,
        api_key: str,
        agent_id: str,
        *,
        voice_id: str = "EXAVITQu4vr4xnSDxMaL",
        model_id: str = "eleven_flash_v2_5",
        language: str = "it",
        first_message: str = "",
        use_signed_url: bool = False,
    ) -> "ElevenLabsConvAIAdapter":
        """Build an adapter pre-configured for Twilio Media Streams.

        Negotiates ``ulaw_8000`` for both ``output_audio_format`` and
        ``input_audio_format`` so ElevenLabs ConvAI emits and accepts μ-law
        @ 8 kHz directly — the exact wire format Twilio uses on its media
        WebSocket. The SDK's stream handler detects this and skips both the
        8 kHz → 16 kHz inbound resample and the 16 kHz → 8 kHz / PCM → μ-law
        outbound transcode that ``TwilioAudioSender`` would otherwise
        perform. Saves ~30–80 ms first-byte plus per-frame CPU on every
        turn.
        """
        return cls(
            api_key=api_key,
            agent_id=agent_id,
            voice_id=voice_id,
            model_id=model_id,
            language=language,
            first_message=first_message,
            output_audio_format="ulaw_8000",
            input_audio_format="ulaw_8000",
            use_signed_url=use_signed_url,
        )

    @classmethod
    def for_telnyx(
        cls,
        api_key: str,
        agent_id: str,
        *,
        voice_id: str = "EXAVITQu4vr4xnSDxMaL",
        model_id: str = "eleven_flash_v2_5",
        language: str = "it",
        first_message: str = "",
        use_signed_url: bool = False,
    ) -> "ElevenLabsConvAIAdapter":
        """Build an adapter pre-configured for Telnyx bidirectional media.

        Telnyx negotiates PCMU @ 8 kHz when ``streaming_start`` sets
        ``stream_bidirectional_codec=PCMU`` (the SDK default). Picking
        ``ulaw_8000`` on both ConvAI directions removes every transcode on
        the audio path — same trade-off as ``for_twilio``.

        If your Telnyx profile is pinned to L16/16000 instead, use the bare
        constructor with the default PCM16 formats.
        """
        return cls(
            api_key=api_key,
            agent_id=agent_id,
            voice_id=voice_id,
            model_id=model_id,
            language=language,
            first_message=first_message,
            output_audio_format="ulaw_8000",
            input_audio_format="ulaw_8000",
            use_signed_url=use_signed_url,
        )

    async def _fetch_signed_url(self) -> str:
        """Fetch a short-lived signed WS URL so we don't have to send the API key."""
        async with httpx.AsyncClient(timeout=15.0) as client:
            resp = await client.get(
                ELEVENLABS_SIGNED_URL,
                params={"agent_id": self.agent_id},
                headers={"xi-api-key": self.api_key},
            )
            resp.raise_for_status()
            data = resp.json()
            signed = data.get("signed_url")
            if not signed:
                raise RuntimeError(
                    "ElevenLabs signed-url response missing 'signed_url' field"
                )
            return signed

    async def connect(self) -> None:
        """Connect to ElevenLabs Conversational AI."""
        if self.use_signed_url:
            if not self.agent_id:
                raise ValueError("use_signed_url=True requires agent_id")
            url = await self._fetch_signed_url()
            # Signed URL embeds auth — no xi-api-key header needed.
            self._ws = await websockets.connect(url)
        else:
            url = ELEVENLABS_CONVAI_URL
            if self.agent_id:
                url = f"{url}?agent_id={self.agent_id}"
            self._ws = await websockets.connect(
                url,
                additional_headers={"xi-api-key": self.api_key},
            )

        self._running = True

        # Build conversation_config_override with all optional fields plumbed.
        agent_cfg: dict[str, Any] = {}
        if self.first_message:
            agent_cfg["first_message"] = self.first_message
        if self.language:
            agent_cfg["language"] = self.language

        override: dict[str, Any] = {"tts": {"voice_id": self.voice_id}}
        if self.output_audio_format:
            override["tts"]["output_format"] = self.output_audio_format
        if self.input_audio_format:
            override["asr"] = {"input_format": self.input_audio_format}
        if agent_cfg:
            override["agent"] = agent_cfg

        config: dict[str, Any] = {
            "type": "conversation_initiation_client_data",
            "conversation_config_override": override,
        }

        await self._ws.send(json.dumps(config))

        # Kick off the background reader that drains the socket and feeds the
        # async-iterator queue exposed via `receive_events`.
        self._events = asyncio.Queue()
        self._reader_task = asyncio.create_task(self._read_loop())

    async def send_client_tool_result(
        self, tool_call_id: str, result: str, *, is_error: bool = False
    ) -> None:
        """Answer a ``client_tool_call`` from the ElevenLabs agent."""
        if self._ws is None:
            return
        await self._ws.send(
            json.dumps(
                {
                    "type": "client_tool_result",
                    "tool_call_id": tool_call_id,
                    "result": result,
                    "is_error": is_error,
                }
            )
        )

    async def send_audio(self, audio_bytes: bytes) -> None:
        """Send user audio to ElevenLabs.

        Per the ConvAI protocol, inbound caller audio is sent as a JSON
        message with a top-level `user_audio_chunk` key holding base64-encoded
        PCM (format must match the agent's `user_input_audio_format`).
        """
        if self._ws is None:
            return
        await self._ws.send(
            json.dumps(
                {"user_audio_chunk": base64.b64encode(audio_bytes).decode("ascii")}
            )
        )

    async def _respond_to_ping(self, event_id: Any, delay_ms: int) -> None:
        """Reply to a `ping` event with the matching `pong`, optionally after ping_ms."""
        try:
            if delay_ms and delay_ms > 0:
                await asyncio.sleep(delay_ms / 1000.0)
            if self._ws is None:
                return
            await self._ws.send(json.dumps({"type": "pong", "event_id": event_id}))
        except Exception as exc:  # pragma: no cover — defensive
            logger.warning("ElevenLabs ConvAI pong send failed: %s", exc)

    async def _emit_response_done_after_silence(self) -> None:
        """Emit `response_done` after `_AGENT_SILENCE_MS` without audio."""
        try:
            await asyncio.sleep(_AGENT_SILENCE_MS / 1000.0)
            if self._agent_speaking and self._events is not None:
                self._agent_speaking = False
                await self._events.put(("response_done", {}))
        except asyncio.CancelledError:
            pass

    def _reset_silence_timer(self) -> None:
        """Cancel any pending silence timer (called on every agent-audio chunk)."""
        if self._silence_task and not self._silence_task.done():
            self._silence_task.cancel()
        self._silence_task = None

    async def _finalize_agent_turn(self) -> None:
        """If an agent turn is in progress, emit `response_done` now."""
        self._reset_silence_timer()
        if self._agent_speaking and self._events is not None:
            self._agent_speaking = False
            await self._events.put(("response_done", {}))

    async def _read_loop(self) -> None:
        """Drain the WS and push parsed events onto the async queue."""
        assert self._events is not None
        try:
            async for raw in self._ws:
                try:
                    data = json.loads(raw)
                except (ValueError, TypeError):
                    continue
                msg_type = data.get("type", "")

                if msg_type == "ping":
                    # Server expects a pong within ~20 s or it terminates the WS.
                    ping_payload = data.get("ping_event") or data.get("ping") or {}
                    event_id = ping_payload.get("event_id") or data.get("event_id")
                    ping_ms = ping_payload.get("ping_ms") or 0
                    asyncio.create_task(self._respond_to_ping(event_id, ping_ms))
                    continue

                if msg_type == "conversation_initiation_metadata":
                    meta = data.get("conversation_initiation_metadata_event") or data
                    self.conversation_id = (
                        meta.get("conversation_id") or self.conversation_id
                    )
                    self.agent_output_audio_format = (
                        meta.get("agent_output_audio_format")
                        or self.agent_output_audio_format
                    )
                    self.user_input_audio_format = (
                        meta.get("user_input_audio_format")
                        or self.user_input_audio_format
                    )
                    # A new turn boundary: close any dangling agent turn.
                    await self._finalize_agent_turn()
                    continue

                if msg_type == "audio":
                    # Audio payload may be nested under `audio_event` per current
                    # ElevenLabs schema, or flat — support both.
                    audio_b64 = ""
                    audio_evt = data.get("audio_event")
                    if isinstance(audio_evt, dict):
                        audio_b64 = audio_evt.get("audio_base_64") or audio_evt.get(
                            "audio", ""
                        )
                    if not audio_b64:
                        audio_b64 = data.get("audio", "")
                    if audio_b64:
                        # Reset silence timer on every agent audio chunk.
                        self._reset_silence_timer()
                        self._agent_speaking = True
                        await self._events.put(("audio", base64.b64decode(audio_b64)))
                        # Schedule silence-based response_done.
                        self._silence_task = asyncio.create_task(
                            self._emit_response_done_after_silence()
                        )
                    continue

                if msg_type == "user_transcript":
                    evt = data.get("user_transcription_event") or data
                    text = evt.get("user_transcript") or evt.get("text", "")
                    # User speaking boundary: close any pending agent turn.
                    await self._finalize_agent_turn()
                    await self._events.put(("transcript_input", text))
                    continue

                if msg_type == "agent_response":
                    evt = data.get("agent_response_event") or data
                    text = evt.get("agent_response") or evt.get("text", "")
                    await self._events.put(("transcript_output", text))
                    # `agent_response` marks the START of the agent turn — do
                    # NOT emit response_done here. It will be emitted by the
                    # silence watcher or on interruption / next turn.
                    self._agent_speaking = True
                    await self._events.put(("response_start", {"text": text}))
                    continue

                if msg_type == "interruption":
                    await self._finalize_agent_turn()
                    await self._events.put(("interruption", None))
                    continue

                if msg_type == "client_tool_call":
                    # The ElevenLabs agent invoked a CLIENT tool — previously
                    # unhandled, so configured client tools stalled until the
                    # provider-side timeout and reported failure. Surface it
                    # as the shared function_call event so the stream handler
                    # routes it through the tool executor.
                    tool = data.get("client_tool_call") or {}
                    await self._events.put(
                        (
                            "function_call",
                            {
                                "call_id": tool.get("tool_call_id", ""),
                                "name": tool.get("tool_name", ""),
                                "arguments": tool.get("parameters", {}) or {},
                            },
                        )
                    )
                    continue

                if msg_type == "error":
                    err_text = (
                        data.get("message") or data.get("error") or json.dumps(data)
                    )
                    logger.error("ElevenLabs ConvAI error: %s", err_text)
                    await self._events.put(("error", err_text))
                    continue

        except websockets.exceptions.ConnectionClosed:
            pass
        except Exception as exc:  # pragma: no cover — defensive
            if self._events is not None:
                await self._events.put(("error", f"read_loop: {exc}"))
        finally:
            self._running = False
            if self._events is not None:
                await self._events.put(None)

    async def receive_events(self):
        """Yield events from ElevenLabs ConvAI.

        Yields tuples of (event_type, data):
        - ("audio", bytes) — agent audio chunk to forward to the caller
        - ("transcript_input", str) — what the user said
        - ("transcript_output", str) — what the AI said (full turn text)
        - ("response_start", dict) — agent begins speaking (start of turn)
        - ("response_done", dict) — agent turn finished (silence or interrupt)
        - ("interruption", None) — user interrupted the agent
        - ("error", str) — server-side error text
        """
        if self._events is None:
            return
        while True:
            evt = await self._events.get()
            if evt is None:
                return
            yield evt

    async def close(self) -> None:
        """Close the connection and cancel the background reader."""
        self.record_session_end()
        self._running = False
        self._reset_silence_timer()
        if self._reader_task and not self._reader_task.done():
            self._reader_task.cancel()
            try:
                await self._reader_task
            except (asyncio.CancelledError, Exception):
                pass
        self._reader_task = None
        if self._ws:
            try:
                await self._ws.close()
            except Exception:
                pass
            self._ws = None
