fix(federation): fix 27 AP bugs, gaps, and inconsistencies
Some checks failed
lint / lint (push) Failing after 9m26s
test / unit (push) Successful in 16m3s

Round 1 — 18 bug fixes:
- remote likes/boosts now persist in engagement tables
- intern_remote_actor uses name@domain, expanded username to VARCHAR(255)
- PgRemoteActorRepository upsert/find now handles all fields
- update_following_status no longer a no-op, count_followers counts all
- accept/reject follow publishes event before DB mark (atomicity)
- fetch_outbox_page follows pagination via next links
- actor URL canonicalized to /users/{uuid}
- content_to_html escapes single quotes
- WebFinger accepts application/ld+json type
- try_from_ap accepts Article and Page object types
- feed SQL uses parameterized viewer UUID instead of format!
- content cap raised from 500 to 5000 chars
- also_known_as changed from Option<String> to Vec<String>
- connections fetch always triggers from page 1

Round 2 — 9 gap fixes:
- on_announce_removed handler deletes boost row on Undo(Announce)
- on_update handles Person/Service/Group actor profile updates
- sync_remote_actor_to_user syncs remote_actors → users on create/update
- FederationBlockPort: block_by_username sends Block activity to remote
- domain RemoteActor gains inbox_url, shared_inbox_url fields
- remote_actors attachment column (JSONB) with read/write
- .well-known/host-meta endpoint
- 256KB body limit on AP inbox routes
- outbox cleanup job (7-day retention, hourly sweep)
This commit is contained in:
2026-05-29 11:28:40 +02:00
parent f9de21dcfa
commit 84edf58de6
32 changed files with 565 additions and 142 deletions

View File

