+
+#[tracing::instrument(skip_all)]
+async fn send_lemmy_activity<Activity, ActorT>(
+ data: &Data<LemmyContext>,
+ activity: Activity,
+ actor: &ActorT,
+ mut inbox: Vec<Url>,
+ sensitive: bool,
+) -> Result<(), LemmyError>
+where
+ Activity: ActivityHandler + Serialize + Send + Sync + Clone,
+ ActorT: Actor,
+ Activity: ActivityHandler<Error = LemmyError>,
+{
+ static CACHE: Lazy<Cache<(), Arc<Vec<String>>>> = Lazy::new(|| {
+ Cache::builder()
+ .max_capacity(1)
+ .time_to_live(DEAD_INSTANCE_LIST_CACHE_DURATION)
+ .build()
+ });
+ let dead_instances = CACHE
+ .try_get_with((), async {
+ Ok::<_, diesel::result::Error>(Arc::new(Instance::dead_instances(&mut data.pool()).await?))
+ })
+ .await?;
+
+ inbox.retain(|i| {
+ let domain = i.domain().expect("has domain").to_string();
+ !dead_instances.contains(&domain)
+ });
+ info!("Sending activity {}", activity.id().to_string());
+ let activity = WithContext::new(activity, CONTEXT.deref().clone());
+
+ let form = SentActivityForm {
+ ap_id: activity.id().clone().into(),
+ data: serde_json::to_value(activity.clone())?,
+ sensitive,
+ };
+ SentActivity::create(&mut data.pool(), form).await?;
+ send_activity(activity, actor, inbox, data).await?;
+
+ Ok(())
+}
+
+pub async fn handle_outgoing_activities(context: Data<LemmyContext>) -> LemmyResult<()> {
+ while let Some(data) = ActivityChannel::retrieve_activity().await {
+ match_outgoing_activities(data, &context.reset_request_count()).await?
+ }
+ Ok(())
+}
+
+pub async fn match_outgoing_activities(
+ data: SendActivityData,
+ context: &Data<LemmyContext>,
+) -> LemmyResult<()> {
+ let fed_task = match data {
+ SendActivityData::CreatePost(post) => {
+ let creator_id = post.creator_id;
+ CreateOrUpdatePage::send(
+ post,
+ creator_id,
+ CreateOrUpdateType::Create,
+ context.reset_request_count(),
+ )
+ }
+ };
+ if *SYNCHRONOUS_FEDERATION {
+ fed_task.await?;
+ } else {
+ spawn_try_task(fed_task);
+ }
+ Ok(())
+}