//! # MockRealtimeBrain — in-process fake OpenAI Realtime WS server (spec §5.3)
//!
//! Behind `--features=mock` only. Runs entirely in-process; no network calls
//! to OpenAI, no API key required. Powers:
//! - the offline dev loop (`cargo run -p rutster-brain-realtime --features=mock`
//!   starts the brain process with this mock as the OpenAI side), and
//! - slice-3's integration test (no real credentials, deterministic canned
//!   responses).
//!
//! # Contract
//!
//! - Accepts a WS client connection (the brain process's OpenAI-side client).
//! - Asserts that the first event the client sends is a `session.update`
//!   with `turn_detection: null` (the S4 load-bearing decision — spec §4.3).
//! - On each `input_audio_buffer.append` event the client sends, replies
//!   with a canned `response.audio.delta` event carrying base64 LE i16 PCM
//!   (480 zeroed samples per 20 ms frame — same wire shape as slice-2's
//!   audio_out, so the brain's translator round-trips through the same
//!   PcmFrame codec real OpenAI would use).
//! - Echoes back `input_audio_buffer.speech_started` / `.speech_stopped`
//!   so the brain's translator can forward them as advisory tap events
//!   (spec §3.2 — these are advisory in slice-3; step 4 wires the FOB
//!   reflex loop).
//! - On `response.function_call_arguments.done`... the brain's translator
//!   consumes those (OpenAI emits them, the brain forwards to core); the
//!   mock doesn't generate unprompted function_call events (the integration
//!   test exercises the function-call round-trip via a deterministic test
//!   hook, not via the mock's own drive).

use std::net::SocketAddr;
use std::sync::Arc;

use futures_util::{SinkExt, StreamExt};
use serde_json::{Value, json};
use tokio::net::TcpListener;
use tokio::sync::{Mutex, oneshot};
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info, warn};

/// A trigger for the advisory schedule. The mock counts
/// `input_audio_buffer.append` events; when the count reaches
/// `after_audio_in_frames`, it emits `event` unprompted (simulating
/// the brain's VAD firing).
#[derive(Debug, Clone)]
pub struct AdvisoryTrigger {
    pub after_audio_in_frames: u32,
    pub event: AdvisoryKind,
}

#[derive(Debug, Clone, Copy)]
pub enum AdvisoryKind {
    SpeechStarted,
    SpeechStopped,
}

/// The handle returned by `MockRealtimeBrain::start`. Drop or call `shutdown`
/// to stop the mock server + abort its accept-loop task.
pub struct MockRealtimeBrain {
    /// The address the mock bound. The brain process dials this URL
    /// (`ws://<addr>/`) instead of the real `wss://api.openai.com/...`
    /// when running with `--features=mock`.
    pub addr: SocketAddr,
    pub shutdown: Option<oneshot::Sender<()>>,
    pub join: JoinHandle<()>,
    /// slice-4: shared advisory schedule. Created empty in `start` and
    /// cloned to every accepted connection; `set_advisory_schedule`
    /// mutates it in place so live connections pick it up.
    advisory_schedule: Arc<Mutex<Vec<AdvisoryTrigger>>>,
}

impl MockRealtimeBrain {
    /// Start the mock on an ephemeral port (caller picks up `addr` for the
    /// brain process to dial). The accept loop spawns per-connection tasks
    /// so multiple sessions can run concurrently in the integration test.
    pub async fn start() -> std::io::Result<Self> {
        let listener = TcpListener::bind("127.0.0.1:0").await?;
        let addr = listener.local_addr()?;
        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
        let schedule = Arc::new(Mutex::new(Vec::new()));
        let join = tokio::spawn(accept_loop(listener, shutdown_rx, schedule.clone()));
        info!(%addr, "MockRealtimeBrain listening (fake OpenAI Realtime)");
        Ok(Self {
            addr,
            shutdown: Some(shutdown_tx),
            join,
            advisory_schedule: schedule,
        })
    }

    /// Build the URL the brain process's OpenAI client should dial instead
    /// of `wss://api.openai.com/v1/realtime?model=...`. The path is `/v1/realtime`
    /// so the brain client's `into_client_request` call works the same as it
    /// does against the real OpenAI endpoint.
    pub fn url(&self) -> String {
        format!("ws://{}/v1/realtime?model=mock", self.addr)
    }

