1 use crate::{signatures::sign_and_send, traits::ActorType};
2 use anyhow::{anyhow, Context, Error};
4 memory_storage::Storage,
12 use lemmy_utils::{location_info, LemmyError};
14 use serde::{Deserialize, Serialize};
15 use std::{env, fmt::Debug, future::Future, pin::Pin};
16 use tracing::{info, warn};
19 pub async fn send_activity(
21 actor: &dyn ActorType,
25 activity_queue: &QueueHandle,
26 ) -> Result<(), LemmyError> {
28 let message = SendActivityTask {
29 activity_id: activity_id.clone(),
31 actor_id: actor.actor_id(),
32 activity: activity.clone(),
33 private_key: actor.private_key().context(location_info!())?,
35 if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
36 do_send(message, client).await?;
38 activity_queue.queue::<SendActivityTask>(message).await?;
45 #[derive(Clone, Debug, Deserialize, Serialize)]
46 struct SendActivityTask {
54 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
55 /// if the delivery failed.
56 impl ActixJob for SendActivityTask {
58 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
59 const NAME: &'static str = "SendActivityTask";
61 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
62 const BACKOFF: Backoff = Backoff::Exponential(2);
64 fn run(self, state: Self::State) -> Self::Future {
65 Box::pin(async move { do_send(self, &state.client).await })
69 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
70 info!("Sending {} to {}", task.activity_id, task.inbox);
71 let result = sign_and_send(
74 task.activity.clone(),
76 task.private_key.to_owned(),
82 if !o.status().is_success() {
84 "Send {} to {} failed with status {}: {}",
94 "Failed to send activity {} to {}: {}",
104 pub fn create_activity_queue() -> Manager {
105 // Configure and start our workers
106 WorkerConfig::new_managed(Storage::new(), |_| MyState {
107 client: Client::default(),
109 .register::<SendActivityTask>()