[ActivityPub] Caching of Following/Followers interactions and collections

Follow interaction:
- Fixed mini-bug where the subscriber profile was being used as the subscribed
- Updated cache subscription-related values in both instances
- Tested and working with local GS instances

Unfollow interaction:
- Updated cache subscription-related values in both instances
- Tested and working with local GS instances

Followers/Following collections:
- Now returning ActivityPub profiles only
- Stored collections in cache

Misc:
- Fix bug concerning the retrieval of public/private-key after in-function generation

plugins/ActivityPub/ActivityPubPlugin.php

@@ -387,7 +387,7 @@ class ActivityPubPlugin extends Plugin
                 ActivityPubPlugin::actor_uri($object->getProfile()),
                 'application/activity+json'
             );
-            $xrd->links[] = clone ($link);
+            $xrd->links[] = clone($link);
         }
     }
 
@@ -559,8 +559,7 @@ class ActivityPubPlugin extends Plugin
      * @throws HTTP_Request2_Exception
      * @author Diogo Cordeiro <diogo@fc.up.pt>
      */
-    public function onStartSubscribe(Profile $profile, Profile $other)
+    public function onStartSubscribe(Profile $profile, Profile $other) {
-    {
         if (!$profile->isLocal() && $other->isLocal()) {
             return true;
         }

plugins/ActivityPub/actions/apactorfollowers.php

@@ -70,10 +70,10 @@ class apActorFollowersAction extends ManagedAction
         }
 
         $since = ($page - 1) * PROFILES_PER_MINILIST;
-        $limit = (($page - 1) == 0 ? 1 : $page) * PROFILES_PER_MINILIST;
+        $limit = PROFILES_PER_MINILIST;
 
         /* Calculate total items */
-        $total_subs  = $profile->subscriberCount();
+        $total_subs  = Activitypub_profile::subscriberCount($profile);
         $total_pages = ceil($total_subs / PROFILES_PER_MINILIST);
 
         $res = [
@@ -118,15 +118,15 @@ class apActorFollowersAction extends ManagedAction
     {
         /* Fetch Followers */
         try {
-            $sub = $profile->getSubscribers($since, $limit);
+            $sub = Activitypub_profile::getSubscribers($profile, $since, $limit);
         } catch (NoResultException $e) {
             // Just let the exception go on its merry way
         }
 
         /* Get followers' URLs */
         $subs = [];
-        while ($sub->fetch()) {
+        foreach ($sub as $s) {
-            $subs[] = ActivityPubPlugin::actor_uri($sub);
+            $subs[] = ActivityPubPlugin::actor_uri($s);
         }
 
         return $subs;

plugins/ActivityPub/actions/apactorfollowing.php

@@ -70,10 +70,10 @@ class apActorFollowingAction extends ManagedAction
         }
 
         $since = ($page - 1) * PROFILES_PER_MINILIST;
-        $limit = (($page - 1) == 0 ? 1 : $page) * PROFILES_PER_MINILIST;
+        $limit = PROFILES_PER_MINILIST;
 
         /* Calculate total items */
-        $total_subs  = $profile->subscriptionCount();
+        $total_subs  = Activitypub_profile::subscriptionCount($profile);
         $total_pages = ceil($total_subs / PROFILES_PER_MINILIST);
 
         $res = [
@@ -118,16 +118,17 @@ class apActorFollowingAction extends ManagedAction
     {
         /* Fetch Following */
         try {
-            $sub = $profile->getSubscribed($since, $limit);
+            $sub = Activitypub_profile::getSubscribed($profile, $since, $limit);
         } catch (NoResultException $e) {
             // Just let the exception go on its merry way
         }
 
         /* Get followed' URLs */
         $subs = [];
-        while ($sub->fetch()) {
+        foreach ($sub as $s) {
-            $subs[] = ActivityPubPlugin::actor_uri($sub);
+            $subs[] = ActivityPubPlugin::actor_uri($s);
         }
+
         return $subs;
     }
 }

plugins/ActivityPub/classes/Activitypub_follow.php

@@ -78,6 +78,7 @@ class Activitypub_follow extends Managed_DataObject
 
         if (!Subscription::exists($actor_profile, $object_profile)) {
             Subscription::start($actor_profile, $object_profile);
+            Activitypub_profile::subscribeCacheUpdate($actor_profile, $object_profile);
             common_debug('ActivityPubPlugin: Accepted Follow request from '.ActivityPubPlugin::actor_uri($actor_profile).' to '.$object);
         } else {
             common_debug('ActivityPubPlugin: Received a repeated Follow request from '.ActivityPubPlugin::actor_uri($actor_profile).' to '.$object);
@@ -85,7 +86,7 @@ class Activitypub_follow extends Managed_DataObject
 
         // Notify remote instance that we have accepted their request
         common_debug('ActivityPubPlugin: Notifying remote instance that we have accepted their Follow request request from '.ActivityPubPlugin::actor_uri($actor_profile).' to '.$object);
-        $postman = new Activitypub_postman($actor_profile, [$actor_aprofile]);
+        $postman = new Activitypub_postman($object_profile, [$actor_aprofile]);
         $postman->accept_follow();
     }
 }

plugins/ActivityPub/classes/Activitypub_profile.php

@@ -473,4 +473,196 @@ class Activitypub_profile extends Managed_DataObject
 
         return $profile;
     }
+
+    /**
+     * Getter for the number of subscribers of a
+     * given local profile
+     *
+     * @param Profile $profile profile object
+     * @return int number of subscribers
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function subscriberCount(Profile $profile): int {
+        $cnt = self::cacheGet(sprintf('activitypub_profile:subscriberCount:%d', $profile->id));
+
+        if ($cnt !== false && is_int($cnt)) {
+            return $cnt;
+        }
+
+        $sub = new Subscription();
+        $sub->subscribed = $profile->id;
+        $sub->joinAdd(['subscriber', 'user:id'], 'LEFT');
+        $sub->joinAdd(['subscriber', 'activitypub_profile:profile_id'], 'LEFT');
+        $sub->whereAdd('subscriber != subscribed');
+        $cnt = $sub->count('distinct subscriber');
+
+        self::cacheSet(sprintf('activitypub_profile:subscriberCount:%d', $profile->id), $cnt);
+
+        return $cnt;
+    }
+
+    /**
+     * Getter for the number of subscriptions of a
+     * given local profile
+     *
+     * @param Profile $profile profile object
+     * @return int number of subscriptions
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function subscriptionCount(Profile $profile): int {
+        $cnt = self::cacheGet(sprintf('activitypub_profile:subscriptionCount:%d', $profile->id));
+
+        if ($cnt !== false && is_int($cnt)) {
+            return $cnt;
+        }
+
+        $sub = new Subscription();
+        $sub->subscriber = $profile->id;
+        $sub->joinAdd(['subscribed', 'user:id'], 'LEFT');
+        $sub->joinAdd(['subscribed', 'activitypub_profile:profile_id'], 'LEFT');
+        $sub->whereAdd('subscriber != subscribed');
+        $cnt = $sub->count('distinct subscribed');
+
+        self::cacheSet(sprintf('activitypub_profile:subscriptionCount:%d', $profile->id), $cnt);
+
+        return $cnt;
+    }
+
+    public static function updateSubscriberCount(Profile $profile, $adder) {
+        $cnt = self::cacheGet(sprintf('activitypub_profile:subscriberCount:%d', $profile->id));
+
+        if ($cnt !== false && is_int($cnt)) {
+            self::cacheSet(sprintf('activitypub_profile:subscriberCount:%d', $profile->id), $cnt+$adder);
+        }
+    }
+
+    public static function updateSubscriptionCount(Profile $profile, $adder) {
+        $cnt = self::cacheGet(sprintf('activitypub_profile:subscriptionCount:%d', $profile->id));
+
+        if ($cnt !== false && is_int($cnt)) {
+            self::cacheSet(sprintf('activitypub_profile:subscriptionCount:%d', $profile->id), $cnt+$adder);
+        }
+    }
+
+    /**
+     * Getter for the subscriber profiles of a
+     * given local profile
+     *
+     * @param Profile $profile profile object
+     * @param int $offset index of the starting row to fetch from
+     * @param int $limit maximum number of rows allowed for fetching
+     * @return array subscriber profile objects
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function getSubscribers(Profile $profile, $offset = 0, $limit = null): array {
+        $cache = false;
+        if ($offset + $limit <= Subscription::CACHE_WINDOW) {
+            $subs = self::cacheGet(sprintf('activitypub_profile:subscriberCollection:%d', $profile->id));
+            if ($subs !== false && is_array($subs)) {
+                return array_slice($subs, $offset, $limit);
+            }
+
+            $cache = true;
+        }
+
+        $subs = Subscription::getSubscriberIDs($profile->id, $offset, $limit);
+        try {
+            $profiles = [];
+
+            $users = User::multiGet('id', $subs);
+            foreach ($users->fetchAll() as $user) {
+                $profiles[$user->id] = $user->getProfile();
+            }
+
+            $ap_profiles = Activitypub_profile::multiGet('profile_id', $subs);
+            foreach ($ap_profiles->fetchAll() as $ap) {
+                $profiles[$ap->getID()] = $ap->local_profile();
+            }
+        } catch (NoResultException $e) {
+            return $e->obj;
+        }
+
+        if ($cache) {
+            self::cacheSet(sprintf('activitypub_profile:subscriberCollection:%d', $profile->id), $profiles);
+        }
+
+        return $profiles;
+    }
+
+    /**
+     * Getter for the subscribed profiles of a
+     * given local profile
+     *
+     * @param Profile $profile profile object
+     * @param int $offset index of the starting row to fetch from
+     * @param int $limit maximum number of rows allowed for fetching
+     * @return array subscribed profile objects
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function getSubscribed(Profile $profile, $offset = 0, $limit = null): array {
+        $cache = false;
+        if ($offset + $limit <= Subscription::CACHE_WINDOW) {
+            $subs = self::cacheGet(sprintf('activitypub_profile:subscribedCollection:%d', $profile->id));
+            if (is_array($subs)) {
+                return array_slice($subs, $offset, $limit);
+            }
+
+            $cache = true;
+        }
+
+        $subs = Subscription::getSubscribedIDs($profile->id, $offset, $limit);
+        try {
+            $profiles = [];
+
+            $users = User::multiGet('id', $subs);
+            foreach ($users->fetchAll() as $user) {
+                $profiles[$user->id] = $user->getProfile();
+            }
+
+            $ap_profiles = Activitypub_profile::multiGet('profile_id', $subs);
+            foreach ($ap_profiles->fetchAll() as $ap) {
+                $profiles[$ap->getID()] = $ap->local_profile();
+            }
+        } catch (NoResultException $e) {
+            return $e->obj;
+        }
+
+        if ($cache) {
+           self::cacheSet(sprintf('activitypub_profile:subscribedCollection:%d', $profile->id), $profiles);
+        }
+
+        return $profiles;
+    }
+
+    /**
+     * Update cached values that are relevant to
+     * the users involved in a subscription
+     *
+     * @param Profile $actor subscriber profile object
+     * @param Profile $other subscribed profile object
+     * @return void
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function subscribeCacheUpdate(Profile $actor, Profile $other) {
+        self::blow('activitypub_profile:subscribedCollection:%d', $actor->getID());
+        self::blow('activitypub_profile:subscriberCollection:%d', $other->id);
+        self::updateSubscriptionCount($actor, +1);
+        self::updateSubscriberCount($other, +1);
+    }
+
+    /**
+     * Update cached values that are relevant to
+     * the users involved in an unsubscription
+     *
+     * @param Profile $actor subscriber profile object
+     * @param Profile $other subscribed profile object
+     * @return void
+     * @author Bruno Casteleiro <brunoccast@fc.up.pt>
+     */
+    public static function unsubscribeCacheUpdate(Profile $actor, Profile $other) {
+        self::blow('activitypub_profile:subscribedCollection:%d', $actor->getID());
+        self::blow('activitypub_profile:subscriberCollection:%d', $other->id);
+        self::updateSubscriptionCount($actor, -1);
+        self::updateSubscriberCount($other, -1);
+    }
 }

plugins/ActivityPub/classes/Activitypub_rsa.php

@@ -75,6 +75,7 @@ class Activitypub_rsa extends Managed_DataObject
             if ($profile->isLocal()) {
                 self::generate_keys($this->private_key, $this->public_key);
                 $this->store_keys();
+                $apRSA->private_key = $this->private_key;
             } else {
                 throw new Exception('This is a remote Profile, there is no Private Key for this Profile.');
             }
@@ -100,6 +101,7 @@ class Activitypub_rsa extends Managed_DataObject
             if ($profile->isLocal()) {
                 self::generate_keys($this->private_key, $this->public_key);
                 $this->store_keys();
+                $apRSA->public_key = $this->public_key;
             } else {
                 // This should never happen, but try to recover!
                 if ($fetch) {

plugins/ActivityPub/lib/inbox_handler.php

@@ -175,10 +175,14 @@ class Activitypub_inbox_handler
     private function handle_accept_follow($actor, $object)
     {
         // Get valid Object profile
+        // Note that, since this an accept_follow, the $object
+        // profile is actually the actor that followed someone
         $object_profile = new Activitypub_explorer;
         $object_profile = $object_profile->lookup($object['object'])[0];
 
-        $pending_list = new Activitypub_pending_follow_requests($actor->getID(), $object_profile->getID());
+        Activitypub_profile::subscribeCacheUpdate($object_profile, $actor);
+
+        $pending_list = new Activitypub_pending_follow_requests($object_profile->getID(), $actor->getID());
         $pending_list->remove();
     }
 
@@ -303,6 +307,7 @@ class Activitypub_inbox_handler
         if (Subscription::exists($actor, $object_profile)) {
             Subscription::cancel($actor, $object_profile);
             // You are no longer following this person.
+            Activitypub_profile::unsubscribeCacheUpdate($actor, $object_profile);
         } else {
             // 409: You are not following this person already.
         }

plugins/ActivityPub/lib/postman.php

@@ -61,7 +61,7 @@ class Activitypub_postman
         $followers = apActorFollowersAction::generate_followers($this->actor, 0, null);
         foreach ($followers as $sub) {
             try {
-                $to[]= Activitypub_profile::from_profile($discovery->lookup($sub)[0]);
+                $this->to[]= Activitypub_profile::from_profile($discovery->lookup($sub)[0]);
             } catch (Exception $e) {
                 // Not an ActivityPub Remote Follower, let it go
             }
@@ -157,6 +157,7 @@ class Activitypub_postman
         $res_body = json_decode($res->getBody());
 
         if ($res->getStatus() == 200 || $res->getStatus() == 202 || $res->getStatus() == 409) {
+            Activitypub_profile::unsubscribeCacheUpdate($this->actor, $this->to[0]->local_profile());
             $pending_list = new Activitypub_pending_follow_requests($this->actor->getID(), $this->to[0]->getID());
             $pending_list->remove();
             return true;
@@ -191,7 +192,7 @@ class Activitypub_postman
         $res_body = json_decode($res->getBody());
 
         if ($res->getStatus() == 200 || $res->getStatus() == 202 || $res->getStatus() == 409) {
-            $pending_list = new Activitypub_pending_follow_requests($this->actor->getID(), $this->to[0]->getID());
+            $pending_list = new Activitypub_pending_follow_requests($this->to[0]->getID(), $this->actor->getID());
             $pending_list->remove();
             return true;
         }