sif_parser/

streaming.rs

// SIF Streaming Protocol v1 — Stream lifecycle, CDC, multiplexing,
// flow control, heartbeat, status, schema evolution.
//
// Built from SIF-STREAMING.md specification.

use std::io::BufRead;
use std::collections::HashMap;

use crate::error::{err, ErrorKind, Result};
use crate::parse::parse_schema_str;
use crate::types::*;

// ── Streaming Types ─────────────────────────────────────────────────

/// Metadata for an active stream.
#[derive(Debug, Clone)]
pub struct StreamInfo {
    pub id: Option<String>,
    pub mode: StreamMode,
    pub schema: Option<Schema>,
    pub record_count: u64,
}

/// Stream mode: append-only (default) or CDC.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StreamMode {
    Append,
    Cdc,
}

/// Status level for `#status` directives.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatusLevel {
    Info,
    Warn,
    Error,
}

/// Events emitted by the streaming reader, extending the core Event
/// set with stream-specific events.
#[derive(Debug, Clone, PartialEq)]
pub enum StreamEvent {
    /// Core SIF event (header, schema, record, directive, block, etc.)
    Core(Event),

    /// `#stream start` with optional id and mode.
    StreamStart {
        id: Option<String>,
        mode: StreamMode,
    },
    /// `#stream end` with optional id and record count.
    StreamEnd {
        id: Option<String>,
        records: Option<u64>,
    },
    /// `#heartbeat` with optional timestamp.
    Heartbeat(Option<String>),
    /// `#status info|warn|error <text>`.
    Status {
        level: StatusLevel,
        message: String,
    },
    /// `#pause`
    Pause,
    /// `#resume`
    Resume,
    /// `#error <text>` (document-level error, non-fatal).
    Error(String),
}

// ── Stream Reader ───────────────────────────────────────────────────

/// A streaming SIF reader with full streaming protocol support.
///
/// Handles stream lifecycle, multiplexed streams, CDC mode,
/// schema evolution, flow control, and heartbeat.
pub struct StreamReader<R: BufRead> {
    source: R,
    line_buf: String,
    line_num: usize,
    header_read: bool,

    /// Active streams keyed by id (None key = unnamed stream).
    streams: HashMap<Option<String>, StreamInfo>,
    /// Currently active stream id.
    active_stream: Option<Option<String>>,
    /// Current schema (from active stream or standalone section).
    current_schema: Option<Schema>,

    /// Block/template state.
    in_block: bool,
    in_template: bool,
}

impl<R: BufRead> StreamReader<R> {
    pub fn new(source: R) -> Self {
        Self {
            source,
            line_buf: String::new(),
            line_num: 0,
            header_read: false,
            streams: HashMap::new(),
            active_stream: None,
            current_schema: None,
            in_block: false,
            in_template: false,
        }
    }

    /// Returns the currently active stream info, if any.
    pub fn active_stream(&self) -> Option<&StreamInfo> {
        self.active_stream
            .as_ref()
            .and_then(|id| self.streams.get(id))
    }

    /// Returns the current schema (from active stream or section).
    pub fn schema(&self) -> Option<&Schema> {
        // Prefer active stream's schema, fall back to section schema.
        if let Some(stream) = self.active_stream() {
            if stream.schema.is_some() {
                return stream.schema.as_ref();
            }
        }
        self.current_schema.as_ref()
    }

    /// Current line number.
    pub fn line_num(&self) -> usize {
        self.line_num
    }

