diff --git a/Cargo.lock b/Cargo.lock index 0edd7148..a0b17504 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -918,6 +918,7 @@ dependencies = [ "config", "domain", "export", + "futures", "http-axum", "image-converter", "importer", diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index 8fcdad59..f428ab80 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -223,3 +223,7 @@ fn _assert_send_sync() { check::(); check::(); } + +#[cfg(test)] +#[path = "tests/consumer.rs"] +mod tests; diff --git a/crates/adapters/nats/src/lib.rs b/crates/adapters/nats/src/lib.rs index a762b365..ca2125fc 100644 --- a/crates/adapters/nats/src/lib.rs +++ b/crates/adapters/nats/src/lib.rs @@ -1,52 +1,41 @@ mod config; mod consumer; mod payload; -mod publisher; +mod relay; mod subject; pub use config::{NatsConfig, NatsMode}; pub use consumer::{NatsCoreConsumer, NatsJetStreamConsumer}; -pub use publisher::NatsEventPublisher; +pub use relay::NatsRelaySink; use std::sync::Arc; -use domain::ports::{EventConsumer, EventPublisher}; +use domain::ports::{EventConsumer, EventRelaySink}; -pub async fn create_publisher(cfg: NatsConfig) -> anyhow::Result> { - let client = async_nats::connect(&cfg.url).await?; - let publisher: Arc = match cfg.mode { - NatsMode::Core => Arc::new(NatsEventPublisher::new_core(client, cfg.subject_prefix)), - NatsMode::JetStream => Arc::new(NatsEventPublisher::new_jetstream( - client, - cfg.subject_prefix, - )), - }; - - tracing::info!("NATS publisher created (mode: {:?})", cfg.mode); - Ok(publisher) -} - -pub async fn create_channel( +/// The worker's NATS side: where handlers read events from, and where the +/// outbox relay writes them to. +pub async fn create_bus( cfg: NatsConfig, -) -> anyhow::Result<(Arc, Arc)> { +) -> anyhow::Result<(Arc, Arc)> { let client = async_nats::connect(&cfg.url).await?; - let publisher: Arc = match cfg.mode { - NatsMode::Core => Arc::new(NatsEventPublisher::new_core( - client.clone(), - cfg.subject_prefix.clone(), - )), - NatsMode::JetStream => Arc::new(NatsEventPublisher::new_jetstream( - client.clone(), - cfg.subject_prefix.clone(), - )), + let (consumer, sink): (Arc, Arc) = match cfg.mode { + NatsMode::Core => { + tracing::info!( + "core NATS mode: no broker-side dedup or redelivery; relying on consumer ledger" + ); + ( + Arc::new(NatsCoreConsumer::create(&cfg, client.clone()).await?), + Arc::new(NatsRelaySink::new_core(client, cfg.subject_prefix)), + ) + } + // The consumer creates the stream the relay publishes into. + NatsMode::JetStream => ( + Arc::new(NatsJetStreamConsumer::create(&cfg, client.clone()).await?), + Arc::new(NatsRelaySink::new_jetstream(client, cfg.subject_prefix)), + ), }; - let consumer: Arc = match cfg.mode { - NatsMode::Core => Arc::new(NatsCoreConsumer::create(&cfg, client).await?), - NatsMode::JetStream => Arc::new(NatsJetStreamConsumer::create(&cfg, client).await?), - }; - - tracing::info!("NATS channel created (mode: {:?})", cfg.mode); - Ok((publisher, consumer)) + tracing::info!("NATS bus created (mode: {:?})", cfg.mode); + Ok((consumer, sink)) } diff --git a/crates/adapters/nats/src/payload.rs b/crates/adapters/nats/src/payload.rs index cc948a08..20248847 100644 --- a/crates/adapters/nats/src/payload.rs +++ b/crates/adapters/nats/src/payload.rs @@ -1,2 +1 @@ pub use event_payload::EventMessage as NatsEventMessage; -pub use event_payload::EventPayload as NatsEventPayload; diff --git a/crates/adapters/nats/src/publisher.rs b/crates/adapters/nats/src/publisher.rs deleted file mode 100644 index d19d5f18..00000000 --- a/crates/adapters/nats/src/publisher.rs +++ /dev/null @@ -1,60 +0,0 @@ -use async_nats::{Client, jetstream}; -use async_trait::async_trait; -use domain::{errors::DomainError, events::DomainEvent, ports::EventPublisher}; - -use crate::{payload::NatsEventPayload, subject::event_to_subject}; - -enum PublisherMode { - Core(Client), - JetStream(jetstream::Context), -} - -pub struct NatsEventPublisher { - mode: PublisherMode, - subject_prefix: String, -} - -impl NatsEventPublisher { - pub fn new_core(client: Client, subject_prefix: String) -> Self { - Self { - mode: PublisherMode::Core(client), - subject_prefix, - } - } - - pub fn new_jetstream(client: Client, subject_prefix: String) -> Self { - Self { - mode: PublisherMode::JetStream(jetstream::new(client)), - subject_prefix, - } - } -} - -#[async_trait] -impl EventPublisher for NatsEventPublisher { - async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> { - let subject = event_to_subject(&self.subject_prefix, event); - let payload = serde_json::to_vec(&NatsEventPayload::from(event)) - .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; - - match &self.mode { - PublisherMode::Core(client) => client - .publish(subject, payload.into()) - .await - .map_err(|e| DomainError::InfrastructureError(e.to_string())), - - PublisherMode::JetStream(js) => js - .publish(subject, payload.into()) - .await - .map_err(|e| DomainError::InfrastructureError(e.to_string()))? - .await - .map(|_| ()) - .map_err(|e| DomainError::InfrastructureError(e.to_string())), - } - } -} - -fn _assert_send_sync() { - fn check() {} - check::(); -} diff --git a/crates/adapters/nats/src/relay.rs b/crates/adapters/nats/src/relay.rs new file mode 100644 index 00000000..52454ffe --- /dev/null +++ b/crates/adapters/nats/src/relay.rs @@ -0,0 +1,104 @@ +use async_nats::{Client, HeaderMap, header::NATS_MESSAGE_ID, jetstream}; +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::{DomainEvent, EventId}, + ports::EventRelaySink, +}; +use event_payload::EventPayload; + +use crate::{payload::NatsEventMessage, subject::event_to_subject}; + +/// What goes on the wire for one outbox row. Built purely from the row, so +/// relaying it again (after a crash between publish and mark-done) yields the +/// same `Nats-Msg-Id` and JetStream drops the duplicate. +pub(crate) struct RelayMessage { + pub subject: String, + pub payload: Vec, + /// `None` for rows queued before identities existed: those get no dedup. + pub msg_id: Option, +} + +pub(crate) fn relay_message( + prefix: &str, + id: Option<&EventId>, + event: &DomainEvent, +) -> Result { + let msg_id = id.map(|id| id.value().to_string()); + let payload = serde_json::to_vec(&NatsEventMessage { + event_id: msg_id.clone(), + payload: EventPayload::from(event), + }) + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + Ok(RelayMessage { + subject: event_to_subject(prefix, event), + payload, + msg_id, + }) +} + +enum Transport { + Core(Client), + JetStream(jetstream::Context), +} + +/// Publishes outbox rows to NATS. JetStream mode waits for the broker ack and +/// sets `Nats-Msg-Id`; core mode is fire-and-forget with no broker-side dedup. +pub struct NatsRelaySink { + transport: Transport, + subject_prefix: String, +} + +impl NatsRelaySink { + pub fn new_core(client: Client, subject_prefix: String) -> Self { + Self { + transport: Transport::Core(client), + subject_prefix, + } + } + + pub fn new_jetstream(client: Client, subject_prefix: String) -> Self { + Self { + transport: Transport::JetStream(jetstream::new(client)), + subject_prefix, + } + } +} + +fn infra(e: impl ToString) -> DomainError { + DomainError::InfrastructureError(e.to_string()) +} + +#[async_trait] +impl EventRelaySink for NatsRelaySink { + async fn relay(&self, id: Option<&EventId>, event: &DomainEvent) -> Result<(), DomainError> { + let message = relay_message(&self.subject_prefix, id, event)?; + match &self.transport { + Transport::Core(client) => client + .publish(message.subject, message.payload.into()) + .await + .map_err(infra), + Transport::JetStream(js) => { + let mut headers = HeaderMap::new(); + if let Some(msg_id) = &message.msg_id { + headers.insert(NATS_MESSAGE_ID, msg_id.as_str()); + } + js.publish_with_headers(message.subject, headers, message.payload.into()) + .await + .map_err(infra)? + .await + .map(|_| ()) + .map_err(infra) + } + } + } +} + +fn _assert_send_sync() { + fn check() {} + check::(); +} + +#[cfg(test)] +#[path = "tests/relay.rs"] +mod tests; diff --git a/crates/adapters/nats/src/tests/consumer.rs b/crates/adapters/nats/src/tests/consumer.rs new file mode 100644 index 00000000..d80c0d0e --- /dev/null +++ b/crates/adapters/nats/src/tests/consumer.rs @@ -0,0 +1,46 @@ +use super::*; +use domain::value_objects::UserId; +use event_payload::{EventMessage, EventPayload}; +use uuid::Uuid; + +fn payload() -> EventPayload { + EventPayload::from(&DomainEvent::UserUpdated { + user_id: UserId::from_uuid(Uuid::new_v4()), + }) +} + +fn encode(event_id: Option<&str>) -> Vec { + serde_json::to_vec(&EventMessage { + event_id: event_id.map(str::to_owned), + payload: payload(), + }) + .unwrap() +} + +#[test] +fn decode_message_reads_the_event_id() { + let id = Uuid::new_v4(); + let (decoded, event) = decode_message(&encode(Some(&id.to_string()))).unwrap(); + assert_eq!(decoded.unwrap().value(), id); + assert!(matches!(event, DomainEvent::UserUpdated { .. })); +} + +#[test] +fn decode_message_accepts_a_legacy_payload_without_an_id() { + let legacy = serde_json::to_vec(&payload()).unwrap(); + let (decoded, _) = decode_message(&legacy).unwrap(); + assert!(decoded.is_none()); + let (decoded, _) = decode_message(&encode(None)).unwrap(); + assert!(decoded.is_none()); +} + +#[test] +fn decode_message_rejects_a_malformed_id() { + let err = decode_message(&encode(Some("not-a-uuid"))).unwrap_err(); + assert!(err.to_string().contains("event_id"), "{err}"); +} + +#[test] +fn decode_message_rejects_garbage() { + assert!(decode_message(b"{not json").is_err()); +} diff --git a/crates/adapters/nats/src/tests/relay.rs b/crates/adapters/nats/src/tests/relay.rs new file mode 100644 index 00000000..b1f741f1 --- /dev/null +++ b/crates/adapters/nats/src/tests/relay.rs @@ -0,0 +1,37 @@ +use super::*; +use domain::value_objects::UserId; +use uuid::Uuid; + +fn event() -> DomainEvent { + DomainEvent::UserUpdated { + user_id: UserId::from_uuid(Uuid::new_v4()), + } +} + +#[test] +fn relaying_the_same_row_twice_sets_the_same_msg_id() { + let id = EventId::new(); + let event = event(); + let first = relay_message("p", Some(&id), &event).unwrap(); + let second = relay_message("p", Some(&id), &event).unwrap(); + assert_eq!(first.msg_id, Some(id.value().to_string())); + assert_eq!(first.msg_id, second.msg_id); + assert_eq!(first.subject, "p.user.updated"); +} + +#[test] +fn a_legacy_row_without_an_id_has_no_msg_id() { + let message = relay_message("p", None, &event()).unwrap(); + assert_eq!(message.msg_id, None); + let decoded: serde_json::Value = serde_json::from_slice(&message.payload).unwrap(); + assert!(decoded.get("event_id").is_none()); +} + +#[test] +fn the_body_carries_the_event_id_for_the_consumer() { + let id = EventId::new(); + let message = relay_message("p", Some(&id), &event()).unwrap(); + let decoded: serde_json::Value = serde_json::from_slice(&message.payload).unwrap(); + assert_eq!(decoded["event_id"], id.value().to_string()); + assert_eq!(decoded["type"], "UserUpdated"); +} diff --git a/crates/bootstrap/Cargo.toml b/crates/bootstrap/Cargo.toml index f580ea8e..1f28bd74 100644 --- a/crates/bootstrap/Cargo.toml +++ b/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 } diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs index 885c59b5..d83486a1 100644 --- a/crates/bootstrap/src/event_bus.rs +++ b/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, + consumer: Arc, + relay: Option<(Arc, Arc)>, +} + +/// 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, Arc), + bus: Option<(Arc, Arc)>, +) -> 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, Arc)> { - 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> { + Box::pin(futures::stream::empty()) + } + } + + fn outbox() -> (Arc, Arc) { + (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 = Arc::new(Nop); + let bus_ptr = Arc::as_ptr(&bus_consumer); + let sink: Arc = 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)); + } } diff --git a/crates/bootstrap/src/lib.rs b/crates/bootstrap/src/lib.rs index 58711843..150daf29 100644 --- a/crates/bootstrap/src/lib.rs +++ b/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; diff --git a/crates/bootstrap/src/outbox_relay.rs b/crates/bootstrap/src/outbox_relay.rs new file mode 100644 index 00000000..5264adf0 --- /dev/null +++ b/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, sink: Arc) -> 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}"); + } + } + } +} diff --git a/crates/bootstrap/tests/outbox_relay_test.rs b/crates/bootstrap/tests/outbox_relay_test.rs new file mode 100644 index 00000000..3cad162e --- /dev/null +++ b/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>>, +} + +#[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, 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) { + 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)); +} diff --git a/crates/domain/src/ports/events.rs b/crates/domain/src/ports/events.rs index 99df3530..4d3f72e3 100644 --- a/crates/domain/src/ports/events.rs +++ b/crates/domain/src/ports/events.rs @@ -2,7 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, - events::{DomainEvent, EventEnvelope}, + events::{DomainEvent, EventEnvelope, EventId}, }; #[async_trait] @@ -10,6 +10,13 @@ pub trait EventPublisher: Send + Sync { async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError>; } +/// Delivers one outbox row to the message bus. `Ok` means the bus accepted it +/// (for a brokered bus: acknowledged), so the row may be marked done. +#[async_trait] +pub trait EventRelaySink: Send + Sync { + async fn relay(&self, id: Option<&EventId>, event: &DomainEvent) -> Result<(), DomainError>; +} + pub trait EventConsumer: Send + Sync { fn consume(&self) -> futures::stream::BoxStream<'_, Result>; } diff --git a/crates/event-payload/tests/event_payload_test.rs b/crates/event-payload/tests/event_payload_test.rs index 79cb331f..5252bedc 100644 --- a/crates/event-payload/tests/event_payload_test.rs +++ b/crates/event-payload/tests/event_payload_test.rs @@ -131,3 +131,194 @@ fn a_message_round_trips_its_event_id_and_decodes_without_one() { let json = serde_json::to_string(&unit).unwrap(); assert_eq!(serde_json::from_str::(&json).unwrap(), unit); } + +/// One sample per `DomainEvent` variant. `assert_covers_every_variant` is an +/// exhaustive match, so adding a variant breaks the build until it is added +/// here too. +fn every_event() -> Vec { + use domain::{ + models::{ExternalPersonId, PersonId}, + value_objects::{FollowTarget, GoalId, PosterPath, SocialIdentity, WrapUpId}, + }; + let user = || UserId::from_uuid(Uuid::new_v4()); + let movie = || MovieId::from_uuid(Uuid::new_v4()); + let local = || SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4())); + let remote = || SocialIdentity::Remote { + actor_url: "https://remote.example/users/a".into(), + }; + let date = |d| chrono::NaiveDate::from_ymd_opt(2026, 1, d).unwrap(); + vec![ + review_logged(), + review_updated(), + movie_discovered(), + DomainEvent::MovieDeleted { + movie_id: movie(), + poster_path: Some(PosterPath::new("p/x.jpg".into()).unwrap()), + }, + DomainEvent::UserUpdated { user_id: user() }, + DomainEvent::ReviewDeleted { + review_id: ReviewId::from_uuid(Uuid::new_v4()), + user_id: user(), + }, + DomainEvent::MovieEnrichmentRequested { + movie_id: movie(), + external_metadata_id: ExternalMetadataId::new("tt7654321".into()).unwrap(), + }, + DomainEvent::PersonEnrichmentRequested { + person_id: PersonId::from_uuid(Uuid::new_v4()), + external_person_id: ExternalPersonId::new("42"), + }, + DomainEvent::ImageStored { key: "k".into() }, + DomainEvent::WatchlistEntryAdded { + user_id: user(), + movie_id: movie(), + movie_title: "Heat".into(), + release_year: 1995, + external_metadata_id: Some("tt0113277".into()), + added_at: fixed_dt(), + }, + DomainEvent::WatchlistEntryRemoved { + user_id: user(), + movie_id: movie(), + }, + DomainEvent::FollowRequested { + follower: user(), + target: FollowTarget::Handle("a@b.example".into()), + }, + DomainEvent::FollowAccepted { + owner: user(), + requester: remote(), + }, + DomainEvent::FollowRejected { + owner: user(), + requester: local(), + }, + DomainEvent::Unfollowed { + follower: user(), + target: remote(), + }, + DomainEvent::FollowerRemoved { + owner: user(), + follower: local(), + }, + DomainEvent::ActorBlocked { + blocker: user(), + target: remote(), + }, + DomainEvent::ActorUnblocked { + blocker: user(), + target: remote(), + }, + DomainEvent::BackfillFollower { + owner_user_id: user(), + follower_inbox_url: "https://remote.example/inbox".into(), + }, + DomainEvent::FederationDeliveryRequested { + inbox_url: "https://remote.example/inbox".into(), + activity_json: serde_json::json!({"type": "Create"}), + signing_actor_id: Uuid::new_v4(), + }, + DomainEvent::WatchEventIngested { + user_id: user(), + title: "Heat".into(), + source: "jellyfin".into(), + }, + DomainEvent::WrapUpRequested { + wrapup_id: WrapUpId::from_uuid(Uuid::new_v4()), + user_id: Some(user()), + start_date: date(1), + end_date: date(2), + }, + DomainEvent::WrapUpCompleted { + wrapup_id: WrapUpId::from_uuid(Uuid::new_v4()), + }, + DomainEvent::SearchReindexRequested, + DomainEvent::PosterSynced { movie_id: movie() }, + DomainEvent::GoalCreated { + goal_id: GoalId::from_uuid(Uuid::new_v4()), + user_id: user(), + year: 2026, + target_count: 50, + }, + DomainEvent::GoalUpdated { + goal_id: GoalId::from_uuid(Uuid::new_v4()), + user_id: user(), + year: 2026, + target_count: 60, + }, + DomainEvent::GoalDeleted { + goal_id: GoalId::from_uuid(Uuid::new_v4()), + user_id: user(), + year: 2026, + }, + DomainEvent::UserDeleted { user_id: user() }, + DomainEvent::UserAccountMoved { + user_id: user(), + new_actor_url: "https://new.example/users/a".into(), + }, + ] +} + +fn assert_covers_every_variant(event: &DomainEvent) { + use DomainEvent::*; + match event { + ReviewLogged { .. } + | ReviewUpdated { .. } + | MovieDiscovered { .. } + | MovieDeleted { .. } + | UserUpdated { .. } + | ReviewDeleted { .. } + | MovieEnrichmentRequested { .. } + | PersonEnrichmentRequested { .. } + | ImageStored { .. } + | WatchlistEntryAdded { .. } + | WatchlistEntryRemoved { .. } + | FollowRequested { .. } + | FollowAccepted { .. } + | FollowRejected { .. } + | Unfollowed { .. } + | FollowerRemoved { .. } + | ActorBlocked { .. } + | ActorUnblocked { .. } + | BackfillFollower { .. } + | FederationDeliveryRequested { .. } + | WatchEventIngested { .. } + | WrapUpRequested { .. } + | WrapUpCompleted { .. } + | SearchReindexRequested + | PosterSynced { .. } + | GoalCreated { .. } + | GoalUpdated { .. } + | GoalDeleted { .. } + | UserDeleted { .. } + | UserAccountMoved { .. } => {} + } +} + +#[test] +fn every_event_round_trips_through_an_event_message_with_and_without_an_id() { + use event_payload::EventMessage; + + let events = every_event(); + events.iter().for_each(assert_covers_every_variant); + let types: std::collections::HashSet<_> = events + .iter() + .map(|e| EventPayload::from(e).event_type()) + .collect(); + assert_eq!(types.len(), events.len(), "one sample per variant"); + + for event in &events { + for id in [Some(Uuid::new_v4().to_string()), None] { + let message = EventMessage { + event_id: id.clone(), + payload: EventPayload::from(event), + }; + let json = serde_json::to_string(&message).unwrap(); + let back: EventMessage = serde_json::from_str(&json) + .unwrap_or_else(|e| panic!("{} decode: {e}", message.payload.event_type())); + assert_eq!(back.event_id, id); + let recovered = DomainEvent::try_from(back.payload).unwrap(); + assert_eq!(EventPayload::from(event), EventPayload::from(&recovered)); + } + } +}