Use collection for moderators, instead of `attributedTo` (ref #1061)

crates/apub/src/extensions/context.rs

@@ -11,7 +11,8 @@ pub(crate) fn lemmy_context() -> Result<Vec<AnyBase>, LemmyError> {
     "comments_enabled": {
       "kind": "sc:Boolean",
       "id": "pt:commentsEnabled"
-    }
+    },
+    "moderators": "as:moderators"
   }))?;
   Ok(vec![AnyBase::from(context()), context_ext])
 }

crates/apub/src/extensions/group_extensions.rs

@@ -1,7 +1,11 @@
-use activitystreams::unparsed::UnparsedMutExt;
+use activitystreams::{
+  collection::{CollectionExt, OrderedCollection},
+  unparsed::UnparsedMutExt,
+};
 use activitystreams_ext::UnparsedExtension;
 use lemmy_utils::LemmyError;
 use serde::{Deserialize, Serialize};
+use url::Url;
 
 /// Activitystreams extension to allow (de)serializing additional Community field
 /// `sensitive` (called 'nsfw' in Lemmy).
@@ -9,12 +13,17 @@ use serde::{Deserialize, Serialize};
 #[serde(rename_all = "camelCase")]
 pub struct GroupExtension {
   pub sensitive: Option<bool>,
+  pub moderators: Option<OrderedCollection>,
 }
 
 impl GroupExtension {
-  pub fn new(sensitive: bool) -> Result<GroupExtension, LemmyError> {
+  pub fn new(sensitive: bool, moderators: Vec<Url>) -> Result<GroupExtension, LemmyError> {
+    let mut mods = OrderedCollection::new();
+    mods.set_total_items(moderators.len() as u64);
+    mods.set_many_items(moderators);
     Ok(GroupExtension {
       sensitive: Some(sensitive),
+      moderators: Some(mods),
     })
   }
 }
@@ -28,11 +37,13 @@ where
   fn try_from_unparsed(unparsed_mut: &mut U) -> Result<Self, Self::Error> {
     Ok(GroupExtension {
       sensitive: unparsed_mut.remove("sensitive")?,
+      moderators: unparsed_mut.remove("moderators")?,
     })
   }
 
   fn try_into_unparsed(self, unparsed_mut: &mut U) -> Result<(), Self::Error> {
     unparsed_mut.insert("sensitive", self.sensitive)?;
+    unparsed_mut.insert("moderators", self.moderators)?;
     Ok(())
   }
 }

crates/apub/src/fetcher/community.rs

