void_core/support/

events.rs

//! Unified event system for progress reporting and observability.
//!
//! This module provides a unified event system that wraps all event categories
//! used throughout void-core. It enables consistent progress reporting and
//! observability across different subsystems.
//!
//! # Event Categories
//!
//! - [`PipelineEvent`] - seal/unseal operations
//! - [`WorkspaceEvent`] - stage/checkout operations
//! - [`OpsEvent`] - fsck/merge operations
//! - [`P2PEvent`] - networking operations
//!
//! # Usage
//!
//! ```rust,ignore
//! use void_core::support::events::{VoidEvent, VoidObserver, emit};
//!
//! struct MyObserver;
//!
//! impl VoidObserver for MyObserver {
//!     fn on_event(&self, event: &VoidEvent) {
//!         println!("Event: {:?}", event);
//!     }
//! }
//! ```

use std::sync::Arc;

// ============================================================================
// Unified Event Type
// ============================================================================

/// Unified event type wrapping all event categories.
///
/// This enum provides a single type for all events emitted by void-core,
/// making it easy to implement observers that handle all event types.
#[derive(Debug, Clone)]
pub enum VoidEvent {
    /// Pipeline events (seal/unseal operations).
    Pipeline(PipelineEvent),
    /// Workspace events (stage/checkout operations).
    Workspace(WorkspaceEvent),
    /// Ops events (fsck/merge operations).
    Ops(OpsEvent),
    /// P2P events (networking operations).
    P2P(P2PEvent),
}

// ============================================================================
// Event Type Enums
// ============================================================================

/// Events emitted during pipeline operations (seal/unseal).
#[derive(Debug, Clone)]
pub enum PipelineEvent {
    /// A file was discovered during workspace traversal.
    FileDiscovered {
        /// Path to the discovered file.
        path: String,
        /// Size of the file in bytes.
        size: u64,
    },
    /// A file was processed (compressed, encrypted, or written).
    FileProcessed {
        /// Path to the processed file.
        path: String,
        /// ID of the shard containing the file.
        shard_id: u64,
    },
    /// A shard was created and stored.
    ShardCreated {
        /// CID of the created shard.
        cid: String,
        /// Size of the shard in bytes.
        size: u64,
        /// Number of files in the shard.
        file_count: usize,
    },
    /// A shard was fetched from storage.
    ShardFetched {
        /// CID of the fetched shard.
        cid: String,
        /// Source from which the shard was fetched.
        source: FetchSource,
    },
    /// Progress update for long-running operations.
    Progress {
        /// Current stage name.
        stage: String,
        /// Current progress value.
        current: u64,
        /// Total expected value.
        total: u64,
    },
    /// Non-fatal warning during operation.
    Warning {
        /// Warning message.
        message: String,
    },
    /// Error occurred (may or may not be recoverable).
    Error {
        /// Error message.
        message: String,
        /// Whether the error is recoverable.
        recoverable: bool,
    },
}

/// Events emitted during workspace operations (stage/checkout).
#[derive(Debug, Clone)]
pub enum WorkspaceEvent {
    /// A file was staged for commit.
    FileStaged {
        /// Path to the staged file.
        path: String,
    },
    /// A file was unstaged.
    FileUnstaged {
        /// Path to the unstaged file.
        path: String,
    },
    /// A file was checked out from a commit.
    FileCheckedOut {
        /// Path to the checked out file.
        path: String,
    },
    /// A file was skipped during checkout.
    FileSkipped {
        /// Path to the skipped file.
        path: String,
        /// Reason for skipping.
        reason: String,
    },
    /// Progress update for workspace operations.
    Progress {
        /// Current stage name.
        stage: String,
        /// Current progress value.
        current: u64,
        /// Total expected value.
        total: u64,
    },
}

