Persistent, performant, reliable federation queue

Cargo.toml

@@ -54,6 +54,7 @@ members = [
   "crates/db_views_actor",
   "crates/db_views_actor",
   "crates/routes",
+  "crates/federate",
 ]
 
 [workspace.dependencies]
@@ -69,7 +70,7 @@ lemmy_db_views_actor = { version = "=0.18.1", path = "./crates/db_views_actor" }
 lemmy_db_views_moderator = { version = "=0.18.1", path = "./crates/db_views_moderator" }
 activitypub_federation = { version = "0.4.6", default-features = false, features = [
   "actix-web",
-] }
+], git = "https://github.com/phiresky/activitypub-federation-rust/", branch = "raw-sending" }
 diesel = "2.1.0"
 diesel_migrations = "2.1.0"
 diesel-async = "0.3.1"
@@ -88,7 +89,6 @@ tracing-error = "0.2.0"
 tracing-log = "0.1.3"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 url = { version = "2.4.0", features = ["serde"] }
-url_serde = "0.2.0"
 reqwest = { version = "0.11.18", features = ["json", "blocking", "gzip"] }
 reqwest-middleware = "0.2.2"
 reqwest-tracing = "0.4.5"
@@ -119,7 +119,6 @@ futures = "0.3.28"
 http = "0.2.9"
 percent-encoding = "2.3.0"
 rosetta-i18n = "0.1.3"
-rand = "0.8.5"
 opentelemetry = { version = "0.19.0", features = ["rt-tokio"] }
 tracing-opentelemetry = { version = "0.19.0" }
 ts-rs = { version = "6.2", features = ["serde-compat", "chrono-impl"] }
@@ -167,3 +166,5 @@ tokio-postgres-rustls = { workspace = true }
 chrono = { workspace = true }
 prometheus = { version = "0.13.3", features = ["process"], optional = true }
 actix-web-prom = { version = "0.6.0", optional = true }
+clap = { version = "4.3.19", features = ["derive"] }
+lemmy_federate = { version = "0.18.1", path = "crates/federate" }

crates/apub/src/activities/block/block_user.rs

@@ -10,7 +10,7 @@ use crate::{
   },
   activity_lists::AnnouncableActivities,
   insert_received_activity,
