Fix federation tests

crates/apub/src/inbox/community_inbox.rs

@@ -134,46 +134,95 @@ pub(crate) async fn community_receive_message(
   let activity_kind = activity.kind().context(location_info!())?;
   let do_announce = match activity_kind {
     CommunityValidTypes::Follow => {
-      handle_follow(any_base.clone(), person, &to_community, &context).await?;
+      Box::pin(handle_follow(
+        any_base.clone(),
+        person,
+        &to_community,
+        &context,
+      ))
+      .await?;
       false
     }
     CommunityValidTypes::Undo => {
-      handle_undo(
+      Box::pin(handle_undo(
         context,
         activity.clone(),
         actor_url,
         &to_community,
         request_counter,
-      )
+      ))
       .await?
     }
     CommunityValidTypes::Create => {
-      receive_create_for_community(context, any_base.clone(), &actor_url, request_counter).await?;
+      Box::pin(receive_create_for_community(
+        context,
+        any_base.clone(),
+        &actor_url,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Update => {
-      receive_update_for_community(context, any_base.clone(), None, &actor_url, request_counter)
+      Box::pin(receive_update_for_community(
-        .await?;
+        context,
+        any_base.clone(),
+        None,
+        &actor_url,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Like => {
-      receive_like_for_community(context, any_base.clone(), &actor_url, request_counter).await?;
+      Box::pin(receive_like_for_community(
+        context,
+        any_base.clone(),
+        &actor_url,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Dislike => {
-      receive_dislike_for_community(context, any_base.clone(), &actor_url, request_counter).await?;
+      Box::pin(receive_dislike_for_community(
+        context,
+        any_base.clone(),
+        &actor_url,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Delete => {
-      receive_delete_for_community(context, any_base.clone(), None, &actor_url).await?;
+      Box::pin(receive_delete_for_community(
+        context,
+        any_base.clone(),
+        None,
+        &actor_url,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Add => {
-      receive_add_for_community(context, any_base.clone(), None, request_counter).await?;
+      Box::pin(receive_add_for_community(
+        context,
+        any_base.clone(),
+        None,
+        request_counter,
+      ))
+      .await?;
       true
     }
     CommunityValidTypes::Remove => {
-      receive_remove_for_community(context, any_base.clone(), None, request_counter).await?;
+      Box::pin(receive_remove_for_community(
+        context,
+        any_base.clone(),
+        None,
+        request_counter,
+      ))
+      .await?;
       true
     }
   };

crates/apub/src/inbox/mod.rs

@@ -58,7 +58,7 @@ pub(crate) async fn is_activity_already_known(
 
 pub(crate) fn get_activity_to_and_cc<T, Kind>(activity: &T) -> Vec<Url>
 where
-  T: AsBase<Kind> + AsObject<Kind> + ActorAndObjectRefExt,
+  T: AsObject<Kind>,
 {
   let mut to_and_cc = vec![];
   if let Some(to) = activity.to() {

crates/apub/src/inbox/person_inbox.rs

@@ -154,19 +154,39 @@ pub(crate) async fn person_receive_message(
       .await?;
     }
     PersonValidTypes::Announce => {
-      receive_announce(&context, any_base, actor, request_counter).await?
+      Box::pin(receive_announce(&context, any_base, actor, request_counter)).await?
     }
     PersonValidTypes::Create => {
-      receive_create(&context, any_base, actor_url, request_counter).await?
+      Box::pin(receive_create(
+        &context,
+        any_base,
+        actor_url,
+        request_counter,
+      ))
+      .await?
     }
     PersonValidTypes::Update => {
-      receive_update(&context, any_base, actor_url, request_counter).await?
+      Box::pin(receive_update(
+        &context,
+        any_base,
+        actor_url,
+        request_counter,
+      ))
+      .await?
     }
     PersonValidTypes::Delete => {
-      receive_delete(context, any_base, &actor_url, request_counter).await?
+      Box::pin(receive_delete(
+        context,
+        any_base,
+        &actor_url,
+        request_counter,
+      ))
+      .await?
     }
-    PersonValidTypes::Undo => receive_undo(context, any_base, &actor_url, request_counter).await?,
+    PersonValidTypes::Undo => {
-    PersonValidTypes::Remove => receive_remove(context, any_base, &actor_url).await?,
+      Box::pin(receive_undo(context, any_base, &actor_url, request_counter)).await?
+    }
+    PersonValidTypes::Remove => Box::pin(receive_remove(context, any_base, &actor_url)).await?,
   };
 
   // TODO: would be logical to move websocket notification code here
@@ -305,7 +325,14 @@ pub async fn receive_announce(
       receive_dislike_for_community(context, inner_activity, &inner_id, request_counter).await
     }
     Some(Delete) => {
-      receive_delete_for_community(context, inner_activity, Some(announce), &inner_id).await
+      receive_delete_for_community(
+        context,
+        inner_activity,
+        Some(announce),
+        &inner_id,
+        request_counter,
+      )
+      .await
     }
     Some(Remove) => {
       receive_remove_for_community(context, inner_activity, Some(announce), request_counter).await

crates/apub/src/inbox/receive_for_community.rs

@@ -35,13 +35,10 @@ use crate::{
     objects::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
     person::get_or_fetch_and_upsert_person,
   },
-  find_object_by_id,
   find_post_or_comment_by_id,
   generate_moderators_url,
   inbox::verify_is_addressed_to_public,
-  ActorType,
   CommunityType,
-  Object,
   PostOrComment,
 };
 use activitystreams::{
@@ -122,7 +119,7 @@ pub(in crate::inbox) async fn receive_update_for_community(
   let update = Update::from_any_base(activity)?.context(location_info!())?;
   verify_activity_domains_valid(&update, &expected_domain, false)?;
   verify_is_addressed_to_public(&update)?;
-  verify_modification_actor_instance(&update, &announce, context).await?;
+  verify_modification_actor_instance(&update, &announce, context, request_counter).await?;
 
   let kind = update
     .object()
@@ -197,11 +194,12 @@ pub(in crate::inbox) async fn receive_delete_for_community(
   activity: AnyBase,
   announce: Option<Announce>,
   expected_domain: &Url,
+  request_counter: &mut i32,
 ) -> Result<(), LemmyError> {
   let delete = Delete::from_any_base(activity)?.context(location_info!())?;
   verify_activity_domains_valid(&delete, &expected_domain, true)?;
   verify_is_addressed_to_public(&delete)?;
-  verify_modification_actor_instance(&delete, &announce, context).await?;
+  verify_modification_actor_instance(&delete, &announce, context, request_counter).await?;
 
   let object = delete
     .object()
@@ -588,6 +586,7 @@ async fn verify_modification_actor_instance<T, Kind>(
   activity: &T,
   announce: &Option<Announce>,
   context: &LemmyContext,
+  request_counter: &mut i32,
 ) -> Result<(), LemmyError>
 where
   T: ActorAndObjectRef + BaseExt<Kind> + AsObject<Kind>,
@@ -603,12 +602,9 @@ where
     .map(|o| o.id())
     .flatten()
     .context(location_info!())?;
-  let original_id = match find_object_by_id(context, object_id.to_owned()).await? {
+  let original_id = match fetch_post_or_comment_by_id(object_id, context, request_counter).await? {
-    Object::Post(p) => p.ap_id.into_inner(),
+    PostOrComment::Post(p) => p.ap_id.into_inner(),
-    Object::Comment(c) => c.ap_id.into_inner(),
+    PostOrComment::Comment(c) => c.ap_id.into_inner(),
-    Object::Community(c) => c.actor_id(),
-    Object::Person(p) => p.actor_id(),
-    Object::PrivateMessage(p) => p.ap_id.into_inner(),
   };
   if actor_id.domain() != original_id.domain() {
     let community = extract_community_from_cc(activity, context).await?;

crates/apub/src/inbox/shared_inbox.rs

@@ -80,13 +80,13 @@ pub async fn shared_inbox(
     let community_activity = CommunityAcceptedActivities::from_any_base(activity_any_base.clone())?
       .context(location_info!())?;
     res = Some(
-      community_receive_message(
+      Box::pin(community_receive_message(
         community_activity,
         community,
         actor.as_ref(),
         &context,
         request_counter,
-      )
+      ))
       .await?,
     );
   } else if is_addressed_to_local_person(&to_and_cc, context.pool()).await? {
@@ -94,13 +94,13 @@ pub async fn shared_inbox(
       .context(location_info!())?;
     // `to_person` is only used for follow activities (which we dont receive here), so no need to pass
     // it in
-    person_receive_message(
+    Box::pin(person_receive_message(
       person_activity,
       None,
       actor.as_ref(),
       &context,
       request_counter,
-    )
+    ))
     .await?;
   } else if is_addressed_to_community_followers(&to_and_cc, context.pool())
     .await?
@@ -109,13 +109,13 @@ pub async fn shared_inbox(
     let person_activity = PersonAcceptedActivities::from_any_base(activity_any_base.clone())?
       .context(location_info!())?;
     res = Some(
-      person_receive_message(
+      Box::pin(person_receive_message(
         person_activity,
         None,
         actor.as_ref(),
         &context,
         request_counter,
-      )
+      ))
       .await?,
     );
   }

crates/apub/src/objects/comment.rs

@@ -165,15 +165,24 @@ impl FromApubToForm<NoteExt> for CommentForm {
     let post_ap_id = in_reply_tos.next().context(location_info!())??;
 
     // This post, or the parent comment might not yet exist on this server yet, fetch them.
-    let post = get_or_fetch_and_insert_post(&post_ap_id, context, request_counter).await?;
+    let post = Box::pin(get_or_fetch_and_insert_post(
+      &post_ap_id,
+      context,
+      request_counter,
+    ))
+    .await?;
 
     // The 2nd item, if it exists, is the parent comment apub_id
     // For deeply nested comments, FromApub automatically gets called recursively
     let parent_id: Option<CommentId> = match in_reply_tos.next() {
       Some(parent_comment_uri) => {
         let parent_comment_ap_id = &parent_comment_uri?;
-        let parent_comment =
+        let parent_comment = Box::pin(get_or_fetch_and_insert_comment(
-          get_or_fetch_and_insert_comment(&parent_comment_ap_id, context, request_counter).await?;
+          &parent_comment_ap_id,
+          context,
+          request_counter,
+        ))
+        .await?;
 
         Some(parent_comment.id)
       }

crates/apub/src/objects/mod.rs

@@ -1,7 +1,8 @@
 use crate::{
   check_is_apub_id_valid,
   fetcher::{community::get_or_fetch_and_upsert_community, person::get_or_fetch_and_upsert_person},
-  inbox::community_inbox::check_community_or_site_ban,
+  inbox::{community_inbox::check_community_or_site_ban, get_activity_to_and_cc},
+  PageExt,
 };
 use activitystreams::{
   base::{AsBase, BaseExt, ExtendsExt},
@@ -230,23 +231,12 @@ where
   check_community_or_site_ban(&person, community_id, context.pool()).await
 }
 
-pub(in crate::objects) async fn get_to_community<T, Kind>(
+pub(in crate::objects) async fn get_community_from_to_or_cc(
-  object: &T,
+  page: &PageExt,
   context: &LemmyContext,
   request_counter: &mut i32,
-) -> Result<Community, LemmyError>
+) -> Result<Community, LemmyError> {
-where
+  for cid in get_activity_to_and_cc(page) {
-  T: ObjectExt<Kind>,
-{
-  let community_ids = object
-    .to()
-    .context(location_info!())?
-    .as_many()
-    .context(location_info!())?
-    .iter()
-    .map(|a| a.as_xsd_any_uri().context(location_info!()))
-    .collect::<Result<Vec<&Url>, anyhow::Error>>()?;
-  for cid in community_ids {
     let community = get_or_fetch_and_upsert_community(&cid, context, request_counter).await;
     if community.is_ok() {
       return community;

crates/apub/src/objects/post.rs

@@ -6,9 +6,9 @@ use crate::{
     check_object_domain,
     check_object_for_community_or_site_ban,
     create_tombstone,
+    get_community_from_to_or_cc,
     get_object_from_apub,
     get_source_markdown_value,
-    get_to_community,
     set_content_and_source,
     FromApub,
     FromApubToForm,
@@ -162,7 +162,7 @@ impl FromApubToForm<PageExt> for PostForm {
     let creator =
       get_or_fetch_and_upsert_person(creator_actor_id, context, request_counter).await?;
 
-    let community = get_to_community(page, context, request_counter).await?;
+    let community = get_community_from_to_or_cc(page, context, request_counter).await?;
 
     let thumbnail_url: Option<Url> = match &page.inner.image() {
       Some(any_image) => Image::from_any_base(