Move code to apub library (#1795)

* Remove dependency of apub_lib on LemmyContext

* Move ApubObject trait to library

* Reorganize files in apub lib

* Move ActorType, signatures, activity_queue to apub library

Cargo.lock

@@ -1711,6 +1711,7 @@ dependencies = [
  "lazy_static",
  "lemmy_api_common",
  "lemmy_apub",
+ "lemmy_apub_lib",
  "lemmy_db_queries",
  "lemmy_db_schema",
  "lemmy_db_views",
@@ -1745,17 +1746,13 @@ dependencies = [
  "anyhow",
  "async-trait",
  "awc",
- "background-jobs",
- "base64 0.13.0",
  "bcrypt",
  "chrono",
  "diesel",
  "futures",
  "http",
  "http-signature-normalization-actix",
- "http-signature-normalization-reqwest",
  "itertools",
- "lazy_static",
  "lemmy_api_common",
  "lemmy_apub_lib",
  "lemmy_db_queries",
@@ -1765,7 +1762,6 @@ dependencies = [
  "lemmy_utils",
  "lemmy_websocket",
  "log",
- "openssl",
  "percent-encoding",
  "rand 0.8.4",
  "reqwest",
@@ -1786,15 +1782,23 @@ name = "lemmy_apub_lib"
 version = "0.13.0"
 dependencies = [
  "activitystreams",
+ "actix-web",
  "anyhow",
  "async-trait",
+ "background-jobs",
+ "base64 0.13.0",
+ "http",
+ "http-signature-normalization-actix",
+ "http-signature-normalization-reqwest",
+ "lazy_static",
  "lemmy_apub_lib_derive",
  "lemmy_utils",
- "lemmy_websocket",
  "log",
+ "openssl",
  "reqwest",
  "serde",
  "serde_json",
+ "sha2",
  "url",
 ]
 
@@ -1837,6 +1841,8 @@ dependencies = [
  "chrono",
  "diesel",
  "diesel-derive-newtype",
+ "lemmy_apub_lib",
+ "lemmy_utils",
  "log",
  "serde",
  "serde_json",
@@ -1925,6 +1931,7 @@ dependencies = [
  "lemmy_api_common",
  "lemmy_api_crud",
  "lemmy_apub",
+ "lemmy_apub_lib",
  "lemmy_db_queries",
  "lemmy_db_schema",
  "lemmy_db_views",

Cargo.toml

@@ -35,6 +35,7 @@ members = [
 lemmy_api = { version = "=0.13.0", path = "./crates/api" }
 lemmy_api_crud = { version = "=0.13.0", path = "./crates/api_crud" }
 lemmy_apub = { version = "=0.13.0", path = "./crates/apub" }
+lemmy_apub_lib = { version = "=0.13.0", path = "./crates/apub_lib" }
 lemmy_utils = { version = "=0.13.0", path = "./crates/utils" }
 lemmy_db_schema = { version = "=0.13.0", path = "./crates/db_schema" }
 lemmy_db_queries = { version = "=0.13.0", path = "./crates/db_queries" }

api_tests/prepare-drone-federation-test.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 set -e
 
-export LEMMY_TEST_SEND_SYNC=1
+export APUB_TESTING_SEND_SYNC=1
 export RUST_BACKTRACE=1
 export RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
 

crates/api_crud/Cargo.toml

@@ -7,6 +7,7 @@ license = "AGPL-3.0"
 
 [dependencies]
 lemmy_apub = { version = "=0.13.0", path = "../apub" }
+lemmy_apub_lib = { version = "=0.13.0", path = "../apub_lib" }
 lemmy_utils = { version = "=0.13.0", path = "../utils" }
 lemmy_db_queries = { version = "=0.13.0", path = "../db_queries" }
 lemmy_db_schema = { version = "=0.13.0", path = "../db_schema" }

crates/api_crud/src/community/create.rs

@@ -7,13 +7,14 @@ use lemmy_api_common::{
   is_admin,
 };
 use lemmy_apub::{
+  fetcher::object_id::ObjectId,
   generate_apub_endpoint,
   generate_followers_url,
   generate_inbox_url,
   generate_shared_inbox_url,
   EndpointType,
 };
-use lemmy_db_queries::{diesel_option_overwrite_to_url, ApubObject, Crud, Followable, Joinable};
+use lemmy_db_queries::{diesel_option_overwrite_to_url, Crud, Followable, Joinable};
 use lemmy_db_schema::source::{
   community::{
     Community,
@@ -67,11 +68,8 @@ impl PerformCrud for CreateCommunity {
       &data.name,
       &context.settings().get_protocol_and_hostname(),
     )?;
-    let actor_id_cloned = community_actor_id.to_owned();
+    let community_actor_id_wrapped = ObjectId::<Community>::new(community_actor_id.clone());
-    let community_dupe = blocking(context.pool(), move |conn| {
+    let community_dupe = community_actor_id_wrapped.dereference_local(context).await;
-      Community::read_from_apub_id(conn, &actor_id_cloned)
-    })
-    .await?;
     if community_dupe.is_ok() {
       return Err(ApiError::err("community_already_exists").into());
     }

crates/api_crud/src/community/read.rs

@@ -1,14 +1,8 @@
 use crate::PerformCrud;
 use actix_web::web::Data;
 use lemmy_api_common::{blocking, community::*, get_local_user_view_from_jwt_opt};
-use lemmy_apub::{build_actor_id_from_shortname, EndpointType};
+use lemmy_apub::{build_actor_id_from_shortname, fetcher::object_id::ObjectId, EndpointType};
-use lemmy_db_queries::{
+use lemmy_db_queries::{from_opt_str_to_opt_enum, DeleteableOrRemoveable, ListingType, SortType};
-  from_opt_str_to_opt_enum,
-  ApubObject,
-  DeleteableOrRemoveable,
-  ListingType,
-  SortType,
-};
 use lemmy_db_schema::source::community::*;
 use lemmy_db_views_actor::{
   community_moderator_view::CommunityModeratorView,
@@ -38,10 +32,9 @@ impl PerformCrud for GetCommunity {
         let community_actor_id =
           build_actor_id_from_shortname(EndpointType::Community, &name, &context.settings())?;
 
-        blocking(context.pool(), move |conn| {
+        ObjectId::<Community>::new(community_actor_id)
-          Community::read_from_apub_id(conn, &community_actor_id)
+          .dereference(context, &mut 0)
-        })
+          .await
-        .await?
           .map_err(|_| ApiError::err("couldnt_find_community"))?
           .id
       }

crates/api_crud/src/user/read.rs

@@ -1,8 +1,8 @@
 use crate::PerformCrud;
 use actix_web::web::Data;
 use lemmy_api_common::{blocking, get_local_user_view_from_jwt_opt, person::*};
-use lemmy_apub::{build_actor_id_from_shortname, EndpointType};
+use lemmy_apub::{build_actor_id_from_shortname, fetcher::object_id::ObjectId, EndpointType};
-use lemmy_db_queries::{from_opt_str_to_opt_enum, ApubObject, SortType};
+use lemmy_db_queries::{from_opt_str_to_opt_enum, SortType};
 use lemmy_db_schema::source::person::*;
 use lemmy_db_views::{comment_view::CommentQueryBuilder, post_view::PostQueryBuilder};
 use lemmy_db_views_actor::{
@@ -45,10 +45,9 @@ impl PerformCrud for GetPersonDetails {
         let actor_id =
           build_actor_id_from_shortname(EndpointType::Person, &name, &context.settings())?;
 
-        let person = blocking(context.pool(), move |conn| {
+        let person = ObjectId::<Person>::new(actor_id)
-          Person::read_from_apub_id(conn, &actor_id)
+          .dereference(context, &mut 0)
-        })
+          .await;
-        .await?;
         person
           .map_err(|_| ApiError::err("couldnt_find_that_username_or_email"))?
           .id

crates/apub/Cargo.toml

@@ -36,11 +36,8 @@ strum = "0.21.0"
 strum_macros = "0.21.1"
 url = { version = "2.2.2", features = ["serde"] }
 percent-encoding = "2.1.0"
-openssl = "0.10.36"
 http = "0.2.5"
 http-signature-normalization-actix = { version = "0.5.0-beta.10", default-features = false, features = ["server", "sha-2"] }
-http-signature-normalization-reqwest = { version = "0.2.0", default-features = false, features = ["sha-2"] }
-base64 = "0.13.0"
 tokio = "1.12.0"
 futures = "0.3.17"
 itertools = "0.10.1"
@@ -49,7 +46,5 @@ sha2 = "0.9.8"
 async-trait = "0.1.51"
 anyhow = "1.0.44"
 thiserror = "1.0.29"
-background-jobs = "0.10.0"
 reqwest = { version = "0.11.4", features = ["json"] }
-lazy_static = "1.4.0"
 

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

@@ -1,22 +1,25 @@
 use crate::{
   activities::{
     comment::{collect_non_local_mentions, get_notif_recipients},
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     extract_community,
     generate_activity_id,
     verify_activity,
     verify_person_in_community,
     CreateOrUpdateType,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{comment::Note, FromApub, ToApub},
-  ActorType,
 };
 use activitystreams::{base::AnyBase, link::Mention, primitives::OneOrMany, unparsed::Unparsed};
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, verify_domains_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+  verify::verify_domains_match,
+};
 use lemmy_db_queries::Crud;
 use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
 use lemmy_utils::LemmyError;
@@ -76,15 +79,17 @@ impl CreateOrUpdateComment {
     };
 
     let activity = AnnouncableActivities::CreateOrUpdateComment(create_or_update);
-    send_to_community_new(activity, &id, actor, &community, maa.inboxes, context).await
+    send_to_community(activity, &id, actor, &community, maa.inboxes, context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for CreateOrUpdateComment {
+  type DataType = LemmyContext;
+
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let community = extract_community(&self.cc, context, request_counter).await?;
@@ -101,7 +106,7 @@ impl ActivityHandler for CreateOrUpdateComment {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let comment =

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

@@ -1,4 +1,4 @@
-use crate::{fetcher::object_id::ObjectId, ActorType};
+use crate::fetcher::object_id::ObjectId;
 use activitystreams::{
   base::BaseExt,
   link::{LinkExt, Mention},
@@ -6,7 +6,7 @@ use activitystreams::{
 use anyhow::anyhow;
 use itertools::Itertools;
 use lemmy_api_common::{blocking, send_local_notifs};
-use lemmy_apub_lib::webfinger::WebfingerResponse;
+use lemmy_apub_lib::{traits::ActorType, webfinger::WebfingerResponse};
 use lemmy_db_queries::{Crud, DbPool};
 use lemmy_db_schema::{
   source::{comment::Comment, community::Community, person::Person, post::Post},
@@ -68,7 +68,7 @@ pub async fn collect_non_local_mentions(
   let parent_creator = get_comment_parent_creator(context.pool(), comment).await?;
   let mut addressed_ccs = vec![community.actor_id(), parent_creator.actor_id()];
   // Note: dont include community inbox here, as we send to it separately with `send_to_community()`
-  let mut inboxes = vec![parent_creator.get_shared_inbox_or_inbox_url()];
+  let mut inboxes = vec![parent_creator.shared_inbox_or_inbox_url()];
 
   // Add the mention tag
   let mut tags = Vec::new();
@@ -88,7 +88,7 @@ pub async fn collect_non_local_mentions(
       addressed_ccs.push(actor_id.to_owned().to_string().parse()?);
 
       let mention_person = actor_id.dereference(context, &mut 0).await?;
-      inboxes.push(mention_person.get_shared_inbox_or_inbox_url());
+      inboxes.push(mention_person.shared_inbox_or_inbox_url());
 
       let mut mention_tag = Mention::new();
       mention_tag

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

@@ -1,17 +1,15 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_add_remove_moderator_target,
     verify_mod_action,
     verify_person_in_community,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   generate_moderators_url,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::AddType,
@@ -20,7 +18,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::{source::community::CommunityModerator_, Joinable};
 use lemmy_db_schema::source::{
   community::{Community, CommunityModerator, CommunityModeratorForm},
@@ -72,28 +74,30 @@ impl AddMod {
     };
 
     let activity = AnnouncableActivities::AddMod(add);
-    let inboxes = vec![added_mod.get_shared_inbox_or_inbox_url()];
+    let inboxes = vec![added_mod.shared_inbox_or_inbox_url()];
-    send_to_community_new(activity, &id, actor, community, inboxes, context).await
+    send_to_community(activity, &id, actor, community, inboxes, context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for AddMod {
+  type DataType = LemmyContext;
+
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person_in_community(&self.actor, &self.cc[0], context, request_counter).await?;
-    verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+    verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
     verify_add_remove_moderator_target(&self.target, &self.cc[0])?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let community = self.cc[0].dereference(context, request_counter).await?;

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

@@ -17,12 +17,11 @@ use crate::{
     verify_community,
     voting::{undo_vote::UndoVote, vote::Vote},
   },
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   http::is_activity_already_known,
   insert_activity,
-  ActorType,
+  send_lemmy_activity,
   CommunityType,
 };
 use activitystreams::{
@@ -31,7 +30,11 @@ use activitystreams::{
   primitives::OneOrMany,
   unparsed::Unparsed,
 };
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_schema::source::community::Community;
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
@@ -40,6 +43,7 @@ use url::Url;
 
 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
 #[serde(untagged)]
+#[activity_handler(LemmyContext)]
 pub enum AnnouncableActivities {
   CreateOrUpdateComment(CreateOrUpdateComment),
   CreateOrUpdatePost(Box<CreateOrUpdatePost>),
@@ -92,15 +96,16 @@ impl AnnounceActivity {
       unparsed: Default::default(),
     };
     let inboxes = list_community_follower_inboxes(community, additional_inboxes, context).await?;
-    send_activity_new(context, &announce, &announce.id, community, inboxes, false).await
+    send_lemmy_activity(context, &announce, &announce.id, community, inboxes, false).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for AnnounceActivity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -111,7 +116,7 @@ impl ActivityHandler for AnnounceActivity {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     if is_activity_already_known(context.pool(), self.object.id_unchecked()).await? {

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

@@ -1,15 +1,13 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_mod_action,
     verify_person_in_community,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::BlockType,
@@ -18,7 +16,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::{Bannable, Followable};
 use lemmy_db_schema::source::{
   community::{
@@ -83,27 +85,28 @@ impl BlockUserFromCommunity {
     let block_id = block.id.clone();
 
     let activity = AnnouncableActivities::BlockUserFromCommunity(block);
-    let inboxes = vec![target.get_shared_inbox_or_inbox_url()];
+    let inboxes = vec![target.shared_inbox_or_inbox_url()];
-    send_to_community_new(activity, &block_id, actor, community, inboxes, context).await
+    send_to_community(activity, &block_id, actor, community, inboxes, context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for BlockUserFromCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person_in_community(&self.actor, &self.cc[0], context, request_counter).await?;
-    verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+    verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let community = self.cc[0].dereference(context, request_counter).await?;

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

@@ -1,5 +1,12 @@
-use crate::{check_is_apub_id_valid, CommunityType};
+use crate::{
+  activities::community::announce::{AnnouncableActivities, AnnounceActivity},
+  check_is_apub_id_valid,
+  insert_activity,
+  send_lemmy_activity,
+  CommunityType,
+};
 use itertools::Itertools;
+use lemmy_apub_lib::traits::ActorType;
 use lemmy_db_schema::source::community::Community;
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
@@ -33,3 +40,24 @@ async fn list_community_follower_inboxes(
     .collect(),
   )
 }
+
+pub(crate) async fn send_to_community<T: ActorType>(
+  activity: AnnouncableActivities,
+  activity_id: &Url,
+  actor: &T,
+  community: &Community,
+  additional_inboxes: Vec<Url>,
+  context: &LemmyContext,
+) -> Result<(), LemmyError> {
+  // if this is a local community, we need to do an announce from the community instead
+  if community.local {
+    insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
+    AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
+  } else {
+    let mut inboxes = additional_inboxes;
+    inboxes.push(community.shared_inbox_or_inbox_url());
+    send_lemmy_activity(context, &activity, activity_id, actor, inboxes, false).await?;
+  }
+
+  Ok(())
+}

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

@@ -1,6 +1,6 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     deletion::{delete::receive_remove_action, verify_delete_activity},
     generate_activity_id,
     verify_activity,
@@ -8,11 +8,9 @@ use crate::{
     verify_mod_action,
     verify_person_in_community,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   generate_moderators_url,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::RemoveType,
@@ -21,7 +19,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::Joinable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityModerator, CommunityModeratorForm},
@@ -74,22 +76,23 @@ impl RemoveMod {
     };
 
     let activity = AnnouncableActivities::RemoveMod(remove);
-    let inboxes = vec![removed_mod.get_shared_inbox_or_inbox_url()];
+    let inboxes = vec![removed_mod.shared_inbox_or_inbox_url()];
-    send_to_community_new(activity, &id, actor, community, inboxes, context).await
+    send_to_community(activity, &id, actor, community, inboxes, context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for RemoveMod {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     if let Some(target) = &self.target {
       verify_person_in_community(&self.actor, &self.cc[0], context, request_counter).await?;
-      verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+      verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
       verify_add_remove_moderator_target(target, &self.cc[0])?;
     } else {
       verify_delete_activity(
@@ -107,7 +110,7 @@ impl ActivityHandler for RemoveMod {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     if self.target.is_some() {

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

@@ -1,15 +1,17 @@
 use crate::{
   activities::{
-    community::{announce::AnnouncableActivities, block_user::BlockUserFromCommunity},
+    community::{
+      announce::AnnouncableActivities,
+      block_user::BlockUserFromCommunity,
+      send_to_community,
+    },
     generate_activity_id,
     verify_activity,
     verify_mod_action,
     verify_person_in_community,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::UndoType,
@@ -18,7 +20,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::Bannable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityPersonBan, CommunityPersonBanForm},
@@ -70,28 +76,29 @@ impl UndoBlockUserFromCommunity {
     };
 
     let activity = AnnouncableActivities::UndoBlockUserFromCommunity(undo);
-    let inboxes = vec![target.get_shared_inbox_or_inbox_url()];
+    let inboxes = vec![target.shared_inbox_or_inbox_url()];
-    send_to_community_new(activity, &id, actor, community, inboxes, context).await
+    send_to_community(activity, &id, actor, community, inboxes, context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoBlockUserFromCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person_in_community(&self.actor, &self.cc[0], context, request_counter).await?;
-    verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+    verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
     self.object.verify(context, request_counter).await?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let community = self.cc[0].dereference(context, request_counter).await?;

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

@@ -1,16 +1,14 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_mod_action,
     verify_person_in_community,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{community::Group, ToApub},
-  ActorType,
 };
 use activitystreams::{
   activity::kind::UpdateType,
@@ -19,8 +17,12 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
-use lemmy_db_queries::{ApubObject, Crud};
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
+use lemmy_db_queries::Crud;
 use lemmy_db_schema::source::{
   community::{Community, CommunityForm},
   person::Person,
@@ -71,33 +73,31 @@ impl UpdateCommunity {
     };
 
     let activity = AnnouncableActivities::UpdateCommunity(Box::new(update));
-    send_to_community_new(activity, &id, actor, community, vec![], context).await
+    send_to_community(activity, &id, actor, community, vec![], context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UpdateCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person_in_community(&self.actor, &self.cc[0], context, request_counter).await?;
-    verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+    verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
-    _request_counter: &mut i32,
+    request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
-    let cc = self.cc[0].clone().into();
+    let cc = self.cc[0].clone();
-    let community = blocking(context.pool(), move |conn| {
+    let community = cc.dereference(context, request_counter).await?;
-      Community::read_from_apub_id(conn, &cc)
-    })
-    .await??;
 
     let updated_community = Group::from_apub_to_form(
       &self.object,

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

@@ -1,6 +1,6 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     deletion::{
       receive_delete_action,
       verify_delete_activity,
@@ -10,10 +10,8 @@ use crate::{
     generate_activity_id,
     verify_activity,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::DeleteType,
@@ -23,7 +21,11 @@ use activitystreams::{
 };
 use anyhow::anyhow;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::{
   source::{comment::Comment_, community::Community_, post::Post_},
   Crud,
@@ -82,9 +84,10 @@ pub struct Delete {
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for Delete {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -102,7 +105,7 @@ impl ActivityHandler for Delete {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     if let Some(reason) = self.summary {
@@ -166,7 +169,7 @@ impl Delete {
     let delete_id = delete.id.clone();
 
     let activity = AnnouncableActivities::Delete(delete);
-    send_to_community_new(activity, &delete_id, actor, community, vec![], context).await
+    send_to_community(activity, &delete_id, actor, community, vec![], context).await
   }
 }
 

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

@@ -5,18 +5,15 @@ use crate::{
     verify_person_in_community,
   },
   fetcher::object_id::ObjectId,
-  ActorType,
 };
+use diesel::PgConnection;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, ActivityFields};
+use lemmy_apub_lib::{
-use lemmy_db_queries::{
+  traits::{ActivityFields, ActorType, ApubObject},
-  source::{comment::Comment_, community::Community_, post::Post_},
+  verify::verify_domains_match,
-  ApubObject,
-};
-use lemmy_db_schema::{
-  source::{comment::Comment, community::Community, person::Person, post::Post},
-  DbUrl,
 };
+use lemmy_db_queries::source::{comment::Comment_, community::Community_, post::Post_};
+use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::{
   send::{send_comment_ws_message_simple, send_community_ws_message, send_post_ws_message},
@@ -70,29 +67,32 @@ impl DeletableObjects {
     ap_id: &Url,
     context: &LemmyContext,
   ) -> Result<DeletableObjects, LemmyError> {
-    let id: DbUrl = ap_id.clone().into();
+    if let Some(c) =
-
+      DeletableObjects::read_type_from_db::<Community>(ap_id.clone(), context).await?
-    if let Some(c) = DeletableObjects::read_type_from_db::<Community>(id.clone(), context).await? {
+    {
       return Ok(DeletableObjects::Community(Box::new(c)));
     }
-    if let Some(p) = DeletableObjects::read_type_from_db::<Post>(id.clone(), context).await? {
+    if let Some(p) = DeletableObjects::read_type_from_db::<Post>(ap_id.clone(), context).await? {
       return Ok(DeletableObjects::Post(Box::new(p)));
     }
-    if let Some(c) = DeletableObjects::read_type_from_db::<Comment>(id.clone(), context).await? {
+    if let Some(c) = DeletableObjects::read_type_from_db::<Comment>(ap_id.clone(), context).await? {
       return Ok(DeletableObjects::Comment(Box::new(c)));
     }
     Err(diesel::NotFound.into())
   }
 
   // TODO: a method like this should be provided by fetcher module
-  async fn read_type_from_db<Type: ApubObject + Send + 'static>(
+  async fn read_type_from_db<Type>(
-    ap_id: DbUrl,
+    ap_id: Url,
     context: &LemmyContext,
-  ) -> Result<Option<Type>, LemmyError> {
+  ) -> Result<Option<Type>, LemmyError>
+  where
+    Type: ApubObject<DataType = PgConnection> + Send + 'static,
+  {
     blocking(context.pool(), move |conn| {
-      Type::read_from_apub_id(conn, &ap_id).ok()
+      Type::read_from_apub_id(conn, ap_id)
     })
-    .await
+    .await?
   }
 }
 
@@ -114,7 +114,13 @@ pub(in crate::activities) async fn verify_delete_activity(
         verify_person_in_community(&actor, community_id, context, request_counter).await?;
       }
       // community deletion is always a mod (or admin) action
-      verify_mod_action(&actor, ObjectId::new(c.actor_id()), context).await?;
+      verify_mod_action(
+        &actor,
+        ObjectId::new(c.actor_id()),
+        context,
+        request_counter,
+      )
+      .await?;
     }
     DeletableObjects::Post(p) => {
       verify_delete_activity_post_or_comment(
@@ -153,7 +159,7 @@ async fn verify_delete_activity_post_or_comment(
   let actor = ObjectId::new(activity.actor().clone());
   verify_person_in_community(&actor, community_id, context, request_counter).await?;
   if is_mod_action {
-    verify_mod_action(&actor, community_id.clone(), context).await?;
+    verify_mod_action(&actor, community_id.clone(), context, request_counter).await?;
   } else {
     // domain of post ap_id and post.creator ap_id are identical, so we just check the former
     verify_domains_match(activity.actor(), object_id)?;

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

@@ -1,6 +1,6 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     deletion::{
       delete::Delete,
       receive_delete_action,
@@ -11,10 +11,8 @@ use crate::{
     generate_activity_id,
     verify_activity,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
 };
 use activitystreams::{
   activity::kind::UndoType,
@@ -24,7 +22,11 @@ use activitystreams::{
 };
 use anyhow::anyhow;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::source::{comment::Comment_, community::Community_, post::Post_};
 use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
 use lemmy_utils::LemmyError;
@@ -54,9 +56,10 @@ pub struct UndoDelete {
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoDelete {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -75,7 +78,7 @@ impl ActivityHandler for UndoDelete {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     if self.object.summary.is_some() {
@@ -124,7 +127,7 @@ impl UndoDelete {
     };
 
     let activity = AnnouncableActivities::UndoDelete(undo);
-    send_to_community_new(activity, &id, actor, community, vec![], context).await
+    send_to_community(activity, &id, actor, community, vec![], context).await
   }
 
   pub(in crate::activities) async fn receive_undo_remove_action(

crates/apub/src/activities/following/accept.rs

@@ -5,10 +5,9 @@ use crate::{
     verify_activity,
     verify_community,
   },
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{
   activity::kind::AcceptType,
@@ -17,8 +16,12 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_urls_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
-use lemmy_db_queries::{ApubObject, Followable};
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::verify_urls_match,
+};
+use lemmy_db_queries::Followable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityFollower},
   person::Person,
@@ -44,18 +47,17 @@ pub struct AcceptFollowCommunity {
 }
 
 impl AcceptFollowCommunity {
-  pub async fn send(follow: FollowCommunity, context: &LemmyContext) -> Result<(), LemmyError> {
+  pub async fn send(
-    let community_id = follow.object.clone();
+    follow: FollowCommunity,
-    let community = blocking(context.pool(), move |conn| {
+    context: &LemmyContext,
-      Community::read_from_apub_id(conn, &community_id.into())
+    request_counter: &mut i32,
-    })
+  ) -> Result<(), LemmyError> {
-    .await??;
+    let community = follow.object.dereference_local(context).await?;
-    let person_id = follow.actor().clone();
+    let person = follow
-    let person = blocking(context.pool(), move |conn| {
+      .actor
-      Person::read_from_apub_id(conn, &person_id.into())
+      .clone()
-    })
+      .dereference(context, request_counter)
-    .await??;
+      .await?;
-
     let accept = AcceptFollowCommunity {
       actor: ObjectId::new(community.actor_id()),
       to: ObjectId::new(person.actor_id()),
@@ -69,15 +71,17 @@ impl AcceptFollowCommunity {
       unparsed: Default::default(),
     };
     let inbox = vec![person.inbox_url.into()];
-    send_activity_new(context, &accept, &accept.id, &community, inbox, true).await
+    send_lemmy_activity(context, &accept, &accept.id, &community, inbox, true).await
   }
 }
+
 /// Handle accepted follows
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for AcceptFollowCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -90,7 +94,7 @@ impl ActivityHandler for AcceptFollowCommunity {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;

crates/apub/src/activities/following/follow.rs

@@ -5,10 +5,9 @@ use crate::{
     verify_activity,
     verify_person,
   },
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{
   activity::kind::FollowType,
@@ -17,7 +16,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_urls_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::verify_urls_match,
+};
 use lemmy_db_queries::Followable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityFollower, CommunityFollowerForm},
@@ -31,7 +34,7 @@ use url::Url;
 #[derive(Clone, Debug, Deserialize, Serialize, ActivityFields)]
 #[serde(rename_all = "camelCase")]
 pub struct FollowCommunity {
-  actor: ObjectId<Person>,
+  pub(in crate::activities::following) actor: ObjectId<Person>,
   // TODO: is there any reason to put the same community id twice, in to and object?
   pub(in crate::activities::following) to: ObjectId<Community>,
   pub(in crate::activities::following) object: ObjectId<Community>,
@@ -80,15 +83,16 @@ impl FollowCommunity {
 
     let follow = FollowCommunity::new(actor, community, context)?;
     let inbox = vec![community.inbox_url.clone().into()];
-    send_activity_new(context, &follow, &follow.id, actor, inbox, true).await
+    send_lemmy_activity(context, &follow, &follow.id, actor, inbox, true).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for FollowCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -99,7 +103,7 @@ impl ActivityHandler for FollowCommunity {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;
@@ -116,6 +120,6 @@ impl ActivityHandler for FollowCommunity {
     })
     .await?;
 
-    AcceptFollowCommunity::send(self, context).await
+    AcceptFollowCommunity::send(self, context, request_counter).await
   }
 }

crates/apub/src/activities/following/undo.rs

@@ -5,10 +5,9 @@ use crate::{
     verify_activity,
     verify_person,
   },
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{
   activity::kind::UndoType,
@@ -17,7 +16,11 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_urls_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::verify_urls_match,
+};
 use lemmy_db_queries::Followable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityFollower, CommunityFollowerForm},
@@ -62,16 +65,17 @@ impl UndoFollowCommunity {
       context: lemmy_context(),
       unparsed: Default::default(),
     };
-    let inbox = vec![community.get_shared_inbox_or_inbox_url()];
+    let inbox = vec![community.shared_inbox_or_inbox_url()];
-    send_activity_new(context, &undo, &undo.id, actor, inbox, true).await
+    send_lemmy_activity(context, &undo, &undo.id, actor, inbox, true).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoFollowCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -84,7 +88,7 @@ impl ActivityHandler for UndoFollowCommunity {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;

crates/apub/src/activities/mod.rs

@@ -6,12 +6,8 @@ use crate::{
 };
 use anyhow::anyhow;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, ActivityFields};
+use lemmy_apub_lib::{traits::ActivityFields, verify::verify_domains_match};
-use lemmy_db_queries::ApubObject;
+use lemmy_db_schema::source::{community::Community, person::Person};
-use lemmy_db_schema::{
-  source::{community::Community, person::Person},
-  DbUrl,
-};
 use lemmy_db_views_actor::community_view::CommunityView;
 use lemmy_utils::{settings::structs::Settings, LemmyError};
 use lemmy_websocket::LemmyContext;
@@ -26,7 +22,6 @@ pub mod deletion;
 pub mod following;
 pub mod post;
 pub mod private_message;
-pub mod send;
 pub mod undo_remove;
 pub mod voting;
 
@@ -104,18 +99,12 @@ pub(crate) async fn verify_mod_action(
   actor_id: &ObjectId<Person>,
   community_id: ObjectId<Community>,
   context: &LemmyContext,
+  request_counter: &mut i32,
 ) -> Result<(), LemmyError> {
-  let community = blocking(context.pool(), move |conn| {
+  let community = community_id.dereference_local(context).await?;
-    Community::read_from_apub_id(conn, &community_id.into())
-  })
-  .await??;
 
   if community.local {
-    let actor_id: DbUrl = actor_id.clone().into();
+    let actor = actor_id.dereference(context, request_counter).await?;
-    let actor = blocking(context.pool(), move |conn| {
-      Person::read_from_apub_id(conn, &actor_id)
-    })
-    .await??;
 
     // Note: this will also return true for admins in addition to mods, but as we dont know about
     //       remote admins, it doesnt make any difference.

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

@@ -1,27 +1,24 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_mod_action,
     verify_person_in_community,
     CreateOrUpdateType,
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{post::Page, FromApub, ToApub},
-  ActorType,
 };
 use activitystreams::{base::AnyBase, primitives::OneOrMany, unparsed::Unparsed};
 use anyhow::anyhow;
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
   values::PublicUrl,
-  verify_domains_match,
+  verify::{verify_domains_match, verify_urls_match},
-  verify_urls_match,
-  ActivityFields,
-  ActivityHandler,
 };
 use lemmy_db_queries::Crud;
 use lemmy_db_schema::source::{community::Community, person::Person, post::Post};
@@ -75,15 +72,16 @@ impl CreateOrUpdatePost {
     };
 
     let activity = AnnouncableActivities::CreateOrUpdatePost(Box::new(create_or_update));
-    send_to_community_new(activity, &id, actor, &community, vec![], context).await
+    send_to_community(activity, &id, actor, &community, vec![], context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for CreateOrUpdatePost {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -104,9 +102,9 @@ impl ActivityHandler for CreateOrUpdatePost {
         }
       }
       CreateOrUpdateType::Update => {
-        let is_mod_action = self.object.is_mod_action(context.pool()).await?;
+        let is_mod_action = self.object.is_mod_action(context).await?;
         if is_mod_action {
-          verify_mod_action(&self.actor, self.cc[0].clone(), context).await?;
+          verify_mod_action(&self.actor, self.cc[0].clone(), context, request_counter).await?;
         } else {
           verify_domains_match(self.actor.inner(), self.object.id_unchecked())?;
           verify_urls_match(self.actor(), self.object.attributed_to.inner())?;
@@ -119,7 +117,7 @@ impl ActivityHandler for CreateOrUpdatePost {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;

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

@@ -1,14 +1,17 @@
 use crate::{
   activities::{generate_activity_id, verify_activity, verify_person, CreateOrUpdateType},
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{private_message::Note, FromApub, ToApub},
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{base::AnyBase, primitives::OneOrMany, unparsed::Unparsed};
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::verify_domains_match,
+};
 use lemmy_db_queries::Crud;
 use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
 use lemmy_utils::LemmyError;
@@ -55,15 +58,16 @@ impl CreateOrUpdatePrivateMessage {
       kind,
       unparsed: Default::default(),
     };
-    let inbox = vec![recipient.get_shared_inbox_or_inbox_url()];
+    let inbox = vec![recipient.shared_inbox_or_inbox_url()];
-    send_activity_new(context, &create_or_update, &id, actor, inbox, true).await
+    send_lemmy_activity(context, &create_or_update, &id, actor, inbox, true).await
   }
 }
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for CreateOrUpdatePrivateMessage {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -75,7 +79,7 @@ impl ActivityHandler for CreateOrUpdatePrivateMessage {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let private_message =

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

@@ -1,9 +1,8 @@
 use crate::{
   activities::{generate_activity_id, verify_activity, verify_person},
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{
   activity::kind::DeleteType,
@@ -12,8 +11,12 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
-use lemmy_db_queries::{source::private_message::PrivateMessage_, ApubObject, Crud};
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::verify_domains_match,
+};
+use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
 use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
@@ -25,7 +28,7 @@ use url::Url;
 pub struct DeletePrivateMessage {
   actor: ObjectId<Person>,
   to: ObjectId<Person>,
-  pub(in crate::activities::private_message) object: Url,
+  pub(in crate::activities::private_message) object: ObjectId<PrivateMessage>,
   #[serde(rename = "type")]
   kind: DeleteType,
   id: Url,
@@ -44,7 +47,7 @@ impl DeletePrivateMessage {
     Ok(DeletePrivateMessage {
       actor: ObjectId::new(actor.actor_id()),
       to: ObjectId::new(actor.actor_id()),
-      object: pm.ap_id.clone().into(),
+      object: ObjectId::new(pm.ap_id.clone()),
       kind: DeleteType::Delete,
       id: generate_activity_id(
         DeleteType::Delete,
@@ -65,34 +68,31 @@ impl DeletePrivateMessage {
     let recipient_id = pm.recipient_id;
     let recipient =
       blocking(context.pool(), move |conn| Person::read(conn, recipient_id)).await??;
-    let inbox = vec![recipient.get_shared_inbox_or_inbox_url()];
+    let inbox = vec![recipient.shared_inbox_or_inbox_url()];
-    send_activity_new(context, &delete, &delete_id, actor, inbox, true).await
+    send_lemmy_activity(context, &delete, &delete_id, actor, inbox, true).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for DeletePrivateMessage {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person(&self.actor, context, request_counter).await?;
-    verify_domains_match(self.actor.inner(), &self.object)?;
+    verify_domains_match(self.actor.inner(), self.object.inner())?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     _request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
-    let ap_id = self.object.clone();
+    let private_message = self.object.dereference_local(context).await?;
-    let private_message = blocking(context.pool(), move |conn| {
-      PrivateMessage::read_from_apub_id(conn, &ap_id.into())
-    })
-    .await??;
     let deleted_private_message = blocking(context.pool(), move |conn| {
       PrivateMessage::update_deleted(conn, private_message.id, true)
     })

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

@@ -5,10 +5,9 @@ use crate::{
     verify_activity,
     verify_person,
   },
-  activity_queue::send_activity_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
+  send_lemmy_activity,
 };
 use activitystreams::{
   activity::kind::UndoType,
@@ -17,8 +16,12 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, verify_urls_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
-use lemmy_db_queries::{source::private_message::PrivateMessage_, ApubObject, Crud};
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  verify::{verify_domains_match, verify_urls_match},
+};
+use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
 use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
@@ -64,36 +67,34 @@ impl UndoDeletePrivateMessage {
       context: lemmy_context(),
       unparsed: Default::default(),
     };
-    let inbox = vec![recipient.get_shared_inbox_or_inbox_url()];
+    let inbox = vec![recipient.shared_inbox_or_inbox_url()];
-    send_activity_new(context, &undo, &id, actor, inbox, true).await
+    send_lemmy_activity(context, &undo, &id, actor, inbox, true).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoDeletePrivateMessage {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
     verify_person(&self.actor, context, request_counter).await?;
     verify_urls_match(self.actor(), self.object.actor())?;
-    verify_domains_match(self.actor(), &self.object.object)?;
+    verify_domains_match(self.actor(), self.object.object.inner())?;
     self.object.verify(context, request_counter).await?;
     Ok(())
   }
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     _request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let ap_id = self.object.object.clone();
-    let private_message = blocking(context.pool(), move |conn| {
+    let private_message = ap_id.dereference_local(context).await?;
-      PrivateMessage::read_from_apub_id(conn, &ap_id.into())
-    })
-    .await??;
 
     let deleted_private_message = blocking(context.pool(), move |conn| {
       PrivateMessage::update_deleted(conn, private_message.id, false)

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

@@ -1,66 +0,0 @@
-use crate::{check_is_apub_id_valid, ActorType, CommunityType};
-use itertools::Itertools;
-use lemmy_api_common::blocking;
-use lemmy_db_queries::DbPool;
-use lemmy_db_schema::source::community::Community;
-use lemmy_db_views_actor::community_follower_view::CommunityFollowerView;
-use lemmy_utils::{settings::structs::Settings, LemmyError};
-use url::Url;
-
-impl ActorType for Community {
-  fn is_local(&self) -> bool {
-    self.local
-  }
-  fn actor_id(&self) -> Url {
-    self.actor_id.to_owned().into()
-  }
-  fn name(&self) -> String {
-    self.name.clone()
-  }
-  fn public_key(&self) -> Option<String> {
-    self.public_key.to_owned()
-  }
-  fn private_key(&self) -> Option<String> {
-    self.private_key.to_owned()
-  }
-
-  fn get_shared_inbox_or_inbox_url(&self) -> Url {
-    self
-      .shared_inbox_url
-      .clone()
-      .unwrap_or_else(|| self.inbox_url.to_owned())
-      .into()
-  }
-}
-
-#[async_trait::async_trait(?Send)]
-impl CommunityType for Community {
-  fn followers_url(&self) -> Url {
-    self.followers_url.clone().into()
-  }
-
-  /// For a given community, returns the inboxes of all followers.
-  async fn get_follower_inboxes(
-    &self,
-    pool: &DbPool,
-    settings: &Settings,
-  ) -> Result<Vec<Url>, LemmyError> {
-    let id = self.id;
-
-    let follows = blocking(pool, move |conn| {
-      CommunityFollowerView::for_community(conn, id)
-    })
-    .await??;
-    let inboxes = follows
-      .into_iter()
-      .filter(|f| !f.follower.local)
-      .map(|f| f.follower.shared_inbox_url.unwrap_or(f.follower.inbox_url))
-      .map(|i| i.into_inner())
-      .unique()
-      // Don't send to blocked instances
-      .filter(|inbox| check_is_apub_id_valid(inbox, false, settings).is_ok())
-      .collect();
-
-    Ok(inboxes)
-  }
-}

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

@@ -1,2 +0,0 @@
-pub(crate) mod community;
-pub(crate) mod person;

crates/apub/src/activities/send/person.rs

@@ -1,31 +0,0 @@
-use crate::ActorType;
-use lemmy_db_schema::source::person::Person;
-use url::Url;
-
-impl ActorType for Person {
-  fn is_local(&self) -> bool {
-    self.local
-  }
-  fn actor_id(&self) -> Url {
-    self.actor_id.to_owned().into_inner()
-  }
-  fn name(&self) -> String {
-    self.name.clone()
-  }
-
-  fn public_key(&self) -> Option<String> {
-    self.public_key.to_owned()
-  }
-
-  fn private_key(&self) -> Option<String> {
-    self.private_key.to_owned()
-  }
-
-  fn get_shared_inbox_or_inbox_url(&self) -> Url {
-    self
-      .shared_inbox_url
-      .clone()
-      .unwrap_or_else(|| self.inbox_url.to_owned())
-      .into()
-  }
-}

crates/apub/src/activities/undo_remove.rs

@@ -12,7 +12,11 @@ use activitystreams::{
   primitives::OneOrMany,
   unparsed::Unparsed,
 };
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler},
+  values::PublicUrl,
+};
 use lemmy_db_schema::source::{community::Community, person::Person};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
@@ -38,9 +42,10 @@ pub struct UndoRemovePostCommentOrCommunity {
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoRemovePostCommentOrCommunity {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -60,7 +65,7 @@ impl ActivityHandler for UndoRemovePostCommentOrCommunity {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     _request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     UndoDelete::receive_undo_remove_action(self.object.object.inner(), context).await

crates/apub/src/activities/voting/undo_vote.rs

@@ -1,6 +1,6 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_person_in_community,
@@ -10,10 +10,8 @@ use crate::{
       vote::{Vote, VoteType},
     },
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
   PostOrComment,
 };
 use activitystreams::{
@@ -23,7 +21,12 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, verify_urls_match, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+  verify::verify_urls_match,
+};
 use lemmy_db_queries::Crud;
 use lemmy_db_schema::{
   source::{community::Community, person::Person},
@@ -80,15 +83,16 @@ impl UndoVote {
       unparsed: Default::default(),
     };
     let activity = AnnouncableActivities::UndoVote(undo_vote);
-    send_to_community_new(activity, &id, actor, &community, vec![], context).await
+    send_to_community(activity, &id, actor, &community, vec![], context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for UndoVote {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -100,7 +104,7 @@ impl ActivityHandler for UndoVote {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;

crates/apub/src/activities/voting/vote.rs

@@ -1,21 +1,23 @@
 use crate::{
   activities::{
-    community::announce::AnnouncableActivities,
+    community::{announce::AnnouncableActivities, send_to_community},
     generate_activity_id,
     verify_activity,
     verify_person_in_community,
     voting::{vote_comment, vote_post},
   },
-  activity_queue::send_to_community_new,
+  context::lemmy_context,
-  extensions::context::lemmy_context,
   fetcher::object_id::ObjectId,
-  ActorType,
   PostOrComment,
 };
 use activitystreams::{base::AnyBase, primitives::OneOrMany, unparsed::Unparsed};
 use anyhow::anyhow;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{values::PublicUrl, ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  traits::{ActivityFields, ActivityHandler, ActorType},
+  values::PublicUrl,
+};
 use lemmy_db_queries::Crud;
 use lemmy_db_schema::{
   source::{community::Community, person::Person},
@@ -106,15 +108,16 @@ impl Vote {
     let vote_id = vote.id.clone();
 
     let activity = AnnouncableActivities::Vote(vote);
-    send_to_community_new(activity, &vote_id, actor, &community, vec![], context).await
+    send_to_community(activity, &vote_id, actor, &community, vec![], context).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
 impl ActivityHandler for Vote {
+  type DataType = LemmyContext;
   async fn verify(
     &self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_activity(self, &context.settings())?;
@@ -124,7 +127,7 @@ impl ActivityHandler for Vote {
 
   async fn receive(
     self,
-    context: &LemmyContext,
+    context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     let actor = self.actor.dereference(context, request_counter).await?;

crates/apub/src/extensions/mod.rs

@@ -1,2 +0,0 @@
-pub mod context;
-pub mod signatures;

crates/apub/src/fetcher/community.rs

@@ -6,7 +6,7 @@ use crate::{
 use activitystreams::collection::{CollectionExt, OrderedCollection};
 use anyhow::Context;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::ActivityHandler;
+use lemmy_apub_lib::{data::Data, traits::ActivityHandler};
 use lemmy_db_queries::Joinable;
 use lemmy_db_schema::source::{
   community::{Community, CommunityModerator, CommunityModeratorForm},
@@ -91,7 +91,9 @@ pub(crate) async fn fetch_community_outbox(
     //       AnnounceActivity as inner type, but that gives me stackoverflow
     let ser = serde_json::to_string(&announce)?;
     let announce: AnnounceActivity = serde_json::from_str(&ser)?;
-    announce.receive(context, recursion_counter).await?;
+    announce
+      .receive(&Data::new(context.clone()), recursion_counter)
+      .await?;
   }
 
   Ok(())

crates/apub/src/fetcher/deletable_apub_object.rs

@@ -6,7 +6,13 @@ use lemmy_db_queries::source::{
   person::Person_,
   post::Post_,
 };
-use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
+use lemmy_db_schema::source::{
+  comment::Comment,
+  community::Community,
+  person::Person,
+  post::Post,
+  private_message::PrivateMessage,
+};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
 
@@ -83,3 +89,11 @@ impl DeletableApubObject for PostOrComment {
     Ok(())
   }
 }
+
+#[async_trait::async_trait(?Send)]
+impl DeletableApubObject for PrivateMessage {
+  async fn delete(self, _context: &LemmyContext) -> Result<(), LemmyError> {
+    // do nothing, because pm can't be fetched over http
+    unimplemented!()
+  }
+}

crates/apub/src/fetcher/fetch.rs

@@ -1,5 +1,6 @@
-use crate::{check_is_apub_id_valid, APUB_JSON_CONTENT_TYPE};
+use crate::check_is_apub_id_valid;
 use anyhow::anyhow;
+use lemmy_apub_lib::APUB_JSON_CONTENT_TYPE;
 use lemmy_utils::{request::retry, settings::structs::Settings, LemmyError};
 use log::info;
 use reqwest::Client;

crates/apub/src/fetcher/mod.rs

@@ -5,8 +5,9 @@ pub mod object_id;
 pub mod post_or_comment;
 pub mod search;
 
-use crate::{fetcher::object_id::ObjectId, ActorType};
+use crate::fetcher::object_id::ObjectId;
 use chrono::NaiveDateTime;
+use lemmy_apub_lib::traits::ActorType;
 use lemmy_db_schema::{
   naive_now,
   source::{community::Community, person::Person},

crates/apub/src/fetcher/object_id.rs

@@ -1,12 +1,12 @@
 use crate::{
   fetcher::{deletable_apub_object::DeletableApubObject, should_refetch_actor},
   objects::FromApub,
-  APUB_JSON_CONTENT_TYPE,
 };
 use anyhow::anyhow;
-use diesel::NotFound;
+use diesel::{NotFound, PgConnection};
 use lemmy_api_common::blocking;
-use lemmy_db_queries::{ApubObject, DbPool};
+use lemmy_apub_lib::{traits::ApubObject, APUB_JSON_CONTENT_TYPE};
+use lemmy_db_queries::DbPool;
 use lemmy_db_schema::DbUrl;
 use lemmy_utils::{request::retry, settings::structs::Settings, LemmyError};
 use lemmy_websocket::LemmyContext;
@@ -26,12 +26,12 @@ static REQUEST_LIMIT: i32 = 25;
 #[derive(Clone, PartialEq, Serialize, Deserialize, Debug)]
 pub struct ObjectId<Kind>(Url, #[serde(skip)] PhantomData<Kind>)
 where
-  Kind: FromApub + ApubObject + DeletableApubObject + Send + 'static,
+  Kind: FromApub + ApubObject<DataType = PgConnection> + DeletableApubObject + Send + 'static,
   for<'de2> <Kind as FromApub>::ApubType: serde::Deserialize<'de2>;
 
 impl<Kind> ObjectId<Kind>
 where
-  Kind: FromApub + ApubObject + DeletableApubObject + Send + 'static,
+  Kind: FromApub + ApubObject<DataType = PgConnection> + DeletableApubObject + Send + 'static,
   for<'de> <Kind as FromApub>::ApubType: serde::Deserialize<'de>,
 {
   pub fn new<T>(url: T) -> Self
@@ -46,12 +46,12 @@ where
   }
 
   /// Fetches an activitypub object, either from local database (if possible), or over http.
-  pub(crate) async fn dereference(
+  pub async fn dereference(
     &self,
     context: &LemmyContext,
     request_counter: &mut i32,
   ) -> Result<Kind, LemmyError> {
-    let db_object = self.dereference_locally(context.pool()).await?;
+    let db_object = self.dereference_from_db(context.pool()).await?;
 
     // if its a local object, only fetch it from the database and not over http
     if self.0.domain() == Some(&Settings::get().get_hostname_without_port()?) {
@@ -66,30 +66,32 @@ where
         // TODO: rename to should_refetch_object()
         if should_refetch_actor(last_refreshed_at) {
           return self
-            .dereference_remotely(context, request_counter, Some(object))
+            .dereference_from_http(context, request_counter, Some(object))
             .await;
         }
       }
       Ok(object)
     } else {
       self
-        .dereference_remotely(context, request_counter, None)
+        .dereference_from_http(context, request_counter, None)
         .await
     }
   }
 
-  /// returning none means the object was not found in local db
+  /// Fetch an object from the local db. Instead of falling back to http, this throws an error if
-  async fn dereference_locally(&self, pool: &DbPool) -> Result<Option<Kind>, LemmyError> {
+  /// the object is not found in the database.
-    let id: DbUrl = self.0.clone().into();
+  pub async fn dereference_local(&self, context: &LemmyContext) -> Result<Kind, LemmyError> {
-    let object = blocking(pool, move |conn| ApubObject::read_from_apub_id(conn, &id)).await?;
+    let object = self.dereference_from_db(context.pool()).await?;
-    match object {
+    object.ok_or_else(|| anyhow!("object not found in database {}", self).into())
-      Ok(o) => Ok(Some(o)),
-      Err(NotFound {}) => Ok(None),
-      Err(e) => Err(e.into()),
-    }
   }
 
-  async fn dereference_remotely(
+  /// returning none means the object was not found in local db
+  async fn dereference_from_db(&self, pool: &DbPool) -> Result<Option<Kind>, LemmyError> {
+    let id = self.0.clone();
+    blocking(pool, move |conn| ApubObject::read_from_apub_id(conn, id)).await?
+  }
+
+  async fn dereference_from_http(
     &self,
     context: &LemmyContext,
     request_counter: &mut i32,
@@ -128,7 +130,7 @@ where
 
 impl<Kind> Display for ObjectId<Kind>
 where
-  Kind: FromApub + ApubObject + DeletableApubObject + Send + 'static,
+  Kind: FromApub + ApubObject<DataType = PgConnection> + DeletableApubObject + Send + 'static,
   for<'de> <Kind as FromApub>::ApubType: serde::Deserialize<'de>,
 {
   fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
@@ -138,7 +140,7 @@ where
 
 impl<Kind> From<ObjectId<Kind>> for Url
 where
-  Kind: FromApub + ApubObject + DeletableApubObject + Send + 'static,
+  Kind: FromApub + ApubObject<DataType = PgConnection> + DeletableApubObject + Send + 'static,
   for<'de> <Kind as FromApub>::ApubType: serde::Deserialize<'de>,
 {
   fn from(id: ObjectId<Kind>) -> Self {
@@ -148,7 +150,7 @@ where
 
 impl<Kind> From<ObjectId<Kind>> for DbUrl
 where
-  Kind: FromApub + ApubObject + DeletableApubObject + Send + 'static,
+  Kind: FromApub + ApubObject<DataType = PgConnection> + DeletableApubObject + Send + 'static,
   for<'de> <Kind as FromApub>::ApubType: serde::Deserialize<'de>,
 {
   fn from(id: ObjectId<Kind>) -> Self {

crates/apub/src/fetcher/objects.rs

@@ -1,107 +0,0 @@
-use crate::{
-  fetcher::fetch::fetch_remote_object,
-  objects::{comment::Note, post::Page, FromApub},
-  PostOrComment,
-};
-use anyhow::anyhow;
-use diesel::result::Error::NotFound;
-use lemmy_api_common::blocking;
-use lemmy_db_queries::{ApubObject, Crud};
-use lemmy_db_schema::source::{comment::Comment, post::Post};
-use lemmy_utils::LemmyError;
-use lemmy_websocket::LemmyContext;
-use log::debug;
-use url::Url;
-
-/// Gets a post by its apub ID. If it exists locally, it is returned directly. Otherwise it is
-/// pulled from its apub ID, inserted and returned.
-///
-/// The parent community is also pulled if necessary. Comments are not pulled.
-pub(crate) async fn get_or_fetch_and_insert_post(
-  post_ap_id: &Url,
-  context: &LemmyContext,
-  recursion_counter: &mut i32,
-) -> Result<Post, LemmyError> {
-  let post_ap_id_owned = post_ap_id.to_owned();
-  let post = blocking(context.pool(), move |conn| {
-    Post::read_from_apub_id(conn, &post_ap_id_owned.into())
-  })
-  .await?;
-
-  match post {
-    Ok(p) => Ok(p),
-    Err(NotFound {}) => {
-      debug!("Fetching and creating remote post: {}", post_ap_id);
-      let page = fetch_remote_object::<Page>(
-        context.client(),
-        &context.settings(),
-        post_ap_id,
-        recursion_counter,
-      )
-      .await?;
-      let post = Post::from_apub(&page, context, post_ap_id, recursion_counter).await?;
-
-      Ok(post)
-    }
-    Err(e) => Err(e.into()),
-  }
-}
-
-/// Gets a comment by its apub ID. If it exists locally, it is returned directly. Otherwise it is
-/// pulled from its apub ID, inserted and returned.
-///
-/// The parent community, post and comment are also pulled if necessary.
-pub(crate) async fn get_or_fetch_and_insert_comment(
-  comment_ap_id: &Url,
-  context: &LemmyContext,
-  recursion_counter: &mut i32,
-) -> Result<Comment, LemmyError> {
-  let comment_ap_id_owned = comment_ap_id.to_owned();
-  let comment = blocking(context.pool(), move |conn| {
-    Comment::read_from_apub_id(conn, &comment_ap_id_owned.into())
-  })
-  .await?;
-
-  match comment {
-    Ok(p) => Ok(p),
-    Err(NotFound {}) => {
-      debug!(
-        "Fetching and creating remote comment and its parents: {}",
-        comment_ap_id
-      );
-      let comment = fetch_remote_object::<Note>(
-        context.client(),
-        &context.settings(),
-        comment_ap_id,
-        recursion_counter,
-      )
-      .await?;
-      let comment = Comment::from_apub(&comment, context, comment_ap_id, recursion_counter).await?;
-
-      let post_id = comment.post_id;
-      let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
-      if post.locked {
-        return Err(anyhow!("Post is locked").into());
-      }
-
-      Ok(comment)
-    }
-    Err(e) => Err(e.into()),
-  }
-}
-
-pub(crate) async fn get_or_fetch_and_insert_post_or_comment(
-  ap_id: &Url,
-  context: &LemmyContext,
-  recursion_counter: &mut i32,
-) -> Result<PostOrComment, LemmyError> {
-  Ok(
-    match get_or_fetch_and_insert_post(ap_id, context, recursion_counter).await {
-      Ok(p) => PostOrComment::Post(Box::new(p)),
-      Err(_) => {
-        let c = get_or_fetch_and_insert_comment(ap_id, context, recursion_counter).await?;
-        PostOrComment::Comment(Box::new(c))
-      }
-    },
-  )
-}

crates/apub/src/fetcher/person.rs

@@ -1,80 +0,0 @@
-use crate::{
-  fetcher::{fetch::fetch_remote_object, is_deleted, should_refetch_actor},
-  objects::{person::Person as ApubPerson, FromApub},
-};
-use anyhow::anyhow;
-use diesel::result::Error::NotFound;
-use lemmy_api_common::blocking;
-use lemmy_db_queries::{source::person::Person_, ApubObject};
-use lemmy_db_schema::source::person::Person;
-use lemmy_utils::LemmyError;
-use lemmy_websocket::LemmyContext;
-use log::debug;
-use url::Url;
-
-/// Get a person from its apub ID.
-///
-/// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
-/// Otherwise it is fetched from the remote instance, stored and returned.
-pub(crate) async fn get_or_fetch_and_upsert_person(
-  apub_id: &Url,
-  context: &LemmyContext,
-  recursion_counter: &mut i32,
-) -> Result<Person, LemmyError> {
-  let apub_id_owned = apub_id.to_owned();
-  let person = blocking(context.pool(), move |conn| {
-    Person::read_from_apub_id(conn, &apub_id_owned.into())
-  })
-  .await?;
-
-  match person {
-    // If its older than a day, re-fetch it
-    Ok(u) if !u.local && should_refetch_actor(u.last_refreshed_at) => {
-      debug!("Fetching and updating from remote person: {}", apub_id);
-      let person = fetch_remote_object::<ApubPerson>(
-        context.client(),
-        &context.settings(),
-        apub_id,
-        recursion_counter,
-      )
-      .await;
-
-      if is_deleted(&person) {
-        // TODO: use Person::update_deleted() once implemented
-        blocking(context.pool(), move |conn| {
-          Person::delete_account(conn, u.id)
-        })
-        .await??;
-        return Err(anyhow!("Person was deleted by remote instance").into());
-      } else if person.is_err() {
-        return Ok(u);
-      }
-
-      let person = Person::from_apub(&person?, context, apub_id, recursion_counter).await?;
-
-      let person_id = person.id;
-      blocking(context.pool(), move |conn| {
-        Person::mark_as_updated(conn, person_id)
-      })
-      .await??;
-
-      Ok(person)
-    }
-    Ok(u) => Ok(u),
-    Err(NotFound {}) => {
-      debug!("Fetching and creating remote person: {}", apub_id);
-      let person = fetch_remote_object::<ApubPerson>(
-        context.client(),
-        &context.settings(),
-        apub_id,
-        recursion_counter,
-      )
-      .await?;
-
-      let person = Person::from_apub(&person, context, apub_id, recursion_counter).await?;
-
-      Ok(person)
-    }
-    Err(e) => Err(e.into()),
-  }
-}

crates/apub/src/fetcher/post_or_comment.rs

@@ -1,13 +1,10 @@
 use crate::objects::{comment::Note, post::Page, FromApub};
 use activitystreams::chrono::NaiveDateTime;
-use diesel::{result::Error, PgConnection};
+use diesel::PgConnection;
-use lemmy_db_queries::ApubObject;
+use lemmy_apub_lib::traits::ApubObject;
-use lemmy_db_schema::{
+use lemmy_db_schema::source::{
-  source::{
   comment::{Comment, CommentForm},
   post::{Post, PostForm},
-  },
-  DbUrl,
 };
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
@@ -33,19 +30,23 @@ pub enum PageOrNote {
 
 #[async_trait::async_trait(?Send)]
 impl ApubObject for PostOrComment {
+  type DataType = PgConnection;
+
   fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
     None
   }
 
   // TODO: this can probably be implemented using a single sql query
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error>
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError>
   where
     Self: Sized,
   {
-    let post = Post::read_from_apub_id(conn, object_id);
+    let post = Post::read_from_apub_id(conn, object_id.clone())?;
     Ok(match post {
-      Ok(o) => PostOrComment::Post(Box::new(o)),
+      Some(o) => Some(PostOrComment::Post(Box::new(o))),
-      Err(_) => PostOrComment::Comment(Box::new(Comment::read_from_apub_id(conn, object_id)?)),
+      None => {
+        Comment::read_from_apub_id(conn, object_id)?.map(|c| PostOrComment::Comment(Box::new(c)))
+      }
     })
   }
 }

crates/apub/src/fetcher/search.rs

@@ -4,19 +4,18 @@ use crate::{
 };
 use activitystreams::chrono::NaiveDateTime;
 use anyhow::anyhow;
-use diesel::{result::Error, PgConnection};
+use diesel::PgConnection;
 use itertools::Itertools;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::webfinger::{webfinger_resolve_actor, WebfingerType};
+use lemmy_apub_lib::{
+  traits::ApubObject,
+  webfinger::{webfinger_resolve_actor, WebfingerType},
+};
 use lemmy_db_queries::{
   source::{community::Community_, person::Person_},
-  ApubObject,
   DbPool,
 };
-use lemmy_db_schema::{
+use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
-  source::{comment::Comment, community::Community, person::Person, post::Post},
-  DbUrl,
-};
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
 use serde::Deserialize;
@@ -102,6 +101,8 @@ pub enum SearchableApubTypes {
 }
 
 impl ApubObject for SearchableObjects {
+  type DataType = PgConnection;
+
   fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
     match self {
       SearchableObjects::Person(p) => p.last_refreshed_at(),
@@ -114,23 +115,26 @@ impl ApubObject for SearchableObjects {
   // TODO: this is inefficient, because if the object is not in local db, it will run 4 db queries
   //       before finally returning an error. it would be nice if we could check all 4 tables in
   //       a single query.
-  //       we could skip this and always return an error, but then it would not be able to mark
+  //       we could skip this and always return an error, but then it would always fetch objects
-  //       objects as deleted that were deleted by remote server.
+  //       over http, and not be able to mark objects as deleted that were deleted by remote server.
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error> {
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
-    let c = Community::read_from_apub_id(conn, object_id);
+    let c = Community::read_from_apub_id(conn, object_id.clone())?;
-    if let Ok(c) = c {
+    if let Some(c) = c {
-      return Ok(SearchableObjects::Community(c));
+      return Ok(Some(SearchableObjects::Community(c)));
     }
-    let p = Person::read_from_apub_id(conn, object_id);
+    let p = Person::read_from_apub_id(conn, object_id.clone())?;
-    if let Ok(p) = p {
+    if let Some(p) = p {
-      return Ok(SearchableObjects::Person(p));
+      return Ok(Some(SearchableObjects::Person(p)));
     }
-    let p = Post::read_from_apub_id(conn, object_id);
+    let p = Post::read_from_apub_id(conn, object_id.clone())?;
-    if let Ok(p) = p {
+    if let Some(p) = p {
-      return Ok(SearchableObjects::Post(p));
+      return Ok(Some(SearchableObjects::Post(p)));
     }
-    let c = Comment::read_from_apub_id(conn, object_id);
+    let c = Comment::read_from_apub_id(conn, object_id)?;
-    Ok(SearchableObjects::Comment(c?))
+    if let Some(c) = c {
+      return Ok(Some(SearchableObjects::Comment(c)));
+    }
+    Ok(None)
   }
 }
 

crates/apub/src/http/community.rs

@@ -4,8 +4,9 @@ use crate::{
     extract_community,
     following::{follow::FollowCommunity, undo::UndoFollowCommunity},
   },
-  extensions::context::lemmy_context,
+  context::lemmy_context,
   generate_moderators_url,
+  generate_outbox_url,
   http::{
     create_apub_response,
     create_apub_tombstone_response,
@@ -13,7 +14,6 @@ use crate::{
     receive_activity,
   },
   objects::ToApub,
-  ActorType,
 };
 use activitystreams::{
   base::{AnyBase, BaseExt},
@@ -22,7 +22,7 @@ use activitystreams::{
 };
 use actix_web::{body::Body, web, web::Payload, HttpRequest, HttpResponse};
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{ActivityFields, ActivityHandler};
+use lemmy_apub_lib::traits::{ActivityFields, ActivityHandler};
 use lemmy_db_queries::source::{activity::Activity_, community::Community_};
 use lemmy_db_schema::source::{activity::Activity, community::Community};
 use lemmy_db_views_actor::{
@@ -60,6 +60,7 @@ pub(crate) async fn get_apub_community_http(
 
 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
 #[serde(untagged)]
+#[activity_handler(LemmyContext)]
 pub enum GroupInboxActivities {
   FollowCommunity(FollowCommunity),
   UndoFollowCommunity(UndoFollowCommunity),
@@ -146,7 +147,7 @@ pub(crate) async fn get_apub_community_outbox(
   collection
     .set_many_items(activities)
     .set_many_contexts(lemmy_context())
-    .set_id(community.get_outbox_url()?)
+    .set_id(generate_outbox_url(&community.actor_id)?.into())
     .set_total_items(len as u64);
   Ok(create_apub_response(&collection))
 }

crates/apub/src/http/mod.rs

@@ -1,13 +1,11 @@
 use crate::{
   check_is_apub_id_valid,
-  extensions::signatures::verify_signature,
   fetcher::get_or_fetch_and_upsert_actor,
   http::{
     community::{receive_group_inbox, GroupInboxActivities},
     person::{receive_person_inbox, PersonInboxActivities},
   },
   insert_activity,
-  APUB_JSON_CONTENT_TYPE,
 };
 use actix_web::{
   body::Body,
@@ -20,7 +18,12 @@ use anyhow::{anyhow, Context};
 use futures::StreamExt;
 use http::StatusCode;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{ActivityFields, ActivityHandler};
+use lemmy_apub_lib::{
+  data::Data,
+  signatures::verify_signature,
+  traits::{ActivityFields, ActivityHandler},
+  APUB_JSON_CONTENT_TYPE,
+};
 use lemmy_db_queries::{source::activity::Activity_, DbPool};
 use lemmy_db_schema::source::activity::Activity;
 use lemmy_utils::{location_info, LemmyError};
@@ -38,6 +41,7 @@ pub mod routes;
 
 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
 #[serde(untagged)]
+#[activity_handler(LemmyContext)]
 pub enum SharedInboxActivities {
   GroupInboxActivities(GroupInboxActivities),
   // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
@@ -80,7 +84,7 @@ async fn receive_activity<'a, T>(
   context: &LemmyContext,
 ) -> Result<HttpResponse, LemmyError>
 where
-  T: ActivityHandler
+  T: ActivityHandler<DataType = LemmyContext>
     + ActivityFields
     + Clone
     + Deserialize<'a>
@@ -100,7 +104,9 @@ where
   }
   check_is_apub_id_valid(activity.actor(), false, &context.settings())?;
   info!("Verifying activity {}", activity.id_unchecked().to_string());
-  activity.verify(context, request_counter).await?;
+  activity
+    .verify(&Data::new(context.clone()), request_counter)
+    .await?;
   assert_activity_not_local(&activity, &context.settings().hostname)?;
 
   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
@@ -115,7 +121,9 @@ where
   .await?;
 
   info!("Receiving activity {}", activity.id_unchecked().to_string());
-  activity.receive(context, request_counter).await?;
+  activity
+    .receive(&Data::new(context.clone()), request_counter)
+    .await?;
   Ok(HttpResponse::Ok().finish())
 }
 

crates/apub/src/http/person.rs

@@ -8,7 +8,8 @@ use crate::{
       undo_delete::UndoDeletePrivateMessage,
     },
   },
-  extensions::context::lemmy_context,
+  context::lemmy_context,
+  generate_outbox_url,
   http::{
     create_apub_response,
     create_apub_tombstone_response,
@@ -16,7 +17,6 @@ use crate::{
     receive_activity,
   },
   objects::ToApub,
-  ActorType,
 };
 use activitystreams::{
   base::BaseExt,
@@ -24,7 +24,7 @@ use activitystreams::{
 };
 use actix_web::{body::Body, web, web::Payload, HttpRequest, HttpResponse};
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::{ActivityFields, ActivityHandler};
+use lemmy_apub_lib::traits::{ActivityFields, ActivityHandler};
 use lemmy_db_queries::source::person::Person_;
 use lemmy_db_schema::source::person::Person;
 use lemmy_utils::LemmyError;
@@ -61,6 +61,7 @@ pub(crate) async fn get_apub_person_http(
 
 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
 #[serde(untagged)]
+#[activity_handler(LemmyContext)]
 pub enum PersonInboxActivities {
   AcceptFollowCommunity(AcceptFollowCommunity),
   /// Some activities can also be sent from user to user, eg a comment with mentions
@@ -104,7 +105,7 @@ pub(crate) async fn get_apub_person_outbox(
   collection
     .set_many_items(Vec::<Url>::new())
     .set_many_contexts(lemmy_context())
-    .set_id(person.get_outbox_url()?)
+    .set_id(generate_outbox_url(&person.actor_id)?.into())
     .set_total_items(0_u64);
   Ok(create_apub_response(&collection))
 }

crates/apub/src/http/routes.rs

@@ -1,5 +1,4 @@
-use crate::{
+use crate::http::{
-  http::{
   comment::get_apub_comment,
   community::{
     community_inbox,
@@ -13,11 +12,10 @@ use crate::{
   person::{get_apub_person_http, get_apub_person_inbox, get_apub_person_outbox, person_inbox},
   post::get_apub_post,
   shared_inbox,
-  },
-  APUB_JSON_CONTENT_TYPE,
 };
 use actix_web::*;
 use http_signature_normalization_actix::digest::middleware::VerifyDigest;
+use lemmy_apub_lib::APUB_JSON_CONTENT_TYPE;
 use lemmy_utils::settings::structs::Settings;
 use sha2::{Digest, Sha256};
 

crates/apub/src/lib.rs

@@ -1,17 +1,14 @@
-#[macro_use]
-extern crate lazy_static;
-
 pub mod activities;
-pub mod activity_queue;
+mod context;
-pub mod extensions;
 pub mod fetcher;
 pub mod http;
 pub mod migrations;
 pub mod objects;
 
-use crate::{extensions::signatures::PublicKey, fetcher::post_or_comment::PostOrComment};
+use crate::fetcher::post_or_comment::PostOrComment;
 use anyhow::{anyhow, Context};
 use lemmy_api_common::blocking;
+use lemmy_apub_lib::{activity_queue::send_activity, traits::ActorType};
 use lemmy_db_queries::{source::activity::Activity_, DbPool};
 use lemmy_db_schema::{
   source::{activity::Activity, person::Person},
@@ -20,12 +17,12 @@ use lemmy_db_schema::{
 };
 use lemmy_db_views_actor::community_person_ban_view::CommunityPersonBanView;
 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
+use lemmy_websocket::LemmyContext;
+use log::info;
 use serde::Serialize;
 use std::net::IpAddr;
 use url::{ParseError, Url};
 
-static APUB_JSON_CONTENT_TYPE: &str = "application/activity+json";
-
 /// Checks if the ID is allowed for sending or receiving.
 ///
 /// In particular, it checks for:
@@ -92,39 +89,6 @@ pub(crate) fn check_is_apub_id_valid(
   Ok(())
 }
 
-/// Common methods provided by ActivityPub actors (community and person). Not all methods are
-/// implemented by all actors.
-trait ActorType {
-  fn is_local(&self) -> bool;
-  fn actor_id(&self) -> Url;
-  fn name(&self) -> String;
-
-  // TODO: every actor should have a public key, so this shouldnt be an option (needs to be fixed in db)
-  fn public_key(&self) -> Option<String>;
-  fn private_key(&self) -> Option<String>;
-
-  fn get_shared_inbox_or_inbox_url(&self) -> Url;
-
-  /// Outbox URL is not generally used by Lemmy, so it can be generated on the fly (but only for
-  /// local actors).
-  fn get_outbox_url(&self) -> Result<Url, LemmyError> {
-    /* TODO
-    if !self.is_local() {
-      return Err(anyhow!("get_outbox_url() called for remote actor").into());
-    }
-    */
-    Ok(Url::parse(&format!("{}/outbox", &self.actor_id()))?)
-  }
-
-  fn get_public_key(&self) -> Result<PublicKey, LemmyError> {
-    Ok(PublicKey {
-      id: format!("{}#main-key", self.actor_id()),
-      owner: self.actor_id(),
-      public_key_pem: self.public_key().context(location_info!())?,
-    })
-  }
-}
-
 #[async_trait::async_trait(?Send)]
 pub trait CommunityType {
   fn followers_url(&self) -> Url;
@@ -192,6 +156,10 @@ pub fn generate_shared_inbox_url(actor_id: &DbUrl) -> Result<DbUrl, LemmyError>
   Ok(Url::parse(&url)?.into())
 }
 
+pub fn generate_outbox_url(actor_id: &DbUrl) -> Result<DbUrl, ParseError> {
+  Ok(Url::parse(&format!("{}/outbox", actor_id))?.into())
+}
+
 fn generate_moderators_url(community_id: &DbUrl) -> Result<DbUrl, LemmyError> {
   Ok(Url::parse(&format!("{}/moderators", community_id))?.into())
 }
@@ -254,3 +222,46 @@ async fn check_community_or_site_ban(
 
   Ok(())
 }
+
+pub(crate) async fn send_lemmy_activity<T: Serialize>(
+  context: &LemmyContext,
+  activity: &T,
+  activity_id: &Url,
+  actor: &dyn ActorType,
+  inboxes: Vec<Url>,
+  sensitive: bool,
+) -> Result<(), LemmyError> {
+  if !context.settings().federation.enabled || inboxes.is_empty() {
+    return Ok(());
+  }
+
+  info!("Sending activity {}", activity_id.to_string());
+
+  // Don't send anything to ourselves
+  // TODO: this should be a debug assert
+  let hostname = context.settings().get_hostname_without_port()?;
+  let inboxes: Vec<&Url> = inboxes
+    .iter()
+    .filter(|i| i.domain().expect("valid inbox url") != hostname)
+    .collect();
+
+  let serialised_activity = serde_json::to_string(&activity)?;
+
+  insert_activity(
+    activity_id,
+    serialised_activity.clone(),
+    true,
+    sensitive,
+    context.pool(),
+  )
+  .await?;
+
+  send_activity(
+    serialised_activity,
+    actor,
+    inboxes,
+    context.client(),
+    context.activity_queue(),
+  )
+  .await
+}

crates/apub/src/objects/comment.rs

@@ -1,10 +1,9 @@
 use crate::{
   activities::verify_person_in_community,
-  extensions::context::lemmy_context,
+  context::lemmy_context,
   fetcher::object_id::ObjectId,
   migrations::CommentInReplyToMigration,
   objects::{create_tombstone, FromApub, Source, ToApub},
-  ActorType,
   PostOrComment,
 };
 use activitystreams::{
@@ -17,8 +16,9 @@ use anyhow::{anyhow, Context};
 use chrono::{DateTime, FixedOffset};
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
+  traits::ActorType,
   values::{MediaTypeHtml, MediaTypeMarkdown, PublicUrl},
-  verify_domains_match,
+  verify::verify_domains_match,
 };
 use lemmy_db_queries::{source::comment::Comment_, Crud, DbPool};
 use lemmy_db_schema::{

crates/apub/src/objects/community.rs

@@ -1,9 +1,11 @@
 use crate::{
-  extensions::{context::lemmy_context, signatures::PublicKey},
+  check_is_apub_id_valid,
+  context::lemmy_context,
   fetcher::community::{fetch_community_outbox, update_community_mods},
   generate_moderators_url,
+  generate_outbox_url,
   objects::{create_tombstone, FromApub, ImageObject, Source, ToApub},
-  ActorType,
+  CommunityType,
 };
 use activitystreams::{
   actor::{kind::GroupType, Endpoints},
@@ -13,16 +15,20 @@ use activitystreams::{
   unparsed::Unparsed,
 };
 use chrono::{DateTime, FixedOffset};
+use itertools::Itertools;
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
+  signatures::PublicKey,
+  traits::ActorType,
   values::{MediaTypeHtml, MediaTypeMarkdown},
-  verify_domains_match,
+  verify::verify_domains_match,
 };
 use lemmy_db_queries::{source::community::Community_, DbPool};
 use lemmy_db_schema::{
   naive_now,
   source::community::{Community, CommunityForm},
 };
+use lemmy_db_views_actor::community_follower_view::CommunityFollowerView;
 use lemmy_utils::{
   settings::structs::Settings,
   utils::{check_slurs, check_slurs_opt, convert_datetime, markdown_to_html},
@@ -143,7 +149,7 @@ impl ToApub for Community {
       sensitive: Some(self.nsfw),
       moderators: Some(generate_moderators_url(&self.actor_id)?.into()),
       inbox: self.inbox_url.clone().into(),
-      outbox: self.get_outbox_url()?,
+      outbox: generate_outbox_url(&self.actor_id)?.into(),
       followers: self.followers_url.clone().into(),
       endpoints: Endpoints {
         shared_inbox: self.shared_inbox_url.clone().map(|s| s.into()),
@@ -189,3 +195,35 @@ impl FromApub for Community {
     Ok(community)
   }
 }
+
+#[async_trait::async_trait(?Send)]
+impl CommunityType for Community {
+  fn followers_url(&self) -> Url {
+    self.followers_url.clone().into()
+  }
+
+  /// For a given community, returns the inboxes of all followers.
+  async fn get_follower_inboxes(
+    &self,
+    pool: &DbPool,
+    settings: &Settings,
+  ) -> Result<Vec<Url>, LemmyError> {
+    let id = self.id;
+
+    let follows = blocking(pool, move |conn| {
+      CommunityFollowerView::for_community(conn, id)
+    })
+    .await??;
+    let inboxes = follows
+      .into_iter()
+      .filter(|f| !f.follower.local)
+      .map(|f| f.follower.shared_inbox_url.unwrap_or(f.follower.inbox_url))
+      .map(|i| i.into_inner())
+      .unique()
+      // Don't send to blocked instances
+      .filter(|inbox| check_is_apub_id_valid(inbox, false, settings).is_ok())
+      .collect();
+
+    Ok(inboxes)
+  }
+}

crates/apub/src/objects/person.rs

@@ -1,8 +1,8 @@
 use crate::{
   check_is_apub_id_valid,
-  extensions::{context::lemmy_context, signatures::PublicKey},
+  context::lemmy_context,
+  generate_outbox_url,
   objects::{FromApub, ImageObject, Source, ToApub},
-  ActorType,
 };
 use activitystreams::{
   actor::Endpoints,
@@ -14,8 +14,10 @@ use activitystreams::{
 };
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
+  signatures::PublicKey,
+  traits::ActorType,
   values::{MediaTypeHtml, MediaTypeMarkdown},
-  verify_domains_match,
+  verify::verify_domains_match,
 };
 use lemmy_db_queries::{source::person::Person_, DbPool};
 use lemmy_db_schema::{
@@ -113,7 +115,7 @@ impl ToApub for DbPerson {
       image,
       matrix_user_id: self.matrix_user_id.clone(),
       published: convert_datetime(self.published),
-      outbox: self.get_outbox_url()?,
+      outbox: generate_outbox_url(&self.actor_id)?.into(),
       endpoints: Endpoints {
         shared_inbox: self.shared_inbox_url.clone().map(|s| s.into()),
         ..Default::default()

crates/apub/src/objects/post.rs

@@ -1,9 +1,8 @@
 use crate::{
   activities::{extract_community, verify_person_in_community},
-  extensions::context::lemmy_context,
+  context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{create_tombstone, FromApub, ImageObject, Source, ToApub},
-  ActorType,
 };
 use activitystreams::{
   base::AnyBase,
@@ -18,10 +17,11 @@ use activitystreams::{
 use chrono::{DateTime, FixedOffset};
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
+  traits::ActorType,
   values::{MediaTypeHtml, MediaTypeMarkdown},
-  verify_domains_match,
+  verify::verify_domains_match,
 };
-use lemmy_db_queries::{source::post::Post_, ApubObject, Crud, DbPool};
+use lemmy_db_queries::{source::post::Post_, Crud, DbPool};
 use lemmy_db_schema::{
   self,
   source::{
@@ -78,12 +78,10 @@ impl Page {
   /// the current value, it is a mod action and needs to be verified as such.
   ///
   /// Both stickied and locked need to be false on a newly created post (verified in [[CreatePost]].
-  pub(crate) async fn is_mod_action(&self, pool: &DbPool) -> Result<bool, LemmyError> {
+  pub(crate) async fn is_mod_action(&self, context: &LemmyContext) -> Result<bool, LemmyError> {
-    let post_id = self.id.clone();
+    let old_post = ObjectId::<Post>::new(self.id.clone())
-    let old_post = blocking(pool, move |conn| {
+      .dereference_local(context)
-      Post::read_from_apub_id(conn, &post_id.into())
+      .await;
-    })
-    .await?;
 
     let is_mod_action = if let Ok(old_post) = old_post {
       self.stickied != Some(old_post.stickied) || self.comments_enabled != Some(!old_post.locked)
@@ -101,7 +99,7 @@ impl Page {
     let community = extract_community(&self.to, context, request_counter).await?;
 
     check_slurs(&self.name, &context.settings().slur_regex())?;
-    verify_domains_match(self.attributed_to.inner(), &self.id)?;
+    verify_domains_match(self.attributed_to.inner(), &self.id.clone())?;
     verify_person_in_community(
       &self.attributed_to,
       &ObjectId::new(community.actor_id()),
@@ -177,7 +175,7 @@ impl FromApub for Post {
   ) -> Result<Post, LemmyError> {
     // We can't verify the domain in case of mod action, because the mod may be on a different
     // instance from the post author.
-    let ap_id = if page.is_mod_action(context.pool()).await? {
+    let ap_id = if page.is_mod_action(context).await? {
       page.id_unchecked()
     } else {
       page.id(expected_domain)?

crates/apub/src/objects/private_message.rs

@@ -1,5 +1,5 @@
 use crate::{
-  extensions::context::lemmy_context,
+  context::lemmy_context,
   fetcher::object_id::ObjectId,
   objects::{create_tombstone, FromApub, Source, ToApub},
 };
@@ -14,7 +14,7 @@ use chrono::{DateTime, FixedOffset};
 use lemmy_api_common::blocking;
 use lemmy_apub_lib::{
   values::{MediaTypeHtml, MediaTypeMarkdown},
-  verify_domains_match,
+  verify::verify_domains_match,
 };
 use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud, DbPool};
 use lemmy_db_schema::source::{

crates/apub_lib/Cargo.toml

@@ -7,7 +7,6 @@ license = "AGPL-3.0"
 
 [dependencies]
 lemmy_utils = { version = "=0.13.0", path = "../utils" }
-lemmy_websocket = { version = "=0.13.0", path = "../websocket" }
 lemmy_apub_lib_derive = { version = "=0.13.0", path = "../apub_lib_derive" }
 activitystreams = "0.7.0-alpha.11"
 serde = { version = "1.0.130", features = ["derive"] }
@@ -17,3 +16,12 @@ serde_json = { version = "1.0.68", features = ["preserve_order"] }
 anyhow = "1.0.44"
 reqwest = { version = "0.11.4", features = ["json"] }
 log = "0.4.14"
+base64 = "0.13.0"
+openssl = "0.10.36"
+lazy_static = "1.4.0"
+http = "0.2.5"
+sha2 = "0.9.8"
+actix-web = { version = "4.0.0-beta.9", default-features = false }
+http-signature-normalization-actix = { version = "0.5.0-beta.10", default-features = false, features = ["server", "sha-2"] }
+http-signature-normalization-reqwest = { version = "0.2.0", default-features = false, features = ["sha-2"] }
+background-jobs = "0.10.0"

crates/apub_lib/src/activity_queue.rs

@@ -1,10 +1,4 @@
-use crate::{
+use crate::{signatures::sign_and_send, traits::ActorType, APUB_JSON_CONTENT_TYPE};
-  activities::community::announce::{AnnouncableActivities, AnnounceActivity},
-  extensions::signatures::sign_and_send,
-  insert_activity,
-  ActorType,
-  APUB_JSON_CONTENT_TYPE,
-};
 use anyhow::{anyhow, Context, Error};
 use background_jobs::{
   create_server,
@@ -15,83 +9,31 @@ use background_jobs::{
   QueueHandle,
   WorkerConfig,
 };
-use lemmy_db_schema::source::community::Community;
 use lemmy_utils::{location_info, LemmyError};
-use lemmy_websocket::LemmyContext;
+use log::warn;
-use log::{info, warn};
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
 use url::Url;
 
-pub(crate) async fn send_to_community_new(
+pub async fn send_activity(
-  activity: AnnouncableActivities,
+  activity: String,
-  activity_id: &Url,
   actor: &dyn ActorType,
-  community: &Community,
+  inboxes: Vec<&Url>,
-  additional_inboxes: Vec<Url>,
+  client: &Client,
-  context: &LemmyContext,
+  activity_queue: &QueueHandle,
 ) -> Result<(), LemmyError> {
-  // if this is a local community, we need to do an announce from the community instead
-  if community.local {
-    insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
-    AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
-  } else {
-    let mut inboxes = additional_inboxes;
-    inboxes.push(community.get_shared_inbox_or_inbox_url());
-    send_activity_new(context, &activity, activity_id, actor, inboxes, false).await?;
-  }
-
-  Ok(())
-}
-
-pub(crate) async fn send_activity_new<T>(
-  context: &LemmyContext,
-  activity: &T,
-  activity_id: &Url,
-  actor: &dyn ActorType,
-  inboxes: Vec<Url>,
-  sensitive: bool,
-) -> Result<(), LemmyError>
-where
-  T: Serialize,
-{
-  if !context.settings().federation.enabled || inboxes.is_empty() {
-    return Ok(());
-  }
-
-  info!("Sending activity {}", activity_id.to_string());
-
-  // Don't send anything to ourselves
-  // TODO: this should be a debug assert
-  let hostname = context.settings().get_hostname_without_port()?;
-  let inboxes: Vec<&Url> = inboxes
-    .iter()
-    .filter(|i| i.domain().expect("valid inbox url") != hostname)
-    .collect();
-
-  let serialised_activity = serde_json::to_string(&activity)?;
-
-  insert_activity(
-    activity_id,
-    serialised_activity.clone(),
-    true,
-    sensitive,
-    context.pool(),
-  )
-  .await?;
-
   for i in inboxes {
     let message = SendActivityTask {
-      activity: serialised_activity.to_owned(),
+      activity: activity.clone(),
       inbox: i.to_owned(),
       actor_id: actor.actor_id(),
       private_key: actor.private_key().context(location_info!())?,
     };
-    if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
+    if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
-      do_send(message, context.client()).await?;
+      do_send(message, client).await?;
     } else {
-      context.activity_queue.queue::<SendActivityTask>(message)?;
+      activity_queue.queue::<SendActivityTask>(message)?;
     }
   }
 

crates/apub_lib/src/data.rs

@@ -0,0 +1,35 @@
+use std::{ops::Deref, sync::Arc};
+
+#[derive(Debug)]
+pub struct Data<T: ?Sized>(Arc<T>);
+
+impl<T> Data<T> {
+  /// Create new `Data` instance.
+  pub fn new(state: T) -> Data<T> {
+    Data(Arc::new(state))
+  }
+
+  /// Get reference to inner app data.
+  pub fn get_ref(&self) -> &T {
+    self.0.as_ref()
+  }
+
+  /// Convert to the internal Arc<T>
+  pub fn into_inner(self) -> Arc<T> {
+    self.0
+  }
+}
+
+impl<T: ?Sized> Deref for Data<T> {
+  type Target = Arc<T>;
+
+  fn deref(&self) -> &Arc<T> {
+    &self.0
+  }
+}
+
+impl<T: ?Sized> Clone for Data<T> {
+  fn clone(&self) -> Data<T> {
+    Data(self.0.clone())
+  }
+}

crates/apub_lib/src/lib.rs

@@ -1,44 +1,12 @@
+#[macro_use]
+extern crate lazy_static;
+
+pub mod activity_queue;
+pub mod data;
+pub mod signatures;
+pub mod traits;
 pub mod values;
-
+pub mod verify;
-use activitystreams::error::DomainError;
-pub use lemmy_apub_lib_derive::*;
-use lemmy_utils::LemmyError;
-use lemmy_websocket::LemmyContext;
-use url::Url;
-
 pub mod webfinger;
 
-pub trait ActivityFields {
+pub static APUB_JSON_CONTENT_TYPE: &str = "application/activity+json";
-  fn id_unchecked(&self) -> &Url;
-  fn actor(&self) -> &Url;
-  fn cc(&self) -> Vec<Url>;
-}
-
-#[async_trait::async_trait(?Send)]
-pub trait ActivityHandler {
-  async fn verify(
-    &self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError>;
-
-  async fn receive(
-    self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError>;
-}
-
-pub fn verify_domains_match(a: &Url, b: &Url) -> Result<(), LemmyError> {
-  if a.domain() != b.domain() {
-    return Err(DomainError.into());
-  }
-  Ok(())
-}
-
-pub fn verify_urls_match(a: &Url, b: &Url) -> Result<(), LemmyError> {
-  if a != b {
-    return Err(DomainError.into());
-  }
-  Ok(())
-}

crates/apub_lib/src/signatures.rs

@@ -23,7 +23,7 @@ lazy_static! {
 
 /// Creates an HTTP post request to `inbox_url`, with the given `client` and `headers`, and
 /// `activity` as request body. The request is signed with `private_key` and then sent.
-pub(crate) async fn sign_and_send(
+pub async fn sign_and_send(
   client: &Client,
   headers: BTreeMap<String, String>,
   inbox_url: &Url,
@@ -62,7 +62,7 @@ pub(crate) async fn sign_and_send(
 }
 
 /// Verifies the HTTP signature on an incoming inbox request.
-pub(crate) fn verify_signature(request: &HttpRequest, public_key: &str) -> Result<(), LemmyError> {
+pub fn verify_signature(request: &HttpRequest, public_key: &str) -> Result<(), LemmyError> {
   let verified = CONFIG2
     .begin_verify(
       request.method(),

crates/apub_lib/src/traits.rs

@@ -0,0 +1,67 @@
+use crate::{data::Data, signatures::PublicKey};
+use activitystreams::chrono::NaiveDateTime;
+use anyhow::Context;
+pub use lemmy_apub_lib_derive::*;
+use lemmy_utils::{location_info, LemmyError};
+use url::Url;
+
+pub trait ActivityFields {
+  fn id_unchecked(&self) -> &Url;
+  fn actor(&self) -> &Url;
+  fn cc(&self) -> Vec<Url>;
+}
+
+#[async_trait::async_trait(?Send)]
+pub trait ActivityHandler {
+  type DataType;
+  async fn verify(
+    &self,
+    data: &Data<Self::DataType>,
+    request_counter: &mut i32,
+  ) -> Result<(), LemmyError>;
+
+  async fn receive(
+    self,
+    data: &Data<Self::DataType>,
+    request_counter: &mut i32,
+  ) -> Result<(), LemmyError>;
+}
+
+pub trait ApubObject {
+  type DataType;
+  /// If this object should be refetched after a certain interval, it should return the last refresh
+  /// time here. This is mainly used to update remote actors.
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime>;
+  /// Try to read the object with given ID from local database. Returns Ok(None) if it doesn't exist.
+  fn read_from_apub_id(data: &Self::DataType, object_id: Url) -> Result<Option<Self>, LemmyError>
+  where
+    Self: Sized;
+}
+
+/// Common methods provided by ActivityPub actors (community and person). Not all methods are
+/// implemented by all actors.
+pub trait ActorType {
+  fn is_local(&self) -> bool;
+  fn actor_id(&self) -> Url;
+  fn name(&self) -> String;
+
+  // TODO: this should not be an option (needs db migration in lemmy)
+  fn public_key(&self) -> Option<String>;
+  fn private_key(&self) -> Option<String>;
+
+  fn inbox_url(&self) -> Url;
+
+  fn shared_inbox_url(&self) -> Option<Url>;
+
+  fn shared_inbox_or_inbox_url(&self) -> Url {
+    self.shared_inbox_url().unwrap_or_else(|| self.inbox_url())
+  }
+
+  fn get_public_key(&self) -> Result<PublicKey, LemmyError> {
+    Ok(PublicKey {
+      id: format!("{}#main-key", self.actor_id()),
+      owner: self.actor_id(),
+      public_key_pem: self.public_key().context(location_info!())?,
+    })
+  }
+}

crates/apub_lib/src/verify.rs

@@ -0,0 +1,17 @@
+use activitystreams::error::DomainError;
+use lemmy_utils::LemmyError;
+use url::Url;
+
+pub fn verify_domains_match(a: &Url, b: &Url) -> Result<(), LemmyError> {
+  if a.domain() != b.domain() {
+    return Err(DomainError.into());
+  }
+  Ok(())
+}
+
+pub fn verify_urls_match(a: &Url, b: &Url) -> Result<(), LemmyError> {
+  if a != b {
+    return Err(DomainError.into());
+  }
+  Ok(())
+}

crates/apub_lib_derive/src/lib.rs

@@ -1,6 +1,6 @@
-use proc_macro2::TokenStream;
+use proc_macro2::{TokenStream, TokenTree};
 use quote::quote;
-use syn::{parse_macro_input, Data, DeriveInput, Fields::Unnamed, Ident, Variant};
+use syn::{parse_macro_input, Attribute, Data, DeriveInput, Fields::Unnamed, Ident, Variant};
 
 /// Generates implementation ActivityHandler for an enum, which looks like the following (handling
 /// all enum variants).
@@ -46,9 +46,29 @@ use syn::{parse_macro_input, Data, DeriveInput, Fields::Unnamed, Ident, Variant}
 ///   }
 ///
 /// ```
-#[proc_macro_derive(ActivityHandler)]
+#[proc_macro_derive(ActivityHandler, attributes(activity_handler))]
 pub fn derive_activity_handler(input: proc_macro::TokenStream) -> proc_macro::TokenStream {
   let input = parse_macro_input!(input as DeriveInput);
+  let attrs: Vec<&Attribute> = input
+    .attrs
+    .iter()
+    .filter(|attr| attr.path.is_ident("activity_handler"))
+    .collect();
+  let attrs: &Vec<TokenStream> = &attrs
+    .first()
+    .unwrap()
+    .tokens
+    .clone()
+    .into_iter()
+    .map(|t| {
+      if let TokenTree::Group(g) = t {
+        g.stream()
+      } else {
+        panic!()
+      }
+    })
+    .collect();
+  let attrs = attrs.first();
 
   let enum_name = input.ident;
 
@@ -71,10 +91,11 @@ pub fn derive_activity_handler(input: proc_macro::TokenStream) -> proc_macro::To
 
   let expanded = quote! {
       #[async_trait::async_trait(?Send)]
-      impl #impl_generics lemmy_apub_lib::ActivityHandler for #enum_name #ty_generics #where_clause {
+      impl #impl_generics lemmy_apub_lib::traits::ActivityHandler for #enum_name #ty_generics #where_clause {
+        type DataType = #attrs;
           async fn verify(
               &self,
-              context: &lemmy_websocket::LemmyContext,
+              context: &lemmy_apub_lib::data::Data<Self::DataType>,
               request_counter: &mut i32,
             ) -> Result<(), lemmy_utils::LemmyError> {
             match self {
@@ -83,7 +104,7 @@ pub fn derive_activity_handler(input: proc_macro::TokenStream) -> proc_macro::To
           }
           async fn receive(
             self,
-            context: &lemmy_websocket::LemmyContext,
+            context: &lemmy_apub_lib::data::Data<Self::DataType>,
             request_counter: &mut i32,
           ) -> Result<(), lemmy_utils::LemmyError> {
             match self {
@@ -128,7 +149,7 @@ pub fn derive_activity_fields(input: proc_macro::TokenStream) -> proc_macro::Tok
         .iter()
         .map(|v| generate_match_arm(&name, v, &quote! {a.cc()}));
       quote! {
-          impl #impl_generics lemmy_apub_lib::ActivityFields for #name #ty_generics #where_clause {
+          impl #impl_generics lemmy_apub_lib::traits::ActivityFields for #name #ty_generics #where_clause {
               fn id_unchecked(&self) -> &url::Url { match self { #(#impl_id)* } }
               fn actor(&self) -> &url::Url { match self { #(#impl_actor)* } }
               fn cc(&self) -> Vec<url::Url> { match self { #(#impl_cc)* } }
@@ -150,7 +171,7 @@ pub fn derive_activity_fields(input: proc_macro::TokenStream) -> proc_macro::Tok
         quote! {vec![]}
       };
       quote! {
-          impl #impl_generics lemmy_apub_lib::ActivityFields for #name #ty_generics #where_clause {
+          impl #impl_generics lemmy_apub_lib::traits::ActivityFields for #name #ty_generics #where_clause {
               fn id_unchecked(&self) -> &url::Url { &self.id }
               fn actor(&self) -> &url::Url { &self.actor.inner() }
               fn cc(&self) -> Vec<url::Url> { #cc_impl }

crates/db_queries/src/lib.rs

@@ -12,7 +12,6 @@ extern crate diesel_migrations;
 #[cfg(test)]
 extern crate serial_test;
 
-use chrono::NaiveDateTime;
 use diesel::{result::Error, *};
 use lemmy_db_schema::{CommunityId, DbUrl, PersonId};
 use lemmy_utils::ApiError;
@@ -155,16 +154,6 @@ pub trait DeleteableOrRemoveable {
   fn blank_out_deleted_or_removed_info(self) -> Self;
 }
 
-// TODO: move this to apub lib
-pub trait ApubObject {
-  /// If this object should be refetched after a certain interval, it should return the last refresh
-  /// time here. This is mainly used to update remote actors.
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime>;
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error>
-  where
-    Self: Sized;
-}
-
 pub trait MaybeOptional<T> {
   fn get_optional(self) -> Option<T>;
 }

crates/db_queries/src/source/comment.rs

@@ -1,5 +1,4 @@
-use crate::{ApubObject, Crud, DeleteableOrRemoveable, Likeable, Saveable};
+use crate::{Crud, DeleteableOrRemoveable, Likeable, Saveable};
-use chrono::NaiveDateTime;
 use diesel::{dsl::*, result::Error, *};
 use lemmy_db_schema::{
   naive_now,
@@ -179,17 +178,6 @@ impl Crud for Comment {
   }
 }
 
-impl ApubObject for Comment {
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
-    None
-  }
-
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error> {
-    use lemmy_db_schema::schema::comment::dsl::*;
-    comment.filter(ap_id.eq(object_id)).first::<Self>(conn)
-  }
-}
-
 impl Likeable for CommentLike {
   type Form = CommentLikeForm;
   type IdType = CommentId;

crates/db_queries/src/source/community.rs

@@ -1,5 +1,4 @@
-use crate::{ApubObject, Bannable, Crud, DeleteableOrRemoveable, Followable, Joinable};
+use crate::{Bannable, Crud, DeleteableOrRemoveable, Followable, Joinable};
-use chrono::NaiveDateTime;
 use diesel::{dsl::*, result::Error, *};
 use lemmy_db_schema::{
   naive_now,
@@ -93,19 +92,6 @@ impl Crud for Community {
   }
 }
 
-impl ApubObject for Community {
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
-    Some(self.last_refreshed_at)
-  }
-
-  fn read_from_apub_id(conn: &PgConnection, for_actor_id: &DbUrl) -> Result<Self, Error> {
-    use lemmy_db_schema::schema::community::dsl::*;
-    community
-      .filter(actor_id.eq(for_actor_id))
-      .first::<Self>(conn)
-  }
-}
-
 pub trait Community_ {
   fn read_from_name(conn: &PgConnection, community_name: &str) -> Result<Community, Error>;
   fn update_deleted(

crates/db_queries/src/source/person.rs

@@ -1,11 +1,9 @@
-use crate::{ApubObject, Crud};
+use crate::Crud;
-use chrono::NaiveDateTime;
 use diesel::{dsl::*, result::Error, *};
 use lemmy_db_schema::{
   naive_now,
   schema::person::dsl::*,
   source::person::{Person, PersonForm},
-  DbUrl,
   PersonId,
 };
 
@@ -181,20 +179,6 @@ impl Crud for Person {
   }
 }
 
-impl ApubObject for Person {
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
-    Some(self.last_refreshed_at)
-  }
-
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error> {
-    use lemmy_db_schema::schema::person::dsl::*;
-    person
-      .filter(deleted.eq(false))
-      .filter(actor_id.eq(object_id))
-      .first::<Self>(conn)
-  }
-}
-
 pub trait Person_ {
   fn ban_person(conn: &PgConnection, person_id: PersonId, ban: bool) -> Result<Person, Error>;
   fn add_admin(conn: &PgConnection, person_id: PersonId, added: bool) -> Result<Person, Error>;

crates/db_queries/src/source/post.rs

@@ -1,5 +1,4 @@
-use crate::{ApubObject, Crud, DeleteableOrRemoveable, Likeable, Readable, Saveable};
+use crate::{Crud, DeleteableOrRemoveable, Likeable, Readable, Saveable};
-use chrono::NaiveDateTime;
 use diesel::{dsl::*, result::Error, *};
 use lemmy_db_schema::{
   naive_now,
@@ -193,17 +192,6 @@ impl Post_ for Post {
   }
 }
 
-impl ApubObject for Post {
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
-    None
-  }
-
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error> {
-    use lemmy_db_schema::schema::post::dsl::*;
-    post.filter(ap_id.eq(object_id)).first::<Self>(conn)
-  }
-}
-
 impl Likeable for PostLike {
   type Form = PostLikeForm;
   type IdType = PostId;

crates/db_queries/src/source/private_message.rs

@@ -1,5 +1,4 @@
-use crate::{ApubObject, Crud, DeleteableOrRemoveable};
+use crate::{Crud, DeleteableOrRemoveable};
-use chrono::NaiveDateTime;
 use diesel::{dsl::*, result::Error, *};
 use lemmy_db_schema::{naive_now, source::private_message::*, DbUrl, PersonId, PrivateMessageId};
 
@@ -30,22 +29,6 @@ impl Crud for PrivateMessage {
   }
 }
 
-impl ApubObject for PrivateMessage {
-  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
-    None
-  }
-
-  fn read_from_apub_id(conn: &PgConnection, object_id: &DbUrl) -> Result<Self, Error>
-  where
-    Self: Sized,
-  {
-    use lemmy_db_schema::schema::private_message::dsl::*;
-    private_message
-      .filter(ap_id.eq(object_id))
-      .first::<Self>(conn)
-  }
-}
-
 pub trait PrivateMessage_ {
   fn update_ap_id(
     conn: &PgConnection,

crates/db_schema/Cargo.toml

@@ -9,6 +9,8 @@ license = "AGPL-3.0"
 doctest = false
 
 [dependencies]
+lemmy_utils = { version = "=0.13.0", path = "../utils" }
+lemmy_apub_lib = { version = "=0.13.0", path = "../apub_lib" }
 diesel = { version = "1.4.8", features = ["postgres","chrono","r2d2","serde_json"] }
 chrono = { version = "0.4.19", features = ["serde"] }
 serde = { version = "1.0.130", features = ["derive"] }

crates/db_schema/src/source/comment.rs

@@ -6,7 +6,12 @@ use crate::{
   PersonId,
   PostId,
 };
+use chrono::NaiveDateTime;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
+use lemmy_apub_lib::traits::ApubObject;
+use lemmy_utils::LemmyError;
 use serde::Serialize;
+use url::Url;
 
 // WITH RECURSIVE MyTree AS (
 //     SELECT * FROM comment WHERE parent_id IS NULL
@@ -104,3 +109,17 @@ pub struct CommentSavedForm {
   pub comment_id: CommentId,
   pub person_id: PersonId,
 }
+
+impl ApubObject for Comment {
+  type DataType = PgConnection;
+
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    None
+  }
+
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
+    use crate::schema::comment::dsl::*;
+    let object_id: DbUrl = object_id.into();
+    Ok(comment.filter(ap_id.eq(object_id)).first::<Self>(conn).ok())
+  }
+}

crates/db_schema/src/source/community.rs

@@ -4,7 +4,12 @@ use crate::{
   DbUrl,
   PersonId,
 };
+use chrono::NaiveDateTime;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
+use lemmy_apub_lib::traits::{ActorType, ApubObject};
+use lemmy_utils::LemmyError;
 use serde::Serialize;
+use url::Url;
 
 #[derive(Clone, Queryable, Identifiable, PartialEq, Debug, Serialize)]
 #[table_name = "community"]
@@ -124,3 +129,48 @@ pub struct CommunityFollowerForm {
   pub person_id: PersonId,
   pub pending: bool,
 }
+
+impl ApubObject for Community {
+  type DataType = PgConnection;
+
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    Some(self.last_refreshed_at)
+  }
+
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
+    use crate::schema::community::dsl::*;
+    let object_id: DbUrl = object_id.into();
+    Ok(
+      community
+        .filter(actor_id.eq(object_id))
+        .first::<Self>(conn)
+        .ok(),
+    )
+  }
+}
+
+impl ActorType for Community {
+  fn is_local(&self) -> bool {
+    self.local
+  }
+  fn actor_id(&self) -> Url {
+    self.actor_id.to_owned().into()
+  }
+  fn name(&self) -> String {
+    self.name.clone()
+  }
+  fn public_key(&self) -> Option<String> {
+    self.public_key.to_owned()
+  }
+  fn private_key(&self) -> Option<String> {
+    self.private_key.to_owned()
+  }
+
+  fn inbox_url(&self) -> Url {
+    self.inbox_url.clone().into()
+  }
+
+  fn shared_inbox_url(&self) -> Option<Url> {
+    self.shared_inbox_url.clone().map(|s| s.into_inner())
+  }
+}

crates/db_schema/src/source/person.rs

@@ -3,7 +3,12 @@ use crate::{
   DbUrl,
   PersonId,
 };
+use chrono::NaiveDateTime;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
+use lemmy_apub_lib::traits::{ActorType, ApubObject};
+use lemmy_utils::LemmyError;
 use serde::Serialize;
+use url::Url;
 
 #[derive(Clone, Queryable, Identifiable, PartialEq, Debug, Serialize)]
 #[table_name = "person"]
@@ -170,3 +175,51 @@ pub struct PersonForm {
   pub admin: Option<bool>,
   pub bot_account: Option<bool>,
 }
+
+impl ApubObject for Person {
+  type DataType = PgConnection;
+
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    Some(self.last_refreshed_at)
+  }
+
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
+    use crate::schema::person::dsl::*;
+    let object_id: DbUrl = object_id.into();
+    Ok(
+      person
+        .filter(deleted.eq(false))
+        .filter(actor_id.eq(object_id))
+        .first::<Self>(conn)
+        .ok(),
+    )
+  }
+}
+
+impl ActorType for Person {
+  fn is_local(&self) -> bool {
+    self.local
+  }
+  fn actor_id(&self) -> Url {
+    self.actor_id.to_owned().into_inner()
+  }
+  fn name(&self) -> String {
+    self.name.clone()
+  }
+
+  fn public_key(&self) -> Option<String> {
+    self.public_key.to_owned()
+  }
+
+  fn private_key(&self) -> Option<String> {
+    self.private_key.to_owned()
+  }
+
+  fn inbox_url(&self) -> Url {
+    self.inbox_url.clone().into()
+  }
+
+  fn shared_inbox_url(&self) -> Option<Url> {
+    self.shared_inbox_url.clone().map(|s| s.into_inner())
+  }
+}

crates/db_schema/src/source/post.rs

@@ -5,7 +5,12 @@ use crate::{
   PersonId,
   PostId,
 };
+use chrono::NaiveDateTime;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
+use lemmy_apub_lib::traits::ApubObject;
+use lemmy_utils::LemmyError;
 use serde::Serialize;
+use url::Url;
 
 #[derive(Clone, Queryable, Identifiable, PartialEq, Debug, Serialize)]
 #[table_name = "post"]
@@ -106,3 +111,17 @@ pub struct PostReadForm {
   pub post_id: PostId,
   pub person_id: PersonId,
 }
+
+impl ApubObject for Post {
+  type DataType = PgConnection;
+
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    None
+  }
+
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
+    use crate::schema::post::dsl::*;
+    let object_id: DbUrl = object_id.into();
+    Ok(post.filter(ap_id.eq(object_id)).first::<Self>(conn).ok())
+  }
+}

crates/db_schema/src/source/private_message.rs

@@ -1,5 +1,10 @@
 use crate::{schema::private_message, DbUrl, PersonId, PrivateMessageId};
+use chrono::NaiveDateTime;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
+use lemmy_apub_lib::traits::ApubObject;
+use lemmy_utils::LemmyError;
 use serde::Serialize;
+use url::Url;
 
 #[derive(Clone, Queryable, Associations, Identifiable, PartialEq, Debug, Serialize)]
 #[table_name = "private_message"]
@@ -29,3 +34,22 @@ pub struct PrivateMessageForm {
   pub ap_id: Option<DbUrl>,
   pub local: Option<bool>,
 }
+
+impl ApubObject for PrivateMessage {
+  type DataType = PgConnection;
+
+  fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
+    None
+  }
+
+  fn read_from_apub_id(conn: &PgConnection, object_id: Url) -> Result<Option<Self>, LemmyError> {
+    use crate::schema::private_message::dsl::*;
+    let object_id: DbUrl = object_id.into();
+    Ok(
+      private_message
+        .filter(ap_id.eq(object_id))
+        .first::<Self>(conn)
+        .ok(),
+    )
+  }
+}

docker/federation/docker-compose.yml

@@ -40,7 +40,7 @@ services:
     volumes:
       - ./lemmy_alpha.hjson:/config/config.hjson
     environment:
-      - LEMMY_TEST_SEND_SYNC=1
+      - APUB_TESTING_SEND_SYNC
       - RUST_BACKTRACE=1
       - RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
     depends_on:
@@ -69,7 +69,7 @@ services:
     volumes:
       - ./lemmy_beta.hjson:/config/config.hjson
     environment:
-      - LEMMY_TEST_SEND_SYNC=1
+      - APUB_TESTING_SEND_SYNC
       - RUST_BACKTRACE=1
       - RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
     depends_on:
@@ -98,7 +98,7 @@ services:
     volumes:
       - ./lemmy_gamma.hjson:/config/config.hjson
     environment:
-      - LEMMY_TEST_SEND_SYNC=1
+      - APUB_TESTING_SEND_SYNC
       - RUST_BACKTRACE=1
       - RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
     depends_on:
@@ -128,7 +128,7 @@ services:
     volumes:
       - ./lemmy_delta.hjson:/config/config.hjson
     environment:
-      - LEMMY_TEST_SEND_SYNC=1
+      - APUB_TESTING_SEND_SYNC
       - RUST_BACKTRACE=1
       - RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
     depends_on:
@@ -158,7 +158,7 @@ services:
     volumes:
       - ./lemmy_epsilon.hjson:/config/config.hjson
     environment:
-      - LEMMY_TEST_SEND_SYNC=1
+      - APUB_TESTING_SEND_SYNC
       - RUST_BACKTRACE=1
       - RUST_LOG="warn,lemmy_server=debug,lemmy_api=debug,lemmy_api_common=debug,lemmy_api_crud=debug,lemmy_apub=debug,lemmy_db_queries=debug,lemmy_db_schema=debug,lemmy_db_views=debug,lemmy_db_views_actor=debug,lemmy_db_views_moderator=debug,lemmy_routes=debug,lemmy_utils=debug,lemmy_websocket=debug"
     depends_on:

src/main.rs

@@ -10,7 +10,7 @@ use diesel::{
 use lemmy_api::match_websocket_operation;
 use lemmy_api_common::blocking;
 use lemmy_api_crud::match_websocket_operation_crud;
-use lemmy_apub::activity_queue::create_activity_queue;
+use lemmy_apub_lib::activity_queue::create_activity_queue;
 use lemmy_db_queries::{get_database_url_from_env, source::secret::Secret_};
 use lemmy_db_schema::source::secret::Secret;
 use lemmy_routes::{feeds, images, nodeinfo, webfinger};