es_entity/

events.rs

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

use chrono::{DateTime, Utc};

use super::{
    error::EntityHydrationError,
    snapshot::{EsSnapshot, NoSnapshot, Replay, SnapshotRecord},
    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>,
    pub forgettable_payload: Option<serde_json::Value>,
}

/// Row type of every loader on a `snapshot` repo (aliased as `Repo__DbEvent`
/// there). `GenericEvent<Id>` itself is untouched — every existing repo's
/// `query_as!` and `.sqlx` entry stay as is; this confines the wider row
/// shape to snapshot repos only.
pub struct SnapshotGenericEvent<Id> {
    pub entity_id: Id,
    pub sequence: i32,
    pub event: Option<serde_json::Value>,
    pub context: Option<crate::ContextData>,
    pub recorded_at: Option<DateTime<Utc>>,
    pub forgettable_payload: Option<serde_json::Value>,
    pub snapshot: Option<serde_json::Value>,
    pub snapshot_sequence: Option<i32>,
    pub snapshot_recorded_at: Option<DateTime<Utc>>,
    pub snapshot_first_recorded_at: Option<DateTime<Utc>>,
    pub snapshot_forgettable_payload: Option<serde_json::Value>,
}

/// Internal common shape loaders normalise into before decoding a container.
#[doc(hidden)]
pub struct HydrationRow<Id> {
    pub entity_id: Id,
    pub sequence: i32,
    pub event: Option<serde_json::Value>,
    pub context: Option<crate::ContextData>,
    pub recorded_at: Option<DateTime<Utc>>,
    pub forgettable_payload: Option<serde_json::Value>,
    pub snapshot: Option<serde_json::Value>,
    pub snapshot_sequence: Option<i32>,
    pub snapshot_recorded_at: Option<DateTime<Utc>>,
    pub snapshot_first_recorded_at: Option<DateTime<Utc>>,
    pub snapshot_forgettable_payload: Option<serde_json::Value>,
}

impl<Id> From<GenericEvent<Id>> for HydrationRow<Id> {
    fn from(e: GenericEvent<Id>) -> Self {
        Self {
            entity_id: e.entity_id,
            sequence: e.sequence,
            event: Some(e.event),
            context: e.context,
            recorded_at: Some(e.recorded_at),
            forgettable_payload: e.forgettable_payload,
            snapshot: None,
            snapshot_sequence: None,
            snapshot_recorded_at: None,
            snapshot_first_recorded_at: None,
            snapshot_forgettable_payload: None,
        }
    }
}

impl<Id> From<SnapshotGenericEvent<Id>> for HydrationRow<Id> {
    fn from(e: SnapshotGenericEvent<Id>) -> Self {
        Self {
            entity_id: e.entity_id,
            sequence: e.sequence,
            event: e.event,
            context: e.context,
            recorded_at: e.recorded_at,
            forgettable_payload: e.forgettable_payload,
            snapshot: e.snapshot,
            snapshot_sequence: e.snapshot_sequence,
            snapshot_recorded_at: e.snapshot_recorded_at,
            snapshot_first_recorded_at: e.snapshot_first_recorded_at,
            snapshot_forgettable_payload: e.snapshot_forgettable_payload,
        }
    }
}

/// 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, S: EsSnapshot = NoSnapshot> {
    /// The entity's id
    pub entity_id: <T as EsEvent>::EntityId,
    /// Sequence of the last event folded into `snapshot` (0 when none). The
    /// tail (`persisted_events`) starts at `base_sequence + 1`.
    base_sequence: usize,
    /// The loaded snapshot, if any.
    snapshot: Option<SnapshotRecord<S>>,
    /// Events that have been persisted in database and marked (the tail,
    /// after the snapshot).
    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, S: EsSnapshot + Clone> Clone for EntityEvents<T, S> {
    fn clone(&self) -> Self {
        Self {
            entity_id: self.entity_id.clone(),
            base_sequence: self.base_sequence,
            snapshot: self.snapshot.clone(),
            persisted_events: self.persisted_events.clone(),
            new_events: self.new_events.clone(),
        }
    }
}

