From c8a1e8bfe4b147c16b2bed8552d744aec1779f34 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 00:40:43 +0100 Subject: [PATCH] feat: atomic review writes with events --- .../sqlite/src/repositories/review/mod.rs | 57 ++++++--- crates/adapters/sqlite/tests/cascade_test.rs | 4 +- crates/adapters/sqlite/tests/repositories.rs | 2 + .../repositories/review_repository_test.rs | 82 ++++++++++++ crates/application/src/diary/delete_review.rs | 21 ++-- crates/application/src/diary/deps.rs | 1 - crates/application/src/diary/edit_review.rs | 25 ++-- crates/application/src/diary/review_logger.rs | 23 ++-- .../tests/diary/delete_review_test.rs | 72 ++++++++++- .../tests/diary/edit_review_test.rs | 119 +++++++++++++----- .../tests/diary/log_review_test.rs | 5 +- .../tests/diary/review_logger_test.rs | 76 ++++++++--- crates/bootstrap/src/build.rs | 1 - crates/domain/src/ports/diary.rs | 16 ++- crates/domain/src/testing/fakes/panics.rs | 6 +- crates/domain/src/testing/fakes/refusing.rs | 6 +- crates/domain/src/testing/fakes/store.rs | 15 +++ .../domain/src/testing/fakes/store_review.rs | 22 +++- .../tests/testing/in_memory_store_test.rs | 4 +- 19 files changed, 440 insertions(+), 117 deletions(-) create mode 100644 crates/adapters/sqlite/tests/repositories/review_repository_test.rs diff --git a/crates/adapters/sqlite/src/repositories/review/mod.rs b/crates/adapters/sqlite/src/repositories/review/mod.rs index ca1f8e69..948ed04c 100644 --- a/crates/adapters/sqlite/src/repositories/review/mod.rs +++ b/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)?; - - Ok(()) + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } async fn get_review_by_id(&self, review_id: &ReviewId) -> Result, 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)?; - - Ok(()) + enqueue_events(&mut tx, events).await?; + 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 = ?") - .bind(&id) - .execute(&self.pool) + let mut tx = 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, DomainError> { diff --git a/crates/adapters/sqlite/tests/cascade_test.rs b/crates/adapters/sqlite/tests/cascade_test.rs index 44c7d2e5..7a07b892 100644 --- a/crates/adapters/sqlite/tests/cascade_test.rs +++ b/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(); diff --git a/crates/adapters/sqlite/tests/repositories.rs b/crates/adapters/sqlite/tests/repositories.rs index 88f5f4e2..aa49c50f 100644 --- a/crates/adapters/sqlite/tests/repositories.rs +++ b/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"] diff --git a/crates/adapters/sqlite/tests/repositories/review_repository_test.rs b/crates/adapters/sqlite/tests/repositories/review_repository_test.rs new file mode 100644 index 00000000..f8166a4b --- /dev/null +++ b/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); +} diff --git a/crates/application/src/diary/delete_review.rs b/crates/application/src/diary/delete_review.rs index 5c6c7000..3a67b26d 100644 --- a/crates/application/src/diary/delete_review.rs +++ b/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?; - - if let Err(e) = deps - .event_publisher - .publish(&DomainEvent::ReviewDeleted { - review_id: review_id.clone(), - user_id: requesting_user_id.clone(), - }) - .await - { - tracing::warn!("failed to publish ReviewDeleted: {e}"); - } + deps.review + .delete_review( + &review_id, + &[DomainEvent::ReviewDeleted { + review_id: review_id.clone(), + user_id: requesting_user_id.clone(), + }], + ) + .await?; let history = deps.diary.get_review_history(&movie_id).await?; if history.viewings().is_empty() { diff --git a/crates/application/src/diary/deps.rs b/crates/application/src/diary/deps.rs index 2f3d0995..bbd56789 100644 --- a/crates/application/src/diary/deps.rs +++ b/crates/application/src/diary/deps.rs @@ -16,7 +16,6 @@ pub struct DeleteReviewDeps { pub struct EditReviewDeps { pub review: Arc, - pub event_publisher: Arc, } pub struct GetMovieSocialPageDeps { diff --git a/crates/application/src/diary/edit_review.rs b/crates/application/src/diary/edit_review.rs index 64f1c159..9e6c7224 100644 --- a/crates/application/src/diary/edit_review.rs +++ b/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?; - - if let Err(e) = deps - .event_publisher - .publish(&DomainEvent::ReviewUpdated { - review_id: updated.id().clone(), - movie_id: updated.movie_id().clone(), - user_id: updated.user_id().clone(), - rating: updated.rating().clone(), - watched_at: *updated.watched_at(), - }) - .await - { - tracing::warn!("failed to publish ReviewUpdated: {e}"); - } - - Ok(()) + let event = DomainEvent::ReviewUpdated { + review_id: updated.id().clone(), + movie_id: updated.movie_id().clone(), + user_id: updated.user_id().clone(), + rating: updated.rating().clone(), + watched_at: *updated.watched_at(), + }; + deps.review.update_review(&updated, &[event]).await } diff --git a/crates/application/src/diary/review_logger.rs b/crates/application/src/diary/review_logger.rs index a5e4683f..d6415392 100644 --- a/crates/application/src/diary/review_logger.rs +++ b/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() { - 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 + Ok(()) } } diff --git a/crates/application/tests/diary/delete_review_test.rs b/crates/application/tests/diary/delete_review_test.rs index 9ef43edc..f9d09e5a 100644 --- a/crates/application/tests/diary/delete_review_test.rs +++ b/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()); +} diff --git a/crates/application/tests/diary/edit_review_test.rs b/crates/application/tests/diary/edit_review_test.rs index 09b0d1b1..9f184dc6 100644 --- a/crates/application/tests/diary/edit_review_test.rs +++ b/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() -> ( - Arc, - Arc, - ReviewId, - UserId, -) { +async fn setup() -> (Arc, 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, events, review_id, user_id) + reviews.save_review(&review, &[]).await.unwrap(); + (reviews, review_id, user_id) } -fn deps(reviews: &Arc, events: &Arc) -> EditReviewDeps { +fn deps(reviews: &Arc) -> 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); + +#[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, 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, domain::errors::DomainError> { + self.0.get_all_reviews_for_user(id).await + } +} + #[tokio::test] -async fn edit_publishes_review_updated_event() { - let (reviews, events, review_id, user_id) = setup().await; +async fn editing_a_review_saves_the_change_and_its_event_together() { + 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(); - assert_eq!(published.len(), 1); + let updated = reviews.get_review_by_id(&review_id).await.unwrap().unwrap(); + 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( diff --git a/crates/application/tests/diary/log_review_test.rs b/crates/application/tests/diary/log_review_test.rs index 3dc95590..ab0ca4c1 100644 --- a/crates/application/tests/diary/log_review_test.rs +++ b/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] diff --git a/crates/application/tests/diary/review_logger_test.rs b/crates/application/tests/diary/review_logger_test.rs index 98e76180..2df444c4 100644 --- a/crates/application/tests/diary/review_logger_test.rs +++ b/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_eq!( - movies.review_count(), - 1, - "review is saved before publishing" + assert!(result.is_err()); + assert_eq!(movies.review_count(), 0); + assert!(movies.recorded_events().is_empty()); +} + +#[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 { .. })) ); } diff --git a/crates/bootstrap/src/build.rs b/crates/bootstrap/src/build.rs index 4fdde3d9..90fd7f6f 100644 --- a/crates/bootstrap/src/build.rs +++ b/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(), diff --git a/crates/domain/src/ports/diary.rs b/crates/domain/src/ports/diary.rs index 5c742f4c..3ae36d54 100644 --- a/crates/domain/src/ports/diary.rs +++ b/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, DomainError>; - async fn update_review(&self, review: &Review) -> Result<(), DomainError>; - async fn delete_review(&self, review_id: &ReviewId) -> Result<(), DomainError>; + async fn update_review( + &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, DomainError>; } diff --git a/crates/domain/src/testing/fakes/panics.rs b/crates/domain/src/testing/fakes/panics.rs index 1039d599..fa9dd781 100644 --- a/crates/domain/src/testing/fakes/panics.rs +++ b/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, 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, DomainError> { diff --git a/crates/domain/src/testing/fakes/refusing.rs b/crates/domain/src/testing/fakes/refusing.rs index 4549d3d4..9ea36bef 100644 --- a/crates/domain/src/testing/fakes/refusing.rs +++ b/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, 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, DomainError> { diff --git a/crates/domain/src/testing/fakes/store.rs b/crates/domain/src/testing/fakes/store.rs index 94f3ca69..71e9c6ca 100644 --- a/crates/domain/src/testing/fakes/store.rs +++ b/crates/domain/src/testing/fakes/store.rs @@ -72,6 +72,7 @@ pub struct InMemoryStore { pub(super) follows: RwLock>, pub(super) blocked: RwLock>, pub(super) processed_events: RwLock>, + pub(super) recorded_events: RwLock>, } 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 { + 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() } diff --git a/crates/domain/src/testing/fakes/store_review.rs b/crates/domain/src/testing/fakes/store_review.rs index cf276ac4..3bf2f5bd 100644 --- a/crates/domain/src/testing/fakes/store_review.rs +++ b/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(()) } diff --git a/crates/domain/tests/testing/in_memory_store_test.rs b/crates/domain/tests/testing/in_memory_store_test.rs index ec8bb068..ecab6148 100644 --- a/crates/domain/tests/testing/in_memory_store_test.rs +++ b/crates/domain/tests/testing/in_memory_store_test.rs @@ -144,7 +144,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (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(); }