From 826e824b58e50d2d9b44ad17028ee31073c83050 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 07:23:21 +0200 Subject: [PATCH] replace tokio::broadcast event bus with SQLite-backed event queue (ADR-0003) - DomainEvent: add Serialize/Deserialize - EventConsumer port: poll_next/ack/nack replacing recv() - EventEnvelope: wraps event with id/retry_count/created_at - SqliteEventPublisher/SqliteEventConsumer: INSERT/poll/DLQ - migration: event_queue + dead_letter_queue tables - InMemoryEventBus: combined publisher+consumer test double - NoopEventConsumer: test stub - webhook_consumer: polls EventConsumer instead of broadcast::Receiver - presentation+mcp wiring: create from SqlitePool --- Cargo.lock | 4 + crates/adapters/event-publisher/Cargo.toml | 4 + crates/adapters/event-publisher/src/lib.rs | 176 +++++++++++++++--- crates/domain/src/events/mod.rs | 56 +++++- crates/domain/src/ports/events.rs | 6 +- crates/domain/src/ports/mod.rs | 2 +- crates/domain/src/testing/in_memory.rs | 68 ++++++- crates/domain/src/testing/noops.rs | 35 +++- crates/mcp/src/main.rs | 8 +- .../src/background/webhook_consumer.rs | 30 +-- crates/presentation/src/factory.rs | 18 +- crates/presentation/src/state.rs | 3 +- .../20260712000001_add_event_queue.sql | 25 +++ 13 files changed, 380 insertions(+), 55 deletions(-) create mode 100644 migrations_sqlite/20260712000001_add_event_queue.sql diff --git a/Cargo.lock b/Cargo.lock index ab153d2..d1f73c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -34,7 +34,11 @@ name = "adapter-event-publisher" version = "0.1.0" dependencies = [ "async-trait", + "chrono", "domain", + "serde", + "serde_json", + "sqlx", "tokio", "tracing", ] diff --git a/crates/adapters/event-publisher/Cargo.toml b/crates/adapters/event-publisher/Cargo.toml index c812897..ce3d788 100644 --- a/crates/adapters/event-publisher/Cargo.toml +++ b/crates/adapters/event-publisher/Cargo.toml @@ -8,3 +8,7 @@ domain = { workspace = true } async-trait = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } +sqlx = { workspace = true, features = ["sqlite"] } +serde = { workspace = true } +serde_json = { workspace = true } +chrono = { workspace = true } diff --git a/crates/adapters/event-publisher/src/lib.rs b/crates/adapters/event-publisher/src/lib.rs index d745c11..effe32f 100644 --- a/crates/adapters/event-publisher/src/lib.rs +++ b/crates/adapters/event-publisher/src/lib.rs @@ -1,42 +1,170 @@ use async_trait::async_trait; use domain::errors::{DomainError, DomainResult}; -use domain::events::DomainEvent; +use domain::events::{DomainEvent, EventEnvelope}; use domain::ports::events::{EventConsumer, EventPublisher}; -use tokio::sync::broadcast; +use sqlx::SqlitePool; -pub struct ChannelEventBus { - tx: broadcast::Sender, +fn event_type_label(event: &DomainEvent) -> &'static str { + match event { + DomainEvent::BroadcastTransition { .. } => "broadcast_transition", + DomainEvent::NoSignal { .. } => "no_signal", + DomainEvent::ScheduleGenerated { .. } => "schedule_generated", + DomainEvent::ChannelCreated { .. } => "channel_created", + DomainEvent::ChannelUpdated { .. } => "channel_updated", + DomainEvent::ChannelDeleted { .. } => "channel_deleted", + DomainEvent::UserRegistered { .. } => "user_registered", + _ => "unknown", + } } -impl ChannelEventBus { - pub fn new(capacity: usize) -> Self { - let (tx, _) = broadcast::channel(capacity); - Self { tx } - } +pub struct SqliteEventPublisher { + pool: SqlitePool, +} - pub fn subscriber(&self) -> broadcast::Receiver { - self.tx.subscribe() - } - - pub fn sender(&self) -> broadcast::Sender { - self.tx.clone() +impl SqliteEventPublisher { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } } } #[async_trait] -impl EventPublisher for ChannelEventBus { +impl EventPublisher for SqliteEventPublisher { async fn publish(&self, event: DomainEvent) -> DomainResult<()> { - let _ = self.tx.send(event); + let event_type = event_type_label(&event); + let payload = serde_json::to_string(&event) + .map_err(|e| DomainError::InfrastructureError(format!("event serialize: {e}")))?; + + sqlx::query( + "INSERT INTO event_queue (event_type, payload, status) VALUES (?, ?, 'pending')", + ) + .bind(event_type) + .bind(&payload) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + Ok(()) } } -#[async_trait] -impl EventConsumer for ChannelEventBus { - async fn recv(&self) -> DomainResult { - let mut rx = self.tx.subscribe(); - rx.recv() - .await - .map_err(|e| DomainError::InfrastructureError(e.to_string())) +pub struct SqliteEventConsumer { + pool: SqlitePool, +} + +impl SqliteEventConsumer { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } } } + +#[derive(sqlx::FromRow)] +struct EventRow { + id: i64, + payload: String, + retry_count: i32, + created_at: String, + max_retries: i32, + event_type: String, +} + +#[async_trait] +impl EventConsumer for SqliteEventConsumer { + async fn poll_next(&self) -> DomainResult> { + let row: Option = sqlx::query_as( + "SELECT id, event_type, payload, retry_count, created_at, max_retries \ + FROM event_queue WHERE status = 'pending' ORDER BY id ASC LIMIT 1", + ) + .fetch_optional(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + + let row = match row { + Some(r) => r, + None => return Ok(None), + }; + + sqlx::query("UPDATE event_queue SET status = 'processing', updated_at = datetime('now') WHERE id = ?") + .bind(row.id) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + + match serde_json::from_str::(&row.payload) { + Ok(event) => Ok(Some(EventEnvelope::from_persistence( + row.id, + event, + row.retry_count as u32, + row.created_at, + ))), + Err(e) => { + move_to_dlq(&self.pool, &row, &e.to_string()).await?; + Ok(None) + } + } + } + + async fn ack(&self, event_id: i64) -> DomainResult<()> { + sqlx::query("DELETE FROM event_queue WHERE id = ?") + .bind(event_id) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + Ok(()) + } + + async fn nack(&self, event_id: i64, error: &str) -> DomainResult<()> { + let row: Option = sqlx::query_as( + "SELECT id, event_type, payload, retry_count, created_at, max_retries FROM event_queue WHERE id = ?", + ) + .bind(event_id) + .fetch_optional(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + + let row = match row { + Some(r) => r, + None => return Ok(()), + }; + + let new_retry = row.retry_count + 1; + if new_retry >= row.max_retries { + move_to_dlq(&self.pool, &row, error).await?; + } else { + sqlx::query( + "UPDATE event_queue SET status = 'pending', retry_count = ?, error_message = ?, updated_at = datetime('now') WHERE id = ?", + ) + .bind(new_retry) + .bind(error) + .bind(event_id) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + } + + Ok(()) + } +} + +async fn move_to_dlq(pool: &SqlitePool, row: &EventRow, error: &str) -> DomainResult<()> { + sqlx::query( + "INSERT INTO dead_letter_queue (original_event_id, event_type, payload, error_message, retry_count, original_created_at) \ + VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(row.id) + .bind(&row.event_type) + .bind(&row.payload) + .bind(error) + .bind(row.retry_count) + .bind(&row.created_at) + .execute(pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + + sqlx::query("DELETE FROM event_queue WHERE id = ?") + .bind(row.id) + .execute(pool) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; + + Ok(()) +} diff --git a/crates/domain/src/events/mod.rs b/crates/domain/src/events/mod.rs index ca38e07..9999a5d 100644 --- a/crates/domain/src/events/mod.rs +++ b/crates/domain/src/events/mod.rs @@ -1,6 +1,8 @@ +use serde::{Deserialize, Serialize}; + use crate::value_objects::{ChannelId, ScheduleId, SlotId}; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Serialize, Deserialize)] #[non_exhaustive] pub enum DomainEvent { BroadcastTransition { @@ -18,6 +20,58 @@ pub enum DomainEvent { UserRegistered { user_id: crate::value_objects::UserId }, } +pub struct EventEnvelope { + id: i64, + event: DomainEvent, + retry_count: u32, + created_at: String, +} + +impl EventEnvelope { + pub fn new(id: i64, event: DomainEvent) -> Self { + Self { + id, + event, + retry_count: 0, + created_at: String::new(), + } + } + + pub fn from_persistence( + id: i64, + event: DomainEvent, + retry_count: u32, + created_at: String, + ) -> Self { + Self { + id, + event, + retry_count, + created_at, + } + } + + pub fn id(&self) -> i64 { + self.id + } + + pub fn event(&self) -> &DomainEvent { + &self.event + } + + pub fn into_event(self) -> DomainEvent { + self.event + } + + pub fn retry_count(&self) -> u32 { + self.retry_count + } + + pub fn created_at(&self) -> &str { + &self.created_at + } +} + #[cfg(test)] #[path = "tests/mod.rs"] mod tests; diff --git a/crates/domain/src/ports/events.rs b/crates/domain/src/ports/events.rs index c87220d..f6c2faf 100644 --- a/crates/domain/src/ports/events.rs +++ b/crates/domain/src/ports/events.rs @@ -1,7 +1,7 @@ use async_trait::async_trait; use crate::errors::DomainResult; -pub use crate::events::DomainEvent; +pub use crate::events::{DomainEvent, EventEnvelope}; #[async_trait] pub trait EventPublisher: Send + Sync { @@ -10,7 +10,9 @@ pub trait EventPublisher: Send + Sync { #[async_trait] pub trait EventConsumer: Send + Sync { - async fn recv(&self) -> DomainResult; + async fn poll_next(&self) -> DomainResult>; + async fn ack(&self, event_id: i64) -> DomainResult<()>; + async fn nack(&self, event_id: i64, error: &str) -> DomainResult<()>; } #[async_trait] diff --git a/crates/domain/src/ports/mod.rs b/crates/domain/src/ports/mod.rs index dced707..c296ed4 100644 --- a/crates/domain/src/ports/mod.rs +++ b/crates/domain/src/ports/mod.rs @@ -13,7 +13,7 @@ pub mod user; pub use activity::{ActivityLogCommand, ActivityLogQuery}; pub use auth::{AuthService, TokenService}; pub use channel::{ChannelCommand, ChannelQuery}; -pub use events::{DomainEvent, EventConsumer, EventHandler, EventPublisher}; +pub use events::{DomainEvent, EventConsumer, EventEnvelope, EventHandler, EventPublisher}; pub use library::{LibraryCommand, LibraryQuery, LibrarySyncAdapter}; pub use media::{ Collection, IMediaProvider, IProviderRegistry, ProviderCapabilities, SeriesSummary, diff --git a/crates/domain/src/testing/in_memory.rs b/crates/domain/src/testing/in_memory.rs index 011047e..a4bbe1c 100644 --- a/crates/domain/src/testing/in_memory.rs +++ b/crates/domain/src/testing/in_memory.rs @@ -1,11 +1,12 @@ use std::cmp::Reverse; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::Mutex; use async_trait::async_trait; use chrono::{DateTime, Utc}; use crate::errors::DomainResult; +use crate::events::{DomainEvent, EventEnvelope}; use crate::models::{ ActivityEvent, Channel, ChannelConfigSnapshot, GeneratedSchedule, LibraryCollection, LibrarySyncLogEntry, LibrarySyncResult, MediaItem, PlaybackRecord, ProviderConfigRow, @@ -13,8 +14,9 @@ use crate::models::{ }; use crate::ports::{ ActivityLogCommand, ActivityLogQuery, AppSettingsRepository, ChannelCommand, ChannelQuery, - LibraryCommand, LibraryQuery, ProviderConfigCommand, ProviderConfigQuery, ScheduleCommand, - ScheduleQuery, TranscodeSettingsRepository, UserCommand, UserQuery, + EventConsumer, EventPublisher, LibraryCommand, LibraryQuery, ProviderConfigCommand, + ProviderConfigQuery, ScheduleCommand, ScheduleQuery, TranscodeSettingsRepository, UserCommand, + UserQuery, }; use crate::value_objects::{ BlockId, ChannelId, ContentType, LibrarySearchFilter, MediaItemId, ScheduleId, SnapshotId, @@ -777,3 +779,63 @@ impl TranscodeSettingsRepository for InMemoryTranscodeSettings { Ok(()) } } + +pub struct InMemoryEventBus { + queue: Mutex>, + next_id: Mutex, +} + +impl InMemoryEventBus { + pub fn new() -> Self { + Self { + queue: Mutex::new(VecDeque::new()), + next_id: Mutex::new(1), + } + } + + pub fn events(&self) -> Vec { + self.queue + .lock() + .unwrap() + .iter() + .map(|(_, e)| e.clone()) + .collect() + } +} + +impl Default for InMemoryEventBus { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl EventPublisher for InMemoryEventBus { + async fn publish(&self, event: DomainEvent) -> DomainResult<()> { + let mut next = self.next_id.lock().unwrap(); + let id = *next; + *next += 1; + self.queue.lock().unwrap().push_back((id, event)); + Ok(()) + } +} + +#[async_trait] +impl EventConsumer for InMemoryEventBus { + async fn poll_next(&self) -> DomainResult> { + let front = self.queue.lock().unwrap().front().cloned(); + Ok(front.map(|(id, event)| EventEnvelope::new(id, event))) + } + + async fn ack(&self, event_id: i64) -> DomainResult<()> { + self.queue + .lock() + .unwrap() + .retain(|(id, _)| *id != event_id); + Ok(()) + } + + async fn nack(&self, _event_id: i64, _error: &str) -> DomainResult<()> { + Ok(()) + } +} diff --git a/crates/domain/src/testing/noops.rs b/crates/domain/src/testing/noops.rs index e4f917a..3ec6ca1 100644 --- a/crates/domain/src/testing/noops.rs +++ b/crates/domain/src/testing/noops.rs @@ -1,13 +1,13 @@ use async_trait::async_trait; use crate::errors::DomainResult; -use crate::events::DomainEvent; +use crate::events::{DomainEvent, EventEnvelope}; use crate::models::{ ActivityEvent, LibrarySyncResult, MediaItem, }; use crate::ports::{ - ActivityLogCommand, ActivityLogQuery, EventPublisher, IMediaProvider, LibrarySyncAdapter, - ProviderCapabilities, StreamQuality, StreamingProtocol, + ActivityLogCommand, ActivityLogQuery, EventConsumer, EventPublisher, IMediaProvider, + LibrarySyncAdapter, ProviderCapabilities, StreamQuality, StreamingProtocol, }; use crate::value_objects::{ChannelId, MediaFilter, MediaItemId}; @@ -32,6 +32,35 @@ impl EventPublisher for NoopEventPublisher { } } +pub struct NoopEventConsumer; + +impl NoopEventConsumer { + pub fn new() -> Self { + Self + } +} + +impl Default for NoopEventConsumer { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl EventConsumer for NoopEventConsumer { + async fn poll_next(&self) -> DomainResult> { + Ok(None) + } + + async fn ack(&self, _event_id: i64) -> DomainResult<()> { + Ok(()) + } + + async fn nack(&self, _event_id: i64, _error: &str) -> DomainResult<()> { + Ok(()) + } +} + pub struct NoopMediaProvider; impl NoopMediaProvider { diff --git a/crates/mcp/src/main.rs b/crates/mcp/src/main.rs index 649a2ea..01560b6 100644 --- a/crates/mcp/src/main.rs +++ b/crates/mcp/src/main.rs @@ -44,8 +44,12 @@ async fn main() -> anyhow::Result<()> { let provider_registry = build_provider_registry().await; - let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(64)); - let event_publisher: Arc = event_bus.clone(); + let sqlite_pool = match &pool { + #[cfg(feature = "sqlite")] + DbPool::Sqlite(p) => p.clone(), + }; + let event_publisher: Arc = + Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool)); let schedule_engine = Arc::new(ScheduleEngineService::new( wire.library_query.clone(), diff --git a/crates/presentation/src/background/webhook_consumer.rs b/crates/presentation/src/background/webhook_consumer.rs index fa9d7de..c845f1e 100644 --- a/crates/presentation/src/background/webhook_consumer.rs +++ b/crates/presentation/src/background/webhook_consumer.rs @@ -3,24 +3,26 @@ use std::sync::Arc; use chrono::Utc; use handlebars::Handlebars; use serde_json::{Value, json}; -use tokio::sync::broadcast; use uuid::Uuid; use domain::events::DomainEvent; -use domain::ports::ChannelQuery; +use domain::ports::{ChannelQuery, EventConsumer}; const DEFAULT_CONTENT_TYPE: &str = "application/json"; +const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); pub async fn run( - mut rx: broadcast::Receiver, + consumer: Arc, channel_query: Arc, client: reqwest::Client, ) { loop { - match rx.recv().await { - Ok(event) => { - let channel_id = event_channel_id(&event); - let payload = build_payload(&event); + match consumer.poll_next().await { + Ok(Some(envelope)) => { + let event_id = envelope.id(); + let event = envelope.event(); + let channel_id = event_channel_id(event); + let payload = build_payload(event); let channel_id_vo = domain::ChannelId::from(channel_id); match channel_query.find_by_id(channel_id_vo).await { @@ -51,13 +53,17 @@ pub async fn run( ); } } + + if let Err(e) = consumer.ack(event_id).await { + tracing::warn!("webhook consumer: ack failed for event {}: {}", event_id, e); + } } - Err(broadcast::error::RecvError::Lagged(n)) => { - tracing::warn!("webhook consumer lagged, {} events dropped", n); + Ok(None) => { + tokio::time::sleep(POLL_INTERVAL).await; } - Err(broadcast::error::RecvError::Closed) => { - tracing::info!("webhook consumer: event bus closed, shutting down"); - break; + Err(e) => { + tracing::warn!("webhook consumer: poll error: {}", e); + tokio::time::sleep(POLL_INTERVAL).await; } } } diff --git a/crates/presentation/src/factory.rs b/crates/presentation/src/factory.rs index d519c83..8d5b2e4 100644 --- a/crates/presentation/src/factory.rs +++ b/crates/presentation/src/factory.rs @@ -17,7 +17,6 @@ use infra_wiring::{Config, ConfigSource, DbPool}; use crate::state::AppState; -const EVENT_BUS_CAPACITY: usize = 64; const DEV_JWT_SECRET: &str = "k-template-dev-secret-not-for-production-use-only"; pub async fn build_app_state(config: Config) -> anyhow::Result { @@ -29,8 +28,14 @@ pub async fn build_app_state(config: Config) -> anyhow::Result { let auth_service: Arc = Arc::new(adapter_auth::PasswordAuthService); - let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(EVENT_BUS_CAPACITY)); - let event_publisher: Arc = event_bus.clone(); + let sqlite_pool = match &pool { + #[cfg(feature = "sqlite")] + DbPool::Sqlite(p) => p.clone(), + }; + let event_publisher: Arc = + Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool.clone())); + let event_consumer: Arc = + Arc::new(adapter_event_publisher::SqliteEventConsumer::new(sqlite_pool)); let provider_registry = build_provider_registry(&config).await; @@ -134,10 +139,10 @@ pub async fn build_app_state(config: Config) -> anyhow::Result { bg_event_publisher, )); - let webhook_rx = event_bus.subscriber(); + let webhook_consumer = event_consumer.clone(); let webhook_channel_query = wire_output.channel_query.clone(); tokio::spawn(crate::background::webhook_consumer::run( - webhook_rx, + webhook_consumer, webhook_channel_query, reqwest::Client::new(), )); @@ -172,7 +177,8 @@ pub async fn build_app_state(config: Config) -> anyhow::Result { #[cfg(feature = "auth-jwt")] jwt_validator, _library_sync: library_sync, - _event_bus: event_bus, + _event_publisher: event_publisher, + _event_consumer: event_consumer, config: config_arc, _sync_trigger: sync_tx, }) diff --git a/crates/presentation/src/state.rs b/crates/presentation/src/state.rs index a61d429..e506177 100644 --- a/crates/presentation/src/state.rs +++ b/crates/presentation/src/state.rs @@ -36,7 +36,8 @@ pub struct AppState { pub jwt_validator: Option>, pub _library_sync: Arc, - pub _event_bus: Arc, + pub _event_publisher: Arc, + pub _event_consumer: Arc, pub config: Arc, pub _sync_trigger: tokio::sync::watch::Sender<()>, diff --git a/migrations_sqlite/20260712000001_add_event_queue.sql b/migrations_sqlite/20260712000001_add_event_queue.sql new file mode 100644 index 0000000..09e9492 --- /dev/null +++ b/migrations_sqlite/20260712000001_add_event_queue.sql @@ -0,0 +1,25 @@ +CREATE TABLE IF NOT EXISTS event_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + retry_count INTEGER NOT NULL DEFAULT 0, + max_retries INTEGER NOT NULL DEFAULT 3, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + error_message TEXT +); + +CREATE TABLE IF NOT EXISTS dead_letter_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + original_event_id INTEGER NOT NULL, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + error_message TEXT NOT NULL, + retry_count INTEGER NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + original_created_at TEXT NOT NULL +); + +CREATE INDEX idx_event_queue_status ON event_queue(status); +CREATE INDEX idx_event_queue_created_at ON event_queue(created_at);