feat(worker): handle FetchActorConnections — resolve and cache remote actor connections
This commit is contained in:
@@ -14,6 +14,7 @@ pub struct FederationEventService {
|
||||
pub base_url: String,
|
||||
pub federation_action: Arc<dyn domain::ports::FederationActionPort>,
|
||||
pub ap_repo: Arc<dyn ActivityPubRepository>,
|
||||
pub remote_actor_connections: Arc<dyn domain::ports::RemoteActorConnectionRepository>,
|
||||
}
|
||||
|
||||
impl FederationEventService {
|
||||
@@ -157,6 +158,52 @@ impl FederationEventService {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
DomainEvent::FetchActorConnections {
|
||||
actor_ap_url,
|
||||
collection_url,
|
||||
connection_type,
|
||||
page,
|
||||
} => {
|
||||
let urls = match self
|
||||
.federation_action
|
||||
.fetch_actor_urls_from_collection(collection_url)
|
||||
.await
|
||||
{
|
||||
Ok(u) => u,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
collection_url,
|
||||
error = %e,
|
||||
"failed to fetch actor connections collection"
|
||||
);
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
if urls.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let summaries = self.federation_action.resolve_actor_profiles(urls).await;
|
||||
|
||||
if summaries.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
count = summaries.len(),
|
||||
connection_type,
|
||||
actor = actor_ap_url,
|
||||
"caching actor connections"
|
||||
);
|
||||
|
||||
self.remote_actor_connections
|
||||
.upsert_connections(actor_ap_url, connection_type, *page, &summaries)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
@@ -255,6 +302,7 @@ mod tests {
|
||||
base_url: "https://example.com".to_string(),
|
||||
federation_action: Arc::new(store.clone()),
|
||||
ap_repo: Arc::new(store.clone()),
|
||||
remote_actor_connections: Arc::new(store.clone()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -592,4 +640,19 @@ mod tests {
|
||||
.unwrap();
|
||||
// TestStore.fetch_outbox_page returns Ok(vec![]) — no notes, no error
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fetch_actor_connections_is_noop_when_collection_empty() {
|
||||
let store = TestStore::default();
|
||||
let spy = Arc::new(SpyPort::default());
|
||||
svc(&store, spy.clone())
|
||||
.process(&DomainEvent::FetchActorConnections {
|
||||
actor_ap_url: "https://mastodon.social/users/alice".into(),
|
||||
collection_url: "https://mastodon.social/users/alice/followers".into(),
|
||||
connection_type: "followers".into(),
|
||||
page: 1,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ use activitypub_base::ActivityPubService;
|
||||
use application::services::{FederationEventService, NotificationEventService};
|
||||
use domain::ports::{ActivityPubRepository, FederationActionPort, OutboundFederationPort};
|
||||
use postgres::activitypub::PgActivityPubRepository;
|
||||
use postgres::remote_actor_connections::PgRemoteActorConnectionRepository;
|
||||
use postgres_federation::{PostgresApUserRepository, PostgresFederationRepository};
|
||||
|
||||
use crate::handlers::{FederationHandler, NotificationHandler};
|
||||
@@ -59,6 +60,8 @@ pub async fn build(
|
||||
let ap_federation = ap_service.clone() as Arc<dyn FederationActionPort>;
|
||||
let ap_repo_worker =
|
||||
Arc::new(PgActivityPubRepository::new(pool.clone())) as Arc<dyn ActivityPubRepository>;
|
||||
let actor_connections = Arc::new(PgRemoteActorConnectionRepository::new(pool.clone()))
|
||||
as Arc<dyn domain::ports::RemoteActorConnectionRepository>;
|
||||
|
||||
// Application services
|
||||
let notification_svc = Arc::new(NotificationEventService {
|
||||
@@ -72,6 +75,7 @@ pub async fn build(
|
||||
base_url: base_url.to_string(),
|
||||
federation_action: ap_federation,
|
||||
ap_repo: ap_repo_worker,
|
||||
remote_actor_connections: actor_connections,
|
||||
});
|
||||
|
||||
// Thin handlers
|
||||
|
||||
Reference in New Issue
Block a user