evento_core/
projection.rs

1//! Projections and event subscriptions.
2//!
3//! This module provides the core building blocks for event sourcing:
4//! - Projections that build read models from events
5//! - Subscriptions that continuously process events
6//! - Loading aggregate state from event streams
7//!
8//! # Key Types
9//!
10//! - [`Projection`] - Defines handlers for building projections
11//! - [`LoadBuilder`] - Loads aggregate state from events
12//! - [`ProjectionSubscription`] - Starts a subscription that keeps a projection up to date
13//! - [`SubscriptionBuilder`] - Builds continuous event subscriptions
14//! - [`Subscription`] - Handle to a running subscription
15//!
16//! # Example
17//!
18//! ```rust,ignore
19//! use evento::projection::Projection;
20//!
21//! // Define a projection with event handlers (no id at construction)
22//! let projection = Projection::<_, AccountView>::new::<Account>()
23//!     .handler(account_opened())
24//!     .handler(money_deposited());
25//!
26//! // Load aggregate state for one id
27//! let result = projection
28//!     .load("account-123")
29//!     .execute(&executor)
30//!     .await?;
31//!
32//! // Or keep the projection auto-updated via a subscription
33//! let subscription = Projection::<_, AccountView>::new::<Account>()
34//!     .handler(account_opened())
35//!     .handler(money_deposited())
36//!     .subscription("account-view")
37//!     .start(&executor)
38//!     .await?;
39//! ```
40
41use std::{
42    collections::HashMap, future::Future, marker::PhantomData, ops::Deref, pin::Pin, sync::Arc,
43    time::Duration,
44};
45
46use crate::{
47    context,
48    cursor::{self, Args, Cursor},
49    subscription::{self, RoutingKey, Subscription, SubscriptionBuilder},
50    Aggregate, AggregateEvent, EventFilter, Executor, WriteBuilder,
51};
52
53/// Handler context providing access to executor and shared data.
54///
55/// `Context` wraps an [`RwContext`](crate::context::RwContext) for type-safe
56/// data storage and provides access to the executor for database operations.
57#[derive(Clone)]
58pub struct Context<'a, E: Executor> {
59    context: context::RwContext,
60    /// Reference to the executor for database operations
61    pub executor: &'a E,
62    pub id: String,
63    revision: u16,
64    aggregate_type: String,
65    aggregators: &'a HashMap<String, String>,
66}
67
68impl<'a, E: Executor> Context<'a, E> {
69    /// Retrieves a stored snapshot for the given ID.
70    ///
71    /// Returns `None` if no snapshot exists.
72    pub async fn get_snapshot<D: bitcode::DecodeOwned + ProjectionCursor>(
73        &self,
74    ) -> anyhow::Result<Option<D>> {
75        let Some((data, cursor)) = self
76            .executor
77            .get_snapshot(
78                self.aggregate_type.to_owned(),
79                self.revision.to_string(),
80                self.id.to_owned(),
81            )
82            .await?
83        else {
84            return Ok(None);
85        };
86
87        let mut data: D = bitcode::decode(&data)?;
88        data.set_cursor(&cursor);
89
90        Ok(Some(data))
91    }
92
93    /// Stores a snapshot for the given ID.
94    ///
95    /// The snapshot cursor is extracted from the data to track the event position.
96    pub async fn take_snapshot<D: bitcode::Encode + ProjectionCursor>(
97        &self,
98        data: &D,
99    ) -> anyhow::Result<()> {
100        let cursor = data.get_cursor();
101        let data = bitcode::encode(data);
102
103        self.executor
104            .save_snapshot(
105                self.aggregate_type.to_owned(),
106                self.revision.to_string(),
107                self.id.to_owned(),
108                data,
109                cursor,
110            )
111            .await
112    }
113
114    /// Deletes the stored snapshot for the given ID.
115    ///
116    /// Idempotent: no error if no snapshot exists.
117    pub async fn drop_snapshot(&self) -> anyhow::Result<()> {
118        self.executor
119            .delete_snapshot(self.aggregate_type.to_owned(), self.id.to_owned())
120            .await
121    }
122
123    /// Returns the aggregate ID for a registered aggregate type.
124    ///
125    /// # Panics
126    ///
127    /// Panics if the aggregate type was not registered via [`LoadBuilder::aggregate`].
128    pub async fn aggregate<A: Aggregate>(&self) -> String {
129        tracing::debug!(
130            "Failed to get `Aggregate id <{}>` For the Aggregate id extractor to work \
131        correctly, register the related aggregator with `.aggregate::<MyAggregator>(id)` on \
132        the load builder. Ensure that types align in both the set and retrieve calls.",
133            A::aggregate_type()
134        );
135
136        self.aggregators
137            .get(A::aggregate_type())
138            .expect("Projection Aggregate not configured correctly. View/enable debug logs for more details.")
139            .to_owned()
140    }
141}
142
143impl<'a, E: Executor> Deref for Context<'a, E> {
144    type Target = context::RwContext;
145
146    fn deref(&self) -> &Self::Target {
147        &self.context
148    }
149}
150
151/// Trait for event handlers.
152///
153/// Handlers process events in two modes:
154/// - `handle`: For subscriptions that perform side effects (send emails, update read models)
155/// - `apply`: For loading aggregate state by replaying events
156///
157/// This trait is typically implemented via the `#[evento::handler]` macro.
158pub trait Handler<P: 'static>: Sync + Send {
159    /// Applies an event to build projection state.
160    ///
161    /// This is called when loading aggregate state by replaying events.
162    /// It should be a pure function that modifies the projection without side effects.
163    fn handle<'a>(
164        &'a self,
165        projection: &'a mut P,
166        event: &'a crate::Event,
167    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
168
169    /// Returns the aggregate type this handler processes.
170    fn aggregate_type(&self) -> &'static str;
171    /// Returns the event name this handler processes.
172    fn event_name(&self) -> &'static str;
173}
174
175/// Trait for types that track their cursor position in the event stream.
176///
177/// This trait is typically derived using the `#[evento::projection]` macro.
178pub trait ProjectionCursor {
179    /// Returns the current cursor position.
180    fn get_cursor(&self) -> cursor::Value;
181    /// Sets the cursor position.
182    fn set_cursor(&mut self, v: &cursor::Value);
183}
184
185/// Trait for projections that can create a [`WriteBuilder`].
186///
187/// Extends [`ProjectionCursor`] to provide aggregate identity and versioning,
188/// enabling projections to emit new events.
189pub trait ProjectionAggregate: ProjectionCursor {
190    /// Returns the aggregate ID for this projection.
191    ///
192    /// Implementors must return the stable identity of the aggregate this
193    /// projection represents (typically a field populated from the first event).
194    fn aggregate_id(&self) -> String;
195
196    /// Returns the current aggregate version from the cursor.
197    ///
198    /// Returns `0` if no cursor is set.
199    fn aggregate_version(&self) -> anyhow::Result<u16> {
200        let value = self.get_cursor();
201        if value == Default::default() {
202            return Ok(0);
203        }
204
205        let cursor = crate::Event::deserialize_cursor(&value)?;
206
207        Ok(cursor.v)
208    }
209
210    /// Creates a [`WriteBuilder`] pre-configured with this projection's ID and version.
211    ///
212    /// Use this to emit new events from a projection (the write gateway).
213    fn write(&self) -> anyhow::Result<WriteBuilder> {
214        Ok(WriteBuilder::new(self.aggregate_id())
215            .original_version(self.aggregate_version()?)
216            .to_owned())
217    }
218}
219
220/// Trait for types that can be restored from snapshots.
221///
222/// Snapshots provide a performance optimization by storing pre-computed
223/// state, avoiding the need to replay all events from the beginning.
224///
225/// This trait is typically implemented via the `#[evento::snapshot]` macro.
226pub trait Snapshot<E: Executor>: ProjectionCursor + Sized {
227    /// Restores state from a snapshot if available.
228    ///
229    /// Returns `None` if no snapshot exists for the given ID.
230    fn restore(
231        _context: &Context<'_, E>,
232    ) -> impl Future<Output = anyhow::Result<Option<Self>>> + Send {
233        Box::pin(async { Ok(None) })
234    }
235
236    /// Stores the current state as a snapshot.
237    ///
238    /// Default implementation does nothing.
239    fn take_snapshot(
240        &self,
241        _context: &Context<'_, E>,
242    ) -> impl Future<Output = anyhow::Result<()>> + Send {
243        Box::pin(async { Ok(()) })
244    }
245
246    /// Drops any stored snapshot for the given aggregate.
247    ///
248    /// Called by a [`ProjectionSubscription`] when the configured tombstone
249    /// event is observed (see [`Projection::tombstone`]). Override this for
250    /// projections backed by a custom table to delete the row; the default
251    /// is a no-op.
252    fn drop_snapshot(_context: &Context<'_, E>) -> impl Future<Output = anyhow::Result<()>> + Send {
253        Box::pin(async { Ok(()) })
254    }
255}
256
257impl<T: bitcode::Encode + bitcode::DecodeOwned + ProjectionCursor + Send + Sync, E: Executor>
258    Snapshot<E> for T
259{
260    async fn restore(context: &Context<'_, E>) -> anyhow::Result<Option<Self>> {
261        context.get_snapshot().await
262    }
263
264    async fn take_snapshot(&self, context: &Context<'_, E>) -> anyhow::Result<()> {
265        context.take_snapshot(self).await
266    }
267
268    async fn drop_snapshot(context: &Context<'_, E>) -> anyhow::Result<()> {
269        context.drop_snapshot().await
270    }
271}
272
273/// Projection definition: a set of handlers for a primary aggregate type.
274///
275/// A `Projection` is constructed without an aggregate id. Terminal operations:
276/// - [`Projection::load`] returns a [`LoadBuilder`] bound to a specific aggregate id.
277/// - [`Projection::subscription`] returns a [`ProjectionSubscription`] that auto-updates
278///   the projection as new events arrive.
279///
280/// # Example
281///
282/// ```rust,ignore
283/// // Load a single aggregate
284/// let result = Projection::<_, AccountView>::new::<Account>()
285///     .handler(account_opened())
286///     .handler(money_deposited())
287///     .data(app_config)
288///     .load("account-123")
289///     .execute(&executor)
290///     .await?;
291///
292/// // Start a subscription that keeps every aggregate up to date
293/// let subscription = Projection::<_, AccountView>::new::<Account>()
294///     .handler(account_opened())
295///     .handler(money_deposited())
296///     .data(app_config)
297///     .subscription("account-view")
298///     .start(&executor)
299///     .await?;
300/// ```
301pub struct Projection<E: Executor, P: Default + 'static> {
302    aggregate_type: &'static str,
303    revision: u16,
304    handlers: HashMap<String, Box<dyn Handler<P>>>,
305    context: context::RwContext,
306    safety_disabled: bool,
307    tombstone: Option<(&'static str, &'static str)>,
308    executor: PhantomData<E>,
309}
310
311impl<E: Executor, P: Snapshot<E> + Default + 'static> Projection<E, P> {
312    /// Creates a new projection definition for the given primary aggregate type.
313    pub fn new<A: Aggregate>() -> Projection<E, P> {
314        Projection {
315            aggregate_type: A::aggregate_type(),
316            context: Default::default(),
317            handlers: HashMap::new(),
318            safety_disabled: true,
319            tombstone: None,
320            executor: PhantomData,
321            revision: 0,
322        }
323    }
324
325    /// Sets the snapshot revision.
326    ///
327    /// Changing the revision invalidates existing snapshots, forcing a full rebuild.
328    pub fn revision(mut self, value: u16) -> Self {
329        self.revision = value;
330
331        self
332    }
333
334    /// Enables safety checks for unhandled events.
335    ///
336    /// When enabled, execution fails if an event is encountered without a handler.
337    pub fn strict(mut self) -> Self {
338        self.safety_disabled = false;
339
340        self
341    }
342
343    /// Declares which event marks an aggregate as deleted (tombstoned).
344    ///
345    /// When set:
346    /// - [`LoadBuilder::execute`] short-circuits and returns `Ok(None)` as soon
347    ///   as it sees a committed tombstone event for the requested id, avoiding
348    ///   any snapshot read or event replay.
349    /// - [`ProjectionSubscription`] routes tombstone events to
350    ///   [`Snapshot::drop_snapshot`] so the user can delete their snapshot row.
351    pub fn tombstone<EV: AggregateEvent + Send + Sync + 'static>(mut self) -> Self {
352        self.tombstone = Some((EV::aggregate_type(), EV::event_name()));
353        self
354    }
355
356    /// Registers an event handler with this projection.
357    ///
358    /// # Panics
359    ///
360    /// Panics if a handler for the same event type is already registered.
361    pub fn handler<H: Handler<P> + 'static>(mut self, h: H) -> Self {
362        let key = format!("{}_{}", h.aggregate_type(), h.event_name());
363        if self.handlers.insert(key.to_owned(), Box::new(h)).is_some() {
364            panic!("Cannot register event handler: key {} already exists", key);
365        }
366        self
367    }
368
369    /// Registers a skip handler with this projection.
370    ///
371    /// # Panics
372    ///
373    /// Panics if a handler for the same event type is already registered.
374    pub fn skip<EV: AggregateEvent + Send + Sync + 'static>(self) -> Self {
375        self.handler(SkipHandler::<EV>(PhantomData))
376    }
377
378    /// Adds shared data to the handler context.
379    ///
380    /// Data added here is accessible in handlers via the context. Data lives
381    /// on the projection definition and is reused for both [`Projection::load`]
382    /// and [`Projection::subscription`].
383    pub fn data<D: Send + Sync + 'static>(self, v: D) -> Self {
384        self.context.insert(v);
385
386        self
387    }
388
389    /// Returns a [`LoadBuilder`] for loading the aggregate with the given id.
390    pub fn load(self, id: impl Into<String>) -> LoadBuilder<E, P> {
391        let id = id.into();
392        let mut aggregators = HashMap::new();
393        aggregators.insert(self.aggregate_type.to_string(), id.to_owned());
394
395        LoadBuilder {
396            projection: self,
397            id,
398            aggregators,
399        }
400    }
401
402    /// Returns a [`LoadBuilder`] for loading state keyed on multiple aggregate ids.
403    ///
404    /// The ids are hashed via [`crate::hash_ids`] to produce a stable snapshot id.
405    pub fn load_ids(self, ids: Vec<impl Into<String>>) -> LoadBuilder<E, P> {
406        self.load(crate::hash_ids(ids))
407    }
408
409    /// Returns a builder for a subscription that keeps this projection up to date.
410    pub fn subscription(self, key: impl Into<String>) -> ProjectionSubscription<E, P> {
411        ProjectionSubscription {
412            projection: self,
413            key: key.into(),
414            routing_key: None,
415            chunk_size: 300,
416            retry: Some(30),
417            delay: None,
418            continue_on_error: false,
419        }
420    }
421
422    async fn load_aggregator(
423        &self,
424        executor: &E,
425        id: &str,
426        extra_aggregators: &HashMap<String, String>,
427    ) -> anyhow::Result<Option<P>> {
428        if let Some((tombstone_type, tombstone_event)) = self.tombstone {
429            let res = executor
430                .read(
431                    Some(vec![EventFilter::exact(
432                        tombstone_type,
433                        id.to_owned(),
434                        tombstone_event,
435                    )]),
436                    None,
437                    Args::backward(1, None),
438                )
439                .await?;
440            if !res.edges.is_empty() {
441                return Ok(None);
442            }
443        }
444
445        let mut aggregators = HashMap::with_capacity(extra_aggregators.len() + 1);
446        aggregators.insert(self.aggregate_type.to_string(), id.to_owned());
447        for (k, v) in extra_aggregators {
448            aggregators.insert(k.to_owned(), v.to_owned());
449        }
450
451        let context = Context {
452            context: self.context.clone(),
453            executor,
454            id: id.to_owned(),
455            aggregate_type: self.aggregate_type.to_string(),
456            aggregators: &aggregators,
457            revision: self.revision,
458        };
459        let snapshot = P::restore(&context).await?;
460        let cursor = snapshot.as_ref().map(|s| s.get_cursor());
461
462        let read_aggregators = self
463            .handlers
464            .values()
465            .map(|h| match aggregators.get(h.aggregate_type()) {
466                Some(id) => EventFilter {
467                    aggregate_type: h.aggregate_type().to_owned(),
468                    aggregate_id: Some(id.to_owned()),
469                    name: if self.safety_disabled {
470                        Some(h.event_name().to_owned())
471                    } else {
472                        None
473                    },
474                },
475                _ => {
476                    if self.safety_disabled {
477                        EventFilter::by_event(h.aggregate_type(), h.event_name())
478                    } else {
479                        EventFilter::by_type(h.aggregate_type())
480                    }
481                }
482            })
483            .collect::<Vec<_>>();
484
485        let events = executor
486            .read(
487                Some(read_aggregators.to_vec()),
488                None,
489                Args::forward(100, cursor.clone()),
490            )
491            .await?;
492
493        if events.edges.is_empty() && snapshot.is_none() {
494            return Ok(None);
495        }
496
497        let mut snapshot = snapshot.unwrap_or_default();
498
499        for event in events.edges.iter() {
500            let key = format!("{}_{}", event.node.aggregate_type, event.node.name);
501
502            let Some(handler) = self.handlers.get(&key) else {
503                if !self.safety_disabled {
504                    anyhow::bail!("no handler k={key}");
505                }
506
507                continue;
508            };
509
510            handler.handle(&mut snapshot, &event.node).await?;
511        }
512
513        if let Some(event) = events.edges.last() {
514            snapshot.set_cursor(&event.cursor);
515            snapshot.take_snapshot(&context).await?;
516        }
517
518        if events.page_info.has_next_page {
519            anyhow::bail!("Too busy");
520        }
521
522        Ok(Some(snapshot))
523    }
524}
525
526/// Builder for loading the projection state of a specific aggregate id.
527///
528/// Created via [`Projection::load`]. Allows registering related aggregates
529/// whose events also feed this projection, then [`LoadBuilder::execute`] runs
530/// the load.
531pub struct LoadBuilder<E: Executor, P: Default + 'static> {
532    projection: Projection<E, P>,
533    id: String,
534    aggregators: HashMap<String, String>,
535}
536
537impl<E: Executor, P: Snapshot<E> + Default + 'static> LoadBuilder<E, P> {
538    /// Adds a related aggregate to load events from.
539    pub fn aggregate<A: Aggregate>(self, id: impl Into<String>) -> Self {
540        self.aggregate_raw(A::aggregate_type().to_owned(), id)
541    }
542
543    /// Adds a related aggregate to load events from (raw aggregate type).
544    pub fn aggregate_raw(
545        mut self,
546        aggregate_type: impl Into<String>,
547        id: impl Into<String>,
548    ) -> Self {
549        self.aggregators.insert(aggregate_type.into(), id.into());
550
551        self
552    }
553
554    /// Executes the load, returning the rebuilt state.
555    ///
556    /// Returns `None` if no events exist for the aggregate, or if a tombstone
557    /// was registered via [`Projection::tombstone`] and the corresponding
558    /// event has been committed for this id.
559    /// Returns `Err` if there are too many events to process in one batch.
560    pub async fn execute(&self, executor: &E) -> anyhow::Result<Option<P>> {
561        self.projection
562            .load_aggregator(executor, &self.id, &self.aggregators)
563            .await
564    }
565}
566
567/// Builder for a subscription that keeps a [`Projection`] auto-updated.
568///
569/// Created via [`Projection::subscription`]. On each incoming event, the
570/// subscription loads the affected aggregate id through the projection,
571/// which in turn re-runs handlers and persists the snapshot.
572pub struct ProjectionSubscription<E: Executor, P: Default + 'static> {
573    projection: Projection<E, P>,
574    key: String,
575    routing_key: Option<RoutingKey>,
576    chunk_size: u16,
577    retry: Option<u8>,
578    delay: Option<Duration>,
579    continue_on_error: bool,
580}
581
582impl<E, P> ProjectionSubscription<E, P>
583where
584    E: Executor + Clone + 'static,
585    P: Snapshot<E> + Default + Send + Sync + 'static,
586{
587    /// Filters events by routing key.
588    ///
589    /// Overrides any executor-level default.
590    pub fn routing_key(mut self, v: impl Into<String>) -> Self {
591        self.routing_key = Some(RoutingKey::Value(Some(v.into())));
592        self
593    }
594
595    /// Processes all events regardless of routing key.
596    ///
597    /// Overrides any executor-level default.
598    pub fn all(mut self) -> Self {
599        self.routing_key = Some(RoutingKey::All);
600        self
601    }
602
603    /// Sets the number of events to process per batch (default 300).
604    pub fn chunk_size(mut self, v: u16) -> Self {
605        self.chunk_size = v;
606        self
607    }
608
609    /// Sets the maximum number of retries on failure (default 30).
610    pub fn retry(mut self, v: u8) -> Self {
611        self.retry = Some(v);
612        self
613    }
614
615    /// Disables retries.
616    pub fn no_retry(mut self) -> Self {
617        self.retry = None;
618        self
619    }
620
621    /// Sets a delay before starting the subscription.
622    pub fn delay(mut self, v: Duration) -> Self {
623        self.delay = Some(v);
624        self
625    }
626
627    /// Allows the subscription to continue after handler failures.
628    pub fn continue_on_error(mut self) -> Self {
629        self.continue_on_error = true;
630        self
631    }
632
633    /// Starts the subscription.
634    ///
635    /// Returns a [`Subscription`] handle that can be used for graceful shutdown.
636    pub async fn start(self, executor: &E) -> anyhow::Result<Subscription> {
637        let ProjectionSubscription {
638            projection,
639            key,
640            routing_key,
641            chunk_size,
642            retry,
643            delay,
644            continue_on_error,
645        } = self;
646
647        let aggregate_type: &'static str = projection.aggregate_type;
648        let tombstone = projection.tombstone;
649        // Capture the event names registered for the primary aggregator type.
650        // Each event_name is &'static str because Handler::event_name returns
651        // &'static str (always derived from the `#[evento::handler]` macro).
652        // Drop any handler that collides with the tombstone — that event is
653        // routed to ProjectionTombstoneHandler instead.
654        let event_names: Vec<&'static str> = projection
655            .handlers
656            .values()
657            .filter(|h| h.aggregate_type() == aggregate_type)
658            .filter(|h| {
659                tombstone
660                    .map(|(t, e)| !(h.aggregate_type() == t && h.event_name() == e))
661                    .unwrap_or(true)
662            })
663            .map(|h| h.event_name())
664            .collect();
665
666        let projection = Arc::new(projection);
667
668        let mut builder: SubscriptionBuilder<E> = SubscriptionBuilder::new(key);
669        builder = match routing_key {
670            Some(RoutingKey::All) => builder.all(),
671            Some(RoutingKey::Value(Some(v))) => builder.routing_key(v),
672            Some(RoutingKey::Value(None)) | None => builder,
673        };
674        builder = builder.chunk_size(chunk_size);
675        builder = match retry {
676            Some(n) => builder.retry(n),
677            None => builder,
678        };
679        if let Some(d) = delay {
680            builder = builder.delay(d);
681        }
682        if continue_on_error {
683            builder = builder.continue_on_error();
684        }
685
686        for event_name in event_names {
687            builder = builder.handler(ProjectionAutoHandler::<E, P> {
688                projection: projection.clone(),
689                aggregate_type,
690                event_name,
691                _marker: PhantomData,
692            });
693        }
694
695        if let Some((tombstone_type, tombstone_event)) = tombstone {
696            builder = builder.handler(ProjectionTombstoneHandler::<E, P> {
697                projection: projection.clone(),
698                aggregate_type: tombstone_type,
699                event_name: tombstone_event,
700                _marker: PhantomData,
701            });
702        }
703
704        if retry.is_none() {
705            builder.no_retry().start(executor).await
706        } else {
707            builder.start(executor).await
708        }
709    }
710}
711
712struct ProjectionAutoHandler<E: Executor, P: Default + 'static> {
713    projection: Arc<Projection<E, P>>,
714    aggregate_type: &'static str,
715    event_name: &'static str,
716    _marker: PhantomData<E>,
717}
718
719impl<E, P> subscription::Handler<E> for ProjectionAutoHandler<E, P>
720where
721    E: Executor,
722    P: Snapshot<E> + Default + Send + Sync + 'static,
723{
724    fn handle<'a>(
725        &'a self,
726        context: &'a subscription::Context<'a, E>,
727        event: &'a crate::Event,
728    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>> {
729        Box::pin(async move {
730            self.projection
731                .load_aggregator(context.executor, &event.aggregate_id, &HashMap::new())
732                .await?;
733            Ok(())
734        })
735    }
736
737    fn aggregate_type(&self) -> &'static str {
738        self.aggregate_type
739    }
740
741    fn event_name(&self) -> &'static str {
742        self.event_name
743    }
744}
745
746struct ProjectionTombstoneHandler<E: Executor, P: Default + 'static> {
747    projection: Arc<Projection<E, P>>,
748    aggregate_type: &'static str,
749    event_name: &'static str,
750    _marker: PhantomData<E>,
751}
752
753impl<E, P> subscription::Handler<E> for ProjectionTombstoneHandler<E, P>
754where
755    E: Executor,
756    P: Snapshot<E> + Default + Send + Sync + 'static,
757{
758    fn handle<'a>(
759        &'a self,
760        context: &'a subscription::Context<'a, E>,
761        event: &'a crate::Event,
762    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>> {
763        Box::pin(async move {
764            let aggregators: HashMap<String, String> = HashMap::new();
765            let ctx = Context {
766                context: self.projection.context.clone(),
767                executor: context.executor,
768                id: event.aggregate_id.clone(),
769                aggregate_type: self.projection.aggregate_type.to_string(),
770                aggregators: &aggregators,
771                revision: self.projection.revision,
772            };
773            P::drop_snapshot(&ctx).await?;
774            Ok(())
775        })
776    }
777
778    fn aggregate_type(&self) -> &'static str {
779        self.aggregate_type
780    }
781
782    fn event_name(&self) -> &'static str {
783        self.event_name
784    }
785}
786
787pub(crate) struct SkipHandler<E: AggregateEvent>(PhantomData<E>);
788
789impl<P: 'static, EV: AggregateEvent + Send + Sync> Handler<P> for SkipHandler<EV> {
790    fn handle<'a>(
791        &'a self,
792        _projection: &'a mut P,
793        _event: &'a crate::Event,
794    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>> {
795        Box::pin(async { Ok(()) })
796    }
797
798    fn aggregate_type(&self) -> &'static str {
799        EV::aggregate_type()
800    }
801
802    fn event_name(&self) -> &'static str {
803        EV::event_name()
804    }
805}
evento_core/projection.rs

evento_core/
projection.rs
