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 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
516 let mut read_aggregators = vec![];
517 for handler in self.handlers.values() {
518 let Some(id) = self.aggregators.get(handler.aggregator_type()) else {
519 anyhow::bail!(
520 "Failed to load projection {}/{}: id not found",
521 handler.aggregator_type(),
522 handler.event_name()
523 );
524 };
525
526 read_aggregators.push(ReadAggregator {
527 aggregator_type: handler.aggregator_type().to_owned(),
528 aggregator_id: Some(id.to_owned()),
529 name: if self.filter_events_by_name {
530 Some(handler.event_name().to_owned())
531 } else {
532 None
533 },
534 });
535 }
536
537 let events = executor
538 .read(
539 Some(read_aggregators.to_vec()),
540 None,
541 Args::forward(100, cursor.clone()),
542 )
543 .await?;
544
545 if events.edges.is_empty() && loaded.is_none() {
546 return Ok(None);
547 }
548
549 let mut snapshot = loaded.unwrap_or_default();
550
551 for event in events.edges.iter() {
552 let key = format!("{}_{}", event.node.aggregator_type, event.node.name);
553 let Some(handler) = self.handlers.get(&key) else {
554 tracing::debug!("No handler found for {}/{key}", self.key);
555 continue;
556 };
557
558 handler.apply(&mut snapshot, &event.node).await?;
559 }
560
561 if events.page_info.has_next_page {
562 anyhow::bail!("Too busy");
563 }
564
565 if let Some(event) = events.edges.last() {
566 version = event.node.version;
567 routing_key = event.node.routing_key.to_owned();
568 }
569
570 Ok(Some(LoadResult {
571 item: snapshot,
572 version,
573 routing_key,
574 }))
575 }
576}
577
578/// Builder for creating event subscriptions.
579///
580/// Created via [`Projection::subscription`], this builder configures
581/// a continuous event processing subscription with retry logic,
582/// routing key filtering, and graceful shutdown support.
583///
584/// # Example
585///
586/// ```rust,ignore
587/// let subscription = projection
588/// .subscription()
589/// .routing_key("accounts")
590/// .chunk_size(100)
591/// .retry(5)
592/// .delay(Duration::from_secs(10))
593/// .start(&executor)
594/// .await?;
595///
596/// // Later, gracefully shutdown
597/// subscription.shutdown().await?;
598/// ```
599pub struct SubscriptionBuilder<P: 'static, E: Executor> {
600 key: String,
601 handlers: HashMap<String, Box<dyn Handler<P, E>>>,
602 context: context::RwContext,
603 routing_key: RoutingKey,
604 delay: Option<Duration>,
605 chunk_size: u16,
606 is_accept_failure: bool,
607 retry: Option<u8>,
608 aggregators: HashMap<String, String>,
609}
610
611impl<P, E: Executor + 'static> SubscriptionBuilder<P, E> {
612 /// Adds shared data to the load context.
613 ///
614 /// Data added here is accessible in handlers via the context.
615 pub fn data<D: Send + Sync + 'static>(self, v: D) -> Self {
616 self.context.insert(v);
617
618 self
619 }
620
621 /// Allows the subscription to continue after handler failures.
622 ///
623 /// By default, subscriptions stop on the first error. With this flag,
624 /// errors are logged but processing continues.
625 pub fn accept_failure(mut self) -> Self {
626 self.is_accept_failure = true;
627
628 self
629 }
630
631 /// Sets the number of events to process per batch.
632 ///
633 /// Default is 300.
634 pub fn chunk_size(mut self, v: u16) -> Self {
635 self.chunk_size = v;
636
637 self
638 }
639
640 /// Sets a delay before starting the subscription.
641 ///
642 /// Useful for staggering subscription starts in multi-node deployments.
643 pub fn delay(mut self, v: Duration) -> Self {
644 self.delay = Some(v);
645
646 self
647 }
648
649 /// Filters events by routing key.
650 ///
651 /// Only events with the matching routing key will be processed.
652 pub fn routing_key(mut self, v: impl Into<String>) -> Self {
653 self.routing_key = RoutingKey::Value(Some(v.into()));
654
655 self
656 }
657
658 /// Sets the maximum number of retries on failure.
659 ///
660 /// Uses exponential backoff. Default is 30.
661 pub fn retry(mut self, v: u8) -> Self {
662 self.retry = Some(v);
663
664 self
665 }
666
667 fn without_retry(mut self) -> Self {
668 self.retry = None;
669
670 self
671 }
672
673 /// Processes all events regardless of routing key.
674 pub fn all(mut self) -> Self {
675 self.routing_key = RoutingKey::All;
676
677 self
678 }
679
680 /// Adds a related aggregate to process events from.
681 pub fn aggregator<A: Aggregator>(mut self, id: impl Into<String>) -> Self {
682 self.aggregators
683 .insert(A::aggregator_type().to_owned(), id.into());
684
685 self
686 }
687
688 fn read_aggregators(&self) -> Vec<ReadAggregator> {
689 self.handlers
690 .values()
691 .map(|h| match self.aggregators.get(h.aggregator_type()) {
692 Some(id) => ReadAggregator {
693 aggregator_type: h.aggregator_type().to_owned(),
694 aggregator_id: Some(id.to_owned()),
695 name: Some(h.event_name().to_owned()),
696 },
697 _ => ReadAggregator::event(h.aggregator_type(), h.event_name()),
698 })
699 .collect()
700 }
701
702 fn key(&self) -> String {
703 if let RoutingKey::Value(Some(ref key)) = self.routing_key {
704 return format!("{key}.{}", self.key);
705 }
706
707 self.key.to_owned()
708 }
709
710 async fn process(
711 &self,
712 executor: &E,
713 id: &Ulid,
714 aggregators: &[ReadAggregator],
715 mut rx: Option<&mut tokio::sync::oneshot::Receiver<()>>,
716 ) -> anyhow::Result<()> {
717 let mut interval = interval_at(
718 Instant::now() - Duration::from_millis(400),
719 Duration::from_millis(300),
720 );
721
722 loop {
723 interval.tick().await;
724
725 if !executor.is_subscriber_running(self.key(), *id).await? {
726 return Ok(());
727 }
728
729 let cursor = executor.get_subscriber_cursor(self.key()).await?;
730
731 let timestamp = executor
732 .read(
733 Some(aggregators.to_vec()),
734 Some(self.routing_key.to_owned()),
735 Args::backward(1, None),
736 )
737 .await?
738 .edges
739 .last()
740 .map(|e| e.node.timestamp)
741 .unwrap_or_default();
742
743 let res = executor
744 .read(
745 Some(aggregators.to_vec()),
746 Some(self.routing_key.to_owned()),
747 Args::forward(self.chunk_size, cursor),
748 )
749 .await?;
750
751 if res.edges.is_empty() {
752 return Ok(());
753 }
754
755 let context = Context {
756 context: self.context.clone(),
757 executor,
758 };
759
760 for event in res.edges {
761 if let Some(rx) = rx.as_mut() {
762 if rx.try_recv().is_ok() {
763 tracing::info!(
764 key = self.key(),
765 "Subscription received shutdown signal, stopping gracefull"
766 );
767
768 return Ok(());
769 }
770 }
771
772 tracing::Span::current().record("aggregator_type", &event.node.aggregator_type);
773 tracing::Span::current().record("aggregator_id", &event.node.aggregator_id);
774 tracing::Span::current().record("event", &event.node.name);
775
776 let key = format!("{}_{}", event.node.aggregator_type, event.node.name);
777 let Some(handler) = self.handlers.get(&key) else {
778 panic!("No handler found for {}/{key}", self.key());
779 };
780
781 handler.handle(&context, &event.node).await?;
782
783 executor
784 .acknowledge(
785 self.key(),
786 event.cursor.to_owned(),
787 timestamp - event.node.timestamp,
788 )
789 .await?;
790 }
791 }
792 }
793
794 /// Starts the subscription without retry logic.
795 ///
796 /// Equivalent to calling `start()` with retries disabled.
797 pub async fn unretry_start(self, executor: &E) -> anyhow::Result<Subscription>
798 where
799 E: Clone,
800 {
801 self.without_retry().start(executor).await
802 }
803
804 /// Starts a continuous background subscription.
805 ///
806 /// Returns a [`Subscription`] handle that can be used for graceful shutdown.
807 /// The subscription runs in a spawned tokio task and polls for new events.
808 pub async fn start(self, executor: &E) -> anyhow::Result<Subscription>
809 where
810 E: Clone,
811 {
812 let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
813 let executor = executor.clone();
814 let id = Ulid::new();
815 let subscription_id = id;
816
817 executor
818 .upsert_subscriber(self.key(), id.to_owned())
819 .await?;
820
821 let task_handle = tokio::spawn(async move {
822 let read_aggregators = self.read_aggregators();
823 let start = self
824 .delay
825 .map(|d| Instant::now() + d)
826 .unwrap_or_else(Instant::now);
827
828 let mut interval = interval_at(
829 start - Duration::from_millis(1200),
830 Duration::from_millis(1000),
831 );
832
833 loop {
834 if shutdown_rx.try_recv().is_ok() {
835 tracing::info!(
836 key = self.key(),
837 "Subscription received shutdown signal, stopping gracefull"
838 );
839
840 break;
841 }
842
843 interval.tick().await;
844
845 let _ = tracing::error_span!(
846 "start",
847 key = self.key(),
848 aggregator_type = tracing::field::Empty,
849 aggregator_id = tracing::field::Empty,
850 event = tracing::field::Empty,
851 )
852 .entered();
853
854 let result = match self.retry {
855 Some(retry) => {
856 (|| async { self.process(&executor, &id, &read_aggregators, None).await })
857 .retry(ExponentialBuilder::default().with_max_times(retry.into()))
858 .sleep(tokio::time::sleep)
859 .notify(|err, dur| {
860 tracing::error!(
861 error = %err,
862 duration = ?dur,
863 "Failed to process event"
864 );
865 })
866 .await
867 }
868 _ => self.process(&executor, &id, &read_aggregators, None).await,
869 };
870
871 let Err(err) = result else {
872 continue;
873 };
874
875 tracing::error!(error = %err, "Failed to process event");
876
877 if !self.is_accept_failure {
878 break;
879 }
880 }
881 });
882
883 Ok(Subscription {
884 id: subscription_id,
885 task_handle,
886 shutdown_tx,
887 })
888 }
889
890 /// Executes the subscription once without retry logic.
891 ///
892 /// Processes all pending events and returns. Does not poll for new events.
893 pub async fn unretry_execute(self, executor: &E) -> anyhow::Result<()> {
894 self.without_retry().execute(executor).await
895 }
896
897 /// Executes the subscription once, processing all pending events.
898 ///
899 /// Unlike `start()`, this does not run continuously. It processes
900 /// all currently pending events and returns.
901 pub async fn execute(&self, executor: &E) -> anyhow::Result<()> {
902 let id = Ulid::new();
903
904 executor
905 .upsert_subscriber(self.key(), id.to_owned())
906 .await?;
907
908 let read_aggregators = self.read_aggregators();
909
910 let _ = tracing::error_span!(
911 "execute",
912 key = self.key(),
913 aggregator_type = tracing::field::Empty,
914 aggregator_id = tracing::field::Empty,
915 event = tracing::field::Empty,
916 )
917 .entered();
918
919 match self.retry {
920 Some(retry) => {
921 (|| async { self.process(executor, &id, &read_aggregators, None).await })
922 .retry(ExponentialBuilder::default().with_max_times(retry.into()))
923 .sleep(tokio::time::sleep)
924 .notify(|err, dur| {
925 tracing::error!(
926 error = %err,
927 duration = ?dur,
928 "Failed to process event"
929 );
930 })
931 .await
932 }
933 _ => self.process(executor, &id, &read_aggregators, None).await,
934 }
935 }
936}
937
938/// Handle to a running event subscription.
939///
940/// Returned by [`SubscriptionBuilder::start`], this handle provides
941/// the subscription ID and a method for graceful shutdown.
942///
943/// # Example
944///
945/// ```rust,ignore
946/// let subscription = projection
947/// .subscription()
948/// .start(&executor)
949/// .await?;
950///
951/// println!("Started subscription: {}", subscription.id);
952///
953/// // On application shutdown
954/// subscription.shutdown().await?;
955/// ```
956#[derive(Debug)]
957pub struct Subscription {
958 /// Unique ID for this subscription instance
959 pub id: Ulid,
960 task_handle: tokio::task::JoinHandle<()>,
961 shutdown_tx: tokio::sync::oneshot::Sender<()>,
962}
963
964impl Subscription {
965 /// Gracefully shuts down the subscription.
966 ///
967 /// Signals the subscription to stop and waits for it to finish
968 /// processing the current event before returning.
969 pub async fn shutdown(self) -> Result<(), tokio::task::JoinError> {
970 let _ = self.shutdown_tx.send(());
971
972 self.task_handle.await
973 }
974}