ito_core/audit/
stream.rs

1//! Poll-based audit event streaming with multi-worktree support.
2//!
3//! Provides a simple polling mechanism that checks routed audit storage for
4//! new events at a configurable interval. Supports monitoring events across
5//! multiple git worktrees without assuming a tracked worktree JSONL file.
6
7use std::collections::HashSet;
8use std::path::Path;
9use std::time::Duration;
10
11use ito_domain::audit::event::AuditEvent;
12
13use super::store::{AuditEventStore, audit_storage_location_key, default_audit_store};
14use super::worktree::discover_worktrees;
15
16/// Configuration for the event stream.
17#[derive(Debug, Clone)]
18pub struct StreamConfig {
19    /// Poll interval (default: 500ms).
20    pub poll_interval: Duration,
21    /// Monitor all worktrees, not just the current one.
22    pub all_worktrees: bool,
23    /// Number of initial events to emit on startup.
24    pub last: usize,
25}
26
27impl Default for StreamConfig {
28    fn default() -> Self {
29        Self {
30            poll_interval: Duration::from_millis(500),
31            all_worktrees: false,
32            last: 10,
33        }
34    }
35}
36
37/// A single stream source: either the main project or a worktree.
38pub struct StreamSource {
39    /// Label for this source (e.g., "main" or worktree branch name).
40    pub label: String,
41    /// Routed audit store retained across polls.
42    store: Box<dyn AuditEventStore>,
43    /// Number of events previously seen.
44    ///
45    /// This offset assumes the store is append-only between polls. If the
46    /// underlying log is truncated or rotated, a shorter read is treated as a
47    /// reset and the next append-only growth resumes from the new length.
48    offset: usize,
49}
50
51/// A streamed event with its source label.
52#[derive(Debug)]
53pub struct StreamEvent {
54    /// The audit event.
55    pub event: AuditEvent,
56    /// Label of the source (e.g., "main" or branch name).
57    pub source: String,
58}
59
60/// Read the initial batch of events for streaming (the last N events).
61///
62/// Returns events from the main project log and, if `all_worktrees` is true,
63/// from all discovered worktrees.
64pub fn read_initial_events(
65    ito_path: &Path,
66    config: &StreamConfig,
67) -> (Vec<StreamEvent>, Vec<StreamSource>) {
68    let mut sources = Vec::new();
69    let mut events = Vec::new();
70    let mut seen_locations = HashSet::new();
71
72    // Main project source
73    let main_store = default_audit_store(ito_path);
74    let main_key = audit_storage_location_key(&main_store.location());
75    let main_events = main_store.read_all();
76    let start = main_events.len().saturating_sub(config.last);
77    for event in &main_events[start..] {
78        events.push(StreamEvent {
79            event: event.clone(),
80            source: "main".to_string(),
81        });
82    }
83    sources.push(StreamSource {
84        label: "main".to_string(),
85        store: main_store,
86        offset: main_events.len(),
87    });
88    seen_locations.insert(main_key);
89
90    // Worktree sources
91    if config.all_worktrees {
92        let worktrees = discover_worktrees(ito_path);
93        for wt in &worktrees {
94            if wt.is_main {
95                continue; // Already handled above
96            }
97            let wt_ito_path = wt.path.join(".ito");
98            if !wt_ito_path.exists() {
99                continue;
100            }
101
102            let wt_store = default_audit_store(&wt_ito_path);
103            let wt_key = audit_storage_location_key(&wt_store.location());
104            if !seen_locations.insert(wt_key) {
105                continue;
106            }
107
108            let wt_events = wt_store.read_all();
109            let label = wt
110                .branch
111                .clone()
112                .unwrap_or_else(|| wt.path.display().to_string());
113            let start = wt_events.len().saturating_sub(config.last);
114            for event in &wt_events[start..] {
115                events.push(StreamEvent {
116                    event: event.clone(),
117                    source: label.clone(),
118                });
119            }
120            sources.push(StreamSource {
121                label,
122                store: wt_store,
123                offset: wt_events.len(),
124            });
125        }
126    }
127
128    (events, sources)
129}
130
131/// Poll all sources for new events since the last check.
132///
133/// Updates the offsets in each source so subsequent polls only return new events.
134///
135/// Offsets are based on event counts, so a store that shrinks between polls is
136/// treated as having reset; the shorter snapshot advances no offset until new
137/// events extend the store again.
138pub fn poll_new_events(sources: &mut [StreamSource]) -> Vec<StreamEvent> {
139    let mut new_events = Vec::new();
140
141    for source in sources.iter_mut() {
142        let current_events = source.store.read_all();
143        if current_events.len() <= source.offset {
144            continue;
145        }
146
147        for event in &current_events[source.offset..] {
148            new_events.push(StreamEvent {
149                event: event.clone(),
150                source: source.label.clone(),
151            });
152        }
153
154        source.offset = current_events.len();
155    }
156
157    new_events
158}
159
160#[cfg(test)]
161#[path = "stream_tests.rs"]
162mod stream_tests;
ito_core/audit/stream.rs

ito_core/audit/
stream.rs
