, DomainError> {
+ use chrono::DateTime;
+
+ let client = reqwest::Client::new();
+ let base: serde_json::Value = client
+ .get(outbox_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 url = base["first"]
+ .as_str()
+ .map(|s| s.to_string())
+ .unwrap_or_else(|| format!("{}?page={}", outbox_url, page));
+
+ 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 empty = vec![];
+ let items = resp["orderedItems"].as_array().unwrap_or(&empty);
+
+ let notes = items
+ .iter()
+ .filter_map(|item| {
+ let note = if item["type"].as_str() == Some("Create") {
+ &item["object"]
+ } else if item["type"].as_str() == Some("Note") {
+ item
+ } else {
+ return None;
+ };
+
+ let to = note["to"].as_array()?;
+ let is_public = to
+ .iter()
+ .any(|t| t.as_str() == Some("https://www.w3.org/ns/activitystreams#Public"));
+ if !is_public {
+ return None;
+ }
+
+ let published =
+ DateTime::parse_from_rfc3339(note["published"].as_str()?)
+ .ok()?
+ .with_timezone(&chrono::Utc);
+
+ let text = note["content"].as_str().unwrap_or("").to_string();
+ let has_attachments = note["attachment"]
+ .as_array()
+ .map(|a| !a.is_empty())
+ .unwrap_or(false);
+
+ let content = if has_attachments {
+ let notice =
+ "📎 Media attachment — not supported
";
+ if text.is_empty() {
+ notice.to_string()
+ } else {
+ format!("{text}{notice}")
+ }
+ } else {
+ text
+ };
+
+ Some(domain::models::remote_note::RemoteNote {
+ ap_id: note["id"].as_str()?.to_string(),
+ content,
+ published,
+ sensitive: note["sensitive"].as_bool().unwrap_or(false),
+ content_warning: note["summary"].as_str().map(|s| s.to_string()),
+ })
+ })
+ .collect();
+
+ Ok(notes)
+ }
+
+ async fn fetch_actor_urls_from_collection(
+ &self,
+ collection_url: &str,
+ ) -> Result, DomainError> {
+ let client = reqwest::Client::new();
+ let base: serde_json::Value = client
+ .get(collection_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 page = if base["orderedItems"].is_null() {
+ if let Some(first_url) = base["first"].as_str() {
+ client
+ .get(first_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()))?
+ } else {
+ base
+ }
+ } else {
+ base
+ };
+
+ let empty = vec![];
+ let items = page["orderedItems"].as_array().unwrap_or(&empty);
+ Ok(items
+ .iter()
+ .filter_map(|v| v.as_str().map(|s| s.to_string()))
+ .collect())
+ }
+
+ async fn resolve_actor_profiles(
+ &self,
+ urls: Vec,
+ ) -> Vec {
+ resolve_actor_profiles_from_urls(urls).await
+ }
+}
+
+// ── FederationFollowPort ──────────────────────────────────────────────────────
+
+#[async_trait]
+impl FederationFollowPort for ApFederationAdapter {
+ async fn follow_remote(&self, local_user_id: &UserId, handle: &str) -> Result<(), DomainError> {
+ self.inner
+ .follow(local_user_id.as_uuid(), handle)
+ .await
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn unfollow_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
+ .unfollow(local_user_id.as_uuid(), &actor_url)
+ .await
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn get_remote_following(
+ &self,
+ user_id: &UserId,
+ ) -> Result, DomainError> {
+ self.inner
+ .get_following(user_id.as_uuid())
+ .await
+ .map(|v| v.into_iter().map(k_ap_actor_to_domain).collect())
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+}
+
+// ── FederationFollowRequestPort ───────────────────────────────────────────────
+
+#[async_trait]
+impl FederationFollowRequestPort for ApFederationAdapter {
+ async fn get_pending_followers(
+ &self,
+ user_id: &UserId,
+ ) -> Result, DomainError> {
+ self.inner
+ .get_pending_followers(user_id.as_uuid())
+ .await
+ .map(|v| v.into_iter().map(k_ap_actor_to_domain).collect())
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn accept_follow_request(
+ &self,
+ user_id: &UserId,
+ actor_url: &str,
+ ) -> Result<(), DomainError> {
+ self.inner
+ .accept_follower(user_id.as_uuid(), actor_url)
+ .await
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn reject_follow_request(
+ &self,
+ user_id: &UserId,
+ actor_url: &str,
+ ) -> Result<(), DomainError> {
+ self.inner
+ .reject_follower(user_id.as_uuid(), actor_url)
+ .await
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn get_remote_followers(
+ &self,
+ user_id: &UserId,
+ ) -> Result, DomainError> {
+ self.inner
+ .get_accepted_followers(user_id.as_uuid())
+ .await
+ .map(|v| v.into_iter().map(k_ap_actor_to_domain).collect())
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+
+ async fn remove_remote_follower(
+ &self,
+ user_id: &UserId,
+ actor_url: &str,
+ ) -> Result<(), DomainError> {
+ self.inner
+ .remove_follower(user_id.as_uuid(), actor_url)
+ .await
+ .map_err(|e| DomainError::ExternalService(e.to_string()))
+ }
+}
+
+// FederationActionPort is a blanket supertrait; no explicit impl needed.
diff --git a/crates/adapters/postgres-federation/Cargo.toml b/crates/adapters/postgres-federation/Cargo.toml
index 0fc67e3..c8582b5 100644
--- a/crates/adapters/postgres-federation/Cargo.toml
+++ b/crates/adapters/postgres-federation/Cargo.toml
@@ -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.0" }
+k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.2" }
sqlx = { workspace = true }
uuid = { workspace = true }
chrono = { workspace = true }
diff --git a/crates/bootstrap/Cargo.toml b/crates/bootstrap/Cargo.toml
index b36542d..6e5e220 100644
--- a/crates/bootstrap/Cargo.toml
+++ b/crates/bootstrap/Cargo.toml
@@ -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.0" }
+k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.2" }
nats = { workspace = true }
event-transport = { workspace = true }
auth = { workspace = true }
diff --git a/crates/bootstrap/src/factory.rs b/crates/bootstrap/src/factory.rs
index 7556bfb..a15883e 100644
--- a/crates/bootstrap/src/factory.rs
+++ b/crates/bootstrap/src/factory.rs
@@ -5,7 +5,7 @@ use async_trait::async_trait;
use sqlx::PgPool;
use std::sync::Arc;
-use activitypub::ThoughtsObjectHandler;
+use activitypub::{ApFederationAdapter, ThoughtsObjectHandler};
use k_ap::ActivityPubService;
use auth::ApiKeyServiceImpl;
use domain::{
@@ -27,7 +27,7 @@ use crate::config::Config;
/// Everything the binary needs to start serving.
pub struct Infrastructure {
pub state: AppState,
- pub ap_service: Arc,
+ pub ap_service: Arc,
}
struct NoOpEventPublisher;
@@ -72,7 +72,9 @@ pub async fn build(cfg: &Config) -> Infrastructure {
};
// 3. ActivityPub federation
- let ap_service = Arc::new(
+ let connections_repo =
+ Arc::new(PgRemoteActorConnectionRepository::new(pool.clone()));
+ let raw_ap_service = Arc::new(
ActivityPubService::builder(
Arc::new(PostgresFederationRepository::new(pool.clone())),
Arc::new(PostgresApUserRepository::new(
@@ -86,7 +88,6 @@ pub async fn build(cfg: &Config) -> Infrastructure {
Arc::new(postgres::tag::PgTagRepository::new(pool.clone())),
)),
cfg.base_url.clone(),
- Arc::new(PgRemoteActorConnectionRepository::new(pool.clone())),
)
.allow_registration(cfg.allow_registration)
.software_name("thoughts")
@@ -95,6 +96,7 @@ pub async fn build(cfg: &Config) -> Infrastructure {
.await
.expect("Failed to build ActivityPubService"),
);
+ let ap_service = Arc::new(ApFederationAdapter::new(raw_ap_service, connections_repo));
// 4. Application state
let state = AppState {
diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml
index 3524676..4f04741 100644
--- a/crates/worker/Cargo.toml
+++ b/crates/worker/Cargo.toml
@@ -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.0" }
+k-ap = { git = "https://git.gabrielkaszewski.dev/GKaszewski/k-ap.git", tag = "v0.1.2" }
activitypub = { workspace = true }
postgres = { workspace = true }
postgres-federation = { workspace = true }
diff --git a/crates/worker/src/factory.rs b/crates/worker/src/factory.rs
index cc8022e..0604a04 100644
--- a/crates/worker/src/factory.rs
+++ b/crates/worker/src/factory.rs
@@ -3,7 +3,7 @@ use postgres::remote_actor_connections::PgRemoteActorConnectionRepository;
use sqlx::PgPool;
use std::sync::Arc;
-use activitypub::ThoughtsObjectHandler;
+use activitypub::{ApFederationAdapter, ThoughtsObjectHandler};
use activitypub::{ActivityPubRepository, OutboundFederationPort};
use k_ap::ActivityPubService;
use application::services::{FederationEventService, NotificationEventService};
@@ -39,7 +39,9 @@ pub async fn build(database_url: &str, base_url: &str, nats_url: &str) -> Worker
));
// ActivityPub service (for federation fan-out)
- let ap_service = Arc::new(
+ let connections_repo_worker =
+ Arc::new(PgRemoteActorConnectionRepository::new(pool.clone()));
+ let raw_ap_service = Arc::new(
ActivityPubService::builder(
Arc::new(PostgresFederationRepository::new(pool.clone())),
Arc::new(PostgresApUserRepository::new(
@@ -53,13 +55,13 @@ pub async fn build(database_url: &str, base_url: &str, nats_url: &str) -> Worker
Arc::new(postgres::tag::PgTagRepository::new(pool.clone())),
)),
base_url,
- Arc::new(PgRemoteActorConnectionRepository::new(pool.clone())),
)
.software_name("thoughts")
.build()
.await
.expect("ActivityPubService build failed"),
);
+ let ap_service = Arc::new(ApFederationAdapter::new(raw_ap_service, connections_repo_worker));
let ap_outbound = ap_service.clone() as Arc;
let ap_repo_worker =
Arc::new(PgActivityPubRepository::new(pool.clone())) as Arc;