umadb_dcb/
lib.rs

1//! API for Dynamic Consistency Boundaries (DCB) event store
2//!
3//! This module provides the core interfaces and data structures for working with
4//! an event store that supports dynamic consistency boundaries.
5
6use async_trait::async_trait;
7use futures_core::Stream;
8use futures_util::StreamExt;
9use std::iter::Iterator;
10use thiserror::Error;
11use uuid::Uuid;
12
13/// Non-async Rust interface for recording and retrieving events
14pub trait DCBEventStoreSync {
15    /// Reads events from the store based on the provided query and constraints
16    ///
17    /// Returns a DCBReadResponseSync that provides an iterator over all events,
18    /// unless 'from' is given then only those with position greater than 'after',
19    /// and unless any query items are given, then only those that match at least one
20    /// query item. An event matches a query item if its type is in the item types or
21    /// there are no item types, and if all the item tags are in the event tags.
22    fn read(
23        &self,
24        query: Option<DCBQuery>,
25        start: Option<u64>,
26        backwards: bool,
27        limit: Option<u32>,
28        subscribe: bool,
29    ) -> DCBResult<Box<dyn DCBReadResponseSync + 'static>>;
30
31    /// Reads events from the store and returns them as a tuple of (Vec<DCBSequencedEvent>, Option<u64>)
32    fn read_with_head(
33        &self,
34        query: Option<DCBQuery>,
35        start: Option<u64>,
36        backwards: bool,
37        limit: Option<u32>,
38    ) -> DCBResult<(Vec<DCBSequencedEvent>, Option<u64>)> {
39        let mut response = self.read(query, start, backwards, limit, false)?;
40        response.collect_with_head()
41    }
42
43    /// Returns the current head position of the event store, or None if empty
44    ///
45    /// Returns the value of last_committed_position, or None if last_committed_position is zero
46    fn head(&self) -> DCBResult<Option<u64>>;
47
48    /// Appends given events to the event store, unless the condition fails
49    ///
50    /// Returns the position of the last appended event
51    fn append(
52        &self,
53        events: Vec<DCBEvent>,
54        condition: Option<DCBAppendCondition>,
55    ) -> DCBResult<u64>;
56}
57
58/// Response from a read operation, providing an iterator over sequenced events
59pub trait DCBReadResponseSync: Iterator<Item = Result<DCBSequencedEvent, DCBError>> {
60    /// Returns the current head position of the event store, or None if empty
61    fn head(&mut self) -> DCBResult<Option<u64>>;
62    /// Returns a vector of events with head
63    fn collect_with_head(&mut self) -> DCBResult<(Vec<DCBSequencedEvent>, Option<u64>)>;
64    /// Returns a batch of events, updating head with the last event in the batch if there is one and if limit.is_some() is true
65    fn next_batch(&mut self) -> DCBResult<Vec<DCBSequencedEvent>>;
66}
67
68/// Async Rust interface for recording and retrieving events
69#[async_trait]
70pub trait DCBEventStoreAsync: Send + Sync {
71    /// Reads events from the store based on the provided query and constraints
72    ///
73    /// Returns a DCBReadResponseSync that provides an iterator over all events,
74    /// unless 'after' is given then only those with position greater than 'after',
75    /// and unless any query items are given, then only those that match at least one
76    /// query item. An event matches a query item if its type is in the item types or
77    /// there are no item types, and if all the item tags are in the event tags.
78    async fn read<'a>(
79        &'a self,
80        query: Option<DCBQuery>,
81        start: Option<u64>,
82        backwards: bool,
83        limit: Option<u32>,
84        subscribe: bool,
85    ) -> DCBResult<Box<dyn DCBReadResponseAsync + Send + 'static>>;
86
87    /// Reads events from the store and returns them as a tuple of (Vec<DCBSequencedEvent>, Option<u64>)
88    async fn read_with_head<'a>(
89        &'a self,
90        query: Option<DCBQuery>,
91        after: Option<u64>,
92        backwards: bool,
93        limit: Option<u32>,
94    ) -> DCBResult<(Vec<DCBSequencedEvent>, Option<u64>)> {
95        let mut response = self.read(query, after, backwards, limit, false).await?;
96        response.collect_with_head().await
97    }
98
99    /// Returns the current head position of the event store, or None if empty
100    ///
101    /// Returns the value of last_committed_position, or None if last_committed_position is zero
102    async fn head(&self) -> DCBResult<Option<u64>>;
103
104    /// Appends given events to the event store, unless the condition fails
105    ///
106    /// Returns the position of the last appended event
107    async fn append(
108        &self,
109        events: Vec<DCBEvent>,
110        condition: Option<DCBAppendCondition>,
111    ) -> DCBResult<u64>;
112}
113
114/// Asynchronous response from a read operation, providing a stream of sequenced events
115#[async_trait]
116pub trait DCBReadResponseAsync: Stream<Item = DCBResult<DCBSequencedEvent>> + Send + Unpin {
117    async fn head(&mut self) -> DCBResult<Option<u64>>;
118
119    async fn collect_with_head(&mut self) -> DCBResult<(Vec<DCBSequencedEvent>, Option<u64>)> {
120        let mut events = Vec::new();
121        while let Some(result) = self.next().await {
122            events.push(result?); // propagate error from stream
123        }
124
125        let head = self.head().await?;
126        Ok((events, head))
127    }
128
129    async fn next_batch(&mut self) -> DCBResult<Vec<DCBSequencedEvent>>;
130}
131
132/// Represents a query item for filtering events
133#[derive(Debug, Clone, Default)]
134pub struct DCBQueryItem {
135    /// Event types to match
136    pub types: Vec<String>,
137    /// Tags that must all be present in the event
138    pub tags: Vec<String>,
139}
140
141impl DCBQueryItem {
142    /// Creates a new query item
143    pub fn new() -> Self {
144        Self {
145            types: vec![],
146            tags: vec![],
147        }
148    }
149
150    /// Sets the types for this query item
151    pub fn types<I, S>(mut self, types: I) -> Self
152    where
153        I: IntoIterator<Item = S>,
154        S: Into<String>,
155    {
156        self.types = types.into_iter().map(|s| s.into()).collect();
157        self
158    }
159
160    /// Sets the tags for this query item
161    pub fn tags<I, S>(mut self, tags: I) -> Self
162    where
163        I: IntoIterator<Item = S>,
164        S: Into<String>,
165    {
166        self.tags = tags.into_iter().map(|s| s.into()).collect();
167        self
168    }
169}
170
171/// A query composed of multiple query items
172#[derive(Debug, Clone, Default)]
173pub struct DCBQuery {
174    /// List of query items, where events matching any item are included in results
175    pub items: Vec<DCBQueryItem>,
176}
177
178impl DCBQuery {
179    /// Creates a new empty query
180    pub fn new() -> Self {
181        Self { items: Vec::new() }
182    }
183
184    /// Creates a query with the specified items
185    pub fn with_items<I>(items: I) -> Self
186    where
187        I: IntoIterator<Item = DCBQueryItem>,
188    {
189        Self {
190            items: items.into_iter().collect(),
191        }
192    }
193
194    /// Adds a query item to this query
195    pub fn item(mut self, item: DCBQueryItem) -> Self {
196        self.items.push(item);
197        self
198    }
199
200    /// Adds multiple query items to this query
201    pub fn items<I>(mut self, items: I) -> Self
202    where
203        I: IntoIterator<Item = DCBQueryItem>,
204    {
205        self.items.extend(items);
206        self
207    }
208}
209
210/// Conditions that must be satisfied for an append operation to succeed
211#[derive(Debug, Clone, Default)]
212pub struct DCBAppendCondition {
213    /// Query that, if matching any events, will cause the append to fail
214    pub fail_if_events_match: DCBQuery,
215    /// Position after which to append; if None, append at the end
216    pub after: Option<u64>,
217}
218
219impl DCBAppendCondition {
220    /// Creates a new empty append condition
221    pub fn new(fail_if_events_match: DCBQuery) -> Self {
222        Self {
223            fail_if_events_match,
224            after: None,
225        }
226    }
227
228    pub fn after(mut self, after: Option<u64>) -> Self {
229        self.after = after;
230        self
231    }
232}
233
234/// Represents an event in the event store
235#[derive(Debug, Clone)]
236pub struct DCBEvent {
237    /// Type of the event
238    pub event_type: String,
239    /// Binary data associated with the event
240    pub data: Vec<u8>,
241    /// Tags associated with the event
242    pub tags: Vec<String>,
243    /// Unique event ID
244    pub uuid: Option<Uuid>,
245}
246
247impl DCBEvent {
248    /// Creates a new event
249    pub fn new() -> Self {
250        Self {
251            event_type: "".to_string(),
252            data: Vec::new(),
253            tags: Vec::new(),
254            uuid: None,
255        }
256    }
257
258    /// Sets the type for this event
259    pub fn event_type<S: Into<String>>(mut self, event_type: S) -> Self {
260        self.event_type = event_type.into();
261        self
262    }
263
264    /// Sets the data for this event
265    pub fn data<D: Into<Vec<u8>>>(mut self, data: D) -> Self {
266        self.data = data.into();
267        self
268    }
269
270    /// Sets the tags for this event
271    pub fn tags<I, S>(mut self, tags: I) -> Self
272    where
273        I: IntoIterator<Item = S>,
274        S: Into<String>,
275    {
276        self.tags = tags.into_iter().map(|s| s.into()).collect();
277        self
278    }
279
280    /// Sets the UUID for this event
281    pub fn uuid(mut self, uuid: Uuid) -> Self {
282        self.uuid = Some(uuid);
283        self
284    }
285}
286
287/// An event with its position in the event sequence
288#[derive(Debug, Clone)]
289pub struct DCBSequencedEvent {
290    /// The event
291    pub event: DCBEvent,
292    /// Position of the event in the sequence
293    pub position: u64,
294}
295
296// Error types
297#[derive(Error, Debug)]
298pub enum DCBError {
299    // Generic/system errors
300    #[error("IO error: {0}")]
301    Io(#[from] std::io::Error),
302
303    // DCB domain errors
304    #[error("Integrity error: condition failed: {0}")]
305    IntegrityError(String),
306    #[error("Corruption detected: {0}")]
307    Corruption(String),
308
309    // LMDB/Storage domain errors (unified into DCBError)
310    #[error("Page not found: {0:?}")]
311    PageNotFound(u64),
312    #[error("Dirty page not found: {0:?}")]
313    DirtyPageNotFound(u64),
314    #[error("Root ID mismatched: old {0:?} new {1:?}")]
315    RootIDMismatch(u64, u64),
316    #[error("Database corrupted: {0}")]
317    DatabaseCorrupted(String),
318    #[error("Internal error: {0}")]
319    InternalError(String),
320    #[error("Serialization error: {0}")]
321    SerializationError(String),
322    #[error("Deserialization error: {0}")]
323    DeserializationError(String),
324    #[error("Page already freed: {0:?}")]
325    PageAlreadyFreed(u64),
326    #[error("Page already dirty: {0:?}")]
327    PageAlreadyDirty(u64),
328    #[error("Transport error: {0}")]
329    TransportError(String),
330    #[error("Cancelled by user")]
331    CancelledByUser(),
332}
333
334pub type DCBResult<T> = Result<T, DCBError>;
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339
340    // A simple implementation of DCBReadResponseSync for testing
341    struct TestReadResponse {
342        events: Vec<DCBSequencedEvent>,
343        current_index: usize,
344        head_position: Option<u64>,
345    }
346
347    impl TestReadResponse {
348        fn new(events: Vec<DCBSequencedEvent>, head_position: Option<u64>) -> Self {
349            Self {
350                events,
351                current_index: 0,
352                head_position,
353            }
354        }
355    }
356
357    impl Iterator for TestReadResponse {
358        type Item = Result<DCBSequencedEvent, DCBError>;
359
360        fn next(&mut self) -> Option<Self::Item> {
361            if self.current_index < self.events.len() {
362                let event = self.events[self.current_index].clone();
363                self.current_index += 1;
364                Some(Ok(event))
365            } else {
366                None
367            }
368        }
369    }
370
371    impl DCBReadResponseSync for TestReadResponse {
372        fn head(&mut self) -> DCBResult<Option<u64>> {
373            Ok(self.head_position)
374        }
375
376        fn collect_with_head(&mut self) -> DCBResult<(Vec<DCBSequencedEvent>, Option<u64>)> {
377            todo!()
378        }
379
380        fn next_batch(&mut self) -> DCBResult<Vec<DCBSequencedEvent>> {
381            let mut batch = Vec::new();
382            while let Some(result) = self.next() {
383                match result {
384                    Ok(event) => batch.push(event),
385                    Err(err) => {
386                        panic!("{}", err);
387                    }
388                }
389            }
390            Ok(batch)
391        }
392    }
393
394    #[test]
395    fn test_dcb_read_response() {
396        // Create some test events
397        let event1 = DCBEvent {
398            event_type: "test_event".to_string(),
399            data: vec![1, 2, 3],
400            tags: vec!["tag1".to_string(), "tag2".to_string()],
401            uuid: None,
402        };
403
404        let event2 = DCBEvent {
405            event_type: "another_event".to_string(),
406            data: vec![4, 5, 6],
407            tags: vec!["tag2".to_string(), "tag3".to_string()],
408            uuid: None,
409        };
410
411        let seq_event1 = DCBSequencedEvent {
412            event: event1,
413            position: 1,
414        };
415
416        let seq_event2 = DCBSequencedEvent {
417            event: event2,
418            position: 2,
419        };
420
421        // Create a test response
422        let mut response =
423            TestReadResponse::new(vec![seq_event1.clone(), seq_event2.clone()], Some(2));
424
425        // Test head position
426        assert_eq!(response.head().unwrap(), Some(2));
427
428        // Test iterator functionality
429        assert_eq!(response.next().unwrap().unwrap().position, 1);
430        assert_eq!(response.next().unwrap().unwrap().position, 2);
431        assert!(response.next().is_none());
432    }
433
434
435    #[test]
436    fn test_event_new() {
437        let event1 = DCBEvent::new().event_type("type1").data(b"data1").tags(["tagX"]);
438
439        println!("Event created with builder API:");
440        println!("  event_type: {}", event1.event_type);
441        println!("  data: {:?}", event1.data);
442        println!("  tags: {:?}", event1.tags);
443        println!("  uuid: {:?}", event1.uuid);
444
445        // Verify the fields match expectations
446        assert_eq!(event1.event_type, "type1");
447        assert_eq!(event1.data, b"data1".to_vec());
448        assert_eq!(event1.tags, vec!["tagX".to_string()]);
449        assert_eq!(event1.uuid, None);
450
451        // Test with multiple tags
452        let event2 = DCBEvent::new().event_type("type2").data(b"data2").tags(["tag1", "tag2", "tag3"]);
453        assert_eq!(event2.tags.len(), 3);
454
455        // Test without data or tags
456        let event3 = DCBEvent::new().event_type("type3");
457        assert_eq!(event3.data.len(), 0);
458        assert_eq!(event3.tags.len(), 0);
459
460        // Test DCBQueryItem builder
461        let query_item = DCBQueryItem::new().types(["type1", "type2"]).tags(["tagA", "tagB"]);
462        assert_eq!(query_item.types.len(), 2);
463        assert_eq!(query_item.tags.len(), 2);
464
465        // Test DCBQuery builder
466        let query = DCBQuery::new().item(query_item);
467        assert_eq!(query.items.len(), 1);
468
469        println!("\nAll builder API tests passed!");
470    }
471
472}
umadb_dcb/lib.rs

umadb_dcb/
lib.rs
