void_core/pipeline/

events.rs

//! Pipeline event system for progress reporting and observability.

use std::sync::Arc;

use crate::support::events::{
    FetchSource as UnifiedFetchSource, LegacyPipelineAdapter,
    PipelineEvent as UnifiedPipelineEvent, VoidEvent,
};

/// Events emitted during pipeline operations.
#[derive(Debug, Clone)]
pub enum PipelineEvent {
    /// A file was discovered during workspace traversal.
    FileDiscovered { path: String, size: u64 },
    /// A file was processed (compressed, encrypted, or written).
    FileProcessed { path: String, shard_id: u64 },
    /// A shard was created and stored.
    ShardCreated {
        cid: String,
        size: u64,
        file_count: usize,
    },
    /// A shard was fetched from storage.
    ShardFetched { cid: String, source: FetchSource },
    /// Progress update for long-running operations.
    Progress {
        stage: String,
        current: u64,
        total: u64,
    },
    /// Non-fatal warning.
    Warning { message: String },
    /// Error occurred (may or may not be recoverable).
    Error { message: String, recoverable: bool },
}

/// Source of a fetched shard.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FetchSource {
    /// Fetched from local object store.
    Local,
    /// Fetched from IPFS.
    Ipfs,
}

/// Observer trait for receiving pipeline events.
pub trait PipelineObserver: Send + Sync {
    /// Called when an event occurs.
    fn on_event(&self, event: &PipelineEvent);
}

/// No-op observer that discards all events.
/// Used for backwards compatibility with non-event-aware code.
pub struct NullObserver;

impl PipelineObserver for NullObserver {
    fn on_event(&self, _event: &PipelineEvent) {
        // Intentionally empty
    }
}

/// Observer that forwards events to multiple observers.
pub struct MultiObserver {
    observers: Vec<Arc<dyn PipelineObserver>>,
}

impl MultiObserver {
    /// Create a new multi-observer from a list of observers.
    pub fn new(observers: Vec<Arc<dyn PipelineObserver>>) -> Self {
        Self { observers }
    }
}

impl PipelineObserver for MultiObserver {
    fn on_event(&self, event: &PipelineEvent) {
        for observer in &self.observers {
            observer.on_event(event);
        }
    }
}

/// Observer that collects events into a vector.
/// Useful for testing.
#[cfg(test)]
pub struct CollectingObserver {
    events: std::sync::Mutex<Vec<PipelineEvent>>,
}

#[cfg(test)]
impl CollectingObserver {
    pub fn new() -> Self {
        Self {
            events: std::sync::Mutex::new(Vec::new()),
        }
    }

    pub fn events(&self) -> Vec<PipelineEvent> {
        self.events.lock().unwrap_or_else(|e| e.into_inner()).clone()
    }
}

#[cfg(test)]
impl PipelineObserver for CollectingObserver {
    fn on_event(&self, event: &PipelineEvent) {
        self.events.lock().unwrap_or_else(|e| e.into_inner()).push(event.clone());
    }
}

// ============================================================================
// Legacy Adapter Implementation
// ============================================================================

/// Convert legacy FetchSource to unified FetchSource.
fn convert_fetch_source(source: FetchSource) -> UnifiedFetchSource {
    match source {
        FetchSource::Local => UnifiedFetchSource::Local,
        FetchSource::Ipfs => UnifiedFetchSource::Ipfs,
    }
}