/// Rebuilds an entity from its container and checks that the entity's own
/// id is the container's. An `Initialized { id }` payload is user data on
/// every path, so nothing else ties it to the id the events are stored
/// under; a mismatch is corrupt state.
#[doc(hidden)]
pub fn hydrate_checked<E>(
    events: EntityEvents<E::Event, E::Snapshot>,
) -> Result<E, EntityHydrationError>
where
    E: EsEntity,
{
    let container_id = events.entity_id.clone();
    let entity = E::try_from_events(events)?;
    if *entity.__entity_id() != container_id {
        return Err(EntityHydrationError::IdMismatch);
    }
    Ok(entity)
}

impl<T, S> EntityEvents<T, S>
where
    T: EsEvent,
    S: EsSnapshot,
{
    /// Builds a container for a brand-new entity. Only
    /// [`EntityEventsBuilder`](crate::EntityEventsBuilder) reaches it, so the
    /// container's id is always the one the repo's id source resolved.
    pub(crate) 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,
            base_sequence: 0,
            snapshot: None,
            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.snapshot
            .as_ref()
            .map(|s| s.first_recorded_at)
            .or_else(|| 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)
            .or_else(|| self.snapshot.as_ref().map(|s| s.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, including those folded into the
    /// snapshot. This is the OCC offset used by every write path.
    pub fn len_persisted(&self) -> usize {
        self.base_sequence + self.persisted_events.len()
    }

    /// Returns the count of events after the snapshot: the persisted tail
    /// plus any events staged for the write in flight. This is what
    /// `HeadSnapshot::capture()` thresholds on.
    pub fn tail_len(&self) -> usize {
        self.persisted_events.len() + self.new_events.len()
    }

    #[doc(hidden)]
    pub fn len_new(&self) -> usize {
        self.new_events.len()
    }

    /// Returns the loaded snapshot, if any.
    pub fn snapshot(&self) -> Option<&SnapshotRecord<S>> {
        self.snapshot.as_ref()
    }

    /// Returns an iterator over the last `n` persisted events after the
    /// snapshot, if any. This is what post-persist hooks receive.
    pub fn last_persisted(&self, n: usize) -> LastPersisted<'_, T> {
        let start = self.persisted_events.len().saturating_sub(n);
        self.persisted_events[start..].iter()
    }

    /// Returns an iterator over the full replay: the snapshot (if any),
    /// first, followed by the tail and any new events, in chronological
    /// order.
    pub fn replay(&self) -> impl DoubleEndedIterator<Item = Replay<'_, T, S>> + Clone {
        self.snapshot
            .iter()
            .map(|r| Replay::Snapshot(&r.state))
            .chain(
                self.persisted_events
                    .iter()
                    .map(|e| Replay::Event(&e.event)),
            )
            .chain(self.new_events.iter().map(|e| Replay::Event(&e.event)))
    }

    /// Like [`replay`][Self::replay] but only over persisted state: the
    /// snapshot record (if any) followed by the persisted tail. No new
    /// events.
    pub fn replay_persisted(
        &self,
    ) -> impl DoubleEndedIterator<Item = Replay<'_, PersistedEvent<T>, SnapshotRecord<S>>> + Clone
    {
        self.snapshot
            .iter()
            .map(Replay::Snapshot)
            .chain(self.persisted_events.iter().map(Replay::Event))
    }

    /// Compacts the tail into a snapshot after a write that took one: the
    /// persisted tail is dropped and replaced by the given state at the
    /// current head. An entity in memory then looks exactly like a reload.
    #[doc(hidden)]
    pub fn compact_to_snapshot(
        &mut self,
        state: S,
        recorded_at: DateTime<Utc>,
        first_recorded_at: DateTime<Utc>,
    ) {
        debug_assert!(self.new_events.is_empty());
        let sequence = self.len_persisted();
        self.persisted_events.clear();
        self.base_sequence = sequence;
        self.snapshot = Some(SnapshotRecord {
            sequence,
            state,
            recorded_at,
            first_recorded_at,
        });
    }

    /// Applies one hydration row (snapshot fields and/or an event) to this
    /// container. Shared by `load_first` and `load_n`.
    fn apply_hydration_row(
        &mut self,
        row: HydrationRow<<T as EsEvent>::EntityId>,
    ) -> Result<(), EntityHydrationError> {
        if let Some(mut snapshot_json) = row.snapshot
            && S::IS_SNAPSHOT
        {
            if let Some(payload) = row.snapshot_forgettable_payload {
                crate::forgettable::inject_forgettable_payload(&mut snapshot_json, payload);
            }
            let sequence = row
                .snapshot_sequence
                .expect("snapshot row missing snapshot_sequence");
            let state: S = serde_json::from_value(snapshot_json)
                .map_err(|source| EntityHydrationError::SnapshotDecode { sequence, source })?;
            self.base_sequence = sequence as usize;
            self.snapshot = Some(SnapshotRecord {
                sequence: sequence as usize,
                state,
                recorded_at: row
                    .snapshot_recorded_at
                    .expect("snapshot row missing snapshot_recorded_at"),
                first_recorded_at: row
                    .snapshot_first_recorded_at
                    .expect("snapshot row missing snapshot_first_recorded_at"),
            });
        }

        match row.event {
            Some(mut event_json) => {
                if self.persisted_events.is_empty()
                    && self.snapshot.is_some()
                    && row.sequence as usize != self.base_sequence + 1
                {
                    return Err(EntityHydrationError::SnapshotGap {
                        snapshot_sequence: self.base_sequence as i32,
                        next_event_sequence: row.sequence,
                    });
                }
                if let Some(payload) = row.forgettable_payload {
                    crate::forgettable::inject_forgettable_payload(&mut event_json, payload);
                }
                self.persisted_events.push(PersistedEvent {
                    entity_id: row.entity_id,
                    recorded_at: row.recorded_at.expect("event row missing recorded_at"),
                    sequence: row.sequence as usize,
                    event: serde_json::from_value(event_json)?,
                    context: row.context,
                });
                Ok(())
            }
            None if self.snapshot.is_some() => Ok(()),
            None => Err(EntityHydrationError::NoEvents),
        }
    }

    /// Loads and reconstructs the first entity from a stream of hydration
    /// rows, marking events as `persisted`.
    ///
    /// Returns `Ok(None)` if no events are present, `Ok(Some(entity))` on success.
    pub fn load_first<E>(
        events: impl IntoIterator<Item = impl Into<HydrationRow<<T as EsEvent>::EntityId>>>,
    ) -> Result<Option<E>, EntityHydrationError>
    where
        E: EsEntity<Event = T, Snapshot = S>,
    {
        let mut current_id = None;
        let mut current: Option<Self> = None;
        for e in events {
            let row: HydrationRow<<T as EsEvent>::EntityId> = e.into();
            if current_id.is_none() {
                current_id = Some(row.entity_id.clone());
                current = Some(Self {
                    entity_id: row.entity_id.clone(),
                    base_sequence: 0,
                    snapshot: None,
                    persisted_events: Vec::new(),
                    new_events: Vec::new(),
                });
            }
            if current_id.as_ref() != Some(&row.entity_id) {
                break;
            }
            let cur = current.as_mut().expect("Could not get current");
            cur.apply_hydration_row(row)?;
        }
        if let Some(current) = current {
            Ok(Some(hydrate_checked(current)?))
        } else {
            Ok(None)
        }
    }

    /// Loads and reconstructs up to `n` entities from a stream of hydration
    /// rows. Assumes the rows 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>(
        events: impl IntoIterator<Item = impl Into<HydrationRow<<T as EsEvent>::EntityId>>>,
        n: usize,
    ) -> Result<(Vec<E>, bool), EntityHydrationError>
    where
        E: EsEntity<Event = T, Snapshot = S>,
    {
        if n == 0 {
            // Asking for zero entities yields zero entities. `has_more` reports
            // whether the stream was non-empty, mirroring the `LIMIT n + 1`
            // over-fetch contract the generated repos rely on: callers query
            // `LIMIT (first + 1)`, so a non-empty stream means a next page exists.
            let has_more = events.into_iter().next().is_some();
            return Ok((Vec::new(), has_more));
        }
        let mut ret: Vec<E> = Vec::new();
        let mut current_id = None;
        let mut current: Option<Self> = None;
        for e in events {
            let row: HydrationRow<<T as EsEvent>::EntityId> = e.into();
            if current_id.as_ref() != Some(&row.entity_id) {
                if let Some(current) = current.take() {
                    ret.push(hydrate_checked(current)?);
                    if ret.len() == n {
                        return Ok((ret, true));
                    }
                }

                current_id = Some(row.entity_id.clone());
                current = Some(Self {
                    entity_id: row.entity_id.clone(),
                    base_sequence: 0,
                    snapshot: None,
                    persisted_events: Vec::new(),
                    new_events: Vec::new(),
                });
            }
            let cur = current.as_mut().expect("Could not get current");
            cur.apply_hydration_row(row)?;
        }
        if let Some(current) = current.take() {
            ret.push(hydrate_checked(current)?);
        }
        Ok((ret, false))
    }

    #[doc(hidden)]
    pub fn iter_new_events(&self) -> impl Iterator<Item = &EventWithContext<T>> {
        self.new_events.iter()
    }

    #[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.len_persisted() + 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 new_event_types(&self) -> Vec<String> {
        self.new_events
            .iter()
            .map(|event| event.event.event_type().to_string())
            .collect()
    }

    #[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()
    }

    /// Forgets all forgettable payloads in persisted events and returns the taken events.
    ///
    /// Applies `forget_fn` to each persisted event, then takes ownership of the event
    /// stream, leaving `self` as an empty shell. The returned `EntityEvents` can be passed
    /// to `TryFromEvents::try_from_events` to rebuild the entity with forgotten fields.
    ///
    /// Only used by non-snapshot repos: snapshot repos rebuild via a
    /// full-history reload (the in-memory tail lacks events folded into the
    /// snapshot).
    #[doc(hidden)]
    pub fn forget_and_take(&mut self, mut forget_fn: impl FnMut(&mut T)) -> Self {
        for persisted in &mut self.persisted_events {
            forget_fn(&mut persisted.event);
        }
        let entity_id = self.entity_id.clone();
        std::mem::replace(
            self,
            Self {
                entity_id,
                base_sequence: 0,
                snapshot: None,
                persisted_events: Vec::new(),
                new_events: Vec::new(),
            },
        )
    }

    #[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
        }
    }
}

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

    /// Returns an iterator over all events (both persisted and new) in
    /// chronological order.
    ///
    /// Only exists for `EntityEvents<T, NoSnapshot>` — a switch to a real
    /// snapshot type is a compile error at every such scan, rather than a
    /// silently-wrong fold that skips whatever the snapshot summarised:
    ///
    /// ```compile_fail,E0599
    /// use es_entity::*;
    /// use serde::{Serialize, Deserialize};
    ///
    /// es_entity::entity_id! { IterAllMeterId }
    ///
    /// #[derive(EsEvent, Debug, Serialize, Deserialize)]
    /// #[serde(tag = "type", rename_all = "snake_case")]
    /// #[es_event(id = "IterAllMeterId")]
    /// pub enum IterAllMeterEvent {
    ///     Initialized { id: IterAllMeterId },
    /// }
    ///
    /// #[derive(EsSnapshot, Debug, Clone, Serialize, Deserialize)]
    /// #[es_snapshot(version = 1)]
    /// pub struct IterAllMeterSnapshot {
    ///     pub id: IterAllMeterId,
    /// }
    ///
    /// fn count_all(events: &EntityEvents<IterAllMeterEvent, IterAllMeterSnapshot>) -> usize {
    ///     // error[E0599]: no method named `iter_all` on this type — it is
    ///     // only defined for `EntityEvents<T, NoSnapshot>`. Use `replay()`,
    ///     // which forces every match to account for `Replay::Snapshot`.
    ///     events.iter_all().count()
    /// }
    /// ```
    pub fn iter_all(&self) -> impl DoubleEndedIterator<Item = &T> + Clone {
        self.persisted_events
            .iter()
            .map(|e| &e.event)
            .chain(self.new_events.iter().map(|e| &e.event))
    }

    /// Widens a freshly-initialized container into one carrying a real
    /// snapshot type. A brand-new entity always has no snapshot, so this
    /// conversion is exact regardless of `S`.
    #[doc(hidden)]
    pub fn widen_snapshot<S: EsSnapshot>(self) -> EntityEvents<T, S> {
        EntityEvents {
            entity_id: self.entity_id,
            base_sequence: 0,
            snapshot: None,
            persisted_events: self.persisted_events,
            new_events: self.new_events,
        }
    }
}

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

    /// Builds a `GenericEvent` whose JSON deserializes to `Created(name)`.
    fn valid_event(id: Uuid, sequence: i32, name: &str) -> GenericEvent<Uuid> {
        GenericEvent {
            entity_id: id,
            sequence,
            event: serde_json::to_value(DummyEntityEvent::Created(name.to_string()))
                .expect("could not serialize"),
            context: None,
            recorded_at: chrono::Utc::now(),
            forgettable_payload: None,
        }
    }

    /// Small bounded JSON strategy covering null/bool/int/float/string plus one
    /// level of array/object nesting. Cheap to shrink and enough to exercise the
    /// serde error paths without ballooning runtime.
    fn json_value() -> impl Strategy<Value = serde_json::Value> {
        let scalar = prop_oneof![
            Just(serde_json::Value::Null),
            any::<bool>().prop_map(serde_json::Value::Bool),
            any::<i64>().prop_map(serde_json::Value::from),
            any::<f64>().prop_map(serde_json::Value::from),
            ".{0,15}".prop_map(serde_json::Value::String),
        ]
        .boxed();
        let nested = prop_oneof![
            proptest::collection::vec(scalar.clone(), 0..4).prop_map(serde_json::Value::Array),
            proptest::collection::vec((".{0,6}", scalar.clone()), 0..4).prop_map(|pairs| {
                let mut m = serde_json::Map::new();
                for (k, v) in pairs {
                    m.insert(k, v);
                }
                serde_json::Value::Object(m)
            },),
        ];
        prop_oneof![scalar, nested]
    }

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

    impl EsEvent for DummyEntityEvent {
        type EntityId = Uuid;
        fn event_context() -> bool {
            true
        }
        fn event_type(&self) -> &'static str {
            match self {
                Self::Created(_) => "created",
            }
        }
    }

    struct DummyEntity {
        name: String,

        events: EntityEvents<DummyEntityEvent>,
    }

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

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

    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 {
        type IdSource = crate::FromCaller;

        fn into_events(
            self,
            builder: crate::EntityEventsBuilder<DummyEntityEvent, crate::FromCaller>,
        ) -> EntityEvents<DummyEntityEvent> {
            builder.initial_events(
                Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap(),
                vec![DummyEntityEvent::Created("".to_owned())],
            )
        }
    }

    #[test]
    fn load_zero_events() {
        let generic_events: Vec<GenericEvent<Uuid>> = 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(),
            forgettable_payload: None,
        }];
        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(),
                forgettable_payload: None,
            },
            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(),
                forgettable_payload: None,
            },
        ];
        let (entity, more): (Vec<DummyEntity>, _) =
            EntityEvents::load_n(generic_events, 2).expect("Could not load");
        assert!(!more);
        assert_eq!(entity.len(), 2);
    }

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

        // n == len works
        assert_eq!(events.last_persisted(1).count(), 1);
        // n > len clamps to all persisted events instead of underflowing
        assert_eq!(events.last_persisted(10).count(), 1);
        // n == 0 yields nothing
        assert_eq!(events.last_persisted(0).count(), 0);
    }

    proptest! {
        #[test]
        fn load_first_empty_input_returns_none(_ in Just(())) {
            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(
                Vec::<GenericEvent<Uuid>>::new(),
            );
            prop_assert!(matches!(res, Ok(None)));
        }

        #[test]
        fn load_first_non_empty_valid_hydrates(
            count in 1u8..8,
            names in proptest::collection::vec(".{0,10}", 1..8),
        ) {
            let id = Uuid::nil();
            let events: Vec<_> = (1..=count as i32)
                .zip(names.iter())
                .map(|(s, n)| valid_event(id, s, n))
                .collect();
            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events);
            prop_assert!(matches!(res, Ok(Some(_))));
        }

        /// Feeds arbitrary JSON through the loader. It may hydrate or error, but
        /// it must never panic — and `Ok(None)` is only possible for empty input.
        #[test]
        fn load_first_arbitrary_json_never_panics(
            raw in proptest::collection::vec(json_value(), 0..10),
        ) {
            let events: Vec<_> = raw
                .into_iter()
                .enumerate()
                .map(|(i, v)| GenericEvent {
                    entity_id: Uuid::nil(),
                    sequence: i as i32,
                    event: v,
                    context: None,
                    recorded_at: chrono::Utc::now(),
                    forgettable_payload: None,
                })
                .collect();
            let is_empty = events.is_empty();
            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events);
            if let Ok(None) = res {
                prop_assert!(is_empty);
            }
        }

        /// With a grouped, ascending stream (the documented precondition) and
        /// `n >= 1`, `load_n` returns `min(n, k)` entities and reports `has_more`
        /// exactly when `n < k`.
        #[test]
        fn load_n_respects_limit_and_more_flag(
            k in 1u8..8,
            per in 1u8..4,
            n in 1u8..12,
        ) {
            let mut events = Vec::new();
            for i in 0..k {
                let id = Uuid::from_u128(i as u128);
                for s in 1..=per as i32 {
                    events.push(valid_event(id, s, &format!("e{i}-{s}")));
                }
            }
            let (entities, has_more) =
                EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, n as usize)
                    .expect("valid events hydrate");
            prop_assert_eq!(entities.len(), (n as usize).min(k as usize));
            prop_assert_eq!(has_more, n < k);
        }

        /// Regression for the `n = 0` degenerate case: previously returned *all*
        /// entities instead of zero. Now returns none and reports `has_more` iff
        /// the stream was non-empty (the `LIMIT n + 1` contract).
        #[test]
        fn load_n_zero_returns_no_entities(k in 0u8..5, per in 1u8..3) {
            let mut events = Vec::new();
            for i in 0..k {
                let id = Uuid::from_u128(i as u128);
                for s in 1..=per as i32 {
                    events.push(valid_event(id, s, &format!("e{i}-{s}")));
                }
            }
            let (entities, has_more) =
                EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, 0)
                    .expect("valid events hydrate");
            prop_assert!(entities.is_empty());
            prop_assert_eq!(has_more, k > 0);
        }

        #[test]
        fn load_n_arbitrary_json_never_panics(
            raw in proptest::collection::vec(json_value(), 0..10),
            n in 0u8..12,
        ) {
            let events: Vec<_> = raw
                .into_iter()
                .enumerate()
                .map(|(i, v)| GenericEvent {
                    entity_id: Uuid::nil(),
                    sequence: i as i32,
                    event: v,
                    context: None,
                    recorded_at: chrono::Utc::now(),
                    forgettable_payload: None,
                })
                .collect();
            let _ = EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, n as usize);
        }

        /// `last_persisted(n)` must clamp to the available events for any `n`,
        /// generalizing the dedicated saturating-sub test above.
        #[test]
        fn last_persisted_clamps_for_any_n(
            p in 1u8..8,
            n in 0u16..12,
        ) {
            let id = Uuid::nil();
            let events: Vec<_> = (1..=p as i32)
                .map(|s| valid_event(id, s, &format!("n{s}")))
                .collect();
            let entity: DummyEntity =
                EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events)
                    .expect("load")
                    .expect("some");
            let count = entity.events().last_persisted(n as usize).count();
            prop_assert_eq!(count, (n as usize).min(p as usize));
        }

        /// Marking new events as persisted must drain `new_events`, leave
        /// `len_persisted` consistent, and assign contiguous 1-based sequences
        /// across repeated calls.
        #[test]
        fn mark_new_events_assigns_contiguous_sequences(
            a in 0u8..5,
            b in 0u8..5,
        ) {
            let id = Uuid::nil();
            let mut events = EntityEvents::init(
                id,
                (0..a).map(|i| DummyEntityEvent::Created(format!("n{i}"))),
            );
            let now = chrono::Utc::now();

            prop_assert_eq!(events.mark_new_events_persisted_at(now), a as usize);
            prop_assert!(!events.any_new());
            prop_assert_eq!(events.len_persisted(), a as usize);
            let seqs: Vec<usize> = events.iter_persisted().map(|e| e.sequence).collect();
            prop_assert_eq!(seqs, (1..=a as usize).collect::<Vec<_>>());

            for i in 0..b {
                events.push(DummyEntityEvent::Created(format!("m{i}")));
            }
            if b > 0 {
                prop_assert!(events.any_new());
            }
            prop_assert_eq!(events.mark_new_events_persisted_at(now), b as usize);
            prop_assert!(!events.any_new());
            prop_assert_eq!(events.len_persisted(), (a + b) as usize);
            let seqs: Vec<usize> = events.iter_persisted().map(|e| e.sequence).collect();
            prop_assert_eq!(seqs, (1..=(a + b) as usize).collect::<Vec<_>>());
        }
    }
}