diff --git a/crates/adapters/sqlite/src/repositories/movie/command.rs b/crates/adapters/sqlite/src/repositories/movie/command.rs index 79946f26..f417676d 100644 --- a/crates/adapters/sqlite/src/repositories/movie/command.rs +++ b/crates/adapters/sqlite/src/repositories/movie/command.rs @@ -1,11 +1,15 @@ use async_trait::async_trait; -use domain::{errors::DomainError, models::Movie, ports::MovieCommand, value_objects::MovieId}; +use domain::{ + errors::DomainError, events::DomainEvent, models::Movie, ports::MovieCommand, + value_objects::MovieId, +}; use super::SqliteMovieRepository; +use crate::shared::enqueue_events; #[async_trait] impl MovieCommand for SqliteMovieRepository { - async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> { + async fn upsert_movie(&self, movie: &Movie, events: &[DomainEvent]) -> Result<(), DomainError> { let id = movie.id().value().to_string(); let external_metadata_id = movie.external_metadata_id().map(|e| e.value().to_string()); let title = movie.title().value(); @@ -13,6 +17,11 @@ impl MovieCommand for SqliteMovieRepository { let director = movie.director(); let poster_path = movie.poster_path().map(|p| p.value().to_string()); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query( "INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path) VALUES (?, ?, ?, ?, ?, ?) @@ -29,20 +38,32 @@ impl MovieCommand for SqliteMovieRepository { .bind(release_year) .bind(director) .bind(&poster_path) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; - - Ok(()) + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } - async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> { + async fn delete_movie( + &self, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let id = movie_id.value().to_string(); - sqlx::query("DELETE FROM movies WHERE id = ?") - .bind(&id) - .execute(&self.pool) + let mut tx = self + .pool + .begin() .await .map_err(crate::shared::map_sqlx_error)?; - Ok(()) + let result = sqlx::query("DELETE FROM movies WHERE id = ?") + .bind(&id) + .execute(&mut *tx) + .await + .map_err(crate::shared::map_sqlx_error)?; + if result.rows_affected() > 0 { + enqueue_events(&mut tx, events).await?; + } + tx.commit().await.map_err(crate::shared::map_sqlx_error) } } diff --git a/crates/adapters/sqlite/src/repositories/watch_event/command.rs b/crates/adapters/sqlite/src/repositories/watch_event/command.rs index 4d20916d..a6c59f92 100644 --- a/crates/adapters/sqlite/src/repositories/watch_event/command.rs +++ b/crates/adapters/sqlite/src/repositories/watch_event/command.rs @@ -1,18 +1,19 @@ use async_trait::async_trait; use domain::{ errors::DomainError, + events::DomainEvent, models::{WatchEvent, WatchEventStatus}, ports::WatchEventCommand, value_objects::WatchEventId, }; -use crate::shared::datetime_to_str; +use crate::shared::{datetime_to_str, enqueue_events}; use super::SqliteWatchEventRepository; #[async_trait] impl WatchEventCommand for SqliteWatchEventRepository { - async fn save(&self, event: &WatchEvent) -> Result<(), DomainError> { + async fn save(&self, event: &WatchEvent, events: &[DomainEvent]) -> Result<(), DomainError> { let id = event.id().value().to_string(); let user_id = event.user_id().value().to_string(); let movie_id = event.movie_id().map(|m| m.value().to_string()); @@ -21,6 +22,11 @@ impl WatchEventCommand for SqliteWatchEventRepository { let status = event.status().to_string(); let created_at = datetime_to_str(event.created_at()); + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; sqlx::query( "INSERT INTO watch_events \ (id, user_id, movie_id, title, year, external_metadata_id, source, watched_at, status, created_at) \ @@ -36,11 +42,11 @@ impl WatchEventCommand for SqliteWatchEventRepository { .bind(&watched_at) .bind(&status) .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_status( diff --git a/crates/adapters/sqlite/src/repositories/wrapup/mod.rs b/crates/adapters/sqlite/src/repositories/wrapup/mod.rs index b694630e..4af56488 100644 --- a/crates/adapters/sqlite/src/repositories/wrapup/mod.rs +++ b/crates/adapters/sqlite/src/repositories/wrapup/mod.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use chrono::NaiveDate; use domain::{ errors::DomainError, + events::DomainEvent, models::wrapup::{WrapUpRecord, WrapUpReport, WrapUpStatus}, ports::WrapUpRepository, value_objects::WrapUpId, @@ -9,6 +10,7 @@ use domain::{ use sqlx::SqlitePool; use uuid::Uuid; +use crate::shared::enqueue_events; use rows::{row_to_record, status_to_str}; mod query; @@ -30,7 +32,11 @@ impl SqliteWrapUpRepository { #[async_trait] impl WrapUpRepository for SqliteWrapUpRepository { - async fn create(&self, record: &WrapUpRecord) -> Result<(), DomainError> { + async fn create( + &self, + record: &WrapUpRecord, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let id = record.id.value().to_string(); let user_id = record.user_id.map(|u| u.to_string()); let status = status_to_str(&record.status); @@ -41,6 +47,11 @@ impl WrapUpRepository for SqliteWrapUpRepository { .completed_at .map(|dt| dt.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 wrap_up_records \ (id, user_id, start_date, end_date, status, report_json, error_message, created_at, completed_at) \ @@ -55,11 +66,11 @@ impl WrapUpRepository for SqliteWrapUpRepository { .bind(&record.error_message) .bind(&created) .bind(&completed) - .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_status( @@ -82,23 +93,36 @@ impl WrapUpRepository for SqliteWrapUpRepository { Ok(()) } - async fn set_complete(&self, id: &WrapUpId, report: &WrapUpReport) -> Result<(), DomainError> { + async fn set_complete( + &self, + id: &WrapUpId, + report: &WrapUpReport, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let id_str = id.value().to_string(); let json = serde_json::to_string(report) .map_err(|e| DomainError::InfrastructureError(e.to_string()))?; - sqlx::query( + let mut tx = self + .pool + .begin() + .await + .map_err(crate::shared::map_sqlx_error)?; + let result = sqlx::query( "UPDATE wrap_up_records \ SET status = 'ready', report_json = ?, completed_at = strftime('%Y-%m-%d %H:%M:%S', 'now') \ WHERE id = ?", ) .bind(&json) .bind(&id_str) - .execute(&self.pool) + .execute(&mut *tx) .await .map_err(crate::shared::map_sqlx_error)?; - - Ok(()) + if result.rows_affected() == 0 { + return Err(DomainError::NotFound("wrapup record".into())); + } + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(crate::shared::map_sqlx_error) } async fn get_by_id(&self, id: &WrapUpId) -> Result, DomainError> { diff --git a/crates/adapters/sqlite/tests/cascade_test.rs b/crates/adapters/sqlite/tests/cascade_test.rs index 176ed7c6..eba3f58b 100644 --- a/crates/adapters/sqlite/tests/cascade_test.rs +++ b/crates/adapters/sqlite/tests/cascade_test.rs @@ -124,7 +124,7 @@ async fn seed_user_with_dependents( let user_id = user.id().clone(); let movie = test_movie("Heat"); parts.user.save(&user).await.unwrap(); - parts.movie_command.upsert_movie(&movie).await.unwrap(); + parts.movie_command.upsert_movie(&movie, &[]).await.unwrap(); parts .watchlist @@ -165,15 +165,18 @@ async fn seed_user_with_dependents( .await .unwrap(); SqliteWatchEventRepository::new(pool.clone()) - .save(&WatchEvent::new( - user_id.clone(), - "Heat".into(), - Some(1995), - None, - WatchEventSource::Jellyfin, - test_watched_at(), - None, - )) + .save( + &WatchEvent::new( + user_id.clone(), + "Heat".into(), + Some(1995), + None, + WatchEventSource::Jellyfin, + test_watched_at(), + None, + ), + &[], + ) .await .unwrap(); let social = SqliteSocialRepository::new( @@ -257,7 +260,7 @@ async fn deleting_a_movie_cascades_to_watchlist_entries_and_every_enrichment_tab let user = test_user("alice"); let movie = test_movie("Heat"); parts.user.save(&user).await.unwrap(); - parts.movie_command.upsert_movie(&movie).await.unwrap(); + parts.movie_command.upsert_movie(&movie, &[]).await.unwrap(); parts .watchlist .add( @@ -272,22 +275,29 @@ async fn deleting_a_movie_cascades_to_watchlist_entries_and_every_enrichment_tab .await .unwrap(); SqliteWatchEventRepository::new(pool.clone()) - .save(&WatchEvent::new( - user.id().clone(), - "Heat".into(), - Some(1995), - None, - WatchEventSource::Plex, - test_watched_at(), - Some(movie.id().clone()), - )) + .save( + &WatchEvent::new( + user.id().clone(), + "Heat".into(), + Some(1995), + None, + WatchEventSource::Plex, + test_watched_at(), + Some(movie.id().clone()), + ), + &[], + ) .await .unwrap(); for table in MOVIE_CASCADE_TABLES { assert!(count(&pool, table).await > 0, "{table} should be seeded"); } - parts.movie_command.delete_movie(movie.id()).await.unwrap(); + parts + .movie_command + .delete_movie(movie.id(), &[]) + .await + .unwrap(); for table in MOVIE_CASCADE_TABLES { assert_eq!(count(&pool, table).await, 0, "{table} should cascade"); @@ -306,14 +316,14 @@ async fn deleting_a_movie_that_has_reviews_is_refused_because_reviews_do_not_cas let user = test_user("alice"); let movie = test_movie("Heat"); parts.user.save(&user).await.unwrap(); - parts.movie_command.upsert_movie(&movie).await.unwrap(); + parts.movie_command.upsert_movie(&movie, &[]).await.unwrap(); parts .review .save_review(&test_review(movie.id(), user.id(), 5), &[]) .await .unwrap(); - let result = parts.movie_command.delete_movie(movie.id()).await; + let result = parts.movie_command.delete_movie(movie.id(), &[]).await; assert!(matches!( result, diff --git a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs index bfdd1f59..ff145b0a 100644 --- a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs +++ b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs @@ -4,7 +4,8 @@ use domain::{ events::DomainEvent, models::{ProfileField, UserProfile, WatchlistEntry}, ports::{ - GoalCommand, MovieCommand, UserProfileFieldsRepository, UserRepository, WatchlistRepository, + GoalCommand, MovieCommand, UserProfileFieldsRepository, UserRepository, WatchEventCommand, + WatchlistRepository, WrapUpRepository, }, testing::{test_goal, test_movie, test_user}, value_objects::UserId, @@ -161,7 +162,7 @@ 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) + .upsert_movie(&movie, &[]) .await .unwrap(); (pool, user.id().clone(), movie) @@ -419,3 +420,295 @@ async fn a_failing_enqueue_rolls_back_the_profile_fields_replacement() { "delete of the old fields rolled back" ); } + +fn movie_discovered(movie: &domain::models::Movie) -> DomainEvent { + DomainEvent::MovieDiscovered { + movie_id: movie.id().clone(), + external_metadata_id: domain::value_objects::ExternalMetadataId::new("tmdb:1".into()) + .unwrap(), + } +} + +fn movie_deleted(movie: &domain::models::Movie) -> DomainEvent { + DomainEvent::MovieDeleted { + movie_id: movie.id().clone(), + poster_path: None, + } +} + +#[tokio::test] +async fn upserting_a_movie_writes_the_movie_and_its_event_in_one_commit() { + let pool = migrated_pool().await; + let movie = test_movie("Heat"); + + SqliteMovieRepository::new(pool.clone()) + .upsert_movie(&movie, &[movie_discovered(&movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "movies").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_movie_upsert_of_a_new_movie() { + let pool = migrated_pool().await; + let movie = test_movie("Heat"); + fail_event_inserts(&pool).await; + + let result = SqliteMovieRepository::new(pool.clone()) + .upsert_movie(&movie, &[movie_discovered(&movie)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "movies").await, 0); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_a_movie_update() { + let pool = migrated_pool().await; + let repo = SqliteMovieRepository::new(pool.clone()); + let mut movie = test_movie("Heat"); + repo.upsert_movie(&movie, &[]).await.unwrap(); + movie.update_poster(domain::value_objects::PosterPath::new("posters/new.jpg".into()).unwrap()); + fail_event_inserts(&pool).await; + + let result = repo.upsert_movie(&movie, &[movie_discovered(&movie)]).await; + + assert!(result.is_err()); + let poster: Option = sqlx::query_scalar("SELECT poster_path FROM movies") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(poster, None, "poster update rolled back"); +} + +#[tokio::test] +async fn deleting_a_movie_writes_the_removal_and_its_event_in_one_commit() { + let pool = migrated_pool().await; + let repo = SqliteMovieRepository::new(pool.clone()); + let movie = test_movie("Heat"); + repo.upsert_movie(&movie, &[]).await.unwrap(); + + repo.delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "movies").await, 0); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_movie_delete() { + let pool = migrated_pool().await; + let repo = SqliteMovieRepository::new(pool.clone()); + let movie = test_movie("Heat"); + repo.upsert_movie(&movie, &[]).await.unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "movies").await, 1, "delete rolled back"); +} + +#[tokio::test] +async fn deleting_an_absent_movie_enqueues_no_event() { + let pool = migrated_pool().await; + let movie = test_movie("Heat"); + + SqliteMovieRepository::new(pool.clone()) + .delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "event_queue").await, 0); +} + +fn wrapup_record(user_id: Option) -> domain::models::wrapup::WrapUpRecord { + use domain::models::wrapup::WrapUpStatus; + domain::models::wrapup::WrapUpRecord { + id: domain::value_objects::WrapUpId::generate(), + user_id, + start_date: chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + end_date: chrono::NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(), + status: WrapUpStatus::Pending, + report: None, + error_message: None, + created_at: chrono::Utc::now().naive_utc(), + completed_at: None, + } +} + +fn wrapup_requested(record: &domain::models::wrapup::WrapUpRecord) -> DomainEvent { + DomainEvent::WrapUpRequested { + wrapup_id: record.id.clone(), + user_id: None, + start_date: record.start_date, + end_date: record.end_date, + } +} + +fn wrapup_completed(record: &domain::models::wrapup::WrapUpRecord) -> DomainEvent { + DomainEvent::WrapUpCompleted { + wrapup_id: record.id.clone(), + } +} + +fn empty_report() -> domain::models::wrapup::WrapUpReport { + serde_json::from_value(serde_json::json!({ + "scope": "Global", + "date_range": { "start": "2024-01-01", "end": "2024-12-31" }, + "generated_at": "2025-01-01T00:00:00Z", + "total_movies": 0, + "total_watch_time_minutes": 0, + "movies_per_month": [], + "rating_distribution": [0, 0, 0, 0, 0], + "top_directors": [], + "top_actors": [], + "director_diversity": 0, + "actor_diversity": 0, + "top_genres": [], + "genre_diversity": 0, + "top_keywords": [], + "language_distribution": [], + "watch_medium_distribution": [], + "total_rewatches": 0, + "poster_paths": [], + "top_cast_profile_paths": [], + })) + .expect("minimal wrap-up report") +} + +#[tokio::test] +async fn creating_a_wrapup_writes_the_record_and_its_event_in_one_commit() { + let pool = migrated_pool().await; + let record = wrapup_record(None); + + sqlite::SqliteWrapUpRepository::new(pool.clone()) + .create(&record, &[wrapup_requested(&record)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "wrap_up_records").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_wrapup_create() { + let pool = migrated_pool().await; + let record = wrapup_record(None); + fail_event_inserts(&pool).await; + + let result = sqlite::SqliteWrapUpRepository::new(pool.clone()) + .create(&record, &[wrapup_requested(&record)]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "wrap_up_records").await, 0); +} + +#[tokio::test] +async fn completing_a_wrapup_writes_the_report_and_its_event_in_one_commit() { + let pool = migrated_pool().await; + let repo = sqlite::SqliteWrapUpRepository::new(pool.clone()); + let record = wrapup_record(None); + repo.create(&record, &[]).await.unwrap(); + + repo.set_complete(&record.id, &empty_report(), &[wrapup_completed(&record)]) + .await + .unwrap(); + + let status: String = sqlx::query_scalar("SELECT status FROM wrap_up_records") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(status, "ready"); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_wrapup_completion() { + let pool = migrated_pool().await; + let repo = sqlite::SqliteWrapUpRepository::new(pool.clone()); + let record = wrapup_record(None); + repo.create(&record, &[]).await.unwrap(); + fail_event_inserts(&pool).await; + + let result = repo + .set_complete(&record.id, &empty_report(), &[wrapup_completed(&record)]) + .await; + + assert!(result.is_err()); + let status: String = sqlx::query_scalar("SELECT status FROM wrap_up_records") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(status, "pending", "completion rolled back"); +} + +#[tokio::test] +async fn completing_a_missing_wrapup_is_not_found_and_enqueues_no_event() { + let pool = migrated_pool().await; + let record = wrapup_record(None); + + let result = sqlite::SqliteWrapUpRepository::new(pool.clone()) + .set_complete(&record.id, &empty_report(), &[wrapup_completed(&record)]) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::NotFound(_)) + )); + assert_eq!(count(&pool, "event_queue").await, 0); +} + +fn watch_event_for(user_id: &UserId) -> domain::models::WatchEvent { + domain::models::WatchEvent::new( + user_id.clone(), + "Heat".into(), + Some(1995), + None, + domain::models::WatchEventSource::Jellyfin, + chrono::Utc::now().naive_utc(), + None, + ) +} + +fn watch_event_ingested(user_id: &UserId) -> DomainEvent { + DomainEvent::WatchEventIngested { + user_id: user_id.clone(), + title: "Heat".into(), + source: "jellyfin".into(), + } +} + +#[tokio::test] +async fn saving_a_watch_event_writes_the_row_and_its_event_in_one_commit() { + let (pool, user) = pool_with_user().await; + let event = watch_event_for(user.id()); + + sqlite::SqliteWatchEventRepository::new(pool.clone()) + .save(&event, &[watch_event_ingested(user.id())]) + .await + .unwrap(); + + assert_eq!(count(&pool, "watch_events").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_the_watch_event_save() { + let (pool, user) = pool_with_user().await; + let event = watch_event_for(user.id()); + fail_event_inserts(&pool).await; + + let result = sqlite::SqliteWatchEventRepository::new(pool.clone()) + .save(&event, &[watch_event_ingested(user.id())]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "watch_events").await, 0); +} diff --git a/crates/adapters/sqlite/tests/repositories/review_repository_test.rs b/crates/adapters/sqlite/tests/repositories/review_repository_test.rs index 557899e6..b8be38ee 100644 --- a/crates/adapters/sqlite/tests/repositories/review_repository_test.rs +++ b/crates/adapters/sqlite/tests/repositories/review_repository_test.rs @@ -28,7 +28,7 @@ async fn saving_a_review_writes_the_review_and_its_event_in_one_commit() { .await .unwrap(); SqliteMovieRepository::new(pool.clone()) - .upsert_movie(&movie) + .upsert_movie(&movie, &[]) .await .unwrap(); let review = test_review(movie.id(), user.id(), 4); @@ -79,7 +79,7 @@ async fn seeded() -> (SqlitePool, domain::models::Review) { .await .unwrap(); SqliteMovieRepository::new(pool.clone()) - .upsert_movie(&movie) + .upsert_movie(&movie, &[]) .await .unwrap(); let review = test_review(movie.id(), user.id(), 4); diff --git a/crates/application/src/diary/delete_review.rs b/crates/application/src/diary/delete_review.rs index 3a67b26d..f3840a1a 100644 --- a/crates/application/src/diary/delete_review.rs +++ b/crates/application/src/diary/delete_review.rs @@ -33,18 +33,15 @@ pub async fn execute(deps: &DeleteReviewDeps, cmd: DeleteReviewCommand) -> Resul let history = deps.diary.get_review_history(&movie_id).await?; if history.viewings().is_empty() { let poster_path = history.movie().poster_path().cloned(); - deps.movie_command.delete_movie(&movie_id).await?; - // best-effort: movie is already deleted, so publish failure is non-fatal - if let Err(e) = deps - .event_publisher - .publish(&DomainEvent::MovieDeleted { - movie_id, - poster_path, - }) - .await - { - tracing::warn!("failed to publish MovieDeleted event: {e}"); - } + deps.movie_command + .delete_movie( + &movie_id, + &[DomainEvent::MovieDeleted { + movie_id: movie_id.clone(), + poster_path, + }], + ) + .await?; } Ok(()) diff --git a/crates/application/src/diary/deps.rs b/crates/application/src/diary/deps.rs index bbd56789..d9613fbc 100644 --- a/crates/application/src/diary/deps.rs +++ b/crates/application/src/diary/deps.rs @@ -1,8 +1,8 @@ use std::sync::Arc; use domain::ports::{ - DiaryExporter, DiaryQuery, EventPublisher, FollowGraphQuery, MovieCommand, - MovieProfileRepository, MovieQuery, ReviewRepository, UserRepository, + DiaryExporter, DiaryQuery, FollowGraphQuery, MovieCommand, MovieProfileRepository, MovieQuery, + ReviewRepository, UserRepository, }; use crate::config::AppSettings; @@ -11,7 +11,6 @@ pub struct DeleteReviewDeps { pub review: Arc, pub diary: Arc, pub movie_command: Arc, - pub event_publisher: Arc, } pub struct EditReviewDeps { diff --git a/crates/application/src/diary/review_logger.rs b/crates/application/src/diary/review_logger.rs index 9a31fa07..7d95fd3c 100644 --- a/crates/application/src/diary/review_logger.rs +++ b/crates/application/src/diary/review_logger.rs @@ -5,10 +5,7 @@ use domain::{ errors::DomainError, events::DomainEvent, models::Review, - ports::{ - EventPublisher, MetadataClient, MovieCommand, MovieQuery, ReviewRepository, - WatchlistRepository, - }, + ports::{MetadataClient, MovieCommand, MovieQuery, ReviewRepository, WatchlistRepository}, value_objects::{Comment, Rating, UserId}, }; @@ -22,7 +19,6 @@ pub struct DefaultReviewLogger { review_repo: Arc, watchlist_repo: Arc, metadata_client: Arc, - event_publisher: Arc, } impl DefaultReviewLogger { @@ -32,7 +28,6 @@ impl DefaultReviewLogger { review_repo: Arc, watchlist_repo: Arc, metadata_client: Arc, - event_publisher: Arc, ) -> Self { Self { movie_command, @@ -40,7 +35,6 @@ impl DefaultReviewLogger { review_repo, watchlist_repo, metadata_client, - event_publisher, } } } @@ -57,13 +51,12 @@ impl ReviewLogger for DefaultReviewLogger { self.movie_command.as_ref(), self.movie_query.as_ref(), self.metadata_client.as_ref(), - self.event_publisher.as_ref(), ) .await?; // Always upsert: even existing movies may have updated metadata if !is_new_movie { - self.movie_command.upsert_movie(&movie).await?; + self.movie_command.upsert_movie(&movie, &[]).await?; } let review = Review::new( diff --git a/crates/application/src/integrations/deps.rs b/crates/application/src/integrations/deps.rs index d7cb1bbe..7a3af90e 100644 --- a/crates/application/src/integrations/deps.rs +++ b/crates/application/src/integrations/deps.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use domain::ports::{EventPublisher, WatchEventCommand, WatchEventQuery, WebhookTokenRepository}; +use domain::ports::{WatchEventCommand, WatchEventQuery, WebhookTokenRepository}; use crate::ports::ReviewLogger; @@ -8,7 +8,6 @@ pub struct IngestWatchEventDeps { pub webhook_token: Arc, pub watch_event_command: Arc, pub watch_event_query: Arc, - pub event_publisher: Arc, } pub struct ConfirmWatchEventsDeps { diff --git a/crates/application/src/integrations/ingest.rs b/crates/application/src/integrations/ingest.rs index d7fe4da7..7fc3755a 100644 --- a/crates/application/src/integrations/ingest.rs +++ b/crates/application/src/integrations/ingest.rs @@ -49,16 +49,16 @@ pub async fn execute( None, ); - deps.watch_event_command.save(&event).await?; - - let _ = deps - .event_publisher - .publish(&DomainEvent::WatchEventIngested { - user_id: event.user_id().clone(), - title: event.title().to_string(), - source: event.source().to_string(), - }) - .await; + deps.watch_event_command + .save( + &event, + &[DomainEvent::WatchEventIngested { + user_id: event.user_id().clone(), + title: event.title().to_string(), + source: event.source().to_string(), + }], + ) + .await?; Ok(()) } diff --git a/crates/application/src/jobs/wrapup.rs b/crates/application/src/jobs/wrapup.rs index 2a2f4040..c5231347 100644 --- a/crates/application/src/jobs/wrapup.rs +++ b/crates/application/src/jobs/wrapup.rs @@ -5,26 +5,17 @@ use async_trait::async_trait; use chrono::Datelike; use domain::{ errors::DomainError, - ports::{EventPublisher, PeriodicJob, UserRepository, WrapUpRepository}, + ports::{PeriodicJob, UserRepository, WrapUpRepository}, }; pub struct WrapUpAutoGenerateJob { user: Arc, wrapup_repo: Arc, - event_publisher: Arc, } impl WrapUpAutoGenerateJob { - pub fn new( - user: Arc, - wrapup_repo: Arc, - event_publisher: Arc, - ) -> Self { - Self { - user, - wrapup_repo, - event_publisher, - } + pub fn new(user: Arc, wrapup_repo: Arc) -> Self { + Self { user, wrapup_repo } } } @@ -60,7 +51,6 @@ impl PeriodicJob for WrapUpAutoGenerateJob { }; let deps = crate::wrapup::deps::GenerateWrapUpDeps { wrapup_repo: self.wrapup_repo.clone(), - event_publisher: self.event_publisher.clone(), }; if let Err(e) = crate::wrapup::generate::execute(&deps, cmd).await { tracing::warn!( @@ -81,7 +71,6 @@ impl PeriodicJob for WrapUpAutoGenerateJob { }; let deps = crate::wrapup::deps::GenerateWrapUpDeps { wrapup_repo: self.wrapup_repo.clone(), - event_publisher: self.event_publisher.clone(), }; if let Err(e) = crate::wrapup::generate::execute(&deps, cmd).await { tracing::warn!("auto-generate global wrapup failed: {e}"); diff --git a/crates/application/src/movies/poster_sync_handler.rs b/crates/application/src/movies/poster_sync_handler.rs index 874f6855..faa99518 100644 --- a/crates/application/src/movies/poster_sync_handler.rs +++ b/crates/application/src/movies/poster_sync_handler.rs @@ -85,20 +85,14 @@ impl PosterSyncHandler { let poster_path = PosterPath::new(stored_path)?; movie.update_poster(poster_path); - self.movie_command.upsert_movie(&movie).await?; - - if let Err(e) = self - .event_publisher - .publish(&DomainEvent::PosterSynced { - movie_id: movie.id().clone(), - }) - .await - { - tracing::warn!( - "failed to emit PosterSynced for {}: {e}", - movie.id().value() - ); - } + self.movie_command + .upsert_movie( + &movie, + &[DomainEvent::PosterSynced { + movie_id: movie.id().clone(), + }], + ) + .await?; Ok(()) } diff --git a/crates/application/src/movies/resolve.rs b/crates/application/src/movies/resolve.rs index 978c1e09..0f188c51 100644 --- a/crates/application/src/movies/resolve.rs +++ b/crates/application/src/movies/resolve.rs @@ -2,7 +2,7 @@ use domain::{ errors::DomainError, events::DomainEvent, models::Movie, - ports::{EventPublisher, MetadataClient, MovieCommand, MovieQuery}, + ports::{MetadataClient, MovieCommand, MovieQuery}, value_objects::MovieId, }; @@ -14,7 +14,6 @@ pub async fn resolve_and_persist_movie( movie_command: &dyn MovieCommand, movie_query: &dyn MovieQuery, metadata_client: &dyn MetadataClient, - event_publisher: &dyn EventPublisher, ) -> Result<(Movie, bool), DomainError> { let (movie, is_new) = if let Some(id) = input.movie_id { let movie_id = MovieId::from_uuid(id); @@ -34,15 +33,15 @@ pub async fn resolve_and_persist_movie( }; if is_new { - movie_command.upsert_movie(&movie).await?; - if let Some(ext_id) = movie.external_metadata_id() { - let _ = event_publisher - .publish(&DomainEvent::MovieDiscovered { - movie_id: movie.id().clone(), - external_metadata_id: ext_id.clone(), - }) - .await; - } + let events: Vec = movie + .external_metadata_id() + .map(|ext_id| DomainEvent::MovieDiscovered { + movie_id: movie.id().clone(), + external_metadata_id: ext_id.clone(), + }) + .into_iter() + .collect(); + movie_command.upsert_movie(&movie, &events).await?; } Ok((movie, is_new)) diff --git a/crates/application/src/movies/sync_poster.rs b/crates/application/src/movies/sync_poster.rs index a778f513..4d50e64e 100644 --- a/crates/application/src/movies/sync_poster.rs +++ b/crates/application/src/movies/sync_poster.rs @@ -59,7 +59,7 @@ pub async fn execute(deps: &SyncPosterDeps, cmd: SyncPosterCommand) -> Result<() let poster_path = PosterPath::new(stored_path)?; movie.update_poster(poster_path); - deps.movie_command.upsert_movie(&movie).await?; + deps.movie_command.upsert_movie(&movie, &[]).await?; // Refresh search index so the new poster_path is reflected immediately. // Fetch existing profile if available for a complete index document. diff --git a/crates/application/src/watchlist/add.rs b/crates/application/src/watchlist/add.rs index c85a714e..8d0d7432 100644 --- a/crates/application/src/watchlist/add.rs +++ b/crates/application/src/watchlist/add.rs @@ -18,7 +18,6 @@ pub async fn execute( deps.movie_command.as_ref(), deps.movie_query.as_ref(), deps.metadata.as_ref(), - deps.event_publisher.as_ref(), ) .await?; diff --git a/crates/application/src/watchlist/deps.rs b/crates/application/src/watchlist/deps.rs index e6154553..5d1fe253 100644 --- a/crates/application/src/watchlist/deps.rs +++ b/crates/application/src/watchlist/deps.rs @@ -1,8 +1,8 @@ use std::sync::Arc; use domain::ports::{ - EventPublisher, MetadataClient, MovieCommand, MovieQuery, RemoteWatchlistRepository, - UserRepository, WatchlistRepository, + MetadataClient, MovieCommand, MovieQuery, RemoteWatchlistRepository, UserRepository, + WatchlistRepository, }; pub struct WatchlistAddDeps { @@ -10,7 +10,6 @@ pub struct WatchlistAddDeps { pub movie_query: Arc, pub metadata: Arc, pub watchlist: Arc, - pub event_publisher: Arc, } pub struct GetWatchlistForOwnerDeps { diff --git a/crates/application/src/wrapup/deps.rs b/crates/application/src/wrapup/deps.rs index 74f9bb99..cc8f7bf5 100644 --- a/crates/application/src/wrapup/deps.rs +++ b/crates/application/src/wrapup/deps.rs @@ -1,10 +1,9 @@ use std::sync::Arc; -use domain::ports::{EventPublisher, WrapUpRepository, WrapUpStatsQuery}; +use domain::ports::{WrapUpRepository, WrapUpStatsQuery}; pub struct HandleWrapUpRequestedDeps { pub wrapup_repo: Arc, - pub event_publisher: Arc, pub wrapup_stats: Arc, } @@ -18,7 +17,6 @@ pub struct DeleteWrapUpDeps { pub struct GenerateWrapUpDeps { pub wrapup_repo: Arc, - pub event_publisher: Arc, } pub struct GetWrapUpDeps { diff --git a/crates/application/src/wrapup/event_handler.rs b/crates/application/src/wrapup/event_handler.rs index 1d5de35d..1615719a 100644 --- a/crates/application/src/wrapup/event_handler.rs +++ b/crates/application/src/wrapup/event_handler.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use async_trait::async_trait; use domain::errors::DomainError; use domain::events::DomainEvent; -use domain::ports::{EventHandler, EventPublisher, WrapUpRepository, WrapUpStatsQuery}; +use domain::ports::{EventHandler, WrapUpRepository, WrapUpStatsQuery}; use tokio::sync::Semaphore; use super::deps::HandleWrapUpRequestedDeps; @@ -16,13 +16,11 @@ pub struct WrapUpEventHandler { impl WrapUpEventHandler { pub fn new( wrapup_repo: Arc, - event_publisher: Arc, wrapup_stats: Arc, ) -> Self { Self { deps: HandleWrapUpRequestedDeps { wrapup_repo, - event_publisher, wrapup_stats, }, semaphore: Arc::new(Semaphore::new(2)), diff --git a/crates/application/src/wrapup/generate.rs b/crates/application/src/wrapup/generate.rs index f560baf6..2f136bc9 100644 --- a/crates/application/src/wrapup/generate.rs +++ b/crates/application/src/wrapup/generate.rs @@ -46,15 +46,16 @@ pub async fn execute( created_at: Utc::now().naive_utc(), completed_at: None, }; - deps.wrapup_repo.create(&record).await?; - - deps.event_publisher - .publish(&DomainEvent::WrapUpRequested { - wrapup_id: id.clone(), - user_id: cmd.user_id.map(UserId::from_uuid), - start_date: date_range.start(), - end_date: date_range.end(), - }) + deps.wrapup_repo + .create( + &record, + &[DomainEvent::WrapUpRequested { + wrapup_id: id.clone(), + user_id: cmd.user_id.map(UserId::from_uuid), + start_date: date_range.start(), + end_date: date_range.end(), + }], + ) .await?; Ok(id) diff --git a/crates/application/src/wrapup/handle_requested.rs b/crates/application/src/wrapup/handle_requested.rs index 70a93bfb..c696ec15 100644 --- a/crates/application/src/wrapup/handle_requested.rs +++ b/crates/application/src/wrapup/handle_requested.rs @@ -37,10 +37,14 @@ pub async fn execute( match compute::execute(deps.wrapup_stats.clone(), query).await { Ok(report) => { - deps.wrapup_repo.set_complete(&wrapup_id, &report).await?; - - deps.event_publisher - .publish(&DomainEvent::WrapUpCompleted { wrapup_id }) + deps.wrapup_repo + .set_complete( + &wrapup_id, + &report, + &[DomainEvent::WrapUpCompleted { + wrapup_id: wrapup_id.clone(), + }], + ) .await?; Ok(()) } diff --git a/crates/application/tests/diary/delete_review_test.rs b/crates/application/tests/diary/delete_review_test.rs index 0230199a..67ba9e45 100644 --- a/crates/application/tests/diary/delete_review_test.rs +++ b/crates/application/tests/diary/delete_review_test.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use domain::{ ports::{MovieCommand, MovieQuery, ReviewRepository}, - testing::{FakeDiaryQuery, InMemoryStore, NoopEventPublisher}, + testing::{FakeDiaryQuery, InMemoryStore, RefusingMovieCommand}, value_objects::{MovieId, UserId}, }; @@ -16,13 +16,12 @@ async fn deleting_a_review_removes_it() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let diary = FakeDiaryQuery::new(); - let events = NoopEventPublisher::new(); let movie = test_movie_with_year("Terminator", 1984); let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); let review = test_review(movie.id(), &user_id, 4); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); reviews.save_review(&review, &[]).await.unwrap(); diary.seed_history(movie.clone(), vec![]); @@ -30,7 +29,6 @@ async fn deleting_a_review_removes_it() { review: Arc::clone(&reviews) as _, diary: diary.clone() as _, movie_command: Arc::clone(&movies) as _, - event_publisher: Arc::clone(&events) as _, }; delete_review::execute( @@ -55,7 +53,6 @@ async fn deleting_another_users_review_is_unauthorized() { let reviews = InMemoryStore::new(); let diary = FakeDiaryQuery::new(); let movies = Arc::clone(&reviews); - let events = NoopEventPublisher::new(); let movie_id = MovieId::from_uuid(uuid::Uuid::new_v4()); let owner_id = UserId::from_uuid(uuid::Uuid::new_v4()); @@ -68,7 +65,6 @@ async fn deleting_another_users_review_is_unauthorized() { review: Arc::clone(&reviews) as _, diary: diary as _, movie_command: movies as _, - event_publisher: Arc::clone(&events) as _, }; let result = delete_review::execute( @@ -88,18 +84,16 @@ async fn deleting_another_users_review_is_unauthorized() { async fn deleting_a_review_removes_it_and_records_its_event_together() { let movies = InMemoryStore::new(); let diary = FakeDiaryQuery::new(); - let events = NoopEventPublisher::new(); let movie = test_movie_with_year("Terminator", 1984); let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); let review = test_review(movie.id(), &user_id, 4); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); movies.save_review(&review, &[]).await.unwrap(); diary.seed_history(movie.clone(), vec![review.clone()]); let deps = DeleteReviewDeps { review: Arc::clone(&movies) as _, diary: diary.clone() as _, movie_command: Arc::clone(&movies) as _, - event_publisher: Arc::clone(&events) as _, }; delete_review::execute( @@ -125,7 +119,6 @@ async fn deleting_a_review_removes_it_and_records_its_event_together() { async fn a_refused_delete_propagates_and_records_nothing() { let movies = InMemoryStore::new(); let diary = FakeDiaryQuery::new(); - let events = NoopEventPublisher::new(); let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); let review = test_review(&MovieId::generate(), &user_id, 4); movies.save_review(&review, &[]).await.unwrap(); @@ -135,7 +128,6 @@ async fn a_refused_delete_propagates_and_records_nothing() { ))), diary: diary as _, movie_command: Arc::clone(&movies) as _, - event_publisher: Arc::clone(&events) as _, }; let result = delete_review::execute( @@ -151,3 +143,114 @@ async fn a_refused_delete_propagates_and_records_nothing() { assert_eq!(movies.review_count(), 1); assert!(movies.recorded_events().is_empty()); } + +#[tokio::test] +async fn deleting_the_last_review_records_movie_deleted_with_the_movie_delete() { + let movies = InMemoryStore::new(); + let diary = FakeDiaryQuery::new(); + let movie = test_movie_with_year("Terminator", 1984); + let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); + let review = test_review(movie.id(), &user_id, 4); + movies.upsert_movie(&movie, &[]).await.unwrap(); + movies.save_review(&review, &[]).await.unwrap(); + diary.seed_history(movie.clone(), vec![]); + let deps = DeleteReviewDeps { + review: Arc::clone(&movies) as _, + diary: diary as _, + movie_command: Arc::clone(&movies) as _, + }; + + delete_review::execute( + &deps, + DeleteReviewCommand { + review_id: review.id().value(), + requesting_user_id: user_id.value(), + }, + ) + .await + .unwrap(); + + assert!(movies.get_movie_by_id(movie.id()).await.unwrap().is_none()); + let recorded = movies.recorded_events(); + assert!(matches!( + recorded[0], + domain::events::DomainEvent::ReviewDeleted { .. } + )); + assert!(matches!( + &recorded[1], + domain::events::DomainEvent::MovieDeleted { movie_id, .. } if movie_id == movie.id() + )); + assert_eq!(recorded.len(), 2); +} + +#[tokio::test] +async fn deleting_a_review_while_viewings_remain_keeps_the_movie_and_records_no_movie_deleted() { + let movies = InMemoryStore::new(); + let diary = FakeDiaryQuery::new(); + let movie = test_movie_with_year("Terminator", 1984); + let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); + let review = test_review(movie.id(), &user_id, 4); + let other = test_review(movie.id(), &UserId::from_uuid(uuid::Uuid::new_v4()), 3); + movies.upsert_movie(&movie, &[]).await.unwrap(); + movies.save_review(&review, &[]).await.unwrap(); + movies.save_review(&other, &[]).await.unwrap(); + diary.seed_history(movie.clone(), vec![other.clone()]); + let deps = DeleteReviewDeps { + review: Arc::clone(&movies) as _, + diary: diary as _, + movie_command: Arc::clone(&movies) as _, + }; + + delete_review::execute( + &deps, + DeleteReviewCommand { + review_id: review.id().value(), + requesting_user_id: user_id.value(), + }, + ) + .await + .unwrap(); + + assert!(movies.get_movie_by_id(movie.id()).await.unwrap().is_some()); + assert!( + !movies + .recorded_events() + .iter() + .any(|e| matches!(e, domain::events::DomainEvent::MovieDeleted { .. })) + ); +} + +#[tokio::test] +async fn a_refused_movie_delete_propagates_and_records_no_movie_deleted() { + let movies = InMemoryStore::new(); + let diary = FakeDiaryQuery::new(); + let movie = test_movie_with_year("Terminator", 1984); + let user_id = UserId::from_uuid(uuid::Uuid::new_v4()); + let review = test_review(movie.id(), &user_id, 4); + movies.upsert_movie(&movie, &[]).await.unwrap(); + movies.save_review(&review, &[]).await.unwrap(); + diary.seed_history(movie.clone(), vec![]); + let deps = DeleteReviewDeps { + review: Arc::clone(&movies) as _, + diary: diary as _, + movie_command: Arc::new(RefusingMovieCommand), + }; + + let result = delete_review::execute( + &deps, + DeleteReviewCommand { + review_id: review.id().value(), + requesting_user_id: user_id.value(), + }, + ) + .await; + + assert!(result.is_err()); + assert!(movies.get_movie_by_id(movie.id()).await.unwrap().is_some()); + assert!( + !movies + .recorded_events() + .iter() + .any(|e| matches!(e, domain::events::DomainEvent::MovieDeleted { .. })) + ); +} diff --git a/crates/application/tests/diary/get_movie_social_page_test.rs b/crates/application/tests/diary/get_movie_social_page_test.rs index 60f91880..0742c0e0 100644 --- a/crates/application/tests/diary/get_movie_social_page_test.rs +++ b/crates/application/tests/diary/get_movie_social_page_test.rs @@ -40,7 +40,7 @@ async fn returns_movie_social_page() { let movie = test_movie_with_year("Social Movie", 2024); let movie_uuid = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = GetMovieSocialPageDeps { movie_query: Arc::clone(&movies) as _, diff --git a/crates/application/tests/diary/log_review_test.rs b/crates/application/tests/diary/log_review_test.rs index ab0ca4c1..95b2b364 100644 --- a/crates/application/tests/diary/log_review_test.rs +++ b/crates/application/tests/diary/log_review_test.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use domain::ports::MovieCommand; -use domain::testing::{InMemoryStore, NoopEventPublisher}; +use domain::testing::InMemoryStore; use application::{ diary::commands::LogReviewCommand, diary::log_review, diary::review_logger::DefaultReviewLogger, @@ -14,7 +14,6 @@ use application::{ fn build_logger( movies: &Arc, reviews: &Arc, - events: &Arc, ) -> Arc { Arc::new(DefaultReviewLogger::new( Arc::clone(movies) as _, @@ -22,7 +21,6 @@ fn build_logger( Arc::clone(reviews) as _, TestContextBuilder::new().watchlist_repo, Arc::new(domain::testing::FakeMetadataClient) as _, - Arc::clone(events) as _, )) } @@ -30,8 +28,7 @@ fn build_logger( async fn logging_a_review_creates_the_movie_and_the_review() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = build_logger(&movies, &reviews, &events); + let logger = build_logger(&movies, &reviews); let user_id = uuid::Uuid::new_v4(); let cmd = log_review_command_manual(user_id, "Blade Runner", 1982); @@ -52,10 +49,9 @@ async fn logging_a_review_reuses_an_existing_movie() { let existing_movie = test_movie_with_year("Alien", 1979); let movie_uuid = existing_movie.id().value(); - movies.upsert_movie(&existing_movie).await.unwrap(); + movies.upsert_movie(&existing_movie, &[]).await.unwrap(); - let events = NoopEventPublisher::new(); - let logger = build_logger(&movies, &reviews, &events); + let logger = build_logger(&movies, &reviews); let cmd = LogReviewCommand { rating: 5, @@ -72,8 +68,7 @@ async fn logging_a_review_reuses_an_existing_movie() { async fn logging_a_review_with_an_invalid_rating_fails() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = build_logger(&movies, &reviews, &events); + let logger = build_logger(&movies, &reviews); let cmd = LogReviewCommand { rating: 6, diff --git a/crates/application/tests/diary/review_logger_test.rs b/crates/application/tests/diary/review_logger_test.rs index 28021ce7..b4e4ebdf 100644 --- a/crates/application/tests/diary/review_logger_test.rs +++ b/crates/application/tests/diary/review_logger_test.rs @@ -8,7 +8,7 @@ use domain::{ models::WatchlistEntry, models::{MetadataSearchCriteria, Movie}, ports::{MetadataClient, MovieCommand, WatchlistRepository}, - testing::{FakeMetadataClient, InMemoryStore, NoopEventPublisher}, + testing::{FakeMetadataClient, InMemoryStore}, value_objects::{ExternalMetadataId, MovieId, PosterUrl, UserId}, }; use uuid::Uuid; @@ -21,7 +21,6 @@ fn make_logger( movies: &Arc, reviews: &Arc, watchlist: &Arc, - events: &Arc, ) -> DefaultReviewLogger { DefaultReviewLogger::new( Arc::clone(movies) as _, @@ -29,7 +28,6 @@ fn make_logger( Arc::clone(reviews) as _, Arc::clone(watchlist) as _, Arc::new(FakeMetadataClient) as _, - Arc::clone(events) as _, ) } @@ -38,8 +36,7 @@ async fn logs_review_with_manual_movie() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let uid = Uuid::new_v4(); let cmd = log_review_command_manual(uid, "Test Film", 2024); @@ -55,8 +52,7 @@ async fn removes_from_watchlist_on_review() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let uid = Uuid::new_v4(); let user_id = UserId::from_uuid(uid); @@ -64,7 +60,7 @@ async fn removes_from_watchlist_on_review() { // Create and store movie let movie = test_movie_with_year("Watchlisted Film", 2024); let movie_id = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); // Add to watchlist let entry = WatchlistEntry::new(user_id.clone(), MovieId::from_uuid(movie_id)); @@ -87,13 +83,6 @@ async fn removes_from_watchlist_on_review() { .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] @@ -101,12 +90,11 @@ async fn logs_review_with_existing_movie_by_id() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let movie = test_movie_with_year("Existing Film", 2020); let movie_uuid = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let cmd = LogReviewCommand { rating: 3, @@ -124,8 +112,7 @@ async fn existing_movie_not_found_returns_error() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let cmd = log_review_command(Uuid::new_v4(), Uuid::new_v4()); @@ -137,8 +124,7 @@ async fn invalid_rating_returns_error() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let cmd = LogReviewCommand { rating: 6, @@ -157,8 +143,7 @@ async fn watchlist_not_present_does_not_publish_removed() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); - let logger = make_logger(&movies, &reviews, &watchlist, &events); + let logger = make_logger(&movies, &reviews, &watchlist); let cmd = log_review_command_manual(Uuid::new_v4(), "No Watchlist Film", 2024); @@ -197,11 +182,10 @@ impl MetadataClient for MetadataClientWithExternalId { } #[tokio::test] -async fn publishes_movie_discovered_for_new_movie_with_external_id() { +async fn movie_discovered_rides_the_movie_write_for_a_new_movie_with_external_id() { let movies = InMemoryStore::new(); let reviews = Arc::clone(&movies); let watchlist = Arc::clone(&movies); - let events = NoopEventPublisher::new(); let logger = DefaultReviewLogger::new( Arc::clone(&movies) as _, @@ -209,7 +193,6 @@ async fn publishes_movie_discovered_for_new_movie_with_external_id() { Arc::clone(&reviews) as _, Arc::clone(&watchlist) as _, Arc::new(MetadataClientWithExternalId) as _, - Arc::clone(&events) as _, ); let cmd = LogReviewCommand { @@ -219,12 +202,12 @@ async fn publishes_movie_discovered_for_new_movie_with_external_id() { logger.log_review(cmd).await.unwrap(); - let published = events.published(); + let published = movies.recorded_events(); assert!( published .iter() .any(|e| matches!(e, domain::events::DomainEvent::MovieDiscovered { .. })), - "should publish MovieDiscovered for new movie with external_metadata_id" + "MovieDiscovered must be recorded with the movie write" ); } @@ -237,7 +220,6 @@ async fn logging_a_review_fails_when_the_review_store_refuses() { Arc::new(domain::testing::RefusingReviewRepository), Arc::clone(&movies) as _, Arc::new(FakeMetadataClient), - NoopEventPublisher::new(), ); let result = logger @@ -256,7 +238,6 @@ async fn logging_a_review_saves_the_review_and_its_event_together() { Arc::clone(&movies) as _, Arc::clone(&movies) as _, Arc::new(FakeMetadataClient), - Arc::new(domain::testing::RefusingEventPublisher), ); logger @@ -282,7 +263,6 @@ async fn a_refused_save_propagates_and_records_nothing() { Arc::new(domain::testing::RefusingReviewRepository), Arc::clone(&movies) as _, Arc::new(FakeMetadataClient), - NoopEventPublisher::new(), ); let result = logger @@ -303,7 +283,6 @@ async fn enrichment_request_rides_the_review_save() { Arc::clone(&movies) as _, Arc::clone(&movies) as _, Arc::new(MetadataClientWithExternalId), - NoopEventPublisher::new(), ); logger diff --git a/crates/application/tests/import/apply_mapping_test.rs b/crates/application/tests/import/apply_mapping_test.rs index c5749e0a..be7cf6fc 100644 --- a/crates/application/tests/import/apply_mapping_test.rs +++ b/crates/application/tests/import/apply_mapping_test.rs @@ -110,7 +110,7 @@ async fn marks_duplicate_by_external_id() { let ext_id = ExternalMetadataId::new("tt1234567".into()).unwrap(); let movie = test_movie_with_external_id("Known Movie", 2020, ext_id); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let parser = Arc::new(DuplicateTestParser { rows: vec![ImportRow { @@ -158,7 +158,7 @@ async fn marks_duplicate_by_title_and_year() { let sessions = Arc::clone(&movies); let movie = test_movie_with_year("Duplicate Film", 2022); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let parser = Arc::new(DuplicateTestParser { rows: vec![ImportRow { diff --git a/crates/application/tests/integrations/confirm_test.rs b/crates/application/tests/integrations/confirm_test.rs index 79430ce2..7c9f0a8a 100644 --- a/crates/application/tests/integrations/confirm_test.rs +++ b/crates/application/tests/integrations/confirm_test.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use domain::models::{WatchEvent, WatchEventSource}; use domain::ports::{MovieCommand, WatchEventCommand}; -use domain::testing::{InMemoryStore, NoopEventPublisher}; +use domain::testing::InMemoryStore; use domain::value_objects::UserId; use uuid::Uuid; @@ -42,7 +42,7 @@ async fn confirms_watch_event_via_review_logger() { None, ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), @@ -86,7 +86,7 @@ async fn confirms_event_with_external_metadata_id_and_no_movie_id() { None, ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), @@ -121,7 +121,7 @@ async fn rejects_other_users_event() { None, ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), @@ -148,7 +148,6 @@ async fn fails_when_event_not_found() { #[tokio::test] async fn confirms_event_with_movie_id() { let watch_events = InMemoryStore::new(); - let events = NoopEventPublisher::new(); let uid = Uuid::new_v4(); let movie_uuid = Uuid::new_v4(); @@ -162,7 +161,7 @@ async fn confirms_event_with_movie_id() { Some(domain::value_objects::MovieId::from_uuid(movie_uuid)), ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); // Also seed movie repo so review_logger can find it let movies = domain::testing::InMemoryStore::new(); @@ -174,7 +173,7 @@ async fn confirms_event_with_movie_id() { None, None, ); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); // Build a real review logger let reviews = domain::testing::InMemoryStore::new(); @@ -186,7 +185,6 @@ async fn confirms_event_with_movie_id() { Arc::clone(&reviews) as _, Arc::clone(&watchlist) as _, Arc::new(domain::testing::FakeMetadataClient) as _, - Arc::clone(&events) as _, )); let result = confirm::execute( @@ -214,7 +212,7 @@ async fn confirms_event_without_movie_id_and_without_external_metadata_id() { None, ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), @@ -260,8 +258,8 @@ async fn confirms_multiple_events() { ); let id2 = event2.id().value(); - watch_events.save(&event1).await.unwrap(); - watch_events.save(&event2).await.unwrap(); + watch_events.save(&event1, &[]).await.unwrap(); + watch_events.save(&event2, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), @@ -302,7 +300,7 @@ async fn confirms_event_without_year() { None, ); let event_id = event.id().value(); - watch_events.save(&event).await.unwrap(); + watch_events.save(&event, &[]).await.unwrap(); let result = confirm::execute( &deps(&watch_events, noop_logger()), diff --git a/crates/application/tests/integrations/dismiss_test.rs b/crates/application/tests/integrations/dismiss_test.rs index d5278125..7b0bfc50 100644 --- a/crates/application/tests/integrations/dismiss_test.rs +++ b/crates/application/tests/integrations/dismiss_test.rs @@ -70,8 +70,8 @@ async fn dismisses_existing_events() { ); let id1 = e1.id().value(); let id2 = e2.id().value(); - watch_events.save(&e1).await.unwrap(); - watch_events.save(&e2).await.unwrap(); + watch_events.save(&e1, &[]).await.unwrap(); + watch_events.save(&e2, &[]).await.unwrap(); let result = dismiss::execute( &deps(&watch_events), diff --git a/crates/application/tests/integrations/get_queue_test.rs b/crates/application/tests/integrations/get_queue_test.rs index 33d3389d..886a318a 100644 --- a/crates/application/tests/integrations/get_queue_test.rs +++ b/crates/application/tests/integrations/get_queue_test.rs @@ -46,7 +46,7 @@ async fn returns_pending_events() { Utc::now().naive_utc(), None, ); - events.save(&event).await.unwrap(); + events.save(&event, &[]).await.unwrap(); let result = get_queue::execute(&deps(&events), GetWatchQueueQuery { user_id }) .await diff --git a/crates/application/tests/integrations/ingest_test.rs b/crates/application/tests/integrations/ingest_test.rs index 9f8cc7b0..b3f67109 100644 --- a/crates/application/tests/integrations/ingest_test.rs +++ b/crates/application/tests/integrations/ingest_test.rs @@ -2,8 +2,7 @@ use crate::support::generate_webhook_token_command; use std::sync::Arc; use domain::models::WatchEventSource; -use domain::ports::EventPublisher; -use domain::testing::{InMemoryStore, NoopEventPublisher}; +use domain::testing::{InMemoryStore, RefusingWatchEventCommand}; use uuid::Uuid; use application::integrations::commands::IngestWatchEventCommand; @@ -27,10 +26,9 @@ impl domain::ports::MediaServerParser for FakeParser { } #[tokio::test] -async fn ingests_watch_event() { +async fn ingesting_saves_the_watch_event_and_records_its_event_together() { let tokens = InMemoryStore::new(); let watch_events = Arc::clone(&tokens); - let event_publisher: Arc = NoopEventPublisher::new(); let user_id = Uuid::new_v4(); let generated = generate_token::execute( @@ -46,7 +44,6 @@ async fn ingests_watch_event() { webhook_token: Arc::clone(&tokens) as _, watch_event_command: Arc::clone(&watch_events) as _, watch_event_query: Arc::clone(&watch_events) as _, - event_publisher: Arc::clone(&event_publisher), }; let result = ingest::execute( @@ -61,19 +58,58 @@ async fn ingests_watch_event() { .await; assert!(result.is_ok()); + assert_eq!(watch_events.rows_in("watch_events"), 1); + let recorded = tokens.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + &recorded[0], + domain::events::DomainEvent::WatchEventIngested { title, .. } if title == "Test" + )); +} + +#[tokio::test] +async fn a_refused_save_propagates_and_records_nothing() { + let tokens = InMemoryStore::new(); + let user_id = Uuid::new_v4(); + let generated = generate_token::execute( + &GenerateWebhookTokenDeps { + webhook_token: Arc::clone(&tokens) as _, + }, + generate_webhook_token_command(user_id), + ) + .await + .unwrap(); + let deps = IngestWatchEventDeps { + webhook_token: Arc::clone(&tokens) as _, + watch_event_command: Arc::new(RefusingWatchEventCommand), + watch_event_query: Arc::clone(&tokens) as _, + }; + + let result = ingest::execute( + &deps, + IngestWatchEventCommand { + token: generated.token_plaintext, + raw_payload: vec![], + source: WatchEventSource::Jellyfin, + }, + &FakeParser, + ) + .await; + + assert!(result.is_err()); + assert_eq!(tokens.rows_in("watch_events"), 0); + assert!(tokens.recorded_events().is_empty()); } #[tokio::test] async fn rejects_invalid_token() { let tokens = InMemoryStore::new(); let watch_events = Arc::clone(&tokens); - let event_publisher: Arc = NoopEventPublisher::new(); let deps = IngestWatchEventDeps { webhook_token: Arc::clone(&tokens) as _, watch_event_command: Arc::clone(&watch_events) as _, watch_event_query: Arc::clone(&watch_events) as _, - event_publisher: Arc::clone(&event_publisher), }; let result = ingest::execute( diff --git a/crates/application/tests/movies.rs b/crates/application/tests/movies.rs index 5d07e747..52328748 100644 --- a/crates/application/tests/movies.rs +++ b/crates/application/tests/movies.rs @@ -6,7 +6,11 @@ mod enrich_movie_test; mod get_movie_profile_test; #[path = "movies/get_movies_test.rs"] mod get_movies_test; +#[path = "movies/poster_sync_handler_test.rs"] +mod poster_sync_handler_test; #[path = "movies/request_enrichment_test.rs"] mod request_enrichment_test; +#[path = "movies/resolve_test.rs"] +mod resolve_test; #[path = "movies/sync_poster_test.rs"] mod sync_poster_test; diff --git a/crates/application/tests/movies/enrich_movie_test.rs b/crates/application/tests/movies/enrich_movie_test.rs index 83d090ab..feb36bf1 100644 --- a/crates/application/tests/movies/enrich_movie_test.rs +++ b/crates/application/tests/movies/enrich_movie_test.rs @@ -19,7 +19,7 @@ async fn stores_profile_and_indexes() { let movie = test_movie_with_year("Test", 2024); let movie_id = MovieId::from_uuid(movie.id().value()); - movie_repo.upsert_movie(&movie).await.unwrap(); + movie_repo.upsert_movie(&movie, &[]).await.unwrap(); let profile = MovieProfile { movie_id: movie_id.clone(), @@ -68,7 +68,7 @@ async fn extracts_and_indexes_persons() { let movie = test_movie_with_year("Cast Movie", 2024); let movie_id = MovieId::from_uuid(movie.id().value()); - movie_repo.upsert_movie(&movie).await.unwrap(); + movie_repo.upsert_movie(&movie, &[]).await.unwrap(); let profile = MovieProfile { movie_id: movie_id.clone(), diff --git a/crates/application/tests/movies/poster_sync_handler_test.rs b/crates/application/tests/movies/poster_sync_handler_test.rs new file mode 100644 index 00000000..2c09508f --- /dev/null +++ b/crates/application/tests/movies/poster_sync_handler_test.rs @@ -0,0 +1,89 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::DomainEvent, + models::Movie, + ports::{EventHandler, MetadataClient, MovieCommand}, + testing::{ + FakePosterFetcher, InMemoryStore, NoopEventPublisher, NoopObjectStorage, + RefusingMovieCommand, test_movie_with_external_id, + }, + value_objects::{ExternalMetadataId, PosterUrl}, +}; + +use application::movies::poster_sync_handler::PosterSyncHandler; + +struct MetadataWithPoster; + +#[async_trait] +impl MetadataClient for MetadataWithPoster { + async fn fetch_movie_metadata( + &self, + _: &domain::models::MetadataSearchCriteria, + ) -> Result { + unimplemented!() + } + async fn get_poster_url( + &self, + _: &ExternalMetadataId, + ) -> Result, DomainError> { + Ok(Some( + PosterUrl::new("https://example.com/poster.jpg".into()).unwrap(), + )) + } +} + +async fn seeded() -> (Arc, Movie, DomainEvent) { + let store = InMemoryStore::new(); + let ext = ExternalMetadataId::new("tmdb:42".into()).unwrap(); + let movie = test_movie_with_external_id("Heat", 1995, ext.clone()); + store.upsert_movie(&movie, &[]).await.unwrap(); + let event = DomainEvent::MovieDiscovered { + movie_id: movie.id().clone(), + external_metadata_id: ext, + }; + (store, movie, event) +} + +fn handler(command: Arc, store: &Arc) -> PosterSyncHandler { + PosterSyncHandler::new( + command, + Arc::clone(store) as _, + Arc::new(MetadataWithPoster), + Arc::new(FakePosterFetcher), + Arc::new(NoopObjectStorage), + NoopEventPublisher::new(), + 0, + ) +} + +#[tokio::test] +async fn poster_synced_rides_the_movie_write() { + let (store, movie, event) = seeded().await; + + handler(Arc::clone(&store) as _, &store) + .handle(&event) + .await + .unwrap(); + + let recorded = store.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + &recorded[0], + DomainEvent::PosterSynced { movie_id } if movie_id == movie.id() + )); +} + +#[tokio::test] +async fn a_refused_upsert_propagates_and_records_nothing() { + let (store, _movie, event) = seeded().await; + + let result = handler(Arc::new(RefusingMovieCommand), &store) + .handle(&event) + .await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused")); + assert!(store.recorded_events().is_empty()); +} diff --git a/crates/application/tests/movies/resolve_test.rs b/crates/application/tests/movies/resolve_test.rs new file mode 100644 index 00000000..99c541f1 --- /dev/null +++ b/crates/application/tests/movies/resolve_test.rs @@ -0,0 +1,127 @@ +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::DomainEvent, + models::{MetadataSearchCriteria, Movie}, + ports::{MetadataClient, MovieCommand}, + testing::{ + FakeMetadataClient, InMemoryStore, RefusingMovieCommand, test_movie_with_external_id, + test_movie_with_year, + }, + value_objects::{ExternalMetadataId, PosterUrl}, +}; + +use application::diary::commands::MovieInput; +use application::movies::resolve::resolve_and_persist_movie; + +struct MetadataWithExternalId; + +#[async_trait] +impl MetadataClient for MetadataWithExternalId { + async fn fetch_movie_metadata(&self, _: &MetadataSearchCriteria) -> Result { + Ok(test_movie_with_external_id( + "Discovered Film", + 2024, + ExternalMetadataId::new("tmdb:99999".into()).unwrap(), + )) + } + + async fn get_poster_url( + &self, + _: &ExternalMetadataId, + ) -> Result, DomainError> { + Ok(None) + } +} + +fn manual_input(title: &str) -> MovieInput { + MovieInput { + movie_id: None, + external_metadata_id: None, + manual_title: Some(title.into()), + manual_release_year: Some(2024), + manual_director: None, + } +} + +#[tokio::test] +async fn a_new_movie_with_an_external_id_is_persisted_with_movie_discovered() { + let store = InMemoryStore::new(); + + let (movie, is_new) = resolve_and_persist_movie( + &manual_input("Discovered Film"), + store.as_ref(), + store.as_ref(), + &MetadataWithExternalId, + ) + .await + .unwrap(); + + assert!(is_new); + assert_eq!(store.movie_count(), 1); + let recorded = store.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + &recorded[0], + DomainEvent::MovieDiscovered { movie_id, .. } if movie_id == movie.id() + )); +} + +#[tokio::test] +async fn a_new_movie_without_an_external_id_records_no_event() { + let store = InMemoryStore::new(); + + let (_, is_new) = resolve_and_persist_movie( + &manual_input("Home Movie"), + store.as_ref(), + store.as_ref(), + &FakeMetadataClient, + ) + .await + .unwrap(); + + assert!(is_new); + assert_eq!(store.movie_count(), 1); + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn an_existing_movie_is_not_rewritten_and_records_no_event() { + let store = InMemoryStore::new(); + let movie = test_movie_with_year("Known", 2001); + store.upsert_movie(&movie, &[]).await.unwrap(); + let input = MovieInput { + movie_id: Some(movie.id().value()), + ..manual_input("ignored") + }; + + let (resolved, is_new) = resolve_and_persist_movie( + &input, + store.as_ref(), + store.as_ref(), + &MetadataWithExternalId, + ) + .await + .unwrap(); + + assert!(!is_new); + assert_eq!(resolved.id(), movie.id()); + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn a_refused_upsert_propagates_and_records_nothing() { + let store = InMemoryStore::new(); + + let result = resolve_and_persist_movie( + &manual_input("Discovered Film"), + &RefusingMovieCommand, + store.as_ref(), + &MetadataWithExternalId, + ) + .await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(m)) if m == "refused")); + assert_eq!(store.movie_count(), 0); + assert!(store.recorded_events().is_empty()); +} diff --git a/crates/application/tests/movies/sync_poster_test.rs b/crates/application/tests/movies/sync_poster_test.rs index 6e2d937c..b8ebff01 100644 --- a/crates/application/tests/movies/sync_poster_test.rs +++ b/crates/application/tests/movies/sync_poster_test.rs @@ -41,7 +41,7 @@ async fn fails_when_no_external_id() { let movies = InMemoryStore::new(); let movie = test_movie_with_year("Test", 2024); let movie_id = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = SyncPosterDeps { movie_command: Arc::clone(&movies) as _, @@ -80,7 +80,7 @@ async fn syncs_poster_for_movie_with_external_id() { let ext_id = ExternalMetadataId::new("tmdb:999".into()).unwrap(); let movie = test_movie_with_external_id("Poster Movie", 2024, ext_id); let movie_id = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = SyncPosterDeps { movie_command: Arc::clone(&movies) as _, diff --git a/crates/application/tests/watchlist/add_test.rs b/crates/application/tests/watchlist/add_test.rs index 995835d4..2d1fb01d 100644 --- a/crates/application/tests/watchlist/add_test.rs +++ b/crates/application/tests/watchlist/add_test.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use domain::{ events::DomainEvent, ports::MovieCommand, - testing::{InMemoryStore, NoopEventPublisher, RefusingWatchlistRepository}, + testing::{InMemoryStore, RefusingWatchlistRepository}, }; use application::watchlist::{add, commands::AddToWatchlistCommand, deps::WatchlistAddDeps}; @@ -16,7 +16,6 @@ fn make_deps(movies: Arc, watchlist: Arc) -> Watch movie_query: movies, metadata: Arc::new(domain::testing::FakeMetadataClient), watchlist, - event_publisher: NoopEventPublisher::new(), } } @@ -27,7 +26,7 @@ async fn adding_to_the_watchlist_resolves_the_movie_and_saves_the_entry() { let movie = test_movie_with_year("The Thing", 1982); let movie_uuid = movie.id().value(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = make_deps(Arc::clone(&movies), Arc::clone(&watchlist)); @@ -46,7 +45,7 @@ async fn adding_a_movie_already_on_the_watchlist_is_idempotent() { let movie = test_movie_with_year("RoboCop", 1987); let movie_uuid = movie.id().value(); let user_id = uuid::Uuid::new_v4(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = make_deps(Arc::clone(&movies), Arc::clone(&watchlist)); @@ -98,7 +97,7 @@ async fn adding_the_same_movie_to_a_watchlist_twice_keeps_a_single_entry_and_suc let movie = test_movie_with_year("Heat", 1995); let movie_uuid = movie.id().value(); let user_id = uuid::Uuid::new_v4(); - movies.upsert_movie(&movie).await.unwrap(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = make_deps(Arc::clone(&movies), Arc::clone(&movies)); for _ in 0..2 { @@ -114,7 +113,7 @@ async fn adding_the_same_movie_to_a_watchlist_twice_keeps_a_single_entry_and_suc 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(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = make_deps(Arc::clone(&movies), Arc::clone(&movies)); add::execute( @@ -137,7 +136,7 @@ async fn adding_an_entry_records_the_entry_and_its_event_together() { 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(); + movies.upsert_movie(&movie, &[]).await.unwrap(); let deps = WatchlistAddDeps { watchlist: Arc::new(RefusingWatchlistRepository), ..make_deps(Arc::clone(&movies), Arc::clone(&movies)) diff --git a/crates/application/tests/wrapup/generate_test.rs b/crates/application/tests/wrapup/generate_test.rs index 0d0657c6..299174e9 100644 --- a/crates/application/tests/wrapup/generate_test.rs +++ b/crates/application/tests/wrapup/generate_test.rs @@ -2,7 +2,7 @@ use crate::support::request_wrap_up_command; use chrono::NaiveDate; use domain::events::DomainEvent; use domain::models::wrapup::{WrapUpRecord, WrapUpStatus}; -use domain::testing::{InMemoryWrapUpRepository, NoopEventPublisher}; +use domain::testing::{InMemoryWrapUpRepository, RefusingWrapUpWrites}; use domain::value_objects::WrapUpId; use uuid::Uuid; @@ -11,13 +11,9 @@ use std::sync::Arc; use application::wrapup::deps::GenerateWrapUpDeps; use application::wrapup::{commands::RequestWrapUpCommand, generate}; -fn deps( - repo: &Arc, - events: &Arc, -) -> GenerateWrapUpDeps { +fn deps(repo: &Arc) -> GenerateWrapUpDeps { GenerateWrapUpDeps { wrapup_repo: Arc::clone(repo) as _, - event_publisher: Arc::clone(events) as _, } } @@ -26,25 +22,22 @@ fn past_cmd() -> RequestWrapUpCommand { } #[tokio::test] -async fn creates_pending_record_and_emits_event() { +async fn creates_pending_record_and_records_its_event_together() { let repo = InMemoryWrapUpRepository::new(); - let events = NoopEventPublisher::new(); - let id = generate::execute(&deps(&repo, &events), past_cmd()) - .await - .unwrap(); + let id = generate::execute(&deps(&repo), past_cmd()).await.unwrap(); let stored = repo.store.lock().unwrap(); assert_eq!(stored.len(), 1); assert_eq!(stored[0].id, id); assert_eq!(stored[0].status, WrapUpStatus::Pending); - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, DomainEvent::WrapUpRequested { .. })) - ); + let recorded = repo.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + &recorded[0], + DomainEvent::WrapUpRequested { wrapup_id, .. } if *wrapup_id == id + )); } #[tokio::test] @@ -62,13 +55,11 @@ async fn reuses_existing_ready_wrapup() { created_at: chrono::Utc::now().naive_utc(), completed_at: None, }); - let events = NoopEventPublisher::new(); - let id = generate::execute(&deps(&repo, &events), past_cmd()) - .await - .unwrap(); + let id = generate::execute(&deps(&repo), past_cmd()).await.unwrap(); assert_eq!(id, existing_id); assert_eq!(repo.store.lock().unwrap().len(), 1); + assert!(repo.recorded_events().is_empty()); } #[tokio::test] @@ -86,11 +77,7 @@ async fn replaces_failed_wrapup() { completed_at: None, }); - let events = NoopEventPublisher::new(); - - let id = generate::execute(&deps(&repo, &events), past_cmd()) - .await - .unwrap(); + let id = generate::execute(&deps(&repo), past_cmd()).await.unwrap(); let stored = repo.store.lock().unwrap(); assert_eq!(stored.len(), 1); @@ -101,10 +88,23 @@ async fn replaces_failed_wrapup() { #[tokio::test] async fn rejects_future_end_date() { let repo = InMemoryWrapUpRepository::new(); - let events = NoopEventPublisher::new(); - let err = generate::execute(&deps(&repo, &events), request_wrap_up_command(None, 2030)) + let err = generate::execute(&deps(&repo), request_wrap_up_command(None, 2030)) .await .unwrap_err(); assert!(err.to_string().contains("future")); } + +#[tokio::test] +async fn a_refused_create_propagates_and_records_nothing() { + let inner = InMemoryWrapUpRepository::new(); + let deps = GenerateWrapUpDeps { + wrapup_repo: RefusingWrapUpWrites::new(Arc::clone(&inner)), + }; + + let result = generate::execute(&deps, past_cmd()).await; + + assert!(result.is_err()); + assert!(inner.store.lock().unwrap().is_empty()); + assert!(inner.recorded_events().is_empty()); +} diff --git a/crates/application/tests/wrapup/get_ready_report_test.rs b/crates/application/tests/wrapup/get_ready_report_test.rs index e8dcef7f..dc05e673 100644 --- a/crates/application/tests/wrapup/get_ready_report_test.rs +++ b/crates/application/tests/wrapup/get_ready_report_test.rs @@ -56,7 +56,7 @@ fn record( async fn ready_report_returns_report_when_ready() { let repo = InMemoryWrapUpRepository::new(); let user_id = Uuid::new_v4(); - repo.create(&record(Some(user_id), 2024, WrapUpStatus::Ready, true)) + repo.create(&record(Some(user_id), 2024, WrapUpStatus::Ready, true), &[]) .await .unwrap(); @@ -72,12 +72,10 @@ async fn ready_report_returns_report_when_ready() { async fn ready_report_is_not_found_when_status_not_ready() { let repo = InMemoryWrapUpRepository::new(); let user_id = Uuid::new_v4(); - repo.create(&record( - Some(user_id), - 2024, - WrapUpStatus::Generating, - false, - )) + repo.create( + &record(Some(user_id), 2024, WrapUpStatus::Generating, false), + &[], + ) .await .unwrap(); @@ -96,9 +94,12 @@ async fn ready_report_is_not_found_when_report_absent() { // Status is Ready but `report` is None — an inconsistent-but-representable // state the old handler guarded against explicitly (`match record.report { // Some(r) => ..., None => 404 }`), separate from the status gate above. - repo.create(&record(Some(user_id), 2024, WrapUpStatus::Ready, false)) - .await - .unwrap(); + repo.create( + &record(Some(user_id), 2024, WrapUpStatus::Ready, false), + &[], + ) + .await + .unwrap(); let deps = GetReadyReportDeps { wrapup_repo: repo }; @@ -122,7 +123,7 @@ async fn ready_report_is_not_found_when_no_record_exists() { #[tokio::test] async fn ready_report_global_scope_ignores_user_id() { let repo = InMemoryWrapUpRepository::new(); - repo.create(&record(None, 2024, WrapUpStatus::Ready, true)) + repo.create(&record(None, 2024, WrapUpStatus::Ready, true), &[]) .await .unwrap(); diff --git a/crates/application/tests/wrapup/handle_requested_test.rs b/crates/application/tests/wrapup/handle_requested_test.rs index c7fb26ce..53177b9a 100644 --- a/crates/application/tests/wrapup/handle_requested_test.rs +++ b/crates/application/tests/wrapup/handle_requested_test.rs @@ -1,7 +1,7 @@ use chrono::{NaiveDate, Utc}; use domain::models::wrapup::{WrapUpRecord, WrapUpStatus}; use domain::ports::WrapUpRepository; -use domain::testing::{InMemoryWrapUpRepository, InMemoryWrapUpStatsQuery, NoopEventPublisher}; +use domain::testing::{InMemoryWrapUpRepository, InMemoryWrapUpStatsQuery, RefusingWrapUpWrites}; use domain::value_objects::WrapUpId; use application::wrapup::{deps::HandleWrapUpRequestedDeps, handle_requested}; @@ -22,11 +22,10 @@ async fn skips_if_already_ready() { created_at: Utc::now().naive_utc(), completed_at: None, }; - repo.create(&record).await.unwrap(); + repo.create(&record, &[]).await.unwrap(); let deps = HandleWrapUpRequestedDeps { wrapup_repo: repo.clone(), - event_publisher: NoopEventPublisher::new(), wrapup_stats: InMemoryWrapUpStatsQuery::new(), }; @@ -46,7 +45,6 @@ async fn skips_if_already_ready() { async fn generates_wrapup_and_marks_complete() { let repo = InMemoryWrapUpRepository::new(); let stats = InMemoryWrapUpStatsQuery::new(); - let events = NoopEventPublisher::new(); let wrapup_id = WrapUpId::generate(); let uid = uuid::Uuid::new_v4(); @@ -61,11 +59,10 @@ async fn generates_wrapup_and_marks_complete() { created_at: Utc::now().naive_utc(), completed_at: None, }; - repo.create(&record).await.unwrap(); + repo.create(&record, &[]).await.unwrap(); let deps = HandleWrapUpRequestedDeps { wrapup_repo: repo.clone(), - event_publisher: events.clone(), wrapup_stats: stats.clone(), }; @@ -85,13 +82,13 @@ async fn generates_wrapup_and_marks_complete() { assert_eq!(final_rec.status, WrapUpStatus::Ready); assert!(final_rec.report.is_some()); - // Verify event was published - let published = events.published(); - assert!( - published - .iter() - .any(|e| matches!(e, domain::events::DomainEvent::WrapUpCompleted { .. })) - ); + // The completion event is recorded together with the completed row. + let recorded = repo.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!( + &recorded[0], + domain::events::DomainEvent::WrapUpCompleted { wrapup_id: id } if *id == wrapup_id + )); } #[tokio::test] @@ -110,11 +107,10 @@ async fn skips_if_already_generating() { created_at: Utc::now().naive_utc(), completed_at: None, }; - repo.create(&record).await.unwrap(); + repo.create(&record, &[]).await.unwrap(); let deps = HandleWrapUpRequestedDeps { wrapup_repo: repo.clone(), - event_publisher: NoopEventPublisher::new(), wrapup_stats: InMemoryWrapUpStatsQuery::new(), }; @@ -132,4 +128,42 @@ async fn skips_if_already_generating() { // Status should still be Generating (not changed to Ready) let final_rec = repo.get_by_id(&wrapup_id).await.unwrap().unwrap(); assert_eq!(final_rec.status, WrapUpStatus::Generating); + assert!(repo.recorded_events().is_empty()); +} + +#[tokio::test] +async fn a_refused_completion_propagates_and_records_nothing() { + let inner = InMemoryWrapUpRepository::new(); + let wrapup_id = WrapUpId::generate(); + let record = WrapUpRecord { + id: wrapup_id.clone(), + user_id: None, + start_date: NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + end_date: NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(), + status: WrapUpStatus::Pending, + report: None, + error_message: None, + created_at: Utc::now().naive_utc(), + completed_at: None, + }; + inner.create(&record, &[]).await.unwrap(); + let deps = HandleWrapUpRequestedDeps { + wrapup_repo: RefusingWrapUpWrites::new(inner.clone()), + wrapup_stats: InMemoryWrapUpStatsQuery::new(), + }; + + let result = handle_requested::execute( + &deps, + wrapup_id.clone(), + None, + NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + NaiveDate::from_ymd_opt(2024, 12, 31).unwrap(), + ) + .await; + + assert!(result.is_err()); + let rec = inner.get_by_id(&wrapup_id).await.unwrap().unwrap(); + assert_ne!(rec.status, WrapUpStatus::Ready); + assert!(rec.report.is_none()); + assert!(inner.recorded_events().is_empty()); } diff --git a/crates/bootstrap/src/build.rs b/crates/bootstrap/src/build.rs index 9003d371..913b002e 100644 --- a/crates/bootstrap/src/build.rs +++ b/crates/bootstrap/src/build.rs @@ -90,7 +90,6 @@ pub fn build_deps( review: repos.review.clone(), diary: repos.diary.clone(), movie_command: repos.movie_command.clone(), - event_publisher: services.event_publisher.clone(), }, edit_review: EditReviewDeps { review: repos.review.clone(), @@ -185,7 +184,6 @@ pub fn build_deps( webhook_token: repos.webhook_token.clone(), watch_event_command: repos.watch_event_command.clone(), watch_event_query: repos.watch_event_query.clone(), - event_publisher: services.event_publisher.clone(), }, confirm_watch_events: ConfirmWatchEventsDeps { watch_event_command: repos.watch_event_command.clone(), @@ -313,7 +311,6 @@ pub fn build_deps( movie_query: repos.movie_query.clone(), metadata: services.metadata.clone(), watchlist: repos.watchlist.clone(), - event_publisher: services.event_publisher.clone(), }, get_watchlist_for_owner: GetWatchlistForOwnerDeps { user: repos.user.clone(), @@ -339,7 +336,6 @@ pub fn build_deps( }, generate: GenerateWrapUpDeps { wrapup_repo: repos.wrapup_repo.clone(), - event_publisher: services.event_publisher.clone(), }, get_wrapup: GetWrapUpDeps { wrapup_repo: repos.wrapup_repo.clone(), @@ -387,7 +383,6 @@ pub fn build_worker_deps( }, handle_requested: HandleWrapUpRequestedDeps { wrapup_repo: db.wrapup_repo.clone(), - event_publisher: services.event_publisher.clone(), wrapup_stats: db.wrapup_stats.clone(), }, } diff --git a/crates/bootstrap/src/server.rs b/crates/bootstrap/src/server.rs index 919e97ca..c3f87a4b 100644 --- a/crates/bootstrap/src/server.rs +++ b/crates/bootstrap/src/server.rs @@ -72,7 +72,6 @@ async fn assemble( Arc::clone(&db.review), Arc::clone(&db.watchlist), Arc::clone(&metadata_client), - Arc::clone(&event_publisher), )); let repos = Repositories { diff --git a/crates/bootstrap/src/worker.rs b/crates/bootstrap/src/worker.rs index c2ba1242..5826e2e0 100644 --- a/crates/bootstrap/src/worker.rs +++ b/crates/bootstrap/src/worker.rs @@ -215,7 +215,6 @@ async fn assemble( Arc::new(application::jobs::WrapUpAutoGenerateJob::new( Arc::clone(&user), Arc::clone(&wrapup_repo), - Arc::clone(&event_publisher), )), Arc::new(application::jobs::WrapUpCleanupJob::new(Arc::clone( &wrapup_repo, @@ -277,7 +276,6 @@ async fn assemble( let wrapup_handler = Arc::new(application::wrapup::event_handler::WrapUpEventHandler::new( worker_deps.handle_requested.wrapup_repo.clone(), - worker_deps.handle_requested.event_publisher.clone(), worker_deps.handle_requested.wrapup_stats.clone(), )) as Arc; diff --git a/crates/domain/src/ports/media_server.rs b/crates/domain/src/ports/media_server.rs index e28e0772..7acbd9bc 100644 --- a/crates/domain/src/ports/media_server.rs +++ b/crates/domain/src/ports/media_server.rs @@ -3,6 +3,7 @@ use chrono::NaiveDateTime; use crate::{ errors::DomainError, + events::DomainEvent, models::{ParsedPlaybackEvent, WatchEvent, WatchEventStatus, WebhookToken}, value_objects::{UserId, WatchEventId, WebhookTokenId}, }; @@ -14,7 +15,7 @@ pub trait MediaServerParser: Send + Sync { #[async_trait] pub trait WatchEventCommand: Send + Sync { - async fn save(&self, event: &WatchEvent) -> Result<(), DomainError>; + async fn save(&self, event: &WatchEvent, events: &[DomainEvent]) -> Result<(), DomainError>; async fn update_status( &self, id: &WatchEventId, diff --git a/crates/domain/src/ports/movie.rs b/crates/domain/src/ports/movie.rs index 5304548c..a25a0ae0 100644 --- a/crates/domain/src/ports/movie.rs +++ b/crates/domain/src/ports/movie.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, models::{ MetadataSearchCriteria, Movie, MovieFilter, MovieProfile, MovieSummary, collections::{PageParams, Paginated}, @@ -11,8 +12,12 @@ use crate::{ #[async_trait] pub trait MovieCommand: Send + Sync { - async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError>; - async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError>; + async fn upsert_movie(&self, movie: &Movie, events: &[DomainEvent]) -> Result<(), DomainError>; + async fn delete_movie( + &self, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError>; } #[async_trait] diff --git a/crates/domain/src/ports/wrapup.rs b/crates/domain/src/ports/wrapup.rs index cd6129c9..187d6774 100644 --- a/crates/domain/src/ports/wrapup.rs +++ b/crates/domain/src/ports/wrapup.rs @@ -4,6 +4,7 @@ use uuid::Uuid; use crate::{ errors::DomainError, + events::DomainEvent, models::wrapup::{ DateRange, WrapUpMovieRow, WrapUpRecord, WrapUpReport, WrapUpScope, WrapUpStatus, }, @@ -12,14 +13,23 @@ use crate::{ #[async_trait] pub trait WrapUpRepository: Send + Sync { - async fn create(&self, record: &WrapUpRecord) -> Result<(), DomainError>; + async fn create( + &self, + record: &WrapUpRecord, + events: &[DomainEvent], + ) -> Result<(), DomainError>; async fn update_status( &self, id: &WrapUpId, status: &WrapUpStatus, error: Option<&str>, ) -> Result<(), DomainError>; - async fn set_complete(&self, id: &WrapUpId, report: &WrapUpReport) -> Result<(), DomainError>; + async fn set_complete( + &self, + id: &WrapUpId, + report: &WrapUpReport, + events: &[DomainEvent], + ) -> Result<(), DomainError>; async fn get_by_id(&self, id: &WrapUpId) -> Result, DomainError>; async fn list_for_user(&self, user_id: Uuid) -> Result, DomainError>; async fn list_global(&self) -> Result, DomainError>; diff --git a/crates/domain/src/testing/fakes/panics.rs b/crates/domain/src/testing/fakes/panics.rs index 2fc6a186..952c4899 100644 --- a/crates/domain/src/testing/fakes/panics.rs +++ b/crates/domain/src/testing/fakes/panics.rs @@ -393,7 +393,11 @@ pub struct PanicWatchEventCommand; #[async_trait] impl crate::ports::WatchEventCommand for PanicWatchEventCommand { - async fn save(&self, _: &crate::models::WatchEvent) -> Result<(), DomainError> { + async fn save( + &self, + _: &crate::models::WatchEvent, + _: &[DomainEvent], + ) -> Result<(), DomainError> { panic!("PanicWatchEventCommand called") } async fn update_status( @@ -487,10 +491,10 @@ pub struct PanicMovieCommand; #[async_trait] impl MovieCommand for PanicMovieCommand { - async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { + async fn upsert_movie(&self, _: &Movie, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicMovieCommand called") } - async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { + async fn delete_movie(&self, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> { panic!("PanicMovieCommand called") } } diff --git a/crates/domain/src/testing/fakes/refusing.rs b/crates/domain/src/testing/fakes/refusing.rs index 70ac8e6e..86864c7c 100644 --- a/crates/domain/src/testing/fakes/refusing.rs +++ b/crates/domain/src/testing/fakes/refusing.rs @@ -1,19 +1,21 @@ //! Fakes whose every call fails with an infrastructure error, for failure-path tests. use async_trait::async_trait; +use chrono::NaiveDateTime; use crate::{ errors::DomainError, events::DomainEvent, models::{ - Goal, Movie, ProfileField, Review, WatchlistEntry, WatchlistWithMovie, + Goal, Movie, ProfileField, Review, WatchEvent, WatchEventStatus, WatchlistEntry, + WatchlistWithMovie, collections::{PageParams, Paginated}, }, ports::{ EventPublisher, GoalCommand, MovieCommand, ReviewRepository, UserProfileFieldsRepository, - WatchlistRepository, + WatchEventCommand, WatchlistRepository, }, - value_objects::{GoalId, MovieId, ReviewId, UserId}, + value_objects::{GoalId, MovieId, ReviewId, UserId, WatchEventId}, }; fn refused() -> DomainError { @@ -45,10 +47,10 @@ pub struct RefusingMovieCommand; #[async_trait] impl MovieCommand for RefusingMovieCommand { - async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { + async fn upsert_movie(&self, _: &Movie, _: &[DomainEvent]) -> Result<(), DomainError> { Err(refused()) } - async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { + async fn delete_movie(&self, _: &MovieId, _: &[DomainEvent]) -> Result<(), DomainError> { Err(refused()) } } @@ -123,3 +125,29 @@ impl EventPublisher for RefusingEventPublisher { Err(refused()) } } + +pub struct RefusingWatchEventCommand; + +#[async_trait] +impl WatchEventCommand for RefusingWatchEventCommand { + async fn save(&self, _: &WatchEvent, _: &[DomainEvent]) -> Result<(), DomainError> { + Err(refused()) + } + async fn update_status( + &self, + _: &WatchEventId, + _: WatchEventStatus, + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn update_status_batch( + &self, + _: &[WatchEventId], + _: WatchEventStatus, + ) -> Result { + Err(refused()) + } + async fn delete_non_pending_older_than(&self, _: NaiveDateTime) -> Result { + Err(refused()) + } +} diff --git a/crates/domain/src/testing/fakes/store_movie.rs b/crates/domain/src/testing/fakes/store_movie.rs index 56852ee8..b70a3877 100644 --- a/crates/domain/src/testing/fakes/store_movie.rs +++ b/crates/domain/src/testing/fakes/store_movie.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, models::{ Movie, MovieFilter, MovieProfile, MovieSummary, PersistedWatchEvent, WatchEvent, collections::{PageParams, Paginated}, @@ -17,7 +18,7 @@ impl MovieCommand for InMemoryStore { /// `movies.external_metadata_id` is UNIQUE: another movie row holding the /// same external id is a constraint violation, which the sqlite adapter /// surfaces as an infrastructure error. - async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> { + async fn upsert_movie(&self, movie: &Movie, events: &[DomainEvent]) -> Result<(), DomainError> { let mut movies = self.movies.write().unwrap(); if let Some(external) = movie.external_metadata_id() { let taken = movies.values().any(|m| { @@ -31,13 +32,19 @@ impl MovieCommand for InMemoryStore { } } movies.insert(movie.id().value(), movie.clone()); + drop(movies); + self.record_events(events); Ok(()) } /// Schema: `reviews.movie_id REFERENCES movies(id)` has no cascade, so a /// movie with reviews cannot be deleted. `watchlist_entries` and the /// `movie_*` profile tables cascade; `watch_events.movie_id` is set to NULL. - async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> { + async fn delete_movie( + &self, + movie_id: &MovieId, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let id = movie_id.value(); if !self.movies.read().unwrap().contains_key(&id) { return Ok(()); @@ -59,12 +66,14 @@ impl MovieCommand for InMemoryStore { .unwrap() .retain(|(_, movie), _| *movie != id); self.movie_profiles.write().unwrap().remove(&id); - let mut events = self.watch_events.write().unwrap(); - for event in events.iter_mut() { + let mut events_guard = self.watch_events.write().unwrap(); + for event in events_guard.iter_mut() { if event.movie_id().map(|m| m.value()) == Some(id) { *event = detached_from_movie(event); } } + drop(events_guard); + self.record_events(events); Ok(()) } } diff --git a/crates/domain/src/testing/fakes/store_watch_event.rs b/crates/domain/src/testing/fakes/store_watch_event.rs index 1bd2625f..750188fa 100644 --- a/crates/domain/src/testing/fakes/store_watch_event.rs +++ b/crates/domain/src/testing/fakes/store_watch_event.rs @@ -4,6 +4,7 @@ use uuid::Uuid; use crate::{ errors::DomainError, + events::DomainEvent, models::{WatchEvent, WatchEventStatus, WebhookToken}, ports::{WatchEventCommand, WatchEventQuery, WebhookTokenRepository}, value_objects::{UserId, WatchEventId, WebhookTokenId}, @@ -13,8 +14,9 @@ use super::InMemoryStore; #[async_trait] impl WatchEventCommand for InMemoryStore { - async fn save(&self, event: &WatchEvent) -> Result<(), DomainError> { + async fn save(&self, event: &WatchEvent, events: &[DomainEvent]) -> Result<(), DomainError> { self.watch_events.write().unwrap().push(event.clone()); + self.record_events(events); Ok(()) } diff --git a/crates/domain/src/testing/fakes/wrapup.rs b/crates/domain/src/testing/fakes/wrapup.rs index 88085418..e771cde2 100644 --- a/crates/domain/src/testing/fakes/wrapup.rs +++ b/crates/domain/src/testing/fakes/wrapup.rs @@ -3,7 +3,9 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use uuid::Uuid; -use crate::{errors::DomainError, ports::WrapUpRepository, value_objects::WrapUpId}; +use crate::{ + errors::DomainError, events::DomainEvent, ports::WrapUpRepository, value_objects::WrapUpId, +}; // ── PanicWrapUpStatsQuery ─────────────────────────────────────────────────── @@ -68,14 +70,22 @@ impl crate::ports::WrapUpStatsQuery for InMemoryWrapUpStatsQuery { pub struct InMemoryWrapUpRepository { pub store: Mutex>, + recorded: Mutex>, } impl InMemoryWrapUpRepository { pub fn new() -> Arc { Arc::new(Self { store: Mutex::new(Vec::new()), + recorded: Mutex::new(Vec::new()), }) } + + /// Events recorded atomically with a change (the outbox rows the sqlite + /// adapter writes in the same transaction). + pub fn recorded_events(&self) -> Vec { + self.recorded.lock().unwrap().clone() + } } #[async_trait] @@ -83,8 +93,10 @@ impl WrapUpRepository for InMemoryWrapUpRepository { async fn create( &self, record: &crate::models::wrapup::WrapUpRecord, + events: &[DomainEvent], ) -> Result<(), DomainError> { self.store.lock().unwrap().push(record.clone()); + self.recorded.lock().unwrap().extend_from_slice(events); Ok(()) } @@ -108,12 +120,15 @@ impl WrapUpRepository for InMemoryWrapUpRepository { &self, id: &WrapUpId, report: &crate::models::wrapup::WrapUpReport, + events: &[DomainEvent], ) -> Result<(), DomainError> { let mut store = self.store.lock().unwrap(); if let Some(rec) = store.iter_mut().find(|r| r.id == *id) { rec.status = crate::models::wrapup::WrapUpStatus::Ready; rec.report = Some(report.clone()); rec.completed_at = Some(chrono::Utc::now().naive_utc()); + drop(store); + self.recorded.lock().unwrap().extend_from_slice(events); Ok(()) } else { Err(DomainError::NotFound("wrapup record".into())) @@ -197,7 +212,11 @@ pub struct PanicWrapUpRepository; #[async_trait] impl WrapUpRepository for PanicWrapUpRepository { - async fn create(&self, _: &crate::models::wrapup::WrapUpRecord) -> Result<(), DomainError> { + async fn create( + &self, + _: &crate::models::wrapup::WrapUpRecord, + _: &[DomainEvent], + ) -> Result<(), DomainError> { panic!("PanicWrapUpRepository called") } async fn update_status( @@ -212,6 +231,7 @@ impl WrapUpRepository for PanicWrapUpRepository { &self, _: &WrapUpId, _: &crate::models::wrapup::WrapUpReport, + _: &[DomainEvent], ) -> Result<(), DomainError> { panic!("PanicWrapUpRepository called") } @@ -245,3 +265,77 @@ impl WrapUpRepository for PanicWrapUpRepository { panic!("PanicWrapUpRepository called") } } + +// ── RefusingWrapUpWrites ─────────────────────────────────────────────────── + +/// Delegates to an in-memory repository but refuses the two writes that carry +/// outbox events (`create`, `set_complete`), leaving `update_status` and reads +/// working so a refusal can be observed mid-flow. +pub struct RefusingWrapUpWrites { + pub inner: Arc, +} + +impl RefusingWrapUpWrites { + pub fn new(inner: Arc) -> Arc { + Arc::new(Self { inner }) + } +} + +#[async_trait] +impl WrapUpRepository for RefusingWrapUpWrites { + async fn create( + &self, + _: &crate::models::wrapup::WrapUpRecord, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(DomainError::InfrastructureError("refused".into())) + } + async fn update_status( + &self, + id: &WrapUpId, + status: &crate::models::wrapup::WrapUpStatus, + error: Option<&str>, + ) -> Result<(), DomainError> { + self.inner.update_status(id, status, error).await + } + async fn set_complete( + &self, + _: &WrapUpId, + _: &crate::models::wrapup::WrapUpReport, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(DomainError::InfrastructureError("refused".into())) + } + async fn get_by_id( + &self, + id: &WrapUpId, + ) -> Result, DomainError> { + self.inner.get_by_id(id).await + } + async fn list_for_user( + &self, + user_id: Uuid, + ) -> Result, DomainError> { + self.inner.list_for_user(user_id).await + } + async fn list_global(&self) -> Result, DomainError> { + self.inner.list_global().await + } + async fn find_existing( + &self, + user_id: Option, + start: chrono::NaiveDate, + end: chrono::NaiveDate, + ) -> Result, DomainError> { + self.inner.find_existing(user_id, start, end).await + } + async fn delete(&self, id: &WrapUpId) -> Result<(), DomainError> { + self.inner.delete(id).await + } + async fn delete_failed_older_than( + &self, + before: chrono::NaiveDateTime, + ) -> Result { + self.inner.delete_failed_older_than(before).await + } +} diff --git a/crates/domain/tests/testing/in_memory_store_test.rs b/crates/domain/tests/testing/in_memory_store_test.rs index a9f68af1..f8b9c69d 100644 --- a/crates/domain/tests/testing/in_memory_store_test.rs +++ b/crates/domain/tests/testing/in_memory_store_test.rs @@ -70,7 +70,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie let movie = test_movie("Heat"); UserRepository::save(&**store, &user).await.unwrap(); UserRepository::save(&**store, &other).await.unwrap(); - store.upsert_movie(&movie).await.unwrap(); + store.upsert_movie(&movie, &[]).await.unwrap(); store .add( @@ -118,6 +118,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie test_watched_at(), None, ), + &[], ) .await .unwrap(); @@ -221,7 +222,7 @@ async fn seed_movie_with_dependents( let user = test_user("alice"); UserRepository::save(&**store, &user).await.unwrap(); let movie = test_movie("Heat"); - store.upsert_movie(&movie).await.unwrap(); + store.upsert_movie(&movie, &[]).await.unwrap(); store .add( &WatchlistEntry::new(user.id().clone(), movie.id().clone()), @@ -241,6 +242,7 @@ async fn seed_movie_with_dependents( test_watched_at(), Some(movie.id().clone()), ), + &[], ) .await .unwrap(); @@ -261,7 +263,7 @@ async fn deleting_a_movie_removes_watchlist_entries_and_enrichment_profile() { assert!(store.rows_in(table) > 0, "{table} should be seeded"); } - store.delete_movie(movie.id()).await.unwrap(); + store.delete_movie(movie.id(), &[]).await.unwrap(); assert_eq!(store.movie_count(), 0); for table in MOVIE_CASCADE_TABLES { @@ -274,7 +276,7 @@ async fn deleting_a_movie_detaches_watch_events_instead_of_removing_them() { let store = InMemoryStore::new(); let (movie, user_id) = seed_movie_with_dependents(&store, false).await; - store.delete_movie(movie.id()).await.unwrap(); + store.delete_movie(movie.id(), &[]).await.unwrap(); let pending = store.list_pending(&user_id).await.unwrap(); assert_eq!(pending.len(), 1); @@ -286,7 +288,7 @@ async fn deleting_a_movie_that_has_reviews_is_refused_because_reviews_do_not_cas let store = InMemoryStore::new(); let (movie, _) = seed_movie_with_dependents(&store, true).await; - let result = store.delete_movie(movie.id()).await; + let result = store.delete_movie(movie.id(), &[]).await; assert!(matches!(result, Err(DomainError::InfrastructureError(_)))); assert_eq!(store.movie_count(), 1); @@ -367,12 +369,15 @@ async fn upserting_a_different_movie_with_a_taken_external_id_is_an_infrastructu let store = InMemoryStore::new(); let external = ExternalMetadataId::new("tt0113277".into()).unwrap(); store - .upsert_movie(&test_movie_with_external_id("Heat", 1995, external.clone())) + .upsert_movie( + &test_movie_with_external_id("Heat", 1995, external.clone()), + &[], + ) .await .unwrap(); let result = store - .upsert_movie(&test_movie_with_external_id("Heat 2", 2030, external)) + .upsert_movie(&test_movie_with_external_id("Heat 2", 2030, external), &[]) .await; assert!(matches!(result, Err(DomainError::InfrastructureError(_)))); @@ -395,3 +400,117 @@ async fn creating_a_refresh_session_with_a_taken_token_is_an_infrastructure_erro assert!(matches!(result, Err(DomainError::InfrastructureError(_)))); } + +fn movie_discovered(movie: &Movie) -> domain::events::DomainEvent { + domain::events::DomainEvent::MovieDiscovered { + movie_id: movie.id().clone(), + external_metadata_id: ExternalMetadataId::new("tmdb:1".into()).unwrap(), + } +} + +fn movie_deleted(movie: &Movie) -> domain::events::DomainEvent { + domain::events::DomainEvent::MovieDeleted { + movie_id: movie.id().clone(), + poster_path: None, + } +} + +#[tokio::test] +async fn upserting_a_movie_records_its_events() { + let store = InMemoryStore::new(); + let movie = test_movie("Heat"); + + store + .upsert_movie(&movie, &[movie_discovered(&movie)]) + .await + .unwrap(); + + assert_eq!(store.movie_count(), 1); + assert_eq!(store.recorded_events().len(), 1); +} + +#[tokio::test] +async fn a_refused_movie_upsert_records_no_events() { + let store = InMemoryStore::new(); + let external = ExternalMetadataId::new("tt0113277".into()).unwrap(); + let first = test_movie_with_external_id("Heat", 1995, external.clone()); + store.upsert_movie(&first, &[]).await.unwrap(); + let clash = test_movie_with_external_id("Heat 2", 2030, external); + + let result = store + .upsert_movie(&clash, &[movie_discovered(&clash)]) + .await; + + assert!(result.is_err()); + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn deleting_a_movie_records_its_events() { + let store = InMemoryStore::new(); + let movie = test_movie("Heat"); + store.upsert_movie(&movie, &[]).await.unwrap(); + + store + .delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await + .unwrap(); + + assert_eq!(store.recorded_events().len(), 1); +} + +#[tokio::test] +async fn deleting_an_absent_movie_records_no_events() { + let store = InMemoryStore::new(); + let movie = test_movie("Heat"); + + store + .delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await + .unwrap(); + + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn a_refused_movie_delete_records_no_events() { + let store = InMemoryStore::new(); + let (movie, _) = seed_movie_with_dependents(&store, true).await; + + let result = store + .delete_movie(movie.id(), &[movie_deleted(&movie)]) + .await; + + assert!(result.is_err()); + assert!(store.recorded_events().is_empty()); +} + +#[tokio::test] +async fn saving_a_watch_event_records_its_events() { + let store = InMemoryStore::new(); + let user_id = UserId::generate(); + let event = WatchEvent::new( + user_id.clone(), + "Heat".into(), + Some(1995), + None, + WatchEventSource::Jellyfin, + test_watched_at(), + None, + ); + + WatchEventCommand::save( + &*store, + &event, + &[domain::events::DomainEvent::WatchEventIngested { + user_id, + title: "Heat".into(), + source: "jellyfin".into(), + }], + ) + .await + .unwrap(); + + assert_eq!(store.rows_in("watch_events"), 1); + assert_eq!(store.recorded_events().len(), 1); +}