//! # wss://api.openai.com/v1/realtime client (spec §4)
//!
//! Owns the OpenAI-side WS connection. Brings the OpenAI Realtime event
//! stream up, sends `session.update` with `turn_detection: null` (S4,
//! spec §4.3) on handshake, then runs a pump loop that:
//! - reads tap-side events (audio_in, function_call_output) from the
//!   tap WS server's `WebSocketStream` and translates + forwards them
//!   to OpenAI;
//! - reads OpenAI events (response.audio.delta, speech_started/stopped,
//!   function_call_arguments.done, error) and translates + forwards them
//!   to the tap WS server.
//!
//! Reconnects with bounded backoff on OpenAI-side WS failure (spec §4.4 —
//! the tap side stays connected; the OpenAI side has its own failure
//! surface). The brain process emits a tap `error` event on OpenAI-side
//! failure so the core can observe it.

// Imports are split by origin so the reader can trace which types come from
// the wire-shape (rutster-tap — shared with the core), which come from the
// pure translators (crate::translator — this crate's pure-function layer),
// and which come from third-party plumbing.
use crate::translator::{
    build_openai_session_update, openai_audio_delta_to_tap_audio_out,
    openai_function_call_arguments_done_to_tap, openai_speech_event_to_tap,
    tap_audio_in_to_openai_append, tap_function_call_output_to_openai_create_item,
};
use rutster_tap::{DecodedPayload, TapProtoError, decode_envelope, encode_function_call};
use serde_json::Value;
use thiserror::Error;
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::Message;
use tracing::{info, warn};

