From 839cababe413553b922e0ef52b4776e4b422b625 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 27 Jul 2026 11:46:24 +0200 Subject: [PATCH] bump k-ap 0.5.0, decompose repo traits, LocalObject, ActivityStreamsType enum, split follow modules --- Cargo.lock | 8 +- crates/adapters/activitypub/Cargo.toml | 2 +- .../activitypub/src/composite_handler.rs | 15 +- .../adapters/activitypub/src/event_handler.rs | 14 +- .../adapters/activitypub/src/goal_handler.rs | 15 +- crates/adapters/activitypub/src/lib.rs | 2 +- crates/adapters/activitypub/src/objects.rs | 20 +- .../activitypub/src/review_handler.rs | 15 +- .../activitypub/src/watchlist_handler.rs | 15 +- .../adapters/postgres-federation/Cargo.toml | 2 +- .../adapters/postgres-federation/src/actor.rs | 29 +-- .../postgres-federation/src/blocklist.rs | 7 +- .../src/{follow.rs => follow/followers.rs} | 168 ++----------- .../src/follow/following.rs | 106 +++++++++ .../src/follow/migration.rs | 27 +++ .../postgres-federation/src/follow/mod.rs | 3 + crates/adapters/sqlite-federation/Cargo.toml | 2 +- .../adapters/sqlite-federation/src/actor.rs | 31 +-- .../sqlite-federation/src/blocklist.rs | 7 +- .../src/{follow.rs => follow/followers.rs} | 224 +++--------------- .../sqlite-federation/src/follow/following.rs | 124 ++++++++++ .../sqlite-federation/src/follow/migration.rs | 47 ++++ .../sqlite-federation/src/follow/mod.rs | 3 + crates/adapters/sqlite-federation/src/lib.rs | 4 - .../src/tests/actor_block_tests.rs | 2 +- .../src/tests/domain_block_tests.rs | 2 +- .../sqlite-federation/src/tests/lib.rs | 2 +- .../sqlite-federation/src/tests/outbox_url.rs | 76 ------ 28 files changed, 475 insertions(+), 497 deletions(-) rename crates/adapters/postgres-federation/src/{follow.rs => follow/followers.rs} (56%) create mode 100644 crates/adapters/postgres-federation/src/follow/following.rs create mode 100644 crates/adapters/postgres-federation/src/follow/migration.rs create mode 100644 crates/adapters/postgres-federation/src/follow/mod.rs rename crates/adapters/sqlite-federation/src/{follow.rs => follow/followers.rs} (56%) create mode 100644 crates/adapters/sqlite-federation/src/follow/following.rs create mode 100644 crates/adapters/sqlite-federation/src/follow/migration.rs create mode 100644 crates/adapters/sqlite-federation/src/follow/mod.rs delete mode 100644 crates/adapters/sqlite-federation/src/tests/outbox_url.rs diff --git a/Cargo.lock b/Cargo.lock index 4998029..e630cc8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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]] diff --git a/crates/adapters/activitypub/Cargo.toml b/crates/adapters/activitypub/Cargo.toml index e7d0186..4fad9e3 100644 --- a/crates/adapters/activitypub/Cargo.toml +++ b/crates/adapters/activitypub/Cargo.toml @@ -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 } diff --git a/crates/adapters/activitypub/src/composite_handler.rs b/crates/adapters/activitypub/src/composite_handler.rs index 22a9333..3abf68a 100644 --- a/crates/adapters/activitypub/src/composite_handler.rs +++ b/crates/adapters/activitypub/src/composite_handler.rs @@ -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>, + before: Option>, limit: usize, - ) -> anyhow::Result)>> { - // Fetch from all three sources (watchlist/goals return all, reviews use DB pagination) + ) -> anyhow::Result> { 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)> = Vec::new(); + let mut all: Vec = 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) } diff --git a/crates/adapters/activitypub/src/event_handler.rs b/crates/adapters/activitypub/src/event_handler.rs index 6864055..6575e57 100644 --- a/crates/adapters/activitypub/src/event_handler.rs +++ b/crates/adapters/activitypub/src/event_handler.rs @@ -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(()) diff --git a/crates/adapters/activitypub/src/goal_handler.rs b/crates/adapters/activitypub/src/goal_handler.rs index e6dd1cf..db826a3 100644 --- a/crates/adapters/activitypub/src/goal_handler.rs +++ b/crates/adapters/activitypub/src/goal_handler.rs @@ -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>, _limit: usize, - ) -> anyhow::Result)>> { + ) -> anyhow::Result> { 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) } diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index cadbe68..fe3ed8f 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -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; diff --git a/crates/adapters/activitypub/src/objects.rs b/crates/adapters/activitypub/src/objects.rs index dbac602..a36237b 100644 --- a/crates/adapters/activitypub/src/objects.rs +++ b/crates/adapters/activitypub/src/objects.rs @@ -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, diff --git a/crates/adapters/activitypub/src/review_handler.rs b/crates/adapters/activitypub/src/review_handler.rs index b1e7d45..d507be4 100644 --- a/crates/adapters/activitypub/src/review_handler.rs +++ b/crates/adapters/activitypub/src/review_handler.rs @@ -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>, limit: usize, - ) -> anyhow::Result)>> { + ) -> anyhow::Result> { 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) } diff --git a/crates/adapters/activitypub/src/watchlist_handler.rs b/crates/adapters/activitypub/src/watchlist_handler.rs index 538fa4e..6e0f721 100644 --- a/crates/adapters/activitypub/src/watchlist_handler.rs +++ b/crates/adapters/activitypub/src/watchlist_handler.rs @@ -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>, _limit: usize, - ) -> anyhow::Result)>> { + ) -> anyhow::Result> { 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) } diff --git a/crates/adapters/postgres-federation/Cargo.toml b/crates/adapters/postgres-federation/Cargo.toml index 932c3a1..7c5758f 100644 --- a/crates/adapters/postgres-federation/Cargo.toml +++ b/crates/adapters/postgres-federation/Cargo.toml @@ -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 } diff --git a/crates/adapters/postgres-federation/src/actor.rs b/crates/adapters/postgres-federation/src/actor.rs index 03a9ec8..41a0e28 100644 --- a/crates/adapters/postgres-federation/src/actor.rs +++ b/crates/adapters/postgres-federation/src/actor.rs @@ -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> { +impl KeypairRepository for PostgresFederationRepository { + async fn get_local_actor_keypair(&self, user_id: uuid::Uuid) -> Result> { 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, diff --git a/crates/adapters/postgres-federation/src/blocklist.rs b/crates/adapters/postgres-federation/src/blocklist.rs index 4e9ef70..0bb22a3 100644 --- a/crates/adapters/postgres-federation/src/blocklist.rs +++ b/crates/adapters/postgres-federation/src/blocklist.rs @@ -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()); diff --git a/crates/adapters/postgres-federation/src/follow.rs b/crates/adapters/postgres-federation/src/follow/followers.rs similarity index 56% rename from crates/adapters/postgres-federation/src/follow.rs rename to crates/adapters/postgres-federation/src/follow/followers.rs index 0a65439..0a1e654 100644 --- a/crates/adapters/postgres-federation/src/follow.rs +++ b/crates/adapters/postgres-federation/src/follow/followers.rs @@ -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> { 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> { 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> { - let uid = local_user_id.to_string(); - let row: Option = 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> { - 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 { - 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> { - 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> { - let uid = local_user_id.to_string(); - let row: Option> = 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> { - let candidates: Vec = 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() - } } diff --git a/crates/adapters/postgres-federation/src/follow/following.rs b/crates/adapters/postgres-federation/src/follow/following.rs new file mode 100644 index 0000000..795487b --- /dev/null +++ b/crates/adapters/postgres-federation/src/follow/following.rs @@ -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> { + let uid = local_user_id.to_string(); + let row: Option = 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> { + 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 { + 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> { + 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()) + } +} diff --git a/crates/adapters/postgres-federation/src/follow/migration.rs b/crates/adapters/postgres-federation/src/follow/migration.rs new file mode 100644 index 0000000..7c990e0 --- /dev/null +++ b/crates/adapters/postgres-federation/src/follow/migration.rs @@ -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> { + let candidates: Vec = 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() + } +} diff --git a/crates/adapters/postgres-federation/src/follow/mod.rs b/crates/adapters/postgres-federation/src/follow/mod.rs new file mode 100644 index 0000000..414bc5c --- /dev/null +++ b/crates/adapters/postgres-federation/src/follow/mod.rs @@ -0,0 +1,3 @@ +mod followers; +mod following; +mod migration; diff --git a/crates/adapters/sqlite-federation/Cargo.toml b/crates/adapters/sqlite-federation/Cargo.toml index ff6d398..13911ab 100644 --- a/crates/adapters/sqlite-federation/Cargo.toml +++ b/crates/adapters/sqlite-federation/Cargo.toml @@ -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 } diff --git a/crates/adapters/sqlite-federation/src/actor.rs b/crates/adapters/sqlite-federation/src/actor.rs index 9e6dd5c..7ed3961 100644 --- a/crates/adapters/sqlite-federation/src/actor.rs +++ b/crates/adapters/sqlite-federation/src/actor.rs @@ -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> { +impl KeypairRepository for SqliteFederationRepository { + async fn get_local_actor_keypair(&self, user_id: uuid::Uuid) -> Result> { 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, diff --git a/crates/adapters/sqlite-federation/src/blocklist.rs b/crates/adapters/sqlite-federation/src/blocklist.rs index 86b9624..2785c9f 100644 --- a/crates/adapters/sqlite-federation/src/blocklist.rs +++ b/crates/adapters/sqlite-federation/src/blocklist.rs @@ -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()); diff --git a/crates/adapters/sqlite-federation/src/follow.rs b/crates/adapters/sqlite-federation/src/follow/followers.rs similarity index 56% rename from crates/adapters/sqlite-federation/src/follow.rs rename to crates/adapters/sqlite-federation/src/follow/followers.rs index 8389e1d..598ff9d 100644 --- a/crates/adapters/sqlite-federation/src/follow.rs +++ b/crates/adapters/sqlite-federation/src/follow/followers.rs @@ -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> { 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> { 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> { - let uid = local_user_id.to_string(); - let row: Option> = 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> { - 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 { - 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> { - 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> { - let uid = local_user_id.to_string(); - let row: Option> = 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> { - let candidates: Vec = 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() - } } diff --git a/crates/adapters/sqlite-federation/src/follow/following.rs b/crates/adapters/sqlite-federation/src/follow/following.rs new file mode 100644 index 0000000..2b69e34 --- /dev/null +++ b/crates/adapters/sqlite-federation/src/follow/following.rs @@ -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> { + let uid = local_user_id.to_string(); + let row: Option> = 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> { + 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 { + 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> { + 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()) + } +} diff --git a/crates/adapters/sqlite-federation/src/follow/migration.rs b/crates/adapters/sqlite-federation/src/follow/migration.rs new file mode 100644 index 0000000..4140f27 --- /dev/null +++ b/crates/adapters/sqlite-federation/src/follow/migration.rs @@ -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> { + let candidates: Vec = 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() + } +} diff --git a/crates/adapters/sqlite-federation/src/follow/mod.rs b/crates/adapters/sqlite-federation/src/follow/mod.rs new file mode 100644 index 0000000..414bc5c --- /dev/null +++ b/crates/adapters/sqlite-federation/src/follow/mod.rs @@ -0,0 +1,3 @@ +mod followers; +mod following; +mod migration; diff --git a/crates/adapters/sqlite-federation/src/lib.rs b/crates/adapters/sqlite-federation/src/lib.rs index 5514b3c..c6fd0e4 100644 --- a/crates/adapters/sqlite-federation/src/lib.rs +++ b/crates/adapters/sqlite-federation/src/lib.rs @@ -102,10 +102,6 @@ pub fn wire(pool: SqlitePool) -> activitypub::FederationRepos { ) } -#[cfg(test)] -#[path = "tests/outbox_url.rs"] -mod outbox_url_tests; - #[cfg(test)] #[path = "tests/actor_block_tests.rs"] mod actor_block_tests; diff --git a/crates/adapters/sqlite-federation/src/tests/actor_block_tests.rs b/crates/adapters/sqlite-federation/src/tests/actor_block_tests.rs index 1dd39f4..c3624ba 100644 --- a/crates/adapters/sqlite-federation/src/tests/actor_block_tests.rs +++ b/crates/adapters/sqlite-federation/src/tests/actor_block_tests.rs @@ -1,5 +1,5 @@ use super::*; -use k_ap::BlocklistRepository; +use k_ap::ActorBlocklist; use sqlx::SqlitePool; async fn test_pool() -> SqlitePool { diff --git a/crates/adapters/sqlite-federation/src/tests/domain_block_tests.rs b/crates/adapters/sqlite-federation/src/tests/domain_block_tests.rs index f592852..cfb7df8 100644 --- a/crates/adapters/sqlite-federation/src/tests/domain_block_tests.rs +++ b/crates/adapters/sqlite-federation/src/tests/domain_block_tests.rs @@ -1,5 +1,5 @@ use super::*; -use k_ap::BlocklistRepository; +use k_ap::DomainBlocklist; use sqlx::SqlitePool; async fn test_pool() -> SqlitePool { diff --git a/crates/adapters/sqlite-federation/src/tests/lib.rs b/crates/adapters/sqlite-federation/src/tests/lib.rs index bc0206b..4b9ad81 100644 --- a/crates/adapters/sqlite-federation/src/tests/lib.rs +++ b/crates/adapters/sqlite-federation/src/tests/lib.rs @@ -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 { diff --git a/crates/adapters/sqlite-federation/src/tests/outbox_url.rs b/crates/adapters/sqlite-federation/src/tests/outbox_url.rs deleted file mode 100644 index 5d6ee37..0000000 --- a/crates/adapters/sqlite-federation/src/tests/outbox_url.rs +++ /dev/null @@ -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); -}