    /// Read the next event.
    pub fn next_event(&mut self) -> Result<Option<StreamEvent>> {
        loop {
            self.line_buf.clear();
            let bytes_read = self
                .source
                .read_line(&mut self.line_buf)
                .map_err(|e| err(ErrorKind::UnexpectedEof, self.line_num, e.to_string()))?;

            if bytes_read == 0 {
                // EOF — treat as implicit #stream end for all open streams.
                return Ok(None);
            }

            self.line_num += 1;
            let line = self.line_buf.trim_end_matches('\n').trim_end_matches('\r');

            // BOM handling on first line
            let line = if self.line_num == 1 {
                line.strip_prefix('\u{FEFF}').unwrap_or(line)
            } else {
                line
            };

            // Inside block
            if self.in_block {
                if line.trim_end() == "#/block" {
                    self.in_block = false;
                    return Ok(Some(StreamEvent::Core(Event::BlockEnd)));
                }
                return Ok(Some(StreamEvent::Core(Event::BlockLine(line.to_string()))));
            }

            // Inside template
            if self.in_template {
                if line.trim_end() == "#/template" {
                    self.in_template = false;
                    return Ok(Some(StreamEvent::Core(Event::TemplateEnd)));
                }
                return Ok(Some(StreamEvent::Core(Event::TemplateLine(line.to_string()))));
            }

            // Skip empty lines
            if line.trim().is_empty() {
                continue;
            }

            // Header
            if !self.header_read {
                self.header_read = true;
                let header = crate::parse::parse_header_public(line, self.line_num)?;
                return Ok(Some(StreamEvent::Core(Event::Header(header))));
            }

            // Skip #! after header
            if line.starts_with("#!") {
                continue;
            }

            let trimmed = line.trim_end();

            // ── Stream-specific directives ──────────────────────

            // #stream start
            if trimmed.starts_with("#stream start") {
                let attrs = parse_stream_attrs(&trimmed[13..]);
                let id = attrs.get("id").cloned();
                let mode = match attrs.get("mode").map(|s| s.as_str()) {
                    Some("cdc") => StreamMode::Cdc,
                    _ => StreamMode::Append,
                };

                // If this stream already exists, this is a switch (reactivation).
                if !self.streams.contains_key(&id) {
                    self.streams.insert(
                        id.clone(),
                        StreamInfo {
                            id: id.clone(),
                            mode,
                            schema: None,
                            record_count: 0,
                        },
                    );
                }
                self.active_stream = Some(id.clone());

                return Ok(Some(StreamEvent::StreamStart { id, mode }));
            }

            // #stream end
            if trimmed.starts_with("#stream end") {
                let attrs = parse_stream_attrs(&trimmed[11..]);
                let id = attrs.get("id").cloned();
                let records = attrs
                    .get("records")
                    .and_then(|s| s.parse::<u64>().ok());

                // Remove the stream.
                self.streams.remove(&id);
                if self.active_stream.as_ref() == Some(&id) {
                    self.active_stream = None;
                }

                return Ok(Some(StreamEvent::StreamEnd { id, records }));
            }

            // #heartbeat
            if trimmed == "#heartbeat" || trimmed.starts_with("#heartbeat ") {
                let ts = if trimmed.len() > 11 {
                    Some(trimmed[11..].trim().to_string())
                } else {
                    None
                };
                return Ok(Some(StreamEvent::Heartbeat(ts)));
            }

            // #status
            if trimmed.starts_with("#status ") {
                let rest = &trimmed[8..];
                let (level, message) = if let Some(msg) = rest.strip_prefix("info ") {
                    (StatusLevel::Info, msg.to_string())
                } else if let Some(msg) = rest.strip_prefix("warn ") {
                    (StatusLevel::Warn, msg.to_string())
                } else if let Some(msg) = rest.strip_prefix("error ") {
                    (StatusLevel::Error, msg.to_string())
                } else {
                    (StatusLevel::Info, rest.to_string())
                };
                return Ok(Some(StreamEvent::Status { level, message }));
            }

            // #pause
            if trimmed == "#pause" {
                return Ok(Some(StreamEvent::Pause));
            }

            // #resume
            if trimmed == "#resume" {
                return Ok(Some(StreamEvent::Resume));
            }

            // #error
            if trimmed.starts_with("#error ") {
                return Ok(Some(StreamEvent::Error(trimmed[7..].to_string())));
            }

            // ── Core SIF directives ─────────────────────────────

            // Section break — schema persists in streams, resets outside
            if trimmed == "---" {
                if self.active_stream.is_none() {
                    self.current_schema = None;
                }
                return Ok(Some(StreamEvent::Core(Event::SectionBreak)));
            }

            // Section identifier
            if trimmed.starts_with('§') {
                let id = &trimmed['§'.len_utf8()..];
                return Ok(Some(StreamEvent::Core(Event::SectionId(
                    id.trim_end().to_string(),
                ))));
            }

            // Block start
            if trimmed.starts_with("#block ") {
                let rest = &trimmed[7..];
                let tokens: Vec<&str> = rest.split_whitespace().collect();
                let block_type = match tokens.first().copied() {
                    Some("code") => BlockType::Code,
                    Some("text") => BlockType::Text,
                    Some("diff") => BlockType::Diff,
                    Some("raw") => BlockType::Raw,
                    Some("template") => BlockType::Template,
                    _ => {
                        return Err(err(
                            ErrorKind::InvalidBlock,
                            self.line_num,
                            "unknown block type",
                        ));
                    }
                };
                let mut attrs = Vec::new();
                for &token in &tokens[1..] {
                    if let Some(eq) = token.find('=') {
                        attrs.push((
                            token[..eq].to_string(),
                            token[eq + 1..].to_string(),
                        ));
                    }
                }
                self.in_block = true;
                return Ok(Some(StreamEvent::Core(Event::BlockStart {
                    block_type,
                    attributes: attrs,
                })));
            }

            // Template start
            if trimmed.starts_with("#template ") {
                let name = trimmed[10..].trim().to_string();
                self.in_template = true;
                return Ok(Some(StreamEvent::Core(Event::TemplateStart(name))));
            }

            // Schema
            if trimmed.starts_with("#schema ") {
                let schema = parse_schema_str(&trimmed[8..], self.line_num)?;

                // Store in active stream if any, otherwise in section.
                if let Some(ref stream_id) = self.active_stream {
                    if let Some(stream) = self.streams.get_mut(stream_id) {
                        stream.schema = Some(schema.clone());
                    }
                } else {
                    self.current_schema = Some(schema.clone());
                }

                return Ok(Some(StreamEvent::Core(Event::Schema(schema))));
            }

            // #recall schema — no-op
            if trimmed == "#recall schema" {
                return Ok(Some(StreamEvent::Core(Event::Directive(
                    Directive::Recall,
                ))));
            }

            // Other directives
            if trimmed.starts_with('#') {
                if let Some(directive) =
                    crate::parse::parse_directive_public(trimmed, self.line_num)?
                {
                    return Ok(Some(StreamEvent::Core(Event::Directive(directive))));
                }
                continue;
            }

            // ── Record ──────────────────────────────────────────

            if let Some(schema) = self.schema().cloned() {
                // In stream mode, malformed records should be skipped
                // with a diagnostic, not cause a fatal error.
                match crate::parse::parse_record_public(trimmed, &schema, self.line_num) {
                    Ok(record) => {
                        // Increment record count for active stream.
                        if let Some(ref stream_id) = self.active_stream {
                            if let Some(stream) = self.streams.get_mut(stream_id) {
                                stream.record_count += 1;
                            }
                        }
                        return Ok(Some(StreamEvent::Core(Event::Record(record))));
                    }
                    Err(e) => {
                        // Stream mode: skip malformed record, emit error.
                        if self.active_stream.is_some() {
                            return Ok(Some(StreamEvent::Error(format!(
                                "Malformed record skipped at line {}: {}",
                                self.line_num, e.message
                            ))));
                        } else {
                            return Err(e);
                        }
                    }
                }
            } else {
                return Err(err(
                    ErrorKind::RecordWithoutSchema,
                    self.line_num,
                    "record found before any #schema",
                ));
            }
        }
    }

