2 activities::community::announce::{AnnouncableActivities, AnnounceActivity},
3 check_is_apub_id_valid,
4 extensions::signatures::sign_and_send,
8 APUB_JSON_CONTENT_TYPE,
10 use activitystreams::{
11 base::{BaseExt, Extends, ExtendsExt},
14 use anyhow::{anyhow, Context, Error};
15 use background_jobs::{
17 memory_storage::Storage,
24 use itertools::Itertools;
25 use lemmy_db_schema::source::{community::Community, person::Person};
26 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
27 use lemmy_websocket::LemmyContext;
28 use log::{debug, info, warn};
30 use serde::{Deserialize, Serialize};
31 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
34 /// Sends a local activity to a single, remote actor.
36 /// * `activity` the apub activity to be sent
37 /// * `creator` the local actor which created the activity
38 /// * `inbox` the inbox url where the activity should be delivered to
39 pub(crate) async fn send_activity_single_dest<T, Kind>(
41 creator: &dyn ActorType,
43 context: &LemmyContext,
44 ) -> Result<(), LemmyError>
46 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
48 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
50 if check_is_apub_id_valid(&inbox, false).is_ok() {
52 "Sending activity {:?} to {}",
53 &activity.id_unchecked().map(ToString::to_string),
56 send_activity_internal(context, activity, creator, vec![inbox], true, true).await?;
62 /// From a local community, send activity to all remote followers.
64 /// * `activity` the apub activity to send
65 /// * `community` the sending community
66 /// * `extra_inbox` actor inbox which should receive the activity, in addition to followers
67 pub(crate) async fn send_to_community_followers<T, Kind>(
69 community: &Community,
70 extra_inbox: Option<Url>,
71 context: &LemmyContext,
72 ) -> Result<(), LemmyError>
74 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
76 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
78 let extra_inbox: Vec<Url> = extra_inbox.into_iter().collect();
79 let follower_inboxes: Vec<Url> = vec![
80 community.get_follower_inboxes(context.pool()).await?,
86 .filter(|inbox| inbox.host_str() != Some(&Settings::get().hostname()))
87 .filter(|inbox| check_is_apub_id_valid(inbox, false).is_ok())
88 .map(|inbox| inbox.to_owned())
91 "Sending activity {:?} to followers of {}",
92 &activity.id_unchecked().map(ToString::to_string),
96 send_activity_internal(context, activity, community, follower_inboxes, true, false).await?;
101 /// Sends an activity from a local person to a remote community.
103 /// * `activity` the activity to send
104 /// * `creator` the creator of the activity
105 /// * `community` the destination community
106 /// * `object_actor` if the object of the activity is an actor, it should be passed here so it can
107 /// be sent directly to the actor
109 pub(crate) async fn send_to_community<T, Kind>(
112 community: &Community,
113 object_actor: Option<Url>,
114 context: &LemmyContext,
115 ) -> Result<(), LemmyError>
117 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
119 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
121 // if this is a local community, we need to do an announce from the community instead
124 .send_announce(activity.into_any_base()?, object_actor, context)
127 let inbox = community.get_shared_inbox_or_inbox_url();
128 check_is_apub_id_valid(&inbox, false)?;
130 "Sending activity {:?} to community {}",
131 &activity.id_unchecked().map(ToString::to_string),
134 // dont send to object_actor here, as that is responsibility of the community itself
135 send_activity_internal(context, activity, creator, vec![inbox], true, false).await?;
141 pub(crate) async fn send_to_community_new(
142 activity: AnnouncableActivities,
144 actor: &dyn ActorType,
145 community: &Community,
146 additional_inboxes: Vec<Url>,
147 context: &LemmyContext,
148 ) -> Result<(), LemmyError> {
149 // if this is a local community, we need to do an announce from the community instead
151 insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
152 AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
154 let mut inboxes = additional_inboxes;
155 inboxes.push(community.get_shared_inbox_or_inbox_url());
156 send_activity_new(context, &activity, activity_id, actor, inboxes, false).await?;
162 pub(crate) async fn send_activity_new<T>(
163 context: &LemmyContext,
166 actor: &dyn ActorType,
169 ) -> Result<(), LemmyError>
173 if !Settings::get().federation().enabled || inboxes.is_empty() {
177 info!("Sending activity {}", activity_id.to_string());
179 // Don't send anything to ourselves
180 // TODO: this should be a debug assert
181 let hostname = Settings::get().get_hostname_without_port()?;
182 let inboxes: Vec<&Url> = inboxes
184 .filter(|i| i.domain().expect("valid inbox url") != hostname)
187 let serialised_activity = serde_json::to_string(&activity)?;
191 serialised_activity.clone(),
199 let message = SendActivityTask {
200 activity: serialised_activity.to_owned(),
202 actor_id: actor.actor_id(),
203 private_key: actor.private_key().context(location_info!())?,
205 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
206 do_send(message, &Client::default()).await?;
208 context.activity_queue.queue::<SendActivityTask>(message)?;
215 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
216 /// handling signing and retrying failed deliveres.
218 /// The caller of this function needs to remove any blocked domains from `to`,
219 /// using `check_is_apub_id_valid()`.
220 async fn send_activity_internal<T, Kind>(
221 context: &LemmyContext,
223 actor: &dyn ActorType,
225 insert_into_db: bool,
227 ) -> Result<(), LemmyError>
229 T: AsObject<Kind> + Extends<Kind> + Debug,
231 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
233 if !Settings::get().federation().enabled || inboxes.is_empty() {
237 // Don't send anything to ourselves
238 let hostname = Settings::get().get_hostname_without_port()?;
239 let inboxes: Vec<&Url> = inboxes
241 .filter(|i| i.domain().expect("valid inbox url") != hostname)
244 let activity = activity.into_any_base()?;
245 let serialised_activity = serde_json::to_string(&activity)?;
247 // This is necessary because send_comment and send_comment_mentions
248 // might send the same ap_id
250 let id = activity.id().context(location_info!())?;
251 insert_activity(id, activity.clone(), true, sensitive, context.pool()).await?;
255 let message = SendActivityTask {
256 activity: serialised_activity.to_owned(),
258 actor_id: actor.actor_id(),
259 private_key: actor.private_key().context(location_info!())?,
261 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
262 do_send(message, &Client::default()).await?;
264 context.activity_queue.queue::<SendActivityTask>(message)?;
271 #[derive(Clone, Debug, Deserialize, Serialize)]
272 struct SendActivityTask {
279 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
280 /// if the delivery failed.
281 impl ActixJob for SendActivityTask {
282 type State = MyState;
283 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
284 const NAME: &'static str = "SendActivityTask";
286 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
287 const BACKOFF: Backoff = Backoff::Exponential(2);
289 fn run(self, state: Self::State) -> Self::Future {
290 Box::pin(async move { do_send(self, &state.client).await })
294 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
295 let mut headers = BTreeMap::<String, String>::new();
296 headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
297 let result = sign_and_send(
301 task.activity.clone(),
303 task.private_key.to_owned(),
307 if let Err(e) = result {
310 "Failed to send activity {} to {}",
318 pub fn create_activity_queue() -> QueueHandle {
319 // Start the application server. This guards access to to the jobs store
320 let queue_handle = create_server(Storage::new());
322 // Configure and start our workers
323 WorkerConfig::new(|| MyState {
324 client: Client::default(),
326 .register::<SendActivityTask>()
327 .start(queue_handle.clone());