Merge branch 'federated_private_messages' into federation

server/migrations/2020-05-05-210233_add_activitypub_for_private_messages/down.sql

@@ -0,0 +1,21 @@
+drop materialized view private_message_mview;
+drop view private_message_view;
+
+alter table private_message 
+drop column ap_id, 
+drop column local;
+
+create view private_message_view as 
+select        
+pm.*,
+u.name as creator_name,
+u.avatar as creator_avatar,
+u2.name as recipient_name,
+u2.avatar as recipient_avatar
+from private_message pm
+inner join user_ u on u.id = pm.creator_id
+inner join user_ u2 on u2.id = pm.recipient_id;
+
+create materialized view private_message_mview as select * from private_message_view;
+
+create unique index idx_private_message_mview_id on private_message_mview (id);

server/migrations/2020-05-05-210233_add_activitypub_for_private_messages/up.sql

@@ -0,0 +1,25 @@
+alter table private_message
+add column ap_id character varying(255) not null default 'changeme', -- This needs to be checked and updated in code, building from the site url if local
+add column local boolean not null default true
+;
+
+drop materialized view private_message_mview;
+drop view private_message_view;
+create view private_message_view as 
+select        
+pm.*,
+u.name as creator_name,
+u.avatar as creator_avatar,
+u.actor_id as creator_actor_id,
+u.local as creator_local,
+u2.name as recipient_name,
+u2.avatar as recipient_avatar,
+u2.actor_id as recipient_actor_id,
+u2.local as recipient_local
+from private_message pm
+inner join user_ u on u.id = pm.creator_id
+inner join user_ u2 on u2.id = pm.recipient_id;
+
+create materialized view private_message_mview as select * from private_message_view;
+
+create unique index idx_private_message_mview_id on private_message_mview (id);

server/src/api/user.rs

