feat: worker relays outbox to nats

crates/bootstrap/Cargo.toml

@@ -34,6 +34,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 axum = { workspace = true }
 tokio = { workspace = true }
+futures = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 

crates/bootstrap/src/event_bus.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 #[cfg(feature = "nats")]
 use anyhow::Context;
-use domain::ports::{EventConsumer, EventPublisher};
+use domain::ports::{EventConsumer, EventPublisher, EventRelaySink};
 
 #[derive(Clone, Copy)]
 pub enum EventBusBackend {
@@ -36,21 +36,114 @@ pub async fn create_publisher(
     sqlite::SqliteEventQueue::create_publisher(db_pool.clone(), &config.events).await
 }
 
+/// What the worker gets from the bus: a publisher for its own events, the
+/// consumer its handlers read, and, when the bus is not the outbox, the relay
+/// that moves outbox rows onto it.
+struct Wiring {
+    publisher: Arc<dyn EventPublisher>,
+    consumer: Arc<dyn EventConsumer>,
+    relay: Option<(Arc<dyn EventConsumer>, Arc<dyn EventRelaySink>)>,
+}
+
+/// The worker always writes to the outbox. With an external bus
+/// (`bus = Some((consumer, sink))`) handlers read from the bus and the relay
+/// drains the outbox into it; without one, handlers drain the outbox directly.
+fn compose(
+    (publisher, outbox): (Arc<dyn EventPublisher>, Arc<dyn EventConsumer>),
+    bus: Option<(Arc<dyn EventConsumer>, Arc<dyn EventRelaySink>)>,
+) -> Wiring {
+    match bus {
+        None => Wiring {
+            publisher,
+            consumer: outbox,
+            relay: None,
+        },
+        Some((consumer, sink)) => Wiring {
+            publisher,
+            consumer,
+            relay: Some((outbox, sink)),
+        },
+    }
+}
+
 pub async fn create_channel(
     db_pool: &sqlx::SqlitePool,
     config: &config::AppConfig,
 ) -> anyhow::Result<(Arc<dyn EventPublisher>, Arc<dyn EventConsumer>)> {
-    match EventBusBackend::from_config(&config.events)? {
+    let outbox = sqlite::SqliteEventQueue::create_channel(db_pool.clone(), &config.events).await?;
+    let wiring = match EventBusBackend::from_config(&config.events)? {
         EventBusBackend::Db => {
             tracing::info!("event bus: DB queue");
-            Ok(sqlite::SqliteEventQueue::create_channel(db_pool.clone(), &config.events).await?)
+            compose(outbox, None)
         }
         #[cfg(feature = "nats")]
         EventBusBackend::Nats => {
             let cfg = nats::NatsConfig::from_config(&config.nats)
                 .context("EVENT_BUS_BACKEND=nats requires NATS_URL to be set")?;
-            tracing::info!("event bus: NATS ({})", cfg.url);
-            Ok(nats::create_channel(cfg).await?)
+            tracing::info!("event bus: NATS ({}), relaying the outbox", cfg.url);
+            compose(outbox, Some(nats::create_bus(cfg).await?))
+        }
+    };
+    if let Some((outbox, sink)) = wiring.relay {
+        crate::outbox_relay::spawn(outbox, sink);
+    }
+    Ok((wiring.publisher, wiring.consumer))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::{
+        errors::DomainError,
+        events::{DomainEvent, EventEnvelope, EventId},
+    };
+    use futures::stream::BoxStream;
+
+    struct Nop;
+
+    #[async_trait::async_trait]
+    impl EventPublisher for Nop {
+        async fn publish(&self, _: &DomainEvent) -> Result<(), DomainError> {
+            Ok(())
         }
     }
+    #[async_trait::async_trait]
+    impl EventRelaySink for Nop {
+        async fn relay(&self, _: Option<&EventId>, _: &DomainEvent) -> Result<(), DomainError> {
+            Ok(())
+        }
+    }
+    impl EventConsumer for Nop {
+        fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+            Box::pin(futures::stream::empty())
+        }
+    }
+
+    fn outbox() -> (Arc<dyn EventPublisher>, Arc<dyn EventConsumer>) {
+        (Arc::new(Nop), Arc::new(Nop))
+    }
+
+    #[test]
+    fn db_mode_does_not_relay_and_consumes_the_outbox() {
+        let (publisher, queue) = outbox();
+        let queue_ptr = Arc::as_ptr(&queue);
+        let wiring = compose((publisher, queue), None);
+        assert!(wiring.relay.is_none());
+        assert!(std::ptr::addr_eq(Arc::as_ptr(&wiring.consumer), queue_ptr));
+    }
+
+    #[test]
+    fn bus_mode_relays_the_outbox_and_consumes_the_bus() {
+        let (publisher, queue) = outbox();
+        let queue_ptr = Arc::as_ptr(&queue);
+        let bus_consumer: Arc<dyn EventConsumer> = Arc::new(Nop);
+        let bus_ptr = Arc::as_ptr(&bus_consumer);
+        let sink: Arc<dyn EventRelaySink> = Arc::new(Nop);
+
+        let wiring = compose((publisher, queue), Some((bus_consumer, sink)));
+
+        let (source, _) = wiring.relay.expect("relay spawned in bus mode");
+        assert!(std::ptr::addr_eq(Arc::as_ptr(&source), queue_ptr));
+        assert!(std::ptr::addr_eq(Arc::as_ptr(&wiring.consumer), bus_ptr));
+    }
 }

crates/bootstrap/src/lib.rs

@@ -10,6 +10,7 @@ mod event_bus;
 mod federation;
 #[cfg(feature = "federation")]
 mod follow_backfill_handler;
+pub mod outbox_relay;
 pub mod repositories;
 mod server;
 mod tracing_setup;

crates/bootstrap/src/outbox_relay.rs

@@ -0,0 +1,48 @@
+//! Moves outbox rows onto the message bus. The worker runs it when the bus is
+//! not the outbox itself (`EVENT_BUS_BACKEND=nats`).
+//!
+//! Rows are claimed by the SQLite queue's own atomic claim, so a row is only
+//! ever relayed by one claimant and failures take the queue's existing
+//! backoff and dead-letter path. A row is marked done only after the sink
+//! reports the bus accepted it; a crash in between re-relays the row, and the
+//! bus de-duplicates on the event id.
+
+use std::sync::Arc;
+
+use domain::{
+    events::EventEnvelope,
+    ports::{EventConsumer, EventRelaySink},
+};
+use futures::StreamExt;
+use tokio::task::JoinHandle;
+
+pub fn spawn(source: Arc<dyn EventConsumer>, sink: Arc<dyn EventRelaySink>) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut rows = source.consume();
+        while let Some(next) = rows.next().await {
+            match next {
+                Ok(envelope) => relay_envelope(envelope, sink.as_ref()).await,
+                // A row that cannot even be decoded stays claimed; there is
+                // no handle to settle it with.
+                Err(e) => tracing::error!("outbox relay: undecodable row: {e}"),
+            }
+        }
+        tracing::warn!("outbox relay: source closed, relay stopped");
+    })
+}
+
+async fn relay_envelope(envelope: EventEnvelope, sink: &dyn EventRelaySink) {
+    match sink.relay(envelope.id(), &envelope.event).await {
+        Ok(()) => {
+            if let Err(e) = envelope.ack().await {
+                tracing::error!("outbox relay: published but could not mark done: {e}");
+            }
+        }
+        Err(e) => {
+            tracing::warn!("outbox relay: publish failed, will retry: {e}");
+            if let Err(e) = envelope.nack().await {
+                tracing::error!("outbox relay: could not nack: {e}");
+            }
+        }
+    }
+}

