//! 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" ); }