Merge branch 'main' into production

.env.example

@@ -8,7 +8,7 @@ USE_HTTPS=true
 DOMAIN=your.domain.here
 EMAIL=your@email.here
 
-# Instance defualt language (see options at bookwyrm/settings.py "LANGUAGES"
+# Instance default language (see options at bookwyrm/settings.py "LANGUAGES"
 LANGUAGE_CODE="en-us"
 # Used for deciding which editions to prefer
 DEFAULT_LANGUAGE="English"

bookwyrm/activitypub/base_activity.py

@@ -127,7 +127,7 @@ def to_model(
         if (
             allow_create
             and hasattr(model, "ignore_activity")
-            and model.ignore_activity(self)
+            and model.ignore_activity(self, allow_external_connections)
         ):
             return None
 
@@ -241,7 +241,7 @@ def serialize(self, **kwargs):
         return data
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 @transaction.atomic
 def set_related_field(
     model_name, origin_model_name, related_field_name, related_remote_id, data
@@ -384,7 +384,8 @@ def get_activitypub_data(url):
         resp = requests.get(
             url,
             headers={
-                "Accept": "application/json; charset=utf-8",
+                # pylint: disable=line-too-long
+                "Accept": 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
                 "Date": now,
                 "Signature": make_signature("get", sender, url, now),
             },

bookwyrm/activitystreams.py

@@ -38,11 +38,14 @@ def get_rank(self, obj):  # pylint: disable=no-self-use
 
     def add_status(self, status, increment_unread=False):
         """add a status to users' feeds"""
+        audience = self.get_audience(status)
         # the pipeline contains all the add-to-stream activities
-        pipeline = self.add_object_to_related_stores(status, execute=False)
+        pipeline = self.add_object_to_stores(
+            status, self.get_stores_for_users(audience), execute=False
+        )
 
         if increment_unread:
-            for user_id in self.get_audience(status):
+            for user_id in audience:
                 # add to the unread status count
                 pipeline.incr(self.unread_id(user_id))
                 # add to the unread status count for status type
@@ -102,9 +105,16 @@ def populate_streams(self, user):
         """go from zero to a timeline"""
         self.populate_store(self.stream_id(user.id))
 
+    @tracer.start_as_current_span("ActivityStream._get_audience")
     def _get_audience(self, status):  # pylint: disable=no-self-use
-        """given a status, what users should see it"""
-        # direct messages don't appeard in feeds, direct comments/reviews/etc do
+        """given a status, what users should see it, excluding the author"""
+        trace.get_current_span().set_attribute("status_type", status.status_type)
+        trace.get_current_span().set_attribute("status_privacy", status.privacy)
+        trace.get_current_span().set_attribute(
+            "status_reply_parent_privacy",
+            status.reply_parent.privacy if status.reply_parent else None,
+        )
+        # direct messages don't appear in feeds, direct comments/reviews/etc do
         if status.privacy == "direct" and status.status_type == "Note":
             return []
 
@@ -119,15 +129,13 @@ def _get_audience(self, status):  # pylint: disable=no-self-use
         # only visible to the poster and mentioned users
         if status.privacy == "direct":
             audience = audience.filter(
-                Q(id=status.user.id)  # if the user is the post's author
-                | Q(id__in=status.mention_users.all())  # if the user is mentioned
+                Q(id__in=status.mention_users.all())  # if the user is mentioned
             )
 
         # don't show replies to statuses the user can't see
         elif status.reply_parent and status.reply_parent.privacy == "followers":
             audience = audience.filter(
-                Q(id=status.user.id)  # if the user is the post's author
-                | Q(id=status.reply_parent.user.id)  # if the user is the OG author
+                Q(id=status.reply_parent.user.id)  # if the user is the OG author
                 | (
                     Q(following=status.user) & Q(following=status.reply_parent.user)
                 )  # if the user is following both authors
@@ -136,19 +144,23 @@ def _get_audience(self, status):  # pylint: disable=no-self-use
         # only visible to the poster's followers and tagged users
         elif 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
+                Q(following=status.user)  # if the user is following the author
             )
         return audience.distinct()
 
     @tracer.start_as_current_span("ActivityStream.get_audience")
     def get_audience(self, status):
         """given a status, what users should see it"""
         trace.get_current_span().set_attribute("stream_id", self.key)
-        return [user.id for user in self._get_audience(status)]
+        audience = self._get_audience(status)
+        status_author = models.User.objects.filter(
+            is_active=True, local=True, id=status.user.id
+        )
+        return list({user.id for user in list(audience) + list(status_author)})
 
-    def get_stores_for_object(self, obj):
-        return [self.stream_id(user_id) for user_id in self.get_audience(obj)]
+    def get_stores_for_users(self, user_ids):
+        """convert a list of user ids into redis store ids"""
+        return [self.stream_id(user_id) for user_id in user_ids]
 
     def get_statuses_for_user(self, user):  # pylint: disable=no-self-use
         """given a user, what statuses should they see on this stream"""
@@ -173,11 +185,13 @@ def get_audience(self, status):
         audience = super()._get_audience(status)
         if not audience:
             return []
-        # if the user is the post's author
-        ids_self = [user.id for user in audience.filter(Q(id=status.user.id))]
         # if the user is following the author
-        ids_following = [user.id for user in audience.filter(Q(following=status.user))]
-        return ids_self + ids_following
+        audience = audience.filter(following=status.user)
+        # if the user is the post's author
+        status_author = models.User.objects.filter(
+            is_active=True, local=True, id=status.user.id
+        )
+        return list({user.id for user in list(audience) + list(status_author)})
 
     def get_statuses_for_user(self, user):
         return models.Status.privacy_filter(
@@ -197,11 +211,11 @@ class LocalStream(ActivityStream):
 
     key = "local"
 
-    def _get_audience(self, status):
+    def get_audience(self, status):
         # this stream wants no part in non-public statuses
         if status.privacy != "public" or not status.user.local:
             return []
-        return super()._get_audience(status)
+        return super().get_audience(status)
 
     def get_statuses_for_user(self, user):
         # all public statuses by a local user
@@ -218,13 +232,6 @@ class BooksStream(ActivityStream):
 
     def _get_audience(self, status):
         """anyone with the mentioned book on their shelves"""
-        # only show public statuses on the books feed,
-        # and only statuses that mention books
-        if status.privacy != "public" or not (
-            status.mention_books.exists() or hasattr(status, "book")
-        ):
-            return []
-
         work = (
             status.book.parent_work
             if hasattr(status, "book")
@@ -236,6 +243,16 @@ def _get_audience(self, status):
             return []
         return audience.filter(shelfbook__book__parent_work=work).distinct()
 
+    def get_audience(self, status):
+        # only show public statuses on the books feed,
+        # and only statuses that mention books
+        if status.privacy != "public" or not (
+            status.mention_books.exists() or hasattr(status, "book")
+        ):
+            return []
+
+        return super().get_audience(status)
+
     def get_statuses_for_user(self, user):
         """any public status that mentions the user's books"""
         books = user.shelfbook_set.values_list(
@@ -480,31 +497,31 @@ def remove_statuses_on_unshelve(sender, instance, *args, **kwargs):
 # ---- TASKS
 
 
-@app.task(queue=LOW, ignore_result=True)
+@app.task(queue=LOW)
 def add_book_statuses_task(user_id, book_id):
     """add statuses related to a book on shelve"""
     user = models.User.objects.get(id=user_id)
     book = models.Edition.objects.get(id=book_id)
     BooksStream().add_book_statuses(user, book)
 
 
-@app.task(queue=LOW, ignore_result=True)
+@app.task(queue=LOW)
 def remove_book_statuses_task(user_id, book_id):
     """remove statuses about a book from a user's books feed"""
     user = models.User.objects.get(id=user_id)
     book = models.Edition.objects.get(id=book_id)
     BooksStream().remove_book_statuses(user, book)
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 def populate_stream_task(stream, user_id):
     """background task for populating an empty activitystream"""
     user = models.User.objects.get(id=user_id)
     stream = streams[stream]
     stream.populate_streams(user)
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 def remove_status_task(status_ids):
     """remove a status from any stream it might be in"""
     # this can take an id or a list of ids
@@ -514,10 +531,12 @@ def remove_status_task(status_ids):
 
     for stream in streams.values():
         for status in statuses:
-            stream.remove_object_from_related_stores(status)
+            stream.remove_object_from_stores(
+                status, stream.get_stores_for_users(stream.get_audience(status))
+            )
 
 
-@app.task(queue=HIGH, ignore_result=True)
+@app.task(queue=HIGH)
 def add_status_task(status_id, increment_unread=False):
     """add a status to any stream it should be in"""
     status = models.Status.objects.select_subclasses().get(id=status_id)
@@ -529,7 +548,7 @@ def add_status_task(status_id, increment_unread=False):
         stream.add_status(status, increment_unread=increment_unread)
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 def remove_user_statuses_task(viewer_id, user_id, stream_list=None):
     """remove all statuses by a user from a viewer's stream"""
     stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
@@ -539,7 +558,7 @@ def remove_user_statuses_task(viewer_id, user_id, stream_list=None):
         stream.remove_user_statuses(viewer, user)
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 def add_user_statuses_task(viewer_id, user_id, stream_list=None):
     """add all statuses by a user to a viewer's stream"""
     stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
@@ -549,7 +568,7 @@ def add_user_statuses_task(viewer_id, user_id, stream_list=None):
         stream.add_user_statuses(viewer, user)
 
 
-@app.task(queue=MEDIUM, ignore_result=True)
+@app.task(queue=MEDIUM)
 def handle_boost_task(boost_id):
     """remove the original post and other, earlier boosts"""
     instance = models.Status.objects.get(id=boost_id)
@@ -563,10 +582,10 @@ def handle_boost_task(boost_id):
 
     for stream in streams.values():
         # people who should see the boost (not people who see the original status)
-        audience = stream.get_stores_for_object(instance)
-        stream.remove_object_from_related_stores(boosted, stores=audience)
+        audience = stream.get_stores_for_users(stream.get_audience(instance))
+        stream.remove_object_from_stores(boosted, audience)
         for status in old_versions:
-            stream.remove_object_from_related_stores(status, stores=audience)
+            stream.remove_object_from_stores(status, audience)
 
 
 def get_status_type(status):

bookwyrm/apps.py

@@ -40,6 +40,7 @@ def ready(self):
             from bookwyrm.telemetry import open_telemetry
 
             open_telemetry.instrumentDjango()
+            open_telemetry.instrumentPostgres()
 
         if settings.ENABLE_PREVIEW_IMAGES and settings.FONTS:
             # Download any fonts that we don't have yet

bookwyrm/connectors/abstract_connector.py

@@ -4,13 +4,16 @@
 import imghdr
 import logging
 import re
+import asyncio
+import requests
+from requests.exceptions import RequestException
+import aiohttp
 
 from django.core.files.base import ContentFile
 from django.db import transaction
-import requests
-from requests.exceptions import RequestException
 
 from bookwyrm import activitypub, models, settings
+from bookwyrm.settings import USER_AGENT
 from .connector_manager import load_more_data, ConnectorException, raise_not_valid_url
 from .format_mappings import format_mappings
 
@@ -52,11 +55,44 @@ def get_search_url(self, query):
         return f"{self.search_url}{quote_plus(query)}"
 
     def process_search_response(self, query, data, min_confidence):
-        """Format the search results based on the formt of the query"""
+        """Format the search results based on the format of the query"""
         if maybe_isbn(query):
             return list(self.parse_isbn_search_data(data))[:10]
         return list(self.parse_search_data(data, min_confidence))[:10]
 
+    async def get_results(self, session, url, min_confidence, query):
+        """try this specific connector"""
+        # pylint: disable=line-too-long
+        headers = {
+            "Accept": (
+                'application/json, application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams"; charset=utf-8'
+            ),
+            "User-Agent": USER_AGENT,
+        }
+        params = {"min_confidence": min_confidence}
+        try:
+            async with session.get(url, headers=headers, params=params) as response:
+                if not response.ok:
+                    logger.info("Unable to connect to %s: %s", url, response.reason)
+                    return
+
+                try:
+                    raw_data = await response.json()
+                except aiohttp.client_exceptions.ContentTypeError as err:
+                    logger.exception(err)
+                    return
+
+                return {
+                    "connector": self,
+                    "results": self.process_search_response(
+                        query, raw_data, min_confidence
+                    ),
+                }
+        except asyncio.TimeoutError:
+            logger.info("Connection timed out for url: %s", url)
+        except aiohttp.ClientError as err:
+            logger.info(err)
+
     @abstractmethod
     def get_or_create_book(self, remote_id):
         """pull up a book record by whatever means possible"""
@@ -321,7 +357,7 @@ def infer_physical_format(format_text):
 
 
 def unique_physical_format(format_text):
-    """only store the format if it isn't diretly in the format mappings"""
+    """only store the format if it isn't directly in the format mappings"""
     format_text = format_text.lower()
     if format_text in format_mappings:
         # try a direct match, so saving this would be redundant