crates/api_common/src/send_activity.rs

   1 use crate::context::LemmyContext;
   2 use activitypub_federation::config::Data;
   3 use futures::future::BoxFuture;
   4 use lemmy_db_schema::{
   5   newtypes::DbUrl,
   6   source::{comment::Comment, community::Community, person::Person, post::Post},
   7 };
   8 use lemmy_utils::{error::LemmyResult, SYNCHRONOUS_FEDERATION};
   9 use once_cell::sync::{Lazy, OnceCell};
  10 use tokio::{
  11   sync::{
  12     mpsc,
  13     mpsc::{UnboundedReceiver, UnboundedSender, WeakUnboundedSender},
  14     Mutex,
  15   },
  16   task::JoinHandle,
  17 };
  18
  19 type MatchOutgoingActivitiesBoxed =
  20   Box<for<'a> fn(SendActivityData, &'a Data<LemmyContext>) -> BoxFuture<'a, LemmyResult<()>>>;
  21
  22 /// This static is necessary so that activities can be sent out synchronously for tests.
  23 pub static MATCH_OUTGOING_ACTIVITIES: OnceCell<MatchOutgoingActivitiesBoxed> = OnceCell::new();
  24
  25 #[derive(Debug)]
  26 pub enum SendActivityData {
  27   CreatePost(Post),
  28   UpdatePost(Post),
  29   CreateComment(Comment),
  30   DeleteComment(Comment, Person, Community),
  31   RemoveComment(Comment, Person, Community, Option<String>),
  32   UpdateComment(Comment),
  33   LikePostOrComment(DbUrl, Person, Community, i16),
  34 }
  35
  36 // TODO: instead of static, move this into LemmyContext. make sure that stopping the process with
  37 //       ctrl+c still works.
  38 static ACTIVITY_CHANNEL: Lazy<ActivityChannel> = Lazy::new(|| {
  39   let (sender, receiver) = mpsc::unbounded_channel();
  40   let weak_sender = sender.downgrade();
  41   ActivityChannel {
  42     weak_sender,
  43     receiver: Mutex::new(receiver),
  44     keepalive_sender: Mutex::new(Some(sender)),
  45   }
  46 });
  47
  48 pub struct ActivityChannel {
  49   weak_sender: WeakUnboundedSender<SendActivityData>,
  50   receiver: Mutex<UnboundedReceiver<SendActivityData>>,
  51   keepalive_sender: Mutex<Option<UnboundedSender<SendActivityData>>>,
  52 }
  53
  54 impl ActivityChannel {
  55   pub async fn retrieve_activity() -> Option<SendActivityData> {
  56     let mut lock = ACTIVITY_CHANNEL.receiver.lock().await;
  57     lock.recv().await
  58   }
  59
  60   pub async fn submit_activity(
  61     data: SendActivityData,
  62     context: &Data<LemmyContext>,
  63   ) -> LemmyResult<()> {
  64     if *SYNCHRONOUS_FEDERATION {
  65       MATCH_OUTGOING_ACTIVITIES
  66         .get()
  67         .expect("retrieve function pointer")(data, context)
  68       .await?;
  69     }
  70     // could do `ACTIVITY_CHANNEL.keepalive_sender.lock()` instead and get rid of weak_sender,
  71     // not sure which way is more efficient
  72     else if let Some(sender) = ACTIVITY_CHANNEL.weak_sender.upgrade() {
  73       sender.send(data)?;
  74     }
  75     Ok(())
  76   }
  77
  78   pub async fn close(outgoing_activities_task: JoinHandle<LemmyResult<()>>) -> LemmyResult<()> {
  79     ACTIVITY_CHANNEL.keepalive_sender.lock().await.take();
  80     outgoing_activities_task.await??;
  81     Ok(())
  82   }
  83 }