From 063bab910b0f3debfb730fa9cc93858a3be70540 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 27 Jul 2026 12:56:30 +0200 Subject: [PATCH] ADR-0003: local follows bypass AP, unified FollowCommand/FollowQuery ports, FederationRepos struct, base_url normalization --- crates/adapters/activitypub/src/lib.rs | 24 +- crates/adapters/activitypub/src/port.rs | 32 -- .../activitypub/src/social_adapter.rs | 247 +++++++++------ .../src/follow_repository.rs | 299 ++++++++++++++++++ .../adapters/postgres-federation/src/lib.rs | 21 +- .../src/follow_repository.rs | 299 ++++++++++++++++++ crates/adapters/sqlite-federation/src/lib.rs | 21 +- crates/domain/src/ports/follow.rs | 80 +++++ crates/domain/src/ports/mod.rs | 2 + crates/domain/src/value_objects/social.rs | 32 ++ crates/infra-wiring/src/config.rs | 6 +- crates/presentation/src/handlers/diary.rs | 3 +- crates/presentation/src/handlers/social.rs | 66 ++-- crates/presentation/src/main.rs | 30 +- crates/worker/src/main.rs | 24 +- docs/adr/0003-local-follow-direct-db.md | 20 ++ 16 files changed, 978 insertions(+), 228 deletions(-) create mode 100644 crates/adapters/postgres-federation/src/follow_repository.rs create mode 100644 crates/adapters/sqlite-federation/src/follow_repository.rs create mode 100644 crates/domain/src/ports/follow.rs create mode 100644 docs/adr/0003-local-follow-direct-db.md diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index fe3ed8f..7e14f8a 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -28,15 +28,17 @@ pub use review_handler::ReviewObjectHandler; pub use social_adapter::CompositeSocialAdapter; pub use user_adapter::DomainUserRepoAdapter; -pub type FederationRepos = ( - std::sync::Arc, - std::sync::Arc, - std::sync::Arc, - std::sync::Arc, - std::sync::Arc, - std::sync::Arc, - std::sync::Arc, -); +pub struct FederationRepos { + pub activity: std::sync::Arc, + pub follow: std::sync::Arc, + pub actor: std::sync::Arc, + pub blocklist: std::sync::Arc, + pub admin_query: std::sync::Arc, + pub review_store: std::sync::Arc, + pub remote_watchlist: std::sync::Arc, + pub follow_command: std::sync::Arc, + pub follow_query: std::sync::Arc, +} pub struct ActivityPubWire { pub service: std::sync::Arc, @@ -60,6 +62,8 @@ pub struct ActivityPubDeps { pub stats_repo: std::sync::Arc, pub user_repo: std::sync::Arc, pub federation_settings: std::sync::Arc, + pub follow_command: std::sync::Arc, + pub follow_query: std::sync::Arc, pub base_url: String, pub allow_registration: bool, pub event_publisher: std::sync::Arc, @@ -82,6 +86,8 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { stats_repo, user_repo, federation_settings, + follow_command: _, + follow_query: _, base_url, allow_registration, event_publisher, diff --git a/crates/adapters/activitypub/src/port.rs b/crates/adapters/activitypub/src/port.rs index a4e1cbb..688ebeb 100644 --- a/crates/adapters/activitypub/src/port.rs +++ b/crates/adapters/activitypub/src/port.rs @@ -6,9 +6,6 @@ use k_ap::{ActivityPubService, BlockedDomain, RemoteActor}; #[async_trait] pub trait ActivityPubPort: Send + Sync { async fn actor_json(&self, user_id: &str) -> anyhow::Result; - async fn count_following(&self, local_user_id: Uuid) -> anyhow::Result; - async fn count_accepted_followers(&self, local_user_id: Uuid) -> anyhow::Result; - async fn get_pending_followers(&self, local_user_id: Uuid) -> anyhow::Result>; async fn follow(&self, local_user_id: Uuid, handle: &str) -> anyhow::Result<()>; async fn unfollow(&self, local_user_id: Uuid, actor_url: &str) -> anyhow::Result<()>; async fn accept_follower( @@ -22,8 +19,6 @@ pub trait ActivityPubPort: Send + Sync { remote_actor_url: &str, ) -> anyhow::Result<()>; async fn get_following(&self, local_user_id: Uuid) -> anyhow::Result>; - async fn get_accepted_followers(&self, local_user_id: Uuid) - -> anyhow::Result>; async fn remove_follower(&self, local_user_id: Uuid, actor_url: &str) -> anyhow::Result<()>; async fn block_actor(&self, local_user_id: Uuid, actor_url: &str) -> anyhow::Result<()>; async fn unblock_actor(&self, local_user_id: Uuid, actor_url: &str) -> anyhow::Result<()>; @@ -54,15 +49,6 @@ impl ActivityPubPort for ActivityPubService { async fn actor_json(&self, user_id: &str) -> anyhow::Result { self.actor_json(user_id).await } - async fn count_following(&self, local_user_id: Uuid) -> anyhow::Result { - self.count_following(local_user_id).await - } - async fn count_accepted_followers(&self, local_user_id: Uuid) -> anyhow::Result { - self.count_accepted_followers(local_user_id).await - } - async fn get_pending_followers(&self, local_user_id: Uuid) -> anyhow::Result> { - self.get_pending_followers(local_user_id).await - } async fn follow(&self, local_user_id: Uuid, handle: &str) -> anyhow::Result<()> { self.follow(local_user_id, handle).await } @@ -86,12 +72,6 @@ impl ActivityPubPort for ActivityPubService { async fn get_following(&self, local_user_id: Uuid) -> anyhow::Result> { self.get_following(local_user_id).await } - async fn get_accepted_followers( - &self, - local_user_id: Uuid, - ) -> anyhow::Result> { - self.get_accepted_followers(local_user_id).await - } async fn remove_follower(&self, local_user_id: Uuid, actor_url: &str) -> anyhow::Result<()> { self.remove_follower(local_user_id, actor_url).await } @@ -147,15 +127,6 @@ impl ActivityPubPort for NoopActivityPubService { async fn actor_json(&self, _: &str) -> anyhow::Result { Ok(String::new()) } - async fn count_following(&self, _: Uuid) -> anyhow::Result { - Ok(0) - } - async fn count_accepted_followers(&self, _: Uuid) -> anyhow::Result { - Ok(0) - } - async fn get_pending_followers(&self, _: Uuid) -> anyhow::Result> { - Ok(vec![]) - } async fn follow(&self, _: Uuid, _: &str) -> anyhow::Result<()> { Ok(()) } @@ -171,9 +142,6 @@ impl ActivityPubPort for NoopActivityPubService { async fn get_following(&self, _: Uuid) -> anyhow::Result> { Ok(vec![]) } - async fn get_accepted_followers(&self, _: Uuid) -> anyhow::Result> { - Ok(vec![]) - } async fn remove_follower(&self, _: Uuid, _: &str) -> anyhow::Result<()> { Ok(()) } diff --git a/crates/adapters/activitypub/src/social_adapter.rs b/crates/adapters/activitypub/src/social_adapter.rs index e316965..4111443 100644 --- a/crates/adapters/activitypub/src/social_adapter.rs +++ b/crates/adapters/activitypub/src/social_adapter.rs @@ -3,17 +3,17 @@ use std::sync::Arc; use async_trait::async_trait; use domain::{ errors::DomainError, - ports::{SocialCommand, SocialQuery, UserRepository}, - value_objects::{FollowTarget, SocialActor, SocialIdentity, UserId}, + ports::{FollowCommand, FollowQuery, SocialCommand, SocialQuery, UserRepository}, + value_objects::{FollowStatus, FollowTarget, SocialActor, SocialIdentity, UserId, Username}, }; -use k_ap::RemoteActor; - use super::ActivityPubPort; pub struct CompositeSocialAdapter { ap_service: Arc, user_repo: Arc, + follow_command: Arc, + follow_query: Arc, base_url: String, } @@ -21,11 +21,15 @@ impl CompositeSocialAdapter { pub fn new( ap_service: Arc, user_repo: Arc, + follow_command: Arc, + follow_query: Arc, base_url: String, ) -> Self { Self { ap_service, user_repo, + follow_command, + follow_query, base_url, } } @@ -41,42 +45,31 @@ impl CompositeSocialAdapter { } } - fn identity_from_actor_url(&self, url: &str) -> SocialIdentity { - let prefix = format!("{}/users/", self.base_url); - if let Some(uuid_str) = url.strip_prefix(&prefix) - && let Ok(uuid) = uuid::Uuid::parse_str(uuid_str) - { - return SocialIdentity::Local(UserId::from_uuid(uuid)); - } - SocialIdentity::Remote { - actor_url: url.to_string(), - } - } - - fn remote_actor_to_social_actor(&self, actor: RemoteActor) -> SocialActor { - let identity = self.identity_from_actor_url(&actor.url); - SocialActor { - identity, - handle: actor.handle, - display_name: actor.display_name, - avatar_url: actor.avatar_url, - } - } - - async fn resolve_handle(&self, identity: &SocialIdentity) -> Result { - match identity { - SocialIdentity::Local(uid) => { - let user = self - .user_repo - .find_by_id(uid) - .await? - .ok_or_else(|| DomainError::NotFound("User not found".into()))?; - let host = url::Url::parse(&self.base_url) - .map(|u| u.host_str().unwrap_or("localhost").to_string()) - .unwrap_or_else(|_| "localhost".to_string()); - Ok(format!("@{}@{}", user.username().value(), host)) + async fn resolve_target_identity( + &self, + target: &FollowTarget, + ) -> Result { + match target { + FollowTarget::Identity(id) => Ok(id.clone()), + FollowTarget::Handle(handle) => { + let host = handle.rsplit_once('@').map(|(_, h)| h).unwrap_or(""); + let local_host = SocialIdentity::host_from_base_url(&self.base_url); + if host == local_host { + let username_str = handle + .trim_start_matches('@') + .split('@') + .next() + .unwrap_or(""); + if let Ok(username) = Username::new(username_str.to_string()) + && let Some(user) = self.user_repo.find_by_username(&username).await? + { + return Ok(SocialIdentity::Local(user.id().clone())); + } + } + Ok(SocialIdentity::Remote { + actor_url: handle.clone(), + }) } - SocialIdentity::Remote { actor_url } => Ok(actor_url.clone()), } } } @@ -88,16 +81,38 @@ fn ap_err(e: anyhow::Error) -> DomainError { #[async_trait] impl SocialCommand for CompositeSocialAdapter { async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError> { - if let FollowTarget::Identity(SocialIdentity::Local(target_id)) = target - && follower == target_id - { - return Err(DomainError::ValidationError( - "Cannot follow yourself".into(), - )); + let identity = self.resolve_target_identity(target).await?; + + if let SocialIdentity::Local(ref target_id) = identity { + if follower == target_id { + return Err(DomainError::ValidationError( + "Cannot follow yourself".into(), + )); + } + let follower_url = self.local_actor_url(follower); + let target_url = self.local_actor_url(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?; + return Ok(()); } + let handle = match target { FollowTarget::Handle(h) => h.clone(), - FollowTarget::Identity(id) => self.resolve_handle(id).await?, + FollowTarget::Identity(id) => match id { + SocialIdentity::Local(uid) => { + let user = self + .user_repo + .find_by_id(uid) + .await? + .ok_or_else(|| DomainError::NotFound("User not found".into()))?; + SocialIdentity::format_local_handle(user.username().value(), &self.base_url) + } + SocialIdentity::Remote { actor_url } => actor_url.clone(), + }, }; self.ap_service .follow(follower.value(), &handle) @@ -111,10 +126,23 @@ impl SocialCommand for CompositeSocialAdapter { target: &SocialIdentity, ) -> Result<(), DomainError> { let actor_url = self.actor_url_from_identity(target); - self.ap_service - .unfollow(follower.value(), &actor_url) - .await - .map_err(ap_err) + match target { + SocialIdentity::Local(target_id) => { + let follower_url = self.local_actor_url(follower); + self.follow_command + .remove_follow(follower.value(), &actor_url) + .await?; + self.follow_command + .remove_follower_record(target_id.value(), &follower_url) + .await?; + Ok(()) + } + SocialIdentity::Remote { .. } => self + .ap_service + .unfollow(follower.value(), &actor_url) + .await + .map_err(ap_err), + } } async fn accept_follow( @@ -123,10 +151,23 @@ impl SocialCommand for CompositeSocialAdapter { requester: &SocialIdentity, ) -> Result<(), DomainError> { let actor_url = self.actor_url_from_identity(requester); - self.ap_service - .accept_follower(owner.value(), &actor_url) - .await - .map_err(ap_err) + match requester { + SocialIdentity::Local(requester_id) => { + let owner_url = self.local_actor_url(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(()) + } + SocialIdentity::Remote { .. } => self + .ap_service + .accept_follower(owner.value(), &actor_url) + .await + .map_err(ap_err), + } } async fn reject_follow( @@ -135,10 +176,23 @@ impl SocialCommand for CompositeSocialAdapter { requester: &SocialIdentity, ) -> Result<(), DomainError> { let actor_url = self.actor_url_from_identity(requester); - self.ap_service - .reject_follower(owner.value(), &actor_url) - .await - .map_err(ap_err) + match requester { + SocialIdentity::Local(requester_id) => { + let owner_url = self.local_actor_url(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(()) + } + SocialIdentity::Remote { .. } => self + .ap_service + .reject_follower(owner.value(), &actor_url) + .await + .map_err(ap_err), + } } async fn remove_follower( @@ -147,10 +201,23 @@ impl SocialCommand for CompositeSocialAdapter { follower: &SocialIdentity, ) -> Result<(), DomainError> { let actor_url = self.actor_url_from_identity(follower); - self.ap_service - .remove_follower(owner.value(), &actor_url) - .await - .map_err(ap_err) + match follower { + SocialIdentity::Local(follower_id) => { + let owner_url = self.local_actor_url(owner); + self.follow_command + .remove_follower_record(owner.value(), &actor_url) + .await?; + self.follow_command + .remove_follow(follower_id.value(), &owner_url) + .await?; + Ok(()) + } + SocialIdentity::Remote { .. } => self + .ap_service + .remove_follower(owner.value(), &actor_url) + .await + .map_err(ap_err), + } } async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> { @@ -173,53 +240,29 @@ impl SocialCommand for CompositeSocialAdapter { #[async_trait] impl SocialQuery for CompositeSocialAdapter { async fn get_following(&self, user: &UserId) -> Result, DomainError> { - let actors = self - .ap_service - .get_following(user.value()) + self.follow_query + .get_following(user.value(), &self.base_url) .await - .map_err(ap_err)?; - Ok(actors - .into_iter() - .map(|a| self.remote_actor_to_social_actor(a)) - .collect()) } async fn get_followers(&self, user: &UserId) -> Result, DomainError> { - let actors = self - .ap_service - .get_accepted_followers(user.value()) + self.follow_query + .get_followers(user.value(), &self.base_url) .await - .map_err(ap_err)?; - Ok(actors - .into_iter() - .map(|a| self.remote_actor_to_social_actor(a)) - .collect()) } async fn get_pending_followers(&self, user: &UserId) -> Result, DomainError> { - let actors = self - .ap_service - .get_pending_followers(user.value()) + self.follow_query + .get_pending_followers(user.value(), &self.base_url) .await - .map_err(ap_err)?; - Ok(actors - .into_iter() - .map(|a| self.remote_actor_to_social_actor(a)) - .collect()) } async fn count_following(&self, user: &UserId) -> Result { - self.ap_service - .count_following(user.value()) - .await - .map_err(ap_err) + self.follow_query.count_following(user.value()).await } async fn count_followers(&self, user: &UserId) -> Result { - self.ap_service - .count_accepted_followers(user.value()) - .await - .map_err(ap_err) + self.follow_query.count_followers(user.value()).await } async fn get_blocked(&self, user: &UserId) -> Result, DomainError> { @@ -230,7 +273,15 @@ impl SocialQuery for CompositeSocialAdapter { .map_err(ap_err)?; Ok(actors .into_iter() - .map(|a| self.remote_actor_to_social_actor(a)) + .map(|a| { + let identity = SocialIdentity::from_actor_url(&a.url, &self.base_url); + SocialActor { + identity, + handle: a.handle, + display_name: a.display_name, + avatar_url: a.avatar_url, + } + }) .collect()) } @@ -239,7 +290,9 @@ impl SocialQuery for CompositeSocialAdapter { follower: &UserId, target: &SocialIdentity, ) -> Result { - let following = self.get_following(follower).await?; - Ok(following.iter().any(|a| a.identity == *target)) + let actor_url = self.actor_url_from_identity(target); + self.follow_query + .is_following(follower.value(), &actor_url) + .await } } diff --git a/crates/adapters/postgres-federation/src/follow_repository.rs b/crates/adapters/postgres-federation/src/follow_repository.rs new file mode 100644 index 0000000..15f7f6a --- /dev/null +++ b/crates/adapters/postgres-federation/src/follow_repository.rs @@ -0,0 +1,299 @@ +use async_trait::async_trait; +use chrono::Utc; +use domain::{ + errors::DomainError, + value_objects::{FollowStatus, SocialActor, SocialIdentity}, +}; +use sqlx::Row; + +use crate::PostgresFederationRepository; +use adapter_common::datetime_to_str; + +fn follow_status_to_str(status: &FollowStatus) -> &'static str { + match status { + FollowStatus::Pending => "pending", + FollowStatus::Accepted => "accepted", + FollowStatus::Rejected => "rejected", + } +} + +fn infra_err(e: impl std::fmt::Display) -> DomainError { + DomainError::InfrastructureError(e.to_string()) +} + +#[async_trait] +impl domain::ports::FollowCommand for PostgresFederationRepository { + async fn add_follow( + &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); + 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::timestamptz, $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::timestamptz, '') + 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(()) + } +} + +fn social_actor_from_row(row: &sqlx::postgres::PgRow, base_url: &str) -> SocialActor { + let actor_url: String = row.get("remote_actor_url"); + let identity = SocialIdentity::from_actor_url(&actor_url, base_url); + + let (handle, display_name, avatar_url) = match &identity { + SocialIdentity::Local(_) => { + let username: Option = row.try_get("local_username").ok().flatten(); + let display: Option = row.try_get("local_display").ok().flatten(); + let avatar: Option = row.try_get("local_avatar").ok().flatten(); + let handle = username + .as_deref() + .map(|u| SocialIdentity::format_local_handle(u, base_url)) + .unwrap_or_else(|| actor_url.clone()); + (handle, display, avatar) + } + SocialIdentity::Remote { .. } => { + let handle: String = row + .try_get("remote_handle") + .ok() + .unwrap_or_else(|| actor_url.clone()); + let display: Option = row.try_get("remote_display").ok().flatten(); + let avatar: Option = row.try_get("remote_avatar").ok().flatten(); + (handle, display, avatar) + } + }; + + SocialActor { + identity, + handle, + display_name, + avatar_url, + } +} + +#[async_trait] +impl domain::ports::FollowQuery for PostgresFederationRepository { + async fn get_following( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_following f + LEFT JOIN users u ON f.remote_actor_url = $1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = $2 AND f.status = 'accepted'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn get_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_followers f + LEFT JOIN users u ON f.remote_actor_url = $1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = $2 AND f.status = 'accepted'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn get_pending_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_followers f + LEFT JOIN users u ON f.remote_actor_url = $1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = $2 AND f.status = 'pending'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn count_following(&self, user_id: uuid::Uuid) -> Result { + let uid = user_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_following WHERE local_user_id = $1 AND status = 'accepted'", + ) + .bind(&uid) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count as usize) + } + + async fn count_followers(&self, user_id: uuid::Uuid) -> Result { + let uid = user_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_followers WHERE local_user_id = $1 AND status = 'accepted'", + ) + .bind(&uid) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count as usize) + } + + async fn is_following( + &self, + follower_id: uuid::Uuid, + target_actor_url: &str, + ) -> Result { + let uid = follower_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_following WHERE local_user_id = $1 AND remote_actor_url = $2 AND status = 'accepted'", + ) + .bind(&uid) + .bind(target_actor_url) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count > 0) + } +} diff --git a/crates/adapters/postgres-federation/src/lib.rs b/crates/adapters/postgres-federation/src/lib.rs index df2bcad..961d2fa 100644 --- a/crates/adapters/postgres-federation/src/lib.rs +++ b/crates/adapters/postgres-federation/src/lib.rs @@ -4,6 +4,7 @@ pub mod ap_content; mod blocklist; mod federated_profile; mod follow; +mod follow_repository; pub mod remote_goals; mod review; mod social; @@ -79,13 +80,15 @@ pub fn create_federated_profile_query( pub fn wire(pool: PgPool) -> activitypub::FederationRepos { let fed = std::sync::Arc::new(PostgresFederationRepository::new(pool)); - ( - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - fed as _, - ) + activitypub::FederationRepos { + activity: std::sync::Arc::clone(&fed) as _, + follow: std::sync::Arc::clone(&fed) as _, + actor: std::sync::Arc::clone(&fed) as _, + blocklist: std::sync::Arc::clone(&fed) as _, + admin_query: std::sync::Arc::clone(&fed) as _, + review_store: std::sync::Arc::clone(&fed) as _, + remote_watchlist: std::sync::Arc::clone(&fed) as _, + follow_command: std::sync::Arc::clone(&fed) as _, + follow_query: fed as _, + } } diff --git a/crates/adapters/sqlite-federation/src/follow_repository.rs b/crates/adapters/sqlite-federation/src/follow_repository.rs new file mode 100644 index 0000000..f981db8 --- /dev/null +++ b/crates/adapters/sqlite-federation/src/follow_repository.rs @@ -0,0 +1,299 @@ +use async_trait::async_trait; +use chrono::Utc; +use domain::{ + errors::DomainError, + value_objects::{FollowStatus, SocialActor, SocialIdentity}, +}; +use sqlx::Row; + +use crate::SqliteFederationRepository; +use adapter_common::datetime_to_str; + +fn follow_status_to_str(status: &FollowStatus) -> &'static str { + match status { + FollowStatus::Pending => "pending", + FollowStatus::Accepted => "accepted", + FollowStatus::Rejected => "rejected", + } +} + +fn infra_err(e: impl std::fmt::Display) -> DomainError { + DomainError::InfrastructureError(e.to_string()) +} + +#[async_trait] +impl domain::ports::FollowCommand for SqliteFederationRepository { + async fn add_follow( + &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); + 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(()) + } +} + +fn social_actor_from_row(row: &sqlx::sqlite::SqliteRow, base_url: &str) -> SocialActor { + let actor_url: String = row.get("remote_actor_url"); + let identity = SocialIdentity::from_actor_url(&actor_url, base_url); + + let (handle, display_name, avatar_url) = match &identity { + SocialIdentity::Local(_) => { + let username: Option = row.try_get("local_username").ok().flatten(); + let display: Option = row.try_get("local_display").ok().flatten(); + let avatar: Option = row.try_get("local_avatar").ok().flatten(); + let handle = username + .as_deref() + .map(|u| SocialIdentity::format_local_handle(u, base_url)) + .unwrap_or_else(|| actor_url.clone()); + (handle, display, avatar) + } + SocialIdentity::Remote { .. } => { + let handle: String = row + .try_get("remote_handle") + .ok() + .unwrap_or_else(|| actor_url.clone()); + let display: Option = row.try_get("remote_display").ok().flatten(); + let avatar: Option = row.try_get("remote_avatar").ok().flatten(); + (handle, display, avatar) + } + }; + + SocialActor { + identity, + handle, + display_name, + avatar_url, + } +} + +#[async_trait] +impl domain::ports::FollowQuery for SqliteFederationRepository { + async fn get_following( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_following f + LEFT JOIN users u ON f.remote_actor_url = ?1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = ?2 AND f.status = 'accepted'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn get_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_followers f + LEFT JOIN users u ON f.remote_actor_url = ?1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = ?2 AND f.status = 'accepted'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn get_pending_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError> { + let uid = user_id.to_string(); + let rows = sqlx::query( + "SELECT f.remote_actor_url, + u.username AS local_username, u.display_name AS local_display, u.avatar_url AS local_avatar, + a.handle AS remote_handle, a.display_name AS remote_display, a.avatar_url AS remote_avatar + FROM ap_followers f + LEFT JOIN users u ON f.remote_actor_url = ?1 || '/users/' || u.id + LEFT JOIN ap_remote_actors a ON a.url = f.remote_actor_url + WHERE f.local_user_id = ?2 AND f.status = 'pending'", + ) + .bind(base_url) + .bind(&uid) + .fetch_all(&self.pool) + .await + .map_err(infra_err)?; + Ok(rows + .iter() + .map(|r| social_actor_from_row(r, base_url)) + .collect()) + } + + async fn count_following(&self, user_id: uuid::Uuid) -> Result { + let uid = user_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_following WHERE local_user_id = ? AND status = 'accepted'", + ) + .bind(&uid) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count as usize) + } + + async fn count_followers(&self, user_id: uuid::Uuid) -> Result { + let uid = user_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_followers WHERE local_user_id = ? AND status = 'accepted'", + ) + .bind(&uid) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count as usize) + } + + async fn is_following( + &self, + follower_id: uuid::Uuid, + target_actor_url: &str, + ) -> Result { + let uid = follower_id.to_string(); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ? AND status = 'accepted'", + ) + .bind(&uid) + .bind(target_actor_url) + .fetch_one(&self.pool) + .await + .map_err(infra_err)?; + Ok(count > 0) + } +} diff --git a/crates/adapters/sqlite-federation/src/lib.rs b/crates/adapters/sqlite-federation/src/lib.rs index c6fd0e4..6430d10 100644 --- a/crates/adapters/sqlite-federation/src/lib.rs +++ b/crates/adapters/sqlite-federation/src/lib.rs @@ -3,6 +3,7 @@ mod actor; mod blocklist; mod federated_profile; mod follow; +mod follow_repository; mod review; mod social; mod watchlist; @@ -91,15 +92,17 @@ pub fn create_federated_profile_query( pub fn wire(pool: SqlitePool) -> activitypub::FederationRepos { let fed = std::sync::Arc::new(SqliteFederationRepository::new(pool)); - ( - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - std::sync::Arc::clone(&fed) as _, - fed as _, - ) + activitypub::FederationRepos { + activity: std::sync::Arc::clone(&fed) as _, + follow: std::sync::Arc::clone(&fed) as _, + actor: std::sync::Arc::clone(&fed) as _, + blocklist: std::sync::Arc::clone(&fed) as _, + admin_query: std::sync::Arc::clone(&fed) as _, + review_store: std::sync::Arc::clone(&fed) as _, + remote_watchlist: std::sync::Arc::clone(&fed) as _, + follow_command: std::sync::Arc::clone(&fed) as _, + follow_query: fed as _, + } } #[cfg(test)] diff --git a/crates/domain/src/ports/follow.rs b/crates/domain/src/ports/follow.rs new file mode 100644 index 0000000..0fa9bcf --- /dev/null +++ b/crates/domain/src/ports/follow.rs @@ -0,0 +1,80 @@ +use async_trait::async_trait; + +use crate::{ + errors::DomainError, + value_objects::{FollowStatus, SocialActor}, +}; + +#[async_trait] +pub trait FollowCommand: Send + Sync { + async fn add_follow( + &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, + ) -> Result<(), DomainError>; +} + +#[async_trait] +pub trait FollowQuery: Send + Sync { + async fn get_following( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError>; + + async fn get_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError>; + + async fn get_pending_followers( + &self, + user_id: uuid::Uuid, + base_url: &str, + ) -> Result, DomainError>; + + async fn count_following(&self, user_id: uuid::Uuid) -> Result; + + async fn count_followers(&self, user_id: uuid::Uuid) -> Result; + + async fn is_following( + &self, + follower_id: uuid::Uuid, + target_actor_url: &str, + ) -> Result; +} diff --git a/crates/domain/src/ports/mod.rs b/crates/domain/src/ports/mod.rs index 71d6dc0..7b4f6f9 100644 --- a/crates/domain/src/ports/mod.rs +++ b/crates/domain/src/ports/mod.rs @@ -2,6 +2,7 @@ pub mod auth; pub mod diary; pub mod events; pub mod federated_profile; +pub mod follow; pub mod goals; pub mod image_fetcher; pub mod images; @@ -21,6 +22,7 @@ pub use auth::*; pub use diary::*; pub use events::*; pub use federated_profile::*; +pub use follow::*; pub use goals::*; pub use image_fetcher::*; pub use images::*; diff --git a/crates/domain/src/value_objects/social.rs b/crates/domain/src/value_objects/social.rs index 8c874d1..41f9eb5 100644 --- a/crates/domain/src/value_objects/social.rs +++ b/crates/domain/src/value_objects/social.rs @@ -7,6 +7,18 @@ pub enum SocialIdentity { } impl SocialIdentity { + pub fn from_actor_url(actor_url: &str, base_url: &str) -> Self { + let prefix = format!("{}/users/", base_url); + if let Some(uuid_str) = actor_url.strip_prefix(&prefix) + && let Ok(uuid) = uuid::Uuid::parse_str(uuid_str) + { + return Self::Local(UserId::from_uuid(uuid)); + } + Self::Remote { + actor_url: actor_url.to_string(), + } + } + pub fn is_local(&self) -> bool { matches!(self, Self::Local(_)) } @@ -14,6 +26,26 @@ impl SocialIdentity { pub fn is_remote(&self) -> bool { matches!(self, Self::Remote { .. }) } + + pub fn format_local_handle(username: &str, base_url: &str) -> String { + let host = Self::host_from_base_url(base_url); + format!("@{}@{}", username, host) + } + + pub fn host_from_base_url(base_url: &str) -> &str { + base_url + .split("://") + .nth(1) + .and_then(|s| s.split('/').next()) + .unwrap_or("localhost") + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FollowStatus { + Pending, + Accepted, + Rejected, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/crates/infra-wiring/src/config.rs b/crates/infra-wiring/src/config.rs index 37d3813..75d1eb9 100644 --- a/crates/infra-wiring/src/config.rs +++ b/crates/infra-wiring/src/config.rs @@ -19,8 +19,10 @@ impl AppConfig { let allow_registration = std::env::var("ALLOW_REGISTRATION") .map(|v| v == "true" || v == "1") .unwrap_or(false); - let base_url = - std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_string()); + let base_url = std::env::var("BASE_URL") + .unwrap_or_else(|_| "http://localhost:3000".to_string()) + .trim_end_matches('/') + .to_string(); let rate_limit = std::env::var("RATE_LIMIT") .ok() .and_then(|v| v.parse().ok()) diff --git a/crates/presentation/src/handlers/diary.rs b/crates/presentation/src/handlers/diary.rs index 0f1cf1a..6dd47c2 100644 --- a/crates/presentation/src/handlers/diary.rs +++ b/crates/presentation/src/handlers/diary.rs @@ -312,8 +312,7 @@ pub async fn get_activity_feed_html( let limit = params.limit.unwrap_or(20); let offset = params.offset.unwrap_or(0); - let filter_following = - cfg!(feature = "federation") && params.filter == "following" && user_id.is_some(); + let filter_following = params.filter == "following" && user_id.is_some(); let filter_str = if filter_following { "following" } else { "all" }; let sort_by_str = match params.sort_by.as_str() { diff --git a/crates/presentation/src/handlers/social.rs b/crates/presentation/src/handlers/social.rs index ce9709c..f61c576 100644 --- a/crates/presentation/src/handlers/social.rs +++ b/crates/presentation/src/handlers/social.rs @@ -189,9 +189,7 @@ pub async fn block_actor_api( &deps, application::social::commands::SocialCmd::Block { blocker_id: user.0.value(), - target: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + target: SocialIdentity::from_actor_url(&body.actor_url, &state.app_ctx.config.base_url), }, ) .await?; @@ -217,9 +215,7 @@ pub async fn unblock_actor_api( &deps, application::social::commands::SocialCmd::Unblock { blocker_id: user.0.value(), - target: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + target: SocialIdentity::from_actor_url(&body.actor_url, &state.app_ctx.config.base_url), }, ) .await?; @@ -381,9 +377,7 @@ pub async fn unfollow( &deps, application::social::commands::SocialCmd::Unfollow { follower_id: user.0.value(), - target: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + target: SocialIdentity::from_actor_url(&body.actor_url, &state.app_ctx.config.base_url), }, ) .await?; @@ -409,9 +403,10 @@ pub async fn accept_follower( &deps, application::social::commands::SocialCmd::AcceptFollow { owner_id: user.0.value(), - requester: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + requester: SocialIdentity::from_actor_url( + &body.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await?; @@ -437,9 +432,10 @@ pub async fn reject_follower( &deps, application::social::commands::SocialCmd::RejectFollow { owner_id: user.0.value(), - requester: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + requester: SocialIdentity::from_actor_url( + &body.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await?; @@ -465,9 +461,10 @@ pub async fn remove_follower( &deps, application::social::commands::SocialCmd::RemoveFollower { owner_id: user.0.value(), - follower: SocialIdentity::Remote { - actor_url: body.actor_url, - }, + follower: SocialIdentity::from_actor_url( + &body.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await?; @@ -563,9 +560,7 @@ pub async fn unfollow_remote_user( &deps, application::social::commands::SocialCmd::Unfollow { follower_id: user_id.value(), - target: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + target: SocialIdentity::from_actor_url(&form.actor_url, &state.app_ctx.config.base_url), }, ) .await @@ -602,9 +597,10 @@ pub async fn accept_follower_html( &deps, application::social::commands::SocialCmd::AcceptFollow { owner_id: user_id.value(), - requester: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + requester: SocialIdentity::from_actor_url( + &form.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await @@ -635,9 +631,10 @@ pub async fn reject_follower_html( &deps, application::social::commands::SocialCmd::RejectFollow { owner_id: user_id.value(), - requester: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + requester: SocialIdentity::from_actor_url( + &form.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await @@ -832,9 +829,10 @@ pub async fn remove_follower_html( &deps, application::social::commands::SocialCmd::RemoveFollower { owner_id: user_id.value(), - follower: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + follower: SocialIdentity::from_actor_url( + &form.actor_url, + &state.app_ctx.config.base_url, + ), }, ) .await @@ -1001,9 +999,7 @@ pub async fn post_block_actor_html( &deps, application::social::commands::SocialCmd::Block { blocker_id: user_id.value(), - target: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + target: SocialIdentity::from_actor_url(&form.actor_url, &state.app_ctx.config.base_url), }, ) .await @@ -1030,9 +1026,7 @@ pub async fn post_unblock_actor( &deps, application::social::commands::SocialCmd::Unblock { blocker_id: user_id.value(), - target: SocialIdentity::Remote { - actor_url: form.actor_url, - }, + target: SocialIdentity::from_actor_url(&form.actor_url, &state.app_ctx.config.base_url), }, ) .await diff --git a/crates/presentation/src/main.rs b/crates/presentation/src/main.rs index aa990f1..8728282 100644 --- a/crates/presentation/src/main.rs +++ b/crates/presentation/src/main.rs @@ -75,15 +75,7 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { social_command_arc, social_query_unified_arc, ) = { - let ( - activity_repo, - follow_repo, - actor_repo, - blocklist_repo, - social_query_arc, - review_store, - remote_watchlist_repo, - ) = match &db_pool { + let fed_repos = match &db_pool { #[cfg(feature = "postgres-federation")] factory::DbPool::Postgres(pool) => postgres_federation::wire(pool.clone()), #[cfg(feature = "sqlite-federation")] @@ -97,12 +89,12 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { let ep = create_event_publisher(event_bus, &db_pool).await?; let ap = activitypub::wire(activitypub::ActivityPubDeps { - activity_repo, - follow_repo, - actor_repo, - blocklist_repo, - review_store, - remote_watchlist_repo: remote_watchlist_repo.clone(), + activity_repo: fed_repos.activity, + follow_repo: fed_repos.follow, + actor_repo: fed_repos.actor, + blocklist_repo: fed_repos.blocklist, + review_store: fed_repos.review_store, + remote_watchlist_repo: fed_repos.remote_watchlist.clone(), remote_goal_repo: Arc::clone(&db.remote_goal), local_ap_content: Arc::clone(&ap_content_repo), movie_repo: Arc::clone(&db.movie_query), @@ -112,6 +104,8 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { stats_repo: Arc::clone(&db.stats), user_repo: Arc::clone(&db.user), federation_settings: std::sync::Arc::clone(&db.federation_settings), + follow_command: Arc::clone(&fed_repos.follow_command), + follow_query: Arc::clone(&fed_repos.follow_query), base_url: app_config.base_url.clone(), allow_registration: app_config.allow_registration, event_publisher: Arc::clone(&ep), @@ -123,6 +117,8 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { let composite_social = Arc::new(activitypub::CompositeSocialAdapter::new( Arc::clone(&ap_service_arc), Arc::clone(&db.user), + fed_repos.follow_command, + fed_repos.follow_query, app_config.base_url.clone(), )); @@ -130,8 +126,8 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { ep, ap_router, ap_service_arc, - social_query_arc, - remote_watchlist_repo, + fed_repos.admin_query, + fed_repos.remote_watchlist, composite_social.clone() as Arc, composite_social as Arc, ) diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index 464b6a5..0015671 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -61,15 +61,7 @@ async fn main() -> anyhow::Result<()> { ); // Wire federation repos early to get remote_watchlist_repo for AppContext. #[cfg(feature = "federation")] - let ( - fed_activity_repo, - fed_follow_repo, - fed_actor_repo, - fed_blocklist_repo, - _fed_social_query, - fed_review_store, - fed_remote_watchlist_repo, - ) = match &db.db_pool { + let fed_repos = match &db.db_pool { #[cfg(feature = "sqlite-federation")] db::DbPool::Sqlite(pool) => sqlite_federation::wire(pool.clone()), #[cfg(feature = "postgres-federation")] @@ -244,12 +236,12 @@ async fn main() -> anyhow::Result<()> { #[cfg(feature = "federation")] { let ap_wire = activitypub::wire(activitypub::ActivityPubDeps { - activity_repo: fed_activity_repo, - follow_repo: fed_follow_repo, - actor_repo: fed_actor_repo, - blocklist_repo: fed_blocklist_repo, - review_store: fed_review_store, - remote_watchlist_repo: fed_remote_watchlist_repo, + activity_repo: fed_repos.activity, + follow_repo: fed_repos.follow, + actor_repo: fed_repos.actor, + blocklist_repo: fed_repos.blocklist, + review_store: fed_repos.review_store, + remote_watchlist_repo: fed_repos.remote_watchlist, remote_goal_repo: Arc::clone(&remote_goal), local_ap_content: fed_ap_content, movie_repo: fed_movie_repo, @@ -258,6 +250,8 @@ async fn main() -> anyhow::Result<()> { goal_repo: fed_goal_repo, stats_repo: fed_stats_repo, user_repo: fed_user_repo, + follow_command: fed_repos.follow_command, + follow_query: fed_repos.follow_query, base_url, allow_registration, event_publisher: Arc::clone(&event_publisher), diff --git a/docs/adr/0003-local-follow-direct-db.md b/docs/adr/0003-local-follow-direct-db.md new file mode 100644 index 0000000..3e30d96 --- /dev/null +++ b/docs/adr/0003-local-follow-direct-db.md @@ -0,0 +1,20 @@ +# Local follows bypass ActivityPub, share storage + +ADR-0002 introduced `SocialIdentity` so the domain never branches on local vs remote, but the adapter (`CompositeSocialAdapter`) still routed everything through k_ap — meaning a local user following another local user triggered WebFinger resolution, HTTP signature verification, and inbox delivery to the same instance. Wasteful on any hardware, unacceptable on an N100. + +## Decision + +**Commands branch in the adapter, queries don't.** + +- `SocialCommand` methods in `CompositeSocialAdapter` check the `SocialIdentity` variant. Local targets get direct SQL writes to the `ap_followers`/`ap_following` tables (via a domain `FollowRepository` port). Remote targets delegate to `k_ap::ActivityPubService` as before. +- `SocialQuery` methods go through the domain port only — a single SQL query that left-joins `ap_followers`/`ap_following` against both `users` (local) and `ap_remote_actors` (remote), returning `SocialActor` directly. +- Local follows store the full actor URL (`https://instance.example/users/{uuid}`) in `remote_actor_url`, same format as remote follows. k_ap's AP collection endpoints read from these tables unchanged, so local relationships are visible to the fediverse automatically. +- Local user metadata (display name, avatar) is resolved from the `users` table at query time — no duplication into `ap_remote_actors`. +- Follow acceptance is required for both local and remote — no behavioral divergence. +- Local follow events (`FollowRequested`, `FollowAccepted`) are not broadcast as AP activities. The fediverse discovers local relationships passively via collection endpoints. + +## Considered Options + +- **Keep routing local through k_ap** — rejected because it wastes CPU/network on self-delivery and creates an unnecessary runtime dependency on federation for local social features. +- **Separate `local_follows` table** — rejected because it creates two sources of truth for the same concept and requires merging in collection endpoints. +- **Insert local users into `ap_remote_actors`** — rejected because it duplicates profile data and requires sync when local users update their profile.