GKaszewski/movies-diary
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: AP bugs — backfill mapping, review activity type, also_known_as parse

Browse Source 
- BackfillRequested now maps to BackfillFollower domain event (not FollowAccepted);
  worker calls run_backfill_for_follower to send LOCAL content to new follower inbox,
  instead of incorrectly trying to import from an inbox URL as if it were an outbox
- reviews broadcast as Create activity instead of Add (semantically correct)
- also_known_as JSON parse failure logs warning + preserves raw string as single-element
  vec instead of silently returning empty
This commit is contained in:
Gabriel Kaszewski
2026-05-29 10:54:11 +02:00
parent 624cfe5799
commit 36d15e1344
9 changed files with 71 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/adapters/activitypub/src/event_handler.rs
View File

@@ -123,7 +123,7 @@ impl ActivityPubEventHandler {
let json = serde_json::to_value(obj)?;
self.ap_service
.broadcast_add_to_followers(user_id.value(), ap_id, json)
.broadcast_create_note(user_id.value(), json, ApVisibility::Public, vec![])
.await?;
Ok(())

12
crates/adapters/activitypub/src/federation_event_bridge.rs
View File

@@ -22,16 +22,10 @@ impl k_ap::EventPublisher for FederationEventBridge {
owner_user_id,
follower_inbox_url,
} => {
tracing::info!(
owner = %owner_user_id,
inbox = %follower_inbox_url,
"federation BackfillRequested → FollowAccepted"
);
self.domain_publisher
.publish(&DomainEvent::FollowAccepted {
local_user_id: UserId::from_uuid(owner_user_id),
remote_actor_url: follower_inbox_url.clone(),
outbox_url: follower_inbox_url,
.publish(&DomainEvent::BackfillFollower {
owner_user_id: UserId::from_uuid(owner_user_id),
follower_inbox_url,
})
.await
.map_err(|e| anyhow::anyhow!(e.to_string()))

7
crates/adapters/activitypub/src/port.rs
View File

@@ -34,6 +34,7 @@ pub trait ActivityPubPort: Send + Sync {
async fn import_remote_outbox(&self, outbox_url: &str, actor_url: &str) -> anyhow::Result<()>;
async fn followers_collection_json(&self, user_id: Uuid, page: Option<u32>) -> anyhow::Result<String>;
async fn following_collection_json(&self, user_id: Uuid, page: Option<u32>) -> anyhow::Result<String>;
async fn run_backfill_for_follower(&self, owner_user_id: Uuid, follower_inbox_url: String) -> anyhow::Result<()>;
}
#[async_trait]
@@ -109,6 +110,9 @@ impl ActivityPubPort for ActivityPubService {
async fn following_collection_json(&self, user_id: Uuid, page: Option<u32>) -> anyhow::Result<String> {
self.following_collection_json(user_id, page).await
}
async fn run_backfill_for_follower(&self, owner_user_id: Uuid, follower_inbox_url: String) -> anyhow::Result<()> {
self.run_backfill_for_follower(owner_user_id, follower_inbox_url).await
}
}
pub struct NoopActivityPubService;
@@ -175,4 +179,7 @@ impl ActivityPubPort for NoopActivityPubService {
async fn following_collection_json(&self, _: Uuid, _: Option<u32>) -> anyhow::Result<String> {
Ok(String::new())
}
async fn run_backfill_for_follower(&self, _: Uuid, _: String) -> anyhow::Result<()> {
Ok(())
}
}

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

@@ -63,6 +63,10 @@ pub enum EventPayload {
remote_actor_url: String,
outbox_url: String,
},
BackfillFollower {
owner_user_id: String,
follower_inbox_url: String,
},
}
impl EventPayload {
@@ -79,6 +83,7 @@ impl EventPayload {
EventPayload::WatchlistEntryAdded { .. } => "WatchlistEntryAdded",
EventPayload::WatchlistEntryRemoved { .. } => "WatchlistEntryRemoved",
EventPayload::FollowAccepted { .. } => "FollowAccepted",
EventPayload::BackfillFollower { .. } => "BackfillFollower",
}
}
}
@@ -181,6 +186,13 @@ impl From<&DomainEvent> for EventPayload {
remote_actor_url: remote_actor_url.clone(),
outbox_url: outbox_url.clone(),
},
DomainEvent::BackfillFollower {
owner_user_id,
follower_inbox_url,
} => EventPayload::BackfillFollower {
owner_user_id: owner_user_id.value().to_string(),
follower_inbox_url: follower_inbox_url.clone(),
},
}
}
}
@@ -281,6 +293,13 @@ impl TryFrom<EventPayload> for DomainEvent {
remote_actor_url,
outbox_url,
}),
EventPayload::BackfillFollower {
owner_user_id,
follower_inbox_url,
} => Ok(DomainEvent::BackfillFollower {
owner_user_id: UserId::from_uuid(parse_uuid(&owner_user_id, "owner_user_id")?),
follower_inbox_url,
}),
}
}
}

