From 4cb2972f633d5a6cd4822098a837f3493091f0fb Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:27:03 +0100 Subject: [PATCH] feat: idempotent handlers, deterministic activity ids --- crates/adapters/activitypub/src/urls.rs | 31 ++ .../adapters/image-converter/src/handler.rs | 11 + .../tests/conversion_handler_test.rs | 30 ++ crates/adapters/object-storage/src/lib.rs | 26 + crates/application/src/movies/enrich_movie.rs | 11 +- .../src/movies/poster_sync_handler.rs | 17 +- .../src/wrapup/handle_requested.rs | 11 +- crates/application/tests/handlers.rs | 2 + .../tests/handlers/idempotency_test.rs | 468 ++++++++++++++++++ .../tests/wrapup/handle_requested_test.rs | 59 ++- .../bootstrap/src/follow_backfill_handler.rs | 72 +++ 11 files changed, 719 insertions(+), 19 deletions(-) create mode 100644 crates/application/tests/handlers.rs create mode 100644 crates/application/tests/handlers/idempotency_test.rs diff --git a/crates/adapters/activitypub/src/urls.rs b/crates/adapters/activitypub/src/urls.rs index 8b36a405..ddb3458b 100644 --- a/crates/adapters/activitypub/src/urls.rs +++ b/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) + ); + } +} diff --git a/crates/adapters/image-converter/src/handler.rs b/crates/adapters/image-converter/src/handler.rs index 4664e687..a0e8a3f9 100644 --- a/crates/adapters/image-converter/src/handler.rs +++ b/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; diff --git a/crates/adapters/image-converter/tests/conversion_handler_test.rs b/crates/adapters/image-converter/tests/conversion_handler_test.rs index b47648a3..c23e48c6 100644 --- a/crates/adapters/image-converter/tests/conversion_handler_test.rs +++ b/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, + Arc::clone(&image_ref) as Arc, + 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" + ); + } +} diff --git a/crates/adapters/object-storage/src/lib.rs b/crates/adapters/object-storage/src/lib.rs index b55fde82..556b6440 100644 --- a/crates/adapters/object-storage/src/lib.rs +++ b/crates/adapters/object-storage/src/lib.rs @@ -136,3 +136,29 @@ pub fn create(config: &::config::StorageConfig) -> anyhow::Result 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(()) } diff --git a/crates/application/src/movies/poster_sync_handler.rs b/crates/application/src/movies/poster_sync_handler.rs index faa99518..cff146b3 100644 --- a/crates/application/src/movies/poster_sync_handler.rs +++ b/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, diff --git a/crates/application/src/wrapup/handle_requested.rs b/crates/application/src/wrapup/handle_requested.rs index c696ec15..0332f5a8 100644 --- a/crates/application/src/wrapup/handle_requested.rs +++ b/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!( - "wrapup {} already {:?}, skipping", - wrapup_id.value(), - rec.status - ); + tracing::debug!("wrapup {} already ready, skipping", wrapup_id.value()); 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) diff --git a/crates/application/tests/handlers.rs b/crates/application/tests/handlers.rs new file mode 100644 index 00000000..8154d2a8 --- /dev/null +++ b/crates/application/tests/handlers.rs @@ -0,0 +1,2 @@ +#[path = "handlers/idempotency_test.rs"] +mod idempotency_test; diff --git a/crates/application/tests/handlers/idempotency_test.rs b/crates/application/tests/handlers/idempotency_test.rs new file mode 100644 index 00000000..0c8bf976 --- /dev/null +++ b/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>, +} + +impl RecordingSearch { + fn new() -> Arc { + 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, + 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, + enrich_writes: AtomicUsize, +} + +impl FakePersons { + fn new(p: Person) -> Arc { + 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, DomainError> { + let p = self.person.lock().unwrap().clone(); + Ok((p.id() == id).then_some(p)) + } + async fn get_by_external_id( + &self, + _: &ExternalPersonId, + ) -> Result, DomainError> { + Ok(None) + } + async fn get_credits(&self, _: &PersonId) -> Result { + unimplemented!() + } + async fn list_orphaned_persons(&self) -> Result, DomainError> { + Ok(vec![self.person.lock().unwrap().id().clone()]) + } + async fn list_page(&self, _limit: u32, offset: u32) -> Result, 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 { + 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 { + 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, DomainError> { + Err(DomainError::NotFound("none".into())) + } +} + +struct MetadataWithPoster; + +#[async_trait] +impl MetadataClient for MetadataWithPoster { + async fn fetch_movie_metadata( + &self, + _: &domain::models::MetadataSearchCriteria, + ) -> Result { + unimplemented!() + } + async fn get_poster_url( + &self, + _: &ExternalMetadataId, + ) -> Result, DomainError> { + Ok(Some( + PosterUrl::new("https://example.com/p.jpg".into()).unwrap(), + )) + } +} + +async fn seeded_movie(store: &Arc) -> (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), + 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, + client: Arc, + search: Arc, +) -> 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"); +} diff --git a/crates/application/tests/wrapup/handle_requested_test.rs b/crates/application/tests/wrapup/handle_requested_test.rs index 53177b9a..ed69f321 100644 --- a/crates/application/tests/wrapup/handle_requested_test.rs +++ b/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!(repo.recorded_events().is_empty()); + assert_eq!(final_rec.status, WrapUpStatus::Ready); + 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] diff --git a/crates/bootstrap/src/follow_backfill_handler.rs b/crates/bootstrap/src/follow_backfill_handler.rs index 37b18070..5cce039b 100644 --- a/crates/bootstrap/src/follow_backfill_handler.rs +++ b/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>, + sent: Mutex>, + } + + #[async_trait] + impl ApBackfillPort for KeyedBackfill { + async fn get_following(&self, _: Uuid) -> Result, 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); + } +}