es_entity/

events.rs

//! Manage events and related operations for event-sourcing.

use chrono::{DateTime, Utc};

use super::{error::EntityHydrationError, traits::*};

/// An alias for iterator over the persisted events
pub type LastPersisted<'a, E> = std::slice::Iter<'a, PersistedEvent<E>>;

/// Represent the events in raw deserialized format when loaded from database
///
/// Events in the database are stored as JSON blobs and loaded initially as `GenericEvents<Id>` where `Id`
/// belongs to the entity the events is a part of. Acts a bridge between database model and
/// domain model when later converted to the `PersistedEvent` type internally
pub struct GenericEvent<Id> {
    pub entity_id: Id,
    pub sequence: i32,
    pub event: serde_json::Value,
    pub context: Option<crate::ContextData>,
    pub recorded_at: DateTime<Utc>,
}

/// Strongly-typed event wrapper with metadata for successfully stored events.
///
/// Contains the event data along with persistence metadata (sequence, timestamp, entity_id).
/// All `new_events` from [`EntityEvents`] are converted to this structure once persisted to construct
/// entities, enabling event sourcing operations and other database operations.
pub struct PersistedEvent<E: EsEvent> {
    /// The identifier of the entity which the event is used to construct
    pub entity_id: <E as EsEvent>::EntityId,
    /// The timestamp which marks event persistence
    pub recorded_at: DateTime<Utc>,
    /// The sequence number of the event in the event stream
    pub sequence: usize,
    /// The event itself
    pub event: E,
    /// The context when the event was persisted
    /// It is only popluated if 'event_context' set on EsEvent
    pub context: Option<crate::ContextData>,
}

impl<E: Clone + EsEvent> Clone for PersistedEvent<E> {
    fn clone(&self) -> Self {
        PersistedEvent {
            entity_id: self.entity_id.clone(),
            recorded_at: self.recorded_at,
            sequence: self.sequence,
            event: self.event.clone(),
            context: self.context.clone(),
        }
    }
}

pub struct EventWithContext<E: EsEvent> {
    pub event: E,
    pub context: Option<crate::ContextData>,
}

impl<E: Clone + EsEvent> Clone for EventWithContext<E> {
    fn clone(&self) -> Self {
        EventWithContext {
            event: self.event.clone(),
            context: self.context.clone(),
        }
    }
}

/// A [`Vec`] wrapper that manages event-stream of an entity with helpers for event-sourcing operations
///
/// Provides event sourcing operations for loading, appending, and persisting events in chronological
/// sequence. Required field for all event-sourced entities to maintain their state change history.
pub struct EntityEvents<T: EsEvent> {
    /// The entity's id
    pub entity_id: <T as EsEvent>::EntityId,
    /// Events that have been persisted in database and marked
    persisted_events: Vec<PersistedEvent<T>>,
    /// New events that are yet to be persisted to track state changes
    new_events: Vec<EventWithContext<T>>,
}

impl<T: Clone + EsEvent> Clone for EntityEvents<T> {
    fn clone(&self) -> Self {
        Self {
            entity_id: self.entity_id.clone(),
            persisted_events: self.persisted_events.clone(),
            new_events: self.new_events.clone(),
        }
    }
}

