sbom_tools/parsers/

streaming.rs

//! Streaming SBOM parser for large files.
//!
//! This module provides memory-efficient parsing for very large SBOMs by:
//! - Using streaming JSON/XML parsing (`serde_json::from_reader`)
//! - Not buffering the entire file into memory as a string
//! - Yielding results via an iterator interface
//! - Supporting progress callbacks
//!
//! Supports CycloneDX 1.4–1.7 (JSON/XML), SPDX 2.2–2.3 (JSON/tag-value/RDF),
//! and SPDX 3.0 (JSON-LD). Note: SPDX 3.0 requires full document loading due
//! to its graph-based element model with cross-references.
//!
//! # Usage
//!
//! ```no_run
//! use sbom_tools::parsers::streaming::{StreamingParser, StreamingConfig, ParseEvent};
//! use std::path::Path;
//!
//! let config = StreamingConfig::default()
//!     .with_chunk_size(64 * 1024)
//!     .with_progress_callback(|p| println!("Progress: {:.1}%", p.percent()));
//!
//! let parser = StreamingParser::new(config);
//! let stream = parser.parse_file(Path::new("large-sbom.json")).unwrap();
//!
//! for event in stream {
//!     match event {
//!         Ok(ParseEvent::Metadata(meta)) => println!("Document: {:?}", meta.document.format),
//!         Ok(ParseEvent::Component(comp)) => println!("Component: {}", comp.name),
//!         Ok(ParseEvent::Dependency(edge)) => println!("Dependency: {} -> {}", edge.from, edge.to),
//!         Ok(ParseEvent::Complete) => println!("Done!"),
//!         Err(e) => eprintln!("Error: {}", e),
//!     }
//! }
//! ```

use super::detection::FormatDetector;
use super::traits::ParseError;
use crate::model::{
    CanonicalId, Component, DependencyEdge, DocumentMetadata, FormatExtensions, NormalizedSbom,
};
use std::collections::VecDeque;
use std::io::{BufRead, BufReader, Read};
use std::path::Path;
use std::sync::Arc;

/// Progress information for streaming parsing
#[derive(Debug, Clone)]
pub struct ParseProgress {
    /// Bytes read so far
    pub bytes_read: u64,
    /// Total bytes (if known)
    pub total_bytes: Option<u64>,
    /// Components parsed so far
    pub components_parsed: usize,
    /// Dependencies parsed so far
    pub dependencies_parsed: usize,
}

impl ParseProgress {
    /// Get progress percentage (0-100), or None if total is unknown
    #[must_use]
    pub fn percent(&self) -> f32 {
        match self.total_bytes {
            Some(total) if total > 0 => (self.bytes_read as f32 / total as f32) * 100.0,
            _ => 0.0,
        }
    }

    /// Check if progress is complete
    #[must_use]
    pub fn is_complete(&self) -> bool {
        self.total_bytes
            .is_some_and(|total| self.bytes_read >= total)
    }
}

/// Progress callback type
pub type ProgressCallback = Arc<dyn Fn(&ParseProgress) + Send + Sync>;

/// Configuration for streaming parser
#[derive(Clone)]
pub struct StreamingConfig {
    /// Chunk size for reading (default: 64KB)
    pub chunk_size: usize,
    /// Buffer size for components (default: 1000)
    pub component_buffer_size: usize,
    /// Progress callback (optional)
    progress_callback: Option<ProgressCallback>,
    /// Whether to validate components during parsing
    pub validate_during_parse: bool,
    /// Skip malformed components instead of erroring
    pub skip_malformed: bool,
}

impl Default for StreamingConfig {
    fn default() -> Self {
        Self {
            chunk_size: 64 * 1024, // 64KB
            component_buffer_size: 1000,
            progress_callback: None,
            validate_during_parse: true,
            skip_malformed: false,
        }
    }
}

impl StreamingConfig {
    /// Set chunk size for reading
    #[must_use]
    pub fn with_chunk_size(mut self, size: usize) -> Self {
        self.chunk_size = size.max(1024); // Minimum 1KB
        self
    }