@@ -186,7 +186,7 @@ pub struct PrivateMessagesResponse {
 
 #[derive(Serialize, Deserialize, Clone)]
 pub struct PrivateMessageResponse {
-  message: PrivateMessageView,
+  pub message: PrivateMessageView,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -861,12 +861,15 @@ impl Perform for Oper<MarkAllAsRead> {
 
     for message in &messages {
       let private_message_form = PrivateMessageForm {
-        content: None,
+        content: message.to_owned().content,
         creator_id: message.to_owned().creator_id,
         recipient_id: message.to_owned().recipient_id,
         deleted: None,
         read: Some(true),
         updated: None,
+        ap_id: message.to_owned().ap_id,
+        local: message.local,
+        published: None,
       };
 
       let _updated_message = match PrivateMessage::update(&conn, message.id, &private_message_form)
@@ -1034,19 +1037,23 @@ impl Perform for Oper<CreatePrivateMessage> {
     let conn = pool.get()?;
 
     // Check for a site ban
-    if UserView::read(&conn, user_id)?.banned {
+    let user = User_::read(&conn, user_id)?;
+    if user.banned {
       return Err(APIError::err("site_ban").into());
     }
 
     let content_slurs_removed = remove_slurs(&data.content.to_owned());
 
     let private_message_form = PrivateMessageForm {
-      content: Some(content_slurs_removed.to_owned()),
+      content: content_slurs_removed.to_owned(),
       creator_id: user_id,
       recipient_id: data.recipient_id,
       deleted: None,
       read: None,
       updated: None,
+      ap_id: "changeme".into(),
+      local: true,
+      published: None,
     };
 
     let inserted_private_message = match PrivateMessage::create(&conn, &private_message_form) {
@@ -1056,6 +1063,14 @@ impl Perform for Oper<CreatePrivateMessage> {
       }
     };
 
+    let updated_private_message =
+      match PrivateMessage::update_ap_id(&conn, inserted_private_message.id) {
+        Ok(private_message) => private_message,
+        Err(_e) => return Err(APIError::err("couldnt_create_private_message").into()),
+      };
+
+    updated_private_message.send_create(&user, &conn)?;
+
     // Send notifications to the recipient
     let recipient_user = User_::read(&conn, data.recipient_id)?;
     if recipient_user.send_notifications_to_email {
@@ -1099,7 +1114,7 @@ impl Perform for Oper<EditPrivateMessage> {
   fn perform(
     &self,
     pool: Pool<ConnectionManager<PgConnection>>,
-    _websocket_info: Option<WebsocketInfo>,
+    websocket_info: Option<WebsocketInfo>,
   ) -> Result<PrivateMessageResponse, Error> {
     let data: &EditPrivateMessage = &self.data;
 
@@ -1115,7 +1130,8 @@ impl Perform for Oper<EditPrivateMessage> {
     let orig_private_message = PrivateMessage::read(&conn, data.edit_id)?;
 
     // Check for a site ban
-    if UserView::read(&conn, user_id)?.banned {
+    let user = User_::read(&conn, user_id)?;
+    if user.banned {
       return Err(APIError::err("site_ban").into());
     }
 
@@ -1127,8 +1143,8 @@ impl Perform for Oper<EditPrivateMessage> {
     }
 
     let content_slurs_removed = match &data.content {
-      Some(content) => Some(remove_slurs(content)),
+      Some(content) => remove_slurs(content),
-      None => None,
+      None => orig_private_message.content,
     };
 
     let private_message_form = PrivateMessageForm {
@@ -1142,17 +1158,41 @@ impl Perform for Oper<EditPrivateMessage> {
       } else {
         Some(naive_now())
       },
+      ap_id: orig_private_message.ap_id,
+      local: orig_private_message.local,
+      published: None,
     };
 
-    let _updated_private_message =
+    let updated_private_message =
       match PrivateMessage::update(&conn, data.edit_id, &private_message_form) {
         Ok(private_message) => private_message,
         Err(_e) => return Err(APIError::err("couldnt_update_private_message").into()),
       };
 
+    if let Some(deleted) = data.deleted.to_owned() {
+      if deleted {
+        updated_private_message.send_delete(&user, &conn)?;
+      } else {
+        updated_private_message.send_undo_delete(&user, &conn)?;
+      }
+    } else {
+      updated_private_message.send_update(&user, &conn)?;
+    }
+
     let message = PrivateMessageView::read(&conn, data.edit_id)?;
 
-    Ok(PrivateMessageResponse { message })
+    let res = PrivateMessageResponse { message };
+
+    if let Some(ws) = websocket_info {
+      ws.chatserver.do_send(SendUserRoomMessage {
+        op: UserOperation::EditPrivateMessage,
+        response: res.clone(),
+        recipient_id: orig_private_message.recipient_id,
+        my_id: ws.id,
+      });
+    }
+
+    Ok(res)
   }
 }
 

server/src/apub/mod.rs

@@ -5,6 +5,7 @@ pub mod community_inbox;
 pub mod extensions;
 pub mod fetcher;
 pub mod post;
+pub mod private_message;
 pub mod shared_inbox;
 pub mod user;
 pub mod user_inbox;
@@ -39,6 +40,7 @@ use url::Url;
 use crate::api::comment::CommentResponse;
 use crate::api::post::PostResponse;
 use crate::api::site::SearchResponse;
+use crate::api::user::PrivateMessageResponse;
 use crate::db::comment::{Comment, CommentForm, CommentLike, CommentLikeForm};
 use crate::db::comment_view::CommentView;
 use crate::db::community::{
@@ -48,13 +50,15 @@ use crate::db::community::{
 use crate::db::community_view::{CommunityFollowerView, CommunityModeratorView, CommunityView};
 use crate::db::post::{Post, PostForm, PostLike, PostLikeForm};
 use crate::db::post_view::PostView;
+use crate::db::private_message::{PrivateMessage, PrivateMessageForm};
+use crate::db::private_message_view::PrivateMessageView;
 use crate::db::user::{UserForm, User_};
 use crate::db::user_view::UserView;
 use crate::db::{activity, Crud, Followable, Joinable, Likeable, SearchType};
 use crate::routes::nodeinfo::{NodeInfo, NodeInfoWellKnown};
 use crate::routes::{ChatServerParam, DbPoolParam};
 use crate::websocket::{
-  server::{SendComment, SendPost},
+  server::{SendComment, SendPost, SendUserRoomMessage},
   UserOperation,
 };
 use crate::{convert_datetime, naive_now, Settings};
@@ -78,6 +82,7 @@ pub enum EndpointType {
   User,
   Post,
   Comment,
+  PrivateMessage,
 }
 
 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
@@ -113,6 +118,7 @@ pub fn make_apub_endpoint(endpoint_type: EndpointType, name: &str) -> Url {
     // TODO I have to change this else my update advanced_migrations crashes the
     // server if a comment exists.
     EndpointType::Comment => "comment",
+    EndpointType::PrivateMessage => "private_message",
   };
 
   Url::parse(&format!(

server/src/apub/private_message.rs

@@ -0,0 +1,234 @@
+use super::*;
+
+impl ToApub for PrivateMessage {
+  type Response = Note;
+
+  fn to_apub(&self, conn: &PgConnection) -> Result<Note, Error> {
+    let mut private_message = Note::default();
+    let oprops: &mut ObjectProperties = private_message.as_mut();
+    let creator = User_::read(&conn, self.creator_id)?;
+    let recipient = User_::read(&conn, self.recipient_id)?;
+
+    oprops
+      .set_context_xsd_any_uri(context())?
+      .set_id(self.ap_id.to_owned())?
+      .set_published(convert_datetime(self.published))?
+      .set_content_xsd_string(self.content.to_owned())?
+      .set_to_xsd_any_uri(recipient.actor_id)?
+      .set_attributed_to_xsd_any_uri(creator.actor_id)?;
+
+    if let Some(u) = self.updated {
+      oprops.set_updated(convert_datetime(u))?;
+    }
+
+    Ok(private_message)
+  }
+
+  fn to_tombstone(&self) -> Result<Tombstone, Error> {
+    create_tombstone(
+      self.deleted,
+      &self.ap_id,
+      self.updated,
+      NoteType.to_string(),
+    )
+  }
+}
+
+impl FromApub for PrivateMessageForm {
+  type ApubType = Note;
+
+  /// Parse an ActivityPub note received from another instance into a Lemmy Private message
+  fn from_apub(note: &Note, conn: &PgConnection) -> Result<PrivateMessageForm, Error> {
+    let oprops = &note.object_props;
+    let creator_actor_id = &oprops.get_attributed_to_xsd_any_uri().unwrap().to_string();
+    let creator = get_or_fetch_and_upsert_remote_user(&creator_actor_id, &conn)?;
+    let recipient_actor_id = &oprops.get_to_xsd_any_uri().unwrap().to_string();
+    let recipient = get_or_fetch_and_upsert_remote_user(&recipient_actor_id, &conn)?;
+
+    Ok(PrivateMessageForm {
+      creator_id: creator.id,
+      recipient_id: recipient.id,
+      content: oprops
+        .get_content_xsd_string()
+        .map(|c| c.to_string())
+        .unwrap(),
+      published: oprops
+        .get_published()
+        .map(|u| u.as_ref().to_owned().naive_local()),
+      updated: oprops
+        .get_updated()
+        .map(|u| u.as_ref().to_owned().naive_local()),
+      deleted: None,
+      read: None,
+      ap_id: oprops.get_id().unwrap().to_string(),
+      local: false,
+    })
+  }
+}
+
+impl ApubObjectType for PrivateMessage {
+  /// Send out information about a newly created private message
+  fn send_create(&self, creator: &User_, conn: &PgConnection) -> Result<(), Error> {
+    let note = self.to_apub(conn)?;
+    let id = format!("{}/create/{}", self.ap_id, uuid::Uuid::new_v4());
+    let recipient = User_::read(&conn, self.recipient_id)?;
+
+    let mut create = Create::new();
+    create
+      .object_props
+      .set_context_xsd_any_uri(context())?
+      .set_id(id)?;
+    let to = format!("{}/inbox", recipient.actor_id);
+
+    create
+      .create_props
+      .set_actor_xsd_any_uri(creator.actor_id.to_owned())?
+      .set_object_base_box(note)?;
+
+    // Insert the sent activity into the activity table
+    let activity_form = activity::ActivityForm {
+      user_id: creator.id,
+      data: serde_json::to_value(&create)?,
+      local: true,
+      updated: None,
+    };
+    activity::Activity::create(&conn, &activity_form)?;
+
+    send_activity(
+      &create,
+      &creator.private_key.as_ref().unwrap(),
+      &creator.actor_id,
+      vec![to],
+    )?;
+    Ok(())
+  }
+
+  /// Send out information about an edited post, to the followers of the community.
+  fn send_update(&self, creator: &User_, conn: &PgConnection) -> Result<(), Error> {
+    let note = self.to_apub(conn)?;
+    let id = format!("{}/update/{}", self.ap_id, uuid::Uuid::new_v4());
+    let recipient = User_::read(&conn, self.recipient_id)?;
+
+    let mut update = Update::new();
+    update
+      .object_props
+      .set_context_xsd_any_uri(context())?
+      .set_id(id)?;
+    let to = format!("{}/inbox", recipient.actor_id);
+
+    update
+      .update_props
+      .set_actor_xsd_any_uri(creator.actor_id.to_owned())?
+      .set_object_base_box(note)?;
+
+    // Insert the sent activity into the activity table
+    let activity_form = activity::ActivityForm {
+      user_id: creator.id,
+      data: serde_json::to_value(&update)?,
+      local: true,
+      updated: None,
+    };
+    activity::Activity::create(&conn, &activity_form)?;
+
+    send_activity(
+      &update,
+      &creator.private_key.as_ref().unwrap(),
+      &creator.actor_id,
+      vec![to],
+    )?;
+    Ok(())
+  }
+
+  fn send_delete(&self, creator: &User_, conn: &PgConnection) -> Result<(), Error> {
+    let note = self.to_apub(conn)?;
+    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
+    let recipient = User_::read(&conn, self.recipient_id)?;
+
+    let mut delete = Delete::new();
+    delete
+      .object_props
+      .set_context_xsd_any_uri(context())?
+      .set_id(id)?;
+    let to = format!("{}/inbox", recipient.actor_id);
+
+    delete
+      .delete_props
+      .set_actor_xsd_any_uri(creator.actor_id.to_owned())?
+      .set_object_base_box(note)?;
+
+    // Insert the sent activity into the activity table
+    let activity_form = activity::ActivityForm {
+      user_id: creator.id,
+      data: serde_json::to_value(&delete)?,
+      local: true,
+      updated: None,
+    };
+    activity::Activity::create(&conn, &activity_form)?;
+
+    send_activity(
+      &delete,
+      &creator.private_key.as_ref().unwrap(),
+      &creator.actor_id,
+      vec![to],
+    )?;
+    Ok(())
+  }
+
+  fn send_undo_delete(&self, creator: &User_, conn: &PgConnection) -> Result<(), Error> {
+    let note = self.to_apub(conn)?;
+    let id = format!("{}/delete/{}", self.ap_id, uuid::Uuid::new_v4());
+    let recipient = User_::read(&conn, self.recipient_id)?;
+
+    let mut delete = Delete::new();
+    delete
+      .object_props
+      .set_context_xsd_any_uri(context())?
+      .set_id(id)?;
+    let to = format!("{}/inbox", recipient.actor_id);
+
+    delete
+      .delete_props
+      .set_actor_xsd_any_uri(creator.actor_id.to_owned())?
+      .set_object_base_box(note)?;
+
+    // TODO
+    // Undo that fake activity
+    let undo_id = format!("{}/undo/delete/{}", self.ap_id, uuid::Uuid::new_v4());
+    let mut undo = Undo::default();
+
+    undo
+      .object_props
+      .set_context_xsd_any_uri(context())?
+      .set_id(undo_id)?;
+
+    undo
+      .undo_props
+      .set_actor_xsd_any_uri(creator.actor_id.to_owned())?
+      .set_object_base_box(delete)?;
+
+    // Insert the sent activity into the activity table
+    let activity_form = activity::ActivityForm {
+      user_id: creator.id,
+      data: serde_json::to_value(&undo)?,
+      local: true,
+      updated: None,
+    };
+    activity::Activity::create(&conn, &activity_form)?;
+
+    send_activity(
+      &undo,
+      &creator.private_key.as_ref().unwrap(),
+      &creator.actor_id,
+      vec![to],
+    )?;
+    Ok(())
+  }
+
+  fn send_remove(&self, _mod_: &User_, _conn: &PgConnection) -> Result<(), Error> {
+    unimplemented!()
+  }
+
+  fn send_undo_remove(&self, _mod_: &User_, _conn: &PgConnection) -> Result<(), Error> {
+    unimplemented!()
+  }
+}

server/src/apub/user_inbox.rs

@@ -3,7 +3,11 @@ use super::*;
 #[serde(untagged)]
 #[derive(Deserialize, Debug)]
 pub enum UserAcceptedObjects {
-  Accept(Accept),
+  Accept(Box<Accept>),
+  Create(Box<Create>),
+  Update(Box<Update>),
+  Delete(Box<Delete>),
+  Undo(Box<Undo>),
 }
 
 /// Handler for all incoming activities to user inboxes.
@@ -12,7 +16,7 @@ pub async fn user_inbox(
   input: web::Json<UserAcceptedObjects>,
   path: web::Path<String>,
   db: DbPoolParam,
-  _chat_server: ChatServerParam,
+  chat_server: ChatServerParam,
 ) -> Result<HttpResponse, Error> {
   // TODO: would be nice if we could do the signature check here, but we cant access the actor property
   let input = input.into_inner();
@@ -21,12 +25,24 @@ pub async fn user_inbox(
   debug!("User {} received activity: {:?}", &username, &input);
 
   match input {
-    UserAcceptedObjects::Accept(a) => handle_accept(&a, &request, &username, &conn),
+    UserAcceptedObjects::Accept(a) => receive_accept(&a, &request, &username, &conn),
+    UserAcceptedObjects::Create(c) => {
+      receive_create_private_message(&c, &request, &conn, chat_server)
+    }
+    UserAcceptedObjects::Update(u) => {
+      receive_update_private_message(&u, &request, &conn, chat_server)
+    }
+    UserAcceptedObjects::Delete(d) => {
+      receive_delete_private_message(&d, &request, &conn, chat_server)
+    }
+    UserAcceptedObjects::Undo(u) => {
+      receive_undo_delete_private_message(&u, &request, &conn, chat_server)
+    }
   }
 }
 
 /// Handle accepted follows.
-fn handle_accept(
+fn receive_accept(
   accept: &Accept,
   request: &HttpRequest,
   username: &str,
@@ -65,3 +81,240 @@ fn handle_accept(
   // TODO: at this point, indicate to the user that they are following the community
   Ok(HttpResponse::Ok().finish())
 }
+
+fn receive_create_private_message(
+  create: &Create,
+  request: &HttpRequest,
+  conn: &PgConnection,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, Error> {
+  let note = create
+    .create_props
+    .get_object_base_box()
+    .to_owned()
+    .unwrap()
+    .to_owned()
+    .into_concrete::<Note>()?;
+
+  let user_uri = create
+    .create_props
+    .get_actor_xsd_any_uri()
+    .unwrap()
+    .to_string();
+
+  let user = get_or_fetch_and_upsert_remote_user(&user_uri, &conn)?;
+  verify(request, &user.public_key.unwrap())?;
+
+  // Insert the received activity into the activity table
+  let activity_form = activity::ActivityForm {
+    user_id: user.id,
+    data: serde_json::to_value(&create)?,
+    local: false,
+    updated: None,
+  };
+  activity::Activity::create(&conn, &activity_form)?;
+
+  let private_message = PrivateMessageForm::from_apub(&note, &conn)?;
+  let inserted_private_message = PrivateMessage::create(&conn, &private_message)?;
+
+  let message = PrivateMessageView::read(&conn, inserted_private_message.id)?;
+
+  let res = PrivateMessageResponse {
+    message: message.to_owned(),
+  };
+
+  chat_server.do_send(SendUserRoomMessage {
+    op: UserOperation::CreatePrivateMessage,
+    response: res,
+    recipient_id: message.recipient_id,
+    my_id: None,
+  });
+
+  Ok(HttpResponse::Ok().finish())
+}
+
+fn receive_update_private_message(
+  update: &Update,
+  request: &HttpRequest,
+  conn: &PgConnection,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, Error> {
+  let note = update
+    .update_props
+    .get_object_base_box()
+    .to_owned()
+    .unwrap()
+    .to_owned()
+    .into_concrete::<Note>()?;
+
+  let user_uri = update
+    .update_props
+    .get_actor_xsd_any_uri()
+    .unwrap()
+    .to_string();
+
+  let user = get_or_fetch_and_upsert_remote_user(&user_uri, &conn)?;
+  verify(request, &user.public_key.unwrap())?;
+
+  // Insert the received activity into the activity table
+  let activity_form = activity::ActivityForm {
+    user_id: user.id,
+    data: serde_json::to_value(&update)?,
+    local: false,
+    updated: None,
+  };
+  activity::Activity::create(&conn, &activity_form)?;
+
+  let private_message = PrivateMessageForm::from_apub(&note, &conn)?;
+  let private_message_id = PrivateMessage::read_from_apub_id(&conn, &private_message.ap_id)?.id;
+  PrivateMessage::update(conn, private_message_id, &private_message)?;
+
+  let message = PrivateMessageView::read(&conn, private_message_id)?;
+
+  let res = PrivateMessageResponse {
+    message: message.to_owned(),
+  };
+
+  chat_server.do_send(SendUserRoomMessage {
+    op: UserOperation::EditPrivateMessage,
+    response: res,
+    recipient_id: message.recipient_id,
+    my_id: None,
+  });
+
+  Ok(HttpResponse::Ok().finish())
+}
+
+fn receive_delete_private_message(
+  delete: &Delete,
+  request: &HttpRequest,
+  conn: &PgConnection,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, Error> {
+  let note = delete
+    .delete_props
+    .get_object_base_box()
+    .to_owned()
+    .unwrap()
+    .to_owned()
+    .into_concrete::<Note>()?;
+
+  let user_uri = delete
+    .delete_props
+    .get_actor_xsd_any_uri()
+    .unwrap()
+    .to_string();
+
+  let user = get_or_fetch_and_upsert_remote_user(&user_uri, &conn)?;
+  verify(request, &user.public_key.unwrap())?;
+
+  // Insert the received activity into the activity table
+  let activity_form = activity::ActivityForm {
+    user_id: user.id,
+    data: serde_json::to_value(&delete)?,
+    local: false,
+    updated: None,
+  };
+  activity::Activity::create(&conn, &activity_form)?;
+
+  let private_message = PrivateMessageForm::from_apub(&note, &conn)?;
+  let private_message_id = PrivateMessage::read_from_apub_id(&conn, &private_message.ap_id)?.id;
+  let private_message_form = PrivateMessageForm {
+    content: private_message.content,
+    recipient_id: private_message.recipient_id,
+    creator_id: private_message.creator_id,
+    deleted: Some(true),
+    read: None,
+    ap_id: private_message.ap_id,
+    local: private_message.local,
+    published: None,
+    updated: Some(naive_now()),
+  };
+  PrivateMessage::update(conn, private_message_id, &private_message_form)?;
+
+  let message = PrivateMessageView::read(&conn, private_message_id)?;
+
+  let res = PrivateMessageResponse {
+    message: message.to_owned(),
+  };
+
+  chat_server.do_send(SendUserRoomMessage {
+    op: UserOperation::EditPrivateMessage,
+    response: res,
+    recipient_id: message.recipient_id,
+    my_id: None,
+  });
+
+  Ok(HttpResponse::Ok().finish())
+}
+
+fn receive_undo_delete_private_message(
+  undo: &Undo,
+  request: &HttpRequest,
+  conn: &PgConnection,
+  chat_server: ChatServerParam,
+) -> Result<HttpResponse, Error> {
+  let delete = undo
+    .undo_props
+    .get_object_base_box()
+    .to_owned()
+    .unwrap()
+    .to_owned()
+    .into_concrete::<Delete>()?;
+
+  let note = delete
+    .delete_props
+    .get_object_base_box()
+    .to_owned()
+    .unwrap()
+    .to_owned()
+    .into_concrete::<Note>()?;
+
+  let user_uri = delete
+    .delete_props
+    .get_actor_xsd_any_uri()
+    .unwrap()
+    .to_string();
+
+  let user = get_or_fetch_and_upsert_remote_user(&user_uri, &conn)?;
+  verify(request, &user.public_key.unwrap())?;
+
+  // Insert the received activity into the activity table
+  let activity_form = activity::ActivityForm {
+    user_id: user.id,
+    data: serde_json::to_value(&delete)?,
+    local: false,
+    updated: None,
+  };
+  activity::Activity::create(&conn, &activity_form)?;
+
+  let private_message = PrivateMessageForm::from_apub(&note, &conn)?;
+  let private_message_id = PrivateMessage::read_from_apub_id(&conn, &private_message.ap_id)?.id;
+  let private_message_form = PrivateMessageForm {
+    content: private_message.content,
+    recipient_id: private_message.recipient_id,
+    creator_id: private_message.creator_id,
+    deleted: Some(false),
+    read: None,
+    ap_id: private_message.ap_id,
+    local: private_message.local,
+    published: None,
+    updated: Some(naive_now()),
+  };
+  PrivateMessage::update(conn, private_message_id, &private_message_form)?;
+
+  let message = PrivateMessageView::read(&conn, private_message_id)?;
+
+  let res = PrivateMessageResponse {
+    message: message.to_owned(),
+  };
+
+  chat_server.do_send(SendUserRoomMessage {
+    op: UserOperation::EditPrivateMessage,
+    response: res,
+    recipient_id: message.recipient_id,
+    my_id: None,
+  });
+
+  Ok(HttpResponse::Ok().finish())
+}

server/src/db/code_migrations.rs

@@ -2,6 +2,7 @@
 use super::comment::Comment;
 use super::community::{Community, CommunityForm};
 use super::post::Post;
+use super::private_message::PrivateMessage;
 use super::user::{UserForm, User_};
 use super::*;
 use crate::apub::extensions::signatures::generate_actor_keypair;
@@ -15,6 +16,7 @@ pub fn run_advanced_migrations(conn: &PgConnection) -> Result<(), Error> {
   community_updates_2020_04_02(conn)?;
   post_updates_2020_04_03(conn)?;
   comment_updates_2020_04_03(conn)?;
+  private_message_updates_2020_05_05(conn)?;
 
   Ok(())
 }
@@ -145,3 +147,23 @@ fn comment_updates_2020_04_03(conn: &PgConnection) -> Result<(), Error> {
 
   Ok(())
 }
+
+fn private_message_updates_2020_05_05(conn: &PgConnection) -> Result<(), Error> {
+  use crate::schema::private_message::dsl::*;
+
+  info!("Running private_message_updates_2020_05_05");
+
+  // Update the ap_id
+  let incorrect_pms = private_message
+    .filter(ap_id.eq("changeme"))
+    .filter(local.eq(true))
+    .load::<PrivateMessage>(conn)?;
+
+  for cpm in &incorrect_pms {
+    PrivateMessage::update_ap_id(&conn, cpm.id)?;
+  }
+
+  info!("{} private message rows updated.", incorrect_pms.len());
+
+  Ok(())
+}

server/src/db/private_message.rs

@@ -1,4 +1,5 @@
 use super::*;
+use crate::apub::{make_apub_endpoint, EndpointType};
 use crate::schema::private_message;
 
 #[derive(Queryable, Identifiable, PartialEq, Debug, Serialize, Deserialize)]
@@ -12,6 +13,8 @@ pub struct PrivateMessage {
   pub read: bool,
   pub published: chrono::NaiveDateTime,
   pub updated: Option<chrono::NaiveDateTime>,
+  pub ap_id: String,
+  pub local: bool,
 }
 
 #[derive(Insertable, AsChangeset, Clone)]
@@ -19,10 +22,13 @@ pub struct PrivateMessage {
 pub struct PrivateMessageForm {
   pub creator_id: i32,
   pub recipient_id: i32,
-  pub content: Option<String>,
+  pub content: String,
   pub deleted: Option<bool>,
   pub read: Option<bool>,
+  pub published: Option<chrono::NaiveDateTime>,
   pub updated: Option<chrono::NaiveDateTime>,
+  pub ap_id: String,
+  pub local: bool,
 }
 
 impl Crud<PrivateMessageForm> for PrivateMessage {
@@ -55,6 +61,28 @@ impl Crud<PrivateMessageForm> for PrivateMessage {
   }
 }
 
+impl PrivateMessage {
+  pub fn update_ap_id(conn: &PgConnection, private_message_id: i32) -> Result<Self, Error> {
+    use crate::schema::private_message::dsl::*;
+
+    let apid = make_apub_endpoint(
+      EndpointType::PrivateMessage,
+      &private_message_id.to_string(),
+    )
+    .to_string();
+    diesel::update(private_message.find(private_message_id))
+      .set(ap_id.eq(apid))
+      .get_result::<Self>(conn)
+  }
+
+  pub fn read_from_apub_id(conn: &PgConnection, object_id: &str) -> Result<Self, Error> {
+    use crate::schema::private_message::dsl::*;
+    private_message
+      .filter(ap_id.eq(object_id))
+      .first::<Self>(conn)
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use super::super::user::*;
@@ -118,12 +146,15 @@ mod tests {
     let inserted_recipient = User_::create(&conn, &recipient_form).unwrap();
 
     let private_message_form = PrivateMessageForm {
-      content: Some("A test private message".into()),
+      content: "A test private message".into(),
       creator_id: inserted_creator.id,
       recipient_id: inserted_recipient.id,
       deleted: None,
       read: None,
+      published: None,
       updated: None,
+      ap_id: "changeme".into(),
+      local: true,
     };
 
     let inserted_private_message = PrivateMessage::create(&conn, &private_message_form).unwrap();
@@ -137,6 +168,8 @@ mod tests {
       read: false,
       updated: None,
       published: inserted_private_message.published,
+      ap_id: "changeme".into(),
+      local: true,
     };
 
     let read_private_message = PrivateMessage::read(&conn, inserted_private_message.id).unwrap();

server/src/db/private_message_view.rs

@@ -12,10 +12,16 @@ table! {
     read -> Bool,
     published -> Timestamp,
     updated -> Nullable<Timestamp>,
+    ap_id -> Text,
+    local -> Bool,
     creator_name -> Varchar,
     creator_avatar -> Nullable<Text>,
+    creator_actor_id -> Text,
+    creator_local -> Bool,
     recipient_name -> Varchar,
     recipient_avatar -> Nullable<Text>,
+    recipient_actor_id -> Text,
+    recipient_local -> Bool,
   }
 }
 
@@ -29,10 +35,16 @@ table! {
     read -> Bool,
     published -> Timestamp,
     updated -> Nullable<Timestamp>,
+    ap_id -> Text,
+    local -> Bool,
     creator_name -> Varchar,
     creator_avatar -> Nullable<Text>,
+    creator_actor_id -> Text,
+    creator_local -> Bool,
     recipient_name -> Varchar,
     recipient_avatar -> Nullable<Text>,
+    recipient_actor_id -> Text,
+    recipient_local -> Bool,
   }
 }
 
@@ -49,10 +61,16 @@ pub struct PrivateMessageView {
   pub read: bool,
   pub published: chrono::NaiveDateTime,
   pub updated: Option<chrono::NaiveDateTime>,
+  pub ap_id: String,
+  pub local: bool,
   pub creator_name: String,
   pub creator_avatar: Option<String>,
+  pub creator_actor_id: String,
+  pub creator_local: bool,
   pub recipient_name: String,
   pub recipient_avatar: Option<String>,
+  pub recipient_actor_id: String,
+  pub recipient_local: bool,
 }
 
 pub struct PrivateMessageQueryBuilder<'a> {

server/src/routes/api.rs

@@ -83,6 +83,14 @@ pub fn config(cfg: &mut web::ServiceConfig, rate_limit: &RateLimit) {
           .route("/like", web::post().to(route_post::<CreateCommentLike>))
           .route("/save", web::put().to(route_post::<SaveComment>)),
       )
+      // Private Message
+      .service(
+        web::scope("/private_message")
+          .wrap(rate_limit.message())
+          .route("/list", web::get().to(route_get::<GetPrivateMessages>))
+          .route("", web::post().to(route_post::<CreatePrivateMessage>))
+          .route("", web::put().to(route_post::<EditPrivateMessage>)),
+      )
       // User
       .service(
         // Account action, I don't like that it's in /user maybe /accounts

server/src/schema.rs

@@ -272,6 +272,8 @@ table! {
         read -> Bool,
         published -> Timestamp,
         updated -> Nullable<Timestamp>,
+        ap_id -> Varchar,
+        local -> Bool,
     }
 }
 

ui/src/api_tests/api.spec.ts

@@ -18,6 +18,10 @@ import {
   GetCommunityResponse,
   CommentLikeForm,
   CreatePostLikeForm,
+  PrivateMessageForm,
+  EditPrivateMessageForm,
+  PrivateMessageResponse,
+  PrivateMessagesResponse,
 } from '../interfaces';
 
 let lemmyAlphaUrl = 'http://localhost:8540';
@@ -158,6 +162,7 @@ describe('main', () => {
           body: wrapper(unfollowForm),
         }
       ).then(d => d.json());
+      expect(unfollowRes.community.local).toBe(false);
 
       // Check that you are unsubscribed to it locally
       let followedCommunitiesResAgain: GetFollowedCommunitiesResponse = await fetch(
@@ -965,6 +970,143 @@ describe('main', () => {
       expect(getCommunityResAgain.community.removed).toBe(false);
     });
   });
+
+  describe('private message', () => {
+    test('/u/lemmy_alpha creates/updates/deletes/undeletes a private_message to /u/lemmy_beta, its on both instances', async () => {
+      let content = 'A jest test federated private message';
+      let privateMessageForm: PrivateMessageForm = {
+        content,
+        recipient_id: 3,
+        auth: lemmyAlphaAuth,
+      };
+
+      let createRes: PrivateMessageResponse = await fetch(
+        `${lemmyAlphaApiUrl}/private_message`,
+        {
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: wrapper(privateMessageForm),
+        }
+      ).then(d => d.json());
+      expect(createRes.message.content).toBe(content);
+      expect(createRes.message.local).toBe(true);
+      expect(createRes.message.creator_local).toBe(true);
+      expect(createRes.message.recipient_local).toBe(false);
+
+      // Get it from beta
+      let getPrivateMessagesUrl = `${lemmyBetaApiUrl}/private_message/list?auth=${lemmyBetaAuth}&unread_only=false`;
+
+      let getPrivateMessagesRes: PrivateMessagesResponse = await fetch(
+        getPrivateMessagesUrl,
+        {
+          method: 'GET',
+        }
+      ).then(d => d.json());
+
+      expect(getPrivateMessagesRes.messages[0].content).toBe(content);
+      expect(getPrivateMessagesRes.messages[0].local).toBe(false);
+      expect(getPrivateMessagesRes.messages[0].creator_local).toBe(false);
+      expect(getPrivateMessagesRes.messages[0].recipient_local).toBe(true);
+
+      // lemmy alpha updates the private message
+      let updatedContent = 'A jest test federated private message edited';
+      let updatePrivateMessageForm: EditPrivateMessageForm = {
+        content: updatedContent,
+        edit_id: createRes.message.id,
+        auth: lemmyAlphaAuth,
+      };
+
+      let updateRes: PrivateMessageResponse = await fetch(
+        `${lemmyAlphaApiUrl}/private_message`,
+        {
+          method: 'PUT',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: wrapper(updatePrivateMessageForm),
+        }
+      ).then(d => d.json());
+
+      expect(updateRes.message.content).toBe(updatedContent);
+
+      // Fetch from beta again
+      let getPrivateMessagesUpdatedRes: PrivateMessagesResponse = await fetch(
+        getPrivateMessagesUrl,
+        {
+          method: 'GET',
+        }
+      ).then(d => d.json());
+
+      expect(getPrivateMessagesUpdatedRes.messages[0].content).toBe(
+        updatedContent
+      );
+
+      // lemmy alpha deletes the private message
+      let deletePrivateMessageForm: EditPrivateMessageForm = {
+        deleted: true,
+        edit_id: createRes.message.id,
+        auth: lemmyAlphaAuth,
+      };
+
+      let deleteRes: PrivateMessageResponse = await fetch(
+        `${lemmyAlphaApiUrl}/private_message`,
+        {
+          method: 'PUT',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: wrapper(deletePrivateMessageForm),
+        }
+      ).then(d => d.json());
+
+      expect(deleteRes.message.deleted).toBe(true);
+
+      // Fetch from beta again
+      let getPrivateMessagesDeletedRes: PrivateMessagesResponse = await fetch(
+        getPrivateMessagesUrl,
+        {
+          method: 'GET',
+        }
+      ).then(d => d.json());
+
+      // The GetPrivateMessages filters out deleted,
+      // even though they are in the actual database.
+      // no reason to show them
+      expect(getPrivateMessagesDeletedRes.messages.length).toBe(0);
+
+      // lemmy alpha undeletes the private message
+      let undeletePrivateMessageForm: EditPrivateMessageForm = {
+        deleted: false,
+        edit_id: createRes.message.id,
+        auth: lemmyAlphaAuth,
+      };
+
+      let undeleteRes: PrivateMessageResponse = await fetch(
+        `${lemmyAlphaApiUrl}/private_message`,
+        {
+          method: 'PUT',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: wrapper(undeletePrivateMessageForm),
+        }
+      ).then(d => d.json());
+
+      expect(undeleteRes.message.deleted).toBe(false);
+
+      // Fetch from beta again
+      let getPrivateMessagesUnDeletedRes: PrivateMessagesResponse = await fetch(
+        getPrivateMessagesUrl,
+        {
+          method: 'GET',
+        }
+      ).then(d => d.json());
+
+      expect(getPrivateMessagesUnDeletedRes.messages[0].deleted).toBe(false);
+    });
+  });
 });
 
 function wrapper(form: any): string {

ui/src/interfaces.ts

@@ -273,10 +273,16 @@ export interface PrivateMessage {
   read: boolean;
   published: string;
   updated?: string;
+  ap_id: string;
+  local: boolean;
   creator_name: string;
   creator_avatar?: string;
+  creator_actor_id: string;
+  creator_local: boolean;
   recipient_name: string;
   recipient_avatar?: string;
+  recipient_actor_id: string;
+  recipient_local: boolean;
 }
 
 export enum BanType {