]> Untitled Git - lemmy.git/blobdiff - crates/apub/src/activities/mod.rs
Split activity table into sent and received parts (fixes #3103) (#3583)
[lemmy.git] / crates / apub / src / activities / mod.rs
index 68a3b535c10bd16ddb5c9b6c7e1d5a2e4c4fcab2..4fd8da536f20f7e01a39cd545c5e6165e7ea1a4a 100644 (file)
@@ -1,26 +1,31 @@
 use crate::{
-  insert_activity,
-  local_instance,
   objects::{community::ApubCommunity, person::ApubPerson},
-  ActorType,
   CONTEXT,
 };
 use activitypub_federation::{
-  core::{activity_queue::send_activity, object_id::ObjectId},
-  deser::context::WithContext,
+  activity_queue::send_activity,
+  config::Data,
+  fetch::object_id::ObjectId,
+  kinds::public,
+  protocol::context::WithContext,
   traits::{ActivityHandler, Actor},
 };
-use activitystreams_kinds::public;
 use anyhow::anyhow;
-use lemmy_api_common::{generate_moderators_url, LemmyContext};
+use lemmy_api_common::context::LemmyContext;
 use lemmy_db_schema::{
   newtypes::CommunityId,
-  source::{community::Community, local_site::LocalSite},
+  source::{
+    activity::{SentActivity, SentActivityForm},
+    community::Community,
+    instance::Instance,
+  },
 };
 use lemmy_db_views_actor::structs::{CommunityPersonBanView, CommunityView};
-use lemmy_utils::error::LemmyError;
+use lemmy_utils::error::{LemmyError, LemmyErrorExt, LemmyErrorType};
+use moka::future::Cache;
+use once_cell::sync::Lazy;
 use serde::Serialize;
-use std::ops::Deref;
+use std::{ops::Deref, sync::Arc, time::Duration};
 use tracing::info;
 use url::{ParseError, Url};
 use uuid::Uuid;
@@ -30,22 +35,24 @@ pub mod community;
 pub mod create_or_update;
 pub mod deletion;
 pub mod following;
+pub mod unfederated;
 pub mod voting;
 
+/// Amount of time that the list of dead instances is cached. This is only updated once a day,
+/// so there is no harm in caching it for a longer time.
+pub static DEAD_INSTANCE_LIST_CACHE_DURATION: Duration = Duration::from_secs(30 * 60);
+
 /// Checks that the specified Url actually identifies a Person (by fetching it), and that the person
 /// doesn't have a site ban.
 #[tracing::instrument(skip_all)]
 async fn verify_person(
   person_id: &ObjectId<ApubPerson>,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  let person = person_id
-    .dereference(context, local_instance(context).await, request_counter)
-    .await?;
+  let person = person_id.dereference(context).await?;
   if person.banned {
-    let err = anyhow!("Person {} is banned", person_id);
-    return Err(LemmyError::from_error_message(err, "banned"));
+    return Err(anyhow!("Person {} is banned", person_id))
+      .with_lemmy_type(LemmyErrorType::CouldntUpdateComment);
   }
   Ok(())
 }
@@ -56,22 +63,19 @@ async fn verify_person(
 pub(crate) async fn verify_person_in_community(
   person_id: &ObjectId<ApubPerson>,
   community: &ApubCommunity,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  let person = person_id
-    .dereference(context, local_instance(context).await, request_counter)
-    .await?;
+  let person = person_id.dereference(context).await?;
   if person.banned {
-    return Err(LemmyError::from_message("Person is banned from site"));
+    return Err(LemmyErrorType::PersonIsBannedFromSite)?;
   }
   let person_id = person.id;
   let community_id = community.id;
-  let is_banned = CommunityPersonBanView::get(context.pool(), person_id, community_id)
+  let is_banned = CommunityPersonBanView::get(&mut context.pool(), person_id, community_id)
     .await
     .is_ok();
   if is_banned {
-    return Err(LemmyError::from_message("Person is banned from community"));
+    return Err(LemmyErrorType::PersonIsBannedFromCommunity)?;
   }
 
   Ok(())
@@ -87,15 +91,12 @@ pub(crate) async fn verify_mod_action(
   mod_id: &ObjectId<ApubPerson>,
   object_id: &Url,
   community_id: CommunityId,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  let mod_ = mod_id
-    .dereference(context, local_instance(context).await, request_counter)
-    .await?;
+  let mod_ = mod_id.dereference(context).await?;
 
   let is_mod_or_admin =
-    CommunityView::is_mod_or_admin(context.pool(), mod_.id, community_id).await?;
+    CommunityView::is_mod_or_admin(&mut context.pool(), mod_.id, community_id).await?;
   if is_mod_or_admin {
     return Ok(());
   }
@@ -107,43 +108,33 @@ pub(crate) async fn verify_mod_action(
     return Ok(());
   }
 
-  Err(LemmyError::from_message("Not a mod"))
-}
-
-/// For Add/Remove community moderator activities, check that the target field actually contains
-/// /c/community/moderators. Any different values are unsupported.
-fn verify_add_remove_moderator_target(
-  target: &Url,
-  community: &ApubCommunity,
-) -> Result<(), LemmyError> {
-  if target != &generate_moderators_url(&community.actor_id)?.into() {
-    return Err(LemmyError::from_message("Unkown target url"));
-  }
-  Ok(())
+  Err(LemmyErrorType::NotAModerator)?
 }
 
 pub(crate) fn verify_is_public(to: &[Url], cc: &[Url]) -> Result<(), LemmyError> {
   if ![to, cc].iter().any(|set| set.contains(&public())) {
-    return Err(LemmyError::from_message("Object is not public"));
+    return Err(LemmyErrorType::ObjectIsNotPublic)?;
   }
   Ok(())
 }
 
-pub(crate) fn verify_community_matches(
-  a: &ApubCommunity,
-  b: CommunityId,
-) -> Result<(), LemmyError> {
-  if a.id != b {
-    return Err(LemmyError::from_message("Invalid community"));
+pub(crate) fn verify_community_matches<T>(
+  a: &ObjectId<ApubCommunity>,
+  b: T,
+) -> Result<(), LemmyError>
+where
+  T: Into<ObjectId<ApubCommunity>>,
+{
+  let b: ObjectId<ApubCommunity> = b.into();
+  if a != &b {
+    return Err(LemmyErrorType::InvalidCommunity)?;
   }
   Ok(())
 }
 
 pub(crate) fn check_community_deleted_or_removed(community: &Community) -> Result<(), LemmyError> {
   if community.deleted || community.removed {
-    Err(LemmyError::from_message(
-      "New post or comment cannot be created in deleted or removed community",
-    ))
+    Err(LemmyErrorType::CannotCreatePostOrCommentInDeletedOrRemovedCommunity)?
   } else {
     Ok(())
   }
@@ -166,39 +157,43 @@ where
 
 #[tracing::instrument(skip_all)]
 async fn send_lemmy_activity<Activity, ActorT>(
-  context: &LemmyContext,
+  data: &Data<LemmyContext>,
   activity: Activity,
   actor: &ActorT,
-  inbox: Vec<Url>,
+  mut inbox: Vec<Url>,
   sensitive: bool,
 ) -> Result<(), LemmyError>
 where
-  Activity: ActivityHandler + Serialize,
-  ActorT: Actor + ActorType,
+  Activity: ActivityHandler + Serialize + Send + Sync + Clone,
+  ActorT: Actor,
   Activity: ActivityHandler<Error = LemmyError>,
 {
-  let federation_enabled = LocalSite::read(context.pool())
-    .await
-    .map(|l| l.federation_enabled)
-    .unwrap_or(false);
-  if !federation_enabled {
-    return Ok(());
-  }
+  static CACHE: Lazy<Cache<(), Arc<Vec<String>>>> = Lazy::new(|| {
+    Cache::builder()
+      .max_capacity(1)
+      .time_to_live(DEAD_INSTANCE_LIST_CACHE_DURATION)
+      .build()
+  });
+  let dead_instances = CACHE
+    .try_get_with((), async {
+      Ok::<_, diesel::result::Error>(Arc::new(Instance::dead_instances(&mut data.pool()).await?))
+    })
+    .await?;
 
+  inbox.retain(|i| {
+    let domain = i.domain().expect("has domain").to_string();
+    !dead_instances.contains(&domain)
+  });
   info!("Sending activity {}", activity.id().to_string());
   let activity = WithContext::new(activity, CONTEXT.deref().clone());
 
-  let object_value = serde_json::to_value(&activity)?;
-  insert_activity(activity.id(), object_value, true, sensitive, context.pool()).await?;
-
-  send_activity(
-    activity,
-    actor.get_public_key(),
-    actor.private_key().expect("actor has private key"),
-    inbox,
-    local_instance(context).await,
-  )
-  .await?;
+  let form = SentActivityForm {
+    ap_id: activity.id().clone().into(),
+    data: serde_json::to_value(activity.clone())?,
+    sensitive,
+  };
+  SentActivity::create(&mut data.pool(), form).await?;
+  send_activity(activity, actor, inbox, data).await?;
 
   Ok(())
 }