Implement federated user following (fixes #752) (#2577)

* Implement federated user following (fixes #752)

* rewrite send_activity_in_community and add docs, remove default for column pending

* improve migration

* replace null values in db migration

crates/api/src/community/block.rs

@@ -4,7 +4,7 @@ use lemmy_api_common::{
   community::{BlockCommunity, BlockCommunityResponse},
   utils::get_local_user_view_from_jwt,
 };
-use lemmy_apub::protocol::activities::following::undo_follow::UndoFollowCommunity;
+use lemmy_apub::protocol::activities::following::undo_follow::UndoFollow;
 use lemmy_db_schema::{
   source::{
     community::{Community, CommunityFollower, CommunityFollowerForm},
@@ -53,7 +53,7 @@ impl Perform for BlockCommunity {
         .await
         .ok();
       let community = Community::read(context.pool(), community_id).await?;
-      UndoFollowCommunity::send(&local_user_view.person.into(), &community.into(), context).await?;
+      UndoFollow::send(&local_user_view.person.into(), &community.into(), context).await?;
     } else {
       CommunityBlock::unblock(context.pool(), &community_block_form)
         .await

crates/api/src/community/follow.rs

@@ -7,8 +7,8 @@ use lemmy_api_common::{
 use lemmy_apub::{
   objects::community::ApubCommunity,
   protocol::activities::following::{
-    follow::FollowCommunity as FollowCommunityApub,
+    follow::Follow as FollowCommunityApub,
-    undo_follow::UndoFollowCommunity,
+    undo_follow::UndoFollow,
   },
 };
 use lemmy_db_schema::{
@@ -60,8 +60,7 @@ impl Perform for FollowCommunity {
       FollowCommunityApub::send(&local_user_view.person.clone().into(), &community, context)
         .await?;
     } else {
-      UndoFollowCommunity::send(&local_user_view.person.clone().into(), &community, context)
+      UndoFollow::send(&local_user_view.person.clone().into(), &community, context).await?;
-        .await?;
       CommunityFollower::unfollow(context.pool(), &community_follower_form)
         .await
         .map_err(|e| LemmyError::from_error_message(e, "community_follower_already_exists"))?;

crates/apub/src/activities/block/block_user.rs

@@ -97,7 +97,7 @@ impl BlockUser {
       SiteOrCommunity::Community(c) => {
         let activity = AnnouncableActivities::BlockUser(block);
         let inboxes = vec![user.shared_inbox_or_inbox()];
-        send_activity_in_community(activity, mod_, c, inboxes, context).await
+        send_activity_in_community(activity, mod_, c, inboxes, true, context).await
       }
     }
   }

crates/apub/src/activities/block/undo_block_user.rs

@@ -63,7 +63,7 @@ impl UndoBlockUser {
       }
       SiteOrCommunity::Community(c) => {
         let activity = AnnouncableActivities::UndoBlockUser(undo);
-        send_activity_in_community(activity, mod_, c, inboxes, context).await
+        send_activity_in_community(activity, mod_, c, inboxes, true, context).await
       }
     }
   }

crates/apub/src/activities/community/add_mod.rs

@@ -59,7 +59,7 @@ impl AddMod {
 
     let activity = AnnouncableActivities::AddMod(add);
     let inboxes = vec![added_mod.shared_inbox_or_inbox()];
-    send_activity_in_community(activity, actor, community, inboxes, context).await
+    send_activity_in_community(activity, actor, community, inboxes, true, context).await
   }
 }
 

crates/apub/src/activities/community/mod.rs

@@ -2,11 +2,11 @@ use crate::{
   activities::send_lemmy_activity,
   activity_lists::AnnouncableActivities,
   local_instance,
-  objects::community::ApubCommunity,
+  objects::{community::ApubCommunity, person::ApubPerson},
   protocol::activities::community::announce::AnnounceActivity,
-  ActorType,
 };
 use activitypub_federation::{core::object_id::ObjectId, traits::Actor};
+use lemmy_db_schema::source::person::PersonFollower;
 use lemmy_utils::error::LemmyError;
 use lemmy_websocket::LemmyContext;
 use url::Url;
@@ -17,22 +17,47 @@ pub mod remove_mod;
 pub mod report;
 pub mod update;
 
-#[tracing::instrument(skip_all)]
+/// This function sends all activities which are happening in a community to the right inboxes.
-pub(crate) async fn send_activity_in_community<ActorT>(
+/// For example Create/Page, Add/Mod etc, but not private messages.
+///
+/// Activities are sent to the community itself if it lives on another instance. If the community
+/// is local, the activity is directly wrapped into Announce and sent to community followers.
+/// Activities are also sent to those who follow the actor (with exception of moderation activities).
+///
+/// * `activity` - The activity which is being sent
+/// * `actor` - The user who is sending the activity
+/// * `community` - Community inside which the activity is sent
+/// * `inboxes` - Any additional inboxes the activity should be sent to (for example,
+///               to the user who is being promoted to moderator)
+/// * `is_mod_activity` - True for things like Add/Mod, these are not sent to user followers
+pub(crate) async fn send_activity_in_community(
   activity: AnnouncableActivities,
-  actor: &ActorT,
+  actor: &ApubPerson,
   community: &ApubCommunity,
-  mut inboxes: Vec<Url>,
+  extra_inboxes: Vec<Url>,
+  is_mod_action: bool,
   context: &LemmyContext,
-) -> Result<(), LemmyError>
+) -> Result<(), LemmyError> {
-where
+  // send to extra_inboxes
-  ActorT: Actor + ActorType,
+  send_lemmy_activity(context, activity.clone(), actor, extra_inboxes, false).await?;
-{
-  inboxes.push(community.shared_inbox_or_inbox());
-  send_lemmy_activity(context, activity.clone(), actor, inboxes, false).await?;
 
   if community.local {
-    AnnounceActivity::send(activity.try_into()?, community, context).await?;
+    // send directly to community followers
+    AnnounceActivity::send(activity.clone().try_into()?, community, context).await?;
+  } else {
+    // send to the community, which will then forward to followers
+    let inbox = vec![community.shared_inbox_or_inbox()];
+    send_lemmy_activity(context, activity.clone(), actor, inbox, false).await?;
+  }
+
+  // send to those who follow `actor`
+  if !is_mod_action {
+    let inboxes = PersonFollower::list_followers(context.pool(), actor.id)
+      .await?
+      .into_iter()
+      .map(|p| ApubPerson(p).shared_inbox_or_inbox())
+      .collect();
+    send_lemmy_activity(context, activity, actor, inboxes, false).await?;
   }
 
   Ok(())

crates/apub/src/activities/community/remove_mod.rs

@@ -59,7 +59,7 @@ impl RemoveMod {
 
     let activity = AnnouncableActivities::RemoveMod(remove);
     let inboxes = vec![removed_mod.shared_inbox_or_inbox()];
-    send_activity_in_community(activity, actor, community, inboxes, context).await
+    send_activity_in_community(activity, actor, community, inboxes, true, context).await
   }
 }
 

crates/apub/src/activities/community/update.rs

@@ -44,7 +44,7 @@ impl UpdateCommunity {
     };
 
     let activity = AnnouncableActivities::UpdateCommunity(update);
-    send_activity_in_community(activity, actor, &community, vec![], context).await
+    send_activity_in_community(activity, actor, &community, vec![], true, context).await
   }
 }
 

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

@@ -87,7 +87,7 @@ impl CreateOrUpdateComment {
     }
 
     let activity = AnnouncableActivities::CreateOrUpdateComment(create_or_update);
-    send_activity_in_community(activity, actor, &community, inboxes, context).await
+    send_activity_in_community(activity, actor, &community, inboxes, false, context).await
   }
 }
 

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

@@ -63,8 +63,9 @@ impl CreateOrUpdatePost {
     let community: ApubCommunity = Community::read(context.pool(), community_id).await?.into();
 
     let create_or_update = CreateOrUpdatePost::new(post, actor, &community, kind, 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, actor, &community, vec![], context).await?;
+    send_activity_in_community(activity, actor, &community, vec![], is_mod_action, context).await?;
     Ok(())
   }
 }