    /// Set a schedule of advisory events the mock emits UNPROMPTED after
    /// observing N `input_audio_buffer.append` events. Used by the slice-4
    /// barge-in e2e test to drive the reflex via the secondary trigger path.
    pub async fn set_advisory_schedule(&mut self, schedule: Vec<AdvisoryTrigger>) {
        let mut locked = self.advisory_schedule.lock().await;
        *locked = schedule;
    }
}

impl Drop for MockRealtimeBrain {
    fn drop(&mut self) {
        // Best-effort shutdown signal — the task may already have exited.
        // `Option::take()` lets us send by-value without moving out of
        // `self` (Drop takes `&mut self`, not `self`).
        if let Some(tx) = self.shutdown.take() {
            let _ = tx.send(());
        }
        self.join.abort();
    }
}

async fn accept_loop(
    listener: TcpListener,
    mut shutdown: oneshot::Receiver<()>,
    schedule: Arc<Mutex<Vec<AdvisoryTrigger>>>,
) {
    loop {
        tokio::select! {
            biased;
            _ = &mut shutdown => {
                info!("MockRealtimeBrain accept loop shutting down");
                return;
            }
            res = listener.accept() => {
                let (stream, peer) = match res {
                    Ok(s) => s,
                    Err(e) => {
                        warn!(error = ?e, "MockRealtimeBrain accept failed; continuing");
                        continue;
                    }
                };
                let schedule = schedule.clone();
                tokio::spawn(handle_connection(stream, peer, schedule));
            }
        }
    }
}

