//! [`Listener`] backed by ElevenLabs' Realtime Speech-to-Text WebSocket.
//!
//! Wire format (per <https://elevenlabs.io/docs/api-reference/speech-to-text/v-1-speech-to-text-realtime>):
//!
//! - **Endpoint:** `wss://api.elevenlabs.io/v1/speech-to-text/realtime`
//!   with query string `?model_id=scribe_v1&audio_format=pcm_16000&commit_strategy=vad&language_code=<bcp47>`.
//! - **Auth:** `xi-api-key: <key>` header.
//! - **Outbound:** JSON messages
//!   `{"message_type":"input_audio_chunk","audio_base_64":"<b64>","commit":bool,"sample_rate":<hz>}`
//!   for every PCM frame the caller pushes via the [`PushHandle`].
//! - **Inbound:** tagged-by-`message_type` JSON union — `session_started`,
//!   `partial_transcript`, `committed_transcript`,
//!   `committed_transcript_with_timestamps`, plus a family of error
//!   variants (`error`, `auth_error`, `quota_exceeded`,
//!   `rate_limited`, etc.).
//!
//! Mapping into the FFI [`VoiceEvent`] taxonomy:
//!
//! | EL `message_type`                        | `VoiceEvent`                      |
//! | ---------------------------------------- | --------------------------------- |
//! | `partial_transcript`                     | `Partial { text, duration_ms }`   |
//! | `committed_transcript`                   | `Transcript { text, duration_ms, role: Unknown }` |
//! | `committed_transcript_with_timestamps`   | `Transcript { text, duration_ms, role: Unknown }` |
//! | `session_started`                        | logged at info; not surfaced      |
//! | any error variant                        | logged at warn / error; the listener treats `auth_error` and `unaccepted_terms` as fatal and ends the session — caller observes channel close. |
//!
//! `duration_ms` is wall-clock from the `session_started` event because
//! ElevenLabs does not include a per-utterance duration on the partial
//! / committed shapes. That is honest: the field documents how long
//! ago the session opened, not how long the speaker spoke.
//!
//! ## Test scope
//!
//! The pure parse / encode helpers (`encode_input_audio_chunk`,
//! `parse_realtime_message`, `is_fatal_realtime_error`) are unit-tested
//! against the AsyncAPI shapes. End-to-end testing requires an
//! `ELEVENLABS_API_KEY` and is therefore out-of-scope for the test
//! suite — the FFI smoke ships behind a `provider: "elevenlabs"`
//! option that consumers must opt into.

use crate::push_listener::PushHandle;
use crate::{Listener, Result, VoiceConfig, VoiceError, VoiceEvent};
use async_trait::async_trait;
use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
use futures::{SinkExt, StreamExt};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::protocol::Message;

const EVENT_CHANNEL_CAPACITY: usize = 64;
const PCM_QUEUE_CAPACITY: usize = 32;
const REALTIME_PATH: &str = "/v1/speech-to-text/realtime";
const DEFAULT_HOST: &str = "api.elevenlabs.io";
const DEFAULT_MODEL_ID: &str = "scribe_v1";

/// Listener that streams pushed PCM frames through ElevenLabs'
/// Realtime STT WebSocket and emits [`VoiceEvent`]s for each transcript
/// the server returns.
pub struct ElevenLabsStreamingListener {
    sample_rate: u32,
    handle: Mutex<Option<PushHandle>>,
    pcm_rx: Mutex<Option<mpsc::Receiver<Vec<u8>>>>,
    running: Arc<AtomicBool>,
    worker: Option<JoinHandle<()>>,
}

impl ElevenLabsStreamingListener {
    /// Construct for a given input sample rate. ElevenLabs supports
    /// `pcm_8000`, `pcm_16000`, `pcm_22050`, `pcm_24000`, `pcm_44100`,
    /// `pcm_48000` — anything else is rejected at `start` with a clear
    /// configuration error rather than silently downsampled.
    pub fn new(sample_rate: u32) -> Self {
        let (tx, rx) = mpsc::channel(PCM_QUEUE_CAPACITY);
        Self {
            sample_rate,
            handle: Mutex::new(Some(PushHandle::from_sender(tx))),
            pcm_rx: Mutex::new(Some(rx)),
            running: Arc::new(AtomicBool::new(false)),
            worker: None,
        }
    }