/// Events emitted during ops operations (fsck/merge).
#[derive(Debug, Clone)]
pub enum OpsEvent {
    /// An object was checked during fsck.
    ObjectChecked {
        /// CID of the checked object.
        cid: String,
        /// Type of the object (commit, shard, etc.).
        object_type: String,
    },
    /// An issue was found during fsck.
    IssueFound {
        /// CID of the problematic object.
        cid: String,
        /// Description of the issue.
        message: String,
        /// Severity of the issue.
        severity: IssueSeverity,
    },
    /// Progress update for merge operations.
    MergeProgress {
        /// Current stage of the merge.
        stage: String,
        /// Number of conflicts resolved.
        conflicts_resolved: u64,
        /// Total number of conflicts.
        conflicts_total: u64,
    },
    /// Progress update for ops operations.
    Progress {
        /// Current stage name.
        stage: String,
        /// Current progress value.
        current: u64,
        /// Total expected value.
        total: u64,
    },
}

/// Events emitted during P2P networking operations.
#[derive(Debug, Clone)]
pub enum P2PEvent {
    /// A peer was discovered on the network.
    PeerDiscovered {
        /// Peer ID of the discovered peer.
        peer_id: String,
        /// Multiaddresses of the peer.
        addresses: Vec<String>,
    },
    /// Connected to a peer.
    PeerConnected {
        /// Peer ID of the connected peer.
        peer_id: String,
    },
    /// Disconnected from a peer.
    PeerDisconnected {
        /// Peer ID of the disconnected peer.
        peer_id: String,
    },
    /// A peer was identified with additional metadata.
    PeerIdentified {
        /// Peer ID of the identified peer.
        peer_id: String,
        /// Agent version string.
        agent_version: String,
    },
    /// Local node started listening on an address.
    Listening {
        /// Address the node is listening on.
        address: String,
    },
    /// A share was received from a peer.
    ShareReceived {
        /// Peer ID of the sender.
        from_peer: String,
        /// CID of the shared content.
        cid: String,
    },
    /// A share was sent to a peer.
    ShareSent {
        /// Peer ID of the recipient.
        to_peer: String,
        /// CID of the shared content.
        cid: String,
    },
    /// A share operation failed.
    ShareFailed {
        /// Peer ID of the intended recipient (if known).
        to_peer: Option<String>,
        /// Error message.
        error: String,
    },
    /// DHT bootstrap completed.
    BootstrapComplete,
    /// DHT record was found.
    DhtRecordFound {
        /// Key of the record.
        key: String,
    },
    /// DHT record was stored.
    DhtRecordStored {
        /// Key of the record.
        key: String,
    },
    /// NAT status changed.
    NatStatusChanged {
        /// New NAT status.
        status: String,
    },
    /// Hole punch succeeded.
    HolePunchSucceeded {
        /// Peer ID of the remote peer.
        peer_id: String,
    },
    /// An inbox head was ignored due to replay detection (seq already seen).
    InboxReplayIgnored {
        /// Peer ID of the sender.
        sender_peer_id: String,
        /// The sequence number that was replayed.
        seq: u64,
        /// The last good sequence number seen.
        last_good_seq: u64,
    },
    /// A fork was detected in the inbox chain (same seq, different prev).
    InboxForkDetected {
        /// Peer ID of the sender.
        sender_peer_id: String,
        /// The sequence number where fork was detected.
        seq: u64,
        /// The expected prev CID.
        expected_prev: Option<String>,
        /// The actual prev CID received.
        actual_prev: Option<String>,
    },
    /// A share send was attempted.
    ShareSendAttempt {
        /// Peer ID of the recipient.
        to_peer: String,
        /// Delivery method being attempted (direct or mailbox).
        method: String,
    },
    /// An envelope fetch was queued for retrieval.
    EnvelopeFetchQueued {
        /// CID of the envelope to fetch.
        cid: String,
        /// Peer ID of the sender.
        from_peer: String,
    },
    /// An envelope fetch failed.
    EnvelopeFetchFailed {
        /// CID of the envelope that failed to fetch.
        cid: String,
        /// Error message.
        error: String,
    },
}

