diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index 3fa93716..096a1056 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -101,7 +101,6 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { movie_repo: std::sync::Arc::clone(&movie_repo), diary_repo, review_store, - event_publisher: std::sync::Arc::clone(&event_publisher), instance: instance.clone(), }); let watchlist_handler = std::sync::Arc::new(watchlist_handler::WatchlistObjectHandler { diff --git a/crates/adapters/activitypub/src/remote_review_repository.rs b/crates/adapters/activitypub/src/remote_review_repository.rs index 682d4384..b246a7cd 100644 --- a/crates/adapters/activitypub/src/remote_review_repository.rs +++ b/crates/adapters/activitypub/src/remote_review_repository.rs @@ -1,7 +1,7 @@ use anyhow::Result; use async_trait::async_trait; use chrono::NaiveDateTime; -use domain::models::Review; +use domain::{events::DomainEvent, models::Review}; pub struct RemoteReviewUpdate<'a> { pub ap_id: &'a str, @@ -16,6 +16,7 @@ pub struct RemoteReviewUpdate<'a> { #[async_trait] pub trait RemoteReviewRepository: Send + Sync { + #[allow(clippy::too_many_arguments)] async fn save_remote_review( &self, review: &Review, @@ -24,6 +25,7 @@ pub trait RemoteReviewRepository: Send + Sync { release_year: u16, external_metadata_id: Option<&str>, poster_url: Option<&str>, + events: &[DomainEvent], ) -> Result<()>; async fn delete_remote_review(&self, ap_id: &str, actor_url: &str) -> Result<()>; diff --git a/crates/adapters/activitypub/src/review_handler.rs b/crates/adapters/activitypub/src/review_handler.rs index 794bc8fb..387f4c5f 100644 --- a/crates/adapters/activitypub/src/review_handler.rs +++ b/crates/adapters/activitypub/src/review_handler.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use domain::{ events::DomainEvent, models::ReviewSource, - ports::{DiaryQuery, EventPublisher, LocalApContentQuery, MovieQuery}, + ports::{DiaryQuery, LocalApContentQuery, MovieQuery}, value_objects::{ Comment, ExternalMetadataId, InstanceIdentity, MovieId, Rating, ReviewId, UserId, }, @@ -21,7 +21,6 @@ pub struct ReviewObjectHandler { pub movie_repo: Arc, pub diary_repo: Arc, pub review_store: Arc, - pub event_publisher: Arc, pub instance: InstanceIdentity, } @@ -160,6 +159,20 @@ impl ApObjectHandler for ReviewObjectHandler { spoilers: obj.contains_spoilers, }); + // The enrichment request rides the review write (one transaction). + let events: Vec = obj + .external_metadata_id + .as_ref() + .and_then(|ext_id| ExternalMetadataId::new(ext_id.clone()).ok()) + .map( + |external_metadata_id| DomainEvent::MovieEnrichmentRequested { + movie_id: movie_id.clone(), + external_metadata_id, + }, + ) + .into_iter() + .collect(); + self.review_store .save_remote_review( &review, @@ -168,21 +181,10 @@ impl ApObjectHandler for ReviewObjectHandler { obj.release_year, obj.external_metadata_id.as_deref(), obj.poster_url.as_deref(), + &events, ) .await?; - if let Some(ref ext_id_str) = obj.external_metadata_id - && let Ok(external_metadata_id) = ExternalMetadataId::new(ext_id_str.clone()) - { - let _ = self - .event_publisher - .publish(&DomainEvent::MovieEnrichmentRequested { - movie_id: movie_id.clone(), - external_metadata_id, - }) - .await; - } - Ok(()) } diff --git a/crates/adapters/activitypub/src/social_adapter.rs b/crates/adapters/activitypub/src/social_adapter.rs index c0a16c78..e0a54ca2 100644 --- a/crates/adapters/activitypub/src/social_adapter.rs +++ b/crates/adapters/activitypub/src/social_adapter.rs @@ -3,7 +3,10 @@ use std::sync::Arc; use async_trait::async_trait; use domain::{ errors::DomainError, - ports::{BlockQuery, FollowGraphQuery, LocalSocial, SocialCommand, UserRepository}, + events::DomainEvent, + ports::{ + BlockQuery, EventPublisher, FollowGraphQuery, LocalSocial, SocialCommand, UserRepository, + }, value_objects::{ FollowRelation, FollowTarget, InstanceIdentity, SocialActor, SocialIdentity, UserId, }, @@ -16,6 +19,7 @@ pub struct CompositeSocialAdapter { ap_service: Arc, user_repo: Arc, instance: InstanceIdentity, + event_publisher: Arc, } impl CompositeSocialAdapter { @@ -24,12 +28,23 @@ impl CompositeSocialAdapter { ap_service: Arc, user_repo: Arc, instance: InstanceIdentity, + event_publisher: Arc, ) -> Self { Self { local, ap_service, user_repo, instance, + event_publisher, + } + } + + /// Best-effort publish for writes k_ap owns (no shared transaction). + async fn announce(&self, events: &[DomainEvent]) { + for event in events { + if let Err(e) = self.event_publisher.publish(event).await { + tracing::warn!("failed to publish {event:?} after k_ap write: {e}"); + } } } } @@ -40,11 +55,19 @@ fn ap_err(e: anyhow::Error) -> DomainError { #[async_trait] impl SocialCommand for CompositeSocialAdapter { - async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError> { + async fn follow( + &self, + follower: &UserId, + target: &FollowTarget, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let identity = self.local.resolve_target(target).await?; if let SocialIdentity::Local(_) = identity { - return self.local.follow_resolved(follower, &identity).await; + return self + .local + .follow_resolved(follower, &identity, events) + .await; } let handle = match target { @@ -61,24 +84,33 @@ impl SocialCommand for CompositeSocialAdapter { SocialIdentity::Remote { actor_url } => actor_url.clone(), }, }; + // Not atomic: k_ap owns the transaction, so the events are published + // best-effort after its write succeeds. self.ap_service .follow(follower.value(), &handle) .await - .map_err(ap_err) + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) } async fn unfollow( &self, follower: &UserId, target: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { match target { - SocialIdentity::Local(_) => self.local.unfollow(follower, target).await, - SocialIdentity::Remote { .. } => self - .ap_service - .unfollow(follower.value(), &self.instance.actor_url_of(target)) - .await - .map_err(ap_err), + SocialIdentity::Local(_) => self.local.unfollow(follower, target, events).await, + // Not atomic: k_ap owns the transaction. + SocialIdentity::Remote { .. } => { + self.ap_service + .unfollow(follower.value(), &self.instance.actor_url_of(target)) + .await + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) + } } } @@ -86,14 +118,19 @@ impl SocialCommand for CompositeSocialAdapter { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { match requester { - SocialIdentity::Local(_) => self.local.accept_follow(owner, requester).await, - SocialIdentity::Remote { .. } => self - .ap_service - .accept_follower(owner.value(), &self.instance.actor_url_of(requester)) - .await - .map_err(ap_err), + SocialIdentity::Local(_) => self.local.accept_follow(owner, requester, events).await, + // Not atomic: k_ap owns the transaction. + SocialIdentity::Remote { .. } => { + self.ap_service + .accept_follower(owner.value(), &self.instance.actor_url_of(requester)) + .await + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) + } } } @@ -101,14 +138,19 @@ impl SocialCommand for CompositeSocialAdapter { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { match requester { - SocialIdentity::Local(_) => self.local.reject_follow(owner, requester).await, - SocialIdentity::Remote { .. } => self - .ap_service - .reject_follower(owner.value(), &self.instance.actor_url_of(requester)) - .await - .map_err(ap_err), + SocialIdentity::Local(_) => self.local.reject_follow(owner, requester, events).await, + // Not atomic: k_ap owns the transaction. + SocialIdentity::Remote { .. } => { + self.ap_service + .reject_follower(owner.value(), &self.instance.actor_url_of(requester)) + .await + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) + } } } @@ -116,31 +158,52 @@ impl SocialCommand for CompositeSocialAdapter { &self, owner: &UserId, follower: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { match follower { - SocialIdentity::Local(_) => self.local.remove_follower(owner, follower).await, - SocialIdentity::Remote { .. } => self - .ap_service - .remove_follower(owner.value(), &self.instance.actor_url_of(follower)) - .await - .map_err(ap_err), + SocialIdentity::Local(_) => self.local.remove_follower(owner, follower, events).await, + // Not atomic: k_ap owns the transaction. + SocialIdentity::Remote { .. } => { + self.ap_service + .remove_follower(owner.value(), &self.instance.actor_url_of(follower)) + .await + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) + } } } - async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> { + /// Not atomic: k_ap owns the transaction. + async fn block( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(target); self.ap_service .block_actor(blocker.value(), &actor_url) .await - .map_err(ap_err) + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) } - async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> { + /// Not atomic: k_ap owns the transaction. + async fn unblock( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(target); self.ap_service .unblock_actor(blocker.value(), &actor_url) .await - .map_err(ap_err) + .map_err(ap_err)?; + self.announce(events).await; + Ok(()) } } diff --git a/crates/adapters/sqlite-federation/Cargo.toml b/crates/adapters/sqlite-federation/Cargo.toml index 0f7549f1..4ce7d6bd 100644 --- a/crates/adapters/sqlite-federation/Cargo.toml +++ b/crates/adapters/sqlite-federation/Cargo.toml @@ -18,4 +18,5 @@ async-trait = { workspace = true } [dev-dependencies] tokio = { workspace = true } +domain = { workspace = true, features = ["test-helpers"] } uuid = { workspace = true } diff --git a/crates/adapters/sqlite-federation/src/review.rs b/crates/adapters/sqlite-federation/src/review.rs index c648c684..8646343c 100644 --- a/crates/adapters/sqlite-federation/src/review.rs +++ b/crates/adapters/sqlite-federation/src/review.rs @@ -1,13 +1,17 @@ use activitypub::{RemoteReviewRepository, RemoteReviewUpdate}; use anyhow::{Result, anyhow}; use async_trait::async_trait; -use domain::models::{Review, ReviewSource}; +use domain::{ + events::DomainEvent, + models::{Review, ReviewSource}, +}; use super::SqliteFederationRepository; -use sqlite::persistence::datetime_to_str; +use sqlite::persistence::{datetime_to_str, enqueue_events}; #[async_trait] impl RemoteReviewRepository for SqliteFederationRepository { + #[allow(clippy::too_many_arguments)] async fn save_remote_review( &self, review: &Review, @@ -16,6 +20,7 @@ impl RemoteReviewRepository for SqliteFederationRepository { release_year: u16, external_metadata_id: Option<&str>, poster_url: Option<&str>, + events: &[DomainEvent], ) -> Result<()> { let actor_url = match review.source() { ReviewSource::Remote { actor_url } => actor_url.clone(), @@ -24,6 +29,7 @@ impl RemoteReviewRepository for SqliteFederationRepository { } }; let movie_id = review.movie_id().value().to_string(); + let mut tx = self.pool.begin().await?; let _ = sqlx::query( "INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path) VALUES (?, ?, ?, ?, NULL, ?) @@ -36,7 +42,7 @@ impl RemoteReviewRepository for SqliteFederationRepository { .bind(movie_title) .bind(release_year.max(1888) as i64) .bind(poster_url) - .execute(&self.pool) + .execute(&mut *tx) .await?; let id = review.id().value().to_string(); @@ -59,8 +65,10 @@ impl RemoteReviewRepository for SqliteFederationRepository { .bind(&actor_url) .bind(ap_id) .bind(review.spoilers()) - .execute(&self.pool) + .execute(&mut *tx) .await?; + enqueue_events(&mut tx, events).await?; + tx.commit().await?; Ok(()) } diff --git a/crates/adapters/sqlite-federation/tests/federation.rs b/crates/adapters/sqlite-federation/tests/federation.rs index 3387415d..e44fcd1d 100644 --- a/crates/adapters/sqlite-federation/tests/federation.rs +++ b/crates/adapters/sqlite-federation/tests/federation.rs @@ -4,3 +4,5 @@ mod actor_block_test; mod announce_test; #[path = "federation/domain_block_test.rs"] mod domain_block_test; +#[path = "federation/remote_review_test.rs"] +mod remote_review_test; diff --git a/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs b/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs new file mode 100644 index 00000000..5e8e083f --- /dev/null +++ b/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs @@ -0,0 +1,115 @@ +//! A remote review and the enrichment request it triggers commit together. + +use activitypub::RemoteReviewRepository; +use domain::{ + events::DomainEvent, + models::{PersistedReview, Review, ReviewSource}, + value_objects::{Comment, ExternalMetadataId, MovieId, Rating, ReviewId, UserId}, +}; +use sqlite_federation::SqliteFederationRepository; +use sqlx::SqlitePool; + +const ACTOR: &str = "https://remote.example/users/ann"; +const AP_ID: &str = "https://remote.example/reviews/1"; + +async fn migrated_pool() -> SqlitePool { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + sqlite::migrate(&pool).await.unwrap(); + pool +} + +async fn count(pool: &SqlitePool, table: &str) -> i64 { + sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}")) + .fetch_one(pool) + .await + .unwrap() +} + +fn remote_review(movie_id: &MovieId) -> Review { + let now = chrono::Utc::now().naive_utc(); + Review::from_persistence(PersistedReview { + id: ReviewId::generate(), + movie_id: movie_id.clone(), + user_id: UserId::generate(), + rating: Rating::new(4).unwrap(), + comment: Some(Comment::new("good".into()).unwrap()), + watched_at: now, + created_at: now, + source: ReviewSource::Remote { + actor_url: ACTOR.into(), + }, + watch_medium: None, + spoilers: false, + }) +} + +fn enrichment_requested(movie_id: &MovieId) -> DomainEvent { + DomainEvent::MovieEnrichmentRequested { + movie_id: movie_id.clone(), + external_metadata_id: ExternalMetadataId::new("tmdb:603".into()).unwrap(), + } +} + +async fn save(pool: &SqlitePool, review: &Review, events: &[DomainEvent]) -> anyhow::Result<()> { + SqliteFederationRepository::new(pool.clone()) + .save_remote_review( + review, + AP_ID, + "The Matrix", + 1999, + Some("tmdb:603"), + None, + events, + ) + .await +} + +#[tokio::test] +async fn saving_a_remote_review_writes_it_and_its_event_in_one_commit() { + let pool = migrated_pool().await; + let movie_id = MovieId::generate(); + let review = remote_review(&movie_id); + + save(&pool, &review, &[enrichment_requested(&movie_id)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "reviews").await, 1); + 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_remote_review_and_its_movie() { + let pool = migrated_pool().await; + sqlx::query( + "CREATE TRIGGER fail_event_insert BEFORE INSERT ON event_queue \ + BEGIN SELECT RAISE(ABORT, 'enqueue failed'); END", + ) + .execute(&pool) + .await + .unwrap(); + let movie_id = MovieId::generate(); + let review = remote_review(&movie_id); + + let result = save(&pool, &review, &[enrichment_requested(&movie_id)]).await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "reviews").await, 0); + assert_eq!(count(&pool, "movies").await, 0); +} + +#[tokio::test] +async fn a_remote_review_without_events_enqueues_nothing() { + let pool = migrated_pool().await; + let movie_id = MovieId::generate(); + + save(&pool, &remote_review(&movie_id), &[]).await.unwrap(); + + assert_eq!(count(&pool, "reviews").await, 1); + assert_eq!(count(&pool, "event_queue").await, 0); +} diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index eeada475..0acc52e4 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -22,7 +22,7 @@ pub use repositories::{ /// Row helpers shared with the `sqlite-federation` crate (which cannot live /// here because it depends on `activitypub`/`k-ap`). pub mod persistence { - pub use crate::shared::datetime_to_str; + pub use crate::shared::{datetime_to_str, enqueue_events}; } pub fn create_profile_fields_repo( diff --git a/crates/adapters/sqlite/src/repositories/social/follow/command.rs b/crates/adapters/sqlite/src/repositories/social/follow/command.rs index 7424c2a7..aa859e7b 100644 --- a/crates/adapters/sqlite/src/repositories/social/follow/command.rs +++ b/crates/adapters/sqlite/src/repositories/social/follow/command.rs @@ -1,130 +1,126 @@ use async_trait::async_trait; use chrono::Utc; -use domain::{errors::DomainError, value_objects::FollowStatus}; +use domain::{ + errors::DomainError, + events::DomainEvent, + ports::{FollowCommand, FollowWrite}, +}; -use crate::shared::datetime_to_str; +use crate::shared::{datetime_to_str, enqueue_events}; use super::infra_err; use super::rows::follow_status_to_str; use crate::repositories::social::SqliteSocialRepository; #[async_trait] -impl domain::ports::FollowCommand for SqliteSocialRepository { - async fn add_follow( +impl FollowCommand for SqliteSocialRepository { + async fn apply( &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - status: FollowStatus, + writes: &[FollowWrite], + events: &[DomainEvent], ) -> Result<(), DomainError> { - let uid = follower_id.to_string(); - let status_str = follow_status_to_str(&status); - let now = datetime_to_str(&Utc::now().naive_utc()); - sqlx::query( - "INSERT INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at, status) - VALUES (?1, ?2, '', ?3, ?4) - ON CONFLICT(local_user_id, remote_actor_url) DO UPDATE SET status = excluded.status", - ) - .bind(&uid) - .bind(target_actor_url) - .bind(&now) - .bind(status_str) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) - } - - async fn update_follow_status( - &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - let uid = follower_id.to_string(); - let status_str = follow_status_to_str(&status); - sqlx::query( - "UPDATE ap_following SET status = ?1 WHERE local_user_id = ?2 AND remote_actor_url = ?3", - ) - .bind(status_str) - .bind(&uid) - .bind(target_actor_url) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) - } - - async fn remove_follow( - &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - ) -> Result<(), DomainError> { - let uid = follower_id.to_string(); - sqlx::query("DELETE FROM ap_following WHERE local_user_id = ?1 AND remote_actor_url = ?2") - .bind(&uid) - .bind(target_actor_url) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) - } - - async fn add_follower( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - let uid = local_user_id.to_string(); - let status_str = follow_status_to_str(&status); - let now = datetime_to_str(&Utc::now().naive_utc()); - sqlx::query( - "INSERT INTO ap_followers (local_user_id, remote_actor_url, status, created_at, follow_activity_id) - VALUES (?1, ?2, ?3, ?4, '') - ON CONFLICT(local_user_id, remote_actor_url) DO UPDATE SET status = excluded.status", - ) - .bind(&uid) - .bind(follower_actor_url) - .bind(status_str) - .bind(&now) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) - } - - async fn update_follower_status( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - let uid = local_user_id.to_string(); - let status_str = follow_status_to_str(&status); - sqlx::query( - "UPDATE ap_followers SET status = ?1 WHERE local_user_id = ?2 AND remote_actor_url = ?3", - ) - .bind(status_str) - .bind(&uid) - .bind(follower_actor_url) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) - } - - async fn remove_follower_record( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, - ) -> Result<(), DomainError> { - let uid = local_user_id.to_string(); - sqlx::query("DELETE FROM ap_followers WHERE local_user_id = ?1 AND remote_actor_url = ?2") - .bind(&uid) - .bind(follower_actor_url) - .execute(&self.pool) - .await - .map_err(infra_err)?; - Ok(()) + let mut tx = self.pool.begin().await.map_err(infra_err)?; + for write in writes { + apply_write(&mut tx, write).await?; + } + enqueue_events(&mut tx, events).await?; + tx.commit().await.map_err(infra_err) } } + +async fn apply_write( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + write: &FollowWrite, +) -> Result<(), DomainError> { + let now = datetime_to_str(&Utc::now().naive_utc()); + match write { + FollowWrite::AddFollow { + follower_id, + target_actor_url, + status, + } => { + sqlx::query( + "INSERT INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at, status) + VALUES (?1, ?2, '', ?3, ?4) + ON CONFLICT(local_user_id, remote_actor_url) DO UPDATE SET status = excluded.status", + ) + .bind(follower_id.to_string()) + .bind(target_actor_url) + .bind(&now) + .bind(follow_status_to_str(status)) + .execute(&mut **tx) + .await + } + FollowWrite::UpdateFollowStatus { + follower_id, + target_actor_url, + status, + } => { + sqlx::query( + "UPDATE ap_following SET status = ?1 WHERE local_user_id = ?2 AND remote_actor_url = ?3", + ) + .bind(follow_status_to_str(status)) + .bind(follower_id.to_string()) + .bind(target_actor_url) + .execute(&mut **tx) + .await + } + FollowWrite::RemoveFollow { + follower_id, + target_actor_url, + } => { + sqlx::query( + "DELETE FROM ap_following WHERE local_user_id = ?1 AND remote_actor_url = ?2", + ) + .bind(follower_id.to_string()) + .bind(target_actor_url) + .execute(&mut **tx) + .await + } + FollowWrite::AddFollower { + local_user_id, + follower_actor_url, + status, + } => { + sqlx::query( + "INSERT INTO ap_followers (local_user_id, remote_actor_url, status, created_at, follow_activity_id) + VALUES (?1, ?2, ?3, ?4, '') + ON CONFLICT(local_user_id, remote_actor_url) DO UPDATE SET status = excluded.status", + ) + .bind(local_user_id.to_string()) + .bind(follower_actor_url) + .bind(follow_status_to_str(status)) + .bind(&now) + .execute(&mut **tx) + .await + } + FollowWrite::UpdateFollowerStatus { + local_user_id, + follower_actor_url, + status, + } => { + sqlx::query( + "UPDATE ap_followers SET status = ?1 WHERE local_user_id = ?2 AND remote_actor_url = ?3", + ) + .bind(follow_status_to_str(status)) + .bind(local_user_id.to_string()) + .bind(follower_actor_url) + .execute(&mut **tx) + .await + } + FollowWrite::RemoveFollowerRecord { + local_user_id, + follower_actor_url, + } => { + sqlx::query( + "DELETE FROM ap_followers WHERE local_user_id = ?1 AND remote_actor_url = ?2", + ) + .bind(local_user_id.to_string()) + .bind(follower_actor_url) + .execute(&mut **tx) + .await + } + } + .map_err(infra_err)?; + Ok(()) +} diff --git a/crates/adapters/sqlite/src/shared.rs b/crates/adapters/sqlite/src/shared.rs index 151839d0..035e240f 100644 --- a/crates/adapters/sqlite/src/shared.rs +++ b/crates/adapters/sqlite/src/shared.rs @@ -13,7 +13,7 @@ use domain::{ /// THE outbox write path: inserts one `event_queue` row per event, each with a /// fresh `EventId`, inside the caller's transaction so a change and the events /// announcing it commit or roll back together. -pub(crate) async fn enqueue_events( +pub async fn enqueue_events( tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, events: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { diff --git a/crates/adapters/sqlite/tests/cascade_test.rs b/crates/adapters/sqlite/tests/cascade_test.rs index eba3f58b..54eac773 100644 --- a/crates/adapters/sqlite/tests/cascade_test.rs +++ b/crates/adapters/sqlite/tests/cascade_test.rs @@ -11,7 +11,8 @@ use domain::models::{ WatchEvent, WatchEventSource, WatchlistEntry, WebhookToken, }; use domain::ports::{ - FollowCommand, UserProfileFieldsRepository, WatchEventCommand, WebhookTokenRepository, + FollowCommand, FollowWrite, UserProfileFieldsRepository, WatchEventCommand, + WebhookTokenRepository, }; use domain::testing::{ MOVIE_CASCADE_TABLES, USER_CASCADE_TABLES, USER_CASCADE_TABLES_OUTSIDE_STORE, test_goal, @@ -184,11 +185,21 @@ async fn seed_user_with_dependents( domain::value_objects::InstanceIdentity::new("https://md.example"), ); social - .add_follow(user_id.value(), REMOTE_ACTOR, FollowStatus::Accepted) - .await - .unwrap(); - social - .add_follower(user_id.value(), REMOTE_ACTOR, FollowStatus::Accepted) + .apply( + &[ + FollowWrite::AddFollow { + follower_id: user_id.value(), + target_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + FollowWrite::AddFollower { + local_user_id: user_id.value(), + follower_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + ], + &[], + ) .await .unwrap(); // No repository in this crate owns these two tables (federation crate). diff --git a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs index ff145b0a..0a64450a 100644 --- a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs +++ b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs @@ -712,3 +712,85 @@ async fn a_failing_enqueue_rolls_back_the_watch_event_save() { assert!(result.is_err()); assert_eq!(count(&pool, "watch_events").await, 0); } + +fn follow_writes(user_id: &UserId) -> Vec { + use domain::{ports::FollowWrite, value_objects::FollowStatus}; + vec![ + FollowWrite::AddFollower { + local_user_id: user_id.value(), + follower_actor_url: "https://md.example/users/other".into(), + status: FollowStatus::Pending, + }, + FollowWrite::AddFollow { + follower_id: user_id.value(), + target_actor_url: "https://md.example/users/other".into(), + status: FollowStatus::Pending, + }, + ] +} + +fn follow_requested(user_id: &UserId) -> DomainEvent { + DomainEvent::FollowRequested { + follower: user_id.clone(), + target: domain::value_objects::FollowTarget::Handle("@other@md.example".into()), + } +} + +fn social_repo(pool: &SqlitePool) -> sqlite::SqliteSocialRepository { + sqlite::SqliteSocialRepository::new( + pool.clone(), + domain::value_objects::InstanceIdentity::new("https://md.example"), + ) +} + +#[tokio::test] +async fn a_local_follow_writes_both_rows_and_its_event_in_one_commit() { + use domain::ports::FollowCommand; + let (pool, user) = pool_with_user().await; + + social_repo(&pool) + .apply(&follow_writes(user.id()), &[follow_requested(user.id())]) + .await + .unwrap(); + + assert_eq!(count(&pool, "ap_followers").await, 1); + assert_eq!(count(&pool, "ap_following").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1); +} + +#[tokio::test] +async fn a_failing_enqueue_rolls_back_both_follow_rows() { + use domain::ports::FollowCommand; + let (pool, user) = pool_with_user().await; + fail_event_inserts(&pool).await; + + let result = social_repo(&pool) + .apply(&follow_writes(user.id()), &[follow_requested(user.id())]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "ap_followers").await, 0); + assert_eq!(count(&pool, "ap_following").await, 0); +} + +#[tokio::test] +async fn a_failing_follow_write_rolls_back_the_rows_already_applied() { + use domain::ports::{FollowCommand, FollowWrite}; + let (pool, user) = pool_with_user().await; + let mut writes = follow_writes(user.id()); + // An unknown local user violates the ap_following foreign key after the + // follower row was already written. + writes.push(FollowWrite::AddFollow { + follower_id: uuid::Uuid::new_v4(), + target_actor_url: "https://md.example/users/other".into(), + status: domain::value_objects::FollowStatus::Pending, + }); + + let result = social_repo(&pool) + .apply(&writes, &[follow_requested(user.id())]) + .await; + + assert!(result.is_err()); + assert_eq!(count(&pool, "ap_followers").await, 0); + assert_eq!(count(&pool, "event_queue").await, 0); +} diff --git a/crates/application/src/social/deps.rs b/crates/application/src/social/deps.rs index 4e9dcf8d..ac9c7055 100644 --- a/crates/application/src/social/deps.rs +++ b/crates/application/src/social/deps.rs @@ -1,10 +1,9 @@ use std::sync::Arc; -use domain::ports::{BlockQuery, EventPublisher, FollowGraphQuery, SocialCommand}; +use domain::ports::{BlockQuery, FollowGraphQuery, SocialCommand}; pub struct SocialCommandDeps { pub social_command: Arc, - pub event_publisher: Arc, } pub struct SocialQueryDeps { diff --git a/crates/application/src/social/execute.rs b/crates/application/src/social/execute.rs index 6b928f0a..a773da24 100644 --- a/crates/application/src/social/execute.rs +++ b/crates/application/src/social/execute.rs @@ -2,61 +2,78 @@ use domain::{errors::DomainError, events::DomainEvent, value_objects::UserId}; use super::{commands::SocialCmd, deps::SocialCommandDeps}; +/// The command's event rides the write: the social port records it with the +/// change (atomically where the adapter owns the transaction). pub async fn execute_command(deps: &SocialCommandDeps, cmd: SocialCmd) -> Result<(), DomainError> { - let event = match cmd { + let command = &deps.social_command; + match cmd { SocialCmd::Follow { follower_id, target, } => { let follower = UserId::from_uuid(follower_id); - deps.social_command.follow(&follower, &target).await?; - DomainEvent::FollowRequested { follower, target } + let events = [DomainEvent::FollowRequested { + follower: follower.clone(), + target: target.clone(), + }]; + command.follow(&follower, &target, &events).await } SocialCmd::Unfollow { follower_id, target, } => { let follower = UserId::from_uuid(follower_id); - deps.social_command.unfollow(&follower, &target).await?; - DomainEvent::Unfollowed { follower, target } + let events = [DomainEvent::Unfollowed { + follower: follower.clone(), + target: target.clone(), + }]; + command.unfollow(&follower, &target, &events).await } SocialCmd::AcceptFollow { owner_id, requester, } => { let owner = UserId::from_uuid(owner_id); - deps.social_command - .accept_follow(&owner, &requester) - .await?; - DomainEvent::FollowAccepted { owner, requester } + let events = [DomainEvent::FollowAccepted { + owner: owner.clone(), + requester: requester.clone(), + }]; + command.accept_follow(&owner, &requester, &events).await } SocialCmd::RejectFollow { owner_id, requester, } => { let owner = UserId::from_uuid(owner_id); - deps.social_command - .reject_follow(&owner, &requester) - .await?; - DomainEvent::FollowRejected { owner, requester } + let events = [DomainEvent::FollowRejected { + owner: owner.clone(), + requester: requester.clone(), + }]; + command.reject_follow(&owner, &requester, &events).await } SocialCmd::RemoveFollower { owner_id, follower } => { let owner = UserId::from_uuid(owner_id); - deps.social_command - .remove_follower(&owner, &follower) - .await?; - DomainEvent::FollowerRemoved { owner, follower } + let events = [DomainEvent::FollowerRemoved { + owner: owner.clone(), + follower: follower.clone(), + }]; + command.remove_follower(&owner, &follower, &events).await } SocialCmd::Block { blocker_id, target } => { let blocker = UserId::from_uuid(blocker_id); - deps.social_command.block(&blocker, &target).await?; - DomainEvent::ActorBlocked { blocker, target } + let events = [DomainEvent::ActorBlocked { + blocker: blocker.clone(), + target: target.clone(), + }]; + command.block(&blocker, &target, &events).await } SocialCmd::Unblock { blocker_id, target } => { let blocker = UserId::from_uuid(blocker_id); - deps.social_command.unblock(&blocker, &target).await?; - DomainEvent::ActorUnblocked { blocker, target } + let events = [DomainEvent::ActorUnblocked { + blocker: blocker.clone(), + target: target.clone(), + }]; + command.unblock(&blocker, &target, &events).await } - }; - deps.event_publisher.publish(&event).await + } } diff --git a/crates/application/src/social/local_service.rs b/crates/application/src/social/local_service.rs index 599362b4..69fcac60 100644 --- a/crates/application/src/social/local_service.rs +++ b/crates/application/src/social/local_service.rs @@ -3,9 +3,10 @@ use std::sync::Arc; use async_trait::async_trait; use domain::{ errors::DomainError, + events::DomainEvent, ports::{ BlockQuery, FollowCommand, FollowGraphQuery, FollowQuery, FollowTargetResolver, - ResolvedFollow, SocialCommand, UserRepository, + FollowWrite, ResolvedFollow, SocialCommand, UserRepository, }, value_objects::{ FollowRelation, FollowStatus, FollowTarget, InstanceIdentity, SocialActor, SocialIdentity, @@ -72,27 +73,41 @@ fn remote_unsupported(actor: &str) -> DomainError { #[async_trait] impl SocialCommand for LocalSocialService { - async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError> { + async fn follow( + &self, + follower: &UserId, + target: &FollowTarget, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let identity = self.resolve_target_identity(target).await?; - self.follow_resolved(follower, &identity).await + self.follow_resolved(follower, &identity, events).await } async fn unfollow( &self, follower: &UserId, target: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(target); match target { SocialIdentity::Local(target_id) => { let follower_url = self.instance.actor_url_for(follower); self.follow_command - .remove_follow(follower.value(), &actor_url) - .await?; - self.follow_command - .remove_follower_record(target_id.value(), &follower_url) - .await?; - Ok(()) + .apply( + &[ + FollowWrite::RemoveFollow { + follower_id: follower.value(), + target_actor_url: actor_url, + }, + FollowWrite::RemoveFollowerRecord { + local_user_id: target_id.value(), + follower_actor_url: follower_url, + }, + ], + events, + ) + .await } SocialIdentity::Remote { .. } => Err(remote_unsupported(&actor_url)), } @@ -102,18 +117,29 @@ impl SocialCommand for LocalSocialService { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(requester); match requester { SocialIdentity::Local(requester_id) => { let owner_url = self.instance.actor_url_for(owner); self.follow_command - .update_follower_status(owner.value(), &actor_url, FollowStatus::Accepted) - .await?; - self.follow_command - .update_follow_status(requester_id.value(), &owner_url, FollowStatus::Accepted) - .await?; - Ok(()) + .apply( + &[ + FollowWrite::UpdateFollowerStatus { + local_user_id: owner.value(), + follower_actor_url: actor_url, + status: FollowStatus::Accepted, + }, + FollowWrite::UpdateFollowStatus { + follower_id: requester_id.value(), + target_actor_url: owner_url, + status: FollowStatus::Accepted, + }, + ], + events, + ) + .await } SocialIdentity::Remote { .. } => Err(remote_unsupported(&actor_url)), } @@ -123,18 +149,28 @@ impl SocialCommand for LocalSocialService { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(requester); match requester { SocialIdentity::Local(requester_id) => { let owner_url = self.instance.actor_url_for(owner); self.follow_command - .update_follower_status(owner.value(), &actor_url, FollowStatus::Rejected) - .await?; - self.follow_command - .remove_follow(requester_id.value(), &owner_url) - .await?; - Ok(()) + .apply( + &[ + FollowWrite::UpdateFollowerStatus { + local_user_id: owner.value(), + follower_actor_url: actor_url, + status: FollowStatus::Rejected, + }, + FollowWrite::RemoveFollow { + follower_id: requester_id.value(), + target_actor_url: owner_url, + }, + ], + events, + ) + .await } SocialIdentity::Remote { .. } => Err(remote_unsupported(&actor_url)), } @@ -144,24 +180,38 @@ impl SocialCommand for LocalSocialService { &self, owner: &UserId, follower: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let actor_url = self.instance.actor_url_of(follower); match follower { SocialIdentity::Local(follower_id) => { let owner_url = self.instance.actor_url_for(owner); self.follow_command - .remove_follower_record(owner.value(), &actor_url) - .await?; - self.follow_command - .remove_follow(follower_id.value(), &owner_url) - .await?; - Ok(()) + .apply( + &[ + FollowWrite::RemoveFollowerRecord { + local_user_id: owner.value(), + follower_actor_url: actor_url, + }, + FollowWrite::RemoveFollow { + follower_id: follower_id.value(), + target_actor_url: owner_url, + }, + ], + events, + ) + .await } SocialIdentity::Remote { .. } => Err(remote_unsupported(&actor_url)), } } - async fn block(&self, _blocker: &UserId, _target: &SocialIdentity) -> Result<(), DomainError> { + async fn block( + &self, + _blocker: &UserId, + _target: &SocialIdentity, + _events: &[DomainEvent], + ) -> Result<(), DomainError> { Err(DomainError::ValidationError( "blocking requires the federation feature".into(), )) @@ -171,6 +221,7 @@ impl SocialCommand for LocalSocialService { &self, _blocker: &UserId, _target: &SocialIdentity, + _events: &[DomainEvent], ) -> Result<(), DomainError> { Err(DomainError::ValidationError( "blocking requires the federation feature".into(), @@ -242,6 +293,7 @@ impl ResolvedFollow for LocalSocialService { &self, follower: &UserId, target: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let SocialIdentity::Local(target_id) = target else { let actor_url = self.instance.actor_url_of(target); @@ -255,11 +307,21 @@ impl ResolvedFollow for LocalSocialService { let follower_url = self.instance.actor_url_for(follower); let target_url = self.instance.actor_url_for(target_id); self.follow_command - .add_follower(target_id.value(), &follower_url, FollowStatus::Pending) - .await?; - self.follow_command - .add_follow(follower.value(), &target_url, FollowStatus::Pending) - .await?; - Ok(()) + .apply( + &[ + FollowWrite::AddFollower { + local_user_id: target_id.value(), + follower_actor_url: follower_url, + status: FollowStatus::Pending, + }, + FollowWrite::AddFollow { + follower_id: follower.value(), + target_actor_url: target_url, + status: FollowStatus::Pending, + }, + ], + events, + ) + .await } } diff --git a/crates/application/tests/social/execute_test.rs b/crates/application/tests/social/execute_test.rs index cf8331c3..f246e0f2 100644 --- a/crates/application/tests/social/execute_test.rs +++ b/crates/application/tests/social/execute_test.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use domain::{ events::DomainEvent, - testing::{InMemoryStore, NoopEventPublisher}, + testing::{InMemoryStore, RefusingSocialCommand}, value_objects::{FollowStatus, FollowTarget, SocialIdentity, UserId}, }; use uuid::Uuid; @@ -13,25 +13,19 @@ use application::social::{ execute::execute_command, }; -fn make_cmd_deps() -> ( - Arc, - Arc, - SocialCommandDeps, -) { +fn make_cmd_deps() -> (Arc, SocialCommandDeps) { let social = InMemoryStore::new(); - let events = NoopEventPublisher::new(); let deps = SocialCommandDeps { social_command: Arc::clone(&social) as _, - event_publisher: Arc::clone(&events) as _, }; - (social, events, deps) + (social, deps) } // ── Follow ────────────────────────────────────────────────────────────────── #[tokio::test] async fn follow_emits_follow_requested_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); execute_command( &deps, @@ -45,7 +39,7 @@ async fn follow_emits_follow_requested_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -55,7 +49,7 @@ async fn follow_emits_follow_requested_event() { #[tokio::test] async fn cannot_follow_yourself() { - let (_social, _events, deps) = make_cmd_deps(); + let (_social, deps) = make_cmd_deps(); let user_id = Uuid::new_v4(); let result = execute_command( @@ -72,7 +66,7 @@ async fn cannot_follow_yourself() { #[tokio::test] async fn cannot_follow_same_target_twice() { - let (_social, _events, deps) = make_cmd_deps(); + let (_social, deps) = make_cmd_deps(); let follower_id = Uuid::new_v4(); let target = FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4()))); @@ -102,7 +96,7 @@ async fn cannot_follow_same_target_twice() { #[tokio::test] async fn unfollow_emits_unfollowed_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); let follower_id = Uuid::new_v4(); let target = SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4())); @@ -126,7 +120,7 @@ async fn unfollow_emits_unfollowed_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -138,7 +132,7 @@ async fn unfollow_emits_unfollowed_event() { #[tokio::test] async fn accept_follow_emits_follow_accepted_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); let follower_id = Uuid::new_v4(); let owner_id = Uuid::new_v4(); let requester = SocialIdentity::Local(UserId::from_uuid(follower_id)); @@ -163,7 +157,7 @@ async fn accept_follow_emits_follow_accepted_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -175,7 +169,7 @@ async fn accept_follow_emits_follow_accepted_event() { #[tokio::test] async fn reject_follow_emits_follow_rejected_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); let follower_id = Uuid::new_v4(); let owner_id = Uuid::new_v4(); @@ -199,7 +193,7 @@ async fn reject_follow_emits_follow_rejected_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -211,7 +205,7 @@ async fn reject_follow_emits_follow_rejected_event() { #[tokio::test] async fn remove_follower_emits_follower_removed_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); let follower_id = Uuid::new_v4(); let owner_id = Uuid::new_v4(); @@ -245,7 +239,7 @@ async fn remove_follower_emits_follower_removed_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -257,7 +251,7 @@ async fn remove_follower_emits_follower_removed_event() { #[tokio::test] async fn block_emits_actor_blocked_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); execute_command( &deps, @@ -269,7 +263,7 @@ async fn block_emits_actor_blocked_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -281,7 +275,7 @@ async fn block_emits_actor_blocked_event() { #[tokio::test] async fn unblock_emits_actor_unblocked_event() { - let (_social, events, deps) = make_cmd_deps(); + let (events, deps) = make_cmd_deps(); let target = SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4())); let blocker_id = Uuid::new_v4(); @@ -299,7 +293,7 @@ async fn unblock_emits_actor_unblocked_event() { .await .unwrap(); - let published = events.published(); + let published = events.recorded_events(); assert!( published .iter() @@ -311,7 +305,7 @@ async fn unblock_emits_actor_unblocked_event() { #[tokio::test] async fn get_following_returns_accepted_targets() { - let (social, _events, cmd_deps) = make_cmd_deps(); + let (social, cmd_deps) = make_cmd_deps(); let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, block_query: Arc::clone(&social) as _, @@ -360,10 +354,8 @@ async fn get_following_returns_accepted_targets() { #[tokio::test] async fn returns_accepted_followers() { let social = InMemoryStore::new(); - let events = NoopEventPublisher::new(); let cmd_deps = SocialCommandDeps { social_command: Arc::clone(&social) as _, - event_publisher: Arc::clone(&events) as _, }; let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, @@ -404,10 +396,8 @@ async fn returns_accepted_followers() { #[tokio::test] async fn returns_only_pending_followers() { let social = InMemoryStore::new(); - let events = NoopEventPublisher::new(); let cmd_deps = SocialCommandDeps { social_command: Arc::clone(&social) as _, - event_publisher: Arc::clone(&events) as _, }; let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, @@ -439,7 +429,7 @@ async fn returns_only_pending_followers() { async fn get_relation_reports_none_for_strangers() { use domain::ports::FollowGraphQuery; - let (social, _events, _deps) = make_cmd_deps(); + let (social, _deps) = make_cmd_deps(); let rel = FollowGraphQuery::get_relation( &*social, &UserId::from_uuid(Uuid::new_v4()), @@ -456,7 +446,7 @@ async fn get_relation_reports_none_for_strangers() { async fn get_relation_reports_pending_before_acceptance() { use domain::ports::FollowGraphQuery; - let (social, _events, deps) = make_cmd_deps(); + let (social, deps) = make_cmd_deps(); let a = Uuid::new_v4(); let b_id = UserId::from_uuid(Uuid::new_v4()); @@ -486,7 +476,7 @@ async fn get_relation_reports_pending_before_acceptance() { async fn get_relation_reports_accepted_in_both_directions_after_mutual_follow() { use domain::ports::FollowGraphQuery; - let (social, _events, deps) = make_cmd_deps(); + let (social, deps) = make_cmd_deps(); let a = Uuid::new_v4(); let b = Uuid::new_v4(); let a_id = UserId::from_uuid(a); @@ -528,7 +518,7 @@ async fn get_relation_reports_accepted_in_both_directions_after_mutual_follow() async fn unaccepted_follow_appears_in_pending_following_not_in_following() { use domain::ports::FollowGraphQuery; - let (social, _events, deps) = make_cmd_deps(); + let (social, deps) = make_cmd_deps(); let viewer = Uuid::new_v4(); let target = UserId::from_uuid(Uuid::new_v4()); @@ -561,7 +551,7 @@ async fn unaccepted_follow_appears_in_pending_following_not_in_following() { #[tokio::test] async fn get_pending_following_returns_unaccepted_targets_and_get_following_does_not() { - let (social, _events, cmd_deps) = make_cmd_deps(); + let (social, cmd_deps) = make_cmd_deps(); let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, block_query: Arc::clone(&social) as _, @@ -599,7 +589,7 @@ async fn get_pending_following_returns_unaccepted_targets_and_get_following_does #[tokio::test] async fn count_pending_followers_drops_as_requests_are_accepted() { - let (social, _events, cmd_deps) = make_cmd_deps(); + let (social, cmd_deps) = make_cmd_deps(); let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, block_query: Arc::clone(&social) as _, @@ -648,7 +638,7 @@ async fn count_pending_followers_drops_as_requests_are_accepted() { #[tokio::test] async fn get_relation_use_case_reports_direction_asymmetrically() { - let (social, _events, cmd_deps) = make_cmd_deps(); + let (social, cmd_deps) = make_cmd_deps(); let query_deps = SocialQueryDeps { follow_graph: Arc::clone(&social) as _, block_query: Arc::clone(&social) as _, @@ -681,3 +671,67 @@ async fn get_relation_use_case_reports_direction_asymmetrically() { ); assert_eq!(rel.followed_by, None, "target has not followed back"); } + +// ── Events ride the write ─────────────────────────────────────────────────── + +#[tokio::test] +async fn a_follow_records_exactly_its_event_with_the_write() { + let (social, deps) = make_cmd_deps(); + + execute_command( + &deps, + SocialCmd::Follow { + follower_id: Uuid::new_v4(), + target: FollowTarget::Identity(SocialIdentity::Local( + UserId::from_uuid(Uuid::new_v4()), + )), + }, + ) + .await + .unwrap(); + + let recorded = social.recorded_events(); + assert_eq!(recorded.len(), 1); + assert!(matches!(recorded[0], DomainEvent::FollowRequested { .. })); +} + +#[tokio::test] +async fn a_rejected_self_follow_records_nothing() { + let (social, deps) = make_cmd_deps(); + let user_id = Uuid::new_v4(); + + let result = execute_command( + &deps, + SocialCmd::Follow { + follower_id: user_id, + target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(user_id))), + }, + ) + .await; + + assert!(result.is_err()); + assert!(social.recorded_events().is_empty()); +} + +#[tokio::test] +async fn a_refused_follow_propagates() { + let deps = SocialCommandDeps { + social_command: Arc::new(RefusingSocialCommand), + }; + + let result = execute_command( + &deps, + SocialCmd::Follow { + follower_id: Uuid::new_v4(), + target: FollowTarget::Identity(SocialIdentity::Local( + UserId::from_uuid(Uuid::new_v4()), + )), + }, + ) + .await; + + assert!(matches!( + result, + Err(domain::errors::DomainError::InfrastructureError(m)) if m == "refused" + )); +} diff --git a/crates/application/tests/social/local_service_test.rs b/crates/application/tests/social/local_service_test.rs index c2e9e7c2..9bacd900 100644 --- a/crates/application/tests/social/local_service_test.rs +++ b/crates/application/tests/social/local_service_test.rs @@ -4,8 +4,10 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use domain::{ errors::DomainError, + events::DomainEvent, ports::{ - BlockQuery, FollowCommand, FollowGraphQuery, FollowQuery, SocialCommand, UserRepository, + BlockQuery, FollowCommand, FollowGraphQuery, FollowQuery, FollowWrite, SocialCommand, + UserRepository, }, testing::InMemoryStore, value_objects::{ @@ -30,6 +32,8 @@ struct FollowFakeStore { follows: Mutex>, // (local_user_id, follower_actor_url, status) — written by add_follower/update_follower_status/remove_follower_record followers: Mutex>, + recorded: Mutex>, + refuse_writes: bool, } impl FollowFakeStore { @@ -37,8 +41,23 @@ impl FollowFakeStore { Arc::new(Self { follows: Mutex::new(Vec::new()), followers: Mutex::new(Vec::new()), + recorded: Mutex::new(Vec::new()), + refuse_writes: false, }) } + + fn refusing() -> Arc { + Arc::new(Self { + follows: Mutex::new(Vec::new()), + followers: Mutex::new(Vec::new()), + recorded: Mutex::new(Vec::new()), + refuse_writes: true, + }) + } + + fn recorded_events(&self) -> Vec { + self.recorded.lock().unwrap().clone() + } } fn fake_actor(url: &str) -> SocialActor { @@ -54,86 +73,63 @@ fn fake_actor(url: &str) -> SocialActor { #[async_trait] impl FollowCommand for FollowFakeStore { - async fn add_follow( + async fn apply( &self, - follower_id: Uuid, - target_actor_url: &str, - status: FollowStatus, + writes: &[FollowWrite], + events: &[DomainEvent], ) -> Result<(), DomainError> { - self.follows - .lock() - .unwrap() - .push((follower_id, target_actor_url.to_string(), status)); - Ok(()) - } - - async fn update_follow_status( - &self, - follower_id: Uuid, - target_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - let mut store = self.follows.lock().unwrap(); - if let Some(entry) = store - .iter_mut() - .find(|(f, t, _)| *f == follower_id && t == target_actor_url) - { - entry.2 = status; + if self.refuse_writes { + return Err(DomainError::InfrastructureError("refused".into())); } - Ok(()) - } - - async fn remove_follow( - &self, - follower_id: Uuid, - target_actor_url: &str, - ) -> Result<(), DomainError> { - self.follows - .lock() - .unwrap() - .retain(|(f, t, _)| !(*f == follower_id && t == target_actor_url)); - Ok(()) - } - - async fn add_follower( - &self, - local_user_id: Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - self.followers.lock().unwrap().push(( - local_user_id, - follower_actor_url.to_string(), - status, - )); - Ok(()) - } - - async fn update_follower_status( - &self, - local_user_id: Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError> { - let mut store = self.followers.lock().unwrap(); - if let Some(entry) = store - .iter_mut() - .find(|(u, f, _)| *u == local_user_id && f == follower_actor_url) - { - entry.2 = status; + let mut follows = self.follows.lock().unwrap(); + let mut followers = self.followers.lock().unwrap(); + for write in writes { + match write { + FollowWrite::AddFollow { + follower_id, + target_actor_url, + status, + } => follows.push((*follower_id, target_actor_url.clone(), *status)), + FollowWrite::UpdateFollowStatus { + follower_id, + target_actor_url, + status, + } => { + if let Some(entry) = follows + .iter_mut() + .find(|(f, t, _)| f == follower_id && t == target_actor_url) + { + entry.2 = *status; + } + } + FollowWrite::RemoveFollow { + follower_id, + target_actor_url, + } => follows.retain(|(f, t, _)| !(f == follower_id && t == target_actor_url)), + FollowWrite::AddFollower { + local_user_id, + follower_actor_url, + status, + } => followers.push((*local_user_id, follower_actor_url.clone(), *status)), + FollowWrite::UpdateFollowerStatus { + local_user_id, + follower_actor_url, + status, + } => { + if let Some(entry) = followers + .iter_mut() + .find(|(u, f, _)| u == local_user_id && f == follower_actor_url) + { + entry.2 = *status; + } + } + FollowWrite::RemoveFollowerRecord { + local_user_id, + follower_actor_url, + } => followers.retain(|(u, f, _)| !(u == local_user_id && f == follower_actor_url)), + } } - Ok(()) - } - - async fn remove_follower_record( - &self, - local_user_id: Uuid, - follower_actor_url: &str, - ) -> Result<(), DomainError> { - self.followers - .lock() - .unwrap() - .retain(|(u, f, _)| !(*u == local_user_id && f == follower_actor_url)); + self.recorded.lock().unwrap().extend_from_slice(events); Ok(()) } } @@ -280,6 +276,7 @@ async fn follow_local_user_writes_both_sides() { svc.follow( &alice_id, &FollowTarget::Handle("@bob@md.example".to_string()), + &[], ) .await .unwrap(); @@ -311,6 +308,7 @@ async fn follow_self_is_rejected() { .follow( &alice_id, &FollowTarget::Handle("@alice@md.example".to_string()), + &[], ) .await; @@ -344,6 +342,7 @@ async fn follow_remote_target_errors_rather_than_silently_succeeding() { .follow( &alice_id, &FollowTarget::Handle("@carol@other.example".to_string()), + &[], ) .await; @@ -386,11 +385,12 @@ async fn reject_follow_updates_follower_status_but_removes_the_follow_row() { svc.follow( &requester_id, &FollowTarget::Handle("@owner@md.example".to_string()), + &[], ) .await .unwrap(); - svc.reject_follow(&owner_id, &SocialIdentity::Local(requester_id.clone())) + svc.reject_follow(&owner_id, &SocialIdentity::Local(requester_id.clone()), &[]) .await .unwrap(); @@ -431,9 +431,60 @@ async fn block_errors_and_get_blocked_is_empty() { actor_url: "https://other.example/users/carol".to_string(), }; - let result = svc.block(&alice_id, &target).await; + let result = svc.block(&alice_id, &target, &[]).await; assert!(matches!(result, Err(DomainError::ValidationError(_)))); let blocked = svc.get_blocked(&alice_id).await.unwrap(); assert_eq!(blocked.len(), 0); } + +#[tokio::test] +async fn a_local_follow_records_both_rows_and_its_event_together() { + let user_repo = InMemoryStore::new(); + let bob_id = register_user(&user_repo, "bob").await; + let alice_id = register_user(&user_repo, "alice").await; + let follow_store = FollowFakeStore::new(); + let svc = service(user_repo, &follow_store, instance()); + let event = DomainEvent::FollowRequested { + follower: alice_id.clone(), + target: FollowTarget::Identity(SocialIdentity::Local(bob_id.clone())), + }; + + svc.follow( + &alice_id, + &FollowTarget::Handle("@bob@md.example".to_string()), + std::slice::from_ref(&event), + ) + .await + .unwrap(); + + assert_eq!(svc.get_followers(&bob_id).await.unwrap().len(), 1); + assert_eq!(svc.get_following(&alice_id).await.unwrap().len(), 1); + assert_eq!(follow_store.recorded_events().len(), 1); +} + +#[tokio::test] +async fn a_refused_local_follow_write_propagates_and_records_nothing() { + let user_repo = InMemoryStore::new(); + let bob_id = register_user(&user_repo, "bob").await; + let alice_id = register_user(&user_repo, "alice").await; + let follow_store = FollowFakeStore::refusing(); + let svc = service(user_repo, &follow_store, instance()); + let event = DomainEvent::FollowRequested { + follower: alice_id.clone(), + target: FollowTarget::Identity(SocialIdentity::Local(bob_id.clone())), + }; + + let result = svc + .follow( + &alice_id, + &FollowTarget::Handle("@bob@md.example".to_string()), + std::slice::from_ref(&event), + ) + .await; + + assert!(matches!(result, Err(DomainError::InfrastructureError(_)))); + assert!(svc.get_followers(&bob_id).await.unwrap().is_empty()); + assert!(svc.get_following(&alice_id).await.unwrap().is_empty()); + assert!(follow_store.recorded_events().is_empty()); +} diff --git a/crates/application/tests/users/get_page_viewer_test.rs b/crates/application/tests/users/get_page_viewer_test.rs index 318c9c78..cf5b8bbe 100644 --- a/crates/application/tests/users/get_page_viewer_test.rs +++ b/crates/application/tests/users/get_page_viewer_test.rs @@ -35,11 +35,11 @@ async fn get_page_viewer_reports_pending_follower_count() { let owner_id = UserId::from_uuid(owner_uuid); let target = FollowTarget::Identity(SocialIdentity::Local(owner_id.clone())); b.social_command - .follow(&UserId::from_uuid(follower1), &target) + .follow(&UserId::from_uuid(follower1), &target, &[]) .await .unwrap(); b.social_command - .follow(&UserId::from_uuid(follower2), &target) + .follow(&UserId::from_uuid(follower2), &target, &[]) .await .unwrap(); diff --git a/crates/bootstrap/src/build.rs b/crates/bootstrap/src/build.rs index 913b002e..9fc9bd50 100644 --- a/crates/bootstrap/src/build.rs +++ b/crates/bootstrap/src/build.rs @@ -241,7 +241,6 @@ pub fn build_deps( social: SocialGroup { command: SocialCommandDeps { social_command: repos.social_command.clone(), - event_publisher: services.event_publisher.clone(), }, query: SocialQueryDeps { follow_graph: repos.follow_graph.clone(), diff --git a/crates/bootstrap/src/federation.rs b/crates/bootstrap/src/federation.rs index 7517a38f..7fbf733b 100644 --- a/crates/bootstrap/src/federation.rs +++ b/crates/bootstrap/src/federation.rs @@ -104,6 +104,7 @@ pub async fn wire_server_social( ap.service, Arc::clone(&db.user), instance.clone(), + Arc::clone(event_publisher), )); Ok(ServerSocial { diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 93792411..318b8161 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -145,3 +145,73 @@ async fn handler_names_are_unique() { // Image conversion ("image_conversion") is disabled in the test config. assert_eq!(sorted, expected); } + +async fn queued(pool: &sqlx::SqlitePool, event_type: &str) -> i64 { + sqlx::query_scalar( + "SELECT COUNT(*) FROM event_queue WHERE event_type = ? AND event_id IS NOT NULL", + ) + .bind(event_type) + .fetch_one(pool) + .await + .unwrap() +} + +/// A local follow writes both follow rows and its event in one commit, in +/// both federation builds. +#[tokio::test] +async fn a_local_follow_lands_with_its_event_in_the_outbox() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + let _bob = support::register(&router, "bob", "bob@example.com").await; + + let (status, _, body) = support::send( + &router, + support::json_request( + axum::http::Method::POST, + "/api/v1/social/follow", + Some(&alice.token), + serde_json::json!({ "handle": "bob@localhost:3000" }), + ), + ) + .await; + assert_eq!(status, 200, "follow: {body}"); + + let following: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ap_following") + .fetch_one(&pool) + .await + .unwrap(); + let followers: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ap_followers") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!((following, followers), (1, 1)); + assert_eq!(queued(&pool, "FollowRequested").await, 1); +} + +/// The k_ap path owns its transaction, so the adapter publishes the event +/// itself once the k_ap write has succeeded. +#[cfg(feature = "federation")] +#[tokio::test] +async fn a_remote_block_publishes_its_event_after_the_k_ap_write() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + + let (status, _, body) = support::send( + &router, + support::json_request( + axum::http::Method::POST, + "/api/v1/social/block", + Some(&alice.token), + serde_json::json!({ "actor_url": "https://remote.example/users/ann" }), + ), + ) + .await; + assert_eq!(status, 204, "block: {body}"); + + let blocked: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blocked_actors") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(blocked, 1); + assert_eq!(queued(&pool, "ActorBlocked").await, 1); +} diff --git a/crates/domain/src/ports/follow.rs b/crates/domain/src/ports/follow.rs index 427ed160..197b5dd9 100644 --- a/crates/domain/src/ports/follow.rs +++ b/crates/domain/src/ports/follow.rs @@ -2,49 +2,52 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, value_objects::{FollowRelation, FollowStatus, SocialActor}, }; +/// One row-level change to the local follow tables. A local follow operation +/// is a batch of these applied in a single transaction with its events. +#[derive(Clone, Debug, PartialEq)] +pub enum FollowWrite { + AddFollow { + follower_id: uuid::Uuid, + target_actor_url: String, + status: FollowStatus, + }, + UpdateFollowStatus { + follower_id: uuid::Uuid, + target_actor_url: String, + status: FollowStatus, + }, + RemoveFollow { + follower_id: uuid::Uuid, + target_actor_url: String, + }, + AddFollower { + local_user_id: uuid::Uuid, + follower_actor_url: String, + status: FollowStatus, + }, + UpdateFollowerStatus { + local_user_id: uuid::Uuid, + follower_actor_url: String, + status: FollowStatus, + }, + RemoveFollowerRecord { + local_user_id: uuid::Uuid, + follower_actor_url: String, + }, +} + #[async_trait] pub trait FollowCommand: Send + Sync { - async fn add_follow( + /// Applies every write and records `events` in one transaction: all rows + /// and events land, or none do. + async fn apply( &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError>; - - async fn update_follow_status( - &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError>; - - async fn remove_follow( - &self, - follower_id: uuid::Uuid, - target_actor_url: &str, - ) -> Result<(), DomainError>; - - async fn add_follower( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError>; - - async fn update_follower_status( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, - status: FollowStatus, - ) -> Result<(), DomainError>; - - async fn remove_follower_record( - &self, - local_user_id: uuid::Uuid, - follower_actor_url: &str, + writes: &[FollowWrite], + events: &[DomainEvent], ) -> Result<(), DomainError>; } diff --git a/crates/domain/src/ports/noop.rs b/crates/domain/src/ports/noop.rs index 28007f00..3c8e0de8 100644 --- a/crates/domain/src/ports/noop.rs +++ b/crates/domain/src/ports/noop.rs @@ -44,25 +44,56 @@ impl super::SocialCommand for NoopSocialCommand { &self, _: &UserId, _: &crate::value_objects::FollowTarget, + _: &[crate::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } - async fn unfollow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn unfollow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } - async fn accept_follow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn accept_follow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } - async fn reject_follow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn reject_follow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } - async fn remove_follower(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn remove_follower( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } - async fn block(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn block( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } - async fn unblock(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> { + async fn unblock( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[crate::events::DomainEvent], + ) -> Result<(), DomainError> { Ok(()) } } diff --git a/crates/domain/src/ports/social.rs b/crates/domain/src/ports/social.rs index 067040aa..5ea51516 100644 --- a/crates/domain/src/ports/social.rs +++ b/crates/domain/src/ports/social.rs @@ -3,6 +3,7 @@ use chrono::NaiveDateTime; use crate::{ errors::DomainError, + events::DomainEvent, models::{ DiaryEntry, FederationFlags, RemoteActorInfo, RemoteGoalEntry, RemoteWatchlistEntry, WatchlistWithMovie, @@ -14,32 +15,54 @@ use crate::{ #[async_trait] pub trait SocialCommand: Send + Sync { - async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError>; + async fn follow( + &self, + follower: &UserId, + target: &FollowTarget, + events: &[DomainEvent], + ) -> Result<(), DomainError>; - async fn unfollow(&self, follower: &UserId, target: &SocialIdentity) - -> Result<(), DomainError>; + async fn unfollow( + &self, + follower: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError>; async fn accept_follow( &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError>; async fn reject_follow( &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError>; async fn remove_follower( &self, owner: &UserId, follower: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError>; - async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError>; + async fn block( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError>; - async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError>; + async fn unblock( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError>; } #[async_trait] @@ -138,6 +161,7 @@ pub trait ResolvedFollow: Send + Sync { &self, follower: &UserId, target: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError>; } diff --git a/crates/domain/src/testing/fakes/refusing.rs b/crates/domain/src/testing/fakes/refusing.rs index 86864c7c..69c10bbd 100644 --- a/crates/domain/src/testing/fakes/refusing.rs +++ b/crates/domain/src/testing/fakes/refusing.rs @@ -12,10 +12,12 @@ use crate::{ collections::{PageParams, Paginated}, }, ports::{ - EventPublisher, GoalCommand, MovieCommand, ReviewRepository, UserProfileFieldsRepository, - WatchEventCommand, WatchlistRepository, + EventPublisher, FollowCommand, FollowWrite, GoalCommand, MovieCommand, ReviewRepository, + SocialCommand, UserProfileFieldsRepository, WatchEventCommand, WatchlistRepository, + }, + value_objects::{ + FollowTarget, GoalId, MovieId, ReviewId, SocialIdentity, UserId, WatchEventId, }, - value_objects::{GoalId, MovieId, ReviewId, UserId, WatchEventId}, }; fn refused() -> DomainError { @@ -151,3 +153,74 @@ impl WatchEventCommand for RefusingWatchEventCommand { Err(refused()) } } + +pub struct RefusingSocialCommand; + +#[async_trait] +impl SocialCommand for RefusingSocialCommand { + async fn follow( + &self, + _: &UserId, + _: &FollowTarget, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn unfollow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn accept_follow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn reject_follow( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn remove_follower( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn block( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } + async fn unblock( + &self, + _: &UserId, + _: &SocialIdentity, + _: &[DomainEvent], + ) -> Result<(), DomainError> { + Err(refused()) + } +} + +pub struct RefusingFollowCommand; + +#[async_trait] +impl FollowCommand for RefusingFollowCommand { + async fn apply(&self, _: &[FollowWrite], _: &[DomainEvent]) -> Result<(), DomainError> { + Err(refused()) + } +} diff --git a/crates/domain/src/testing/fakes/store_social.rs b/crates/domain/src/testing/fakes/store_social.rs index 2d5395ff..5941cfae 100644 --- a/crates/domain/src/testing/fakes/store_social.rs +++ b/crates/domain/src/testing/fakes/store_social.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use crate::{ errors::DomainError, + events::DomainEvent, ports::{BlockQuery, FollowGraphQuery, SocialCommand}, value_objects::{FollowRelation, FollowStatus, SocialActor, SocialIdentity, UserId}, }; @@ -28,6 +29,7 @@ impl SocialCommand for InMemoryStore { &self, follower: &UserId, target: &crate::value_objects::FollowTarget, + events: &[DomainEvent], ) -> Result<(), DomainError> { let identity = match target { crate::value_objects::FollowTarget::Identity(id) => id.clone(), @@ -50,6 +52,8 @@ impl SocialCommand for InMemoryStore { return Err(DomainError::ValidationError("Already following".into())); } store.push((follower.value(), identity, FollowState::Pending)); + drop(store); + self.record_events(events); Ok(()) } @@ -57,6 +61,7 @@ impl SocialCommand for InMemoryStore { &self, follower: &UserId, target: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let mut store = self.follows.write().unwrap(); let before = store.len(); @@ -66,6 +71,8 @@ impl SocialCommand for InMemoryStore { "Follow relationship not found".into(), )); } + drop(store); + self.record_events(events); Ok(()) } @@ -73,6 +80,7 @@ impl SocialCommand for InMemoryStore { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let mut store = self.follows.write().unwrap(); let target_identity = SocialIdentity::Local(owner.clone()); @@ -92,6 +100,8 @@ impl SocialCommand for InMemoryStore { }; if requester_matches && *t == target_identity && *state == FollowState::Pending { *state = FollowState::Accepted; + drop(store); + self.record_events(events); return Ok(()); } } @@ -104,6 +114,7 @@ impl SocialCommand for InMemoryStore { &self, owner: &UserId, requester: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let mut store = self.follows.write().unwrap(); let target_identity = SocialIdentity::Local(owner.clone()); @@ -123,6 +134,8 @@ impl SocialCommand for InMemoryStore { "Pending follow request not found".into(), )); } + drop(store); + self.record_events(events); Ok(()) } @@ -130,6 +143,7 @@ impl SocialCommand for InMemoryStore { &self, owner: &UserId, follower: &SocialIdentity, + events: &[DomainEvent], ) -> Result<(), DomainError> { let mut store = self.follows.write().unwrap(); let target_identity = SocialIdentity::Local(owner.clone()); @@ -144,21 +158,38 @@ impl SocialCommand for InMemoryStore { if store.len() == before { return Err(DomainError::NotFound("Follower not found".into())); } + drop(store); + self.record_events(events); Ok(()) } - async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> { + async fn block( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let mut store = self.blocked.write().unwrap(); store.push((blocker.value(), target.clone())); // Also remove any existing follow relationships let mut follows = self.follows.write().unwrap(); follows.retain(|(f, t, _)| !(*f == blocker.value() && t == target)); + drop(follows); + drop(store); + self.record_events(events); Ok(()) } - async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> { + async fn unblock( + &self, + blocker: &UserId, + target: &SocialIdentity, + events: &[DomainEvent], + ) -> Result<(), DomainError> { let mut store = self.blocked.write().unwrap(); store.retain(|(b, t)| !(*b == blocker.value() && t == target)); + drop(store); + self.record_events(events); Ok(()) } } diff --git a/crates/domain/tests/ports/social_test.rs b/crates/domain/tests/ports/social_test.rs index 4a023a8b..89fef2a9 100644 --- a/crates/domain/tests/ports/social_test.rs +++ b/crates/domain/tests/ports/social_test.rs @@ -14,6 +14,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::FollowTarget, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -21,6 +22,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -28,6 +30,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -35,6 +38,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -42,6 +46,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -49,6 +54,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -56,6 +62,7 @@ impl SocialCommand for Stub { &self, _: &domain::value_objects::UserId, _: &domain::value_objects::SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } @@ -145,6 +152,7 @@ impl ResolvedFollow for Stub { &self, _: &domain::value_objects::UserId, _: &SocialIdentity, + _: &[domain::events::DomainEvent], ) -> Result<(), DomainError> { Ok(()) } diff --git a/crates/domain/tests/testing/in_memory_store_test.rs b/crates/domain/tests/testing/in_memory_store_test.rs index f8b9c69d..ca462742 100644 --- a/crates/domain/tests/testing/in_memory_store_test.rs +++ b/crates/domain/tests/testing/in_memory_store_test.rs @@ -128,6 +128,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie &FollowTarget::Identity(SocialIdentity::Remote { actor_url: REMOTE_ACTOR.into(), }), + &[], ) .await .unwrap(); @@ -135,6 +136,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie .follow( other.id(), &FollowTarget::Identity(SocialIdentity::Local(user_id.clone())), + &[], ) .await .unwrap(); @@ -144,6 +146,7 @@ async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie &SocialIdentity::Remote { actor_url: REMOTE_ACTOR.into(), }, + &[], ) .await .unwrap();