    /// Collect all events.
    pub fn collect_events(&mut self) -> Result<Vec<StreamEvent>> {
        let mut events = Vec::new();
        while let Some(event) = self.next_event()? {
            events.push(event);
        }
        Ok(events)
    }
}

/// Create a StreamReader from a string.
impl StreamReader<std::io::BufReader<std::io::Cursor<String>>> {
    pub fn from_str(input: &str) -> Self {
        let cursor = std::io::Cursor::new(input.to_string());
        Self::new(std::io::BufReader::new(cursor))
    }
}

// ── Stream Emitter ──────────────────────────────────────────────────

/// Extension methods for emitting streaming directives.
impl<W: std::io::Write> crate::emit::Emitter<W> {
    /// Write `#stream start` with optional attributes.
    pub fn emit_stream_start(
        &mut self,
        id: Option<&str>,
        mode: StreamMode,
    ) -> std::io::Result<()> {
        let w = self.writer_mut();
        write!(w, "#stream start")?;
        if let Some(id) = id {
            write!(w, " id={}", id)?;
        }
        if mode == StreamMode::Cdc {
            write!(w, " mode=cdc")?;
        }
        writeln!(w)
    }

    /// Write `#stream end` with optional attributes.
    pub fn emit_stream_end(
        &mut self,
        id: Option<&str>,
        records: Option<u64>,
    ) -> std::io::Result<()> {
        let w = self.writer_mut();
        write!(w, "#stream end")?;
        if let Some(id) = id {
            write!(w, " id={}", id)?;
        }
        if let Some(n) = records {
            write!(w, " records={}", n)?;
        }
        writeln!(w)
    }

