finish this

lemmy_apub/src/activities/receive/comment.rs

@@ -1,6 +1,7 @@
 use crate::{
   activities::receive::get_actor_as_user,
   fetcher::get_or_fetch_and_insert_comment,
+  objects::FromApub,
   ActorType,
   NoteExt,
 };
@@ -10,16 +11,15 @@ use activitystreams::{
 };
 use anyhow::{anyhow, Context};
 use lemmy_db::{
-  comment::{Comment, CommentForm, CommentLike, CommentLikeForm},
+  comment::{Comment, CommentLike, CommentLikeForm},
   comment_view::CommentView,
   post::Post,
-  Crud,
   Likeable,
 };
 use lemmy_structs::{blocking, comment::CommentResponse, send_local_notifs};
 use lemmy_utils::{location_info, utils::scrape_text_for_mentions, LemmyError};
 use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperation};
-use crate::objects::FromApub;
+use url::Url;
 
 pub(crate) async fn receive_create_comment(
   create: Create,
@@ -30,8 +30,8 @@ pub(crate) async fn receive_create_comment(
   let note = NoteExt::from_any_base(create.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let comment =
+  // TODO: need to do the check for locked post before calling this
-    CommentForm::from_apub(&note, context, Some(user.actor_id()?), request_counter).await?;
+  let comment = Comment::from_apub(&note, context, Some(user.actor_id()?), request_counter).await?;
 
   let post_id = comment.post_id;
   let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
@@ -39,27 +39,17 @@ pub(crate) async fn receive_create_comment(
     return Err(anyhow!("Post is locked").into());
   }
 
-  let inserted_comment =
-    blocking(context.pool(), move |conn| Comment::upsert(conn, &comment)).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 mentions = scrape_text_for_mentions(&comment.content);
-  let recipient_ids = send_local_notifs(
+  let recipient_ids =
-    mentions,
+    send_local_notifs(mentions, comment.clone(), &user, post, context.pool(), true).await?;
-    inserted_comment.clone(),
-    &user,
-    post,
-    context.pool(),
-    true,
-  )
-  .await?;
 
   // Refetch the view
   let comment_view = blocking(context.pool(), move |conn| {
-    CommentView::read(conn, inserted_comment.id, None)
+    CommentView::read(conn, comment.id, None)
   })
   .await??;
 
@@ -87,32 +77,20 @@ pub(crate) async fn receive_update_comment(
     .context(location_info!())?;
   let user = get_actor_as_user(&update, context, request_counter).await?;
 
-  let comment =
+  let comment = Comment::from_apub(&note, context, Some(user.actor_id()?), request_counter).await?;
-    CommentForm::from_apub(&note, context, Some(user.actor_id()?), request_counter).await?;
 
+  // TODO: why fetch?
   let original_comment_id =
-    get_or_fetch_and_insert_comment(&comment.get_ap_id()?, context, request_counter)
+    get_or_fetch_and_insert_comment(&Url::parse(&comment.ap_id)?, context, request_counter)
       .await?
       .id;
 
-  let updated_comment = blocking(context.pool(), move |conn| {
+  let post_id = comment.post_id;
-    Comment::update(conn, original_comment_id, &comment)
-  })
-  .await??;
-
-  let post_id = updated_comment.post_id;
   let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
 
-  let mentions = scrape_text_for_mentions(&updated_comment.content);
+  let mentions = scrape_text_for_mentions(&comment.content);
-  let recipient_ids = send_local_notifs(
+  let recipient_ids =
-    mentions,
+    send_local_notifs(mentions, comment, &user, post, context.pool(), false).await?;
-    updated_comment,
-    &user,
-    post,
-    context.pool(),
-    false,
-  )
-  .await?;
 
   // Refetch the view
   let comment_view = blocking(context.pool(), move |conn| {
@@ -144,9 +122,11 @@ pub(crate) async fn receive_like_comment(
     .context(location_info!())?;
   let user = get_actor_as_user(&like, context, request_counter).await?;
 
-  let comment = CommentForm::from_apub(&note, context, None, request_counter).await?;
+  let comment = Comment::from_apub(&note, context, None, request_counter).await?;
 
-  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, context, request_counter)
+  // TODO: why do we need to fetch here if we already have the comment?
+  let comment_id =
+    get_or_fetch_and_insert_comment(&Url::parse(&comment.ap_id)?, context, request_counter)
       .await?
       .id;
 
@@ -201,9 +181,11 @@ pub(crate) async fn receive_dislike_comment(
   .context(location_info!())?;
   let user = get_actor_as_user(&dislike, context, request_counter).await?;
 
-  let comment = CommentForm::from_apub(&note, context, None, request_counter).await?;
+  let comment = Comment::from_apub(&note, context, None, request_counter).await?;
 
-  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, context, request_counter)
+  // TODO: same as above, why fetch here?
+  let comment_id =
+    get_or_fetch_and_insert_comment(&Url::parse(&comment.ap_id)?, context, request_counter)
       .await?
       .id;
 

lemmy_apub/src/activities/receive/comment_undo.rs

@@ -1,19 +1,20 @@
 use crate::{
   activities::receive::get_actor_as_user,
   fetcher::get_or_fetch_and_insert_comment,
+  objects::FromApub,
   NoteExt,
 };
-use crate::objects::FromApub;
 use activitystreams::{activity::*, prelude::*};
 use anyhow::Context;
 use lemmy_db::{
-  comment::{Comment, CommentForm, CommentLike},
+  comment::{Comment, CommentLike},
   comment_view::CommentView,
   Likeable,
 };
 use lemmy_structs::{blocking, comment::CommentResponse};
 use lemmy_utils::{location_info, LemmyError};
 use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperation};
+use url::Url;
 
 pub(crate) async fn receive_undo_like_comment(
   like: &Like,
@@ -24,9 +25,11 @@ pub(crate) async fn receive_undo_like_comment(
   let note = NoteExt::from_any_base(like.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let comment = CommentForm::from_apub(&note, context, None, request_counter).await?;
+  let comment = Comment::from_apub(&note, context, None, request_counter).await?;
 
-  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, context, request_counter)
+  // TODO: why?
+  let comment_id =
+    get_or_fetch_and_insert_comment(&Url::parse(&comment.ap_id)?, context, request_counter)
       .await?
       .id;
 
@@ -74,9 +77,11 @@ pub(crate) async fn receive_undo_dislike_comment(
   )?
   .context(location_info!())?;
 
-  let comment = CommentForm::from_apub(&note, context, None, request_counter).await?;
+  let comment = Comment::from_apub(&note, context, None, request_counter).await?;
 
-  let comment_id = get_or_fetch_and_insert_comment(&comment.get_ap_id()?, context, request_counter)
+  // TODO
+  let comment_id =
+    get_or_fetch_and_insert_comment(&Url::parse(&comment.ap_id)?, context, request_counter)
       .await?
       .id;
 

lemmy_apub/src/activities/receive/post.rs

@@ -1,24 +1,24 @@
 use crate::{
   activities::receive::get_actor_as_user,
   fetcher::get_or_fetch_and_insert_post,
+  objects::FromApub,
   ActorType,
   PageExt,
 };
-use crate::objects::FromApub;
 use activitystreams::{
   activity::{Create, Dislike, Like, Remove, Update},
   prelude::*,
 };
 use anyhow::Context;
 use lemmy_db::{
-  post::{Post, PostForm, PostLike, PostLikeForm},
+  post::{Post, PostLike, PostLikeForm},
   post_view::PostView,
-  Crud,
   Likeable,
 };
 use lemmy_structs::{blocking, post::PostResponse};
 use lemmy_utils::{location_info, LemmyError};
 use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperation};
+use url::Url;
 
 pub(crate) async fn receive_create_post(
   create: Create,
@@ -29,16 +29,12 @@ pub(crate) async fn receive_create_post(
   let page = PageExt::from_any_base(create.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, Some(user.actor_id()?), request_counter).await?;
+  let post = Post::from_apub(&page, context, Some(user.actor_id()?), request_counter).await?;
-
-  // Using an upsert, since likes (which fetch the post), sometimes come in before the create
-  // resulting in double posts.
-  let inserted_post = blocking(context.pool(), move |conn| Post::upsert(conn, &post)).await??;
 
   // Refetch the view
-  let inserted_post_id = inserted_post.id;
+  let post_id = post.id;
   let post_view = blocking(context.pool(), move |conn| {
-    PostView::read(conn, inserted_post_id, None)
+    PostView::read(conn, post_id, None)
   })
   .await??;
 
@@ -62,17 +58,14 @@ pub(crate) async fn receive_update_post(
   let page = PageExt::from_any_base(update.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, Some(user.actor_id()?), request_counter).await?;
+  let post = Post::from_apub(&page, context, Some(user.actor_id()?), request_counter).await?;
 
-  let original_post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, context, request_counter)
+  // TODO: why?
+  let original_post_id =
+    get_or_fetch_and_insert_post(&Url::parse(&post.ap_id)?, context, request_counter)
       .await?
       .id;
 
-  blocking(context.pool(), move |conn| {
-    Post::update(conn, original_post_id, &post)
-  })
-  .await??;
-
   // Refetch the view
   let post_view = blocking(context.pool(), move |conn| {
     PostView::read(conn, original_post_id, None)
@@ -99,9 +92,10 @@ pub(crate) async fn receive_like_post(
   let page = PageExt::from_any_base(like.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, None, request_counter).await?;
+  let post = Post::from_apub(&page, context, None, request_counter).await?;
 
-  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, context, request_counter)
+  // TODO: why?
+  let post_id = get_or_fetch_and_insert_post(&Url::parse(&post.ap_id)?, context, request_counter)
     .await?
     .id;
 
@@ -149,9 +143,9 @@ pub(crate) async fn receive_dislike_post(
   )?
   .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, None, request_counter).await?;
+  let post = Post::from_apub(&page, context, None, request_counter).await?;
 
-  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, context, request_counter)
+  let post_id = get_or_fetch_and_insert_post(&Url::parse(&post.ap_id)?, context, request_counter)
     .await?
     .id;
 

lemmy_apub/src/activities/receive/post_undo.rs

@@ -1,19 +1,20 @@
 use crate::{
   activities::receive::get_actor_as_user,
   fetcher::get_or_fetch_and_insert_post,
+  objects::FromApub,
   PageExt,
 };
-use crate::objects::FromApub;
 use activitystreams::{activity::*, prelude::*};
 use anyhow::Context;
 use lemmy_db::{
-  post::{Post, PostForm, PostLike},
+  post::{Post, PostLike},
   post_view::PostView,
   Likeable,
 };
 use lemmy_structs::{blocking, post::PostResponse};
 use lemmy_utils::{location_info, LemmyError};
 use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperation};
+use url::Url;
 
 pub(crate) async fn receive_undo_like_post(
   like: &Like,
@@ -24,9 +25,10 @@ pub(crate) async fn receive_undo_like_post(
   let page = PageExt::from_any_base(like.object().to_owned().one().context(location_info!())?)?
     .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, None, request_counter).await?;
+  let post = Post::from_apub(&page, context, None, request_counter).await?;
 
-  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, context, request_counter)
+  // TODO: why?
+  let post_id = get_or_fetch_and_insert_post(&Url::parse(&post.ap_id)?, context, request_counter)
     .await?
     .id;
 
@@ -68,9 +70,10 @@ pub(crate) async fn receive_undo_dislike_post(
   )?
   .context(location_info!())?;
 
-  let post = PostForm::from_apub(&page, context, None, request_counter).await?;
+  let post = Post::from_apub(&page, context, None, request_counter).await?;
 
-  let post_id = get_or_fetch_and_insert_post(&post.get_ap_id()?, context, request_counter)
+  // TODO: why?
+  let post_id = get_or_fetch_and_insert_post(&Url::parse(&post.ap_id)?, context, request_counter)
     .await?
     .id;
 

lemmy_apub/src/activities/receive/private_message.rs

@@ -3,9 +3,9 @@ use crate::{
   check_is_apub_id_valid,
   fetcher::get_or_fetch_and_upsert_user,
   inbox::get_activity_to_and_cc,
+  objects::FromApub,
   NoteExt,
 };
-use crate::objects::FromApub;
 use activitystreams::{
   activity::{ActorAndObjectRefExt, Create, Delete, Undo, Update},
   base::{AsBase, ExtendsExt},
@@ -13,11 +13,7 @@ use activitystreams::{
   public,
 };
 use anyhow::{anyhow, Context};
-use lemmy_db::{
+use lemmy_db::{private_message::PrivateMessage, private_message_view::PrivateMessageView};
-  private_message::{PrivateMessage, PrivateMessageForm},
-  private_message_view::PrivateMessageView,
-  Crud,
-};
 use lemmy_structs::{blocking, user::PrivateMessageResponse};
 use lemmy_utils::{location_info, LemmyError};
 use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperation};
@@ -41,15 +37,10 @@ pub(crate) async fn receive_create_private_message(
   .context(location_info!())?;
 
   let private_message =
-    PrivateMessageForm::from_apub(&note, context, Some(expected_domain), request_counter).await?;
+    PrivateMessage::from_apub(&note, context, Some(expected_domain), request_counter).await?;
-
-  let inserted_private_message = blocking(&context.pool(), move |conn| {
-    PrivateMessage::create(conn, &private_message)
-  })
-  .await??;
 
   let message = blocking(&context.pool(), move |conn| {
-    PrivateMessageView::read(conn, inserted_private_message.id)
+    PrivateMessageView::read(conn, private_message.id)
   })
   .await??;
 
@@ -82,24 +73,8 @@ pub(crate) async fn receive_update_private_message(
     .to_owned();
   let note = NoteExt::from_any_base(object)?.context(location_info!())?;
 
-  let private_message_form =
+  let private_message =
-    PrivateMessageForm::from_apub(&note, context, Some(expected_domain), request_counter).await?;
+    PrivateMessage::from_apub(&note, context, Some(expected_domain), request_counter).await?;
-
-  let private_message_ap_id = private_message_form
-    .ap_id
-    .as_ref()
-    .context(location_info!())?
-    .clone();
-  let private_message = blocking(&context.pool(), move |conn| {
-    PrivateMessage::read_from_apub_id(conn, &private_message_ap_id)
-  })
-  .await??;
-
-  let private_message_id = private_message.id;
-  blocking(&context.pool(), move |conn| {
-    PrivateMessage::update(conn, private_message_id, &private_message_form)
-  })
-  .await??;
 
   let private_message_id = private_message.id;
   let message = blocking(&context.pool(), move |conn| {

lemmy_apub/src/activities/send/comment.rs

@@ -3,6 +3,7 @@ use crate::{
   activity_queue::{send_comment_mentions, send_to_community},
   extensions::context::lemmy_context,
   fetcher::get_or_fetch_and_upsert_user,
+  objects::ToApub,
   ActorType,
   ApubLikeableType,
   ApubObjectType,
@@ -38,7 +39,6 @@ use log::debug;
 use reqwest::Client;
 use serde_json::Error;
 use url::Url;
-use crate::objects::ToApub;
 
 #[async_trait::async_trait(?Send)]
 impl ApubObjectType for Comment {

lemmy_apub/src/activities/send/post.rs

@@ -2,11 +2,11 @@ use crate::{
   activities::send::generate_activity_id,
   activity_queue::send_to_community,
   extensions::context::lemmy_context,
+  objects::ToApub,
   ActorType,
   ApubLikeableType,
   ApubObjectType,
 };
-use crate::objects::ToApub;
 use activitystreams::{
   activity::{
     kind::{CreateType, DeleteType, DislikeType, LikeType, RemoveType, UndoType, UpdateType},

lemmy_apub/src/activities/send/private_message.rs

@@ -2,10 +2,10 @@ use crate::{
   activities::send::generate_activity_id,
   activity_queue::send_activity_single_dest,
   extensions::context::lemmy_context,
+  objects::ToApub,
   ActorType,
   ApubObjectType,
 };
-use crate::objects::ToApub;
 use activitystreams::{
   activity::{
     kind::{CreateType, DeleteType, UndoType, UpdateType},

lemmy_apub/src/fetcher.rs

@@ -1,5 +1,6 @@
 use crate::{
   check_is_apub_id_valid,
+  objects::FromApub,
   ActorType,
   GroupExt,
   NoteExt,
@@ -12,16 +13,15 @@ use anyhow::{anyhow, Context};
 use chrono::NaiveDateTime;
 use diesel::result::Error::NotFound;
 use lemmy_db::{
-  comment::{Comment, CommentForm},
+  comment::Comment,
   comment_view::CommentView,
-  community::{Community, CommunityForm, CommunityModerator, CommunityModeratorForm},
+  community::{Community, CommunityModerator, CommunityModeratorForm},
   community_view::CommunityView,
   naive_now,
-  post::{Post, PostForm},
+  post::Post,
   post_view::PostView,
-  user::{UserForm, User_},
+  user::User_,
   user_view::UserView,
-  Crud,
   Joinable,
   SearchType,
 };
@@ -38,7 +38,6 @@ use reqwest::Client;
 use serde::Deserialize;
 use std::{fmt::Debug, time::Duration};
 use url::Url;
-use crate::objects::FromApub;
 
 static ACTOR_REFETCH_INTERVAL_SECONDS: i64 = 24 * 60 * 60;
 static ACTOR_REFETCH_INTERVAL_SECONDS_DEBUG: i64 = 10;
@@ -184,22 +183,16 @@ pub async fn search_by_apub_id(
       response
     }
     SearchAcceptedObjects::Page(p) => {
-      let post_form = PostForm::from_apub(&p, context, Some(query_url), recursion_counter).await?;
+      let p = Post::from_apub(&p, context, Some(query_url), recursion_counter).await?;
 
-      let p = blocking(context.pool(), move |conn| Post::upsert(conn, &post_form)).await??;
       response.posts =
         vec![blocking(context.pool(), move |conn| PostView::read(conn, p.id, None)).await??];
 
       response
     }
     SearchAcceptedObjects::Comment(c) => {
-      let comment_form =
+      let c = Comment::from_apub(&c, context, Some(query_url), recursion_counter).await?;
-        CommentForm::from_apub(&c, context, Some(query_url), recursion_counter).await?;
 
-      let c = blocking(context.pool(), move |conn| {
-        Comment::upsert(conn, &comment_form)
-      })
-      .await??;
       response.comments = vec![
         blocking(context.pool(), move |conn| {
           CommentView::read(conn, c.id, None)
@@ -258,15 +251,15 @@ pub(crate) async fn get_or_fetch_and_upsert_user(
         return Ok(u);
       }
 
-      let mut uf = UserForm::from_apub(
+      let user = User_::from_apub(
         &person?,
         context,
         Some(apub_id.to_owned()),
         recursion_counter,
       )
       .await?;
-      uf.last_refreshed_at = Some(naive_now());
+      // TODO: do we need to set this? would need a separate db call
-      let user = blocking(context.pool(), move |conn| User_::update(conn, u.id, &uf)).await??;
+      //uf.last_refreshed_at = Some(naive_now());
 
       Ok(user)
     }
@@ -276,14 +269,13 @@ pub(crate) async fn get_or_fetch_and_upsert_user(
       let person =
         fetch_remote_object::<PersonExt>(context.client(), apub_id, recursion_counter).await?;
 
-      let uf = UserForm::from_apub(
+      let user = User_::from_apub(
         &person,
         context,
         Some(apub_id.to_owned()),
         recursion_counter,
       )
       .await?;
-      let user = blocking(context.pool(), move |conn| User_::upsert(conn, &uf)).await??;
 
       Ok(user)
     }
@@ -354,9 +346,8 @@ async fn fetch_remote_community(
   }
 
   let group = group?;
-  let cf =
+  let community =
-    CommunityForm::from_apub(&group, context, Some(apub_id.to_owned()), recursion_counter).await?;
+    Community::from_apub(&group, context, Some(apub_id.to_owned()), recursion_counter).await?;
-  let community = blocking(context.pool(), move |conn| Community::upsert(conn, &cf)).await??;
 
   // Also add the community moderators too
   let attributed_to = group.inner.attributed_to().context(location_info!())?;
@@ -409,20 +400,10 @@ async fn fetch_remote_community(
 
     // The post creator may be from a blocked instance,
     // if it errors, then continue
-    let post = match PostForm::from_apub(&page, context, None, recursion_counter).await {
+    match Post::from_apub(&page, context, None, recursion_counter).await {
       Ok(post) => post,
       Err(_) => continue,
     };
-    let post_ap_id = post.ap_id.as_ref().context(location_info!())?.clone();
-    // Check whether the post already exists in the local db
-    let existing = blocking(context.pool(), move |conn| {
-      Post::read_from_apub_id(conn, &post_ap_id)
-    })
-    .await?;
-    match existing {
-      Ok(e) => blocking(context.pool(), move |conn| Post::update(conn, e.id, &post)).await??,
-      Err(_) => blocking(context.pool(), move |conn| Post::upsert(conn, &post)).await??,
-    };
     // TODO: we need to send a websocket update here
   }
 
@@ -448,18 +429,16 @@ pub(crate) async fn get_or_fetch_and_insert_post(
     Ok(p) => Ok(p),
     Err(NotFound {}) => {
       debug!("Fetching and creating remote post: {}", post_ap_id);
-      let post =
+      let page =
         fetch_remote_object::<PageExt>(context.client(), post_ap_id, recursion_counter).await?;
-      let post_form = PostForm::from_apub(
+      let post = Post::from_apub(
-        &post,
+        &page,
         context,
         Some(post_ap_id.to_owned()),
         recursion_counter,
       )
       .await?;
 
-      let post = blocking(context.pool(), move |conn| Post::upsert(conn, &post_form)).await??;
-
       Ok(post)
     }
     Err(e) => Err(e.into()),
@@ -490,7 +469,7 @@ pub(crate) async fn get_or_fetch_and_insert_comment(
       );
       let comment =
         fetch_remote_object::<NoteExt>(context.client(), comment_ap_id, recursion_counter).await?;
-      let comment_form = CommentForm::from_apub(
+      let comment = Comment::from_apub(
         &comment,
         context,
         Some(comment_ap_id.to_owned()),
@@ -498,17 +477,12 @@ pub(crate) async fn get_or_fetch_and_insert_comment(
       )
       .await?;
 
-      let post_id = comment_form.post_id;
+      let post_id = comment.post_id;
       let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
       if post.locked {
         return Err(anyhow!("Post is locked").into());
       }
 
-      let comment = blocking(context.pool(), move |conn| {
-        Comment::upsert(conn, &comment_form)
-      })
-      .await??;
-
       Ok(comment)
     }
     Err(e) => Err(e.into()),

lemmy_apub/src/http/comment.rs

@@ -1,5 +1,6 @@
 use crate::{
   http::{create_apub_response, create_apub_tombstone_response},
+  objects::ToApub,
 };
 use actix_web::{body::Body, web, web::Path, HttpResponse};
 use diesel::result::Error::NotFound;
@@ -8,7 +9,6 @@ use lemmy_structs::blocking;
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
 use serde::Deserialize;
-use crate::objects::ToApub;
 
 #[derive(Deserialize)]
 pub struct CommentQuery {

lemmy_apub/src/http/community.rs

@@ -1,6 +1,7 @@
 use crate::{
   extensions::context::lemmy_context,
   http::{create_apub_response, create_apub_tombstone_response},
+  objects::ToApub,
   ActorType,
 };
 use activitystreams::{
@@ -13,7 +14,6 @@ use lemmy_structs::blocking;
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
 use serde::Deserialize;
-use crate::objects::ToApub;
 
 #[derive(Deserialize)]
 pub struct CommunityQuery {

lemmy_apub/src/http/post.rs

@@ -1,5 +1,6 @@
 use crate::{
   http::{create_apub_response, create_apub_tombstone_response},
+  objects::ToApub,
 };
 use actix_web::{body::Body, web, HttpResponse};
 use diesel::result::Error::NotFound;
@@ -8,7 +9,6 @@ use lemmy_structs::blocking;
 use lemmy_utils::LemmyError;
 use lemmy_websocket::LemmyContext;
 use serde::Deserialize;
-use crate::objects::ToApub;
 
 #[derive(Deserialize)]
 pub struct PostQuery {

lemmy_apub/src/http/user.rs

@@ -1,9 +1,13 @@
-use crate::{extensions::context::lemmy_context, http::create_apub_response, ActorType};
+use crate::{
+  extensions::context::lemmy_context,
+  http::create_apub_response,
+  objects::ToApub,
+  ActorType,
+};
 use activitystreams::{
   base::BaseExt,
   collection::{CollectionExt, OrderedCollection},
 };
-use crate::objects::ToApub;
 use actix_web::{body::Body, web, HttpResponse};
 use lemmy_db::user::User_;
 use lemmy_structs::blocking;

lemmy_apub/src/objects/comment.rs

@@ -10,6 +10,9 @@ use crate::{
     create_tombstone,
     get_source_markdown_value,
     set_content_and_source,
+    FromApub,
+    FromApubToForm,
+    ToApub,
   },
   NoteExt,
 };
@@ -29,13 +32,12 @@ use lemmy_db::{
 use lemmy_structs::blocking;
 use lemmy_utils::{
   location_info,
+  settings::Settings,
   utils::{convert_datetime, remove_slurs},
   LemmyError,
 };
 use lemmy_websocket::LemmyContext;
 use url::Url;
-use crate::objects::{FromApub, ToApub, FromApubToForm};
-use lemmy_utils::settings::Settings;
 
 #[async_trait::async_trait(?Send)]
 impl ToApub for Comment {
@@ -99,7 +101,9 @@ impl FromApub for Comment {
     expected_domain: Option<Url>,
     request_counter: &mut i32,
   ) -> Result<Comment, LemmyError> {
-    let comment_id = note.id_unchecked().context(location_info!())?;
+    // TODO: we should move read_from_apub_id() and upsert() into traits so we can make a generic
+    //       function to handle all this (shared with User_::from_apub etc)
+    let comment_id = note.id_unchecked().context(location_info!())?.to_owned();
     let domain = comment_id.domain().context(location_info!())?;
     if domain == Settings::get().hostname {
       let comment = blocking(context.pool(), move |conn| {
@@ -108,9 +112,10 @@ impl FromApub for Comment {
       .await??;
       Ok(comment)
     } else {
-      let comment_form = CommentForm::from_apub(note, context, expected_domain, request_counter)?;
+      let comment_form =
+        CommentForm::from_apub(note, context, expected_domain, request_counter).await?;
       let comment = blocking(context.pool(), move |conn| {
-        Comment::upsert(conn, comment_form)
+        Comment::upsert(conn, &comment_form)
       })
       .await??;
       Ok(comment)

lemmy_apub/src/objects/community.rs

@@ -6,6 +6,9 @@ use crate::{
     create_tombstone,
     get_source_markdown_value,
     set_content_and_source,
+    FromApub,
+    FromApubToForm,
+    ToApub,
   },
   ActorType,
   GroupExt,
@@ -27,10 +30,10 @@ use lemmy_db::{
 use lemmy_structs::blocking;
 use lemmy_utils::{
   location_info,
+  settings::Settings,
   utils::{check_slurs, check_slurs_opt, convert_datetime},
   LemmyError,
 };
-use crate::objects::{FromApub, ToApub, FromApubToForm};
 use lemmy_websocket::LemmyContext;
 use url::Url;
 
@@ -117,7 +120,23 @@ impl FromApub for Community {
     expected_domain: Option<Url>,
     request_counter: &mut i32,
   ) -> Result<Community, LemmyError> {
-    todo!()
+    let community_id = group.id_unchecked().context(location_info!())?.to_owned();
+    let domain = community_id.domain().context(location_info!())?;
+    if domain == Settings::get().hostname {
+      let community = blocking(context.pool(), move |conn| {
+        Community::read_from_actor_id(conn, community_id.as_str())
+      })
+      .await??;
+      Ok(community)
+    } else {
+      let community_form =
+        CommunityForm::from_apub(group, context, expected_domain, request_counter).await?;
+      let community = blocking(context.pool(), move |conn| {
+        Community::upsert(conn, &community_form)
+      })
+      .await??;
+      Ok(community)
+    }
   }
 }
 

lemmy_apub/src/objects/mod.rs

@@ -7,10 +7,10 @@ use activitystreams::{
 };
 use anyhow::{anyhow, Context};
 use chrono::NaiveDateTime;
-use lemmy_utils::{location_info, utils::convert_datetime, LemmyError};
-use url::Url;
-use lemmy_websocket::LemmyContext;
 use lemmy_db::DbPool;
+use lemmy_utils::{location_info, utils::convert_datetime, LemmyError};
+use lemmy_websocket::LemmyContext;
+use url::Url;
 
 pub(crate) mod comment;
 pub(crate) mod community;

lemmy_apub/src/objects/post.rs

@@ -6,6 +6,9 @@ use crate::{
     create_tombstone,
     get_source_markdown_value,
     set_content_and_source,
+    FromApub,
+    FromApubToForm,
+    ToApub,
   },
   PageExt,
 };
@@ -27,13 +30,13 @@ use lemmy_structs::blocking;
 use lemmy_utils::{
   location_info,
   request::fetch_iframely_and_pictrs_data,
+  settings::Settings,
   utils::{check_slurs, convert_datetime, remove_slurs},
   LemmyError,
 };
 use lemmy_websocket::LemmyContext;
 use log::error;
 use url::Url;
-use crate::objects::{FromApubToForm, ToApub, FromApub};
 
 #[async_trait::async_trait(?Send)]
 impl ToApub for Post {
@@ -104,12 +107,24 @@ impl FromApub for Post {
   ///
   /// If the post's community or creator are not known locally, these are also fetched.
   async fn from_apub(
-    note: &PageExt,
+    page: &PageExt,
     context: &LemmyContext,
     expected_domain: Option<Url>,
     request_counter: &mut i32,
   ) -> Result<Post, LemmyError> {
-    todo!()
+    let post_id = page.id_unchecked().context(location_info!())?.to_owned();
+    let domain = post_id.domain().context(location_info!())?;
+    if domain == Settings::get().hostname {
+      let post = blocking(context.pool(), move |conn| {
+        Post::read_from_apub_id(conn, post_id.as_str())
+      })
+      .await??;
+      Ok(post)
+    } else {
+      let post_form = PostForm::from_apub(page, context, expected_domain, request_counter).await?;
+      let post = blocking(context.pool(), move |conn| Post::upsert(conn, &post_form)).await??;
+      Ok(post)
+    }
   }
 }
 

lemmy_apub/src/objects/private_message.rs

@@ -7,10 +7,12 @@ use crate::{
     create_tombstone,
     get_source_markdown_value,
     set_content_and_source,
+    FromApub,
+    FromApubToForm,
+    ToApub,
   },
   NoteExt,
 };
-use crate::objects::{FromApubToForm, ToApub, FromApub};
 use activitystreams::{
   object::{kind::NoteType, ApObject, Note, Tombstone},
   prelude::*,
@@ -23,7 +25,7 @@ use lemmy_db::{
   DbPool,
 };
 use lemmy_structs::blocking;
-use lemmy_utils::{location_info, utils::convert_datetime, LemmyError};
+use lemmy_utils::{location_info, settings::Settings, utils::convert_datetime, LemmyError};
 use lemmy_websocket::LemmyContext;
 use url::Url;
 
@@ -71,7 +73,23 @@ impl FromApub for PrivateMessage {
     expected_domain: Option<Url>,
     request_counter: &mut i32,
   ) -> Result<PrivateMessage, LemmyError> {
-    todo!()
+    let private_message_id = note.id_unchecked().context(location_info!())?.to_owned();
+    let domain = private_message_id.domain().context(location_info!())?;
+    if domain == Settings::get().hostname {
+      let private_message = blocking(context.pool(), move |conn| {
+        PrivateMessage::read_from_apub_id(conn, private_message_id.as_str())
+      })
+      .await??;
+      Ok(private_message)
+    } else {
+      let private_message_form =
+        PrivateMessageForm::from_apub(note, context, expected_domain, request_counter).await?;
+      let private_message = blocking(context.pool(), move |conn| {
+        PrivateMessage::upsert(conn, &private_message_form)
+      })
+      .await??;
+      Ok(private_message)
+    }
   }
 }
 

lemmy_apub/src/objects/user.rs

@@ -1,10 +1,16 @@
 use crate::{
   extensions::context::lemmy_context,
-  objects::{check_object_domain, get_source_markdown_value, set_content_and_source},
+  objects::{
+    check_object_domain,
+    get_source_markdown_value,
+    set_content_and_source,
+    FromApub,
+    FromApubToForm,
+    ToApub,
+  },
   ActorType,
   PersonExt,
 };
-use crate::objects::{FromApubToForm, ToApub, FromApub};
 use activitystreams::{
   actor::{ApActor, Endpoints, Person},
   object::{ApObject, Image, Tombstone},
@@ -17,8 +23,10 @@ use lemmy_db::{
   user::{UserForm, User_},
   DbPool,
 };
+use lemmy_structs::blocking;
 use lemmy_utils::{
   location_info,
+  settings::Settings,
   utils::{check_slurs, check_slurs_opt, convert_datetime},
   LemmyError,
 };
@@ -84,12 +92,25 @@ impl FromApub for User_ {
   type ApubType = PersonExt;
 
   async fn from_apub(
-    note: &PersonExt,
+    person: &PersonExt,
     context: &LemmyContext,
     expected_domain: Option<Url>,
     request_counter: &mut i32,
   ) -> Result<User_, LemmyError> {
-    todo!()
+    let user_id = person.id_unchecked().context(location_info!())?.to_owned();
+    let domain = user_id.domain().context(location_info!())?;
+    if domain == Settings::get().hostname {
+      let user = blocking(context.pool(), move |conn| {
+        User_::read_from_actor_id(conn, user_id.as_str())
+      })
+      .await??;
+      Ok(user)
+    } else {
+      let user_form =
+        UserForm::from_apub(person, context, expected_domain, request_counter).await?;
+      let user = blocking(context.pool(), move |conn| User_::upsert(conn, &user_form)).await??;
+      Ok(user)
+    }
   }
 }