//! SSE stream helpers — parses `data:` frames into typed event structs.

use async_stream::try_stream;
use futures::Stream;
use futures::StreamExt;

use crate::client::Kranth;
use crate::errors::{Error, Result};
use crate::types::{DebateEvent, ReconEvent, SimEvent};

pub(crate) fn stream_sim_events(
    client: Kranth,
    sim_id: String,
) -> impl Stream<Item = Result<SimEvent>> {
    try_stream! {
        // Mint a short-lived stream token (SSE endpoint can't accept Bearer headers).
        let token_url = format!("{}/v1/sims/{}/stream-token", client.base_url(), sim_id);
        let token_resp = client
            .http()
            .post(&token_url)
            .bearer_auth(client.api_key())
            .send()
            .await?;
        let token_status = token_resp.status();
        let token_bytes = token_resp.bytes().await?;
        if !token_status.is_success() {
            Err(Error::Api {
                status: token_status.as_u16(),
                code: "stream_token_failed".into(),
                message: String::from_utf8_lossy(&token_bytes).into_owned(),
            })?;
        }
        let token_json: serde_json::Value = serde_json::from_slice(&token_bytes)?;
        let token = token_json["token"].as_str()
            .ok_or_else(|| Error::Api { status: 502, code: "stream_token_missing".into(), message: "no token in response".into() })?
            .to_string();

        let url = format!("{}/v1/sims/{}/events?token={}", client.base_url(), sim_id, token);
        let resp = client
            .http()
            .get(url)
            .header("accept", "text/event-stream")
            .send()
            .await?;
        let status = resp.status();
        let mut stream = if !status.is_success() {
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status: status.as_u16(),
                code: "stream_open_failed".into(),
                message: body,
            })?;
        } else {
            resp.bytes_stream()
        };
        let mut buf: Vec<u8> = Vec::with_capacity(8 * 1024);
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            buf.extend_from_slice(&chunk);
            // Frames end on a blank line (\n\n or \r\n\r\n).
            while let Some(pos) = find_frame_end(&buf) {
                let frame = buf[..pos].to_vec();
                let drop_n = pos + delim_len(&buf[pos..]);
                buf.drain(..drop_n);
                let payload = extract_data_lines(&frame);
                if payload.is_empty() || payload == "ping" {
                    continue;
                }
                match serde_json::from_str::<SimEvent>(&payload) {
                    Ok(ev) => yield ev,
                    Err(_) => { /* skip non-JSON keep-alives */ }
                }
            }
        }
    }
}

// ─── recon SSE ─────────────────────────────────────────────────

pub(crate) fn stream_recon_events(
    client: Kranth,
    recon_id: String,
) -> impl Stream<Item = Result<ReconEvent>> {
    try_stream! {
        let token_url = format!("{}/v1/recon/{}/stream-token", client.base_url(), recon_id);
        let token_resp = client
            .http()
            .post(&token_url)
            .bearer_auth(client.api_key())
            .send()
            .await?;
        let token_status = token_resp.status();
        let token_bytes = token_resp.bytes().await?;
        if !token_status.is_success() {
            Err(Error::Api {
                status: token_status.as_u16(),
                code: "stream_token_failed".into(),
                message: String::from_utf8_lossy(&token_bytes).into_owned(),
            })?;
        }
        let token_json: serde_json::Value = serde_json::from_slice(&token_bytes)?;
        let token = token_json["token"].as_str()
            .ok_or_else(|| Error::Api { status: 502, code: "stream_token_missing".into(), message: "no token in response".into() })?
            .to_string();

        let url = format!("{}/v1/recon/{}/events?token={}", client.base_url(), recon_id, token);
        let resp = client
            .http()
            .get(url)
            .header("accept", "text/event-stream")
            .send()
            .await?;
        let status = resp.status();
        let mut stream = if !status.is_success() {
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status: status.as_u16(),
                code: "stream_open_failed".into(),
                message: body,
            })?;
        } else {
            resp.bytes_stream()
        };
        let mut buf: Vec<u8> = Vec::with_capacity(8 * 1024);
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            buf.extend_from_slice(&chunk);
            while let Some(pos) = find_frame_end(&buf) {
                let frame = buf[..pos].to_vec();
                let drop_n = pos + delim_len(&buf[pos..]);
                buf.drain(..drop_n);
                let payload = extract_data_lines(&frame);
                if payload.is_empty() || payload == "ping" {
                    continue;
                }
                match serde_json::from_str::<ReconEvent>(&payload) {
                    Ok(ev) => yield ev,
                    Err(_) => { /* skip non-JSON keep-alives */ }
                }
            }
        }
    }
}

// ─── debate SSE ────────────────────────────────────────────────

pub(crate) fn stream_debate_events(
    client: Kranth,
    debate_id: String,
) -> impl Stream<Item = Result<DebateEvent>> {
    try_stream! {
        let token_url = format!("{}/v1/debates/{}/stream-token", client.base_url(), debate_id);
        let token_resp = client
            .http()
            .post(&token_url)
            .bearer_auth(client.api_key())
            .send()
            .await?;
        let token_status = token_resp.status();
        let token_bytes = token_resp.bytes().await?;
        if !token_status.is_success() {
            Err(Error::Api {
                status: token_status.as_u16(),
                code: "stream_token_failed".into(),
                message: String::from_utf8_lossy(&token_bytes).into_owned(),
            })?;
        }
        let token_json: serde_json::Value = serde_json::from_slice(&token_bytes)?;
        let token = token_json["token"].as_str()
            .ok_or_else(|| Error::Api { status: 502, code: "stream_token_missing".into(), message: "no token in response".into() })?
            .to_string();

        let url = format!("{}/v1/debates/{}/events?token={}", client.base_url(), debate_id, token);
        let resp = client
            .http()
            .get(url)
            .header("accept", "text/event-stream")
            .send()
            .await?;
        let status = resp.status();
        let mut stream = if !status.is_success() {
            let body = resp.text().await.unwrap_or_default();
            return Err(Error::Api {
                status: status.as_u16(),
                code: "stream_open_failed".into(),
                message: body,
            })?;
        } else {
            resp.bytes_stream()
        };
        let mut buf: Vec<u8> = Vec::with_capacity(8 * 1024);
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            buf.extend_from_slice(&chunk);
            while let Some(pos) = find_frame_end(&buf) {
                let frame = buf[..pos].to_vec();
                let drop_n = pos + delim_len(&buf[pos..]);
                buf.drain(..drop_n);
                let payload = extract_data_lines(&frame);
                if payload.is_empty() || payload == "ping" {
                    continue;
                }
                match serde_json::from_str::<DebateEvent>(&payload) {
                    Ok(ev) => yield ev,
                    Err(_) => { /* skip non-JSON keep-alives */ }
                }
            }
        }
    }
}

fn find_frame_end(buf: &[u8]) -> Option<usize> {
    // Look for "\n\n" or "\r\n\r\n" — return start index of the delimiter.
    if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
        return Some(i);
    }
    buf.windows(2).position(|w| w == b"\n\n")
}

fn delim_len(after_start: &[u8]) -> usize {
    if after_start.starts_with(b"\r\n\r\n") {
        4
    } else {
        2
    }
}

fn extract_data_lines(frame: &[u8]) -> String {
    let text = std::str::from_utf8(frame).unwrap_or_default();
    let mut out = String::new();
    for line in text.split('\n') {
        let line = line.trim_end_matches('\r');
        if let Some(rest) = line.strip_prefix("data:") {
            if !out.is_empty() {
                out.push('\n');
            }
            out.push_str(rest.trim_start());
        }
    }
    out
}
