//! Live debate driver — pairs with `simulator.rs` and exposes the same
//! accessors. Behind the scenes a tokio task POSTs `/v1/debates`, opens the
//! `/v1/debates/:id/events` SSE stream, and drains events into a mutex-
//! protected snapshot. The cinema's render code reads the snapshot exactly
//! like it reads the simulator — same shapes, same names.
//!
//! v0.1 of the wiring: the verdict pane fills with real LLM-streamed text;
//! persona grid + sparklines stay on the local cascade timing for visual
//! continuity. Real persona-to-archetype mapping (and real sentiment
//! tracking from synthesis events) lands in the next brick.

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use anyhow::Context;
use futures::StreamExt;
use serde::{Deserialize, Serialize};

use super::simulator::{Phase, PersonaState, Reaction, Sentiment, Verdict};

const NUM_PERSONAS: usize = 10;
const WAKE_DURATION_MS: u128 = 5_000;
const HISTORY_SAMPLE_MS: u128 = 80;

pub struct LiveDebate {
    started: Instant,
    elapsed_ms: u128,
    last_history_tick_ms: u128,
    idea: String,
    /// Local persona-activation cascade — purely visual continuity with sim
    /// mode until per-speaker archetype mapping lands.
    personas: [PersonaState; NUM_PERSONAS],
    hostile_history: Vec<u64>,
    neutral_history: Vec<u64>,
    aligned_history: Vec<u64>,
    /// Shared with the tokio driver task — populated as SSE events arrive.
    inner: Arc<Mutex<Inner>>,
}

#[derive(Default)]
struct Inner {
    /// Server-confirmed phase. Defaults to `Waking` while we wait for the
    /// first event from the wire.
    phase: Option<Phase>,
    /// Token stream from `text.fragment` events, joined into one body — we
    /// render this directly in the right pane instead of the per-archetype
    /// reaction cards (those land in the next brick).
    streaming_text: String,
    /// Accumulated reaction summaries by speaker, finalised on each
    /// `turn.end` event. Each tuple is (speaker_id, role, text).
    finalised_turns: Vec<TurnRecord>,
    /// Surfaced into the verdict pane when set — fatal failure, e.g.
    /// 401/402/422/network. The cinema renders this and Esc returns the
    /// user to the splash.
    error: Option<String>,
    /// Brain's synthesis (verdict text) when the debate completes.
    verdict_text: Option<String>,
}

#[derive(Clone)]
#[allow(dead_code)] // Consumed in v0.2 once we map speaker_id → archetype + render per-turn cards.
pub struct TurnRecord {
    pub speaker_id: String,
    pub role: String,
    pub text: String,
}

#[derive(Deserialize)]
struct WireEvent {
    #[serde(default)]
    kind: Option<String>,
    #[serde(default)]
    speaker_id: Option<String>,
    #[serde(default)]
    role: Option<String>,
    #[serde(default)]
    text: Option<String>,
    #[serde(default, rename = "turn_seq")]
    _turn_seq: Option<i32>,
}

#[derive(Serialize)]
struct CreateRequest<'a> {
    topic: &'a str,
    mode: &'a str,
    public: bool,
    training_opt_in: bool,
    voice_enabled: bool,
}

#[derive(Deserialize)]
struct CreateResponse {
    id: String,
}

#[derive(Deserialize)]
struct DebateGet {
    #[serde(default)]
    synthesis: Option<SynthesisShape>,
    #[serde(default)]
    transcript: Vec<TurnShape>,
}

#[derive(Deserialize)]
struct SynthesisShape {
    #[serde(default)]
    summary: Option<String>,
    #[serde(default)]
    strongest_turn: Option<String>,
}

#[derive(Deserialize)]
struct TurnShape {
    #[serde(default)]
    audio_url: Option<String>,
    #[serde(default)]
    turn_seq: i32,
}

