We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Goodreads import floods activity streams
When a user does a goodreads import, the activitstreams manager isn't checking if the publication date is older than the oldest status, and adds all the statuses to the feeds, pushing out newer statuses.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `bookwyrm/activitystreams.py`
Content:
```
1 """ access the activity streams stored in redis """
2 from abc import ABC
3 from django.dispatch import receiver
4 from django.db.models import signals, Q
5 import redis
6 
7 from bookwyrm import models, settings
8 from bookwyrm.views.helpers import privacy_filter
9 
10 r = redis.Redis(
11     host=settings.REDIS_ACTIVITY_HOST, port=settings.REDIS_ACTIVITY_PORT, db=0
12 )
13 
14 
15 class ActivityStream(ABC):
16     """ a category of activity stream (like home, local, federated) """
17 
18     def stream_id(self, user):
19         """ the redis key for this user's instance of this stream """
20         return "{}-{}".format(user.id, self.key)
21 
22     def unread_id(self, user):
23         """ the redis key for this user's unread count for this stream """
24         return "{}-unread".format(self.stream_id(user))
25 
26     def add_status(self, status):
27         """ add a status to users' feeds """
28         # we want to do this as a bulk operation, hence "pipeline"
29         pipeline = r.pipeline()
30         for user in self.stream_users(status):
31             # add the status to the feed
32             pipeline.lpush(self.stream_id(user), status.id)
33             pipeline.ltrim(self.stream_id(user), 0, settings.MAX_STREAM_LENGTH)
34 
35             # add to the unread status count
36             pipeline.incr(self.unread_id(user))
37         # and go!
38         pipeline.execute()
39 
40     def remove_status(self, status):
41         """ remove a status from all feeds """
42         pipeline = r.pipeline()
43         for user in self.stream_users(status):
44             pipeline.lrem(self.stream_id(user), -1, status.id)
45         pipeline.execute()
46 
47     def add_user_statuses(self, viewer, user):
48         """ add a user's statuses to another user's feed """
49         pipeline = r.pipeline()
50         for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:
51             pipeline.lpush(self.stream_id(viewer), status.id)
52         pipeline.execute()
53 
54     def remove_user_statuses(self, viewer, user):
55         """ remove a user's status from another user's feed """
56         pipeline = r.pipeline()
57         for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:
58             pipeline.lrem(self.stream_id(viewer), -1, status.id)
59         pipeline.execute()
60 
61     def get_activity_stream(self, user):
62         """ load the ids for statuses to be displayed """
63         # clear unreads for this feed
64         r.set(self.unread_id(user), 0)
65 
66         statuses = r.lrange(self.stream_id(user), 0, -1)
67         return (
68             models.Status.objects.select_subclasses()
69             .filter(id__in=statuses)
70             .order_by("-published_date")
71         )
72 
73     def get_unread_count(self, user):
74         """ get the unread status count for this user's feed """
75         return int(r.get(self.unread_id(user)))
76 
77     def populate_stream(self, user):
78         """ go from zero to a timeline """
79         pipeline = r.pipeline()
80         statuses = self.stream_statuses(user)
81 
82         stream_id = self.stream_id(user)
83         for status in statuses.all()[: settings.MAX_STREAM_LENGTH]:
84             pipeline.lpush(stream_id, status.id)
85         pipeline.execute()
86 
87     def stream_users(self, status):  # pylint: disable=no-self-use
88         """ given a status, what users should see it """
89         # direct messages don't appeard in feeds, direct comments/reviews/etc do
90         if status.privacy == "direct" and status.status_type == "Note":
91             return []
92 
93         # everybody who could plausibly see this status
94         audience = models.User.objects.filter(
95             is_active=True,
96             local=True,  # we only create feeds for users of this instance
97         ).exclude(
98             Q(id__in=status.user.blocks.all()) | Q(blocks=status.user)  # not blocked
99         )
100 
101         # only visible to the poster and mentioned users
102         if status.privacy == "direct":
103             audience = audience.filter(
104                 Q(id=status.user.id)  # if the user is the post's author
105                 | Q(id__in=status.mention_users.all())  # if the user is mentioned
106             )
107         # only visible to the poster's followers and tagged users
108         elif status.privacy == "followers":
109             audience = audience.filter(
110                 Q(id=status.user.id)  # if the user is the post's author
111                 | Q(following=status.user)  # if the user is following the author
112             )
113         return audience.distinct()
114 
115     def stream_statuses(self, user):  # pylint: disable=no-self-use
116         """ given a user, what statuses should they see on this stream """
117         return privacy_filter(
118             user,
119             models.Status.objects.select_subclasses(),
120             privacy_levels=["public", "unlisted", "followers"],
121         )
122 
123 
124 class HomeStream(ActivityStream):
125     """ users you follow """
126 
127     key = "home"
128 
129     def stream_users(self, status):
130         audience = super().stream_users(status)
131         if not audience:
132             return []
133         return audience.filter(
134             Q(id=status.user.id)  # if the user is the post's author
135             | Q(following=status.user)  # if the user is following the author
136         ).distinct()
137 
138     def stream_statuses(self, user):
139         return privacy_filter(
140             user,
141             models.Status.objects.select_subclasses(),
142             privacy_levels=["public", "unlisted", "followers"],
143             following_only=True,
144         )
145 
146 
147 class LocalStream(ActivityStream):
148     """ users you follow """
149 
150     key = "local"
151 
152     def stream_users(self, status):
153         # this stream wants no part in non-public statuses
154         if status.privacy != "public" or not status.user.local:
155             return []
156         return super().stream_users(status)
157 
158     def stream_statuses(self, user):
159         # all public statuses by a local user
160         return privacy_filter(
161             user,
162             models.Status.objects.select_subclasses().filter(user__local=True),
163             privacy_levels=["public"],
164         )
165 
166 
167 class FederatedStream(ActivityStream):
168     """ users you follow """
169 
170     key = "federated"
171 
172     def stream_users(self, status):
173         # this stream wants no part in non-public statuses
174         if status.privacy != "public":
175             return []
176         return super().stream_users(status)
177 
178     def stream_statuses(self, user):
179         return privacy_filter(
180             user,
181             models.Status.objects.select_subclasses(),
182             privacy_levels=["public"],
183         )
184 
185 
186 streams = {
187     "home": HomeStream(),
188     "local": LocalStream(),
189     "federated": FederatedStream(),
190 }
191 
192 
193 @receiver(signals.post_save)
194 # pylint: disable=unused-argument
195 def add_status_on_create(sender, instance, created, *args, **kwargs):
196     """ add newly created statuses to activity feeds """
197     # we're only interested in new statuses
198     if not issubclass(sender, models.Status):
199         return
200 
201     if instance.deleted:
202         for stream in streams.values():
203             stream.remove_status(instance)
204         return
205 
206     if not created:
207         return
208 
209     # iterates through Home, Local, Federated
210     for stream in streams.values():
211         stream.add_status(instance)
212 
213 
214 @receiver(signals.post_delete, sender=models.Boost)
215 # pylint: disable=unused-argument
216 def remove_boost_on_delete(sender, instance, *args, **kwargs):
217     """ boosts are deleted """
218     # we're only interested in new statuses
219     for stream in streams.values():
220         stream.remove_status(instance)
221 
222 
223 @receiver(signals.post_save, sender=models.UserFollows)
224 # pylint: disable=unused-argument
225 def add_statuses_on_follow(sender, instance, created, *args, **kwargs):
226     """ add a newly followed user's statuses to feeds """
227     if not created or not instance.user_subject.local:
228         return
229     HomeStream().add_user_statuses(instance.user_subject, instance.user_object)
230 
231 
232 @receiver(signals.post_delete, sender=models.UserFollows)
233 # pylint: disable=unused-argument
234 def remove_statuses_on_unfollow(sender, instance, *args, **kwargs):
235     """ remove statuses from a feed on unfollow """
236     if not instance.user_subject.local:
237         return
238     HomeStream().remove_user_statuses(instance.user_subject, instance.user_object)
239 
240 
241 @receiver(signals.post_save, sender=models.UserBlocks)
242 # pylint: disable=unused-argument
243 def remove_statuses_on_block(sender, instance, *args, **kwargs):
244     """ remove statuses from all feeds on block """
245     # blocks apply ot all feeds
246     if instance.user_subject.local:
247         for stream in streams.values():
248             stream.remove_user_statuses(instance.user_subject, instance.user_object)
249 
250     # and in both directions
251     if instance.user_object.local:
252         for stream in streams.values():
253             stream.remove_user_statuses(instance.user_object, instance.user_subject)
254 
255 
256 @receiver(signals.post_delete, sender=models.UserBlocks)
257 # pylint: disable=unused-argument
258 def add_statuses_on_unblock(sender, instance, *args, **kwargs):
259     """ remove statuses from all feeds on block """
260     public_streams = [LocalStream(), FederatedStream()]
261     # add statuses back to streams with statuses from anyone
262     if instance.user_subject.local:
263         for stream in public_streams:
264             stream.add_user_statuses(instance.user_subject, instance.user_object)
265 
266     # add statuses back to streams with statuses from anyone
267     if instance.user_object.local:
268         for stream in public_streams:
269             stream.add_user_statuses(instance.user_object, instance.user_subject)
270 
271 
272 @receiver(signals.post_save, sender=models.User)
273 # pylint: disable=unused-argument
274 def populate_feed_on_account_create(sender, instance, created, *args, **kwargs):
275     """ build a user's feeds when they join """
276     if not created or not instance.local:
277         return
278 
279     for stream in streams.values():
280         stream.populate_stream(instance)
281
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/bookwyrm/activitystreams.py b/bookwyrm/activitystreams.py
--- a/bookwyrm/activitystreams.py
+++ b/bookwyrm/activitystreams.py
@@ -23,15 +23,21 @@
         """ the redis key for this user's unread count for this stream """
         return "{}-unread".format(self.stream_id(user))
 
+    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 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.lpush(self.stream_id(user), status.id)
-            pipeline.ltrim(self.stream_id(user), 0, settings.MAX_STREAM_LENGTH)
-
+            pipeline.zadd(self.stream_id(user), value)
+            pipeline.zremrangebyrank(
+                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1
+            )
             # add to the unread status count
             pipeline.incr(self.unread_id(user))
         # and go!
@@ -47,8 +53,13 @@
     def add_user_statuses(self, viewer, user):
         """ add a user's statuses to another user's feed """
         pipeline = r.pipeline()
-        for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:
-            pipeline.lpush(self.stream_id(viewer), status.id)
+        statuses = user.status_set.all()[: settings.MAX_STREAM_LENGTH]
+        for status in statuses:
+            pipeline.zadd(self.stream_id(viewer), self.get_value(status))
+        if statuses:
+            pipeline.zremrangebyrank(
+                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1
+            )
         pipeline.execute()
 
     def remove_user_statuses(self, viewer, user):
@@ -63,7 +74,7 @@
         # clear unreads for this feed
         r.set(self.unread_id(user), 0)
 
-        statuses = r.lrange(self.stream_id(user), 0, -1)
+        statuses = r.zrevrange(self.stream_id(user), 0, -1)
         return (
             models.Status.objects.select_subclasses()
             .filter(id__in=statuses)
@@ -81,7 +92,11 @@
 
         stream_id = self.stream_id(user)
         for status in statuses.all()[: settings.MAX_STREAM_LENGTH]:
-            pipeline.lpush(stream_id, status.id)
+            pipeline.zadd(stream_id, self.get_value(status))
+
+        # only trim the stream if statuses were added
+        if statuses.exists():
+            pipeline.zremrangebyrank(stream_id, settings.MAX_STREAM_LENGTH, -1)
         pipeline.execute()
 
     def stream_users(self, status):  # pylint: disable=no-self-use
@@ -271,7 +286,7 @@
 
 @receiver(signals.post_save, sender=models.User)
 # pylint: disable=unused-argument
-def populate_feed_on_account_create(sender, instance, created, *args, **kwargs):
+def populate_streams_on_account_create(sender, instance, created, *args, **kwargs):
     """ build a user's feeds when they join """
     if not created or not instance.local:
         return

{"golden_diff": "diff --git a/bookwyrm/activitystreams.py b/bookwyrm/activitystreams.py\n--- a/bookwyrm/activitystreams.py\n+++ b/bookwyrm/activitystreams.py\n@@ -23,15 +23,21 @@\n         \"\"\" the redis key for this user's unread count for this stream \"\"\"\n         return \"{}-unread\".format(self.stream_id(user))\n \n+    def get_value(self, status):  # pylint: disable=no-self-use\n+        \"\"\" the status id and the rank (ie, published date) \"\"\"\n+        return {status.id: status.published_date.timestamp()}\n+\n     def add_status(self, status):\n         \"\"\" add a status to users' feeds \"\"\"\n+        value = self.get_value(status)\n         # we want to do this as a bulk operation, hence \"pipeline\"\n         pipeline = r.pipeline()\n         for user in self.stream_users(status):\n             # add the status to the feed\n-            pipeline.lpush(self.stream_id(user), status.id)\n-            pipeline.ltrim(self.stream_id(user), 0, settings.MAX_STREAM_LENGTH)\n-\n+            pipeline.zadd(self.stream_id(user), value)\n+            pipeline.zremrangebyrank(\n+                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1\n+            )\n             # add to the unread status count\n             pipeline.incr(self.unread_id(user))\n         # and go!\n@@ -47,8 +53,13 @@\n     def add_user_statuses(self, viewer, user):\n         \"\"\" add a user's statuses to another user's feed \"\"\"\n         pipeline = r.pipeline()\n-        for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:\n-            pipeline.lpush(self.stream_id(viewer), status.id)\n+        statuses = user.status_set.all()[: settings.MAX_STREAM_LENGTH]\n+        for status in statuses:\n+            pipeline.zadd(self.stream_id(viewer), self.get_value(status))\n+        if statuses:\n+            pipeline.zremrangebyrank(\n+                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1\n+            )\n         pipeline.execute()\n \n     def remove_user_statuses(self, viewer, user):\n@@ -63,7 +74,7 @@\n         # clear unreads for this feed\n         r.set(self.unread_id(user), 0)\n \n-        statuses = r.lrange(self.stream_id(user), 0, -1)\n+        statuses = r.zrevrange(self.stream_id(user), 0, -1)\n         return (\n             models.Status.objects.select_subclasses()\n             .filter(id__in=statuses)\n@@ -81,7 +92,11 @@\n \n         stream_id = self.stream_id(user)\n         for status in statuses.all()[: settings.MAX_STREAM_LENGTH]:\n-            pipeline.lpush(stream_id, status.id)\n+            pipeline.zadd(stream_id, self.get_value(status))\n+\n+        # only trim the stream if statuses were added\n+        if statuses.exists():\n+            pipeline.zremrangebyrank(stream_id, settings.MAX_STREAM_LENGTH, -1)\n         pipeline.execute()\n \n     def stream_users(self, status):  # pylint: disable=no-self-use\n@@ -271,7 +286,7 @@\n \n @receiver(signals.post_save, sender=models.User)\n # pylint: disable=unused-argument\n-def populate_feed_on_account_create(sender, instance, created, *args, **kwargs):\n+def populate_streams_on_account_create(sender, instance, created, *args, **kwargs):\n     \"\"\" build a user's feeds when they join \"\"\"\n     if not created or not instance.local:\n         return\n", "issue": "Goodreads import floods activity streams\nWhen a user does a goodreads import, the activitstreams manager isn't checking if the publication date is older than the oldest status, and adds all the statuses to the feeds, pushing out newer statuses.\n", "before_files": [{"content": "\"\"\" access the activity streams stored in redis \"\"\"\nfrom abc import ABC\nfrom django.dispatch import receiver\nfrom django.db.models import signals, Q\nimport redis\n\nfrom bookwyrm import models, settings\nfrom bookwyrm.views.helpers import privacy_filter\n\nr = redis.Redis(\n    host=settings.REDIS_ACTIVITY_HOST, port=settings.REDIS_ACTIVITY_PORT, db=0\n)\n\n\nclass ActivityStream(ABC):\n    \"\"\" a category of activity stream (like home, local, federated) \"\"\"\n\n    def stream_id(self, user):\n        \"\"\" the redis key for this user's instance of this stream \"\"\"\n        return \"{}-{}\".format(user.id, self.key)\n\n    def unread_id(self, user):\n        \"\"\" the redis key for this user's unread count for this stream \"\"\"\n        return \"{}-unread\".format(self.stream_id(user))\n\n    def add_status(self, status):\n        \"\"\" add a status to users' feeds \"\"\"\n        # we want to do this as a bulk operation, hence \"pipeline\"\n        pipeline = r.pipeline()\n        for user in self.stream_users(status):\n            # add the status to the feed\n            pipeline.lpush(self.stream_id(user), status.id)\n            pipeline.ltrim(self.stream_id(user), 0, settings.MAX_STREAM_LENGTH)\n\n            # add to the unread status count\n            pipeline.incr(self.unread_id(user))\n        # and go!\n        pipeline.execute()\n\n    def remove_status(self, status):\n        \"\"\" remove a status from all feeds \"\"\"\n        pipeline = r.pipeline()\n        for user in self.stream_users(status):\n            pipeline.lrem(self.stream_id(user), -1, status.id)\n        pipeline.execute()\n\n    def add_user_statuses(self, viewer, user):\n        \"\"\" add a user's statuses to another user's feed \"\"\"\n        pipeline = r.pipeline()\n        for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:\n            pipeline.lpush(self.stream_id(viewer), status.id)\n        pipeline.execute()\n\n    def remove_user_statuses(self, viewer, user):\n        \"\"\" remove a user's status from another user's feed \"\"\"\n        pipeline = r.pipeline()\n        for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:\n            pipeline.lrem(self.stream_id(viewer), -1, status.id)\n        pipeline.execute()\n\n    def get_activity_stream(self, user):\n        \"\"\" load the ids for statuses to be displayed \"\"\"\n        # clear unreads for this feed\n        r.set(self.unread_id(user), 0)\n\n        statuses = r.lrange(self.stream_id(user), 0, -1)\n        return (\n            models.Status.objects.select_subclasses()\n            .filter(id__in=statuses)\n            .order_by(\"-published_date\")\n        )\n\n    def get_unread_count(self, user):\n        \"\"\" get the unread status count for this user's feed \"\"\"\n        return int(r.get(self.unread_id(user)))\n\n    def populate_stream(self, user):\n        \"\"\" go from zero to a timeline \"\"\"\n        pipeline = r.pipeline()\n        statuses = self.stream_statuses(user)\n\n        stream_id = self.stream_id(user)\n        for status in statuses.all()[: settings.MAX_STREAM_LENGTH]:\n            pipeline.lpush(stream_id, status.id)\n        pipeline.execute()\n\n    def stream_users(self, status):  # pylint: disable=no-self-use\n        \"\"\" given a status, what users should see it \"\"\"\n        # direct messages don't appeard in feeds, direct comments/reviews/etc do\n        if status.privacy == \"direct\" and status.status_type == \"Note\":\n            return []\n\n        # everybody who could plausibly see this status\n        audience = models.User.objects.filter(\n            is_active=True,\n            local=True,  # we only create feeds for users of this instance\n        ).exclude(\n            Q(id__in=status.user.blocks.all()) | Q(blocks=status.user)  # not blocked\n        )\n\n        # only visible to the poster and mentioned users\n        if status.privacy == \"direct\":\n            audience = audience.filter(\n                Q(id=status.user.id)  # if the user is the post's author\n                | Q(id__in=status.mention_users.all())  # if the user is mentioned\n            )\n        # only visible to the poster's followers and tagged users\n        elif status.privacy == \"followers\":\n            audience = audience.filter(\n                Q(id=status.user.id)  # if the user is the post's author\n                | Q(following=status.user)  # if the user is following the author\n            )\n        return audience.distinct()\n\n    def stream_statuses(self, user):  # pylint: disable=no-self-use\n        \"\"\" given a user, what statuses should they see on this stream \"\"\"\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\", \"unlisted\", \"followers\"],\n        )\n\n\nclass HomeStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"home\"\n\n    def stream_users(self, status):\n        audience = super().stream_users(status)\n        if not audience:\n            return []\n        return audience.filter(\n            Q(id=status.user.id)  # if the user is the post's author\n            | Q(following=status.user)  # if the user is following the author\n        ).distinct()\n\n    def stream_statuses(self, user):\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\", \"unlisted\", \"followers\"],\n            following_only=True,\n        )\n\n\nclass LocalStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"local\"\n\n    def stream_users(self, status):\n        # this stream wants no part in non-public statuses\n        if status.privacy != \"public\" or not status.user.local:\n            return []\n        return super().stream_users(status)\n\n    def stream_statuses(self, user):\n        # all public statuses by a local user\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses().filter(user__local=True),\n            privacy_levels=[\"public\"],\n        )\n\n\nclass FederatedStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"federated\"\n\n    def stream_users(self, status):\n        # this stream wants no part in non-public statuses\n        if status.privacy != \"public\":\n            return []\n        return super().stream_users(status)\n\n    def stream_statuses(self, user):\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\"],\n        )\n\n\nstreams = {\n    \"home\": HomeStream(),\n    \"local\": LocalStream(),\n    \"federated\": FederatedStream(),\n}\n\n\n@receiver(signals.post_save)\n# pylint: disable=unused-argument\ndef add_status_on_create(sender, instance, created, *args, **kwargs):\n    \"\"\" add newly created statuses to activity feeds \"\"\"\n    # we're only interested in new statuses\n    if not issubclass(sender, models.Status):\n        return\n\n    if instance.deleted:\n        for stream in streams.values():\n            stream.remove_status(instance)\n        return\n\n    if not created:\n        return\n\n    # iterates through Home, Local, Federated\n    for stream in streams.values():\n        stream.add_status(instance)\n\n\n@receiver(signals.post_delete, sender=models.Boost)\n# pylint: disable=unused-argument\ndef remove_boost_on_delete(sender, instance, *args, **kwargs):\n    \"\"\" boosts are deleted \"\"\"\n    # we're only interested in new statuses\n    for stream in streams.values():\n        stream.remove_status(instance)\n\n\n@receiver(signals.post_save, sender=models.UserFollows)\n# pylint: disable=unused-argument\ndef add_statuses_on_follow(sender, instance, created, *args, **kwargs):\n    \"\"\" add a newly followed user's statuses to feeds \"\"\"\n    if not created or not instance.user_subject.local:\n        return\n    HomeStream().add_user_statuses(instance.user_subject, instance.user_object)\n\n\n@receiver(signals.post_delete, sender=models.UserFollows)\n# pylint: disable=unused-argument\ndef remove_statuses_on_unfollow(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from a feed on unfollow \"\"\"\n    if not instance.user_subject.local:\n        return\n    HomeStream().remove_user_statuses(instance.user_subject, instance.user_object)\n\n\n@receiver(signals.post_save, sender=models.UserBlocks)\n# pylint: disable=unused-argument\ndef remove_statuses_on_block(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from all feeds on block \"\"\"\n    # blocks apply ot all feeds\n    if instance.user_subject.local:\n        for stream in streams.values():\n            stream.remove_user_statuses(instance.user_subject, instance.user_object)\n\n    # and in both directions\n    if instance.user_object.local:\n        for stream in streams.values():\n            stream.remove_user_statuses(instance.user_object, instance.user_subject)\n\n\n@receiver(signals.post_delete, sender=models.UserBlocks)\n# pylint: disable=unused-argument\ndef add_statuses_on_unblock(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from all feeds on block \"\"\"\n    public_streams = [LocalStream(), FederatedStream()]\n    # add statuses back to streams with statuses from anyone\n    if instance.user_subject.local:\n        for stream in public_streams:\n            stream.add_user_statuses(instance.user_subject, instance.user_object)\n\n    # add statuses back to streams with statuses from anyone\n    if instance.user_object.local:\n        for stream in public_streams:\n            stream.add_user_statuses(instance.user_object, instance.user_subject)\n\n\n@receiver(signals.post_save, sender=models.User)\n# pylint: disable=unused-argument\ndef populate_feed_on_account_create(sender, instance, created, *args, **kwargs):\n    \"\"\" build a user's feeds when they join \"\"\"\n    if not created or not instance.local:\n        return\n\n    for stream in streams.values():\n        stream.populate_stream(instance)\n", "path": "bookwyrm/activitystreams.py"}], "after_files": [{"content": "\"\"\" access the activity streams stored in redis \"\"\"\nfrom abc import ABC\nfrom django.dispatch import receiver\nfrom django.db.models import signals, Q\nimport redis\n\nfrom bookwyrm import models, settings\nfrom bookwyrm.views.helpers import privacy_filter\n\nr = redis.Redis(\n    host=settings.REDIS_ACTIVITY_HOST, port=settings.REDIS_ACTIVITY_PORT, db=0\n)\n\n\nclass ActivityStream(ABC):\n    \"\"\" a category of activity stream (like home, local, federated) \"\"\"\n\n    def stream_id(self, user):\n        \"\"\" the redis key for this user's instance of this stream \"\"\"\n        return \"{}-{}\".format(user.id, self.key)\n\n    def unread_id(self, user):\n        \"\"\" the redis key for this user's unread count for this stream \"\"\"\n        return \"{}-unread\".format(self.stream_id(user))\n\n    def get_value(self, status):  # pylint: disable=no-self-use\n        \"\"\" the status id and the rank (ie, published date) \"\"\"\n        return {status.id: status.published_date.timestamp()}\n\n    def add_status(self, status):\n        \"\"\" add a status to users' feeds \"\"\"\n        value = self.get_value(status)\n        # we want to do this as a bulk operation, hence \"pipeline\"\n        pipeline = r.pipeline()\n        for user in self.stream_users(status):\n            # add the status to the feed\n            pipeline.zadd(self.stream_id(user), value)\n            pipeline.zremrangebyrank(\n                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1\n            )\n            # add to the unread status count\n            pipeline.incr(self.unread_id(user))\n        # and go!\n        pipeline.execute()\n\n    def remove_status(self, status):\n        \"\"\" remove a status from all feeds \"\"\"\n        pipeline = r.pipeline()\n        for user in self.stream_users(status):\n            pipeline.lrem(self.stream_id(user), -1, status.id)\n        pipeline.execute()\n\n    def add_user_statuses(self, viewer, user):\n        \"\"\" add a user's statuses to another user's feed \"\"\"\n        pipeline = r.pipeline()\n        statuses = user.status_set.all()[: settings.MAX_STREAM_LENGTH]\n        for status in statuses:\n            pipeline.zadd(self.stream_id(viewer), self.get_value(status))\n        if statuses:\n            pipeline.zremrangebyrank(\n                self.stream_id(user), settings.MAX_STREAM_LENGTH, -1\n            )\n        pipeline.execute()\n\n    def remove_user_statuses(self, viewer, user):\n        \"\"\" remove a user's status from another user's feed \"\"\"\n        pipeline = r.pipeline()\n        for status in user.status_set.all()[: settings.MAX_STREAM_LENGTH]:\n            pipeline.lrem(self.stream_id(viewer), -1, status.id)\n        pipeline.execute()\n\n    def get_activity_stream(self, user):\n        \"\"\" load the ids for statuses to be displayed \"\"\"\n        # clear unreads for this feed\n        r.set(self.unread_id(user), 0)\n\n        statuses = r.zrevrange(self.stream_id(user), 0, -1)\n        return (\n            models.Status.objects.select_subclasses()\n            .filter(id__in=statuses)\n            .order_by(\"-published_date\")\n        )\n\n    def get_unread_count(self, user):\n        \"\"\" get the unread status count for this user's feed \"\"\"\n        return int(r.get(self.unread_id(user)))\n\n    def populate_stream(self, user):\n        \"\"\" go from zero to a timeline \"\"\"\n        pipeline = r.pipeline()\n        statuses = self.stream_statuses(user)\n\n        stream_id = self.stream_id(user)\n        for status in statuses.all()[: settings.MAX_STREAM_LENGTH]:\n            pipeline.zadd(stream_id, self.get_value(status))\n\n        # only trim the stream if statuses were added\n        if statuses.exists():\n            pipeline.zremrangebyrank(stream_id, settings.MAX_STREAM_LENGTH, -1)\n        pipeline.execute()\n\n    def stream_users(self, status):  # pylint: disable=no-self-use\n        \"\"\" given a status, what users should see it \"\"\"\n        # direct messages don't appeard in feeds, direct comments/reviews/etc do\n        if status.privacy == \"direct\" and status.status_type == \"Note\":\n            return []\n\n        # everybody who could plausibly see this status\n        audience = models.User.objects.filter(\n            is_active=True,\n            local=True,  # we only create feeds for users of this instance\n        ).exclude(\n            Q(id__in=status.user.blocks.all()) | Q(blocks=status.user)  # not blocked\n        )\n\n        # only visible to the poster and mentioned users\n        if status.privacy == \"direct\":\n            audience = audience.filter(\n                Q(id=status.user.id)  # if the user is the post's author\n                | Q(id__in=status.mention_users.all())  # if the user is mentioned\n            )\n        # only visible to the poster's followers and tagged users\n        elif status.privacy == \"followers\":\n            audience = audience.filter(\n                Q(id=status.user.id)  # if the user is the post's author\n                | Q(following=status.user)  # if the user is following the author\n            )\n        return audience.distinct()\n\n    def stream_statuses(self, user):  # pylint: disable=no-self-use\n        \"\"\" given a user, what statuses should they see on this stream \"\"\"\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\", \"unlisted\", \"followers\"],\n        )\n\n\nclass HomeStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"home\"\n\n    def stream_users(self, status):\n        audience = super().stream_users(status)\n        if not audience:\n            return []\n        return audience.filter(\n            Q(id=status.user.id)  # if the user is the post's author\n            | Q(following=status.user)  # if the user is following the author\n        ).distinct()\n\n    def stream_statuses(self, user):\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\", \"unlisted\", \"followers\"],\n            following_only=True,\n        )\n\n\nclass LocalStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"local\"\n\n    def stream_users(self, status):\n        # this stream wants no part in non-public statuses\n        if status.privacy != \"public\" or not status.user.local:\n            return []\n        return super().stream_users(status)\n\n    def stream_statuses(self, user):\n        # all public statuses by a local user\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses().filter(user__local=True),\n            privacy_levels=[\"public\"],\n        )\n\n\nclass FederatedStream(ActivityStream):\n    \"\"\" users you follow \"\"\"\n\n    key = \"federated\"\n\n    def stream_users(self, status):\n        # this stream wants no part in non-public statuses\n        if status.privacy != \"public\":\n            return []\n        return super().stream_users(status)\n\n    def stream_statuses(self, user):\n        return privacy_filter(\n            user,\n            models.Status.objects.select_subclasses(),\n            privacy_levels=[\"public\"],\n        )\n\n\nstreams = {\n    \"home\": HomeStream(),\n    \"local\": LocalStream(),\n    \"federated\": FederatedStream(),\n}\n\n\n@receiver(signals.post_save)\n# pylint: disable=unused-argument\ndef add_status_on_create(sender, instance, created, *args, **kwargs):\n    \"\"\" add newly created statuses to activity feeds \"\"\"\n    # we're only interested in new statuses\n    if not issubclass(sender, models.Status):\n        return\n\n    if instance.deleted:\n        for stream in streams.values():\n            stream.remove_status(instance)\n        return\n\n    if not created:\n        return\n\n    # iterates through Home, Local, Federated\n    for stream in streams.values():\n        stream.add_status(instance)\n\n\n@receiver(signals.post_delete, sender=models.Boost)\n# pylint: disable=unused-argument\ndef remove_boost_on_delete(sender, instance, *args, **kwargs):\n    \"\"\" boosts are deleted \"\"\"\n    # we're only interested in new statuses\n    for stream in streams.values():\n        stream.remove_status(instance)\n\n\n@receiver(signals.post_save, sender=models.UserFollows)\n# pylint: disable=unused-argument\ndef add_statuses_on_follow(sender, instance, created, *args, **kwargs):\n    \"\"\" add a newly followed user's statuses to feeds \"\"\"\n    if not created or not instance.user_subject.local:\n        return\n    HomeStream().add_user_statuses(instance.user_subject, instance.user_object)\n\n\n@receiver(signals.post_delete, sender=models.UserFollows)\n# pylint: disable=unused-argument\ndef remove_statuses_on_unfollow(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from a feed on unfollow \"\"\"\n    if not instance.user_subject.local:\n        return\n    HomeStream().remove_user_statuses(instance.user_subject, instance.user_object)\n\n\n@receiver(signals.post_save, sender=models.UserBlocks)\n# pylint: disable=unused-argument\ndef remove_statuses_on_block(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from all feeds on block \"\"\"\n    # blocks apply ot all feeds\n    if instance.user_subject.local:\n        for stream in streams.values():\n            stream.remove_user_statuses(instance.user_subject, instance.user_object)\n\n    # and in both directions\n    if instance.user_object.local:\n        for stream in streams.values():\n            stream.remove_user_statuses(instance.user_object, instance.user_subject)\n\n\n@receiver(signals.post_delete, sender=models.UserBlocks)\n# pylint: disable=unused-argument\ndef add_statuses_on_unblock(sender, instance, *args, **kwargs):\n    \"\"\" remove statuses from all feeds on block \"\"\"\n    public_streams = [LocalStream(), FederatedStream()]\n    # add statuses back to streams with statuses from anyone\n    if instance.user_subject.local:\n        for stream in public_streams:\n            stream.add_user_statuses(instance.user_subject, instance.user_object)\n\n    # add statuses back to streams with statuses from anyone\n    if instance.user_object.local:\n        for stream in public_streams:\n            stream.add_user_statuses(instance.user_object, instance.user_subject)\n\n\n@receiver(signals.post_save, sender=models.User)\n# pylint: disable=unused-argument\ndef populate_streams_on_account_create(sender, instance, created, *args, **kwargs):\n    \"\"\" build a user's feeds when they join \"\"\"\n    if not created or not instance.local:\n        return\n\n    for stream in streams.values():\n        stream.populate_stream(instance)\n", "path": "bookwyrm/activitystreams.py"}]}