crates/apub/src/activity_queue.rs

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