Move the other bulk edit actions through the new taks tracking route

src/documents/bulk_edit.py

@@ -475,9 +475,9 @@ def run_workflow(
     """
     Manually run a single workflow against the given documents.
 
-    Done asynchronously via a Celery task.
+    Called from the tracked bulk operation task.
     """
-    run_workflows_for_documents.delay(
+    run_workflows_for_documents(
         document_ids=doc_ids,
         workflow_id=workflow,
         filter_documents=filter_documents,

src/documents/bulk_operations.py

@@ -1,4 +1,4 @@
-"""Queued document metadata edits."""
+"""Queued document bulk operations."""
 
 from typing import Any
 
@@ -12,6 +12,7 @@ from rest_framework.exceptions import PermissionDenied
 from documents import bulk_edit
 from documents.models import CustomFieldInstance
 from documents.models import Document
+from documents.models import PaperlessTask
 from documents.permissions import has_document_operation_permissions
 from documents.tasks import bulk_update_documents
 
@@ -25,7 +26,35 @@ MODIFIED_FIELDS = {
     "modify_custom_fields": "custom_fields",
     "set_permissions": None,
 }
-QUEUED_OPERATIONS = frozenset(MODIFIED_FIELDS)
+LEGACY_MODIFIED_FIELDS = {"delete": "deleted_at", "reprocess": "checksum"}
+QUEUED_OPERATIONS = frozenset(
+    (
+        *MODIFIED_FIELDS,
+        "delete",
+        "reprocess",
+        "run_workflow",
+        "rotate",
+        "merge",
+        "merge_as_versions",
+        "split",
+        "delete_pages",
+        "edit_pdf",
+        "remove_password",
+    ),
+)
+USER_OPERATIONS = frozenset(
+    {
+        "split",
+        "merge",
+        "rotate",
+        "delete_pages",
+        "edit_pdf",
+        "remove_password",
+        "merge_as_versions",
+    },
+)
+# merge_as_versions doesn't queue any consume tasks
+TRIGGER_SOURCE_OPERATIONS = USER_OPERATIONS - {"merge_as_versions"}
 
 
 def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
@@ -66,6 +95,7 @@ def execute_bulk_operation(
     document_ids: list[int],
     parameters: dict,
     user_id: int,
+    trigger_source: str = PaperlessTask.TriggerSource.API_UPLOAD,
 ) -> dict:
     if operation not in QUEUED_OPERATIONS:
         raise ValueError(f"Unsupported bulk operation: {operation}")
@@ -79,6 +109,39 @@ def execute_bulk_operation(
     ):
         raise PermissionDenied("Insufficient permissions to execute bulk operation")
 
+    if operation == "run_workflow" and not user.has_perm("documents.view_workflow"):
+        raise PermissionDenied("Insufficient permissions to run this workflow")
+
+    if operation not in MODIFIED_FIELDS:
+        execution_parameters = parameters.copy()
+        if operation in USER_OPERATIONS:
+            execution_parameters["user"] = user
+        if operation in TRIGGER_SOURCE_OPERATIONS:
+            execution_parameters["trigger_source"] = trigger_source
+        modified_field = LEGACY_MODIFIED_FIELDS.get(operation)
+        with set_actor(user):
+            old_values = (
+                snapshot_field(document_ids, modified_field)
+                if settings.AUDIT_LOG_ENABLED and modified_field
+                else {}
+            )
+            method(document_ids, **execution_parameters)
+            if settings.AUDIT_LOG_ENABLED and modified_field:
+                from auditlog.models import LogEntry
+
+                new_values = snapshot_field(document_ids, modified_field)
+                for doc in Document.objects.filter(pk__in=document_ids):
+                    LogEntry.objects.log_create(
+                        instance=doc,
+                        changes={
+                            modified_field: [old_values[doc.pk], new_values[doc.pk]],
+                        },
+                        action=LogEntry.Action.UPDATE,
+                        actor=user,
+                        additional_data={"reason": f"Bulk edit: {operation}"},
+                    )
+        return {"operation": operation, "document_count": len(document_ids)}
+
     affected_ids: list[int] = []
     modified_field = MODIFIED_FIELDS[operation]
     # Keep metadata writes and their audit entries together. Follow-up processing

src/documents/signals/handlers.py

@@ -1206,6 +1206,8 @@ def before_task_publish_handler(
                 "document_count": len(task_kwargs["document_ids"]),
             }
             owner_id = task_kwargs["user_id"]
+            if task_kwargs["operation"] == "delete":
+                task_type = PaperlessTask.TaskType.BULK_DELETE
         else:
             input_data = _extract_input_data(task_type, task_kwargs)
             owner_id = _extract_owner_id(task_type, task_kwargs)

src/documents/views.py

@@ -107,9 +107,9 @@ from documents import bulk_edit
 from documents.bulk_download import ArchiveOnlyStrategy
 from documents.bulk_download import OriginalAndArchiveStrategy
 from documents.bulk_download import OriginalsOnlyStrategy
-from documents.bulk_operations import QUEUED_OPERATIONS
+from documents.bulk_operations import TRIGGER_SOURCE_OPERATIONS
+from documents.bulk_operations import USER_OPERATIONS
 from documents.bulk_operations import execute_bulk_operation
-from documents.bulk_operations import snapshot_field
 from documents.caching import get_llm_suggestion_cache
 from documents.caching import get_metadata_cache
 from documents.caching import get_suggestion_cache
@@ -2974,38 +2974,30 @@ class DocumentSelectionMixin:
 class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
     permission_classes = (IsAuthenticated,)
     parser_classes = (parsers.JSONParser,)
-    METHOD_NAMES_REQUIRING_USER = {
-        "split",
-        "merge",
-        "rotate",
-        "delete_pages",
-        "edit_pdf",
-        "remove_password",
-        "merge_as_versions",
-    }
-    # merge_as_versions doesn't queue any consume tasks
-    METHOD_NAMES_REQUIRING_TRIGGER_SOURCE = METHOD_NAMES_REQUIRING_USER - {
-        "merge_as_versions",
-    }
 
     def _queue_bulk_operation(self, method, documents, parameters, from_webui):
         if "update_callback" in parameters:
             raise ValueError("Unsupported bulk operation parameter: update_callback")
-        signature(method).bind(documents, **parameters)
+        trigger_source = (
+            PaperlessTask.TriggerSource.WEB_UI
+            if from_webui
+            else PaperlessTask.TriggerSource.API_UPLOAD
+        )
+        final_parameters = parameters.copy()
+        if method.__name__ in USER_OPERATIONS:
+            final_parameters["user"] = self.request.user
+        if method.__name__ in TRIGGER_SOURCE_OPERATIONS:
+            final_parameters["trigger_source"] = trigger_source
+        signature(method).bind(documents, **final_parameters)
         task = execute_bulk_operation.apply_async(
             kwargs={
                 "operation": method.__name__,
                 "document_ids": documents,
                 "parameters": parameters,
                 "user_id": self.request.user.pk,
+                "trigger_source": trigger_source,
             },
-            headers={
-                "trigger_source": (
-                    PaperlessTask.TriggerSource.WEB_UI
-                    if from_webui
-                    else PaperlessTask.TriggerSource.API_UPLOAD
-                ),
-            },
+            headers={"trigger_source": trigger_source},
         )
         return Response({"result": "OK", "task_id": task.id})
 
@@ -3129,9 +3121,9 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         user = self.request.user
         from_webui = validated_data.get("from_webui", False)
 
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_USER:
+        if method.__name__ in USER_OPERATIONS:
             parameters["user"] = user
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE:
+        if method.__name__ in TRIGGER_SOURCE_OPERATIONS:
             parameters["trigger_source"] = (
                 PaperlessTask.TriggerSource.WEB_UI
                 if from_webui
@@ -3169,19 +3161,13 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
                 name="BulkEditDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
-                    "task_id": serializers.CharField(required=False),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
     ),
 )
 class BulkEditView(DocumentOperationPermissionMixin):
-    LEGACY_MODIFIED_FIELDS = {
-        "delete": "deleted_at",
-        "reprocess": "checksum",
-        "remove_password": None,
-    }
-
     serializer_class = BulkEditSerializer
 
     def post(self, request, *args, **kwargs):
@@ -3211,14 +3197,6 @@ class BulkEditView(DocumentOperationPermissionMixin):
             user=user,
             validated_data=serializer.validated_data,
         )
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_USER:
-            parameters["user"] = user
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE:
-            parameters["trigger_source"] = (
-                PaperlessTask.TriggerSource.WEB_UI
-                if from_webui
-                else PaperlessTask.TriggerSource.API_UPLOAD
-            )
         if not has_document_operation_permissions(
             user=user,
             documents=documents,
@@ -3228,38 +3206,12 @@ class BulkEditView(DocumentOperationPermissionMixin):
             return HttpResponseForbidden("Insufficient permissions")
 
         try:
-            if method.__name__ in QUEUED_OPERATIONS:
-                return self._queue_bulk_operation(
-                    method,
-                    documents,
-                    parameters,
-                    from_webui,
-                )
-            modified_field = self.LEGACY_MODIFIED_FIELDS.get(method.__name__)
-            if settings.AUDIT_LOG_ENABLED and modified_field:
-                old_values = snapshot_field(documents, modified_field)
-
-            result = method(documents, **parameters)
-
-            if settings.AUDIT_LOG_ENABLED and modified_field:
-                new_values = snapshot_field(documents, modified_field)
-                for doc in Document.objects.filter(pk__in=documents):
-                    LogEntry.objects.log_create(
-                        instance=doc,
-                        changes={
-                            modified_field: [
-                                old_values[doc.pk],
-                                new_values[doc.pk],
-                            ],
-                        },
-                        action=LogEntry.Action.UPDATE,
-                        actor=user,
-                        additional_data={
-                            "reason": f"Bulk edit: {method.__name__}",
-                        },
-                    )
-
-            return Response({"result": result})
+            return self._queue_bulk_operation(
+                method,
+                documents,
+                parameters,
+                from_webui,
+            )
         except Exception as e:
             logger.warning(f"An error occurred performing bulk edit: {e!s}")
             return HttpResponseBadRequest(