Rewrite remaining activities (#1712)

[lemmy.git] / crates / apub / src / http / mod.rs

diff --git a/crates/apub/src/http/mod.rs b/crates/apub/src/http/mod.rs
index 477345fd4700f72d5c55ccbd3e1e7e0619f4cdb1..6fe8d182ea88df53b0ebfeabb6bdf2f985e6c6b7 100644 (file)
--- a/crates/apub/src/http/mod.rs
+++ b/crates/apub/src/http/mod.rs
@@ -2,7 +2,10 @@ use crate::{
   check_is_apub_id_valid,
   extensions::signatures::verify_signature,
   fetcher::get_or_fetch_and_upsert_actor,
-  http::inbox_enums::SharedInboxActivities,
+  http::{
+    community::{receive_group_inbox, GroupInboxActivities},
+    person::{receive_person_inbox, PersonInboxActivities},
+  },
   insert_activity,
   APUB_JSON_CONTENT_TYPE,
 };
@@ -17,29 +20,47 @@ use anyhow::{anyhow, Context};
 use futures::StreamExt;
 use http::StatusCode;
 use lemmy_api_common::blocking;
-use lemmy_apub_lib::ActivityHandler;
+use lemmy_apub_lib::{ActivityFields, ActivityHandler};
 use lemmy_db_queries::{source::activity::Activity_, DbPool};
 use lemmy_db_schema::source::activity::Activity;
 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
 use lemmy_websocket::LemmyContext;
+use log::{info, trace};
 use serde::{Deserialize, Serialize};
 use std::{fmt::Debug, io::Read};
 use url::Url;
 
 mod comment;
 mod community;
-mod inbox_enums;
 mod person;
 mod post;
 pub mod routes;
 
+#[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
+#[serde(untagged)]
+pub enum SharedInboxActivities {
+  GroupInboxActivities(GroupInboxActivities),
+  // 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.
+  PersonInboxActivities(PersonInboxActivities),
+}
+
 pub async fn shared_inbox(
   request: HttpRequest,
   payload: Payload,
   context: web::Data<LemmyContext>,
 ) -> Result<HttpResponse, LemmyError> {
   let unparsed = payload_to_string(payload).await?;
-  receive_activity::<SharedInboxActivities>(request, &unparsed, context).await
+  trace!("Received shared inbox activity {}", unparsed);
+  let activity = serde_json::from_str::<SharedInboxActivities>(&unparsed)?;
+  match activity {
+    SharedInboxActivities::GroupInboxActivities(g) => {
+      receive_group_inbox(g, request, &context).await
+    }
+    SharedInboxActivities::PersonInboxActivities(p) => {
+      receive_person_inbox(p, request, &context).await
+    }
+  }
 }
 
 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
@@ -55,36 +76,36 @@ async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
 // TODO: move most of this code to library
 async fn receive_activity<'a, T>(
   request: HttpRequest,
-  activity: &'a str,
-  context: web::Data<LemmyContext>,
+  activity: T,
+  context: &LemmyContext,
 ) -> Result<HttpResponse, LemmyError>
 where
-  T: ActivityHandler + Clone + Deserialize<'a> + Serialize + std::fmt::Debug + Send + 'static,
+  T: ActivityHandler
+    + ActivityFields
+    + Clone
+    + Deserialize<'a>
+    + Serialize
+    + std::fmt::Debug
+    + Send
+    + 'static,
 {
-  let activity = serde_json::from_str::<T>(activity)?;
-  let activity_data = activity.common();
-
   let request_counter = &mut 0;
-  let actor =
-    get_or_fetch_and_upsert_actor(&activity_data.actor, &context, request_counter).await?;
+  let actor = get_or_fetch_and_upsert_actor(activity.actor(), context, request_counter).await?;
   verify_signature(&request, &actor.public_key().context(location_info!())?)?;
 
   // Do nothing if we received the same activity before
-  if is_activity_already_known(context.pool(), activity_data.id_unchecked()).await? {
+  if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
     return Ok(HttpResponse::Ok().finish());
   }
-  check_is_apub_id_valid(&activity_data.actor, false)?;
-  println!(
-    "Verifying activity {}",
-    activity_data.id_unchecked().to_string()
-  );
-  activity.verify(&context, request_counter).await?;
+  check_is_apub_id_valid(activity.actor(), false)?;
+  info!("Verifying activity {}", activity.id_unchecked().to_string());
+  activity.verify(context, request_counter).await?;
   assert_activity_not_local(&activity)?;
 
   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
   // if we receive the same activity twice in very quick succession.
   insert_activity(
-    activity_data.id_unchecked(),
+    activity.id_unchecked(),
     activity.clone(),
     false,
     true,
@@ -92,11 +113,8 @@ where
   )
   .await?;
 
-  println!(
-    "Receiving activity {}",
-    activity_data.id_unchecked().to_string()
-  );
-  activity.receive(&context, request_counter).await?;
+  info!("Receiving activity {}", activity.id_unchecked().to_string());
+  activity.receive(context, request_counter).await?;
   Ok(HttpResponse::Ok().finish())
 }
 
@@ -168,12 +186,8 @@ pub(crate) async fn is_activity_already_known(
   }
 }
 
-fn assert_activity_not_local<T: Debug + ActivityHandler>(activity: &T) -> Result<(), LemmyError> {
-  let activity_domain = activity
-    .common()
-    .id_unchecked()
-    .domain()
-    .context(location_info!())?;
+fn assert_activity_not_local<T: Debug + ActivityFields>(activity: &T) -> Result<(), LemmyError> {
+  let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
 
   if activity_domain == Settings::get().hostname {
     return Err(