feat: processed-event ledger, idempotent dispatch

crates/application/src/jobs/mod.rs

@@ -1,6 +1,7 @@
 mod enrichment_staleness;
 mod import_cleanup;
 mod movie_dedup;
+mod processed_events_cleanup;
 mod refresh_session_cleanup;
 mod watch_event_cleanup;
 mod wrapup;
@@ -8,6 +9,7 @@ mod wrapup;
 pub use enrichment_staleness::EnrichmentStalenessJob;
 pub use import_cleanup::ImportSessionCleanupJob;
 pub use movie_dedup::MovieDeduplicationJob;
+pub use processed_events_cleanup::{PROCESSED_EVENTS_RETENTION_DAYS, ProcessedEventsCleanupJob};
 pub use refresh_session_cleanup::RefreshSessionCleanupJob;
 pub use watch_event_cleanup::WatchEventCleanupJob;
 pub use wrapup::{WrapUpAutoGenerateJob, WrapUpCleanupJob};

crates/application/src/jobs/processed_events_cleanup.rs

@@ -0,0 +1,39 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    ports::{PeriodicJob, ProcessedEventLedger},
+};
+
+/// Ledger rows older than this can no longer matter: the queue gives up on an
+/// event long before, so it cannot be redelivered.
+pub const PROCESSED_EVENTS_RETENTION_DAYS: i64 = 30;
+
+pub struct ProcessedEventsCleanupJob {
+    ledger: Arc<dyn ProcessedEventLedger>,
+}
+
+impl ProcessedEventsCleanupJob {
+    pub fn new(ledger: Arc<dyn ProcessedEventLedger>) -> Self {
+        Self { ledger }
+    }
+}
+
+#[async_trait]
+impl PeriodicJob for ProcessedEventsCleanupJob {
+    fn interval(&self) -> Duration {
+        Duration::from_secs(24 * 3600)
+    }
+
+    async fn run(&self) -> Result<(), DomainError> {
+        let cutoff = chrono::Utc::now().naive_utc()
+            - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS);
+        let n = self.ledger.delete_older_than(cutoff).await?;
+        if n > 0 {
+            tracing::info!("processed events cleanup: removed {n} ledger rows");
+        }
+        Ok(())
+    }
+}

crates/application/src/movies/discovery_indexer.rs