    /// Write `#heartbeat` with optional timestamp.
    pub fn emit_heartbeat(&mut self, timestamp: Option<&str>) -> std::io::Result<()> {
        let w = self.writer_mut();
        if let Some(ts) = timestamp {
            writeln!(w, "#heartbeat {}", ts)
        } else {
            writeln!(w, "#heartbeat")
        }
    }

    /// Write `#status <level> <message>`.
    pub fn emit_status(&mut self, level: StatusLevel, message: &str) -> std::io::Result<()> {
        let w = self.writer_mut();
        let lvl = match level {
            StatusLevel::Info => "info",
            StatusLevel::Warn => "warn",
            StatusLevel::Error => "error",
        };
        writeln!(w, "#status {} {}", lvl, message)
    }

    /// Write `#pause`.
    pub fn emit_pause(&mut self) -> std::io::Result<()> {
        writeln!(self.writer_mut(), "#pause")
    }

    /// Write `#resume`.
    pub fn emit_resume(&mut self) -> std::io::Result<()> {
        writeln!(self.writer_mut(), "#resume")
    }

    /// Write `#error <message>`.
    pub fn emit_error(&mut self, message: &str) -> std::io::Result<()> {
        writeln!(self.writer_mut(), "#error {}", message)
    }
}

// ── Helpers ─────────────────────────────────────────────────────────

fn parse_stream_attrs(rest: &str) -> HashMap<String, String> {
    let mut attrs = HashMap::new();
    for token in rest.split_whitespace() {
        if let Some(eq) = token.find('=') {
            let key = &token[..eq];
            let val = &token[eq + 1..];
            attrs.insert(key.to_string(), val.to_string());
        }
    }
    attrs
}

