crates/apub/src/activity_queue.rs

   1 use crate::{
   2   activities::community::announce::{AnnouncableActivities, AnnounceActivity},
   3   check_is_apub_id_valid,
   4   extensions::signatures::sign_and_send,
   5   insert_activity,
   6   ActorType,
   7   CommunityType,
   8   APUB_JSON_CONTENT_TYPE,
   9 };
  10 use activitystreams::{
  11   base::{BaseExt, Extends, ExtendsExt},
  12   object::AsObject,
  13 };
  14 use anyhow::{anyhow, Context, Error};
  15 use background_jobs::{
  16   create_server,
  17   memory_storage::Storage,
  18   ActixJob,
  19   Backoff,
  20   MaxRetries,
  21   QueueHandle,
  22   WorkerConfig,
  23 };
  24 use itertools::Itertools;
  25 use lemmy_db_schema::source::{community::Community, person::Person};
  26 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
  27 use lemmy_websocket::LemmyContext;
  28 use log::{debug, info, warn};
  29 use reqwest::Client;
  30 use serde::{Deserialize, Serialize};
  31 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
  32 use url::Url;
  33
  34 /// Sends a local activity to a single, remote actor.
  35 ///
  36 /// * `activity` the apub activity to be sent
  37 /// * `creator` the local actor which created the activity
  38 /// * `inbox` the inbox url where the activity should be delivered to
  39 pub(crate) async fn send_activity_single_dest<T, Kind>(
  40   activity: T,
  41   creator: &dyn ActorType,
  42   inbox: Url,
  43   context: &LemmyContext,
  44 ) -> Result<(), LemmyError>
  45 where
  46   T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
  47   Kind: Serialize,
  48   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
  49 {
  50   if check_is_apub_id_valid(&inbox, false).is_ok() {
  51     debug!(
  52       "Sending activity {:?} to {}",
  53       &activity.id_unchecked().map(ToString::to_string),
  54       &inbox
  55     );
  56     send_activity_internal(context, activity, creator, vec![inbox], true, true).await?;
  57   }
  58
  59   Ok(())
  60 }
  61
  62 /// From a local community, send activity to all remote followers.
  63 ///
  64 /// * `activity` the apub activity to send
  65 /// * `community` the sending community
  66 /// * `extra_inbox` actor inbox which should receive the activity, in addition to followers
  67 pub(crate) async fn send_to_community_followers<T, Kind>(
  68   activity: T,
  69   community: &Community,
  70   extra_inbox: Option<Url>,
  71   context: &LemmyContext,
  72 ) -> Result<(), LemmyError>
  73 where
  74   T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
  75   Kind: Serialize,
  76   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
  77 {
  78   let extra_inbox: Vec<Url> = extra_inbox.into_iter().collect();
  79   let follower_inboxes: Vec<Url> = vec![
  80     community.get_follower_inboxes(context.pool()).await?,
  81     extra_inbox,
  82   ]
  83   .iter()
  84   .flatten()
  85   .unique()
  86   .filter(|inbox| inbox.host_str() != Some(&Settings::get().hostname))
  87   .filter(|inbox| check_is_apub_id_valid(inbox, false).is_ok())
  88   .map(|inbox| inbox.to_owned())
  89   .collect();
  90   debug!(
  91     "Sending activity {:?} to followers of {}",
  92     &activity.id_unchecked().map(ToString::to_string),
  93     &community.actor_id
  94   );
  95
  96   send_activity_internal(context, activity, community, follower_inboxes, true, false).await?;
  97
  98   Ok(())
  99 }
 100
 101 /// Sends an activity from a local person to a remote community.
 102 ///
 103 /// * `activity` the activity to send
 104 /// * `creator` the creator of the activity
 105 /// * `community` the destination community
 106 /// * `object_actor` if the object of the activity is an actor, it should be passed here so it can
 107 ///                  be sent directly to the actor
 108 ///
 109 pub(crate) async fn send_to_community<T, Kind>(
 110   activity: T,
 111   creator: &Person,
 112   community: &Community,
 113   object_actor: Option<Url>,
 114   context: &LemmyContext,
 115 ) -> Result<(), LemmyError>
 116 where
 117   T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
 118   Kind: Serialize,
 119   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
 120 {
 121   // if this is a local community, we need to do an announce from the community instead
 122   if community.local {
 123     community
 124       .send_announce(activity.into_any_base()?, object_actor, context)
 125       .await?;
 126   } else {
 127     let inbox = community.get_shared_inbox_or_inbox_url();
 128     check_is_apub_id_valid(&inbox, false)?;
 129     debug!(
 130       "Sending activity {:?} to community {}",
 131       &activity.id_unchecked().map(ToString::to_string),
 132       &community.actor_id
 133     );
 134     // dont send to object_actor here, as that is responsibility of the community itself
 135     send_activity_internal(context, activity, creator, vec![inbox], true, false).await?;
 136   }
 137
 138   Ok(())
 139 }
 140
 141 pub(crate) async fn send_to_community_new(
 142   activity: AnnouncableActivities,
 143   activity_id: &Url,
 144   actor: &dyn ActorType,
 145   community: &Community,
 146   additional_inboxes: Vec<Url>,
 147   context: &LemmyContext,
 148 ) -> Result<(), LemmyError> {
 149   // if this is a local community, we need to do an announce from the community instead
 150   if community.local {
 151     insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
 152     AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
 153   } else {
 154     let mut inboxes = additional_inboxes;
 155     inboxes.push(community.get_shared_inbox_or_inbox_url());
 156     send_activity_new(context, &activity, activity_id, actor, inboxes, false).await?;
 157   }
 158
 159   Ok(())
 160 }
 161
 162 pub(crate) async fn send_activity_new<T>(
 163   context: &LemmyContext,
 164   activity: &T,
 165   activity_id: &Url,
 166   actor: &dyn ActorType,
 167   inboxes: Vec<Url>,
 168   sensitive: bool,
 169 ) -> Result<(), LemmyError>
 170 where
 171   T: Serialize,
 172 {
 173   if !Settings::get().federation.enabled || inboxes.is_empty() {
 174     return Ok(());
 175   }
 176
 177   info!("Sending activity {}", activity_id.to_string());
 178
 179   // Don't send anything to ourselves
 180   // TODO: this should be a debug assert
 181   let hostname = Settings::get().get_hostname_without_port()?;
 182   let inboxes: Vec<&Url> = inboxes
 183     .iter()
 184     .filter(|i| i.domain().expect("valid inbox url") != hostname)
 185     .collect();
 186
 187   let serialised_activity = serde_json::to_string(&activity)?;
 188
 189   insert_activity(
 190     activity_id,
 191     serialised_activity.clone(),
 192     true,
 193     sensitive,
 194     context.pool(),
 195   )
 196   .await?;
 197
 198   for i in inboxes {
 199     let message = SendActivityTask {
 200       activity: serialised_activity.to_owned(),
 201       inbox: i.to_owned(),
 202       actor_id: actor.actor_id(),
 203       private_key: actor.private_key().context(location_info!())?,
 204     };
 205     if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
 206       do_send(message, &Client::default()).await?;
 207     } else {
 208       context.activity_queue.queue::<SendActivityTask>(message)?;
 209     }
 210   }
 211
 212   Ok(())
 213 }
 214
 215 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
 216 /// handling signing and retrying failed deliveres.
 217 ///
 218 /// The caller of this function needs to remove any blocked domains from `to`,
 219 /// using `check_is_apub_id_valid()`.
 220 async fn send_activity_internal<T, Kind>(
 221   context: &LemmyContext,
 222   activity: T,
 223   actor: &dyn ActorType,
 224   inboxes: Vec<Url>,
 225   insert_into_db: bool,
 226   sensitive: bool,
 227 ) -> Result<(), LemmyError>
 228 where
 229   T: AsObject<Kind> + Extends<Kind> + Debug,
 230   Kind: Serialize,
 231   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
 232 {
 233   if !Settings::get().federation.enabled || inboxes.is_empty() {
 234     return Ok(());
 235   }
 236
 237   // Don't send anything to ourselves
 238   let hostname = Settings::get().get_hostname_without_port()?;
 239   let inboxes: Vec<&Url> = inboxes
 240     .iter()
 241     .filter(|i| i.domain().expect("valid inbox url") != hostname)
 242     .collect();
 243
 244   let activity = activity.into_any_base()?;
 245   let serialised_activity = serde_json::to_string(&activity)?;
 246
 247   // This is necessary because send_comment and send_comment_mentions
 248   // might send the same ap_id
 249   if insert_into_db {
 250     let id = activity.id().context(location_info!())?;
 251     insert_activity(id, activity.clone(), true, sensitive, context.pool()).await?;
 252   }
 253
 254   for i in inboxes {
 255     let message = SendActivityTask {
 256       activity: serialised_activity.to_owned(),
 257       inbox: i.to_owned(),
 258       actor_id: actor.actor_id(),
 259       private_key: actor.private_key().context(location_info!())?,
 260     };
 261     if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
 262       do_send(message, &Client::default()).await?;
 263     } else {
 264       context.activity_queue.queue::<SendActivityTask>(message)?;
 265     }
 266   }
 267
 268   Ok(())
 269 }
 270
 271 #[derive(Clone, Debug, Deserialize, Serialize)]
 272 struct SendActivityTask {
 273   activity: String,
 274   inbox: Url,
 275   actor_id: Url,
 276   private_key: String,
 277 }
 278
 279 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
 280 /// if the delivery failed.
 281 impl ActixJob for SendActivityTask {
 282   type State = MyState;
 283   type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
 284   const NAME: &'static str = "SendActivityTask";
 285
 286   const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
 287   const BACKOFF: Backoff = Backoff::Exponential(2);
 288
 289   fn run(self, state: Self::State) -> Self::Future {
 290     Box::pin(async move { do_send(self, &state.client).await })
 291   }
 292 }
 293
 294 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
 295   let mut headers = BTreeMap::<String, String>::new();
 296   headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
 297   let result = sign_and_send(
 298     client,
 299     headers,
 300     &task.inbox,
 301     task.activity.clone(),
 302     &task.actor_id,
 303     task.private_key.to_owned(),
 304   )
 305   .await;
 306
 307   if let Err(e) = result {
 308     warn!("{}", e);
 309     return Err(anyhow!(
 310       "Failed to send activity {} to {}",
 311       &task.activity,
 312       task.inbox
 313     ));
 314   }
 315   Ok(())
 316 }
 317
 318 pub fn create_activity_queue() -> QueueHandle {
 319   // Start the application server. This guards access to to the jobs store
 320   let queue_handle = create_server(Storage::new());
 321
 322   // Configure and start our workers
 323   WorkerConfig::new(|| MyState {
 324     client: Client::default(),
 325   })
 326   .register::<SendActivityTask>()
 327   .start(queue_handle.clone());
 328
 329   queue_handle
 330 }
 331
 332 #[derive(Clone)]
 333 struct MyState {
 334   pub client: Client,
 335 }