crates/apub_lib/src/activity_queue.rs

   1 use crate::{signatures::sign_and_send, traits::ActorType, APUB_JSON_CONTENT_TYPE};
   2 use anyhow::{anyhow, Context, Error};
   3 use background_jobs::{
   4   create_server,
   5   memory_storage::Storage,
   6   ActixJob,
   7   Backoff,
   8   MaxRetries,
   9   QueueHandle,
  10   WorkerConfig,
  11 };
  12 use lemmy_utils::{location_info, LemmyError};
  13 use log::warn;
  14 use reqwest::Client;
  15 use serde::{Deserialize, Serialize};
  16 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
  17 use url::Url;
  18
  19 pub async fn send_activity(
  20   activity: String,
  21   actor: &dyn ActorType,
  22   inboxes: Vec<&Url>,
  23   client: &Client,
  24   activity_queue: &QueueHandle,
  25 ) -> Result<(), LemmyError> {
  26   for i in inboxes {
  27     let message = SendActivityTask {
  28       activity: activity.clone(),
  29       inbox: i.to_owned(),
  30       actor_id: actor.actor_id(),
  31       private_key: actor.private_key().context(location_info!())?,
  32     };
  33     if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
  34       do_send(message, client).await?;
  35     } else {
  36       activity_queue.queue::<SendActivityTask>(message)?;
  37     }
  38   }
  39
  40   Ok(())
  41 }
  42
  43 #[derive(Clone, Debug, Deserialize, Serialize)]
  44 struct SendActivityTask {
  45   activity: String,
  46   inbox: Url,
  47   actor_id: Url,
  48   private_key: String,
  49 }
  50
  51 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
  52 /// if the delivery failed.
  53 impl ActixJob for SendActivityTask {
  54   type State = MyState;
  55   type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
  56   const NAME: &'static str = "SendActivityTask";
  57
  58   const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
  59   const BACKOFF: Backoff = Backoff::Exponential(2);
  60
  61   fn run(self, state: Self::State) -> Self::Future {
  62     Box::pin(async move { do_send(self, &state.client).await })
  63   }
  64 }
  65
  66 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
  67   let mut headers = BTreeMap::<String, String>::new();
  68   headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
  69   let result = sign_and_send(
  70     client,
  71     headers,
  72     &task.inbox,
  73     task.activity.clone(),
  74     &task.actor_id,
  75     task.private_key.to_owned(),
  76   )
  77   .await;
  78
  79   if let Err(e) = result {
  80     warn!("{}", e);
  81     return Err(anyhow!(
  82       "Failed to send activity {} to {}",
  83       &task.activity,
  84       task.inbox
  85     ));
  86   }
  87   Ok(())
  88 }
  89
  90 pub fn create_activity_queue() -> QueueHandle {
  91   // Start the application server. This guards access to to the jobs store
  92   let queue_handle = create_server(Storage::new());
  93   let arbiter = actix_web::rt::Arbiter::new();
  94
  95   // Configure and start our workers
  96   WorkerConfig::new(|| MyState {
  97     client: Client::default(),
  98   })
  99   .register::<SendActivityTask>()
 100   .start_in_arbiter(&arbiter, queue_handle.clone());
 101
 102   queue_handle
 103 }
 104
 105 #[derive(Clone)]
 106 struct MyState {
 107   pub client: Client,
 108 }