arc_core/
event_store.rs

1//! # Event Store Module
2//!
3//! Defines the [`EventStore`] trait for persisting and retrieving events.
4//!
5//! ## Design Principles
6//!
7//! - **Append-only**: events can only be added, never modified or deleted
8//! - **Optimistic concurrency**: version-based conflict detection
9//! - **Stream-based**: events can be loaded by aggregate or streamed globally
10//! - **Audited**: every event must carry valid [`AuditMetadata`](crate::audit::AuditMetadata)
11//!   when appended (HIPAA §164.312(b))
12//! - **Pluggable**: multiple implementations (SQLite, Postgres, in-memory)
13//!
14//! ## HIPAA defense-in-depth
15//!
16//! `EventStore::append` MUST call `event.audit.validate()?` for each event
17//! before persisting. The `CommandBus` validates first, but the store is the
18//! durable boundary — it must not trust upstream.
19
20use crate::audit::AuditError;
21use crate::event::Event;
22use crate::integrity::IntegrityError;
23use crate::snapshot::Snapshot;
24use async_trait::async_trait;
25use thiserror::Error;
26
27/// Version check strategy for optimistic concurrency control.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum VersionCheck {
30    /// First event for this aggregate (expected version is 0)
31    New,
32    /// Require aggregate to be at this exact version
33    Expected(i64),
34    /// Automatically load and use current version (use sparingly)
35    Auto,
36}
37
38impl VersionCheck {
39    pub fn version(&self) -> Option<i64> {
40        match self {
41            VersionCheck::New => Some(0),
42            VersionCheck::Expected(v) => Some(*v),
43            VersionCheck::Auto => None,
44        }
45    }
46}
47
48/// Errors that can occur during event store operations.
49#[derive(Debug, Error)]
50pub enum EventStoreError {
51    /// Optimistic concurrency conflict.
52    #[error("Concurrency conflict: expected version {expected}, but aggregate is at version {actual} (aggregate_id: {aggregate_id})")]
53    ConcurrencyConflict {
54        aggregate_id: String,
55        expected: i64,
56        actual: i64,
57    },
58
59    #[error("Aggregate not found: {aggregate_id}")]
60    AggregateNotFound { aggregate_id: String },
61
62    #[error(
63        "Invalid event sequence: expected {expected}, got {actual} (aggregate_id: {aggregate_id})"
64    )]
65    InvalidSequence {
66        aggregate_id: String,
67        expected: i64,
68        actual: i64,
69    },
70
71    /// One or more events in an `append` batch had invalid audit metadata.
72    /// Defense-in-depth: the bus should have caught this first.
73    #[error(
74        "Audit metadata validation failed for event {event_index} (aggregate_id: {aggregate_id}): {source}"
75    )]
76    InvalidAudit {
77        aggregate_id: String,
78        event_index: usize,
79        #[source]
80        source: AuditError,
81    },
82
83    /// Stored event signatures are missing or do not match the recomputed
84    /// integrity chain.
85    #[error("Integrity validation failed: {source}")]
86    Integrity {
87        #[from]
88        source: IntegrityError,
89    },
90
91    #[error("Database error: {message}")]
92    DatabaseError { message: String },
93
94    #[error("Serialization error: {message}")]
95    SerializationError { message: String },
96
97    #[error("I/O error: {0}")]
98    IoError(#[from] std::io::Error),
99
100    /// The store does not implement snapshot persistence. Callers fall back to
101    /// replaying the full event stream.
102    #[error("snapshots not supported by this store")]
103    Unsupported,
104
105    #[error("Event store error: {message}")]
106    Other { message: String },
107}
108
109impl EventStoreError {
110    pub fn database(message: impl Into<String>) -> Self {
111        EventStoreError::DatabaseError {
112            message: message.into(),
113        }
114    }
115
116    pub fn serialization(message: impl Into<String>) -> Self {
117        EventStoreError::SerializationError {
118            message: message.into(),
119        }
120    }
121
122    pub fn other(message: impl Into<String>) -> Self {
123        EventStoreError::Other {
124            message: message.into(),
125        }
126    }
127
128    pub fn invalid_audit(
129        aggregate_id: impl Into<String>,
130        event_index: usize,
131        source: AuditError,
132    ) -> Self {
133        EventStoreError::InvalidAudit {
134            aggregate_id: aggregate_id.into(),
135            event_index,
136            source,
137        }
138    }
139}
140
141/// Result type for event store operations.
142pub type EventStoreResult<T> = Result<T, EventStoreError>;
143
144/// Helper for store implementations: validate every event's audit before persisting.
145/// Returns `Err(EventStoreError::InvalidAudit)` on the first failure.
146pub fn validate_audit_batch(aggregate_id: &str, events: &[Event]) -> EventStoreResult<()> {
147    for (idx, ev) in events.iter().enumerate() {
148        ev.audit
149            .validate()
150            .map_err(|e| EventStoreError::invalid_audit(aggregate_id, idx, e))?;
151    }
152    Ok(())
153}
154
155/// Trait for event store implementations.
156///
157/// `append` MUST invoke `validate_audit_batch` before persisting (HIPAA defense
158/// in depth). The `CommandBus` also validates upstream — both layers run.
159#[async_trait]
160pub trait EventStore: Send + Sync {
161    /// Append events to the store for a specific aggregate.
162    ///
163    /// Implementations must call `validate_audit_batch(aggregate_id, &events)?`
164    /// before any persistence work.
165    async fn append(
166        &self,
167        aggregate_id: &str,
168        version_check: VersionCheck,
169        events: Vec<Event>,
170    ) -> EventStoreResult<()>;
171
172    async fn load(&self, aggregate_id: &str) -> EventStoreResult<Vec<Event>>;
173
174    async fn load_from(
175        &self,
176        aggregate_id: &str,
177        from_sequence: i64,
178    ) -> EventStoreResult<Vec<Event>>;
179
180    async fn stream_all(&self, from_position: i64) -> EventStoreResult<Vec<Event>>;
181
182    async fn get_version(&self, aggregate_id: &str) -> EventStoreResult<i64>;
183
184    /// Persist an aggregate snapshot (upsert by `aggregate_id`).
185    ///
186    /// Default returns `EventStoreError::Unsupported` so stores that have not
187    /// implemented snapshotting compile unchanged and fail loudly if a caller
188    /// tries to save one.
189    async fn save_snapshot(&self, snapshot: &Snapshot) -> EventStoreResult<()> {
190        let _ = snapshot;
191        Err(EventStoreError::Unsupported)
192    }
193
194    /// Load the latest snapshot for an aggregate, if one exists.
195    ///
196    /// Default returns `Ok(None)` — safe because the caller then replays the
197    /// stream from sequence 0, which is always correct, just slower.
198    async fn load_snapshot(&self, aggregate_id: &str) -> EventStoreResult<Option<Snapshot>> {
199        let _ = aggregate_id;
200        Ok(None)
201    }
202}
203
204// ─────────────────────────────────────────────────────────────────────────────
205// In-memory implementation, public for downstream test code.
206// ─────────────────────────────────────────────────────────────────────────────
207
208#[cfg(any(test, feature = "test-utils"))]
209mod in_memory {
210    use super::*;
211    use std::collections::HashMap;
212    use std::sync::Arc;
213    use tokio::sync::Mutex as TokioMutex;
214
215    /// In-memory event store. Available to downstream crates via the
216    /// `test-utils` feature flag.
217    ///
218    /// Validates audit metadata on every append (same contract as production
219    /// stores) so behavior matches what real implementations enforce.
220    #[derive(Clone, Default)]
221    pub struct InMemoryEventStore {
222        events: Arc<TokioMutex<Vec<Event>>>,
223        snapshots: Arc<TokioMutex<HashMap<String, Snapshot>>>,
224    }
225
226    impl InMemoryEventStore {
227        pub fn new() -> Self {
228            Self::default()
229        }
230    }
231
232    #[async_trait]
233    impl EventStore for InMemoryEventStore {
234        async fn append(
235            &self,
236            aggregate_id: &str,
237            version_check: VersionCheck,
238            events: Vec<Event>,
239        ) -> EventStoreResult<()> {
240            validate_audit_batch(aggregate_id, &events)?;
241
242            let mut store = self.events.lock().await;
243
244            let current_version = store
245                .iter()
246                .filter(|e| e.aggregate_id == aggregate_id)
247                .map(|e| e.sequence)
248                .max()
249                .unwrap_or(0);
250
251            if let Some(expected) = version_check.version() {
252                if current_version != expected {
253                    return Err(EventStoreError::ConcurrencyConflict {
254                        aggregate_id: aggregate_id.to_string(),
255                        expected,
256                        actual: current_version,
257                    });
258                }
259            }
260
261            store.extend(events);
262            Ok(())
263        }
264
265        async fn load(&self, aggregate_id: &str) -> EventStoreResult<Vec<Event>> {
266            let store = self.events.lock().await;
267            Ok(store
268                .iter()
269                .filter(|e| e.aggregate_id == aggregate_id)
270                .cloned()
271                .collect())
272        }
273
274        async fn load_from(
275            &self,
276            aggregate_id: &str,
277            from_sequence: i64,
278        ) -> EventStoreResult<Vec<Event>> {
279            let store = self.events.lock().await;
280            Ok(store
281                .iter()
282                .filter(|e| e.aggregate_id == aggregate_id && e.sequence >= from_sequence)
283                .cloned()
284                .collect())
285        }
286
287        async fn stream_all(&self, from_position: i64) -> EventStoreResult<Vec<Event>> {
288            let store = self.events.lock().await;
289            Ok(store.iter().skip(from_position as usize).cloned().collect())
290        }
291
292        async fn get_version(&self, aggregate_id: &str) -> EventStoreResult<i64> {
293            let store = self.events.lock().await;
294            Ok(store
295                .iter()
296                .filter(|e| e.aggregate_id == aggregate_id)
297                .map(|e| e.sequence)
298                .max()
299                .unwrap_or(0))
300        }
301
302        async fn save_snapshot(&self, snapshot: &Snapshot) -> EventStoreResult<()> {
303            let mut snapshots = self.snapshots.lock().await;
304            snapshots.insert(snapshot.aggregate_id.clone(), snapshot.clone());
305            Ok(())
306        }
307
308        async fn load_snapshot(&self, aggregate_id: &str) -> EventStoreResult<Option<Snapshot>> {
309            let snapshots = self.snapshots.lock().await;
310            Ok(snapshots.get(aggregate_id).cloned())
311        }
312    }
313}
314
315#[cfg(any(test, feature = "test-utils"))]
316pub use in_memory::InMemoryEventStore;
317
318#[cfg(test)]
319mod tests {
320    use super::*;
321    use crate::audit::AuditMetadata;
322    use crate::event::Event;
323    use serde_json::json;
324
325    #[test]
326    fn test_version_check_new() {
327        assert_eq!(VersionCheck::New.version(), Some(0));
328    }
329
330    #[test]
331    fn test_version_check_expected() {
332        assert_eq!(VersionCheck::Expected(5).version(), Some(5));
333    }
334
335    #[test]
336    fn test_version_check_auto() {
337        assert_eq!(VersionCheck::Auto.version(), None);
338    }
339
340    #[test]
341    fn test_error_messages() {
342        let error = EventStoreError::ConcurrencyConflict {
343            aggregate_id: "user-123".to_string(),
344            expected: 5,
345            actual: 6,
346        };
347        let msg = error.to_string();
348        assert!(msg.contains("expected version 5"));
349        assert!(msg.contains("aggregate is at version 6"));
350        assert!(msg.contains("user-123"));
351
352        assert!(EventStoreError::database("X").to_string().contains("X"));
353        assert!(EventStoreError::serialization("Y")
354            .to_string()
355            .contains("Y"));
356    }
357
358    #[test]
359    fn test_validate_audit_batch_rejects_pending() {
360        let mut e = Event::new("User", "u1", 1, "X", json!({}));
361        e.audit = AuditMetadata::pending();
362        let err = validate_audit_batch("u1", &[e]).unwrap_err();
363        assert!(matches!(
364            err,
365            EventStoreError::InvalidAudit { event_index: 0, .. }
366        ));
367    }
368
369    #[test]
370    fn test_validate_audit_batch_passes_stamped() {
371        let e =
372            Event::new("User", "u1", 1, "X", json!({})).with_audit(AuditMetadata::test_default());
373        validate_audit_batch("u1", &[e]).expect("stamped audit must pass");
374    }
375
376    #[tokio::test]
377    async fn test_in_memory_store_rejects_pending_audit() {
378        let store = InMemoryEventStore::new();
379        let e = Event::new("User", "u1", 1, "X", json!({})); // pending
380        let err = store
381            .append("u1", VersionCheck::New, vec![e])
382            .await
383            .unwrap_err();
384        assert!(matches!(err, EventStoreError::InvalidAudit { .. }));
385    }
386
387    #[tokio::test]
388    async fn test_in_memory_store_persists_stamped_event() {
389        let store = InMemoryEventStore::new();
390        let e =
391            Event::new("User", "u1", 1, "X", json!({})).with_audit(AuditMetadata::test_default());
392        store
393            .append("u1", VersionCheck::New, vec![e])
394            .await
395            .unwrap();
396        let loaded = store.load("u1").await.unwrap();
397        assert_eq!(loaded.len(), 1);
398    }
399
400    #[tokio::test]
401    async fn test_in_memory_snapshot_save_then_load() {
402        let store = InMemoryEventStore::new();
403        let snap = Snapshot::new("u1", "User", 3, json!({ "name": "Alice" }));
404        store.save_snapshot(&snap).await.unwrap();
405        let loaded = store.load_snapshot("u1").await.unwrap();
406        assert_eq!(loaded, Some(snap));
407    }
408
409    #[tokio::test]
410    async fn test_in_memory_load_snapshot_unknown_returns_none() {
411        let store = InMemoryEventStore::new();
412        assert_eq!(store.load_snapshot("missing").await.unwrap(), None);
413    }
414
415    #[tokio::test]
416    async fn test_in_memory_snapshot_overwrites_on_resave() {
417        let store = InMemoryEventStore::new();
418        store
419            .save_snapshot(&Snapshot::new("u1", "User", 3, json!({ "v": 3 })))
420            .await
421            .unwrap();
422        let newer = Snapshot::new("u1", "User", 9, json!({ "v": 9 }));
423        store.save_snapshot(&newer).await.unwrap();
424        let loaded = store.load_snapshot("u1").await.unwrap().unwrap();
425        assert_eq!(loaded.version, 9);
426        assert_eq!(loaded.state["v"], 9);
427    }
428}
arc_core/event_store.rs

arc_core/
event_store.rs
