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