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

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
Details

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-07-27 11:46:24 +02:00
parent fdf286ca94
commit 839cababe4
28 changed files with 475 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
crates/adapters/postgres-federation/src/actor.rs
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,

7
crates/adapters/postgres-federation/src/blocklist.rs
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());

168
crates/adapters/postgres-federation/src/follow.rs → crates/adapters/postgres-federation/src/follow/followers.rs
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()
}
}

106
crates/adapters/postgres-federation/src/follow/following.rs Normal file
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())
}
}

27
crates/adapters/postgres-federation/src/follow/migration.rs Normal file
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()
}
}

3
crates/adapters/postgres-federation/src/follow/mod.rs Normal file
View File

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