fix: real AP id tests, exact poster key check, exists port, list_movies fake

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"] }

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

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<Vec<WatchlistWithMovie>, DomainError> {
+        Ok(vec![])
+    }
+    async fn get_local_reviews_for_movie(
+        &self,
+        _: &MovieId,
+    ) -> Result<Vec<DiaryEntry>, DomainError> {
+        Ok(vec![])
+    }
+    async fn get_local_reviews_page(
+        &self,
+        _: &UserId,
+        _: Option<chrono::NaiveDateTime>,
+        _: usize,
+    ) -> Result<Vec<DiaryEntry>, DomainError> {
+        Ok(vec![])
+    }
+}
+
+struct Capture(mpsc::UnboundedSender<serde_json::Value>);
+
+#[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<serde_json::Value>,
+    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);
+}

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

@@ -48,14 +48,10 @@ 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 {
+        if !self.storage.exists(&key).await? {
-            Ok(_) => {}
+            tracing::warn!(%key, "original image already gone, treating conversion as done");
-            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/object-storage/src/lib.rs

@@ -82,6 +82,14 @@ impl ObjectStorage for ObjectStorageAdapter {
         Ok(Box::pin(stream))
     }
 
+    async fn exists(&self, key: &str) -> Result<bool, DomainError> {
+        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 {

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");
+        // The movie row is written with the event, so a missing row means it was
-        deps.movie_profile.upsert(&cmd.profile).await?;
+        // 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(());
     };
 

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
+                // The poster is stored under the movie id (the converter later
-                // updated atomically with `PosterSynced`; a poster already keyed
+                // appends `.webp`/`.avif`) and the movie row is updated
-                // by this movie means an earlier delivery completed.
+                // 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(());

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<String, DomainError> {
+        Ok(key.to_string())
+    }
+    async fn get(&self, _: &str) -> Result<Vec<u8>, DomainError> {
+        Ok(vec![])
+    }
+    async fn get_stream(
+        &self,
+        _: &str,
+    ) -> Result<futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>, DomainError>
+    {
+        Ok(Box::pin(futures::stream::empty()))
+    }
+    async fn delete(&self, _: &str) -> Result<(), DomainError> {
+        Ok(())
+    }
+    async fn exists(&self, _: &str) -> Result<bool, DomainError> {
+        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());
+}

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
+    /// The handler has no state of its own: idempotency lives behind the port
-    /// deterministic activity ids, so a repeat lands on the same key.
+    /// (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 {
+    struct CallLog(Mutex<Vec<String>>);
-        imported: Mutex<HashSet<(String, String)>>,
-        sent: Mutex<HashSet<(Uuid, String)>>,
-    }
 
     #[async_trait]
-    impl ApBackfillPort for KeyedBackfill {
+    impl ApBackfillPort for CallLog {
         async fn get_following(&self, _: Uuid) -> Result<Vec<FollowedActorInfo>, 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() {
+    async fn follow_backfill_replays_identical_port_calls_when_run_twice() {
-        let port = Arc::new(KeyedBackfill::default());
+        let port = Arc::new(CallLog::default());
         let handler = FollowBackfillHandler {
             backfill: port.clone(),
         };
         let owner = UserId::from_uuid(Uuid::new_v4());
-        let accepted = DomainEvent::FollowAccepted {
+        let events = [
+            DomainEvent::FollowAccepted {
                 owner: owner.clone(),
                 requester: SocialIdentity::Remote {
                     actor_url: "https://remote.example/users/a".into(),
                 },
-        };
+            },
-        let backfill = DomainEvent::BackfillFollower {
+            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);
     }
 }

crates/domain/src/ports/images.rs

@@ -11,6 +11,8 @@ pub trait ObjectStorage: Send + Sync {
         key: &str,
     ) -> Result<futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>, 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<bool, DomainError>;
 }
 
 #[async_trait]

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<bool, DomainError> {
+        Ok(true)
+    }
 }
 
 // Re-export production noop types so test code that imports from

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<bool, DomainError> {
+        panic!("PanicObjectStorage called")
+    }
 }
 
 pub struct PanicAuthService;

crates/domain/src/testing/fakes/store_movie.rs

@@ -152,14 +152,53 @@ impl MovieQuery for InMemoryStore {
 
     async fn list_movies(
         &self,
-        _page: &PageParams,
+        page: &PageParams,
-        _filter: &MovieFilter,
+        filter: &MovieFilter,
     ) -> Result<Paginated<MovieSummary>, 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<MovieSummary> = movies
+            .values()
+            .filter_map(|m| {
+                let profile = profiles.get(&m.id().value());
+                let genres: Vec<String> = 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![],
+            items,
-            total_count: 0,
+            total_count,
-            limit: 10,
+            limit: page.limit,
-            offset: 0,
+            offset: page.offset,
         })
     }