feat: atomic goal, watchlist, profile writes

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();
+}

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<ProfileField> {
+    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"
+    );
+}

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);
+}

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();