crates/apub/src/collections/community_outbox.rs

   1 use crate::{
   2   activity_lists::AnnouncableActivities,
   3   collections::CommunityContext,
   4   generate_outbox_url,
   5   objects::post::ApubPost,
   6   protocol::{
   7     activities::community::announce::AnnounceActivity,
   8     collections::group_outbox::GroupOutbox,
   9   },
  10 };
  11 use activitypub_federation::{
  12   data::Data,
  13   traits::{ActivityHandler, ApubObject},
  14   utils::verify_domains_match,
  15 };
  16 use activitystreams_kinds::collection::OrderedCollectionType;
  17 use chrono::NaiveDateTime;
  18 use futures::future::join_all;
  19 use lemmy_api_common::utils::blocking;
  20 use lemmy_db_schema::source::post::Post;
  21 use lemmy_utils::error::LemmyError;
  22 use url::Url;
  23
  24 #[derive(Clone, Debug)]
  25 pub(crate) struct ApubCommunityOutbox(Vec<ApubPost>);
  26
  27 #[async_trait::async_trait(?Send)]
  28 impl ApubObject for ApubCommunityOutbox {
  29   type DataType = CommunityContext;
  30   type ApubType = GroupOutbox;
  31   type Error = LemmyError;
  32
  33   fn last_refreshed_at(&self) -> Option<NaiveDateTime> {
  34     None
  35   }
  36
  37   #[tracing::instrument(skip_all)]
  38   async fn read_from_apub_id(
  39     _object_id: Url,
  40     data: &Self::DataType,
  41   ) -> Result<Option<Self>, LemmyError> {
  42     // Only read from database if its a local community, otherwise fetch over http
  43     if data.0.local {
  44       let community_id = data.0.id;
  45       let post_list: Vec<ApubPost> = blocking(data.1.pool(), move |conn| {
  46         Post::list_for_community(conn, community_id)
  47       })
  48       .await??
  49       .into_iter()
  50       .map(Into::into)
  51       .collect();
  52       Ok(Some(ApubCommunityOutbox(post_list)))
  53     } else {
  54       Ok(None)
  55     }
  56   }
  57
  58   async fn delete(self, _data: &Self::DataType) -> Result<(), LemmyError> {
  59     // do nothing (it gets deleted automatically with the community)
  60     Ok(())
  61   }
  62
  63   #[tracing::instrument(skip_all)]
  64   async fn into_apub(self, data: &Self::DataType) -> Result<Self::ApubType, LemmyError> {
  65     let mut ordered_items = vec![];
  66     for post in self.0 {
  67       let page = post.into_apub(&data.1).await?;
  68       let announcable = AnnouncableActivities::Page(page);
  69       let announce = AnnounceActivity::new(announcable, &data.0, &data.1)?;
  70       ordered_items.push(announce);
  71     }
  72
  73     Ok(GroupOutbox {
  74       r#type: OrderedCollectionType::OrderedCollection,
  75       id: generate_outbox_url(&data.0.actor_id)?.into(),
  76       total_items: ordered_items.len() as i32,
  77       ordered_items,
  78     })
  79   }
  80
  81   #[tracing::instrument(skip_all)]
  82   async fn verify(
  83     group_outbox: &GroupOutbox,
  84     expected_domain: &Url,
  85     _context: &CommunityContext,
  86     _request_counter: &mut i32,
  87   ) -> Result<(), LemmyError> {
  88     verify_domains_match(expected_domain, &group_outbox.id)?;
  89     Ok(())
  90   }
  91
  92   #[tracing::instrument(skip_all)]
  93   async fn from_apub(
  94     apub: Self::ApubType,
  95     data: &Self::DataType,
  96     _request_counter: &mut i32,
  97   ) -> Result<Self, LemmyError> {
  98     let mut outbox_activities = apub.ordered_items;
  99     if outbox_activities.len() > 20 {
 100       outbox_activities = outbox_activities[0..20].to_vec();
 101     }
 102
 103     // We intentionally ignore errors here. This is because the outbox might contain posts from old
 104     // Lemmy versions, or from other software which we cant parse. In that case, we simply skip the
 105     // item and only parse the ones that work.
 106     let data = Data::new(data.1.clone());
 107     // process items in parallel, to avoid long delay from fetch_site_metadata() and other processing
 108     join_all(outbox_activities.into_iter().map(|activity| {
 109       async {
 110         // use separate request counter for each item, otherwise there will be problems with
 111         // parallel processing
 112         let request_counter = &mut 0;
 113         let verify = activity.verify(&data, request_counter).await;
 114         if verify.is_ok() {
 115           activity.receive(&data, request_counter).await.ok();
 116         }
 117       }
 118     }))
 119     .await;
 120
 121     // This return value is unused, so just set an empty vec
 122     Ok(ApubCommunityOutbox(Vec::new()))
 123   }
 124
 125   type DbType = ();
 126 }