feat: idempotent handlers, deterministic activity ids

crates/adapters/activitypub/src/urls.rs

@@ -38,3 +38,34 @@ pub fn watchlist_entry_url(
     ))
     .expect("base_url is always a valid URL prefix")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// k-ap derives the Create activity id from the object id (UUIDv5), so
+    /// re-handling an event only re-sends the same activity id if our object
+    /// ids are pure functions of the entity. Update/Delete/Move ids are minted
+    /// randomly inside k-ap 0.5 with no override (see ADR-0015): those repeat
+    /// semantically idempotent activities under a fresh id.
+    #[test]
+    fn activitypub_object_ids_are_stable_across_event_redelivery() {
+        let instance = InstanceIdentity::new("https://diary.example");
+        let review = ReviewId::from_uuid(uuid::Uuid::new_v4());
+        let user = uuid::Uuid::new_v4();
+        let movie = uuid::Uuid::new_v4();
+
+        assert_eq!(
+            review_url(&instance, &review),
+            review_url(&instance, &review)
+        );
+        assert_eq!(
+            goal_url(&instance, user, 2026),
+            goal_url(&instance, user, 2026)
+        );
+        assert_eq!(
+            watchlist_entry_url(&instance, user, movie),
+            watchlist_entry_url(&instance, user, movie)
+        );
+    }
+}

crates/adapters/image-converter/src/handler.rs

@@ -45,6 +45,17 @@ impl EventHandler for ImageConversionHandler {
             return Ok(());
         }
 
+        // `get` falls back to `{key}.webp` / `{key}.avif`, which would re-encode
+        // an already converted image on redelivery. Probe the exact key first:
+        // once the original is gone, an earlier delivery finished the conversion.
+        match self.storage.get_stream(&key).await {
+            Ok(_) => {}
+            Err(DomainError::NotFound(_)) => {
+                tracing::debug!("original {key} already converted, skipping");
+                return Ok(());
+            }
+            Err(e) => return Err(e),
+        }
         let bytes = self.storage.get(&key).await?;
         let format = self.format;
 

crates/adapters/image-converter/tests/conversion_handler_test.rs

@@ -162,3 +162,33 @@ async fn converts_jpeg_to_webp_and_swaps_key() {
     );
     assert!(storage.get("avatars/u1.webp").await.is_ok());
 }
+
+#[tokio::test]
+async fn image_conversion_is_idempotent_when_run_twice() {
+    for format in [Format::Webp, Format::Avif] {
+        let storage = in_memory_storage();
+        storage.store("posters/m1", &tiny_jpeg()).await.unwrap();
+        let image_ref = MockImageRef::new();
+        let handler = ImageConversionHandler::new(
+            Arc::clone(&storage) as Arc<dyn ObjectStorage>,
+            Arc::clone(&image_ref) as Arc<dyn ImageRefCommand>,
+            format,
+        );
+        let event = DomainEvent::ImageStored {
+            key: "posters/m1".into(),
+        };
+        let new_key = format!("posters/m1{}", format.extension());
+
+        handler.handle(&event).await.unwrap();
+        let converted = storage.get(&new_key).await.unwrap();
+        // Redelivery after the original was swapped and deleted.
+        handler.handle(&event).await.unwrap();
+
+        assert_eq!(image_ref.swaps().len(), 1, "one swap");
+        assert_eq!(
+            storage.get(&new_key).await.unwrap(),
+            converted,
+            "converted bytes are not re-encoded"
+        );
+    }
+}

crates/adapters/object-storage/src/lib.rs

@@ -136,3 +136,29 @@ pub fn create(config: &::config::StorageConfig) -> anyhow::Result<Arc<dyn Object
         StorageConfig::from_config(config)?,
     )))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use domain::value_objects::{MovieId, PosterPath};
+    use object_store::memory::InMemory;
+
+    #[tokio::test]
+    async fn image_cleanup_is_idempotent_when_run_twice() {
+        let storage = Arc::new(ObjectStorageAdapter::new(Arc::new(InMemory::new())));
+        storage.store("posters/m1.webp", b"img").await.unwrap();
+        let handler = ImageCleanupHandler::new(storage.clone());
+        let event = DomainEvent::MovieDeleted {
+            movie_id: MovieId::from_uuid(uuid::Uuid::new_v4()),
+            poster_path: Some(PosterPath::new("posters/m1.webp".into()).unwrap()),
+        };
+
+        handler.handle(&event).await.unwrap();
+        handler.handle(&event).await.unwrap();
+
+        assert!(matches!(
+            storage.get("posters/m1.webp").await,
+            Err(DomainError::NotFound(_))
+        ));
+    }
+}

