2 check_is_apub_id_valid,
3 community::do_announce,
4 extensions::signatures::sign,
9 base::{Extends, ExtendsExt},
12 use anyhow::{anyhow, Context, Error};
13 use background_jobs::{
15 memory_storage::Storage,
22 use itertools::Itertools;
23 use lemmy_db::{community::Community, user::User_, DbPool};
24 use lemmy_utils::{location_info, settings::Settings, LemmyError};
25 use lemmy_websocket::LemmyContext;
28 use serde::{Deserialize, Serialize};
29 use std::{collections::BTreeMap, future::Future, pin::Pin};
32 pub async fn send_activity_single_dest<T, Kind>(
34 creator: &dyn ActorType,
36 context: &LemmyContext,
37 ) -> Result<(), LemmyError>
39 T: AsObject<Kind> + Extends<Kind>,
41 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
43 if check_is_apub_id_valid(&to).is_ok() {
44 send_activity_internal(
45 context.activity_queue(),
57 pub async fn send_to_community_followers<T, Kind>(
59 community: &Community,
60 context: &LemmyContext,
61 sender_shared_inbox: Option<Url>,
62 ) -> Result<(), LemmyError>
64 T: AsObject<Kind> + Extends<Kind>,
66 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
68 // dont send to the local instance, nor to the instance where the activity originally came from,
69 // because that would result in a database error (same data inserted twice)
70 let community_shared_inbox = community.get_shared_inbox_url()?;
71 let to: Vec<Url> = community
72 .get_follower_inboxes(context.pool())
75 .filter(|inbox| Some(inbox) != sender_shared_inbox.as_ref().as_ref())
76 .filter(|inbox| inbox != &&community_shared_inbox)
77 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
79 .map(|inbox| inbox.to_owned())
82 send_activity_internal(
83 context.activity_queue(),
94 pub async fn send_to_community<T, Kind>(
96 community: &Community,
98 context: &LemmyContext,
99 ) -> Result<(), LemmyError>
101 T: AsObject<Kind> + Extends<Kind>,
103 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
105 // if this is a local community, we need to do an announce from the community instead
107 do_announce(activity.into_any_base()?, &community, creator, context).await?;
109 let inbox = community.get_shared_inbox_url()?;
110 check_is_apub_id_valid(&inbox)?;
111 send_activity_internal(
112 context.activity_queue(),
124 pub async fn send_comment_mentions<T, Kind>(
128 context: &LemmyContext,
129 ) -> Result<(), LemmyError>
131 T: AsObject<Kind> + Extends<Kind>,
133 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
135 let mentions = mentions
137 .filter(|inbox| check_is_apub_id_valid(inbox).is_ok())
138 .map(|i| i.to_owned())
140 send_activity_internal(
141 context.activity_queue(),
151 /// Asynchronously sends the given `activity` from `actor` to every inbox URL in `to`.
153 /// The caller of this function needs to remove any blocked domains from `to`,
154 /// using `check_is_apub_id_valid()`.
155 async fn send_activity_internal<T, Kind>(
156 activity_sender: &QueueHandle,
158 actor: &dyn ActorType,
161 ) -> Result<(), LemmyError>
163 T: AsObject<Kind> + Extends<Kind>,
165 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
167 if !Settings::get().federation.enabled {
172 assert!(check_is_apub_id_valid(&to_url).is_ok());
175 let activity = activity.into_any_base()?;
176 let serialised_activity = serde_json::to_string(&activity)?;
177 insert_activity(actor.user_id(), serialised_activity.clone(), true, pool).await?;
179 // TODO: it would make sense to create a separate task for each destination server
180 let message = SendActivityTask {
181 activity: serialised_activity,
183 actor_id: actor.actor_id()?,
184 private_key: actor.private_key().context(location_info!())?,
187 activity_sender.queue::<SendActivityTask>(message)?;
192 #[derive(Clone, Debug, Deserialize, Serialize)]
193 struct SendActivityTask {
200 impl ActixJob for SendActivityTask {
201 type State = MyState;
202 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
203 const NAME: &'static str = "SendActivityTask";
205 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
206 const BACKOFF: Backoff = Backoff::Exponential(2);
208 fn run(self, state: Self::State) -> Self::Future {
209 Box::pin(async move {
210 for to_url in &self.to {
211 let mut headers = BTreeMap::<String, String>::new();
212 headers.insert("Content-Type".into(), "application/json".into());
217 self.activity.clone(),
219 self.private_key.to_owned(),
223 if let Err(e) = result {
226 "Failed to send activity {} to {}",
237 pub fn create_activity_queue() -> QueueHandle {
238 // Start the application server. This guards access to to the jobs store
239 let queue_handle = create_server(Storage::new());
241 // Configure and start our workers
242 WorkerConfig::new(|| MyState {
243 client: Client::default(),
245 .register::<SendActivityTask>()
246 .start(queue_handle.clone());