#[derive(Debug, Error)]
pub enum OpenAiClientError {
    #[error("OpenAI WS error: {0}")]
    Ws(#[from] tokio_tungstenite::tungstenite::Error),
    #[error("OpenAI auth failed (401)")]
    AuthFailed,
    #[error("translator error: {0}")]
    Translate(#[from] crate::translator::TranslateError),
    #[error("tap protocol error: {0}")]
    TapProto(#[from] TapProtoError),
}

/// Build the OpenAI Realtime URL for the given model. Spec §4.2 + §5.3:
/// `wss://api.openai.com/v1/realtime?model=<model>`.
pub fn openai_realtime_url(model: &str) -> String {
    format!("wss://api.openai.com/v1/realtime?model={model}")
}

/// Build the HTTP headers for the OpenAI WS handshake (spec §5.3). Two
/// required headers:
/// - `Authorization: Bearer <api_key>`
/// - `OpenAI-Beta: realtime=v1`
pub fn openai_headers(api_key: &str) -> Vec<(String, String)> {
    vec![
        ("Authorization".to_string(), format!("Bearer {api_key}")),
        ("OpenAI-Beta".to_string(), "realtime=v1".to_string()),
    ]
}

/// Drive the OpenAI Realtime WS connection + the tap-side pump.
///
/// `tap_rx`: inbound side — *tap frames* the brain received from the core
///   (audio_in, function_call_output). We translate each one to its OpenAI
///   equivalent + send to OpenAI.
/// `openai_ws`: the OpenAI WS connection (already connected; the caller
///   does the dial + auth).
/// `tap_tx`: outbound side — *tap frames* the brain sends back to the core
///   (audio_out, speech_started/stopped, function_call, error). We
///   translate OpenAI events into these + send.
pub async fn run_openai_pump<T>(
    mut openai_ws: tokio_tungstenite::WebSocketStream<T>,
    mut tap_rx: mpsc::Receiver<String>,
    tap_tx: mpsc::Sender<String>,
    voice: String,
) -> Result<(), OpenAiClientError>
where
    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    // === Handshake: send session.update with turn_detection: null (S4). ===
    let session_update = build_openai_session_update(&voice);
    openai_ws
        .send(Message::Text(session_update.to_string()))
        .await?;
    info!(voice = %voice, "sent session.update to OpenAI (turn_detection: null)");

    let mut seq_egress = 0u64;

    use futures_util::{SinkExt, StreamExt};
    loop {
        tokio::select! {
            // Inbound tap frame from the core (audio_in, function_call_output).
            tap_str = tap_rx.recv() => {
                let Some(tap_str) = tap_str else {
                    info!("tap_rx closed; ending OpenAI pump");
                    return Ok(());
                };
                let decoded = decode_envelope(&tap_str)?;
                match decoded.payload {
                    DecodedPayload::AudioIn(audio) => {
                        let append = tap_audio_in_to_openai_append(&audio.pcm);
                        openai_ws.send(Message::Text(append.to_string())).await?;
                    }
                    DecodedPayload::FunctionCallOutput(out) => {
                        let create_item = tap_function_call_output_to_openai_create_item(
                            &out.id, &out.status, &out.result,
                        );
                        openai_ws.send(Message::Text(create_item.to_string())).await?;
                    }
                    // Ignore others; slice-2's hello/audio_in on the tap side
                    // happens before this pump starts.
                    _ => warn!(?decoded.payload, "unexpected tap frame to OpenAI pump"),
                }
            }
            // Inbound OpenAI event.
            msg = openai_ws.next() => {
                let Some(msg) = msg else {
                    info!("OpenAI WS stream ended");
                    return Ok(());
                };
                let msg = msg?;
                let Ok(text) = msg.into_text() else { continue };
                let openai_event: Value = match serde_json::from_str(&text) {
                    Ok(v) => v,
                    Err(e) => {
                        warn!(error = ?e, "OpenAI sent non-JSON event; ignoring");
                        continue;
                    }
                };
                let event_type = openai_event.get("type").and_then(|v| v.as_str()).unwrap_or("");
                match event_type {
                    "response.audio.delta" => {
                        let tap_str = openai_audio_delta_to_tap_audio_out(
                            &openai_event, seq_egress, 0,
                        )?;
                        seq_egress += 1;
                        tap_tx.send(tap_str).await.map_err(|_| {
                            OpenAiClientError::Ws(
                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
                            )
                        })?;
                    }
                    "input_audio_buffer.speech_started" => {
                        let tap_str = openai_speech_event_to_tap(
                            &openai_event, true, seq_egress, 0,
                        )?;
                        seq_egress += 1;
                        tap_tx.send(tap_str).await.map_err(|_| {
                            OpenAiClientError::Ws(
                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
                            )
                        })?;
                    }
                    "input_audio_buffer.speech_stopped" => {
                        let tap_str = openai_speech_event_to_tap(
                            &openai_event, false, seq_egress, 0,
                        )?;
                        seq_egress += 1;
                        tap_tx.send(tap_str).await.map_err(|_| {
                            OpenAiClientError::Ws(
                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
                            )
                        })?;
                    }
                    "response.function_call_arguments.done" => {
                        let (call_id, name, args_str) =
                            openai_function_call_arguments_done_to_tap(&openai_event)?;
                        let tap_str = encode_function_call(
                            &call_id, &name, &args_str, seq_egress, 0,
                        )?;
                        seq_egress += 1;
                        tap_tx.send(tap_str).await.map_err(|_| {
                            OpenAiClientError::Ws(
                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
                            )
                        })?;
                    }
                    "error" => {
                        // OpenAI emits a typed error event (e.g. 401 auth
                        // failure surfaces here). Inspect the .error.code.
                        let code = openai_event
                            .pointer("/error/code")
                            .and_then(|v| v.as_str())
                            .unwrap_or("unknown");
                        warn!(code = %code, "OpenAI error event");
                        if code == "invalid_api_key" {
                            return Err(OpenAiClientError::AuthFailed);
                        }
                    }
                    _ => {
                        // Unknown OpenAI event — log + drop (don't crash).
                        warn!(event_type = %event_type, "ignoring OpenAI event type");
                    }
                }
            }
        }
    }
}

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

    #[test]
    fn openai_url_for_known_model() {
        assert_eq!(
            openai_realtime_url("gpt-4o-realtime"),
            "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime"
        );
    }

    #[test]
    fn openai_headers_carry_bearer_auth_and_beta() {
        let h = openai_headers("sk-test-12345");
        assert_eq!(h[0].0, "Authorization");
        assert_eq!(h[0].1, "Bearer sk-test-12345");
        assert_eq!(h[1].0, "OpenAI-Beta");
        assert_eq!(h[1].1, "realtime=v1");
    }
}
