Compare commits

...

4 Commits

Author SHA1 Message Date
22b1dd3f56 fix: portal wrapup share card to escape transform stacking context
All checks were successful
CI / Check / Test (push) Successful in 1h5m29s
2026-07-27 14:07:36 +02:00
813778dc7e fix: users table has avatar_path not avatar_url, resolve to full URL
Some checks failed
CI / Check / Test (push) Has been cancelled
2026-07-27 13:08:45 +02:00
063bab910b ADR-0003: local follows bypass AP, unified FollowCommand/FollowQuery ports, FederationRepos struct, base_url normalization
Some checks failed
CI / Check / Test (push) Has been cancelled
2026-07-27 12:56:30 +02:00
839cababe4 bump k-ap 0.5.0, decompose repo traits, LocalObject, ActivityStreamsType enum, split follow modules
Some checks failed
CI / Check / Test (push) Has been cancelled
2026-07-27 11:46:24 +02:00
43 changed files with 1465 additions and 727 deletions

8
Cargo.lock generated
View File

@@ -2892,9 +2892,9 @@ dependencies = [
[[package]]
name = "k-ap"
version = "0.4.6"
version = "0.5.0"
source = "sparse+https://git.gabrielkaszewski.dev/api/packages/GKaszewski/cargo/"
checksum = "97c1d000cf9c80891973bb5f5adb96c90e0aea9c929c7fa6cb794ad28938bc50"
checksum = "ab6066cccc6ae8aaa2f6262ac7d471e58930a04be3266a2e367d6bdd8aaaba29"
dependencies = [
"activitypub_federation",
"anyhow",
@@ -2903,9 +2903,11 @@ dependencies = [
"chrono",
"enum_delegate",
"futures",
"paste",
"reqwest 0.13.3",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tracing",
"url",
@@ -6714,7 +6716,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.48.0",
]
[[package]]

View File

@@ -4,7 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
k-ap = { version = "0.4.6", registry = "gitea" }
k-ap = { version = "0.5.0", registry = "gitea" }
domain = { workspace = true }
axum = { workspace = true }
serde = { workspace = true }

View File

@@ -1,8 +1,7 @@
use std::sync::Arc;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use k_ap::{ApContentReader, ApObjectHandler};
use k_ap::{ApContentReader, ApObjectHandler, LocalObject};
use url::Url;
use crate::{
@@ -21,10 +20,9 @@ impl ApContentReader for CompositeObjectHandler {
async fn get_local_objects_page(
&self,
user_id: uuid::Uuid,
before: Option<DateTime<Utc>>,
before: Option<chrono::DateTime<chrono::Utc>>,
limit: usize,
) -> anyhow::Result<Vec<(Url, serde_json::Value, DateTime<Utc>)>> {
// Fetch from all three sources (watchlist/goals return all, reviews use DB pagination)
) -> anyhow::Result<Vec<LocalObject>> {
let fetch_limit = limit * 3;
let reviews = self
.review
@@ -39,16 +37,15 @@ impl ApContentReader for CompositeObjectHandler {
.get_local_objects_page(user_id, None, usize::MAX)
.await?;
let mut all: Vec<(Url, serde_json::Value, DateTime<Utc>)> = Vec::new();
let mut all: Vec<LocalObject> = Vec::new();
all.extend(reviews);
all.extend(watchlist);
all.extend(goals);
// Apply before filter and sort descending by timestamp
if let Some(before_ts) = before {
all.retain(|(_, _, ts)| *ts < before_ts);
all.retain(|obj| obj.published_at < before_ts);
}
all.sort_by_key(|b| std::cmp::Reverse(b.2));
all.sort_by_key(|obj| std::cmp::Reverse(obj.published_at));
all.truncate(limit);
Ok(all)
}

View File

@@ -217,7 +217,7 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_create_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_create(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
let year = review.watched_at().year() as u16;
@@ -283,7 +283,7 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_update_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_update(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
Ok(())
@@ -349,7 +349,7 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_create_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_create(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
Ok(())
}
@@ -416,7 +416,7 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_update_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_update(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
}
@@ -462,7 +462,7 @@ impl ActivityPubEventHandler {
);
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_update_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_update(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
Ok(())
}
@@ -494,11 +494,11 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
if is_create {
self.ap_service
.broadcast_create_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_create(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
} else {
self.ap_service
.broadcast_update_note(user_id.value(), json, ApVisibility::Public, vec![])
.broadcast_update(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
}
Ok(())

View File

@@ -7,7 +7,7 @@ use domain::{
ports::{GoalQuery, RemoteGoalRepository},
value_objects::UserId,
};
use k_ap::{ApContentReader, ApObjectHandler};
use k_ap::{AS_PUBLIC, ApContentReader, ApObjectHandler, LocalObject};
use url::Url;
use crate::objects::{GoalObject, goal_to_ap_object};
@@ -26,7 +26,7 @@ impl ApContentReader for GoalObjectHandler {
user_id: uuid::Uuid,
_before: Option<DateTime<chrono::Utc>>,
_limit: usize,
) -> anyhow::Result<Vec<(Url, serde_json::Value, DateTime<chrono::Utc>)>> {
) -> anyhow::Result<Vec<LocalObject>> {
let uid = UserId::from_uuid(user_id);
let goals = self
.goal_repo
@@ -35,6 +35,7 @@ impl ApContentReader for GoalObjectHandler {
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
let actor = actor_url(&self.base_url, user_id);
let follower_cc = format!("{}/followers", actor);
let mut results = Vec::new();
for goal in goals {
let ap_id = goal_url(&self.base_url, user_id, goal.year());
@@ -47,7 +48,15 @@ impl ApContentReader for GoalObjectHandler {
0,
&self.base_url,
);
results.push((ap_id, serde_json::to_value(obj)?, published));
results.push(LocalObject {
ap_id,
object: serde_json::to_value(obj)?,
published_at: published,
to: vec![AS_PUBLIC.to_string()],
cc: vec![follower_cc.clone()],
bto: vec![],
bcc: vec![],
});
}
Ok(results)
}

View File

@@ -18,7 +18,7 @@ pub const INSTANCE_ACTOR_ID: uuid::Uuid =
pub use k_ap::{
ActivityPubService, ActivityRepository, ActorRepository, ApContentReader, ApFederationConfig,
ApObjectHandler, ApUser, ApUserRepository, BlocklistRepository, FederationData,
FollowRepository, Follower, FollowerStatus, FollowingStatus, RemoteActor,
FollowRepository, Follower, FollowerStatus, FollowingStatus, LocalObject, RemoteActor,
};
pub use event_handler::ActivityPubEventHandler;
@@ -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<dyn ActivityRepository>,
std::sync::Arc<dyn FollowRepository>,
std::sync::Arc<dyn ActorRepository>,
std::sync::Arc<dyn BlocklistRepository>,
std::sync::Arc<dyn domain::ports::FederationAdminQuery>,
std::sync::Arc<dyn RemoteReviewRepository>,
std::sync::Arc<dyn domain::ports::RemoteWatchlistRepository>,
);
pub struct FederationRepos {
pub activity: std::sync::Arc<dyn ActivityRepository>,
pub follow: std::sync::Arc<dyn FollowRepository>,
pub actor: std::sync::Arc<dyn ActorRepository>,
pub blocklist: std::sync::Arc<dyn BlocklistRepository>,
pub admin_query: std::sync::Arc<dyn domain::ports::FederationAdminQuery>,
pub review_store: std::sync::Arc<dyn RemoteReviewRepository>,
pub remote_watchlist: std::sync::Arc<dyn domain::ports::RemoteWatchlistRepository>,
pub follow_command: std::sync::Arc<dyn domain::ports::FollowCommand>,
pub follow_query: std::sync::Arc<dyn domain::ports::FollowQuery>,
}
pub struct ActivityPubWire {
pub service: std::sync::Arc<dyn ActivityPubPort>,
@@ -60,6 +62,8 @@ pub struct ActivityPubDeps {
pub stats_repo: std::sync::Arc<dyn domain::ports::StatsRepository>,
pub user_repo: std::sync::Arc<dyn domain::ports::UserRepository>,
pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
pub follow_command: std::sync::Arc<dyn domain::ports::FollowCommand>,
pub follow_query: std::sync::Arc<dyn domain::ports::FollowQuery>,
pub base_url: String,
pub allow_registration: bool,
pub event_publisher: std::sync::Arc<dyn domain::ports::EventPublisher>,
@@ -82,6 +86,8 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
stats_repo,
user_repo,
federation_settings,
follow_command: _,
follow_query: _,
base_url,
allow_registration,
event_publisher,

View File

@@ -1,11 +1,17 @@
use chrono::{DateTime, Utc};
use k_ap::AS_PUBLIC;
use k_ap::NoteType;
use serde::{Deserialize, Serialize};
use url::Url;
use domain::models::Review;
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
pub(crate) enum ActivityStreamsType {
#[default]
Note,
Article,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ApAttachment {
@@ -34,7 +40,7 @@ pub(crate) fn normalize_hashtag(title: &str) -> String {
#[serde(rename_all = "camelCase")]
pub struct ReviewObject {
#[serde(rename = "type")]
pub(crate) kind: NoteType,
pub(crate) kind: ActivityStreamsType,
pub(crate) id: Url,
pub(crate) attributed_to: Url,
pub(crate) content: String,
@@ -125,7 +131,7 @@ pub fn review_to_ap_object(review: &Review, input: ReviewApInput) -> ReviewObjec
};
ReviewObject {
kind: NoteType::default(),
kind: ActivityStreamsType::default(),
id: ap_id,
attributed_to: actor_url.clone(),
content,
@@ -150,7 +156,7 @@ pub fn review_to_ap_object(review: &Review, input: ReviewApInput) -> ReviewObjec
#[serde(rename_all = "camelCase")]
pub struct WatchlistObject {
#[serde(rename = "type")]
pub(crate) kind: NoteType,
pub(crate) kind: ActivityStreamsType,
pub(crate) id: Url,
pub(crate) attributed_to: Url,
pub(crate) content: String,
@@ -218,7 +224,7 @@ pub fn watchlist_to_ap_object(input: WatchlistApInput) -> WatchlistObject {
];
WatchlistObject {
kind: NoteType::default(),
kind: ActivityStreamsType::default(),
id: ap_id,
attributed_to: actor_url.clone(),
content,
@@ -240,7 +246,7 @@ pub fn watchlist_to_ap_object(input: WatchlistApInput) -> WatchlistObject {
#[serde(rename_all = "camelCase")]
pub struct GoalObject {
#[serde(rename = "type")]
pub(crate) kind: NoteType,
pub(crate) kind: ActivityStreamsType,
pub(crate) id: Url,
pub(crate) attributed_to: Url,
pub(crate) content: String,
@@ -277,7 +283,7 @@ pub fn goal_to_ap_object(
}];
GoalObject {
kind: NoteType::default(),
kind: ActivityStreamsType::default(),
id: ap_id,
attributed_to: actor_url.clone(),
content,

View File

@@ -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<String>;
async fn count_following(&self, local_user_id: Uuid) -> anyhow::Result<usize>;
async fn count_accepted_followers(&self, local_user_id: Uuid) -> anyhow::Result<usize>;
async fn get_pending_followers(&self, local_user_id: Uuid) -> anyhow::Result<Vec<RemoteActor>>;
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<Vec<RemoteActor>>;
async fn get_accepted_followers(&self, local_user_id: Uuid)
-> anyhow::Result<Vec<RemoteActor>>;
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<String> {
self.actor_json(user_id).await
}
async fn count_following(&self, local_user_id: Uuid) -> anyhow::Result<usize> {
self.count_following(local_user_id).await
}
async fn count_accepted_followers(&self, local_user_id: Uuid) -> anyhow::Result<usize> {
self.count_accepted_followers(local_user_id).await
}
async fn get_pending_followers(&self, local_user_id: Uuid) -> anyhow::Result<Vec<RemoteActor>> {
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<Vec<RemoteActor>> {
self.get_following(local_user_id).await
}
async fn get_accepted_followers(
&self,
local_user_id: Uuid,
) -> anyhow::Result<Vec<RemoteActor>> {
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<String> {
Ok(String::new())
}
async fn count_following(&self, _: Uuid) -> anyhow::Result<usize> {
Ok(0)
}
async fn count_accepted_followers(&self, _: Uuid) -> anyhow::Result<usize> {
Ok(0)
}
async fn get_pending_followers(&self, _: Uuid) -> anyhow::Result<Vec<RemoteActor>> {
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<Vec<RemoteActor>> {
Ok(vec![])
}
async fn get_accepted_followers(&self, _: Uuid) -> anyhow::Result<Vec<RemoteActor>> {
Ok(vec![])
}
async fn remove_follower(&self, _: Uuid, _: &str) -> anyhow::Result<()> {
Ok(())
}

View File

@@ -7,7 +7,7 @@ use domain::{
ports::{DiaryQuery, EventPublisher, LocalApContentQuery, MovieQuery},
value_objects::{Comment, ExternalMetadataId, MovieId, Rating, ReviewId, UserId},
};
use k_ap::{ApContentReader, ApObjectHandler};
use k_ap::{AS_PUBLIC, ApContentReader, ApObjectHandler, LocalObject};
use url::Url;
use crate::objects::{ReviewApInput, ReviewObject, review_to_ap_object};
@@ -30,7 +30,7 @@ impl ApContentReader for ReviewObjectHandler {
user_id: uuid::Uuid,
before: Option<chrono::DateTime<chrono::Utc>>,
limit: usize,
) -> anyhow::Result<Vec<(url::Url, serde_json::Value, chrono::DateTime<chrono::Utc>)>> {
) -> anyhow::Result<Vec<LocalObject>> {
let domain_user_id = UserId::from_uuid(user_id);
let before_naive = before.map(|dt| dt.naive_utc());
let entries = self
@@ -65,7 +65,16 @@ impl ApContentReader for ReviewObjectHandler {
base_url: self.base_url.clone(),
},
);
results.push((ap_id, serde_json::to_value(obj)?, published));
let follower_cc = format!("{}/followers", actor);
results.push(LocalObject {
ap_id,
object: serde_json::to_value(obj)?,
published_at: published,
to: vec![AS_PUBLIC.to_string()],
cc: vec![follower_cc],
bto: vec![],
bcc: vec![],
});
}
Ok(results)
}

View File

@@ -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<dyn ActivityPubPort>,
user_repo: Arc<dyn UserRepository>,
follow_command: Arc<dyn FollowCommand>,
follow_query: Arc<dyn FollowQuery>,
base_url: String,
}
@@ -21,11 +21,15 @@ impl CompositeSocialAdapter {
pub fn new(
ap_service: Arc<dyn ActivityPubPort>,
user_repo: Arc<dyn UserRepository>,
follow_command: Arc<dyn FollowCommand>,
follow_query: Arc<dyn FollowQuery>,
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<String, DomainError> {
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<SocialIdentity, DomainError> {
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<Vec<SocialActor>, 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<Vec<SocialActor>, 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<Vec<SocialActor>, 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<usize, DomainError> {
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<usize, DomainError> {
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<Vec<SocialActor>, 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<bool, DomainError> {
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
}
}

View File

@@ -7,7 +7,7 @@ use domain::{
ports::{LocalApContentQuery, RemoteWatchlistRepository},
value_objects::UserId,
};
use k_ap::{ApContentReader, ApObjectHandler};
use k_ap::{AS_PUBLIC, ApContentReader, ApObjectHandler, LocalObject};
use url::Url;
use crate::objects::{WatchlistApInput, WatchlistObject, watchlist_to_ap_object};
@@ -26,7 +26,7 @@ impl ApContentReader for WatchlistObjectHandler {
user_id: uuid::Uuid,
_before: Option<DateTime<chrono::Utc>>,
_limit: usize,
) -> anyhow::Result<Vec<(Url, serde_json::Value, DateTime<chrono::Utc>)>> {
) -> anyhow::Result<Vec<LocalObject>> {
let uid = UserId::from_uuid(user_id);
let entries = self
.content_query
@@ -35,6 +35,7 @@ impl ApContentReader for WatchlistObjectHandler {
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
let actor = actor_url(&self.base_url, user_id);
let follower_cc = format!("{}/followers", actor);
let mut results = Vec::new();
for WatchlistWithMovie { entry, movie } in entries {
let ap_id = watchlist_entry_url(&self.base_url, user_id, entry.movie_id.value());
@@ -54,7 +55,15 @@ impl ApContentReader for WatchlistObjectHandler {
added_at: published,
base_url: self.base_url.clone(),
});
results.push((ap_id, serde_json::to_value(obj)?, published));
results.push(LocalObject {
ap_id,
object: serde_json::to_value(obj)?,
published_at: published,
to: vec![AS_PUBLIC.to_string()],
cc: vec![follower_cc.clone()],
bto: vec![],
bcc: vec![],
});
}
Ok(results)
}

View File

@@ -13,7 +13,7 @@ sqlx = { version = "0.8.6", features = [
] }
activitypub = { workspace = true }
adapter-common = { workspace = true }
k-ap = { version = "0.4.6", registry = "gitea" }
k-ap = { version = "0.5.0", registry = "gitea" }
domain = { workspace = true }
uuid = { workspace = true }
chrono = { workspace = true }

View File

@@ -1,43 +1,41 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{ActorRepository, RemoteActor};
use k_ap::{AnnounceRepository, Keypair, KeypairRepository, RemoteActor, RemoteActorCache};
use sqlx::Row;
use super::{PG_ACTOR_COLS, PostgresFederationRepository, pg_remote_actor};
use adapter_common::datetime_to_str;
#[async_trait]
impl ActorRepository for PostgresFederationRepository {
async fn get_local_actor_keypair(
&self,
user_id: uuid::Uuid,
) -> Result<Option<(String, String)>> {
impl KeypairRepository for PostgresFederationRepository {
async fn get_local_actor_keypair(&self, user_id: uuid::Uuid) -> Result<Option<Keypair>> {
let uid = user_id.to_string();
let row =
sqlx::query("SELECT public_key, private_key FROM ap_local_actors WHERE user_id = $1")
.bind(&uid)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.get("public_key"), r.get("private_key"))))
Ok(row.map(|r| Keypair {
public_key: r.get("public_key"),
private_key: r.get("private_key"),
}))
}
async fn save_local_actor_keypair(
&self,
user_id: uuid::Uuid,
public_key: String,
private_key: String,
) -> Result<()> {
async fn save_local_actor_keypair(&self, user_id: uuid::Uuid, keypair: Keypair) -> Result<()> {
let uid = user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
sqlx::query(
"INSERT INTO ap_local_actors (user_id, public_key, private_key, created_at) VALUES ($1, $2, $3, $4::timestamptz)
ON CONFLICT(user_id) DO UPDATE SET public_key = EXCLUDED.public_key, private_key = EXCLUDED.private_key",
).bind(&uid).bind(&public_key).bind(&private_key).bind(&created_at).execute(&self.pool).await?;
).bind(&uid).bind(&keypair.public_key).bind(&keypair.private_key).bind(&created_at).execute(&self.pool).await?;
Ok(())
}
}
#[async_trait]
impl RemoteActorCache for PostgresFederationRepository {
async fn upsert_remote_actor(&self, actor: RemoteActor) -> Result<()> {
let now = Utc::now().naive_utc();
let fetched_at = datetime_to_str(&now);
@@ -68,7 +66,10 @@ impl ActorRepository for PostgresFederationRepository {
.await?;
Ok(row.as_ref().map(|r| pg_remote_actor(r, "url")))
}
}
#[async_trait]
impl AnnounceRepository for PostgresFederationRepository {
async fn add_announce(
&self,
activity_id: &str,

View File

@@ -1,14 +1,14 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{BlockedDomain, BlocklistRepository};
use k_ap::{ActorBlocklist, BlockedDomain, DomainBlocklist};
use sqlx::Row;
use super::PostgresFederationRepository;
use adapter_common::datetime_to_str;
#[async_trait]
impl BlocklistRepository for PostgresFederationRepository {
impl DomainBlocklist for PostgresFederationRepository {
async fn add_blocked_domain(&self, domain: &str, reason: Option<&str>) -> Result<()> {
let ts = datetime_to_str(&Utc::now().naive_utc());
sqlx::query("INSERT INTO blocked_domains (domain, reason, blocked_at) VALUES ($1, $2, $3) ON CONFLICT(domain) DO UPDATE SET reason = EXCLUDED.reason")
@@ -48,7 +48,10 @@ impl BlocklistRepository for PostgresFederationRepository {
.await?;
Ok(count > 0)
}
}
#[async_trait]
impl ActorBlocklist for PostgresFederationRepository {
async fn add_blocked_actor(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
let ts = datetime_to_str(&Utc::now().naive_utc());

View File

@@ -1,18 +1,16 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{
ActorRepository, FollowRepository, Follower, FollowerStatus, FollowingStatus, RemoteActor,
};
use k_ap::{Follower, FollowerReader, FollowerStatus, FollowerWriter, RemoteActor};
use sqlx::Row;
use super::{
use crate::{
PG_ACTOR_COLS, PostgresFederationRepository, pg_remote_actor, status_to_str, str_to_status,
};
use adapter_common::datetime_to_str;
#[async_trait]
impl FollowRepository for PostgresFederationRepository {
impl FollowerWriter for PostgresFederationRepository {
async fn add_follower(
&self,
local_user_id: uuid::Uuid,
@@ -59,6 +57,25 @@ impl FollowRepository for PostgresFederationRepository {
Ok(())
}
async fn update_follower_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowerStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = status_to_str(&status);
let result = sqlx::query("UPDATE ap_followers SET status = $1 WHERE local_user_id = $2 AND remote_actor_url = $3")
.bind(status_str).bind(&uid).bind(remote_actor_url).execute(&self.pool).await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_follower_status: no row found");
}
Ok(())
}
}
#[async_trait]
impl FollowerReader for PostgresFederationRepository {
async fn get_followers(&self, local_user_id: uuid::Uuid) -> Result<Vec<Follower>> {
let uid = local_user_id.to_string();
let q = format!(
@@ -116,22 +133,6 @@ impl FollowRepository for PostgresFederationRepository {
Ok(count as usize)
}
async fn update_follower_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowerStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = status_to_str(&status);
let result = sqlx::query("UPDATE ap_followers SET status = $1 WHERE local_user_id = $2 AND remote_actor_url = $3")
.bind(status_str).bind(&uid).bind(remote_actor_url).execute(&self.pool).await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_follower_status: no row found");
}
Ok(())
}
async fn get_pending_followers(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let q = format!(
@@ -193,129 +194,4 @@ impl FollowRepository for PostgresFederationRepository {
.map(|row| pg_remote_actor(row, "remote_actor_url"))
.collect())
}
async fn add_following(
&self,
local_user_id: uuid::Uuid,
actor: RemoteActor,
follow_activity_id: &str,
) -> Result<()> {
let uid = local_user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
ActorRepository::upsert_remote_actor(self, actor.clone()).await?;
sqlx::query("INSERT INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at) VALUES ($1, $2, $3, $4::timestamptz) ON CONFLICT DO NOTHING")
.bind(&uid).bind(&actor.url).bind(follow_activity_id).bind(&created_at).execute(&self.pool).await?;
Ok(())
}
async fn get_follow_activity_id(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<String> = sqlx::query_scalar("SELECT follow_activity_id FROM ap_following WHERE local_user_id = $1 AND remote_actor_url = $2")
.bind(&uid).bind(remote_actor_url).fetch_optional(&self.pool).await?;
Ok(row)
}
async fn remove_following(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
sqlx::query("DELETE FROM ap_following WHERE local_user_id = $1 AND remote_actor_url = $2")
.bind(&uid)
.bind(actor_url)
.execute(&self.pool)
.await?;
Ok(())
}
async fn get_following(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let q = format!(
"SELECT a.url, {PG_ACTOR_COLS} FROM ap_following f INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.status = 'accepted'"
);
let rows = sqlx::query(&q).bind(&uid).fetch_all(&self.pool).await?;
Ok(rows.iter().map(|row| pg_remote_actor(row, "url")).collect())
}
async fn count_following(&self, local_user_id: uuid::Uuid) -> Result<usize> {
let uid = local_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?;
Ok(count as usize)
}
async fn get_following_page(
&self,
local_user_id: uuid::Uuid,
offset: u32,
limit: usize,
) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let q = format!(
"SELECT a.url, {PG_ACTOR_COLS} FROM ap_following f INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.status = 'accepted' ORDER BY f.created_at ASC LIMIT $2 OFFSET $3"
);
let rows = sqlx::query(&q)
.bind(&uid)
.bind(limit as i64)
.bind(offset as i64)
.fetch_all(&self.pool)
.await?;
Ok(rows.iter().map(|row| pg_remote_actor(row, "url")).collect())
}
async fn update_following_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowingStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = match status {
FollowingStatus::Pending => "pending",
FollowingStatus::Accepted => "accepted",
};
let result = sqlx::query("UPDATE ap_following SET status = $1 WHERE local_user_id = $2 AND remote_actor_url = $3")
.bind(status_str).bind(&uid).bind(remote_actor_url).execute(&self.pool).await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_following_status: no row found");
}
Ok(())
}
async fn get_following_outbox_url(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<Option<String>> = sqlx::query_scalar(
"SELECT a.outbox_url FROM ap_following f INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.remote_actor_url = $2",
).bind(&uid).bind(remote_actor_url).fetch_optional(&self.pool).await?;
Ok(row.flatten())
}
async fn migrate_follower_actor(
&self,
old_actor_url: &str,
new_actor_url: &str,
) -> Result<Vec<uuid::Uuid>> {
let candidates: Vec<String> = sqlx::query_scalar(
"SELECT local_user_id FROM ap_following WHERE remote_actor_url = $1 AND local_user_id NOT IN (SELECT local_user_id FROM ap_following WHERE remote_actor_url = $2)",
).bind(old_actor_url).bind(new_actor_url).fetch_all(&self.pool).await?;
if candidates.is_empty() {
return Ok(vec![]);
}
sqlx::query("UPDATE ap_following SET remote_actor_url = $1 WHERE remote_actor_url = $2 AND local_user_id NOT IN (SELECT local_user_id FROM ap_following WHERE remote_actor_url = $1)")
.bind(new_actor_url).bind(old_actor_url).execute(&self.pool).await?;
candidates
.into_iter()
.map(|s| uuid::Uuid::parse_str(&s).map_err(|e| anyhow::anyhow!(e)))
.collect()
}
}

View File

@@ -0,0 +1,106 @@
use crate::{PG_ACTOR_COLS, PostgresFederationRepository, pg_remote_actor};
use adapter_common::datetime_to_str;
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{FollowingReader, FollowingStatus, FollowingWriter, RemoteActor, RemoteActorCache};
#[async_trait]
impl FollowingWriter for PostgresFederationRepository {
async fn add_following(
&self,
local_user_id: uuid::Uuid,
actor: RemoteActor,
follow_activity_id: &str,
) -> Result<()> {
let uid = local_user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
RemoteActorCache::upsert_remote_actor(self, actor.clone()).await?;
sqlx::query("INSERT INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at) VALUES ($1, $2, $3, $4::timestamptz) ON CONFLICT DO NOTHING")
.bind(&uid).bind(&actor.url).bind(follow_activity_id).bind(&created_at).execute(&self.pool).await?;
Ok(())
}
async fn get_follow_activity_id(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<String> = sqlx::query_scalar("SELECT follow_activity_id FROM ap_following WHERE local_user_id = $1 AND remote_actor_url = $2")
.bind(&uid).bind(remote_actor_url).fetch_optional(&self.pool).await?;
Ok(row)
}
async fn remove_following(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
sqlx::query("DELETE FROM ap_following WHERE local_user_id = $1 AND remote_actor_url = $2")
.bind(&uid)
.bind(actor_url)
.execute(&self.pool)
.await?;
Ok(())
}
async fn update_following_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowingStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = match status {
FollowingStatus::Pending => "pending",
FollowingStatus::Accepted => "accepted",
};
let result = sqlx::query("UPDATE ap_following SET status = $1 WHERE local_user_id = $2 AND remote_actor_url = $3")
.bind(status_str).bind(&uid).bind(remote_actor_url).execute(&self.pool).await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_following_status: no row found");
}
Ok(())
}
}
#[async_trait]
impl FollowingReader for PostgresFederationRepository {
async fn get_following(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let q = format!(
"SELECT a.url, {PG_ACTOR_COLS} FROM ap_following f INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.status = 'accepted'"
);
let rows = sqlx::query(&q).bind(&uid).fetch_all(&self.pool).await?;
Ok(rows.iter().map(|row| pg_remote_actor(row, "url")).collect())
}
async fn count_following(&self, local_user_id: uuid::Uuid) -> Result<usize> {
let uid = local_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?;
Ok(count as usize)
}
async fn get_following_page(
&self,
local_user_id: uuid::Uuid,
offset: u32,
limit: usize,
) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let q = format!(
"SELECT a.url, {PG_ACTOR_COLS} FROM ap_following f INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.status = 'accepted' ORDER BY f.created_at ASC LIMIT $2 OFFSET $3"
);
let rows = sqlx::query(&q)
.bind(&uid)
.bind(limit as i64)
.bind(offset as i64)
.fetch_all(&self.pool)
.await?;
Ok(rows.iter().map(|row| pg_remote_actor(row, "url")).collect())
}
}

View File

@@ -0,0 +1,27 @@
use anyhow::Result;
use async_trait::async_trait;
use k_ap::FollowMigration;
use crate::PostgresFederationRepository;
#[async_trait]
impl FollowMigration for PostgresFederationRepository {
async fn migrate_follower_actor(
&self,
old_actor_url: &str,
new_actor_url: &str,
) -> Result<Vec<uuid::Uuid>> {
let candidates: Vec<String> = sqlx::query_scalar(
"SELECT local_user_id FROM ap_following WHERE remote_actor_url = $1 AND local_user_id NOT IN (SELECT local_user_id FROM ap_following WHERE remote_actor_url = $2)",
).bind(old_actor_url).bind(new_actor_url).fetch_all(&self.pool).await?;
if candidates.is_empty() {
return Ok(vec![]);
}
sqlx::query("UPDATE ap_following SET remote_actor_url = $1 WHERE remote_actor_url = $2 AND local_user_id NOT IN (SELECT local_user_id FROM ap_following WHERE remote_actor_url = $1)")
.bind(new_actor_url).bind(old_actor_url).execute(&self.pool).await?;
candidates
.into_iter()
.map(|s| uuid::Uuid::parse_str(&s).map_err(|e| anyhow::anyhow!(e)))
.collect()
}
}

View File

@@ -0,0 +1,3 @@
mod followers;
mod following;
mod migration;

View File

@@ -0,0 +1,303 @@
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<String> = row.try_get("local_username").ok().flatten();
let display: Option<String> = row.try_get("local_display").ok().flatten();
let avatar: Option<String> = row
.try_get::<Option<String>, _>("local_avatar_path")
.ok()
.flatten()
.map(|p| format!("{}/images/{}", base_url, p));
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<String> = row.try_get("remote_display").ok().flatten();
let avatar: Option<String> = 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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<usize, DomainError> {
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<usize, DomainError> {
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<bool, DomainError> {
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)
}
}

View File

@@ -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 _,
}
}

View File

@@ -7,7 +7,7 @@ edition = "2024"
sqlx = { workspace = true }
activitypub = { workspace = true }
adapter-common = { workspace = true }
k-ap = { version = "0.4.6", registry = "gitea" }
k-ap = { version = "0.5.0", registry = "gitea" }
domain = { workspace = true }
anyhow = { workspace = true }
serde_json = { workspace = true }

View File

@@ -1,33 +1,28 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{ActorRepository, RemoteActor};
use k_ap::{AnnounceRepository, Keypair, KeypairRepository, RemoteActor, RemoteActorCache};
use sqlx::Row;
use super::{SqliteFederationRepository, remote_actor_from_row};
use adapter_common::datetime_to_str;
#[async_trait]
impl ActorRepository for SqliteFederationRepository {
async fn get_local_actor_keypair(
&self,
user_id: uuid::Uuid,
) -> Result<Option<(String, String)>> {
impl KeypairRepository for SqliteFederationRepository {
async fn get_local_actor_keypair(&self, user_id: uuid::Uuid) -> Result<Option<Keypair>> {
let uid = user_id.to_string();
let row =
sqlx::query("SELECT public_key, private_key FROM ap_local_actors WHERE user_id = ?")
.bind(&uid)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.get("public_key"), r.get("private_key"))))
Ok(row.map(|r| Keypair {
public_key: r.get("public_key"),
private_key: r.get("private_key"),
}))
}
async fn save_local_actor_keypair(
&self,
user_id: uuid::Uuid,
public_key: String,
private_key: String,
) -> Result<()> {
async fn save_local_actor_keypair(&self, user_id: uuid::Uuid, keypair: Keypair) -> Result<()> {
let uid = user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
@@ -39,14 +34,17 @@ impl ActorRepository for SqliteFederationRepository {
private_key = excluded.private_key",
)
.bind(&uid)
.bind(&public_key)
.bind(&private_key)
.bind(&keypair.public_key)
.bind(&keypair.private_key)
.bind(&created_at)
.execute(&self.pool)
.await?;
Ok(())
}
}
#[async_trait]
impl RemoteActorCache for SqliteFederationRepository {
async fn upsert_remote_actor(&self, actor: RemoteActor) -> Result<()> {
let now = Utc::now().naive_utc();
let fetched_at = datetime_to_str(&now);
@@ -84,7 +82,10 @@ impl ActorRepository for SqliteFederationRepository {
).bind(actor_url).fetch_optional(&self.pool).await?;
Ok(row.as_ref().map(|r| remote_actor_from_row(r, "url")))
}
}
#[async_trait]
impl AnnounceRepository for SqliteFederationRepository {
async fn add_announce(
&self,
activity_id: &str,

View File

@@ -1,14 +1,14 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{BlockedDomain, BlocklistRepository};
use k_ap::{ActorBlocklist, BlockedDomain, DomainBlocklist};
use sqlx::Row;
use super::SqliteFederationRepository;
use adapter_common::datetime_to_str;
#[async_trait]
impl BlocklistRepository for SqliteFederationRepository {
impl DomainBlocklist for SqliteFederationRepository {
async fn add_blocked_domain(&self, domain: &str, reason: Option<&str>) -> Result<()> {
let now = Utc::now().naive_utc();
let ts = datetime_to_str(&now);
@@ -56,7 +56,10 @@ impl BlocklistRepository for SqliteFederationRepository {
.await?;
Ok(count > 0)
}
}
#[async_trait]
impl ActorBlocklist for SqliteFederationRepository {
async fn add_blocked_actor(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
let ts = datetime_to_str(&Utc::now().naive_utc());

View File

@@ -1,16 +1,14 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{
ActorRepository, FollowRepository, Follower, FollowerStatus, FollowingStatus, RemoteActor,
};
use k_ap::{Follower, FollowerReader, FollowerStatus, FollowerWriter, RemoteActor};
use sqlx::Row;
use super::{SqliteFederationRepository, remote_actor_from_row, status_to_str, str_to_status};
use crate::{SqliteFederationRepository, remote_actor_from_row, status_to_str, str_to_status};
use adapter_common::datetime_to_str;
#[async_trait]
impl FollowRepository for SqliteFederationRepository {
impl FollowerWriter for SqliteFederationRepository {
async fn add_follower(
&self,
local_user_id: uuid::Uuid,
@@ -69,6 +67,31 @@ impl FollowRepository for SqliteFederationRepository {
Ok(())
}
async fn update_follower_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowerStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = status_to_str(&status);
let result = sqlx::query(
"UPDATE ap_followers SET status = ? WHERE local_user_id = ? AND remote_actor_url = ?",
)
.bind(status_str)
.bind(&uid)
.bind(remote_actor_url)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_follower_status: no row found");
}
Ok(())
}
}
#[async_trait]
impl FollowerReader for SqliteFederationRepository {
async fn get_followers(&self, local_user_id: uuid::Uuid) -> Result<Vec<Follower>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
@@ -138,28 +161,6 @@ impl FollowRepository for SqliteFederationRepository {
Ok(count as usize)
}
async fn update_follower_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowerStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = status_to_str(&status);
let result = sqlx::query(
"UPDATE ap_followers SET status = ? WHERE local_user_id = ? AND remote_actor_url = ?",
)
.bind(status_str)
.bind(&uid)
.bind(remote_actor_url)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_follower_status: no row found");
}
Ok(())
}
async fn get_pending_followers(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
@@ -232,173 +233,4 @@ impl FollowRepository for SqliteFederationRepository {
.map(|row| remote_actor_from_row(row, "remote_actor_url"))
.collect())
}
async fn add_following(
&self,
local_user_id: uuid::Uuid,
actor: RemoteActor,
follow_activity_id: &str,
) -> Result<()> {
let uid = local_user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
ActorRepository::upsert_remote_actor(self, actor.clone()).await?;
sqlx::query(
"INSERT OR IGNORE INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at)
VALUES (?, ?, ?, ?)",
).bind(&uid).bind(&actor.url).bind(follow_activity_id).bind(&created_at).execute(&self.pool).await?;
Ok(())
}
async fn get_follow_activity_id(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<Option<String>> = sqlx::query_scalar(
"SELECT follow_activity_id FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ?",
).bind(&uid).bind(remote_actor_url).fetch_optional(&self.pool).await?;
Ok(row.flatten())
}
async fn remove_following(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
sqlx::query("DELETE FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ?")
.bind(&uid)
.bind(actor_url)
.execute(&self.pool)
.await?;
Ok(())
}
async fn get_following(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
"SELECT a.url, a.handle, a.inbox_url, a.shared_inbox_url, a.display_name, a.avatar_url,
a.outbox_url, a.bio, a.banner_url, a.followers_url, a.following_url, a.also_known_as, a.fetched_at
FROM ap_following f
INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'accepted'",
).bind(&uid).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|row| remote_actor_from_row(row, "url"))
.collect())
}
async fn count_following(&self, local_user_id: uuid::Uuid) -> Result<usize> {
let uid = local_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?;
Ok(count as usize)
}
async fn get_following_page(
&self,
local_user_id: uuid::Uuid,
offset: u32,
limit: usize,
) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
"SELECT a.url, a.handle, a.inbox_url, a.shared_inbox_url, a.display_name, a.avatar_url,
a.outbox_url, a.bio, a.banner_url, a.followers_url, a.following_url, a.also_known_as, a.fetched_at
FROM ap_following f
INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'accepted'
ORDER BY f.created_at ASC LIMIT ? OFFSET ?",
).bind(&uid).bind(limit as i64).bind(offset as i64).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|row| remote_actor_from_row(row, "url"))
.collect())
}
async fn update_following_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowingStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = match status {
FollowingStatus::Pending => "pending",
FollowingStatus::Accepted => "accepted",
};
let result = sqlx::query(
"UPDATE ap_following SET status = ? WHERE local_user_id = ? AND remote_actor_url = ?",
)
.bind(status_str)
.bind(&uid)
.bind(remote_actor_url)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_following_status: no row found");
}
Ok(())
}
async fn get_following_outbox_url(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<Option<String>> = sqlx::query_scalar(
"SELECT a.outbox_url
FROM ap_following f
INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.remote_actor_url = ?",
)
.bind(&uid)
.bind(remote_actor_url)
.fetch_optional(&self.pool)
.await?;
Ok(row.flatten())
}
async fn migrate_follower_actor(
&self,
old_actor_url: &str,
new_actor_url: &str,
) -> Result<Vec<uuid::Uuid>> {
let candidates: Vec<String> = sqlx::query_scalar(
"SELECT local_user_id FROM ap_following
WHERE remote_actor_url = ?1
AND local_user_id NOT IN (
SELECT local_user_id FROM ap_following WHERE remote_actor_url = ?2
)",
)
.bind(old_actor_url)
.bind(new_actor_url)
.fetch_all(&self.pool)
.await?;
if candidates.is_empty() {
return Ok(vec![]);
}
sqlx::query(
"UPDATE ap_following SET remote_actor_url = ?1
WHERE remote_actor_url = ?2
AND local_user_id NOT IN (
SELECT local_user_id FROM ap_following WHERE remote_actor_url = ?1
)",
)
.bind(new_actor_url)
.bind(old_actor_url)
.execute(&self.pool)
.await?;
candidates
.into_iter()
.map(|s| uuid::Uuid::parse_str(&s).map_err(|e| anyhow::anyhow!(e)))
.collect()
}
}

View File

@@ -0,0 +1,124 @@
use anyhow::Result;
use async_trait::async_trait;
use chrono::Utc;
use k_ap::{FollowingReader, FollowingStatus, FollowingWriter, RemoteActor, RemoteActorCache};
use crate::{SqliteFederationRepository, remote_actor_from_row};
use adapter_common::datetime_to_str;
#[async_trait]
impl FollowingWriter for SqliteFederationRepository {
async fn add_following(
&self,
local_user_id: uuid::Uuid,
actor: RemoteActor,
follow_activity_id: &str,
) -> Result<()> {
let uid = local_user_id.to_string();
let now = Utc::now().naive_utc();
let created_at = datetime_to_str(&now);
RemoteActorCache::upsert_remote_actor(self, actor.clone()).await?;
sqlx::query(
"INSERT OR IGNORE INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, created_at)
VALUES (?, ?, ?, ?)",
).bind(&uid).bind(&actor.url).bind(follow_activity_id).bind(&created_at).execute(&self.pool).await?;
Ok(())
}
async fn get_follow_activity_id(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
) -> Result<Option<String>> {
let uid = local_user_id.to_string();
let row: Option<Option<String>> = sqlx::query_scalar(
"SELECT follow_activity_id FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ?",
).bind(&uid).bind(remote_actor_url).fetch_optional(&self.pool).await?;
Ok(row.flatten())
}
async fn remove_following(&self, local_user_id: uuid::Uuid, actor_url: &str) -> Result<()> {
let uid = local_user_id.to_string();
sqlx::query("DELETE FROM ap_following WHERE local_user_id = ? AND remote_actor_url = ?")
.bind(&uid)
.bind(actor_url)
.execute(&self.pool)
.await?;
Ok(())
}
async fn update_following_status(
&self,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowingStatus,
) -> Result<()> {
let uid = local_user_id.to_string();
let status_str = match status {
FollowingStatus::Pending => "pending",
FollowingStatus::Accepted => "accepted",
};
let result = sqlx::query(
"UPDATE ap_following SET status = ? WHERE local_user_id = ? AND remote_actor_url = ?",
)
.bind(status_str)
.bind(&uid)
.bind(remote_actor_url)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
tracing::warn!(local_user_id = %local_user_id, remote_actor_url, "update_following_status: no row found");
}
Ok(())
}
}
#[async_trait]
impl FollowingReader for SqliteFederationRepository {
async fn get_following(&self, local_user_id: uuid::Uuid) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
"SELECT a.url, a.handle, a.inbox_url, a.shared_inbox_url, a.display_name, a.avatar_url,
a.outbox_url, a.bio, a.banner_url, a.followers_url, a.following_url, a.also_known_as, a.fetched_at
FROM ap_following f
INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'accepted'",
).bind(&uid).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|row| remote_actor_from_row(row, "url"))
.collect())
}
async fn count_following(&self, local_user_id: uuid::Uuid) -> Result<usize> {
let uid = local_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?;
Ok(count as usize)
}
async fn get_following_page(
&self,
local_user_id: uuid::Uuid,
offset: u32,
limit: usize,
) -> Result<Vec<RemoteActor>> {
let uid = local_user_id.to_string();
let rows = sqlx::query(
"SELECT a.url, a.handle, a.inbox_url, a.shared_inbox_url, a.display_name, a.avatar_url,
a.outbox_url, a.bio, a.banner_url, a.followers_url, a.following_url, a.also_known_as, a.fetched_at
FROM ap_following f
INNER JOIN ap_remote_actors a ON a.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'accepted'
ORDER BY f.created_at ASC LIMIT ? OFFSET ?",
).bind(&uid).bind(limit as i64).bind(offset as i64).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|row| remote_actor_from_row(row, "url"))
.collect())
}
}

View File

@@ -0,0 +1,47 @@
use anyhow::Result;
use async_trait::async_trait;
use k_ap::FollowMigration;
use crate::SqliteFederationRepository;
#[async_trait]
impl FollowMigration for SqliteFederationRepository {
async fn migrate_follower_actor(
&self,
old_actor_url: &str,
new_actor_url: &str,
) -> Result<Vec<uuid::Uuid>> {
let candidates: Vec<String> = sqlx::query_scalar(
"SELECT local_user_id FROM ap_following
WHERE remote_actor_url = ?1
AND local_user_id NOT IN (
SELECT local_user_id FROM ap_following WHERE remote_actor_url = ?2
)",
)
.bind(old_actor_url)
.bind(new_actor_url)
.fetch_all(&self.pool)
.await?;
if candidates.is_empty() {
return Ok(vec![]);
}
sqlx::query(
"UPDATE ap_following SET remote_actor_url = ?1
WHERE remote_actor_url = ?2
AND local_user_id NOT IN (
SELECT local_user_id FROM ap_following WHERE remote_actor_url = ?1
)",
)
.bind(new_actor_url)
.bind(old_actor_url)
.execute(&self.pool)
.await?;
candidates
.into_iter()
.map(|s| uuid::Uuid::parse_str(&s).map_err(|e| anyhow::anyhow!(e)))
.collect()
}
}

View File

@@ -0,0 +1,3 @@
mod followers;
mod following;
mod migration;

View File

@@ -0,0 +1,303 @@
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<String> = row.try_get("local_username").ok().flatten();
let display: Option<String> = row.try_get("local_display").ok().flatten();
let avatar: Option<String> = row
.try_get::<Option<String>, _>("local_avatar_path")
.ok()
.flatten()
.map(|p| format!("{}/images/{}", base_url, p));
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<String> = row.try_get("remote_display").ok().flatten();
let avatar: Option<String> = 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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<Vec<SocialActor>, 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_path AS local_avatar_path,
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<usize, DomainError> {
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<usize, DomainError> {
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<bool, DomainError> {
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)
}
}

View File

@@ -3,6 +3,7 @@ mod actor;
mod blocklist;
mod federated_profile;
mod follow;
mod follow_repository;
mod review;
mod social;
mod watchlist;
@@ -91,21 +92,19 @@ 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)]
#[path = "tests/outbox_url.rs"]
mod outbox_url_tests;
#[cfg(test)]
#[path = "tests/actor_block_tests.rs"]
mod actor_block_tests;

View File

@@ -1,5 +1,5 @@
use super::*;
use k_ap::BlocklistRepository;
use k_ap::ActorBlocklist;
use sqlx::SqlitePool;
async fn test_pool() -> SqlitePool {

View File

@@ -1,5 +1,5 @@
use super::*;
use k_ap::BlocklistRepository;
use k_ap::DomainBlocklist;
use sqlx::SqlitePool;
async fn test_pool() -> SqlitePool {

View File

@@ -1,7 +1,7 @@
use super::*;
use chrono::Utc;
use domain::ports::FederationAdminQuery;
use k_ap::ActorRepository;
use k_ap::AnnounceRepository;
use sqlx::SqlitePool;
async fn test_pool() -> SqlitePool {

View File

@@ -1,76 +0,0 @@
use super::*;
use k_ap::{FollowRepository, FollowingStatus, RemoteActor};
async fn setup_pool() -> SqlitePool {
let pool = SqlitePool::connect(":memory:").await.unwrap();
sqlx::query(
"CREATE TABLE ap_remote_actors (
url TEXT PRIMARY KEY, handle TEXT NOT NULL, inbox_url TEXT NOT NULL,
shared_inbox_url TEXT, display_name TEXT, avatar_url TEXT,
outbox_url TEXT, bio TEXT, banner_url TEXT, followers_url TEXT,
following_url TEXT, also_known_as TEXT, fetched_at TEXT NOT NULL
);
CREATE TABLE ap_following (
local_user_id TEXT NOT NULL, remote_actor_url TEXT NOT NULL,
follow_activity_id TEXT, created_at TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
PRIMARY KEY (local_user_id, remote_actor_url)
);",
)
.execute(&pool)
.await
.unwrap();
pool
}
#[tokio::test]
async fn get_following_outbox_url_returns_stored_url() {
let pool = setup_pool().await;
let repo = SqliteFederationRepository::new(pool);
let local_user = uuid::Uuid::new_v4();
let actor = RemoteActor {
url: "https://remote.example/users/alice".to_string(),
handle: "alice@remote.example".to_string(),
inbox_url: "https://remote.example/users/alice/inbox".to_string(),
shared_inbox_url: None,
display_name: None,
avatar_url: None,
outbox_url: Some("https://remote.example/users/alice/outbox".to_string()),
bio: None,
banner_url: None,
followers_url: None,
following_url: None,
also_known_as: vec![],
fetched_at: None,
};
repo.add_following(local_user, actor, "https://local/activities/1")
.await
.unwrap();
repo.update_following_status(
local_user,
"https://remote.example/users/alice",
FollowingStatus::Accepted,
)
.await
.unwrap();
let result = repo
.get_following_outbox_url(local_user, "https://remote.example/users/alice")
.await
.unwrap();
assert_eq!(
result,
Some("https://remote.example/users/alice/outbox".to_string())
);
}
#[tokio::test]
async fn get_following_outbox_url_returns_none_when_not_following() {
let pool = setup_pool().await;
let repo = SqliteFederationRepository::new(pool);
let result = repo
.get_following_outbox_url(uuid::Uuid::new_v4(), "https://remote.example/users/alice")
.await
.unwrap();
assert_eq!(result, None);
}

View File

@@ -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<Vec<SocialActor>, DomainError>;
async fn get_followers(
&self,
user_id: uuid::Uuid,
base_url: &str,
) -> Result<Vec<SocialActor>, DomainError>;
async fn get_pending_followers(
&self,
user_id: uuid::Uuid,
base_url: &str,
) -> Result<Vec<SocialActor>, DomainError>;
async fn count_following(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
async fn count_followers(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
async fn is_following(
&self,
follower_id: uuid::Uuid,
target_actor_url: &str,
) -> Result<bool, DomainError>;
}

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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<dyn domain::ports::SocialCommand>,
composite_social as Arc<dyn domain::ports::SocialQuery>,
)

View File

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

View File

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

View File

@@ -1,4 +1,5 @@
import { useEffect, useRef, useState } from "react"
import { createPortal } from "react-dom"
import { useTranslation } from "react-i18next"
import { Download, Share2, X } from "lucide-react"
import html2canvas from "html2canvas-pro"
@@ -59,7 +60,7 @@ export function WrapUpShareCard({ report, onClose }: Props) {
}
}
return (
return createPortal(
<div className="fixed inset-0 z-50 flex flex-col items-center justify-center bg-black/80 p-4">
<div className="mb-4 flex w-full max-w-sm items-center justify-between">
<Button variant="ghost" size="icon" onClick={onClose} className="text-white">
@@ -138,7 +139,8 @@ export function WrapUpShareCard({ report, onClose }: Props) {
</div>
</div>
</div>
</div>
</div>,
document.body,
)
}