//! A minimal, dependency-light `text/event-stream` (SSE) parser.
//!
//! It consumes raw byte chunks from a streaming HTTP body and yields one
//! [`SseFrame`] per complete event (frames are separated by a blank line).
//! Only the subset of the spec the Imaige job-event stream uses is handled:
//! `event:`, `data:` (which may span multiple lines), and `id:`. Comment
//! lines (`:`-prefixed) and unknown fields are ignored.

/// One fully-assembled SSE event.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SseFrame {
    /// The `event:` field, if present (defaults to `message` per the spec,
    /// but we keep it `None` when absent so callers can decide).
    pub event: Option<String>,
    /// The concatenated `data:` lines, joined by `\n`.
    pub data: String,
    /// The `id:` field, if present.
    pub id: Option<String>,
}

impl SseFrame {
    fn is_empty(&self) -> bool {
        self.event.is_none() && self.data.is_empty() && self.id.is_none()
    }
}

/// Incremental SSE decoder. Feed it byte chunks; it returns completed frames.
#[derive(Debug, Default)]
pub struct SseDecoder {
    /// Bytes received but not yet split into a complete line.
    buf: Vec<u8>,
    /// The frame currently being assembled across lines.
    current: SseFrame,
}

impl SseDecoder {
    /// Create a fresh decoder.
    pub fn new() -> Self {
        Self::default()
    }

    /// Push a chunk of bytes; returns every event completed by this chunk.
    pub fn push(&mut self, chunk: &[u8]) -> Vec<SseFrame> {
        self.buf.extend_from_slice(chunk);
        let mut frames = Vec::new();

        // Process complete lines (terminated by `\n`; tolerate a trailing
        // `\r`). We retain any partial trailing line in `self.buf`.
        while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
            // Split off the line including the newline; keep the rest.
            let line_bytes: Vec<u8> = self.buf.drain(..=nl).collect();
            // Strip the trailing `\n` and an optional `\r`.
            let mut end = line_bytes.len() - 1;
            if end > 0 && line_bytes[end - 1] == b'\r' {
                end -= 1;
            }
            let line = String::from_utf8_lossy(&line_bytes[..end]).into_owned();
            if let Some(frame) = self.consume_line(&line) {
                frames.push(frame);
            }
        }
        frames
    }

    /// Flush any frame assembled from a final, unterminated line. Call once
    /// the underlying body has ended.
    pub fn finish(&mut self) -> Option<SseFrame> {
        // Treat any leftover bytes as a final line without a newline.
        if !self.buf.is_empty() {
            let line = String::from_utf8_lossy(&self.buf).into_owned();
            self.buf.clear();
            if let Some(frame) = self.consume_line(&line) {
                return Some(frame);
            }
        }
        if !self.current.is_empty() {
            return Some(std::mem::take(&mut self.current));
        }
        None
    }

    /// Apply one line to the in-progress frame. A blank line dispatches the
    /// accumulated frame.
    fn consume_line(&mut self, line: &str) -> Option<SseFrame> {
        if line.is_empty() {
            // End of event: dispatch if non-empty.
            if self.current.is_empty() {
                return None;
            }
            return Some(std::mem::take(&mut self.current));
        }
        // Comment line.
        if line.starts_with(':') {
            return None;
        }
        let (field, value) = match line.split_once(':') {
            Some((f, v)) => (f, v.strip_prefix(' ').unwrap_or(v)),
            // A line with no colon is a field name with an empty value.
            None => (line, ""),
        };
        match field {
            "event" => self.current.event = Some(value.to_string()),
            "id" => self.current.id = Some(value.to_string()),
            "data" => {
                if !self.current.data.is_empty() {
                    self.current.data.push('\n');
                }
                self.current.data.push_str(value);
            }
            _ => {} // ignore unknown fields (retry:, etc.)
        }
        None
    }
}

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

    #[test]
    fn parses_single_frame() {
        let mut d = SseDecoder::new();
        let frames = d.push(b"event: progress\ndata: {\"p\":0.5}\n\n");
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0].event.as_deref(), Some("progress"));
        assert_eq!(frames[0].data, "{\"p\":0.5}");
    }

    #[test]
    fn joins_multiline_data() {
        let mut d = SseDecoder::new();
        let frames = d.push(b"data: line1\ndata: line2\n\n");
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0].data, "line1\nline2");
    }

    #[test]
    fn handles_split_across_chunks() {
        let mut d = SseDecoder::new();
        assert!(d.push(b"event: done\nda").is_empty());
        assert!(d.push(b"ta: hi\n").is_empty());
        let frames = d.push(b"\n");
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0].event.as_deref(), Some("done"));
        assert_eq!(frames[0].data, "hi");
    }

    #[test]
    fn tolerates_crlf_and_comments() {
        let mut d = SseDecoder::new();
        let frames = d.push(b": keepalive\r\ndata: x\r\n\r\n");
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0].data, "x");
    }

    #[test]
    fn finish_flushes_unterminated() {
        let mut d = SseDecoder::new();
        assert!(d.push(b"data: tail").is_empty());
        let f = d.finish().expect("frame");
        assert_eq!(f.data, "tail");
    }
}
