Refactor inbox, simplify and split into multiple files

server/src/api/comment.rs

@@ -176,7 +176,7 @@ 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(mentions, updated_comment.clone(), user.clone(), post, pool).await?;
+      send_local_notifs(mentions, updated_comment.clone(), &user, post, pool).await?;
 
     // You like your own comment by default
     let like_form = CommentLikeForm {
@@ -407,7 +407,7 @@ impl Perform for Oper<EditComment> {
     let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;
 
     let mentions = scrape_text_for_mentions(&comment_form.content);
-    let recipient_ids = send_local_notifs(mentions, updated_comment, user, post, pool).await?;
+    let recipient_ids = send_local_notifs(mentions, updated_comment, &user, post, pool).await?;
 
     let edit_id = data.edit_id;
     let comment_view = blocking(pool, move |conn| {
@@ -672,12 +672,13 @@ impl Perform for Oper<GetComments> {
 pub async fn send_local_notifs(
   mentions: Vec<MentionData>,
   comment: Comment,
-  user: User_,
+  user: &User_,
   post: Post,
   pool: &DbPool,
 ) -> 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_local_notifs(conn, &mentions, &comment, &user2, &post)
   })
   .await?;
 

server/src/apub/community.rs

@@ -318,6 +318,10 @@ impl ActorType for Community {
   ) -> Result<(), LemmyError> {
     unimplemented!()
   }
+
+  fn user_id(&self) -> i32 {
+    self.creator_id
+  }
 }
 
 #[async_trait::async_trait(?Send)]
@@ -427,10 +431,10 @@ 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
@@ -450,5 +454,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},
@@ -192,6 +200,19 @@ pub async fn search_by_apub_id(
   Ok(response)
 }
 
+pub async fn get_or_fetch_and_upsert_remote_actor(
+  apub_id: &Url,
+  client: &Client,
+  pool: &DbPool,
+) -> Result<Box<dyn ActorType>, LemmyError> {
+  let user = get_or_fetch_and_upsert_remote_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_remote_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(
   apub_id: &Url,

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

@@ -0,0 +1,56 @@
+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,
+    },
+    inbox::shared_inbox::receive_unhandled_activity,
+  },
+  routes::ChatServerParam,
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::*, prelude::ExtendsExt};
+use actix_web::{client::Client, HttpResponse};
+use activitystreams_new::base::AnyBase;
+
+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,124 @@
+use crate::{
+  api::{
+    comment::{send_local_notifs, CommentResponse},
+    post::PostResponse,
+  },
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    ActorType,
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Create, 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;
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::{announce_if_community_is_local};
+
+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, &user.actor_id()?).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, &user.actor_id()?).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).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,
+  };
+
+  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,222 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::inbox::
+  shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Delete, 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,
+};
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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, &user.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_remote_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, &user.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_remote_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,
+  };
+
+  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, &user.actor_id()?)
+    .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,132 @@
+use crate::{
+  api::{comment::CommentResponse, post::PostResponse},
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Dislike, 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,
+};
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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, &user.actor_id()?).await?;
+
+  let post_id = get_or_fetch_and_insert_remote_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, &user.actor_id()?).await?;
+
+  let comment_id = get_or_fetch_and_insert_remote_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,
+  };
+
+  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,132 @@
+use crate::{
+  api::{comment::CommentResponse, post::PostResponse},
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Like, 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,
+};
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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, &user.actor_id()?).await?;
+
+  let post_id = get_or_fetch_and_insert_remote_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, &user.actor_id()?).await?;
+
+  let comment_id = get_or_fetch_and_insert_remote_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,
+  };
+
+  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,221 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::Remove, 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,
+};
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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, &mod_.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_remote_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, &mod_.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_remote_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,
+  };
+
+  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, &mod_.actor_id()?)
+    .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,550 @@
+use crate::{
+  api::{comment::CommentResponse, community::CommunityResponse, post::PostResponse},
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    GroupExt,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendCommunityRoomMessage, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{activity::*, 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,
+};
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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();
+  match type_ {
+    // TODO: handle dislike
+    d => Err(format_err!("Undo Delete type {} not supported", d).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, &user.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_remote_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,
+  };
+
+  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, &mod_.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let comment = get_or_fetch_and_insert_remote_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,
+  };
+
+  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, &user.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_remote_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, &mod_.actor_id()?)
+    .await?
+    .get_ap_id()?;
+
+  let post = get_or_fetch_and_insert_remote_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, &user.actor_id()?)
+    .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, &mod_.actor_id()?)
+    .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, &user.actor_id()?).await?;
+
+  let comment_id = get_or_fetch_and_insert_remote_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,
+  };
+
+  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, &user.actor_id()?).await?;
+
+  let post_id = get_or_fetch_and_insert_remote_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,128 @@
+use crate::{
+  api::{
+    comment::{send_local_notifs, CommentResponse},
+    post::PostResponse,
+  },
+  apub::inbox::shared_inbox::{get_user_from_activity, receive_unhandled_activity},
+  apub::{
+    fetcher::{get_or_fetch_and_insert_remote_comment, get_or_fetch_and_insert_remote_post},
+    ActorType,
+    FromApub,
+    PageExt,
+  },
+  blocking,
+  routes::ChatServerParam,
+  websocket::{
+    server::{SendComment, SendPost},
+    UserOperation,
+  },
+  DbPool,
+  LemmyError,
+};
+use activitystreams_new::{
+  activity::{Update},
+  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;
+use activitystreams_new::base::AnyBase;
+use crate::apub::inbox::shared_inbox::announce_if_community_is_local;
+
+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, &user.actor_id()?).await?;
+
+  let post_id = get_or_fetch_and_insert_remote_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, &user.actor_id()?).await?;
+
+  let comment_id = get_or_fetch_and_insert_remote_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).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,
+  };
+
+  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/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,136 @@
+use crate::{
+  apub::{
+    extensions::signatures::verify,
+    fetcher::{
+      get_or_fetch_and_upsert_remote_actor,
+      get_or_fetch_and_upsert_remote_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},
+  prelude::*,
+};
+use actix_web::{client::Client, web, HttpRequest, HttpResponse};
+use lemmy_db::{user::User_};
+use log::debug;
+use std::fmt::Debug;
+use crate::apub::fetcher::get_or_fetch_and_upsert_remote_community;
+use activitystreams_new::object::AsObject;
+use crate::apub::community::do_announce;
+use activitystreams_new::base::Extends;
+use serde::Serialize;
+
+#[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_remote_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_remote_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_uri = cc.first().unwrap().as_xsd_any_uri().unwrap();
+  let community = get_or_fetch_and_upsert_remote_community(&community_uri, client, pool).await?;
+
+  if community.local {
+    do_announce(activity.into_any_base()?, &community, &user, client, pool).await?;
+  }
+  Ok(())
+}

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::{
@@ -219,6 +217,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/user.rs

@@ -186,6 +186,10 @@ 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)]

server/src/routes/federation.rs

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