src/apub/inbox/user_inbox.rs

   1 use crate::{
   2   apub::{
   3     check_is_apub_id_valid,
   4     extensions::signatures::verify,
   5     fetcher::{get_or_fetch_and_upsert_actor, get_or_fetch_and_upsert_community},
   6     insert_activity,
   7     FromApub,
   8   },
   9   LemmyContext,
  10 };
  11 use activitystreams::{
  12   activity::{Accept, ActorAndObject, Create, Delete, Undo, Update},
  13   base::AnyBase,
  14   object::Note,
  15   prelude::*,
  16 };
  17 use actix_web::{web, HttpRequest, HttpResponse};
  18 use anyhow::Context;
  19 use lemmy_db::{
  20   community::{CommunityFollower, CommunityFollowerForm},
  21   naive_now,
  22   private_message::{PrivateMessage, PrivateMessageForm},
  23   private_message_view::PrivateMessageView,
  24   user::User_,
  25   Crud,
  26   Followable,
  27 };
  28 use lemmy_structs::{
  29   blocking,
  30   user::PrivateMessageResponse,
  31   websocket::{SendUserRoomMessage, UserOperation},
  32 };
  33 use lemmy_utils::{location_info, LemmyError};
  34 use log::debug;
  35 use serde::{Deserialize, Serialize};
  36 use std::fmt::Debug;
  37
  38 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Deserialize, Serialize)]
  39 #[serde(rename_all = "PascalCase")]
  40 pub enum ValidTypes {
  41   Accept,
  42   Create,
  43   Update,
  44   Delete,
  45   Undo,
  46 }
  47
  48 pub type AcceptedActivities = ActorAndObject<ValidTypes>;
  49
  50 /// Handler for all incoming activities to user inboxes.
  51 pub async fn user_inbox(
  52   request: HttpRequest,
  53   input: web::Json<AcceptedActivities>,
  54   path: web::Path<String>,
  55   context: web::Data<LemmyContext>,
  56 ) -> Result<HttpResponse, LemmyError> {
  57   let activity = input.into_inner();
  58   let username = path.into_inner();
  59   debug!("User {} received activity: {:?}", &username, &activity);
  60
  61   let actor_uri = activity
  62     .actor()?
  63     .as_single_xsd_any_uri()
  64     .context(location_info!())?;
  65
  66   check_is_apub_id_valid(actor_uri)?;
  67
  68   let actor = get_or_fetch_and_upsert_actor(actor_uri, &context).await?;
  69   verify(&request, actor.as_ref())?;
  70
  71   let any_base = activity.clone().into_any_base()?;
  72   let kind = activity.kind().context(location_info!())?;
  73   let res = match kind {
  74     ValidTypes::Accept => receive_accept(any_base, username, &context).await,
  75     ValidTypes::Create => receive_create_private_message(any_base, &context).await,
  76     ValidTypes::Update => receive_update_private_message(any_base, &context).await,
  77     ValidTypes::Delete => receive_delete_private_message(any_base, &context).await,
  78     ValidTypes::Undo => receive_undo_delete_private_message(any_base, &context).await,
  79   };
  80
  81   insert_activity(actor.user_id(), activity.clone(), false, context.pool()).await?;
  82   res
  83 }
  84
  85 /// Handle accepted follows.
  86 async fn receive_accept(
  87   activity: AnyBase,
  88   username: String,
  89   context: &LemmyContext,
  90 ) -> Result<HttpResponse, LemmyError> {
  91   let accept = Accept::from_any_base(activity)?.context(location_info!())?;
  92   let community_uri = accept
  93     .actor()?
  94     .to_owned()
  95     .single_xsd_any_uri()
  96     .context(location_info!())?;
  97
  98   let community = get_or_fetch_and_upsert_community(&community_uri, context).await?;
  99
 100   let user = blocking(&context.pool(), move |conn| {
 101     User_::read_from_name(conn, &username)
 102   })
 103   .await??;
 104
 105   // Now you need to add this to the community follower
 106   let community_follower_form = CommunityFollowerForm {
 107     community_id: community.id,
 108     user_id: user.id,
 109   };
 110
 111   // This will fail if they're already a follower
 112   blocking(&context.pool(), move |conn| {
 113     CommunityFollower::follow(conn, &community_follower_form).ok()
 114   })
 115   .await?;
 116
 117   // TODO: make sure that we actually requested a follow
 118   Ok(HttpResponse::Ok().finish())
 119 }
 120
 121 async fn receive_create_private_message(
 122   activity: AnyBase,
 123   context: &LemmyContext,
 124 ) -> Result<HttpResponse, LemmyError> {
 125   let create = Create::from_any_base(activity)?.context(location_info!())?;
 126   let note = Note::from_any_base(
 127     create
 128       .object()
 129       .as_one()
 130       .context(location_info!())?
 131       .to_owned(),
 132   )?
 133   .context(location_info!())?;
 134
 135   let domain = Some(create.id_unchecked().context(location_info!())?.to_owned());
 136   let private_message = PrivateMessageForm::from_apub(&note, context, domain).await?;
 137
 138   let inserted_private_message = blocking(&context.pool(), move |conn| {
 139     PrivateMessage::create(conn, &private_message)
 140   })
 141   .await??;
 142
 143   let message = blocking(&context.pool(), move |conn| {
 144     PrivateMessageView::read(conn, inserted_private_message.id)
 145   })
 146   .await??;
 147
 148   let res = PrivateMessageResponse { message };
 149
 150   let recipient_id = res.message.recipient_id;
 151
 152   context.chat_server().do_send(SendUserRoomMessage {
 153     op: UserOperation::CreatePrivateMessage,
 154     response: res,
 155     recipient_id,
 156     websocket_id: None,
 157   });
 158
 159   Ok(HttpResponse::Ok().finish())
 160 }
 161
 162 async fn receive_update_private_message(
 163   activity: AnyBase,
 164   context: &LemmyContext,
 165 ) -> Result<HttpResponse, LemmyError> {
 166   let update = Update::from_any_base(activity)?.context(location_info!())?;
 167   let note = Note::from_any_base(
 168     update
 169       .object()
 170       .as_one()
 171       .context(location_info!())?
 172       .to_owned(),
 173   )?
 174   .context(location_info!())?;
 175
 176   let domain = Some(update.id_unchecked().context(location_info!())?.to_owned());
 177   let private_message_form = PrivateMessageForm::from_apub(&note, context, domain).await?;
 178
 179   let private_message_ap_id = private_message_form
 180     .ap_id
 181     .as_ref()
 182     .context(location_info!())?
 183     .clone();
 184   let private_message = blocking(&context.pool(), move |conn| {
 185     PrivateMessage::read_from_apub_id(conn, &private_message_ap_id)
 186   })
 187   .await??;
 188
 189   let private_message_id = private_message.id;
 190   blocking(&context.pool(), move |conn| {
 191     PrivateMessage::update(conn, private_message_id, &private_message_form)
 192   })
 193   .await??;
 194
 195   let private_message_id = private_message.id;
 196   let message = blocking(&context.pool(), move |conn| {
 197     PrivateMessageView::read(conn, private_message_id)
 198   })
 199   .await??;
 200
 201   let res = PrivateMessageResponse { message };
 202
 203   let recipient_id = res.message.recipient_id;
 204
 205   context.chat_server().do_send(SendUserRoomMessage {
 206     op: UserOperation::EditPrivateMessage,
 207     response: res,
 208     recipient_id,
 209     websocket_id: None,
 210   });
 211
 212   Ok(HttpResponse::Ok().finish())
 213 }
 214
 215 async fn receive_delete_private_message(
 216   activity: AnyBase,
 217   context: &LemmyContext,
 218 ) -> Result<HttpResponse, LemmyError> {
 219   let delete = Delete::from_any_base(activity)?.context(location_info!())?;
 220   let note = Note::from_any_base(
 221     delete
 222       .object()
 223       .as_one()
 224       .context(location_info!())?
 225       .to_owned(),
 226   )?
 227   .context(location_info!())?;
 228
 229   let domain = Some(delete.id_unchecked().context(location_info!())?.to_owned());
 230   let private_message_form = PrivateMessageForm::from_apub(&note, context, domain).await?;
 231
 232   let private_message_ap_id = private_message_form.ap_id.context(location_info!())?;
 233   let private_message = blocking(&context.pool(), move |conn| {
 234     PrivateMessage::read_from_apub_id(conn, &private_message_ap_id)
 235   })
 236   .await??;
 237
 238   let private_message_form = PrivateMessageForm {
 239     content: private_message_form.content,
 240     recipient_id: private_message.recipient_id,
 241     creator_id: private_message.creator_id,
 242     deleted: Some(true),
 243     read: None,
 244     ap_id: Some(private_message.ap_id),
 245     local: private_message.local,
 246     published: None,
 247     updated: Some(naive_now()),
 248   };
 249
 250   let private_message_id = private_message.id;
 251   blocking(&context.pool(), move |conn| {
 252     PrivateMessage::update(conn, private_message_id, &private_message_form)
 253   })
 254   .await??;
 255
 256   let private_message_id = private_message.id;
 257   let message = blocking(&context.pool(), move |conn| {
 258     PrivateMessageView::read(&conn, private_message_id)
 259   })
 260   .await??;
 261
 262   let res = PrivateMessageResponse { message };
 263
 264   let recipient_id = res.message.recipient_id;
 265
 266   context.chat_server().do_send(SendUserRoomMessage {
 267     op: UserOperation::EditPrivateMessage,
 268     response: res,
 269     recipient_id,
 270     websocket_id: None,
 271   });
 272
 273   Ok(HttpResponse::Ok().finish())
 274 }
 275
 276 async fn receive_undo_delete_private_message(
 277   activity: AnyBase,
 278   context: &LemmyContext,
 279 ) -> Result<HttpResponse, LemmyError> {
 280   let undo = Undo::from_any_base(activity)?.context(location_info!())?;
 281   let delete = Delete::from_any_base(undo.object().as_one().context(location_info!())?.to_owned())?
 282     .context(location_info!())?;
 283   let note = Note::from_any_base(
 284     delete
 285       .object()
 286       .as_one()
 287       .context(location_info!())?
 288       .to_owned(),
 289   )?
 290   .context(location_info!())?;
 291
 292   let domain = Some(undo.id_unchecked().context(location_info!())?.to_owned());
 293   let private_message = PrivateMessageForm::from_apub(&note, context, domain).await?;
 294
 295   let private_message_ap_id = private_message
 296     .ap_id
 297     .as_ref()
 298     .context(location_info!())?
 299     .clone();
 300   let private_message_id = blocking(&context.pool(), move |conn| {
 301     PrivateMessage::read_from_apub_id(conn, &private_message_ap_id).map(|pm| pm.id)
 302   })
 303   .await??;
 304
 305   let private_message_form = PrivateMessageForm {
 306     content: private_message.content,
 307     recipient_id: private_message.recipient_id,
 308     creator_id: private_message.creator_id,
 309     deleted: Some(false),
 310     read: None,
 311     ap_id: private_message.ap_id,
 312     local: private_message.local,
 313     published: None,
 314     updated: Some(naive_now()),
 315   };
 316
 317   blocking(&context.pool(), move |conn| {
 318     PrivateMessage::update(conn, private_message_id, &private_message_form)
 319   })
 320   .await??;
 321
 322   let message = blocking(&context.pool(), move |conn| {
 323     PrivateMessageView::read(&conn, private_message_id)
 324   })
 325   .await??;
 326
 327   let res = PrivateMessageResponse { message };
 328
 329   let recipient_id = res.message.recipient_id;
 330
 331   context.chat_server().do_send(SendUserRoomMessage {
 332     op: UserOperation::EditPrivateMessage,
 333     response: res,
 334     recipient_id,
 335     websocket_id: None,
 336   });
 337
 338   Ok(HttpResponse::Ok().finish())
 339 }