impl<T> EntityEvents<T>
where
    T: EsEvent,
{
    /// Initializes a new `EntityEvents` instance with the given entity ID and initial events which is returned by [`IntoEvents`] method
    pub fn init(id: <T as EsEvent>::EntityId, initial_events: impl IntoIterator<Item = T>) -> Self {
        let context = if <T as EsEvent>::event_context() {
            Some(crate::EventContext::data_for_storing())
        } else {
            None
        };
        let new_events = initial_events
            .into_iter()
            .map(|event| EventWithContext {
                event,
                context: context.clone(),
            })
            .collect();
        Self {
            entity_id: id,
            persisted_events: Vec::new(),
            new_events,
        }
    }

    /// Returns a reference to the entity's identifier
    pub fn id(&self) -> &<T as EsEvent>::EntityId {
        &self.entity_id
    }

    /// Returns the timestamp of the first persisted event, indicating when the entity was created
    pub fn entity_first_persisted_at(&self) -> Option<DateTime<Utc>> {
        self.persisted_events.first().map(|e| e.recorded_at)
    }

    /// Returns the timestamp of the last persisted event, indicating when the entity was last modified
    pub fn entity_last_modified_at(&self) -> Option<DateTime<Utc>> {
        self.persisted_events.last().map(|e| e.recorded_at)
    }

    /// Appends a single new event to the entity's event stream to be persisted later
    pub fn push(&mut self, event: T) {
        let context = if <T as EsEvent>::event_context() {
            Some(crate::EventContext::data_for_storing())
        } else {
            None
        };
        self.new_events.push(EventWithContext { event, context });
    }

    /// Appends multiple new events to the entity's event stream to be persisted later
    pub fn extend(&mut self, events: impl IntoIterator<Item = T>) {
        let context = if <T as EsEvent>::event_context() {
            Some(crate::EventContext::data_for_storing())
        } else {
            None
        };
        self.new_events
            .extend(events.into_iter().map(|event| EventWithContext {
                event,
                context: context.clone(),
            }));
    }

    /// Returns true if there are any unpersisted events waiting to be saved
    pub fn any_new(&self) -> bool {
        !self.new_events.is_empty()
    }

    /// Returns the count of persisted events
    pub fn len_persisted(&self) -> usize {
        self.persisted_events.len()
    }

    /// Returns an iterator over all persisted events
    pub fn iter_persisted(&self) -> impl DoubleEndedIterator<Item = &PersistedEvent<T>> {
        self.persisted_events.iter()
    }

    /// Returns an iterator over the last `n` persisted events
    pub fn last_persisted(&self, n: usize) -> LastPersisted<'_, T> {
        let start = self.persisted_events.len() - n;
        self.persisted_events[start..].iter()
    }

    /// Returns an iterator over all events (both persisted and new) in chronological order
    pub fn iter_all(&self) -> impl DoubleEndedIterator<Item = &T> {
        self.persisted_events
            .iter()
            .map(|e| &e.event)
            .chain(self.new_events.iter().map(|e| &e.event))
    }

    /// Loads and reconstructs the first entity from a stream of GenericEvents, marking events as `persisted`.
    ///
    /// Returns `Ok(None)` if no events are present, `Ok(Some(entity))` on success.
    pub fn load_first<E: EsEntity<Event = T>>(
        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
    ) -> Result<Option<E>, EntityHydrationError> {
        let mut current_id = None;
        let mut current = None;
        for e in events {
            if current_id.is_none() {
                current_id = Some(e.entity_id.clone());
                current = Some(Self {
                    entity_id: e.entity_id.clone(),
                    persisted_events: Vec::new(),
                    new_events: Vec::new(),
                });
            }
            if current_id.as_ref() != Some(&e.entity_id) {
                break;
            }
            let cur = current.as_mut().expect("Could not get current");
            cur.persisted_events.push(PersistedEvent {
                entity_id: e.entity_id,
                recorded_at: e.recorded_at,
                sequence: e.sequence as usize,
                event: serde_json::from_value(e.event)?,
                context: e.context,
            });
        }
        if let Some(current) = current {
            Ok(Some(E::try_from_events(current)?))
        } else {
            Ok(None)
        }
    }

    /// Loads and reconstructs up to `n` entities from a stream of GenericEvents.
    /// Assumes the events are grouped by `id` and ordered by `sequence` per `id`.
    ///
    /// Returns both the entities and a flag indicating whether more entities were available in the stream.
    pub fn load_n<E: EsEntity<Event = T>>(
        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
        n: usize,
    ) -> Result<(Vec<E>, bool), EntityHydrationError> {
        let mut ret: Vec<E> = Vec::new();
        let mut current_id = None;
        let mut current = None;
        for e in events {
            if current_id.as_ref() != Some(&e.entity_id) {
                if let Some(current) = current.take() {
                    ret.push(E::try_from_events(current)?);
                    if ret.len() == n {
                        return Ok((ret, true));
                    }
                }

                current_id = Some(e.entity_id.clone());
                current = Some(Self {
                    entity_id: e.entity_id.clone(),
                    persisted_events: Vec::new(),
                    new_events: Vec::new(),
                });
            }
            let cur = current.as_mut().expect("Could not get current");
            cur.persisted_events.push(PersistedEvent {
                entity_id: e.entity_id,
                recorded_at: e.recorded_at,
                sequence: e.sequence as usize,
                event: serde_json::from_value(e.event)?,
                context: e.context,
            });
        }
        if let Some(current) = current.take() {
            ret.push(E::try_from_events(current)?);
        }
        Ok((ret, false))
    }

    #[doc(hidden)]
    pub fn mark_new_events_persisted_at(
        &mut self,
        recorded_at: chrono::DateTime<chrono::Utc>,
    ) -> usize {
        let n = self.new_events.len();
        let offset = self.persisted_events.len() + 1;
        self.persisted_events
            .extend(
                self.new_events
                    .drain(..)
                    .enumerate()
                    .map(|(i, event)| PersistedEvent {
                        entity_id: self.entity_id.clone(),
                        recorded_at,
                        sequence: i + offset,
                        event: event.event,
                        context: event.context,
                    }),
            );
        n
    }

    #[doc(hidden)]
    pub fn serialize_new_events(&self) -> Vec<serde_json::Value> {
        self.new_events
            .iter()
            .map(|event| serde_json::to_value(&event.event).expect("Failed to serialize event"))
            .collect()
    }

    #[doc(hidden)]
    pub fn serialize_new_event_contexts(&self) -> Option<Vec<crate::ContextData>> {
        if <T as EsEvent>::event_context() {
            let contexts = self
                .new_events
                .iter()
                .map(|event| event.context.clone().expect("Missing context"))
                .collect();

            Some(contexts)
        } else {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use uuid::Uuid;

    #[derive(Debug, serde::Serialize, serde::Deserialize)]
    enum DummyEntityEvent {
        Created(String),
    }

    impl EsEvent for DummyEntityEvent {
        type EntityId = Uuid;
        fn event_context() -> bool {
            true
        }
    }

    struct DummyEntity {
        name: String,

        events: EntityEvents<DummyEntityEvent>,
    }

    impl EsEntity for DummyEntity {
        type Event = DummyEntityEvent;
        type New = NewDummyEntity;

        fn events_mut(&mut self) -> &mut EntityEvents<DummyEntityEvent> {
            &mut self.events
        }
        fn events(&self) -> &EntityEvents<DummyEntityEvent> {
            &self.events
        }
    }

    impl TryFromEvents<DummyEntityEvent> for DummyEntity {
        fn try_from_events(
            events: EntityEvents<DummyEntityEvent>,
        ) -> Result<Self, EntityHydrationError> {
            let name = events
                .iter_persisted()
                .map(|e| match &e.event {
                    DummyEntityEvent::Created(name) => name.clone(),
                })
                .next()
                .expect("Could not find name");
            Ok(Self { name, events })
        }
    }

    struct NewDummyEntity {}

    impl IntoEvents<DummyEntityEvent> for NewDummyEntity {
        fn into_events(self) -> EntityEvents<DummyEntityEvent> {
            EntityEvents::init(
                Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap(),
                vec![DummyEntityEvent::Created("".to_owned())],
            )
        }
    }

    #[test]
    fn load_zero_events() {
        let generic_events = vec![];
        let res = EntityEvents::load_first::<DummyEntity>(generic_events);
        assert!(matches!(res, Ok(None)));
    }

    #[test]
    fn load_first() {
        let generic_events = vec![GenericEvent {
            entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
            sequence: 1,
            event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
                .expect("Could not serialize"),
            context: None,
            recorded_at: chrono::Utc::now(),
        }];
        let entity: DummyEntity = EntityEvents::load_first(generic_events)
            .expect("Could not load")
            .expect("No entity found");
        assert!(entity.name == "dummy-name");
    }

    #[test]
    fn load_n() {
        let generic_events = vec![
            GenericEvent {
                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap(),
                sequence: 1,
                event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
                    .expect("Could not serialize"),
                context: None,
                recorded_at: chrono::Utc::now(),
            },
            GenericEvent {
                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000003").unwrap(),
                sequence: 1,
                event: serde_json::to_value(DummyEntityEvent::Created("other-name".to_owned()))
                    .expect("Could not serialize"),
                context: None,
                recorded_at: chrono::Utc::now(),
            },
        ];
        let (entity, more): (Vec<DummyEntity>, _) =
            EntityEvents::load_n(generic_events, 2).expect("Could not load");
        assert!(!more);
        assert_eq!(entity.len(), 2);
    }
}