Enhancement: “live” document updates (#12141)

src/documents/apps.py

@@ -15,6 +15,7 @@ class DocumentsConfig(AppConfig):
         from documents.signals.handlers import add_to_index
         from documents.signals.handlers import run_workflows_added
         from documents.signals.handlers import run_workflows_updated
+        from documents.signals.handlers import send_websocket_document_updated
         from documents.signals.handlers import set_correspondent
         from documents.signals.handlers import set_document_type
         from documents.signals.handlers import set_storage_path
@@ -29,6 +30,7 @@ class DocumentsConfig(AppConfig):
         document_consumption_finished.connect(run_workflows_added)
         document_consumption_finished.connect(add_or_update_document_in_llm_index)
         document_updated.connect(run_workflows_updated)
+        document_updated.connect(send_websocket_document_updated)
 
         import documents.schema  # noqa: F401
 

src/documents/plugins/helpers.py

@@ -1,4 +1,5 @@
 import enum
+from collections.abc import Mapping
 from typing import TYPE_CHECKING
 
 from asgiref.sync import async_to_sync
@@ -47,7 +48,7 @@ class BaseStatusManager:
             async_to_sync(self._channel.flush)
             self._channel = None
 
-    def send(self, payload: dict[str, str | int | None]) -> None:
+    def send(self, payload: Mapping[str, object]) -> None:
         # Ensure the layer is open
         self.open()
 
@@ -73,26 +74,28 @@ class ProgressManager(BaseStatusManager):
         max_progress: int,
         extra_args: dict[str, str | int | None] | None = None,
     ) -> None:
-        payload = {
-            "type": "status_update",
-            "data": {
-                "filename": self.filename,
-                "task_id": self.task_id,
-                "current_progress": current_progress,
-                "max_progress": max_progress,
-                "status": status,
-                "message": message,
-            },
+        data: dict[str, object] = {
+            "filename": self.filename,
+            "task_id": self.task_id,
+            "current_progress": current_progress,
+            "max_progress": max_progress,
+            "status": status,
+            "message": message,
         }
         if extra_args is not None:
-            payload["data"].update(extra_args)
+            data.update(extra_args)
+
+        payload: dict[str, object] = {
+            "type": "status_update",
+            "data": data,
+        }
 
         self.send(payload)
 
 
 class DocumentsStatusManager(BaseStatusManager):
     def send_documents_deleted(self, documents: list[int]) -> None:
-        payload = {
+        payload: dict[str, object] = {
             "type": "documents_deleted",
             "data": {
                 "documents": documents,
@@ -100,3 +103,25 @@ class DocumentsStatusManager(BaseStatusManager):
         }
 
         self.send(payload)
+
+    def send_document_updated(
+        self,
+        *,
+        document_id: int,
+        modified: str,
+        owner_id: int | None = None,
+        users_can_view: list[int] | None = None,
+        groups_can_view: list[int] | None = None,
+    ) -> None:
+        payload: dict[str, object] = {
+            "type": "document_updated",
+            "data": {
+                "document_id": document_id,
+                "modified": modified,
+                "owner_id": owner_id,
+                "users_can_view": users_can_view or [],
+                "groups_can_view": groups_can_view or [],
+            },
+        }
+
+        self.send(payload)

src/documents/signals/handlers.py

@@ -24,6 +24,7 @@ from django.db.models import Q
 from django.dispatch import receiver
 from django.utils import timezone
 from filelock import FileLock
+from rest_framework import serializers
 
 from documents import matching
 from documents.caching import clear_document_caches
@@ -48,6 +49,7 @@ from documents.models import WorkflowAction
 from documents.models import WorkflowRun
 from documents.models import WorkflowTrigger
 from documents.permissions import get_objects_for_user_owner_aware
+from documents.plugins.helpers import DocumentsStatusManager
 from documents.templating.utils import convert_format_str_to_template_format
 from documents.workflows.actions import build_workflow_action_context
 from documents.workflows.actions import execute_email_action
@@ -69,6 +71,7 @@ if TYPE_CHECKING:
     from documents.data_models import DocumentMetadataOverrides
 
 logger = logging.getLogger("paperless.handlers")
+DRF_DATETIME_FIELD = serializers.DateTimeField()
 
 
 def add_inbox_tags(sender, document: Document, logging_group=None, **kwargs) -> None:
@@ -764,6 +767,28 @@ def run_workflows_updated(
     )
 
 
+def send_websocket_document_updated(
+    sender,
+    document: Document,
+    **kwargs,
+) -> None:
+    # At this point, workflows may already have applied additional changes.
+    document.refresh_from_db()
+
+    from documents.data_models import DocumentMetadataOverrides
+
+    doc_overrides = DocumentMetadataOverrides.from_document(document)
+
+    with DocumentsStatusManager() as status_mgr:
+        status_mgr.send_document_updated(
+            document_id=document.id,
+            modified=DRF_DATETIME_FIELD.to_representation(document.modified),
+            owner_id=doc_overrides.owner_id,
+            users_can_view=doc_overrides.view_users,
+            groups_can_view=doc_overrides.view_groups,
+        )
+
+
 def run_workflows(
     trigger_type: WorkflowTrigger.WorkflowTriggerType,
     document: Document | ConsumableDocument,
@@ -1045,7 +1070,11 @@ def add_or_update_document_in_llm_index(sender, document, **kwargs):
 
 
 @receiver(models.signals.post_delete, sender=Document)
-def delete_document_from_llm_index(sender, instance: Document, **kwargs):
+def delete_document_from_llm_index(
+    sender: Any,
+    instance: Document,
+    **kwargs: Any,
+) -> None:
     """
     Delete a document from the LLM index when it is deleted.
     """

src/documents/tasks.py

@@ -62,6 +62,7 @@ from documents.sanity_checker import SanityCheckFailedException
 from documents.signals import document_updated
 from documents.signals.handlers import cleanup_document_deletion
 from documents.signals.handlers import run_workflows
+from documents.signals.handlers import send_websocket_document_updated
 from documents.workflows.utils import get_workflows_for_trigger
 from paperless.config import AIConfig
 from paperless_ai.indexing import llm_index_add_or_update_document
@@ -572,6 +573,11 @@ def check_scheduled_workflows() -> None:
                             workflow_to_run=workflow,
                             document=document,
                         )
+                        # Scheduled workflows dont send document_updated signal, so send a websocket update here to ensure clients are updated
+                        send_websocket_document_updated(
+                            sender=None,
+                            document=document,
+                        )
 
 
 def update_document_parent_tags(tag: Tag, new_parent: Tag) -> None:

src/documents/tests/test_api_documents.py

@@ -1270,7 +1270,11 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         input_doc, overrides = self.get_last_consume_delay_call_args()
 
         self.assertEqual(input_doc.original_file.name, "simple.pdf")
-        self.assertIn(Path(settings.SCRATCH_DIR), input_doc.original_file.parents)
+        self.assertTrue(
+            input_doc.original_file.resolve(strict=False).is_relative_to(
+                Path(settings.SCRATCH_DIR).resolve(strict=False),
+            ),
+        )
         self.assertIsNone(overrides.title)
         self.assertIsNone(overrides.correspondent_id)
         self.assertIsNone(overrides.document_type_id)
@@ -1351,7 +1355,11 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         input_doc, overrides = self.get_last_consume_delay_call_args()
 
         self.assertEqual(input_doc.original_file.name, "simple.pdf")
-        self.assertIn(Path(settings.SCRATCH_DIR), input_doc.original_file.parents)
+        self.assertTrue(
+            input_doc.original_file.resolve(strict=False).is_relative_to(
+                Path(settings.SCRATCH_DIR).resolve(strict=False),
+            ),
+        )
         self.assertIsNone(overrides.title)
         self.assertIsNone(overrides.correspondent_id)
         self.assertIsNone(overrides.document_type_id)

src/documents/tests/test_workflows.py

@@ -643,7 +643,9 @@ class TestWorkflows(
 
         expected_str = f"Document did not match {w}"
         self.assertIn(expected_str, cm.output[0])
-        expected_str = f"Document path {test_file} does not match"
+        expected_str = (
+            f"Document path {Path(test_file).resolve(strict=False)} does not match"
+        )
         self.assertIn(expected_str, cm.output[1])
 
     def test_workflow_no_match_mail_rule(self) -> None:
@@ -2010,6 +2012,36 @@ class TestWorkflows(
         doc.refresh_from_db()
         self.assertEqual(doc.owner, self.user2)
 
+    @mock.patch("documents.tasks.send_websocket_document_updated")
+    def test_workflow_scheduled_trigger_sends_websocket_update(
+        self,
+        mock_send_websocket_document_updated,
+    ) -> None:
+        trigger = WorkflowTrigger.objects.create(
+            type=WorkflowTrigger.WorkflowTriggerType.SCHEDULED,
+            schedule_offset_days=1,
+            schedule_date_field=WorkflowTrigger.ScheduleDateField.CREATED,
+        )
+        action = WorkflowAction.objects.create(assign_owner=self.user2)
+        workflow = Workflow.objects.create(name="Workflow 1", order=0)
+        workflow.triggers.add(trigger)
+        workflow.actions.add(action)
+
+        doc = Document.objects.create(
+            title="sample test",
+            correspondent=self.c,
+            original_filename="sample.pdf",
+            created=timezone.now() - timedelta(days=2),
+        )
+
+        tasks.check_scheduled_workflows()
+
+        self.assertEqual(mock_send_websocket_document_updated.call_count, 1)
+        self.assertEqual(
+            mock_send_websocket_document_updated.call_args.kwargs["document"].pk,
+            doc.pk,
+        )
+
     def test_workflow_scheduled_trigger_added(self) -> None:
         """
         GIVEN:

src/paperless/consumers.py

@@ -1,4 +1,5 @@
 import json
+from typing import Any
 
 from asgiref.sync import async_to_sync
 from channels.exceptions import AcceptConnection
@@ -52,3 +53,10 @@ class StatusConsumer(WebsocketConsumer):
             self.close()
         else:
             self.send(json.dumps(event))
+
+    def document_updated(self, event: Any) -> None:
+        if not self._authenticated():
+            self.close()
+        else:
+            if self._can_view(event["data"]):
+                self.send(json.dumps(event))

src/paperless/settings/__init__.py

@@ -493,18 +493,24 @@ TEMPLATES = [
     },
 ]
 
+_CHANNELS_BACKEND = os.environ.get(
+    "PAPERLESS_CHANNELS_BACKEND",
+    "channels_redis.pubsub.RedisPubSubChannelLayer",
+)
 CHANNEL_LAYERS = {
     "default": {
-        "BACKEND": "channels_redis.pubsub.RedisPubSubChannelLayer",
-        "CONFIG": {
-            "hosts": [_CHANNELS_REDIS_URL],
-            "capacity": 2000,  # default 100
-            "expiry": 15,  # default 60
-            "prefix": _REDIS_KEY_PREFIX,
-        },
+        "BACKEND": _CHANNELS_BACKEND,
     },
 }
 
+if _CHANNELS_BACKEND.startswith("channels_redis."):
+    CHANNEL_LAYERS["default"]["CONFIG"] = {
+        "hosts": [_CHANNELS_REDIS_URL],
+        "capacity": 2000,  # default 100
+        "expiry": 15,  # default 60
+        "prefix": _REDIS_KEY_PREFIX,
+    }
+
 ###############################################################################
 # Email (SMTP) Backend                                                        #
 ###############################################################################

src/paperless/tests/test_websockets.py

@@ -48,6 +48,20 @@ class TestWebSockets(TestCase):
         mock_close.assert_called_once()
         mock_close.reset_mock()
 
+        message = {
+            "type": "document_updated",
+            "data": {"document_id": 10, "modified": "2026-02-17T00:00:00Z"},
+        }
+
+        await channel_layer.group_send(
+            "status_updates",
+            message,
+        )
+        await communicator.receive_nothing()
+
+        mock_close.assert_called_once()
+        mock_close.reset_mock()
+
         message = {"type": "documents_deleted", "data": {"documents": [1, 2, 3]}}
 
         await channel_layer.group_send(
@@ -158,6 +172,40 @@ class TestWebSockets(TestCase):
 
         await communicator.disconnect()
 
+    @mock.patch("paperless.consumers.StatusConsumer._can_view")
+    @mock.patch("paperless.consumers.StatusConsumer._authenticated")
+    async def test_receive_document_updated(self, _authenticated, _can_view) -> None:
+        _authenticated.return_value = True
+        _can_view.return_value = True
+
+        communicator = WebsocketCommunicator(application, "/ws/status/")
+        connected, _ = await communicator.connect()
+        self.assertTrue(connected)
+
+        message = {
+            "type": "document_updated",
+            "data": {
+                "document_id": 10,
+                "modified": "2026-02-17T00:00:00Z",
+                "owner_id": 1,
+                "users_can_view": [1],
+                "groups_can_view": [],
+            },
+        }
+
+        channel_layer = get_channel_layer()
+        assert channel_layer is not None
+        await channel_layer.group_send(
+            "status_updates",
+            message,
+        )
+
+        response = await communicator.receive_json_from()
+
+        self.assertEqual(response, message)
+
+        await communicator.disconnect()
+
     @mock.patch("channels.layers.InMemoryChannelLayer.group_send")
     def test_manager_send_progress(self, mock_group_send) -> None:
         with ProgressManager(task_id="test") as manager:
@@ -190,7 +238,10 @@ class TestWebSockets(TestCase):
         )
 
     @mock.patch("channels.layers.InMemoryChannelLayer.group_send")
-    def test_manager_send_documents_deleted(self, mock_group_send) -> None:
+    def test_manager_send_documents_deleted(
+        self,
+        mock_group_send: mock.MagicMock,
+    ) -> None:
         with DocumentsStatusManager() as manager:
             manager.send_documents_deleted([1, 2, 3])