impl LiveDebate {
    pub fn new(
        api_url: String,
        api_key: String,
        idea: String,
        mode: String,
        voice_enabled: bool,
    ) -> Self {
        let inner = Arc::new(Mutex::new(Inner::default()));
        let inner_clone = inner.clone();
        let idea_clone = idea.clone();
        let mode_clone = mode.clone();

        // Spawn the driver task. tokio runtime is active because we're
        // inside main.rs::block_on (cinema::run is sync but the parent is
        // running it inside the runtime).
        tokio::spawn(async move {
            if let Err(err) = drive(
                api_url,
                api_key,
                idea_clone,
                mode_clone,
                voice_enabled,
                inner_clone.clone(),
            )
            .await
            {
                if let Ok(mut state) = inner_clone.lock() {
                    state.error = Some(format!("{err:#}"));
                    state.phase = Some(Phase::Done);
                }
            }
        });

        Self {
            started: Instant::now(),
            elapsed_ms: 0,
            last_history_tick_ms: 0,
            idea,
            personas: [PersonaState {
                activated_at_ms: None,
                sentiment: Sentiment::Neutral,
                reacted_at_ms: None,
            }; NUM_PERSONAS],
            hostile_history: Vec::with_capacity(240),
            neutral_history: Vec::with_capacity(240),
            aligned_history: Vec::with_capacity(240),
            inner,
        }
    }

    pub fn idea(&self) -> &str {
        &self.idea
    }

    pub fn elapsed(&self) -> Duration {
        self.started.elapsed()
    }

    pub fn phase(&self) -> Phase {
        let server_phase = self.inner.lock().ok().and_then(|s| s.phase);
        match server_phase {
            Some(p) => p,
            None => {
                if self.elapsed_ms < WAKE_DURATION_MS {
                    Phase::Waking
                } else {
                    Phase::Streaming
                }
            }
        }
    }

    pub fn personas(&self) -> [PersonaState; NUM_PERSONAS] {
        self.personas
    }

    pub fn reactions(&self) -> Vec<Reaction> {
        // For v0.1 wiring we don't have a real archetype-to-speaker map, so
        // we surface an empty reactions list — the cinema renders the
        // streaming token body instead via `streaming_text()`.
        Vec::new()
    }

    pub fn streaming_text(&self) -> String {
        self.inner
            .lock()
            .ok()
            .map(|s| s.streaming_text.clone())
            .unwrap_or_default()
    }

    #[allow(dead_code)]
    pub fn finalised_turns(&self) -> Vec<TurnRecord> {
        self.inner
            .lock()
            .ok()
            .map(|s| s.finalised_turns.clone())
            .unwrap_or_default()
    }

    pub fn error(&self) -> Option<String> {
        self.inner.lock().ok().and_then(|s| s.error.clone())
    }

    pub fn verdict(&self) -> Option<Verdict> {
        let state = self.inner.lock().ok()?;
        if matches!(state.phase, Some(Phase::Done)) && state.error.is_none() {
            Some(Verdict {
                aligned_pct: 0,
                hostile_pct: 0,
                neutral_pct: 0,
                headline: "COMPLETE",
            })
        } else {
            None
        }
    }

    pub fn verdict_text(&self) -> Option<String> {
        self.inner.lock().ok().and_then(|s| s.verdict_text.clone())
    }

    pub fn hostile_history(&self) -> &[u64] {
        &self.hostile_history
    }
    pub fn neutral_history(&self) -> &[u64] {
        &self.neutral_history
    }
    pub fn aligned_history(&self) -> &[u64] {
        &self.aligned_history
    }

