Creates activity stream class

bookwyrm/activitystreams.py

@@ -0,0 +1,176 @@
+""" access the activity streams stored in redis """
+from abc import ABC
+from django.dispatch import receiver
+from django.db.models import signals
+from django.db.models import Q
+import redis
+
+from bookwyrm import models, settings
+from bookwyrm.views.helpers import privacy_filter
+
+r = redis.Redis(
+    host=settings.REDIS_ACTIVITY_HOST, port=settings.REDIS_ACTIVITY_PORT, db=0
+)
+
+
+class ActivityStream(ABC):
+    """ a category of activity stream (like home, local, federated) """
+
+    def stream_id(self, user):
+        """ the redis key for this user's instance of this stream """
+        return '{}-{}'.format(user.id, self.key)
+
+    def unread_id(self, user):
+        """ the redis key for this user's unread count for this stream """
+        return '{}-unread'.format(self.stream_id(user))
+
+    def add_status(self, status):
+        """ add a status to users' feeds """
+        value = self.get_value(status)
+        # we want to do this as a bulk operation, hence "pipeline"
+        pipeline = r.pipeline()
+        for user in self.stream_users(status):
+            # add the status to the feed
+            pipeline.zadd(self.stream_id(user), value)
+
+            # add to the unread status count
+            pipeline.incr(self.unread_id(user))
+        # and go!
+        pipeline.execute()
+
+
+    def get_value(self, status):  # pylint: disable=no-self-use
+        """ the status id and the rank (ie, published date) """
+        return {status.id: status.published_date.timestamp()}
+
+
+    def get_activity_stream(self, user):
+        """ load the ids for statuses to be displayed """
+        # clear unreads for this feed
+        r.set(self.unread_id(user), 0)
+
+        statuses = r.zrevrange(self.stream_id(user), 0, -1)
+        return models.Status.objects.select_subclasses().filter(
+            id__in=statuses
+        ).order_by('-published_date')
+
+
+    def populate_stream(self, user):
+        ''' go from zero to a timeline '''
+        pipeline = r.pipeline()
+        statuses = self.stream_statuses(user)
+
+        stream_id = self.stream_id(user)
+        for status in statuses.all()[:settings.MAX_STREAM_LENGTH]:
+            pipeline.zadd(stream_id, self.get_value(status))
+        pipeline.execute()
+
+
+    def stream_users(self, status):  # pylint: disable=no-self-use
+        """ given a status, what users should see it """
+        # direct messages don't appeard in feeds.
+        if status.privacy == 'direct':
+            return None
+
+        # everybody who could plausibly see this status
+        audience = models.User.objects.filter(
+            is_active=True,
+            local=True  # we only create feeds for users of this instance
+        ).exclude(
+            Q(id__in=status.user.blocks.all()) | Q(blocks=status.user)  # not blocked
+        )
+
+        # only visible to the poster's followers and tagged users
+        if status.privacy == 'followers':
+            audience = audience.filter(
+                Q(id=status.user.id)  # if the user is the post's author
+                | Q(following=status.user) # if the user is following the author
+            )
+        return audience
+
+
+    def stream_statuses(self, user):  # pylint: disable=no-self-use
+        """ given a user, what statuses should they see on this stream """
+        return privacy_filter(
+            user,
+            models.Status.objects.select_subclasses(),
+            privacy_levels=["public", 'unlisted', 'followers'],
+        )
+
+
+class HomeStream(ActivityStream):
+    """ users you follow """
+    key = 'home'
+
+    def stream_users(self, status):
+        audience = super().stream_users(status)
+        return audience.filter(
+            Q(id=status.user.id)  # if the user is the post's author
+            | Q(following=status.user) # if the user is following the author
+            | Q(id__in=status.mention_users.all()) # or the user is mentioned
+        )
+
+    def stream_statuses(self, user):
+        return privacy_filter(
+            user,
+            models.Status.objects.select_subclasses(),
+            privacy_levels=["public", 'unlisted', 'followers'],
+            following_only=True
+        )
+
+
+class LocalStream(ActivityStream):
+    """ users you follow """
+    key = 'local'
+
+    def stream_users(self, status):
+        # this stream wants no part in non-public statuses
+        if status.privacy != 'public':
+            return None
+        return super().stream_users(status)
+
+    def stream_statuses(self, user):
+        # all public statuses by a local user
+        return privacy_filter(
+            user,
+            models.Status.objects.select_subclasses().filter(user__local=True),
+            privacy_levels=["public"],
+        )
+
+
+class FederatedStream(ActivityStream):
+    """ users you follow """
+    key = 'federated'
+
+    def stream_users(self, status):
+        # this stream wants no part in non-public statuses
+        if status.privacy != 'public':
+            return None
+        return super().stream_users(status)
+
+    def stream_statuses(self, user):
+        return privacy_filter(
+            user,
+            models.Status.objects.select_subclasses(),
+            privacy_levels=["public"],
+        )
+
+
+
+streams = {
+    'home': HomeStream(),
+    'local': LocalStream(),
+    'federated': FederatedStream(),
+}
+
+@receiver(signals.post_save)
+# pylint: disable=unused-argument
+def update_feeds(sender, instance, created, *args, **kwargs):
+    """ add statuses to activity feeds """
+    # we're only interested in new statuses that aren't dms
+    if not created or not issubclass(sender, models.Status) or \
+            instance.privacy == 'direct':
+        return
+
+    for stream in streams.values():
+        stream.add_status(instance)