use std::sync::{Arc, Mutex};
use std::time::SystemTime;
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum AuditEventKind {
VmCreated {
id: String,
image: Option<String>,
},
VmStarted {
id: String,
},
VmStopped {
id: String,
exit_code: Option<i32>,
},
VmRemoved {
id: String,
},
ExecStarted {
vm_id: String,
command: String,
exec_id: String,
},
ExecCompleted {
vm_id: String,
exec_id: String,
exit_code: i32,
duration_ms: u64,
},
SnapshotCreated {
vm_id: String,
snapshot_id: String,
},
SnapshotRestored {
vm_id: String,
snapshot_id: String,
},
FileCopied {
vm_id: String,
direction: CopyDirection,
path: String,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum CopyDirection {
In,
Out,
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct AuditEvent {
pub timestamp: SystemTime,
pub kind: AuditEventKind,
}
impl AuditEvent {
#[must_use]
pub fn now(kind: AuditEventKind) -> Self {
Self {
timestamp: SystemTime::now(),
kind,
}
}
}
pub trait EventListener: Send + Sync {
fn on_event(&self, event: &AuditEvent);
}
#[derive(Default)]
pub struct EventDispatcher {
listeners: Mutex<Vec<Arc<dyn EventListener>>>,
}
impl std::fmt::Debug for EventDispatcher {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let count = self.listeners.lock().map_or(0, |l| l.len());
f.debug_struct("EventDispatcher")
.field("listener_count", &count)
.finish()
}
}
impl EventDispatcher {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn add_listener(&self, listener: Arc<dyn EventListener>) {
if let Ok(mut listeners) = self.listeners.lock() {
listeners.push(listener);
}
}
#[allow(
clippy::needless_pass_by_value,
reason = "event is consumed after broadcast"
)]
pub fn emit(&self, event: AuditEvent) {
if let Ok(listeners) = self.listeners.lock() {
for listener in listeners.iter() {
listener.on_event(&event);
}
}
}
}
pub struct RingBufferListener {
inner: Mutex<RingBuffer>,
capacity: usize,
}
struct RingBuffer {
slots: Vec<Option<AuditEvent>>,
write_pos: usize,
total: u64,
}
impl std::fmt::Debug for RingBufferListener {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RingBufferListener")
.field("capacity", &self.capacity)
.field("total_events", &self.total_events())
.finish_non_exhaustive()
}
}
impl RingBufferListener {
#[must_use]
pub fn new(capacity: usize) -> Self {
let cap = capacity.max(1);
Self {
inner: Mutex::new(RingBuffer {
slots: vec![None; cap],
write_pos: 0,
total: 0,
}),
capacity: cap,
}
}
#[must_use]
pub fn total_events(&self) -> u64 {
self.inner.lock().map_or(0, |g| g.total)
}
#[must_use]
pub fn recent(&self, limit: usize) -> Vec<AuditEvent> {
let Ok(guard) = self.inner.lock() else {
return Vec::new();
};
#[allow(
clippy::cast_possible_truncation,
reason = "capacity is usize, so truncation is acceptable"
)]
let total = guard.total as usize;
let available = total.min(self.capacity);
let take = limit.min(available);
if take == 0 {
return Vec::new();
}
let write_pos = guard.write_pos;
let mut events = Vec::with_capacity(take);
let start = if write_pos >= take {
write_pos - take
} else {
self.capacity - (take - write_pos)
};
for i in 0..take {
let idx = (start + i) % self.capacity;
if let Some(event) = guard.slots.get(idx).and_then(Option::as_ref) {
events.push(event.clone());
}
}
events
}
}
impl EventListener for RingBufferListener {
fn on_event(&self, event: &AuditEvent) {
if let Ok(mut guard) = self.inner.lock() {
let pos = guard.write_pos;
if let Some(slot) = guard.slots.get_mut(pos % self.capacity) {
*slot = Some(event.clone());
}
guard.write_pos = (pos + 1) % self.capacity;
guard.total += 1;
}
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::indexing_slicing,
reason = "test assertions use unwrap/indexing for clarity"
)]
mod tests {
use super::*;
fn make_event(id: &str) -> AuditEvent {
AuditEvent::now(AuditEventKind::VmCreated {
id: id.to_owned(),
image: None,
})
}
#[test]
fn ring_buffer_stores_and_retrieves() {
let ring = RingBufferListener::new(3);
ring.on_event(&make_event("vm1"));
ring.on_event(&make_event("vm2"));
let events = ring.recent(10);
assert_eq!(events.len(), 2);
assert_eq!(ring.total_events(), 2);
}
#[test]
fn ring_buffer_wraps_around() {
let ring = RingBufferListener::new(2);
ring.on_event(&make_event("vm1"));
ring.on_event(&make_event("vm2"));
ring.on_event(&make_event("vm3"));
assert_eq!(ring.total_events(), 3);
let events = ring.recent(10);
assert_eq!(events.len(), 2);
if let AuditEventKind::VmCreated { ref id, .. } = events[0].kind {
assert_eq!(id, "vm2");
}
if let AuditEventKind::VmCreated { ref id, .. } = events[1].kind {
assert_eq!(id, "vm3");
}
}
#[test]
fn file_copied_variant_round_trip() {
let event = AuditEvent::now(AuditEventKind::FileCopied {
vm_id: "vm1".into(),
direction: CopyDirection::In,
path: "/tmp/x".into(),
});
assert!(matches!(
event.kind,
AuditEventKind::FileCopied {
ref vm_id,
direction: CopyDirection::In,
ref path,
} if vm_id == "vm1" && path == "/tmp/x"
));
}
#[test]
fn snapshot_restored_variant_round_trip() {
let event = AuditEvent::now(AuditEventKind::SnapshotRestored {
vm_id: "vm1".into(),
snapshot_id: "snap1".into(),
});
assert!(matches!(
event.kind,
AuditEventKind::SnapshotRestored {
ref vm_id,
ref snapshot_id,
} if vm_id == "vm1" && snapshot_id == "snap1"
));
}
#[test]
fn dispatcher_fans_out() {
let dispatcher = EventDispatcher::new();
let ring = Arc::new(RingBufferListener::new(10));
#[allow(
clippy::clone_on_ref_ptr,
reason = "coercion to dyn trait requires .clone()"
)]
let listener: Arc<dyn EventListener> = ring.clone();
dispatcher.add_listener(listener);
dispatcher.emit(make_event("vm1"));
dispatcher.emit(make_event("vm2"));
assert_eq!(ring.total_events(), 2);
}
}