implement upsert for user/community

server/lemmy_db/src/community.rs

@@ -160,6 +160,15 @@ impl Community {
       .unwrap_or_default()
       .contains(&user_id)
   }
+
+  pub fn upsert(conn: &PgConnection, community_form: &CommunityForm) -> Result<Community, Error> {
+    let existing = Self::read_from_actor_id(conn, &community_form.actor_id);
+    match existing {
+      Err(NotFound {}) => Ok(Self::create(conn, &community_form)?),
+      Ok(c) => Ok(Self::update(conn, c.id, &community_form)?),
+      Err(e) => Err(e),
+    }
+  }
 }
 
 #[derive(Identifiable, Queryable, Associations, PartialEq, Debug)]

server/lemmy_db/src/user.rs

@@ -152,6 +152,15 @@ impl User_ {
   pub fn get_profile_url(&self, hostname: &str) -> String {
     format!("https://{}/u/{}", hostname, self.name)
   }
+
+  pub fn upsert(conn: &PgConnection, user_form: &UserForm) -> Result<User_, Error> {
+    let existing = Self::read_from_actor_id(conn, &user_form.actor_id);
+    match existing {
+      Err(NotFound {}) => Ok(Self::create(conn, &user_form)?),
+      Ok(u) => Ok(Self::update(conn, u.id, &user_form)?),
+      Err(e) => Err(e),
+    }
+  }
 }
 
 #[cfg(test)]

server/src/apub/fetcher.rs

@@ -230,7 +230,7 @@ pub async fn get_or_fetch_and_upsert_user(
       let person = fetch_remote_object::<PersonExt>(context.client(), apub_id).await?;
 
       let uf = UserForm::from_apub(&person, context, Some(apub_id.to_owned())).await?;
-      let user = blocking(context.pool(), move |conn| User_::create(conn, &uf)).await??;
+      let user = blocking(context.pool(), move |conn| User_::upsert(conn, &uf)).await??;
 
       Ok(user)
     }
@@ -286,14 +286,7 @@ async fn fetch_remote_community(
   let group = fetch_remote_object::<GroupExt>(context.client(), apub_id).await?;
 
   let cf = CommunityForm::from_apub(&group, context, Some(apub_id.to_owned())).await?;
-  let community = blocking(context.pool(), move |conn| {
-    if let Some(cid) = community_id {
-      Community::update(conn, cid, &cf)
-    } else {
-      Community::create(conn, &cf)
-    }
-  })
-  .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!())?;

server/src/apub/inbox/shared_inbox.rs

@@ -66,6 +66,8 @@ pub async fn shared_inbox(
   let json = serde_json::to_string(&activity)?;
   debug!("Shared inbox received activity: {}", json);
 
+  // TODO: if we already received an activity with identical ID, then ignore this (same in other inboxes)
+
   let sender = &activity
     .actor()?
     .to_owned()