Remove SendActivity and Perform traits, rely on channel (#3596)

* Remove SendActivity and Perform traits, rely on channel

These traits arent necessary anymore now that websocket is removed.
Removing them allows us to use normal actix http handler methods
which are much more flexible, and allow using different middlewares
as well as setting response attributes.

* compiling and create post federating

* clippy

* rename methods, join outgoing activities task

* fix api tests

* no unwrap

* conditional compile

* add back getrandom

* make crates optional

* fmt

Cargo.lock

@@ -2633,6 +2633,7 @@ dependencies = [
 name = "lemmy_api_common"
 version = "0.18.1"
 dependencies = [
+ "activitypub_federation",
  "actix-web",
  "anyhow",
  "chrono",
@@ -2644,6 +2645,7 @@ dependencies = [
  "lemmy_db_views_actor",
  "lemmy_db_views_moderator",
  "lemmy_utils",
+ "once_cell",
  "percent-encoding",
  "regex",
  "reqwest",

crates/api_common/Cargo.toml

@@ -22,6 +22,7 @@ full = [
   "lemmy_db_views/full",
   "lemmy_db_views_actor/full",
   "lemmy_db_views_moderator/full",
+  "activitypub_federation",
   "percent-encoding",
   "encoding",
   "reqwest-middleware",
@@ -32,6 +33,7 @@ full = [
   "reqwest",
   "actix-web",
   "futures",
+  "once_cell",
 ]
 
 [dependencies]
@@ -40,6 +42,7 @@ lemmy_db_views_moderator = { workspace = true }
 lemmy_db_views_actor = { workspace = true }
 lemmy_db_schema = { workspace = true }
 lemmy_utils = { workspace = true, optional = true }
+activitypub_federation = { workspace = true, optional = true }
 serde = { workspace = true }
 serde_with = { workspace = true }
 url = { workspace = true }
@@ -59,5 +62,7 @@ uuid = { workspace = true, optional = true }
 tokio = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true }
 ts-rs = { workspace = true, optional = true }
+once_cell = { workspace = true, optional = true }
 actix-web = { workspace = true, optional = true }
+# necessary for wasmt compilation
 getrandom = { version = "0.2.10", features = ["js"] }

crates/api_common/src/build_response.rs

@@ -64,7 +64,7 @@ pub async fn build_community_response(
 }
 
 pub async fn build_post_response(
-  context: &Data<LemmyContext>,
+  context: &LemmyContext,
   community_id: CommunityId,
   person_id: PersonId,
   post_id: PostId,

crates/api_common/src/lib.rs

@@ -10,6 +10,8 @@ pub mod post;
 pub mod private_message;
 #[cfg(feature = "full")]
 pub mod request;
+#[cfg(feature = "full")]
+pub mod send_activity;
 pub mod sensitive;
 pub mod site;
 #[cfg(feature = "full")]

crates/api_common/src/send_activity.rs

@@ -0,0 +1,58 @@
+use crate::context::LemmyContext;
+use activitypub_federation::config::Data;
+use futures::future::BoxFuture;
+use lemmy_db_schema::source::post::Post;
+use lemmy_utils::{error::LemmyResult, SYNCHRONOUS_FEDERATION};
+use once_cell::sync::{Lazy, OnceCell};
+use tokio::sync::{
+  mpsc,
+  mpsc::{UnboundedReceiver, UnboundedSender},
+  Mutex,
+};
+
+type MatchOutgoingActivitiesBoxed =
+  Box<for<'a> fn(SendActivityData, &'a Data<LemmyContext>) -> BoxFuture<'a, LemmyResult<()>>>;
+
+/// This static is necessary so that activities can be sent out synchronously for tests.
+pub static MATCH_OUTGOING_ACTIVITIES: OnceCell<MatchOutgoingActivitiesBoxed> = OnceCell::new();
+
+#[derive(Debug)]
+pub enum SendActivityData {
+  CreatePost(Post),
+}
+
+static ACTIVITY_CHANNEL: Lazy<ActivityChannel> = Lazy::new(|| {
+  let (sender, receiver) = mpsc::unbounded_channel();
+  ActivityChannel {
+    sender,
+    receiver: Mutex::new(receiver),
+  }
+});
+
+pub struct ActivityChannel {
+  sender: UnboundedSender<SendActivityData>,
+  receiver: Mutex<UnboundedReceiver<SendActivityData>>,
+}
+
+impl ActivityChannel {
+  pub async fn retrieve_activity() -> Option<SendActivityData> {
+    let mut lock = ACTIVITY_CHANNEL.receiver.lock().await;
+    lock.recv().await
+  }
+
+  pub async fn submit_activity(
+    data: SendActivityData,
+    context: &Data<LemmyContext>,
+  ) -> LemmyResult<()> {
+    if *SYNCHRONOUS_FEDERATION {
+      MATCH_OUTGOING_ACTIVITIES
+        .get()
+        .expect("retrieve function pointer")(data, context)
+      .await?;
+    } else {
+      let lock = &ACTIVITY_CHANNEL.sender;
+      lock.send(data)?;
+    }
+    Ok(())
+  }
+}

crates/api_crud/src/lib.rs

@@ -5,7 +5,7 @@ use lemmy_utils::error::LemmyError;
 mod comment;
 mod community;
 mod custom_emoji;
-mod post;
+pub mod post;
 mod private_message;
 mod site;
 mod user;

crates/api_crud/src/post/create.rs

@@ -1,10 +1,11 @@
-use crate::PerformCrud;
+use activitypub_federation::config::Data;
-use actix_web::web::Data;
+use actix_web::web::Json;
 use lemmy_api_common::{
   build_response::build_post_response,
   context::LemmyContext,
   post::{CreatePost, PostResponse},
   request::fetch_site_data,
+  send_activity::{ActivityChannel, SendActivityData},
   utils::{
     check_community_ban,
     check_community_deleted_or_removed,
@@ -40,147 +41,145 @@ use tracing::Instrument;
 use url::Url;
 use webmention::{Webmention, WebmentionError};
 
-#[async_trait::async_trait(?Send)]
+#[tracing::instrument(skip(context))]
-impl PerformCrud for CreatePost {
+pub async fn create_post(
-  type Response = PostResponse;
+  data: Json<CreatePost>,
+  context: Data<LemmyContext>,
+) -> Result<Json<PostResponse>, LemmyError> {
+  let local_user_view = local_user_view_from_jwt(&data.auth, &context).await?;
+  let local_site = LocalSite::read(&mut context.pool()).await?;
 
-  #[tracing::instrument(skip(context))]
+  let slur_regex = local_site_to_slur_regex(&local_site);
-  async fn perform(&self, context: &Data<LemmyContext>) -> Result<PostResponse, LemmyError> {
+  check_slurs(&data.name, &slur_regex)?;
-    let data: &CreatePost = self;
+  check_slurs_opt(&data.body, &slur_regex)?;
-    let local_user_view = local_user_view_from_jwt(&data.auth, context).await?;
+  honeypot_check(&data.honeypot)?;
-    let local_site = LocalSite::read(&mut context.pool()).await?;
 
-    let slur_regex = local_site_to_slur_regex(&local_site);
+  let data_url = data.url.as_ref();
-    check_slurs(&data.name, &slur_regex)?;
+  let url = data_url.map(clean_url_params).map(Into::into); // TODO no good way to handle a "clear"
-    check_slurs_opt(&data.body, &slur_regex)?;
-    honeypot_check(&data.honeypot)?;
 
-    let data_url = data.url.as_ref();
+  is_valid_post_title(&data.name)?;
-    let url = data_url.map(clean_url_params).map(Into::into); // TODO no good way to handle a "clear"
+  is_valid_body_field(&data.body, true)?;
+  check_url_scheme(&data.url)?;
 
-    is_valid_post_title(&data.name)?;
+  check_community_ban(
-    is_valid_body_field(&data.body, true)?;
+    local_user_view.person.id,
-    check_url_scheme(&data.url)?;
+    data.community_id,
-
+    &mut context.pool(),
-    check_community_ban(
+  )
-      local_user_view.person.id,
+  .await?;
-      data.community_id,
+  check_community_deleted_or_removed(data.community_id, &mut context.pool()).await?;
-      &mut context.pool(),
-    )
-    .await?;
-    check_community_deleted_or_removed(data.community_id, &mut context.pool()).await?;
 
+  let community_id = data.community_id;
+  let community = Community::read(&mut context.pool(), community_id).await?;
+  if community.posting_restricted_to_mods {
     let community_id = data.community_id;
-    let community = Community::read(&mut context.pool(), community_id).await?;
+    let is_mod = CommunityView::is_mod_or_admin(
-    if community.posting_restricted_to_mods {
-      let community_id = data.community_id;
-      let is_mod = CommunityView::is_mod_or_admin(
-        &mut context.pool(),
-        local_user_view.local_user.person_id,
-        community_id,
-      )
-      .await?;
-      if !is_mod {
-        return Err(LemmyErrorType::OnlyModsCanPostInCommunity)?;
-      }
-    }
-
-    // Fetch post links and pictrs cached image
-    let (metadata_res, thumbnail_url) =
-      fetch_site_data(context.client(), context.settings(), data_url, true).await;
-    let (embed_title, embed_description, embed_video_url) = metadata_res
-      .map(|u| (u.title, u.description, u.embed_video_url))
-      .unwrap_or_default();
-
-    let language_id = match data.language_id {
-      Some(lid) => Some(lid),
-      None => {
-        default_post_language(
-          &mut context.pool(),
-          community_id,
-          local_user_view.local_user.id,
-        )
-        .await?
-      }
-    };
-    CommunityLanguage::is_allowed_community_language(
       &mut context.pool(),
-      language_id,
+      local_user_view.local_user.person_id,
       community_id,
     )
     .await?;
+    if !is_mod {
+      return Err(LemmyErrorType::OnlyModsCanPostInCommunity)?;
+    }
+  }
 
-    let post_form = PostInsertForm::builder()
+  // Fetch post links and pictrs cached image
-      .name(data.name.trim().to_owned())
+  let (metadata_res, thumbnail_url) =
-      .url(url)
+    fetch_site_data(context.client(), context.settings(), data_url, true).await;
-      .body(data.body.clone())
+  let (embed_title, embed_description, embed_video_url) = metadata_res
-      .community_id(data.community_id)
+    .map(|u| (u.title, u.description, u.embed_video_url))
-      .creator_id(local_user_view.person.id)
+    .unwrap_or_default();
-      .nsfw(data.nsfw)
-      .embed_title(embed_title)
-      .embed_description(embed_description)
-      .embed_video_url(embed_video_url)
-      .language_id(language_id)
-      .thumbnail_url(thumbnail_url)
-      .build();
 
-    let inserted_post = Post::create(&mut context.pool(), &post_form)
+  let language_id = match data.language_id {
-      .await
+    Some(lid) => Some(lid),
-      .with_lemmy_type(LemmyErrorType::CouldntCreatePost)?;
+    None => {
+      default_post_language(
+        &mut context.pool(),
+        community_id,
+        local_user_view.local_user.id,
+      )
+      .await?
+    }
+  };
+  CommunityLanguage::is_allowed_community_language(&mut context.pool(), language_id, community_id)
+    .await?;
 
-    let inserted_post_id = inserted_post.id;
+  let post_form = PostInsertForm::builder()
-    let protocol_and_hostname = context.settings().get_protocol_and_hostname();
+    .name(data.name.trim().to_owned())
-    let apub_id = generate_local_apub_endpoint(
+    .url(url)
-      EndpointType::Post,
+    .body(data.body.clone())
-      &inserted_post_id.to_string(),
+    .community_id(data.community_id)
-      &protocol_and_hostname,
+    .creator_id(local_user_view.person.id)
-    )?;
+    .nsfw(data.nsfw)
-    let updated_post = Post::update(
+    .embed_title(embed_title)
-      &mut context.pool(),
+    .embed_description(embed_description)
-      inserted_post_id,
+    .embed_video_url(embed_video_url)
-      &PostUpdateForm::builder().ap_id(Some(apub_id)).build(),
+    .language_id(language_id)
-    )
+    .thumbnail_url(thumbnail_url)
+    .build();
+
+  let inserted_post = Post::create(&mut context.pool(), &post_form)
     .await
     .with_lemmy_type(LemmyErrorType::CouldntCreatePost)?;
 
-    // They like their own post by default
+  let inserted_post_id = inserted_post.id;
-    let person_id = local_user_view.person.id;
+  let protocol_and_hostname = context.settings().get_protocol_and_hostname();
-    let post_id = inserted_post.id;
+  let apub_id = generate_local_apub_endpoint(
-    let like_form = PostLikeForm {
+    EndpointType::Post,
-      post_id,
+    &inserted_post_id.to_string(),
-      person_id,
+    &protocol_and_hostname,
-      score: 1,
+  )?;
-    };
+  let updated_post = Post::update(
+    &mut context.pool(),
+    inserted_post_id,
+    &PostUpdateForm::builder().ap_id(Some(apub_id)).build(),
+  )
+  .await
+  .with_lemmy_type(LemmyErrorType::CouldntCreatePost)?;
 
-    PostLike::like(&mut context.pool(), &like_form)
+  // They like their own post by default
-      .await
+  let person_id = local_user_view.person.id;
-      .with_lemmy_type(LemmyErrorType::CouldntLikePost)?;
+  let post_id = inserted_post.id;
+  let like_form = PostLikeForm {
+    post_id,
+    person_id,
+    score: 1,
+  };
 
-    // Mark the post as read
+  PostLike::like(&mut context.pool(), &like_form)
-    mark_post_as_read(person_id, post_id, &mut context.pool()).await?;
+    .await
+    .with_lemmy_type(LemmyErrorType::CouldntLikePost)?;
 
-    if let Some(url) = updated_post.url.clone() {
+  ActivityChannel::submit_activity(SendActivityData::CreatePost(updated_post.clone()), &context)
-      let task = async move {
+    .await?;
-        let mut webmention =
+
-          Webmention::new::<Url>(updated_post.ap_id.clone().into(), url.clone().into())?;
+  // Mark the post as read
-        webmention.set_checked(true);
+  mark_post_as_read(person_id, post_id, &mut context.pool()).await?;
-        match webmention
+
-          .send()
+  if let Some(url) = updated_post.url.clone() {
-          .instrument(tracing::info_span!("Sending webmention"))
+    let task = async move {
-          .await
+      let mut webmention =
-        {
+        Webmention::new::<Url>(updated_post.ap_id.clone().into(), url.clone().into())?;
-          Err(WebmentionError::NoEndpointDiscovered(_)) => Ok(()),
+      webmention.set_checked(true);
-          Ok(_) => Ok(()),
+      match webmention
-          Err(e) => Err(e).with_lemmy_type(LemmyErrorType::CouldntSendWebmention),
+        .send()
-        }
+        .instrument(tracing::info_span!("Sending webmention"))
-      };
+        .await
-      if *SYNCHRONOUS_FEDERATION {
+      {
-        task.await?;
+        Err(WebmentionError::NoEndpointDiscovered(_)) => Ok(()),
-      } else {
+        Ok(_) => Ok(()),
-        spawn_try_task(task);
+        Err(e) => Err(e).with_lemmy_type(LemmyErrorType::CouldntSendWebmention),
       }
     };
+    if *SYNCHRONOUS_FEDERATION {
+      task.await?;
+    } else {
+      spawn_try_task(task);
+    }
+  };
 
-    build_post_response(context, community_id, person_id, post_id).await
+  Ok(Json(
-  }
+    build_post_response(&context, community_id, person_id, post_id).await?,
+  ))
 }

crates/api_crud/src/post/mod.rs

@@ -1,4 +1,4 @@
-mod create;
+pub mod create;
 mod delete;
 mod read;
 mod remove;

crates/apub/src/activities/create_or_update/post.rs

@@ -24,7 +24,7 @@ use activitypub_federation::{
 };
 use lemmy_api_common::{
   context::LemmyContext,
-  post::{CreatePost, EditPost, PostResponse},
+  post::{EditPost, PostResponse},
 };
 use lemmy_db_schema::{
   aggregates::structs::PostAggregates,
@@ -39,25 +39,6 @@ use lemmy_db_schema::{
 use lemmy_utils::error::{LemmyError, LemmyErrorType};
 use url::Url;
 
-#[async_trait::async_trait]
-impl SendActivity for CreatePost {
-  type Response = PostResponse;
-
-  async fn send_activity(
-    _request: &Self,
-    response: &Self::Response,
-    context: &Data<LemmyContext>,
-  ) -> Result<(), LemmyError> {
-    CreateOrUpdatePage::send(
-      &response.post_view.post,
-      response.post_view.creator.id,
-      CreateOrUpdateType::Create,
-      context,
-    )
-    .await
-  }
-}
-
 #[async_trait::async_trait]
 impl SendActivity for EditPost {
   type Response = PostResponse;
@@ -68,10 +49,10 @@ impl SendActivity for EditPost {
     context: &Data<LemmyContext>,
   ) -> Result<(), LemmyError> {
     CreateOrUpdatePage::send(
-      &response.post_view.post,
+      response.post_view.post.clone(),
       response.post_view.creator.id,
       CreateOrUpdateType::Update,
-      context,
+      context.reset_request_count(),
     )
     .await
   }
@@ -102,12 +83,12 @@ impl CreateOrUpdatePage {
 
   #[tracing::instrument(skip_all)]
   pub(crate) async fn send(
-    post: &Post,
+    post: Post,
     person_id: PersonId,
     kind: CreateOrUpdateType,
-    context: &Data<LemmyContext>,
+    context: Data<LemmyContext>,
   ) -> Result<(), LemmyError> {
-    let post = ApubPost(post.clone());
+    let post = ApubPost(post);
     let community_id = post.community_id;
     let person: ApubPerson = Person::read(&mut context.pool(), person_id).await?.into();
     let community: ApubCommunity = Community::read(&mut context.pool(), community_id)
@@ -115,8 +96,8 @@ impl CreateOrUpdatePage {
       .into();
 
     let create_or_update =
-      CreateOrUpdatePage::new(post, &person, &community, kind, context).await?;
+      CreateOrUpdatePage::new(post, &person, &community, kind, &context).await?;
-    let is_mod_action = create_or_update.object.is_mod_action(context).await?;
+    let is_mod_action = create_or_update.object.is_mod_action(&context).await?;
     let activity = AnnouncableActivities::CreateOrUpdatePost(create_or_update);
     send_activity_in_community(
       activity,
@@ -124,7 +105,7 @@ impl CreateOrUpdatePage {
       &community,
       vec![],
       is_mod_action,
-      context,
+      &context,
     )
     .await?;
     Ok(())

crates/apub/src/activities/mod.rs

@@ -1,5 +1,6 @@
 use crate::{
   objects::{community::ApubCommunity, person::ApubPerson},
+  protocol::activities::{create_or_update::page::CreateOrUpdatePage, CreateOrUpdateType},
   CONTEXT,
 };
 use activitypub_federation::{
@@ -11,7 +12,10 @@ use activitypub_federation::{
   traits::{ActivityHandler, Actor},
 };
 use anyhow::anyhow;
-use lemmy_api_common::context::LemmyContext;
+use lemmy_api_common::{
+  context::LemmyContext,
+  send_activity::{ActivityChannel, SendActivityData},
+};
 use lemmy_db_schema::{
   newtypes::CommunityId,
   source::{
@@ -21,7 +25,11 @@ use lemmy_db_schema::{
   },
 };
 use lemmy_db_views_actor::structs::{CommunityPersonBanView, CommunityView};
-use lemmy_utils::error::{LemmyError, LemmyErrorExt, LemmyErrorType};
+use lemmy_utils::{
+  error::{LemmyError, LemmyErrorExt, LemmyErrorType, LemmyResult},
+  spawn_try_task,
+  SYNCHRONOUS_FEDERATION,
+};
 use moka::future::Cache;
 use once_cell::sync::Lazy;
 use serde::Serialize;
@@ -197,3 +205,33 @@ where
 
   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(())
+}

src/api_routes_http.rs

@@ -52,7 +52,6 @@ use lemmy_api_common::{
     VerifyEmail,
   },
   post::{
-    CreatePost,
     CreatePostLike,
     CreatePostReport,
     DeletePost,
@@ -93,7 +92,7 @@ use lemmy_api_common::{
     PurgePost,
   },
 };
-use lemmy_api_crud::PerformCrud;
+use lemmy_api_crud::{post::create::create_post, PerformCrud};
 use lemmy_apub::{
   api::{
     list_comments::list_comments,
@@ -175,7 +174,7 @@ pub fn config(cfg: &mut web::ServiceConfig, rate_limit: &RateLimitCell) {
         web::resource("/post")
           .guard(guard::Post())
           .wrap(rate_limit.post())
-          .route(web::post().to(route_post_crud::<CreatePost>)),
+          .route(web::post().to(create_post)),
       )
       .service(
         web::scope("/post")

src/lib.rs

@@ -21,12 +21,17 @@ use lemmy_api_common::{
   context::LemmyContext,
   lemmy_db_views::structs::SiteView,
   request::build_user_agent,
+  send_activity::MATCH_OUTGOING_ACTIVITIES,
   utils::{
     check_private_instance_and_federation_enabled,
     local_site_rate_limit_to_rate_limit_config,
   },
 };
-use lemmy_apub::{VerifyUrlData, FEDERATION_HTTP_FETCH_LIMIT};
+use lemmy_apub::{
+  activities::{handle_outgoing_activities, match_outgoing_activities},
+  VerifyUrlData,
+  FEDERATION_HTTP_FETCH_LIMIT,
+};
 use lemmy_db_schema::{
   source::secret::Secret,
   utils::{build_db_pool, get_database_url, run_migrations},
@@ -165,9 +170,17 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
     .build()
     .expect("Should always be buildable");
 
+  MATCH_OUTGOING_ACTIVITIES
+    .set(Box::new(move |d, c| {
+      Box::pin(match_outgoing_activities(d, c))
+    }))
+    .expect("set function pointer");
+  let request_data = federation_config.to_request_data();
+  let outgoing_activities_task = tokio::task::spawn(handle_outgoing_activities(request_data));
+
   // Create Http server with websocket support
   HttpServer::new(move || {
-    let cors_origin = std::env::var("LEMMY_CORS_ORIGIN");
+    let cors_origin = env::var("LEMMY_CORS_ORIGIN");
     let cors_config = match (cors_origin, cfg!(debug_assertions)) {
       (Ok(origin), false) => Cors::default()
         .allowed_origin(&origin)
@@ -213,6 +226,9 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
   .run()
   .await?;
 
+  // Wait for outgoing apub sends to complete
+  outgoing_activities_task.await??;
+
   Ok(())
 }