feat: atomic review writes with events

crates/adapters/sqlite/src/repositories/review/mod.rs

@@ -1,6 +1,7 @@
 use async_trait::async_trait;
 use domain::{
     errors::DomainError,
+    events::DomainEvent,
     models::{Review, ReviewSource},
     ports::ReviewRepository,
     value_objects::{ReviewId, UserId},
@@ -10,7 +11,7 @@ use sqlx::SqlitePool;
 use rows::ReviewRow;
 
 pub(crate) mod rows;
-use crate::shared::datetime_to_str;
+use crate::shared::{datetime_to_str, enqueue_events};
 
 pub struct SqliteReviewRepository {
     pool: SqlitePool,
@@ -24,7 +25,11 @@ impl SqliteReviewRepository {
 
 #[async_trait]
 impl ReviewRepository for SqliteReviewRepository {
-    async fn save_review(&self, review: &Review) -> Result<(), DomainError> {
+    async fn save_review(
+        &self,
+        review: &Review,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         let id = review.id().value().to_string();
         let movie_id = review.movie_id().value().to_string();
         let user_id = review.user_id().value().to_string();
@@ -37,6 +42,11 @@ impl ReviewRepository for SqliteReviewRepository {
             ReviewSource::Remote { actor_url } => Some(actor_url.clone()),
         };
 
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .map_err(crate::shared::map_sqlx_error)?;
         sqlx::query(
             "INSERT INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url, watch_medium, spoilers)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
@@ -51,11 +61,11 @@ impl ReviewRepository for SqliteReviewRepository {
         .bind(&remote_actor_url)
         .bind(review.watch_medium().map(|wm| wm.to_string()))
         .bind(review.spoilers())
-        .execute(&self.pool)
+        .execute(&mut *tx)
         .await
         .map_err(crate::shared::map_sqlx_error)?;
-
+        enqueue_events(&mut tx, events).await?;
-        Ok(())
+        tx.commit().await.map_err(crate::shared::map_sqlx_error)
     }
 
     async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError> {
@@ -72,13 +82,22 @@ impl ReviewRepository for SqliteReviewRepository {
         .transpose()
     }
 
-    async fn update_review(&self, review: &Review) -> Result<(), DomainError> {
+    async fn update_review(
+        &self,
+        review: &Review,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         let id = review.id().value().to_string();
         let rating = review.rating().value() as i64;
         let comment = review.comment().map(|c| c.value().to_string());
         let watched_at = datetime_to_str(review.watched_at());
         let watch_medium = review.watch_medium().map(|wm| wm.to_string());
 
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .map_err(crate::shared::map_sqlx_error)?;
         sqlx::query(
             "UPDATE reviews SET rating = ?, comment = ?, watched_at = ?, watch_medium = ?, spoilers = ? WHERE id = ?",
         )
@@ -88,21 +107,31 @@ impl ReviewRepository for SqliteReviewRepository {
         .bind(&watch_medium)
         .bind(review.spoilers())
         .bind(&id)
-        .execute(&self.pool)
+        .execute(&mut *tx)
         .await
         .map_err(crate::shared::map_sqlx_error)?;
-
+        enqueue_events(&mut tx, events).await?;
-        Ok(())
+        tx.commit().await.map_err(crate::shared::map_sqlx_error)
     }
 
-    async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError> {
+    async fn delete_review(
+        &self,
+        review_id: &ReviewId,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         let id = review_id.value().to_string();
-        sqlx::query("DELETE FROM reviews WHERE id = ?")
+        let mut tx = self
-            .bind(&id)
+            .pool
-            .execute(&self.pool)
+            .begin()
             .await
             .map_err(crate::shared::map_sqlx_error)?;
-        Ok(())
+        sqlx::query("DELETE FROM reviews WHERE id = ?")
+            .bind(&id)
+            .execute(&mut *tx)
+            .await
+            .map_err(crate::shared::map_sqlx_error)?;
+        enqueue_events(&mut tx, events).await?;
+        tx.commit().await.map_err(crate::shared::map_sqlx_error)
     }
 
     async fn get_all_reviews_for_user(&self, user_id: &UserId) -> Result<Vec<Review>, DomainError> {

crates/adapters/sqlite/tests/cascade_test.rs

@@ -199,7 +199,7 @@ async fn seed_user_with_dependents(
 
     parts
         .review
-        .save_review(&test_review(movie.id(), &user_id, 4))
+        .save_review(&test_review(movie.id(), &user_id, 4), &[])
         .await
         .unwrap();
     (user_id, movie)
@@ -302,7 +302,7 @@ async fn deleting_a_movie_that_has_reviews_is_refused_because_reviews_do_not_cas
     parts.movie_command.upsert_movie(&movie).await.unwrap();
     parts
         .review
-        .save_review(&test_review(movie.id(), user.id(), 5))
+        .save_review(&test_review(movie.id(), user.id(), 5), &[])
         .await
         .unwrap();
 

crates/adapters/sqlite/tests/repositories.rs

@@ -8,6 +8,8 @@ mod movie_dedup_test;
 mod movie_profile_repository_test;
 #[path = "repositories/person_adapter_test.rs"]
 mod person_adapter_test;
+#[path = "repositories/review_repository_test.rs"]
+mod review_repository_test;
 #[path = "repositories/search_adapter_test.rs"]
 mod search_adapter_test;
 #[path = "repositories/social_repository_test.rs"]

crates/adapters/sqlite/tests/repositories/review_repository_test.rs

@@ -0,0 +1,82 @@
+use std::str::FromStr;
+
+use domain::{
+    events::DomainEvent,
+    ports::{MovieCommand, ReviewRepository, UserRepository},
+    testing::{test_movie, test_review, test_user},
+};
+use sqlite::{SqliteMovieRepository, SqliteReviewRepository, SqliteUserRepository};
+use sqlx::{
+    SqlitePool,
+    sqlite::{SqliteConnectOptions, SqlitePoolOptions},
+};
+
+async fn migrated_pool() -> SqlitePool {
+    let opts = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
+    let pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .connect_with(opts)
+        .await
+        .unwrap();
+    sqlite::migrate(&pool).await.unwrap();
+    pool
+}
+
+async fn count(pool: &SqlitePool, table: &str) -> i64 {
+    sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}"))
+        .fetch_one(pool)
+        .await
+        .unwrap()
+}
+
+fn logged(review: &domain::models::Review) -> DomainEvent {
+    DomainEvent::ReviewLogged {
+        review_id: review.id().clone(),
+        movie_id: review.movie_id().clone(),
+        user_id: review.user_id().clone(),
+        rating: review.rating().clone(),
+        watched_at: *review.watched_at(),
+    }
+}
+
+#[tokio::test]
+async fn saving_a_review_writes_the_review_and_its_event_in_one_commit() {
+    let pool = migrated_pool().await;
+    let movie = test_movie("Heat");
+    let user = test_user("alice");
+    SqliteUserRepository::new(pool.clone())
+        .save(&user)
+        .await
+        .unwrap();
+    SqliteMovieRepository::new(pool.clone())
+        .upsert_movie(&movie)
+        .await
+        .unwrap();
+    let review = test_review(movie.id(), user.id(), 4);
+
+    SqliteReviewRepository::new(pool.clone())
+        .save_review(&review, &[logged(&review)])
+        .await
+        .unwrap();
+
+    assert_eq!(count(&pool, "reviews").await, 1);
+    assert_eq!(count(&pool, "event_queue").await, 1);
+}
+
+#[tokio::test]
+async fn a_failed_review_save_leaves_no_event_row() {
+    let pool = migrated_pool().await;
+    let review = test_review(
+        &domain::value_objects::MovieId::generate(),
+        &domain::value_objects::UserId::generate(),
+        4,
+    );
+
+    let result = SqliteReviewRepository::new(pool.clone())
+        .save_review(&review, &[logged(&review)])
+        .await;
+
+    assert!(result.is_err(), "FK violation must fail the save");
+    assert_eq!(count(&pool, "reviews").await, 0);
+    assert_eq!(count(&pool, "event_queue").await, 0);
+}

crates/application/src/diary/delete_review.rs

@@ -20,18 +20,15 @@ pub async fn execute(deps: &DeleteReviewDeps, cmd: DeleteReviewCommand) -> Resul
     }
 
     let movie_id = review.movie_id().clone();
-    deps.review.delete_review(&review_id).await?;
+    deps.review
-
+        .delete_review(
-    if let Err(e) = deps
+            &review_id,
-        .event_publisher
+            &[DomainEvent::ReviewDeleted {
-        .publish(&DomainEvent::ReviewDeleted {
+                review_id: review_id.clone(),
-            review_id: review_id.clone(),
+                user_id: requesting_user_id.clone(),
-            user_id: requesting_user_id.clone(),
+            }],
-        })
+        )
-        .await
+        .await?;
-    {
-        tracing::warn!("failed to publish ReviewDeleted: {e}");
-    }
 
     let history = deps.diary.get_review_history(&movie_id).await?;
     if history.viewings().is_empty() {

crates/application/src/diary/deps.rs

@@ -16,7 +16,6 @@ pub struct DeleteReviewDeps {
 
 pub struct EditReviewDeps {
     pub review: Arc<dyn ReviewRepository>,
-    pub event_publisher: Arc<dyn EventPublisher>,
 }
 
 pub struct GetMovieSocialPageDeps {

crates/application/src/diary/edit_review.rs

@@ -37,21 +37,12 @@ pub async fn execute(deps: &EditReviewDeps, cmd: EditReviewCommand) -> Result<()
         spoilers: cmd.spoilers,
     });
 
-    deps.review.update_review(&updated).await?;
+    let event = DomainEvent::ReviewUpdated {
-
+        review_id: updated.id().clone(),
-    if let Err(e) = deps
+        movie_id: updated.movie_id().clone(),
-        .event_publisher
+        user_id: updated.user_id().clone(),
-        .publish(&DomainEvent::ReviewUpdated {
+        rating: updated.rating().clone(),
-            review_id: updated.id().clone(),
+        watched_at: *updated.watched_at(),
-            movie_id: updated.movie_id().clone(),
+    };
-            user_id: updated.user_id().clone(),
+    deps.review.update_review(&updated, &[event]).await
-            rating: updated.rating().clone(),
-            watched_at: *updated.watched_at(),
-        })
-        .await
-    {
-        tracing::warn!("failed to publish ReviewUpdated: {e}");
-    }
-
-    Ok(())
 }

crates/application/src/diary/review_logger.rs

@@ -75,7 +75,6 @@ impl ReviewLogger for DefaultReviewLogger {
             cmd.watch_medium,
             cmd.spoilers,
         )?;
-        self.review_repo.save_review(&review).await?;
         let review_event = DomainEvent::ReviewLogged {
             review_id: review.id().clone(),
             movie_id: review.movie_id().clone(),
@@ -84,6 +83,17 @@ impl ReviewLogger for DefaultReviewLogger {
             watched_at: *review.watched_at(),
         };
 
+        // Events ride the review save (the decisive write) so they commit with it.
+        let mut events = Vec::with_capacity(2);
+        if let Some(ext_id) = movie.external_metadata_id() {
+            events.push(DomainEvent::MovieEnrichmentRequested {
+                movie_id: movie.id().clone(),
+                external_metadata_id: ext_id.clone(),
+            });
+        }
+        events.push(review_event);
+        self.review_repo.save_review(&review, &events).await?;
+
         let was_on_watchlist = self
             .watchlist_repo
             .remove_if_present(review.user_id(), review.movie_id())
@@ -98,15 +108,6 @@ impl ReviewLogger for DefaultReviewLogger {
                 .await;
         }
 
-        if let Some(ext_id) = movie.external_metadata_id() {
+        Ok(())
-            self.event_publisher
-                .publish(&DomainEvent::MovieEnrichmentRequested {
-                    movie_id: movie.id().clone(),
-                    external_metadata_id: ext_id.clone(),
-                })
-                .await?;
-        }
-
-        self.event_publisher.publish(&review_event).await
     }
 }

crates/application/tests/diary/delete_review_test.rs

@@ -23,7 +23,7 @@ async fn deleting_a_review_removes_it() {
     let review = test_review(movie.id(), &user_id, 4);
 
     movies.upsert_movie(&movie).await.unwrap();
-    reviews.save_review(&review).await.unwrap();
+    reviews.save_review(&review, &[]).await.unwrap();
     diary.seed_history(movie.clone(), vec![]);
 
     let deps = DeleteReviewDeps {
@@ -62,7 +62,7 @@ async fn deleting_another_users_review_is_unauthorized() {
     let other_id = uuid::Uuid::new_v4();
     let review = test_review(&movie_id, &owner_id, 4);
 
-    reviews.save_review(&review).await.unwrap();
+    reviews.save_review(&review, &[]).await.unwrap();
 
     let deps = DeleteReviewDeps {
         review: Arc::clone(&reviews) as _,
@@ -83,3 +83,71 @@ async fn deleting_another_users_review_is_unauthorized() {
     assert!(result.is_err(), "wrong user should not be able to delete");
     assert_eq!(reviews.review_count(), 1, "review should still exist");
 }
+
+#[tokio::test]
+async fn deleting_a_review_removes_it_and_records_its_event_together() {
+    let movies = InMemoryStore::new();
+    let diary = FakeDiaryQuery::new();
+    let events = NoopEventPublisher::new();
+    let movie = test_movie_with_year("Terminator", 1984);
+    let user_id = UserId::from_uuid(uuid::Uuid::new_v4());
+    let review = test_review(movie.id(), &user_id, 4);
+    movies.upsert_movie(&movie).await.unwrap();
+    movies.save_review(&review, &[]).await.unwrap();
+    diary.seed_history(movie.clone(), vec![review.clone()]);
+    let deps = DeleteReviewDeps {
+        review: Arc::clone(&movies) as _,
+        diary: diary.clone() as _,
+        movie_command: Arc::clone(&movies) as _,
+        event_publisher: Arc::clone(&events) as _,
+    };
+
+    delete_review::execute(
+        &deps,
+        DeleteReviewCommand {
+            review_id: review.id().value(),
+            requesting_user_id: user_id.value(),
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(movies.review_count(), 0);
+    let recorded = movies.recorded_events();
+    assert_eq!(recorded.len(), 1);
+    assert!(matches!(
+        recorded[0],
+        domain::events::DomainEvent::ReviewDeleted { .. }
+    ));
+}
+
+#[tokio::test]
+async fn a_refused_delete_records_neither_the_removal_nor_its_event() {
+    let movies = InMemoryStore::new();
+    let diary = FakeDiaryQuery::new();
+    let events = NoopEventPublisher::new();
+    let user_id = UserId::from_uuid(uuid::Uuid::new_v4());
+    let review = test_review(&MovieId::generate(), &user_id, 4);
+    movies.save_review(&review, &[]).await.unwrap();
+    let deps = DeleteReviewDeps {
+        review: Arc::new(crate::edit_review_test::WriteRefusingReviews(Arc::clone(
+            &movies,
+        ))),
+        diary: diary as _,
+        movie_command: Arc::clone(&movies) as _,
+        event_publisher: Arc::clone(&events) as _,
+    };
+
+    let result = delete_review::execute(
+        &deps,
+        DeleteReviewCommand {
+            review_id: review.id().value(),
+            requesting_user_id: user_id.value(),
+        },
+    )
+    .await;
+
+    assert!(result.is_err());
+    assert_eq!(movies.review_count(), 1);
+    assert!(movies.recorded_events().is_empty());
+}

crates/application/tests/diary/edit_review_test.rs

@@ -4,20 +4,14 @@ use std::sync::Arc;
 use domain::{
     models::ReviewEdit,
     ports::ReviewRepository,
-    testing::{InMemoryStore, NoopEventPublisher, test_review},
+    testing::{InMemoryStore, test_review},
     value_objects::{Comment, MovieId, ReviewId, UserId, WatchMedium},
 };
 
 use application::diary::{commands::EditReviewCommand, deps::EditReviewDeps, edit_review};
 
-async fn setup() -> (
+async fn setup() -> (Arc<InMemoryStore>, ReviewId, UserId) {
-    Arc<InMemoryStore>,
-    Arc<NoopEventPublisher>,
-    ReviewId,
-    UserId,
-) {
     let reviews = InMemoryStore::new();
-    let events = NoopEventPublisher::new();
     let user_id = UserId::generate();
     let review = test_review(&MovieId::generate(), &user_id, 3).apply_edit(ReviewEdit {
         comment: Some(Some(Comment::new("original comment".into()).unwrap())),
@@ -25,23 +19,22 @@ async fn setup() -> (
         ..Default::default()
     });
     let review_id = review.id().clone();
-    reviews.save_review(&review).await.unwrap();
+    reviews.save_review(&review, &[]).await.unwrap();
-    (reviews, events, review_id, user_id)
+    (reviews, review_id, user_id)
 }
 
-fn deps(reviews: &Arc<InMemoryStore>, events: &Arc<NoopEventPublisher>) -> EditReviewDeps {
+fn deps(reviews: &Arc<InMemoryStore>) -> EditReviewDeps {
     EditReviewDeps {
         review: Arc::clone(reviews) as _,
-        event_publisher: Arc::clone(events) as _,
     }
 }
 
 #[tokio::test]
 async fn edit_own_review_updates_rating() {
-    let (reviews, events, review_id, user_id) = setup().await;
+    let (reviews, review_id, user_id) = setup().await;
 
     edit_review::execute(
-        &deps(&reviews, &events),
+        &deps(&reviews),
         EditReviewCommand {
             review_id: review_id.value(),
             requesting_user_id: user_id.value(),
@@ -60,9 +53,8 @@ async fn edit_own_review_updates_rating() {
 #[tokio::test]
 async fn edit_nonexistent_review_returns_not_found() {
     let reviews = InMemoryStore::new();
-    let events = NoopEventPublisher::new();
 
-    let err = edit_review::execute(&deps(&reviews, &events), edit_review_command(5))
+    let err = edit_review::execute(&deps(&reviews), edit_review_command(5))
         .await
         .unwrap_err();
 
@@ -71,10 +63,10 @@ async fn edit_nonexistent_review_returns_not_found() {
 
 #[tokio::test]
 async fn edit_other_users_review_returns_forbidden() {
-    let (reviews, events, review_id, _user_id) = setup().await;
+    let (reviews, review_id, _user_id) = setup().await;
 
     let err = edit_review::execute(
-        &deps(&reviews, &events),
+        &deps(&reviews),
         EditReviewCommand {
             review_id: review_id.value(),
             ..edit_review_command(5)
@@ -86,12 +78,59 @@ async fn edit_other_users_review_returns_forbidden() {
     assert!(matches!(err, domain::errors::DomainError::Forbidden(_)));
 }
 
+/// Loads from the store but refuses every write.
+pub(crate) struct WriteRefusingReviews(pub(crate) Arc<InMemoryStore>);
+
+#[async_trait::async_trait]
+impl ReviewRepository for WriteRefusingReviews {
+    async fn save_review(
+        &self,
+        _: &domain::models::Review,
+        _: &[domain::events::DomainEvent],
+    ) -> Result<(), domain::errors::DomainError> {
+        Err(domain::errors::DomainError::InfrastructureError(
+            "refused".into(),
+        ))
+    }
+    async fn get_review_by_id(
+        &self,
+        id: &ReviewId,
+    ) -> Result<Option<domain::models::Review>, domain::errors::DomainError> {
+        self.0.get_review_by_id(id).await
+    }
+    async fn update_review(
+        &self,
+        _: &domain::models::Review,
+        _: &[domain::events::DomainEvent],
+    ) -> Result<(), domain::errors::DomainError> {
+        Err(domain::errors::DomainError::InfrastructureError(
+            "refused".into(),
+        ))
+    }
+    async fn delete_review(
+        &self,
+        _: &ReviewId,
+        _: &[domain::events::DomainEvent],
+    ) -> Result<(), domain::errors::DomainError> {
+        Err(domain::errors::DomainError::InfrastructureError(
+            "refused".into(),
+        ))
+    }
+    async fn get_all_reviews_for_user(
+        &self,
+        id: &UserId,
+    ) -> Result<Vec<domain::models::Review>, domain::errors::DomainError> {
+        self.0.get_all_reviews_for_user(id).await
+    }
+}
+
 #[tokio::test]
-async fn edit_publishes_review_updated_event() {
+async fn editing_a_review_saves_the_change_and_its_event_together() {
-    let (reviews, events, review_id, user_id) = setup().await;
+    let (reviews, review_id, user_id) = setup().await;
+    assert!(reviews.recorded_events().is_empty());
 
     edit_review::execute(
-        &deps(&reviews, &events),
+        &deps(&reviews),
         EditReviewCommand {
             review_id: review_id.value(),
             requesting_user_id: user_id.value(),
@@ -101,20 +140,45 @@ async fn edit_publishes_review_updated_event() {
     .await
     .unwrap();
 
-    let published = events.published();
+    let updated = reviews.get_review_by_id(&review_id).await.unwrap().unwrap();
-    assert_eq!(published.len(), 1);
+    assert_eq!(updated.rating().value(), 1);
+    let recorded = reviews.recorded_events();
+    assert_eq!(recorded.len(), 1);
     assert!(matches!(
-        published[0],
+        recorded[0],
         domain::events::DomainEvent::ReviewUpdated { .. }
     ));
 }
 
+#[tokio::test]
+async fn a_refused_edit_records_neither_the_change_nor_its_event() {
+    let (reviews, review_id, user_id) = setup().await;
+    let deps = EditReviewDeps {
+        review: Arc::new(WriteRefusingReviews(Arc::clone(&reviews))),
+    };
+
+    let result = edit_review::execute(
+        &deps,
+        EditReviewCommand {
+            review_id: review_id.value(),
+            requesting_user_id: user_id.value(),
+            ..edit_review_command(1)
+        },
+    )
+    .await;
+
+    assert!(result.is_err());
+    let unchanged = reviews.get_review_by_id(&review_id).await.unwrap().unwrap();
+    assert_eq!(unchanged.rating().value(), 3);
+    assert!(reviews.recorded_events().is_empty());
+}
+
 #[tokio::test]
 async fn edit_sets_watch_medium() {
-    let (reviews, events, review_id, user_id) = setup().await;
+    let (reviews, review_id, user_id) = setup().await;
 
     edit_review::execute(
-        &deps(&reviews, &events),
+        &deps(&reviews),
         EditReviewCommand {
             rating: None,
             review_id: review_id.value(),
@@ -132,10 +196,9 @@ async fn edit_sets_watch_medium() {
 
 #[tokio::test]
 async fn editing_a_review_fails_when_the_store_refuses_to_load_it() {
-    let (_, events, review_id, user_id) = setup().await;
+    let (_, review_id, user_id) = setup().await;
     let deps = EditReviewDeps {
         review: Arc::new(domain::testing::RefusingReviewRepository),
-        event_publisher: Arc::clone(&events) as _,
     };
 
     let result = edit_review::execute(

crates/application/tests/diary/log_review_test.rs

@@ -39,7 +39,10 @@ async fn logging_a_review_creates_the_movie_and_the_review() {
     log_review::execute(&logger, cmd).await.unwrap();
 
     assert_eq!(reviews.review_count(), 1, "review should be saved");
-    assert!(!events.published().is_empty(), "events should be published");
+    assert!(
+        !movies.recorded_events().is_empty(),
+        "events should be recorded with the review"
+    );
 }
 
 #[tokio::test]

crates/application/tests/diary/review_logger_test.rs

@@ -212,13 +212,6 @@ async fn publishes_movie_discovered_for_new_movie_with_external_id() {
             .any(|e| matches!(e, domain::events::DomainEvent::MovieDiscovered { .. })),
         "should publish MovieDiscovered for new movie with external_metadata_id"
     );
-    assert!(
-        published.iter().any(|e| matches!(
-            e,
-            domain::events::DomainEvent::MovieEnrichmentRequested { .. }
-        )),
-        "should publish MovieEnrichmentRequested"
-    );
 }
 
 #[tokio::test]
@@ -241,8 +234,7 @@ async fn logging_a_review_fails_when_the_review_store_refuses() {
 }
 
 #[tokio::test]
-async fn logging_a_review_when_event_publishing_fails_returns_the_error_but_keeps_the_saved_review()
+async fn logging_a_review_saves_the_review_and_its_event_together() {
-{
     let movies = InMemoryStore::new();
     let logger = DefaultReviewLogger::new(
         Arc::clone(&movies) as _,
@@ -253,14 +245,70 @@ async fn logging_a_review_when_event_publishing_fails_returns_the_error_but_keep
         Arc::new(domain::testing::RefusingEventPublisher),
     );
 
+    logger
+        .log_review(log_review_command_manual(Uuid::new_v4(), "Test Film", 2024))
+        .await
+        .unwrap();
+
+    assert_eq!(movies.review_count(), 1);
+    let recorded = movies.recorded_events();
+    assert_eq!(recorded.len(), 1);
+    assert!(matches!(
+        recorded[0],
+        domain::events::DomainEvent::ReviewLogged { .. }
+    ));
+}
+
+#[tokio::test]
+async fn a_refused_save_records_neither_the_review_nor_its_event() {
+    let movies = InMemoryStore::new();
+    let logger = DefaultReviewLogger::new(
+        Arc::clone(&movies) as _,
+        Arc::clone(&movies) as _,
+        Arc::new(domain::testing::RefusingReviewRepository),
+        Arc::clone(&movies) as _,
+        Arc::new(FakeMetadataClient),
+        NoopEventPublisher::new(),
+    );
+
     let result = logger
         .log_review(log_review_command_manual(Uuid::new_v4(), "Test Film", 2024))
         .await;
 
-    assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused"));
+    assert!(result.is_err());
-    assert_eq!(
+    assert_eq!(movies.review_count(), 0);
-        movies.review_count(),
+    assert!(movies.recorded_events().is_empty());
-        1,
+}
-        "review is saved before publishing"
+
+#[tokio::test]
+async fn enrichment_request_rides_the_review_save() {
+    let movies = InMemoryStore::new();
+    let logger = DefaultReviewLogger::new(
+        Arc::clone(&movies) as _,
+        Arc::clone(&movies) as _,
+        Arc::clone(&movies) as _,
+        Arc::clone(&movies) as _,
+        Arc::new(MetadataClientWithExternalId),
+        NoopEventPublisher::new(),
+    );
+
+    logger
+        .log_review(log_review_command_manual(
+            Uuid::new_v4(),
+            "Discovered Film",
+            2024,
+        ))
+        .await
+        .unwrap();
+
+    let recorded = movies.recorded_events();
+    assert!(recorded.iter().any(|e| matches!(
+        e,
+        domain::events::DomainEvent::MovieEnrichmentRequested { .. }
+    )));
+    assert!(
+        recorded
+            .iter()
+            .any(|e| matches!(e, domain::events::DomainEvent::ReviewLogged { .. }))
     );
 }

crates/bootstrap/src/build.rs

@@ -94,7 +94,6 @@ pub fn build_deps(
             },
             edit_review: EditReviewDeps {
                 review: repos.review.clone(),
-                event_publisher: services.event_publisher.clone(),
             },
             get_movie_social_page: GetMovieSocialPageDeps {
                 movie_query: repos.movie_query.clone(),

crates/domain/src/ports/diary.rs

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 
 use crate::{
     errors::DomainError,
+    events::DomainEvent,
     models::{
         DiaryEntry, DiaryFilter, ExportFormat, FeedEntry, FeedSortBy, FollowingFilter, MovieStats,
         Review, ReviewHistory, UserStats, UserTrends,
@@ -42,10 +43,19 @@ pub trait DiaryQuery: Send + Sync {
 
 #[async_trait]
 pub trait ReviewRepository: Send + Sync {
-    async fn save_review(&self, review: &Review) -> Result<(), DomainError>;
+    async fn save_review(&self, review: &Review, events: &[DomainEvent])
+    -> Result<(), DomainError>;
     async fn get_review_by_id(&self, review_id: &ReviewId) -> Result<Option<Review>, DomainError>;
-    async fn update_review(&self, review: &Review) -> Result<(), DomainError>;
+    async fn update_review(
-    async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError>;
+        &self,
+        review: &Review,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError>;
+    async fn delete_review(
+        &self,
+        review_id: &ReviewId,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError>;
     async fn get_all_reviews_for_user(&self, user_id: &UserId) -> Result<Vec<Review>, DomainError>;
 }
 

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

@@ -538,16 +538,16 @@ pub struct PanicReviewRepository;
 
 #[async_trait]
 impl ReviewRepository for PanicReviewRepository {
-    async fn save_review(&self, _: &Review) -> Result<(), DomainError> {
+    async fn save_review(&self, _: &Review, _: &[DomainEvent]) -> Result<(), DomainError> {
         panic!("PanicReviewRepository called")
     }
     async fn get_review_by_id(&self, _: &ReviewId) -> Result<Option<Review>, DomainError> {
         panic!("PanicReviewRepository called")
     }
-    async fn update_review(&self, _: &Review) -> Result<(), DomainError> {
+    async fn update_review(&self, _: &Review, _: &[DomainEvent]) -> Result<(), DomainError> {
         panic!("PanicReviewRepository called")
     }
-    async fn delete_review(&self, _: &ReviewId) -> Result<(), DomainError> {
+    async fn delete_review(&self, _: &ReviewId, _: &[DomainEvent]) -> Result<(), DomainError> {
         panic!("PanicReviewRepository called")
     }
     async fn get_all_reviews_for_user(&self, _: &UserId) -> Result<Vec<Review>, DomainError> {

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

@@ -21,16 +21,16 @@ pub struct RefusingReviewRepository;
 
 #[async_trait]
 impl ReviewRepository for RefusingReviewRepository {
-    async fn save_review(&self, _: &Review) -> Result<(), DomainError> {
+    async fn save_review(&self, _: &Review, _: &[DomainEvent]) -> Result<(), DomainError> {
         Err(refused())
     }
     async fn get_review_by_id(&self, _: &ReviewId) -> Result<Option<Review>, DomainError> {
         Err(refused())
     }
-    async fn update_review(&self, _: &Review) -> Result<(), DomainError> {
+    async fn update_review(&self, _: &Review, _: &[DomainEvent]) -> Result<(), DomainError> {
         Err(refused())
     }
-    async fn delete_review(&self, _: &ReviewId) -> Result<(), DomainError> {
+    async fn delete_review(&self, _: &ReviewId, _: &[DomainEvent]) -> Result<(), DomainError> {
         Err(refused())
     }
     async fn get_all_reviews_for_user(&self, _: &UserId) -> Result<Vec<Review>, DomainError> {

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

@@ -72,6 +72,7 @@ pub struct InMemoryStore {
     pub(super) follows: RwLock<Vec<(Uuid, SocialIdentity, FollowState)>>,
     pub(super) blocked: RwLock<Vec<(Uuid, SocialIdentity)>>,
     pub(super) processed_events: RwLock<HashMap<(Uuid, String), LedgerRow>>,
+    pub(super) recorded_events: RwLock<Vec<crate::events::DomainEvent>>,
 }
 
 impl InMemoryStore {
@@ -94,6 +95,7 @@ impl InMemoryStore {
             follows: RwLock::default(),
             blocked: RwLock::default(),
             processed_events: RwLock::default(),
+            recorded_events: RwLock::default(),
         })
     }
 
@@ -103,6 +105,19 @@ impl InMemoryStore {
         store
     }
 
+    /// Events recorded atomically with a change (the outbox rows the sqlite
+    /// adapter would have written in the same transaction), in write order.
+    pub fn recorded_events(&self) -> Vec<crate::events::DomainEvent> {
+        self.recorded_events.read().unwrap().clone()
+    }
+
+    pub(super) fn record_events(&self, events: &[crate::events::DomainEvent]) {
+        self.recorded_events
+            .write()
+            .unwrap()
+            .extend_from_slice(events);
+    }
+
     pub fn movie_count(&self) -> usize {
         self.movies.read().unwrap().len()
     }

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

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 
 use crate::{
     errors::DomainError,
+    events::DomainEvent,
     models::Review,
     ports::ReviewRepository,
     value_objects::{ReviewId, UserId},
@@ -11,11 +12,16 @@ use super::InMemoryStore;
 
 #[async_trait]
 impl ReviewRepository for InMemoryStore {
-    async fn save_review(&self, review: &Review) -> Result<(), DomainError> {
+    async fn save_review(
+        &self,
+        review: &Review,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         self.reviews
             .write()
             .unwrap()
             .insert(review.id().value(), review.clone());
+        self.record_events(events);
         Ok(())
     }
 
@@ -28,16 +34,26 @@ impl ReviewRepository for InMemoryStore {
             .cloned())
     }
 
-    async fn update_review(&self, review: &Review) -> Result<(), DomainError> {
+    async fn update_review(
+        &self,
+        review: &Review,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         self.reviews
             .write()
             .unwrap()
             .insert(review.id().value(), review.clone());
+        self.record_events(events);
         Ok(())
     }
 
-    async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError> {
+    async fn delete_review(
+        &self,
+        review_id: &ReviewId,
+        events: &[DomainEvent],
+    ) -> Result<(), DomainError> {
         self.reviews.write().unwrap().remove(&review_id.value());
+        self.record_events(events);
         Ok(())
     }
 

crates/domain/tests/testing/in_memory_store_test.rs

@@ -144,7 +144,7 @@ async fn seed_user_with_dependents(store: &Arc<InMemoryStore>) -> (UserId, Movie
         .unwrap();
 
     store
-        .save_review(&test_review(movie.id(), &user_id, 4))
+        .save_review(&test_review(movie.id(), &user_id, 4), &[])
         .await
         .unwrap();
     store
@@ -239,7 +239,7 @@ async fn seed_movie_with_dependents(
     .unwrap();
     if with_review {
         store
-            .save_review(&test_review(movie.id(), user.id(), 5))
+            .save_review(&test_review(movie.id(), user.id(), 5), &[])
             .await
             .unwrap();
     }