GKaszewski/movies-diary
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: atomic social local writes, remote review enrichment

Browse Source 
SocialCommand methods take events. Local follow ops = one FollowCommand::apply
tx (rows + events). k_ap path (remote, block/unblock): best-effort publish in
CompositeSocialAdapter after k_ap write; not atomic, k_ap owns tx.
save_remote_review takes events, enqueued in same tx via
sqlite::persistence::enqueue_events. Deps drop event_publisher: social
command, ReviewObjectHandler.

Remaining .publish( sites:
announce-only: users/delete_account UserDeleted, person/get + get_credits
PersonEnrichmentRequested, jobs/enrichment_staleness, http-axum search,
ImageStored (sync_poster, poster_sync_handler, image-converter backfill,
update_profile; object storage), federation_event_bridge (handler reaction),
nats relay (transport).
non-atomic by necessity: CompositeSocialAdapter::announce (k_ap-owned write).
Everything else rides its write.
This commit is contained in:
Gabriel Kaszewski
2026-10-26 01:12:58 +01:00
parent 6c295866a0
commit 9a4d0c1714
29 changed files with 1115 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/adapters/activitypub/src/lib.rs
View File

@@ -101,7 +101,6 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
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 {

4
crates/adapters/activitypub/src/remote_review_repository.rs
View File

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

30
crates/adapters/activitypub/src/review_handler.rs
View File

@@ -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<dyn MovieQuery>,
pub diary_repo: Arc<dyn DiaryQuery>,
pub review_store: Arc<dyn RemoteReviewRepository>,
pub event_publisher: Arc<dyn EventPublisher>,
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<DomainEvent> = 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(())
}

127
crates/adapters/activitypub/src/social_adapter.rs
View File

@@ -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<ActivityPubService>,
user_repo: Arc<dyn UserRepository>,
instance: InstanceIdentity,
event_publisher: Arc<dyn EventPublisher>,
}
impl CompositeSocialAdapter {
@@ -24,12 +28,23 @@ impl CompositeSocialAdapter {
ap_service: Arc<ActivityPubService>,
user_repo: Arc<dyn UserRepository>,
instance: InstanceIdentity,
event_publisher: Arc<dyn EventPublisher>,
) -> 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(())
}
}

1
crates/adapters/sqlite-federation/Cargo.toml
View File

@@ -18,4 +18,5 @@ async-trait = { workspace = true }
[dev-dependencies]
tokio = { workspace = true }
domain = { workspace = true, features = ["test-helpers"] }
uuid = { workspace = true }

16
crates/adapters/sqlite-federation/src/review.rs
View File

@@ -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(())
}

2
crates/adapters/sqlite-federation/tests/federation.rs
View File

@@ -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;

115
crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs Normal file
View File

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

2
crates/adapters/sqlite/src/lib.rs
View File

@@ -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(

230
crates/adapters/sqlite/src/repositories/social/follow/command.rs
View File

@@ -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(())
}

2
crates/adapters/sqlite/src/shared.rs
View File

@@ -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> {

23
crates/adapters/sqlite/tests/cascade_test.rs
View File

@@ -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).

82
crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs
View File

@@ -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<domain::ports::FollowWrite> {
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);
}