// ============================================================================
// Supporting Types
// ============================================================================

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

/// Severity level for issues found during fsck.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IssueSeverity {
    /// Informational message, not an actual problem.
    Info,
    /// Warning that may indicate a problem.
    Warning,
    /// Error that indicates data corruption or loss.
    Error,
}

// ============================================================================
// Observer Trait
// ============================================================================

/// Observer trait for receiving void events.
///
/// Implement this trait to receive events from void-core operations.
/// The observer must be thread-safe (`Send + Sync`) as it may be called
/// from multiple threads during parallel operations.
pub trait VoidObserver: Send + Sync {
    /// Called when an event occurs.
    ///
    /// # Arguments
    ///
    /// * `event` - The event that occurred.
    fn on_event(&self, event: &VoidEvent);
}

// ============================================================================
// Built-in Observers
// ============================================================================

/// No-op observer that discards all events.
///
/// Used when event handling is not needed or for backwards compatibility
/// with code that doesn't use events.
#[derive(Debug, Clone, Copy, Default)]
pub struct NullObserver;

impl VoidObserver for NullObserver {
    fn on_event(&self, _event: &VoidEvent) {
        // Intentionally empty - discards all events
    }
}

/// Observer that forwards events to multiple child observers.
///
/// Useful for composing multiple observers, such as logging to both
/// a progress bar and a log file.
pub struct MultiObserver {
    observers: Vec<Arc<dyn VoidObserver>>,
}

impl MultiObserver {
    /// Create a new multi-observer from a list of child observers.
    ///
    /// # Arguments
    ///
    /// * `observers` - List of observers to forward events to.
    pub fn new(observers: Vec<Arc<dyn VoidObserver>>) -> Self {
        Self { observers }
    }

    /// Add an observer to the list.
    ///
    /// # Arguments
    ///
    /// * `observer` - Observer to add.
    pub fn add(&mut self, observer: Arc<dyn VoidObserver>) {
        self.observers.push(observer);
    }

    /// Returns the number of child observers.
    pub fn len(&self) -> usize {
        self.observers.len()
    }

    /// Returns true if there are no child observers.
    pub fn is_empty(&self) -> bool {
        self.observers.is_empty()
    }
}

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

impl std::fmt::Debug for MultiObserver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MultiObserver")
            .field("observer_count", &self.observers.len())
            .finish()
    }
}

/// Observer that collects events into a vector for testing.
///
/// This observer is only available in test builds and is useful for
/// verifying that the correct events are emitted during operations.
#[cfg(any(test, feature = "test-utils"))]
pub struct CollectingObserver {
    events: std::sync::Mutex<Vec<VoidEvent>>,
}

#[cfg(any(test, feature = "test-utils"))]
impl CollectingObserver {
    /// Create a new collecting observer.
    pub fn new() -> Self {
        Self {
            events: std::sync::Mutex::new(Vec::new()),
        }
    }

    /// Get a copy of all collected events.
    pub fn events(&self) -> Vec<VoidEvent> {
        self.events.lock().unwrap().clone()
    }

    /// Clear all collected events.
    pub fn clear(&self) {
        self.events.lock().unwrap().clear();
    }

    /// Returns the number of collected events.
    pub fn len(&self) -> usize {
        self.events.lock().unwrap().len()
    }

