lemmy_apub/src/activity_queue.rs

   1 use crate::{check_is_apub_id_valid, extensions::signatures::sign_and_send, ActorType};
   2 use activitystreams::{
   3   base::{Extends, ExtendsExt},
   4   object::AsObject,
   5 };
   6 use anyhow::{anyhow, Context, Error};
   7 use background_jobs::{
   8   create_server,
   9   memory_storage::Storage,
  10   ActixJob,
  11   Backoff,
  12   MaxRetries,
  13   QueueHandle,
  14   WorkerConfig,
  15 };
  16 use lemmy_utils::{location_info, settings::Settings, LemmyError};
  17 use log::warn;
  18 use reqwest::Client;
  19 use serde::{Deserialize, Serialize};
  20 use std::{collections::BTreeMap, 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
  54   activity_sender.queue::<SendActivityTask>(message)?;
  55
  56   Ok(())
  57 }
  58
  59 #[derive(Clone, Debug, Deserialize, Serialize)]
  60 struct SendActivityTask {
  61   activity: String,
  62   to: Vec<Url>,
  63   actor_id: Url,
  64   private_key: String,
  65 }
  66
  67 impl ActixJob for SendActivityTask {
  68   type State = MyState;
  69   type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
  70   const NAME: &'static str = "SendActivityTask";
  71
  72   const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
  73   const BACKOFF: Backoff = Backoff::Exponential(2);
  74
  75   fn run(self, state: Self::State) -> Self::Future {
  76     Box::pin(async move {
  77       for to_url in &self.to {
  78         let mut headers = BTreeMap::<String, String>::new();
  79         headers.insert("Content-Type".into(), "application/json".into());
  80         let result = sign_and_send(
  81           &state.client,
  82           headers,
  83           to_url,
  84           self.activity.clone(),
  85           &self.actor_id,
  86           self.private_key.to_owned(),
  87         )
  88         .await;
  89
  90         if let Err(e) = result {
  91           warn!("{}", e);
  92           return Err(anyhow!(
  93             "Failed to send activity {} to {}",
  94             &self.activity,
  95             to_url
  96           ));
  97         }
  98       }
  99       Ok(())
 100     })
 101   }
 102 }
 103
 104 pub fn create_activity_queue() -> QueueHandle {
 105   // Start the application server. This guards access to to the jobs store
 106   let queue_handle = create_server(Storage::new());
 107
 108   // Configure and start our workers
 109   WorkerConfig::new(|| MyState {
 110     client: Client::default(),
 111   })
 112   .register::<SendActivityTask>()
 113   .start(queue_handle.clone());
 114
 115   queue_handle
 116 }
 117
 118 #[derive(Clone)]
 119 struct MyState {
 120   pub client: Client,
 121 }