dig_events_protocol/
traits.rs

1//! The two shape traits the engine implements and apps depend on (SPEC §7).
2//!
3//! These define the SHAPE of emit + catch-up without any machinery: [`EventEmitter`] is what the
4//! engine's event bus offers to producers, and [`CatchUp`] is what a subscriber calls once after a
5//! gap to backfill the missed range before resuming the live stream. The concrete bus, persisted
6//! delta store, and live subscription streams live in the engine (dig-wallet-backend) — this crate
7//! only fixes the interface so a second implementation is interchangeable.
8
9use async_trait::async_trait;
10use enumset::EnumSet;
11
12use crate::cursor::{Cursor, EmittedEvent};
13use crate::event::WalletEvent;
14use crate::kind::EventKind;
15
16/// A sink that accepts emitted events and stamps each with a monotonic [`Cursor`].
17///
18/// The engine implements this over its event bus; a producer calls [`publish`](EventEmitter::publish)
19/// and gets back the cursor the event was assigned (so it can correlate or persist a checkpoint).
20pub trait EventEmitter {
21    /// Publish an event, returning the monotonic [`Cursor`] stamped on it.
22    fn publish(&self, event: WalletEvent) -> Cursor;
23}
24
25/// The backfill half of the subscription contract: a subscriber that fell behind calls this ONCE to
26/// fetch the events it missed, then resumes the live stream.
27///
28/// The associated [`Error`](CatchUp::Error) type is generic so this leaf crate needs no error
29/// dependency — the engine picks its own error type when it implements the trait.
30#[async_trait]
31pub trait CatchUp {
32    /// The implementer's backfill error type.
33    type Error;
34
35    /// Return every [`EmittedEvent`] with a cursor STRICTLY GREATER than `since`, in cursor order,
36    /// optionally narrowed to the subscriber's `filter`. Passing an empty filter is the caller's
37    /// choice to receive nothing; pass [`EnumSet::all`] to backfill every kind.
38    async fn catch_up(
39        &self,
40        since: Cursor,
41        filter: EnumSet<EventKind>,
42    ) -> Result<Vec<EmittedEvent>, Self::Error>;
43}
44
45/// Retain only the [`EmittedEvent`]s whose event kind passes `filter`, preserving cursor order.
46///
47/// The shared, drift-free filtering rule used on BOTH sides: the engine narrows a live stream with
48/// it, and a [`CatchUp`] implementer applies the same rule to its backfill so live and catch-up
49/// deliver an identical filtered view.
50pub fn filter_events(
51    events: impl IntoIterator<Item = EmittedEvent>,
52    filter: EnumSet<EventKind>,
53) -> Vec<EmittedEvent> {
54    events
55        .into_iter()
56        .filter(|e| e.event.matches(filter))
57        .collect()
58}
dig_events_protocol/traits.rs

dig_events_protocol/
traits.rs