crates/application/src/movies/enrich_movie.rs

@@ -8,15 +8,14 @@ use domain::{
 use crate::movies::{commands::EnrichMovieCommand, deps::EnrichMovieDeps};
 
 pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<(), DomainError> {
-    deps.movie_profile.upsert(&cmd.profile).await?;
-
     let persons = extract_persons(&cmd.profile.cast, &cmd.profile.crew);
     if !persons.is_empty() {
         deps.person_command.upsert_batch(&persons).await?;
     }
 
     let Some(movie) = deps.movie_query.get_movie_by_id(&cmd.movie_id).await? else {
-        tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found after profile upsert");
+        tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found, storing profile only");
+        deps.movie_profile.upsert(&cmd.profile).await?;
         return Ok(());
     };
 
@@ -37,10 +36,14 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<
             .await?;
     }
 
+    // Written last: a stored fresh profile is what makes a redelivered
+    // enrichment skip, so it must only exist once every other effect landed.
+    deps.movie_profile.upsert(&cmd.profile).await?;
+
     tracing::info!(
         movie_id = %cmd.movie_id.value(),
         persons = persons.len(),
-        "enrich_movie: profile stored and search index updated"
+        "enrich_movie: search index updated and profile stored"
     );
     Ok(())
 }

crates/application/src/movies/poster_sync_handler.rs