1
crates/adapters/nats/src/subject.rs
View File

@@ -13,6 +13,7 @@ pub fn event_to_subject(prefix: &str, event: &DomainEvent) -> String {
DomainEvent::WatchlistEntryAdded { .. } => "watchlist.entry.added",
DomainEvent::WatchlistEntryRemoved { .. } => "watchlist.entry.removed",
DomainEvent::FollowAccepted { .. } => "follow.accepted",
DomainEvent::BackfillFollower { .. } => "backfill.follower",
};
format!("{prefix}.{suffix}")
}

7
crates/adapters/postgres-federation/src/lib.rs
View File

@@ -58,7 +58,12 @@ fn pg_remote_actor(row: &sqlx::postgres::PgRow, url_col: &str) -> RemoteActor {
.try_get::<Option<String>, _>("also_known_as")
.ok()
.flatten()
.map(|s| serde_json::from_str(&s).unwrap_or_default())
.map(|s| {
serde_json::from_str::<Vec<String>>(&s).unwrap_or_else(|e| {
tracing::warn!(raw = %s, error = %e, "failed to parse also_known_as JSON");
vec![s]
})
})
.unwrap_or_default(),
}
}

7
crates/adapters/sqlite-federation/src/lib.rs
View File

@@ -58,7 +58,12 @@ fn remote_actor_from_row(row: &sqlx::sqlite::SqliteRow, url_col: &str) -> Remote
.try_get::<Option<String>, _>("also_known_as")
.ok()
.flatten()
.map(|s| serde_json::from_str(&s).unwrap_or_default())
.map(|s| {
serde_json::from_str::<Vec<String>>(&s).unwrap_or_else(|e| {
tracing::warn!(raw = %s, error = %e, "failed to parse also_known_as JSON");
vec![s]
})
})
.unwrap_or_default(),
}
}

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

@@ -61,6 +61,10 @@ pub enum DomainEvent {
remote_actor_url: String,
outbox_url: String,
},
BackfillFollower {
owner_user_id: UserId,
follower_inbox_url: String,
},
}
#[async_trait]

37
crates/worker/src/follow_backfill_handler.rs
View File

@@ -10,18 +10,29 @@ pub struct FollowBackfillHandler {
#[async_trait]
impl EventHandler for FollowBackfillHandler {
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
let DomainEvent::FollowAccepted {
remote_actor_url,
outbox_url,
..
} = event
else {
return Ok(());
};
tracing::info!(actor = %remote_actor_url, outbox = %outbox_url, "starting outbox backfill");
self.ap_service
.import_remote_outbox(outbox_url, remote_actor_url)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
match event {
DomainEvent::FollowAccepted {
remote_actor_url,
outbox_url,
..
} => {
tracing::info!(actor = %remote_actor_url, outbox = %outbox_url, "importing remote outbox");
self.ap_service
.import_remote_outbox(outbox_url, remote_actor_url)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
DomainEvent::BackfillFollower {
owner_user_id,
follower_inbox_url,
} => {
tracing::info!(owner = %owner_user_id.value(), inbox = %follower_inbox_url, "backfilling local content to new follower");
self.ap_service
.run_backfill_for_follower(owner_user_id.value(), follower_inbox_url.clone())
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
_ => Ok(()),
}
}
}