Rewrite private message apub and merge create/update

crates/api_crud/src/private_message/create.rs

@@ -6,7 +6,14 @@ use lemmy_api_common::{
   person::{CreatePrivateMessage, PrivateMessageResponse},
   send_email_to_user,
 };
-use lemmy_apub::{generate_apub_endpoint, ApubObjectType, EndpointType};
+use lemmy_apub::{
+  activities::{
+    private_message::create_or_update::CreateOrUpdatePrivateMessage,
+    CreateOrUpdateType,
+  },
+  generate_apub_endpoint,
+  EndpointType,
+};
 use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
 use lemmy_db_schema::source::private_message::{PrivateMessage, PrivateMessageForm};
 use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
@@ -63,9 +70,13 @@ impl PerformCrud for CreatePrivateMessage {
     .await?
     .map_err(|_| ApiError::err("couldnt_create_private_message"))?;
 
-    updated_private_message
-      .send_create(&local_user_view.person, context)
-      .await?;
+    CreateOrUpdatePrivateMessage::send(
+      &updated_private_message,
+      &local_user_view.person,
+      CreateOrUpdateType::Create,
+      context,
+    )
+    .await?;
 
     let private_message_view = blocking(context.pool(), move |conn| {
       PrivateMessageView::read(conn, inserted_private_message.id)

crates/api_crud/src/private_message/update.rs

@@ -5,7 +5,10 @@ use lemmy_api_common::{
   get_local_user_view_from_jwt,
   person::{EditPrivateMessage, PrivateMessageResponse},
 };
-use lemmy_apub::ApubObjectType;
+use lemmy_apub::activities::{
+  private_message::create_or_update::CreateOrUpdatePrivateMessage,
+  CreateOrUpdateType,
+};
 use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud, DeleteableOrRemoveable};
 use lemmy_db_schema::source::private_message::PrivateMessage;
 use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
@@ -44,9 +47,13 @@ impl PerformCrud for EditPrivateMessage {
     .map_err(|_| ApiError::err("couldnt_update_private_message"))?;
 
     // Send the apub update
-    updated_private_message
-      .send_update(&local_user_view.person, context)
-      .await?;
+    CreateOrUpdatePrivateMessage::send(
+      &updated_private_message,
+      &local_user_view.person,
+      CreateOrUpdateType::Update,
+      context,
+    )
+    .await?;
 
     let private_message_id = data.private_message_id;
     let mut private_message_view = blocking(context.pool(), move |conn| {

crates/apub/src/activities/private_message/create.rs

@@ -1,64 +0,0 @@
-use crate::{
-  activities::{private_message::send_websocket_message, verify_activity, verify_person},
-  objects::FromApub,
-  NoteExt,
-};
-use activitystreams::{activity::kind::CreateType, base::BaseExt};
-use lemmy_apub_lib::{verify_domains_match_opt, ActivityCommonFields, ActivityHandler};
-use lemmy_db_schema::source::private_message::PrivateMessage;
-use lemmy_utils::LemmyError;
-use lemmy_websocket::{LemmyContext, UserOperationCrud};
-use url::Url;
-
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct CreatePrivateMessage {
-  to: Url,
-  object: NoteExt,
-  #[serde(rename = "type")]
-  kind: CreateType,
-  #[serde(flatten)]
-  common: ActivityCommonFields,
-}
-
-#[async_trait::async_trait(?Send)]
-impl ActivityHandler for CreatePrivateMessage {
-  async fn verify(
-    &self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError> {
-    verify_activity(self.common())?;
-    verify_person(&self.common.actor, context, request_counter).await?;
-    verify_domains_match_opt(&self.common.actor, self.object.id_unchecked())?;
-    Ok(())
-  }
-
-  async fn receive(
-    &self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError> {
-    let private_message = PrivateMessage::from_apub(
-      &self.object,
-      context,
-      self.common.actor.clone(),
-      request_counter,
-      false,
-    )
-    .await?;
-
-    send_websocket_message(
-      private_message.id,
-      UserOperationCrud::CreatePrivateMessage,
-      context,
-    )
-    .await?;
-
-    Ok(())
-  }
-
-  fn common(&self) -> &ActivityCommonFields {
-    &self.common
-  }
-}

crates/apub/src/activities/private_message/create_or_update.rs

@@ -0,0 +1,107 @@
+use crate::{
+  activities::{
+    generate_activity_id,
+    private_message::send_websocket_message,
+    verify_activity,
+    verify_person,
+    CreateOrUpdateType,
+  },
+  activity_queue::send_activity_new,
+  extensions::context::lemmy_context,
+  objects::{private_message::Note, FromApub, ToApub},
+  ActorType,
+};
+use lemmy_api_common::blocking;
+use lemmy_apub_lib::{verify_domains_match, ActivityCommonFields, ActivityHandler};
+use lemmy_db_queries::Crud;
+use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
+use lemmy_utils::LemmyError;
+use lemmy_websocket::{LemmyContext, UserOperationCrud};
+use url::Url;
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateOrUpdatePrivateMessage {
+  to: Url,
+  object: Note,
+  #[serde(rename = "type")]
+  kind: CreateOrUpdateType,
+  #[serde(flatten)]
+  common: ActivityCommonFields,
+}
+
+impl CreateOrUpdatePrivateMessage {
+  pub async fn send(
+    private_message: &PrivateMessage,
+    actor: &Person,
+    kind: CreateOrUpdateType,
+    context: &LemmyContext,
+  ) -> Result<(), LemmyError> {
+    let recipient_id = private_message.recipient_id;
+    let recipient =
+      blocking(context.pool(), move |conn| Person::read(conn, recipient_id)).await??;
+
+    let id = generate_activity_id(kind.clone())?;
+    let create_or_update = CreateOrUpdatePrivateMessage {
+      to: recipient.actor_id(),
+      object: private_message.to_apub(context.pool()).await?,
+      kind,
+      common: ActivityCommonFields {
+        context: lemmy_context(),
+        id: id.clone(),
+        actor: actor.actor_id(),
+        unparsed: Default::default(),
+      },
+    };
+    send_activity_new(
+      context,
+      &create_or_update,
+      &id,
+      actor,
+      vec![recipient.get_shared_inbox_or_inbox_url()],
+      true,
+    )
+    .await
+  }
+}
+#[async_trait::async_trait(?Send)]
+impl ActivityHandler for CreateOrUpdatePrivateMessage {
+  async fn verify(
+    &self,
+    context: &LemmyContext,
+    request_counter: &mut i32,
+  ) -> Result<(), LemmyError> {
+    verify_activity(self.common())?;
+    verify_person(&self.common.actor, context, request_counter).await?;
+    verify_domains_match(&self.common.actor, &self.object.id)?;
+    self.object.verify(context, request_counter).await?;
+    Ok(())
+  }
+
+  async fn receive(
+    &self,
+    context: &LemmyContext,
+    request_counter: &mut i32,
+  ) -> Result<(), LemmyError> {
+    let private_message = PrivateMessage::from_apub(
+      &self.object,
+      context,
+      self.common.actor.clone(),
+      request_counter,
+      false,
+    )
+    .await?;
+
+    let notif_type = match self.kind {
+      CreateOrUpdateType::Create => UserOperationCrud::CreatePrivateMessage,
+      CreateOrUpdateType::Update => UserOperationCrud::EditPrivateMessage,
+    };
+    send_websocket_message(private_message.id, notif_type, context).await?;
+
+    Ok(())
+  }
+
+  fn common(&self) -> &ActivityCommonFields {
+    &self.common
+  }
+}

crates/apub/src/activities/private_message/mod.rs

@@ -4,10 +4,9 @@ use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::Priva
 use lemmy_utils::LemmyError;
 use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperationCrud};
 
-pub mod create;
+pub mod create_or_update;
 pub mod delete;
 pub mod undo_delete;
-pub mod update;
 
 async fn send_websocket_message(
   private_message_id: PrivateMessageId,

crates/apub/src/activities/private_message/update.rs

@@ -1,64 +0,0 @@
-use crate::{
-  activities::{private_message::send_websocket_message, verify_activity, verify_person},
-  objects::FromApub,
-  NoteExt,
-};
-use activitystreams::{activity::kind::UpdateType, base::BaseExt};
-use lemmy_apub_lib::{verify_domains_match_opt, ActivityCommonFields, ActivityHandler};
-use lemmy_db_schema::source::private_message::PrivateMessage;
-use lemmy_utils::LemmyError;
-use lemmy_websocket::{LemmyContext, UserOperationCrud};
-use url::Url;
-
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct UpdatePrivateMessage {
-  to: Url,
-  object: NoteExt,
-  #[serde(rename = "type")]
-  kind: UpdateType,
-  #[serde(flatten)]
-  common: ActivityCommonFields,
-}
-
-#[async_trait::async_trait(?Send)]
-impl ActivityHandler for UpdatePrivateMessage {
-  async fn verify(
-    &self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError> {
-    verify_activity(self.common())?;
-    verify_person(&self.common.actor, context, request_counter).await?;
-    verify_domains_match_opt(&self.common.actor, self.object.id_unchecked())?;
-    Ok(())
-  }
-
-  async fn receive(
-    &self,
-    context: &LemmyContext,
-    request_counter: &mut i32,
-  ) -> Result<(), LemmyError> {
-    let private_message = PrivateMessage::from_apub(
-      &self.object,
-      context,
-      self.common.actor.clone(),
-      request_counter,
-      false,
-    )
-    .await?;
-
-    send_websocket_message(
-      private_message.id,
-      UserOperationCrud::EditPrivateMessage,
-      context,
-    )
-    .await?;
-
-    Ok(())
-  }
-
-  fn common(&self) -> &ActivityCommonFields {
-    &self.common
-  }
-}

crates/apub/src/activities/send/community.rs

@@ -54,7 +54,7 @@ impl ActorType for Community {
     self.local
   }
   fn actor_id(&self) -> Url {
-    self.actor_id.to_owned().into_inner()
+    self.actor_id.to_owned().into()
   }
   fn name(&self) -> String {
     self.name.clone()
@@ -78,7 +78,7 @@ impl ActorType for Community {
 #[async_trait::async_trait(?Send)]
 impl CommunityType for Community {
   fn followers_url(&self) -> Url {
-    self.followers_url.clone().into_inner()
+    self.followers_url.clone().into()
   }
 
   /// As a local community, accept the follow request from a remote person.

crates/apub/src/activities/send/private_message.rs

@@ -2,19 +2,17 @@ use crate::{
   activities::generate_activity_id,
   activity_queue::send_activity_single_dest,
   extensions::context::lemmy_context,
-  objects::ToApub,
   ActorType,
   ApubObjectType,
 };
 use activitystreams::{
   activity::{
-    kind::{CreateType, DeleteType, UndoType, UpdateType},
-    Create,
+    kind::{DeleteType, UndoType},
     Delete,
     Undo,
-    Update,
   },
-  prelude::*,
+  base::{BaseExt, ExtendsExt},
+  object::ObjectExt,
 };
 use lemmy_api_common::blocking;
 use lemmy_db_queries::Crud;
@@ -24,47 +22,20 @@ use lemmy_websocket::LemmyContext;
 
 #[async_trait::async_trait(?Send)]
 impl ApubObjectType for PrivateMessage {
-  /// Send out information about a newly created private message
-  async fn send_create(&self, creator: &Person, context: &LemmyContext) -> Result<(), LemmyError> {
-    let note = self.to_apub(context.pool()).await?;
-
-    let recipient_id = self.recipient_id;
-    let recipient =
-      blocking(context.pool(), move |conn| Person::read(conn, recipient_id)).await??;
-
-    let mut create = Create::new(
-      creator.actor_id.to_owned().into_inner(),
-      note.into_any_base()?,
-    );
-
-    create
-      .set_many_contexts(lemmy_context())
-      .set_id(generate_activity_id(CreateType::Create)?)
-      .set_to(recipient.actor_id());
-
-    send_activity_single_dest(create, creator, recipient.inbox_url.into(), context).await?;
-    Ok(())
+  async fn send_create(
+    &self,
+    _creator: &Person,
+    _context: &LemmyContext,
+  ) -> Result<(), LemmyError> {
+    unimplemented!()
   }
 
-  /// Send out information about an edited private message, to the followers of the community.
-  async fn send_update(&self, creator: &Person, context: &LemmyContext) -> Result<(), LemmyError> {
-    let note = self.to_apub(context.pool()).await?;
-
-    let recipient_id = self.recipient_id;
-    let recipient =
-      blocking(context.pool(), move |conn| Person::read(conn, recipient_id)).await??;
-
-    let mut update = Update::new(
-      creator.actor_id.to_owned().into_inner(),
-      note.into_any_base()?,
-    );
-    update
-      .set_many_contexts(lemmy_context())
-      .set_id(generate_activity_id(UpdateType::Update)?)
-      .set_to(recipient.actor_id());
-
-    send_activity_single_dest(update, creator, recipient.inbox_url.into(), context).await?;
-    Ok(())
+  async fn send_update(
+    &self,
+    _creator: &Person,
+    _context: &LemmyContext,
+  ) -> Result<(), LemmyError> {
+    unimplemented!()
   }
 
   async fn send_delete(&self, creator: &Person, context: &LemmyContext) -> Result<(), LemmyError> {

crates/apub/src/http/community.rs

@@ -155,7 +155,7 @@ pub(crate) async fn get_apub_community_moderators(
 
   let moderators: Vec<Url> = moderators
     .into_iter()
-    .map(|m| m.moderator.actor_id.into_inner())
+    .map(|m| m.moderator.actor_id.into())
     .collect();
   let mut collection = OrderedCollection::new();
   collection

crates/apub/src/http/inbox_enums.rs

@@ -11,10 +11,9 @@ use crate::activities::{
   following::{accept::AcceptFollowCommunity, follow::FollowCommunity, undo::UndoFollowCommunity},
   post::create_or_update::CreateOrUpdatePost,
   private_message::{
-    create::CreatePrivateMessage,
+    create_or_update::CreateOrUpdatePrivateMessage,
     delete::DeletePrivateMessage,
     undo_delete::UndoDeletePrivateMessage,
-    update::UpdatePrivateMessage,
   },
   removal::{
     remove::RemovePostCommentCommunityOrMod,
@@ -36,8 +35,7 @@ use serde::{Deserialize, Serialize};
 #[serde(untagged)]
 pub enum PersonInboxActivities {
   AcceptFollowCommunity(AcceptFollowCommunity),
-  CreatePrivateMessage(CreatePrivateMessage),
-  UpdatePrivateMessage(UpdatePrivateMessage),
+  CreateOrUpdatePrivateMessage(CreateOrUpdatePrivateMessage),
   DeletePrivateMessage(DeletePrivateMessage),
   UndoDeletePrivateMessage(UndoDeletePrivateMessage),
   AnnounceActivity(Box<AnnounceActivity>),
@@ -71,7 +69,7 @@ pub enum SharedInboxActivities {
   FollowCommunity(FollowCommunity),
   UndoFollowCommunity(UndoFollowCommunity),
   CreateOrUpdateComment(CreateOrUpdateComment),
-  CreateOrUpdatePost(CreateOrUpdatePost),
+  CreateOrUpdatePost(Box<CreateOrUpdatePost>),
   LikePostOrComment(LikePostOrComment),
   DislikePostOrComment(DislikePostOrComment),
   UndoDislikePostOrComment(UndoDislikePostOrComment),
@@ -88,8 +86,7 @@ pub enum SharedInboxActivities {
   AcceptFollowCommunity(AcceptFollowCommunity),
   // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
   // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
-  CreatePrivateMessage(CreatePrivateMessage),
-  UpdatePrivateMessage(UpdatePrivateMessage),
+  CreateOrUpdatePrivateMessage(CreateOrUpdatePrivateMessage),
   DeletePrivateMessage(DeletePrivateMessage),
   UndoDeletePrivateMessage(UndoDeletePrivateMessage),
   AnnounceActivity(Box<AnnounceActivity>),

crates/apub/src/lib.rs

@@ -20,7 +20,7 @@ use activitystreams::{
   activity::Follow,
   actor,
   base::AnyBase,
-  object::{ApObject, AsObject, Note, ObjectExt},
+  object::{ApObject, AsObject, ObjectExt},
 };
 use activitystreams_ext::Ext2;
 use anyhow::{anyhow, Context};
@@ -53,7 +53,6 @@ pub type GroupExt =
 type PersonExt =
   Ext2<actor::ApActor<ApObject<actor::Actor<UserTypes>>>, PersonExtension, PublicKeyExtension>;
 pub type SiteExt = actor::ApActor<ApObject<actor::Service>>;
-pub type NoteExt = ApObject<Note>;
 
 #[derive(Clone, Copy, Debug, serde::Deserialize, serde::Serialize, PartialEq)]
 pub enum UserTypes {
@@ -314,7 +313,7 @@ pub fn generate_inbox_url(actor_id: &DbUrl) -> Result<DbUrl, ParseError> {
 }
 
 pub fn generate_shared_inbox_url(actor_id: &DbUrl) -> Result<DbUrl, LemmyError> {
-  let actor_id = actor_id.clone().into_inner();
+  let actor_id: Url = actor_id.clone().into();
   let url = format!(
     "{}://{}{}/inbox",
     &actor_id.scheme(),

crates/apub/src/objects/private_message.rs

@@ -1,60 +1,93 @@
 use crate::{
   extensions::context::lemmy_context,
   fetcher::person::get_or_fetch_and_upsert_person,
-  objects::{
-    check_object_domain,
-    create_tombstone,
-    get_object_from_apub,
-    get_source_markdown_value,
-    set_content_and_source,
-    FromApub,
-    FromApubToForm,
-    ToApub,
-  },
-  NoteExt,
+  objects::{create_tombstone, FromApub, Source, ToApub},
 };
 use activitystreams::{
-  object::{kind::NoteType, ApObject, Note, Tombstone},
-  prelude::*,
+  base::AnyBase,
+  object::{kind::NoteType, Tombstone},
+  primitives::OneOrMany,
+  unparsed::Unparsed,
 };
-use anyhow::Context;
+use anyhow::anyhow;
+use chrono::{DateTime, FixedOffset};
 use lemmy_api_common::blocking;
-use lemmy_db_queries::{Crud, DbPool};
+use lemmy_apub_lib::{
+  values::{MediaTypeHtml, MediaTypeMarkdown},
+  verify_domains_match,
+};
+use lemmy_db_queries::{ApubObject, Crud, DbPool};
 use lemmy_db_schema::source::{
   person::Person,
   private_message::{PrivateMessage, PrivateMessageForm},
 };
-use lemmy_utils::{location_info, utils::convert_datetime, LemmyError};
+use lemmy_utils::{utils::convert_datetime, LemmyError};
 use lemmy_websocket::LemmyContext;
+use serde::{Deserialize, Serialize};
 use url::Url;
 
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Note {
+  #[serde(rename = "@context")]
+  context: OneOrMany<AnyBase>,
+  r#type: NoteType,
+  pub(crate) id: Url,
+  pub(crate) attributed_to: Url,
+  to: Url,
+  content: String,
+  media_type: MediaTypeHtml,
+  source: Source,
+  published: DateTime<FixedOffset>,
+  updated: Option<DateTime<FixedOffset>>,
+  #[serde(flatten)]
+  unparsed: Unparsed,
+}
+
+impl Note {
+  pub(crate) async fn verify(
+    &self,
+    context: &LemmyContext,
+    request_counter: &mut i32,
+  ) -> Result<(), LemmyError> {
+    verify_domains_match(&self.attributed_to, &self.id)?;
+    let person =
+      get_or_fetch_and_upsert_person(&self.attributed_to, context, request_counter).await?;
+    if person.banned {
+      return Err(anyhow!("Person is banned from site").into());
+    }
+    Ok(())
+  }
+}
+
 #[async_trait::async_trait(?Send)]
 impl ToApub for PrivateMessage {
-  type ApubType = NoteExt;
-
-  async fn to_apub(&self, pool: &DbPool) -> Result<NoteExt, LemmyError> {
-    let mut private_message = ApObject::new(Note::new());
+  type ApubType = Note;
 
+  async fn to_apub(&self, pool: &DbPool) -> Result<Note, LemmyError> {
     let creator_id = self.creator_id;
     let creator = blocking(pool, move |conn| Person::read(conn, creator_id)).await??;
 
     let recipient_id = self.recipient_id;
     let recipient = blocking(pool, move |conn| Person::read(conn, recipient_id)).await??;
 
-    private_message
-      .set_many_contexts(lemmy_context())
-      .set_id(self.ap_id.to_owned().into_inner())
-      .set_published(convert_datetime(self.published))
-      .set_to(recipient.actor_id.into_inner())
-      .set_attributed_to(creator.actor_id.into_inner());
-
-    set_content_and_source(&mut private_message, &self.content)?;
-
-    if let Some(u) = self.updated {
-      private_message.set_updated(convert_datetime(u));
-    }
-
-    Ok(private_message)
+    let note = Note {
+      context: lemmy_context(),
+      r#type: NoteType::Note,
+      id: self.ap_id.clone().into(),
+      attributed_to: creator.actor_id.into_inner(),
+      to: recipient.actor_id.into(),
+      content: self.content.clone(),
+      media_type: MediaTypeHtml::Html,
+      source: Source {
+        content: self.content.clone(),
+        media_type: MediaTypeMarkdown::Markdown,
+      },
+      published: convert_datetime(self.published),
+      updated: self.updated.map(convert_datetime),
+      unparsed: Default::default(),
+    };
+    Ok(note)
   }
 
   fn to_tombstone(&self) -> Result<Tombstone, LemmyError> {
@@ -69,66 +102,35 @@ impl ToApub for PrivateMessage {
 
 #[async_trait::async_trait(?Send)]
 impl FromApub for PrivateMessage {
-  type ApubType = NoteExt;
+  type ApubType = Note;
 
   async fn from_apub(
-    note: &NoteExt,
+    note: &Note,
     context: &LemmyContext,
-    expected_domain: Url,
-    request_counter: &mut i32,
-    mod_action_allowed: bool,
-  ) -> Result<PrivateMessage, LemmyError> {
-    get_object_from_apub(
-      note,
-      context,
-      expected_domain,
-      request_counter,
-      mod_action_allowed,
-    )
-    .await
-  }
-}
-
-#[async_trait::async_trait(?Send)]
-impl FromApubToForm<NoteExt> for PrivateMessageForm {
-  async fn from_apub(
-    note: &NoteExt,
-    context: &LemmyContext,
-    expected_domain: Url,
+    _expected_domain: Url,
     request_counter: &mut i32,
     _mod_action_allowed: bool,
-  ) -> Result<PrivateMessageForm, LemmyError> {
-    let creator_actor_id = note
-      .attributed_to()
-      .context(location_info!())?
-      .clone()
-      .single_xsd_any_uri()
-      .context(location_info!())?;
-
+  ) -> Result<PrivateMessage, LemmyError> {
     let creator =
-      get_or_fetch_and_upsert_person(&creator_actor_id, context, request_counter).await?;
-    let recipient_actor_id = note
-      .to()
-      .context(location_info!())?
-      .clone()
-      .single_xsd_any_uri()
-      .context(location_info!())?;
-    let recipient =
-      get_or_fetch_and_upsert_person(&recipient_actor_id, context, request_counter).await?;
-    let ap_id = Some(check_object_domain(note, expected_domain, false)?);
-
-    let content = get_source_markdown_value(note)?.context(location_info!())?;
+      get_or_fetch_and_upsert_person(&note.attributed_to, context, request_counter).await?;
+    let recipient = get_or_fetch_and_upsert_person(&note.to, context, request_counter).await?;
 
-    Ok(PrivateMessageForm {
+    let form = PrivateMessageForm {
       creator_id: creator.id,
       recipient_id: recipient.id,
-      content,
-      published: note.published().map(|u| u.to_owned().naive_local()),
-      updated: note.updated().map(|u| u.to_owned().naive_local()),
+      content: note.source.content.clone(),
+      published: Some(note.published.naive_local()),
+      updated: note.updated.map(|u| u.to_owned().naive_local()),
       deleted: None,
       read: None,
-      ap_id,
+      ap_id: Some(note.id.clone().into()),
       local: Some(false),
-    })
+    };
+    Ok(
+      blocking(context.pool(), move |conn| {
+        PrivateMessage::upsert(conn, &form)
+      })
+      .await??,
+    )
   }
 }

crates/db_schema/src/lib.rs

@@ -92,6 +92,7 @@ where
 }
 
 impl DbUrl {
+  // TODO: remove this method and just use into()
   pub fn into_inner(self) -> Url {
     self.0
   }
@@ -99,7 +100,7 @@ impl DbUrl {
 
 impl Display for DbUrl {
   fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-    self.to_owned().into_inner().fmt(f)
+    self.to_owned().0.fmt(f)
   }
 }