Moving upserts to lemmy_db.

server/lemmy_db/src/comment.rs

@@ -148,6 +148,18 @@ impl Comment {
       .set((content.eq(new_content), updated.eq(naive_now())))
       .get_result::<Self>(conn)
   }
+
+  pub fn upsert(
+    conn: &PgConnection,
+    comment_form: &CommentForm,
+  ) -> Result<Self, Error> {
+    let existing = Self::read_from_apub_id(conn, &comment_form.ap_id);
+    match existing {
+      Err(NotFound {}) => Ok(Self::create(conn, &comment_form)?),
+      Ok(p) => Ok(Self::update(conn, p.id, &comment_form)?),
+      Err(e) => Err(e),
+    }
+  }
 }
 
 #[derive(Identifiable, Queryable, Associations, PartialEq, Debug, Clone)]

server/lemmy_db/src/post.rs

@@ -155,6 +155,15 @@ impl Post {
   pub fn is_post_creator(user_id: i32, post_creator_id: i32) -> bool {
     user_id == post_creator_id
   }
+
+  pub fn upsert(conn: &PgConnection, post_form: &PostForm) -> Result<Post, Error> {
+    let existing = Self::read_from_apub_id(conn, &post_form.ap_id);
+    match existing {
+      Err(NotFound {}) => Ok(Self::create(conn, &post_form)?),
+      Ok(p) => Ok(Self::update(conn, p.id, &post_form)?),
+      Err(e) => Err(e),
+    }
+  }
 }
 
 impl Crud<PostForm> for Post {

server/src/apub/fetcher.rs

@@ -18,7 +18,7 @@ use activitystreams::{base::BaseExt, collection::OrderedCollection, object::Note
 use actix_web::client::Client;
 use anyhow::{anyhow, Context};
 use chrono::NaiveDateTime;
-use diesel::{result::Error::NotFound, PgConnection};
+use diesel::result::Error::NotFound;
 use lemmy_db::{
   comment::{Comment, CommentForm},
   comment_view::CommentView,
@@ -158,7 +158,7 @@ pub async fn search_by_apub_id(
     SearchAcceptedObjects::Page(p) => {
       let post_form = PostForm::from_apub(&p, client, pool, Some(query_url)).await?;
 
-      let p = blocking(pool, move |conn| upsert_post(&post_form, conn)).await??;
+      let p = blocking(pool, move |conn| Post::upsert(conn, &post_form)).await??;
       response.posts = vec![blocking(pool, move |conn| PostView::read(conn, p.id, None)).await??];
 
       response
@@ -166,7 +166,7 @@ pub async fn search_by_apub_id(
     SearchAcceptedObjects::Comment(c) => {
       let comment_form = CommentForm::from_apub(&c, client, pool, Some(query_url)).await?;
 
-      let c = blocking(pool, move |conn| upsert_comment(&comment_form, conn)).await??;
+      let c = blocking(pool, move |conn| Comment::upsert(conn, &comment_form)).await??;
       response.comments =
         vec![blocking(pool, move |conn| CommentView::read(conn, c.id, None)).await??];
 
@@ -341,15 +341,6 @@ async fn fetch_remote_community(
   Ok(community)
 }
 
-fn upsert_post(post_form: &PostForm, conn: &PgConnection) -> Result<Post, LemmyError> {
-  let existing = Post::read_from_apub_id(conn, &post_form.ap_id);
-  match existing {
-    Err(NotFound {}) => Ok(Post::create(conn, &post_form)?),
-    Ok(p) => Ok(Post::update(conn, p.id, &post_form)?),
-    Err(e) => Err(e.into()),
-  }
-}
-
 pub async fn get_or_fetch_and_insert_post(
   post_ap_id: &Url,
   client: &Client,
@@ -376,15 +367,6 @@ pub async fn get_or_fetch_and_insert_post(
   }
 }
 
-fn upsert_comment(comment_form: &CommentForm, conn: &PgConnection) -> Result<Comment, LemmyError> {
-  let existing = Comment::read_from_apub_id(conn, &comment_form.ap_id);
-  match existing {
-    Err(NotFound {}) => Ok(Comment::create(conn, &comment_form)?),
-    Ok(p) => Ok(Comment::update(conn, p.id, &comment_form)?),
-    Err(e) => Err(e.into()),
-  }
-}
-
 pub async fn get_or_fetch_and_insert_comment(
   comment_ap_id: &Url,
   client: &Client,

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

@@ -30,7 +30,6 @@ use lemmy_db::{
   comment_view::CommentView,
   post::{Post, PostForm},
   post_view::PostView,
-  Crud,
 };
 use lemmy_utils::{location_info, scrape_text_for_mentions};
 
@@ -65,7 +64,9 @@ async fn receive_create_post(
 
   let post = PostForm::from_apub(&page, client, pool, Some(user.actor_id()?)).await?;
 
-  let inserted_post = blocking(pool, move |conn| Post::create(conn, &post)).await??;
+  // Using an upsert, since likes (which fetch the post), sometimes come in before the create
+  // resulting in double posts.
+  let inserted_post = blocking(pool, move |conn| Post::upsert(conn, &post)).await??;
 
   // Refetch the view
   let inserted_post_id = inserted_post.id;
@@ -98,7 +99,7 @@ async fn receive_create_comment(
 
   let comment = CommentForm::from_apub(&note, client, pool, Some(user.actor_id()?)).await?;
 
-  let inserted_comment = blocking(pool, move |conn| Comment::create(conn, &comment)).await??;
+  let inserted_comment = blocking(pool, move |conn| Comment::upsert(conn, &comment)).await??;
 
   let post_id = inserted_comment.post_id;
   let post = blocking(pool, move |conn| Post::read(conn, post_id)).await??;