crates/apub/src/fetcher/community.rs

   1 use crate::{
   2   fetcher::{
   3     fetch::fetch_remote_object,
   4     is_deleted,
   5     person::get_or_fetch_and_upsert_person,
   6     should_refetch_actor,
   7   },
   8   inbox::person_inbox::receive_announce,
   9   objects::FromApub,
  10   GroupExt,
  11 };
  12 use activitystreams::{
  13   actor::ApActorExt,
  14   collection::{CollectionExt, OrderedCollection},
  15 };
  16 use anyhow::Context;
  17 use diesel::result::Error::NotFound;
  18 use lemmy_api_structs::blocking;
  19 use lemmy_db_queries::{source::community::Community_, ApubObject, Joinable};
  20 use lemmy_db_schema::{
  21   source::community::{Community, CommunityModerator, CommunityModeratorForm},
  22   DbUrl,
  23 };
  24 use lemmy_db_views_actor::community_moderator_view::CommunityModeratorView;
  25 use lemmy_utils::{location_info, LemmyError};
  26 use lemmy_websocket::LemmyContext;
  27 use log::debug;
  28 use url::Url;
  29
  30 /// Get a community from its apub ID.
  31 ///
  32 /// If it exists locally and `!should_refetch_actor()`, it is returned directly from the database.
  33 /// Otherwise it is fetched from the remote instance, stored and returned.
  34 pub(crate) async fn get_or_fetch_and_upsert_community(
  35   apub_id: &Url,
  36   context: &LemmyContext,
  37   recursion_counter: &mut i32,
  38 ) -> Result<Community, LemmyError> {
  39   let apub_id_owned = apub_id.to_owned();
  40   let community = blocking(context.pool(), move |conn| {
  41     Community::read_from_apub_id(conn, &apub_id_owned.into())
  42   })
  43   .await?;
  44
  45   match community {
  46     Ok(c) if !c.local && should_refetch_actor(c.last_refreshed_at) => {
  47       debug!("Fetching and updating from remote community: {}", apub_id);
  48       fetch_remote_community(apub_id, context, Some(c), recursion_counter).await
  49     }
  50     Ok(c) => Ok(c),
  51     Err(NotFound {}) => {
  52       debug!("Fetching and creating remote community: {}", apub_id);
  53       fetch_remote_community(apub_id, context, None, recursion_counter).await
  54     }
  55     Err(e) => Err(e.into()),
  56   }
  57 }
  58
  59 /// Request a community by apub ID from a remote instance, including moderators. If `old_community`,
  60 /// is set, this is an update for a community which is already known locally. If not, we don't know
  61 /// the community yet and also pull the outbox, to get some initial posts.
  62 async fn fetch_remote_community(
  63   apub_id: &Url,
  64   context: &LemmyContext,
  65   old_community: Option<Community>,
  66   request_counter: &mut i32,
  67 ) -> Result<Community, LemmyError> {
  68   let group = fetch_remote_object::<GroupExt>(context.client(), apub_id, request_counter).await;
  69
  70   if let Some(c) = old_community.to_owned() {
  71     if is_deleted(&group) {
  72       blocking(context.pool(), move |conn| {
  73         Community::update_deleted(conn, c.id, true)
  74       })
  75       .await??;
  76     } else if group.is_err() {
  77       // If fetching failed, return the existing data.
  78       return Ok(c);
  79     }
  80   }
  81
  82   let group = group?;
  83   let community =
  84     Community::from_apub(&group, context, apub_id.to_owned(), request_counter, false).await?;
  85
  86   update_community_mods(&group, &community, context, request_counter).await?;
  87
  88   // only fetch outbox for new communities, otherwise this can create an infinite loop
  89   if old_community.is_none() {
  90     let outbox = group.inner.outbox()?.context(location_info!())?;
  91     fetch_community_outbox(context, outbox, &community, request_counter).await?
  92   }
  93
  94   Ok(community)
  95 }
  96
  97 async fn update_community_mods(
  98   group: &GroupExt,
  99   community: &Community,
 100   context: &LemmyContext,
 101   request_counter: &mut i32,
 102 ) -> Result<(), LemmyError> {
 103   let new_moderators = fetch_community_mods(context, group, request_counter).await?;
 104   let community_id = community.id;
 105   let current_moderators = blocking(context.pool(), move |conn| {
 106     CommunityModeratorView::for_community(&conn, community_id)
 107   })
 108   .await??;
 109   // Remove old mods from database which arent in the moderators collection anymore
 110   for mod_user in &current_moderators {
 111     if !new_moderators.contains(&&mod_user.moderator.actor_id.clone().into()) {
 112       let community_moderator_form = CommunityModeratorForm {
 113         community_id: mod_user.community.id,
 114         person_id: mod_user.moderator.id,
 115       };
 116       blocking(context.pool(), move |conn| {
 117         CommunityModerator::leave(conn, &community_moderator_form)
 118       })
 119       .await??;
 120     }
 121   }
 122
 123   // Add new mods to database which have been added to moderators collection
 124   for mod_uri in new_moderators {
 125     let mod_user = get_or_fetch_and_upsert_person(&mod_uri, context, request_counter).await?;
 126     let current_mod_uris: Vec<DbUrl> = current_moderators
 127       .clone()
 128       .iter()
 129       .map(|c| c.moderator.actor_id.clone())
 130       .collect();
 131     if !current_mod_uris.contains(&mod_user.actor_id) {
 132       let community_moderator_form = CommunityModeratorForm {
 133         community_id: community.id,
 134         person_id: mod_user.id,
 135       };
 136       blocking(context.pool(), move |conn| {
 137         CommunityModerator::join(conn, &community_moderator_form)
 138       })
 139       .await??;
 140     }
 141   }
 142
 143   Ok(())
 144 }
 145
 146 async fn fetch_community_outbox(
 147   context: &LemmyContext,
 148   outbox: &Url,
 149   community: &Community,
 150   recursion_counter: &mut i32,
 151 ) -> Result<(), LemmyError> {
 152   let outbox =
 153     fetch_remote_object::<OrderedCollection>(context.client(), outbox, recursion_counter).await?;
 154   let outbox_activities = outbox.items().context(location_info!())?.clone();
 155   let mut outbox_activities = outbox_activities.many().context(location_info!())?;
 156   if outbox_activities.len() > 20 {
 157     outbox_activities = outbox_activities[0..20].to_vec();
 158   }
 159
 160   for activity in outbox_activities {
 161     receive_announce(context, activity, community, recursion_counter).await?;
 162   }
 163
 164   Ok(())
 165 }
 166
 167 pub(crate) async fn fetch_community_mods(
 168   context: &LemmyContext,
 169   group: &GroupExt,
 170   recursion_counter: &mut i32,
 171 ) -> Result<Vec<Url>, LemmyError> {
 172   if let Some(mods_url) = &group.ext_one.moderators {
 173     let mods =
 174       fetch_remote_object::<OrderedCollection>(context.client(), mods_url, recursion_counter)
 175         .await?;
 176     let mods = mods
 177       .items()
 178       .map(|i| i.as_many())
 179       .flatten()
 180       .context(location_info!())?
 181       .iter()
 182       .filter_map(|i| i.as_xsd_any_uri())
 183       .map(|u| u.to_owned())
 184       .collect();
 185     Ok(mods)
 186   } else {
 187     Ok(vec![])
 188   }
 189 }