/// Convert legacy PipelineEvent to unified PipelineEvent.
fn convert_pipeline_event(event: &PipelineEvent) -> UnifiedPipelineEvent {
    match event {
        PipelineEvent::FileDiscovered { path, size } => UnifiedPipelineEvent::FileDiscovered {
            path: path.clone(),
            size: *size,
        },
        PipelineEvent::FileProcessed { path, shard_id } => UnifiedPipelineEvent::FileProcessed {
            path: path.clone(),
            shard_id: *shard_id,
        },
        PipelineEvent::ShardCreated {
            cid,
            size,
            file_count,
        } => UnifiedPipelineEvent::ShardCreated {
            cid: cid.clone(),
            size: *size,
            file_count: *file_count,
        },
        PipelineEvent::ShardFetched { cid, source } => UnifiedPipelineEvent::ShardFetched {
            cid: cid.clone(),
            source: convert_fetch_source(*source),
        },
        PipelineEvent::Progress {
            stage,
            current,
            total,
        } => UnifiedPipelineEvent::Progress {
            stage: stage.clone(),
            current: *current,
            total: *total,
        },
        PipelineEvent::Warning { message } => UnifiedPipelineEvent::Warning {
            message: message.clone(),
        },
        PipelineEvent::Error {
            message,
            recoverable,
        } => UnifiedPipelineEvent::Error {
            message: message.clone(),
            recoverable: *recoverable,
        },
    }
}

impl PipelineObserver for LegacyPipelineAdapter {
    fn on_event(&self, event: &PipelineEvent) {
        let unified_event = convert_pipeline_event(event);
        self.inner().on_event(&VoidEvent::Pipeline(unified_event));
    }
}

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

    #[test]
    fn null_observer_accepts_events() {
        let observer = NullObserver;
        observer.on_event(&PipelineEvent::Progress {
            stage: "test".into(),
            current: 1,
            total: 10,
        });
        // Should not panic
    }

    #[test]
    fn multi_observer_forwards_to_all() {
        let collector1 = Arc::new(CollectingObserver::new());
        let collector2 = Arc::new(CollectingObserver::new());

        let multi = MultiObserver::new(vec![
            collector1.clone() as Arc<dyn PipelineObserver>,
            collector2.clone() as Arc<dyn PipelineObserver>,
        ]);

        multi.on_event(&PipelineEvent::FileDiscovered {
            path: "test.rs".into(),
            size: 100,
        });

        assert_eq!(collector1.events().len(), 1);
        assert_eq!(collector2.events().len(), 1);
    }

    #[test]
    fn collecting_observer_stores_events() {
        let collector = CollectingObserver::new();

        collector.on_event(&PipelineEvent::Progress {
            stage: "seal".into(),
            current: 5,
            total: 10,
        });
        collector.on_event(&PipelineEvent::ShardCreated {
            cid: "bafk...".into(),
            size: 1024,
            file_count: 3,
        });

        let events = collector.events();
        assert_eq!(events.len(), 2);
    }

    #[test]
    fn legacy_adapter_implements_pipeline_observer() {
        use crate::support::events::{CollectingObserver as UnifiedCollector, VoidObserver};

        // Create a unified VoidObserver to capture events
        let unified_collector = Arc::new(UnifiedCollector::new());

        // Wrap it in LegacyPipelineAdapter
        let adapter =
            LegacyPipelineAdapter::new(unified_collector.clone() as Arc<dyn VoidObserver>);

        // Use it as a PipelineObserver (the impl we just added)
        let observer: &dyn PipelineObserver = &adapter;

        // Send a legacy PipelineEvent through the trait
        observer.on_event(&PipelineEvent::FileDiscovered {
            path: "test.rs".into(),
            size: 1024,
        });
        observer.on_event(&PipelineEvent::ShardCreated {
            cid: "bafk...".into(),
            size: 2048,
            file_count: 5,
        });

        // Verify events were received by the unified collector
        let events = unified_collector.events();
        assert_eq!(events.len(), 2);

        // Verify first event converted correctly
        match &events[0] {
            VoidEvent::Pipeline(UnifiedPipelineEvent::FileDiscovered { path, size }) => {
                assert_eq!(path, "test.rs");
                assert_eq!(*size, 1024);
            }
            _ => panic!("Expected Pipeline(FileDiscovered), got {:?}", events[0]),
        }

        // Verify second event converted correctly
        match &events[1] {
            VoidEvent::Pipeline(UnifiedPipelineEvent::ShardCreated {
                cid,
                size,
                file_count,
            }) => {
                assert_eq!(cid, "bafk...");
                assert_eq!(*size, 2048);
                assert_eq!(*file_count, 5);
            }
            _ => panic!("Expected Pipeline(ShardCreated), got {:?}", events[1]),
        }
    }
}