    /// Set component buffer size
    #[must_use]
    pub fn with_buffer_size(mut self, size: usize) -> Self {
        self.component_buffer_size = size.max(10);
        self
    }

    /// Set progress callback
    #[must_use]
    pub fn with_progress_callback<F>(mut self, callback: F) -> Self
    where
        F: Fn(&ParseProgress) + Send + Sync + 'static,
    {
        self.progress_callback = Some(Arc::new(callback));
        self
    }

    /// Enable/disable validation during parsing
    #[must_use]
    pub const fn with_validation(mut self, validate: bool) -> Self {
        self.validate_during_parse = validate;
        self
    }

    /// Enable/disable skipping malformed components
    #[must_use]
    pub const fn with_skip_malformed(mut self, skip: bool) -> Self {
        self.skip_malformed = skip;
        self
    }
}

impl std::fmt::Debug for StreamingConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamingConfig")
            .field("chunk_size", &self.chunk_size)
            .field("component_buffer_size", &self.component_buffer_size)
            .field("has_progress_callback", &self.progress_callback.is_some())
            .field("validate_during_parse", &self.validate_during_parse)
            .field("skip_malformed", &self.skip_malformed)
            .finish()
    }
}

/// Events emitted during streaming parsing
#[derive(Debug, Clone)]
pub enum ParseEvent {
    /// Document metadata has been parsed.
    ///
    /// Carries `primary_component_id`/`extensions` alongside the document so
    /// a consumer reconstructing a full `NormalizedSbom` from the event
    /// stream (see [`StreamingIterator::collect_sbom`]) does not silently
    /// lose them — both live on `NormalizedSbom` itself, not on
    /// `DocumentMetadata`, and previously had no event to travel in at all.
    /// Boxed to keep this variant from dominating `ParseEvent`'s size.
    Metadata(Box<ParsedMetadata>),
    /// A component has been parsed
    Component(Box<Component>),
    /// A dependency relationship has been parsed
    Dependency(DependencyEdge),
    /// Parsing is complete
    Complete,
}

/// Payload for [`ParseEvent::Metadata`].
#[derive(Debug, Clone)]
pub struct ParsedMetadata {
    pub document: DocumentMetadata,
    pub primary_component_id: Option<CanonicalId>,
    pub extensions: FormatExtensions,
}

/// Streaming parser for large SBOMs
#[derive(Debug)]
pub struct StreamingParser {
    config: StreamingConfig,
}

impl StreamingParser {
    /// Create a new streaming parser with the given configuration
    #[must_use]
    pub const fn new(config: StreamingConfig) -> Self {
        Self { config }
    }

    /// Create a streaming parser with default configuration
    #[must_use]
    pub fn default_config() -> Self {
        Self::new(StreamingConfig::default())
    }

    /// Parse a file and return an iterator of events
    pub fn parse_file(&self, path: &Path) -> Result<StreamingIterator, ParseError> {
        let file = std::fs::File::open(path)
            .map_err(|e| ParseError::IoError(format!("Failed to open file: {e}")))?;

        let total_bytes = file.metadata().map(|m| m.len()).ok();
        let reader = BufReader::with_capacity(self.config.chunk_size, file);

        self.parse_reader(reader, total_bytes)
    }

    /// Parse from a reader and return an iterator of events
    pub fn parse_reader<R: Read + Send + 'static>(
        &self,
        reader: BufReader<R>,
        total_bytes: Option<u64>,
    ) -> Result<StreamingIterator, ParseError> {
        Ok(StreamingIterator::new(
            reader,
            total_bytes,
            self.config.clone(),
        ))
    }

    /// Parse from string content
    pub fn parse_str(&self, content: &str) -> Result<StreamingIterator, ParseError> {
        let cursor = std::io::Cursor::new(content.to_string());
        let total_bytes = Some(content.len() as u64);
        let reader = BufReader::new(cursor);
        self.parse_reader(reader, total_bytes)
    }

    /// Collect all events into a `NormalizedSbom` (for convenience)
    ///
    /// Note: This loads the entire SBOM into memory, negating the
    /// streaming benefits. Use the iterator directly for large files.
    pub fn parse_to_sbom(&self, path: &Path) -> Result<NormalizedSbom, ParseError> {
        let mut stream = self.parse_file(path)?;
        stream.collect_sbom()
    }
}