@@ -109,7 +109,22 @@ impl EventHandler for PosterSyncHandler {
             DomainEvent::MovieDiscovered {
                 movie_id,
                 external_metadata_id,
-            } => (movie_id.value(), external_metadata_id.value().to_owned()),
+            } => {
+                // The poster is stored under the movie id and the movie row is
+                // updated atomically with `PosterSynced`; a poster already keyed
+                // by this movie means an earlier delivery completed.
+                let id_str = movie_id.value().to_string();
+                let already_synced = self
+                    .movie_query
+                    .get_movie_by_id(&MovieId::from_uuid(movie_id.value()))
+                    .await?
+                    .and_then(|m| m.poster_path().map(|p| p.value().contains(&id_str)))
+                    .unwrap_or(false);
+                if already_synced {
+                    return Ok(());
+                }
+                (movie_id.value(), external_metadata_id.value().to_owned())
+            }
             DomainEvent::MovieEnrichmentRequested {
                 movie_id,
                 external_metadata_id,

crates/application/src/wrapup/handle_requested.rs

@@ -12,15 +12,14 @@ pub async fn execute(
     end_date: chrono::NaiveDate,
 ) -> Result<(), DomainError> {
     if let Ok(Some(rec)) = deps.wrapup_repo.get_by_id(&wrapup_id).await
-        && (rec.status == WrapUpStatus::Ready || rec.status == WrapUpStatus::Generating)
+        && rec.status == WrapUpStatus::Ready
     {
-        tracing::debug!(
+        tracing::debug!("wrapup {} already ready, skipping", wrapup_id.value());
-            "wrapup {} already {:?}, skipping",
-            wrapup_id.value(),
-            rec.status
-        );
         return Ok(());
     }
+    // A record still `Generating` is an earlier delivery that died mid-way. The
+    // ledger guarantees no concurrent run, so resume it: generation is a pure
+    // read of stats and `set_complete` is the single atomic completion write.
 
     deps.wrapup_repo
         .update_status(&wrapup_id, &WrapUpStatus::Generating, None)

crates/application/tests/handlers.rs

@@ -0,0 +1,2 @@
+#[path = "handlers/idempotency_test.rs"]
+mod idempotency_test;

crates/application/tests/handlers/idempotency_test.rs

@@ -0,0 +1,468 @@
+//! Every event handler may run twice for the same event (lease takeover after
+//! a crash mid-handler). Running it twice must leave the world as if it ran once.
+
+use std::collections::HashMap;
+use std::sync::{
+    Arc, Mutex,
+    atomic::{AtomicUsize, Ordering},
+};
+
+use async_trait::async_trait;
+use chrono::Utc;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    models::{
+        EntityType, ExternalPersonId, IndexableDocument, Movie, MovieProfile, Person,
+        PersonCredits, PersonEnrichmentData, PersonId,
+    },
+    ports::{
+        EventHandler, MetadataClient, MovieCommand, MovieEnrichmentClient, MovieProfileRepository,
+        PersonCommand, PersonEnrichmentClient, PersonQuery, SearchCommand,
+    },
+    testing::{
+        FakePosterFetcher, InMemoryStore, NoopEventPublisher, NoopObjectStorage, NoopPersonCommand,
+        test_movie_with_external_id,
+    },
+    value_objects::{ExternalMetadataId, MovieId, PosterUrl},
+};
+
+use application::movies::{
+    deps::ReindexSearchDeps, discovery_indexer::MovieDiscoveryIndexer,
+    event_handler::MovieEnrichmentHandler, poster_sync_handler::PosterSyncHandler,
+    reindex_search::SearchReindexHandler, search_cleanup::SearchCleanupHandler,
+};
+use application::person::event_handler::PersonEnrichmentHandler;
+
+// ── fakes ───────────────────────────────────────────────────────────────────
+
+/// Search index keyed like a real one: indexing the same id twice is one doc.
+#[derive(Default)]
+struct RecordingSearch {
+    docs: Mutex<HashMap<(String, String), u32>>,
+}
+
+impl RecordingSearch {
+    fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+    fn snapshot(&self) -> Vec<(String, String)> {
+        let mut keys: Vec<_> = self.docs.lock().unwrap().keys().cloned().collect();
+        keys.sort();
+        keys
+    }
+}
+
+fn doc_key(doc: &IndexableDocument) -> (String, String) {
+    match doc {
+        IndexableDocument::Movie { id, .. } => ("movie".into(), id.value().to_string()),
+        IndexableDocument::Person { id, .. } => ("person".into(), id.value().to_string()),
+    }
+}
+
+#[async_trait]
+impl SearchCommand for RecordingSearch {
+    async fn index(&self, doc: IndexableDocument) -> Result<(), DomainError> {
+        self.docs.lock().unwrap().insert(doc_key(&doc), 1);
+        Ok(())
+    }
+    async fn remove(&self, entity_type: EntityType, id: &str) -> Result<(), DomainError> {
+        let kind = match entity_type {
+            EntityType::Movie => "movie",
+            EntityType::Person => "person",
+        };
+        self.docs
+            .lock()
+            .unwrap()
+            .remove(&(kind.to_string(), id.to_string()));
+        Ok(())
+    }
+}
+
+/// Fails the first `index` call, succeeds afterwards.
+struct FlakySearch {
+    inner: Arc<RecordingSearch>,
+    failed_once: AtomicUsize,
+}
+
+#[async_trait]
+impl SearchCommand for FlakySearch {
+    async fn index(&self, doc: IndexableDocument) -> Result<(), DomainError> {
+        if self.failed_once.fetch_add(1, Ordering::SeqCst) == 0 {
+            return Err(DomainError::InfrastructureError("index down".into()));
+        }
+        self.inner.index(doc).await
+    }
+    async fn remove(&self, t: EntityType, id: &str) -> Result<(), DomainError> {
+        self.inner.remove(t, id).await
+    }
+}
+
+fn person(tmdb: u64, name: &str) -> Person {
+    let ext = ExternalPersonId::new(format!("tmdb:{tmdb}"));
+    Person::basic(PersonId::from_external(&ext), ext, name.into(), None, None)
+}
+
+/// One person whose `enriched_at` is set by `update_enrichment`, like the real store.
+struct FakePersons {
+    person: Mutex<Person>,
+    enrich_writes: AtomicUsize,
+}
+
+impl FakePersons {
+    fn new(p: Person) -> Arc<Self> {
+        Arc::new(Self {
+            person: Mutex::new(p),
+            enrich_writes: AtomicUsize::new(0),
+        })
+    }
+}
+
+#[async_trait]
+impl PersonQuery for FakePersons {
+    async fn get_by_id(&self, id: &PersonId) -> Result<Option<Person>, DomainError> {
+        let p = self.person.lock().unwrap().clone();
+        Ok((p.id() == id).then_some(p))
+    }
+    async fn get_by_external_id(
+        &self,
+        _: &ExternalPersonId,
+    ) -> Result<Option<Person>, DomainError> {
+        Ok(None)
+    }
+    async fn get_credits(&self, _: &PersonId) -> Result<PersonCredits, DomainError> {
+        unimplemented!()
+    }
+    async fn list_orphaned_persons(&self) -> Result<Vec<PersonId>, DomainError> {
+        Ok(vec![self.person.lock().unwrap().id().clone()])
+    }
+    async fn list_page(&self, _limit: u32, offset: u32) -> Result<Vec<Person>, DomainError> {
+        if offset == 0 {
+            Ok(vec![self.person.lock().unwrap().clone()])
+        } else {
+            Ok(vec![])
+        }
+    }
+}
+
+#[async_trait]
+impl PersonCommand for FakePersons {
+    async fn upsert_batch(&self, _: &[Person]) -> Result<(), DomainError> {
+        Ok(())
+    }
+    async fn backfill_from_credits_batch(&self, _: u32) -> Result<(u64, bool), DomainError> {
+        Ok((0, false))
+    }
+    async fn update_enrichment(
+        &self,
+        id: &PersonId,
+        data: &PersonEnrichmentData,
+    ) -> Result<(), DomainError> {
+        self.enrich_writes.fetch_add(1, Ordering::SeqCst);
+        let mut p = self.person.lock().unwrap();
+        assert_eq!(p.id(), id);
+        *p = Person::new(
+            p.id().clone(),
+            p.external_id().clone(),
+            p.name().to_string(),
+            p.known_for_department().map(str::to_string),
+            p.profile_path().map(str::to_string),
+            data.biography.clone(),
+            data.birthday,
+            data.deathday,
+            data.place_of_birth.clone(),
+            data.also_known_as.clone(),
+            data.homepage.clone(),
+            data.imdb_id.clone(),
+            Some(Utc::now()),
+        );
+        Ok(())
+    }
+}
+
+struct CountingPersonClient(AtomicUsize);
+
+#[async_trait]
+impl PersonEnrichmentClient for CountingPersonClient {
+    async fn fetch_details(&self, _: &str) -> Result<PersonEnrichmentData, DomainError> {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Ok(PersonEnrichmentData {
+            biography: Some("bio".into()),
+            birthday: None,
+            deathday: None,
+            place_of_birth: None,
+            also_known_as: vec![],
+            homepage: None,
+            imdb_id: None,
+        })
+    }
+}
+
+fn profile_for(movie_id: &MovieId) -> MovieProfile {
+    MovieProfile {
+        movie_id: movie_id.clone(),
+        tmdb_id: 42,
+        imdb_id: None,
+        overview: None,
+        tagline: None,
+        runtime_minutes: None,
+        budget_usd: None,
+        revenue_usd: None,
+        vote_average: None,
+        vote_count: None,
+        original_language: None,
+        collection_name: None,
+        genres: vec![],
+        keywords: vec![],
+        cast: vec![],
+        crew: vec![],
+        enriched_at: Utc::now(),
+    }
+}
+
+struct CountingMovieClient(AtomicUsize);
+
+#[async_trait]
+impl MovieEnrichmentClient for CountingMovieClient {
+    async fn fetch_profile(&self, movie_id: MovieId, _: &str) -> Result<MovieProfile, DomainError> {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Ok(profile_for(&movie_id))
+    }
+}
+
+struct NoImages;
+
+#[async_trait]
+impl domain::ports::ImageFetcher for NoImages {
+    async fn fetch_image(&self, _: &str) -> Result<Vec<u8>, DomainError> {
+        Err(DomainError::NotFound("none".into()))
+    }
+}
+
+struct MetadataWithPoster;
+
+#[async_trait]
+impl MetadataClient for MetadataWithPoster {
+    async fn fetch_movie_metadata(
+        &self,
+        _: &domain::models::MetadataSearchCriteria,
+    ) -> Result<Movie, DomainError> {
+        unimplemented!()
+    }
+    async fn get_poster_url(
+        &self,
+        _: &ExternalMetadataId,
+    ) -> Result<Option<PosterUrl>, DomainError> {
+        Ok(Some(
+            PosterUrl::new("https://example.com/p.jpg".into()).unwrap(),
+        ))
+    }
+}
+
+async fn seeded_movie(store: &Arc<InMemoryStore>) -> (Movie, ExternalMetadataId) {
+    let ext = ExternalMetadataId::new("tmdb:42".into()).unwrap();
+    let movie = test_movie_with_external_id("Heat", 1995, ext.clone());
+    store.upsert_movie(&movie, &[]).await.unwrap();
+    (movie, ext)
+}
+
+// ── person_enrichment ───────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn person_enrichment_is_idempotent_when_run_twice() {
+    let p = person(7, "Al Pacino");
+    let persons = FakePersons::new(p.clone());
+    let client = Arc::new(CountingPersonClient(AtomicUsize::new(0)));
+    let handler = PersonEnrichmentHandler::new(
+        persons.clone(),
+        Some(client.clone() as Arc<dyn PersonEnrichmentClient>),
+        persons.clone(),
+    );
+    let event = DomainEvent::PersonEnrichmentRequested {
+        person_id: p.id().clone(),
+        external_person_id: p.external_id().clone(),
+    };
+
+    handler.handle(&event).await.unwrap();
+    handler.handle(&event).await.unwrap();
+
+    assert_eq!(client.0.load(Ordering::SeqCst), 1);
+    assert_eq!(persons.enrich_writes.load(Ordering::SeqCst), 1);
+}
+
+// ── movie_enrichment ────────────────────────────────────────────────────────
+
+fn enrichment_handler(
+    store: &Arc<InMemoryStore>,
+    client: Arc<CountingMovieClient>,
+    search: Arc<dyn SearchCommand>,
+) -> MovieEnrichmentHandler {
+    MovieEnrichmentHandler::new(
+        client,
+        Arc::clone(store) as _,
+        Arc::clone(store) as _,
+        Arc::new(NoopPersonCommand),
+        search,
+        Arc::new(NoopObjectStorage),
+        Arc::new(NoImages),
+    )
+}
+
+#[tokio::test]
+async fn movie_enrichment_is_idempotent_when_run_twice() {
+    let store = InMemoryStore::new();
+    let (movie, ext) = seeded_movie(&store).await;
+    let search = RecordingSearch::new();
+    let client = Arc::new(CountingMovieClient(AtomicUsize::new(0)));
+    let handler = enrichment_handler(&store, client.clone(), search.clone());
+    let event = DomainEvent::MovieEnrichmentRequested {
+        movie_id: movie.id().clone(),
+        external_metadata_id: ext,
+    };
+
+    handler.handle(&event).await.unwrap();
+    let after_first = search.snapshot();
+    handler.handle(&event).await.unwrap();
+
+    assert_eq!(search.snapshot(), after_first);
+    assert_eq!(client.0.load(Ordering::SeqCst), 1, "no second remote fetch");
+}
+
+#[tokio::test]
+async fn a_movie_enrichment_that_failed_midway_is_redone_on_redelivery() {
+    let store = InMemoryStore::new();
+    let (movie, ext) = seeded_movie(&store).await;
+    let search = RecordingSearch::new();
+    let flaky = Arc::new(FlakySearch {
+        inner: search.clone(),
+        failed_once: AtomicUsize::new(0),
+    });
+    let client = Arc::new(CountingMovieClient(AtomicUsize::new(0)));
+    let handler = enrichment_handler(&store, client, flaky);
+    let event = DomainEvent::MovieEnrichmentRequested {
+        movie_id: movie.id().clone(),
+        external_metadata_id: ext,
+    };
+
+    assert!(handler.handle(&event).await.is_err());
+    // The profile row is the "enrichment done" marker: it must not exist yet,
+    // or the redelivery below would skip and the search index would stay empty.
+    assert!(store.get_by_movie_id(movie.id()).await.unwrap().is_none());
+
+    handler.handle(&event).await.unwrap();
+    assert_eq!(search.snapshot().len(), 1);
+    assert!(store.get_by_movie_id(movie.id()).await.unwrap().is_some());
+}
+
+// ── search_cleanup ──────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn search_cleanup_is_idempotent_when_run_twice() {
+    let orphan = person(9, "Orphan");
+    let persons = FakePersons::new(orphan.clone());
+    let search = RecordingSearch::new();
+    let movie_id = MovieId::from_uuid(uuid::Uuid::new_v4());
+    search
+        .index(IndexableDocument::Person {
+            id: orphan.id().clone(),
+            person: Box::new(orphan.clone()),
+        })
+        .await
+        .unwrap();
+    let handler = SearchCleanupHandler::new(search.clone(), persons);
+    let event = DomainEvent::MovieDeleted {
+        movie_id,
+        poster_path: None,
+    };
+
+    handler.handle(&event).await.unwrap();
+    let after_first = search.snapshot();
+    handler.handle(&event).await.unwrap();
+
+    assert!(after_first.is_empty());
+    assert_eq!(search.snapshot(), after_first);
+}
+
+// ── movie_discovery_indexer ─────────────────────────────────────────────────
+
+#[tokio::test]
+async fn movie_discovery_indexer_is_idempotent_when_run_twice() {
+    let store = InMemoryStore::new();
+    let (movie, ext) = seeded_movie(&store).await;
+    let search = RecordingSearch::new();
+    let handler = MovieDiscoveryIndexer::new(store.clone(), search.clone());
+    let event = DomainEvent::MovieDiscovered {
+        movie_id: movie.id().clone(),
+        external_metadata_id: ext,
+    };
+
+    handler.handle(&event).await.unwrap();
+    let after_first = search.snapshot();
+    handler.handle(&event).await.unwrap();
+
+    assert_eq!(after_first.len(), 1);
+    assert_eq!(search.snapshot(), after_first);
+}
+
+// ── search_reindex ──────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn search_reindex_is_idempotent_when_run_twice() {
+    let store = InMemoryStore::new();
+    seeded_movie(&store).await;
+    let persons = FakePersons::new(person(3, "Someone"));
+    let search = RecordingSearch::new();
+    let handler = SearchReindexHandler::new(ReindexSearchDeps {
+        movie_query: store.clone(),
+        movie_profile: store.clone(),
+        search_command: search.clone(),
+        person_command: persons.clone(),
+        person_query: persons,
+    });
+
+    handler
+        .handle(&DomainEvent::SearchReindexRequested)
+        .await
+        .unwrap();
+    let after_first = search.snapshot();
+    handler
+        .handle(&DomainEvent::SearchReindexRequested)
+        .await
+        .unwrap();
+
+    assert!(!after_first.is_empty());
+    assert_eq!(search.snapshot(), after_first);
+}
+
+// ── poster_sync ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn poster_sync_is_idempotent_when_run_twice() {
+    let store = InMemoryStore::new();
+    let (movie, ext) = seeded_movie(&store).await;
+    let publisher = NoopEventPublisher::new();
+    let handler = PosterSyncHandler::new(
+        store.clone(),
+        store.clone(),
+        Arc::new(MetadataWithPoster),
+        Arc::new(FakePosterFetcher),
+        Arc::new(NoopObjectStorage),
+        publisher.clone(),
+        0,
+    );
+    let event = DomainEvent::MovieDiscovered {
+        movie_id: movie.id().clone(),
+        external_metadata_id: ext,
+    };
+
+    handler.handle(&event).await.unwrap();
+    handler.handle(&event).await.unwrap();
+
+    let synced = store
+        .recorded_events()
+        .iter()
+        .filter(|e| matches!(e, DomainEvent::PosterSynced { .. }))
+        .count();
+    assert_eq!(synced, 1, "PosterSynced fans out to federation: emit once");
+    assert_eq!(publisher.published().len(), 1, "one ImageStored, not two");
+}