crates/apub/src/activities/deletion/mod.rs

@@ -65,6 +65,7 @@ pub async fn send_apub_delete_in_community(
   context: &LemmyContext,
 ) -> Result<(), LemmyError> {
   let actor = ApubPerson::from(actor);
+  let is_mod_action = reason.is_some();
   let activity = if deleted {
     let delete = Delete::new(&actor, object, public(), Some(&community), reason, context)?;
     AnnouncableActivities::Delete(delete)
@@ -72,7 +73,15 @@ pub async fn send_apub_delete_in_community(
     let undo = UndoDelete::new(&actor, object, public(), Some(&community), reason, context)?;
     AnnouncableActivities::UndoDelete(undo)
   };
-  send_activity_in_community(activity, &actor, &community.into(), vec![], context).await
+  send_activity_in_community(
+    activity,
+    &actor,
+    &community.into(),
+    vec![],
+    is_mod_action,
+    context,
+  )
+  .await
 }
 
 #[tracing::instrument(skip_all)]

crates/apub/src/activities/following/accept.rs

@@ -1,7 +1,7 @@
 use crate::{
   activities::{generate_activity_id, send_lemmy_activity},
   local_instance,
-  protocol::activities::following::{accept::AcceptFollowCommunity, follow::FollowCommunity},
+  protocol::activities::following::{accept::AcceptFollow, follow::Follow},
   ActorType,
 };
 use activitypub_federation::{
@@ -19,21 +19,21 @@ use lemmy_utils::error::LemmyError;
 use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperation};
 use url::Url;
 
-impl AcceptFollowCommunity {
+impl AcceptFollow {
   #[tracing::instrument(skip_all)]
   pub async fn send(
-    follow: FollowCommunity,
+    follow: Follow,
     context: &LemmyContext,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
-    let community = follow.object.dereference_local(context).await?;
+    let user_or_community = follow.object.dereference_local(context).await?;
     let person = follow
       .actor
       .clone()
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
-    let accept = AcceptFollowCommunity {
+    let accept = AcceptFollow {
-      actor: ObjectId::new(community.actor_id()),
+      actor: ObjectId::new(user_or_community.actor_id()),
       object: follow,
       kind: AcceptType::Accept,
       id: generate_activity_id(
@@ -42,13 +42,13 @@ impl AcceptFollowCommunity {
       )?,
     };
     let inbox = vec![person.shared_inbox_or_inbox()];
-    send_lemmy_activity(context, accept, &community, inbox, true).await
+    send_lemmy_activity(context, accept, &user_or_community, inbox, true).await
   }
 }
 
 /// Handle accepted follows
 #[async_trait::async_trait(?Send)]
-impl ActivityHandler for AcceptFollowCommunity {
+impl ActivityHandler for AcceptFollow {
   type DataType = LemmyContext;
   type Error = LemmyError;
 

crates/apub/src/activities/following/follow.rs

@@ -5,9 +5,10 @@ use crate::{
     verify_person,
     verify_person_in_community,
   },
+  fetcher::user_or_community::UserOrCommunity,
   local_instance,
   objects::{community::ApubCommunity, person::ApubPerson},
-  protocol::activities::following::{accept::AcceptFollowCommunity, follow::FollowCommunity},
+  protocol::activities::following::{accept::AcceptFollow, follow::Follow},
   ActorType,
 };
 use activitypub_federation::{
@@ -17,20 +18,23 @@ use activitypub_federation::{
 };
 use activitystreams_kinds::activity::FollowType;
 use lemmy_db_schema::{
-  source::community::{CommunityFollower, CommunityFollowerForm},
+  source::{
+    community::{CommunityFollower, CommunityFollowerForm},
+    person::{PersonFollower, PersonFollowerForm},
+  },
   traits::Followable,
 };
 use lemmy_utils::error::LemmyError;
 use lemmy_websocket::LemmyContext;
 use url::Url;
 
-impl FollowCommunity {
+impl Follow {
   pub(in crate::activities::following) fn new(
     actor: &ApubPerson,
     community: &ApubCommunity,
     context: &LemmyContext,
-  ) -> Result<FollowCommunity, LemmyError> {
+  ) -> Result<Follow, LemmyError> {
-    Ok(FollowCommunity {
+    Ok(Follow {
       actor: ObjectId::new(actor.actor_id()),
       object: ObjectId::new(community.actor_id()),
       kind: FollowType::Follow,
@@ -56,14 +60,14 @@ impl FollowCommunity {
       .await
       .ok();
 
-    let follow = FollowCommunity::new(actor, community, context)?;
+    let follow = Follow::new(actor, community, context)?;
     let inbox = vec![community.shared_inbox_or_inbox()];
     send_lemmy_activity(context, follow, actor, inbox, true).await
   }
 }
 
 #[async_trait::async_trait(?Send)]
-impl ActivityHandler for FollowCommunity {
+impl ActivityHandler for Follow {
   type DataType = LemmyContext;
   type Error = LemmyError;
 
@@ -82,11 +86,13 @@ impl ActivityHandler for FollowCommunity {
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
     verify_person(&self.actor, context, request_counter).await?;
-    let community = self
+    let object = self
       .object
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
-    verify_person_in_community(&self.actor, &community, context, request_counter).await?;
+    if let UserOrCommunity::Community(c) = object {
+      verify_person_in_community(&self.actor, &c, context, request_counter).await?;
+    }
     Ok(())
   }
 
@@ -96,25 +102,33 @@ impl ActivityHandler for FollowCommunity {
     context: &Data<LemmyContext>,
     request_counter: &mut i32,
   ) -> Result<(), LemmyError> {
-    let person = self
+    let actor = self
       .actor
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
-    let community = self
+    let object = self
       .object
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
-    let community_follower_form = CommunityFollowerForm {
+    match object {
-      community_id: community.id,
+      UserOrCommunity::User(u) => {
-      person_id: person.id,
+        let form = PersonFollowerForm {
-      pending: false,
+          person_id: u.id,
-    };
+          follower_id: actor.id,
+          pending: false,
+        };
+        PersonFollower::follow(context.pool(), &form).await?;
+      }
+      UserOrCommunity::Community(c) => {
+        let form = CommunityFollowerForm {
+          community_id: c.id,
+          person_id: actor.id,
+          pending: false,
+        };
+        CommunityFollower::follow(context.pool(), &form).await?;
+      }
+    }
 
-    // This will fail if they're already a follower, but ignore the error.
+    AcceptFollow::send(self, context, request_counter).await
-    CommunityFollower::follow(context.pool(), &community_follower_form)
-      .await
-      .ok();
-
-    AcceptFollowCommunity::send(self, context, request_counter).await
   }
 }

crates/apub/src/activities/following/undo_follow.rs

@@ -1,8 +1,9 @@
 use crate::{
   activities::{generate_activity_id, send_lemmy_activity, verify_person},
+  fetcher::user_or_community::UserOrCommunity,
   local_instance,
   objects::{community::ApubCommunity, person::ApubPerson},
-  protocol::activities::following::{follow::FollowCommunity, undo_follow::UndoFollowCommunity},
+  protocol::activities::following::{follow::Follow, undo_follow::UndoFollow},
   ActorType,
 };
 use activitypub_federation::{
@@ -13,22 +14,25 @@ use activitypub_federation::{
 };
 use activitystreams_kinds::activity::UndoType;
 use lemmy_db_schema::{
-  source::community::{CommunityFollower, CommunityFollowerForm},
+  source::{
+    community::{CommunityFollower, CommunityFollowerForm},
+    person::{PersonFollower, PersonFollowerForm},
+  },
   traits::Followable,
 };
 use lemmy_utils::error::LemmyError;
 use lemmy_websocket::LemmyContext;
 use url::Url;
 
-impl UndoFollowCommunity {
+impl UndoFollow {
   #[tracing::instrument(skip_all)]
   pub async fn send(
     actor: &ApubPerson,
     community: &ApubCommunity,
     context: &LemmyContext,
   ) -> Result<(), LemmyError> {
-    let object = FollowCommunity::new(actor, community, context)?;
+    let object = Follow::new(actor, community, context)?;
-    let undo = UndoFollowCommunity {
+    let undo = UndoFollow {
       actor: ObjectId::new(actor.actor_id()),
       object,
       kind: UndoType::Undo,
@@ -43,7 +47,7 @@ impl UndoFollowCommunity {
 }
 
 #[async_trait::async_trait(?Send)]
-impl ActivityHandler for UndoFollowCommunity {
+impl ActivityHandler for UndoFollow {
   type DataType = LemmyContext;
   type Error = LemmyError;
 
@@ -77,22 +81,31 @@ impl ActivityHandler for UndoFollowCommunity {
       .actor
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
-    let community = self
+    let object = self
       .object
       .object
       .dereference(context, local_instance(context).await, request_counter)
       .await?;
 
-    let community_follower_form = CommunityFollowerForm {
+    match object {
-      community_id: community.id,
+      UserOrCommunity::User(u) => {
-      person_id: person.id,
+        let form = PersonFollowerForm {
-      pending: false,
+          person_id: u.id,
-    };
+          follower_id: person.id,
+          pending: false,
+        };
+        PersonFollower::unfollow(context.pool(), &form).await?;
+      }
+      UserOrCommunity::Community(c) => {
+        let form = CommunityFollowerForm {
+          community_id: c.id,
+          person_id: person.id,
+          pending: false,
+        };
+        CommunityFollower::unfollow(context.pool(), &form).await?;
+      }
+    }
 
-    // This will fail if they aren't a follower, but ignore the error.
-    CommunityFollower::unfollow(context.pool(), &community_follower_form)
-      .await
-      .ok();
     Ok(())
   }
 }

crates/apub/src/activities/voting/undo_vote.rs

@@ -53,7 +53,7 @@ impl UndoVote {
       id: id.clone(),
     };
     let activity = AnnouncableActivities::UndoVote(undo_vote);
-    send_activity_in_community(activity, actor, &community, vec![], context).await
+    send_activity_in_community(activity, actor, &community, vec![], false, context).await
   }
 }
 

crates/apub/src/activities/voting/vote.rs

@@ -52,7 +52,7 @@ impl Vote {
     let vote = Vote::new(object, actor, kind, context)?;
 
     let activity = AnnouncableActivities::Vote(vote);
-    send_activity_in_community(activity, actor, &community, vec![], context).await
+    send_activity_in_community(activity, actor, &community, vec![], false, context).await
   }
 }
 

crates/apub/src/activity_lists.rs

@@ -17,11 +17,7 @@ use crate::{
         private_message::CreateOrUpdatePrivateMessage,
       },
       deletion::{delete::Delete, delete_user::DeleteUser, undo_delete::UndoDelete},
-      following::{
+      following::{accept::AcceptFollow, follow::Follow, undo_follow::UndoFollow},
-        accept::AcceptFollowCommunity,
-        follow::FollowCommunity,
-        undo_follow::UndoFollowCommunity,
-      },
       voting::{undo_vote::UndoVote, vote::Vote},
     },
     objects::page::Page,
@@ -45,8 +41,8 @@ pub enum SharedInboxActivities {
 #[serde(untagged)]
 #[enum_delegate::implement(ActivityHandler)]
 pub enum GroupInboxActivities {
-  FollowCommunity(FollowCommunity),
+  Follow(Follow),
-  UndoFollowCommunity(UndoFollowCommunity),
+  UndoFollow(UndoFollow),
   Report(Report),
   // This is a catch-all and needs to be last
   AnnouncableActivities(RawAnnouncableActivities),
@@ -56,7 +52,9 @@ pub enum GroupInboxActivities {
 #[serde(untagged)]
 #[enum_delegate::implement(ActivityHandler)]
 pub enum PersonInboxActivities {
-  AcceptFollowCommunity(AcceptFollowCommunity),
+  AcceptFollow(AcceptFollow),
+  UndoFollow(UndoFollow),
+  FollowCommunity(Follow),
   CreateOrUpdatePrivateMessage(CreateOrUpdatePrivateMessage),
   Delete(Delete),
   UndoDelete(UndoDelete),

crates/apub/src/fetcher/user_or_community.rs

@@ -1,6 +1,7 @@
 use crate::{
   objects::{community::ApubCommunity, person::ApubPerson},
   protocol::objects::{group::Group, person::Person},
+  ActorType,
 };
 use activitypub_federation::traits::{Actor, ApubObject};
 use chrono::NaiveDateTime;
@@ -114,3 +115,19 @@ impl Actor for UserOrCommunity {
     unimplemented!()
   }
 }
+
+impl ActorType for UserOrCommunity {
+  fn actor_id(&self) -> Url {
+    match self {
+      UserOrCommunity::User(u) => u.actor_id(),
+      UserOrCommunity::Community(c) => c.actor_id(),
+    }
+  }
+
+  fn private_key(&self) -> Option<String> {
+    match self {
+      UserOrCommunity::User(u) => u.private_key(),
+      UserOrCommunity::Community(c) => c.private_key(),
+    }
+  }
+}

crates/apub/src/objects/person.rs

@@ -37,7 +37,7 @@ use std::ops::Deref;
 use url::Url;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
-pub struct ApubPerson(DbPerson);
+pub struct ApubPerson(pub(crate) DbPerson);
 
 impl Deref for ApubPerson {
   type Target = DbPerson;

crates/apub/src/protocol/activities/following/accept.rs

@@ -1,7 +1,4 @@
-use crate::{
+use crate::{objects::community::ApubCommunity, protocol::activities::following::follow::Follow};
-  objects::community::ApubCommunity,
-  protocol::activities::following::follow::FollowCommunity,
-};
 use activitypub_federation::core::object_id::ObjectId;
 use activitystreams_kinds::activity::AcceptType;
 use serde::{Deserialize, Serialize};
@@ -9,9 +6,9 @@ use url::Url;
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub struct AcceptFollowCommunity {
+pub struct AcceptFollow {
   pub(crate) actor: ObjectId<ApubCommunity>,
-  pub(crate) object: FollowCommunity,
+  pub(crate) object: Follow,
   #[serde(rename = "type")]
   pub(crate) kind: AcceptType,
   pub(crate) id: Url,

crates/apub/src/protocol/activities/following/follow.rs

@@ -1,4 +1,4 @@
-use crate::objects::{community::ApubCommunity, person::ApubPerson};
+use crate::{fetcher::user_or_community::UserOrCommunity, objects::person::ApubPerson};
 use activitypub_federation::core::object_id::ObjectId;
 use activitystreams_kinds::activity::FollowType;
 use serde::{Deserialize, Serialize};
@@ -6,9 +6,9 @@ use url::Url;
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub struct FollowCommunity {
+pub struct Follow {
   pub(crate) actor: ObjectId<ApubPerson>,
-  pub(crate) object: ObjectId<ApubCommunity>,
+  pub(crate) object: ObjectId<UserOrCommunity>,
   #[serde(rename = "type")]
   pub(crate) kind: FollowType,
   pub(crate) id: Url,

crates/apub/src/protocol/activities/following/mod.rs

@@ -5,23 +5,15 @@ pub mod undo_follow;
 #[cfg(test)]
 mod tests {
   use crate::protocol::{
-    activities::following::{
+    activities::following::{accept::AcceptFollow, follow::Follow, undo_follow::UndoFollow},
-      accept::AcceptFollowCommunity,
-      follow::FollowCommunity,
-      undo_follow::UndoFollowCommunity,
-    },
     tests::test_parse_lemmy_item,
   };
 
   #[test]
   fn test_parse_lemmy_accept_follow() {
-    test_parse_lemmy_item::<FollowCommunity>("assets/lemmy/activities/following/follow.json")
+    test_parse_lemmy_item::<Follow>("assets/lemmy/activities/following/follow.json").unwrap();
+    test_parse_lemmy_item::<AcceptFollow>("assets/lemmy/activities/following/accept.json").unwrap();
+    test_parse_lemmy_item::<UndoFollow>("assets/lemmy/activities/following/undo_follow.json")
       .unwrap();
-    test_parse_lemmy_item::<AcceptFollowCommunity>("assets/lemmy/activities/following/accept.json")
-      .unwrap();
-    test_parse_lemmy_item::<UndoFollowCommunity>(
-      "assets/lemmy/activities/following/undo_follow.json",
-    )
-    .unwrap();
   }
 }

crates/apub/src/protocol/activities/following/undo_follow.rs

@@ -1,7 +1,4 @@
-use crate::{
+use crate::{objects::person::ApubPerson, protocol::activities::following::follow::Follow};
-  objects::person::ApubPerson,
-  protocol::activities::following::follow::FollowCommunity,
-};
 use activitypub_federation::core::object_id::ObjectId;
 use activitystreams_kinds::activity::UndoType;
 use serde::{Deserialize, Serialize};
@@ -9,9 +6,9 @@ use url::Url;
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub struct UndoFollowCommunity {
+pub struct UndoFollow {
   pub(crate) actor: ObjectId<ApubPerson>,
-  pub(crate) object: FollowCommunity,
+  pub(crate) object: Follow,
   #[serde(rename = "type")]
   pub(crate) kind: UndoType,
   pub(crate) id: Url,

crates/apub/src/protocol/activities/mod.rs

@@ -21,7 +21,7 @@ mod tests {
       community::announce::AnnounceActivity,
       create_or_update::{comment::CreateOrUpdateComment, post::CreateOrUpdatePost},
       deletion::delete::Delete,
-      following::{follow::FollowCommunity, undo_follow::UndoFollowCommunity},
+      following::{follow::Follow, undo_follow::UndoFollow},
       voting::{undo_vote::UndoVote, vote::Vote},
     },
     tests::test_json,
@@ -36,15 +36,15 @@ mod tests {
   fn test_parse_pleroma_activities() {
     test_json::<CreateOrUpdateComment>("assets/pleroma/activities/create_note.json").unwrap();
     test_json::<Delete>("assets/pleroma/activities/delete.json").unwrap();
-    test_json::<FollowCommunity>("assets/pleroma/activities/follow.json").unwrap();
+    test_json::<Follow>("assets/pleroma/activities/follow.json").unwrap();
   }
 
   #[test]
   fn test_parse_mastodon_activities() {
     test_json::<CreateOrUpdateComment>("assets/mastodon/activities/create_note.json").unwrap();
     test_json::<Delete>("assets/mastodon/activities/delete.json").unwrap();
-    test_json::<FollowCommunity>("assets/mastodon/activities/follow.json").unwrap();
+    test_json::<Follow>("assets/mastodon/activities/follow.json").unwrap();
-    test_json::<UndoFollowCommunity>("assets/mastodon/activities/undo_follow.json").unwrap();
+    test_json::<UndoFollow>("assets/mastodon/activities/undo_follow.json").unwrap();
     test_json::<Vote>("assets/mastodon/activities/like_page.json").unwrap();
     test_json::<UndoVote>("assets/mastodon/activities/undo_like_page.json").unwrap();
   }

crates/db_schema/src/impls/community.rs

@@ -20,13 +20,7 @@ use crate::{
   utils::{functions::lower, get_conn, DbPool},
   SubscribedType,
 };
-use diesel::{
+use diesel::{dsl::insert_into, result::Error, ExpressionMethods, QueryDsl, TextExpressionMethods};
-  dsl::{exists, insert_into},
-  result::Error,
-  ExpressionMethods,
-  QueryDsl,
-  TextExpressionMethods,
-};
 use diesel_async::RunQueryDsl;
 
 mod safe_type {
@@ -265,7 +259,7 @@ impl CommunityFollower {
   pub fn to_subscribed_type(follower: &Option<Self>) -> SubscribedType {
     match follower {
       Some(f) => {
-        if f.pending.unwrap_or(false) {
+        if f.pending {
           SubscribedType::Pending
         } else {
           SubscribedType::Subscribed
@@ -280,17 +274,14 @@ impl CommunityFollower {
 #[async_trait]
 impl Followable for CommunityFollower {
   type Form = CommunityFollowerForm;
-  async fn follow(
+  async fn follow(pool: &DbPool, form: &CommunityFollowerForm) -> Result<Self, Error> {
-    pool: &DbPool,
-    community_follower_form: &CommunityFollowerForm,
-  ) -> Result<Self, Error> {
     use crate::schema::community_follower::dsl::{community_follower, community_id, person_id};
     let conn = &mut get_conn(pool).await?;
     insert_into(community_follower)
-      .values(community_follower_form)
+      .values(form)
       .on_conflict((community_id, person_id))
       .do_update()
-      .set(community_follower_form)
+      .set(form)
       .get_result::<Self>(conn)
       .await
   }
@@ -315,31 +306,17 @@ impl Followable for CommunityFollower {
     .get_result::<Self>(conn)
     .await
   }
-  async fn unfollow(
+  async fn unfollow(pool: &DbPool, form: &CommunityFollowerForm) -> Result<usize, Error> {
-    pool: &DbPool,
-    community_follower_form: &CommunityFollowerForm,
-  ) -> Result<usize, Error> {
     use crate::schema::community_follower::dsl::{community_follower, community_id, person_id};
     let conn = &mut get_conn(pool).await?;
     diesel::delete(
       community_follower
-        .filter(community_id.eq(&community_follower_form.community_id))
+        .filter(community_id.eq(&form.community_id))
-        .filter(person_id.eq(&community_follower_form.person_id)),
+        .filter(person_id.eq(&form.person_id)),
     )
     .execute(conn)
     .await
   }
-  // TODO: this function name only makes sense if you call it with a remote community. for a local
-  //       community, it will also return true if only remote followers exist
-  async fn has_local_followers(pool: &DbPool, community_id_: CommunityId) -> Result<bool, Error> {
-    use crate::schema::community_follower::dsl::{community_follower, community_id};
-    let conn = &mut get_conn(pool).await?;
-    diesel::select(exists(
-      community_follower.filter(community_id.eq(community_id_)),
-    ))
-    .get_result(conn)
-    .await
-  }
 }
 
 #[async_trait]
@@ -472,7 +449,7 @@ mod tests {
       id: inserted_community_follower.id,
       community_id: inserted_community.id,
       person_id: inserted_person.id,
-      pending: Some(false),
+      pending: false,
       published: inserted_community_follower.published,
     };
 

crates/db_schema/src/impls/person.rs

@@ -1,5 +1,5 @@
 use crate::{
-  newtypes::{DbUrl, PersonId},
+  newtypes::{CommunityId, DbUrl, PersonId},
   schema::person::dsl::{
     actor_id,
     avatar,
@@ -13,8 +13,14 @@ use crate::{
     person,
     updated,
   },
-  source::person::{Person, PersonInsertForm, PersonUpdateForm},
+  source::person::{
-  traits::{ApubActor, Crud},
+    Person,
+    PersonFollower,
+    PersonFollowerForm,
+    PersonInsertForm,
+    PersonUpdateForm,
+  },
+  traits::{ApubActor, Crud, Followable},
   utils::{functions::lower, get_conn, naive_now, DbPool},
 };
 use diesel::{dsl::insert_into, result::Error, ExpressionMethods, QueryDsl, TextExpressionMethods};
@@ -219,14 +225,57 @@ impl ApubActor for Person {
   }
 }
 
+#[async_trait]
+impl Followable for PersonFollower {
+  type Form = PersonFollowerForm;
+  async fn follow(pool: &DbPool, form: &PersonFollowerForm) -> Result<Self, Error> {
+    use crate::schema::person_follower::dsl::{follower_id, person_follower, person_id};
+    let conn = &mut get_conn(pool).await?;
+    insert_into(person_follower)
+      .values(form)
+      .on_conflict((follower_id, person_id))
+      .do_update()
+      .set(form)
+      .get_result::<Self>(conn)
+      .await
+  }
+  async fn follow_accepted(_: &DbPool, _: CommunityId, _: PersonId) -> Result<Self, Error> {
+    unimplemented!()
+  }
+  async fn unfollow(pool: &DbPool, form: &PersonFollowerForm) -> Result<usize, Error> {
+    use crate::schema::person_follower::dsl::{follower_id, person_follower, person_id};
+    let conn = &mut get_conn(pool).await?;
+    diesel::delete(
+      person_follower
+        .filter(follower_id.eq(&form.follower_id))
+        .filter(person_id.eq(&form.person_id)),
+    )
+    .execute(conn)
+    .await
+  }
+}
+
+impl PersonFollower {
+  pub async fn list_followers(pool: &DbPool, person_id_: PersonId) -> Result<Vec<Person>, Error> {
+    use crate::schema::{person, person_follower, person_follower::person_id};
+    let conn = &mut get_conn(pool).await?;
+    person_follower::table
+      .inner_join(person::table)
+      .filter(person_id.eq(person_id_))
+      .select(person::all_columns)
+      .load(conn)
+      .await
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use crate::{
     source::{
       instance::Instance,
-      person::{Person, PersonInsertForm, PersonUpdateForm},
+      person::{Person, PersonFollower, PersonFollowerForm, PersonInsertForm, PersonUpdateForm},
     },
-    traits::Crud,
+    traits::{Crud, Followable},
     utils::build_db_pool_for_tests,
   };
   use serial_test::serial;
@@ -288,4 +337,42 @@ mod tests {
     assert_eq!(expected_person, updated_person);
     assert_eq!(1, num_deleted);
   }
+
+  #[tokio::test]
+  #[serial]
+  async fn follow() {
+    let pool = &build_db_pool_for_tests().await;
+    let inserted_instance = Instance::create(pool, "my_domain.tld").await.unwrap();
+
+    let person_form_1 = PersonInsertForm::builder()
+      .name("erich".into())
+      .public_key("pubkey".to_string())
+      .instance_id(inserted_instance.id)
+      .build();
+    let person_1 = Person::create(pool, &person_form_1).await.unwrap();
+    let person_form_2 = PersonInsertForm::builder()
+      .name("michele".into())
+      .public_key("pubkey".to_string())
+      .instance_id(inserted_instance.id)
+      .build();
+    let person_2 = Person::create(pool, &person_form_2).await.unwrap();
+
+    let follow_form = PersonFollowerForm {
+      person_id: person_1.id,
+      follower_id: person_2.id,
+      pending: false,
+    };
+    let person_follower = PersonFollower::follow(pool, &follow_form).await.unwrap();
+    assert_eq!(person_1.id, person_follower.person_id);
+    assert_eq!(person_2.id, person_follower.follower_id);
+    assert!(!person_follower.pending);
+
+    let followers = PersonFollower::list_followers(pool, person_1.id)
+      .await
+      .unwrap();
+    assert_eq!(vec![person_2], followers);
+
+    let unfollow = PersonFollower::unfollow(pool, &follow_form).await.unwrap();
+    assert_eq!(1, unfollow);
+  }
 }

crates/db_schema/src/schema.rs

@@ -125,7 +125,7 @@ table! {
         community_id -> Int4,
         person_id -> Int4,
         published -> Timestamp,
-        pending -> Nullable<Bool>,
+        pending -> Bool,
     }
 }
 
@@ -729,6 +729,16 @@ table! {
   }
 }
 
+table! {
+    person_follower (id) {
+        id -> Int4,
+        person_id -> Int4,
+        follower_id -> Int4,
+        published -> Timestamp,
+        pending -> Bool,
+    }
+}
+
 joinable!(person_block -> person (person_id));
 
 joinable!(comment -> person (creator_id));
@@ -797,6 +807,7 @@ joinable!(site_language -> language (language_id));
 joinable!(site_language -> site (site_id));
 joinable!(community_language -> language (language_id));
 joinable!(community_language -> community (community_id));
+joinable!(person_follower -> person (follower_id));
 
 joinable!(admin_purge_comment -> person (admin_person_id));
 joinable!(admin_purge_comment -> post (post_id));
@@ -873,4 +884,5 @@ allow_tables_to_appear_in_same_query!(
   federation_blocklist,
   local_site,
   local_site_rate_limit,
+  person_follower
 );

crates/db_schema/src/source/community.rs

@@ -170,7 +170,7 @@ pub struct CommunityFollower {
   pub community_id: CommunityId,
   pub person_id: PersonId,
   pub published: chrono::NaiveDateTime,
-  pub pending: Option<bool>,
+  pub pending: bool,
 }
 
 #[derive(Clone)]

crates/db_schema/src/source/person.rs

@@ -1,6 +1,6 @@
 use crate::newtypes::{DbUrl, InstanceId, PersonId};
 #[cfg(feature = "full")]
-use crate::schema::person;
+use crate::schema::{person, person_follower};
 use serde::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 
@@ -113,3 +113,24 @@ pub struct PersonUpdateForm {
   pub bot_account: Option<bool>,
   pub ban_expires: Option<Option<chrono::NaiveDateTime>>,
 }
+
+#[derive(PartialEq, Eq, Debug)]
+#[cfg_attr(feature = "full", derive(Identifiable, Queryable, Associations))]
+#[cfg_attr(feature = "full", diesel(belongs_to(crate::source::person::Person)))]
+#[cfg_attr(feature = "full", diesel(table_name = person_follower))]
+pub struct PersonFollower {
+  pub id: i32,
+  pub person_id: PersonId,
+  pub follower_id: PersonId,
+  pub published: chrono::NaiveDateTime,
+  pub pending: bool,
+}
+
+#[derive(Clone)]
+#[cfg_attr(feature = "full", derive(Insertable, AsChangeset))]
+#[cfg_attr(feature = "full", diesel(table_name = person_follower))]
+pub struct PersonFollowerForm {
+  pub person_id: PersonId,
+  pub follower_id: PersonId,
+  pub pending: bool,
+}

crates/db_schema/src/traits.rs

@@ -44,7 +44,6 @@ pub trait Followable {
   async fn unfollow(pool: &DbPool, form: &Self::Form) -> Result<usize, Error>
   where
     Self: Sized;
-  async fn has_local_followers(pool: &DbPool, community_id: CommunityId) -> Result<bool, Error>;
 }
 
 #[async_trait]

migrations/2022-11-21-204256_user-following/down.sql

@@ -0,0 +1,3 @@
+drop table person_follower;
+
+alter table community_follower alter column pending drop not null;

migrations/2022-11-21-204256_user-following/up.sql

@@ -0,0 +1,12 @@
+-- create user follower table with two references to persons
+create table person_follower (
+    id serial primary key,
+    person_id int references person on update cascade on delete cascade not null,
+    follower_id int references person on update cascade on delete cascade not null,
+    published timestamp not null default now(),
+    pending boolean not null,
+    unique (follower_id, person_id)
+);
+
+update community_follower set pending = false where pending is null;
+alter table community_follower alter column pending set not null;