@@ -1,10 +1,5 @@
 use crate::{
-  fetcher::{
-    fetch::fetch_remote_object,
-    get_or_fetch_and_upsert_user,
-    is_deleted,
-    should_refetch_actor,
-  },
+  fetcher::{fetch::fetch_remote_object, is_deleted, should_refetch_actor},
   inbox::user_inbox::receive_announce,
   objects::FromApub,
   GroupExt,
@@ -12,13 +7,12 @@ use crate::{
 use activitystreams::{
   actor::ApActorExt,
   collection::{CollectionExt, OrderedCollection},
-  object::ObjectExt,
 };
 use anyhow::Context;
 use diesel::result::Error::NotFound;
 use lemmy_api_structs::blocking;
-use lemmy_db_queries::{source::community::Community_, ApubObject, Joinable};
-use lemmy_db_schema::source::community::{Community, CommunityModerator, CommunityModeratorForm};
+use lemmy_db_queries::{source::community::Community_, ApubObject};
+use lemmy_db_schema::source::community::Community;
 use lemmy_utils::{location_info, LemmyError};
 use lemmy_websocket::LemmyContext;
 use log::debug;
@@ -80,40 +74,6 @@ async fn fetch_remote_community(
   let community =
     Community::from_apub(&group, context, apub_id.to_owned(), recursion_counter).await?;
 
-  // Also add the community moderators too
-  let attributed_to = group.inner.attributed_to().context(location_info!())?;
-  let creator_and_moderator_uris: Vec<&Url> = attributed_to
-    .as_many()
-    .context(location_info!())?
-    .iter()
-    .map(|a| a.as_xsd_any_uri().context(""))
-    .collect::<Result<Vec<&Url>, anyhow::Error>>()?;
-
-  let mut creator_and_moderators = Vec::new();
-
-  for uri in creator_and_moderator_uris {
-    let c_or_m = get_or_fetch_and_upsert_user(uri, context, recursion_counter).await?;
-
-    creator_and_moderators.push(c_or_m);
-  }
-
-  // TODO: need to make this work to update mods of existing communities
-  if old_community.is_none() {
-    let community_id = community.id;
-    blocking(context.pool(), move |conn| {
-      for mod_ in creator_and_moderators {
-        let community_moderator_form = CommunityModeratorForm {
-          community_id,
-          user_id: mod_.id,
-        };
-
-        CommunityModerator::join(conn, &community_moderator_form)?;
-      }
-      Ok(()) as Result<(), LemmyError>
-    })
-    .await??;
-  }
-
   // only fetch outbox for new communities, otherwise this can create an infinite loop
   if old_community.is_none() {
     let outbox = group.inner.outbox()?.context(location_info!())?;

crates/apub/src/objects/community.rs

@@ -23,10 +23,11 @@ use activitystreams::{
 use activitystreams_ext::Ext2;
 use anyhow::Context;
 use lemmy_api_structs::blocking;
-use lemmy_db_queries::DbPool;
+use lemmy_db_queries::{DbPool, Joinable};
 use lemmy_db_schema::{
   naive_now,
-  source::community::{Community, CommunityForm},
+  source::community::{Community, CommunityForm, CommunityModerator, CommunityModeratorForm},
+  DbUrl,
 };
 use lemmy_db_views_actor::community_moderator_view::CommunityModeratorView;
 use lemmy_utils::{
@@ -51,18 +52,13 @@ impl ToApub for Community {
       CommunityModeratorView::for_community(&conn, id)
     })
     .await??;
-    let moderators: Vec<Url> = moderators
-      .into_iter()
-      .map(|m| m.moderator.actor_id.into_inner())
-      .collect();
 
     let mut group = ApObject::new(Group::new());
     group
       .set_many_contexts(lemmy_context()?)
       .set_id(self.actor_id.to_owned().into())
       .set_name(self.title.to_owned())
-      .set_published(convert_datetime(self.published))
-      .set_many_attributed_tos(moderators);
+      .set_published(convert_datetime(self.published));
 
     if let Some(u) = self.updated.to_owned() {
       group.set_updated(convert_datetime(u));
@@ -93,9 +89,14 @@ impl ToApub for Community {
         ..Default::default()
       });
 
+    let moderators: Vec<Url> = moderators
+      .into_iter()
+      .map(|m| m.moderator.actor_id.into_inner())
+      .collect();
+
     Ok(Ext2::new(
       ap_actor,
-      GroupExtension::new(self.nsfw)?,
+      GroupExtension::new(self.nsfw, moderators)?,
       self.get_public_key_ext()?,
     ))
   }
@@ -114,14 +115,57 @@ impl ToApub for Community {
 impl FromApub for Community {
   type ApubType = GroupExt;
 
-  /// Converts a `Group` to `Community`.
+  /// Converts a `Group` to `Community`, inserts it into the database and updates moderators.
   async fn from_apub(
     group: &GroupExt,
     context: &LemmyContext,
     expected_domain: Url,
     request_counter: &mut i32,
   ) -> Result<Community, LemmyError> {
-    get_object_from_apub(group, context, expected_domain, request_counter).await
+    let community: Community =
+      get_object_from_apub(group, context, expected_domain, request_counter).await?;
+
+    let new_moderators = get_community_moderators(group)?;
+    let community_id = community.id;
+    let current_moderators = blocking(context.pool(), move |conn| {
+      CommunityModeratorView::for_community(&conn, community_id)
+    })
+    .await??;
+    // Remove old mods from database which arent in the moderators collection anymore
+    for mod_user in &current_moderators {
+      if !new_moderators.contains(&&mod_user.moderator.actor_id.clone().into()) {
+        let community_moderator_form = CommunityModeratorForm {
+          community_id: mod_user.community.id,
+          user_id: mod_user.moderator.id,
+        };
+        blocking(context.pool(), move |conn| {
+          CommunityModerator::leave(conn, &community_moderator_form)
+        })
+        .await??;
+      }
+    }
+
+    // Add new mods to database which have been added to moderators collection
+    for mod_uri in new_moderators {
+      let mod_user = get_or_fetch_and_upsert_user(&mod_uri, context, request_counter).await?;
+      let current_mod_uris: Vec<DbUrl> = current_moderators
+        .clone()
+        .iter()
+        .map(|c| c.moderator.actor_id.clone())
+        .collect();
+      if !current_mod_uris.contains(&mod_user.actor_id) {
+        let community_moderator_form = CommunityModeratorForm {
+          community_id: community.id,
+          user_id: mod_user.id,
+        };
+        blocking(context.pool(), move |conn| {
+          CommunityModerator::join(conn, &community_moderator_form)
+        })
+        .await??;
+      }
+    }
+
+    Ok(community)
   }
 }
 
@@ -133,15 +177,8 @@ impl FromApubToForm<GroupExt> for CommunityForm {
     expected_domain: Url,
     request_counter: &mut i32,
   ) -> Result<Self, LemmyError> {
-    let creator_and_moderator_uris = group.inner.attributed_to().context(location_info!())?;
-    let creator_uri = creator_and_moderator_uris
-      .as_many()
-      .context(location_info!())?
-      .iter()
-      .next()
-      .context(location_info!())?
-      .as_xsd_any_uri()
-      .context(location_info!())?;
+    let moderator_uris = get_community_moderators(group)?;
+    let creator_uri = moderator_uris.first().context(location_info!())?;
 
     let creator = get_or_fetch_and_upsert_user(creator_uri, context, request_counter).await?;
     let name = group
@@ -226,3 +263,20 @@ impl FromApubToForm<GroupExt> for CommunityForm {
     })
   }
 }
+
+fn get_community_moderators(group: &GroupExt) -> Result<Vec<&Url>, LemmyError> {
+  if let Some(moderators) = &group.ext_one.moderators {
+    Ok(
+      moderators
+        .items()
+        .map(|i| i.as_many())
+        .flatten()
+        .context(location_info!())?
+        .iter()
+        .filter_map(|i| i.as_xsd_any_uri())
+        .collect(),
+    )
+  } else {
+    Ok(vec![])
+  }
+}

docker/federation/docker-compose.yml

@@ -29,7 +29,7 @@ services:
       - ./volumes/pictrs_alpha:/mnt
 
   lemmy-alpha-ui:
-    image: dessalines/lemmy-ui:0.9.9
+    image: lemmy-ui:test
     environment:
       - LEMMY_INTERNAL_HOST=lemmy-alpha:8541
       - LEMMY_EXTERNAL_HOST=localhost:8541
@@ -58,7 +58,7 @@ services:
       - ./volumes/postgres_alpha:/var/lib/postgresql/data
 
   lemmy-beta-ui:
-    image: dessalines/lemmy-ui:0.9.9
+    image: lemmy-ui:test
     environment:
       - LEMMY_INTERNAL_HOST=lemmy-beta:8551
       - LEMMY_EXTERNAL_HOST=localhost:8551

docker/federation/lemmy_alpha.hjson

@@ -1,4 +1,5 @@
 {
+  hostname: lemmy-alpha:8541
   port: 8541
   tls_enabled: false
   jwt_secret: changeme