    /// Cloneable producer handle. `transcribe_stream_push` resolves
    /// against this via the same side registry the regular
    /// [`crate::PushListener`] uses.
    pub fn handle(&self) -> Option<PushHandle> {
        self.handle.lock().unwrap().as_ref().cloned()
    }
}

#[async_trait]
impl Listener for ElevenLabsStreamingListener {
    async fn start(&mut self, config: VoiceConfig) -> Result<mpsc::Receiver<VoiceEvent>> {
        if self.running.load(Ordering::SeqCst) {
            return Err(VoiceError::AlreadyRunning);
        }
        let audio_format = audio_format_for_rate(self.sample_rate)?;

        // Resolve API key with the same precedence the rest of car-voice
        // uses: explicit config field > ELEVENLABS_API_KEY env var >
        // OS keychain.
        let api_key = config
            .elevenlabs_api_key
            .clone()
            .filter(|k| !k.is_empty())
            .or_else(|| car_secrets::resolve_env_or_keychain("ELEVENLABS_API_KEY"))
            .ok_or_else(|| {
                VoiceError::Config(
                    "ELEVENLABS_API_KEY not set; set the env var or store it via \
                     `car secrets put ELEVENLABS_API_KEY`"
                        .into(),
                )
            })?;

        let lang_code = match config.language.as_str() {
            "en" => "eng".to_string(),
            other => other.to_string(),
        };
        let url = format!(
            "wss://{host}{path}?model_id={model}&audio_format={fmt}&commit_strategy=vad&language_code={lang}",
            host = DEFAULT_HOST,
            path = REALTIME_PATH,
            model = DEFAULT_MODEL_ID,
            fmt = audio_format,
            lang = lang_code,
        );

        let mut request = url
            .into_client_request()
            .map_err(|e| VoiceError::Config(format!("realtime URL build: {e}")))?;
        request.headers_mut().insert(
            "xi-api-key",
            api_key.parse().map_err(|e| {
                VoiceError::Config(format!("api key contained invalid header bytes: {e}"))
            })?,
        );

        let (ws_stream, _resp) = tokio_tungstenite::connect_async(request)
            .await
            .map_err(|e| VoiceError::Stt(format!("elevenlabs realtime connect: {e}")))?;
        let (mut ws_tx, mut ws_rx) = ws_stream.split();

        self.running.store(true, Ordering::SeqCst);
        let (event_tx, event_rx) = mpsc::channel::<VoiceEvent>(EVENT_CHANNEL_CAPACITY);
        let mut pcm_rx = self
            .pcm_rx
            .lock()
            .unwrap()
            .take()
            .ok_or_else(|| VoiceError::Config("listener has no rx half".into()))?;
        let running = self.running.clone();
        let sample_rate = self.sample_rate;

        // Single worker — half pumps PCM up, half pumps transcripts
        // down. They share the websocket via the split halves and exit
        // together when either side closes.
        let event_tx_for_send = event_tx.clone();
        let event_tx_for_recv = event_tx.clone();
        let running_for_send = running.clone();
        let session_started = Instant::now();

        let worker = crate::voice_runtime_handle().spawn(async move {
            tracing::info!("[voice] elevenlabs realtime listener started");

            let send_task = tokio::spawn(async move {
                while running_for_send.load(Ordering::SeqCst) {
                    let Some(bytes) = pcm_rx.recv().await else { break };
                    let payload = encode_input_audio_chunk(&bytes, sample_rate);
                    if ws_tx.send(Message::text(payload)).await.is_err() {
                        break;
                    }
                }
                // Best-effort polite close. Not all errors here matter.
                let _ = ws_tx.close().await;
                drop(event_tx_for_send);
            });

            while let Some(frame) = ws_rx.next().await {
                let frame = match frame {
                    Ok(f) => f,
                    Err(e) => {
                        tracing::warn!("[voice] elevenlabs realtime ws error: {e}");
                        break;
                    }
                };
                let text = match frame {
                    Message::Text(t) => t.to_string(),
                    Message::Binary(b) => String::from_utf8_lossy(&b).to_string(),
                    Message::Close(_) => break,
                    _ => continue,
                };
                let parsed = match parse_realtime_message(&text) {
                    Ok(p) => p,
                    Err(e) => {
                        tracing::debug!(
                            "[voice] elevenlabs realtime: ignoring non-protocol message: {e}"
                        );
                        continue;
                    }
                };
                let duration_ms = session_started.elapsed().as_millis() as u64;
                match parsed {
                    RealtimeMessage::SessionStarted => {
                        tracing::info!("[voice] elevenlabs realtime session started");
                    }
                    RealtimeMessage::Partial { text } => {
                        let _ = event_tx_for_recv
                            .send(VoiceEvent::Partial { text, duration_ms })
                            .await;
                    }
                    RealtimeMessage::Committed { text } => {
                        let _ = event_tx_for_recv
                            .send(VoiceEvent::Transcript {
                                text,
                                duration_ms,
                                role: crate::enrollment::TranscriptRole::Unknown,
                            })
                            .await;
                    }
                    RealtimeMessage::Error { kind, message } => {
                        if is_fatal_realtime_error(&kind) {
                            tracing::error!(
                                "[voice] elevenlabs realtime fatal {kind}: {message}; ending session"
                            );
                            break;
                        }
                        tracing::warn!(
                            "[voice] elevenlabs realtime non-fatal {kind}: {message}"
                        );
                    }
                }
            }
            send_task.abort();
            running.store(false, Ordering::SeqCst);
            tracing::info!("[voice] elevenlabs realtime listener exited");
        });
        self.worker = Some(worker);
        Ok(event_rx)
    }