@@ -27,6 +27,10 @@ impl MovieDiscoveryIndexer {
 
 #[async_trait]
 impl EventHandler for MovieDiscoveryIndexer {
+    fn name(&self) -> &'static str {
+        "movie_discovery_indexer"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         let movie_id = match event {
             DomainEvent::MovieDiscovered { movie_id, .. } => movie_id.clone(),

crates/application/src/movies/event_handler.rs

@@ -70,6 +70,10 @@ impl MovieEnrichmentHandler {
 
 #[async_trait]
 impl EventHandler for MovieEnrichmentHandler {
+    fn name(&self) -> &'static str {
+        "movie_enrichment"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         let (movie_id, external_metadata_id) = match event {
             DomainEvent::MovieEnrichmentRequested {

crates/application/src/movies/poster_sync_handler.rs

@@ -106,6 +106,10 @@ impl PosterSyncHandler {
 
 #[async_trait]
 impl EventHandler for PosterSyncHandler {
+    fn name(&self) -> &'static str {
+        "poster_sync"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         let (movie_id, external_metadata_id) = match event {
             DomainEvent::MovieDiscovered {

crates/application/src/movies/reindex_search.rs

@@ -132,6 +132,10 @@ impl SearchReindexHandler {
 
 #[async_trait]
 impl EventHandler for SearchReindexHandler {
+    fn name(&self) -> &'static str {
+        "search_reindex"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         if !matches!(event, DomainEvent::SearchReindexRequested) {
             return Ok(());

crates/application/src/movies/search_cleanup.rs

@@ -24,6 +24,10 @@ impl SearchCleanupHandler {
 
 #[async_trait]
 impl EventHandler for SearchCleanupHandler {
+    fn name(&self) -> &'static str {
+        "search_cleanup"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         let movie_id = match event {
             DomainEvent::MovieDeleted { movie_id, .. } => movie_id.value().to_string(),

crates/application/src/person/event_handler.rs

@@ -31,6 +31,10 @@ impl PersonEnrichmentHandler {
 
 #[async_trait]
 impl EventHandler for PersonEnrichmentHandler {
+    fn name(&self) -> &'static str {
+        "person_enrichment"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         let (person_id, external_person_id) = match event {
             DomainEvent::PersonEnrichmentRequested {

crates/application/src/worker.rs

@@ -1,8 +1,9 @@
 use std::sync::Arc;
 
 use domain::{
+    errors::DomainError,
     events::EventEnvelope,
-    ports::{EventConsumer, EventHandler},
+    ports::{EventConsumer, EventHandler, ProcessedEventLedger},
 };
 use futures::StreamExt;
 use tokio::sync::Semaphore;
@@ -12,20 +13,42 @@ const DEFAULT_CONCURRENCY: usize = 4;
 pub struct WorkerService {
     consumer: Arc<dyn EventConsumer>,
     handlers: Vec<Arc<dyn EventHandler>>,
+    ledger: Arc<dyn ProcessedEventLedger>,
     semaphore: Arc<Semaphore>,
 }
 
 impl WorkerService {
-    pub fn new(consumer: Arc<dyn EventConsumer>, handlers: Vec<Arc<dyn EventHandler>>) -> Self {
-        Self {
+    /// Fails if two handlers share a name: the name keys the processed-event
+    /// ledger, so a collision would make one handler's success skip the other.
+    pub fn new(
+        consumer: Arc<dyn EventConsumer>,
+        handlers: Vec<Arc<dyn EventHandler>>,
+        ledger: Arc<dyn ProcessedEventLedger>,
+    ) -> Result<Self, DomainError> {
+        let mut seen = std::collections::HashSet::new();
+        for handler in &handlers {
+            if !seen.insert(handler.name()) {
+                return Err(DomainError::ValidationError(format!(
+                    "duplicate event handler name `{}`",
+                    handler.name()
+                )));
+            }
+        }
+        Ok(Self {
             consumer,
             handlers,
+            ledger,
             semaphore: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY)),
-        }
+        })
+    }
+
+    pub fn handler_names(&self) -> Vec<&'static str> {
+        self.handlers.iter().map(|h| h.name()).collect()
     }
 
     pub async fn run(self, mut shutdown: tokio::sync::watch::Receiver<bool>) {
         let handlers = Arc::new(self.handlers);
+        let ledger = self.ledger;
         let mut tasks = tokio::task::JoinSet::new();
         let mut stream = self.consumer.consume();
 
@@ -43,8 +66,9 @@ impl WorkerService {
                             let permit = self.semaphore.clone().acquire_owned().await;
                             let Ok(permit) = permit else { break };
                             let h = Arc::clone(&handlers);
+                            let ledger = Arc::clone(&ledger);
                             tasks.spawn(async move {
-                                dispatch(h, envelope).await;
+                                dispatch(h, ledger, envelope).await;
                                 drop(permit);
                             });
                         }
@@ -64,15 +88,48 @@ impl WorkerService {
     }
 }
 
-async fn dispatch(handlers: Arc<Vec<Arc<dyn EventHandler>>>, envelope: EventEnvelope) {
+async fn dispatch(
+    handlers: Arc<Vec<Arc<dyn EventHandler>>>,
+    ledger: Arc<dyn ProcessedEventLedger>,
+    envelope: EventEnvelope,
+) {
+    let id = envelope.id().copied();
     let mut any_transient = false;
     for handler in handlers.iter() {
-        if let Err(e) = handler.handle(&envelope.event).await {
-            if e.is_transient() {
-                tracing::warn!("transient handler error, will retry: {e}");
+        let name = handler.name();
+        if let Some(id) = &id {
+            match ledger.has_processed(id, name).await {
+                Ok(true) => {
+                    tracing::debug!(handler = name, "already processed, skipping");
+                    continue;
+                }
+                Ok(false) => {}
+                Err(e) => {
+                    // Running anyway could repeat a side effect; retry the delivery instead.
+                    tracing::warn!(handler = name, "ledger lookup failed, will retry: {e}");
+                    any_transient = true;
+                    continue;
+                }
+            }
+        }
+        match handler.handle(&envelope.event).await {
+            Ok(()) => {
+                if let Some(id) = &id
+                    && let Err(e) = ledger.record_processed(id, name).await
+                {
+                    // The work is done; failing the delivery would only repeat it.
+                    tracing::warn!(handler = name, "recording processed event failed: {e}");
+                }
+            }
+            Err(e) if e.is_transient() => {
+                tracing::warn!(handler = name, "transient handler error, will retry: {e}");
                 any_transient = true;
-            } else {
-                tracing::warn!("permanent handler error (not retrying): {e}");
+            }
+            Err(e) => {
+                tracing::warn!(
+                    handler = name,
+                    "permanent handler error (not retrying): {e}"
+                );
             }
         }
     }

crates/application/src/wrapup/event_handler.rs

@@ -32,6 +32,10 @@ impl WrapUpEventHandler {
 
 #[async_trait]
 impl EventHandler for WrapUpEventHandler {
+    fn name(&self) -> &'static str {
+        "wrapup"
+    }
+
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
         match event {
             DomainEvent::WrapUpRequested {