es_entity/
events.rs

1//! Manage events and related operations for event-sourcing.
2
3use chrono::{DateTime, Utc};
4
5use super::{error::EntityHydrationError, 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    pub forgettable_payload: Option<serde_json::Value>,
22}
23
24/// Strongly-typed event wrapper with metadata for successfully stored events.
25///
26/// Contains the event data along with persistence metadata (sequence, timestamp, entity_id).
27/// All `new_events` from [`EntityEvents`] are converted to this structure once persisted to construct
28/// entities, enabling event sourcing operations and other database operations.
29pub struct PersistedEvent<E: EsEvent> {
30    /// The identifier of the entity which the event is used to construct
31    pub entity_id: <E as EsEvent>::EntityId,
32    /// The timestamp which marks event persistence
33    pub recorded_at: DateTime<Utc>,
34    /// The sequence number of the event in the event stream
35    pub sequence: usize,
36    /// The event itself
37    pub event: E,
38    /// The context when the event was persisted
39    /// It is only popluated if 'event_context' set on EsEvent
40    pub context: Option<crate::ContextData>,
41}
42
43impl<E: Clone + EsEvent> Clone for PersistedEvent<E> {
44    fn clone(&self) -> Self {
45        PersistedEvent {
46            entity_id: self.entity_id.clone(),
47            recorded_at: self.recorded_at,
48            sequence: self.sequence,
49            event: self.event.clone(),
50            context: self.context.clone(),
51        }
52    }
53}
54
55pub struct EventWithContext<E: EsEvent> {
56    pub event: E,
57    pub context: Option<crate::ContextData>,
58}
59
60impl<E: Clone + EsEvent> Clone for EventWithContext<E> {
61    fn clone(&self) -> Self {
62        EventWithContext {
63            event: self.event.clone(),
64            context: self.context.clone(),
65        }
66    }
67}
68
69/// A [`Vec`] wrapper that manages event-stream of an entity with helpers for event-sourcing operations
70///
71/// Provides event sourcing operations for loading, appending, and persisting events in chronological
72/// sequence. Required field for all event-sourced entities to maintain their state change history.
73pub struct EntityEvents<T: EsEvent> {
74    /// The entity's id
75    pub entity_id: <T as EsEvent>::EntityId,
76    /// Events that have been persisted in database and marked
77    persisted_events: Vec<PersistedEvent<T>>,
78    /// New events that are yet to be persisted to track state changes
79    new_events: Vec<EventWithContext<T>>,
80}
81
82impl<T: Clone + EsEvent> Clone for EntityEvents<T> {
83    fn clone(&self) -> Self {
84        Self {
85            entity_id: self.entity_id.clone(),
86            persisted_events: self.persisted_events.clone(),
87            new_events: self.new_events.clone(),
88        }
89    }
90}
91
92impl<T> EntityEvents<T>
93where
94    T: EsEvent,
95{
96    /// Initializes a new `EntityEvents` instance with the given entity ID and initial events which is returned by [`IntoEvents`] method
97    pub fn init(id: <T as EsEvent>::EntityId, initial_events: impl IntoIterator<Item = T>) -> Self {
98        let context = if <T as EsEvent>::event_context() {
99            Some(crate::EventContext::data_for_storing())
100        } else {
101            None
102        };
103        let new_events = initial_events
104            .into_iter()
105            .map(|event| EventWithContext {
106                event,
107                context: context.clone(),
108            })
109            .collect();
110        Self {
111            entity_id: id,
112            persisted_events: Vec::new(),
113            new_events,
114        }
115    }
116
117    /// Returns a reference to the entity's identifier
118    pub fn id(&self) -> &<T as EsEvent>::EntityId {
119        &self.entity_id
120    }
121
122    /// Returns the timestamp of the first persisted event, indicating when the entity was created
123    pub fn entity_first_persisted_at(&self) -> Option<DateTime<Utc>> {
124        self.persisted_events.first().map(|e| e.recorded_at)
125    }
126
127    /// Returns the timestamp of the last persisted event, indicating when the entity was last modified
128    pub fn entity_last_modified_at(&self) -> Option<DateTime<Utc>> {
129        self.persisted_events.last().map(|e| e.recorded_at)
130    }
131
132    /// Appends a single new event to the entity's event stream to be persisted later
133    pub fn push(&mut self, event: T) {
134        let context = if <T as EsEvent>::event_context() {
135            Some(crate::EventContext::data_for_storing())
136        } else {
137            None
138        };
139        self.new_events.push(EventWithContext { event, context });
140    }
141
142    /// Appends multiple new events to the entity's event stream to be persisted later
143    pub fn extend(&mut self, events: impl IntoIterator<Item = T>) {
144        let context = if <T as EsEvent>::event_context() {
145            Some(crate::EventContext::data_for_storing())
146        } else {
147            None
148        };
149        self.new_events
150            .extend(events.into_iter().map(|event| EventWithContext {
151                event,
152                context: context.clone(),
153            }));
154    }
155
156    /// Returns true if there are any unpersisted events waiting to be saved
157    pub fn any_new(&self) -> bool {
158        !self.new_events.is_empty()
159    }
160
161    /// Returns the count of persisted events
162    pub fn len_persisted(&self) -> usize {
163        self.persisted_events.len()
164    }
165
166    /// Returns an iterator over all persisted events
167    pub fn iter_persisted(&self) -> impl DoubleEndedIterator<Item = &PersistedEvent<T>> + Clone {
168        self.persisted_events.iter()
169    }
170
171    /// Returns an iterator over the last `n` persisted events
172    ///
173    /// If fewer than `n` events have been persisted, all persisted events are
174    /// returned instead of panicking.
175    pub fn last_persisted(&self, n: usize) -> LastPersisted<'_, T> {
176        let start = self.persisted_events.len().saturating_sub(n);
177        self.persisted_events[start..].iter()
178    }
179
180    /// Returns an iterator over all events (both persisted and new) in chronological order
181    pub fn iter_all(&self) -> impl DoubleEndedIterator<Item = &T> + Clone {
182        self.persisted_events
183            .iter()
184            .map(|e| &e.event)
185            .chain(self.new_events.iter().map(|e| &e.event))
186    }
187
188    /// Loads and reconstructs the first entity from a stream of GenericEvents, marking events as `persisted`.
189    ///
190    /// Returns `Ok(None)` if no events are present, `Ok(Some(entity))` on success.
191    pub fn load_first<E: EsEntity<Event = T>>(
192        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
193    ) -> Result<Option<E>, EntityHydrationError> {
194        let mut current_id = None;
195        let mut current = None;
196        for e in events {
197            if current_id.is_none() {
198                current_id = Some(e.entity_id.clone());
199                current = Some(Self {
200                    entity_id: e.entity_id.clone(),
201                    persisted_events: Vec::new(),
202                    new_events: Vec::new(),
203                });
204            }
205            if current_id.as_ref() != Some(&e.entity_id) {
206                break;
207            }
208            let cur = current.as_mut().expect("Could not get current");
209            let mut event_json = e.event;
210            if let Some(payload) = e.forgettable_payload {
211                crate::forgettable::inject_forgettable_payload(&mut event_json, payload);
212            }
213            cur.persisted_events.push(PersistedEvent {
214                entity_id: e.entity_id,
215                recorded_at: e.recorded_at,
216                sequence: e.sequence as usize,
217                event: serde_json::from_value(event_json)?,
218                context: e.context,
219            });
220        }
221        if let Some(current) = current {
222            Ok(Some(E::try_from_events(current)?))
223        } else {
224            Ok(None)
225        }
226    }
227
228    /// Loads and reconstructs up to `n` entities from a stream of GenericEvents.
229    /// Assumes the events are grouped by `id` and ordered by `sequence` per `id`.
230    ///
231    /// Returns both the entities and a flag indicating whether more entities were available in the stream.
232    pub fn load_n<E: EsEntity<Event = T>>(
233        events: impl IntoIterator<Item = GenericEvent<<T as EsEvent>::EntityId>>,
234        n: usize,
235    ) -> Result<(Vec<E>, bool), EntityHydrationError> {
236        if n == 0 {
237            // Asking for zero entities yields zero entities. `has_more` reports
238            // whether the stream was non-empty, mirroring the `LIMIT n + 1`
239            // over-fetch contract the generated repos rely on: callers query
240            // `LIMIT (first + 1)`, so a non-empty stream means a next page exists.
241            let has_more = events.into_iter().next().is_some();
242            return Ok((Vec::new(), has_more));
243        }
244        let mut ret: Vec<E> = Vec::new();
245        let mut current_id = None;
246        let mut current = None;
247        for e in events {
248            if current_id.as_ref() != Some(&e.entity_id) {
249                if let Some(current) = current.take() {
250                    ret.push(E::try_from_events(current)?);
251                    if ret.len() == n {
252                        return Ok((ret, true));
253                    }
254                }
255
256                current_id = Some(e.entity_id.clone());
257                current = Some(Self {
258                    entity_id: e.entity_id.clone(),
259                    persisted_events: Vec::new(),
260                    new_events: Vec::new(),
261                });
262            }
263            let cur = current.as_mut().expect("Could not get current");
264            let mut event_json = e.event;
265            if let Some(payload) = e.forgettable_payload {
266                crate::forgettable::inject_forgettable_payload(&mut event_json, payload);
267            }
268            cur.persisted_events.push(PersistedEvent {
269                entity_id: e.entity_id,
270                recorded_at: e.recorded_at,
271                sequence: e.sequence as usize,
272                event: serde_json::from_value(event_json)?,
273                context: e.context,
274            });
275        }
276        if let Some(current) = current.take() {
277            ret.push(E::try_from_events(current)?);
278        }
279        Ok((ret, false))
280    }
281
282    #[doc(hidden)]
283    pub fn iter_new_events(&self) -> impl Iterator<Item = &EventWithContext<T>> {
284        self.new_events.iter()
285    }
286
287    #[doc(hidden)]
288    pub fn mark_new_events_persisted_at(
289        &mut self,
290        recorded_at: chrono::DateTime<chrono::Utc>,
291    ) -> usize {
292        let n = self.new_events.len();
293        let offset = self.persisted_events.len() + 1;
294        self.persisted_events
295            .extend(
296                self.new_events
297                    .drain(..)
298                    .enumerate()
299                    .map(|(i, event)| PersistedEvent {
300                        entity_id: self.entity_id.clone(),
301                        recorded_at,
302                        sequence: i + offset,
303                        event: event.event,
304                        context: event.context,
305                    }),
306            );
307        n
308    }
309
310    #[doc(hidden)]
311    pub fn new_event_types(&self) -> Vec<String> {
312        self.new_events
313            .iter()
314            .map(|event| event.event.event_type().to_string())
315            .collect()
316    }
317
318    #[doc(hidden)]
319    pub fn serialize_new_events(&self) -> Vec<serde_json::Value> {
320        self.new_events
321            .iter()
322            .map(|event| serde_json::to_value(&event.event).expect("Failed to serialize event"))
323            .collect()
324    }
325
326    /// Forgets all forgettable payloads in persisted events and returns the taken events.
327    ///
328    /// Applies `forget_fn` to each persisted event, then takes ownership of the event
329    /// stream, leaving `self` as an empty shell. The returned `EntityEvents` can be passed
330    /// to `TryFromEvents::try_from_events` to rebuild the entity with forgotten fields.
331    #[doc(hidden)]
332    pub fn forget_and_take(&mut self, mut forget_fn: impl FnMut(&mut T)) -> Self {
333        for persisted in &mut self.persisted_events {
334            forget_fn(&mut persisted.event);
335        }
336        let entity_id = self.entity_id.clone();
337        std::mem::replace(
338            self,
339            Self {
340                entity_id,
341                persisted_events: Vec::new(),
342                new_events: Vec::new(),
343            },
344        )
345    }
346
347    #[doc(hidden)]
348    pub fn serialize_new_event_contexts(&self) -> Option<Vec<crate::ContextData>> {
349        if <T as EsEvent>::event_context() {
350            let contexts = self
351                .new_events
352                .iter()
353                .map(|event| event.context.clone().expect("Missing context"))
354                .collect();
355
356            Some(contexts)
357        } else {
358            None
359        }
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::*;
366    use proptest::prelude::*;
367    use uuid::Uuid;
368
369    /// Builds a `GenericEvent` whose JSON deserializes to `Created(name)`.
370    fn valid_event(id: Uuid, sequence: i32, name: &str) -> GenericEvent<Uuid> {
371        GenericEvent {
372            entity_id: id,
373            sequence,
374            event: serde_json::to_value(DummyEntityEvent::Created(name.to_string()))
375                .expect("could not serialize"),
376            context: None,
377            recorded_at: chrono::Utc::now(),
378            forgettable_payload: None,
379        }
380    }
381
382    /// Small bounded JSON strategy covering null/bool/int/float/string plus one
383    /// level of array/object nesting. Cheap to shrink and enough to exercise the
384    /// serde error paths without ballooning runtime.
385    fn json_value() -> impl Strategy<Value = serde_json::Value> {
386        let scalar = prop_oneof![
387            Just(serde_json::Value::Null),
388            any::<bool>().prop_map(serde_json::Value::Bool),
389            any::<i64>().prop_map(serde_json::Value::from),
390            any::<f64>().prop_map(serde_json::Value::from),
391            ".{0,15}".prop_map(serde_json::Value::String),
392        ]
393        .boxed();
394        let nested = prop_oneof![
395            proptest::collection::vec(scalar.clone(), 0..4).prop_map(serde_json::Value::Array),
396            proptest::collection::vec((".{0,6}", scalar.clone()), 0..4).prop_map(|pairs| {
397                let mut m = serde_json::Map::new();
398                for (k, v) in pairs {
399                    m.insert(k, v);
400                }
401                serde_json::Value::Object(m)
402            },),
403        ];
404        prop_oneof![scalar, nested]
405    }
406
407    #[derive(Debug, serde::Serialize, serde::Deserialize)]
408    enum DummyEntityEvent {
409        Created(String),
410    }
411
412    impl EsEvent for DummyEntityEvent {
413        type EntityId = Uuid;
414        fn event_context() -> bool {
415            true
416        }
417        fn event_type(&self) -> &'static str {
418            match self {
419                Self::Created(_) => "created",
420            }
421        }
422    }
423
424    struct DummyEntity {
425        name: String,
426
427        events: EntityEvents<DummyEntityEvent>,
428    }
429
430    impl EsEntity for DummyEntity {
431        type Event = DummyEntityEvent;
432        type New = NewDummyEntity;
433
434        fn events_mut(&mut self) -> &mut EntityEvents<DummyEntityEvent> {
435            &mut self.events
436        }
437        fn events(&self) -> &EntityEvents<DummyEntityEvent> {
438            &self.events
439        }
440    }
441
442    impl TryFromEvents<DummyEntityEvent> for DummyEntity {
443        fn try_from_events(
444            events: EntityEvents<DummyEntityEvent>,
445        ) -> Result<Self, EntityHydrationError> {
446            let name = events
447                .iter_persisted()
448                .map(|e| match &e.event {
449                    DummyEntityEvent::Created(name) => name.clone(),
450                })
451                .next()
452                .expect("Could not find name");
453            Ok(Self { name, events })
454        }
455    }
456
457    struct NewDummyEntity {}
458
459    impl IntoEvents<DummyEntityEvent> for NewDummyEntity {
460        fn into_events(self) -> EntityEvents<DummyEntityEvent> {
461            EntityEvents::init(
462                Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap(),
463                vec![DummyEntityEvent::Created("".to_owned())],
464            )
465        }
466    }
467
468    #[test]
469    fn load_zero_events() {
470        let generic_events = vec![];
471        let res = EntityEvents::load_first::<DummyEntity>(generic_events);
472        assert!(matches!(res, Ok(None)));
473    }
474
475    #[test]
476    fn load_first() {
477        let generic_events = vec![GenericEvent {
478            entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
479            sequence: 1,
480            event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
481                .expect("Could not serialize"),
482            context: None,
483            recorded_at: chrono::Utc::now(),
484            forgettable_payload: None,
485        }];
486        let entity: DummyEntity = EntityEvents::load_first(generic_events)
487            .expect("Could not load")
488            .expect("No entity found");
489        assert!(entity.name == "dummy-name");
490    }
491
492    #[test]
493    fn load_n() {
494        let generic_events = vec![
495            GenericEvent {
496                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap(),
497                sequence: 1,
498                event: serde_json::to_value(DummyEntityEvent::Created("dummy-name".to_owned()))
499                    .expect("Could not serialize"),
500                context: None,
501                recorded_at: chrono::Utc::now(),
502                forgettable_payload: None,
503            },
504            GenericEvent {
505                entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000003").unwrap(),
506                sequence: 1,
507                event: serde_json::to_value(DummyEntityEvent::Created("other-name".to_owned()))
508                    .expect("Could not serialize"),
509                context: None,
510                recorded_at: chrono::Utc::now(),
511                forgettable_payload: None,
512            },
513        ];
514        let (entity, more): (Vec<DummyEntity>, _) =
515            EntityEvents::load_n(generic_events, 2).expect("Could not load");
516        assert!(!more);
517        assert_eq!(entity.len(), 2);
518    }
519
520    #[test]
521    fn last_persisted_does_not_panic_when_n_exceeds_len() {
522        let generic_events = vec![GenericEvent {
523            entity_id: Uuid::parse_str("00000000-0000-0000-0000-000000000004").unwrap(),
524            sequence: 1,
525            event: serde_json::to_value(DummyEntityEvent::Created("dummy".to_owned()))
526                .expect("Could not serialize"),
527            context: None,
528            recorded_at: chrono::Utc::now(),
529            forgettable_payload: None,
530        }];
531        let entity: DummyEntity = EntityEvents::load_first(generic_events)
532            .expect("Could not load")
533            .expect("No entity found");
534        let events = entity.events();
535
536        // n == len works
537        assert_eq!(events.last_persisted(1).count(), 1);
538        // n > len clamps to all persisted events instead of underflowing
539        assert_eq!(events.last_persisted(10).count(), 1);
540        // n == 0 yields nothing
541        assert_eq!(events.last_persisted(0).count(), 0);
542    }
543
544    proptest! {
545        #[test]
546        fn load_first_empty_input_returns_none(_ in Just(())) {
547            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(vec![]);
548            prop_assert!(matches!(res, Ok(None)));
549        }
550
551        #[test]
552        fn load_first_non_empty_valid_hydrates(
553            count in 1u8..8,
554            names in proptest::collection::vec(".{0,10}", 1..8),
555        ) {
556            let id = Uuid::nil();
557            let events: Vec<_> = (1..=count as i32)
558                .zip(names.iter())
559                .map(|(s, n)| valid_event(id, s, n))
560                .collect();
561            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events);
562            prop_assert!(matches!(res, Ok(Some(_))));
563        }
564
565        /// Feeds arbitrary JSON through the loader. It may hydrate or error, but
566        /// it must never panic — and `Ok(None)` is only possible for empty input.
567        #[test]
568        fn load_first_arbitrary_json_never_panics(
569            raw in proptest::collection::vec(json_value(), 0..10),
570        ) {
571            let events: Vec<_> = raw
572                .into_iter()
573                .enumerate()
574                .map(|(i, v)| GenericEvent {
575                    entity_id: Uuid::nil(),
576                    sequence: i as i32,
577                    event: v,
578                    context: None,
579                    recorded_at: chrono::Utc::now(),
580                    forgettable_payload: None,
581                })
582                .collect();
583            let is_empty = events.is_empty();
584            let res = EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events);
585            if let Ok(None) = res {
586                prop_assert!(is_empty);
587            }
588        }
589
590        /// With a grouped, ascending stream (the documented precondition) and
591        /// `n >= 1`, `load_n` returns `min(n, k)` entities and reports `has_more`
592        /// exactly when `n < k`.
593        #[test]
594        fn load_n_respects_limit_and_more_flag(
595            k in 1u8..8,
596            per in 1u8..4,
597            n in 1u8..12,
598        ) {
599            let mut events = Vec::new();
600            for i in 0..k {
601                let id = Uuid::from_u128(i as u128);
602                for s in 1..=per as i32 {
603                    events.push(valid_event(id, s, &format!("e{i}-{s}")));
604                }
605            }
606            let (entities, has_more) =
607                EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, n as usize)
608                    .expect("valid events hydrate");
609            prop_assert_eq!(entities.len(), (n as usize).min(k as usize));
610            prop_assert_eq!(has_more, n < k);
611        }
612
613        /// Regression for the `n = 0` degenerate case: previously returned *all*
614        /// entities instead of zero. Now returns none and reports `has_more` iff
615        /// the stream was non-empty (the `LIMIT n + 1` contract).
616        #[test]
617        fn load_n_zero_returns_no_entities(k in 0u8..5, per in 1u8..3) {
618            let mut events = Vec::new();
619            for i in 0..k {
620                let id = Uuid::from_u128(i as u128);
621                for s in 1..=per as i32 {
622                    events.push(valid_event(id, s, &format!("e{i}-{s}")));
623                }
624            }
625            let (entities, has_more) =
626                EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, 0)
627                    .expect("valid events hydrate");
628            prop_assert!(entities.is_empty());
629            prop_assert_eq!(has_more, k > 0);
630        }
631
632        #[test]
633        fn load_n_arbitrary_json_never_panics(
634            raw in proptest::collection::vec(json_value(), 0..10),
635            n in 0u8..12,
636        ) {
637            let events: Vec<_> = raw
638                .into_iter()
639                .enumerate()
640                .map(|(i, v)| GenericEvent {
641                    entity_id: Uuid::nil(),
642                    sequence: i as i32,
643                    event: v,
644                    context: None,
645                    recorded_at: chrono::Utc::now(),
646                    forgettable_payload: None,
647                })
648                .collect();
649            let _ = EntityEvents::<DummyEntityEvent>::load_n::<DummyEntity>(events, n as usize);
650        }
651
652        /// `last_persisted(n)` must clamp to the available events for any `n`,
653        /// generalizing the dedicated saturating-sub test above.
654        #[test]
655        fn last_persisted_clamps_for_any_n(
656            p in 1u8..8,
657            n in 0u16..12,
658        ) {
659            let id = Uuid::nil();
660            let events: Vec<_> = (1..=p as i32)
661                .map(|s| valid_event(id, s, &format!("n{s}")))
662                .collect();
663            let entity: DummyEntity =
664                EntityEvents::<DummyEntityEvent>::load_first::<DummyEntity>(events)
665                    .expect("load")
666                    .expect("some");
667            let count = entity.events().last_persisted(n as usize).count();
668            prop_assert_eq!(count, (n as usize).min(p as usize));
669        }
670
671        /// Marking new events as persisted must drain `new_events`, leave
672        /// `len_persisted` consistent, and assign contiguous 1-based sequences
673        /// across repeated calls.
674        #[test]
675        fn mark_new_events_assigns_contiguous_sequences(
676            a in 0u8..5,
677            b in 0u8..5,
678        ) {
679            let id = Uuid::nil();
680            let mut events = EntityEvents::init(
681                id,
682                (0..a).map(|i| DummyEntityEvent::Created(format!("n{i}"))),
683            );
684            let now = chrono::Utc::now();
685
686            prop_assert_eq!(events.mark_new_events_persisted_at(now), a as usize);
687            prop_assert!(!events.any_new());
688            prop_assert_eq!(events.len_persisted(), a as usize);
689            let seqs: Vec<usize> = events.iter_persisted().map(|e| e.sequence).collect();
690            prop_assert_eq!(seqs, (1..=a as usize).collect::<Vec<_>>());
691
692            for i in 0..b {
693                events.push(DummyEntityEvent::Created(format!("m{i}")));
694            }
695            if b > 0 {
696                prop_assert!(events.any_new());
697            }
698            prop_assert_eq!(events.mark_new_events_persisted_at(now), b as usize);
699            prop_assert!(!events.any_new());
700            prop_assert_eq!(events.len_persisted(), (a + b) as usize);
701            let seqs: Vec<usize> = events.iter_persisted().map(|e| e.sequence).collect();
702            prop_assert_eq!(seqs, (1..=(a + b) as usize).collect::<Vec<_>>());
703        }
704    }
705}
es_entity/events.rs

es_entity/
events.rs