-  objects::{instance::remote_instance_inboxes, person::ApubPerson},
+  objects::person::ApubPerson,
   protocol::activities::block::block_user::BlockUser,
 };
 use activitypub_federation::{
@@ -27,6 +27,7 @@ use lemmy_api_common::{
 };
 use lemmy_db_schema::{
   source::{
+    activity::ActivitySendTargets,
     community::{
       CommunityFollower,
       CommunityFollowerForm,
@@ -97,12 +98,12 @@ impl BlockUser {
 
     match target {
       SiteOrCommunity::Site(_) => {
-        let inboxes = remote_instance_inboxes(&mut context.pool()).await?;
+        let inboxes = ActivitySendTargets::to_all_instances();
         send_lemmy_activity(context, block, mod_, inboxes, false).await
       }
       SiteOrCommunity::Community(c) => {
         let activity = AnnouncableActivities::BlockUser(block);
-        let inboxes = vec![user.shared_inbox_or_inbox()];
+        let inboxes = ActivitySendTargets::to_inbox(user.shared_inbox_or_inbox());
         send_activity_in_community(activity, mod_, c, inboxes, true, context).await
       }
     }

crates/apub/src/activities/block/undo_block_user.rs

@@ -8,7 +8,7 @@ use crate::{
   },
   activity_lists::AnnouncableActivities,
   insert_received_activity,
-  objects::{instance::remote_instance_inboxes, person::ApubPerson},
+  objects::person::ApubPerson,
   protocol::activities::block::{block_user::BlockUser, undo_block_user::UndoBlockUser},
 };
 use activitypub_federation::{
@@ -20,6 +20,7 @@ use activitypub_federation::{
 use lemmy_api_common::{context::LemmyContext, utils::sanitize_html_opt};
 use lemmy_db_schema::{
   source::{
+    activity::ActivitySendTargets,
     community::{CommunityPersonBan, CommunityPersonBanForm},
     moderator::{ModBan, ModBanForm, ModBanFromCommunity, ModBanFromCommunityForm},
     person::{Person, PersonUpdateForm},
@@ -59,10 +60,10 @@ impl UndoBlockUser {
       audience,
     };
 
-    let mut inboxes = vec![user.shared_inbox_or_inbox()];
+    let mut inboxes = ActivitySendTargets::to_inbox(user.shared_inbox_or_inbox());
     match target {
       SiteOrCommunity::Site(_) => {
-        inboxes.append(&mut remote_instance_inboxes(&mut context.pool()).await?);
+        inboxes.set_all_instances();
         send_lemmy_activity(context, undo, mod_, inboxes, false).await
       }
       SiteOrCommunity::Community(c) => {

crates/apub/src/activities/community/announce.rs

@@ -21,6 +21,7 @@ use activitypub_federation::{
   traits::{ActivityHandler, Actor},
 };
 use lemmy_api_common::context::LemmyContext;
+use lemmy_db_schema::source::activity::ActivitySendTargets;
 use lemmy_utils::error::{LemmyError, LemmyErrorType};
 use serde_json::Value;
 use url::Url;
@@ -94,7 +95,7 @@ impl AnnounceActivity {
     context: &Data<LemmyContext>,
   ) -> Result<(), LemmyError> {
     let announce = AnnounceActivity::new(object.clone(), community, context)?;
-    let inboxes = community.get_follower_inboxes(context).await?;
+    let inboxes = ActivitySendTargets::to_local_community_followers(community.id);
     send_lemmy_activity(context, announce, community, inboxes.clone(), false).await?;
 
     // Pleroma and Mastodon can't handle activities like Announce/Create/Page. So for

crates/apub/src/activities/community/collection_add.rs

@@ -28,6 +28,7 @@ use lemmy_db_schema::{
   impls::community::CollectionType,
   newtypes::{CommunityId, PersonId},
   source::{
+    activity::ActivitySendTargets,
     community::{Community, CommunityModerator, CommunityModeratorForm},
     moderator::{ModAddCommunity, ModAddCommunityForm},
     person::Person,
@@ -62,7 +63,7 @@ impl CollectionAdd {
     };
 
     let activity = AnnouncableActivities::CollectionAdd(add);
-    let inboxes = vec![added_mod.shared_inbox_or_inbox()];
+    let inboxes = ActivitySendTargets::to_inbox(added_mod.shared_inbox_or_inbox());
     send_activity_in_community(activity, actor, community, inboxes, true, context).await
   }
 
@@ -87,7 +88,15 @@ impl CollectionAdd {
       audience: Some(community.id().into()),
     };
     let activity = AnnouncableActivities::CollectionAdd(add);
-    send_activity_in_community(activity, actor, community, vec![], true, context).await
+    send_activity_in_community(
+      activity,
+      actor,
+      community,
+      ActivitySendTargets::empty(),
+      true,
+      context,
+    )
+    .await
   }
 }
 

crates/apub/src/activities/community/collection_remove.rs

@@ -24,6 +24,7 @@ use lemmy_api_common::{
 use lemmy_db_schema::{
   impls::community::CollectionType,
   source::{
+    activity::ActivitySendTargets,
     community::{Community, CommunityModerator, CommunityModeratorForm},
     moderator::{ModAddCommunity, ModAddCommunityForm},
     post::{Post, PostUpdateForm},
@@ -57,7 +58,7 @@ impl CollectionRemove {
     };
 
     let activity = AnnouncableActivities::CollectionRemove(remove);
-    let inboxes = vec![removed_mod.shared_inbox_or_inbox()];
+    let inboxes = ActivitySendTargets::to_inbox(removed_mod.shared_inbox_or_inbox());
     send_activity_in_community(activity, actor, community, inboxes, true, context).await
   }
 
@@ -82,7 +83,15 @@ impl CollectionRemove {
       audience: Some(community.id().into()),
     };
     let activity = AnnouncableActivities::CollectionRemove(remove);
-    send_activity_in_community(activity, actor, community, vec![], true, context).await
+    send_activity_in_community(
+      activity,
+      actor,
+      community,
+      ActivitySendTargets::empty(),
+      true,
+      context,
+    )
+    .await
   }
 }
 

crates/apub/src/activities/community/lock_page.rs

@@ -24,6 +24,7 @@ use activitypub_federation::{
 use lemmy_api_common::context::LemmyContext;
 use lemmy_db_schema::{
   source::{
+    activity::ActivitySendTargets,
     community::Community,
     person::Person,
     post::{Post, PostUpdateForm},
@@ -141,6 +142,14 @@ pub(crate) async fn send_lock_post(
     };
     AnnouncableActivities::UndoLockPost(undo)
   };
-  send_activity_in_community(activity, &actor.into(), &community, vec![], true, &context).await?;
+  send_activity_in_community(
+    activity,
+    &actor.into(),
+    &community,
+    ActivitySendTargets::empty(),
+    true,
+    &context,
+  )
+  .await?;
   Ok(())
 }

crates/apub/src/activities/community/mod.rs

@@ -6,9 +6,8 @@ use crate::{
 };
 use activitypub_federation::{config::Data, traits::Actor};
 use lemmy_api_common::context::LemmyContext;
-use lemmy_db_schema::source::person::PersonFollower;
+use lemmy_db_schema::source::{activity::ActivitySendTargets, person::PersonFollower};
 use lemmy_utils::error::LemmyError;
-use url::Url;
 
 pub mod announce;
 pub mod collection_add;
@@ -34,7 +33,7 @@ pub(crate) async fn send_activity_in_community(
   activity: AnnouncableActivities,
   actor: &ApubPerson,
   community: &ApubCommunity,
-  extra_inboxes: Vec<Url>,
+  extra_inboxes: ActivitySendTargets,
   is_mod_action: bool,
   context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
@@ -43,8 +42,8 @@ pub(crate) async fn send_activity_in_community(
 
   // send to user followers
   if !is_mod_action {
-    inboxes.extend(
-      &mut PersonFollower::list_followers(&mut context.pool(), actor.id)
+    inboxes.add_inboxes(
+      PersonFollower::list_followers(&mut context.pool(), actor.id)
         .await?
         .into_iter()
         .map(|p| ApubPerson(p).shared_inbox_or_inbox()),
@@ -56,7 +55,7 @@ pub(crate) async fn send_activity_in_community(
     AnnounceActivity::send(activity.clone().try_into()?, community, context).await?;
   } else {
     // send to the community, which will then forward to followers
-    inboxes.push(community.shared_inbox_or_inbox());
+    inboxes.add_inbox(community.shared_inbox_or_inbox());
   }
 
   send_lemmy_activity(context, activity.clone(), actor, inboxes, false).await?;

crates/apub/src/activities/community/report.rs

@@ -14,6 +14,7 @@ use activitypub_federation::{
 use lemmy_api_common::{context::LemmyContext, utils::sanitize_html};
 use lemmy_db_schema::{
   source::{
+    activity::ActivitySendTargets,
     comment_report::{CommentReport, CommentReportForm},
     community::Community,
     person::Person,
@@ -49,8 +50,8 @@ impl Report {
       id: id.clone(),
       audience: Some(community.id().into()),
     };
-
-    let inbox = vec![community.shared_inbox_or_inbox()];
+    // todo: this should probably filter and only send if the community is remote?
+    let inbox = ActivitySendTargets::to_inbox(community.shared_inbox_or_inbox());
     send_lemmy_activity(&context, report, &actor, inbox, false).await
   }
 }

crates/apub/src/activities/community/update.rs

@@ -18,7 +18,7 @@ use activitypub_federation::{
 };
 use lemmy_api_common::context::LemmyContext;
 use lemmy_db_schema::{
-  source::{community::Community, person::Person},
+  source::{activity::ActivitySendTargets, community::Community, person::Person},
   traits::Crud,
 };
 use lemmy_utils::error::LemmyError;
@@ -46,7 +46,15 @@ pub(crate) async fn send_update_community(
   };
 
   let activity = AnnouncableActivities::UpdateCommunity(update);
-  send_activity_in_community(activity, &actor, &community, vec![], true, &context).await
+  send_activity_in_community(
+    activity,
+    &actor,
+    &community,
+    ActivitySendTargets::empty(),
+    true,
+    &context,
+  )
+  .await
 }
 
 #[async_trait::async_trait]

crates/apub/src/activities/create_or_update/comment.rs

@@ -31,6 +31,7 @@ use lemmy_db_schema::{
   aggregates::structs::CommentAggregates,
   newtypes::PersonId,
   source::{
+    activity::ActivitySendTargets,
     comment::{Comment, CommentLike, CommentLikeForm},
     community::Community,
     person::Person,
@@ -88,10 +89,10 @@ impl CreateOrUpdateNote {
       .map(|t| t.href.clone())
       .map(ObjectId::from)
       .collect();
-    let mut inboxes = vec![];
+    let mut inboxes = ActivitySendTargets::empty();
     for t in tagged_users {
       let person = t.dereference(&context).await?;
-      inboxes.push(person.shared_inbox_or_inbox());
+      inboxes.add_inbox(person.shared_inbox_or_inbox());
     }
 
     let activity = AnnouncableActivities::CreateOrUpdateComment(create_or_update);

crates/apub/src/activities/create_or_update/post.rs

@@ -26,6 +26,7 @@ use lemmy_db_schema::{
   aggregates::structs::PostAggregates,
   newtypes::PersonId,
   source::{
+    activity::ActivitySendTargets,
     community::Community,
     person::Person,
     post::{Post, PostLike, PostLikeForm},
@@ -80,7 +81,7 @@ impl CreateOrUpdatePage {
       activity,
       &person,
       &community,
-      vec![],
+      ActivitySendTargets::empty(),
       is_mod_action,
       &context,
     )

crates/apub/src/activities/create_or_update/private_message.rs

@@ -13,6 +13,7 @@ use activitypub_federation::{
   traits::{ActivityHandler, Actor, Object},
 };
 use lemmy_api_common::context::LemmyContext;
+use lemmy_db_schema::source::activity::ActivitySendTargets;
 use lemmy_db_views::structs::PrivateMessageView;
 use lemmy_utils::error::LemmyError;
 use url::Url;
@@ -38,7 +39,7 @@ pub(crate) async fn send_create_or_update_pm(
       .await?,
     kind,
   };
-  let inbox = vec![recipient.shared_inbox_or_inbox()];
+  let inbox = ActivitySendTargets::to_inbox(recipient.shared_inbox_or_inbox());
   send_lemmy_activity(&context, create_or_update, &actor, inbox, true).await
 }
 

crates/apub/src/activities/deletion/delete_user.rs

@@ -1,7 +1,7 @@
 use crate::{
   activities::{generate_activity_id, send_lemmy_activity, verify_is_public, verify_person},
   insert_received_activity,
-  objects::{instance::remote_instance_inboxes, person::ApubPerson},
+  objects::person::ApubPerson,
   protocol::activities::deletion::delete_user::DeleteUser,
 };
 use activitypub_federation::{
@@ -11,7 +11,7 @@ use activitypub_federation::{
   traits::{ActivityHandler, Actor},
 };
 use lemmy_api_common::{context::LemmyContext, utils::delete_user_account};
-use lemmy_db_schema::source::person::Person;
+use lemmy_db_schema::source::{activity::ActivitySendTargets, person::Person};
 use lemmy_utils::error::LemmyError;
 use url::Url;
 
@@ -38,7 +38,8 @@ pub async fn delete_user(person: Person, context: Data<LemmyContext>) -> Result<
     cc: vec![],
   };
 
-  let inboxes = remote_instance_inboxes(&mut context.pool()).await?;
+  let inboxes = ActivitySendTargets::to_all_instances();
+
   send_lemmy_activity(&context, delete, &actor, inboxes, true).await?;
   Ok(())
 }