//! Minimal `text/event-stream` parser shared by the streamable HTTP and
//! legacy SSE transports.
//!
//! The HTML Living Standard's SSE grammar is small enough that a hand
//! rolled parser is shorter and easier to audit than another crate
//! dependency. We feed byte slices in one chunk at a time and pop
//! complete events as they materialize. The parser is split out of the
//! transport modules because both  (streamable HTTP) and
//! (legacy SSE) need it.

use std::collections::VecDeque;

/// One decoded SSE event.
///
/// Field semantics mirror the spec: `event` defaults to `"message"` when
/// the stream omits it, `data` is the concatenation of every `data:` line
/// in the event separated by `\n`, and `id` is the `id:` line if any.
/// `retry` is captured so the transport's reconnect loop can honor a
/// server-supplied retry hint.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct SseEvent {
    /// Event type. Defaults to `"message"` when the stream omits it.
    pub event: String,
    /// Concatenated `data:` payload, joined by `\n`.
    pub data: String,
    /// `id:` value or `None` if the event did not carry one.
    pub id: Option<String>,
    /// `retry:` value in milliseconds, if the stream provided one.
    pub retry: Option<u64>,
}

/// Incremental SSE parser. Feed bytes in, drain finished events out.
///
/// Keeps a single internal byte buffer; complete lines are consumed in
/// place. Stateful between calls so callers can stream from
/// `reqwest::Response::bytes_stream()` without buffering the whole body.
#[derive(Debug, Default)]
pub(crate) struct SseParser {
    /// Bytes read but not yet split into complete lines.
    pending: Vec<u8>,
    /// Currently-building event.
    cur: SseEvent,
    /// Events that have completed parsing and are ready to emit.
    ready: VecDeque<SseEvent>,
    /// True once we have parsed any field for the current event. Tells
    /// `take_ready` whether to emit the in-progress event on stream end.
    cur_dirty: bool,
}

impl SseParser {
    /// Construct an empty parser.
    pub(crate) fn new() -> Self {
        Self::default()
    }

    /// Append a chunk of bytes from the underlying transport.
    pub(crate) fn feed(&mut self, chunk: &[u8]) {
        self.pending.extend_from_slice(chunk);
        // Drain every complete line we have available. A line ends in
        // \n, \r, or \r\n per the SSE spec.
        loop {
            let nl = self.pending.iter().position(|b| *b == b'\n' || *b == b'\r');
            let Some(idx) = nl else { break };
            let line: Vec<u8> = self.pending.drain(..idx).collect();
            // Drain the newline byte itself.
            let consumed = self.pending.remove(0);
            // Handle a CRLF as a single newline by also dropping the LF
            // when we just consumed a CR. The remove(0) above only
            // pulled one byte; check the next for the paired half.
            if consumed == b'\r' && self.pending.first() == Some(&b'\n') {
                self.pending.remove(0);
            }
            self.handle_line(&line);
        }
    }

    /// Pop every event the parser has completed since the last call.
    pub(crate) fn take_ready(&mut self) -> Vec<SseEvent> {
        self.ready.drain(..).collect()
    }

    /// Process a single decoded line.
    ///
    /// Empty line dispatches the in-progress event; otherwise the line
    /// is split into `field:value` (ignoring lines that start with `:`,
    /// which the spec treats as comments).
    fn handle_line(&mut self, line: &[u8]) {
        if line.is_empty() {
            if self.cur_dirty {
                self.finish_event();
            }
            return;
        }
        if line[0] == b':' {
            // Comment line per the SSE spec. Used as a keepalive.
            return;
        }
        let (field, value) = split_field(line);
        let value = strip_one_leading_space(value);
        let value_str = String::from_utf8_lossy(value);
        match field {
            b"event" => {
                self.cur.event = value_str.into_owned();
                self.cur_dirty = true;
            }
            b"data" => {
                if !self.cur.data.is_empty() {
                    self.cur.data.push('\n');
                }
                self.cur.data.push_str(&value_str);
                self.cur_dirty = true;
            }
            // The SSE spec says null bytes in `id` are ignored. UTF-8
            // strings cannot represent a null byte without explicit
            // handling, so the guard rejects raw `\0` before we copy.
            b"id" if !value.contains(&0) => {
                self.cur.id = Some(value_str.into_owned());
                self.cur_dirty = true;
            }
            b"retry" => {
                if let Ok(ms) = value_str.parse::<u64>() {
                    self.cur.retry = Some(ms);
                    self.cur_dirty = true;
                }
            }
            _ => {
                // Unknown field. The spec says ignore.
            }
        }
    }

    /// Finalize the in-progress event and stage it for the next
    /// `take_ready` call.
    fn finish_event(&mut self) {
        let mut event = std::mem::take(&mut self.cur);
        if event.event.is_empty() {
            event.event = "message".into();
        }
        self.ready.push_back(event);
        self.cur_dirty = false;
    }
}

/// Split a `field:value` line. Returns the whole line as the field with
/// an empty value when the line has no colon (per the SSE spec).
fn split_field(line: &[u8]) -> (&[u8], &[u8]) {
    match line.iter().position(|b| *b == b':') {
        Some(idx) => (&line[..idx], &line[idx + 1..]),
        None => (line, &[]),
    }
}

/// Strip exactly one leading space, per the SSE spec.
fn strip_one_leading_space(value: &[u8]) -> &[u8] {
    if value.first() == Some(&b' ') {
        &value[1..]
    } else {
        value
    }
}

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

    fn drain(p: &mut SseParser) -> Vec<SseEvent> {
        p.take_ready()
    }

    #[test]
    fn parses_a_single_event() {
        let mut p = SseParser::new();
        p.feed(b"event: message\ndata: hello\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].event, "message");
        assert_eq!(evs[0].data, "hello");
    }

    #[test]
    fn concatenates_multiple_data_lines_with_newlines() {
        let mut p = SseParser::new();
        p.feed(b"data: line one\ndata: line two\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs[0].data, "line one\nline two");
    }

    #[test]
    fn defaults_event_field_to_message_when_omitted() {
        let mut p = SseParser::new();
        p.feed(b"data: x\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs[0].event, "message");
    }

    #[test]
    fn captures_id_and_retry_fields() {
        let mut p = SseParser::new();
        p.feed(b"id: 42\nretry: 1500\ndata: x\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs[0].id.as_deref(), Some("42"));
        assert_eq!(evs[0].retry, Some(1500));
    }

    #[test]
    fn ignores_comment_lines() {
        let mut p = SseParser::new();
        p.feed(b": this is a keepalive\ndata: x\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].data, "x");
    }

    #[test]
    fn handles_chunk_boundaries_inside_a_line() {
        let mut p = SseParser::new();
        p.feed(b"data: he");
        p.feed(b"llo\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs[0].data, "hello");
    }

    #[test]
    fn handles_crlf_line_endings() {
        let mut p = SseParser::new();
        p.feed(b"data: x\r\n\r\n");
        let evs = drain(&mut p);
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].data, "x");
    }

    #[test]
    fn back_to_back_events_are_each_emitted() {
        let mut p = SseParser::new();
        p.feed(b"data: one\n\ndata: two\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs.len(), 2);
        assert_eq!(evs[0].data, "one");
        assert_eq!(evs[1].data, "two");
    }

    #[test]
    fn line_without_colon_is_treated_as_field_with_empty_value() {
        let mut p = SseParser::new();
        // `data` with no value still dirties the event so the blank
        // line dispatches it; the data string remains empty.
        p.feed(b"data\n\n");
        let evs = drain(&mut p);
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].data, "");
    }
}
