evento_core/
subscription.rs

1//! Continuous event subscriptions.
2//!
3//! This module provides infrastructure for processing events continuously
4//! in the background with retry logic, routing key filtering, and graceful
5//! shutdown support.
6//!
7//! # Key Types
8//!
9//! - [`SubscriptionBuilder`] - Builds and configures event subscriptions
10//! - [`Subscription`] - Handle to a running subscription
11//! - [`Handler`] - Trait for event handlers
12//! - [`Context`] - Handler context with executor access
13//! - [`RoutingKey`] - Filter for event routing
14//!
15//! # Example
16//!
17//! ```rust,ignore
18//! use evento::subscription::SubscriptionBuilder;
19//!
20//! // Build a subscription with handlers
21//! let subscription = SubscriptionBuilder::new("my-subscription")
22//!     .handler(account_opened_handler)
23//!     .handler(money_deposited_handler)
24//!     .routing_key("accounts")
25//!     .chunk_size(100)
26//!     .retry(5)
27//!     .start(&executor)
28//!     .await?;
29//!
30//! // Later, gracefully shutdown
31//! subscription.shutdown().await?;
32//! ```
33
34use backon::{ExponentialBuilder, Retryable};
35use std::{
36    collections::HashMap, future::Future, marker::PhantomData, ops::Deref, pin::Pin, time::Duration,
37};
38use tokio::{
39    sync::{oneshot::Receiver, Mutex},
40    time::{interval_at, Instant},
41};
42use tracing::field::Empty;
43use ulid::Ulid;
44
45use crate::{context, cursor::Args, Aggregator, AggregatorEvent, Executor, ReadAggregator};
46
47/// Filter for events by routing key.
48///
49/// Routing keys allow partitioning events for parallel processing
50/// or filtering subscriptions to specific event streams.
51#[derive(Clone)]
52pub enum RoutingKey {
53    /// Match all events regardless of routing key
54    All,
55    /// Match events with a specific routing key (or no key if `None`)
56    Value(Option<String>),
57}
58
59/// Handler context providing access to executor and shared data.
60///
61/// `Context` wraps an [`RwContext`](crate::context::RwContext) for type-safe
62/// data storage and provides access to the executor for database operations.
63///
64/// # Example
65///
66/// ```rust,ignore
67/// #[evento::handler]
68/// async fn my_handler<E: Executor>(
69///     event: Event<MyEventData>,
70///     action: Action<'_, MyView, E>,
71/// ) -> anyhow::Result<()> {
72///     if let Action::Handle(ctx) = action {
73///         // Access shared data
74///         let config: Data<AppConfig> = ctx.extract();
75///
76///         // Use executor for queries
77///         let events = ctx.executor.read(...).await?;
78///     }
79///     Ok(())
80/// }
81/// ```
82#[derive(Clone)]
83pub struct Context<'a, E: Executor> {
84    context: context::RwContext,
85    /// Reference to the executor for database operations
86    pub executor: &'a E,
87}
88
89impl<'a, E: Executor> Deref for Context<'a, E> {
90    type Target = context::RwContext;
91
92    fn deref(&self) -> &Self::Target {
93        &self.context
94    }
95}
96
97/// Trait for event handlers.
98///
99/// Handlers process events in two modes:
100/// - `handle`: For subscriptions that perform side effects (send emails, update read models)
101/// - `apply`: For loading aggregate state by replaying events
102///
103/// This trait is typically implemented via the `#[evento::handler]` macro.
104pub trait Handler<E: Executor>: Sync + Send {
105    /// Handles an event during subscription processing.
106    ///
107    /// This is called when processing events in a subscription context,
108    /// where side effects like database updates or API calls are appropriate.
109    fn handle<'a>(
110        &'a self,
111        context: &'a Context<'a, E>,
112        event: &'a crate::Event,
113    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
114
115    /// Returns the aggregator type this handler processes.
116    fn aggregator_type(&self) -> &'static str;
117    /// Returns the event name this handler processes.
118    fn event_name(&self) -> &'static str;
119}
120
121/// Builder for creating event subscriptions.
122///
123/// Created via [`Projection::subscription`], this builder configures
124/// a continuous event processing subscription with retry logic,
125/// routing key filtering, and graceful shutdown support.
126///
127/// # Example
128///
129/// ```rust,ignore
130/// let subscription = projection
131///     .subscription()
132///     .routing_key("accounts")
133///     .chunk_size(100)
134///     .retry(5)
135///     .delay(Duration::from_secs(10))
136///     .start(&executor)
137///     .await?;
138///
139/// // Later, gracefully shutdown
140/// subscription.shutdown().await?;
141/// ```
142pub struct SubscriptionBuilder<E: Executor> {
143    key: String,
144    handlers: HashMap<String, Box<dyn Handler<E>>>,
145    context: context::RwContext,
146    routing_key: Option<RoutingKey>,
147    prefix_key: Option<String>,
148    delay: Option<Duration>,
149    chunk_size: u16,
150    is_accept_failure: bool,
151    retry: Option<u8>,
152    aggregators: HashMap<String, String>,
153    safety_disabled: bool,
154    shutdown_rx: Option<Mutex<Receiver<()>>>,
155}
156
157impl<E: Executor + 'static> SubscriptionBuilder<E> {
158    /// Creates a new projection with the given key.
159    ///
160    /// The key is used as the subscription identifier for cursor tracking.
161    pub fn new(key: impl Into<String>) -> Self {
162        Self {
163            key: key.into(),
164            handlers: HashMap::new(),
165            safety_disabled: true,
166            context: Default::default(),
167            delay: None,
168            retry: Some(30),
169            chunk_size: 300,
170            is_accept_failure: false,
171            routing_key: None,
172            prefix_key: None,
173            aggregators: Default::default(),
174            shutdown_rx: None,
175        }
176    }
177
178    /// Enables safety checks for unhandled events.
179    ///
180    /// When enabled, processing fails if an event is encountered without a handler.
181    pub fn safety_check(mut self) -> Self {
182        self.safety_disabled = false;
183
184        self
185    }
186
187    /// Registers an event handler with this subscription.
188    ///
189    /// # Panics
190    ///
191    /// Panics if a handler for the same event type is already registered.
192    pub fn handler<H: Handler<E> + 'static>(mut self, h: H) -> Self {
193        let key = format!("{}_{}", h.aggregator_type(), h.event_name());
194        if self.handlers.insert(key.to_owned(), Box::new(h)).is_some() {
195            panic!("Cannot register event handler: key {} already exists", key);
196        }
197        self
198    }
199
200    /// Registers a skip handler for an event type.
201    ///
202    /// Events of this type will be acknowledged but not processed.
203    ///
204    /// # Panics
205    ///
206    /// Panics if a handler for the same event type is already registered.
207    pub fn skip<EV: AggregatorEvent + Send + Sync + 'static>(self) -> Self {
208        self.handler(SkipHandler::<EV>(PhantomData))
209    }
210
211    /// Adds shared data to the subscription context.
212    ///
213    /// Data added here is accessible in handlers via the context.
214    pub fn data<D: Send + Sync + 'static>(self, v: D) -> Self {
215        self.context.insert(v);
216
217        self
218    }
219
220    /// Allows the subscription to continue after handler failures.
221    ///
222    /// By default, subscriptions stop on the first error. With this flag,
223    /// errors are logged but processing continues.
224    pub fn accept_failure(mut self) -> Self {
225        self.is_accept_failure = true;
226
227        self
228    }
229
230    /// Sets the number of events to process per batch.
231    ///
232    /// Default is 300.
233    pub fn chunk_size(mut self, v: u16) -> Self {
234        self.chunk_size = v;
235
236        self
237    }
238
239    /// Sets a delay before starting the subscription.
240    ///
241    /// Useful for staggering subscription starts in multi-node deployments.
242    pub fn delay(mut self, v: Duration) -> Self {
243        self.delay = Some(v);
244
245        self
246    }
247
248    /// Filters events by routing key.
249    ///
250    /// Only events with the matching routing key will be processed.
251    /// Overrides any executor-level default.
252    pub fn routing_key(mut self, v: impl Into<String>) -> Self {
253        self.routing_key = Some(RoutingKey::Value(Some(v.into())));
254
255        self
256    }
257
258    /// Sets the maximum number of retries on failure.
259    ///
260    /// Uses exponential backoff. Default is 30.
261    pub fn retry(mut self, v: u8) -> Self {
262        self.retry = Some(v);
263
264        self
265    }
266
267    /// Processes all events regardless of routing key.
268    ///
269    /// Overrides any executor-level default.
270    pub fn all(mut self) -> Self {
271        self.routing_key = Some(RoutingKey::All);
272
273        self
274    }
275
276    /// Adds a related aggregate to process events from.
277    pub fn aggregator<A: Aggregator>(mut self, id: impl Into<String>) -> Self {
278        self.aggregators
279            .insert(A::aggregator_type().to_owned(), id.into());
280
281        self
282    }
283
284    fn read_aggregators(&self) -> Vec<ReadAggregator> {
285        self.handlers
286            .values()
287            .map(|h| match self.aggregators.get(h.aggregator_type()) {
288                Some(id) => ReadAggregator {
289                    aggregator_type: h.aggregator_type().to_owned(),
290                    aggregator_id: Some(id.to_owned()),
291                    name: if self.safety_disabled {
292                        Some(h.event_name().to_owned())
293                    } else {
294                        None
295                    },
296                },
297                _ => {
298                    if self.safety_disabled {
299                        ReadAggregator::event(h.aggregator_type(), h.event_name())
300                    } else {
301                        ReadAggregator::aggregator(h.aggregator_type())
302                    }
303                }
304            })
305            .collect()
306    }
307
308    fn key(&self) -> String {
309        let prefix = match &self.routing_key {
310            Some(RoutingKey::Value(Some(k))) => Some(k.as_str()),
311            Some(RoutingKey::All) => self.prefix_key.as_deref(),
312            _ => None,
313        };
314        match prefix {
315            Some(p) => format!("{p}.{}", self.key),
316            None => self.key.to_owned(),
317        }
318    }
319
320    /// Resolves an unset routing key from the executor's default and captures
321    /// the default as a storage-key prefix.
322    ///
323    /// Called once at the top of `start()` / `execute()` before any other
324    /// method reads `self.routing_key`. After this, `routing_key` is always
325    /// `Some(_)`.
326    ///
327    /// `prefix_key` is captured from `executor.default_routing_key()` even
328    /// when the user has already called `.all()`, so the storage key remains
329    /// scoped to the executor's tenant — otherwise two executors with
330    /// different defaults would share one row in the subscriber table.
331    fn resolve_routing_key(&mut self, executor: &E) {
332        if self.prefix_key.is_none() {
333            self.prefix_key = executor.default_routing_key().map(|s| s.to_owned());
334        }
335        if self.routing_key.is_some() {
336            return;
337        }
338        self.routing_key = Some(match executor.default_routing_key() {
339            Some(k) => RoutingKey::Value(Some(k.to_owned())),
340            None => RoutingKey::Value(None),
341        });
342    }
343
344    fn effective_routing_key(&self) -> RoutingKey {
345        self.routing_key.clone().unwrap_or(RoutingKey::Value(None))
346    }
347
348    #[tracing::instrument(
349        skip_all,
350        fields(
351            subscription = Empty,
352            aggregator_type = Empty,
353            aggregator_id = Empty,
354            event = Empty,
355        )
356    )]
357    async fn process(
358        &self,
359        executor: &E,
360        id: &Ulid,
361        aggregators: &[ReadAggregator],
362    ) -> anyhow::Result<bool> {
363        let mut interval = interval_at(
364            Instant::now() - Duration::from_millis(400),
365            Duration::from_millis(300),
366        );
367
368        tracing::Span::current().record("subscription", self.key());
369
370        loop {
371            interval.tick().await;
372
373            if !executor.is_subscriber_running(self.key(), *id).await? {
374                return Ok(false);
375            }
376
377            let cursor = executor.get_subscriber_cursor(self.key()).await?;
378
379            let res = executor
380                .read(
381                    Some(aggregators.to_vec()),
382                    Some(self.effective_routing_key()),
383                    Args::forward(self.chunk_size, cursor),
384                )
385                .await?;
386
387            if res.edges.is_empty() {
388                return Ok(false);
389            }
390
391            let timestamp = executor
392                .latest_timestamp(
393                    Some(aggregators.to_vec()),
394                    Some(self.effective_routing_key()),
395                )
396                .await?;
397
398            let context = Context {
399                context: self.context.clone(),
400                executor,
401            };
402
403            for event in res.edges {
404                if let Some(ref rx) = self.shutdown_rx {
405                    let mut rx = rx.lock().await;
406                    if rx.try_recv().is_ok() {
407                        tracing::info!(
408                            key = self.key(),
409                            "Subscription received shutdown signal, stopping gracefull"
410                        );
411
412                        return Ok(true);
413                    }
414                    drop(rx);
415                }
416
417                tracing::Span::current().record("aggregator_type", &event.node.aggregator_type);
418                tracing::Span::current().record("aggregator_id", &event.node.aggregator_id);
419                tracing::Span::current().record("event", &event.node.name);
420
421                let all_key = format!("{}_all", event.node.aggregator_type);
422                let key = format!("{}_{}", event.node.aggregator_type, event.node.name);
423                let Some(handler) = self.handlers.get(&all_key).or(self.handlers.get(&key)) else {
424                    if !self.safety_disabled {
425                        anyhow::bail!("no handler s={} k={key}", self.key());
426                    }
427
428                    continue;
429                };
430
431                if let Err(err) = handler.handle(&context, &event.node).await {
432                    tracing::error!("failed");
433
434                    return Err(err);
435                }
436
437                tracing::debug!("completed");
438
439                executor
440                    .acknowledge(
441                        self.key(),
442                        event.cursor.to_owned(),
443                        timestamp.saturating_sub(event.node.timestamp),
444                    )
445                    .await?;
446            }
447        }
448    }
449
450    /// Starts the subscription without retry logic.
451    ///
452    /// Equivalent to calling `start()` with retries disabled.
453    pub async fn unretry_start(mut self, executor: &E) -> anyhow::Result<Subscription>
454    where
455        E: Clone,
456    {
457        self.retry = None;
458        self.start(executor).await
459    }
460
461    /// Starts a continuous background subscription.
462    ///
463    /// Returns a [`Subscription`] handle that can be used for graceful shutdown.
464    /// The subscription runs in a spawned tokio task and polls for new events.
465    #[tracing::instrument(skip_all, fields(
466        subscription = self.key(),
467        aggregator_type = tracing::field::Empty,
468        aggregator_id = tracing::field::Empty,
469        event = tracing::field::Empty,
470    ))]
471    pub async fn start(mut self, executor: &E) -> anyhow::Result<Subscription>
472    where
473        E: Clone,
474    {
475        self.resolve_routing_key(executor);
476        let executor = executor.clone();
477        let id = Ulid::new();
478        let subscription_id = id;
479        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
480        self.shutdown_rx = Some(Mutex::new(shutdown_rx));
481
482        executor
483            .upsert_subscriber(self.key(), id.to_owned())
484            .await?;
485
486        let task_handle = tokio::spawn(async move {
487            let read_aggregators = self.read_aggregators();
488            let start = self
489                .delay
490                .map(|d| Instant::now() + d)
491                .unwrap_or_else(Instant::now);
492
493            let mut interval = interval_at(
494                start - Duration::from_millis(1200),
495                Duration::from_millis(1000),
496            );
497
498            loop {
499                interval.tick().await;
500
501                if let Some(ref rx) = self.shutdown_rx {
502                    let mut rx = rx.lock().await;
503                    if rx.try_recv().is_ok() {
504                        tracing::info!(
505                            key = self.key(),
506                            "Subscription received shutdown signal, stopping gracefull"
507                        );
508
509                        break;
510                    }
511                    drop(rx);
512                }
513
514                let result = match self.retry {
515                    Some(retry) => {
516                        (|| async { self.process(&executor, &id, &read_aggregators).await })
517                            .retry(ExponentialBuilder::default().with_max_times(retry.into()))
518                            .sleep(tokio::time::sleep)
519                            .notify(|err, dur| {
520                                tracing::error!(
521                                    error = %err,
522                                    duration = ?dur,
523                                    "Failed to process event"
524                                );
525                            })
526                            .await
527                    }
528                    _ => self.process(&executor, &id, &read_aggregators).await,
529                };
530
531                match result {
532                    Ok(shutdown) => {
533                        if shutdown {
534                            break;
535                        }
536                    }
537                    Err(err) => {
538                        tracing::error!(error = %err, "Failed to process event");
539
540                        if !self.is_accept_failure {
541                            break;
542                        }
543                    }
544                };
545            }
546        });
547
548        Ok(Subscription {
549            id: subscription_id,
550            task_handle,
551            shutdown_tx,
552        })
553    }
554
555    /// Executes the subscription once without retry logic.
556    ///
557    /// Processes all pending events and returns. Does not poll for new events.
558    pub async fn unretry_execute(mut self, executor: &E) -> anyhow::Result<()> {
559        self.retry = None;
560        self.execute(executor).await
561    }
562
563    /// Executes the subscription once, processing all pending events.
564    ///
565    /// Unlike `start()`, this does not run continuously. It processes
566    /// all currently pending events and returns.
567    #[tracing::instrument(skip_all, fields(
568        subscription = self.key(),
569        aggregator_type = tracing::field::Empty,
570        aggregator_id = tracing::field::Empty,
571        event = tracing::field::Empty,
572    ))]
573    pub async fn execute(&mut self, executor: &E) -> anyhow::Result<()> {
574        self.resolve_routing_key(executor);
575        let id = Ulid::new();
576
577        executor
578            .upsert_subscriber(self.key(), id.to_owned())
579            .await?;
580
581        let read_aggregators = self.read_aggregators();
582
583        match self.retry {
584            Some(retry) => {
585                (|| async { self.process(executor, &id, &read_aggregators).await })
586                    .retry(ExponentialBuilder::default().with_max_times(retry.into()))
587                    .sleep(tokio::time::sleep)
588                    .notify(|err, dur| {
589                        tracing::error!(
590                            error = %err,
591                            duration = ?dur,
592                            "Failed to process event"
593                        );
594                    })
595                    .await
596            }
597            _ => self.process(executor, &id, &read_aggregators).await,
598        }?;
599
600        Ok(())
601    }
602}
603
604/// Handle to a running event subscription.
605///
606/// Returned by [`SubscriptionBuilder::start`], this handle provides
607/// the subscription ID and a method for graceful shutdown.
608///
609/// # Example
610///
611/// ```rust,ignore
612/// let subscription = projection
613///     .subscription()
614///     .start(&executor)
615///     .await?;
616///
617/// println!("Started subscription: {}", subscription.id);
618///
619/// // On application shutdown
620/// subscription.shutdown().await?;
621/// ```
622#[derive(Debug)]
623pub struct Subscription {
624    /// Unique ID for this subscription instance
625    pub id: Ulid,
626    task_handle: tokio::task::JoinHandle<()>,
627    shutdown_tx: tokio::sync::oneshot::Sender<()>,
628}
629
630impl Subscription {
631    /// Gracefully shuts down the subscription.
632    ///
633    /// Signals the subscription to stop and waits for it to finish
634    /// processing the current event before returning.
635    pub async fn shutdown(self) -> Result<(), tokio::task::JoinError> {
636        let _ = self.shutdown_tx.send(());
637
638        self.task_handle.await
639    }
640}
641
642struct SkipHandler<E: AggregatorEvent>(PhantomData<E>);
643
644impl<E: Executor, EV: AggregatorEvent + Send + Sync> Handler<E> for SkipHandler<EV> {
645    fn handle<'a>(
646        &'a self,
647        _context: &'a Context<'a, E>,
648        _event: &'a crate::Event,
649    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>> {
650        Box::pin(async { Ok(()) })
651    }
652
653    fn aggregator_type(&self) -> &'static str {
654        EV::aggregator_type()
655    }
656
657    fn event_name(&self) -> &'static str {
658        EV::event_name()
659    }
660}
evento_core/subscription.rs

evento_core/
subscription.rs
