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

refactor(domain): remove FetchRemoteActorPosts/FetchActorConnections from DomainEvent; add FederationSchedulerPort

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-05-15 13:28:19 +02:00
parent e935c8973e
commit a902154777
13 changed files with 1310 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
crates/adapters/activitypub-base/src/service.rs
View File

@@ -1592,6 +1592,39 @@ impl domain::ports::OutboundFederationPort for ActivityPubService {
}
}
#[async_trait::async_trait]
impl domain::ports::FederationSchedulerPort for ActivityPubService {
async fn schedule_actor_posts_fetch(
&self,
actor_ap_url: &str,
outbox_url: &str,
) -> Result<(), domain::errors::DomainError> {
tracing::debug!(
actor = actor_ap_url,
outbox = outbox_url,
"schedule_actor_posts_fetch: deferred"
);
Ok(())
}
async fn schedule_connections_fetch(
&self,
actor_ap_url: &str,
collection_url: &str,
connection_type: &str,
page: u32,
) -> Result<(), domain::errors::DomainError> {
tracing::debug!(
actor = actor_ap_url,
collection = collection_url,
connection_type,
page,
"schedule_connections_fetch: deferred"
);
Ok(())
}
}
#[async_trait::async_trait]
impl domain::ports::FederationActionPort for ActivityPubService {
async fn lookup_actor(

58
crates/adapters/event-payload/src/lib.rs
View File

@@ -71,16 +71,6 @@ pub enum EventPayload {
ProfileUpdated {
user_id: String,
},
FetchRemoteActorPosts {
actor_ap_url: String,
outbox_url: String,
},
FetchActorConnections {
actor_ap_url: String,
collection_url: String,
connection_type: String,
page: u32,
},
MentionReceived {
thought_id: String,
mentioned_user_id: String,
@@ -107,8 +97,6 @@ impl EventPayload {
Self::UserUnblocked { .. } => "users.unblocked",
Self::UserRegistered { .. } => "users.registered",
Self::ProfileUpdated { .. } => "users.profile_updated",
Self::FetchRemoteActorPosts { .. } => "federation.fetch_outbox",
Self::FetchActorConnections { .. } => "federation.fetch_connections",
Self::MentionReceived { .. } => "mentions.received",
}
}
@@ -222,24 +210,6 @@ impl From<&DomainEvent> for EventPayload {
DomainEvent::ProfileUpdated { user_id } => Self::ProfileUpdated {
user_id: user_id.to_string(),
},
DomainEvent::FetchRemoteActorPosts {
actor_ap_url,
outbox_url,
} => Self::FetchRemoteActorPosts {
actor_ap_url: actor_ap_url.clone(),
outbox_url: outbox_url.clone(),
},
DomainEvent::FetchActorConnections {
actor_ap_url,
collection_url,
connection_type,
page,
} => Self::FetchActorConnections {
actor_ap_url: actor_ap_url.clone(),
collection_url: collection_url.clone(),
connection_type: connection_type.clone(),
page: *page,
},
DomainEvent::MentionReceived {
thought_id,
mentioned_user_id,
@@ -370,24 +340,6 @@ impl TryFrom<EventPayload> for DomainEvent {
EventPayload::ProfileUpdated { user_id } => DomainEvent::ProfileUpdated {
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
},
EventPayload::FetchRemoteActorPosts {
actor_ap_url,
outbox_url,
} => DomainEvent::FetchRemoteActorPosts {
actor_ap_url,
outbox_url,
},
EventPayload::FetchActorConnections {
actor_ap_url,
collection_url,
connection_type,
page,
} => DomainEvent::FetchActorConnections {
actor_ap_url,
collection_url,
connection_type,
page,
},
EventPayload::MentionReceived {
thought_id,
mentioned_user_id,
@@ -481,16 +433,6 @@ mod tests {
EventPayload::UserRegistered {
user_id: "a".into(),
},
EventPayload::FetchRemoteActorPosts {
actor_ap_url: "https://mastodon.social/users/alice".into(),
outbox_url: "https://mastodon.social/users/alice/outbox".into(),
},
EventPayload::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,
},
];
let mut subjects: Vec<&str> = samples.iter().map(|p| p.subject()).collect();
subjects.sort();

139
crates/application/src/services/federation_event.rs
View File

@@ -12,9 +12,7 @@ pub struct FederationEventService {
pub users: Arc<dyn UserRepository>,
pub ap: Arc<dyn OutboundFederationPort>,
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 {
@@ -148,112 +146,6 @@ impl FederationEventService {
.await
}
DomainEvent::FetchRemoteActorPosts {
actor_ap_url,
outbox_url,
} => {
let notes = match self
.federation_action
.fetch_outbox_page(outbox_url, 1)
.await
{
Ok(n) => n,
Err(e) => {
tracing::warn!(outbox_url, error = %e, "failed to fetch remote outbox");
return Ok(());
}
};
let actor_url = url::Url::parse(actor_ap_url)
.map_err(|e| DomainError::ExternalService(e.to_string()))?;
let author_id = self.ap_repo.intern_remote_actor(&actor_url).await?;
// Resolve and cache display info so thought cards show proper names.
let profiles = self
.federation_action
.resolve_actor_profiles(vec![actor_ap_url.clone()])
.await;
if let Some(profile) = profiles.into_iter().next() {
let _ = self
.ap_repo
.update_remote_actor_display(
&author_id,
profile.display_name.as_deref(),
profile.avatar_url.as_deref(),
)
.await;
}
for note in notes {
let ap_id = match url::Url::parse(&note.ap_id) {
Ok(u) => u,
Err(_) => continue,
};
let _ = self
.ap_repo
.accept_note(
&ap_id,
&author_id,
&note.content,
note.published,
note.sensitive,
note.content_warning,
"public",
None,
)
.await;
}
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(())
}
DomainEvent::LikeAdded {
like_id: _,
user_id,
@@ -438,9 +330,7 @@ mod tests {
users: Arc::new(store.clone()),
ap: spy,
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()),
}
}
@@ -772,35 +662,6 @@ mod tests {
assert!(spy.updated.lock().unwrap().is_empty());
}
#[tokio::test]
async fn fetch_remote_actor_posts_is_noop_when_outbox_empty() {
let store = TestStore::default();
let spy = Arc::new(SpyPort::default());
svc(&store, spy.clone())
.process(&DomainEvent::FetchRemoteActorPosts {
actor_ap_url: "https://mastodon.social/users/alice".into(),
outbox_url: "https://mastodon.social/users/alice/outbox".into(),
})
.await
.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();
}
#[tokio::test]
async fn like_added_local_user_remote_thought_broadcasts_like() {
let store = TestStore::default();

25
crates/application/src/use_cases/federation_management.rs
View File

@@ -1,14 +1,13 @@
use domain::{
errors::DomainError,
events::DomainEvent,
models::{
actor_connection_summary::ActorConnectionSummary,
feed::{FeedEntry, PageParams, Paginated},
remote_actor::RemoteActor,
},
ports::{
ActivityPubRepository, EventPublisher, FederationActionPort, FeedRepository,
FollowRepository, RemoteActorConnectionRepository, UserRepository,
ActivityPubRepository, EventPublisher, FederationActionPort, FederationSchedulerPort,
FeedRepository, FollowRepository, RemoteActorConnectionRepository, UserRepository,
},
value_objects::UserId,
};
@@ -75,7 +74,7 @@ pub async fn get_remote_actor_posts(
federation: &dyn FederationActionPort,
ap_repo: &dyn ActivityPubRepository,
feed: &dyn FeedRepository,
events: &dyn EventPublisher,
scheduler: &dyn FederationSchedulerPort,
handle: &str,
page: PageParams,
viewer_id: Option<&UserId>,
@@ -88,11 +87,8 @@ pub async fn get_remote_actor_posts(
};
let result = feed.user_feed(&author_id, &page, viewer_id).await?;
if let Some(outbox_url) = actor.outbox_url {
let _ = events
.publish(&DomainEvent::FetchRemoteActorPosts {
actor_ap_url: actor.url,
outbox_url,
})
let _ = scheduler
.schedule_actor_posts_fetch(&actor.url, &outbox_url)
.await;
}
Ok(result)
@@ -103,7 +99,7 @@ const ACTOR_CONNECTIONS_CACHE_TTL_SECS: i64 = 3600;
pub async fn get_actor_connections_page(
federation: &dyn FederationActionPort,
connections: &dyn RemoteActorConnectionRepository,
events: &dyn EventPublisher,
scheduler: &dyn FederationSchedulerPort,
handle: &str,
connection_type: &str,
page: u32,
@@ -128,13 +124,8 @@ pub async fn get_actor_connections_page(
}
};
if stale {
let _ = events
.publish(&DomainEvent::FetchActorConnections {
actor_ap_url: actor.url,
collection_url,
connection_type: connection_type.to_string(),
page,
})
let _ = scheduler
.schedule_connections_fetch(&actor.url, &collection_url, connection_type, page)
.await;
}
let has_more = items.len() >= PAGE_SIZE;

1
crates/bootstrap/src/factory.rs
View File

@@ -116,6 +116,7 @@ pub async fn build(cfg: &Config) -> Infrastructure {
federation: ap_service.clone() as Arc<dyn domain::ports::FederationActionPort>,
ap_repo: Arc::new(PgActivityPubRepository::new(pool.clone())),
remote_actor_connections: Arc::new(PgRemoteActorConnectionRepository::new(pool.clone())),
federation_scheduler: ap_service.clone() as Arc<dyn domain::ports::FederationSchedulerPort>,
};
Infrastructure { state, ap_service }

10
crates/domain/src/events.rs
View File

@@ -63,16 +63,6 @@ pub enum DomainEvent {
ProfileUpdated {
user_id: UserId,
},
FetchRemoteActorPosts {
actor_ap_url: String,
outbox_url: String,
},
FetchActorConnections {
actor_ap_url: String,
collection_url: String,
connection_type: String,
page: u32,
},
MentionReceived {
thought_id: ThoughtId,
mentioned_user_id: UserId,

17
crates/domain/src/ports.rs
View File

@@ -497,3 +497,20 @@ pub trait OutboundFederationPort: Send + Sync {
/// Fan out an Update(Actor) to all accepted followers after a profile change.
async fn broadcast_actor_update(&self, user_id: &UserId) -> Result<(), DomainError>;
}
#[async_trait]
pub trait FederationSchedulerPort: Send + Sync {
async fn schedule_actor_posts_fetch(
&self,
actor_ap_url: &str,
outbox_url: &str,
) -> Result<(), DomainError>;
async fn schedule_connections_fetch(
&self,
actor_ap_url: &str,
collection_url: &str,
connection_type: &str,
page: u32,
) -> Result<(), DomainError>;
}

16
crates/domain/src/testing.rs
View File

@@ -903,6 +903,22 @@ impl ActivityPubRepository for TestStore {
}
}
#[async_trait]
impl FederationSchedulerPort for TestStore {
async fn schedule_actor_posts_fetch(&self, _: &str, _: &str) -> Result<(), DomainError> {
Ok(())
}
async fn schedule_connections_fetch(
&self,
_: &str,
_: &str,
_: &str,
_: u32,
) -> Result<(), DomainError> {
Ok(())
}
}
#[async_trait]
impl EventPublisher for TestStore {
async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {

4
crates/presentation/src/handlers/federation_actors.rs
View File

@@ -29,7 +29,7 @@ pub async fn remote_actor_posts_handler(
&*s.federation,
&*s.ap_repo,
&*s.feed,
&*s.events,
&*s.federation_scheduler,
&handle,
page,
viewer.as_ref(),
@@ -68,7 +68,7 @@ async fn actor_connections_handler(
let (items, has_more) = get_actor_connections_page(
&*s.federation,
&*s.remote_actor_connections,
&*s.events,
&*s.federation_scheduler,
&handle,
connection_type,
page,

1
crates/presentation/src/state.rs
View File

@@ -22,4 +22,5 @@ pub struct AppState {
pub federation: Arc<dyn FederationActionPort>,
pub ap_repo: Arc<dyn ActivityPubRepository>,
pub remote_actor_connections: Arc<dyn RemoteActorConnectionRepository>,
pub federation_scheduler: Arc<dyn FederationSchedulerPort>,
}

1
crates/presentation/src/testing.rs
View File

@@ -51,5 +51,6 @@ pub fn make_state() -> AppState {
federation: store.clone(),
ap_repo: store.clone(),
remote_actor_connections: store.clone(),
federation_scheduler: store.clone(),
}
}

8
crates/worker/src/factory.rs
View File

@@ -4,9 +4,8 @@ use std::sync::Arc;
use activitypub::ThoughtsObjectHandler;
use activitypub_base::ActivityPubService;
use application::services::{FederationEventService, NotificationEventService};
use domain::ports::{ActivityPubRepository, FederationActionPort, OutboundFederationPort};
use domain::ports::{ActivityPubRepository, OutboundFederationPort};
use postgres::activitypub::PgActivityPubRepository;
use postgres::remote_actor_connections::PgRemoteActorConnectionRepository;
use postgres_federation::{PostgresApUserRepository, PostgresFederationRepository};
use crate::handlers::{FederationHandler, NotificationHandler};
@@ -58,11 +57,8 @@ pub async fn build(
.expect("ActivityPubService build failed"),
);
let ap_outbound = ap_service.clone() as Arc<dyn OutboundFederationPort>;
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 {
@@ -74,9 +70,7 @@ pub async fn build(
users,
ap: ap_outbound,
base_url: base_url.to_string(),
federation_action: ap_federation,
ap_repo: ap_repo_worker,
remote_actor_connections: actor_connections,
});
// Thin handlers

1230
docs/superpowers/plans/2026-05-15-domain-application-refactor.md Normal file
View File

File diff suppressed because it is too large Load Diff