es_entity/
events.rs

1//! Manage events and related operations for event-sourcing.
2
3use chrono::{DateTime, Utc};
4
5use super::{error::EsEntityError, traits::*};
6
7/// An alias for iterator over the persisted events
8pub type LastPersisted<'a, E> = std::slice::Iter<'a, PersistedEvent<E>>;
9
10/// Represent the events in raw deserialized format when loaded from database
11///
12/// Events in the database are stored as JSON blobs and loaded initially as `GenericEvents<Id>` where `Id`
13/// belongs to the entity the events is a part of. Acts a bridge between database model and
14/// domain model when later converted to the `PersistedEvent` type internally
15pub struct GenericEvent<Id> {
16    pub entity_id: Id,
17    pub sequence: i32,
18    pub event: serde_json::Value,
19    pub context: Option<crate::ContextData>,
20    pub recorded_at: DateTime<Utc>,
21}
22
23/// Strongly-typed event wrapper with metadata for successfully stored events.
24///
25/// Contains the event data along with persistence metadata (sequence, timestamp, entity_id).
26/// All `new_events` from [`EntityEvents`] are converted to this structure once persisted to construct
27/// entities, enabling event sourcing operations and other database operations.
28pub struct PersistedEvent<E: EsEvent> {
29    /// The identifier of the entity which the event is used to construct
30    pub entity_id: <E as EsEvent>::EntityId,
31    /// The timestamp which marks event persistence
32    pub recorded_at: DateTime<Utc>,
33    /// The sequence number of the event in the event stream
34    pub sequence: usize,
35    /// The event itself
36    pub event: E,
37    /// The context when the event was persisted
38    /// It is only popluated if 'event_context' set on EsEvent
39    pub context: Option<crate::ContextData>,
40}
41
42impl<E: Clone + EsEvent> Clone for PersistedEvent<E> {
43    fn clone(&self) -> Self {
44        PersistedEvent {
45            entity_id: self.entity_id.clone(),
46            recorded_at: self.recorded_at,
47            sequence: self.sequence,
48            event: self.event.clone(),
49            context: self.context.clone(),
50        }
51    }
52}
53
54pub struct EventWithContext<E: EsEvent> {
55    pub event: E,
56    pub context: Option<crate::ContextData>,
57}
58
59impl<E: Clone + EsEvent> Clone for EventWithContext<E> {
60    fn clone(&self) -> Self {
61        EventWithContext {
62            event: self.event.clone(),
63            context: self.context.clone(),
64        }
65    }
66}
67
68/// A [`Vec`] wrapper that manages event-stream of an entity with helpers for event-sourcing operations
69///
70/// Provides event sourcing operations for loading, appending, and persisting events in chronological
71/// sequence. Required field for all event-sourced entities to maintain their state change history.
72pub struct EntityEvents<T: EsEvent> {
73    /// The entity's id
74    pub entity_id: <T as EsEvent>::EntityId,
75    /// Events that have been persisted in database and marked
76    persisted_events: Vec<PersistedEvent<T>>,
77    /// New events that are yet to be persisted to track state changes
78    new_events: Vec<EventWithContext<T>>,
79}
80
81impl<T: Clone + EsEvent> Clone for EntityEvents<T> {
82    fn clone(&self) -> Self {
83        Self {
84            entity_id: self.entity_id.clone(),
85            persisted_events: self.persisted_events.clone(),
86            new_events: self.new_events.clone(),
87        }
88    }
89}
90
91impl<T> EntityEvents<T>
92where
93    T: EsEvent,
94{
95    /// Initializes a new `EntityEvents` instance with the given entity ID and initial events which is returned by [`IntoEvents`] method
96    pub fn init(id: <T as EsEvent>::EntityId, initial_events: impl IntoIterator<Item = T>) -> Self {
97        let context = if <T as EsEvent>::event_context() {
98            Some(crate::EventContext::data_for_storing())
99        } else {
100            None
101        };
102        let new_events = initial_events
103            .into_iter()
104            .map(|event| EventWithContext {
105                event,
106                context: context.clone(),
107            })
108            .collect();
109        Self {
110            entity_id: id,
111            persisted_events: Vec::new(),
112            new_events,
113        }
114    }
115
116    /// Returns a reference to the entity's identifier
117    pub fn id(&self) -> &<T as EsEvent>::EntityId {
118        &self.entity_id
119    }
120
121    /// Returns the timestamp of the first persisted event, indicating when the entity was created
122    pub fn entity_first_persisted_at(&self) -> Option<DateTime<Utc>> {
123        self.persisted_events.first().map(|e| e.recorded_at)
124    }
125
126    /// Returns the timestamp of the last persisted event, indicating when the entity was last modified
127    pub fn entity_last_modified_at(&self) -> Option<DateTime<Utc>> {
128        self.persisted_events.last().map(|e| e.recorded_at)
129    }
130
131    /// Appends a single new event to the entity's event stream to be persisted later
132    pub fn push(&mut self, event: T) {
133        let context = if <T as EsEvent>::event_context() {
134            Some(crate::EventContext::data_for_storing())
135        } else {
136            None
137        };
138        self.new_events.push(EventWithContext { event, context });
139    }
140
141    /// Appends multiple new events to the entity's event stream to be persisted later
142    pub fn extend(&mut self, events: impl IntoIterator<Item = T>) {
143        let context = if <T as EsEvent>::event_context() {
144            Some(crate::EventContext::data_for_storing())
145        } else {
146            None
147        };
148        self.new_events
149            .extend(events.into_iter().map(|event| EventWithContext {
150                event,
151                context: context.clone(),
152            }));
153    }
154
155    /// Returns true if there are any unpersisted events waiting to be saved
156    pub fn any_new(&self) -> bool {
157        !self.new_events.is_empty()
158    }
159
160    /// Returns the count of persisted events
161    pub fn len_persisted(&self) -> usize {
162        self.persisted_events.len()
163    }
164
165    /// Returns an iterator over all persisted events
166    pub fn iter_persisted(&self) -> impl DoubleEndedIterator<Item = &PersistedEvent<T>> {
167        self.persisted_events.iter()
168    }
169
170    /// Returns an iterator over the last `n` persisted events
171    pub fn last_persisted(&self, n: usize) -> LastPersisted<'_, T> {
172        let start = self.persisted_events.len() - n;
173        self.persisted_events[start..].iter()
174    }
175
176    /// Returns an iterator over all events (both persisted and new) in chronological order
177    pub fn iter_all(&self) -> impl DoubleEndedIterator<Item = &T> {
178        self.persisted_events
179            .iter()
180            .map(|e| &e.event)
181            .chain(self.new_events.iter().map(|e| &e.event))
182    }
183
184    /// Loads and reconstructs the first entity from a stream of GenericEvents, marking events as `persisted`
185    pub fn load_first<E: EsEntity<Event = T>>(
186        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
187    ) -> Result<E, EsEntityError> {
188        let mut current_id = None;
189        let mut current = None;
190        for e in events {
191            if current_id.is_none() {
192                current_id = Some(e.entity_id.clone());
193                current = Some(Self {
194                    entity_id: e.entity_id.clone(),
195                    persisted_events: Vec::new(),
196                    new_events: Vec::new(),
197                });
198            }
199            if current_id.as_ref() != Some(&e.entity_id) {
200                break;
201            }
202            let cur = current.as_mut().expect("Could not get current");
203            cur.persisted_events.push(PersistedEvent {
204                entity_id: e.entity_id,
205                recorded_at: e.recorded_at,
206                sequence: e.sequence as usize,
207                event: serde_json::from_value(e.event)?,
208                context: e.context,
209            });
210        }
211        if let Some(current) = current {
212            E::try_from_events(current)
213        } else {
214            Err(EsEntityError::NotFound)
215        }
216    }
217
218    /// Loads and reconstructs up to `n` entities from a stream of GenericEvents.
219    /// Assumes the events are grouped by `id` and ordered by `sequence` per `id`.
220    ///
221    /// Returns both the entities and a flag indicating whether more entities were available in the stream.
222    pub fn load_n<E: EsEntity<Event = T>>(
223        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
224        n: usize,
225    ) -> Result<(Vec<E>, bool), EsEntityError> {
226        let mut ret: Vec<E> = Vec::new();
227        let mut current_id = None;
228        let mut current = None;
229        for e in events {
230            if current_id.as_ref() != Some(&e.entity_id) {
231                if let Some(current) = current.take() {
232                    ret.push(E::try_from_events(current)?);
233                    if ret.len() == n {
234                        return Ok((ret, true));
235                    }
236                }
237
238                current_id = Some(e.entity_id.clone());
239                current = Some(Self {
240                    entity_id: e.entity_id.clone(),
241                    persisted_events: Vec::new(),
242                    new_events: Vec::new(),
243                });
244            }
245            let cur = current.as_mut().expect("Could not get current");
246            cur.persisted_events.push(PersistedEvent {
247                entity_id: e.entity_id,
248                recorded_at: e.recorded_at,
249                sequence: e.sequence as usize,
250                event: serde_json::from_value(e.event)?,
251                context: e.context,
252            });
253        }
254        if let Some(current) = current.take() {
255            ret.push(E::try_from_events(current)?);
256        }
257        Ok((ret, false))
258    }
259
260    #[doc(hidden)]
261    pub fn mark_new_events_persisted_at(
262        &mut self,
263        recorded_at: chrono::DateTime<chrono::Utc>,
264    ) -> usize {
265        let n = self.new_events.len();
266        let offset = self.persisted_events.len() + 1;
267        self.persisted_events
268            .extend(
269                self.new_events
270                    .drain(..)
271                    .enumerate()
272                    .map(|(i, event)| PersistedEvent {
273                        entity_id: self.entity_id.clone(),
274                        recorded_at,
275                        sequence: i + offset,
276                        event: event.event,
277                        context: event.context,
278                    }),
279            );
280        n
281    }
282
283    #[doc(hidden)]
284    pub fn serialize_new_events(&self) -> Vec<serde_json::Value> {
285        self.new_events
286            .iter()
287            .map(|event| serde_json::to_value(&event.event).expect("Failed to serialize event"))
288            .collect()
289    }
290
291    #[doc(hidden)]
292    pub fn serialize_new_event_contexts(&self) -> Option<Vec<crate::ContextData>> {
293        if <T as EsEvent>::event_context() {
294            let contexts = self
295                .new_events
296                .iter()
297                .map(|event| event.context.clone().expect("Missing context"))
298                .collect();
299
300            Some(contexts)
301        } else {
302            None
303        }
304    }
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310    use uuid::Uuid;
311
312    #[derive(Debug, serde::Serialize, serde::Deserialize)]
313    enum DummyEntityEvent {
314        Created(String),
315    }
316
317    impl EsEvent for DummyEntityEvent {
318        type EntityId = Uuid;
319        fn event_context() -> bool {
320            true
321        }
322    }
323
324    struct DummyEntity {
325        name: String,
326
327        events: EntityEvents<DummyEntityEvent>,
328    }
329
330    impl EsEntity for DummyEntity {
331        type Event = DummyEntityEvent;
332        type New = NewDummyEntity;
333
334        fn events_mut(&mut self) -> &mut EntityEvents<DummyEntityEvent> {
335            &mut self.events
336        }
337        fn events(&self) -> &EntityEvents<DummyEntityEvent> {
338            &self.events
339        }
340    }
341
342    impl TryFromEvents<DummyEntityEvent> for DummyEntity {
343        fn try_from_events(events: EntityEvents<DummyEntityEvent>) -> Result<Self, EsEntityError> {
344            let name = events
345                .iter_persisted()
346                .map(|e| match &e.event {
347                    DummyEntityEvent::Created(name) => name.clone(),
348                })
349                .next()
350                .expect("Could not find name");
351            Ok(Self { name, events })
352        }
353    }
354
355    struct NewDummyEntity {}
356
357    impl IntoEvents<DummyEntityEvent> for NewDummyEntity {
358        fn into_events(self) -> EntityEvents<DummyEntityEvent> {
359            EntityEvents::init(
360                Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap(),
361                vec![DummyEntityEvent::Created("".to_owned())],
362            )
363        }
364    }
365
366    #[test]
367    fn load_zero_events() {
368        let generic_events = vec![];
369        let res = EntityEvents::load_first::<DummyEntity>(generic_events);
370        assert!(matches!(res, Err(EsEntityError::NotFound)));
371    }
372
373    #[test]
374    fn load_first() {
375        let generic_events = vec![GenericEvent {
376            entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
377            sequence: 1,
378            event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
379                .expect("Could not serialize"),
380            context: None,
381            recorded_at: chrono::Utc::now(),
382        }];
383        let entity: DummyEntity = EntityEvents::load_first(generic_events).expect("Could not load");
384        assert!(entity.name == "dummy-name");
385    }
386
387    #[test]
388    fn load_n() {
389        let generic_events = vec![
390            GenericEvent {
391                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap(),
392                sequence: 1,
393                event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
394                    .expect("Could not serialize"),
395                context: None,
396                recorded_at: chrono::Utc::now(),
397            },
398            GenericEvent {
399                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000003").unwrap(),
400                sequence: 1,
401                event: serde_json::to_value(DummyEntityEvent::Created("other-name".to_owned()))
402                    .expect("Could not serialize"),
403                context: None,
404                recorded_at: chrono::Utc::now(),
405            },
406        ];
407        let (entity, more): (Vec<DummyEntity>, _) =
408            EntityEvents::load_n(generic_events, 2).expect("Could not load");
409        assert!(!more);
410        assert_eq!(entity.len(), 2);
411    }
412}
es_entity/events.rs

es_entity/
events.rs
