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, Aggregate, AggregateEvent, EventFilter, Executor};
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, Debug, serde::Serialize, serde::Deserialize)]
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::subscription]
68/// async fn my_handler<E: Executor>(
69/// context: &Context<'_, E>,
70/// event: Event<MyEventData>,
71/// ) -> anyhow::Result<()> {
72/// // Access shared data
73/// let config: Data<AppConfig> = context.extract();
74///
75/// // Use executor for queries
76/// let events = context.executor.read(...).await?;
77/// Ok(())
78/// }
79/// ```
80#[derive(Clone)]
81pub struct Context<'a, E: Executor> {
82 context: context::RwContext,
83 /// Reference to the executor for database operations
84 pub executor: &'a E,
85}
86
87impl<'a, E: Executor> Deref for Context<'a, E> {
88 type Target = context::RwContext;
89
90 fn deref(&self) -> &Self::Target {
91 &self.context
92 }
93}
94
95/// Trait for event handlers.
96///
97/// Handlers process events in two modes:
98/// - `handle`: For subscriptions that perform side effects (send emails, update read models)
99/// - `apply`: For loading aggregate state by replaying events
100///
101/// This trait is typically implemented via the `#[evento::handler]` macro.
102pub trait Handler<E: Executor>: Sync + Send {
103 /// Handles an event during subscription processing.
104 ///
105 /// This is called when processing events in a subscription context,
106 /// where side effects like database updates or API calls are appropriate.
107 fn handle<'a>(
108 &'a self,
109 context: &'a Context<'a, E>,
110 event: &'a crate::Event,
111 ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
112
113 /// Returns the aggregate type this handler processes.
114 fn aggregate_type(&self) -> &'static str;
115 /// Returns the event name this handler processes.
116 fn event_name(&self) -> &'static str;
117}
118
119/// Builder for creating event subscriptions.
120///
121/// Created via [`Projection::subscription`](crate::projection::Projection::subscription), this builder configures
122/// a continuous event processing subscription with retry logic,
123/// routing key filtering, and graceful shutdown support.
124///
125/// # Example
126///
127/// ```rust,ignore
128/// let subscription = projection
129/// .subscription()
130/// .routing_key("accounts")
131/// .chunk_size(100)
132/// .retry(5)
133/// .delay(Duration::from_secs(10))
134/// .start(&executor)
135/// .await?;
136///
137/// // Later, gracefully shutdown
138/// subscription.shutdown().await?;
139/// ```
140pub struct SubscriptionBuilder<E: Executor> {
141 key: String,
142 handlers: HashMap<String, Box<dyn Handler<E>>>,
143 context: context::RwContext,
144 routing_key: Option<RoutingKey>,
145 prefix_key: Option<String>,
146 delay: Option<Duration>,
147 chunk_size: u16,
148 continue_on_error: bool,
149 retry: Option<u8>,
150 aggregators: HashMap<String, String>,
151 safety_disabled: bool,
152 shutdown_rx: Option<Mutex<Receiver<()>>>,
153}
154
155impl<E: Executor + 'static> SubscriptionBuilder<E> {
156 /// Creates a new projection with the given key.
157 ///
158 /// The key is used as the subscription identifier for cursor tracking.
159 pub fn new(key: impl Into<String>) -> Self {
160 Self {
161 key: key.into(),
162 handlers: HashMap::new(),
163 safety_disabled: true,
164 context: Default::default(),
165 delay: None,
166 retry: Some(30),
167 chunk_size: 300,
168 continue_on_error: false,
169 routing_key: None,
170 prefix_key: None,
171 aggregators: Default::default(),
172 shutdown_rx: None,
173 }
174 }
175
176 /// Enables safety checks for unhandled events.
177 ///
178 /// When enabled, processing fails if an event is encountered without a handler.
179 pub fn strict(mut self) -> Self {
180 self.safety_disabled = false;
181
182 self
183 }
184
185 /// Registers an event handler with this subscription.
186 ///
187 /// # Panics
188 ///
189 /// Panics if a handler for the same event type is already registered.
190 pub fn handler<H: Handler<E> + 'static>(mut self, h: H) -> Self {
191 let key = format!("{}_{}", h.aggregate_type(), h.event_name());
192 if self.handlers.insert(key.to_owned(), Box::new(h)).is_some() {
193 panic!("Cannot register event handler: key {} already exists", key);
194 }
195 self
196 }
197
198 /// Registers a skip handler for an event type.
199 ///
200 /// Events of this type will be acknowledged but not processed.
201 ///
202 /// # Panics
203 ///
204 /// Panics if a handler for the same event type is already registered.
205 pub fn skip<EV: AggregateEvent + Send + Sync + 'static>(self) -> Self {
206 self.handler(SkipHandler::<EV>(PhantomData))
207 }
208
209 /// Adds shared data to the subscription context.
210 ///
211 /// Data added here is accessible in handlers via the context.
212 pub fn data<D: Send + Sync + 'static>(self, v: D) -> Self {
213 self.context.insert(v);
214
215 self
216 }
217
218 /// Allows the subscription to continue after handler failures.
219 ///
220 /// By default, subscriptions stop on the first error. With this flag,
221 /// errors are logged but processing continues.
222 pub fn continue_on_error(mut self) -> Self {
223 self.continue_on_error = true;
224
225 self
226 }
227
228 /// Sets the number of events to process per batch.
229 ///
230 /// Default is 300.
231 pub fn chunk_size(mut self, v: u16) -> Self {
232 self.chunk_size = v;
233
234 self
235 }
236
237 /// Sets a delay before starting the subscription.
238 ///
239 /// Useful for staggering subscription starts in multi-node deployments.
240 pub fn delay(mut self, v: Duration) -> Self {
241 self.delay = Some(v);
242
243 self
244 }
245
246 /// Filters events by routing key.
247 ///
248 /// Only events with the matching routing key will be processed.
249 /// Overrides any executor-level default.
250 pub fn routing_key(mut self, v: impl Into<String>) -> Self {
251 self.routing_key = Some(RoutingKey::Value(Some(v.into())));
252
253 self
254 }
255
256 /// Sets the maximum number of retries on failure.
257 ///
258 /// Uses exponential backoff. Default is 30.
259 pub fn retry(mut self, v: u8) -> Self {
260 self.retry = Some(v);
261
262 self
263 }
264
265 /// Processes all events regardless of routing key.
266 ///
267 /// Overrides any executor-level default.
268 pub fn all(mut self) -> Self {
269 self.routing_key = Some(RoutingKey::All);
270
271 self
272 }
273
274 /// Adds a related aggregate to process events from.
275 pub fn aggregate<A: Aggregate>(mut self, id: impl Into<String>) -> Self {
276 self.aggregators
277 .insert(A::aggregate_type().to_owned(), id.into());
278
279 self
280 }
281
282 fn read_aggregators(&self) -> Vec<EventFilter> {
283 self.handlers
284 .values()
285 .map(|h| {
286 // `#[subscription_all]` handlers report the sentinel name "all" and
287 // match every event of the type, so they must read by type rather
288 // than by a literal event name (which would match nothing).
289 let by_name = self.safety_disabled && h.event_name() != "all";
290 match self.aggregators.get(h.aggregate_type()) {
291 Some(id) => EventFilter {
292 aggregate_type: h.aggregate_type().to_owned(),
293 aggregate_id: Some(id.to_owned()),
294 name: by_name.then(|| h.event_name().to_owned()),
295 },
296 _ => {
297 if by_name {
298 EventFilter::by_event(h.aggregate_type(), h.event_name())
299 } else {
300 EventFilter::by_type(h.aggregate_type())
301 }
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()` / `run_once()` 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 aggregate_type = Empty,
353 aggregate_id = Empty,
354 event = Empty,
355 )
356 )]
357 async fn process(
358 &self,
359 executor: &E,
360 id: &Ulid,
361 aggregators: &[EventFilter],
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 gracefully"
410 );
411
412 return Ok(true);
413 }
414 drop(rx);
415 }
416
417 tracing::Span::current().record("aggregate_type", &event.node.aggregate_type);
418 tracing::Span::current().record("aggregate_id", &event.node.aggregate_id);
419 tracing::Span::current().record("event", &event.node.name);
420
421 let all_key = format!("{}_all", event.node.aggregate_type);
422 let key = format!("{}_{}", event.node.aggregate_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 /// Disables retry-on-failure for this subscription.
451 ///
452 /// By default failed batches are retried with exponential backoff (see
453 /// [`retry`](Self::retry)). Combine this with [`start`](Self::start) or
454 /// [`run_once`](Self::run_once) to process without retries.
455 pub fn no_retry(mut self) -> Self {
456 self.retry = None;
457
458 self
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 aggregate_type = tracing::field::Empty,
468 aggregate_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 gracefully"
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.continue_on_error {
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 /// Processes all currently pending events once, then returns.
556 ///
557 /// Unlike [`start`](Self::start), this does not run continuously or spawn a
558 /// background task — it drains the events available now and returns. Pair with
559 /// [`no_retry`](Self::no_retry) to run a single pass without retries.
560 #[tracing::instrument(skip_all, fields(
561 subscription = self.key(),
562 aggregate_type = tracing::field::Empty,
563 aggregate_id = tracing::field::Empty,
564 event = tracing::field::Empty,
565 ))]
566 pub async fn run_once(&mut self, executor: &E) -> anyhow::Result<()> {
567 self.resolve_routing_key(executor);
568 let id = Ulid::new();
569
570 executor
571 .upsert_subscriber(self.key(), id.to_owned())
572 .await?;
573
574 let read_aggregators = self.read_aggregators();
575
576 match self.retry {
577 Some(retry) => {
578 (|| async { self.process(executor, &id, &read_aggregators).await })
579 .retry(ExponentialBuilder::default().with_max_times(retry.into()))
580 .sleep(tokio::time::sleep)
581 .notify(|err, dur| {
582 tracing::error!(
583 error = %err,
584 duration = ?dur,
585 "Failed to process event"
586 );
587 })
588 .await
589 }
590 _ => self.process(executor, &id, &read_aggregators).await,
591 }?;
592
593 Ok(())
594 }
595}
596
597/// Handle to a running event subscription.
598///
599/// Returned by [`SubscriptionBuilder::start`], this handle provides
600/// the subscription ID and a method for graceful shutdown.
601///
602/// # Example
603///
604/// ```rust,ignore
605/// let subscription = projection
606/// .subscription()
607/// .start(&executor)
608/// .await?;
609///
610/// println!("Started subscription: {}", subscription.id);
611///
612/// // On application shutdown
613/// subscription.shutdown().await?;
614/// ```
615#[derive(Debug)]
616pub struct Subscription {
617 /// Unique ID for this subscription instance
618 pub id: Ulid,
619 task_handle: tokio::task::JoinHandle<()>,
620 shutdown_tx: tokio::sync::oneshot::Sender<()>,
621}
622
623impl Subscription {
624 /// Gracefully shuts down the subscription.
625 ///
626 /// Signals the subscription to stop and waits for it to finish
627 /// processing the current event before returning.
628 pub async fn shutdown(self) -> Result<(), tokio::task::JoinError> {
629 let _ = self.shutdown_tx.send(());
630
631 self.task_handle.await
632 }
633}
634
635struct SkipHandler<E: AggregateEvent>(PhantomData<E>);
636
637impl<E: Executor, EV: AggregateEvent + Send + Sync> Handler<E> for SkipHandler<EV> {
638 fn handle<'a>(
639 &'a self,
640 _context: &'a Context<'a, E>,
641 _event: &'a crate::Event,
642 ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>> {
643 Box::pin(async { Ok(()) })
644 }
645
646 fn aggregate_type(&self) -> &'static str {
647 EV::aggregate_type()
648 }
649
650 fn event_name(&self) -> &'static str {
651 EV::event_name()
652 }
653}