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 /// From a local community, send activity to all remote followers.
36 /// * `activity` the apub activity to send
37 /// * `community` the sending community
38 /// * `extra_inbox` actor inbox which should receive the activity, in addition to followers
39 pub(crate) async fn send_to_community_followers<T, Kind>(
41 community: &Community,
42 extra_inbox: Option<Url>,
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 let extra_inbox: Vec<Url> = extra_inbox.into_iter().collect();
51 let follower_inboxes: Vec<Url> = vec![
52 community.get_follower_inboxes(context.pool()).await?,
58 .filter(|inbox| inbox.host_str() != Some(&Settings::get().hostname))
59 .filter(|inbox| check_is_apub_id_valid(inbox, false).is_ok())
60 .map(|inbox| inbox.to_owned())
63 "Sending activity {:?} to followers of {}",
64 &activity.id_unchecked().map(ToString::to_string),
68 send_activity_internal(context, activity, community, follower_inboxes, true, false).await?;
73 /// Sends an activity from a local person to a remote community.
75 /// * `activity` the activity to send
76 /// * `creator` the creator of the activity
77 /// * `community` the destination community
78 /// * `object_actor` if the object of the activity is an actor, it should be passed here so it can
79 /// be sent directly to the actor
81 pub(crate) async fn send_to_community<T, Kind>(
84 community: &Community,
85 object_actor: Option<Url>,
86 context: &LemmyContext,
87 ) -> Result<(), LemmyError>
89 T: AsObject<Kind> + Extends<Kind> + Debug + BaseExt<Kind>,
91 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
93 // if this is a local community, we need to do an announce from the community instead
96 .send_announce(activity.into_any_base()?, object_actor, context)
99 let inbox = community.get_shared_inbox_or_inbox_url();
100 check_is_apub_id_valid(&inbox, false)?;
102 "Sending activity {:?} to community {}",
103 &activity.id_unchecked().map(ToString::to_string),
106 // dont send to object_actor here, as that is responsibility of the community itself
107 send_activity_internal(context, activity, creator, vec![inbox], true, false).await?;
113 pub(crate) async fn send_to_community_new(
114 activity: AnnouncableActivities,
116 actor: &dyn ActorType,
117 community: &Community,
118 additional_inboxes: Vec<Url>,
119 context: &LemmyContext,
120 ) -> Result<(), LemmyError> {
121 // if this is a local community, we need to do an announce from the community instead
123 insert_activity(activity_id, activity.clone(), true, false, context.pool()).await?;
124 AnnounceActivity::send(activity, community, additional_inboxes, context).await?;
126 let mut inboxes = additional_inboxes;
127 inboxes.push(community.get_shared_inbox_or_inbox_url());
128 send_activity_new(context, &activity, activity_id, actor, inboxes, false).await?;
134 pub(crate) async fn send_activity_new<T>(
135 context: &LemmyContext,
138 actor: &dyn ActorType,
141 ) -> Result<(), LemmyError>
145 if !Settings::get().federation.enabled || inboxes.is_empty() {
149 info!("Sending activity {}", activity_id.to_string());
151 // Don't send anything to ourselves
152 // TODO: this should be a debug assert
153 let hostname = Settings::get().get_hostname_without_port()?;
154 let inboxes: Vec<&Url> = inboxes
156 .filter(|i| i.domain().expect("valid inbox url") != hostname)
159 let serialised_activity = serde_json::to_string(&activity)?;
163 serialised_activity.clone(),
171 let message = SendActivityTask {
172 activity: serialised_activity.to_owned(),
174 actor_id: actor.actor_id(),
175 private_key: actor.private_key().context(location_info!())?,
177 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
178 do_send(message, &Client::default()).await?;
180 context.activity_queue.queue::<SendActivityTask>(message)?;
187 /// Create new `SendActivityTasks`, which will deliver the given activity to inboxes, as well as
188 /// handling signing and retrying failed deliveres.
190 /// The caller of this function needs to remove any blocked domains from `to`,
191 /// using `check_is_apub_id_valid()`.
192 async fn send_activity_internal<T, Kind>(
193 context: &LemmyContext,
195 actor: &dyn ActorType,
197 insert_into_db: bool,
199 ) -> Result<(), LemmyError>
201 T: AsObject<Kind> + Extends<Kind> + Debug,
203 <T as Extends<Kind>>::Error: From<serde_json::Error> + Send + Sync + 'static,
205 if !Settings::get().federation.enabled || inboxes.is_empty() {
209 // Don't send anything to ourselves
210 let hostname = Settings::get().get_hostname_without_port()?;
211 let inboxes: Vec<&Url> = inboxes
213 .filter(|i| i.domain().expect("valid inbox url") != hostname)
216 let activity = activity.into_any_base()?;
217 let serialised_activity = serde_json::to_string(&activity)?;
219 // This is necessary because send_comment and send_comment_mentions
220 // might send the same ap_id
222 let id = activity.id().context(location_info!())?;
223 insert_activity(id, activity.clone(), true, sensitive, context.pool()).await?;
227 let message = SendActivityTask {
228 activity: serialised_activity.to_owned(),
230 actor_id: actor.actor_id(),
231 private_key: actor.private_key().context(location_info!())?,
233 if env::var("LEMMY_TEST_SEND_SYNC").is_ok() {
234 do_send(message, &Client::default()).await?;
236 context.activity_queue.queue::<SendActivityTask>(message)?;
243 #[derive(Clone, Debug, Deserialize, Serialize)]
244 struct SendActivityTask {
251 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
252 /// if the delivery failed.
253 impl ActixJob for SendActivityTask {
254 type State = MyState;
255 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
256 const NAME: &'static str = "SendActivityTask";
258 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
259 const BACKOFF: Backoff = Backoff::Exponential(2);
261 fn run(self, state: Self::State) -> Self::Future {
262 Box::pin(async move { do_send(self, &state.client).await })
266 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
267 let mut headers = BTreeMap::<String, String>::new();
268 headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
269 let result = sign_and_send(
273 task.activity.clone(),
275 task.private_key.to_owned(),
279 if let Err(e) = result {
282 "Failed to send activity {} to {}",
290 pub fn create_activity_queue() -> QueueHandle {
291 // Start the application server. This guards access to to the jobs store
292 let queue_handle = create_server(Storage::new());
294 // Configure and start our workers
295 WorkerConfig::new(|| MyState {
296 client: Client::default(),
298 .register::<SendActivityTask>()
299 .start(queue_handle.clone());