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;
22#[cfg(test)]
23use crate::event::NewEvent;
24use crate::integrity::IntegrityError;
25use crate::snapshot::Snapshot;
26use async_trait::async_trait;
27use thiserror::Error;
28
29/// Version check strategy for optimistic concurrency control.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum VersionCheck {
32    /// First event for this aggregate (expected version is 0)
33    New,
34    /// Require aggregate to be at this exact version
35    Expected(i64),
36    /// Automatically load and use current version (use sparingly)
37    Auto,
38}
39
40impl VersionCheck {
41    pub fn version(&self) -> Option<i64> {
42        match self {
43            VersionCheck::New => Some(0),
44            VersionCheck::Expected(v) => Some(*v),
45            VersionCheck::Auto => None,
46        }
47    }
48}
49
50/// Errors that can occur during event store operations.
51#[derive(Debug, Error)]
52pub enum EventStoreError {
53    /// Optimistic concurrency conflict.
54    #[error("Concurrency conflict: expected version {expected}, but aggregate is at version {actual} (aggregate_id: {aggregate_id})")]
55    ConcurrencyConflict {
56        aggregate_id: String,
57        expected: i64,
58        actual: i64,
59    },
60
61    #[error("Aggregate not found: {aggregate_id}")]
62    AggregateNotFound { aggregate_id: String },
63
64    #[error(
65        "Invalid event sequence: expected {expected}, got {actual} (aggregate_id: {aggregate_id})"
66    )]
67    InvalidSequence {
68        aggregate_id: String,
69        expected: i64,
70        actual: i64,
71    },
72
73    /// One or more events in an `append` batch had invalid audit metadata.
74    /// Defense-in-depth: the bus should have caught this first.
75    #[error(
76        "Audit metadata validation failed for event {event_index} (aggregate_id: {aggregate_id}): {source}"
77    )]
78    InvalidAudit {
79        aggregate_id: String,
80        event_index: usize,
81        #[source]
82        source: AuditError,
83    },
84
85    /// Stored event signatures are missing or do not match the recomputed
86    /// integrity chain.
87    #[error("Integrity validation failed: {source}")]
88    Integrity {
89        #[from]
90        source: IntegrityError,
91    },
92
93    #[error("Database error: {message}")]
94    DatabaseError { message: String },
95
96    #[error("Serialization error: {message}")]
97    SerializationError { message: String },
98
99    #[error("I/O error: {0}")]
100    IoError(#[from] std::io::Error),
101
102    /// The store does not implement snapshot persistence. Callers fall back to
103    /// replaying the full event stream.
104    #[error("snapshots not supported by this store")]
105    Unsupported,
106
107    #[error("Event store error: {message}")]
108    Other { message: String },
109}
110
111impl EventStoreError {
112    pub fn database(message: impl Into<String>) -> Self {
113        EventStoreError::DatabaseError {
114            message: message.into(),
115        }
116    }
117
118    pub fn serialization(message: impl Into<String>) -> Self {
119        EventStoreError::SerializationError {
120            message: message.into(),
121        }
122    }
123
124    pub fn other(message: impl Into<String>) -> Self {
125        EventStoreError::Other {
126            message: message.into(),
127        }
128    }
129
130    pub fn invalid_audit(
131        aggregate_id: impl Into<String>,
132        event_index: usize,
133        source: AuditError,
134    ) -> Self {
135        EventStoreError::InvalidAudit {
136            aggregate_id: aggregate_id.into(),
137            event_index,
138            source,
139        }
140    }
141}
142
143/// Result type for event store operations.
144pub type EventStoreResult<T> = Result<T, EventStoreError>;
145
146/// Helper for store implementations: validate every event's audit before persisting.
147/// Returns `Err(EventStoreError::InvalidAudit)` on the first failure.
148pub fn validate_audit_batch(aggregate_id: &str, events: &[Event]) -> EventStoreResult<()> {
149    for (idx, ev) in events.iter().enumerate() {
150        ev.audit
151            .validate()
152            .map_err(|e| EventStoreError::invalid_audit(aggregate_id, idx, e))?;
153    }
154    Ok(())
155}
156
157/// Trait for event store implementations.
158///
159/// `append` MUST invoke `validate_audit_batch` before persisting (HIPAA defense
160/// in depth). The `CommandBus` also validates upstream — both layers run.
161#[async_trait]
162pub trait EventStore: Send + Sync {
163    /// Append events to the store for a specific aggregate.
164    ///
165    /// Implementations must call `validate_audit_batch(aggregate_id, &events)?`
166    /// before any persistence work.
167    async fn append(
168        &self,
169        aggregate_id: &str,
170        version_check: VersionCheck,
171        events: Vec<Event>,
172    ) -> EventStoreResult<()>;
173
174    /// Append to the stream identified by aggregate type and instance ID.
175    ///
176    /// The default preserves compatibility with stores that historically keyed
177    /// streams only by instance ID. Multi-aggregate stores override this to
178    /// enforce `(aggregate_type, aggregate_id, sequence)` identity.
179    async fn append_to(
180        &self,
181        aggregate_type: &str,
182        aggregate_id: &str,
183        version_check: VersionCheck,
184        events: Vec<Event>,
185    ) -> EventStoreResult<()> {
186        let _ = aggregate_type;
187        self.append(aggregate_id, version_check, events).await
188    }
189
190    async fn load(&self, aggregate_id: &str) -> EventStoreResult<Vec<Event>>;
191
192    async fn load_stream(
193        &self,
194        aggregate_type: &str,
195        aggregate_id: &str,
196    ) -> EventStoreResult<Vec<Event>> {
197        let _ = aggregate_type;
198        self.load(aggregate_id).await
199    }
200
201    async fn load_from(
202        &self,
203        aggregate_id: &str,
204        from_sequence: i64,
205    ) -> EventStoreResult<Vec<Event>>;
206
207    async fn load_stream_from(
208        &self,
209        aggregate_type: &str,
210        aggregate_id: &str,
211        from_sequence: i64,
212    ) -> EventStoreResult<Vec<Event>> {
213        let _ = aggregate_type;
214        self.load_from(aggregate_id, from_sequence).await
215    }
216
217    async fn stream_all(&self, from_position: i64) -> EventStoreResult<Vec<Event>>;
218
219    async fn get_version(&self, aggregate_id: &str) -> EventStoreResult<i64>;
220
221    async fn get_stream_version(
222        &self,
223        aggregate_type: &str,
224        aggregate_id: &str,
225    ) -> EventStoreResult<i64> {
226        let _ = aggregate_type;
227        self.get_version(aggregate_id).await
228    }
229
230    /// Persist an aggregate snapshot (upsert by `aggregate_id`).
231    ///
232    /// Default returns `EventStoreError::Unsupported` so stores that have not
233    /// implemented snapshotting compile unchanged and fail loudly if a caller
234    /// tries to save one.
235    async fn save_snapshot(&self, snapshot: &Snapshot) -> EventStoreResult<()> {
236        let _ = snapshot;
237        Err(EventStoreError::Unsupported)
238    }
239
240    /// Load the latest snapshot for an aggregate, if one exists.
241    ///
242    /// Default returns `Ok(None)` — safe because the caller then replays the
243    /// stream from sequence 0, which is always correct, just slower.
244    async fn load_snapshot(&self, aggregate_id: &str) -> EventStoreResult<Option<Snapshot>> {
245        let _ = aggregate_id;
246        Ok(None)
247    }
248
249    async fn load_snapshot_for(
250        &self,
251        aggregate_type: &str,
252        aggregate_id: &str,
253    ) -> EventStoreResult<Option<Snapshot>> {
254        let _ = aggregate_type;
255        self.load_snapshot(aggregate_id).await
256    }
257}
258
259// ─────────────────────────────────────────────────────────────────────────────
260// In-memory implementation, public for downstream test code.
261// ─────────────────────────────────────────────────────────────────────────────
262
263#[cfg(any(test, feature = "test-utils"))]
264mod in_memory {
265    use super::*;
266    use std::collections::HashMap;
267    use std::sync::Arc;
268    use tokio::sync::Mutex as TokioMutex;
269
270    /// In-memory event store. Available to downstream crates via the
271    /// `test-utils` feature flag.
272    ///
273    /// Validates audit metadata on every append (same contract as production
274    /// stores) so behavior matches what real implementations enforce.
275    #[derive(Clone, Default)]
276    pub struct InMemoryEventStore {
277        events: Arc<TokioMutex<Vec<Event>>>,
278        snapshots: Arc<TokioMutex<HashMap<(String, String), Snapshot>>>,
279    }
280
281    impl InMemoryEventStore {
282        pub fn new() -> Self {
283            Self::default()
284        }
285    }
286
287    #[async_trait]
288    impl EventStore for InMemoryEventStore {
289        async fn append(
290            &self,
291            aggregate_id: &str,
292            version_check: VersionCheck,
293            events: Vec<Event>,
294        ) -> EventStoreResult<()> {
295            validate_audit_batch(aggregate_id, &events)?;
296
297            let mut store = self.events.lock().await;
298
299            let current_version = store
300                .iter()
301                .filter(|e| e.aggregate_id == aggregate_id)
302                .map(|e| e.sequence)
303                .max()
304                .unwrap_or(0);
305
306            if let Some(expected) = version_check.version() {
307                if current_version != expected {
308                    return Err(EventStoreError::ConcurrencyConflict {
309                        aggregate_id: aggregate_id.to_string(),
310                        expected,
311                        actual: current_version,
312                    });
313                }
314            }
315
316            store.extend(events);
317            Ok(())
318        }
319
320        async fn append_to(
321            &self,
322            aggregate_type: &str,
323            aggregate_id: &str,
324            version_check: VersionCheck,
325            events: Vec<Event>,
326        ) -> EventStoreResult<()> {
327            validate_audit_batch(aggregate_id, &events)?;
328            let mut store = self.events.lock().await;
329            let current_version = store
330                .iter()
331                .filter(|event| {
332                    event.aggregate_type == aggregate_type && event.aggregate_id == aggregate_id
333                })
334                .map(|event| event.sequence)
335                .max()
336                .unwrap_or(0);
337
338            if let Some(expected) = version_check.version() {
339                if current_version != expected {
340                    return Err(EventStoreError::ConcurrencyConflict {
341                        aggregate_id: aggregate_id.to_string(),
342                        expected,
343                        actual: current_version,
344                    });
345                }
346            }
347
348            store.extend(events);
349            Ok(())
350        }
351
352        async fn load(&self, aggregate_id: &str) -> EventStoreResult<Vec<Event>> {
353            let store = self.events.lock().await;
354            Ok(store
355                .iter()
356                .filter(|e| e.aggregate_id == aggregate_id)
357                .cloned()
358                .collect())
359        }
360
361        async fn load_stream(
362            &self,
363            aggregate_type: &str,
364            aggregate_id: &str,
365        ) -> EventStoreResult<Vec<Event>> {
366            let store = self.events.lock().await;
367            Ok(store
368                .iter()
369                .filter(|event| {
370                    event.aggregate_type == aggregate_type && event.aggregate_id == aggregate_id
371                })
372                .cloned()
373                .collect())
374        }
375
376        async fn load_from(
377            &self,
378            aggregate_id: &str,
379            from_sequence: i64,
380        ) -> EventStoreResult<Vec<Event>> {
381            let store = self.events.lock().await;
382            Ok(store
383                .iter()
384                .filter(|e| e.aggregate_id == aggregate_id && e.sequence >= from_sequence)
385                .cloned()
386                .collect())
387        }
388
389        async fn load_stream_from(
390            &self,
391            aggregate_type: &str,
392            aggregate_id: &str,
393            from_sequence: i64,
394        ) -> EventStoreResult<Vec<Event>> {
395            let store = self.events.lock().await;
396            Ok(store
397                .iter()
398                .filter(|event| {
399                    event.aggregate_type == aggregate_type
400                        && event.aggregate_id == aggregate_id
401                        && event.sequence >= from_sequence
402                })
403                .cloned()
404                .collect())
405        }
406
407        async fn stream_all(&self, from_position: i64) -> EventStoreResult<Vec<Event>> {
408            let store = self.events.lock().await;
409            Ok(store.iter().skip(from_position as usize).cloned().collect())
410        }
411
412        async fn get_version(&self, aggregate_id: &str) -> EventStoreResult<i64> {
413            let store = self.events.lock().await;
414            Ok(store
415                .iter()
416                .filter(|e| e.aggregate_id == aggregate_id)
417                .map(|e| e.sequence)
418                .max()
419                .unwrap_or(0))
420        }
421
422        async fn get_stream_version(
423            &self,
424            aggregate_type: &str,
425            aggregate_id: &str,
426        ) -> EventStoreResult<i64> {
427            let store = self.events.lock().await;
428            Ok(store
429                .iter()
430                .filter(|event| {
431                    event.aggregate_type == aggregate_type && event.aggregate_id == aggregate_id
432                })
433                .map(|event| event.sequence)
434                .max()
435                .unwrap_or(0))
436        }
437
438        async fn save_snapshot(&self, snapshot: &Snapshot) -> EventStoreResult<()> {
439            let mut snapshots = self.snapshots.lock().await;
440            snapshots.insert(
441                (
442                    snapshot.aggregate_type.clone(),
443                    snapshot.aggregate_id.clone(),
444                ),
445                snapshot.clone(),
446            );
447            Ok(())
448        }
449
450        async fn load_snapshot(&self, aggregate_id: &str) -> EventStoreResult<Option<Snapshot>> {
451            let snapshots = self.snapshots.lock().await;
452            Ok(snapshots
453                .iter()
454                .find(|((_, id), _)| id == aggregate_id)
455                .map(|(_, snapshot)| snapshot.clone()))
456        }
457
458        async fn load_snapshot_for(
459            &self,
460            aggregate_type: &str,
461            aggregate_id: &str,
462        ) -> EventStoreResult<Option<Snapshot>> {
463            let snapshots = self.snapshots.lock().await;
464            Ok(snapshots
465                .get(&(aggregate_type.to_string(), aggregate_id.to_string()))
466                .cloned())
467        }
468    }
469}
470
471#[cfg(any(test, feature = "test-utils"))]
472pub use in_memory::InMemoryEventStore;
473
474#[cfg(test)]
475mod tests {
476    use super::*;
477    use crate::audit::AuditMetadata;
478    use crate::event::Event;
479    use serde_json::json;
480
481    #[test]
482    fn test_version_check_new() {
483        assert_eq!(VersionCheck::New.version(), Some(0));
484    }
485
486    #[test]
487    fn test_version_check_expected() {
488        assert_eq!(VersionCheck::Expected(5).version(), Some(5));
489    }
490
491    #[test]
492    fn test_version_check_auto() {
493        assert_eq!(VersionCheck::Auto.version(), None);
494    }
495
496    #[test]
497    fn test_error_messages() {
498        let error = EventStoreError::ConcurrencyConflict {
499            aggregate_id: "user-123".to_string(),
500            expected: 5,
501            actual: 6,
502        };
503        let msg = error.to_string();
504        assert!(msg.contains("expected version 5"));
505        assert!(msg.contains("aggregate is at version 6"));
506        assert!(msg.contains("user-123"));
507
508        assert!(EventStoreError::database("X").to_string().contains("X"));
509        assert!(EventStoreError::serialization("Y")
510            .to_string()
511            .contains("Y"));
512    }
513
514    #[test]
515    fn test_validate_audit_batch_rejects_pending() {
516        let mut e = Event::new(NewEvent {
517            aggregate_type: "User",
518            aggregate_id: "u1",
519            sequence: 1,
520            event_type: "X",
521            payload: json!({}),
522        });
523        e.audit = AuditMetadata::pending();
524        let err = validate_audit_batch("u1", &[e]).unwrap_err();
525        assert!(matches!(
526            err,
527            EventStoreError::InvalidAudit { event_index: 0, .. }
528        ));
529    }
530
531    #[test]
532    fn test_validate_audit_batch_passes_stamped() {
533        let e = Event::new(NewEvent {
534            aggregate_type: "User",
535            aggregate_id: "u1",
536            sequence: 1,
537            event_type: "X",
538            payload: json!({}),
539        })
540        .with_audit(AuditMetadata::test_default());
541        validate_audit_batch("u1", &[e]).expect("stamped audit must pass");
542    }
543
544    #[tokio::test]
545    async fn test_in_memory_store_rejects_pending_audit() {
546        let store = InMemoryEventStore::new();
547        let e = Event::new(NewEvent {
548            aggregate_type: "User",
549            aggregate_id: "u1",
550            sequence: 1,
551            event_type: "X",
552            payload: json!({}),
553        }); // pending
554        let err = store
555            .append("u1", VersionCheck::New, vec![e])
556            .await
557            .unwrap_err();
558        assert!(matches!(err, EventStoreError::InvalidAudit { .. }));
559    }
560
561    #[tokio::test]
562    async fn test_in_memory_store_persists_stamped_event() {
563        let store = InMemoryEventStore::new();
564        let e = Event::new(NewEvent {
565            aggregate_type: "User",
566            aggregate_id: "u1",
567            sequence: 1,
568            event_type: "X",
569            payload: json!({}),
570        })
571        .with_audit(AuditMetadata::test_default());
572        store
573            .append("u1", VersionCheck::New, vec![e])
574            .await
575            .unwrap();
576        let loaded = store.load("u1").await.unwrap();
577        assert_eq!(loaded.len(), 1);
578    }
579
580    #[tokio::test]
581    async fn test_in_memory_snapshot_save_then_load() {
582        let store = InMemoryEventStore::new();
583        let snap = Snapshot::new("u1", "User", 3, json!({ "name": "Alice" }));
584        store.save_snapshot(&snap).await.unwrap();
585        let loaded = store.load_snapshot("u1").await.unwrap();
586        assert_eq!(loaded, Some(snap));
587    }
588
589    #[tokio::test]
590    async fn test_in_memory_load_snapshot_unknown_returns_none() {
591        let store = InMemoryEventStore::new();
592        assert_eq!(store.load_snapshot("missing").await.unwrap(), None);
593    }
594
595    #[tokio::test]
596    async fn test_in_memory_snapshot_overwrites_on_resave() {
597        let store = InMemoryEventStore::new();
598        store
599            .save_snapshot(&Snapshot::new("u1", "User", 3, json!({ "v": 3 })))
600            .await
601            .unwrap();
602        let newer = Snapshot::new("u1", "User", 9, json!({ "v": 9 }));
603        store.save_snapshot(&newer).await.unwrap();
604        let loaded = store.load_snapshot("u1").await.unwrap().unwrap();
605        assert_eq!(loaded.version, 9);
606        assert_eq!(loaded.state["v"], 9);
607    }
608}
arc_core/event_store.rs

arc_core/
event_store.rs
