2 check_is_apub_id_valid,
3 extensions::signatures::sign_and_send,
6 APUB_JSON_CONTENT_TYPE,
9 base::{BaseExt, Extends, ExtendsExt},
12 use anyhow::{anyhow, Context, Error};
13 use background_jobs::{
15 memory_storage::Storage,
22 use itertools::Itertools;
23 use lemmy_db_queries::DbPool;
24 use lemmy_db_schema::source::{community::Community, user::User_};
25 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
26 use lemmy_websocket::LemmyContext;
27 use log::{debug, warn};
29 use serde::{Deserialize, Serialize};
30 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
33 /// Sends a local activity to a single, remote actor.
35 /// * `activity` the apub activity to be sent
36 /// * `creator` the local actor which created the activity
37 /// * `inbox` the inbox url where the activity should be delivered to
38 pub(crate) async fn send_activity_single_dest<T, Kind>(
40 creator: &dyn ActorType,
42 context: &LemmyContext,
43 ) -> Result<(), LemmyError>
45 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
47 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
49 if check_is_apub_id_valid(&inbox).is_ok() {
51 "Sending activity {:?} to {}",
52 &activity.id_unchecked(),
55 send_activity_internal(
56 context.activity_queue(),
70 /// From a local community, send activity to all remote followers.
72 /// * `activity` the apub activity to send
73 /// * `community` the sending community
74 /// * `sender_shared_inbox` in case of an announce, this should be the shared inbox of the inner
75 /// activities creator, as receiving a known activity will cause an error
76 pub(crate) async fn send_to_community_followers<T, Kind>(
78 community: &Community,
79 context: &LemmyContext,
80 ) -> Result<(), LemmyError>
82 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
84 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
86 let follower_inboxes: Vec<Url> = community
87 .get_follower_inboxes(context.pool())
91 .filter(|inbox| inbox.host_str() != Some(&Settings::get().hostname()))
92 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
93 .map(|inbox| inbox.to_owned())
96 "Sending activity {:?} to followers of {}",
97 &activity.id_unchecked().map(|i| i.to_string()),
101 send_activity_internal(
102 context.activity_queue(),
115 /// Sends an activity from a local user to a remote community.
117 /// * `activity` the activity to send
118 /// * `creator` the creator of the activity
119 /// * `community` the destination community
121 pub(crate) async fn send_to_community<T, Kind>(
124 community: &Community,
125 context: &LemmyContext,
126 ) -> Result<(), LemmyError>
128 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
130 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
132 // if this is a local community, we need to do an announce from the community instead
135 .send_announce(activity.into_any_base()?, context)
138 let inbox = community.get_shared_inbox_or_inbox_url();
139 check_is_apub_id_valid(&inbox)?;
141 "Sending activity {:?} to community {}",
142 &activity.id_unchecked(),
145 send_activity_internal(
146 context.activity_queue(),
160 /// Sends notification to any users mentioned in a comment
162 /// * `creator` user who created the comment
163 /// * `mentions` list of inboxes of users which are mentioned in the comment
164 /// * `activity` either a `Create/Note` or `Update/Note`
165 pub(crate) async fn send_comment_mentions<T, Kind>(
169 context: &LemmyContext,
170 ) -> Result<(), LemmyError>
172 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
174 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
177 "Sending mentions activity {:?} to {:?}",
178 &activity.id_unchecked(),
181 let mentions = mentions
183 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
184 .map(|i| i.to_owned())
186 send_activity_internal(
187 context.activity_queue(),
192 false, // Don't create a new DB row
199 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
200 /// handling signing and retrying failed deliveres.
202 /// The caller of this function needs to remove any blocked domains from `to`,
203 /// using `check_is_apub_id_valid()`.
204 async fn send_activity_internal<T, Kind>(
205 activity_sender: &QueueHandle,
207 actor: &dyn ActorType,
210 insert_into_db: bool,
212 ) -> Result<(), LemmyError>
214 T: AsObject<Kind> + Extends<Kind> + Debug,
216 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
218 if !Settings::get().federation().enabled || inboxes.is_empty() {
222 // Don't send anything to ourselves
223 let hostname = Settings::get().get_hostname_without_port()?;
224 let inboxes: Vec<&Url> = inboxes
226 .filter(|i| i.domain().unwrap() != hostname)
229 let activity = activity.into_any_base()?;
230 let serialised_activity = serde_json::to_string(&activity)?;
232 // This is necessary because send_comment and send_comment_mentions
233 // might send the same ap_id
235 let id = activity.id().context(location_info!())?;
236 insert_activity(id, activity.clone(), true, sensitive, pool).await?;
240 let message = SendActivityTask {
241 activity: serialised_activity.to_owned(),
243 actor_id: actor.actor_id(),
244 private_key: actor.private_key().context(location_info!())?,
246 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
247 do_send(message, &Client::default()).await?;
249 activity_sender.queue::<SendActivityTask>(message)?;
256 #[derive(Clone, Debug, Deserialize, Serialize)]
257 struct SendActivityTask {
264 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
265 /// if the delivery failed.
266 impl ActixJob for SendActivityTask {
267 type State = MyState;
268 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
269 const NAME: &'static str = "SendActivityTask";
271 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
272 const BACKOFF: Backoff = Backoff::Exponential(2);
274 fn run(self, state: Self::State) -> Self::Future {
275 Box::pin(async move { do_send(self, &state.client).await })
279 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
280 let mut headers = BTreeMap::<String, String>::new();
281 headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
282 let result = sign_and_send(
286 task.activity.clone(),
288 task.private_key.to_owned(),
292 if let Err(e) = result {
295 "Failed to send activity {} to {}",
303 pub fn create_activity_queue() -> QueueHandle {
304 // Start the application server. This guards access to to the jobs store
305 let queue_handle = create_server(Storage::new());
307 // Configure and start our workers
308 WorkerConfig::new(|| MyState {
309 client: Client::default(),
311 .register::<SendActivityTask>()
312 .start(queue_handle.clone());