/// Handle one brain-process → mock-OpenAI WS connection. Each connection is
/// one OpenAI Realtime session (stateless across reconnects — same as the
/// brain process's own contract for the tap side).
async fn handle_connection(
    stream: tokio::net::TcpStream,
    peer: SocketAddr,
    schedule: Arc<Mutex<Vec<AdvisoryTrigger>>>,
) {
    let ws = tokio_tungstenite::accept_async(stream).await;
    let Ok(mut ws) = ws else {
        warn!(%peer, "MockRealtimeBrain WS handshake failed");
        return;
    };
    debug!(%peer, "MockRealtimeBrain connection accepted");

    // slice-4: per-connection counter for advisory schedule triggers.
    let mut audio_in_count: u32 = 0;

    // First event MUST be session.update with turn_detection: null.
    // Wait for it (bounded 2s — the brain process sends it immediately
    // after the WS handshake per `run_openai_pump`).
    let first = tokio::time::timeout(std::time::Duration::from_secs(2), ws.next()).await;
    match first {
        Ok(Some(Ok(msg))) => {
            let Ok(text) = msg.into_text() else {
                warn!(%peer, "MockRealtimeBrain first frame not text; closing");
                let _ = ws.close(None).await;
                return;
            };
            let event: Value = match serde_json::from_str(&text) {
                Ok(v) => v,
                Err(e) => {
                    warn!(%peer, error = ?e, "MockRealtimeBrain first frame not JSON; closing");
                    let _ = ws.close(None).await;
                    return;
                }
            };
            if event.get("type").and_then(|v| v.as_str()) != Some("session.update") {
                warn!(%peer, "MockRealtimeBrain first event not session.update; closing");
                let _ = ws.close(None).await;
                return;
            }
            // S4 load-bearing assertion (spec §4.3). Do NOT silently accept
            // a session.update that has turn_detection != null — that would
            // mask a regression in the brain's translator.
            let td = event.pointer("/session/turn_detection");
            if td != Some(&Value::Null) {
                warn!(%peer, td = ?td, "MockRealtimeBrain S4 violation: turn_detection != null; closing");
                // Echo back an OpenAI-shaped error so the brain surfaces it.
                let err = json!({
                    "type": "error",
                    "error": {
                        "code": "invalid_session_update",
                        "message": "MockRealtimeBrain: turn_detection must be null (S4)"
                    }
                });
                let _ = ws.send(Message::Text(err.to_string())).await;
                let _ = ws.close(None).await;
                return;
            }
            info!(%peer, "MockRealtimeBrain session.update accepted (turn_detection=null)");
        }
        _ => {
            warn!(%peer, "MockRealtimeBrain no session.update within 2s; closing");
            let _ = ws.close(None).await;
            return;
        }
    }

    // Pump loop: reply to input_audio_buffer.append with canned audio.
    while let Some(msg_res) = ws.next().await {
        let msg = match msg_res {
            Ok(m) => m,
            Err(e) => {
                debug!(%peer, error = ?e, "MockRealtimeBrain WS recv error; closing");
                return;
            }
        };
        let Ok(text) = msg.into_text() else { continue };
        let event: Value = match serde_json::from_str(&text) {
            Ok(v) => v,
            Err(e) => {
                debug!(%peer, error = ?e, "MockRealtimeBrain non-JSON event; ignoring");
                continue;
            }
        };
        let event_type = event.get("type").and_then(|v| v.as_str()).unwrap_or("");
        match event_type {
            "input_audio_buffer.append" => {
                // Canned response: one `response.audio.delta` per append
                // carrying 480 zeroed samples as base64 LE i16 PCM (the same
                // wire shape as slice-2's audio_out — verified by the
                // translator's existing round-trip test). Identical bytes
                // every time: deterministic for test assertions.
                use base64::Engine;
                let pcm_zeros = vec![0u8; 960]; // 480 i16 samples × 2 bytes
                let pcm_b64 = base64::engine::general_purpose::STANDARD.encode(&pcm_zeros);
                let delta = json!({
                    "type": "response.audio.delta",
                    "delta": pcm_b64
                });
                if let Err(e) = ws.send(Message::Text(delta.to_string())).await {
                    debug!(%peer, error = ?e, "MockRealtimeBrain send delta failed; closing");
                    return;
                }

                // slice-4: count the append + emit any scheduled advisories.
                audio_in_count += 1;
                for trigger in schedule.lock().await.iter() {
                    if trigger.after_audio_in_frames == audio_in_count {
                        let evt = match trigger.event {
                            AdvisoryKind::SpeechStarted => {
                                json!({ "type": "input_audio_buffer.speech_started" })
                            }
                            AdvisoryKind::SpeechStopped => {
                                json!({ "type": "input_audio_buffer.speech_stopped" })
                            }
                        };
                        if let Err(e) = ws.send(Message::Text(evt.to_string())).await {
                            debug!(%peer, error = ?e, "MockRealtimeBrain send advisory failed; closing");
                            return;
                        }
                    }
                }
            }
            "input_audio_buffer.speech_started" => {
                // Pass-through echo so the brain's translator forwards it as
                // a tap `speech_started` advisory event.
                let evt = json!({ "type": "input_audio_buffer.speech_started" });
                let _ = ws.send(Message::Text(evt.to_string())).await;
            }
            "input_audio_buffer.speech_stopped" => {
                let evt = json!({ "type": "input_audio_buffer.speech_stopped" });
                let _ = ws.send(Message::Text(evt.to_string())).await;
            }
            "conversation.item.create" => {
                // The brain sends this to relay a function_call_output (tool
                // reply) back to OpenAI. The mock just acknowledges by
                // echoing a typed `conversation.item.created` event so the
                // brain's OpenAI pump doesn't treat the silence as an error.
                let evt = json!({
                    "type": "conversation.item.created",
                    "item": event.get("item").cloned().unwrap_or(Value::Null)
                });
                let _ = ws.send(Message::Text(evt.to_string())).await;
            }
            _ => {
                debug!(%peer, event_type = %event_type, "MockRealtimeBrain ignoring event");
            }
        }
    }
    debug!(%peer, "MockRealtimeBrain connection closed");
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;
    use tokio_tungstenite::tungstenite::client::IntoClientRequest;

    /// The S4 turn-ownership decision is load-bearing (spec §4.3 + ADR-0008).
    /// The mock asserts the brain's session.update has turn_detection: null;
    /// this test confirms a session.update WITH turn_detection: null is
    /// accepted (the green path).
    #[tokio::test]
    async fn accepts_session_update_with_turn_detection_null() {
        let mock = MockRealtimeBrain::start().await.unwrap();
        let url = mock.url();
        let req = url.as_str().into_client_request().unwrap();
        let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();

        // Send session.update with turn_detection: null.
        let session_update = json!({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",
                "turn_detection": null
            }
        });
        ws.send(Message::Text(session_update.to_string()))
            .await
            .unwrap();

        // Send input_audio_buffer.append — must receive a canned
        // response.audio.delta (the brain's translator round-trips this
        // through encode_audio_out, so the wire shape must be exactly what
        // real OpenAI sends: base64 PCM16 in a `delta` field).
        let append = json!({
            "type": "input_audio_buffer.append",
            "audio": "AAAA"
        });
        ws.send(Message::Text(append.to_string())).await.unwrap();

        let delta_msg = tokio::time::timeout(Duration::from_millis(500), ws.next())
            .await
            .expect("delta within 500ms")
            .unwrap()
            .unwrap();
        let delta_text = delta_msg.into_text().unwrap();
        let delta: Value = serde_json::from_str(&delta_text).unwrap();
        assert_eq!(delta["type"], "response.audio.delta");
        assert!(delta["delta"].is_string());
        // The mock's canned PCM is 480 zeroed samples; the base64 string
        // must decode to exactly 960 bytes (480 × 2 bytes/i16).
        use base64::Engine;
        let bytes = base64::engine::general_purpose::STANDARD
            .decode(delta["delta"].as_str().unwrap())
            .unwrap();
        assert_eq!(bytes.len(), 960);
    }

    /// The S4 violation: session.update with turn_detection != null
    /// surfaces a typed OpenAI-shaped error so the brain's pump fails fast
    /// + visibly. Catches a future regression where the brain process
    /// forgets to set turn_detection: null.
    #[tokio::test]
    async fn rejects_session_update_with_turn_detection_set() {
        let mock = MockRealtimeBrain::start().await.unwrap();
        let url = mock.url();
        let req = url.as_str().into_client_request().unwrap();
        let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();

        let session_update = json!({
            "type": "session.update",
            "session": {
                "voice": "alloy",
                "turn_detection": { "type": "server_vad" }
            }
        });
        ws.send(Message::Text(session_update.to_string()))
            .await
            .unwrap();

        let err_msg = tokio::time::timeout(Duration::from_millis(500), ws.next())
            .await
            .expect("error within 500ms")
            .unwrap()
            .unwrap();
        let err: Value = serde_json::from_str(&err_msg.into_text().unwrap()).unwrap();
        assert_eq!(err["type"], "error");
        assert_eq!(err["error"]["code"], "invalid_session_update");
    }

    /// slice-4: MockRealtimeBrain can emit `speech_started`/`speech_stopped`
    /// on a programmable schedule, simulating the brain's VAD firing. This
    /// is what the slice-4 barge-in e2e test drives.
    #[tokio::test]
    async fn emits_speech_started_on_schedule_after_n_audio_in_frames() {
        let mut mock = MockRealtimeBrain::start().await.unwrap();
        mock.set_advisory_schedule(vec![
            AdvisoryTrigger {
                after_audio_in_frames: 2,
                event: AdvisoryKind::SpeechStarted,
            },
            AdvisoryTrigger {
                after_audio_in_frames: 4,
                event: AdvisoryKind::SpeechStopped,
            },
        ])
        .await;
        let url = mock.url();
        let req = url.as_str().into_client_request().unwrap();
        let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();

        // Send session.update first (the mock's contract).
        let session_update = json!({
            "type": "session.update",
            "session": { "turn_detection": null }
        });
        ws.send(Message::Text(session_update.to_string()))
            .await
            .unwrap();

        // Send 2 audio_in appends → expect a speech_started.
        for _ in 0..2 {
            let append = json!({ "type": "input_audio_buffer.append", "audio": "AAAA" });
            ws.send(Message::Text(append.to_string())).await.unwrap();
        }

        // Skip the canned response.audio.delta replies; wait for speech_started.
        let mut saw_started = false;
        for _ in 0..10 {
            let msg = tokio::time::timeout(Duration::from_millis(500), ws.next())
                .await
                .expect("event within 500ms")
                .unwrap()
                .unwrap();
            let text = msg.into_text().unwrap();
            if text.contains("speech_started") {
                saw_started = true;
                break;
            }
        }
        assert!(saw_started, "mock must emit speech_started after N appends");
    }
}
