2 check_is_apub_id_valid,
3 extensions::signatures::sign_and_send,
8 base::{BaseExt, Extends, ExtendsExt},
11 use anyhow::{anyhow, Context, Error};
12 use background_jobs::{
14 memory_storage::Storage,
21 use itertools::Itertools;
23 use lemmy_db_schema::source::{community::Community, user::User_};
24 use lemmy_utils::{location_info, settings::Settings, LemmyError};
25 use lemmy_websocket::LemmyContext;
26 use log::{debug, warn};
28 use serde::{export::fmt::Debug, Deserialize, Serialize};
29 use std::{collections::BTreeMap, env, future::Future, pin::Pin};
32 /// Sends a local activity to a single, remote actor.
34 /// * `activity` the apub activity to be sent
35 /// * `creator` the local actor which created the activity
36 /// * `inbox` the inbox url where the activity should be delivered to
37 pub(crate) async fn send_activity_single_dest<T, Kind>(
39 creator: &dyn ActorType,
41 context: &LemmyContext,
42 ) -> Result<(), LemmyError>
44 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
46 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
48 if check_is_apub_id_valid(&inbox).is_ok() {
50 "Sending activity {:?} to {}",
51 &activity.id_unchecked(),
54 send_activity_internal(
55 context.activity_queue(),
69 /// From a local community, send activity to all remote followers.
71 /// * `activity` the apub activity to send
72 /// * `community` the sending community
73 /// * `sender_shared_inbox` in case of an announce, this should be the shared inbox of the inner
74 /// activities creator, as receiving a known activity will cause an error
75 pub(crate) async fn send_to_community_followers<T, Kind>(
77 community: &Community,
78 context: &LemmyContext,
79 ) -> Result<(), LemmyError>
81 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
83 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
85 let follower_inboxes: Vec<Url> = community
86 .get_follower_inboxes(context.pool())
90 .filter(|inbox| inbox.host_str() != Some(&Settings::get().hostname))
91 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
92 .map(|inbox| inbox.to_owned())
95 "Sending activity {:?} to followers of {}",
96 &activity.id_unchecked(),
100 send_activity_internal(
101 context.activity_queue(),
114 /// Sends an activity from a local user to a remote community.
116 /// * `activity` the activity to send
117 /// * `creator` the creator of the activity
118 /// * `community` the destination community
120 pub(crate) async fn send_to_community<T, Kind>(
123 community: &Community,
124 context: &LemmyContext,
125 ) -> Result<(), LemmyError>
127 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
129 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
131 // if this is a local community, we need to do an announce from the community instead
134 .send_announce(activity.into_any_base()?, context)
137 let inbox = community.get_shared_inbox_url()?;
138 check_is_apub_id_valid(&inbox)?;
140 "Sending activity {:?} to community {}",
141 &activity.id_unchecked(),
144 send_activity_internal(
145 context.activity_queue(),
159 /// Sends notification to any users mentioned in a comment
161 /// * `creator` user who created the comment
162 /// * `mentions` list of inboxes of users which are mentioned in the comment
163 /// * `activity` either a `Create/Note` or `Update/Note`
164 pub(crate) async fn send_comment_mentions<T, Kind>(
168 context: &LemmyContext,
169 ) -> Result<(), LemmyError>
171 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
173 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
176 "Sending mentions activity {:?} to {:?}",
177 &activity.id_unchecked(),
180 let mentions = mentions
182 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
183 .map(|i| i.to_owned())
185 send_activity_internal(
186 context.activity_queue(),
191 false, // Don't create a new DB row
198 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
199 /// handling signing and retrying failed deliveres.
201 /// The caller of this function needs to remove any blocked domains from `to`,
202 /// using `check_is_apub_id_valid()`.
203 async fn send_activity_internal<T, Kind>(
204 activity_sender: &QueueHandle,
206 actor: &dyn ActorType,
209 insert_into_db: bool,
211 ) -> Result<(), LemmyError>
213 T: AsObject<Kind> + Extends<Kind> + Debug,
215 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
217 if !Settings::get().federation.enabled || inboxes.is_empty() {
221 let activity = activity.into_any_base()?;
222 let serialised_activity = serde_json::to_string(&activity)?;
224 // This is necessary because send_comment and send_comment_mentions
225 // might send the same ap_id
227 let id = activity.id().context(location_info!())?;
228 insert_activity(id, activity.clone(), true, sensitive, pool).await?;
232 let message = SendActivityTask {
233 activity: serialised_activity.to_owned(),
235 actor_id: actor.actor_id()?,
236 private_key: actor.private_key().context(location_info!())?,
238 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
239 do_send(message, &Client::default()).await?;
241 activity_sender.queue::<SendActivityTask>(message)?;
248 #[derive(Clone, Debug, Deserialize, Serialize)]
249 struct SendActivityTask {
256 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
257 /// if the delivery failed.
258 impl ActixJob for SendActivityTask {
259 type State = MyState;
260 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
261 const NAME: &'static str = "SendActivityTask";
263 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
264 const BACKOFF: Backoff = Backoff::Exponential(2);
266 fn run(self, state: Self::State) -> Self::Future {
267 Box::pin(async move { do_send(self, &state.client).await })
271 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
272 let mut headers = BTreeMap::<String, String>::new();
273 headers.insert("Content-Type".into(), "application/json".into());
274 let result = sign_and_send(
278 task.activity.clone(),
280 task.private_key.to_owned(),
284 if let Err(e) = result {
287 "Failed to send activity {} to {}",
295 pub fn create_activity_queue() -> QueueHandle {
296 // Start the application server. This guards access to to the jobs store
297 let queue_handle = create_server(Storage::new());
299 // Configure and start our workers
300 WorkerConfig::new(|| MyState {
301 client: Client::default(),
303 .register::<SendActivityTask>()
304 .start(queue_handle.clone());