    pub fn tick(&mut self, dt: Duration) {
        self.elapsed_ms = self.elapsed_ms.saturating_add(dt.as_millis());

        // Activate persona tiles on the same 500ms cascade as sim mode so the
        // grid still feels alive while we wait for real speaker-to-archetype
        // mapping (next brick).
        for (i, persona) in self.personas.iter_mut().enumerate() {
            let activation_ms = (i as u128) * 500;
            if persona.activated_at_ms.is_none() && self.elapsed_ms >= activation_ms {
                persona.activated_at_ms = Some(self.elapsed_ms);
            }
        }

        // Sample sentiment history — all aligned for now so the sparkline
        // shows movement rather than a flat line.
        if self.elapsed_ms.saturating_sub(self.last_history_tick_ms) >= HISTORY_SAMPLE_MS {
            self.last_history_tick_ms = self.elapsed_ms;
            let active = self
                .personas
                .iter()
                .filter(|p| p.activated_at_ms.is_some())
                .count() as u64;
            push_capped(&mut self.aligned_history, active, 240);
            push_capped(&mut self.neutral_history, 0, 240);
            push_capped(&mut self.hostile_history, 0, 240);
        }
    }
}

fn push_capped(v: &mut Vec<u64>, value: u64, cap: usize) {
    if v.len() >= cap {
        v.remove(0);
    }
    v.push(value);
}

async fn drive(
    api_url: String,
    api_key: String,
    idea: String,
    mode: String,
    voice_enabled: bool,
    inner: Arc<Mutex<Inner>>,
) -> anyhow::Result<()> {
    let http = reqwest::Client::builder()
        .timeout(Duration::from_secs(120))
        .build()
        .context("init http client")?;

    // 1) POST /v1/debates
    let create_resp = http
        .post(format!("{api_url}/v1/debates"))
        .bearer_auth(&api_key)
        .json(&CreateRequest {
            topic: &idea,
            mode: &mode,
            public: false,
            training_opt_in: false,
            voice_enabled,
        })
        .send()
        .await
        .context("contacting api")?;

    if !create_resp.status().is_success() {
        let status = create_resp.status();
        let body = create_resp.text().await.unwrap_or_default();
        anyhow::bail!("create debate failed ({status}): {body}");
    }

    let created: CreateResponse = create_resp
        .json()
        .await
        .context("parse create response")?;
    let debate_id = created.id;

    if let Ok(mut state) = inner.lock() {
        state.phase = Some(Phase::Streaming);
    }

    // 2) Open SSE — reqwest plus eventsource-stream to decode the wire.
    let sse_resp = http
        .get(format!("{api_url}/v1/debates/{debate_id}/events"))
        .bearer_auth(&api_key)
        .send()
        .await
        .context("opening sse stream")?;
    if !sse_resp.status().is_success() {
        anyhow::bail!("sse stream open failed: {}", sse_resp.status());
    }

    let mut events =
        eventsource_stream::EventStream::new(sse_resp.bytes_stream().map(|r| match r {
            Ok(b) => Ok(b),
            Err(e) => Err(eventsource_stream::EventStreamError::Transport(e)),
        }));

    let mut current_turn_text = String::new();
    while let Some(evt) = events.next().await {
        let evt = match evt {
            Ok(e) => e,
            Err(err) => {
                if let Ok(mut state) = inner.lock() {
                    state.error = Some(format!("sse: {err}"));
                    state.phase = Some(Phase::Done);
                }
                return Ok(());
            }
        };
        let parsed: WireEvent = match serde_json::from_str(&evt.data) {
            Ok(v) => v,
            Err(_) => continue,
        };
        let kind = parsed.kind.as_deref().unwrap_or("");
        match kind {
            "text.fragment" => {
                let text = parsed.text.unwrap_or_default();
                current_turn_text.push_str(&text);
                if let Ok(mut state) = inner.lock() {
                    state.streaming_text.push_str(&text);
                }
            }
            "turn.end" => {
                let speaker_id = parsed.speaker_id.unwrap_or_default();
                let role = parsed.role.unwrap_or_default();
                if let Ok(mut state) = inner.lock() {
                    state.finalised_turns.push(TurnRecord {
                        speaker_id,
                        role,
                        text: std::mem::take(&mut current_turn_text),
                    });
                    state.streaming_text.push('\n');
                    state.streaming_text.push('\n');
                }
            }
            "debate.end" => {
                // Fetch the full debate row — synthesis + transcript (with
                // per-turn audio_url when voice_enabled and Cartesia rendered).
                let mut audio_urls: Vec<String> = Vec::new();
                if let Ok(resp) = http
                    .get(format!("{api_url}/v1/debates/{debate_id}"))
                    .bearer_auth(&api_key)
                    .send()
                    .await
                {
                    if let Ok(body) = resp.json::<DebateGet>().await {
                        if let Some(syn) = body.synthesis {
                            if let Some(summary) = syn.summary {
                                if let Ok(mut state) = inner.lock() {
                                    state.verdict_text = Some(summary);
                                }
                            } else if let Some(strongest) = syn.strongest_turn {
                                if let Ok(mut state) = inner.lock() {
                                    state.verdict_text = Some(strongest);
                                }
                            }
                        }
                        if voice_enabled {
                            let mut turns = body.transcript;
                            turns.sort_by_key(|t| t.turn_seq);
                            audio_urls = turns
                                .into_iter()
                                .filter_map(|t| t.audio_url)
                                .filter(|u| !u.is_empty())
                                .collect();
                        }
                    }
                }
                if let Ok(mut state) = inner.lock() {
                    state.phase = Some(Phase::VerdictReveal);
                }
                // Hold the verdict reveal for a beat, then transition to
                // Done so the user can hit Esc and queue another idea.
                tokio::time::sleep(Duration::from_secs(3)).await;
                if let Ok(mut state) = inner.lock() {
                    state.phase = Some(Phase::Done);
                }

                // Best-effort voice playback. Spawned as a detached task so
                // it doesn't block the verdict transition or the cinema's
                // input loop. Player binary is auto-detected per OS.
                if voice_enabled && !audio_urls.is_empty() {
                    tokio::spawn(play_audio_sequence(audio_urls));
                }
                return Ok(());
            }
            _ => {}
        }
    }
    Ok(())
}

