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)
 }
 
-pub 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,18 +367,6 @@ pub async fn get_or_fetch_and_insert_post(
   }
 }
 
-pub 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

@@ -4,7 +4,6 @@ use crate::{
     post::PostResponse,
   },
   apub::{
-    fetcher::{upsert_comment, upsert_post},
     inbox::shared_inbox::{
       announce_if_community_is_local,
       get_user_from_activity,
@@ -27,7 +26,7 @@ use activitystreams::{activity::Create, base::AnyBase, object::Note, prelude::*}
 use actix_web::{client::Client, HttpResponse};
 use anyhow::Context;
 use lemmy_db::{
-  comment::CommentForm,
+  comment::{Comment, CommentForm},
   comment_view::CommentView,
   post::{Post, PostForm},
   post_view::PostView,
@@ -67,7 +66,7 @@ async fn receive_create_post(
 
   // 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| upsert_post(&post, conn)).await??;
+  let inserted_post = blocking(pool, move |conn| Post::upsert(conn, &post)).await??;
 
   // Refetch the view
   let inserted_post_id = inserted_post.id;
@@ -100,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| upsert_comment(&comment, conn)).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??;