    async fn stop(&mut self) -> Result<()> {
        if !self.running.swap(false, Ordering::SeqCst) {
            return Err(VoiceError::NotRunning);
        }
        // Drop the push handle so the send half observes channel close.
        *self.handle.lock().unwrap() = None;
        if let Some(task) = self.worker.take() {
            let _ = task.await;
        }
        Ok(())
    }

    fn is_running(&self) -> bool {
        self.running.load(Ordering::SeqCst)
    }
}

/// Map a sample rate to the `audio_format` query value ElevenLabs
/// requires. Only the documented PCM rates are accepted; an
/// out-of-range rate fails fast rather than silently mismatching the
/// server-side decoder.
fn audio_format_for_rate(sample_rate: u32) -> Result<&'static str> {
    Ok(match sample_rate {
        8_000 => "pcm_8000",
        16_000 => "pcm_16000",
        22_050 => "pcm_22050",
        24_000 => "pcm_24000",
        44_100 => "pcm_44100",
        48_000 => "pcm_48000",
        other => {
            return Err(VoiceError::Config(format!(
                "elevenlabs realtime requires one of 8000/16000/22050/24000/44100/48000 Hz, got {other}"
            )))
        }
    })
}

/// Build the JSON payload for one `input_audio_chunk` message.
pub(crate) fn encode_input_audio_chunk(pcm_bytes: &[u8], sample_rate: u32) -> String {
    let payload = serde_json::json!({
        "message_type": "input_audio_chunk",
        "audio_base_64": B64.encode(pcm_bytes),
        // We rely on server-side VAD (`commit_strategy=vad` on the
        // connect URL); per-chunk `commit: false` is what every
        // example in the EL realtime docs uses for that strategy.
        "commit": false,
        "sample_rate": sample_rate,
    });
    payload.to_string()
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum RealtimeMessage {
    SessionStarted,
    Partial { text: String },
    Committed { text: String },
    Error { kind: String, message: String },
}

/// Parse one inbound websocket text frame against the AsyncAPI
/// subscribe schema. Unknown `message_type`s become `Err(...)` so the
/// caller can log + skip.
pub(crate) fn parse_realtime_message(json: &str) -> std::result::Result<RealtimeMessage, String> {
    let value: serde_json::Value =
        serde_json::from_str(json).map_err(|e| format!("invalid JSON: {e}"))?;
    let kind = value
        .get("message_type")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "missing message_type".to_string())?;
    let text_field = || -> std::result::Result<String, String> {
        value
            .get("text")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string())
            .ok_or_else(|| format!("{kind} missing text"))
    };
    let error_field = || -> String {
        value
            .get("error")
            .and_then(|v| v.as_str())
            .unwrap_or("(no error message)")
            .to_string()
    };
    match kind {
        "session_started" => Ok(RealtimeMessage::SessionStarted),
        "partial_transcript" => Ok(RealtimeMessage::Partial {
            text: text_field()?,
        }),
        "committed_transcript" | "committed_transcript_with_timestamps" => {
            Ok(RealtimeMessage::Committed {
                text: text_field()?,
            })
        }
        // All the documented error shapes carry an `error` string and
        // identify themselves by their tag.
        "error"
        | "auth_error"
        | "quota_exceeded"
        | "commit_throttled"
        | "unaccepted_terms"
        | "rate_limited"
        | "queue_overflow"
        | "resource_exhausted"
        | "session_time_limit_exceeded"
        | "input_error"
        | "chunk_size_exceeded"
        | "insufficient_audio_activity"
        | "transcriber_error" => Ok(RealtimeMessage::Error {
            kind: kind.to_string(),
            message: error_field(),
        }),
        other => Err(format!("unknown message_type {other}")),
    }
}