/// Best-effort sequential MP3 playback for completed-debate audio.
/// Picks the platform's native CLI audio player. Errors are silent because
/// audio is a polish feature — the debate's text already won.
async fn play_audio_sequence(urls: Vec<String>) {
    use std::process::{Command, Stdio};

    let player: Option<(&'static str, Vec<&'static str>)> = if cfg!(target_os = "macos") {
        Some(("afplay", vec![]))
    } else if cfg!(target_os = "linux") {
        if which_in_path("ffplay") {
            Some(("ffplay", vec!["-nodisp", "-autoexit", "-loglevel", "quiet"]))
        } else if which_in_path("mpg123") {
            Some(("mpg123", vec!["-q"]))
        } else if which_in_path("paplay") {
            Some(("paplay", vec![]))
        } else {
            None
        }
    } else if cfg!(target_os = "windows") {
        Some((
            "powershell",
            vec![
                "-NoProfile",
                "-Command",
                "(New-Object Media.SoundPlayer ($args[0])).PlaySync()",
            ],
        ))
    } else {
        None
    };

    let Some((bin, base_args)) = player else {
        return; // unsupported OS — silent no-op
    };
    let base_args: Vec<String> = base_args.iter().map(|s| s.to_string()).collect();

    for url in urls {
        let bin = bin.to_string();
        let mut args = base_args.clone();
        args.push(url);
        let _ = tokio::task::spawn_blocking(move || {
            Command::new(&bin)
                .args(&args)
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .status()
        })
        .await;
    }
}

fn which_in_path(bin: &str) -> bool {
    let Some(path) = std::env::var_os("PATH") else {
        return false;
    };
    for dir in std::env::split_paths(&path) {
        if dir.join(bin).exists() {
            return true;
        }
    }
    false
}