impl Default for StreamingParser {
    fn default() -> Self {
        Self::default_config()
    }
}

/// Iterator over streaming parse events
#[allow(dead_code)]
pub struct StreamingIterator {
    /// Internal state
    state: StreamingState,
    /// Configuration
    config: StreamingConfig,
    /// Progress tracking
    progress: ParseProgress,
    /// Pending events
    pending: VecDeque<ParseEvent>,
    /// Whether parsing is complete
    complete: bool,
}

enum StreamingState {
    /// Initial state - need to detect format and parse
    Initial(Box<dyn BufRead + Send>),
    /// Parsing complete, emitting events from parsed SBOM
    Emitting {
        sbom: Box<NormalizedSbom>,
        component_index: usize,
        dependency_index: usize,
        metadata_emitted: bool,
    },
    /// Finished
    Done,
}

impl StreamingIterator {
    fn new<R: Read + Send + 'static>(
        reader: BufReader<R>,
        total_bytes: Option<u64>,
        config: StreamingConfig,
    ) -> Self {
        Self {
            state: StreamingState::Initial(Box::new(reader)),
            config,
            progress: ParseProgress {
                bytes_read: 0,
                total_bytes,
                components_parsed: 0,
                dependencies_parsed: 0,
            },
            pending: VecDeque::new(),
            complete: false,
        }
    }

    /// Collect all events into a `NormalizedSbom`
    pub fn collect_sbom(&mut self) -> Result<NormalizedSbom, ParseError> {
        let mut metadata: Option<DocumentMetadata> = None;
        let mut primary_component_id: Option<CanonicalId> = None;
        let mut extensions: Option<FormatExtensions> = None;
        let mut components = Vec::new();
        let mut edges = Vec::new();

        for event in self.by_ref() {
            match event {
                Ok(ParseEvent::Metadata(meta)) => {
                    metadata = Some(meta.document);
                    primary_component_id = meta.primary_component_id;
                    extensions = Some(meta.extensions);
                }
                Ok(ParseEvent::Component(comp)) => components.push(*comp),
                Ok(ParseEvent::Dependency(edge)) => edges.push(edge),
                Ok(ParseEvent::Complete) => break,
                Err(e) => return Err(e),
            }
        }

        let document = metadata.unwrap_or_default();
        let mut sbom = NormalizedSbom::new(document);
        sbom.primary_component_id = primary_component_id;
        sbom.extensions = extensions.unwrap_or_default();

        for comp in components {
            sbom.add_component(comp);
        }
        for edge in edges {
            sbom.add_edge(edge);
        }

        sbom.calculate_content_hash();
        Ok(sbom)
    }

    fn report_progress(&self) {
        if let Some(ref callback) = self.config.progress_callback {
            callback(&self.progress);
        }
    }

    fn advance(&mut self) -> Option<Result<ParseEvent, ParseError>> {
        // Return pending events first
        if let Some(event) = self.pending.pop_front() {
            return Some(Ok(event));
        }

        if self.complete {
            return None;
        }

        // Process based on state
        match std::mem::replace(&mut self.state, StreamingState::Done) {
            StreamingState::Initial(reader) => {
                // Use centralized FormatDetector for consistent detection
                let detector = FormatDetector::new();

                // Parse using the detector which handles format detection and parsing
                match detector.parse_reader(reader) {
                    Ok(sbom) => {
                        self.progress.bytes_read = self.progress.total_bytes.unwrap_or(0);
                        self.report_progress();
                        self.state = StreamingState::Emitting {
                            sbom: Box::new(sbom),
                            component_index: 0,
                            dependency_index: 0,
                            metadata_emitted: false,
                        };
                        self.advance()
                    }
                    Err(e) => Some(Err(e)),
                }
            }
            StreamingState::Emitting {
                sbom,
                component_index,
                dependency_index,
                metadata_emitted,
            } => {
                // Emit metadata first
                if !metadata_emitted {
                    let document = sbom.document.clone();
                    let primary_component_id = sbom.primary_component_id.clone();
                    let extensions = sbom.extensions.clone();
                    self.state = StreamingState::Emitting {
                        sbom,
                        component_index,
                        dependency_index,
                        metadata_emitted: true,
                    };
                    return Some(Ok(ParseEvent::Metadata(Box::new(ParsedMetadata {
                        document,
                        primary_component_id,
                        extensions,
                    }))));
                }

                // IndexMap allows O(1) positional access; cloning the whole
                // component map on every advance() (once per emitted event)
                // was O(n²) time and allocation churn for a streaming parser.
                let edges_len = sbom.edges.len();

                // Emit components
                if let Some(comp) = sbom
                    .components
                    .get_index(component_index)
                    .map(|(_, c)| c.clone())
                {
                    self.progress.components_parsed += 1;
                    if self.progress.components_parsed.is_multiple_of(100) {
                        self.report_progress();
                    }
                    self.state = StreamingState::Emitting {
                        sbom,
                        component_index: component_index + 1,
                        dependency_index,
                        metadata_emitted,
                    };
                    return Some(Ok(ParseEvent::Component(Box::new(comp))));
                }

                // Emit dependencies
                if dependency_index < edges_len {
                    let edge = sbom.edges[dependency_index].clone();
                    self.progress.dependencies_parsed += 1;
                    self.state = StreamingState::Emitting {
                        sbom,
                        component_index,
                        dependency_index: dependency_index + 1,
                        metadata_emitted,
                    };
                    return Some(Ok(ParseEvent::Dependency(edge)));
                }

                // All done
                self.complete = true;
                self.report_progress();
                self.state = StreamingState::Done;
                Some(Ok(ParseEvent::Complete))
            }
            StreamingState::Done => {
                self.complete = true;
                None
            }
        }
    }
}

