termivar_scanner/
event_bus.rs

1//! Event Bus - Publish/Subscribe Event System
2//!
3//! ## Runtime scope
4//!
5//! - **Build:** opt-in via `legacy-scanner`.
6//! - **Execution:** `ScanRunner` publishes `Event` values.
7//! - **Default `termivar scan`:** no.
8//! - **Support:** implemented.
9//!
10//! See `docs/internals/runtime-map.md`.
11//!
12//! Central event infrastructure for scanning lifecycle and plugin coordination.
13
14use dashmap::DashMap;
15use std::sync::Arc;
16pub use termivar_core::{Event, EventBuilder, EventSeverity, EventType};
17
18/// Event handler callback type
19pub type EventHandler = Arc<dyn Fn(&Event) + Send + Sync>;
20
21/// Event subscription
22pub struct Subscription {
23    pub event_type: EventType,
24    pub handler_id: String,
25}
26
27/// Event Bus - Central publish/subscribe system
28pub struct EventBus {
29    subscribers: Arc<DashMap<EventType, Vec<(String, EventHandler)>>>,
30    event_history: Arc<DashMap<String, Vec<Event>>>,
31    event_count: Arc<std::sync::atomic::AtomicU64>,
32}
33
34impl EventBus {
35    /// Creates new event bus
36    pub fn new() -> Self {
37        Self {
38            subscribers: Arc::new(DashMap::new()),
39            event_history: Arc::new(DashMap::new()),
40            event_count: Arc::new(std::sync::atomic::AtomicU64::new(0)),
41        }
42    }
43
44    /// Subscribes to an event type
45    pub fn subscribe(
46        &self,
47        event_type: EventType,
48        handler_id: impl Into<String>,
49        handler: EventHandler,
50    ) {
51        let handler_id = handler_id.into();
52        self.subscribers
53            .entry(event_type.clone())
54            .or_default()
55            .push((handler_id, handler));
56    }
57
58    /// Unsubscribes from an event type
59    pub fn unsubscribe(&self, event_type: &EventType, handler_id: &str) {
60        if let Some(mut handlers) = self.subscribers.get_mut(event_type) {
61            handlers.retain(|(id, _)| id != handler_id);
62        }
63    }
64
65    /// Publishes event to all subscribers
66    pub fn publish(&self, event: Event) {
67        self.event_count
68            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
69
70        // Store in history
71        self.event_history
72            .entry(event.event_type.as_str().to_string())
73            .or_default()
74            .push(event.clone());
75
76        // Call subscribers
77        if let Some(handlers) = self.subscribers.get(&event.event_type) {
78            for (_, handler) in handlers.iter() {
79                let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
80                    handler(&event);
81                }));
82            }
83        }
84    }
85
86    /// Gets subscriber count for event type
87    pub fn subscriber_count(&self, event_type: &EventType) -> usize {
88        self.subscribers
89            .get(event_type)
90            .map(|h| h.len())
91            .unwrap_or(0)
92    }
93
94    /// Gets event history for type
95    pub fn get_history(&self, event_type: &EventType) -> Vec<Event> {
96        self.event_history
97            .get(event_type.as_str())
98            .map(|h| h.clone())
99            .unwrap_or_default()
100    }
101
102    /// Gets total event count
103    pub fn total_events(&self) -> u64 {
104        self.event_count.load(std::sync::atomic::Ordering::SeqCst)
105    }
106
107    /// Clears event history
108    pub fn clear_history(&self) {
109        self.event_history.clear();
110    }
111
112    /// Gets all events
113    pub fn get_all_events(&self) -> Vec<Event> {
114        self.event_history
115            .iter()
116            .flat_map(|ref_multi| ref_multi.value().clone())
117            .collect()
118    }
119
120    /// Gets all events for a specific correlation ID (scan)
121    pub fn get_events_by_correlation(&self, correlation_id: &str) -> Vec<Event> {
122        self.get_all_events()
123            .into_iter()
124            .filter(|e| e.correlation_id == correlation_id)
125            .collect()
126    }
127
128    /// Gets events for correlation ID within time range (ms)
129    pub fn get_events_by_correlation_and_time(
130        &self,
131        correlation_id: &str,
132        start_ms: u64,
133        end_ms: u64,
134    ) -> Vec<Event> {
135        self.get_all_events()
136            .into_iter()
137            .filter(|e| {
138                e.correlation_id == correlation_id
139                    && e.timestamp_ms >= start_ms
140                    && e.timestamp_ms <= end_ms
141            })
142            .collect()
143    }
144
145    /// Gets all events sorted by timestamp
146    pub fn get_events_sorted(&self) -> Vec<Event> {
147        let mut events = self.get_all_events();
148        events.sort_by_key(|event| event.timestamp_ms);
149        events
150    }
151}
152
153impl Default for EventBus {
154    fn default() -> Self {
155        Self::new()
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162    use std::sync::atomic::AtomicBool;
163
164    #[test]
165    fn test_event_creation() {
166        let event = Event::new(EventType::ScanStarted, "scanner");
167        assert_eq!(event.event_type, EventType::ScanStarted);
168        assert_eq!(event.source, "scanner");
169        assert_eq!(event.version, 1);
170        assert!(!event.correlation_id.is_empty());
171        assert!(!event.event_id.is_empty());
172    }
173
174    #[test]
175    fn test_event_with_data() {
176        let event = Event::new(EventType::FindingFound, "plugin")
177            .with_data("severity", "HIGH")
178            .with_data("type", "XSS");
179
180        assert_eq!(event.data.get("severity"), Some(&"HIGH".to_string()));
181        assert_eq!(event.data.get("type"), Some(&"XSS".to_string()));
182    }
183
184    #[test]
185    fn test_event_severity() {
186        let event =
187            Event::new(EventType::ScanFailed, "worker").with_severity(EventSeverity::Critical);
188
189        assert_eq!(event.severity, EventSeverity::Critical);
190    }
191
192    #[test]
193    fn test_event_versioning() {
194        let event = Event::new(EventType::ScanStarted, "scanner").with_version(2);
195
196        assert_eq!(event.version, 2);
197    }
198
199    #[test]
200    fn test_event_correlation_id() {
201        let scan_id = "scan_12345";
202        let event = Event::new(EventType::FindingFound, "scanner").with_correlation_id(scan_id);
203
204        assert_eq!(event.correlation_id, scan_id);
205    }
206
207    #[test]
208    fn test_event_builder() {
209        let scan_id = "scan_abc";
210        let event = Event::builder(EventType::ScanStarted, "scanner")
211            .correlation_id(scan_id)
212            .version(1)
213            .severity(EventSeverity::Info)
214            .data("target", "http://example.com")
215            .build();
216
217        assert_eq!(event.correlation_id, scan_id);
218        assert_eq!(event.version, 1);
219        assert_eq!(event.severity, EventSeverity::Info);
220        assert_eq!(
221            event.data.get("target"),
222            Some(&"http://example.com".to_string())
223        );
224    }
225
226    #[test]
227    fn test_event_bus_creation() {
228        let bus = EventBus::new();
229        assert_eq!(bus.total_events(), 0);
230    }
231
232    #[test]
233    fn test_event_publishing() {
234        let bus = EventBus::new();
235        let event = Event::new(EventType::ScanStarted, "test");
236
237        bus.publish(event);
238        assert_eq!(bus.total_events(), 1);
239    }
240
241    #[test]
242    fn test_event_subscription() {
243        let bus = EventBus::new();
244        let called = Arc::new(AtomicBool::new(false));
245        let called_clone = called.clone();
246
247        bus.subscribe(
248            EventType::ScanStarted,
249            "test_handler",
250            Arc::new(move |_| {
251                called_clone.store(true, std::sync::atomic::Ordering::SeqCst);
252            }),
253        );
254
255        assert_eq!(bus.subscriber_count(&EventType::ScanStarted), 1);
256
257        let event = Event::new(EventType::ScanStarted, "test");
258        bus.publish(event);
259
260        assert!(called.load(std::sync::atomic::Ordering::SeqCst));
261    }
262
263    #[test]
264    fn test_multiple_subscribers() {
265        let bus = EventBus::new();
266        let count = Arc::new(std::sync::atomic::AtomicU32::new(0));
267
268        for i in 0..3 {
269            let count_clone = count.clone();
270            bus.subscribe(
271                EventType::FindingFound,
272                format!("handler_{i}"),
273                Arc::new(move |_| {
274                    count_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
275                }),
276            );
277        }
278
279        assert_eq!(bus.subscriber_count(&EventType::FindingFound), 3);
280
281        let event = Event::new(EventType::FindingFound, "test");
282        bus.publish(event);
283
284        assert_eq!(count.load(std::sync::atomic::Ordering::SeqCst), 3);
285    }
286
287    #[test]
288    fn test_event_unsubscribe() {
289        let bus = EventBus::new();
290
291        bus.subscribe(EventType::ScanCompleted, "handler1", Arc::new(|_| {}));
292        bus.subscribe(EventType::ScanCompleted, "handler2", Arc::new(|_| {}));
293
294        assert_eq!(bus.subscriber_count(&EventType::ScanCompleted), 2);
295
296        bus.unsubscribe(&EventType::ScanCompleted, "handler1");
297        assert_eq!(bus.subscriber_count(&EventType::ScanCompleted), 1);
298    }
299
300    #[test]
301    fn test_event_history() {
302        let bus = EventBus::new();
303
304        for i in 0..5 {
305            let event = Event::new(EventType::WorkerFinished, format!("worker_{i}"));
306            bus.publish(event);
307        }
308
309        let history = bus.get_history(&EventType::WorkerFinished);
310        assert_eq!(history.len(), 5);
311    }
312
313    #[test]
314    fn test_event_severity_ordering() {
315        assert!(EventSeverity::Critical > EventSeverity::Error);
316        assert!(EventSeverity::Error > EventSeverity::Warning);
317        assert!(EventSeverity::Warning > EventSeverity::Info);
318        assert!(EventSeverity::Info > EventSeverity::Debug);
319    }
320
321    #[test]
322    fn test_get_all_events() {
323        let bus = EventBus::new();
324
325        bus.publish(Event::new(EventType::ScanStarted, "test"));
326        bus.publish(Event::new(EventType::FindingFound, "test"));
327        bus.publish(Event::new(EventType::WorkerFinished, "test"));
328
329        let all = bus.get_all_events();
330        assert_eq!(all.len(), 3);
331    }
332
333    #[test]
334    fn test_clear_history() {
335        let bus = EventBus::new();
336
337        bus.publish(Event::new(EventType::ScanStarted, "test"));
338        bus.publish(Event::new(EventType::FindingFound, "test"));
339
340        assert!(bus.total_events() > 0);
341
342        bus.clear_history();
343        assert_eq!(bus.get_all_events().len(), 0);
344    }
345
346    #[test]
347    fn test_events_by_correlation_id() {
348        let bus = EventBus::new();
349        let scan_id = "scan_001";
350
351        // Create events for same scan
352        bus.publish(Event::new(EventType::ScanStarted, "test").with_correlation_id(scan_id));
353        bus.publish(Event::new(EventType::FindingFound, "test").with_correlation_id(scan_id));
354        bus.publish(Event::new(EventType::FindingFound, "test").with_correlation_id(scan_id));
355
356        // Create events for different scan
357        bus.publish(Event::new(EventType::ScanStarted, "test").with_correlation_id("scan_002"));
358
359        let events = bus.get_events_by_correlation(scan_id);
360        assert_eq!(events.len(), 3);
361        assert!(events.iter().all(|e| e.correlation_id == scan_id));
362    }
363
364    #[test]
365    fn test_events_by_correlation_and_time() {
366        let bus = EventBus::new();
367        let scan_id = "scan_001";
368        let now = std::time::SystemTime::now()
369            .duration_since(std::time::UNIX_EPOCH)
370            .unwrap()
371            .as_millis() as u64;
372
373        // Event at current time
374        let event1 = Event::builder(EventType::ScanStarted, "test")
375            .correlation_id(scan_id)
376            .timestamp_ms(now)
377            .build();
378        bus.publish(event1);
379
380        // Event 1 second later
381        let event2 = Event::builder(EventType::FindingFound, "test")
382            .correlation_id(scan_id)
383            .timestamp_ms(now + 1000)
384            .build();
385        bus.publish(event2);
386
387        // Query events within time range
388        let events = bus.get_events_by_correlation_and_time(scan_id, now, now + 1000);
389        assert_eq!(events.len(), 2);
390    }
391
392    #[test]
393    fn test_events_sorted_by_timestamp() {
394        let bus = EventBus::new();
395
396        let now = std::time::SystemTime::now()
397            .duration_since(std::time::UNIX_EPOCH)
398            .unwrap()
399            .as_millis() as u64;
400
401        // Add events in random order
402        bus.publish(
403            Event::builder(EventType::FindingFound, "test")
404                .timestamp_ms(now + 2000)
405                .build(),
406        );
407        bus.publish(
408            Event::builder(EventType::ScanStarted, "test")
409                .timestamp_ms(now)
410                .build(),
411        );
412        bus.publish(
413            Event::builder(EventType::WorkerFinished, "test")
414                .timestamp_ms(now + 1000)
415                .build(),
416        );
417
418        let sorted = bus.get_events_sorted();
419        assert_eq!(sorted.len(), 3);
420        assert!(sorted[0].timestamp_ms <= sorted[1].timestamp_ms);
421        assert!(sorted[1].timestamp_ms <= sorted[2].timestamp_ms);
422    }
423
424    #[test]
425    fn test_event_uniqueness() {
426        let event1 = Event::new(EventType::ScanStarted, "test");
427        let event2 = Event::new(EventType::ScanStarted, "test");
428
429        // Each event has unique ID and correlation ID
430        assert_ne!(event1.event_id, event2.event_id);
431        assert_ne!(event1.correlation_id, event2.correlation_id);
432    }
433
434    #[test]
435    fn test_timestamp_str() {
436        let event = Event::builder(EventType::ScanStarted, "test")
437            .timestamp_ms(1234567890123)
438            .build();
439
440        assert_eq!(event.timestamp_str(), "1234567890.123");
441    }
442
443    // ============ CONCURRENCY & STRESS TESTS ============
444
445    #[test]
446    fn test_concurrent_publications_1000() {
447        let bus = Arc::new(EventBus::new());
448        let mut handles = vec![];
449
450        for i in 0..1000 {
451            let bus_clone = bus.clone();
452            let handle = std::thread::spawn(move || {
453                let event = Event::builder(EventType::FindingFound, format!("thread_{i}"))
454                    .correlation_id("scan_concurrent")
455                    .build();
456                bus_clone.publish(event);
457            });
458            handles.push(handle);
459        }
460
461        // Wait for all threads
462        for handle in handles {
463            handle.join().unwrap();
464        }
465
466        // Verify all events published
467        assert_eq!(bus.total_events(), 1000);
468        let events = bus.get_events_by_correlation("scan_concurrent");
469        assert_eq!(events.len(), 1000);
470    }
471
472    #[tokio::test]
473    async fn test_concurrent_async_publications() {
474        let bus = Arc::new(EventBus::new());
475        let mut tasks = vec![];
476
477        for i in 0..500 {
478            let bus_clone = bus.clone();
479            let task = tokio::spawn(async move {
480                let event =
481                    Event::builder(EventType::WorkerFinished, format!("async_worker_{i}")).build();
482                bus_clone.publish(event);
483            });
484            tasks.push(task);
485        }
486
487        // Wait for all tasks
488        for task in tasks {
489            task.await.unwrap();
490        }
491
492        assert_eq!(bus.total_events(), 500);
493    }
494
495    #[test]
496    fn test_subscriber_panic_isolation() {
497        let bus = Arc::new(EventBus::new());
498        bus.subscribe(
499            EventType::ScanStarted,
500            "panicking_handler",
501            Arc::new(|_| {
502                panic!("subscriber fixture panic");
503            }),
504        );
505
506        let normal_called = Arc::new(std::sync::atomic::AtomicBool::new(false));
507        let normal_clone = normal_called.clone();
508        bus.subscribe(
509            EventType::ScanStarted,
510            "normal_handler",
511            Arc::new(move |_| {
512                normal_clone.store(true, std::sync::atomic::Ordering::SeqCst);
513            }),
514        );
515
516        bus.publish(Event::new(EventType::ScanStarted, "test"));
517
518        assert!(normal_called.load(std::sync::atomic::Ordering::SeqCst));
519        assert_eq!(bus.total_events(), 1);
520    }
521
522    #[test]
523    fn test_unsubscribe_memory_cleanup() {
524        let bus = EventBus::new();
525
526        // Subscribe many handlers
527        for i in 0..100 {
528            bus.subscribe(
529                EventType::FindingFound,
530                format!("handler_{i}"),
531                Arc::new(|_| {}),
532            );
533        }
534
535        assert_eq!(bus.subscriber_count(&EventType::FindingFound), 100);
536
537        // Unsubscribe all
538        for i in 0..100 {
539            bus.unsubscribe(&EventType::FindingFound, &format!("handler_{i}"));
540        }
541
542        // Memory should be cleaned
543        assert_eq!(bus.subscriber_count(&EventType::FindingFound), 0);
544    }
545
546    #[test]
547    fn test_event_order_preservation() {
548        let bus = EventBus::new();
549        let scan_id = "order_test";
550        let now = std::time::SystemTime::now()
551            .duration_since(std::time::UNIX_EPOCH)
552            .unwrap()
553            .as_millis() as u64;
554
555        // Publish events with increasing timestamps
556        for i in 0..100 {
557            let event = Event::builder(EventType::FindingFound, "test")
558                .correlation_id(scan_id)
559                .timestamp_ms(now + i as u64)
560                .build();
561            bus.publish(event);
562        }
563
564        let events = bus.get_events_sorted();
565
566        // Verify order
567        for i in 0..events.len() - 1 {
568            assert!(
569                events[i].timestamp_ms <= events[i + 1].timestamp_ms,
570                "Events not in chronological order"
571            );
572        }
573    }
574
575    #[tokio::test]
576    async fn test_slow_subscriber_nonblocking() {
577        let bus = Arc::new(EventBus::new());
578        let slow_counter = Arc::new(std::sync::atomic::AtomicU32::new(0));
579        let fast_counter = Arc::new(std::sync::atomic::AtomicU32::new(0));
580
581        // Slow subscriber
582        let slow_clone = slow_counter.clone();
583        bus.subscribe(
584            EventType::WorkerFinished,
585            "slow",
586            Arc::new(move |_| {
587                // Simulate slow work
588                std::thread::sleep(std::time::Duration::from_millis(10));
589                slow_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
590            }),
591        );
592
593        // Fast subscriber
594        let fast_clone = fast_counter.clone();
595        bus.subscribe(
596            EventType::WorkerFinished,
597            "fast",
598            Arc::new(move |_| {
599                fast_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
600            }),
601        );
602
603        // Publish events
604        for _ in 0..10 {
605            let event = Event::new(EventType::WorkerFinished, "test");
606            bus.publish(event);
607        }
608
609        // Fast subscriber should be called same number of times
610        assert_eq!(fast_counter.load(std::sync::atomic::Ordering::SeqCst), 10);
611        assert_eq!(slow_counter.load(std::sync::atomic::Ordering::SeqCst), 10);
612    }
613
614    #[test]
615    fn test_concurrent_subscribe_unsubscribe() {
616        let bus = Arc::new(EventBus::new());
617        let mut handles = vec![];
618
619        // Half subscribe, half unsubscribe concurrently
620        for i in 0..200 {
621            let bus_clone = bus.clone();
622            let handle = std::thread::spawn(move || {
623                if i % 2 == 0 {
624                    // Subscribe
625                    bus_clone.subscribe(
626                        EventType::ScanCompleted,
627                        format!("handler_{i}"),
628                        Arc::new(|_| {}),
629                    );
630                } else if i > 1 {
631                    // Unsubscribe
632                    bus_clone.unsubscribe(&EventType::ScanCompleted, &format!("handler_{}", i - 1));
633                }
634            });
635            handles.push(handle);
636        }
637
638        for handle in handles {
639            handle.join().unwrap();
640        }
641
642        // Should not crash and have reasonable subscriber count
643        let count = bus.subscriber_count(&EventType::ScanCompleted);
644        assert!(count <= 200);
645    }
646
647    #[test]
648    fn test_large_event_payload() {
649        let bus = EventBus::new();
650        let large_data = "x".repeat(1_000_000); // 1MB data
651
652        let event = Event::builder(EventType::FindingFound, "test")
653            .data("large_payload", large_data.clone())
654            .build();
655
656        bus.publish(event);
657
658        let retrieved = bus.get_all_events();
659        assert_eq!(retrieved.len(), 1);
660        assert_eq!(retrieved[0].data.get("large_payload"), Some(&large_data));
661    }
662
663    #[test]
664    fn test_many_subscribers_single_event() {
665        let bus = Arc::new(EventBus::new());
666        let counter = Arc::new(std::sync::atomic::AtomicU32::new(0));
667
668        // Subscribe 500 handlers to same event
669        for i in 0..500 {
670            let counter_clone = counter.clone();
671            bus.subscribe(
672                EventType::AlertTriggered,
673                format!("listener_{i}"),
674                Arc::new(move |_| {
675                    counter_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
676                }),
677            );
678        }
679
680        assert_eq!(bus.subscriber_count(&EventType::AlertTriggered), 500);
681
682        let event = Event::new(EventType::AlertTriggered, "test");
683        bus.publish(event);
684
685        // All subscribers called
686        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 500);
687    }
688
689    #[test]
690    fn test_event_history_memory_pressure() {
691        let bus = EventBus::new();
692
693        // Publish 10,000 events
694        for i in 0..10_000 {
695            let event = Event::builder(EventType::FindingFound, "test")
696                .data("index", i.to_string())
697                .build();
698            bus.publish(event);
699        }
700
701        assert_eq!(bus.total_events(), 10_000);
702
703        // Query subset should work efficiently
704        let all = bus.get_all_events();
705        assert_eq!(all.len(), 10_000);
706
707        // Clear should free memory
708        bus.clear_history();
709        assert_eq!(bus.get_all_events().len(), 0);
710        assert_eq!(bus.total_events(), 10_000); // Count not reset, just history cleared
711    }
712
713    #[tokio::test]
714    async fn test_concurrent_correlation_queries() {
715        let bus = Arc::new(EventBus::new());
716
717        // Publish to multiple correlation IDs
718        for scan_id in 0..10 {
719            for _ in 0..100 {
720                let event = Event::builder(EventType::FindingFound, "test")
721                    .correlation_id(format!("scan_{scan_id}"))
722                    .build();
723                bus.publish(event);
724            }
725        }
726
727        // Query concurrently
728        let mut tasks = vec![];
729        for scan_id in 0..10 {
730            let bus_clone = bus.clone();
731            let task = tokio::spawn(async move {
732                let events = bus_clone.get_events_by_correlation(&format!("scan_{scan_id}"));
733                assert_eq!(events.len(), 100);
734            });
735            tasks.push(task);
736        }
737
738        for task in tasks {
739            task.await.unwrap();
740        }
741    }
742}
termivar_scanner/event_bus.rs

termivar_scanner/
event_bus.rs
