⚡(search) use bulk API and prefetching for reindex_all (#595)

The previous implementation indexed each thread and message with individual
OpenSearch HTTP calls and triggered N+1 DB queries for accesses, recipients
and sender lookups. This rewrites reindex_all to batch documents via
opensearchpy.helpers.bulk and prefetch related objects in a single queryset,
drastically reducing both DB round-trips and HTTP overhead.

Author: jbpenrath

src/backend/core/management/commands/search_reindex.py

@@ -8,6 +8,7 @@
 from core.services.search import create_index_if_not_exists, delete_index
 from core.services.search.tasks import (
     _reindex_all_base,
+    _reindex_mailbox_base,
     reindex_all,
     reindex_mailbox_task,
     reindex_thread_task,
@@ -152,7 +153,15 @@ def _reindex_mailbox(self, mailbox_id, async_mode):
                 self.style.SUCCESS(f"Reindexing task scheduled (ID: {task.id})")
             )
         else:
-            result = reindex_mailbox_task(mailbox_id)  # pylint: disable=no-value-for-parameter
+
+            def update_progress(current, total, success_count, failure_count):
+                """Update progress in the console."""
+                self.stdout.write(
+                    f"Progress: {current}/{total} threads processed "
+                    f"({success_count} succeeded, {failure_count} failed)"
+                )
+
+            result = _reindex_mailbox_base(str(mailbox_uuid), update_progress)
             self.stdout.write(
                 self.style.SUCCESS(
                     f"Reindexing completed: {result.get('success_count', 0)} succeeded, "

src/backend/core/models.py

@@ -1306,10 +1306,8 @@ def thread_unread_filter(user, mailbox_id=None):
 
     @staticmethod
     def unread_filter():
-        """Return a `Q` for filtering a `ThreadAccess` queryset to unread entries.
-
-        Used by `MailboxSerializer._get_cached_counts()` and
-        `compute_unread_mailboxes()` in the search index.
+        """
+        Return a `Q` for filtering a `ThreadAccess` queryset to unread entries.
         """
         return Q(read_at__isnull=True, thread__messaged_at__isnull=False) | Q(
             read_at__lt=F("thread__messaged_at")
@@ -1347,7 +1345,7 @@ def thread_starred_filter(user, mailbox_id=None):
     def starred_filter():
         """Return a `Q` for filtering a `ThreadAccess` queryset to starred entries.
 
-        Used by `compute_starred_mailboxes()` in the search index.
+        Used by `_compute_unread_starred_from_accesses()` in the search index.
         """
         return Q(starred_at__isnull=False)