    /// Returns true if no events have been collected.
    pub fn is_empty(&self) -> bool {
        self.events.lock().unwrap().is_empty()
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl Default for CollectingObserver {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl VoidObserver for CollectingObserver {
    fn on_event(&self, event: &VoidEvent) {
        self.events.lock().unwrap().push(event.clone());
    }
}

#[cfg(any(test, feature = "test-utils"))]
impl std::fmt::Debug for CollectingObserver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CollectingObserver")
            .field("event_count", &self.len())
            .finish()
    }
}

// ============================================================================
// Helper Functions
// ============================================================================

/// Emit a void event to an optional observer.
///
/// This helper function handles the common pattern of emitting events
/// to an optional observer, doing nothing if the observer is None.
///
/// # Arguments
///
/// * `observer` - Optional observer to emit to.
/// * `event` - Event to emit.
///
/// # Example
///
/// ```rust,ignore
/// use void_core::support::events::{VoidEvent, VoidObserver, emit, PipelineEvent};
///
/// fn process_file(observer: &Option<Arc<dyn VoidObserver>>, path: &str) {
///     emit(observer, VoidEvent::Pipeline(PipelineEvent::FileDiscovered {
///         path: path.to_string(),
///         size: 1024,
///     }));
/// }
/// ```
pub fn emit(observer: &Option<Arc<dyn VoidObserver>>, event: VoidEvent) {
    if let Some(obs) = observer {
        obs.on_event(&event);
    }
}

/// Emit a workspace event to an optional observer.
///
/// Convenience function that wraps a WorkspaceEvent in a VoidEvent.
///
/// # Arguments
///
/// * `observer` - Optional observer to emit to.
/// * `event` - Workspace event to emit.
pub fn emit_workspace(observer: &Option<Arc<dyn VoidObserver>>, event: WorkspaceEvent) {
    emit(observer, VoidEvent::Workspace(event));
}

/// Emit an ops event to an optional observer.
///
/// Convenience function that wraps an OpsEvent in a VoidEvent.
///
/// # Arguments
///
/// * `observer` - Optional observer to emit to.
/// * `event` - Ops event to emit.
pub fn emit_ops(observer: &Option<Arc<dyn VoidObserver>>, event: OpsEvent) {
    emit(observer, VoidEvent::Ops(event));
}

/// Emit a pipeline event to an optional observer.
///
/// Convenience function that wraps a PipelineEvent in a VoidEvent.
///
/// # Arguments
///
/// * `observer` - Optional observer to emit to.
/// * `event` - Pipeline event to emit.
pub fn emit_pipeline(observer: &Option<Arc<dyn VoidObserver>>, event: PipelineEvent) {
    emit(observer, VoidEvent::Pipeline(event));
}

/// Emit a P2P event to an optional observer.
///
/// Convenience function that wraps a P2PEvent in a VoidEvent.
///
/// # Arguments
///
/// * `observer` - Optional observer to emit to.
/// * `event` - P2P event to emit.
pub fn emit_p2p(observer: &Option<Arc<dyn VoidObserver>>, event: P2PEvent) {
    emit(observer, VoidEvent::P2P(event));
}

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

/// Adapter for legacy PipelineObserver compatibility.
///
/// This struct wraps a VoidObserver and can be used where the legacy
/// PipelineObserver trait is expected. The actual PipelineObserver impl
/// is in pipeline/events.rs to avoid circular dependencies.
pub struct LegacyPipelineAdapter {
    observer: Arc<dyn VoidObserver>,
}

impl LegacyPipelineAdapter {
    /// Create a new legacy adapter wrapping a VoidObserver.
    ///
    /// # Arguments
    ///
    /// * `observer` - The VoidObserver to wrap.
    pub fn new(observer: Arc<dyn VoidObserver>) -> Self {
        Self { observer }
    }

    /// Get a reference to the inner observer.
    pub fn inner(&self) -> &Arc<dyn VoidObserver> {
        &self.observer
    }

    /// Emit a pipeline event through the wrapped observer.
    ///
    /// This method converts the legacy PipelineEvent to the unified
    /// event system and forwards it to the wrapped observer.
    pub fn emit(&self, event: PipelineEvent) {
        self.observer.on_event(&VoidEvent::Pipeline(event));
    }
}

impl std::fmt::Debug for LegacyPipelineAdapter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LegacyPipelineAdapter").finish()
    }
}