@@ -10,7 +10,7 @@ use url::Url;
use crate::note::{ThoughtNote, ThoughtNoteInput};
use crate::port::{AcceptNoteInput, ActivityPubRepository};
use crate::urls::ThoughtsUrls;
use domain::ports::{EventPublisher, TagRepository};
use domain::ports::{BoostRepository, EventPublisher, LikeRepository, TagRepository};
use domain::value_objects::UserId;
use k_ap::{ApContentReader, ApObjectHandler};
@@ -19,6 +19,8 @@ pub struct ThoughtsObjectHandler {
urls: ThoughtsUrls,
event_publisher: Option<Arc<dyn EventPublisher>>,
tag_repo: Arc<dyn TagRepository>,
likes: Arc<dyn LikeRepository>,
boosts: Arc<dyn BoostRepository>,
}
impl ThoughtsObjectHandler {
@@ -27,12 +29,16 @@ impl ThoughtsObjectHandler {
base_url: &str,
event_publisher: Option<Arc<dyn EventPublisher>>,
tag_repo: Arc<dyn TagRepository>,
likes: Arc<dyn LikeRepository>,
boosts: Arc<dyn BoostRepository>,
) -> Self {
Self {
repo,
urls: ThoughtsUrls::new(base_url),
event_publisher,
tag_repo,
likes,
boosts,
}
}
}
@@ -106,6 +112,10 @@ impl ApObjectHandler for ThoughtsObjectHandler {
.intern_remote_actor(actor_url.as_str())
.await
.map_err(|e| anyhow!("{e}"))?;
let _ = self
.repo
.sync_remote_actor_to_user(actor_url.as_str())
.await;
let as_public = "https://www.w3.org/ns/activitystreams#Public";
let in_to = note.to.iter().any(|s| s == as_public);
@@ -194,17 +204,50 @@ impl ApObjectHandler for ThoughtsObjectHandler {
async fn on_update(
&self,
ap_id: &Url,
_actor_url: &Url,
actor_url: &Url,
object: serde_json::Value,
) -> Result<()> {
let Some((note, _)) = ThoughtNote::try_from_ap(object) else {
tracing::debug!(ap_id = %ap_id, "on_update: skipping non-Note object");
return Ok(());
};
self.repo
.apply_note_update(ap_id.as_str(), &note.content)
.await
.map_err(|e| anyhow!("{e}"))
let obj_type = object.get("type").and_then(|v| v.as_str()).unwrap_or("");
match obj_type {
"Note" | "Article" | "Page" => {
let Some((note, _)) = ThoughtNote::try_from_ap(object) else {
return Ok(());
};
self.repo
.apply_note_update(ap_id.as_str(), &note.content)
.await
.map_err(|e| anyhow!("{e}"))
}
"Person" | "Service" | "Application" | "Group" | "Organization" => {
let display_name = object.get("name").and_then(|v| v.as_str());
let avatar_url = object
.get("icon")
.and_then(|v| v.get("url"))
.and_then(|v| v.as_str());
self.repo
.update_remote_actor_display(
&self
.repo
.find_remote_actor_id(actor_url.as_str())
.await
.map_err(|e| anyhow!("{e}"))?
.ok_or_else(|| anyhow!("unknown actor"))?,
display_name,
avatar_url,
)
.await
.map_err(|e| anyhow!("{e}"))?;
let _ = self
.repo
.sync_remote_actor_to_user(actor_url.as_str())
.await;
Ok(())
}
_ => {
tracing::debug!(ap_id = %ap_id, obj_type, "on_update: skipping");
Ok(())
}
}
}
async fn on_delete(&self, ap_id: &Url, _actor_url: &Url) -> Result<()> {
@@ -245,14 +288,24 @@ impl ApObjectHandler for ThoughtsObjectHandler {
let actor_user_id = match actor_user_id {
Some(id) => id,
None => {
tracing::debug!(actor = %actor_url, "on_like: remote actor not interned, skipping notification");
tracing::debug!(actor = %actor_url, "on_like: remote actor not interned, skipping");
return Ok(());
}
};
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let like_id = domain::value_objects::LikeId::new();
let like = domain::models::social::Like {
id: like_id.clone(),
user_id: actor_user_id.clone(),
thought_id: thought_id.clone(),
ap_id: Some(object_url.to_string()),
created_at: Utc::now(),
};
let _ = self.likes.save(&like).await;
if let Some(ep) = &self.event_publisher {
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let like_id = domain::value_objects::LikeId::new();
ep.publish(&domain::events::DomainEvent::LikeAdded {
like_id,
user_id: actor_user_id,
@@ -294,10 +347,13 @@ impl ApObjectHandler for ThoughtsObjectHandler {
}
};
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let _ = self.likes.delete(&actor_user_id, &thought_id).await;
if let Some(ep) = &self.event_publisher {
ep.publish(&domain::events::DomainEvent::LikeRemoved {
user_id: actor_user_id,
thought_id: domain::value_objects::ThoughtId::from_uuid(thought_uuid),
thought_id,
})
.await
.map_err(|e| anyhow!("{e}"))?;
@@ -369,9 +425,19 @@ impl ApObjectHandler for ThoughtsObjectHandler {
None => return Ok(()),
};
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let boost_id = domain::value_objects::BoostId::new();
let boost = domain::models::social::Boost {
id: boost_id.clone(),
user_id: actor_user_id.clone(),
thought_id: thought_id.clone(),
ap_id: Some(object_url.to_string()),
created_at: Utc::now(),
};
let _ = self.boosts.save(&boost).await;
if let Some(ep) = &self.event_publisher {
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let boost_id = domain::value_objects::BoostId::new();
ep.publish(&domain::events::DomainEvent::BoostAdded {
boost_id,
user_id: actor_user_id,
@@ -384,6 +450,44 @@ impl ApObjectHandler for ThoughtsObjectHandler {
Ok(())
}
async fn on_announce_removed(&self, object_url: &Url, actor_url: &Url) -> Result<()> {
let thought_uuid = object_url
.path()
.strip_prefix(THOUGHTS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());
let thought_uuid = match thought_uuid {
Some(u) => u,
None => return Ok(()),
};
let actor_user_id = self
.repo
.find_remote_actor_id(actor_url.as_str())
.await
.map_err(|e| anyhow!("{e}"))?;
let actor_user_id = match actor_user_id {
Some(id) => id,
None => return Ok(()),
};
let thought_id = domain::value_objects::ThoughtId::from_uuid(thought_uuid);
let _ = self.boosts.delete(&actor_user_id, &thought_id).await;
if let Some(ep) = &self.event_publisher {
ep.publish(&domain::events::DomainEvent::BoostRemoved {
user_id: actor_user_id,
thought_id,
})
.await
.map_err(|e| anyhow!("{e}"))?;
}
Ok(())
}
async fn on_announce_of_remote(&self, _object_url: &Url, _actor_url: &Url) -> Result<()> {
Ok(())
}

View File

@@ -74,11 +74,15 @@ pub struct ThoughtNoteInput {
impl ThoughtNote {
/// Returns `(note, extensions)` if `value` is a Note object, `None` otherwise.
pub fn try_from_ap(value: serde_json::Value) -> Option<(Self, Option<serde_json::Value>)> {
if value.get("type").and_then(|v| v.as_str()) != Some("Note") {
pub fn try_from_ap(mut value: serde_json::Value) -> Option<(Self, Option<serde_json::Value>)> {
let obj_type = value.get("type").and_then(|v| v.as_str());
if !matches!(obj_type, Some("Note" | "Article" | "Page")) {
return None;
}
let extensions = extract_extensions(&value);
if let Some(obj) = value.as_object_mut() {
obj.insert("type".to_string(), serde_json::json!("Note"));
}
serde_json::from_value(value)
.ok()
.map(|note| (note, extensions))

View File

@@ -101,6 +101,9 @@ pub trait ActivityPubRepository: Send + Sync {
/// Returns None for users that have not been federated.
async fn get_actor_ap_urls(&self, user_id: &UserId)
-> Result<Option<ActorApUrls>, DomainError>;
/// Sync display_name + avatar_url from remote_actors to users table.
async fn sync_remote_actor_to_user(&self, actor_ap_url: &str) -> Result<(), DomainError>;
}
#[async_trait]

View File

@@ -23,7 +23,8 @@ fn content_to_html(text: &str) -> String {
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;");
.replace('"', "&quot;")
.replace('\'', "&#39;");
let paragraphs: Vec<&str> = escaped.split('\n').filter(|s| !s.is_empty()).collect();
if paragraphs.is_empty() {
format!("<p>{}</p>", escaped)
@@ -116,13 +117,17 @@ fn k_ap_actor_to_domain(a: k_ap::RemoteActor) -> DomainRemoteActor {
last_fetched_at: chrono::Utc::now(),
bio: a.bio,
banner_url: a.banner_url,
also_known_as: a.also_known_as.into_iter().next(),
also_known_as: a.also_known_as,
followers_url: a.followers_url,
following_url: a.following_url,
inbox_url: Some(a.inbox_url),
shared_inbox_url: a.shared_inbox_url,
attachment: vec![],
}
}
// TODO: these fetches are unsigned — fails on instances with authorized-fetch (Secure Mode).
// Fix requires exposing k-ap's signed HTTP client.
async fn resolve_actor_profiles_from_urls(
urls: Vec<String>,
) -> Vec<domain::models::actor_connection_summary::ActorConnectionSummary> {
@@ -201,7 +206,9 @@ async fn webfinger_resolve_actor_url(handle: &str) -> anyhow::Result<String> {
.and_then(|links| {
links.iter().find(|l| {
l["rel"].as_str() == Some("self")
&& l["type"].as_str() == Some("application/activity+json")
&& l["type"].as_str().is_some_and(|t| {
t == "application/activity+json" || t.starts_with("application/ld+json")
})
})
})
.and_then(|l| l["href"].as_str())
@@ -415,11 +422,8 @@ impl FederationSchedulerPort for ApFederationAdapter {
actor_ap_url: &str,
collection_url: &str,
connection_type: &str,
page: u32,
_page: u32,
) -> Result<(), DomainError> {
if page != 1 {
return Ok(());
}
let actor = actor_ap_url.to_string();
let collection = collection_url.to_string();
let conn_type = connection_type.to_string();
@@ -536,9 +540,15 @@ impl FederationLookupPort for ApFederationAdapter {
last_fetched_at: chrono::Utc::now(),
bio: actor.bio,
banner_url: actor.banner_url.as_ref().map(|u| u.to_string()),
also_known_as: actor.also_known_as.into_iter().next(),
also_known_as: actor
.also_known_as
.into_iter()
.map(|u| u.to_string())
.collect(),
followers_url: actor.followers_url.as_ref().map(|u| u.to_string()),
following_url: actor.following_url.as_ref().map(|u| u.to_string()),
inbox_url: None,
shared_inbox_url: None,
attachment: actor
.attachment
.into_iter()
@@ -599,20 +609,36 @@ impl FederationFetchPort for ApFederationAdapter {
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
let url = base["first"]
let first_url = base["first"]
.as_str()
.map(|s| s.to_string())
.unwrap_or_else(|| format!("{}?page={}", outbox_url, page));
.unwrap_or_else(|| format!("{}?page=1", outbox_url));
let resp: serde_json::Value = client
.get(&url)
.header("Accept", "application/activity+json, application/ld+json")
.send()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?
.json()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
let mut current_url = first_url;
let mut hops = 0u32;
let target_page = page.max(1);
let max_hops = 10u32;
let resp: serde_json::Value = loop {
let page_resp: serde_json::Value = client
.get(&current_url)
.header("Accept", "application/activity+json, application/ld+json")
.send()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?
.json()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
hops += 1;
if hops >= target_page || hops >= max_hops {
break page_resp;
}
match page_resp["next"].as_str() {
Some(next) => current_url = next.to_string(),
None => break page_resp,
}
};
let empty = vec![];
let items = resp["orderedItems"].as_array().unwrap_or(&empty);
@@ -850,4 +876,33 @@ impl FederationFollowRequestPort for ApFederationAdapter {
}
}
// ── FederationBlockPort ──────────────────────────────────────────────────────
#[async_trait]
impl domain::ports::FederationBlockPort for ApFederationAdapter {
async fn block_remote(&self, local_user_id: &UserId, handle: &str) -> Result<(), DomainError> {
let actor_url = webfinger_resolve_actor_url(handle)
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
self.inner
.block_actor(local_user_id.as_uuid(), &actor_url)
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))
}
async fn unblock_remote(
&self,
local_user_id: &UserId,
handle: &str,
) -> Result<(), DomainError> {
let actor_url = webfinger_resolve_actor_url(handle)
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
self.inner
.unblock_actor(local_user_id.as_uuid(), &actor_url)
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))
}
}
// FederationActionPort is a blanket supertrait; no explicit impl needed.

View File

@@ -11,24 +11,24 @@ impl ThoughtsUrls {
}
}
pub fn user_url(&self, username: &str) -> Url {
Url::parse(&format!("{}/users/{}", self.base_url, username)).expect("valid URL")
pub fn user_url(&self, id: &str) -> Url {
Url::parse(&format!("{}/users/{}", self.base_url, id)).expect("valid URL")
}
pub fn thought_url(&self, thought_id: uuid::Uuid) -> Url {
Url::parse(&format!("{}/thoughts/{}", self.base_url, thought_id)).expect("valid URL")
}
pub fn user_inbox(&self, username: &str) -> Url {
Url::parse(&format!("{}/users/{}/inbox", self.base_url, username)).expect("valid URL")
pub fn user_inbox(&self, id: &str) -> Url {
Url::parse(&format!("{}/users/{}/inbox", self.base_url, id)).expect("valid URL")
}
pub fn user_outbox(&self, username: &str) -> Url {
Url::parse(&format!("{}/users/{}/outbox", self.base_url, username)).expect("valid URL")
pub fn user_outbox(&self, id: &str) -> Url {
Url::parse(&format!("{}/users/{}/outbox", self.base_url, id)).expect("valid URL")
}
pub fn user_followers(&self, username: &str) -> Url {
Url::parse(&format!("{}/users/{}/followers", self.base_url, username)).expect("valid URL")
pub fn user_followers(&self, id: &str) -> Url {
Url::parse(&format!("{}/users/{}/followers", self.base_url, id)).expect("valid URL")
}
}

View File

@@ -222,13 +222,12 @@ impl FollowRepository for PostgresFederationRepository {
}
async fn count_followers(&self, local_user_id: uuid::Uuid) -> Result<usize> {
let n: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM federation_followers WHERE local_user_id=$1 AND status='accepted'",
)
.bind(local_user_id)
.fetch_one(&self.pool)
.await
.map_err(|e| anyhow!(e))?;
let n: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM federation_followers WHERE local_user_id=$1")
.bind(local_user_id)
.fetch_one(&self.pool)
.await
.map_err(|e| anyhow!(e))?;
Ok(n as usize)
}
@@ -428,11 +427,24 @@ impl FollowRepository for PostgresFederationRepository {
async fn update_following_status(
&self,
_local_user_id: uuid::Uuid,
_remote_actor_url: &str,
_status: FollowingStatus,
local_user_id: uuid::Uuid,
remote_actor_url: &str,
status: FollowingStatus,
) -> Result<()> {
Ok(())
let s = match status {
FollowingStatus::Pending => "pending",
FollowingStatus::Accepted => "accepted",
};
sqlx::query(
"UPDATE federation_following SET status=$3 WHERE local_user_id=$1 AND remote_actor_url=$2",
)
.bind(local_user_id)
.bind(remote_actor_url)
.bind(s)
.execute(&self.pool)
.await
.map_err(|e| anyhow!(e))
.map(|_| ())
}
async fn get_following_outbox_url(
@@ -743,7 +755,7 @@ impl PostgresApUserRepository {
}
fn row_to_ap_user(&self, r: UserRow) -> ApUser {
let profile_url = url::Url::parse(&format!("{}/users/{}", self.base_url, r.username)).ok();
let profile_url = url::Url::parse(&format!("{}/users/{}", self.base_url, r.id)).ok();
let avatar_url = r.avatar_url.and_then(|u| url::Url::parse(&u).ok());
let banner_url = r.header_url.and_then(|u| url::Url::parse(&u).ok());
ApUser {

View File

@@ -0,0 +1 @@
ALTER TABLE users ALTER COLUMN username TYPE VARCHAR(255);

View File

@@ -0,0 +1,2 @@
ALTER TABLE federation_following
ADD COLUMN IF NOT EXISTS status TEXT NOT NULL DEFAULT 'accepted';

View File

@@ -0,0 +1 @@
ALTER TABLE remote_actors ADD COLUMN IF NOT EXISTS attachment JSONB DEFAULT '[]'::jsonb;

View File

@@ -1,7 +1,7 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
const MAX_REMOTE_CONTENT_CHARS: usize = 500;
const MAX_REMOTE_CONTENT_CHARS: usize = 5000;
const THOUGHTS_PATH_PREFIX: &str = "/thoughts/";
use chrono::{DateTime, Utc};
use sqlx::PgPool;
@@ -155,24 +155,28 @@ impl ActivityPubRepository for PgActivityPubRepository {
return Ok(id);
}
let new_id = uuid::Uuid::new_v4();
// Use the last path segment as username (e.g. /users/alice → "alice").
// Falls back to a random short id for long segments (e.g. UUID-based actor URLs).
// username column is VARCHAR(32).
let last_seg = url::Url::parse(actor_ap_url)
.ok()
let parsed = url::Url::parse(actor_ap_url).ok();
let domain_str = parsed
.as_ref()
.and_then(|u| u.host_str().map(|s| s.to_string()))
.unwrap_or_default();
let last_seg = parsed
.and_then(|u| {
u.path_segments()
.and_then(|mut s| s.next_back().map(|s| s.to_string()))
})
.unwrap_or_default();
let handle = if last_seg.is_empty() {
format!("remote_{}", &new_id.to_string()[..13])
} else if last_seg.len() <= 32 {
last_seg
let handle = if last_seg.is_empty() || domain_str.is_empty() {
format!("r_{}", &new_id.to_string()[..13])
} else {
format!("remote_{}", &new_id.to_string()[..13])
let candidate = format!("{}@{}", last_seg, domain_str);
if candidate.len() <= 255 {
candidate
} else {
format!("r_{}", &new_id.to_string()[..13])
}
};
sqlx::query(
let result = sqlx::query(
"INSERT INTO users(id,username,email,password_hash,local,ap_id,created_at,updated_at)
VALUES($1,$2,$3,'',false,$4,NOW(),NOW()) ON CONFLICT(ap_id) DO NOTHING",
)
@@ -181,9 +185,24 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(format!("{}@remote", new_id))
.bind(actor_ap_url)
.execute(&self.pool)
.await
.into_domain()?;
// Re-fetch to get whichever id won the race
.await;
if result.is_err() {
let fallback = format!("r_{}", &new_id.to_string()[..13]);
let new_id2 = uuid::Uuid::new_v4();
sqlx::query(
"INSERT INTO users(id,username,email,password_hash,local,ap_id,created_at,updated_at)
VALUES($1,$2,$3,'',false,$4,NOW(),NOW()) ON CONFLICT(ap_id) DO NOTHING",
)
.bind(new_id2)
.bind(&fallback)
.bind(format!("{}@remote", new_id2))
.bind(actor_ap_url)
.execute(&self.pool)
.await
.into_domain()?;
}
self.find_remote_actor_id(actor_ap_url)
.await?
.ok_or_else(|| {
@@ -345,6 +364,19 @@ impl ActivityPubRepository for PgActivityPubRepository {
.into_domain()
.map(|opt| opt.map(|(ap_id, inbox_url)| ActorApUrls { ap_id, inbox_url }))
}
async fn sync_remote_actor_to_user(&self, actor_ap_url: &str) -> Result<(), DomainError> {
sqlx::query(
"UPDATE users SET display_name = ra.display_name, avatar_url = ra.avatar_url, updated_at = NOW()
FROM remote_actors ra
WHERE users.ap_id = ra.url AND users.ap_id = $1 AND users.local = false",
)
.bind(actor_ap_url)
.execute(&self.pool)
.await
.into_domain()
.map(|_| ())
}
}
#[cfg(test)]

View File

@@ -113,14 +113,14 @@ impl<'a> FeedSqlBuilder<'a> {
}
}
fn select(&self) -> String {
fn select(&self, viewer_param: &str) -> String {
let (viewer_cols, viewer_joins) = match self.viewer {
Some(uid) => (
Some(_) => (
"(lv.thought_id IS NOT NULL) AS liked_by_viewer,
(bv.thought_id IS NOT NULL) AS boosted_by_viewer".to_string(),
format!(
"LEFT JOIN (SELECT thought_id FROM likes WHERE user_id='{uid}') lv ON lv.thought_id = t.id
LEFT JOIN (SELECT thought_id FROM boosts WHERE user_id='{uid}') bv ON bv.thought_id = t.id"
"LEFT JOIN (SELECT thought_id FROM likes WHERE user_id={viewer_param}) lv ON lv.thought_id = t.id
LEFT JOIN (SELECT thought_id FROM boosts WHERE user_id={viewer_param}) bv ON bv.thought_id = t.id"
),
),
None => (
@@ -164,13 +164,13 @@ impl<'a> FeedSqlBuilder<'a> {
)
}
fn fed_clause(&self) -> String {
fn fed_clause(&self, viewer_param: &str) -> String {
match self.viewer {
Some(fid) => format!(
Some(_) => format!(
" OR t.user_id IN (
SELECT u2.id FROM users u2
JOIN federation_following ff ON u2.ap_id = ff.remote_actor_url
WHERE ff.local_user_id = '{fid}'
WHERE ff.local_user_id = {viewer_param}
)"
),
None => String::new(),
@@ -217,7 +217,7 @@ impl<'a> FeedSqlBuilder<'a> {
);
let data = format!(
"{} WHERE t.local=true AND t.visibility='public'{} {} LIMIT $1 OFFSET $2",
self.select(),
self.select("$3"),
filter,
order
);
@@ -225,17 +225,16 @@ impl<'a> FeedSqlBuilder<'a> {
}
fn home(&self) -> (String, String) {
let fed = self.fed_clause();
let filter = self.filter_sql();
let order = self.order_sql();
let count = format!(
let count = format!(
"SELECT COUNT(*) FROM thoughts t WHERE (t.user_id=ANY($1){}) AND t.visibility != 'direct'{}",
fed, filter
self.fed_clause("$2"), filter
);
let data =
format!(
"{} WHERE (t.user_id=ANY($1){}) AND t.visibility != 'direct'{} {} LIMIT $2 OFFSET $3",
self.select(), fed, filter, order
self.select("$4"), self.fed_clause("$4"), filter, order
);
(count, data)
}
@@ -249,7 +248,7 @@ impl<'a> FeedSqlBuilder<'a> {
);
let data = format!(
"{} WHERE t.content % $1 AND t.visibility='public'{} {} LIMIT $2 OFFSET $3",
self.select(),
self.select("$4"),
filter,
order
);
@@ -271,7 +270,7 @@ impl<'a> FeedSqlBuilder<'a> {
JOIN thought_tags tt ON tt.thought_id = t.id
JOIN tags tg ON tg.id = tt.tag_id
WHERE tg.name = $1 AND t.visibility = 'public'{} {} LIMIT $2 OFFSET $3",
self.select(),
self.select("$4"),
filter,
order
);
@@ -287,7 +286,7 @@ impl<'a> FeedSqlBuilder<'a> {
);
let data = format!(
"{} WHERE t.user_id = $1 AND ($4::uuid = $1 OR (t.visibility != 'direct' AND (t.visibility IN ('public', 'unlisted') OR (t.visibility = 'followers' AND EXISTS(SELECT 1 FROM follows WHERE follower_id = $4 AND following_id = $1 AND state = 'accepted'))))){} {} LIMIT $2 OFFSET $3",
self.select(), filter, order
self.select("$4"), filter, order
);
(count, data)
}
@@ -300,12 +299,15 @@ impl FeedRepository for PgFeedRepository {
let page = &req.query.page;
let builder = FeedSqlBuilder::new(&req.options, &req.query.scope, viewer);
let viewer_uuid = viewer.unwrap_or(uuid::Uuid::nil());
match &req.query.scope {
FeedScope::Home { following_ids } => {
let ids: Vec<uuid::Uuid> = following_ids.iter().map(|id| id.as_uuid()).collect();
let (count_sql, data_sql) = builder.home();
let total: i64 = sqlx::query_scalar(&count_sql)
.bind(&ids)
.bind(viewer_uuid)
.fetch_one(&self.pool)
.await
.into_domain()?;
@@ -313,6 +315,7 @@ impl FeedRepository for PgFeedRepository {
.bind(&ids)
.bind(page.limit())
.bind(page.offset())
.bind(viewer_uuid)
.fetch_all(&self.pool)
.await
.into_domain()?;
@@ -336,6 +339,7 @@ impl FeedRepository for PgFeedRepository {
let rows = sqlx::query_as::<_, FeedRow>(&data_sql)
.bind(page.limit())
.bind(page.offset())
.bind(viewer_uuid)
.fetch_all(&self.pool)
.await
.into_domain()?;
@@ -361,6 +365,7 @@ impl FeedRepository for PgFeedRepository {
.bind(query)
.bind(page.limit())
.bind(page.offset())
.bind(viewer_uuid)
.fetch_all(&self.pool)
.await
.into_domain()?;
@@ -386,6 +391,7 @@ impl FeedRepository for PgFeedRepository {
.bind(tag_name)
.bind(page.limit())
.bind(page.offset())
.bind(viewer_uuid)
.fetch_all(&self.pool)
.await
.into_domain()?;
@@ -402,7 +408,6 @@ impl FeedRepository for PgFeedRepository {
FeedScope::User { user_id } => {
let uid = user_id.as_uuid();
let viewer_uuid = viewer.unwrap_or(uuid::Uuid::nil());
let (count_sql, data_sql) = builder.user();
let total: i64 = sqlx::query_scalar(&count_sql)
.bind(uid)

View File

@@ -18,14 +18,44 @@ impl PgRemoteActorRepository {
#[async_trait]
impl RemoteActorRepository for PgRemoteActorRepository {
async fn upsert(&self, a: &RemoteActor) -> Result<(), DomainError> {
let also_known_as: Option<Vec<&str>> = if a.also_known_as.is_empty() {
None
} else {
Some(a.also_known_as.iter().map(|s| s.as_str()).collect())
};
let attachment_json: serde_json::Value = a
.attachment
.iter()
.map(|(n, v)| serde_json::json!({"name": n, "value": v}))
.collect();
sqlx::query(
"INSERT INTO remote_actors(url,handle,display_name,avatar_url,last_fetched_at)
VALUES($1,$2,$3,$4,$5)
ON CONFLICT(url) DO UPDATE SET handle=EXCLUDED.handle,display_name=EXCLUDED.display_name,
avatar_url=EXCLUDED.avatar_url,last_fetched_at=EXCLUDED.last_fetched_at"
"INSERT INTO remote_actors(url,handle,display_name,avatar_url,last_fetched_at,
bio,banner_url,outbox_url,followers_url,following_url,also_known_as,attachment)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
ON CONFLICT(url) DO UPDATE SET
handle=EXCLUDED.handle,display_name=EXCLUDED.display_name,
avatar_url=EXCLUDED.avatar_url,last_fetched_at=EXCLUDED.last_fetched_at,
bio=EXCLUDED.bio,banner_url=EXCLUDED.banner_url,
outbox_url=EXCLUDED.outbox_url,followers_url=EXCLUDED.followers_url,
following_url=EXCLUDED.following_url,also_known_as=EXCLUDED.also_known_as,
attachment=EXCLUDED.attachment",
)
.bind(&a.url).bind(&a.handle).bind(&a.display_name).bind(&a.avatar_url).bind(a.last_fetched_at)
.execute(&self.pool).await.into_domain().map(|_| ())
.bind(&a.url)
.bind(&a.handle)
.bind(&a.display_name)
.bind(&a.avatar_url)
.bind(a.last_fetched_at)
.bind(&a.bio)
.bind(&a.banner_url)
.bind(&a.outbox_url)
.bind(&a.followers_url)
.bind(&a.following_url)
.bind(also_known_as.as_deref())
.bind(&attachment_json)
.execute(&self.pool)
.await
.into_domain()
.map(|_| ())
}
async fn find_by_url(&self, url: &str) -> Result<Option<RemoteActor>, DomainError> {
@@ -36,24 +66,55 @@ impl RemoteActorRepository for PgRemoteActorRepository {
display_name: Option<String>,
avatar_url: Option<String>,
last_fetched_at: DateTime<Utc>,
bio: Option<String>,
banner_url: Option<String>,
outbox_url: Option<String>,
followers_url: Option<String>,
following_url: Option<String>,
also_known_as: Option<Vec<String>>,
inbox_url: Option<String>,
shared_inbox_url: Option<String>,
attachment: Option<serde_json::Value>,
}
sqlx::query_as::<_, Row>(
"SELECT url,handle,display_name,avatar_url,last_fetched_at FROM remote_actors WHERE url=$1"
).bind(url).fetch_optional(&self.pool).await
"SELECT url,handle,display_name,avatar_url,last_fetched_at,
bio,banner_url,outbox_url,followers_url,following_url,also_known_as,
inbox_url,shared_inbox_url,attachment
FROM remote_actors WHERE url=$1",
)
.bind(url)
.fetch_optional(&self.pool)
.await
.into_domain()
.map(|o| o.map(|r| RemoteActor {
url: r.url,
handle: r.handle,
display_name: r.display_name,
avatar_url: r.avatar_url,
last_fetched_at: r.last_fetched_at,
bio: None,
banner_url: None,
also_known_as: None,
outbox_url: None,
followers_url: None,
following_url: None,
attachment: vec![],
}))
.map(|o| {
o.map(|r| RemoteActor {
url: r.url,
handle: r.handle,
display_name: r.display_name,
avatar_url: r.avatar_url,
last_fetched_at: r.last_fetched_at,
bio: r.bio,
banner_url: r.banner_url,
also_known_as: r.also_known_as.unwrap_or_default(),
outbox_url: r.outbox_url,
followers_url: r.followers_url,
following_url: r.following_url,
inbox_url: r.inbox_url,
shared_inbox_url: r.shared_inbox_url,
attachment: r
.attachment
.and_then(|v| v.as_array().cloned())
.map(|arr| {
arr.into_iter()
.filter_map(|item| {
let name = item.get("name")?.as_str()?.to_string();
let value = item.get("value")?.as_str()?.to_string();
Some((name, value))
})
.collect()
})
.unwrap_or_default(),
})
})
}
}