+) -> 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 context = context.reset_request_count();
+ let fed_task = async {
+ use SendActivityData::*;
+ match data {
+ CreatePost(post) => {
+ let creator_id = post.creator_id;
+ CreateOrUpdatePage::send(post, creator_id, CreateOrUpdateType::Create, context).await
+ }
+ UpdatePost(post) => {
+ let creator_id = post.creator_id;
+ CreateOrUpdatePage::send(post, creator_id, CreateOrUpdateType::Update, context).await
+ }
+ DeletePost(post, person, data) => {
+ send_apub_delete_in_community_new(
+ person,
+ post.community_id,
+ DeletableObjects::Post(post.into()),
+ None,
+ data.deleted,
+ context,
+ )
+ .await
+ }
+ RemovePost(post, person, data) => {
+ send_apub_delete_in_community_new(
+ person,
+ post.community_id,
+ DeletableObjects::Post(post.into()),
+ data.reason.or_else(|| Some(String::new())),
+ data.removed,
+ context,
+ )
+ .await
+ }
+ LockPost(post, actor, locked) => send_lock_post(post, actor, locked, context).await,
+ FeaturePost(post, actor, featured) => send_feature_post(post, actor, featured, context).await,
+ CreateComment(comment) => {
+ let creator_id = comment.creator_id;
+ CreateOrUpdateNote::send(comment, creator_id, CreateOrUpdateType::Create, context).await
+ }
+ UpdateComment(comment) => {
+ let creator_id = comment.creator_id;
+ CreateOrUpdateNote::send(comment, creator_id, CreateOrUpdateType::Update, context).await
+ }
+ DeleteComment(comment, actor, community) => {
+ let is_deleted = comment.deleted;
+ let deletable = DeletableObjects::Comment(comment.into());
+ send_apub_delete_in_community(actor, community, deletable, None, is_deleted, &context).await
+ }
+ RemoveComment(comment, actor, community, reason) => {
+ let is_removed = comment.removed;
+ let deletable = DeletableObjects::Comment(comment.into());
+ send_apub_delete_in_community(actor, community, deletable, reason, is_removed, &context)
+ .await
+ }
+ LikePostOrComment(object_id, person, community, score) => {
+ send_like_activity(object_id, person, community, score, context).await
+ }
+ FollowCommunity(community, person, follow) => {
+ send_follow_community(community, person, follow, &context).await
+ }
+ UpdateCommunity(actor, community) => send_update_community(community, actor, context).await,
+ DeleteCommunity(actor, community, removed) => {
+ let deletable = DeletableObjects::Community(community.clone().into());
+ send_apub_delete_in_community(actor, community, deletable, None, removed, &context).await
+ }
+ RemoveCommunity(actor, community, reason, removed) => {
+ let deletable = DeletableObjects::Community(community.clone().into());
+ send_apub_delete_in_community(
+ actor,
+ community,
+ deletable,
+ reason.clone().or_else(|| Some(String::new())),
+ removed,
+ &context,
+ )
+ .await
+ }
+ AddModToCommunity(actor, community_id, updated_mod_id, added) => {
+ send_add_mod_to_community(actor, community_id, updated_mod_id, added, context).await
+ }
+ BanFromCommunity(mod_, community_id, target, data) => {
+ send_ban_from_community(mod_, community_id, target, data, context).await
+ }
+ BanFromSite(mod_, target, data) => send_ban_from_site(mod_, target, data, context).await,
+ CreatePrivateMessage(pm) => {
+ send_create_or_update_pm(pm, CreateOrUpdateType::Create, context).await
+ }
+ UpdatePrivateMessage(pm) => {
+ send_create_or_update_pm(pm, CreateOrUpdateType::Update, context).await
+ }
+ DeletePrivateMessage(person, pm, deleted) => {
+ send_apub_delete_private_message(&person.into(), pm, deleted, context).await
+ }
+ DeleteUser(person) => delete_user(person, context).await,
+ CreateReport(url, actor, community, reason) => {
+ Report::send(ObjectId::from(url), actor, community, reason, context).await
+ }
+ }
+ };
+ if *SYNCHRONOUS_FEDERATION {
+ fed_task.await?;
+ } else {
+ spawn_try_task(fed_task);