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 /// Sends notification to any persons mentioned in a comment
 142 ///
 143 /// * `creator` person who created the comment
 144 /// * `mentions` list of inboxes of persons which are mentioned in the comment
 145 /// * `activity` either a `Create/Note` or `Update/Note`
 146 pub(crate) async fn send_comment_mentions<T, Kind>(
 147   creator: &Person,
 148   mentions: Vec<Url>,
 149   activity: T,
 150   context: &LemmyContext,
 151 ) -> Result<(), LemmyError>
 152 where
 153   T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
 154   Kind: Serialize,
 155   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
 156 {
 157   debug!(
 158     "Sending mentions activity {:?} to {:?}",
 159     &activity.id_unchecked(),
 160     &mentions
 161   );
 162   let mentions = mentions
 163     .iter()
 164     .filter(|inbox| check_is_apub_id_valid(inbox, false).is_ok())
 165     .map(|i| i.to_owned())
 166     .collect();
 167   send_activity_internal(
 168     context, activity, creator, mentions, false, // Don't create a new DB row
 169     false,
 170   )
 171   .await?;
 172   Ok(())
 173 }
 174
 175 pub(crate) async fn send_to_community_new(
 176   activity: AnnouncableActivities,
 177   activity_id: &Url,
 178   actor: &dyn ActorType,
 179   community: &Community,
 180   additional_inboxes: Vec<Url>,
 181   context: &LemmyContext,
 182 ) -> Result<(), LemmyError> {
 183   // if this is a local community, we need to do an announce from the community instead
 184   if community.local {
 185     insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
 186     AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
 187   } else {
 188     let mut inboxes = additional_inboxes;
 189     inboxes.push(community.get_shared_inbox_or_inbox_url());
 190     send_activity_new(context, &activity, activity_id, actor, inboxes, false).await?;
 191   }
 192
 193   Ok(())
 194 }
 195
 196 pub(crate) async fn send_activity_new<T>(
 197   context: &LemmyContext,
 198   activity: &T,
 199   activity_id: &Url,
 200   actor: &dyn ActorType,
 201   inboxes: Vec<Url>,
 202   sensitive: bool,
 203 ) -> Result<(), LemmyError>
 204 where
 205   T: Serialize,
 206 {
 207   if !Settings::get().federation().enabled || inboxes.is_empty() {
 208     return Ok(());
 209   }
 210
 211   info!("Sending activity {}", activity_id.to_string());
 212
 213   // Don't send anything to ourselves
 214   // TODO: this should be a debug assert
 215   let hostname = Settings::get().get_hostname_without_port()?;
 216   let inboxes: Vec<&Url> = inboxes
 217     .iter()
 218     .filter(|i| i.domain().expect("valid inbox url") != hostname)
 219     .collect();
 220
 221   let serialised_activity = serde_json::to_string(&activity)?;
 222
 223   insert_activity(
 224     activity_id,
 225     serialised_activity.clone(),
 226     true,
 227     sensitive,
 228     context.pool(),
 229   )
 230   .await?;
 231
 232   for i in inboxes {
 233     let message = SendActivityTask {
 234       activity: serialised_activity.to_owned(),
 235       inbox: i.to_owned(),
 236       actor_id: actor.actor_id(),
 237       private_key: actor.private_key().context(location_info!())?,
 238     };
 239     if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
 240       do_send(message, &Client::default()).await?;
 241     } else {
 242       context.activity_queue.queue::<SendActivityTask>(message)?;
 243     }
 244   }
 245
 246   Ok(())
 247 }
 248
 249 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
 250 /// handling signing and retrying failed deliveres.
 251 ///
 252 /// The caller of this function needs to remove any blocked domains from `to`,
 253 /// using `check_is_apub_id_valid()`.
 254 async fn send_activity_internal<T, Kind>(
 255   context: &LemmyContext,
 256   activity: T,
 257   actor: &dyn ActorType,
 258   inboxes: Vec<Url>,
 259   insert_into_db: bool,
 260   sensitive: bool,
 261 ) -> Result<(), LemmyError>
 262 where
 263   T: AsObject<Kind> + Extends<Kind> + Debug,
 264   Kind: Serialize,
 265   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
 266 {
 267   if !Settings::get().federation().enabled || inboxes.is_empty() {
 268     return Ok(());
 269   }
 270
 271   // Don't send anything to ourselves
 272   let hostname = Settings::get().get_hostname_without_port()?;
 273   let inboxes: Vec<&Url> = inboxes
 274     .iter()
 275     .filter(|i| i.domain().expect("valid inbox url") != hostname)
 276     .collect();
 277
 278   let activity = activity.into_any_base()?;
 279   let serialised_activity = serde_json::to_string(&activity)?;
 280
 281   // This is necessary because send_comment and send_comment_mentions
 282   // might send the same ap_id
 283   if insert_into_db {
 284     let id = activity.id().context(location_info!())?;
 285     insert_activity(id, activity.clone(), true, sensitive, context.pool()).await?;
 286   }
 287
 288   for i in inboxes {
 289     let message = SendActivityTask {
 290       activity: serialised_activity.to_owned(),
 291       inbox: i.to_owned(),
 292       actor_id: actor.actor_id(),
 293       private_key: actor.private_key().context(location_info!())?,
 294     };
 295     if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
 296       do_send(message, &Client::default()).await?;
 297     } else {
 298       context.activity_queue.queue::<SendActivityTask>(message)?;
 299     }
 300   }
 301
 302   Ok(())
 303 }
 304
 305 #[derive(Clone, Debug, Deserialize, Serialize)]
 306 struct SendActivityTask {
 307   activity: String,
 308   inbox: Url,
 309   actor_id: Url,
 310   private_key: String,
 311 }
 312
 313 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
 314 /// if the delivery failed.
 315 impl ActixJob for SendActivityTask {
 316   type State = MyState;
 317   type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
 318   const NAME: &'static str = "SendActivityTask";
 319
 320   const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
 321   const BACKOFF: Backoff = Backoff::Exponential(2);
 322
 323   fn run(self, state: Self::State) -> Self::Future {
 324     Box::pin(async move { do_send(self, &state.client).await })
 325   }
 326 }
 327
 328 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
 329   let mut headers = BTreeMap::<String, String>::new();
 330   headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
 331   let result = sign_and_send(
 332     client,
 333     headers,
 334     &task.inbox,
 335     task.activity.clone(),
 336     &task.actor_id,
 337     task.private_key.to_owned(),
 338   )
 339   .await;
 340
 341   if let Err(e) = result {
 342     warn!("{}", e);
 343     return Err(anyhow!(
 344       "Failed to send activity {} to {}",
 345       &task.activity,
 346       task.inbox
 347     ));
 348   }
 349   Ok(())
 350 }
 351
 352 pub fn create_activity_queue() -> QueueHandle {
 353   // Start the application server. This guards access to to the jobs store
 354   let queue_handle = create_server(Storage::new());
 355
 356   // Configure and start our workers
 357   WorkerConfig::new(|| MyState {
 358     client: Client::default(),
 359   })
 360   .register::<SendActivityTask>()
 361   .start(queue_handle.clone());
 362
 363   queue_handle
 364 }
 365
 366 #[derive(Clone)]
 367 struct MyState {
 368   pub client: Client,
 369 }