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