GKaszewski/movies-diary
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: event ids, outbox identity migration

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-10-25 23:57:24 +01:00
parent c31aa033cc
commit 32dfc84697
14 changed files with 287 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/adapters/event-publisher/src/lib.rs
View File

@@ -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))
}))
}

30
crates/adapters/nats/src/consumer.rs
View File

@@ -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<EventEnvelope, DomainError> {
let payload: NatsEventPayload = serde_json::from_slice(&msg.payload)
fn decode_message(bytes: &[u8]) -> Result<(Option<EventId>, 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<EventEnvelope, DomainError> {
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<EventEnvelope, DomainError> {
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 ──────────────────────────────────

1
crates/adapters/nats/src/payload.rs
View File

@@ -1 +1,2 @@
pub use event_payload::EventMessage as NatsEventMessage;
pub use event_payload::EventPayload as NatsEventPayload;

23
crates/adapters/sqlite/migrations/0101_outbox_identity.sql Normal file
View File

@@ -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)
);

33
crates/adapters/sqlite/src/event_queue/mod.rs
View File

@@ -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<String>,
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(),

81
crates/adapters/sqlite/src/event_queue/tests.rs
View File

@@ -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<Option<String>> =
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());
}
}

22
crates/adapters/sqlite/src/shared.rs
View File

@@ -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())

1
crates/adapters/sqlite/tests/fixtures/migration_ledger.txt vendored
View File

@@ -29,3 +29,4 @@
29 56F4F7B41A09F07211D7543204F0EAC93B7BFB15AFD9BC90DF2654DAE5BE96B1E358B50E2190BF25739C5F0B1FD4B806
30 DAAA0AF35B4ACB271A9D8068B82F02A72EEC62F3AC07ED504C7A7887976C2C8BACB35FFF477B63A931DF15F79018390D
100 310ADC9ABF0CC2F8E9EDAB33168D009D64228FF5BD8D2A14185513699EBB782606A1EEA8FEE6EC6AF65D72396153F13F
101 D4712BF715F8BD86502ED8162488998B1BAFF41C241072173BC2210A3676C05B5B373DD3ED33A16B473A2EC983211302

6
crates/application/tests/worker/worker_test.rs
View File

@@ -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<EventEnvelope, DomainError>> {
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<EventEnvelope, DomainError>> {
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<EventEnvelope, DomainError>> {
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<EventEnvelope, DomainError>> {
let envelope = EventEnvelope::new(
None,
self.event.clone(),
Box::new(TrackingAck {
acked: Arc::clone(&self.acked),

17
crates/bootstrap/src/event_bus.rs
View File

@@ -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<Arc<dyn EventPublisher>> {
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(

40
crates/bootstrap/tests/bootstrap_test.rs
View File

@@ -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);
}

34
crates/domain/src/events.rs
View File

@@ -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<EventId>,
ack: Box<dyn AckHandle>,
}
impl EventEnvelope {
pub fn new(event: DomainEvent, ack: Box<dyn AckHandle>) -> Self {
Self { event, ack }
/// `id` is `None` only for events queued before identities existed.
pub fn new(id: Option<EventId>, event: DomainEvent, ack: Box<dyn AckHandle>) -> Self {
Self { event, id, ack }
}
pub fn id(&self) -> Option<&EventId> {
self.id.as_ref()
}
pub async fn ack(self) -> Result<(), DomainError> {

10
crates/event-payload/src/lib.rs
View File

@@ -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<String>,
#[serde(flatten)]
pub payload: EventPayload,
}
impl EventPayload {
pub fn event_type(&self) -> &'static str {
match self {

26
crates/event-payload/tests/event_payload_test.rs
View File

@@ -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::<EventMessage>(&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::<EventMessage>(&json).unwrap(), unit);
}