Use different channels, refactor because now its not just consumer

src/documents/bulk_edit.py

@@ -24,8 +24,7 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.permissions import set_permissions_for_object
-from documents.plugins.helpers import ProgressManager
-from documents.plugins.helpers import ProgressStatusOptions
+from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
 from documents.tasks import update_document_content_maybe_archive_file
@@ -222,13 +221,8 @@ def delete(doc_ids: list[int]) -> Literal["OK"]:
             for id in doc_ids:
                 index.remove_document_by_id(writer, id)
 
-        status_mgr = ProgressManager()
-        status_mgr.send_progress(
-            status=ProgressStatusOptions.DELETED,
-            message="Documents deleted",
-            current_progress=1,
-            max_progress=1,
-        )
+        status_mgr = DocumentsStatusManager()
+        status_mgr.send_documents_deleted(doc_ids)
     except Exception as e:
         if "Data too long for column" in str(e):
             logger.warning(

src/documents/plugins/helpers.py

@@ -13,19 +13,16 @@ class ProgressStatusOptions(str, enum.Enum):
     WORKING = "WORKING"
     SUCCESS = "SUCCESS"
     FAILED = "FAILED"
-    DELETED = "DELETED"
 
 
-class ProgressManager:
+class BaseStatusManager:
     """
     Handles sending of progress information via the channel layer, with proper management
     of the open/close of the layer to ensure messages go out and everything is cleaned up
     """
 
-    def __init__(self, filename: str | None = None, task_id: str | None = None) -> None:
-        self.filename = filename
+    def __init__(self) -> None:
         self._channel: RedisPubSubChannelLayer | None = None
-        self.task_id = task_id
 
     def __enter__(self):
         self.open()
@@ -50,6 +47,24 @@ class ProgressManager:
             async_to_sync(self._channel.flush)
             self._channel = None
 
+    def send(self, payload: dict[str, str | int | None]) -> None:
+        # Ensure the layer is open
+        self.open()
+
+        # Just for IDEs
+        if TYPE_CHECKING:
+            assert self._channel is not None
+
+        # Construct and send the update
+        async_to_sync(self._channel.group_send)("status_updates", payload)
+
+
+class ProgressManager(BaseStatusManager):
+    def __init__(self, filename: str | None = None, task_id: str | None = None) -> None:
+        super().__init__()
+        self.filename = filename
+        self.task_id = task_id
+
     def send_progress(
         self,
         status: ProgressStatusOptions,
@@ -58,13 +73,6 @@ class ProgressManager:
         max_progress: int,
         extra_args: dict[str, str | int | None] | None = None,
     ) -> None:
-        # Ensure the layer is open
-        self.open()
-
-        # Just for IDEs
-        if TYPE_CHECKING:
-            assert self._channel is not None
-
         payload = {
             "type": "status_update",
             "data": {
@@ -79,5 +87,16 @@ class ProgressManager:
         if extra_args is not None:
             payload["data"].update(extra_args)
 
-        # Construct and send the update
-        async_to_sync(self._channel.group_send)("status_updates", payload)
+        self.send(payload)
+
+
+class DocumentsStatusManager(BaseStatusManager):
+    def send_documents_deleted(self, documents: list[int]) -> None:
+        payload = {
+            "type": "documents_deleted",
+            "data": {
+                "documents": documents,
+            },
+        }
+
+        self.send(payload)

src/paperless/consumers.py

@@ -41,4 +41,10 @@ class StatusConsumer(WebsocketConsumer):
             self.close()
         else:
             if self._is_owner_or_unowned(event["data"]):
-                self.send(json.dumps(event["data"]))
+                self.send(json.dumps(event))
+
+    def documents_deleted(self, event):
+        if not self._authenticated():
+            self.close()
+        else:
+            self.send(json.dumps(event))

src/paperless/tests/test_websockets.py

@@ -40,12 +40,12 @@ class TestWebSockets(TestCase):
         connected, subprotocol = await communicator.connect()
         self.assertTrue(connected)
 
-        message = {"task_id": "test"}
+        message = {"type": "status_update", "data": {"task_id": "test"}}
 
         channel_layer = get_channel_layer()
         await channel_layer.group_send(
             "status_updates",
-            {"type": "status_update", "data": message},
+            message,
         )
 
         response = await communicator.receive_json_from()