feat: atomic wrapup, movie, integration writes

Ride writes: WrapUpRequested/Completed (wrapup create/set_complete),
MovieDiscovered (resolve upsert, new+ext id only), PosterSynced (poster
handler upsert), MovieDeleted (delete_movie), WatchEventIngested (save).
Deps drop event_publisher: wrapup, ingest, delete_review, watchlist add,
review logger.

Remaining .publish( sites:
announce-only: users/delete_account UserDeleted, person/get +
get_credits PersonEnrichmentRequested, jobs/enrichment_staleness,
http-axum search, movies/sync_poster + poster_sync_handler +
image-converter backfill + users/update_profile ImageStored (object
storage), activitypub federation_event_bridge (reacting to events).
rides write, deferred: social/execute (see report),
activitypub review_handler MovieEnrichmentRequested (after
save_remote_review).

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

@@ -393,7 +393,11 @@ pub struct PanicWatchEventCommand;
 
 #[async_trait]
 impl crate::ports::WatchEventCommand for PanicWatchEventCommand {
-    async fn save(&self, _: &crate::models::WatchEvent) -> Result<(), DomainError> {
+    async fn save(
+        &self,
+        _: &crate::models::WatchEvent,
+        _: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         panic!("PanicWatchEventCommand called")
     }
     async fn update_status(
@@ -487,10 +491,10 @@ pub struct PanicMovieCommand;
 
 #[async_trait]
 impl MovieCommand for PanicMovieCommand {
-    async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> {
+    async fn upsert_movie(&self, _: &Movie, _: &[DomainEvent]) -> Result<(), DomainError> {
         panic!("PanicMovieCommand called")
     }
-    async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> {
+    async fn delete_movie(&self, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> {
         panic!("PanicMovieCommand called")
     }
 }

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

@@ -1,19 +1,21 @@
 //! Fakes whose every call fails with an infrastructure error, for failure-path tests.
 
 use async_trait::async_trait;
+use chrono::NaiveDateTime;
 
 use crate::{
     errors::DomainError,
     events::DomainEvent,
     models::{
-        Goal, Movie, ProfileField, Review, WatchlistEntry, WatchlistWithMovie,
+        Goal, Movie, ProfileField, Review, WatchEvent, WatchEventStatus, WatchlistEntry,
+        WatchlistWithMovie,
         collections::{PageParams, Paginated},
     },
     ports::{
         EventPublisher, GoalCommand, MovieCommand, ReviewRepository, UserProfileFieldsRepository,
-        WatchlistRepository,
+        WatchEventCommand, WatchlistRepository,
     },
-    value_objects::{GoalId, MovieId, ReviewId, UserId},
+    value_objects::{GoalId, MovieId, ReviewId, UserId, WatchEventId},
 };
 
 fn refused() -> DomainError {
@@ -45,10 +47,10 @@ pub struct RefusingMovieCommand;
 
 #[async_trait]
 impl MovieCommand for RefusingMovieCommand {
-    async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> {
+    async fn upsert_movie(&self, _: &Movie, _: &[DomainEvent]) -> Result<(), DomainError> {
         Err(refused())
     }
-    async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> {
+    async fn delete_movie(&self, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> {
         Err(refused())
     }
 }
@@ -123,3 +125,29 @@ impl EventPublisher for RefusingEventPublisher {
         Err(refused())
     }
 }
+
+pub struct RefusingWatchEventCommand;
+
+#[async_trait]
+impl WatchEventCommand for RefusingWatchEventCommand {
+    async fn save(&self, _: &WatchEvent, _: &[DomainEvent]) -> Result<(), DomainError> {
+        Err(refused())
+    }
+    async fn update_status(
+        &self,
+        _: &WatchEventId,
+        _: WatchEventStatus,
+    ) -> Result<(), DomainError> {
+        Err(refused())
+    }
+    async fn update_status_batch(
+        &self,
+        _: &[WatchEventId],
+        _: WatchEventStatus,
+    ) -> Result<u64, DomainError> {
+        Err(refused())
+    }
+    async fn delete_non_pending_older_than(&self, _: NaiveDateTime) -> Result<u64, DomainError> {
+        Err(refused())
+    }
+}

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

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 
 use crate::{
     errors::DomainError,
+    events::DomainEvent,
     models::{
         Movie, MovieFilter, MovieProfile, MovieSummary, PersistedWatchEvent, WatchEvent,
         collections::{PageParams, Paginated},
@@ -17,7 +18,7 @@ impl MovieCommand for InMemoryStore {
     /// `movies.external_metadata_id` is UNIQUE: another movie row holding the
     /// same external id is a constraint violation, which the sqlite adapter
     /// surfaces as an infrastructure error.
-    async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> {
+    async fn upsert_movie(&self, movie: &Movie, events: &[DomainEvent]) -> Result<(), DomainError> {
         let mut movies = self.movies.write().unwrap();
         if let Some(external) = movie.external_metadata_id() {
             let taken = movies.values().any(|m| {
@@ -31,13 +32,19 @@ impl MovieCommand for InMemoryStore {
             }
         }
         movies.insert(movie.id().value(), movie.clone());
+        drop(movies);
+        self.record_events(events);
         Ok(())
     }
 
     /// Schema: `reviews.movie_id REFERENCES movies(id)` has no cascade, so a
     /// movie with reviews cannot be deleted. `watchlist_entries` and the
     /// `movie_*` profile tables cascade; `watch_events.movie_id` is set to NULL.
-    async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> {
+    async fn delete_movie(
+        &self,
+        movie_id: &MovieId,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         let id = movie_id.value();
         if !self.movies.read().unwrap().contains_key(&id) {
             return Ok(());
@@ -59,12 +66,14 @@ impl MovieCommand for InMemoryStore {
             .unwrap()
             .retain(|(_, movie), _| *movie != id);
         self.movie_profiles.write().unwrap().remove(&id);
-        let mut events = self.watch_events.write().unwrap();
-        for event in events.iter_mut() {
+        let mut events_guard = self.watch_events.write().unwrap();
+        for event in events_guard.iter_mut() {
             if event.movie_id().map(|m| m.value()) == Some(id) {
                 *event = detached_from_movie(event);
             }
         }
+        drop(events_guard);
+        self.record_events(events);
         Ok(())
     }
 }

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

@@ -4,6 +4,7 @@ use uuid::Uuid;
 
 use crate::{
     errors::DomainError,
+    events::DomainEvent,
     models::{WatchEvent, WatchEventStatus, WebhookToken},
     ports::{WatchEventCommand, WatchEventQuery, WebhookTokenRepository},
     value_objects::{UserId, WatchEventId, WebhookTokenId},
@@ -13,8 +14,9 @@ use super::InMemoryStore;
 
 #[async_trait]
 impl WatchEventCommand for InMemoryStore {
-    async fn save(&self, event: &WatchEvent) -> Result<(), DomainError> {
+    async fn save(&self, event: &WatchEvent, events: &[DomainEvent]) -> Result<(), DomainError> {
         self.watch_events.write().unwrap().push(event.clone());
+        self.record_events(events);
         Ok(())
     }
 

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

@@ -3,7 +3,9 @@ use std::sync::{Arc, Mutex};
 use async_trait::async_trait;
 use uuid::Uuid;
 
-use crate::{errors::DomainError, ports::WrapUpRepository, value_objects::WrapUpId};
+use crate::{
+    errors::DomainError, events::DomainEvent, ports::WrapUpRepository, value_objects::WrapUpId,
+};
 
 // ── PanicWrapUpStatsQuery ───────────────────────────────────────────────────
 
@@ -68,14 +70,22 @@ impl crate::ports::WrapUpStatsQuery for InMemoryWrapUpStatsQuery {
 
 pub struct InMemoryWrapUpRepository {
     pub store: Mutex<Vec<crate::models::wrapup::WrapUpRecord>>,
+    recorded: Mutex<Vec<DomainEvent>>,
 }
 
 impl InMemoryWrapUpRepository {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
             store: Mutex::new(Vec::new()),
+            recorded: Mutex::new(Vec::new()),
         })
     }
+
+    /// Events recorded atomically with a change (the outbox rows the sqlite
+    /// adapter writes in the same transaction).
+    pub fn recorded_events(&self) -> Vec<DomainEvent> {
+        self.recorded.lock().unwrap().clone()
+    }
 }
 
 #[async_trait]
@@ -83,8 +93,10 @@ impl WrapUpRepository for InMemoryWrapUpRepository {
     async fn create(
         &self,
         record: &crate::models::wrapup::WrapUpRecord,
+        events: &[DomainEvent],
     ) -> Result<(), DomainError> {
         self.store.lock().unwrap().push(record.clone());
+        self.recorded.lock().unwrap().extend_from_slice(events);
         Ok(())
     }
 
@@ -108,12 +120,15 @@ impl WrapUpRepository for InMemoryWrapUpRepository {
         &self,
         id: &WrapUpId,
         report: &crate::models::wrapup::WrapUpReport,
+        events: &[DomainEvent],
     ) -> Result<(), DomainError> {
         let mut store = self.store.lock().unwrap();
         if let Some(rec) = store.iter_mut().find(|r| r.id == *id) {
             rec.status = crate::models::wrapup::WrapUpStatus::Ready;
             rec.report = Some(report.clone());
             rec.completed_at = Some(chrono::Utc::now().naive_utc());
+            drop(store);
+            self.recorded.lock().unwrap().extend_from_slice(events);
             Ok(())
         } else {
             Err(DomainError::NotFound("wrapup record".into()))
@@ -197,7 +212,11 @@ pub struct PanicWrapUpRepository;
 
 #[async_trait]
 impl WrapUpRepository for PanicWrapUpRepository {
-    async fn create(&self, _: &crate::models::wrapup::WrapUpRecord) -> Result<(), DomainError> {
+    async fn create(
+        &self,
+        _: &crate::models::wrapup::WrapUpRecord,
+        _: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         panic!("PanicWrapUpRepository called")
     }
     async fn update_status(
@@ -212,6 +231,7 @@ impl WrapUpRepository for PanicWrapUpRepository {
         &self,
         _: &WrapUpId,
         _: &crate::models::wrapup::WrapUpReport,
+        _: &[DomainEvent],
     ) -> Result<(), DomainError> {
         panic!("PanicWrapUpRepository called")
     }
@@ -245,3 +265,77 @@ impl WrapUpRepository for PanicWrapUpRepository {
         panic!("PanicWrapUpRepository called")
     }
 }
+
+// ── RefusingWrapUpWrites ───────────────────────────────────────────────────
+
+/// Delegates to an in-memory repository but refuses the two writes that carry
+/// outbox events (`create`, `set_complete`), leaving `update_status` and reads
+/// working so a refusal can be observed mid-flow.
+pub struct RefusingWrapUpWrites {
+    pub inner: Arc<InMemoryWrapUpRepository>,
+}
+
+impl RefusingWrapUpWrites {
+    pub fn new(inner: Arc<InMemoryWrapUpRepository>) -> Arc<Self> {
+        Arc::new(Self { inner })
+    }
+}
+
+#[async_trait]
+impl WrapUpRepository for RefusingWrapUpWrites {
+    async fn create(
+        &self,
+        _: &crate::models::wrapup::WrapUpRecord,
+        _: &[DomainEvent],
+    ) -> Result<(), DomainError> {
+        Err(DomainError::InfrastructureError("refused".into()))
+    }
+    async fn update_status(
+        &self,
+        id: &WrapUpId,
+        status: &crate::models::wrapup::WrapUpStatus,
+        error: Option<&str>,
+    ) -> Result<(), DomainError> {
+        self.inner.update_status(id, status, error).await
+    }
+    async fn set_complete(
+        &self,
+        _: &WrapUpId,
+        _: &crate::models::wrapup::WrapUpReport,
+        _: &[DomainEvent],
+    ) -> Result<(), DomainError> {
+        Err(DomainError::InfrastructureError("refused".into()))
+    }
+    async fn get_by_id(
+        &self,
+        id: &WrapUpId,
+    ) -> Result<Option<crate::models::wrapup::WrapUpRecord>, DomainError> {
+        self.inner.get_by_id(id).await
+    }
+    async fn list_for_user(
+        &self,
+        user_id: Uuid,
+    ) -> Result<Vec<crate::models::wrapup::WrapUpRecord>, DomainError> {
+        self.inner.list_for_user(user_id).await
+    }
+    async fn list_global(&self) -> Result<Vec<crate::models::wrapup::WrapUpRecord>, DomainError> {
+        self.inner.list_global().await
+    }
+    async fn find_existing(
+        &self,
+        user_id: Option<Uuid>,
+        start: chrono::NaiveDate,
+        end: chrono::NaiveDate,
+    ) -> Result<Option<crate::models::wrapup::WrapUpRecord>, DomainError> {
+        self.inner.find_existing(user_id, start, end).await
+    }
+    async fn delete(&self, id: &WrapUpId) -> Result<(), DomainError> {
+        self.inner.delete(id).await
+    }
+    async fn delete_failed_older_than(
+        &self,
+        before: chrono::NaiveDateTime,
+    ) -> Result<u64, DomainError> {
+        self.inner.delete_failed_older_than(before).await
+    }
+}