src/apub/activity_queue.rs

   1 use crate::apub::{check_is_apub_id_valid, extensions::signatures::sign, ActorType};
   2 use activitystreams::{
   3   base::{Extends, ExtendsExt},
   4   object::AsObject,
   5 };
   6 use anyhow::{anyhow, Context, Error};
   7 use awc::Client;
   8 use background_jobs::{
   9   create_server,
  10   memory_storage::Storage,
  11   ActixJob,
  12   Backoff,
  13   MaxRetries,
  14   QueueHandle,
  15   WorkerConfig,
  16 };
  17 use lemmy_utils::{location_info, settings::Settings, LemmyError};
  18 use log::warn;
  19 use serde::{Deserialize, Serialize};
  20 use std::{future::Future, pin::Pin};
  21 use url::Url;
  22
  23 pub fn send_activity<T, Kind>(
  24   activity_sender: &QueueHandle,
  25   activity: T,
  26   actor: &dyn ActorType,
  27   to: Vec<Url>,
  28 ) -> Result<(), LemmyError>
  29 where
  30   T: AsObject<Kind>,
  31   T: Extends<Kind>,
  32   Kind: Serialize,
  33   <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
  34 {
  35   if !Settings::get().federation.enabled {
  36     return Ok(());
  37   }
  38
  39   let activity = activity.into_any_base()?;
  40   let serialised_activity = serde_json::to_string(&activity)?;
  41
  42   for to_url in &to {
  43     check_is_apub_id_valid(&to_url)?;
  44   }
  45
  46   // TODO: it would make sense to create a separate task for each destination server
  47   let message = SendActivityTask {
  48     activity: serialised_activity,
  49     to,
  50     actor_id: actor.actor_id()?,
  51     private_key: actor.private_key().context(location_info!())?,
  52   };
  53   activity_sender.queue::<SendActivityTask>(message)?;
  54
  55   Ok(())
  56 }
  57
  58 #[derive(Clone, Debug, Deserialize, Serialize)]
  59 struct SendActivityTask {
  60   activity: String,
  61   to: Vec<Url>,
  62   actor_id: Url,
  63   private_key: String,
  64 }
  65
  66 impl ActixJob for SendActivityTask {
  67   type State = MyState;
  68   type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
  69   const NAME: &'static str = "SendActivityTask";
  70
  71   const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
  72   const BACKOFF: Backoff = Backoff::Exponential(2);
  73
  74   fn run(self, state: Self::State) -> Self::Future {
  75     Box::pin(async move {
  76       for to_url in &self.to {
  77         let request = state
  78           .client
  79           .post(to_url.as_str())
  80           .header("Content-Type", "application/json");
  81
  82         // TODO: i believe we have to do the signing in here because it is only valid for a few seconds
  83         let signed = sign(
  84           request,
  85           self.activity.clone(),
  86           &self.actor_id,
  87           self.private_key.to_owned(),
  88         )
  89         .await;
  90         let signed = match signed {
  91           Ok(s) => s,
  92           Err(e) => {
  93             warn!("{}", e);
  94             // dont return an error because retrying would probably not fix the signing
  95             return Ok(());
  96           }
  97         };
  98         if let Err(e) = signed.send().await {
  99           warn!("{}", e);
 100           return Err(anyhow!(
 101             "Failed to send activity {} to {}",
 102             &self.activity,
 103             to_url
 104           ));
 105         }
 106       }
 107
 108       Ok(())
 109     })
 110   }
 111 }
 112
 113 pub fn create_activity_queue() -> QueueHandle {
 114   // Start the application server. This guards access to to the jobs store
 115   let queue_handle = create_server(Storage::new());
 116
 117   // Configure and start our workers
 118   WorkerConfig::new(|| MyState {
 119     client: Client::default(),
 120   })
 121   .register::<SendActivityTask>()
 122   .start(queue_handle.clone());
 123
 124   queue_handle
 125 }
 126
 127 #[derive(Clone)]
 128 struct MyState {
 129   pub client: Client,
 130 }