moviewyrm/bookwyrm/activitystreams.py
2021-03-22 18:54:17 -07:00

179 lines
5.6 KiB
Python

""" 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)