/// Errors that should end the listener — recovery from these requires
/// human action (rotate key, accept terms, raise quota), not just a
/// retry. Non-fatal errors get logged and the session continues.
pub(crate) fn is_fatal_realtime_error(kind: &str) -> bool {
    matches!(
        kind,
        "auth_error" | "unaccepted_terms" | "quota_exceeded" | "session_time_limit_exceeded"
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn audio_format_for_rate_accepts_documented_rates() {
        assert_eq!(audio_format_for_rate(16_000).unwrap(), "pcm_16000");
        assert_eq!(audio_format_for_rate(48_000).unwrap(), "pcm_48000");
        assert!(audio_format_for_rate(11_025).is_err());
    }

    #[test]
    fn encode_input_audio_chunk_round_trips_base64() {
        let raw = vec![0x01u8, 0x02, 0x03, 0x04];
        let json = encode_input_audio_chunk(&raw, 16_000);
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["message_type"], "input_audio_chunk");
        assert_eq!(parsed["sample_rate"], 16_000);
        assert_eq!(parsed["commit"], false);
        let b64 = parsed["audio_base_64"].as_str().unwrap();
        let back = B64.decode(b64).unwrap();
        assert_eq!(back, raw);
    }

    #[test]
    fn parse_partial_transcript_message() {
        let m =
            parse_realtime_message(r#"{"message_type":"partial_transcript","text":"hello wor"}"#)
                .unwrap();
        assert_eq!(
            m,
            RealtimeMessage::Partial {
                text: "hello wor".into()
            }
        );
    }

    #[test]
    fn parse_committed_transcript_message() {
        let m = parse_realtime_message(
            r#"{"message_type":"committed_transcript","text":"hello world"}"#,
        )
        .unwrap();
        assert_eq!(
            m,
            RealtimeMessage::Committed {
                text: "hello world".into()
            }
        );
    }

    #[test]
    fn parse_committed_with_timestamps_treated_as_committed() {
        // We don't surface word-level timestamps yet — both the bare
        // committed shape and the timestamped shape map to the same
        // VoiceEvent::Transcript today.
        let m = parse_realtime_message(
            r#"{"message_type":"committed_transcript_with_timestamps","text":"hi","words":null}"#,
        )
        .unwrap();
        assert_eq!(m, RealtimeMessage::Committed { text: "hi".into() });
    }

    #[test]
    fn parse_session_started_message() {
        let m = parse_realtime_message(
            r#"{"message_type":"session_started","session_id":"s","config":{}}"#,
        )
        .unwrap();
        assert_eq!(m, RealtimeMessage::SessionStarted);
    }

    #[test]
    fn parse_auth_error_carries_kind_and_message() {
        let m =
            parse_realtime_message(r#"{"message_type":"auth_error","error":"invalid api key"}"#)
                .unwrap();
        match m {
            RealtimeMessage::Error { kind, message } => {
                assert_eq!(kind, "auth_error");
                assert_eq!(message, "invalid api key");
            }
            _ => panic!("expected error variant"),
        }
    }

    #[test]
    fn parse_rejects_missing_message_type() {
        assert!(parse_realtime_message(r#"{"text":"hi"}"#).is_err());
    }

    #[test]
    fn parse_rejects_unknown_message_type() {
        assert!(parse_realtime_message(r#"{"message_type":"made_up","x":1}"#).is_err());
    }

    #[test]
    fn fatal_realtime_errors_are_only_account_level() {
        assert!(is_fatal_realtime_error("auth_error"));
        assert!(is_fatal_realtime_error("unaccepted_terms"));
        assert!(is_fatal_realtime_error("quota_exceeded"));
        assert!(is_fatal_realtime_error("session_time_limit_exceeded"));
        // Non-fatal: throttling and rate-limiting should be retried by
        // the caller, not surfaced as session death.
        assert!(!is_fatal_realtime_error("commit_throttled"));
        assert!(!is_fatal_realtime_error("rate_limited"));
        assert!(!is_fatal_realtime_error("queue_overflow"));
        assert!(!is_fatal_realtime_error("transcriber_error"));
    }
}
