diff --git a/crates/adapters/activitypub/Cargo.toml b/crates/adapters/activitypub/Cargo.toml index c00dcf72..98e1ef85 100644 --- a/crates/adapters/activitypub/Cargo.toml +++ b/crates/adapters/activitypub/Cargo.toml @@ -18,3 +18,7 @@ async-trait = { workspace = true } url = { version = "2", features = ["serde"] } ammonia = "4" + +[dev-dependencies] +tokio = { workspace = true } +domain = { workspace = true, features = ["test-helpers"] } diff --git a/crates/adapters/activitypub/src/urls.rs b/crates/adapters/activitypub/src/urls.rs index ddb3458b..8b36a405 100644 --- a/crates/adapters/activitypub/src/urls.rs +++ b/crates/adapters/activitypub/src/urls.rs @@ -38,34 +38,3 @@ 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/activitypub/tests/activity_ids.rs b/crates/adapters/activitypub/tests/activity_ids.rs new file mode 100644 index 00000000..4a52d46e --- /dev/null +++ b/crates/adapters/activitypub/tests/activity_ids.rs @@ -0,0 +1,193 @@ +//! Outbound activity ids across a redelivered event, through the real k-ap +//! `ActivityPubService` (delivery captured via its `EventPublisher` seam). +//! +//! k-ap 0.5.0 derives Create ids from the object id (UUIDv5, +//! `service/broadcast.rs:285-288`) but mints Update/Delete/Move/actor-update ids +//! randomly (`broadcast.rs:81,129,157,210,321,343,377` -> `url_scheme.rs:53`, +//! `Uuid::new_v4`) with no override. + +use std::sync::Arc; +use std::time::Duration; + +use activitypub::ActivityPubEventHandler; +use async_trait::async_trait; +use chrono::NaiveDate; +use domain::{ + errors::DomainError, + events::DomainEvent, + models::{DiaryEntry, WatchlistWithMovie}, + ports::{EventHandler, LocalApContentQuery, MovieCommand}, + testing::{ + InMemoryStore, PanicGoalQuery, PanicReviewRepository, PanicStatsRepository, test_movie, + }, + value_objects::{InstanceIdentity, MovieId, UserId}, +}; +use k_ap::{ + ActivityPubService, ApUser, FederationEvent, + testing::{MockActorRepoBuilder, MockFollowRepoBuilder, MockUserRepoBuilder}, +}; +use tokio::sync::mpsc; + +struct NoContent; + +#[async_trait] +impl LocalApContentQuery for NoContent { + async fn get_local_watchlist_for_user( + &self, + _: &UserId, + ) -> Result, DomainError> { + Ok(vec![]) + } + async fn get_local_reviews_for_movie( + &self, + _: &MovieId, + ) -> Result, DomainError> { + Ok(vec![]) + } + async fn get_local_reviews_page( + &self, + _: &UserId, + _: Option, + _: usize, + ) -> Result, DomainError> { + Ok(vec![]) + } +} + +struct Capture(mpsc::UnboundedSender); + +#[async_trait] +impl k_ap::EventPublisher for Capture { + async fn publish(&self, event: FederationEvent) -> anyhow::Result<()> { + if let FederationEvent::DeliveryRequested { activity, .. } = event { + let _ = self.0.send(activity); + } + Ok(()) + } +} + +struct Harness { + handler: ActivityPubEventHandler, + delivered: mpsc::UnboundedReceiver, + user: UserId, + movie_id: MovieId, +} + +async fn harness() -> Harness { + let user = UserId::from_uuid(uuid::Uuid::new_v4()); + let uid = user.value(); + let (tx, rx) = mpsc::unbounded_channel(); + let service = ActivityPubService::builder("https://diary.example") + .activity_repo(k_ap::testing::MockActivityRepoBuilder::new().build()) + .follow_repo( + MockFollowRepoBuilder::new() + .on_get_accepted_follower_inboxes(|_| { + Ok(vec!["https://remote.example/inbox".to_string()]) + }) + .build(), + ) + .actor_repo(MockActorRepoBuilder::new().build()) + .blocklist_repo(k_ap::testing::MockBlocklistRepoBuilder::new().build()) + .user_repo( + MockUserRepoBuilder::new() + .on_find_by_id(move |id| { + Ok((id == uid).then(|| ApUser { + id, + username: "alice".into(), + display_name: None, + bio: None, + avatar_url: None, + banner_url: None, + also_known_as: vec![], + profile_url: None, + attachment: vec![], + manually_approves_followers: false, + discoverable: true, + actor_type: Default::default(), + featured_url: None, + })) + }) + .build(), + ) + .content_reader(k_ap::testing::MockContentReaderBuilder::new().build()) + .object_handler(k_ap::testing::MockObjectHandlerBuilder::new().build()) + .event_publisher(Arc::new(Capture(tx))) + .build() + .await + .expect("service builds"); + + let store = InMemoryStore::new(); + let movie = test_movie("Heat"); + store.upsert_movie(&movie, &[]).await.unwrap(); + let handler = ActivityPubEventHandler::new( + Arc::new(service), + Arc::new(NoContent), + Arc::new(PanicReviewRepository), + store.clone(), + Arc::new(PanicGoalQuery), + Arc::new(PanicStatsRepository), + store, + InstanceIdentity::new("https://diary.example"), + ); + Harness { + handler, + delivered: rx, + user, + movie_id: movie.id().clone(), + } +} + +impl Harness { + async fn next_activity_id(&mut self) -> String { + let activity = tokio::time::timeout(Duration::from_secs(5), self.delivered.recv()) + .await + .expect("an activity was delivered") + .expect("channel open"); + activity["id"] + .as_str() + .expect("activity has an id") + .to_string() + } +} + +#[tokio::test] +async fn activitypub_is_idempotent_when_run_twice() { + let mut h = harness().await; + let event = DomainEvent::WatchlistEntryAdded { + user_id: h.user.clone(), + movie_id: h.movie_id.clone(), + movie_title: "Heat".into(), + release_year: 1995, + external_metadata_id: None, + added_at: NaiveDate::from_ymd_opt(2026, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(), + }; + + h.handler.handle(&event).await.unwrap(); + let first = h.next_activity_id().await; + h.handler.handle(&event).await.unwrap(); + let second = h.next_activity_id().await; + + assert!(first.starts_with("https://diary.example/activities/create/")); + assert_eq!(first, second, "a redelivered Create keeps its activity id"); +} + +/// Characterises the k-ap 0.5.0 limitation recorded in ADR-0015. If k-ap starts +/// minting deterministic Delete ids this fails: upgrade the ADR and flip it. +#[tokio::test] +async fn redelivered_delete_gets_a_fresh_activity_id_known_k_ap_limitation() { + let mut h = harness().await; + let event = DomainEvent::WatchlistEntryRemoved { + user_id: h.user.clone(), + movie_id: h.movie_id.clone(), + }; + + h.handler.handle(&event).await.unwrap(); + let first = h.next_activity_id().await; + h.handler.handle(&event).await.unwrap(); + let second = h.next_activity_id().await; + + assert_ne!(first, second); +} diff --git a/crates/adapters/image-converter/src/handler.rs b/crates/adapters/image-converter/src/handler.rs index a0e8a3f9..046562cb 100644 --- a/crates/adapters/image-converter/src/handler.rs +++ b/crates/adapters/image-converter/src/handler.rs @@ -48,13 +48,9 @@ impl EventHandler for ImageConversionHandler { // `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), + if !self.storage.exists(&key).await? { + tracing::warn!(%key, "original image already gone, treating conversion as done"); + return Ok(()); } let bytes = self.storage.get(&key).await?; let format = self.format; diff --git a/crates/adapters/object-storage/src/lib.rs b/crates/adapters/object-storage/src/lib.rs index 556b6440..5f8190ae 100644 --- a/crates/adapters/object-storage/src/lib.rs +++ b/crates/adapters/object-storage/src/lib.rs @@ -82,6 +82,14 @@ impl ObjectStorage for ObjectStorageAdapter { Ok(Box::pin(stream)) } + async fn exists(&self, key: &str) -> Result { + match self.store.head(&Path::from(key)).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(e) => Err(DomainError::InfrastructureError(e.to_string())), + } + } + async fn delete(&self, key: &str) -> Result<(), DomainError> { let path = Path::from(key); match self.store.delete(&path).await { diff --git a/crates/application/src/movies/enrich_movie.rs b/crates/application/src/movies/enrich_movie.rs index 1cf14f6a..c0510e20 100644 --- a/crates/application/src/movies/enrich_movie.rs +++ b/crates/application/src/movies/enrich_movie.rs @@ -14,8 +14,9 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result< } 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, storing profile only"); - deps.movie_profile.upsert(&cmd.profile).await?; + // The movie row is written with the event, so a missing row means it was + // deleted. Do not store the profile: it is the "enrichment done" marker. + tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found, skipping"); return Ok(()); }; diff --git a/crates/application/src/movies/poster_sync_handler.rs b/crates/application/src/movies/poster_sync_handler.rs index cff146b3..1f9e4620 100644 --- a/crates/application/src/movies/poster_sync_handler.rs +++ b/crates/application/src/movies/poster_sync_handler.rs @@ -110,15 +110,24 @@ impl EventHandler for PosterSyncHandler { movie_id, external_metadata_id, } => { - // 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. + // The poster is stored under the movie id (the converter later + // appends `.webp`/`.avif`) and the movie row is updated + // atomically with `PosterSynced`; exactly that path on the movie + // means an earlier delivery completed. let id_str = movie_id.value().to_string(); + let canonical = [ + id_str.clone(), + format!("{id_str}.webp"), + format!("{id_str}.avif"), + ]; 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))) + .and_then(|m| { + m.poster_path() + .map(|p| canonical.iter().any(|k| k == p.value())) + }) .unwrap_or(false); if already_synced { return Ok(()); diff --git a/crates/application/tests/handlers/idempotency_test.rs b/crates/application/tests/handlers/idempotency_test.rs index 0c8bf976..0028e324 100644 --- a/crates/application/tests/handlers/idempotency_test.rs +++ b/crates/application/tests/handlers/idempotency_test.rs @@ -24,7 +24,7 @@ use domain::{ FakePosterFetcher, InMemoryStore, NoopEventPublisher, NoopObjectStorage, NoopPersonCommand, test_movie_with_external_id, }, - value_objects::{ExternalMetadataId, MovieId, PosterUrl}, + value_objects::{ExternalMetadataId, MovieId, PosterPath, PosterUrl}, }; use application::movies::{ @@ -239,6 +239,32 @@ impl domain::ports::ImageFetcher for NoImages { } } +/// Returns the key it was given, like the real adapter. +struct KeyStorage; + +#[async_trait] +impl domain::ports::ObjectStorage for KeyStorage { + async fn store(&self, key: &str, _: &[u8]) -> Result { + Ok(key.to_string()) + } + async fn get(&self, _: &str) -> Result, DomainError> { + Ok(vec![]) + } + async fn get_stream( + &self, + _: &str, + ) -> Result>, DomainError> + { + Ok(Box::pin(futures::stream::empty())) + } + async fn delete(&self, _: &str) -> Result<(), DomainError> { + Ok(()) + } + async fn exists(&self, _: &str) -> Result { + Ok(true) + } +} + struct MetadataWithPoster; #[async_trait] @@ -409,7 +435,7 @@ async fn movie_discovery_indexer_is_idempotent_when_run_twice() { #[tokio::test] async fn search_reindex_is_idempotent_when_run_twice() { let store = InMemoryStore::new(); - seeded_movie(&store).await; + let (movie, _) = seeded_movie(&store).await; let persons = FakePersons::new(person(3, "Someone")); let search = RecordingSearch::new(); let handler = SearchReindexHandler::new(ReindexSearchDeps { @@ -430,7 +456,11 @@ async fn search_reindex_is_idempotent_when_run_twice() { .await .unwrap(); - assert!(!after_first.is_empty()); + assert!( + after_first.contains(&("movie".to_string(), movie.id().value().to_string())), + "the movie doc is indexed" + ); + assert!(after_first.iter().any(|(k, _)| k == "person")); assert_eq!(search.snapshot(), after_first); } @@ -446,7 +476,7 @@ async fn poster_sync_is_idempotent_when_run_twice() { store.clone(), Arc::new(MetadataWithPoster), Arc::new(FakePosterFetcher), - Arc::new(NoopObjectStorage), + Arc::new(KeyStorage), publisher.clone(), 0, ); @@ -466,3 +496,54 @@ async fn poster_sync_is_idempotent_when_run_twice() { assert_eq!(synced, 1, "PosterSynced fans out to federation: emit once"); assert_eq!(publisher.published().len(), 1, "one ImageStored, not two"); } + +#[tokio::test] +async fn poster_sync_skips_converted_poster_keys_but_not_lookalike_paths() { + let store = InMemoryStore::new(); + let (mut movie, ext) = seeded_movie(&store).await; + let id = movie.id().value().to_string(); + let publisher = NoopEventPublisher::new(); + let handler = PosterSyncHandler::new( + store.clone(), + store.clone(), + Arc::new(MetadataWithPoster), + Arc::new(FakePosterFetcher), + Arc::new(KeyStorage), + publisher.clone(), + 0, + ); + let event = DomainEvent::MovieDiscovered { + movie_id: movie.id().clone(), + external_metadata_id: ext, + }; + + // The converter's output is the canonical synced key: skipped. + movie.update_poster(PosterPath::new(format!("{id}.webp")).unwrap()); + store.upsert_movie(&movie, &[]).await.unwrap(); + handler.handle(&event).await.unwrap(); + assert!(publisher.published().is_empty()); + + // A foreign path that merely contains the id is not a synced poster. + movie.update_poster(PosterPath::new(format!("https://cdn.example/{id}/p.jpg")).unwrap()); + store.upsert_movie(&movie, &[]).await.unwrap(); + handler.handle(&event).await.unwrap(); + assert_eq!(publisher.published().len(), 1); +} + +#[tokio::test] +async fn enrichment_of_a_deleted_movie_stores_no_profile() { + let store = InMemoryStore::new(); + let ghost = MovieId::from_uuid(uuid::Uuid::new_v4()); + let search = RecordingSearch::new(); + let client = Arc::new(CountingMovieClient(AtomicUsize::new(0))); + let handler = enrichment_handler(&store, client, search.clone()); + let event = DomainEvent::MovieEnrichmentRequested { + movie_id: ghost.clone(), + external_metadata_id: ExternalMetadataId::new("tmdb:1".into()).unwrap(), + }; + + handler.handle(&event).await.unwrap(); + + assert!(store.get_by_movie_id(&ghost).await.unwrap().is_none()); + assert!(search.snapshot().is_empty()); +} diff --git a/crates/bootstrap/src/follow_backfill_handler.rs b/crates/bootstrap/src/follow_backfill_handler.rs index 5cce039b..96f91c2f 100644 --- a/crates/bootstrap/src/follow_backfill_handler.rs +++ b/crates/bootstrap/src/follow_backfill_handler.rs @@ -54,7 +54,6 @@ impl EventHandler for FollowBackfillHandler { #[cfg(test)] mod tests { - use std::collections::HashSet; use std::sync::Mutex; use domain::models::FollowedActorInfo; @@ -63,16 +62,16 @@ mod tests { 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. + /// The handler has no state of its own: idempotency lives behind the port + /// (remote imports are keyed upserts; backfill Creates use the deterministic + /// k-ap id `activities/create/{uuidv5(object id)}`, see activitypub + /// `tests/activity_ids.rs`). What this proves is that a redelivery replays + /// exactly the same port calls, with no accumulated or derived state. #[derive(Default)] - struct KeyedBackfill { - imported: Mutex>, - sent: Mutex>, - } + struct CallLog(Mutex>); #[async_trait] - impl ApBackfillPort for KeyedBackfill { + impl ApBackfillPort for CallLog { async fn get_following(&self, _: Uuid) -> Result, DomainError> { Ok(vec![FollowedActorInfo { url: "https://remote.example/users/a".into(), @@ -80,10 +79,10 @@ mod tests { }]) } async fn import_remote_outbox(&self, outbox: &str, actor: &str) -> Result<(), DomainError> { - self.imported + self.0 .lock() .unwrap() - .insert((outbox.into(), actor.into())); + .push(format!("import {outbox} {actor}")); Ok(()) } async fn run_backfill_for_follower( @@ -91,35 +90,42 @@ mod tests { owner: Uuid, inbox: String, ) -> Result<(), DomainError> { - self.sent.lock().unwrap().insert((owner, inbox)); + self.0 + .lock() + .unwrap() + .push(format!("backfill {owner} {inbox}")); Ok(()) } } #[tokio::test] - async fn follow_backfill_is_idempotent_when_run_twice() { - let port = Arc::new(KeyedBackfill::default()); + async fn follow_backfill_replays_identical_port_calls_when_run_twice() { + let port = Arc::new(CallLog::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 events = [ + 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(), - }; + DomainEvent::BackfillFollower { + owner_user_id: owner, + follower_inbox_url: "https://remote.example/inbox".into(), + }, + ]; - for event in [&accepted, &backfill] { + for event in &events { + let start = port.0.lock().unwrap().len(); handler.handle(event).await.unwrap(); + let mid = port.0.lock().unwrap().len(); handler.handle(event).await.unwrap(); + let log = port.0.lock().unwrap().clone(); + assert!(mid > start, "the event reaches the port"); + assert_eq!(log[start..mid], log[mid..]); } - - assert_eq!(port.imported.lock().unwrap().len(), 1); - assert_eq!(port.sent.lock().unwrap().len(), 1); } } diff --git a/crates/domain/src/ports/images.rs b/crates/domain/src/ports/images.rs index a16fe31b..be06f48a 100644 --- a/crates/domain/src/ports/images.rs +++ b/crates/domain/src/ports/images.rs @@ -11,6 +11,8 @@ pub trait ObjectStorage: Send + Sync { key: &str, ) -> Result>, DomainError>; async fn delete(&self, key: &str) -> Result<(), DomainError>; + /// Whether exactly `key` exists. Unlike `get`, no extension fallback. + async fn exists(&self, key: &str) -> Result; } #[async_trait] diff --git a/crates/domain/src/testing/fakes/noops.rs b/crates/domain/src/testing/fakes/noops.rs index 6d6d5aaf..341385d4 100644 --- a/crates/domain/src/testing/fakes/noops.rs +++ b/crates/domain/src/testing/fakes/noops.rs @@ -61,6 +61,10 @@ impl ObjectStorage for NoopObjectStorage { async fn delete(&self, _key: &str) -> Result<(), DomainError> { Ok(()) } + + async fn exists(&self, _key: &str) -> Result { + Ok(true) + } } // Re-export production noop types so test code that imports from diff --git a/crates/domain/src/testing/fakes/panics.rs b/crates/domain/src/testing/fakes/panics.rs index 54c43d44..d2f754a2 100644 --- a/crates/domain/src/testing/fakes/panics.rs +++ b/crates/domain/src/testing/fakes/panics.rs @@ -609,6 +609,9 @@ impl ObjectStorage for PanicObjectStorage { async fn delete(&self, _: &str) -> Result<(), DomainError> { panic!("PanicObjectStorage called") } + async fn exists(&self, _: &str) -> Result { + panic!("PanicObjectStorage called") + } } pub struct PanicAuthService; diff --git a/crates/domain/src/testing/fakes/store_movie.rs b/crates/domain/src/testing/fakes/store_movie.rs index f260d632..139cdae0 100644 --- a/crates/domain/src/testing/fakes/store_movie.rs +++ b/crates/domain/src/testing/fakes/store_movie.rs @@ -152,14 +152,53 @@ impl MovieQuery for InMemoryStore { async fn list_movies( &self, - _page: &PageParams, - _filter: &MovieFilter, + page: &PageParams, + filter: &MovieFilter, ) -> Result, DomainError> { + // Mirrors the sqlite adapter: filter, ORDER BY title ASC, then page. + let movies = self.movies.read().unwrap(); + let profiles = self.movie_profiles.read().unwrap(); + let search = filter.search.as_deref().map(str::to_lowercase); + let mut rows: Vec = movies + .values() + .filter_map(|m| { + let profile = profiles.get(&m.id().value()); + let genres: Vec = profile + .map(|p| p.genres.iter().map(|g| g.name.clone()).collect()) + .unwrap_or_default(); + let language = profile.and_then(|p| p.original_language.clone()); + let keep = search + .as_ref() + .is_none_or(|q| m.title().value().to_lowercase().contains(q)) + && filter + .language + .as_ref() + .is_none_or(|l| language.as_ref() == Some(l)) + && filter.genre.as_ref().is_none_or(|g| { + genres.iter().any(|n| n.to_lowercase() == g.to_lowercase()) + }); + keep.then(|| MovieSummary { + movie: m.clone(), + genres, + runtime_minutes: profile.and_then(|p| p.runtime_minutes), + original_language: language, + overview: profile.and_then(|p| p.overview.clone()), + collection_name: profile.and_then(|p| p.collection_name.clone()), + }) + }) + .collect(); + rows.sort_by(|a, b| a.movie.title().value().cmp(b.movie.title().value())); + let total_count = rows.len() as u64; + let items = rows + .into_iter() + .skip(page.offset as usize) + .take(page.limit as usize) + .collect(); Ok(Paginated { - items: vec![], - total_count: 0, - limit: 10, - offset: 0, + items, + total_count, + limit: page.limit, + offset: page.offset, }) }