crates/application/tests/wrapup/handle_requested_test.rs

@@ -92,7 +92,9 @@ async fn generates_wrapup_and_marks_complete() {
 }
 
 #[tokio::test]
-async fn skips_if_already_generating() {
+async fn a_wrapup_left_generating_is_completed_on_redelivery() {
+    // A crash after `Generating` was written (or a failed `set_complete`) leaves
+    // the record stuck; the redelivered event must finish it, not skip it.
     let repo = InMemoryWrapUpRepository::new();
     let wrapup_id = WrapUpId::generate();
 
@@ -114,21 +116,62 @@ async fn skips_if_already_generating() {
         wrapup_stats: InMemoryWrapUpStatsQuery::new(),
     };
 
-    let result = handle_requested::execute(
+    handle_requested::execute(
         &deps,
         wrapup_id.clone(),
         None,
         NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
         NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(),
     )
-    .await;
+    .await
+    .unwrap();
 
-    assert!(result.is_ok());
-
-    // Status should still be Generating (not changed to Ready)
     let final_rec = repo.get_by_id(&wrapup_id).await.unwrap().unwrap();
-    assert_eq!(final_rec.status, WrapUpStatus::Generating);
+    assert_eq!(final_rec.status, WrapUpStatus::Ready);
-    assert!(repo.recorded_events().is_empty());
+    assert!(final_rec.report.is_some());
+    assert_eq!(repo.recorded_events().len(), 1);
+}
+
+#[tokio::test]
+async fn wrapup_is_idempotent_when_run_twice() {
+    let repo = InMemoryWrapUpRepository::new();
+    let wrapup_id = WrapUpId::generate();
+    let record = WrapUpRecord {
+        id: wrapup_id.clone(),
+        user_id: None,
+        start_date: NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
+        end_date: NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(),
+        status: WrapUpStatus::Pending,
+        report: None,
+        error_message: None,
+        created_at: Utc::now().naive_utc(),
+        completed_at: None,
+    };
+    repo.create(&record, &[]).await.unwrap();
+    let deps = HandleWrapUpRequestedDeps {
+        wrapup_repo: repo.clone(),
+        wrapup_stats: InMemoryWrapUpStatsQuery::new(),
+    };
+
+    for _ in 0..2 {
+        handle_requested::execute(
+            &deps,
+            wrapup_id.clone(),
+            None,
+            NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
+            NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(),
+        )
+        .await
+        .unwrap();
+    }
+
+    let rec = repo.get_by_id(&wrapup_id).await.unwrap().unwrap();
+    assert_eq!(rec.status, WrapUpStatus::Ready);
+    assert_eq!(
+        repo.recorded_events().len(),
+        1,
+        "WrapUpCompleted emitted once"
+    );
 }
 
 #[tokio::test]

crates/bootstrap/src/follow_backfill_handler.rs

@@ -51,3 +51,75 @@ impl EventHandler for FollowBackfillHandler {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashSet;
+    use std::sync::Mutex;
+
+    use domain::models::FollowedActorInfo;
+    use domain::value_objects::UserId;
+    use uuid::Uuid;
+
+    use super::*;
+
+    /// Models the real port: imports are keyed upserts and backfill sends carry
+    /// deterministic activity ids, so a repeat lands on the same key.
+    #[derive(Default)]
+    struct KeyedBackfill {
+        imported: Mutex<HashSet<(String, String)>>,
+        sent: Mutex<HashSet<(Uuid, String)>>,
+    }
+
+    #[async_trait]
+    impl ApBackfillPort for KeyedBackfill {
+        async fn get_following(&self, _: Uuid) -> Result<Vec<FollowedActorInfo>, DomainError> {
+            Ok(vec![FollowedActorInfo {
+                url: "https://remote.example/users/a".into(),
+                outbox_url: Some("https://remote.example/users/a/outbox".into()),
+            }])
+        }
+        async fn import_remote_outbox(&self, outbox: &str, actor: &str) -> Result<(), DomainError> {
+            self.imported
+                .lock()
+                .unwrap()
+                .insert((outbox.into(), actor.into()));
+            Ok(())
+        }
+        async fn run_backfill_for_follower(
+            &self,
+            owner: Uuid,
+            inbox: String,
+        ) -> Result<(), DomainError> {
+            self.sent.lock().unwrap().insert((owner, inbox));
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn follow_backfill_is_idempotent_when_run_twice() {
+        let port = Arc::new(KeyedBackfill::default());
+        let handler = FollowBackfillHandler {
+            backfill: port.clone(),
+        };
+        let owner = UserId::from_uuid(Uuid::new_v4());
+        let accepted = DomainEvent::FollowAccepted {
+            owner: owner.clone(),
+            requester: SocialIdentity::Remote {
+                actor_url: "https://remote.example/users/a".into(),
+            },
+        };
+        let backfill = DomainEvent::BackfillFollower {
+            owner_user_id: owner,
+            follower_inbox_url: "https://remote.example/inbox".into(),
+        };
+
+        for event in [&accepted, &backfill] {
+            handler.handle(event).await.unwrap();
+            handler.handle(event).await.unwrap();
+        }
+
+        assert_eq!(port.imported.lock().unwrap().len(), 1);
+        assert_eq!(port.sent.lock().unwrap().len(), 1);
+    }
+}