impl Iterator for StreamingIterator {
    type Item = Result<ParseEvent, ParseError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.advance()
    }
}

/// Estimate the number of components in an SBOM file without full parsing
///
/// This performs a quick scan to estimate component count, useful for
/// progress reporting and memory allocation.
pub fn estimate_component_count(path: &Path) -> Result<ComponentEstimate, ParseError> {
    let file = std::fs::File::open(path)
        .map_err(|e| ParseError::IoError(format!("Failed to open file: {e}")))?;

    let file_size = file.metadata().map(|m| m.len()).unwrap_or(0);

    let mut count = 0;
    let mut bytes_sampled = 0usize;
    let sample_limit = 1024 * 1024; // Sample first 1MB

    // Cap the whole read at the sample limit (+1 to see we hit it) BEFORE
    // splitting into lines. Minified SBOMs are single-line JSON, so
    // `reader.lines()` on the raw file would allocate the ENTIRE file (e.g.
    // 10 GB) into one String before the sample check could fire.
    let reader = BufReader::new(file.take(sample_limit as u64 + 1));

    for line in reader.lines() {
        let line = line.map_err(|e| ParseError::IoError(e.to_string()))?;
        bytes_sampled += line.len();

        // Count component markers
        if line.contains("\"bom-ref\"") || line.contains("\"SPDXID\"") {
            count += 1;
        }

        if bytes_sampled > sample_limit {
            break;
        }
    }

    // Extrapolate if we only sampled part of the file
    let estimated = if bytes_sampled < file_size as usize && bytes_sampled > 0 {
        (count as f64 * (file_size as f64 / bytes_sampled as f64)) as usize
    } else {
        count
    };

    Ok(ComponentEstimate {
        estimated_count: estimated,
        sampled_count: count,
        file_size,
        bytes_sampled,
        is_extrapolated: bytes_sampled < file_size as usize,
    })
}

