Split activity table into sent and received parts (fixes #3103) (#3583)

[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 e2fb0b4292acecd6209ee2bc079efff5583f4a2e..c261d9e4929c363385e10756d491967a25f29148 100644 (file)
--- a/crates/apub/src/http/mod.rs
+++ b/crates/apub/src/http/mod.rs
@@ -1,29 +1,22 @@
 use crate::{
   activity_lists::SharedInboxActivities,
   fetcher::user_or_community::UserOrCommunity,
-  insert_activity,
-  local_instance,
   protocol::objects::tombstone::Tombstone,
   CONTEXT,
 };
 use activitypub_federation::{
-  core::inbox::receive_activity,
-  data::Data,
-  deser::context::WithContext,
-  traits::{ActivityHandler, Actor, ApubObject},
-  APUB_JSON_CONTENT_TYPE,
+  actix_web::inbox::receive_activity,
+  config::Data,
+  protocol::context::WithContext,
+  FEDERATION_CONTENT_TYPE,
 };
-use actix_web::{web, HttpRequest, HttpResponse};
+use actix_web::{web, web::Bytes, HttpRequest, HttpResponse};
 use http::StatusCode;
-use lemmy_api_common::utils::blocking;
-use lemmy_db_schema::source::activity::Activity;
-use lemmy_utils::error::LemmyError;
-use lemmy_websocket::LemmyContext;
-use once_cell::sync::OnceCell;
-use serde::{de::DeserializeOwned, Deserialize, Serialize};
-use serde_json::Value;
+use lemmy_api_common::context::LemmyContext;
+use lemmy_db_schema::source::activity::SentActivity;
+use lemmy_utils::error::{LemmyError, LemmyErrorType, LemmyResult};
+use serde::{Deserialize, Serialize};
 use std::ops::Deref;
-use tracing::{debug, log::info};
 use url::Url;
 
 mod comment;
@@ -33,73 +26,46 @@ mod post;
 pub mod routes;
 pub mod site;
 
-#[tracing::instrument(skip_all)]
 pub async fn shared_inbox(
   request: HttpRequest,
-  payload: String,
-  context: web::Data<LemmyContext>,
-) -> Result<HttpResponse, LemmyError> {
-  receive_lemmy_activity::<SharedInboxActivities, UserOrCommunity>(request, payload, context).await
-}
-
-pub async fn receive_lemmy_activity<Activity, ActorT>(
-  request: HttpRequest,
-  payload: String,
-  context: web::Data<LemmyContext>,
-) -> Result<HttpResponse, LemmyError>
-where
-  Activity: ActivityHandler<DataType = LemmyContext, Error = LemmyError>
-    + DeserializeOwned
-    + Send
-    + 'static,
-  ActorT: ApubObject<DataType = LemmyContext, Error = LemmyError> + Actor + Send + 'static,
-  for<'de2> <ActorT as ApubObject>::ApubType: serde::Deserialize<'de2>,
-{
-  let activity_value: Value = serde_json::from_str(&payload)?;
-  debug!("Received activity {:#}", payload.as_str());
-  let activity: Activity = serde_json::from_value(activity_value.clone())?;
-  // Log the activity, so we avoid receiving and parsing it twice.
-  let insert = insert_activity(activity.id(), activity_value, false, true, context.pool()).await?;
-  if !insert {
-    debug!("Received duplicate activity {}", activity.id().to_string());
-    return Ok(HttpResponse::BadRequest().finish());
-  }
-  info!("Received activity {}", payload);
-
-  static DATA: OnceCell<Data<LemmyContext>> = OnceCell::new();
-  let data = DATA.get_or_init(|| Data::new(context.get_ref().clone()));
-  receive_activity::<Activity, ActorT, LemmyContext>(
-    request,
-    activity,
-    local_instance(&context),
-    data,
-  )
-  .await
+  body: Bytes,
+  data: Data<LemmyContext>,
+) -> LemmyResult<HttpResponse> {
+  receive_activity::<SharedInboxActivities, UserOrCommunity, LemmyContext>(request, body, &data)
+    .await
 }
 
 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
 /// headers.
-fn create_apub_response<T>(data: &T) -> HttpResponse
+///
+/// actix-web doesn't allow pretty-print for json so we need to do this manually.
+fn create_apub_response<T>(data: &T) -> LemmyResult<HttpResponse>
 where
   T: Serialize,
 {
-  HttpResponse::Ok()
-    .content_type(APUB_JSON_CONTENT_TYPE)
-    .json(WithContext::new(data, CONTEXT.deref().clone()))
-}
+  let json = serde_json::to_string_pretty(&WithContext::new(data, CONTEXT.clone()))?;
 
-fn create_json_apub_response(data: serde_json::Value) -> HttpResponse {
-  HttpResponse::Ok()
-    .content_type(APUB_JSON_CONTENT_TYPE)
-    .json(data)
+  Ok(
+    HttpResponse::Ok()
+      .content_type(FEDERATION_CONTENT_TYPE)
+      .body(json),
+  )
 }
 
-fn create_apub_tombstone_response<T: Into<Url>>(id: T) -> HttpResponse {
+fn create_apub_tombstone_response<T: Into<Url>>(id: T) -> LemmyResult<HttpResponse> {
   let tombstone = Tombstone::new(id.into());
-  HttpResponse::Gone()
-    .content_type(APUB_JSON_CONTENT_TYPE)
-    .status(StatusCode::GONE)
-    .json(WithContext::new(tombstone, CONTEXT.deref().clone()))
+  let json = serde_json::to_string_pretty(&WithContext::new(tombstone, CONTEXT.deref().clone()))?;
+
+  Ok(
+    HttpResponse::Gone()
+      .content_type(FEDERATION_CONTENT_TYPE)
+      .status(StatusCode::GONE)
+      .body(json),
+  )
+}
+
+fn err_object_not_local() -> LemmyError {
+  LemmyErrorType::ObjectNotLocal.into()
 }
 
 #[derive(Deserialize)]
@@ -122,15 +88,12 @@ pub(crate) async fn get_activity(
     info.id
   ))?
   .into();
-  let activity = blocking(context.pool(), move |conn| {
-    Activity::read_from_apub_id(conn, &activity_id)
-  })
-  .await??;
+  let activity = SentActivity::read_from_apub_id(&mut context.pool(), &activity_id).await?;
 
-  let sensitive = activity.sensitive.unwrap_or(true);
-  if !activity.local || sensitive {
-    Ok(HttpResponse::NotFound().finish())
+  let sensitive = activity.sensitive;
+  if sensitive {
+    Ok(HttpResponse::Forbidden().finish())
   } else {
-    Ok(create_json_apub_response(activity.data))
+    create_apub_response(&activity.data)
   }
 }