cratefield_core/
events.rs

1//! The in-process event bus (issue #4, architecture section 4).
2//!
3//! It exists so `waitlist.confirmed` can feed `email-signup` without a crate
4//! dependency between them. Handlers run through the request's [`Scope`]
5//! defer (`wait_until` on Workers); errors are logged with the event name
6//! and never fail the request. There is **no ordering guarantee and no
7//! persistence** — if the isolate dies between response and deferred run,
8//! the event is lost. Durable workflows are out of scope for the harness.
9
10use crate::scope::Scope;
11use futures_core::future::BoxFuture;
12use serde_json::Value;
13use std::sync::Arc;
14use tracing::{error, warn};
15
16/// Error type for handler and scheduled-work results.
17pub type AnyError = Box<dyn std::error::Error + Send + Sync + 'static>;
18
19/// Event names are `"<module>.<event>"`, e.g. `waitlist.confirmed`.
20pub type EventName = String;
21
22/// A registered handler: receives the emitting request's scope and the
23/// payload.
24pub type EventHandler =
25    Arc<dyn Fn(&Scope, Value) -> BoxFuture<'static, Result<(), AnyError>> + Send + Sync>;
26
27/// Registry of handlers, built once by `Harness::build` from every module's
28/// `events()`. Cheap to clone (one `Arc`).
29#[derive(Clone, Default)]
30pub struct EventBus {
31    handlers: Arc<Vec<(EventName, EventHandler)>>,
32}
33
34impl EventBus {
35    /// An empty bus (a harness with no subscriptions).
36    pub fn new() -> Self {
37        Self::default()
38    }
39
40    /// Builds a bus from collected `(name, handler)` pairs, appending the
41    /// new pairs after any existing ones (copy-on-write if the bus is
42    /// already shared). Modules register via `Module::events()` at
43    /// `Harness::build`.
44    #[must_use]
45    pub fn on(self, name: impl Into<EventName>, handler: EventHandler) -> Self {
46        let mut handlers: Vec<(EventName, EventHandler)> = match Arc::try_unwrap(self.handlers) {
47            Ok(handlers) => handlers,
48            Err(shared) => (*shared).clone(),
49        };
50        handlers.push((name.into(), handler));
51        Self {
52            handlers: Arc::new(handlers),
53        }
54    }
55
56    /// The registered (name, handler) pairs, in registration order.
57    pub fn handlers(&self) -> &[(EventName, EventHandler)] {
58        &self.handlers
59    }
60
61    /// Runs every handler registered for `name` through the scope's defer,
62    /// in the emitting request's `wait_until`. Never fails the request;
63    /// handler errors are logged with the event name.
64    // The by-value payload is the API fixed by issue #4; handlers each get
65    // a clone.
66    #[allow(clippy::needless_pass_by_value)]
67    pub fn emit_in(&self, scope: &Scope, name: &str, payload: Value) {
68        let matched: Vec<&(EventName, EventHandler)> = self
69            .handlers
70            .iter()
71            .filter(|(handler_name, _)| handler_name == name)
72            .collect();
73        for (_, handler) in &matched {
74            let fut = handler(scope, payload.clone());
75            let event = name.to_string();
76            scope.defer.wait_until(Box::pin(async move {
77                if let Err(err) = fut.await {
78                    error!(event = %event, error = %err, "event handler failed");
79                    crate::logging::forward_internal_error(&format!(
80                        "event handler failed for {event}: {err}"
81                    ));
82                }
83            }));
84        }
85        if matched.is_empty() && !self.handlers.is_empty() {
86            warn!(event = %name, "emitted event has no registered handler");
87        }
88    }
89}
cratefield_core/events.rs

cratefield_core/
events.rs