// ── Tests ───────────────────────────────────────────────────────────

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

    #[test]
    fn test_stream_lifecycle() {
        let input = "\
#!sif v1
#stream start id=build
#schema timestamp:str message:str
2026-03-08T09:00:00Z\tCompiling
2026-03-08T09:00:01Z\tDone
#stream end id=build records=2
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let starts: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::StreamStart { .. }))
            .collect();
        assert_eq!(starts.len(), 1);

        let ends: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::StreamEnd { .. }))
            .collect();
        assert_eq!(ends.len(), 1);
        if let StreamEvent::StreamEnd { records, .. } = &ends[0] {
            assert_eq!(*records, Some(2));
        }

        let records: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Core(Event::Record(_))))
            .collect();
        assert_eq!(records.len(), 2);
    }

    #[test]
    fn test_cdc_stream() {
        let input = "\
#!sif v1
#stream start id=inventory mode=cdc
#schema id:uint:id product:str stock:uint
1\tWidget A\t100
2\tWidget B\t50
Δ1\tWidget A\t97
∅2
#stream end id=inventory records=4
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let records: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Core(Event::Record(r)) => Some(r),
                _ => None,
            })
            .collect();
        assert_eq!(records.len(), 4);
        assert_eq!(records[0].cdc_op, CdcOp::Insert);
        assert_eq!(records[2].cdc_op, CdcOp::Update);
        assert_eq!(records[3].cdc_op, CdcOp::Delete);
    }

    #[test]
    fn test_multiplexed_streams() {
        let input = "\
#!sif v1
#stream start id=build
#schema step:str status:str
#stream start id=tests
#schema name:str passed:bool
#stream start id=build
compile\tok
#stream start id=tests
api_auth\tT
#stream end id=build
#stream end id=tests
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let records: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Core(Event::Record(_))))
            .collect();
        assert_eq!(records.len(), 2);
    }

    #[test]
    fn test_flow_control() {
        let input = "\
#!sif v1
#stream start id=deploy
#schema step:str status:str
build\tcomplete
#pause
#resume
deploy\tin_progress
#stream end id=deploy
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        assert!(events.iter().any(|e| matches!(e, StreamEvent::Pause)));
        assert!(events.iter().any(|e| matches!(e, StreamEvent::Resume)));
    }

    #[test]
    fn test_heartbeat() {
        let input = "\
#!sif v1
#stream start
#schema x:uint
#heartbeat
#heartbeat 2026-03-08T09:00:30Z
1
#stream end
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let heartbeats: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Heartbeat(ts) => Some(ts.clone()),
                _ => None,
            })
            .collect();
        assert_eq!(heartbeats.len(), 2);
        assert_eq!(heartbeats[0], None);
        assert_eq!(
            heartbeats[1],
            Some("2026-03-08T09:00:30Z".to_string())
        );
    }

    #[test]
    fn test_status_levels() {
        let input = "\
#!sif v1
#stream start
#schema x:uint
#status info Stream initialized
#status warn Degraded
#status error Connection lost
#stream end
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let statuses: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Status { level, message } => Some((*level, message.clone())),
                _ => None,
            })
            .collect();
        assert_eq!(statuses.len(), 3);
        assert_eq!(statuses[0].0, StatusLevel::Info);
        assert_eq!(statuses[1].0, StatusLevel::Warn);
        assert_eq!(statuses[2].0, StatusLevel::Error);
    }

    #[test]
    fn test_schema_evolution() {
        let input = "\
#!sif v1
#stream start id=metrics
#schema timestamp:str cpu:float
2026-03-08T09:00:00Z\t42.5
#schema timestamp:str cpu:float gpu:float
2026-03-08T09:00:01Z\t43.0\t85.0
#stream end id=metrics
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        let schemas: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Core(Event::Schema(_))))
            .collect();
        assert_eq!(schemas.len(), 2);

        let records: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Core(Event::Record(r)) => Some(r),
                _ => None,
            })
            .collect();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].values.len(), 2);
        assert_eq!(records[1].values.len(), 3);
    }

    #[test]
    fn test_malformed_record_recovery() {
        let input = "\
#!sif v1
#stream start
#schema id:uint name:str
1\talice
bad record with wrong fields and not a uint
2\tbob
#stream end
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        // Should have 2 valid records and 1 error for the malformed one.
        let records: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Core(Event::Record(_))))
            .collect();
        let errors: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Error(_)))
            .collect();
        assert_eq!(records.len(), 2);
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn test_section_breaks_in_stream() {
        let input = "\
#!sif v1
#stream start
#schema x:uint
1
---
2
#stream end
";
        let mut reader = StreamReader::from_str(input);
        let events = reader.collect_events().unwrap();

        // Schema persists across section breaks inside a stream.
        let records: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Core(Event::Record(_))))
            .collect();
        assert_eq!(records.len(), 2);
    }
}