Merge pull request 'Refactor inbox, simplify and split into multiple files' (#72) from inbox-refactoring into main

Reviewed-on: https://yerbamate.dev/LemmyNet/lemmy/pulls/72

server/src/api/comment.rs

@@ -196,15 +196,8 @@ impl Perform for Oper<CreateComment> {
 
     // Scan the comment for user mentions, add those rows
     let mentions = scrape_text_for_mentions(&comment_form.content);
-    let recipient_ids = send_local_notifs(
+    let recipient_ids =
-      mentions,
+      send_local_notifs(mentions, updated_comment.clone(), &user, post, pool, true).await?;
-      updated_comment.clone(),
-      user.clone(),
-      post,
-      pool,
-      true,
-    )
-    .await?;
 
     // You like your own comment by default
     let like_form = CommentLikeForm {
@@ -313,7 +306,7 @@ impl Perform for Oper<EditComment> {
     let updated_comment_content = updated_comment.content.to_owned();
     let mentions = scrape_text_for_mentions(&updated_comment_content);
     let recipient_ids =
-      send_local_notifs(mentions, updated_comment, user, post, pool, false).await?;
+      send_local_notifs(mentions, updated_comment, &user, post, pool, false).await?;
 
     let edit_id = data.edit_id;
     let comment_view = blocking(pool, move |conn| {
@@ -418,7 +411,7 @@ impl Perform for Oper<DeleteComment> {
     let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;
     let mentions = vec![];
     let recipient_ids =
-      send_local_notifs(mentions, updated_comment, user, post, pool, false).await?;
+      send_local_notifs(mentions, updated_comment, &user, post, pool, false).await?;
 
     let mut res = CommentResponse {
       comment: comment_view,
@@ -524,7 +517,7 @@ impl Perform for Oper<RemoveComment> {
     let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;
     let mentions = vec![];
     let recipient_ids =
-      send_local_notifs(mentions, updated_comment, user, post, pool, false).await?;
+      send_local_notifs(mentions, updated_comment, &user, post, pool, false).await?;
 
     let mut res = CommentResponse {
       comment: comment_view,
@@ -870,13 +863,14 @@ impl Perform for Oper<GetComments> {
 pub async fn send_local_notifs(
   mentions: Vec<MentionData>,
   comment: Comment,
-  user: User_,
+  user: &User_,
   post: Post,
   pool: &DbPool,
   do_send_email: bool,
 ) -> Result<Vec<i32>, LemmyError> {
+  let user2 = user.clone();
   let ids = blocking(pool, move |conn| {
-    do_send_local_notifs(conn, &mentions, &comment, &user, &post, do_send_email)
+    do_send_local_notifs(conn, &mentions, &comment, &user2, &post, do_send_email)
   })
   .await?;
 

server/src/apub/activities.rs

@@ -13,8 +13,10 @@ use crate::{
 use activitystreams_new::base::AnyBase;
 use actix_web::client::Client;
 use lemmy_db::{community::Community, user::User_};
+use lemmy_utils::{get_apub_protocol_string, settings::Settings};
 use log::debug;
-use url::Url;
+use url::{ParseError, Url};
+use uuid::Uuid;
 
 pub async fn send_activity_to_community(
   creator: &User_,
@@ -68,3 +70,17 @@ pub async fn send_activity(
 
   Ok(())
 }
+
+pub(in crate::apub) fn generate_activity_id<T>(kind: T) -> Result<Url, ParseError>
+where
+  T: ToString,
+{
+  let id = format!(
+    "{}://{}/activities/{}/{}",
+    get_apub_protocol_string(),
+    Settings::get().hostname,
+    kind.to_string().to_lowercase(),
+    Uuid::new_v4()
+  );
+  Url::parse(&id)
+}

server/src/apub/comment.rs

@@ -1,14 +1,14 @@
 use crate::{
   apub::{
-    activities::send_activity_to_community,
+    activities::{generate_activity_id, send_activity_to_community},
     create_apub_response,
     create_apub_tombstone_response,
     create_tombstone,
     fetch_webfinger_url,
     fetcher::{
-      get_or_fetch_and_insert_remote_comment,
+      get_or_fetch_and_insert_comment,
-      get_or_fetch_and_insert_remote_post,
+      get_or_fetch_and_insert_post,
-      get_or_fetch_and_upsert_remote_user,
+      get_or_fetch_and_upsert_user,
     },
     ActorType,
     ApubLikeableType,
@@ -22,7 +22,16 @@ use crate::{
   LemmyError,
 };
 use activitystreams_new::{
-  activity::{Create, Delete, Dislike, Like, Remove, Undo, Update},
+  activity::{
+    kind::{CreateType, DeleteType, DislikeType, LikeType, RemoveType, UndoType, UpdateType},
+    Create,
+    Delete,
+    Dislike,
+    Like,
+    Remove,
+    Undo,
+    Update,
+  },
   base::AnyBase,
   context,
   link::Mention,
@@ -109,12 +118,7 @@ impl ToApub for Comment {
   }
 
   fn to_tombstone(&self) -> Result<Tombstone, LemmyError> {
-    create_tombstone(
+    create_tombstone(self.deleted, &self.ap_id, self.updated, NoteType::Note)
-      self.deleted,
-      &self.ap_id,
-      self.updated,
-      NoteType::Note.to_string(),
-    )
   }
 }
 
@@ -127,7 +131,6 @@ impl FromApub for CommentForm {
     note: &Note,
     client: &Client,
     pool: &DbPool,
-    actor_id: &Url,
   ) -> Result<CommentForm, LemmyError> {
     let creator_actor_id = &note
       .attributed_to()
@@ -135,7 +138,7 @@ impl FromApub for CommentForm {
       .as_single_xsd_any_uri()
       .unwrap();
 
-    let creator = get_or_fetch_and_upsert_remote_user(creator_actor_id, client, pool).await?;
+    let creator = get_or_fetch_and_upsert_user(creator_actor_id, client, pool).await?;
 
     let mut in_reply_tos = note
       .in_reply_to()
@@ -148,7 +151,7 @@ impl FromApub for CommentForm {
     let post_ap_id = in_reply_tos.next().unwrap();
 
     // This post, or the parent comment might not yet exist on this server yet, fetch them.
-    let post = get_or_fetch_and_insert_remote_post(&post_ap_id, client, pool).await?;
+    let post = get_or_fetch_and_insert_post(&post_ap_id, client, pool).await?;
 
     // The 2nd item, if it exists, is the parent comment apub_id
     // For deeply nested comments, FromApub automatically gets called recursively
@@ -156,7 +159,7 @@ impl FromApub for CommentForm {
       Some(parent_comment_uri) => {
         let parent_comment_ap_id = &parent_comment_uri;
         let parent_comment =
-          get_or_fetch_and_insert_remote_comment(&parent_comment_ap_id, client, pool).await?;
+          get_or_fetch_and_insert_comment(&parent_comment_ap_id, client, pool).await?;
 
         Some(parent_comment.id)
       }
@@ -178,7 +181,7 @@ impl FromApub for CommentForm {
       published: note.published().map(|u| u.to_owned().naive_local()),
       updated: note.updated().map(|u| u.to_owned().naive_local()),
       deleted: None,
-      ap_id: note.id(actor_id.domain().unwrap())?.unwrap().to_string(),
+      ap_id: note.id_unchecked().unwrap().to_string(),
       local: false,
     })
   }
@@ -204,11 +207,10 @@ impl ApubObjectType for Comment {
     let maa =
       collect_non_local_mentions_and_addresses(&self.content, &community, client, pool).await?;
 
-    let id = format!("{}/create/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut create = Create::new(creator.actor_id.to_owned(), note.into_any_base()?);
     create
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(CreateType::Create)?)
       .set_to(public())
       .set_many_ccs(maa.addressed_ccs.to_owned())
       // Set the mention tags
@@ -244,11 +246,10 @@ impl ApubObjectType for Comment {
     let maa =
       collect_non_local_mentions_and_addresses(&self.content, &community, client, pool).await?;
 
-    let id = format!("{}/update/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut update = Update::new(creator.actor_id.to_owned(), note.into_any_base()?);
     update
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(UpdateType::Update)?)
       .set_to(public())
       .set_many_ccs(maa.addressed_ccs.to_owned())
       // Set the mention tags
@@ -280,11 +281,10 @@ impl ApubObjectType for Comment {
     let community_id = post.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut delete = Delete::new(creator.actor_id.to_owned(), note.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -315,21 +315,18 @@ impl ApubObjectType for Comment {
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
     // Generate a fake delete activity, with the correct object
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut delete = Delete::new(creator.actor_id.to_owned(), note.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), delete.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -359,11 +356,10 @@ impl ApubObjectType for Comment {
     let community_id = post.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut remove = Remove::new(mod_.actor_id.to_owned(), note.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -394,20 +390,18 @@ impl ApubObjectType for Comment {
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
     // Generate a fake delete activity, with the correct object
-    let id = format!("{}/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut remove = Remove::new(mod_.actor_id.to_owned(), note.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
     // Undo that fake activity
-    let undo_id = format!("{}/undo/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(mod_.actor_id.to_owned(), remove.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -440,12 +434,10 @@ impl ApubLikeableType for Comment {
     let community_id = post.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/like/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut like = Like::new(creator.actor_id.to_owned(), note.into_any_base()?);
     like
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(LikeType::Like)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -475,12 +467,10 @@ impl ApubLikeableType for Comment {
     let community_id = post.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/dislike/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut dislike = Dislike::new(creator.actor_id.to_owned(), note.into_any_base()?);
     dislike
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DislikeType::Dislike)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -510,22 +500,18 @@ impl ApubLikeableType for Comment {
     let community_id = post.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/dislike/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut like = Like::new(creator.actor_id.to_owned(), note.into_any_base()?);
     like
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DislikeType::Dislike)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/like/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), like.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -586,7 +572,7 @@ async fn collect_non_local_mentions_and_addresses(
       debug!("mention actor_id: {}", actor_id);
       addressed_ccs.push(actor_id.to_owned().to_string());
 
-      let mention_user = get_or_fetch_and_upsert_remote_user(&actor_id, client, pool).await?;
+      let mention_user = get_or_fetch_and_upsert_user(&actor_id, client, pool).await?;
       let shared_inbox = mention_user.get_shared_inbox_url();
 
       mention_inboxes.push(shared_inbox);

server/src/apub/community.rs

@@ -1,11 +1,11 @@
 use crate::{
   apub::{
-    activities::send_activity,
+    activities::{generate_activity_id, send_activity},
     create_apub_response,
     create_apub_tombstone_response,
     create_tombstone,
     extensions::group_extensions::GroupExtension,
-    fetcher::get_or_fetch_and_upsert_remote_user,
+    fetcher::get_or_fetch_and_upsert_user,
     get_shared_inbox,
     insert_activity,
     ActorType,
@@ -20,7 +20,15 @@ use crate::{
 };
 use activitystreams_ext::Ext2;
 use activitystreams_new::{
-  activity::{Accept, Announce, Delete, Follow, Remove, Undo},
+  activity::{
+    kind::{AcceptType, AnnounceType, DeleteType, LikeType, RemoveType, UndoType},
+    Accept,
+    Announce,
+    Delete,
+    Follow,
+    Remove,
+    Undo,
+  },
   actor::{kind::GroupType, ApActor, Endpoints, Group},
   base::{AnyBase, BaseExt},
   collection::UnorderedCollection,
@@ -107,12 +115,7 @@ impl ToApub for Community {
   }
 
   fn to_tombstone(&self) -> Result<Tombstone, LemmyError> {
-    create_tombstone(
+    create_tombstone(self.deleted, &self.actor_id, self.updated, GroupType::Group)
-      self.deleted,
-      &self.actor_id,
-      self.updated,
-      GroupType::Group.to_string(),
-    )
   }
 }
 
@@ -137,13 +140,12 @@ impl ActorType for Community {
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
     let actor_uri = follow.actor()?.as_single_xsd_any_uri().unwrap().to_string();
-    let id = format!("{}/accept/{}", self.actor_id, uuid::Uuid::new_v4());
 
     let mut accept = Accept::new(self.actor_id.to_owned(), follow.into_any_base()?);
     let to = format!("{}/inbox", actor_uri);
     accept
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(AcceptType::Accept)?)
       .set_to(to.clone());
 
     insert_activity(self.creator_id, accept.clone(), true, pool).await?;
@@ -160,12 +162,10 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     let group = self.to_apub(pool).await?;
 
-    let id = format!("{}/delete/{}", self.actor_id, uuid::Uuid::new_v4());
-
     let mut delete = Delete::new(creator.actor_id.to_owned(), group.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
@@ -188,22 +188,19 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     let group = self.to_apub(pool).await?;
 
-    let id = format!("{}/delete/{}", self.actor_id, uuid::Uuid::new_v4());
-
     let mut delete = Delete::new(creator.actor_id.to_owned(), group.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
     // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/delete/{}", self.actor_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), delete.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
@@ -226,12 +223,10 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     let group = self.to_apub(pool).await?;
 
-    let id = format!("{}/remove/{}", self.actor_id, uuid::Uuid::new_v4());
-
     let mut remove = Remove::new(mod_.actor_id.to_owned(), group.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
@@ -254,21 +249,18 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     let group = self.to_apub(pool).await?;
 
-    let id = format!("{}/remove/{}", self.actor_id, uuid::Uuid::new_v4());
-
     let mut remove = Remove::new(mod_.actor_id.to_owned(), group.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
     // Undo that fake activity
-    let undo_id = format!("{}/undo/remove/{}", self.actor_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(mod_.actor_id.to_owned(), remove.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(LikeType::Like)?)
       .set_to(public())
       .set_many_ccs(vec![self.get_followers_url()]);
 
@@ -318,6 +310,10 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     unimplemented!()
   }
+
+  fn user_id(&self) -> i32 {
+    self.creator_id
+  }
 }
 
 #[async_trait::async_trait(?Send)]
@@ -325,12 +321,7 @@ impl FromApub for CommunityForm {
   type ApubType = GroupExt;
 
   /// Parse an ActivityPub group received from another instance into a Lemmy community.
-  async fn from_apub(
+  async fn from_apub(group: &GroupExt, client: &Client, pool: &DbPool) -> Result<Self, LemmyError> {
-    group: &GroupExt,
-    client: &Client,
-    pool: &DbPool,
-    actor_id: &Url,
-  ) -> Result<Self, LemmyError> {
     let creator_and_moderator_uris = group.inner.attributed_to().unwrap();
     let creator_uri = creator_and_moderator_uris
       .as_many()
@@ -341,7 +332,7 @@ impl FromApub for CommunityForm {
       .as_xsd_any_uri()
       .unwrap();
 
-    let creator = get_or_fetch_and_upsert_remote_user(creator_uri, client, pool).await?;
+    let creator = get_or_fetch_and_upsert_user(creator_uri, client, pool).await?;
 
     Ok(CommunityForm {
       name: group
@@ -367,11 +358,7 @@ impl FromApub for CommunityForm {
       updated: group.inner.updated().map(|u| u.to_owned().naive_local()),
       deleted: None,
       nsfw: group.ext_one.sensitive,
-      actor_id: group
+      actor_id: group.inner.id_unchecked().unwrap().to_string(),
-        .inner
-        .id(actor_id.domain().unwrap())?
-        .unwrap()
-        .to_string(),
       local: false,
       private_key: None,
       public_key: Some(group.ext_two.to_owned().public_key.public_key_pem),
@@ -427,15 +414,14 @@ pub async fn get_apub_community_followers(
 pub async fn do_announce(
   activity: AnyBase,
   community: &Community,
-  sender: &dyn ActorType,
+  sender: &User_,
   client: &Client,
   pool: &DbPool,
-) -> Result<HttpResponse, LemmyError> {
+) -> Result<(), LemmyError> {
-  let id = format!("{}/announce/{}", community.actor_id, uuid::Uuid::new_v4());
   let mut announce = Announce::new(community.actor_id.to_owned(), activity);
   announce
     .set_context(context())
-    .set_id(Url::parse(&id)?)
+    .set_id(generate_activity_id(AnnounceType::Announce)?)
     .set_to(public())
     .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -450,5 +436,5 @@ pub async fn do_announce(
 
   send_activity(client, &announce.into_any_base()?, community, to).await?;
 
-  Ok(HttpResponse::Ok().finish())
+  Ok(())
 }

server/src/apub/fetcher.rs

@@ -1,6 +1,14 @@
 use crate::{
   api::site::SearchResponse,
-  apub::{is_apub_id_valid, FromApub, GroupExt, PageExt, PersonExt, APUB_JSON_CONTENT_TYPE},
+  apub::{
+    is_apub_id_valid,
+    ActorType,
+    FromApub,
+    GroupExt,
+    PageExt,
+    PersonExt,
+    APUB_JSON_CONTENT_TYPE,
+  },
   blocking,
   request::{retry, RecvError},
   routes::nodeinfo::{NodeInfo, NodeInfoWellKnown},
@@ -141,7 +149,7 @@ pub async fn search_by_apub_id(
     SearchAcceptedObjects::Person(p) => {
       let user_uri = p.inner.id(domain)?.unwrap();
 
-      let user = get_or_fetch_and_upsert_remote_user(&user_uri, client, pool).await?;
+      let user = get_or_fetch_and_upsert_user(&user_uri, client, pool).await?;
 
       response.users = vec![blocking(pool, move |conn| UserView::read(conn, user.id)).await??];
 
@@ -150,7 +158,7 @@ pub async fn search_by_apub_id(
     SearchAcceptedObjects::Group(g) => {
       let community_uri = g.inner.id(domain)?.unwrap();
 
-      let community = get_or_fetch_and_upsert_remote_community(community_uri, client, pool).await?;
+      let community = get_or_fetch_and_upsert_community(community_uri, client, pool).await?;
 
       // TODO Maybe at some point in the future, fetch all the history of a community
       // fetch_community_outbox(&c, conn)?;
@@ -164,7 +172,7 @@ pub async fn search_by_apub_id(
       response
     }
     SearchAcceptedObjects::Page(p) => {
-      let post_form = PostForm::from_apub(&p, client, pool, &query_url).await?;
+      let post_form = PostForm::from_apub(&p, client, pool).await?;
 
       let p = blocking(pool, move |conn| upsert_post(&post_form, conn)).await??;
       response.posts = vec![blocking(pool, move |conn| PostView::read(conn, p.id, None)).await??];
@@ -177,8 +185,8 @@ pub async fn search_by_apub_id(
       // TODO: also fetch parent comments if any
       let x = post_url.first().unwrap().as_xsd_any_uri().unwrap();
       let post = fetch_remote_object(client, x).await?;
-      let post_form = PostForm::from_apub(&post, client, pool, &query_url).await?;
+      let post_form = PostForm::from_apub(&post, client, pool).await?;
-      let comment_form = CommentForm::from_apub(&c, client, pool, &query_url).await?;
+      let comment_form = CommentForm::from_apub(&c, client, pool).await?;
 
       blocking(pool, move |conn| upsert_post(&post_form, conn)).await??;
       let c = blocking(pool, move |conn| upsert_comment(&comment_form, conn)).await??;
@@ -192,8 +200,21 @@ pub async fn search_by_apub_id(
   Ok(response)
 }
 
+pub async fn get_or_fetch_and_upsert_actor(
+  apub_id: &Url,
+  client: &Client,
+  pool: &DbPool,
+) -> Result<Box<dyn ActorType>, LemmyError> {
+  let user = get_or_fetch_and_upsert_user(apub_id, client, pool).await;
+  let actor: Box<dyn ActorType> = match user {
+    Ok(u) => Box::new(u),
+    Err(_) => Box::new(get_or_fetch_and_upsert_community(apub_id, client, pool).await?),
+  };
+  Ok(actor)
+}
+
 /// Check if a remote user exists, create if not found, if its too old update it.Fetch a user, insert/update it in the database and return the user.
-pub async fn get_or_fetch_and_upsert_remote_user(
+pub async fn get_or_fetch_and_upsert_user(
   apub_id: &Url,
   client: &Client,
   pool: &DbPool,
@@ -210,7 +231,7 @@ pub async fn get_or_fetch_and_upsert_remote_user(
       debug!("Fetching and updating from remote user: {}", apub_id);
       let person = fetch_remote_object::<PersonExt>(client, apub_id).await?;
 
-      let mut uf = UserForm::from_apub(&person, client, pool, apub_id).await?;
+      let mut uf = UserForm::from_apub(&person, client, pool).await?;
       uf.last_refreshed_at = Some(naive_now());
       let user = blocking(pool, move |conn| User_::update(conn, u.id, &uf)).await??;
 
@@ -221,7 +242,7 @@ pub async fn get_or_fetch_and_upsert_remote_user(
       debug!("Fetching and creating remote user: {}", apub_id);
       let person = fetch_remote_object::<PersonExt>(client, apub_id).await?;
 
-      let uf = UserForm::from_apub(&person, client, pool, apub_id).await?;
+      let uf = UserForm::from_apub(&person, client, pool).await?;
       let user = blocking(pool, move |conn| User_::create(conn, &uf)).await??;
 
       Ok(user)
@@ -245,7 +266,7 @@ fn should_refetch_actor(last_refreshed: NaiveDateTime) -> bool {
 }
 
 /// Check if a remote community exists, create if not found, if its too old update it.Fetch a community, insert/update it in the database and return the community.
-pub async fn get_or_fetch_and_upsert_remote_community(
+pub async fn get_or_fetch_and_upsert_community(
   apub_id: &Url,
   client: &Client,
   pool: &DbPool,
@@ -261,7 +282,7 @@ pub async fn get_or_fetch_and_upsert_remote_community(
       debug!("Fetching and updating from remote community: {}", apub_id);
       let group = fetch_remote_object::<GroupExt>(client, apub_id).await?;
 
-      let mut cf = CommunityForm::from_apub(&group, client, pool, apub_id).await?;
+      let mut cf = CommunityForm::from_apub(&group, client, pool).await?;
       cf.last_refreshed_at = Some(naive_now());
       let community = blocking(pool, move |conn| Community::update(conn, c.id, &cf)).await??;
 
@@ -272,7 +293,7 @@ pub async fn get_or_fetch_and_upsert_remote_community(
       debug!("Fetching and creating remote community: {}", apub_id);
       let group = fetch_remote_object::<GroupExt>(client, apub_id).await?;
 
-      let cf = CommunityForm::from_apub(&group, client, pool, apub_id).await?;
+      let cf = CommunityForm::from_apub(&group, client, pool).await?;
       let community = blocking(pool, move |conn| Community::create(conn, &cf)).await??;
 
       // Also add the community moderators too
@@ -287,7 +308,7 @@ pub async fn get_or_fetch_and_upsert_remote_community(
       let mut creator_and_moderators = Vec::new();
 
       for uri in creator_and_moderator_uris {
-        let c_or_m = get_or_fetch_and_upsert_remote_user(uri, client, pool).await?;
+        let c_or_m = get_or_fetch_and_upsert_user(uri, client, pool).await?;
 
         creator_and_moderators.push(c_or_m);
       }
@@ -321,7 +342,7 @@ fn upsert_post(post_form: &PostForm, conn: &PgConnection) -> Result<Post, LemmyE
   }
 }
 
-pub async fn get_or_fetch_and_insert_remote_post(
+pub async fn get_or_fetch_and_insert_post(
   post_ap_id: &Url,
   client: &Client,
   pool: &DbPool,
@@ -337,7 +358,7 @@ pub async fn get_or_fetch_and_insert_remote_post(
     Err(NotFound {}) => {
       debug!("Fetching and creating remote post: {}", post_ap_id);
       let post = fetch_remote_object::<PageExt>(client, post_ap_id).await?;
-      let post_form = PostForm::from_apub(&post, client, pool, post_ap_id).await?;
+      let post_form = PostForm::from_apub(&post, client, pool).await?;
 
       let post = blocking(pool, move |conn| Post::create(conn, &post_form)).await??;
 
@@ -356,7 +377,7 @@ fn upsert_comment(comment_form: &CommentForm, conn: &PgConnection) -> Result<Com
   }
 }
 
-pub async fn get_or_fetch_and_insert_remote_comment(
+pub async fn get_or_fetch_and_insert_comment(
   comment_ap_id: &Url,
   client: &Client,
   pool: &DbPool,
@@ -375,7 +396,7 @@ pub async fn get_or_fetch_and_insert_remote_comment(
         comment_ap_id
       );
       let comment = fetch_remote_object::<Note>(client, comment_ap_id).await?;
-      let comment_form = CommentForm::from_apub(&comment, client, pool, comment_ap_id).await?;
+      let comment_form = CommentForm::from_apub(&comment, client, pool).await?;
 
       let comment = blocking(pool, move |conn| Comment::create(conn, &comment_form)).await??;
 

server/src/apub/inbox/activities/announce.rs

@@ -0,0 +1,41 @@
+use crate::{
+  apub::inbox::{
+    activities::{
+      create::receive_create,
+      delete::receive_delete,
+      dislike::receive_dislike,
+      like::receive_like,
+      remove::receive_remove,
+      undo::receive_undo,
+      update::receive_update,
+    },
+    shared_inbox::receive_unhandled_activity,
+  },
+  routes::ChatServerParam,
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::*, base::AnyBase, prelude::ExtendsExt};
+use actix_web::{client::Client, HttpResponse};
+
+pub async fn receive_announce(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let announce = Announce::from_any_base(activity)?.unwrap();
+  let kind = announce.object().as_single_kind_str();
+  let object = announce.object();
+  let object2 = object.clone().one().unwrap();
+  match kind {
+    Some("Create") => receive_create(object2, client, pool, chat_server).await,
+    Some("Update") => receive_update(object2, client, pool, chat_server).await,
+    Some("Like") => receive_like(object2, client, pool, chat_server).await,
+    Some("Dislike") => receive_dislike(object2, client, pool, chat_server).await,
+    Some("Delete") => receive_delete(object2, client, pool, chat_server).await,
+    Some("Remove") => receive_remove(object2, client, pool, chat_server).await,
+    Some("Undo") => receive_undo(object2, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(announce),
+  }
+}

server/src/apub/inbox/activities/create.rs

@@ -0,0 +1,126 @@
+use crate::{
+  api::{
+    comment::{send_local_notifs, CommentResponse},
+    post::PostResponse,
+  },
+  apub::{
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Create, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{Comment, CommentForm},
+  comment_view::CommentView,
+  post::{Post, PostForm},
+  post_view::PostView,
+  Crud,
+};
+use lemmy_utils::scrape_text_for_mentions;
+
+pub async fn receive_create(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let create = Create::from_any_base(activity)?.unwrap();
+  dbg!(create.object().as_single_kind_str());
+  match create.object().as_single_kind_str() {
+    Some("Page") => receive_create_post(create, client, pool, chat_server).await,
+    Some("Note") => receive_create_comment(create, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(create),
+  }
+}
+
+async fn receive_create_post(
+  create: Create,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&create, client, pool).await?;
+  let page = PageExt::from_any_base(create.object().to_owned().one().unwrap())?.unwrap();
+
+  let post = PostForm::from_apub(&page, client, pool).await?;
+
+  let inserted_post = blocking(pool, move |conn| Post::create(conn, &post)).await??;
+
+  // Refetch the view
+  let inserted_post_id = inserted_post.id;
+  let post_view = blocking(pool, move |conn| {
+    PostView::read(conn, inserted_post_id, None)
+  })
+  .await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::CreatePost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(create, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_create_comment(
+  create: Create,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&create, client, pool).await?;
+  let note = Note::from_any_base(create.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment = CommentForm::from_apub(&note, client, pool).await?;
+
+  let inserted_comment = blocking(pool, move |conn| Comment::create(conn, &comment)).await??;
+
+  let post_id = inserted_comment.post_id;
+  let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;
+
+  // Note:
+  // Although mentions could be gotten from the post tags (they are included there), or the ccs,
+  // Its much easier to scrape them from the comment body, since the API has to do that
+  // anyway.
+  let mentions = scrape_text_for_mentions(&inserted_comment.content);
+  let recipient_ids =
+    send_local_notifs(mentions, inserted_comment.clone(), &user, post, pool, true).await?;
+
+  // Refetch the view
+  let comment_view = blocking(pool, move |conn| {
+    CommentView::read(conn, inserted_comment.id, None)
+  })
+  .await??;
+
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::CreateComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(create, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/delete.rs

@@ -0,0 +1,223 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Delete, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{Comment, CommentForm},
+  comment_view::CommentView,
+  community::{Community, CommunityForm},
+  community_view::CommunityView,
+  naive_now,
+  post::{Post, PostForm},
+  post_view::PostView,
+  Crud,
+};
+
+pub async fn receive_delete(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let delete = Delete::from_any_base(activity)?.unwrap();
+  match delete.object().as_single_kind_str() {
+    Some("Page") => receive_delete_post(delete, client, pool, chat_server).await,
+    Some("Note") => receive_delete_comment(delete, client, pool, chat_server).await,
+    Some("Group") => receive_delete_community(delete, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(delete),
+  }
+}
+
+async fn receive_delete_post(
+  delete: Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&delete, client, pool).await?;
+  let page = PageExt::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+
+  let post_ap_id = PostForm::from_apub(&page, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_post(&post_ap_id, client, pool).await?;
+
+  let post_form = PostForm {
+    name: post.name.to_owned(),
+    url: post.url.to_owned(),
+    body: post.body.to_owned(),
+    creator_id: post.creator_id.to_owned(),
+    community_id: post.community_id,
+    removed: None,
+    deleted: Some(true),
+    nsfw: post.nsfw,
+    locked: None,
+    stickied: None,
+    updated: Some(naive_now()),
+    embed_title: post.embed_title,
+    embed_description: post.embed_description,
+    embed_html: post.embed_html,
+    thumbnail_url: post.thumbnail_url,
+    ap_id: post.ap_id,
+    local: post.local,
+    published: None,
+  };
+  let post_id = post.id;
+  blocking(pool, move |conn| Post::update(conn, post_id, &post_form)).await??;
+
+  // Refetch the view
+  let post_id = post.id;
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::EditPost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(delete, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_delete_comment(
+  delete: Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&delete, client, pool).await?;
+  let note = Note::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment_ap_id = CommentForm::from_apub(&note, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_comment(&comment_ap_id, client, pool).await?;
+
+  let comment_form = CommentForm {
+    content: comment.content.to_owned(),
+    parent_id: comment.parent_id,
+    post_id: comment.post_id,
+    creator_id: comment.creator_id,
+    removed: None,
+    deleted: Some(true),
+    read: None,
+    published: None,
+    updated: Some(naive_now()),
+    ap_id: comment.ap_id,
+    local: comment.local,
+  };
+  let comment_id = comment.id;
+  blocking(pool, move |conn| {
+    Comment::update(conn, comment_id, &comment_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_id = comment.id;
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::EditComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(delete, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_delete_community(
+  delete: Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let group = GroupExt::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+  let user = get_user_from_activity(&delete, client, pool).await?;
+
+  let community_actor_id = CommunityForm::from_apub(&group, client, pool)
+    .await?
+    .actor_id;
+
+  let community = blocking(pool, move |conn| {
+    Community::read_from_actor_id(conn, &community_actor_id)
+  })
+  .await??;
+
+  let community_form = CommunityForm {
+    name: community.name.to_owned(),
+    title: community.title.to_owned(),
+    description: community.description.to_owned(),
+    category_id: community.category_id, // Note: need to keep this due to foreign key constraint
+    creator_id: community.creator_id,   // Note: need to keep this due to foreign key constraint
+    removed: None,
+    published: None,
+    updated: Some(naive_now()),
+    deleted: Some(true),
+    nsfw: community.nsfw,
+    actor_id: community.actor_id,
+    local: community.local,
+    private_key: community.private_key,
+    public_key: community.public_key,
+    last_refreshed_at: None,
+  };
+
+  let community_id = community.id;
+  blocking(pool, move |conn| {
+    Community::update(conn, community_id, &community_form)
+  })
+  .await??;
+
+  let community_id = community.id;
+  let res = CommunityResponse {
+    community: blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, None)
+    })
+    .await??,
+  };
+
+  let community_id = res.community.id;
+
+  chat_server.do_send(SendCommunityRoomMessage {
+    op: UserOperation::EditCommunity,
+    response: res,
+    community_id,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(delete, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/dislike.rs

@@ -0,0 +1,134 @@
+use crate::{
+  api::{comment::CommentResponse, post::PostResponse},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Dislike, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{CommentForm, CommentLike, CommentLikeForm},
+  comment_view::CommentView,
+  post::{PostForm, PostLike, PostLikeForm},
+  post_view::PostView,
+  Likeable,
+};
+
+pub async fn receive_dislike(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let dislike = Dislike::from_any_base(activity)?.unwrap();
+  match dislike.object().as_single_kind_str() {
+    Some("Page") => receive_dislike_post(dislike, client, pool, chat_server).await,
+    Some("Note") => receive_dislike_comment(dislike, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(dislike),
+  }
+}
+
+async fn receive_dislike_post(
+  dislike: Dislike,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&dislike, client, pool).await?;
+  let page = PageExt::from_any_base(dislike.object().to_owned().one().unwrap())?.unwrap();
+
+  let post = PostForm::from_apub(&page, client, pool).await?;
+
+  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = PostLikeForm {
+    post_id,
+    user_id: user.id,
+    score: -1,
+  };
+  blocking(pool, move |conn| {
+    PostLike::remove(conn, &like_form)?;
+    PostLike::like(conn, &like_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::CreatePostLike,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(dislike, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_dislike_comment(
+  dislike: Dislike,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let note = Note::from_any_base(dislike.object().to_owned().one().unwrap())?.unwrap();
+  let user = get_user_from_activity(&dislike, client, pool).await?;
+
+  let comment = CommentForm::from_apub(&note, client, pool).await?;
+
+  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = CommentLikeForm {
+    comment_id,
+    post_id: comment.post_id,
+    user_id: user.id,
+    score: -1,
+  };
+  blocking(pool, move |conn| {
+    CommentLike::remove(conn, &like_form)?;
+    CommentLike::like(conn, &like_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::CreateCommentLike,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(dislike, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/like.rs

@@ -0,0 +1,134 @@
+use crate::{
+  api::{comment::CommentResponse, post::PostResponse},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Like, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{CommentForm, CommentLike, CommentLikeForm},
+  comment_view::CommentView,
+  post::{PostForm, PostLike, PostLikeForm},
+  post_view::PostView,
+  Likeable,
+};
+
+pub async fn receive_like(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let like = Like::from_any_base(activity)?.unwrap();
+  match like.object().as_single_kind_str() {
+    Some("Page") => receive_like_post(like, client, pool, chat_server).await,
+    Some("Note") => receive_like_comment(like, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(like),
+  }
+}
+
+async fn receive_like_post(
+  like: Like,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&like, client, pool).await?;
+  let page = PageExt::from_any_base(like.object().to_owned().one().unwrap())?.unwrap();
+
+  let post = PostForm::from_apub(&page, client, pool).await?;
+
+  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = PostLikeForm {
+    post_id,
+    user_id: user.id,
+    score: 1,
+  };
+  blocking(pool, move |conn| {
+    PostLike::remove(conn, &like_form)?;
+    PostLike::like(conn, &like_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::CreatePostLike,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(like, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_like_comment(
+  like: Like,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let note = Note::from_any_base(like.object().to_owned().one().unwrap())?.unwrap();
+  let user = get_user_from_activity(&like, client, pool).await?;
+
+  let comment = CommentForm::from_apub(&note, client, pool).await?;
+
+  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = CommentLikeForm {
+    comment_id,
+    post_id: comment.post_id,
+    user_id: user.id,
+    score: 1,
+  };
+  blocking(pool, move |conn| {
+    CommentLike::remove(conn, &like_form)?;
+    CommentLike::like(conn, &like_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::CreateCommentLike,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(like, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/mod.rs

@@ -0,0 +1,8 @@
+pub mod announce;
+pub mod create;
+pub mod delete;
+pub mod dislike;
+pub mod like;
+pub mod remove;
+pub mod undo;
+pub mod update;

server/src/apub/inbox/activities/remove.rs

@@ -0,0 +1,223 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Remove, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{Comment, CommentForm},
+  comment_view::CommentView,
+  community::{Community, CommunityForm},
+  community_view::CommunityView,
+  naive_now,
+  post::{Post, PostForm},
+  post_view::PostView,
+  Crud,
+};
+
+pub async fn receive_remove(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let remove = Remove::from_any_base(activity)?.unwrap();
+  match remove.object().as_single_kind_str() {
+    Some("Page") => receive_remove_post(remove, client, pool, chat_server).await,
+    Some("Note") => receive_remove_comment(remove, client, pool, chat_server).await,
+    Some("Group") => receive_remove_community(remove, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(remove),
+  }
+}
+
+async fn receive_remove_post(
+  remove: Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(&remove, client, pool).await?;
+  let page = PageExt::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let post_ap_id = PostForm::from_apub(&page, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_post(&post_ap_id, client, pool).await?;
+
+  let post_form = PostForm {
+    name: post.name.to_owned(),
+    url: post.url.to_owned(),
+    body: post.body.to_owned(),
+    creator_id: post.creator_id.to_owned(),
+    community_id: post.community_id,
+    removed: Some(true),
+    deleted: None,
+    nsfw: post.nsfw,
+    locked: None,
+    stickied: None,
+    updated: Some(naive_now()),
+    embed_title: post.embed_title,
+    embed_description: post.embed_description,
+    embed_html: post.embed_html,
+    thumbnail_url: post.thumbnail_url,
+    ap_id: post.ap_id,
+    local: post.local,
+    published: None,
+  };
+  let post_id = post.id;
+  blocking(pool, move |conn| Post::update(conn, post_id, &post_form)).await??;
+
+  // Refetch the view
+  let post_id = post.id;
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::EditPost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(remove, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_remove_comment(
+  remove: Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(&remove, client, pool).await?;
+  let note = Note::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment_ap_id = CommentForm::from_apub(&note, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_comment(&comment_ap_id, client, pool).await?;
+
+  let comment_form = CommentForm {
+    content: comment.content.to_owned(),
+    parent_id: comment.parent_id,
+    post_id: comment.post_id,
+    creator_id: comment.creator_id,
+    removed: Some(true),
+    deleted: None,
+    read: None,
+    published: None,
+    updated: Some(naive_now()),
+    ap_id: comment.ap_id,
+    local: comment.local,
+  };
+  let comment_id = comment.id;
+  blocking(pool, move |conn| {
+    Comment::update(conn, comment_id, &comment_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_id = comment.id;
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::EditComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(remove, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_remove_community(
+  remove: Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(&remove, client, pool).await?;
+  let group = GroupExt::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let community_actor_id = CommunityForm::from_apub(&group, client, pool)
+    .await?
+    .actor_id;
+
+  let community = blocking(pool, move |conn| {
+    Community::read_from_actor_id(conn, &community_actor_id)
+  })
+  .await??;
+
+  let community_form = CommunityForm {
+    name: community.name.to_owned(),
+    title: community.title.to_owned(),
+    description: community.description.to_owned(),
+    category_id: community.category_id, // Note: need to keep this due to foreign key constraint
+    creator_id: community.creator_id,   // Note: need to keep this due to foreign key constraint
+    removed: Some(true),
+    published: None,
+    updated: Some(naive_now()),
+    deleted: None,
+    nsfw: community.nsfw,
+    actor_id: community.actor_id,
+    local: community.local,
+    private_key: community.private_key,
+    public_key: community.public_key,
+    last_refreshed_at: None,
+  };
+
+  let community_id = community.id;
+  blocking(pool, move |conn| {
+    Community::update(conn, community_id, &community_form)
+  })
+  .await??;
+
+  let community_id = community.id;
+  let res = CommunityResponse {
+    community: blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, None)
+    })
+    .await??,
+  };
+
+  let community_id = res.community.id;
+
+  chat_server.do_send(SendCommunityRoomMessage {
+    op: UserOperation::EditCommunity,
+    response: res,
+    community_id,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(remove, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/undo.rs

@@ -0,0 +1,551 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::*, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{Comment, CommentForm, CommentLike, CommentLikeForm},
+  comment_view::CommentView,
+  community::{Community, CommunityForm},
+  community_view::CommunityView,
+  naive_now,
+  post::{Post, PostForm, PostLike, PostLikeForm},
+  post_view::PostView,
+  Crud,
+  Likeable,
+};
+
+pub async fn receive_undo(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let undo = Undo::from_any_base(activity)?.unwrap();
+  match undo.object().as_single_kind_str() {
+    Some("Delete") => receive_undo_delete(undo, client, pool, chat_server).await,
+    Some("Remove") => receive_undo_remove(undo, client, pool, chat_server).await,
+    Some("Like") => receive_undo_like(undo, client, pool, chat_server).await,
+    Some("Dislike") => receive_undo_dislike(undo, client, pool, chat_server).await,
+    // TODO: handle undo_dislike?
+    _ => receive_unhandled_activity(undo),
+  }
+}
+
+async fn receive_undo_delete(
+  undo: Undo,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let delete = Delete::from_any_base(undo.object().to_owned().one().unwrap())?.unwrap();
+  let type_ = delete.object().as_single_kind_str().unwrap();
+  match type_ {
+    "Note" => receive_undo_delete_comment(undo, &delete, client, pool, chat_server).await,
+    "Page" => receive_undo_delete_post(undo, &delete, client, pool, chat_server).await,
+    "Group" => receive_undo_delete_community(undo, &delete, client, pool, chat_server).await,
+    d => Err(format_err!("Undo Delete type {} not supported", d).into()),
+  }
+}
+
+async fn receive_undo_remove(
+  undo: Undo,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let remove = Remove::from_any_base(undo.object().to_owned().one().unwrap())?.unwrap();
+
+  let type_ = remove.object().as_single_kind_str().unwrap();
+  match type_ {
+    "Note" => receive_undo_remove_comment(undo, &remove, client, pool, chat_server).await,
+    "Page" => receive_undo_remove_post(undo, &remove, client, pool, chat_server).await,
+    "Group" => receive_undo_remove_community(undo, &remove, client, pool, chat_server).await,
+    d => Err(format_err!("Undo Delete type {} not supported", d).into()),
+  }
+}
+
+async fn receive_undo_like(
+  undo: Undo,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let like = Like::from_any_base(undo.object().to_owned().one().unwrap())?.unwrap();
+
+  let type_ = like.object().as_single_kind_str().unwrap();
+  match type_ {
+    "Note" => receive_undo_like_comment(undo, &like, client, pool, chat_server).await,
+    "Page" => receive_undo_like_post(undo, &like, client, pool, chat_server).await,
+    d => Err(format_err!("Undo Delete type {} not supported", d).into()),
+  }
+}
+
+async fn receive_undo_dislike(
+  undo: Undo,
+  _client: &Client,
+  _pool: &DbPool,
+  _chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let dislike = Dislike::from_any_base(undo.object().to_owned().one().unwrap())?.unwrap();
+
+  let type_ = dislike.object().as_single_kind_str().unwrap();
+  Err(format_err!("Undo Delete type {} not supported", type_).into())
+}
+
+async fn receive_undo_delete_comment(
+  undo: Undo,
+  delete: &Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(delete, client, pool).await?;
+  let note = Note::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment_ap_id = CommentForm::from_apub(&note, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_comment(&comment_ap_id, client, pool).await?;
+
+  let comment_form = CommentForm {
+    content: comment.content.to_owned(),
+    parent_id: comment.parent_id,
+    post_id: comment.post_id,
+    creator_id: comment.creator_id,
+    removed: None,
+    deleted: Some(false),
+    read: None,
+    published: None,
+    updated: Some(naive_now()),
+    ap_id: comment.ap_id,
+    local: comment.local,
+  };
+  let comment_id = comment.id;
+  blocking(pool, move |conn| {
+    Comment::update(conn, comment_id, &comment_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_id = comment.id;
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::EditComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_remove_comment(
+  undo: Undo,
+  remove: &Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(remove, client, pool).await?;
+  let note = Note::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment_ap_id = CommentForm::from_apub(&note, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_comment(&comment_ap_id, client, pool).await?;
+
+  let comment_form = CommentForm {
+    content: comment.content.to_owned(),
+    parent_id: comment.parent_id,
+    post_id: comment.post_id,
+    creator_id: comment.creator_id,
+    removed: Some(false),
+    deleted: None,
+    read: None,
+    published: None,
+    updated: Some(naive_now()),
+    ap_id: comment.ap_id,
+    local: comment.local,
+  };
+  let comment_id = comment.id;
+  blocking(pool, move |conn| {
+    Comment::update(conn, comment_id, &comment_form)
+  })
+  .await??;
+
+  // Refetch the view
+  let comment_id = comment.id;
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::EditComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_delete_post(
+  undo: Undo,
+  delete: &Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(delete, client, pool).await?;
+  let page = PageExt::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+
+  let post_ap_id = PostForm::from_apub(&page, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_post(&post_ap_id, client, pool).await?;
+
+  let post_form = PostForm {
+    name: post.name.to_owned(),
+    url: post.url.to_owned(),
+    body: post.body.to_owned(),
+    creator_id: post.creator_id.to_owned(),
+    community_id: post.community_id,
+    removed: None,
+    deleted: Some(false),
+    nsfw: post.nsfw,
+    locked: None,
+    stickied: None,
+    updated: Some(naive_now()),
+    embed_title: post.embed_title,
+    embed_description: post.embed_description,
+    embed_html: post.embed_html,
+    thumbnail_url: post.thumbnail_url,
+    ap_id: post.ap_id,
+    local: post.local,
+    published: None,
+  };
+  let post_id = post.id;
+  blocking(pool, move |conn| Post::update(conn, post_id, &post_form)).await??;
+
+  // Refetch the view
+  let post_id = post.id;
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::EditPost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_remove_post(
+  undo: Undo,
+  remove: &Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(remove, client, pool).await?;
+  let page = PageExt::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let post_ap_id = PostForm::from_apub(&page, client, pool)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_post(&post_ap_id, client, pool).await?;
+
+  let post_form = PostForm {
+    name: post.name.to_owned(),
+    url: post.url.to_owned(),
+    body: post.body.to_owned(),
+    creator_id: post.creator_id.to_owned(),
+    community_id: post.community_id,
+    removed: Some(false),
+    deleted: None,
+    nsfw: post.nsfw,
+    locked: None,
+    stickied: None,
+    updated: Some(naive_now()),
+    embed_title: post.embed_title,
+    embed_description: post.embed_description,
+    embed_html: post.embed_html,
+    thumbnail_url: post.thumbnail_url,
+    ap_id: post.ap_id,
+    local: post.local,
+    published: None,
+  };
+  let post_id = post.id;
+  blocking(pool, move |conn| Post::update(conn, post_id, &post_form)).await??;
+
+  // Refetch the view
+  let post_id = post.id;
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::EditPost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_delete_community(
+  undo: Undo,
+  delete: &Delete,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(delete, client, pool).await?;
+  let group = GroupExt::from_any_base(delete.object().to_owned().one().unwrap())?.unwrap();
+
+  let community_actor_id = CommunityForm::from_apub(&group, client, pool)
+    .await?
+    .actor_id;
+
+  let community = blocking(pool, move |conn| {
+    Community::read_from_actor_id(conn, &community_actor_id)
+  })
+  .await??;
+
+  let community_form = CommunityForm {
+    name: community.name.to_owned(),
+    title: community.title.to_owned(),
+    description: community.description.to_owned(),
+    category_id: community.category_id, // Note: need to keep this due to foreign key constraint
+    creator_id: community.creator_id,   // Note: need to keep this due to foreign key constraint
+    removed: None,
+    published: None,
+    updated: Some(naive_now()),
+    deleted: Some(false),
+    nsfw: community.nsfw,
+    actor_id: community.actor_id,
+    local: community.local,
+    private_key: community.private_key,
+    public_key: community.public_key,
+    last_refreshed_at: None,
+  };
+
+  let community_id = community.id;
+  blocking(pool, move |conn| {
+    Community::update(conn, community_id, &community_form)
+  })
+  .await??;
+
+  let community_id = community.id;
+  let res = CommunityResponse {
+    community: blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, None)
+    })
+    .await??,
+  };
+
+  let community_id = res.community.id;
+
+  chat_server.do_send(SendCommunityRoomMessage {
+    op: UserOperation::EditCommunity,
+    response: res,
+    community_id,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_remove_community(
+  undo: Undo,
+  remove: &Remove,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let mod_ = get_user_from_activity(remove, client, pool).await?;
+  let group = GroupExt::from_any_base(remove.object().to_owned().one().unwrap())?.unwrap();
+
+  let community_actor_id = CommunityForm::from_apub(&group, client, pool)
+    .await?
+    .actor_id;
+
+  let community = blocking(pool, move |conn| {
+    Community::read_from_actor_id(conn, &community_actor_id)
+  })
+  .await??;
+
+  let community_form = CommunityForm {
+    name: community.name.to_owned(),
+    title: community.title.to_owned(),
+    description: community.description.to_owned(),
+    category_id: community.category_id, // Note: need to keep this due to foreign key constraint
+    creator_id: community.creator_id,   // Note: need to keep this due to foreign key constraint
+    removed: Some(false),
+    published: None,
+    updated: Some(naive_now()),
+    deleted: None,
+    nsfw: community.nsfw,
+    actor_id: community.actor_id,
+    local: community.local,
+    private_key: community.private_key,
+    public_key: community.public_key,
+    last_refreshed_at: None,
+  };
+
+  let community_id = community.id;
+  blocking(pool, move |conn| {
+    Community::update(conn, community_id, &community_form)
+  })
+  .await??;
+
+  let community_id = community.id;
+  let res = CommunityResponse {
+    community: blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, None)
+    })
+    .await??,
+  };
+
+  let community_id = res.community.id;
+
+  chat_server.do_send(SendCommunityRoomMessage {
+    op: UserOperation::EditCommunity,
+    response: res,
+    community_id,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &mod_, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_like_comment(
+  undo: Undo,
+  like: &Like,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(like, client, pool).await?;
+  let note = Note::from_any_base(like.object().to_owned().one().unwrap())?.unwrap();
+
+  let comment = CommentForm::from_apub(&note, client, pool).await?;
+
+  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = CommentLikeForm {
+    comment_id,
+    post_id: comment.post_id,
+    user_id: user.id,
+    score: 0,
+  };
+  blocking(pool, move |conn| CommentLike::remove(conn, &like_form)).await??;
+
+  // Refetch the view
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  // TODO get those recipient actor ids from somewhere
+  let recipient_ids = vec![];
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::CreateCommentLike,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_undo_like_post(
+  undo: Undo,
+  like: &Like,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(like, client, pool).await?;
+  let page = PageExt::from_any_base(like.object().to_owned().one().unwrap())?.unwrap();
+
+  let post = PostForm::from_apub(&page, client, pool).await?;
+
+  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let like_form = PostLikeForm {
+    post_id,
+    user_id: user.id,
+    score: 1,
+  };
+  blocking(pool, move |conn| PostLike::remove(conn, &like_form)).await??;
+
+  // Refetch the view
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::CreatePostLike,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(undo, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/activities/update.rs

@@ -0,0 +1,127 @@
+use crate::{
+  api::{
+    comment::{send_local_notifs, CommentResponse},
+    post::PostResponse,
+  },
+  apub::{
+    fetcher::{get_or_fetch_and_insert_comment, get_or_fetch_and_insert_post},
+    inbox::shared_inbox::{
+      announce_if_community_is_local,
+      get_user_from_activity,
+      receive_unhandled_activity,
+    },
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Update, base::AnyBase, object::Note, prelude::*};
+use actix_web::{client::Client, HttpResponse};
+use lemmy_db::{
+  comment::{Comment, CommentForm},
+  comment_view::CommentView,
+  post::{Post, PostForm},
+  post_view::PostView,
+  Crud,
+};
+use lemmy_utils::scrape_text_for_mentions;
+
+pub async fn receive_update(
+  activity: AnyBase,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let update = Update::from_any_base(activity)?.unwrap();
+  match update.object().as_single_kind_str() {
+    Some("Page") => receive_update_post(update, client, pool, chat_server).await,
+    Some("Note") => receive_update_comment(update, client, pool, chat_server).await,
+    _ => receive_unhandled_activity(update),
+  }
+}
+
+async fn receive_update_post(
+  update: Update,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let user = get_user_from_activity(&update, client, pool).await?;
+  let page = PageExt::from_any_base(update.object().to_owned().one().unwrap())?.unwrap();
+
+  let post = PostForm::from_apub(&page, client, pool).await?;
+
+  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  blocking(pool, move |conn| Post::update(conn, post_id, &post)).await??;
+
+  // Refetch the view
+  let post_view = blocking(pool, move |conn| PostView::read(conn, post_id, None)).await??;
+
+  let res = PostResponse { post: post_view };
+
+  chat_server.do_send(SendPost {
+    op: UserOperation::EditPost,
+    post: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(update, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}
+
+async fn receive_update_comment(
+  update: Update,
+  client: &Client,
+  pool: &DbPool,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let note = Note::from_any_base(update.object().to_owned().one().unwrap())?.unwrap();
+  let user = get_user_from_activity(&update, client, pool).await?;
+
+  let comment = CommentForm::from_apub(&note, client, pool).await?;
+
+  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, client, pool)
+    .await?
+    .id;
+
+  let updated_comment = blocking(pool, move |conn| {
+    Comment::update(conn, comment_id, &comment)
+  })
+  .await??;
+
+  let post_id = updated_comment.post_id;
+  let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;
+
+  let mentions = scrape_text_for_mentions(&updated_comment.content);
+  let recipient_ids =
+    send_local_notifs(mentions, updated_comment, &user, post, pool, false).await?;
+
+  // Refetch the view
+  let comment_view =
+    blocking(pool, move |conn| CommentView::read(conn, comment_id, None)).await??;
+
+  let res = CommentResponse {
+    comment: comment_view,
+    recipient_ids,
+    form_id: None,
+  };
+
+  chat_server.do_send(SendComment {
+    op: UserOperation::EditComment,
+    comment: res,
+    my_id: None,
+  });
+
+  announce_if_community_is_local(update, &user, client, pool).await?;
+  Ok(HttpResponse::Ok().finish())
+}

server/src/apub/inbox/community_inbox.rs

@@ -1,7 +1,7 @@
 use crate::{
   apub::{
     extensions::signatures::verify,
-    fetcher::{get_or_fetch_and_upsert_remote_community, get_or_fetch_and_upsert_remote_user},
+    fetcher::{get_or_fetch_and_upsert_community, get_or_fetch_and_upsert_user},
     insert_activity,
     ActorType,
   },
@@ -72,8 +72,8 @@ pub async fn community_inbox(
   let user_uri = follow.actor()?.as_single_xsd_any_uri().unwrap();
   let community_uri = follow.object().as_single_xsd_any_uri().unwrap();
 
-  let user = get_or_fetch_and_upsert_remote_user(&user_uri, &client, &db).await?;
+  let user = get_or_fetch_and_upsert_user(&user_uri, &client, &db).await?;
-  let community = get_or_fetch_and_upsert_remote_community(community_uri, &client, &db).await?;
+  let community = get_or_fetch_and_upsert_community(community_uri, &client, &db).await?;
 
   verify(&request, &user)?;
 

server/src/apub/inbox/mod.rs

@@ -0,0 +1,4 @@
+pub mod activities;
+pub mod community_inbox;
+pub mod shared_inbox;
+pub mod user_inbox;

server/src/apub/inbox/shared_inbox.rs

@@ -0,0 +1,141 @@
+use crate::{
+  apub::{
+    community::do_announce,
+    extensions::signatures::verify,
+    fetcher::{
+      get_or_fetch_and_upsert_actor,
+      get_or_fetch_and_upsert_community,
+      get_or_fetch_and_upsert_user,
+    },
+    inbox::activities::{
+      announce::receive_announce,
+      create::receive_create,
+      delete::receive_delete,
+      dislike::receive_dislike,
+      like::receive_like,
+      remove::receive_remove,
+      undo::receive_undo,
+      update::receive_update,
+    },
+    insert_activity,
+  },
+  routes::{ChatServerParam, DbPoolParam},
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{
+  activity::{ActorAndObject, ActorAndObjectRef},
+  base::{AsBase, Extends},
+  object::AsObject,
+  prelude::*,
+};
+use actix_web::{client::Client, web, HttpRequest, HttpResponse};
+use lemmy_db::user::User_;
+use log::debug;
+use serde::Serialize;
+use std::fmt::Debug;
+use url::Url;
+
+#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum ValidTypes {
+  Create,
+  Update,
+  Like,
+  Dislike,
+  Delete,
+  Undo,
+  Remove,
+  Announce,
+}
+
+// TODO: this isnt entirely correct, cause some of these activities are not ActorAndObject,
+//       but it might still work due to the anybase conversion
+pub type AcceptedActivities = ActorAndObject<ValidTypes>;
+
+/// Handler for all incoming activities to user inboxes.
+pub async fn shared_inbox(
+  request: HttpRequest,
+  input: web::Json<AcceptedActivities>,
+  client: web::Data<Client>,
+  pool: DbPoolParam,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, LemmyError> {
+  let activity = input.into_inner();
+
+  let json = serde_json::to_string(&activity)?;
+  debug!("Shared inbox received activity: {}", json);
+
+  let sender = &activity.actor()?.to_owned().single_xsd_any_uri().unwrap();
+
+  // TODO: pass this actor in instead of using get_user_from_activity()
+  let actor = get_or_fetch_and_upsert_actor(sender, &client, &pool).await?;
+  verify(&request, actor.as_ref())?;
+
+  insert_activity(actor.user_id(), activity.clone(), false, &pool).await?;
+
+  let any_base = activity.clone().into_any_base()?;
+  let kind = activity.kind().unwrap();
+  dbg!(kind);
+  match kind {
+    ValidTypes::Announce => receive_announce(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Create => receive_create(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Update => receive_update(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Like => receive_like(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Dislike => receive_dislike(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Remove => receive_remove(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Delete => receive_delete(any_base, &client, &pool, chat_server).await,
+    ValidTypes::Undo => receive_undo(any_base, &client, &pool, chat_server).await,
+  }
+}
+
+pub(in crate::apub::inbox) fn receive_unhandled_activity<A>(
+  activity: A,
+) -> Result<HttpResponse, LemmyError>
+where
+  A: Debug,
+{
+  debug!("received unhandled activity type: {:?}", activity);
+  Ok(HttpResponse::NotImplemented().finish())
+}
+
+pub(in crate::apub::inbox) async fn get_user_from_activity<T, A>(
+  activity: &T,
+  client: &Client,
+  pool: &DbPool,
+) -> Result<User_, LemmyError>
+where
+  T: AsBase<A> + ActorAndObjectRef,
+{
+  let actor = activity.actor()?;
+  let user_uri = actor.as_single_xsd_any_uri().unwrap();
+  get_or_fetch_and_upsert_user(&user_uri, client, pool).await
+}
+
+pub(in crate::apub::inbox) async fn announce_if_community_is_local<T, Kind>(
+  activity: T,
+  user: &User_,
+  client: &Client,
+  pool: &DbPool,
+) -> Result<(), LemmyError>
+where
+  T: AsObject<Kind>,
+  T: Extends<Kind>,
+  Kind: Serialize,
+  <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
+{
+  let cc = activity.cc().unwrap();
+  let cc = cc.as_many().unwrap();
+  let community_followers_uri = cc.first().unwrap().as_xsd_any_uri().unwrap();
+  // TODO: this is hacky but seems to be the only way to get the community ID
+  let community_uri = community_followers_uri
+    .to_string()
+    .replace("/followers", "");
+  let community =
+    get_or_fetch_and_upsert_community(&Url::parse(&community_uri)?, client, pool).await?;
+
+  if community.local {
+    do_announce(activity.into_any_base()?, &community, &user, client, pool).await?;
+  }
+  Ok(())
+}

server/src/apub/inbox/user_inbox.rs

@@ -2,7 +2,7 @@ use crate::{
   api::user::PrivateMessageResponse,
   apub::{
     extensions::signatures::verify,
-    fetcher::{get_or_fetch_and_upsert_remote_community, get_or_fetch_and_upsert_remote_user},
+    fetcher::{get_or_fetch_and_upsert_community, get_or_fetch_and_upsert_user},
     insert_activity,
     FromApub,
   },
@@ -82,7 +82,7 @@ async fn receive_accept(
 ) -> Result<HttpResponse, LemmyError> {
   let community_uri = accept.actor()?.to_owned().single_xsd_any_uri().unwrap();
 
-  let community = get_or_fetch_and_upsert_remote_community(&community_uri, client, pool).await?;
+  let community = get_or_fetch_and_upsert_community(&community_uri, client, pool).await?;
   verify(request, &community)?;
 
   let username = username.to_owned();
@@ -116,12 +116,12 @@ async fn receive_create_private_message(
   let user_uri = &create.actor()?.to_owned().single_xsd_any_uri().unwrap();
   let note = Note::from_any_base(create.object().as_one().unwrap().to_owned())?.unwrap();
 
-  let user = get_or_fetch_and_upsert_remote_user(user_uri, client, pool).await?;
+  let user = get_or_fetch_and_upsert_user(user_uri, client, pool).await?;
   verify(request, &user)?;
 
   insert_activity(user.id, create, false, pool).await?;
 
-  let private_message = PrivateMessageForm::from_apub(&note, client, pool, user_uri).await?;
+  let private_message = PrivateMessageForm::from_apub(&note, client, pool).await?;
 
   let inserted_private_message = blocking(pool, move |conn| {
     PrivateMessage::create(conn, &private_message)
@@ -157,12 +157,12 @@ async fn receive_update_private_message(
   let user_uri = &update.actor()?.to_owned().single_xsd_any_uri().unwrap();
   let note = Note::from_any_base(update.object().as_one().unwrap().to_owned())?.unwrap();
 
-  let user = get_or_fetch_and_upsert_remote_user(&user_uri, client, pool).await?;
+  let user = get_or_fetch_and_upsert_user(&user_uri, client, pool).await?;
   verify(request, &user)?;
 
   insert_activity(user.id, update, false, pool).await?;
 
-  let private_message_form = PrivateMessageForm::from_apub(&note, client, pool, user_uri).await?;
+  let private_message_form = PrivateMessageForm::from_apub(&note, client, pool).await?;
 
   let private_message_ap_id = private_message_form.ap_id.clone();
   let private_message = blocking(pool, move |conn| {
@@ -206,12 +206,12 @@ async fn receive_delete_private_message(
   let user_uri = &delete.actor()?.to_owned().single_xsd_any_uri().unwrap();
   let note = Note::from_any_base(delete.object().as_one().unwrap().to_owned())?.unwrap();
 
-  let user = get_or_fetch_and_upsert_remote_user(&user_uri, client, pool).await?;
+  let user = get_or_fetch_and_upsert_user(&user_uri, client, pool).await?;
   verify(request, &user)?;
 
   insert_activity(user.id, delete, false, pool).await?;
 
-  let private_message_form = PrivateMessageForm::from_apub(&note, client, pool, user_uri).await?;
+  let private_message_form = PrivateMessageForm::from_apub(&note, client, pool).await?;
 
   let private_message_ap_id = private_message_form.ap_id;
   let private_message = blocking(pool, move |conn| {
@@ -268,12 +268,12 @@ async fn receive_undo_delete_private_message(
   let note = Note::from_any_base(delete.object().as_one().unwrap().to_owned())?.unwrap();
   let user_uri = &delete.actor()?.to_owned().single_xsd_any_uri().unwrap();
 
-  let user = get_or_fetch_and_upsert_remote_user(&user_uri, client, pool).await?;
+  let user = get_or_fetch_and_upsert_user(&user_uri, client, pool).await?;
   verify(request, &user)?;
 
   insert_activity(user.id, delete, false, pool).await?;
 
-  let private_message = PrivateMessageForm::from_apub(&note, client, pool, user_uri).await?;
+  let private_message = PrivateMessageForm::from_apub(&note, client, pool).await?;
 
   let private_message_ap_id = private_message.ap_id.clone();
   let private_message_id = blocking(pool, move |conn| {

server/src/apub/mod.rs

@@ -1,14 +1,12 @@
 pub mod activities;
 pub mod comment;
 pub mod community;
-pub mod community_inbox;
 pub mod extensions;
 pub mod fetcher;
+pub mod inbox;
 pub mod post;
 pub mod private_message;
-pub mod shared_inbox;
 pub mod user;
-pub mod user_inbox;
 
 use crate::{
   apub::extensions::{
@@ -103,17 +101,20 @@ pub trait ToApub {
 }
 
 /// Updated is actually the deletion time
-fn create_tombstone(
+fn create_tombstone<T>(
   deleted: bool,
   object_id: &str,
   updated: Option<NaiveDateTime>,
-  former_type: String,
+  former_type: T,
-) -> Result<Tombstone, LemmyError> {
+) -> Result<Tombstone, LemmyError>
+where
+  T: ToString,
+{
   if deleted {
     if let Some(updated) = updated {
       let mut tombstone = Tombstone::new();
       tombstone.set_id(object_id.parse()?);
-      tombstone.set_former_type(former_type);
+      tombstone.set_former_type(former_type.to_string());
       tombstone.set_deleted(convert_datetime(updated));
       Ok(tombstone)
     } else {
@@ -131,7 +132,6 @@ pub trait FromApub {
     apub: &Self::ApubType,
     client: &Client,
     pool: &DbPool,
-    actor_id: &Url,
   ) -> Result<Self, LemmyError>
   where
     Self: Sized;
@@ -219,6 +219,9 @@ pub trait ActorType {
   fn public_key(&self) -> String;
   fn private_key(&self) -> String;
 
+  /// numeric id in the database, used for insert_activity
+  fn user_id(&self) -> i32;
+
   // These two have default impls, since currently a community can't follow anything,
   // and a user can't be followed (yet)
   #[allow(unused_variables)]

server/src/apub/post.rs

@@ -1,11 +1,11 @@
 use crate::{
   apub::{
-    activities::send_activity_to_community,
+    activities::{generate_activity_id, send_activity_to_community},
     create_apub_response,
     create_apub_tombstone_response,
     create_tombstone,
     extensions::page_extension::PageExtension,
-    fetcher::{get_or_fetch_and_upsert_remote_community, get_or_fetch_and_upsert_remote_user},
+    fetcher::{get_or_fetch_and_upsert_community, get_or_fetch_and_upsert_user},
     ActorType,
     ApubLikeableType,
     ApubObjectType,
@@ -20,7 +20,16 @@ use crate::{
 };
 use activitystreams_ext::Ext1;
 use activitystreams_new::{
-  activity::{Create, Delete, Dislike, Like, Remove, Undo, Update},
+  activity::{
+    kind::{CreateType, DeleteType, DislikeType, LikeType, RemoveType, UndoType, UpdateType},
+    Create,
+    Delete,
+    Dislike,
+    Like,
+    Remove,
+    Undo,
+    Update,
+  },
   context,
   object::{kind::PageType, Image, Page, Tombstone},
   prelude::*,
@@ -132,12 +141,7 @@ impl ToApub for Post {
   }
 
   fn to_tombstone(&self) -> Result<Tombstone, LemmyError> {
-    create_tombstone(
+    create_tombstone(self.deleted, &self.ap_id, self.updated, PageType::Page)
-      self.deleted,
-      &self.ap_id,
-      self.updated,
-      PageType::Page.to_string(),
-    )
   }
 }
 
@@ -150,7 +154,6 @@ impl FromApub for PostForm {
     page: &PageExt,
     client: &Client,
     pool: &DbPool,
-    actor_id: &Url,
   ) -> Result<PostForm, LemmyError> {
     let ext = &page.ext_one;
     let creator_actor_id = page
@@ -161,7 +164,7 @@ impl FromApub for PostForm {
       .as_single_xsd_any_uri()
       .unwrap();
 
-    let creator = get_or_fetch_and_upsert_remote_user(creator_actor_id, client, pool).await?;
+    let creator = get_or_fetch_and_upsert_user(creator_actor_id, client, pool).await?;
 
     let community_actor_id = page
       .inner
@@ -171,8 +174,7 @@ impl FromApub for PostForm {
       .as_single_xsd_any_uri()
       .unwrap();
 
-    let community =
+    let community = get_or_fetch_and_upsert_community(community_actor_id, client, pool).await?;
-      get_or_fetch_and_upsert_remote_community(community_actor_id, client, pool).await?;
 
     let thumbnail_url = match &page.inner.image() {
       Some(any_image) => Image::from_any_base(any_image.to_owned().as_one().unwrap().to_owned())?
@@ -243,11 +245,7 @@ impl FromApub for PostForm {
       embed_description,
       embed_html,
       thumbnail_url,
-      ap_id: page
+      ap_id: page.inner.id_unchecked().unwrap().to_string(),
-        .inner
-        .id(actor_id.domain().unwrap())?
-        .unwrap()
-        .to_string(),
       local: false,
     })
   }
@@ -267,12 +265,10 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/create/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut create = Create::new(creator.actor_id.to_owned(), page.into_any_base()?);
     create
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(CreateType::Create)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -300,12 +296,10 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/update/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut update = Update::new(creator.actor_id.to_owned(), page.into_any_base()?);
     update
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(UpdateType::Update)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -332,11 +326,10 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut delete = Delete::new(creator.actor_id.to_owned(), page.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -363,21 +356,18 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut delete = Delete::new(creator.actor_id.to_owned(), page.into_any_base()?);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), delete.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -404,11 +394,10 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut remove = Remove::new(mod_.actor_id.to_owned(), page.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -435,20 +424,18 @@ impl ApubObjectType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut remove = Remove::new(mod_.actor_id.to_owned(), page.into_any_base()?);
     remove
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(RemoveType::Remove)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
     // Undo that fake activity
-    let undo_id = format!("{}/undo/remove/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(mod_.actor_id.to_owned(), remove.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -478,12 +465,10 @@ impl ApubLikeableType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/like/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut like = Like::new(creator.actor_id.to_owned(), page.into_any_base()?);
     like
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(LikeType::Like)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -510,12 +495,10 @@ impl ApubLikeableType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/dislike/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut dislike = Dislike::new(creator.actor_id.to_owned(), page.into_any_base()?);
     dislike
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DislikeType::Dislike)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
@@ -542,22 +525,18 @@ impl ApubLikeableType for Post {
     let community_id = self.community_id;
     let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
 
-    let id = format!("{}/like/{}", self.ap_id, uuid::Uuid::new_v4());
-
     let mut like = Like::new(creator.actor_id.to_owned(), page.into_any_base()?);
     like
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(LikeType::Like)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/like/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), like.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(public())
       .set_many_ccs(vec![community.get_followers_url()]);
 

server/src/apub/private_message.rs

@@ -1,8 +1,8 @@
 use crate::{
   apub::{
-    activities::send_activity,
+    activities::{generate_activity_id, send_activity},
     create_tombstone,
-    fetcher::get_or_fetch_and_upsert_remote_user,
+    fetcher::get_or_fetch_and_upsert_user,
     insert_activity,
     ApubObjectType,
     FromApub,
@@ -13,7 +13,13 @@ use crate::{
   LemmyError,
 };
 use activitystreams_new::{
-  activity::{Create, Delete, Undo, Update},
+  activity::{
+    kind::{CreateType, DeleteType, UndoType, UpdateType},
+    Create,
+    Delete,
+    Undo,
+    Update,
+  },
   context,
   object::{kind::NoteType, Note, Tombstone},
   prelude::*,
@@ -56,12 +62,7 @@ impl ToApub for PrivateMessage {
   }
 
   fn to_tombstone(&self) -> Result<Tombstone, LemmyError> {
-    create_tombstone(
+    create_tombstone(self.deleted, &self.ap_id, self.updated, NoteType::Note)
-      self.deleted,
-      &self.ap_id,
-      self.updated,
-      NoteType::Note.to_string(),
-    )
   }
 }
 
@@ -74,7 +75,6 @@ impl FromApub for PrivateMessageForm {
     note: &Note,
     client: &Client,
     pool: &DbPool,
-    actor_id: &Url,
   ) -> Result<PrivateMessageForm, LemmyError> {
     let creator_actor_id = note
       .attributed_to()
@@ -83,11 +83,11 @@ impl FromApub for PrivateMessageForm {
       .single_xsd_any_uri()
       .unwrap();
 
-    let creator = get_or_fetch_and_upsert_remote_user(&creator_actor_id, client, pool).await?;
+    let creator = get_or_fetch_and_upsert_user(&creator_actor_id, client, pool).await?;
 
     let recipient_actor_id = note.to().unwrap().clone().single_xsd_any_uri().unwrap();
 
-    let recipient = get_or_fetch_and_upsert_remote_user(&recipient_actor_id, client, pool).await?;
+    let recipient = get_or_fetch_and_upsert_user(&recipient_actor_id, client, pool).await?;
 
     Ok(PrivateMessageForm {
       creator_id: creator.id,
@@ -102,7 +102,7 @@ impl FromApub for PrivateMessageForm {
       updated: note.updated().map(|u| u.to_owned().naive_local()),
       deleted: None,
       read: None,
-      ap_id: note.id(actor_id.domain().unwrap())?.unwrap().to_string(),
+      ap_id: note.id_unchecked().unwrap().to_string(),
       local: false,
     })
   }
@@ -118,7 +118,6 @@ impl ApubObjectType for PrivateMessage {
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
     let note = self.to_apub(pool).await?;
-    let id = format!("{}/create/{}", self.ap_id, uuid::Uuid::new_v4());
 
     let recipient_id = self.recipient_id;
     let recipient = blocking(pool, move |conn| User_::read(conn, recipient_id)).await??;
@@ -127,7 +126,7 @@ impl ApubObjectType for PrivateMessage {
     let to = format!("{}/inbox", recipient.actor_id);
     create
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(CreateType::Create)?)
       .set_to(to.clone());
 
     insert_activity(creator.id, create.clone(), true, pool).await?;
@@ -144,7 +143,6 @@ impl ApubObjectType for PrivateMessage {
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
     let note = self.to_apub(pool).await?;
-    let id = format!("{}/update/{}", self.ap_id, uuid::Uuid::new_v4());
 
     let recipient_id = self.recipient_id;
     let recipient = blocking(pool, move |conn| User_::read(conn, recipient_id)).await??;
@@ -153,7 +151,7 @@ impl ApubObjectType for PrivateMessage {
     let to = format!("{}/inbox", recipient.actor_id);
     update
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(UpdateType::Update)?)
       .set_to(to.clone());
 
     insert_activity(creator.id, update.clone(), true, pool).await?;
@@ -169,7 +167,6 @@ impl ApubObjectType for PrivateMessage {
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
     let note = self.to_apub(pool).await?;
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
 
     let recipient_id = self.recipient_id;
     let recipient = blocking(pool, move |conn| User_::read(conn, recipient_id)).await??;
@@ -178,7 +175,7 @@ impl ApubObjectType for PrivateMessage {
     let to = format!("{}/inbox", recipient.actor_id);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(to.clone());
 
     insert_activity(creator.id, delete.clone(), true, pool).await?;
@@ -194,7 +191,6 @@ impl ApubObjectType for PrivateMessage {
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
     let note = self.to_apub(pool).await?;
-    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
 
     let recipient_id = self.recipient_id;
     let recipient = blocking(pool, move |conn| User_::read(conn, recipient_id)).await??;
@@ -203,16 +199,14 @@ impl ApubObjectType for PrivateMessage {
     let to = format!("{}/inbox", recipient.actor_id);
     delete
       .set_context(context())
-      .set_id(Url::parse(&id)?)
+      .set_id(generate_activity_id(DeleteType::Delete)?)
       .set_to(to.clone());
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/delete/{}", self.ap_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(creator.actor_id.to_owned(), delete.into_any_base()?);
     undo
       .set_context(context())
-      .set_id(Url::parse(&undo_id)?)
+      .set_id(generate_activity_id(UndoType::Undo)?)
       .set_to(to.clone());
 
     insert_activity(creator.id, undo.clone(), true, pool).await?;

server/src/apub/user.rs

@@ -1,6 +1,6 @@
 use crate::{
   apub::{
-    activities::send_activity,
+    activities::{generate_activity_id, send_activity},
     create_apub_response,
     insert_activity,
     ActorType,
@@ -15,7 +15,11 @@ use crate::{
 };
 use activitystreams_ext::Ext1;
 use activitystreams_new::{
-  activity::{Follow, Undo},
+  activity::{
+    kind::{FollowType, UndoType},
+    Follow,
+    Undo,
+  },
   actor::{ApActor, Endpoints, Person},
   context,
   object::{Image, Tombstone},
@@ -102,9 +106,10 @@ impl ActorType for User_ {
     client: &Client,
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
-    let id = format!("{}/follow/{}", self.actor_id, uuid::Uuid::new_v4());
     let mut follow = Follow::new(self.actor_id.to_owned(), follow_actor_id);
-    follow.set_context(context()).set_id(id.parse()?);
+    follow
+      .set_context(context())
+      .set_id(generate_activity_id(FollowType::Follow)?);
     let to = format!("{}/inbox", follow_actor_id);
 
     insert_activity(self.id, follow.clone(), true, pool).await?;
@@ -119,17 +124,18 @@ impl ActorType for User_ {
     client: &Client,
     pool: &DbPool,
   ) -> Result<(), LemmyError> {
-    let id = format!("{}/follow/{}", self.actor_id, uuid::Uuid::new_v4());
     let mut follow = Follow::new(self.actor_id.to_owned(), follow_actor_id);
-    follow.set_context(context()).set_id(id.parse()?);
+    follow
+      .set_context(context())
+      .set_id(generate_activity_id(FollowType::Follow)?);
 
     let to = format!("{}/inbox", follow_actor_id);
 
-    // TODO
     // Undo that fake activity
-    let undo_id = format!("{}/undo/follow/{}", self.actor_id, uuid::Uuid::new_v4());
     let mut undo = Undo::new(Url::parse(&self.actor_id)?, follow.into_any_base()?);
-    undo.set_context(context()).set_id(undo_id.parse()?);
+    undo
+      .set_context(context())
+      .set_id(generate_activity_id(UndoType::Undo)?);
 
     insert_activity(self.id, undo.clone(), true, pool).await?;
 
@@ -185,18 +191,17 @@ impl ActorType for User_ {
   async fn get_follower_inboxes(&self, _pool: &DbPool) -> Result<Vec<String>, LemmyError> {
     unimplemented!()
   }
+
+  fn user_id(&self) -> i32 {
+    self.id
+  }
 }
 
 #[async_trait::async_trait(?Send)]
 impl FromApub for UserForm {
   type ApubType = PersonExt;
   /// Parse an ActivityPub person received from another instance into a Lemmy user.
-  async fn from_apub(
+  async fn from_apub(person: &PersonExt, _: &Client, _: &DbPool) -> Result<Self, LemmyError> {
-    person: &PersonExt,
-    _: &Client,
-    _: &DbPool,
-    actor_id: &Url,
-  ) -> Result<Self, LemmyError> {
     let avatar = match person.icon() {
       Some(any_image) => Image::from_any_base(any_image.as_one().unwrap().clone())
         .unwrap()
@@ -232,7 +237,7 @@ impl FromApub for UserForm {
       show_avatars: false,
       send_notifications_to_email: false,
       matrix_user_id: None,
-      actor_id: person.id(actor_id.domain().unwrap())?.unwrap().to_string(),
+      actor_id: person.id_unchecked().unwrap().to_string(),
       bio: person
         .inner
         .summary()

server/src/routes/federation.rs

@@ -1,11 +1,9 @@
 use crate::apub::{
   comment::get_apub_comment,
   community::*,
-  community_inbox::community_inbox,
+  inbox::{community_inbox::community_inbox, shared_inbox::shared_inbox, user_inbox::user_inbox},
   post::get_apub_post,
-  shared_inbox::shared_inbox,
   user::*,
-  user_inbox::user_inbox,
   APUB_JSON_CONTENT_TYPE,
 };
 use actix_web::*;

ui/src/api_tests/api.spec.ts

@@ -618,6 +618,37 @@ describe('main', () => {
     });
   });
 
+  describe('federated comment like', () => {
+    test('/u/lemmy_beta likes a comment from /u/lemmy_alpha, the like is on both instances', async () => {
+      // Do a like, to test it (its also been unliked, so its at 0)
+      let likeCommentForm: CommentLikeForm = {
+        comment_id: 1,
+        score: 1,
+        auth: lemmyBetaAuth,
+      };
+
+      let likeCommentRes: CommentResponse = await fetch(
+        `${lemmyBetaApiUrl}/comment/like`,
+        {
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: wrapper(likeCommentForm),
+        }
+      ).then(d => d.json());
+
+      expect(likeCommentRes.comment.score).toBe(1);
+
+      let getPostUrl = `${lemmyAlphaApiUrl}/post?id=2`;
+      let getPostRes: GetPostResponse = await fetch(getPostUrl, {
+        method: 'GET',
+      }).then(d => d.json());
+
+      expect(getPostRes.comments[2].score).toBe(1);
+    });
+  });
+
   describe('delete things', () => {
     test('/u/lemmy_beta deletes and undeletes a federated comment, post, and community, lemmy_alpha sees its deleted.', async () => {
       // Create a test community