From 62ad9f401745a5deebe24e7359fd9c61fb854108 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 00:49:28 +0100 Subject: [PATCH] feat: atomic goal, watchlist, profile writes --- crates/adapters/sqlite/src/lib.rs | 4 +- .../sqlite/src/repositories/goals/mod.rs | 45 +- .../src/repositories/profile_fields/mod.rs | 19 +- .../sqlite/src/repositories/users/mod.rs | 11 +- .../sqlite/src/repositories/watchlist/mod.rs | 48 +- crates/adapters/sqlite/tests/cascade_test.rs | 13 +- crates/adapters/sqlite/tests/repositories.rs | 4 + .../tests/repositories/outbox_support.rs | 38 ++ .../tests/repositories/outbox_writes_test.rs | 421 ++++++++++++++++++ .../repositories/review_repository_test.rs | 125 +++++- .../repositories/user_repository_test.rs | 4 +- crates/application/src/diary/review_logger.rs | 20 +- crates/application/src/goals/create.rs | 17 +- crates/application/src/goals/delete.rs | 15 +- crates/application/src/goals/deps.rs | 3 +- crates/application/src/goals/update.rs | 17 +- crates/application/src/users/deps.rs | 1 - .../application/src/users/update_profile.rs | 27 +- .../src/users/update_profile_fields.rs | 11 +- crates/application/src/watchlist/add.rs | 22 +- crates/application/src/watchlist/deps.rs | 1 - crates/application/src/watchlist/remove.rs | 11 +- .../tests/diary/delete_review_test.rs | 2 +- .../tests/diary/edit_review_test.rs | 2 +- .../tests/diary/review_logger_test.rs | 20 +- crates/application/tests/goals/create_test.rs | 55 ++- crates/application/tests/goals/delete_test.rs | 69 ++- crates/application/tests/goals/get_test.rs | 1 - crates/application/tests/goals/list_test.rs | 1 - crates/application/tests/goals/update_test.rs | 60 ++- .../tests/users/get_profile_settings_test.rs | 1 + .../tests/users/update_profile_fields_test.rs | 42 +- .../tests/users/update_profile_test.rs | 129 +++++- .../application/tests/watchlist/add_test.rs | 50 ++- .../application/tests/watchlist/is_on_test.rs | 8 +- .../tests/watchlist/remove_test.rs | 68 ++- crates/bootstrap/src/build.rs | 3 - .../bootstrap/tests/http_handlers/support.rs | 1 + crates/domain/src/ports/auth.rs | 2 + crates/domain/src/ports/goals.rs | 12 +- crates/domain/src/ports/watchlist.rs | 12 +- crates/domain/src/testing/fakes/noops.rs | 5 +- crates/domain/src/testing/fakes/panics.rs | 18 +- crates/domain/src/testing/fakes/refusing.rs | 52 ++- crates/domain/src/testing/fakes/store_goal.rs | 17 +- crates/domain/src/testing/fakes/store_user.rs | 5 + .../src/testing/fakes/store_watchlist.rs | 19 +- .../tests/testing/in_memory_store_test.rs | 29 +- 48 files changed, 1296 insertions(+), 264 deletions(-) create mode 100644 crates/adapters/sqlite/tests/repositories/outbox_support.rs create mode 100644 crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index 7a5c486f..eeada475 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -6,8 +6,9 @@ mod repositories; mod shared; pub use event_queue::{DbEventQueueConfig, SqliteEventQueue}; +use repositories::SqliteUserSettingsRepository; pub use repositories::{ - SqliteApContentQuery, SqliteDiaryRepository, SqliteImageRefAdapter, + SqliteApContentQuery, SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter, SqliteImportProfileRepository, SqliteImportSessionRepository, SqliteInsightsQuery, SqliteMovieDeduplicator, SqliteMovieProfileRepository, SqliteMovieRepository, SqlitePersonAdapter, SqliteProcessedEventLedger, SqliteProfileFieldsRepository, @@ -17,7 +18,6 @@ pub use repositories::{ SqliteWrapUpRepository, SqliteWrapUpStatsQuery, create_federated_profile_query, create_image_ref, create_person_adapter, create_search_adapter, }; -use repositories::{SqliteGoalRepository, SqliteUserSettingsRepository}; /// Row helpers shared with the `sqlite-federation` crate (which cannot live /// here because it depends on `activitypub`/`k-ap`). diff --git a/crates/adapters/sqlite/src/repositories/goals/mod.rs b/crates/adapters/sqlite/src/repositories/goals/mod.rs index a8934cae..d4c82f7b 100644 --- a/crates/adapters/sqlite/src/repositories/goals/mod.rs +++ b/crates/adapters/sqlite/src/repositories/goals/mod.rs @@ -1,12 +1,15 @@ use async_trait::async_trait; use domain::{ errors::DomainError, + events::DomainEvent, models::{Goal, GoalType}, ports::{GoalCommand, GoalQuery}, value_objects::{GoalId, UserId}, }; use sqlx::{Row, SqlitePool}; +use crate::shared::enqueue_events; + pub struct SqliteGoalRepository { pool: SqlitePool, } @@ -19,7 +22,7 @@ impl SqliteGoalRepository { #[async_trait] impl GoalCommand for SqliteGoalRepository { - async fn save(&self, goal: &Goal) -> Result<(), DomainError> { + async fn save(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError> { let id = goal.id().value().to_string(); let user_id = goal.user_id().value().to_string(); let year = goal.year() as i64; @@ -27,6 +30,11 @@ impl GoalCommand for SqliteGoalRepository { let goal_type = goal.goal_type().as_str(); let created_at = goal.created_at().format("%Y-%m-%d %H:%M:%S").to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query( "INSERT INTO goals (id, user_id, year, target_count, goal_type, created_at) \ VALUES (?, ?, ?, ?, ?, ?)", @@ -37,45 +45,62 @@ impl GoalCommand for SqliteGoalRepository { .bind(target) .bind(goal_type) .bind(&created_at) - .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 update(&self, goal: &Goal) -> Result<(), DomainError> { + async fn update(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError> { let id = goal.id().value().to_string(); let target = goal.target_count() as i64; + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; let result = sqlx::query("UPDATE goals SET target_count = ? WHERE id = ?") .bind(target) .bind(&id) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; if result.rows_affected() == 0 { return Err(DomainError::NotFound("Goal not found".into())); } - Ok(()) + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } - async fn delete(&self, id: &GoalId, user_id: &UserId) -> Result<(), DomainError> { + async fn delete( + &self, + id: &GoalId, + user_id: &UserId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let gid = id.value().to_string(); let uid = user_id.value().to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; let result = sqlx::query("DELETE FROM goals WHERE id = ? AND user_id = ?") .bind(&gid) .bind(&uid) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; if result.rows_affected() == 0 { return Err(DomainError::NotFound("Goal not found".into())); } - Ok(()) + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } } diff --git a/crates/adapters/sqlite/src/repositories/profile_fields/mod.rs b/crates/adapters/sqlite/src/repositories/profile_fields/mod.rs index 9f11241f..9dd32339 100644 --- a/crates/adapters/sqlite/src/repositories/profile_fields/mod.rs +++ b/crates/adapters/sqlite/src/repositories/profile_fields/mod.rs @@ -2,10 +2,12 @@ use async_trait::async_trait; use sqlx::{Row, SqlitePool}; use domain::{ - errors::DomainError, models::ProfileField, ports::UserProfileFieldsRepository, - value_objects::UserId, + errors::DomainError, events::DomainEvent, models::ProfileField, + ports::UserProfileFieldsRepository, value_objects::UserId, }; +use crate::shared::enqueue_events; + pub struct SqliteProfileFieldsRepository { pool: SqlitePool, } @@ -41,12 +43,18 @@ impl UserProfileFieldsRepository for SqliteProfileFieldsRepository { &self, user_id: &UserId, fields: Vec, + events: &[DomainEvent], ) -> Result<(), DomainError> { let id_str = user_id.value().to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query("DELETE FROM user_profile_fields WHERE user_id = ?") .bind(&id_str) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; @@ -61,11 +69,12 @@ impl UserProfileFieldsRepository for SqliteProfileFieldsRepository { .bind(&field.name) .bind(&field.value) .bind(position) - .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) } } diff --git a/crates/adapters/sqlite/src/repositories/users/mod.rs b/crates/adapters/sqlite/src/repositories/users/mod.rs index f6990409..e856ea06 100644 --- a/crates/adapters/sqlite/src/repositories/users/mod.rs +++ b/crates/adapters/sqlite/src/repositories/users/mod.rs @@ -161,8 +161,14 @@ impl UserRepository for SqliteUserRepository { &self, user_id: &UserId, profile: &domain::models::UserProfile, + events: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { let id_str = user_id.value().to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query( "UPDATE users SET display_name = ?, bio = ?, avatar_path = ?, banner_path = ?, also_known_as = ? WHERE id = ?", ) @@ -172,10 +178,11 @@ impl UserRepository for SqliteUserRepository { .bind(&profile.banner_path) .bind(&profile.also_known_as) .bind(&id_str) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; - Ok(()) + crate::shared::enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } async fn list_with_stats(&self) -> Result, DomainError> { diff --git a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs index 2485dd53..a6046ce3 100644 --- a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs +++ b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs @@ -1,6 +1,7 @@ use async_trait::async_trait; use domain::{ errors::DomainError, + events::DomainEvent, models::{ WatchlistEntry, WatchlistWithMovie, collections::{PageParams, Paginated}, @@ -13,7 +14,7 @@ use sqlx::{Row, SqlitePool}; use rows::WatchlistRow; mod rows; -use crate::shared::datetime_to_str; +use crate::shared::{datetime_to_str, enqueue_events}; pub struct SqliteWatchlistRepository { pool: SqlitePool, @@ -27,12 +28,17 @@ impl SqliteWatchlistRepository { #[async_trait] impl WatchlistRepository for SqliteWatchlistRepository { - async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError> { + async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError> { let id = entry.id.value().to_string(); let user_id = entry.user_id.value().to_string(); let movie_id = entry.movie_id.value().to_string(); let added_at = datetime_to_str(&entry.added_at); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query( "INSERT OR IGNORE INTO watchlist_entries (id, user_id, movie_id, added_at) \ VALUES (?, ?, ?, ?)", @@ -41,22 +47,32 @@ impl WatchlistRepository for SqliteWatchlistRepository { .bind(&user_id) .bind(&movie_id) .bind(&added_at) - .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 remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError> { + async fn remove( + &self, + user_id: &UserId, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let uid = user_id.value().to_string(); let mid = movie_id.value().to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; let result = sqlx::query("DELETE FROM watchlist_entries WHERE user_id = ? AND movie_id = ?") .bind(&uid) .bind(&mid) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; @@ -66,24 +82,36 @@ impl WatchlistRepository for SqliteWatchlistRepository { mid ))); } - Ok(()) + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } async fn remove_if_present( &self, user_id: &UserId, movie_id: &MovieId, + events: &[DomainEvent], ) -> Result { let uid = user_id.value().to_string(); let mid = movie_id.value().to_string(); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; let result = sqlx::query("DELETE FROM watchlist_entries WHERE user_id = ? AND movie_id = ?") .bind(&uid) .bind(&mid) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; - Ok(result.rows_affected() > 0) + let removed = result.rows_affected() > 0; + if removed { + enqueue_events(&mut tx, events).await?; + } + tx.commit().await.map_err(crate::shared::map_sqlx_error)?; + Ok(removed) } async fn get_for_user( diff --git a/crates/adapters/sqlite/tests/cascade_test.rs b/crates/adapters/sqlite/tests/cascade_test.rs index 7a07b892..176ed7c6 100644 --- a/crates/adapters/sqlite/tests/cascade_test.rs +++ b/crates/adapters/sqlite/tests/cascade_test.rs @@ -128,12 +128,15 @@ async fn seed_user_with_dependents( parts .watchlist - .add(&WatchlistEntry::new(user_id.clone(), movie.id().clone())) + .add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) .await .unwrap(); parts .goal_command - .save(&test_goal(&user_id, 2024, 50)) + .save(&test_goal(&user_id, 2024, 50), &[]) .await .unwrap(); parts @@ -148,6 +151,7 @@ async fn seed_user_with_dependents( name: "Site".into(), value: "https://example.com".into(), }], + &[], ) .await .unwrap(); @@ -256,7 +260,10 @@ async fn deleting_a_movie_cascades_to_watchlist_entries_and_every_enrichment_tab parts.movie_command.upsert_movie(&movie).await.unwrap(); parts .watchlist - .add(&WatchlistEntry::new(user.id().clone(), movie.id().clone())) + .add( + &WatchlistEntry::new(user.id().clone(), movie.id().clone()), + &[], + ) .await .unwrap(); parts diff --git a/crates/adapters/sqlite/tests/repositories.rs b/crates/adapters/sqlite/tests/repositories.rs index aa49c50f..3b49c84d 100644 --- a/crates/adapters/sqlite/tests/repositories.rs +++ b/crates/adapters/sqlite/tests/repositories.rs @@ -6,6 +6,10 @@ mod image_ref_adapter_test; mod movie_dedup_test; #[path = "repositories/movie_profile_repository_test.rs"] mod movie_profile_repository_test; +#[path = "repositories/outbox_support.rs"] +mod outbox_support; +#[path = "repositories/outbox_writes_test.rs"] +mod outbox_writes_test; #[path = "repositories/person_adapter_test.rs"] mod person_adapter_test; #[path = "repositories/review_repository_test.rs"] diff --git a/crates/adapters/sqlite/tests/repositories/outbox_support.rs b/crates/adapters/sqlite/tests/repositories/outbox_support.rs new file mode 100644 index 00000000..60b4ff3d --- /dev/null +++ b/crates/adapters/sqlite/tests/repositories/outbox_support.rs @@ -0,0 +1,38 @@ +//! Shared helpers for the transactional-outbox repository tests. + +use std::str::FromStr; + +use sqlx::{ + SqlitePool, + sqlite::{SqliteConnectOptions, SqlitePoolOptions}, +}; + +pub 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 +} + +pub async fn count(pool: &SqlitePool, table: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}")) + .fetch_one(pool) + .await + .unwrap() +} + +/// Makes every later `event_queue` insert abort, so a write whose own row +/// already landed inside the transaction must be rolled back with it. +pub async fn fail_event_inserts(pool: &SqlitePool) { + sqlx::query( + "CREATE TRIGGER fail_event_insert BEFORE INSERT ON event_queue \ + BEGIN SELECT RAISE(ABORT, 'enqueue failed'); END", + ) + .execute(pool) + .await + .unwrap(); +} diff --git a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs new file mode 100644 index 00000000..bfdd1f59 --- /dev/null +++ b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs @@ -0,0 +1,421 @@ +//! Goal, watchlist and profile writes commit with their events or not at all. + +use domain::{ + events::DomainEvent, + models::{ProfileField, UserProfile, WatchlistEntry}, + ports::{ + GoalCommand, MovieCommand, UserProfileFieldsRepository, UserRepository, WatchlistRepository, + }, + testing::{test_goal, test_movie, test_user}, + value_objects::UserId, +}; +use sqlite::{ + SqliteGoalRepository, SqliteMovieRepository, SqliteProfileFieldsRepository, + SqliteUserRepository, SqliteWatchlistRepository, +}; +use sqlx::SqlitePool; + +use super::outbox_support::{count, fail_event_inserts, migrated_pool}; + +async fn pool_with_user() -> (SqlitePool, domain::models::User) { + let pool = migrated_pool().await; + let user = test_user("alice"); + SqliteUserRepository::new(pool.clone()) + .save(&user) + .await + .unwrap(); + (pool, user) +} + +fn goal_created(goal: &domain::models::Goal) -> DomainEvent { + DomainEvent::GoalCreated { + goal_id: goal.id().clone(), + user_id: goal.user_id().clone(), + year: goal.year(), + target_count: goal.target_count(), + } +} + +fn goal_updated(goal: &domain::models::Goal) -> DomainEvent { + DomainEvent::GoalUpdated { + goal_id: goal.id().clone(), + user_id: goal.user_id().clone(), + year: goal.year(), + target_count: goal.target_count(), + } +} + +fn goal_deleted(goal: &domain::models::Goal) -> DomainEvent { + DomainEvent::GoalDeleted { + goal_id: goal.id().clone(), + user_id: goal.user_id().clone(), + year: goal.year(), + } +} + +#[tokio::test] +async fn saving_a_goal_writes_the_goal_and_its_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + let goal = test_goal(user.id(), 2025, 50); + + SqliteGoalRepository::new(pool.clone()) + .save(&goal, &[goal_created(&goal)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "goals").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_goal_save() { + let (pool, user) = pool_with_user().await; + let goal = test_goal(user.id(), 2025, 50); + fail_event_inserts(&pool).await; + + let result = SqliteGoalRepository::new(pool.clone()) + .save(&goal, &[goal_created(&goal)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "goals").await, 0); +} + +#[tokio::test] +async fn updating_a_goal_writes_the_change_and_its_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + let repo = SqliteGoalRepository::new(pool.clone()); + let mut goal = test_goal(user.id(), 2025, 50); + repo.save(&goal, &[]).await.unwrap(); + goal.update_target(80).unwrap(); + + repo.update(&goal, &[goal_updated(&goal)]).await.unwrap(); + + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_goal_update() { + let (pool, user) = pool_with_user().await; + let repo = SqliteGoalRepository::new(pool.clone()); + let mut goal = test_goal(user.id(), 2025, 50); + repo.save(&goal, &[]).await.unwrap(); + goal.update_target(80).unwrap(); + fail_event_inserts(&pool).await; + + let result = repo.update(&goal, &[goal_updated(&goal)]).await; + + assert!(result.is_err()); + let target: i64 = sqlx::query_scalar("SELECT target_count FROM goals") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(target, 50, "update rolled back"); +} + +#[tokio::test] +async fn updating_a_missing_goal_enqueues_no_event() { + let (pool, user) = pool_with_user().await; + let goal = test_goal(user.id(), 2025, 50); + + let result = SqliteGoalRepository::new(pool.clone()) + .update(&goal, &[goal_updated(&goal)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "event_queue").await, 0); +} + +#[tokio::test] +async fn deleting_a_goal_writes_the_removal_and_its_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + let repo = SqliteGoalRepository::new(pool.clone()); + let goal = test_goal(user.id(), 2025, 50); + repo.save(&goal, &[]).await.unwrap(); + + repo.delete(goal.id(), user.id(), &[goal_deleted(&goal)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "goals").await, 0); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_goal_delete() { + let (pool, user) = pool_with_user().await; + let repo = SqliteGoalRepository::new(pool.clone()); + let goal = test_goal(user.id(), 2025, 50); + repo.save(&goal, &[]).await.unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .delete(goal.id(), user.id(), &[goal_deleted(&goal)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "goals").await, 1, "delete rolled back"); +} + +async fn watchlist_fixture() -> (SqlitePool, UserId, domain::models::Movie) { + let (pool, user) = pool_with_user().await; + let movie = test_movie("Heat"); + SqliteMovieRepository::new(pool.clone()) + .upsert_movie(&movie) + .await + .unwrap(); + (pool, user.id().clone(), movie) +} + +fn watchlist_added(entry: &WatchlistEntry, movie: &domain::models::Movie) -> DomainEvent { + DomainEvent::WatchlistEntryAdded { + user_id: entry.user_id.clone(), + movie_id: movie.id().clone(), + movie_title: movie.title().value().to_string(), + release_year: movie.release_year().value(), + external_metadata_id: None, + added_at: entry.added_at, + } +} + +fn watchlist_removed(user_id: &UserId, movie: &domain::models::Movie) -> DomainEvent { + DomainEvent::WatchlistEntryRemoved { + user_id: user_id.clone(), + movie_id: movie.id().clone(), + } +} + +#[tokio::test] +async fn adding_a_watchlist_entry_writes_the_entry_and_its_event_in_one_commit() { + let (pool, user_id, movie) = watchlist_fixture().await; + let entry = WatchlistEntry::new(user_id, movie.id().clone()); + + SqliteWatchlistRepository::new(pool.clone()) + .add(&entry, &[watchlist_added(&entry, &movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "watchlist_entries").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_watchlist_add() { + let (pool, user_id, movie) = watchlist_fixture().await; + let entry = WatchlistEntry::new(user_id, movie.id().clone()); + fail_event_inserts(&pool).await; + + let result = SqliteWatchlistRepository::new(pool.clone()) + .add(&entry, &[watchlist_added(&entry, &movie)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "watchlist_entries").await, 0); +} + +#[tokio::test] +async fn removing_a_watchlist_entry_writes_the_removal_and_its_event_in_one_commit() { + let (pool, user_id, movie) = watchlist_fixture().await; + let repo = SqliteWatchlistRepository::new(pool.clone()); + repo.add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) + .await + .unwrap(); + + repo.remove(&user_id, movie.id(), &[watchlist_removed(&user_id, &movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "watchlist_entries").await, 0); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_watchlist_remove() { + let (pool, user_id, movie) = watchlist_fixture().await; + let repo = SqliteWatchlistRepository::new(pool.clone()); + repo.add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) + .await + .unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .remove(&user_id, movie.id(), &[watchlist_removed(&user_id, &movie)]) + .await; + + assert!(result.is_err()); + assert_eq!( + count(&pool, "watchlist_entries").await, + 1, + "remove rolled back" + ); +} + +#[tokio::test] +async fn removing_an_absent_watchlist_entry_enqueues_no_event() { + let (pool, user_id, movie) = watchlist_fixture().await; + + let result = SqliteWatchlistRepository::new(pool.clone()) + .remove(&user_id, movie.id(), &[watchlist_removed(&user_id, &movie)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "event_queue").await, 0); +} + +#[tokio::test] +async fn remove_if_present_enqueues_its_event_only_when_a_row_was_removed() { + let (pool, user_id, movie) = watchlist_fixture().await; + let repo = SqliteWatchlistRepository::new(pool.clone()); + let event = watchlist_removed(&user_id, &movie); + + let absent = repo + .remove_if_present(&user_id, movie.id(), std::slice::from_ref(&event)) + .await + .unwrap(); + assert!(!absent); + assert_eq!(count(&pool, "event_queue").await, 0); + + repo.add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) + .await + .unwrap(); + let present = repo + .remove_if_present(&user_id, movie.id(), std::slice::from_ref(&event)) + .await + .unwrap(); + + assert!(present); + assert_eq!(count(&pool, "watchlist_entries").await, 0); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_remove_if_present() { + let (pool, user_id, movie) = watchlist_fixture().await; + let repo = SqliteWatchlistRepository::new(pool.clone()); + repo.add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) + .await + .unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .remove_if_present(&user_id, movie.id(), &[watchlist_removed(&user_id, &movie)]) + .await; + + assert!(result.is_err()); + assert_eq!( + count(&pool, "watchlist_entries").await, + 1, + "remove rolled back" + ); +} + +#[tokio::test] +async fn updating_a_profile_writes_the_change_and_its_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + let repo = SqliteUserRepository::new(pool.clone()); + let profile = UserProfile { + bio: Some("hello".into()), + ..Default::default() + }; + + repo.update_profile( + user.id(), + &profile, + &[DomainEvent::UserUpdated { + user_id: user.id().clone(), + }], + ) + .await + .unwrap(); + + let found = repo.find_by_id(user.id()).await.unwrap().unwrap(); + assert_eq!(found.bio(), Some("hello")); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_profile_update() { + let (pool, user) = pool_with_user().await; + let repo = SqliteUserRepository::new(pool.clone()); + fail_event_inserts(&pool).await; + + let result = repo + .update_profile( + user.id(), + &UserProfile { + bio: Some("hello".into()), + ..Default::default() + }, + &[DomainEvent::UserUpdated { + user_id: user.id().clone(), + }], + ) + .await; + + assert!(result.is_err()); + let found = repo.find_by_id(user.id()).await.unwrap().unwrap(); + assert_eq!(found.bio(), None, "profile update rolled back"); +} + +fn fields() -> Vec { + vec![ProfileField { + name: "Site".into(), + value: "https://example.com".into(), + }] +} + +#[tokio::test] +async fn setting_profile_fields_writes_them_and_the_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + + SqliteProfileFieldsRepository::new(pool.clone()) + .set_fields( + user.id(), + fields(), + &[DomainEvent::UserUpdated { + user_id: user.id().clone(), + }], + ) + .await + .unwrap(); + + assert_eq!(count(&pool, "user_profile_fields").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_profile_fields_replacement() { + let (pool, user) = pool_with_user().await; + let repo = SqliteProfileFieldsRepository::new(pool.clone()); + repo.set_fields(user.id(), fields(), &[]).await.unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .set_fields( + user.id(), + vec![], + &[DomainEvent::UserUpdated { + user_id: user.id().clone(), + }], + ) + .await; + + assert!(result.is_err()); + assert_eq!( + repo.get_fields(user.id()).await.unwrap().len(), + 1, + "delete of the old fields rolled back" + ); +} diff --git a/crates/adapters/sqlite/tests/repositories/review_repository_test.rs b/crates/adapters/sqlite/tests/repositories/review_repository_test.rs index f8166a4b..9f9cc136 100644 --- a/crates/adapters/sqlite/tests/repositories/review_repository_test.rs +++ b/crates/adapters/sqlite/tests/repositories/review_repository_test.rs @@ -1,33 +1,12 @@ -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}, -}; +use sqlx::SqlitePool; -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() -} +use super::outbox_support::{count, fail_event_inserts, migrated_pool}; fn logged(review: &domain::models::Review) -> DomainEvent { DomainEvent::ReviewLogged { @@ -80,3 +59,103 @@ async fn a_failed_review_save_leaves_no_event_row() { assert_eq!(count(&pool, "reviews").await, 0); assert_eq!(count(&pool, "event_queue").await, 0); } + +fn updated(review: &domain::models::Review) -> DomainEvent { + DomainEvent::ReviewUpdated { + 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(), + } +} + +async fn seeded() -> (SqlitePool, domain::models::Review) { + 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); + (pool, review) +} + +#[tokio::test] +async fn a_failing_enqueue_after_the_review_insert_rolls_the_review_back() { + let (pool, review) = seeded().await; + fail_event_inserts(&pool).await; + + let result = SqliteReviewRepository::new(pool.clone()) + .save_review(&review, &[logged(&review)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "reviews").await, 0); + assert_eq!(count(&pool, "event_queue").await, 0); +} + +#[tokio::test] +async fn updating_a_review_writes_the_change_and_its_event_in_one_commit() { + let (pool, review) = seeded().await; + let repo = SqliteReviewRepository::new(pool.clone()); + repo.save_review(&review, &[]).await.unwrap(); + let event = updated(&review); + + repo.update_review(&review, &[event]).await.unwrap(); + + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_after_the_review_update_rolls_the_update_back() { + let (pool, review) = seeded().await; + let repo = SqliteReviewRepository::new(pool.clone()); + repo.save_review(&review, &[]).await.unwrap(); + fail_event_inserts(&pool).await; + let event = updated(&review); + + let result = repo.update_review(&review, &[event]).await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "event_queue").await, 0); +} + +#[tokio::test] +async fn deleting_a_review_writes_the_removal_and_its_event_in_one_commit() { + let (pool, review) = seeded().await; + let repo = SqliteReviewRepository::new(pool.clone()); + repo.save_review(&review, &[]).await.unwrap(); + let event = DomainEvent::ReviewDeleted { + review_id: review.id().clone(), + user_id: review.user_id().clone(), + }; + + repo.delete_review(review.id(), &[event]).await.unwrap(); + + assert_eq!(count(&pool, "reviews").await, 0); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_after_the_review_delete_rolls_the_delete_back() { + let (pool, review) = seeded().await; + let repo = SqliteReviewRepository::new(pool.clone()); + repo.save_review(&review, &[]).await.unwrap(); + fail_event_inserts(&pool).await; + let event = DomainEvent::ReviewDeleted { + review_id: review.id().clone(), + user_id: review.user_id().clone(), + }; + + let result = repo.delete_review(review.id(), &[event]).await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "reviews").await, 1, "delete rolled back"); + assert_eq!(count(&pool, "event_queue").await, 0); +} diff --git a/crates/adapters/sqlite/tests/repositories/user_repository_test.rs b/crates/adapters/sqlite/tests/repositories/user_repository_test.rs index fa6c3290..42586f4b 100644 --- a/crates/adapters/sqlite/tests/repositories/user_repository_test.rs +++ b/crates/adapters/sqlite/tests/repositories/user_repository_test.rs @@ -67,6 +67,7 @@ async fn update_profile_persists_bio_and_avatar() { avatar_path: Some("avatars/user1".to_string()), ..Default::default() }, + &[], ) .await .unwrap(); @@ -88,10 +89,11 @@ async fn update_profile_clears_fields_with_none() { avatar_path: Some("path".to_string()), ..Default::default() }, + &[], ) .await .unwrap(); - repo.update_profile(user.id(), &UserProfile::default()) + repo.update_profile(user.id(), &UserProfile::default(), &[]) .await .unwrap(); diff --git a/crates/application/src/diary/review_logger.rs b/crates/application/src/diary/review_logger.rs index d6415392..9a31fa07 100644 --- a/crates/application/src/diary/review_logger.rs +++ b/crates/application/src/diary/review_logger.rs @@ -94,19 +94,17 @@ impl ReviewLogger for DefaultReviewLogger { 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()) - .await?; - if was_on_watchlist { - let _ = self - .event_publisher - .publish(&DomainEvent::WatchlistEntryRemoved { + // The removal event is recorded only if an entry was actually removed. + self.watchlist_repo + .remove_if_present( + review.user_id(), + review.movie_id(), + &[DomainEvent::WatchlistEntryRemoved { user_id: review.user_id().clone(), movie_id: review.movie_id().clone(), - }) - .await; - } + }], + ) + .await?; Ok(()) } diff --git a/crates/application/src/goals/create.rs b/crates/application/src/goals/create.rs index 4ada4d4d..ae4bc708 100644 --- a/crates/application/src/goals/create.rs +++ b/crates/application/src/goals/create.rs @@ -29,19 +29,16 @@ pub async fn execute( cmd.target_count, GoalType::Movies, )?; - deps.goal_command.save(&g).await?; + let event = DomainEvent::GoalCreated { + goal_id: g.id().clone(), + user_id: user_id.clone(), + year: cmd.year, + target_count: cmd.target_count, + }; + deps.goal_command.save(&g, &[event]).await?; let current_count = deps.stats.count_reviews_in_year(&user_id, cmd.year).await?; - deps.event_publisher - .publish(&DomainEvent::GoalCreated { - goal_id: g.id().clone(), - user_id, - year: cmd.year, - target_count: cmd.target_count, - }) - .await?; - Ok(GoalWithProgress { goal: g, current_count, diff --git a/crates/application/src/goals/delete.rs b/crates/application/src/goals/delete.rs index ce092a77..41529167 100644 --- a/crates/application/src/goals/delete.rs +++ b/crates/application/src/goals/delete.rs @@ -11,15 +11,12 @@ pub async fn execute(deps: &GoalCommandDeps, cmd: DeleteGoalCommand) -> Result<( .await? .ok_or_else(|| DomainError::NotFound(format!("Goal for year {}", cmd.year)))?; - deps.goal_command.delete(g.id(), &user_id).await?; - - deps.event_publisher - .publish(&DomainEvent::GoalDeleted { - goal_id: g.id().clone(), - user_id, - year: cmd.year, - }) - .await?; + let event = DomainEvent::GoalDeleted { + goal_id: g.id().clone(), + user_id: user_id.clone(), + year: cmd.year, + }; + deps.goal_command.delete(g.id(), &user_id, &[event]).await?; Ok(()) } diff --git a/crates/application/src/goals/deps.rs b/crates/application/src/goals/deps.rs index cfe4b9a3..36ed1a96 100644 --- a/crates/application/src/goals/deps.rs +++ b/crates/application/src/goals/deps.rs @@ -1,12 +1,11 @@ use std::sync::Arc; -use domain::ports::{EventPublisher, GoalCommand, GoalQuery, StatsRepository}; +use domain::ports::{GoalCommand, GoalQuery, StatsRepository}; pub struct GoalCommandDeps { pub goal_command: Arc, pub goal_query: Arc, pub stats: Arc, - pub event_publisher: Arc, } pub struct GoalQueryDeps { diff --git a/crates/application/src/goals/update.rs b/crates/application/src/goals/update.rs index d0fbe230..4274e5ba 100644 --- a/crates/application/src/goals/update.rs +++ b/crates/application/src/goals/update.rs @@ -17,19 +17,16 @@ pub async fn execute( .ok_or_else(|| DomainError::NotFound(format!("Goal for year {}", cmd.year)))?; g.update_target(cmd.target_count)?; - deps.goal_command.update(&g).await?; + let event = DomainEvent::GoalUpdated { + goal_id: g.id().clone(), + user_id: user_id.clone(), + year: cmd.year, + target_count: cmd.target_count, + }; + deps.goal_command.update(&g, &[event]).await?; let current_count = deps.stats.count_reviews_in_year(&user_id, cmd.year).await?; - deps.event_publisher - .publish(&DomainEvent::GoalUpdated { - goal_id: g.id().clone(), - user_id, - year: cmd.year, - target_count: cmd.target_count, - }) - .await?; - Ok(GoalWithProgress { goal: g, current_count, diff --git a/crates/application/src/users/deps.rs b/crates/application/src/users/deps.rs index 4394599c..f60dd06f 100644 --- a/crates/application/src/users/deps.rs +++ b/crates/application/src/users/deps.rs @@ -58,7 +58,6 @@ pub struct GetCurrentProfileDeps { pub struct UpdateProfileFieldsDeps { pub profile_fields: Arc, - pub event_publisher: Arc, } pub struct GetSettingsDeps { diff --git a/crates/application/src/users/update_profile.rs b/crates/application/src/users/update_profile.rs index e2168254..afae932b 100644 --- a/crates/application/src/users/update_profile.rs +++ b/crates/application/src/users/update_profile.rs @@ -91,6 +91,16 @@ pub async fn execute( } }); + let mut events = vec![DomainEvent::UserUpdated { + user_id: user_id.clone(), + }]; + if let Some(new_actor_url) = moved_to { + events.push(DomainEvent::UserAccountMoved { + user_id: user_id.clone(), + new_actor_url, + }); + } + deps.user .update_profile( &user_id, @@ -102,24 +112,9 @@ pub async fn execute( also_known_as: cmd.also_known_as, profile_fields: vec![], }, + &events, ) .await?; - deps.event_publisher - .publish(&DomainEvent::UserUpdated { - user_id: user_id.clone(), - }) - .await?; - - if let Some(new_actor_url) = moved_to { - let _ = deps - .event_publisher - .publish(&DomainEvent::UserAccountMoved { - user_id, - new_actor_url, - }) - .await; - } - Ok(()) } diff --git a/crates/application/src/users/update_profile_fields.rs b/crates/application/src/users/update_profile_fields.rs index 250b497c..c1a5cd1d 100644 --- a/crates/application/src/users/update_profile_fields.rs +++ b/crates/application/src/users/update_profile_fields.rs @@ -11,9 +11,10 @@ pub async fn execute( ) -> Result<(), DomainError> { UserProfile::validate_custom_fields(&cmd.fields)?; let user_id = UserId::from_uuid(cmd.user_id); - deps.profile_fields.set_fields(&user_id, cmd.fields).await?; - deps.event_publisher - .publish(&DomainEvent::UserUpdated { user_id }) - .await?; - Ok(()) + let event = DomainEvent::UserUpdated { + user_id: user_id.clone(), + }; + deps.profile_fields + .set_fields(&user_id, cmd.fields, &[event]) + .await } diff --git a/crates/application/src/watchlist/add.rs b/crates/application/src/watchlist/add.rs index 4d3fa2d7..c85a714e 100644 --- a/crates/application/src/watchlist/add.rs +++ b/crates/application/src/watchlist/add.rs @@ -23,19 +23,15 @@ pub async fn execute( .await?; let entry = WatchlistEntry::new(user_id.clone(), movie.id().clone()); - deps.watchlist.add(&entry).await?; - - let _ = deps - .event_publisher - .publish(&DomainEvent::WatchlistEntryAdded { - user_id, - movie_id: movie.id().clone(), - movie_title: movie.title().value().to_string(), - release_year: movie.release_year().value(), - external_metadata_id: movie.external_metadata_id().map(|e| e.value().to_string()), - added_at: entry.added_at, - }) - .await; + let event = DomainEvent::WatchlistEntryAdded { + user_id, + movie_id: movie.id().clone(), + movie_title: movie.title().value().to_string(), + release_year: movie.release_year().value(), + external_metadata_id: movie.external_metadata_id().map(|e| e.value().to_string()), + added_at: entry.added_at, + }; + deps.watchlist.add(&entry, &[event]).await?; Ok(()) } diff --git a/crates/application/src/watchlist/deps.rs b/crates/application/src/watchlist/deps.rs index acb4ef30..e6154553 100644 --- a/crates/application/src/watchlist/deps.rs +++ b/crates/application/src/watchlist/deps.rs @@ -29,5 +29,4 @@ pub struct IsOnWatchlistDeps { pub struct RemoveFromWatchlistDeps { pub watchlist: Arc, - pub event_publisher: Arc, } diff --git a/crates/application/src/watchlist/remove.rs b/crates/application/src/watchlist/remove.rs index 6d84ff82..7c0980e2 100644 --- a/crates/application/src/watchlist/remove.rs +++ b/crates/application/src/watchlist/remove.rs @@ -13,12 +13,11 @@ pub async fn execute( ) -> Result<(), DomainError> { let user_id = UserId::from_uuid(cmd.user_id); let movie_id = MovieId::from_uuid(cmd.movie_id); - deps.watchlist.remove(&user_id, &movie_id).await?; - - let _ = deps - .event_publisher - .publish(&DomainEvent::WatchlistEntryRemoved { user_id, movie_id }) - .await; + let event = DomainEvent::WatchlistEntryRemoved { + user_id: user_id.clone(), + movie_id: movie_id.clone(), + }; + deps.watchlist.remove(&user_id, &movie_id, &[event]).await?; Ok(()) } diff --git a/crates/application/tests/diary/delete_review_test.rs b/crates/application/tests/diary/delete_review_test.rs index f9d09e5a..0230199a 100644 --- a/crates/application/tests/diary/delete_review_test.rs +++ b/crates/application/tests/diary/delete_review_test.rs @@ -122,7 +122,7 @@ async fn deleting_a_review_removes_it_and_records_its_event_together() { } #[tokio::test] -async fn a_refused_delete_records_neither_the_removal_nor_its_event() { +async fn a_refused_delete_propagates_and_records_nothing() { let movies = InMemoryStore::new(); let diary = FakeDiaryQuery::new(); let events = NoopEventPublisher::new(); diff --git a/crates/application/tests/diary/edit_review_test.rs b/crates/application/tests/diary/edit_review_test.rs index 9f184dc6..065a5f08 100644 --- a/crates/application/tests/diary/edit_review_test.rs +++ b/crates/application/tests/diary/edit_review_test.rs @@ -151,7 +151,7 @@ async fn editing_a_review_saves_the_change_and_its_event_together() { } #[tokio::test] -async fn a_refused_edit_records_neither_the_change_nor_its_event() { +async fn a_refused_edit_propagates_and_records_nothing() { let (reviews, review_id, user_id) = setup().await; let deps = EditReviewDeps { review: Arc::new(WriteRefusingReviews(Arc::clone(&reviews))), diff --git a/crates/application/tests/diary/review_logger_test.rs b/crates/application/tests/diary/review_logger_test.rs index 2df444c4..28021ce7 100644 --- a/crates/application/tests/diary/review_logger_test.rs +++ b/crates/application/tests/diary/review_logger_test.rs @@ -68,7 +68,7 @@ async fn removes_from_watchlist_on_review() { // Add to watchlist let entry = WatchlistEntry::new(user_id.clone(), MovieId::from_uuid(movie_id)); - watchlist.add(&entry).await.unwrap(); + watchlist.add(&entry, &[]).await.unwrap(); assert_eq!(watchlist.watchlist_count(), 1); // Log review for same movie @@ -80,6 +80,20 @@ async fn removes_from_watchlist_on_review() { logger.log_review(cmd).await.unwrap(); assert_eq!(watchlist.watchlist_count(), 0); + assert!( + movies + .recorded_events() + .iter() + .any(|e| matches!(e, domain::events::DomainEvent::WatchlistEntryRemoved { .. })), + "the removal event rides the watchlist write" + ); + assert!( + !events + .published() + .iter() + .any(|e| matches!(e, domain::events::DomainEvent::WatchlistEntryRemoved { .. })), + "no longer published out of band" + ); } #[tokio::test] @@ -150,7 +164,7 @@ async fn watchlist_not_present_does_not_publish_removed() { logger.log_review(cmd).await.unwrap(); - let published = events.published(); + let published = movies.recorded_events(); assert!( !published .iter() @@ -260,7 +274,7 @@ async fn logging_a_review_saves_the_review_and_its_event_together() { } #[tokio::test] -async fn a_refused_save_records_neither_the_review_nor_its_event() { +async fn a_refused_save_propagates_and_records_nothing() { let movies = InMemoryStore::new(); let logger = DefaultReviewLogger::new( Arc::clone(&movies) as _, diff --git a/crates/application/tests/goals/create_test.rs b/crates/application/tests/goals/create_test.rs index 7bc05398..38a3b3a8 100644 --- a/crates/application/tests/goals/create_test.rs +++ b/crates/application/tests/goals/create_test.rs @@ -1,8 +1,9 @@ use crate::support::create_goal_command; use std::sync::Arc; +use domain::errors::DomainError; use domain::events::DomainEvent; -use domain::testing::{FakeStatsRepository, InMemoryStore, NoopEventPublisher}; +use domain::testing::{FakeStatsRepository, InMemoryStore, RefusingGoalCommand}; use uuid::Uuid; use crate::support::TestContextBuilder; @@ -13,12 +14,10 @@ use application::goals::deps::GoalCommandDeps; async fn creates_goal_and_returns_progress() { let goals = InMemoryStore::new(); let stats = FakeStatsRepository::new(); - let events = NoopEventPublisher::new(); let deps = GoalCommandDeps { goal_command: Arc::clone(&goals) as _, goal_query: Arc::clone(&goals) as _, stats: Arc::clone(&stats) as _, - event_publisher: Arc::clone(&events) as _, }; let result = create::execute(&deps, create_goal_command(2025, 50)) @@ -36,12 +35,10 @@ async fn creates_goal_with_review_count() { let goals = InMemoryStore::new(); let stats = FakeStatsRepository::new(); stats.set_review_count(Uuid::nil(), 2025, 5); - let events = NoopEventPublisher::new(); let deps = GoalCommandDeps { goal_command: Arc::clone(&goals) as _, goal_query: Arc::clone(&goals) as _, stats: Arc::clone(&stats) as _, - event_publisher: Arc::clone(&events) as _, }; let result = create::execute(&deps, create_goal_command(2025, 50)) @@ -53,26 +50,45 @@ async fn creates_goal_with_review_count() { } #[tokio::test] -async fn emits_goal_created_event() { - let b = TestContextBuilder::new(); - let events = NoopEventPublisher::new(); +async fn creating_a_goal_records_the_goal_and_its_event_together() { + let goals = InMemoryStore::new(); let deps = GoalCommandDeps { - goal_command: b.goal_command.clone(), - goal_query: b.goal_query.clone(), - stats: b.stats_repo.clone(), - event_publisher: Arc::clone(&events) as _, + goal_command: Arc::clone(&goals) as _, + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, }; create::execute(&deps, create_goal_command(2025, 10)) .await .unwrap(); - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, DomainEvent::GoalCreated { year: 2025, .. })) - ); + assert_eq!(goals.goal_count(), 1); + let recorded = goals.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + recorded[0], + DomainEvent::GoalCreated { + year: 2025, + target_count: 10, + .. + } + )); +} + +#[tokio::test] +async fn a_refused_create_propagates_and_records_nothing() { + let goals = InMemoryStore::new(); + let deps = GoalCommandDeps { + goal_command: Arc::new(RefusingGoalCommand), + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, + }; + + let result = create::execute(&deps, create_goal_command(2025, 10)).await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused")); + assert_eq!(goals.goal_count(), 0); + assert!(goals.recorded_events().is_empty()); } #[tokio::test] @@ -82,7 +98,6 @@ async fn rejects_duplicate_year() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let cmd = create_goal_command(2025, 10); @@ -100,7 +115,6 @@ async fn rejects_year_before_2020() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let result = create::execute(&deps, create_goal_command(2019, 10)).await; @@ -114,7 +128,6 @@ async fn rejects_zero_target() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let result = create::execute(&deps, create_goal_command(2025, 0)).await; diff --git a/crates/application/tests/goals/delete_test.rs b/crates/application/tests/goals/delete_test.rs index 86d8de0f..415d6fb1 100644 --- a/crates/application/tests/goals/delete_test.rs +++ b/crates/application/tests/goals/delete_test.rs @@ -1,7 +1,9 @@ use crate::support::create_goal_command; use std::sync::Arc; -use domain::testing::{FakeStatsRepository, InMemoryStore, NoopEventPublisher}; +use domain::errors::DomainError; +use domain::events::DomainEvent; +use domain::testing::{FakeStatsRepository, InMemoryStore, RefusingGoalCommand}; use uuid::Uuid; use crate::support::TestContextBuilder; @@ -12,12 +14,10 @@ use application::goals::{commands::DeleteGoalCommand, create, delete}; async fn deletes_existing_goal() { let goals = InMemoryStore::new(); let stats = FakeStatsRepository::new(); - let events = NoopEventPublisher::new(); let deps = GoalCommandDeps { goal_command: Arc::clone(&goals) as _, goal_query: Arc::clone(&goals) as _, stats: Arc::clone(&stats) as _, - event_publisher: Arc::clone(&events) as _, }; create::execute(&deps, create_goal_command(2025, 10)) @@ -45,7 +45,6 @@ async fn fails_when_not_found() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let result = delete::execute( &deps, @@ -58,3 +57,65 @@ async fn fails_when_not_found() { assert!(result.is_err()); } + +#[tokio::test] +async fn deleting_a_goal_removes_it_and_records_its_event_together() { + let goals = InMemoryStore::new(); + let deps = GoalCommandDeps { + goal_command: Arc::clone(&goals) as _, + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, + }; + create::execute(&deps, create_goal_command(2025, 10)) + .await + .unwrap(); + + delete::execute( + &deps, + DeleteGoalCommand { + user_id: Uuid::nil(), + year: 2025, + }, + ) + .await + .unwrap(); + + assert_eq!(goals.goal_count(), 0); + let recorded = goals.recorded_events(); + assert_eq!(recorded.len(), 2, "created + deleted"); + assert!(matches!( + recorded[1], + DomainEvent::GoalDeleted { year: 2025, .. } + )); +} + +#[tokio::test] +async fn a_refused_delete_propagates_and_records_nothing() { + let goals = InMemoryStore::new(); + let seed = GoalCommandDeps { + goal_command: Arc::clone(&goals) as _, + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, + }; + create::execute(&seed, create_goal_command(2025, 10)) + .await + .unwrap(); + let before = goals.recorded_events().len(); + let deps = GoalCommandDeps { + goal_command: Arc::new(RefusingGoalCommand), + ..seed + }; + + let result = delete::execute( + &deps, + DeleteGoalCommand { + user_id: Uuid::nil(), + year: 2025, + }, + ) + .await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused")); + assert_eq!(goals.goal_count(), 1); + assert_eq!(goals.recorded_events().len(), before); +} diff --git a/crates/application/tests/goals/get_test.rs b/crates/application/tests/goals/get_test.rs index a4562bdf..b001896e 100644 --- a/crates/application/tests/goals/get_test.rs +++ b/crates/application/tests/goals/get_test.rs @@ -12,7 +12,6 @@ async fn returns_goal_when_exists() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let query_deps = GoalQueryDeps { goal_query: b.goal_query.clone(), diff --git a/crates/application/tests/goals/list_test.rs b/crates/application/tests/goals/list_test.rs index 430c3d45..2ae07992 100644 --- a/crates/application/tests/goals/list_test.rs +++ b/crates/application/tests/goals/list_test.rs @@ -31,7 +31,6 @@ async fn returns_all_goals_for_user() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let query_deps = GoalQueryDeps { goal_query: b.goal_query.clone(), diff --git a/crates/application/tests/goals/update_test.rs b/crates/application/tests/goals/update_test.rs index 7d7bac10..48643cfe 100644 --- a/crates/application/tests/goals/update_test.rs +++ b/crates/application/tests/goals/update_test.rs @@ -1,5 +1,11 @@ use crate::support::{create_goal_command, update_goal_command}; +use std::sync::Arc; + +use domain::errors::DomainError; +use domain::events::DomainEvent; +use domain::testing::{FakeStatsRepository, InMemoryStore, RefusingGoalCommand}; + use crate::support::TestContextBuilder; use application::goals::deps::GoalCommandDeps; use application::goals::{create, update}; @@ -11,7 +17,6 @@ async fn updates_target_count() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; create::execute(&deps, create_goal_command(2025, 10)) @@ -32,7 +37,6 @@ async fn fails_when_goal_not_found() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let result = update::execute(&deps, update_goal_command(2025, 10)).await; @@ -46,7 +50,6 @@ async fn rejects_zero_target() { goal_command: b.goal_command.clone(), goal_query: b.goal_query.clone(), stats: b.stats_repo.clone(), - event_publisher: b.event_publisher.clone(), }; create::execute(&deps, create_goal_command(2025, 10)) @@ -57,3 +60,54 @@ async fn rejects_zero_target() { assert!(result.is_err()); } + +#[tokio::test] +async fn updating_a_goal_records_the_change_and_its_event_together() { + let goals = InMemoryStore::new(); + let deps = GoalCommandDeps { + goal_command: Arc::clone(&goals) as _, + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, + }; + create::execute(&deps, create_goal_command(2025, 10)) + .await + .unwrap(); + + update::execute(&deps, update_goal_command(2025, 100)) + .await + .unwrap(); + + let recorded = goals.recorded_events(); + assert_eq!(recorded.len(), 2, "created + updated"); + assert!(matches!( + recorded[1], + DomainEvent::GoalUpdated { + year: 2025, + target_count: 100, + .. + } + )); +} + +#[tokio::test] +async fn a_refused_update_propagates_and_records_nothing() { + let goals = InMemoryStore::new(); + let seed = GoalCommandDeps { + goal_command: Arc::clone(&goals) as _, + goal_query: Arc::clone(&goals) as _, + stats: FakeStatsRepository::new() as _, + }; + create::execute(&seed, create_goal_command(2025, 10)) + .await + .unwrap(); + let before = goals.recorded_events().len(); + let deps = GoalCommandDeps { + goal_command: Arc::new(RefusingGoalCommand), + ..seed + }; + + let result = update::execute(&deps, update_goal_command(2025, 100)).await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused")); + assert_eq!(goals.recorded_events().len(), before); +} diff --git a/crates/application/tests/users/get_profile_settings_test.rs b/crates/application/tests/users/get_profile_settings_test.rs index d8ca2c49..58496c12 100644 --- a/crates/application/tests/users/get_profile_settings_test.rs +++ b/crates/application/tests/users/get_profile_settings_test.rs @@ -37,6 +37,7 @@ async fn get_profile_settings_includes_profile_fields() { name: "pronouns".into(), value: "they/them".into(), }], + &[], ) .await .unwrap(); diff --git a/crates/application/tests/users/update_profile_fields_test.rs b/crates/application/tests/users/update_profile_fields_test.rs index eb00c1fa..c406829a 100644 --- a/crates/application/tests/users/update_profile_fields_test.rs +++ b/crates/application/tests/users/update_profile_fields_test.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use domain::events::DomainEvent; use domain::models::ProfileField; -use domain::testing::{InMemoryStore, NoopEventPublisher}; +use domain::testing::{InMemoryStore, RefusingProfileFieldsRepository}; use uuid::Uuid; use application::users::{ @@ -11,15 +11,10 @@ use application::users::{ }; #[tokio::test] -async fn saves_profile_fields() { +async fn saving_profile_fields_records_the_change_and_its_event_together() { let fields_repo = InMemoryStore::new(); - let events = NoopEventPublisher::new(); - let b = TestContextBuilder::new() - .with_profile_fields(Arc::clone(&fields_repo) as _) - .with_event_publisher(Arc::clone(&events) as _); let deps = UpdateProfileFieldsDeps { - profile_fields: b.profile_fields_repo.clone(), - event_publisher: b.event_publisher.clone(), + profile_fields: Arc::clone(&fields_repo) as _, }; update_profile_fields::execute( @@ -41,12 +36,9 @@ async fn saves_profile_fields() { .await .unwrap(); - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, DomainEvent::UserUpdated { .. })) - ); + let recorded = fields_repo.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!(recorded[0], DomainEvent::UserUpdated { .. })); } #[tokio::test] @@ -54,7 +46,6 @@ async fn rejects_more_than_four_fields() { let b = TestContextBuilder::new(); let deps = UpdateProfileFieldsDeps { profile_fields: b.profile_fields_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let fields: Vec = (0..5) @@ -75,3 +66,24 @@ async fn rejects_more_than_four_fields() { assert!(result.is_err()); } + +#[tokio::test] +async fn refused_profile_fields_propagate_and_record_nothing() { + let deps = UpdateProfileFieldsDeps { + profile_fields: Arc::new(RefusingProfileFieldsRepository), + }; + + let result = update_profile_fields::execute( + &deps, + UpdateProfileFieldsCommand { + user_id: Uuid::nil(), + fields: vec![], + }, + ) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::InfrastructureError(m)) if m == "refused" + )); +} diff --git a/crates/application/tests/users/update_profile_test.rs b/crates/application/tests/users/update_profile_test.rs index 14540b62..7bde7f73 100644 --- a/crates/application/tests/users/update_profile_test.rs +++ b/crates/application/tests/users/update_profile_test.rs @@ -31,7 +31,7 @@ async fn register_user(b: &TestContextBuilder, users: &Arc) -> Uu } #[tokio::test] -async fn updates_display_name() { +async fn updating_the_profile_records_the_change_and_its_event_together() { let users = InMemoryStore::new(); let events = NoopEventPublisher::new(); let b = TestContextBuilder::new() @@ -54,12 +54,9 @@ async fn updates_display_name() { .await .unwrap(); - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, DomainEvent::UserUpdated { .. })) - ); + let recorded = users.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!(recorded[0], DomainEvent::UserUpdated { .. })); } #[tokio::test] @@ -111,14 +108,15 @@ async fn uploads_avatar() { .await .unwrap(); - let published = events.published(); assert!( - published + users + .recorded_events() .iter() .any(|e| matches!(e, DomainEvent::UserUpdated { .. })) ); assert!( - published + events + .published() .iter() .any(|e| matches!(e, DomainEvent::ImageStored { .. })) ); @@ -149,14 +147,15 @@ async fn uploads_banner() { .await .unwrap(); - let published = events.published(); assert!( - published + users + .recorded_events() .iter() .any(|e| matches!(e, DomainEvent::UserUpdated { .. })) ); assert!( - published + events + .published() .iter() .any(|e| matches!(e, DomainEvent::ImageStored { .. })) ); @@ -232,16 +231,114 @@ async fn text_only_update_emits_user_updated_no_image_stored() { .await .unwrap(); - let published = events.published(); assert!( - published + users + .recorded_events() .iter() .any(|e| matches!(e, DomainEvent::UserUpdated { .. })) ); assert!( - !published + !events + .published() .iter() .any(|e| matches!(e, DomainEvent::ImageStored { .. })), "text-only update should not emit ImageStored" ); } + +#[tokio::test] +async fn changing_also_known_as_records_the_move_with_the_update() { + let users = InMemoryStore::new(); + let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _); + let deps = UpdateProfileDeps { + user: b.user_repo.clone(), + object_storage: b.object_storage.clone(), + event_publisher: b.event_publisher.clone(), + }; + let uid = register_user(&b, &users).await; + + update_profile::execute( + &deps, + UpdateProfileCommand { + also_known_as: Some("https://other.example/users/alice".into()), + ..update_profile_command(uid) + }, + ) + .await + .unwrap(); + + let recorded = users.recorded_events(); + assert_eq!(recorded.len(), 2); + assert!(matches!(recorded[0], DomainEvent::UserUpdated { .. })); + assert!(matches!(recorded[1], DomainEvent::UserAccountMoved { .. })); +} + +/// Loads from the store but refuses the profile write. +struct ProfileWriteRefusingUsers(Arc); + +#[async_trait::async_trait] +impl UserRepository for ProfileWriteRefusingUsers { + async fn find_by_email( + &self, + e: &domain::value_objects::Email, + ) -> Result, domain::errors::DomainError> { + self.0.find_by_email(e).await + } + async fn find_by_username( + &self, + u: &domain::value_objects::Username, + ) -> Result, domain::errors::DomainError> { + self.0.find_by_username(u).await + } + async fn save(&self, u: &domain::models::User) -> Result<(), domain::errors::DomainError> { + self.0.save(u).await + } + async fn find_by_id( + &self, + id: &domain::value_objects::UserId, + ) -> Result, domain::errors::DomainError> { + self.0.find_by_id(id).await + } + async fn list_with_stats( + &self, + ) -> Result, domain::errors::DomainError> { + self.0.list_with_stats().await + } + async fn update_profile( + &self, + _: &domain::value_objects::UserId, + _: &domain::models::UserProfile, + _: &[DomainEvent], + ) -> Result<(), domain::errors::DomainError> { + Err(domain::errors::DomainError::InfrastructureError( + "refused".into(), + )) + } +} + +#[tokio::test] +async fn a_refused_profile_update_propagates_and_records_nothing() { + let users = InMemoryStore::new(); + let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _); + let uid = register_user(&b, &users).await; + let deps = UpdateProfileDeps { + user: Arc::new(ProfileWriteRefusingUsers(Arc::clone(&users))), + object_storage: b.object_storage.clone(), + event_publisher: b.event_publisher.clone(), + }; + + let result = update_profile::execute( + &deps, + UpdateProfileCommand { + display_name: Some("Alice W.".into()), + ..update_profile_command(uid) + }, + ) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::InfrastructureError(m)) if m == "refused" + )); + assert!(users.recorded_events().is_empty()); +} diff --git a/crates/application/tests/watchlist/add_test.rs b/crates/application/tests/watchlist/add_test.rs index 20c3572c..995835d4 100644 --- a/crates/application/tests/watchlist/add_test.rs +++ b/crates/application/tests/watchlist/add_test.rs @@ -3,8 +3,9 @@ use domain::testing::test_movie_with_year; use std::sync::Arc; use domain::{ + events::DomainEvent, ports::MovieCommand, - testing::{InMemoryStore, NoopEventPublisher}, + testing::{InMemoryStore, NoopEventPublisher, RefusingWatchlistRepository}, }; use application::watchlist::{add, commands::AddToWatchlistCommand, deps::WatchlistAddDeps}; @@ -108,3 +109,50 @@ async fn adding_the_same_movie_to_a_watchlist_twice_keeps_a_single_entry_and_suc assert_eq!(movies.watchlist_count(), 1); } + +#[tokio::test] +async fn adding_an_entry_records_the_entry_and_its_event_together() { + let movies = InMemoryStore::new(); + let movie = test_movie_with_year("Heat", 1995); + movies.upsert_movie(&movie).await.unwrap(); + let deps = make_deps(Arc::clone(&movies), Arc::clone(&movies)); + + add::execute( + &deps, + add_to_watchlist_command(uuid::Uuid::new_v4(), movie.id().value()), + ) + .await + .unwrap(); + + assert_eq!(movies.watchlist_count(), 1); + let recorded = movies.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + recorded[0], + DomainEvent::WatchlistEntryAdded { .. } + )); +} + +#[tokio::test] +async fn a_refused_add_propagates_and_records_nothing() { + let movies = InMemoryStore::new(); + let movie = test_movie_with_year("Heat", 1995); + movies.upsert_movie(&movie).await.unwrap(); + let deps = WatchlistAddDeps { + watchlist: Arc::new(RefusingWatchlistRepository), + ..make_deps(Arc::clone(&movies), Arc::clone(&movies)) + }; + + let result = add::execute( + &deps, + add_to_watchlist_command(uuid::Uuid::new_v4(), movie.id().value()), + ) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::InfrastructureError(m)) if m == "refused" + )); + assert_eq!(movies.watchlist_count(), 0); + assert!(movies.recorded_events().is_empty()); +} diff --git a/crates/application/tests/watchlist/is_on_test.rs b/crates/application/tests/watchlist/is_on_test.rs index 2b59f127..55650397 100644 --- a/crates/application/tests/watchlist/is_on_test.rs +++ b/crates/application/tests/watchlist/is_on_test.rs @@ -15,10 +15,10 @@ async fn returns_true_when_present() { let uid = Uuid::new_v4(); let mid = Uuid::new_v4(); watchlist - .add(&WatchlistEntry::new( - UserId::from_uuid(uid), - MovieId::from_uuid(mid), - )) + .add( + &WatchlistEntry::new(UserId::from_uuid(uid), MovieId::from_uuid(mid)), + &[], + ) .await .unwrap(); diff --git a/crates/application/tests/watchlist/remove_test.rs b/crates/application/tests/watchlist/remove_test.rs index 0616e298..d0e85fe8 100644 --- a/crates/application/tests/watchlist/remove_test.rs +++ b/crates/application/tests/watchlist/remove_test.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use domain::events::DomainEvent; use domain::models::WatchlistEntry; use domain::ports::WatchlistRepository; -use domain::testing::{InMemoryStore, NoopEventPublisher}; +use domain::testing::{InMemoryStore, RefusingWatchlistRepository}; use domain::value_objects::{MovieId, UserId}; use uuid::Uuid; @@ -13,22 +13,20 @@ use application::watchlist::{ }; #[tokio::test] -async fn removes_entry_and_emits_event() { +async fn removing_an_entry_records_the_removal_and_its_event_together() { let watchlist = InMemoryStore::new(); - let events = NoopEventPublisher::new(); let uid = Uuid::new_v4(); let mid = Uuid::new_v4(); watchlist - .add(&WatchlistEntry::new( - UserId::from_uuid(uid), - MovieId::from_uuid(mid), - )) + .add( + &WatchlistEntry::new(UserId::from_uuid(uid), MovieId::from_uuid(mid)), + &[], + ) .await .unwrap(); let deps = RemoveFromWatchlistDeps { watchlist: Arc::clone(&watchlist) as _, - event_publisher: Arc::clone(&events) as _, }; remove::execute( &deps, @@ -41,12 +39,12 @@ async fn removes_entry_and_emits_event() { .unwrap(); assert_eq!(watchlist.watchlist_count(), 0); - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, DomainEvent::WatchlistEntryRemoved { .. })) - ); + let recorded = watchlist.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + recorded[0], + DomainEvent::WatchlistEntryRemoved { .. } + )); } #[tokio::test] @@ -54,7 +52,6 @@ async fn fails_when_not_on_watchlist() { let b = TestContextBuilder::new(); let deps = RemoveFromWatchlistDeps { watchlist: b.watchlist_repo.clone(), - event_publisher: b.event_publisher.clone(), }; let result = remove::execute( &deps, @@ -67,3 +64,44 @@ async fn fails_when_not_on_watchlist() { assert!(result.is_err()); } + +#[tokio::test] +async fn a_failed_removal_records_no_event() { + let store = InMemoryStore::new(); + let deps = RemoveFromWatchlistDeps { + watchlist: Arc::clone(&store) as _, + }; + + let result = remove::execute( + &deps, + RemoveFromWatchlistCommand { + user_id: Uuid::new_v4(), + movie_id: Uuid::new_v4(), + }, + ) + .await; + + assert!(result.is_err()); + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn a_refused_remove_propagates_the_refusal() { + let deps = RemoveFromWatchlistDeps { + watchlist: Arc::new(RefusingWatchlistRepository), + }; + + let result = remove::execute( + &deps, + RemoveFromWatchlistCommand { + user_id: Uuid::new_v4(), + movie_id: Uuid::new_v4(), + }, + ) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::InfrastructureError(m)) if m == "refused" + )); +} diff --git a/crates/bootstrap/src/build.rs b/crates/bootstrap/src/build.rs index 90fd7f6f..9003d371 100644 --- a/crates/bootstrap/src/build.rs +++ b/crates/bootstrap/src/build.rs @@ -125,7 +125,6 @@ pub fn build_deps( goal_command: repos.goal_command.clone(), goal_query: repos.goal_query.clone(), stats: repos.stats.clone(), - event_publisher: services.event_publisher.clone(), }, query: GoalQueryDeps { goal_query: repos.goal_query.clone(), @@ -294,7 +293,6 @@ pub fn build_deps( }, update_profile_fields: UpdateProfileFieldsDeps { profile_fields: repos.profile_fields.clone(), - event_publisher: services.event_publisher.clone(), }, get_settings: GetSettingsDeps { user_settings: repos.user_settings.clone(), @@ -330,7 +328,6 @@ pub fn build_deps( }, remove_from_watchlist: RemoveFromWatchlistDeps { watchlist: repos.watchlist.clone(), - event_publisher: services.event_publisher.clone(), }, }, wrapup: WrapupGroup { diff --git a/crates/bootstrap/tests/http_handlers/support.rs b/crates/bootstrap/tests/http_handlers/support.rs index 5d4715d9..b5477228 100644 --- a/crates/bootstrap/tests/http_handlers/support.rs +++ b/crates/bootstrap/tests/http_handlers/support.rs @@ -223,6 +223,7 @@ impl UserRepository for ErroringUserRepo { &self, _: &UserId, _: &domain::models::UserProfile, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { panic!() } diff --git a/crates/domain/src/ports/auth.rs b/crates/domain/src/ports/auth.rs index a1e95bfd..b4337cb2 100644 --- a/crates/domain/src/ports/auth.rs +++ b/crates/domain/src/ports/auth.rs @@ -23,6 +23,7 @@ pub trait UserRepository: Send + Sync { &self, user_id: &UserId, profile: &crate::models::UserProfile, + events: &[crate::events::DomainEvent], ) -> Result<(), DomainError>; } @@ -36,6 +37,7 @@ pub trait UserProfileFieldsRepository: Send + Sync { &self, user_id: &UserId, fields: Vec, + events: &[crate::events::DomainEvent], ) -> Result<(), DomainError>; } diff --git a/crates/domain/src/ports/goals.rs b/crates/domain/src/ports/goals.rs index c7691f45..17cb63ee 100644 --- a/crates/domain/src/ports/goals.rs +++ b/crates/domain/src/ports/goals.rs @@ -2,15 +2,21 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, models::Goal, value_objects::{GoalId, UserId}, }; #[async_trait] pub trait GoalCommand: Send + Sync { - async fn save(&self, goal: &Goal) -> Result<(), DomainError>; - async fn update(&self, goal: &Goal) -> Result<(), DomainError>; - async fn delete(&self, id: &GoalId, user_id: &UserId) -> Result<(), DomainError>; + async fn save(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError>; + async fn update(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError>; + async fn delete( + &self, + id: &GoalId, + user_id: &UserId, + events: &[DomainEvent], + ) -> Result<(), DomainError>; } #[async_trait] diff --git a/crates/domain/src/ports/watchlist.rs b/crates/domain/src/ports/watchlist.rs index ff87adc4..7fe6ace9 100644 --- a/crates/domain/src/ports/watchlist.rs +++ b/crates/domain/src/ports/watchlist.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, models::{ WatchlistEntry, WatchlistWithMovie, collections::{PageParams, Paginated}, @@ -11,14 +12,21 @@ use crate::{ #[async_trait] pub trait WatchlistRepository: Send + Sync { - async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError>; + async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError>; - async fn remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError>; + async fn remove( + &self, + user_id: &UserId, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError>; + /// `events` are recorded only when an entry was actually removed. async fn remove_if_present( &self, user_id: &UserId, movie_id: &MovieId, + events: &[DomainEvent], ) -> Result; async fn get_for_user( diff --git a/crates/domain/src/testing/fakes/noops.rs b/crates/domain/src/testing/fakes/noops.rs index 4a633de4..6d6d5aaf 100644 --- a/crates/domain/src/testing/fakes/noops.rs +++ b/crates/domain/src/testing/fakes/noops.rs @@ -74,16 +74,17 @@ pub struct NoopGoalCommand; #[async_trait] impl crate::ports::GoalCommand for NoopGoalCommand { - async fn save(&self, _: &crate::models::Goal) -> Result<(), DomainError> { + async fn save(&self, _: &crate::models::Goal, _: &[DomainEvent]) -> Result<(), DomainError> { Ok(()) } - async fn update(&self, _: &crate::models::Goal) -> Result<(), DomainError> { + async fn update(&self, _: &crate::models::Goal, _: &[DomainEvent]) -> Result<(), DomainError> { Ok(()) } async fn delete( &self, _: &crate::value_objects::GoalId, _: &UserId, + _: &[DomainEvent], ) -> Result<(), DomainError> { Ok(()) } diff --git a/crates/domain/src/testing/fakes/panics.rs b/crates/domain/src/testing/fakes/panics.rs index fa9dd781..2fc6a186 100644 --- a/crates/domain/src/testing/fakes/panics.rs +++ b/crates/domain/src/testing/fakes/panics.rs @@ -295,13 +295,18 @@ pub struct PanicWatchlistRepository; #[async_trait] impl WatchlistRepository for PanicWatchlistRepository { - async fn add(&self, _: &WatchlistEntry) -> Result<(), DomainError> { + async fn add(&self, _: &WatchlistEntry, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicWatchlistRepository called") } - async fn remove(&self, _: &UserId, _: &MovieId) -> Result<(), DomainError> { + async fn remove(&self, _: &UserId, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicWatchlistRepository called") } - async fn remove_if_present(&self, _: &UserId, _: &MovieId) -> Result { + async fn remove_if_present( + &self, + _: &UserId, + _: &MovieId, + _: &[DomainEvent], + ) -> Result { panic!("PanicWatchlistRepository called") } async fn get_for_user( @@ -357,6 +362,7 @@ impl UserProfileFieldsRepository for PanicProfileFieldsRepo { &self, _: &UserId, _: Vec, + _: &[DomainEvent], ) -> Result<(), DomainError> { panic!("PanicProfileFieldsRepo called") } @@ -645,6 +651,7 @@ impl UserRepository for PanicUserRepository { &self, _: &UserId, _: &crate::models::UserProfile, + _: &[DomainEvent], ) -> Result<(), DomainError> { panic!("PanicUserRepository called") } @@ -675,16 +682,17 @@ pub struct PanicGoalCommand; #[async_trait] impl crate::ports::GoalCommand for PanicGoalCommand { - async fn save(&self, _: &crate::models::Goal) -> Result<(), DomainError> { + async fn save(&self, _: &crate::models::Goal, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicGoalCommand called") } - async fn update(&self, _: &crate::models::Goal) -> Result<(), DomainError> { + async fn update(&self, _: &crate::models::Goal, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicGoalCommand called") } async fn delete( &self, _: &crate::value_objects::GoalId, _: &crate::value_objects::UserId, + _: &[DomainEvent], ) -> Result<(), DomainError> { panic!("PanicGoalCommand called") } diff --git a/crates/domain/src/testing/fakes/refusing.rs b/crates/domain/src/testing/fakes/refusing.rs index 9ea36bef..70ac8e6e 100644 --- a/crates/domain/src/testing/fakes/refusing.rs +++ b/crates/domain/src/testing/fakes/refusing.rs @@ -6,11 +6,14 @@ use crate::{ errors::DomainError, events::DomainEvent, models::{ - Movie, Review, WatchlistEntry, WatchlistWithMovie, + Goal, Movie, ProfileField, Review, WatchlistEntry, WatchlistWithMovie, collections::{PageParams, Paginated}, }, - ports::{EventPublisher, MovieCommand, ReviewRepository, WatchlistRepository}, - value_objects::{MovieId, ReviewId, UserId}, + ports::{ + EventPublisher, GoalCommand, MovieCommand, ReviewRepository, UserProfileFieldsRepository, + WatchlistRepository, + }, + value_objects::{GoalId, MovieId, ReviewId, UserId}, }; fn refused() -> DomainError { @@ -54,13 +57,18 @@ pub struct RefusingWatchlistRepository; #[async_trait] impl WatchlistRepository for RefusingWatchlistRepository { - async fn add(&self, _: &WatchlistEntry) -> Result<(), DomainError> { + async fn add(&self, _: &WatchlistEntry, _: &[DomainEvent]) -> Result<(), DomainError> { Err(refused()) } - async fn remove(&self, _: &UserId, _: &MovieId) -> Result<(), DomainError> { + async fn remove(&self, _: &UserId, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> { Err(refused()) } - async fn remove_if_present(&self, _: &UserId, _: &MovieId) -> Result { + async fn remove_if_present( + &self, + _: &UserId, + _: &MovieId, + _: &[DomainEvent], + ) -> Result { Err(refused()) } async fn get_for_user( @@ -75,6 +83,38 @@ impl WatchlistRepository for RefusingWatchlistRepository { } } +pub struct RefusingGoalCommand; + +#[async_trait] +impl GoalCommand for RefusingGoalCommand { + async fn save(&self, _: &Goal, _: &[DomainEvent]) -> Result<(), DomainError> { + Err(refused()) + } + async fn update(&self, _: &Goal, _: &[DomainEvent]) -> Result<(), DomainError> { + Err(refused()) + } + async fn delete(&self, _: &GoalId, _: &UserId, _: &[DomainEvent]) -> Result<(), DomainError> { + Err(refused()) + } +} + +pub struct RefusingProfileFieldsRepository; + +#[async_trait] +impl UserProfileFieldsRepository for RefusingProfileFieldsRepository { + async fn get_fields(&self, _: &UserId) -> Result, DomainError> { + Err(refused()) + } + async fn set_fields( + &self, + _: &UserId, + _: Vec, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } +} + pub struct RefusingEventPublisher; #[async_trait] diff --git a/crates/domain/src/testing/fakes/store_goal.rs b/crates/domain/src/testing/fakes/store_goal.rs index c927457e..9d9e1845 100644 --- a/crates/domain/src/testing/fakes/store_goal.rs +++ b/crates/domain/src/testing/fakes/store_goal.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, models::Goal, ports::{GoalCommand, GoalQuery}, value_objects::{GoalId, UserId}, @@ -12,7 +13,7 @@ use super::InMemoryStore; #[async_trait] impl GoalCommand for InMemoryStore { /// `goals` is UNIQUE(user_id, year). - async fn save(&self, goal: &Goal) -> Result<(), DomainError> { + async fn save(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError> { let mut goals = self.goals.write().unwrap(); let duplicate = goals.values().any(|g| { g.id() == goal.id() || (g.user_id() == goal.user_id() && g.year() == goal.year()) @@ -23,14 +24,18 @@ impl GoalCommand for InMemoryStore { )); } goals.insert(goal.id().value(), goal.clone()); + drop(goals); + self.record_events(events); Ok(()) } - async fn update(&self, goal: &Goal) -> Result<(), DomainError> { + async fn update(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError> { let mut store = self.goals.write().unwrap(); match store.entry(goal.id().value()) { std::collections::hash_map::Entry::Occupied(mut e) => { e.insert(goal.clone()); + drop(store); + self.record_events(events); Ok(()) } std::collections::hash_map::Entry::Vacant(_) => { @@ -39,8 +44,14 @@ impl GoalCommand for InMemoryStore { } } - async fn delete(&self, id: &GoalId, _user_id: &UserId) -> Result<(), DomainError> { + async fn delete( + &self, + id: &GoalId, + _user_id: &UserId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { self.goals.write().unwrap().remove(&id.value()); + self.record_events(events); Ok(()) } } diff --git a/crates/domain/src/testing/fakes/store_user.rs b/crates/domain/src/testing/fakes/store_user.rs index 0d76498f..83f07176 100644 --- a/crates/domain/src/testing/fakes/store_user.rs +++ b/crates/domain/src/testing/fakes/store_user.rs @@ -84,11 +84,14 @@ impl UserRepository for InMemoryStore { &self, user_id: &UserId, profile: &crate::models::UserProfile, + events: &[crate::events::DomainEvent], ) -> Result<(), DomainError> { let mut store = self.users.write().unwrap(); if let Some(user) = store.get_mut(&user_id.value()) { user.update_profile(profile.clone()); } + drop(store); + self.record_events(events); Ok(()) } } @@ -104,11 +107,13 @@ impl UserProfileFieldsRepository for InMemoryStore { &self, user_id: &UserId, fields: Vec, + events: &[crate::events::DomainEvent], ) -> Result<(), DomainError> { self.profile_fields .write() .unwrap() .insert(user_id.value(), fields); + self.record_events(events); Ok(()) } } diff --git a/crates/domain/src/testing/fakes/store_watchlist.rs b/crates/domain/src/testing/fakes/store_watchlist.rs index 57c2a7ba..c19d6820 100644 --- a/crates/domain/src/testing/fakes/store_watchlist.rs +++ b/crates/domain/src/testing/fakes/store_watchlist.rs @@ -3,6 +3,7 @@ use uuid::Uuid; use crate::{ errors::DomainError, + events::DomainEvent, models::{ RemoteWatchlistEntry, WatchlistEntry, WatchlistWithMovie, collections::{PageParams, Paginated}, @@ -16,23 +17,30 @@ use super::InMemoryStore; #[async_trait] impl WatchlistRepository for InMemoryStore { /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op. - async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError> { + async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError> { let key = (entry.user_id.value(), entry.movie_id.value()); self.watchlist .write() .unwrap() .entry(key) .or_insert_with(|| entry.clone()); + self.record_events(events); Ok(()) } - async fn remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError> { + async fn remove( + &self, + user_id: &UserId, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let key = (user_id.value(), movie_id.value()); self.watchlist .write() .unwrap() .remove(&key) .ok_or_else(|| DomainError::NotFound("watchlist entry".into()))?; + self.record_events(events); Ok(()) } @@ -40,9 +48,14 @@ impl WatchlistRepository for InMemoryStore { &self, user_id: &UserId, movie_id: &MovieId, + events: &[DomainEvent], ) -> Result { let key = (user_id.value(), movie_id.value()); - Ok(self.watchlist.write().unwrap().remove(&key).is_some()) + let removed = self.watchlist.write().unwrap().remove(&key).is_some(); + if removed { + self.record_events(events); + } + Ok(removed) } async fn get_for_user( diff --git a/crates/domain/tests/testing/in_memory_store_test.rs b/crates/domain/tests/testing/in_memory_store_test.rs index ecab6148..a9f68af1 100644 --- a/crates/domain/tests/testing/in_memory_store_test.rs +++ b/crates/domain/tests/testing/in_memory_store_test.rs @@ -73,10 +73,13 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie store.upsert_movie(&movie).await.unwrap(); store - .add(&WatchlistEntry::new(user_id.clone(), movie.id().clone())) + .add( + &WatchlistEntry::new(user_id.clone(), movie.id().clone()), + &[], + ) .await .unwrap(); - GoalCommand::save(&**store, &test_goal(&user_id, 2024, 50)) + GoalCommand::save(&**store, &test_goal(&user_id, 2024, 50), &[]) .await .unwrap(); UserSettingsRepository::save(&**store, &UserSettings::new(user_id.clone())) @@ -89,6 +92,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie name: "Site".into(), value: "https://example.com".into(), }], + &[], ) .await .unwrap(); @@ -197,10 +201,10 @@ async fn deleting_a_user_leaves_other_users_rows_alone() { let bystander = test_user("carol"); UserRepository::save(&*store, &bystander).await.unwrap(); store - .add(&WatchlistEntry::new( - bystander.id().clone(), - movie.id().clone(), - )) + .add( + &WatchlistEntry::new(bystander.id().clone(), movie.id().clone()), + &[], + ) .await .unwrap(); @@ -219,7 +223,10 @@ async fn seed_movie_with_dependents( let movie = test_movie("Heat"); store.upsert_movie(&movie).await.unwrap(); store - .add(&WatchlistEntry::new(user.id().clone(), movie.id().clone())) + .add( + &WatchlistEntry::new(user.id().clone(), movie.id().clone()), + &[], + ) .await .unwrap(); store.upsert(&movie_profile(movie.id())).await.unwrap(); @@ -327,11 +334,11 @@ async fn saving_a_user_with_a_taken_email_is_a_validation_error() { async fn saving_a_second_goal_for_the_same_user_and_year_is_an_infrastructure_error() { let store = InMemoryStore::new(); let user_id = UserId::generate(); - GoalCommand::save(&*store, &test_goal(&user_id, 2024, 10)) + GoalCommand::save(&*store, &test_goal(&user_id, 2024, 10), &[]) .await .unwrap(); - let result = GoalCommand::save(&*store, &test_goal(&user_id, 2024, 20)).await; + let result = GoalCommand::save(&*store, &test_goal(&user_id, 2024, 20), &[]).await; assert!(matches!(result, Err(DomainError::InfrastructureError(_)))); assert_eq!(store.goal_count(), 1); @@ -344,11 +351,11 @@ async fn adding_the_same_movie_to_a_watchlist_twice_keeps_one_entry() { let movie_id = MovieId::generate(); store - .add(&WatchlistEntry::new(user_id.clone(), movie_id.clone())) + .add(&WatchlistEntry::new(user_id.clone(), movie_id.clone()), &[]) .await .unwrap(); store - .add(&WatchlistEntry::new(user_id, movie_id)) + .add(&WatchlistEntry::new(user_id, movie_id), &[]) .await .unwrap();