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//! - [`SubscriptionBuilder`] - Builds continuous event subscriptions
13//! - [`Subscription`] - Handle to a running subscription
14//! - [`EventData`] - Typed event with deserialized data and metadata
15//!
16//! # Example
17//!
18//! ```rust,ignore
19//! use evento::projection::Projection;
20//!
21//! // Define a projection with event handlers
22//! let projection = Projection::<AccountView, _>::new("accounts")
23//! .handler(account_opened)
24//! .handler(money_deposited);
25//!
26//! // Load aggregate state
27//! let result = projection
28//! .load::<Account>("account-123")
29//! .execute(&executor)
30//! .await?;
31//!
32//! // Or start a subscription
33//! let subscription = projection
34//! .subscription()
35//! .routing_key("accounts")
36//! .start(&executor)
37//! .await?;
38//! ```
39
40use std::{
41 collections::HashMap,
42 future::Future,
43 ops::{Deref, DerefMut},
44 pin::Pin,
45 time::Duration,
46};
47use tokio::time::{interval_at, Instant};
48use ulid::Ulid;
49
50use backon::{ExponentialBuilder, Retryable};
51
52use crate::{
53 context,
54 cursor::{Args, Cursor},
55 Executor, ReadAggregator,
56};
57
58/// Filter for events by routing key.
59///
60/// Routing keys allow partitioning events for parallel processing
61/// or filtering subscriptions to specific event streams.
62#[derive(Clone)]
63pub enum RoutingKey {
64 /// Match all events regardless of routing key
65 All,
66 /// Match events with a specific routing key (or no key if `None`)
67 Value(Option<String>),
68}
69
70/// Handler context providing access to executor and shared data.
71///
72/// `Context` wraps an [`RwContext`](crate::context::RwContext) for type-safe
73/// data storage and provides access to the executor for database operations.
74///
75/// # Example
76///
77/// ```rust,ignore
78/// #[evento::handler]
79/// async fn my_handler<E: Executor>(
80/// event: Event<MyEventData>,
81/// action: Action<'_, MyView, E>,
82/// ) -> anyhow::Result<()> {
83/// if let Action::Handle(ctx) = action {
84/// // Access shared data
85/// let config: Data<AppConfig> = ctx.extract();
86///
87/// // Use executor for queries
88/// let events = ctx.executor.read(...).await?;
89/// }
90/// Ok(())
91/// }
92/// ```
93#[derive(Clone)]
94pub struct Context<'a, E: Executor> {
95 context: context::RwContext,
96 /// Reference to the executor for database operations
97 pub executor: &'a E,
98}
99
100impl<'a, E: Executor> Deref for Context<'a, E> {
101 type Target = context::RwContext;
102
103 fn deref(&self) -> &Self::Target {
104 &self.context
105 }
106}
107
108/// Trait for aggregate types.
109///
110/// Aggregates are the root entities in event sourcing. Each aggregate
111/// type has a unique identifier string used for event storage and routing.
112///
113/// This trait is typically derived using the `#[evento::aggregator]` macro.
114///
115/// # Example
116///
117/// ```rust,ignore
118/// #[evento::aggregator("myapp/Account")]
119/// #[derive(Default)]
120/// pub struct Account {
121/// pub balance: i64,
122/// pub owner: String,
123/// }
124/// ```
125pub trait Aggregator: Default {
126 /// Returns the unique type identifier for this aggregate (e.g., "myapp/Account")
127 fn aggregator_type() -> &'static str;
128}
129
130/// Trait for event types.
131///
132/// Events represent state changes that have occurred. Each event type
133/// has a name and belongs to an aggregator type.
134///
135/// This trait is typically derived using the `#[evento::aggregator]` macro.
136///
137/// # Example
138///
139/// ```rust,ignore
140/// #[evento::aggregator("myapp/Account")]
141/// #[derive(bitcode::Encode, bitcode::Decode)]
142/// pub struct AccountOpened {
143/// pub owner: String,
144/// }
145/// ```
146pub trait Event: Aggregator {
147 /// Returns the event name (e.g., "AccountOpened")
148 fn event_name() -> &'static str;
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, E: Executor>: Sync + Send {
159 /// Handles an event during subscription processing.
160 ///
161 /// This is called when processing events in a subscription context,
162 /// where side effects like database updates or API calls are appropriate.
163 fn handle<'a>(
164 &'a self,
165 context: &'a Context<'a, E>,
166 event: &'a crate::Event,
167 ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
168
169 /// Applies an event to build projection state.
170 ///
171 /// This is called when loading aggregate state by replaying events.
172 /// It should be a pure function that modifies the projection without side effects.
173 fn apply<'a>(
174 &'a self,
175 projection: &'a mut P,
176 event: &'a crate::Event,
177 ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
178
179 /// Returns the aggregator type this handler processes.
180 fn aggregator_type(&self) -> &'static str;
181 /// Returns the event name this handler processes.
182 fn event_name(&self) -> &'static str;
183}
184
185/// Action passed to event handlers.
186///
187/// Determines whether the handler should apply state changes or
188/// handle the event with side effects.
189pub enum Action<'a, P: 'static, E: Executor> {
190 /// Apply event to projection state (for loading)
191 Apply(&'a mut P),
192 /// Handle event with context (for subscriptions)
193 Handle(&'a Context<'a, E>),
194}
195
196/// Typed event with deserialized data and metadata.
197///
198/// `EventData` wraps a raw [`Event`](crate::Event) and provides typed access
199/// to the deserialized event data and metadata. It implements `Deref` to
200/// provide access to the underlying event fields (id, timestamp, version, etc.).
201///
202/// # Type Parameters
203///
204/// - `D`: The event data type (e.g., `AccountOpened`)
205/// - `M`: The metadata type (defaults to `bool` for no metadata)
206///
207/// # Example
208///
209/// ```rust,ignore
210/// use evento::metadata::Event;
211///
212/// #[evento::handler]
213/// async fn handle_deposit<E: Executor>(
214/// event: Event<MoneyDeposited>,
215/// action: Action<'_, AccountView, E>,
216/// ) -> anyhow::Result<()> {
217/// // Access typed data
218/// println!("Amount: {}", event.data.amount);
219///
220/// // Access metadata
221/// if let Ok(user) = event.metadata.user() {
222/// println!("By user: {}", user);
223/// }
224///
225/// // Access underlying event fields via Deref
226/// println!("Event ID: {}", event.id);
227/// println!("Version: {}", event.version);
228///
229/// Ok(())
230/// }
231/// ```
232pub struct EventData<D, M = bool> {
233 event: crate::Event,
234 /// The typed event data
235 pub data: D,
236 /// The typed event metadata
237 pub metadata: M,
238}
239
240impl<D, M> Deref for EventData<D, M> {
241 type Target = crate::Event;
242
243 fn deref(&self) -> &Self::Target {
244 &self.event
245 }
246}
247
248impl<D, M> TryFrom<&crate::Event> for EventData<D, M>
249where
250 D: bitcode::DecodeOwned,
251 M: bitcode::DecodeOwned,
252{
253 type Error = bitcode::Error;
254
255 fn try_from(value: &crate::Event) -> Result<Self, Self::Error> {
256 let data = bitcode::decode::<D>(&value.data)?;
257 let metadata = bitcode::decode::<M>(&value.metadata)?;
258 Ok(EventData {
259 data,
260 metadata,
261 event: value.clone(),
262 })
263 }
264}
265
266/// Container for event handlers that build a projection.
267///
268/// A `Projection` groups related event handlers together and provides
269/// methods to load aggregate state or create subscriptions.
270///
271/// # Type Parameters
272///
273/// - `P`: The projection/view type being built
274/// - `E`: The executor type for database operations
275///
276/// # Example
277///
278/// ```rust,ignore
279/// let projection = Projection::<AccountView, _>::new("accounts")
280/// .handler(account_opened)
281/// .handler(money_deposited)
282/// .handler(money_withdrawn);
283///
284/// // Use for loading state
285/// let state = projection.clone()
286/// .load::<Account>("account-123")
287/// .execute(&executor)
288/// .await?;
289///
290/// // Or create a subscription
291/// let sub = projection
292/// .subscription()
293/// .start(&executor)
294/// .await?;
295/// ```
296pub struct Projection<P: 'static, E: Executor> {
297 key: String,
298 handlers: HashMap<String, Box<dyn Handler<P, E>>>,
299}
300
301impl<P: 'static, E: Executor> Projection<P, E> {
302 /// Creates a new projection with the given key.
303 ///
304 /// The key is used as the subscription identifier for cursor tracking.
305 pub fn new(key: impl Into<String>) -> Self {
306 Self {
307 key: key.into(),
308 handlers: HashMap::new(),
309 }
310 }
311
312 /// Registers an event handler with this projection.
313 ///
314 /// # Panics
315 ///
316 /// Panics if a handler for the same event type is already registered.
317 pub fn handler<H: Handler<P, E> + 'static>(mut self, h: H) -> Self {
318 let key = format!("{}_{}", h.aggregator_type(), h.event_name());
319 if self.handlers.insert(key.to_owned(), Box::new(h)).is_some() {
320 panic!("Cannot register event handler: key {} already exists", key);
321 }
322 self
323 }
324
325 /// Creates a builder for loading aggregate state.
326 ///
327 /// This consumes the projection and returns a [`LoadBuilder`] configured
328 /// to load the state for the specified aggregate.
329 ///
330 /// # Type Parameters
331 ///
332 /// - `A`: The aggregate type to load
333 pub fn load<A: Aggregator>(self, id: impl Into<String>) -> LoadBuilder<P, E>
334 where
335 P: Snapshot + Default,
336 {
337 let id = id.into();
338 let mut aggregators = HashMap::new();
339 aggregators.insert(A::aggregator_type().to_owned(), id.to_owned());
340
341 LoadBuilder {
342 key: self.key.to_owned(),
343 id,
344 aggregators,
345 handlers: self.handlers,
346 context: Default::default(),
347 filter_events_by_name: true,
348 }
349 }
350
351 /// Creates a builder for a continuous event subscription.
352 ///
353 /// This consumes the projection and returns a [`SubscriptionBuilder`]
354 /// that can be configured and started.
355 pub fn subscription(self) -> SubscriptionBuilder<P, E> {
356 SubscriptionBuilder {
357 key: self.key.to_owned(),
358 context: Default::default(),
359 handlers: self.handlers,
360 delay: None,
361 retry: Some(30),
362 chunk_size: 300,
363 is_accept_failure: false,
364 routing_key: RoutingKey::Value(None),
365 aggregators: Default::default(),
366 }
367 }
368}
369
370/// Result of loading an aggregate's state.
371///
372/// Contains the rebuilt projection state along with the current version
373/// and routing key. Implements `Deref` and `DerefMut` for transparent
374/// access to the inner item.
375///
376/// # Example
377///
378/// ```rust,ignore
379/// let result: LoadResult<AccountView> = projection
380/// .load::<Account>("account-123")
381/// .execute(&executor)
382/// .await?
383/// .expect("Account not found");
384///
385/// // Access inner item via Deref
386/// println!("Balance: {}", result.balance);
387///
388/// // Access metadata
389/// println!("Version: {}", result.version);
390/// println!("Routing key: {:?}", result.routing_key);
391/// ```
392#[derive(Debug, Clone, Default)]
393pub struct LoadResult<A> {
394 /// The loaded projection/view state
395 pub item: A,
396 /// Current version of the aggregate
397 pub version: u16,
398 /// Routing key for the aggregate (if set)
399 pub routing_key: Option<String>,
400}
401
402impl<A> Deref for LoadResult<A> {
403 type Target = A;
404 fn deref(&self) -> &Self::Target {
405 &self.item
406 }
407}
408
409impl<A> DerefMut for LoadResult<A> {
410 fn deref_mut(&mut self) -> &mut Self::Target {
411 &mut self.item
412 }
413}
414
415/// Trait for types that can be restored from snapshots.
416///
417/// Snapshots provide a performance optimization by storing pre-computed
418/// state, avoiding the need to replay all events from the beginning.
419///
420/// This trait is typically implemented via the `#[evento::snapshot]` macro.
421///
422/// # Example
423///
424/// ```rust,ignore
425/// #[evento::snapshot]
426/// #[derive(Default)]
427/// pub struct AccountView {
428/// pub balance: i64,
429/// pub owner: String,
430/// }
431///
432/// // The macro generates the restore implementation that loads
433/// // from a snapshot table if available
434/// ```
435pub trait Snapshot: Sized {
436 /// Restores state from a snapshot if available.
437 ///
438 /// Returns `None` if no snapshot exists for the given ID.
439 fn restore<'a>(
440 context: &'a context::RwContext,
441 id: String,
442 ) -> Pin<Box<dyn Future<Output = anyhow::Result<Option<Self>>> + Send + 'a>>;
443}
444
445/// Builder for loading aggregate state from events.
446///
447/// Created via [`Projection::load`], this builder configures how to
448/// load an aggregate's state by replaying events.
449///
450/// # Example
451///
452/// ```rust,ignore
453/// let result = projection
454/// .load::<Account>("account-123")
455/// .data(app_config) // Add shared data
456/// .aggregator::<User>("user-456") // Add related aggregate
457/// .execute(&executor)
458/// .await?;
459/// ```
460pub struct LoadBuilder<P: Snapshot + Default + 'static, E: Executor> {
461 key: String,
462 id: String,
463 aggregators: HashMap<String, String>,
464 handlers: HashMap<String, Box<dyn Handler<P, E>>>,
465 context: context::RwContext,
466 filter_events_by_name: bool,
467}
468
469impl<P: Snapshot + Default + 'static, E: Executor> LoadBuilder<P, E> {
470 /// Adds shared data to the load context.
471 ///
472 /// Data added here is accessible in handlers via the context.
473 pub fn data<D: Send + Sync + 'static>(&mut self, v: D) -> &mut Self {
474 self.context.insert(v);
475
476 self
477 }
478
479 /// Adds a related aggregate to load events from.
480 ///
481 /// Use this when the projection needs events from multiple aggregates.
482 pub fn aggregator<A: Aggregator>(&mut self, id: impl Into<String>) -> &mut Self {
483 self.aggregators
484 .insert(A::aggregator_type().to_owned(), id.into());
485
486 self
487 }
488
489 pub fn filter_events_by_name(&mut self, v: bool) -> &mut Self {
490 self.filter_events_by_name = v;
491
492 self
493 }
494
495 /// Executes the load operation, returning the rebuilt state.
496 ///
497 /// Returns `None` if no events exist for the aggregate.
498 /// Returns `Err` if there are too many events to process in one batch.
499 pub async fn execute(&self, executor: &E) -> anyhow::Result<Option<LoadResult<P>>> {
500 let context = Context {
501 context: self.context.clone(),
502 executor,
503 };
504
505 let mut cursor = executor.get_subscriber_cursor(self.key.to_owned()).await?;
506 let (mut version, mut routing_key) = match cursor {
507 Some(ref cursor) => {
508 let cursor = crate::Event::deserialize_cursor(cursor)?;
509
510 (cursor.v, cursor.r)
511 }
512 _ => (0, None),
513 };
514 let loaded = P::restore(&context, self.id.to_owned()).await?;
515 if loaded.is_none() {
516 cursor = None;
517 }
518
519 let mut read_aggregators = vec![];
520 for handler in self.handlers.values() {
521 let Some(id) = self.aggregators.get(handler.aggregator_type()) else {
522 anyhow::bail!(
523 "Failed to load projection {}/{}: id not found",
524 handler.aggregator_type(),
525 handler.event_name()
526 );
527 };
528
529 read_aggregators.push(ReadAggregator {
530 aggregator_type: handler.aggregator_type().to_owned(),
531 aggregator_id: Some(id.to_owned()),
532 name: if self.filter_events_by_name {
533 Some(handler.event_name().to_owned())
534 } else {
535 None
536 },
537 });
538 }
539
540 let events = executor
541 .read(
542 Some(read_aggregators.to_vec()),
543 None,
544 Args::forward(100, cursor.clone()),
545 )
546 .await?;
547
548 if events.edges.is_empty() && loaded.is_none() {
549 return Ok(None);
550 }
551
552 let mut snapshot = loaded.unwrap_or_default();
553
554 for event in events.edges.iter() {
555 let key = format!("{}_{}", event.node.aggregator_type, event.node.name);
556 let Some(handler) = self.handlers.get(&key) else {
557 tracing::debug!("No handler found for {}/{key}", self.key);
558 continue;
559 };
560
561 handler.apply(&mut snapshot, &event.node).await?;
562 }
563
564 if events.page_info.has_next_page {
565 anyhow::bail!("Too busy");
566 }
567
568 if let Some(event) = events.edges.last() {
569 version = event.node.version;
570 routing_key = event.node.routing_key.to_owned();
571 }
572
573 Ok(Some(LoadResult {
574 item: snapshot,
575 version,
576 routing_key,
577 }))
578 }
579}
580
581/// Builder for creating event subscriptions.
582///
583/// Created via [`Projection::subscription`], this builder configures
584/// a continuous event processing subscription with retry logic,
585/// routing key filtering, and graceful shutdown support.
586///
587/// # Example
588///
589/// ```rust,ignore
590/// let subscription = projection
591/// .subscription()
592/// .routing_key("accounts")
593/// .chunk_size(100)
594/// .retry(5)
595/// .delay(Duration::from_secs(10))
596/// .start(&executor)
597/// .await?;
598///
599/// // Later, gracefully shutdown
600/// subscription.shutdown().await?;
601/// ```
602pub struct SubscriptionBuilder<P: 'static, E: Executor> {
603 key: String,
604 handlers: HashMap<String, Box<dyn Handler<P, E>>>,
605 context: context::RwContext,
606 routing_key: RoutingKey,
607 delay: Option<Duration>,
608 chunk_size: u16,
609 is_accept_failure: bool,
610 retry: Option<u8>,
611 aggregators: HashMap<String, String>,
612}
613
614impl<P, E: Executor + 'static> SubscriptionBuilder<P, E> {
615 /// Adds shared data to the load context.
616 ///
617 /// Data added here is accessible in handlers via the context.
618 pub fn data<D: Send + Sync + 'static>(self, v: D) -> Self {
619 self.context.insert(v);
620
621 self
622 }
623
624 /// Allows the subscription to continue after handler failures.
625 ///
626 /// By default, subscriptions stop on the first error. With this flag,
627 /// errors are logged but processing continues.
628 pub fn accept_failure(mut self) -> Self {
629 self.is_accept_failure = true;
630
631 self
632 }
633
634 /// Sets the number of events to process per batch.
635 ///
636 /// Default is 300.
637 pub fn chunk_size(mut self, v: u16) -> Self {
638 self.chunk_size = v;
639
640 self
641 }
642
643 /// Sets a delay before starting the subscription.
644 ///
645 /// Useful for staggering subscription starts in multi-node deployments.
646 pub fn delay(mut self, v: Duration) -> Self {
647 self.delay = Some(v);
648
649 self
650 }
651
652 /// Filters events by routing key.
653 ///
654 /// Only events with the matching routing key will be processed.
655 pub fn routing_key(mut self, v: impl Into<String>) -> Self {
656 self.routing_key = RoutingKey::Value(Some(v.into()));
657
658 self
659 }
660
661 /// Sets the maximum number of retries on failure.
662 ///
663 /// Uses exponential backoff. Default is 30.
664 pub fn retry(mut self, v: u8) -> Self {
665 self.retry = Some(v);
666
667 self
668 }
669
670 fn without_retry(mut self) -> Self {
671 self.retry = None;
672
673 self
674 }
675
676 /// Processes all events regardless of routing key.
677 pub fn all(mut self) -> Self {
678 self.routing_key = RoutingKey::All;
679
680 self
681 }
682
683 /// Adds a related aggregate to process events from.
684 pub fn aggregator<A: Aggregator>(mut self, id: impl Into<String>) -> Self {
685 self.aggregators
686 .insert(A::aggregator_type().to_owned(), id.into());
687
688 self
689 }
690
691 fn read_aggregators(&self) -> Vec<ReadAggregator> {
692 self.handlers
693 .values()
694 .map(|h| match self.aggregators.get(h.aggregator_type()) {
695 Some(id) => ReadAggregator {
696 aggregator_type: h.aggregator_type().to_owned(),
697 aggregator_id: Some(id.to_owned()),
698 name: Some(h.event_name().to_owned()),
699 },
700 _ => ReadAggregator::event(h.aggregator_type(), h.event_name()),
701 })
702 .collect()
703 }
704
705 fn key(&self) -> String {
706 if let RoutingKey::Value(Some(ref key)) = self.routing_key {
707 return format!("{key}.{}", self.key);
708 }
709
710 self.key.to_owned()
711 }
712
713 async fn process(
714 &self,
715 executor: &E,
716 id: &Ulid,
717 aggregators: &[ReadAggregator],
718 mut rx: Option<&mut tokio::sync::oneshot::Receiver<()>>,
719 ) -> anyhow::Result<()> {
720 let mut interval = interval_at(
721 Instant::now() - Duration::from_millis(400),
722 Duration::from_millis(300),
723 );
724
725 loop {
726 interval.tick().await;
727
728 if !executor.is_subscriber_running(self.key(), *id).await? {
729 return Ok(());
730 }
731
732 let cursor = executor.get_subscriber_cursor(self.key()).await?;
733
734 let timestamp = executor
735 .read(
736 Some(aggregators.to_vec()),
737 Some(self.routing_key.to_owned()),
738 Args::backward(1, None),
739 )
740 .await?
741 .edges
742 .last()
743 .map(|e| e.node.timestamp)
744 .unwrap_or_default();
745
746 let res = executor
747 .read(
748 Some(aggregators.to_vec()),
749 Some(self.routing_key.to_owned()),
750 Args::forward(self.chunk_size, cursor),
751 )
752 .await?;
753
754 if res.edges.is_empty() {
755 return Ok(());
756 }
757
758 let context = Context {
759 context: self.context.clone(),
760 executor,
761 };
762
763 for event in res.edges {
764 if let Some(rx) = rx.as_mut() {
765 if rx.try_recv().is_ok() {
766 tracing::info!(
767 key = self.key(),
768 "Subscription received shutdown signal, stopping gracefull"
769 );
770
771 return Ok(());
772 }
773 }
774
775 tracing::Span::current().record("aggregator_type", &event.node.aggregator_type);
776 tracing::Span::current().record("aggregator_id", &event.node.aggregator_id);
777 tracing::Span::current().record("event", &event.node.name);
778
779 let key = format!("{}_{}", event.node.aggregator_type, event.node.name);
780 let Some(handler) = self.handlers.get(&key) else {
781 panic!("No handler found for {}/{key}", self.key());
782 };
783
784 handler.handle(&context, &event.node).await?;
785
786 executor
787 .acknowledge(
788 self.key(),
789 event.cursor.to_owned(),
790 timestamp - event.node.timestamp,
791 )
792 .await?;
793 }
794 }
795 }
796
797 /// Starts the subscription without retry logic.
798 ///
799 /// Equivalent to calling `start()` with retries disabled.
800 pub async fn unretry_start(self, executor: &E) -> anyhow::Result<Subscription>
801 where
802 E: Clone,
803 {
804 self.without_retry().start(executor).await
805 }
806
807 /// Starts a continuous background subscription.
808 ///
809 /// Returns a [`Subscription`] handle that can be used for graceful shutdown.
810 /// The subscription runs in a spawned tokio task and polls for new events.
811 pub async fn start(self, executor: &E) -> anyhow::Result<Subscription>
812 where
813 E: Clone,
814 {
815 let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
816 let executor = executor.clone();
817 let id = Ulid::new();
818 let subscription_id = id;
819
820 executor
821 .upsert_subscriber(self.key(), id.to_owned())
822 .await?;
823
824 let task_handle = tokio::spawn(async move {
825 let read_aggregators = self.read_aggregators();
826 let start = self
827 .delay
828 .map(|d| Instant::now() + d)
829 .unwrap_or_else(Instant::now);
830
831 let mut interval = interval_at(
832 start - Duration::from_millis(1200),
833 Duration::from_millis(1000),
834 );
835
836 loop {
837 if shutdown_rx.try_recv().is_ok() {
838 tracing::info!(
839 key = self.key(),
840 "Subscription received shutdown signal, stopping gracefull"
841 );
842
843 break;
844 }
845
846 interval.tick().await;
847
848 let _ = tracing::error_span!(
849 "start",
850 key = self.key(),
851 aggregator_type = tracing::field::Empty,
852 aggregator_id = tracing::field::Empty,
853 event = tracing::field::Empty,
854 )
855 .entered();
856
857 let result = match self.retry {
858 Some(retry) => {
859 (|| async { self.process(&executor, &id, &read_aggregators, None).await })
860 .retry(ExponentialBuilder::default().with_max_times(retry.into()))
861 .sleep(tokio::time::sleep)
862 .notify(|err, dur| {
863 tracing::error!(
864 error = %err,
865 duration = ?dur,
866 "Failed to process event"
867 );
868 })
869 .await
870 }
871 _ => self.process(&executor, &id, &read_aggregators, None).await,
872 };
873
874 let Err(err) = result else {
875 continue;
876 };
877
878 tracing::error!(error = %err, "Failed to process event");
879
880 if !self.is_accept_failure {
881 break;
882 }
883 }
884 });
885
886 Ok(Subscription {
887 id: subscription_id,
888 task_handle,
889 shutdown_tx,
890 })
891 }
892
893 /// Executes the subscription once without retry logic.
894 ///
895 /// Processes all pending events and returns. Does not poll for new events.
896 pub async fn unretry_execute(self, executor: &E) -> anyhow::Result<()> {
897 self.without_retry().execute(executor).await
898 }
899
900 /// Executes the subscription once, processing all pending events.
901 ///
902 /// Unlike `start()`, this does not run continuously. It processes
903 /// all currently pending events and returns.
904 pub async fn execute(&self, executor: &E) -> anyhow::Result<()> {
905 let id = Ulid::new();
906
907 executor
908 .upsert_subscriber(self.key(), id.to_owned())
909 .await?;
910
911 let read_aggregators = self.read_aggregators();
912
913 let _ = tracing::error_span!(
914 "execute",
915 key = self.key(),
916 aggregator_type = tracing::field::Empty,
917 aggregator_id = tracing::field::Empty,
918 event = tracing::field::Empty,
919 )
920 .entered();
921
922 match self.retry {
923 Some(retry) => {
924 (|| async { self.process(executor, &id, &read_aggregators, None).await })
925 .retry(ExponentialBuilder::default().with_max_times(retry.into()))
926 .sleep(tokio::time::sleep)
927 .notify(|err, dur| {
928 tracing::error!(
929 error = %err,
930 duration = ?dur,
931 "Failed to process event"
932 );
933 })
934 .await
935 }
936 _ => self.process(executor, &id, &read_aggregators, None).await,
937 }
938 }
939}
940
941/// Handle to a running event subscription.
942///
943/// Returned by [`SubscriptionBuilder::start`], this handle provides
944/// the subscription ID and a method for graceful shutdown.
945///
946/// # Example
947///
948/// ```rust,ignore
949/// let subscription = projection
950/// .subscription()
951/// .start(&executor)
952/// .await?;
953///
954/// println!("Started subscription: {}", subscription.id);
955///
956/// // On application shutdown
957/// subscription.shutdown().await?;
958/// ```
959#[derive(Debug)]
960pub struct Subscription {
961 /// Unique ID for this subscription instance
962 pub id: Ulid,
963 task_handle: tokio::task::JoinHandle<()>,
964 shutdown_tx: tokio::sync::oneshot::Sender<()>,
965}
966
967impl Subscription {
968 /// Gracefully shuts down the subscription.
969 ///
970 /// Signals the subscription to stop and waits for it to finish
971 /// processing the current event before returning.
972 pub async fn shutdown(self) -> Result<(), tokio::task::JoinError> {
973 let _ = self.shutdown_tx.send(());
974
975 self.task_handle.await
976 }
977}