crates/bootstrap/tests/outbox_relay_test.rs

@@ -0,0 +1,115 @@
+//! The relay claims outbox rows with the SQLite queue's own atomic claim and
+//! settles each row from the sink's verdict, with fakes at the sink boundary.
+
+mod support;
+
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::{DomainEvent, EventId},
+    ports::{EventPublisher, EventRelaySink},
+    value_objects::UserId,
+};
+use sqlite::SqliteEventQueue;
+use uuid::Uuid;
+
+#[derive(Default)]
+struct FakeSink {
+    fail: bool,
+    seen: Mutex<Vec<Option<EventId>>>,
+}
+
+#[async_trait]
+impl EventRelaySink for FakeSink {
+    async fn relay(&self, id: Option<&EventId>, _event: &DomainEvent) -> Result<(), DomainError> {
+        self.seen.lock().unwrap().push(id.cloned());
+        if self.fail {
+            Err(DomainError::InfrastructureError("bus down".into()))
+        } else {
+            Ok(())
+        }
+    }
+}
+
+async fn outbox() -> (sqlx::SqlitePool, Arc<dyn EventPublisher>, SqliteEventQueue) {
+    let pool = support::memory_pool().await;
+    sqlite::migrate(&pool).await.unwrap();
+    let config = config::EventsConfig {
+        queue_poll_interval_ms: 10,
+        ..Default::default()
+    };
+    let queue = SqliteEventQueue::create(
+        pool.clone(),
+        sqlite::DbEventQueueConfig::from_config(&config),
+    );
+    (pool.clone(), Arc::new(queue.clone()), queue)
+}
+
+async fn row(pool: &sqlx::SqlitePool) -> (String, i32, Option<String>) {
+    sqlx::query_as("SELECT status, attempts, event_id FROM event_queue")
+        .fetch_one(pool)
+        .await
+        .unwrap()
+}
+
+async fn wait_for(pool: &sqlx::SqlitePool, status: &str) {
+    for _ in 0..200 {
+        if row(pool).await.0 == status {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("row never reached {status}: {:?}", row(pool).await);
+}
+
+fn event() -> DomainEvent {
+    DomainEvent::UserUpdated {
+        user_id: UserId::from_uuid(Uuid::new_v4()),
+    }
+}
+
+#[tokio::test]
+async fn relay_marks_rows_done_only_after_publish_succeeds() {
+    let (pool, publisher, queue) = outbox().await;
+    publisher.publish(&event()).await.unwrap();
+    let sink = Arc::new(FakeSink::default());
+
+    let relay = bootstrap::outbox_relay::spawn(Arc::new(queue), sink.clone());
+    wait_for(&pool, "done").await;
+    relay.abort();
+
+    let (_, attempts, event_id) = row(&pool).await;
+    assert_eq!(attempts, 0);
+    let seen = sink.seen.lock().unwrap();
+    assert_eq!(seen.len(), 1);
+    assert_eq!(
+        seen[0].as_ref().map(|id| id.value().to_string()),
+        event_id,
+        "the sink gets the row's own event id"
+    );
+}
+
+#[tokio::test]
+async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
+    let (pool, publisher, queue) = outbox().await;
+    publisher.publish(&event()).await.unwrap();
+    let sink = Arc::new(FakeSink {
+        fail: true,
+        ..Default::default()
+    });
+
+    let relay = bootstrap::outbox_relay::spawn(Arc::new(queue), sink.clone());
+    for _ in 0..200 {
+        if row(&pool).await.1 == 1 {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    relay.abort();
+
+    let (status, attempts, _) = row(&pool).await;
+    assert_eq!((status.as_str(), attempts), ("pending", 1));
+}