oxirs_stream/event_sourcing/
mod.rs

1//! # Event Sourcing Framework
2//!
3//! Complete event sourcing implementation for OxiRS Stream providing event storage,
4//! replay capabilities, snapshots, and temporal queries. This forms the foundation
5//! for CQRS patterns and enables advanced temporal analytics.
6
7pub mod rdf_store_mod;
8mod simple;
9mod store;
10
11#[cfg(test)]
12mod simple_tests;
13#[cfg(test)]
14mod tests;
15
16// Re-export all public types from sub-modules
17pub use rdf_store_mod as rdf_store;
18pub use simple::{
19    EventStreamIter, ProjectionRunner, SimpleEvent, SimpleEventBus, SimpleEventHandler,
20    SimpleEventStore, SimpleSnapshot, SimpleSnapshotStore,
21};
22pub use store::{EventIndexes, EventMetadataAccessor, EventStore, PersistenceManager};
23
24use crate::StreamEvent;
25use chrono::{DateTime, Duration as ChronoDuration, Utc};
26use serde::{Deserialize, Serialize};
27use std::collections::HashMap;
28use std::sync::atomic::AtomicU64;
29use uuid::Uuid;
30
31/// Event store configuration
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct EventStoreConfig {
34    /// Maximum events to keep in memory
35    pub max_memory_events: usize,
36    /// Enable persistent storage
37    pub enable_persistence: bool,
38    /// Persistence backend type
39    pub persistence_backend: PersistenceBackend,
40    /// Snapshot configuration
41    pub snapshot_config: SnapshotConfig,
42    /// Retention policy
43    pub retention_policy: RetentionPolicy,
44    /// Indexing configuration
45    pub indexing_config: IndexingConfig,
46    /// Enable compression for stored events
47    pub enable_compression: bool,
48    /// Batch size for persistence operations
49    pub persistence_batch_size: usize,
50}
51
52impl Default for EventStoreConfig {
53    fn default() -> Self {
54        Self {
55            max_memory_events: 1_000_000,
56            enable_persistence: true,
57            persistence_backend: PersistenceBackend::FileSystem {
58                base_path: std::env::temp_dir()
59                    .join("oxirs-event-store")
60                    .to_string_lossy()
61                    .into_owned(),
62            },
63            snapshot_config: SnapshotConfig::default(),
64            retention_policy: RetentionPolicy::default(),
65            indexing_config: IndexingConfig::default(),
66            enable_compression: true,
67            persistence_batch_size: 1000,
68        }
69    }
70}
71
72/// Persistence backend options
73#[derive(Debug, Clone, Serialize, Deserialize)]
74pub enum PersistenceBackend {
75    /// File system based storage
76    FileSystem { base_path: String },
77    /// Database storage
78    Database { connection_string: String },
79    /// S3-compatible object storage
80    ObjectStorage {
81        endpoint: String,
82        bucket: String,
83        access_key: String,
84        secret_key: String,
85    },
86    /// In-memory only (no persistence)
87    Memory,
88}
89
90/// Snapshot configuration
91#[derive(Debug, Clone, Serialize, Deserialize)]
92pub struct SnapshotConfig {
93    /// Enable automatic snapshots
94    pub enable_snapshots: bool,
95    /// Snapshot interval (number of events)
96    pub snapshot_interval: usize,
97    /// Maximum snapshots to keep
98    pub max_snapshots: usize,
99    /// Snapshot compression
100    pub compress_snapshots: bool,
101}
102
103impl Default for SnapshotConfig {
104    fn default() -> Self {
105        Self {
106            enable_snapshots: true,
107            snapshot_interval: 10000,
108            max_snapshots: 10,
109            compress_snapshots: true,
110        }
111    }
112}
113
114/// Event retention policy
115#[derive(Debug, Clone, Serialize, Deserialize)]
116pub struct RetentionPolicy {
117    /// Maximum age of events to keep
118    pub max_age: Option<ChronoDuration>,
119    /// Maximum number of events to keep
120    pub max_events: Option<u64>,
121    /// Archive old events instead of deleting
122    pub enable_archiving: bool,
123    /// Archive backend
124    pub archive_backend: Option<PersistenceBackend>,
125}
126
127impl Default for RetentionPolicy {
128    fn default() -> Self {
129        Self {
130            max_age: Some(ChronoDuration::days(365)), // 1 year
131            max_events: Some(10_000_000),             // 10M events
132            enable_archiving: true,
133            archive_backend: None,
134        }
135    }
136}
137
138/// Indexing configuration
139#[derive(Debug, Clone, Serialize, Deserialize)]
140pub struct IndexingConfig {
141    /// Enable event type indexing
142    pub index_by_event_type: bool,
143    /// Enable timestamp indexing
144    pub index_by_timestamp: bool,
145    /// Enable source indexing
146    pub index_by_source: bool,
147    /// Enable custom field indexing
148    pub custom_indexes: Vec<CustomIndex>,
149}
150
151impl Default for IndexingConfig {
152    fn default() -> Self {
153        Self {
154            index_by_event_type: true,
155            index_by_timestamp: true,
156            index_by_source: true,
157            custom_indexes: Vec::new(),
158        }
159    }
160}
161
162/// Custom index definition
163#[derive(Debug, Clone, Serialize, Deserialize)]
164pub struct CustomIndex {
165    /// Index name
166    pub name: String,
167    /// Field path to index
168    pub field_path: String,
169    /// Index type
170    pub index_type: IndexType,
171}
172
173/// Index type
174#[derive(Debug, Clone, Serialize, Deserialize)]
175pub enum IndexType {
176    /// Hash index for exact matches
177    Hash,
178    /// B-tree index for range queries
179    BTree,
180    /// Full-text search index
181    FullText,
182}
183
184/// Stored event with metadata
185#[derive(Debug, Clone, Serialize, Deserialize)]
186pub struct StoredEvent {
187    /// Unique event ID
188    pub event_id: Uuid,
189    /// Event sequence number (global order)
190    pub sequence_number: u64,
191    /// Stream ID (for grouping related events)
192    pub stream_id: String,
193    /// Event version within the stream
194    pub stream_version: u64,
195    /// Original event data
196    pub event_data: StreamEvent,
197    /// Storage timestamp
198    pub stored_at: DateTime<Utc>,
199    /// Storage metadata
200    pub storage_metadata: StorageMetadata,
201}
202
203/// Storage metadata
204#[derive(Debug, Clone, Serialize, Deserialize)]
205pub struct StorageMetadata {
206    /// Checksum for integrity verification
207    pub checksum: String,
208    /// Compressed size (if compressed)
209    pub compressed_size: Option<usize>,
210    /// Original size
211    pub original_size: usize,
212    /// Storage location
213    pub storage_location: String,
214    /// Persistence status
215    pub persistence_status: PersistenceStatus,
216}
217
218/// Persistence status
219#[derive(Debug, Clone, Serialize, Deserialize)]
220pub enum PersistenceStatus {
221    /// Only in memory
222    InMemory,
223    /// Persisted to disk
224    Persisted,
225    /// Archived to long-term storage
226    Archived,
227    /// Failed to persist
228    Failed { error: String },
229}
230
231/// Event stream snapshot
232#[derive(Debug, Clone, Serialize, Deserialize)]
233pub struct EventSnapshot {
234    /// Snapshot ID
235    pub snapshot_id: Uuid,
236    /// Stream ID
237    pub stream_id: String,
238    /// Stream version at snapshot time
239    pub stream_version: u64,
240    /// Sequence number at snapshot time
241    pub sequence_number: u64,
242    /// Snapshot timestamp
243    pub created_at: DateTime<Utc>,
244    /// Aggregated state data
245    pub state_data: Vec<u8>,
246    /// Snapshot metadata
247    pub metadata: SnapshotMetadata,
248}
249
250/// Snapshot metadata
251#[derive(Debug, Clone, Serialize, Deserialize)]
252pub struct SnapshotMetadata {
253    /// Compression algorithm used
254    pub compression: Option<String>,
255    /// Original state size
256    pub original_size: usize,
257    /// Compressed size
258    pub compressed_size: usize,
259    /// Checksum for integrity
260    pub checksum: String,
261}
262
263/// Query criteria for event retrieval
264#[derive(Debug, Clone, Serialize, Deserialize)]
265pub struct EventQuery {
266    /// Stream ID filter
267    pub stream_id: Option<String>,
268    /// Event type filter
269    pub event_types: Option<Vec<String>>,
270    /// Time range filter
271    pub time_range: Option<TimeRange>,
272    /// Sequence number range
273    pub sequence_range: Option<SequenceRange>,
274    /// Source filter
275    pub source: Option<String>,
276    /// Custom field filters
277    pub custom_filters: HashMap<String, String>,
278    /// Maximum number of events to return
279    pub limit: Option<usize>,
280    /// Ordering preference
281    pub order: QueryOrder,
282}
283
284/// Time range for queries
285#[derive(Debug, Clone, Serialize, Deserialize)]
286pub struct TimeRange {
287    pub start: DateTime<Utc>,
288    pub end: DateTime<Utc>,
289}
290
291/// Sequence number range for queries
292#[derive(Debug, Clone, Serialize, Deserialize)]
293pub struct SequenceRange {
294    pub start: u64,
295    pub end: u64,
296}
297
298/// Query ordering
299#[derive(Debug, Clone, Serialize, Deserialize)]
300pub enum QueryOrder {
301    /// Ascending by sequence number
302    SequenceAsc,
303    /// Descending by sequence number
304    SequenceDesc,
305    /// Ascending by timestamp
306    TimestampAsc,
307    /// Descending by timestamp
308    TimestampDesc,
309}
310
311/// Event sourcing statistics
312#[derive(Debug, Default)]
313pub struct EventSourcingStats {
314    pub total_events_stored: AtomicU64,
315    pub total_events_retrieved: AtomicU64,
316    pub snapshots_created: AtomicU64,
317    pub events_archived: AtomicU64,
318    pub persistence_operations: AtomicU64,
319    pub failed_operations: AtomicU64,
320    pub memory_usage_bytes: AtomicU64,
321    pub disk_usage_bytes: AtomicU64,
322    pub average_store_latency_ms: AtomicU64,
323    pub average_retrieve_latency_ms: AtomicU64,
324}
325
326/// EventStore trait for abstracting event storage
327#[async_trait::async_trait]
328pub trait EventStoreTrait: Send + Sync {
329    async fn store_event(
330        &self,
331        stream_id: String,
332        event: StreamEvent,
333    ) -> anyhow::Result<StoredEvent>;
334    async fn query_events(&self, query: EventQuery) -> anyhow::Result<Vec<StoredEvent>>;
335    async fn get_stream_events(
336        &self,
337        stream_id: &str,
338        from_version: Option<u64>,
339    ) -> anyhow::Result<Vec<StoredEvent>>;
340    async fn replay_from_timestamp(
341        &self,
342        timestamp: DateTime<Utc>,
343    ) -> anyhow::Result<Vec<StoredEvent>>;
344    async fn get_latest_snapshot(&self, stream_id: &str) -> anyhow::Result<Option<EventSnapshot>>;
345    async fn rebuild_stream_state(&self, stream_id: &str) -> anyhow::Result<Vec<u8>>;
346    async fn append_events(
347        &self,
348        aggregate_id: &str,
349        events: &[StreamEvent],
350        expected_version: Option<u64>,
351    ) -> anyhow::Result<u64>;
352}
353
354/// Event stream trait for streaming events
355#[async_trait::async_trait]
356pub trait EventStream: Send + Sync {
357    async fn next_event(&mut self) -> Option<StoredEvent>;
358    async fn has_events(&self) -> bool;
359    async fn read_events_from_position(
360        &self,
361        position: u64,
362        max_events: usize,
363    ) -> anyhow::Result<Vec<StoredEvent>>;
364}
365
366/// Snapshot store trait for managing snapshots
367#[async_trait::async_trait]
368pub trait SnapshotStore: Send + Sync {
369    async fn store_snapshot(&self, snapshot: EventSnapshot) -> anyhow::Result<()>;
370    async fn get_snapshot(
371        &self,
372        stream_id: &str,
373        version: Option<u64>,
374    ) -> anyhow::Result<Option<EventSnapshot>>;
375    async fn list_snapshots(&self, stream_id: &str) -> anyhow::Result<Vec<EventSnapshot>>;
376}
377
378/// Persistence operation
379#[derive(Debug, Clone)]
380pub enum PersistenceOperation {
381    /// Store event
382    StoreEvent(Box<StoredEvent>),
383    /// Store snapshot
384    StoreSnapshot(EventSnapshot),
385    /// Archive events
386    ArchiveEvents(Vec<StoredEvent>),
387    /// Delete events
388    DeleteEvents(Vec<u64>),
389}
390
391/// Persistence statistics
392#[derive(Debug, Default)]
393pub struct PersistenceStats {
394    pub operations_queued: AtomicU64,
395    pub operations_completed: AtomicU64,
396    pub operations_failed: AtomicU64,
397    pub bytes_written: AtomicU64,
398    pub bytes_read: AtomicU64,
399}
oxirs_stream/event_sourcing/mod.rs

oxirs_stream/event_sourcing/
mod.rs
