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