/// Estimate of component count
#[derive(Debug, Clone)]
pub struct ComponentEstimate {
    /// Estimated total component count
    pub estimated_count: usize,
    /// Components found in sampled region
    pub sampled_count: usize,
    /// Total file size in bytes
    pub file_size: u64,
    /// Bytes that were sampled
    pub bytes_sampled: usize,
    /// Whether the estimate was extrapolated
    pub is_extrapolated: bool,
}

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

    #[test]
    fn test_progress_percent() {
        let progress = ParseProgress {
            bytes_read: 50,
            total_bytes: Some(100),
            components_parsed: 5,
            dependencies_parsed: 3,
        };
        assert_eq!(progress.percent(), 50.0);
        assert!(!progress.is_complete());

        let complete = ParseProgress {
            bytes_read: 100,
            total_bytes: Some(100),
            components_parsed: 10,
            dependencies_parsed: 5,
        };
        assert_eq!(complete.percent(), 100.0);
        assert!(complete.is_complete());
    }

    #[test]
    fn test_streaming_config_builder() {
        let config = StreamingConfig::default()
            .with_chunk_size(128 * 1024)
            .with_buffer_size(500)
            .with_validation(false)
            .with_skip_malformed(true);

        assert_eq!(config.chunk_size, 128 * 1024);
        assert_eq!(config.component_buffer_size, 500);
        assert!(!config.validate_during_parse);
        assert!(config.skip_malformed);
    }

    #[test]
    fn test_streaming_parser_creation() {
        let parser = StreamingParser::default_config();
        assert_eq!(parser.config.chunk_size, 64 * 1024);
    }

    /// The streaming path must not lose `primary_component_id`/`extensions`
    /// relative to the non-streaming parse of the same input — both live on
    /// `NormalizedSbom`, not `DocumentMetadata`, and previously had no event
    /// to travel through at all, so `collect_sbom()` always produced `None`/
    /// default regardless of what the source document actually declared.
    #[test]
    fn collect_sbom_preserves_primary_component_and_extensions() {
        let cdx = r#"{
            "bomFormat": "CycloneDX", "specVersion": "1.5",
            "metadata": {
                "component": {
                    "type": "application", "bom-ref": "pkg:npm/app@1.0",
                    "name": "app", "version": "1.0"
                }
            },
            "components": [
                {"type": "library", "bom-ref": "pkg:npm/lib@1.0", "name": "lib", "version": "1.0"}
            ]
        }"#;

        let non_streaming = crate::parsers::parse_sbom_str(cdx).expect("non-streaming parse");
        let mut stream = StreamingParser::default_config()
            .parse_str(cdx)
            .expect("stream start");
        let streamed = stream.collect_sbom().expect("collect_sbom");

        assert!(
            non_streaming.primary_component_id.is_some(),
            "fixture must actually exercise a primary component"
        );
        assert_eq!(
            streamed.primary_component_id, non_streaming.primary_component_id,
            "streaming must preserve primary_component_id"
        );
        assert_eq!(streamed.component_count(), non_streaming.component_count());
    }

    /// The no-primary-component case must round-trip as `None`, not panic
    /// or silently become `Some(default)`.
    #[test]
    fn collect_sbom_preserves_absent_primary_component_as_none() {
        let cdx = r#"{
            "bomFormat": "CycloneDX", "specVersion": "1.5",
            "components": [
                {"type": "library", "bom-ref": "pkg:npm/lib@1.0", "name": "lib", "version": "1.0"}
            ]
        }"#;

        let non_streaming = crate::parsers::parse_sbom_str(cdx).expect("non-streaming parse");
        assert!(
            non_streaming.primary_component_id.is_none(),
            "fixture must have no metadata.component"
        );

        let mut stream = StreamingParser::default_config()
            .parse_str(cdx)
            .expect("stream start");
        let streamed = stream.collect_sbom().expect("collect_sbom");
        assert!(
            streamed.primary_component_id.is_none(),
            "absent primary component must round-trip as None, not Some(default)"
        );
    }
}