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

@@ -34,7 +34,11 @@ name = "adapter-event-publisher"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "chrono",
  "domain",
+ "serde",
+ "serde_json",
+ "sqlx",
  "tokio",
  "tracing",
 ]

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 }

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<DomainEvent>,
+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<DomainEvent> {
-        self.tx.subscribe()
-    }
-
-    pub fn sender(&self) -> broadcast::Sender<DomainEvent> {
-        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<DomainEvent> {
-        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<Option<EventEnvelope>> {
+        let row: Option<EventRow> = 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::<DomainEvent>(&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<EventRow> = 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(())
+}

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;

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<DomainEvent>;
+    async fn poll_next(&self) -> DomainResult<Option<EventEnvelope>>;
+    async fn ack(&self, event_id: i64) -> DomainResult<()>;
+    async fn nack(&self, event_id: i64, error: &str) -> DomainResult<()>;
 }
 
 #[async_trait]

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,

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<VecDeque<(i64, DomainEvent)>>,
+    next_id: Mutex<i64>,
+}
+
+impl InMemoryEventBus {
+    pub fn new() -> Self {
+        Self {
+            queue: Mutex::new(VecDeque::new()),
+            next_id: Mutex::new(1),
+        }
+    }
+
+    pub fn events(&self) -> Vec<DomainEvent> {
+        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<Option<EventEnvelope>> {
+        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(())
+    }
+}

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<Option<EventEnvelope>> {
+        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 {

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<dyn domain::ports::EventPublisher> = event_bus.clone();
+    let sqlite_pool = match &pool {
+        #[cfg(feature = "sqlite")]
+        DbPool::Sqlite(p) => p.clone(),
+    };
+    let event_publisher: Arc<dyn domain::ports::EventPublisher> =
+        Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool));
 
     let schedule_engine = Arc::new(ScheduleEngineService::new(
         wire.library_query.clone(),

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<DomainEvent>,
+    consumer: Arc<dyn EventConsumer>,
     channel_query: Arc<dyn ChannelQuery>,
     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;
             }
         }
     }

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<AppState> {
@@ -29,8 +28,14 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
     let auth_service: Arc<dyn domain::ports::AuthService> =
         Arc::new(adapter_auth::PasswordAuthService);
 
-    let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(EVENT_BUS_CAPACITY));
-    let event_publisher: Arc<dyn domain::ports::EventPublisher> = event_bus.clone();
+    let sqlite_pool = match &pool {
+        #[cfg(feature = "sqlite")]
+        DbPool::Sqlite(p) => p.clone(),
+    };
+    let event_publisher: Arc<dyn domain::ports::EventPublisher> =
+        Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool.clone()));
+    let event_consumer: Arc<dyn domain::ports::EventConsumer> =
+        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<AppState> {
         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<AppState> {
         #[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,
     })

crates/presentation/src/state.rs

@@ -36,7 +36,8 @@ pub struct AppState {
     pub jwt_validator: Option<Arc<adapter_auth::JwtValidator>>,
 
     pub _library_sync: Arc<dyn domain::ports::LibrarySyncAdapter>,
-    pub _event_bus: Arc<adapter_event_publisher::ChannelEventBus>,
+    pub _event_publisher: Arc<dyn domain::ports::EventPublisher>,
+    pub _event_consumer: Arc<dyn domain::ports::EventConsumer>,
 
     pub config: Arc<infra_wiring::Config>,
     pub _sync_trigger: tokio::sync::watch::Sender<()>,

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