crates/apub_lib/src/activity_queue.rs

   1 use crate::{signatures::sign_and_send, traits::ActorType};
   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::{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 result = sign_and_send(
  68     client,
  69     &task.inbox,
  70     task.activity.clone(),
  71     &task.actor_id,
  72     task.private_key.to_owned(),
  73   )
  74   .await;
  75
  76   if let Err(e) = result {
  77     warn!("{}", e);
  78     return Err(anyhow!(
  79       "Failed to send activity {} to {}",
  80       &task.activity,
  81       task.inbox
  82     ));
  83   }
  84   Ok(())
  85 }
  86
  87 pub fn create_activity_queue() -> QueueHandle {
  88   // Start the application server. This guards access to to the jobs store
  89   let queue_handle = create_server(Storage::new());
  90   let arbiter = actix_web::rt::Arbiter::new();
  91
  92   // Configure and start our workers
  93   WorkerConfig::new(|| MyState {
  94     client: Client::default(),
  95   })
  96   .register::<SendActivityTask>()
  97   .start_in_arbiter(&arbiter, queue_handle.clone());
  98
  99   queue_handle
 100 }
 101
 102 #[derive(Clone)]
 103 struct MyState {
 104   pub client: Client,
 105 }