GKaszewski/thoughts
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update dependencies to k-ap v0.1.7 and add profileHref utility for user links

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-05-27 23:37:22 +02:00
parent 6f65742284
commit f6893b19dc
18 changed files with 254 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/adapters/activitypub/Cargo.toml
View File

@@ -4,7 +4,7 @@ version = "0.1.0"
edition = "2021"
[dependencies]
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.3" }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.7" }
domain = { workspace = true }
url = { workspace = true }
serde = { workspace = true }

84
crates/adapters/activitypub/src/handler.rs
View File

@@ -14,36 +14,6 @@ use domain::ports::{EventPublisher, TagRepository};
use domain::value_objects::UserId;
use k_ap::ApObjectHandler;
fn extract_note_extensions(obj: &serde_json::Value) -> Option<serde_json::Value> {
const STANDARD: &[&str] = &[
"type",
"id",
"attributedTo",
"content",
"published",
"to",
"cc",
"inReplyTo",
"sensitive",
"summary",
"tag",
"url",
"@context",
"mediaType",
];
let extensions: serde_json::Map<String, serde_json::Value> = obj
.as_object()?
.iter()
.filter(|(k, _)| !STANDARD.contains(&k.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
if extensions.is_empty() {
None
} else {
Some(serde_json::Value::Object(extensions))
}
}
pub struct ThoughtsObjectHandler {
repo: Arc<dyn ActivityPubRepository>,
urls: ThoughtsUrls,
@@ -148,8 +118,10 @@ impl ApObjectHandler for ThoughtsObjectHandler {
actor_url: &Url,
object: serde_json::Value,
) -> Result<()> {
let note_extensions = extract_note_extensions(&object);
let note: ThoughtNote = serde_json::from_value(object)?;
let Some((note, note_extensions)) = ThoughtNote::try_from_ap(object) else {
tracing::debug!(ap_id = %ap_id, "on_create: skipping non-Note object");
return Ok(());
};
let author_id = self
.repo
.intern_remote_actor(actor_url.as_str())
@@ -249,7 +221,10 @@ impl ApObjectHandler for ThoughtsObjectHandler {
_actor_url: &Url,
object: serde_json::Value,
) -> Result<()> {
let note: ThoughtNote = serde_json::from_value(object)?;
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
@@ -440,46 +415,3 @@ impl ApObjectHandler for ThoughtsObjectHandler {
.map_err(|e| anyhow!("{e}"))
}
}
#[cfg(test)]
mod extract_tests {
use super::extract_note_extensions;
#[test]
fn extracts_non_standard_fields() {
let obj = serde_json::json!({
"type": "Note",
"id": "https://example.com/notes/1",
"content": "hello",
"published": "2025-01-01T00:00:00Z",
"movieTitle": "Dune",
"rating": 5,
"posterUrl": "https://example.com/poster.jpg"
});
let ext = extract_note_extensions(&obj).unwrap();
assert_eq!(ext["movieTitle"], "Dune");
assert_eq!(ext["rating"], 5);
assert_eq!(ext["posterUrl"], "https://example.com/poster.jpg");
assert!(ext.get("type").is_none());
assert!(ext.get("content").is_none());
assert!(ext.get("id").is_none());
}
#[test]
fn returns_none_for_standard_only_note() {
let obj = serde_json::json!({
"type": "Note",
"content": "hello",
"published": "2025-01-01T00:00:00Z",
"to": ["https://www.w3.org/ns/activitystreams#Public"],
"tag": []
});
assert!(extract_note_extensions(&obj).is_none());
}
#[test]
fn returns_none_for_non_object() {
let obj = serde_json::json!("not an object");
assert!(extract_note_extensions(&obj).is_none());
}
}

42
crates/adapters/activitypub/src/note/mod.rs
View File

@@ -4,6 +4,37 @@ use k_ap::AS_PUBLIC;
use serde::{Deserialize, Serialize};
use url::Url;
const STANDARD_NOTE_FIELDS: &[&str] = &[
"type",
"id",
"attributedTo",
"content",
"published",
"to",
"cc",
"inReplyTo",
"sensitive",
"summary",
"tag",
"url",
"@context",
"mediaType",
];
pub fn extract_extensions(obj: &serde_json::Value) -> Option<serde_json::Value> {
let extensions: serde_json::Map<String, serde_json::Value> = obj
.as_object()?
.iter()
.filter(|(k, _)| !STANDARD_NOTE_FIELDS.contains(&k.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
if extensions.is_empty() {
None
} else {
Some(serde_json::Value::Object(extensions))
}
}
/// AP Note representing a Thought.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -42,6 +73,17 @@ 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") {
return None;
}
let extensions = extract_extensions(&value);
serde_json::from_value(value)
.ok()
.map(|note| (note, extensions))
}
pub fn new_public(p: ThoughtNoteInput) -> Self {
Self {
kind: Default::default(),

50
crates/adapters/activitypub/src/note/tests.rs
View File

@@ -1,5 +1,55 @@
use super::*;
#[test]
fn extract_extensions_picks_up_non_standard_fields() {
let obj = serde_json::json!({
"type": "Note",
"id": "https://example.com/notes/1",
"content": "hello",
"published": "2025-01-01T00:00:00Z",
"movieTitle": "Dune",
"rating": 5,
"posterUrl": "https://example.com/poster.jpg"
});
let ext = extract_extensions(&obj).unwrap();
assert_eq!(ext["movieTitle"], "Dune");
assert_eq!(ext["rating"], 5);
assert_eq!(ext["posterUrl"], "https://example.com/poster.jpg");
assert!(ext.get("type").is_none());
assert!(ext.get("content").is_none());
assert!(ext.get("id").is_none());
}
#[test]
fn extract_extensions_returns_none_for_standard_only_note() {
let obj = serde_json::json!({
"type": "Note",
"content": "hello",
"published": "2025-01-01T00:00:00Z",
"to": ["https://www.w3.org/ns/activitystreams#Public"],
"tag": []
});
assert!(extract_extensions(&obj).is_none());
}
#[test]
fn extract_extensions_returns_none_for_non_object() {
let obj = serde_json::json!("not an object");
assert!(extract_extensions(&obj).is_none());
}
#[test]
fn try_from_ap_returns_none_for_person() {
let person = serde_json::json!({ "type": "Person", "id": "https://example.com/users/1" });
assert!(ThoughtNote::try_from_ap(person).is_none());
}
#[test]
fn try_from_ap_returns_none_for_missing_type() {
let obj = serde_json::json!({ "content": "hello" });
assert!(ThoughtNote::try_from_ap(obj).is_none());
}
#[test]
fn note_serializes_with_public_audience() {
let note = ThoughtNote::new_public(super::ThoughtNoteInput {

79
crates/adapters/activitypub/src/service.rs
View File

@@ -502,72 +502,29 @@ impl FederationSchedulerPort for ApFederationAdapter {
#[async_trait]
impl FederationLookupPort for ApFederationAdapter {
async fn lookup_actor(&self, handle: &str) -> Result<DomainRemoteActor, DomainError> {
let normalized = handle.trim_start_matches('@');
let at = normalized
.rfind('@')
.ok_or_else(|| DomainError::InvalidInput("handle must be user@domain".into()))?;
let (user, domain_str) = (&normalized[..at], &normalized[at + 1..]);
let wf_url = format!(
"https://{}/.well-known/webfinger?resource=acct:{}@{}",
domain_str, user, domain_str
);
let wf: serde_json::Value = reqwest::Client::new()
.get(&wf_url)
.header("Accept", "application/jrd+json, application/json")
.send()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?
.json()
let actor = self
.inner
.lookup_actor_by_handle(handle)
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
let self_href = wf["links"]
.as_array()
.and_then(|links| {
links.iter().find(|l| {
l["rel"].as_str() == Some("self")
&& l["type"].as_str() == Some("application/activity+json")
})
})
.and_then(|l| l["href"].as_str())
.ok_or(DomainError::NotFound)?
.to_owned();
let actor_json: serde_json::Value = reqwest::Client::new()
.get(&self_href)
.header("Accept", "application/activity+json")
.send()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?
.json()
.await
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
let ap_url = actor_json["id"].as_str().unwrap_or(&self_href).to_string();
let preferred_username = actor_json["preferredUsername"]
.as_str()
.unwrap_or("")
.to_string();
let domain_part = url::Url::parse(&ap_url)
.ok()
.and_then(|u| u.host_str().map(|s| s.to_string()))
.unwrap_or_default();
let full_handle = format!("{}@{}", preferred_username, domain_part);
Ok(DomainRemoteActor {
url: ap_url.clone(),
handle: full_handle,
display_name: actor_json["name"].as_str().map(|s| s.to_string()),
avatar_url: actor_json["icon"]["url"].as_str().map(|s| s.to_string()),
outbox_url: actor_json["outbox"].as_str().map(|s| s.to_string()),
url: actor.ap_url.to_string(),
handle: actor.handle,
display_name: actor.display_name,
avatar_url: actor.avatar_url.as_ref().map(|u| u.to_string()),
outbox_url: actor.outbox_url.as_ref().map(|u| u.to_string()),
last_fetched_at: chrono::Utc::now(),
bio: actor_json["summary"].as_str().map(|s| s.to_string()),
banner_url: actor_json["image"]["url"].as_str().map(|s| s.to_string()),
also_known_as: None,
followers_url: actor_json["followers"].as_str().map(|s| s.to_string()),
following_url: actor_json["following"].as_str().map(|s| s.to_string()),
attachment: vec![],
bio: actor.bio,
banner_url: actor.banner_url.as_ref().map(|u| u.to_string()),
also_known_as: actor.also_known_as,
followers_url: actor.followers_url.as_ref().map(|u| u.to_string()),
following_url: actor.following_url.as_ref().map(|u| u.to_string()),
attachment: actor
.attachment
.into_iter()
.map(|f| (f.name, f.value))
.collect(),
})
}

2
crates/adapters/postgres-federation/Cargo.toml
View File

@@ -4,7 +4,7 @@ version = "0.1.0"
edition = "2021"
[dependencies]
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.3" }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.7" }
sqlx = { workspace = true }
uuid = { workspace = true }
chrono = { workspace = true }

32
crates/application/src/services/federation_event/mod.rs
View File

@@ -41,7 +41,10 @@ impl FederationEventService {
{
t
}
_ => return Ok(()),
_ => {
tracing::debug!(thought_id = %thought_id, "federation: skipping ThoughtCreated (remote or non-public)");
return Ok(());
}
};
let user = match self.users.find_by_id(user_id).await? {
Some(u) => u,
@@ -58,6 +61,7 @@ impl FederationEventService {
} else {
None
};
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Create(Note)");
self.ap
.broadcast_create(
user_id,
@@ -72,8 +76,7 @@ impl FederationEventService {
thought_id,
user_id,
} => {
// No DB lookup — thought is already deleted when this event fires.
// No locality guard: delete commands only reach local thoughts via the use case.
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Delete");
let ap_id = format!("{}/thoughts/{}", self.base_url, thought_id);
self.ap.broadcast_delete(user_id, &ap_id).await
}
@@ -106,6 +109,7 @@ impl FederationEventService {
} else {
None
};
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Update(Note)");
self.ap
.broadcast_update(
user_id,
@@ -121,16 +125,19 @@ impl FederationEventService {
user_id,
thought_id,
} => {
// Only fan-out if the booster is a local user. Remote boosts must not be re-broadcast.
let booster = match self.users.find_by_id(user_id).await? {
Some(u) if u.local => u,
_ => return Ok(()),
_ => {
tracing::debug!(user_id = %user_id, "federation: skipping BoostAdded (remote user)");
return Ok(());
}
};
let _ = booster;
if self.thoughts.find_by_id(thought_id).await?.is_none() {
return Ok(());
}
let object_ap_id = self.object_ap_id(thought_id).await?;
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Announce");
self.ap.broadcast_announce(user_id, &object_ap_id).await
}
@@ -142,6 +149,7 @@ impl FederationEventService {
return Ok(());
}
let object_ap_id = self.object_ap_id(thought_id).await?;
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Undo(Announce)");
self.ap
.broadcast_undo_announce(user_id, &object_ap_id)
.await
@@ -152,10 +160,12 @@ impl FederationEventService {
user_id,
thought_id,
} => {
// Only federate: local liker + remote thought (has ap_id) + author has inbox.
let liker = match self.users.find_by_id(user_id).await? {
Some(u) if u.local => u,
_ => return Ok(()),
_ => {
tracing::debug!(user_id = %user_id, "federation: skipping LikeAdded (remote user)");
return Ok(());
}
};
let _ = liker;
let thought = match self.thoughts.find_by_id(thought_id).await? {
@@ -164,12 +174,16 @@ impl FederationEventService {
};
let thought_ap_id = match self.ap_repo.get_thought_ap_id(thought_id).await? {
Some(id) => id,
None => return Ok(()), // local thought — no federation needed
None => {
tracing::debug!(thought_id = %thought_id, "federation: skipping LikeAdded (local thought)");
return Ok(());
}
};
let actor_urls = match self.ap_repo.get_actor_ap_urls(&thought.user_id).await? {
Some(u) => u,
None => return Ok(()),
};
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Like");
self.ap
.broadcast_like(user_id, &thought_ap_id, &actor_urls.inbox_url)
.await
@@ -196,12 +210,14 @@ impl FederationEventService {
Some(u) => u,
None => return Ok(()),
};
tracing::info!(thought_id = %thought_id, user_id = %user_id, "federation: broadcasting Undo(Like)");
self.ap
.broadcast_undo_like(user_id, &thought_ap_id, &actor_urls.inbox_url)
.await
}
DomainEvent::ProfileUpdated { user_id } => {
tracing::info!(user_id = %user_id, "federation: broadcasting actor update");
self.ap.broadcast_actor_update(user_id).await
}

5
crates/application/src/services/notification_event/mod.rs
View File

@@ -35,6 +35,7 @@ impl NotificationEventService {
if is_self_action(&thought.user_id, user_id) {
return Ok(());
}
tracing::info!(from = %user_id, to = %thought.user_id, thought_id = %thought_id, "notification: Like");
self.notifications
.save(&Notification {
id: NotificationId::new(),
@@ -60,6 +61,7 @@ impl NotificationEventService {
if is_self_action(&thought.user_id, user_id) {
return Ok(());
}
tracing::info!(from = %user_id, to = %thought.user_id, thought_id = %thought_id, "notification: Boost");
self.notifications
.save(&Notification {
id: NotificationId::new(),
@@ -77,6 +79,7 @@ impl NotificationEventService {
follower_id,
following_id,
} => {
tracing::info!(from = %follower_id, to = %following_id, "notification: Follow");
self.notifications
.save(&Notification {
id: NotificationId::new(),
@@ -105,6 +108,7 @@ impl NotificationEventService {
if is_self_action(&original.user_id, user_id) {
return Ok(());
}
tracing::info!(from = %user_id, to = %original.user_id, thought_id = %thought_id, "notification: Reply");
self.notifications
.save(&Notification {
id: NotificationId::new(),
@@ -123,6 +127,7 @@ impl NotificationEventService {
mentioned_user_id,
author_user_id,
} => {
tracing::info!(from = %author_user_id, to = %mentioned_user_id, thought_id = %thought_id, "notification: Mention");
self.notifications
.save(&Notification {
id: NotificationId::new(),

2
crates/bootstrap/Cargo.toml
View File

@@ -14,7 +14,7 @@ postgres = { workspace = true }
postgres-search = { workspace = true }
postgres-federation = { workspace = true }
activitypub = { workspace = true }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.3" }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.7" }
nats = { workspace = true }
event-transport = { workspace = true }
auth = { workspace = true }

2
crates/worker/Cargo.toml
View File

@@ -13,7 +13,7 @@ application = { workspace = true }
nats = { workspace = true }
event-transport = { workspace = true }
event-payload = { workspace = true }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.3" }
k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.7" }
activitypub = { workspace = true }
postgres = { workspace = true }
postgres-federation = { workspace = true }