1 use crate::{check_is_apub_id_valid, extensions::signatures::sign_and_send, ActorType};
3 base::{Extends, ExtendsExt},
6 use anyhow::{anyhow, Context, Error};
9 memory_storage::Storage,
16 use lemmy_utils::{location_info, settings::Settings, LemmyError};
19 use serde::{Deserialize, Serialize};
20 use std::{collections::BTreeMap, future::Future, pin::Pin};
23 pub fn send_activity<T, Kind>(
24 activity_sender: &QueueHandle,
26 actor: &dyn ActorType,
28 ) -> Result<(), LemmyError>
33 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
35 if !Settings::get().federation.enabled {
39 let activity = activity.into_any_base()?;
40 let serialised_activity = serde_json::to_string(&activity)?;
43 check_is_apub_id_valid(&to_url)?;
46 // TODO: it would make sense to create a separate task for each destination server
47 let message = SendActivityTask {
48 activity: serialised_activity,
50 actor_id: actor.actor_id()?,
51 private_key: actor.private_key().context(location_info!())?,
54 activity_sender.queue::<SendActivityTask>(message)?;
59 #[derive(Clone, Debug, Deserialize, Serialize)]
60 struct SendActivityTask {
67 impl ActixJob for SendActivityTask {
69 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
70 const NAME: &'static str = "SendActivityTask";
72 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
73 const BACKOFF: Backoff = Backoff::Exponential(2);
75 fn run(self, state: Self::State) -> Self::Future {
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(
84 self.activity.clone(),
86 self.private_key.to_owned(),
90 if let Err(e) = result {
93 "Failed to send activity {} to {}",
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());
108 // Configure and start our workers
109 WorkerConfig::new(|| MyState {
110 client: Client::default(),
112 .register::<SendActivityTask>()
113 .start(queue_handle.clone());