From 32dfc84697d8248bacca76dcc13d702c85385ab7 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 25 Oct 2026 23:57:24 +0100 Subject: [PATCH] feat: event ids, outbox identity migration --- crates/adapters/event-publisher/src/lib.rs | 2 +- crates/adapters/nats/src/consumer.rs | 30 ++++--- crates/adapters/nats/src/payload.rs | 1 + .../migrations/0101_outbox_identity.sql | 23 ++++++ crates/adapters/sqlite/src/event_queue/mod.rs | 33 +++++--- .../adapters/sqlite/src/event_queue/tests.rs | 81 +++++++++++++++++++ crates/adapters/sqlite/src/shared.rs | 22 +++++ .../tests/fixtures/migration_ledger.txt | 1 + .../application/tests/worker/worker_test.rs | 6 +- crates/bootstrap/src/event_bus.rs | 17 +--- crates/bootstrap/tests/bootstrap_test.rs | 40 +++++++++ crates/domain/src/events.rs | 34 +++++++- crates/event-payload/src/lib.rs | 10 +++ .../event-payload/tests/event_payload_test.rs | 26 ++++++ 14 files changed, 287 insertions(+), 39 deletions(-) create mode 100644 crates/adapters/sqlite/migrations/0101_outbox_identity.sql diff --git a/crates/adapters/event-publisher/src/lib.rs b/crates/adapters/event-publisher/src/lib.rs index 64ba21d2..8c68fc40 100644 --- a/crates/adapters/event-publisher/src/lib.rs +++ b/crates/adapters/event-publisher/src/lib.rs @@ -58,7 +58,7 @@ impl EventConsumer for ChannelEventConsumer { let receiver = Arc::clone(&self.receiver); Box::pin(stream::unfold(receiver, |rx| async move { let event = rx.lock().await.recv().await?; - let envelope = EventEnvelope::new(event, Box::new(NoopAck)); + let envelope = EventEnvelope::new(None, event, Box::new(NoopAck)); Some((Ok(envelope), rx)) })) } diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index 278fc83e..8fcdad59 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -5,7 +5,7 @@ use async_nats::{ use async_trait::async_trait; use domain::{ errors::DomainError, - events::{AckHandle, DomainEvent, EventEnvelope}, + events::{AckHandle, DomainEvent, EventEnvelope, EventId}, ports::EventConsumer, }; use futures::{ @@ -15,7 +15,7 @@ use futures::{ use std::sync::Arc; use tokio::sync::{Mutex, mpsc}; -use crate::{config::NatsConfig, payload::NatsEventPayload, subject::consumer_subject_filter}; +use crate::{config::NatsConfig, payload::NatsEventMessage, subject::consumer_subject_filter}; // ── JetStream ack handle ───────────────────────────────────────────────────── @@ -66,21 +66,33 @@ impl AckHandle for NoopAck { // ── Envelope construction helpers ──────────────────────────────────────────── -fn decode_js(msg: async_nats::jetstream::Message) -> Result { - let payload: NatsEventPayload = serde_json::from_slice(&msg.payload) +fn decode_message(bytes: &[u8]) -> Result<(Option, DomainEvent), DomainError> { + let message: NatsEventMessage = serde_json::from_slice(bytes) .map_err(|e| DomainError::InfrastructureError(format!("deserialize: {e}")))?; - let event = DomainEvent::try_from(payload)?; + let id = message + .event_id + .as_deref() + .map(|s| { + uuid::Uuid::parse_str(s) + .map(EventId::from_uuid) + .map_err(|e| DomainError::InfrastructureError(format!("event_id: {e}"))) + }) + .transpose()?; + Ok((id, DomainEvent::try_from(message.payload)?)) +} + +fn decode_js(msg: async_nats::jetstream::Message) -> Result { + let (id, event) = decode_message(&msg.payload)?; Ok(EventEnvelope::new( + id, event, Box::new(NatsJetStreamAckHandle { message: msg }), )) } fn decode_core(msg: async_nats::Message) -> Result { - let payload: NatsEventPayload = serde_json::from_slice(&msg.payload) - .map_err(|e| DomainError::InfrastructureError(format!("deserialize: {e}")))?; - let event = DomainEvent::try_from(payload)?; - Ok(EventEnvelope::new(event, Box::new(NoopAck))) + let (id, event) = decode_message(&msg.payload)?; + Ok(EventEnvelope::new(id, event, Box::new(NoopAck))) } // ── Channel-bridge shared by both consumers ────────────────────────────────── diff --git a/crates/adapters/nats/src/payload.rs b/crates/adapters/nats/src/payload.rs index 2ca19ecc..cc948a08 100644 --- a/crates/adapters/nats/src/payload.rs +++ b/crates/adapters/nats/src/payload.rs @@ -1 +1,2 @@ +pub use event_payload::EventMessage as NatsEventMessage; pub use event_payload::EventPayload as NatsEventPayload; diff --git a/crates/adapters/sqlite/migrations/0101_outbox_identity.sql b/crates/adapters/sqlite/migrations/0101_outbox_identity.sql new file mode 100644 index 00000000..caeabf65 --- /dev/null +++ b/crates/adapters/sqlite/migrations/0101_outbox_identity.sql @@ -0,0 +1,23 @@ +-- Outbox identity: every queued event carries a stable id; handlers record +-- (event_id, handler) pairs so redeliveries are skipped. +ALTER TABLE event_queue ADD COLUMN event_id TEXT; + +-- Legacy rows get a random v4-shaped uuid. The subquery references the row id +-- so it is re-evaluated per row rather than cached. +UPDATE event_queue SET event_id = ( + SELECT lower( + substr(h, 1, 8) || '-' || substr(h, 9, 4) || '-4' || substr(h, 14, 3) || '-' + || substr('89ab', (abs(random()) % 4) + 1, 1) || substr(h, 18, 3) || '-' || substr(h, 21, 12) + ) + FROM (SELECT hex(randomblob(16)) AS h, event_queue.id AS row_id) +); + +CREATE UNIQUE INDEX idx_event_queue_event_id + ON event_queue (event_id) WHERE event_id IS NOT NULL; + +CREATE TABLE processed_events ( + event_id TEXT NOT NULL, + handler TEXT NOT NULL, + processed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + PRIMARY KEY (event_id, handler) +); diff --git a/crates/adapters/sqlite/src/event_queue/mod.rs b/crates/adapters/sqlite/src/event_queue/mod.rs index 32dc370e..a26ff578 100644 --- a/crates/adapters/sqlite/src/event_queue/mod.rs +++ b/crates/adapters/sqlite/src/event_queue/mod.rs @@ -6,14 +6,15 @@ use std::time::Duration; use async_trait::async_trait; use domain::{ errors::DomainError, - events::{AckHandle, DomainEvent, EventEnvelope}, + events::{AckHandle, DomainEvent, EventEnvelope, EventId}, ports::{EventConsumer, EventPublisher}, }; use futures::stream::{self, BoxStream}; use sqlx::SqlitePool; use tokio::sync::{Mutex, mpsc}; -use payload::DbEventPayload; +use crate::shared::enqueue_events; +pub(crate) use payload::DbEventPayload; pub struct DbEventQueueConfig { pub poll_interval_ms: u64, @@ -65,17 +66,15 @@ impl SqliteEventQueue { #[async_trait] impl EventPublisher for SqliteEventQueue { async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> { - let db_payload = DbEventPayload::from(event); - let event_type = db_payload.event_type(); - let payload_json = serde_json::to_string(&db_payload) - .map_err(|e| DomainError::InfrastructureError(format!("serialize: {e}")))?; - - sqlx::query("INSERT INTO event_queue (event_type, payload) VALUES (?, ?)") - .bind(event_type) - .bind(payload_json) - .execute(&self.pool) + let mut tx = self + .pool + .begin() .await - .map_err(|e| DomainError::InfrastructureError(format!("insert event: {e}")))?; + .map_err(|e| DomainError::InfrastructureError(format!("begin: {e}")))?; + enqueue_events(&mut tx, std::slice::from_ref(event)).await?; + tx.commit() + .await + .map_err(|e| DomainError::InfrastructureError(format!("commit: {e}")))?; Ok(()) } @@ -126,6 +125,7 @@ impl EventConsumer for SqliteEventQueue { struct QueueRow { id: i64, payload: String, + event_id: Option, attempts: i32, next_attempt_at: String, } @@ -146,7 +146,7 @@ async fn claim_batch( ORDER BY next_attempt_at ASC LIMIT ? ) - RETURNING id, payload, attempts, next_attempt_at", + RETURNING id, payload, event_id, attempts, next_attempt_at", ) .bind(config.batch_size) .fetch_all(pool) @@ -166,7 +166,14 @@ fn decode_row( let db_payload: DbEventPayload = serde_json::from_str(&row.payload) .map_err(|e| DomainError::InfrastructureError(format!("deserialize: {e}")))?; let event = DomainEvent::try_from(db_payload)?; + let id = row + .event_id + .as_deref() + .map(crate::shared::parse_uuid) + .transpose()? + .map(EventId::from_uuid); Ok(EventEnvelope::new( + id, event, Box::new(DbAckHandle { pool: pool.clone(), diff --git a/crates/adapters/sqlite/src/event_queue/tests.rs b/crates/adapters/sqlite/src/event_queue/tests.rs index 0b52662d..79b66e9f 100644 --- a/crates/adapters/sqlite/src/event_queue/tests.rs +++ b/crates/adapters/sqlite/src/event_queue/tests.rs @@ -109,3 +109,84 @@ async fn claimed_rows_are_marked_processing_and_not_claimed_twice() { let payloads: Vec<&str> = first.iter().map(|r| r.payload.as_str()).collect(); assert_eq!(payloads, ["p0", "p1", "p2"], "oldest first"); } + +mod identity { + use domain::{ + events::DomainEvent, + ports::{EventConsumer, EventPublisher}, + value_objects::UserId, + }; + use futures::StreamExt; + use sqlx::SqlitePool; + + use super::{DbEventQueueConfig, config}; + use crate::event_queue::SqliteEventQueue; + + async fn queue() -> (SqliteEventQueue, SqlitePool) { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + crate::migrate(&pool).await.unwrap(); + let cfg: DbEventQueueConfig = config(10); + (SqliteEventQueue::create(pool.clone(), cfg), pool) + } + + fn event() -> DomainEvent { + DomainEvent::UserUpdated { + user_id: UserId::generate(), + } + } + + #[tokio::test] + async fn published_events_get_a_unique_event_id() { + let (q, pool) = queue().await; + + q.publish(&event()).await.unwrap(); + q.publish(&event()).await.unwrap(); + + let ids: Vec> = + sqlx::query_scalar("SELECT event_id FROM event_queue ORDER BY id") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(ids.len(), 2); + let a = ids[0].as_deref().expect("first event has an id"); + let b = ids[1].as_deref().expect("second event has an id"); + assert_ne!(a, b); + uuid::Uuid::parse_str(a).expect("event id is a uuid"); + } + + #[tokio::test] + async fn the_consumer_surfaces_the_published_event_id() { + let (q, pool) = queue().await; + q.publish(&event()).await.unwrap(); + let stored: String = sqlx::query_scalar("SELECT event_id FROM event_queue") + .fetch_one(&pool) + .await + .unwrap(); + + let envelope = q.consume().next().await.unwrap().unwrap(); + + assert_eq!(envelope.id().unwrap().value().to_string(), stored); + } + + #[tokio::test] + async fn legacy_queue_rows_without_event_id_still_decode() { + let (q, pool) = queue().await; + let payload = + serde_json::to_string(&crate::event_queue::DbEventPayload::from(&event())).unwrap(); + sqlx::query( + "INSERT INTO event_queue (event_type, payload, event_id) VALUES ('UserUpdated', ?, NULL)", + ) + .bind(payload) + .execute(&pool) + .await + .unwrap(); + + let envelope = q.consume().next().await.unwrap().unwrap(); + + assert!(envelope.id().is_none()); + } +} diff --git a/crates/adapters/sqlite/src/shared.rs b/crates/adapters/sqlite/src/shared.rs index 13ae4078..151839d0 100644 --- a/crates/adapters/sqlite/src/shared.rs +++ b/crates/adapters/sqlite/src/shared.rs @@ -10,6 +10,28 @@ use domain::{ }, }; +/// THE outbox write path: inserts one `event_queue` row per event, each with a +/// fresh `EventId`, inside the caller's transaction so a change and the events +/// announcing it commit or roll back together. +pub(crate) async fn enqueue_events( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + events: &[domain::events::DomainEvent], +) -> Result<(), DomainError> { + for event in events { + let payload = event_payload::EventPayload::from(event); + let payload_json = serde_json::to_string(&payload) + .map_err(|e| DomainError::InfrastructureError(format!("serialize: {e}")))?; + sqlx::query("INSERT INTO event_queue (event_type, payload, event_id) VALUES (?, ?, ?)") + .bind(payload.event_type()) + .bind(payload_json) + .bind(domain::events::EventId::new().value().to_string()) + .execute(&mut **tx) + .await + .map_err(|e| DomainError::InfrastructureError(format!("insert event: {e}")))?; + } + Ok(()) +} + pub(crate) fn map_sqlx_error(e: sqlx::Error) -> DomainError { tracing::error!("Database error: {:?}", e); DomainError::InfrastructureError("Database operation failed".into()) diff --git a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt index 98d343ed..93d4a5c3 100644 --- a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt +++ b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt @@ -29,3 +29,4 @@ 29 56F4F7B41A09F07211D7543204F0EAC93B7BFB15AFD9BC90DF2654DAE5BE96B1E358B50E2190BF25739C5F0B1FD4B806 30 DAAA0AF35B4ACB271A9D8068B82F02A72EEC62F3AC07ED504C7A7887976C2C8BACB35FFF477B63A931DF15F79018390D 100 310ADC9ABF0CC2F8E9EDAB33168D009D64228FF5BD8D2A14185513699EBB782606A1EEA8FEE6EC6AF65D72396153F13F +101 D4712BF715F8BD86502ED8162488998B1BAFF41C241072173BC2210A3676C05B5B373DD3ED33A16B473A2EC983211302 diff --git a/crates/application/tests/worker/worker_test.rs b/crates/application/tests/worker/worker_test.rs index 56324c71..61f4fc43 100644 --- a/crates/application/tests/worker/worker_test.rs +++ b/crates/application/tests/worker/worker_test.rs @@ -40,7 +40,7 @@ impl EventConsumer for VecConsumer { .events .iter() .cloned() - .map(|e| Ok(EventEnvelope::new(e, Box::new(NoopAckHandle)))) + .map(|e| Ok(EventEnvelope::new(None, e, Box::new(NoopAckHandle)))) .collect(); Box::pin(stream::iter(envelopes)) } @@ -130,6 +130,7 @@ async fn acks_when_all_handlers_succeed() { impl EventConsumer for TrackingConsumer { fn consume(&self) -> BoxStream<'_, Result> { let envelope = EventEnvelope::new( + None, self.event.clone(), Box::new(TrackingAck { acked: Arc::clone(&self.acked), @@ -167,6 +168,7 @@ async fn nacks_on_transient_handler_failure() { impl EventConsumer for TrackingConsumer { fn consume(&self) -> BoxStream<'_, Result> { let envelope = EventEnvelope::new( + None, self.event.clone(), Box::new(TrackingAck { acked: Arc::clone(&self.acked), @@ -212,6 +214,7 @@ async fn acks_on_permanent_handler_failure() { impl EventConsumer for TrackingConsumer { fn consume(&self) -> BoxStream<'_, Result> { let envelope = EventEnvelope::new( + None, self.event.clone(), Box::new(TrackingAck { acked: Arc::clone(&self.acked), @@ -260,6 +263,7 @@ async fn nacks_if_any_handler_is_transient() { impl EventConsumer for TrackingConsumer { fn consume(&self) -> BoxStream<'_, Result> { let envelope = EventEnvelope::new( + None, self.event.clone(), Box::new(TrackingAck { acked: Arc::clone(&self.acked), diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs index 222832d2..885c59b5 100644 --- a/crates/bootstrap/src/event_bus.rs +++ b/crates/bootstrap/src/event_bus.rs @@ -26,23 +26,14 @@ impl EventBusBackend { } } +/// The server's publisher is always the outbox writer: it never talks to the +/// bus, whatever `EVENT_BUS_BACKEND` says. The worker owns delivery. pub async fn create_publisher( db_pool: &sqlx::SqlitePool, config: &config::AppConfig, ) -> anyhow::Result> { - match EventBusBackend::from_config(&config.events)? { - EventBusBackend::Db => { - tracing::info!("event bus: DB queue"); - Ok(sqlite::SqliteEventQueue::create_publisher(db_pool.clone(), &config.events).await?) - } - #[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_publisher(cfg).await?) - } - } + tracing::info!("event publisher: outbox (DB queue)"); + sqlite::SqliteEventQueue::create_publisher(db_pool.clone(), &config.events).await } pub async fn create_channel( diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 33a2e027..5f0fc821 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -67,3 +67,43 @@ async fn federation_off_build_wires_local_social() { "follower should be alice: {body}" ); } + +/// The server never talks to the bus: whatever `EVENT_BUS_BACKEND` says, a +/// change and its event land in the outbox table. No NATS server is reachable here. +#[tokio::test] +async fn server_publishes_to_the_outbox_even_when_bus_is_nats() { + let pool = support::memory_pool().await; + let mut config = support::test_config(); + config.events.bus_backend = "nats".into(); + config.nats.url = Some("nats://127.0.0.1:1".into()); + let router = bootstrap::build_server_on(&config, pool.clone()) + .await + .expect("server builds without a reachable NATS"); + let session = support::register(&router, "alice", "alice@example.com").await; + + let (status, _, body) = support::send( + &router, + support::json_request( + axum::http::Method::POST, + "/api/v1/reviews", + Some(&session.token), + serde_json::json!({ + "manual_title": "Outbox", + "manual_release_year": 2001, + "rating": 4, + "comment": "queued", + "watched_at": "2024-01-01" + }), + ), + ) + .await; + assert_eq!(status, 201, "log review: {body}"); + + let queued: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM event_queue WHERE event_type = 'ReviewLogged' AND event_id IS NOT NULL", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(queued, 1); +} diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 43bc964a..113d5fe8 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -150,14 +150,44 @@ pub trait AckHandle: Send + Sync { async fn nack(&self) -> Result<(), DomainError>; } +/// Stable identity of one queued event; the key for de-duplicating deliveries. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct EventId(uuid::Uuid); + +impl EventId { + pub fn new() -> Self { + Self(uuid::Uuid::new_v4()) + } + + pub fn from_uuid(uuid: uuid::Uuid) -> Self { + Self(uuid) + } + + pub fn value(&self) -> uuid::Uuid { + self.0 + } +} + +impl Default for EventId { + fn default() -> Self { + Self::new() + } +} + pub struct EventEnvelope { pub event: DomainEvent, + id: Option, ack: Box, } impl EventEnvelope { - pub fn new(event: DomainEvent, ack: Box) -> Self { - Self { event, ack } + /// `id` is `None` only for events queued before identities existed. + pub fn new(id: Option, event: DomainEvent, ack: Box) -> Self { + Self { event, id, ack } + } + + pub fn id(&self) -> Option<&EventId> { + self.id.as_ref() } pub async fn ack(self) -> Result<(), DomainError> { diff --git a/crates/event-payload/src/lib.rs b/crates/event-payload/src/lib.rs index 92c32bff..e63059d0 100644 --- a/crates/event-payload/src/lib.rs +++ b/crates/event-payload/src/lib.rs @@ -154,6 +154,16 @@ pub enum EventPayload { }, } +/// Wire form of an event: the tagged payload plus its optional identity. +/// Messages without `event_id` (queued before identities existed) still decode. +#[derive(Debug, Serialize, Deserialize, PartialEq)] +pub struct EventMessage { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_id: Option, + #[serde(flatten)] + pub payload: EventPayload, +} + impl EventPayload { pub fn event_type(&self) -> &'static str { match self { diff --git a/crates/event-payload/tests/event_payload_test.rs b/crates/event-payload/tests/event_payload_test.rs index 0f941bc5..79cb331f 100644 --- a/crates/event-payload/tests/event_payload_test.rs +++ b/crates/event-payload/tests/event_payload_test.rs @@ -105,3 +105,29 @@ fn image_stored_event_type() { }); assert_eq!(payload.event_type(), "ImageStored"); } + +#[test] +fn a_message_round_trips_its_event_id_and_decodes_without_one() { + use event_payload::EventMessage; + + let with_id = EventMessage { + event_id: Some(Uuid::new_v4().to_string()), + payload: EventPayload::from(&review_logged()), + }; + let json = serde_json::to_string(&with_id).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + with_id + ); + + let legacy = serde_json::to_string(&EventPayload::from(&review_logged())).unwrap(); + let decoded: EventMessage = serde_json::from_str(&legacy).unwrap(); + assert_eq!(decoded.event_id, None); + + let unit = EventMessage { + event_id: Some("x".into()), + payload: EventPayload::SearchReindexRequested, + }; + let json = serde_json::to_string(&unit).unwrap(); + assert_eq!(serde_json::from_str::(&json).unwrap(), unit); +}