This was a leftover

src-ui/src/app/components/admin/tasks/tasks.component.spec.ts

@@ -7,10 +7,15 @@ import {
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { FormsModule } from '@angular/forms'
 import { By } from '@angular/platform-browser'
-import { Router } from '@angular/router'
+import {
+  ActivatedRoute,
+  convertToParamMap,
+  ParamMap,
+  Router,
+} from '@angular/router'
 import { RouterTestingModule } from '@angular/router/testing'
 import { NgbModal, NgbModalRef, NgbModule } from '@ng-bootstrap/ng-bootstrap'
-import { of, throwError } from 'rxjs'
+import { BehaviorSubject, of, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
 import {
   PaperlessTask,
@@ -159,6 +164,7 @@ const sectionCountResponse = {
 }
 
 describe('TasksComponent', () => {
+  let queryParams: BehaviorSubject<ParamMap>
   let component: TasksComponent
   let fixture: ComponentFixture<TasksComponent>
   let tasksService: TasksService
@@ -169,6 +175,7 @@ describe('TasksComponent', () => {
   let toastService: ToastService
 
   beforeEach(async () => {
+    queryParams = new BehaviorSubject(convertToParamMap({}))
     TestBed.configureTestingModule({
       imports: [
         NgbModule,
@@ -181,6 +188,17 @@ describe('TasksComponent', () => {
         ConfirmDialogComponent,
       ],
       providers: [
+        {
+          provide: ActivatedRoute,
+          useValue: {
+            queryParamMap: queryParams.asObservable(),
+            snapshot: {
+              get queryParamMap() {
+                return queryParams.value
+              },
+            },
+          },
+        },
         {
           provide: PermissionsService,
           useValue: {
@@ -238,6 +256,71 @@ describe('TasksComponent', () => {
     flushTaskLists(25, count)
   }
 
+  it('consumes a task link on load into the normal resettable Task ID filter', () => {
+    jest
+      .spyOn(tasksService, 'list')
+      .mockReturnValue(of({ count: 0, results: [] }))
+    jest
+      .spyOn(tasksService, 'statusCounts')
+      .mockReturnValue(of(sectionCountResponse))
+    jest.spyOn(router, 'navigate').mockResolvedValue(true)
+    queryParams.next(convertToParamMap({ task_id: 'linked-task' }))
+    fixture.destroy()
+    fixture = TestBed.createComponent(TasksComponent)
+    component = fixture.componentInstance
+    fixture.detectChanges()
+    expect(router.navigate).toHaveBeenCalledTimes(1)
+    expect(router.navigate).toHaveBeenCalledWith(
+      [],
+      expect.objectContaining({
+        queryParams: { task_id: null },
+        replaceUrl: true,
+      })
+    )
+    expect(component.filterTargetName).toBe('Task ID')
+    expect(component.filterText).toBe('linked-task')
+    expect(component.isFiltered).toBe(true)
+    component.setSection(TaskSection.InProgress)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      task_id: 'linked-task',
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    component.resetFilters()
+    expect(component.filterText).toBe('')
+    expect(component.isFiltered).toBe(false)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      acknowledged: false,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    expect(router.navigate).toHaveBeenCalledTimes(1)
+  })
+
+  it('supports manually filtering by Task ID and clearing the text filter', () => {
+    jest
+      .spyOn(tasksService, 'list')
+      .mockReturnValue(of({ count: 0, results: [] }))
+    jest
+      .spyOn(tasksService, 'statusCounts')
+      .mockReturnValue(of(sectionCountResponse))
+    component.setSection(TaskSection.InProgress)
+    component.setFilterTarget(TaskFilterTargetID.TaskId)
+    component.filterText = tasks[0].task_id
+    jest.advanceTimersByTime(150)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      task_id: tasks[0].task_id,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    component.pagedTasks.set(tasks)
+    expect(component.tasksForSection(TaskSection.NeedsAttention)).toEqual([
+      tasks[0],
+    ])
+    component.resetFilter()
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      acknowledged: false,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+  })
+
   it('should display task sections with counts', () => {
     expect(component.selectedSection).toBe(TaskSection.All)
     expect(component.selectedTaskType).toBeNull()

src-ui/src/app/components/admin/tasks/tasks.component.ts

@@ -1,7 +1,7 @@
 import { JsonPipe, NgTemplateOutlet } from '@angular/common'
 import { Component, inject, OnDestroy, OnInit, signal } from '@angular/core'
 import { FormsModule, ReactiveFormsModule } from '@angular/forms'
-import { Router, RouterLink } from '@angular/router'
+import { ActivatedRoute, Router, RouterLink } from '@angular/router'
 import {
   NgbCollapseModule,
   NgbDropdownModule,
@@ -47,11 +47,13 @@ export enum TaskSection {
 export enum TaskFilterTargetID {
   Name,
   Result,
+  TaskId,
 }
 
 const FILTER_TARGETS = [
   { id: TaskFilterTargetID.Name, name: $localize`Name` },
   { id: TaskFilterTargetID.Result, name: $localize`Result` },
+  { id: TaskFilterTargetID.TaskId, name: $localize`Task ID` },
 ]
 
 const SECTION_LABELS = {
@@ -161,6 +163,7 @@ export class TasksComponent
   tasksService = inject(TasksService)
   private modalService = inject(NgbModal)
   private readonly router = inject(Router)
+  private readonly route = inject(ActivatedRoute)
   private readonly toastService = inject(ToastService)
 
   readonly TaskSection = TaskSection
@@ -275,6 +278,17 @@ export class TasksComponent
   }
 
   ngOnInit() {
+    const taskId = this.route.snapshot.queryParamMap.get('task_id')
+    if (taskId) {
+      this.filterTargetID = TaskFilterTargetID.TaskId
+      this._filterText = taskId
+      this.router.navigate([], {
+        relativeTo: this.route,
+        queryParams: { task_id: null },
+        queryParamsHandling: 'merge',
+        replaceUrl: true,
+      })
+    }
     this.tasksService.reload()
     this.reloadPage()
     timer(5000, 5000)
@@ -636,6 +650,9 @@ export class TasksComponent
 
     const query = this._filterText.toLowerCase()
 
+    if (this.filterTargetID === TaskFilterTargetID.TaskId) {
+      return task.task_id.toLowerCase() === query
+    }
     if (this.filterTargetID == TaskFilterTargetID.Name) {
       return [
         this.taskDisplayName(task),
@@ -706,9 +723,15 @@ export class TasksComponent
     }
 
     if (this._filterText.length) {
-      params[
+      const parameter = {
-        this.filterTargetID === TaskFilterTargetID.Name ? 'name' : 'result'
+        [TaskFilterTargetID.Name]: 'name',
-      ] = this._filterText
+        [TaskFilterTargetID.Result]: 'result',
+        [TaskFilterTargetID.TaskId]: 'task_id',
+      }[this.filterTargetID]
+      params[parameter] = this._filterText
+      if (this.filterTargetID === TaskFilterTargetID.TaskId) {
+        delete params.acknowledged
+      }
     }
 
     return params

src-ui/src/app/components/app-frame/app-frame.component.spec.ts

@@ -11,8 +11,10 @@ import { ActivatedRoute, Router } from '@angular/router'
 import { RouterTestingModule } from '@angular/router/testing'
 import { NgbModal, NgbModalModule, NgbModule } from '@ng-bootstrap/ng-bootstrap'
 import { provideUiTour } from 'ngx-ui-tour-ng-bootstrap'
-import { of, throwError } from 'rxjs'
+import { of, Subject, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
+import { BulkOperationStatus } from 'src/app/data/bulk-operation'
+import { PaperlessTaskStatus } from 'src/app/data/paperless-task'
 import { SavedView } from 'src/app/data/saved-view'
 import { HideableSidebarItemID, SETTINGS_KEYS } from 'src/app/data/ui-settings'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
@@ -21,10 +23,11 @@ import {
   DjangoMessageLevel,
   DjangoMessagesService,
 } from 'src/app/services/django-messages.service'
+import { DocumentListViewService } from 'src/app/services/document-list-view.service'
 import { OpenDocumentsService } from 'src/app/services/open-documents.service'
 import {
-  PermissionType,
   PermissionsService,
+  PermissionType,
 } from 'src/app/services/permissions.service'
 import { RemoteVersionService } from 'src/app/services/rest/remote-version.service'
 import { SavedViewService } from 'src/app/services/rest/saved-view.service'
@@ -95,6 +98,7 @@ describe('AppFrameComponent', () => {
   let savedViewSpy
   let modalService: NgbModal
   let maybeRefreshSpy
+  let bulkEvents: Subject<BulkOperationStatus>
   let tasksService: TasksService
 
   beforeEach(async () => {
@@ -182,6 +186,10 @@ describe('AppFrameComponent', () => {
     savedViewSpy = jest.spyOn(savedViewService, 'reload')
     maybeRefreshSpy = jest.spyOn(savedViewService, 'maybeRefreshDocumentCounts')
 
+    bulkEvents = new Subject<BulkOperationStatus>()
+    jest
+      .spyOn(TestBed.inject(WebsocketStatusService), 'onBulkOperationStatus')
+      .mockReturnValue(bulkEvents.asObservable())
     fixture = TestBed.createComponent(AppFrameComponent)
     component = fixture.componentInstance
 
@@ -190,6 +198,70 @@ describe('AppFrameComponent', () => {
     fixture.detectChanges()
   })
 
+  it('refreshes and notifies on completion without client-side task tracking', () => {
+    const list = TestBed.inject(DocumentListViewService)
+    const reload = jest.spyOn(list, 'reload').mockImplementation(() => {})
+    jest.spyOn(list, 'reduceSelectionToFilter').mockImplementation(() => {})
+    const tasksReload = jest
+      .spyOn(tasksService, 'reload')
+      .mockImplementation(() => {})
+    const success = jest.spyOn(toastService, 'showSuccess')
+    const refresh = jest
+      .spyOn(openDocumentsService, 'refreshDocument')
+      .mockImplementation(() => {})
+    jest
+      .spyOn(openDocumentsService, 'getOpenDocuments')
+      .mockReturnValue([document])
+    tasksReload.mockClear()
+    maybeRefreshSpy.mockClear()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Started,
+    })
+    expect(reload).not.toHaveBeenCalled()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(reload).toHaveBeenCalledTimes(1)
+    expect(refresh).toHaveBeenCalledWith(document.id)
+    expect(maybeRefreshSpy).toHaveBeenCalledTimes(1)
+    expect(tasksReload).toHaveBeenCalledTimes(1)
+    expect(success).toHaveBeenCalledWith('Bulk operation completed.')
+    fixture.destroy()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(reload).toHaveBeenCalledTimes(1)
+  })
+
+  it('refreshes and reports failed or revoked bulk operations', () => {
+    const list = TestBed.inject(DocumentListViewService)
+    const reload = jest.spyOn(list, 'reload').mockImplementation(() => {})
+    jest.spyOn(list, 'reduceSelectionToFilter').mockImplementation(() => {})
+    jest.spyOn(tasksService, 'reload').mockImplementation(() => {})
+    const error = jest.spyOn(toastService, 'showError')
+    const success = jest.spyOn(toastService, 'showSuccess')
+    for (const status of [
+      PaperlessTaskStatus.Failure,
+      PaperlessTaskStatus.Revoked,
+    ]) {
+      bulkEvents.next({
+        task_id: 'bulk-task',
+        operation: 'set_permissions',
+        status,
+        error: 'Could not update',
+      })
+    }
+    expect(reload).toHaveBeenCalledTimes(2)
+    expect(error).toHaveBeenCalledWith(expect.any(String), 'Could not update')
+    expect(success).not.toHaveBeenCalled()
+  })
+
   it('should initialize the saved view service', () => {
     expect(savedViewSpy).toHaveBeenCalled()
   })

src-ui/src/app/components/app-frame/app-frame.component.ts

@@ -8,12 +8,14 @@ import {
 import { NgClass } from '@angular/common'
 import {
   Component,
+  DestroyRef,
   HostListener,
   inject,
   OnInit,
   signal,
   viewChild,
 } from '@angular/core'
+import { takeUntilDestroyed } from '@angular/core/rxjs-interop'
 import { FormsModule } from '@angular/forms'
 import { ActivatedRoute, Router, RouterModule } from '@angular/router'
 import {
@@ -24,10 +26,11 @@ import {
   NgbPopoverModule,
 } from '@ng-bootstrap/ng-bootstrap'
 import { TourNgBootstrap } from 'ngx-ui-tour-ng-bootstrap'
-import { Observable } from 'rxjs'
+import { filter, Observable } from 'rxjs'
 import { first } from 'rxjs/operators'
 import { BootstrapIconComponent } from 'src/app/components/common/bootstrap-icon/bootstrap-icon.component'
 import { Document } from 'src/app/data/document'
+import { PaperlessTaskStatus } from 'src/app/data/paperless-task'
 import { SavedView } from 'src/app/data/saved-view'
 import {
   CollapsibleSection,
@@ -41,6 +44,7 @@ import {
   DjangoMessageLevel,
   DjangoMessagesService,
 } from 'src/app/services/django-messages.service'
+import { DocumentListViewService } from 'src/app/services/document-list-view.service'
 import { OpenDocumentsService } from 'src/app/services/open-documents.service'
 import {
   PermissionAction,
@@ -101,6 +105,8 @@ export class AppFrameComponent
   extends ComponentWithPermissions
   implements OnInit, ComponentCanDeactivate
 {
+  private destroyRef = inject(DestroyRef)
+  private documentList = inject(DocumentListViewService)
   router = inject(Router)
   private activatedRoute = inject(ActivatedRoute)
   private openDocumentsService = inject(OpenDocumentsService)
@@ -164,6 +170,44 @@ export class AppFrameComponent
   }
 
   ngOnInit(): void {
+    this.websocketStatusService
+      .onBulkOperationStatus()
+      .pipe(
+        filter(({ status }) =>
+          [
+            PaperlessTaskStatus.Success,
+            PaperlessTaskStatus.Failure,
+            PaperlessTaskStatus.Revoked,
+          ].includes(status)
+        ),
+        takeUntilDestroyed(this.destroyRef)
+      )
+      .subscribe(({ status, error }) => {
+        this.documentList.reload()
+        this.documentList.reduceSelectionToFilter()
+        // maybe someday we can optimize to only refresh affected documents
+        this.openDocumentsService.getOpenDocuments().forEach((doc) => {
+          this.openDocumentsService.refreshDocument(doc.id)
+        })
+        this.savedViewService.maybeRefreshDocumentCounts()
+        if (
+          this.permissionsService.currentUserCan(
+            PermissionAction.View,
+            PermissionType.PaperlessTask
+          )
+        ) {
+          this.tasksService.reload()
+        }
+        if (status === PaperlessTaskStatus.Success) {
+          this.toastService.showSuccess($localize`Bulk operation completed.`)
+        } else {
+          this.toastService.showError(
+            $localize`Bulk operation failed. Some changes may already have been applied.`,
+            error
+          )
+        }
+      })
+
     this.lastScrollY = window.scrollY
     this.detectClassicScrollbars()
 

src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.spec.ts

@@ -559,18 +559,19 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'modify_tags',
       parameters: { add_tags: [101], remove_tags: [] },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should execute modify tags bulk operation for all filtered documents', () => {
@@ -607,17 +608,18 @@ describe('BulkEditorComponent', () => {
     const req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       all: true,
       filters: { title_search: 'apple' },
       excluded_documents: [4],
       method: 'modify_tags',
       parameters: { add_tags: [101], remove_tags: [] },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
   })
 
   it('should execute modify tags bulk operation with confirmation dialog if enabled', () => {
@@ -644,12 +646,12 @@ describe('BulkEditorComponent', () => {
     httpTestingController
       .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
       .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should set modal dialog text accordingly for tag edit confirmation', () => {
@@ -729,18 +731,19 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'set_correspondent',
       parameters: { correspondent: 101 },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should execute modify correspondent bulk operation with confirmation dialog if enabled', () => {
@@ -767,12 +770,12 @@ describe('BulkEditorComponent', () => {
     httpTestingController
       .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
       .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should set modal dialog text accordingly for correspondent edit confirmation', () => {
@@ -827,18 +830,19 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'set_document_type',
       parameters: { document_type: 101 },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should execute modify document type bulk operation with confirmation dialog if enabled', () => {
@@ -865,12 +869,12 @@ describe('BulkEditorComponent', () => {
     httpTestingController
       .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
       .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should set modal dialog text accordingly for document type edit confirmation', () => {
@@ -925,18 +929,19 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'set_storage_path',
       parameters: { storage_path: 101 },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should execute modify storage path bulk operation with confirmation dialog if enabled', () => {
@@ -963,12 +968,12 @@ describe('BulkEditorComponent', () => {
     httpTestingController
       .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
       .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should set modal dialog text accordingly for storage path edit confirmation', () => {
@@ -1023,18 +1028,19 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'modify_custom_fields',
       parameters: { add_custom_fields: [101], remove_custom_fields: [102] },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should execute modify custom fields bulk operation with confirmation dialog if enabled', () => {
@@ -1061,12 +1067,12 @@ describe('BulkEditorComponent', () => {
     httpTestingController
       .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
       .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
 
     // coverage for modal messages
     component.setCustomFields({
@@ -1540,6 +1546,28 @@ describe('BulkEditorComponent', () => {
     )
   })
 
+  it('closes a queued permissions dialog without refreshing before completion', () => {
+    let modal: NgbModalRef
+    modalService.activeInstances.subscribe((instances) => {
+      if (instances.length) modal = instances[0]
+    })
+    const reload = jest.spyOn(documentListViewService, 'reload')
+    component.setPermissions()
+    const close = jest.spyOn(modal, 'close')
+    modal.componentInstance.confirmClicked.emit({
+      permissions: {},
+      merge: true,
+    })
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}documents/bulk_edit/`
+    )
+    req.flush({ result: 'OK', task_id: 'queued-operation' })
+    expect(close).toHaveBeenCalledTimes(1)
+    expect(reload).not.toHaveBeenCalled()
+    expect(documentListViewService.selected.size).toBe(0)
+    expect(documentListViewService.allSelected).toBe(false)
+  })
+
   it('should support bulk permissions update', () => {
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
@@ -1572,8 +1600,9 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: [3, 4],
       method: 'set_permissions',
       parameters: {
@@ -1581,12 +1610,12 @@ describe('BulkEditorComponent', () => {
         merge: true,
       },
     })
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should not attempt to retrieve objects if user does not have permissions', () => {
@@ -1887,15 +1916,18 @@ describe('BulkEditorComponent', () => {
     expect(toastServiceShowErrorSpy).toHaveBeenCalled()
     expect(listReloadSpy).not.toHaveBeenCalled()
 
-    modal.componentInstance.succeeded.emit()
+    modal.componentInstance.succeeded.emit({
-    expect(toastServiceShowInfoSpy).toHaveBeenCalled()
+      result: 'OK',
-    expect(listReloadSpy).toHaveBeenCalled()
+      task_id: 'custom-fields-task',
-    httpTestingController.match(
+    })
+    expect(toastServiceShowInfoSpy).not.toHaveBeenCalled()
+    expect(listReloadSpy).not.toHaveBeenCalled()
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
-    httpTestingController.match(
+    httpTestingController.expectNone(
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
   })
 
   it('should create share link bundle and enable manage callback', () => {

src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts

@@ -104,11 +104,11 @@ export class BulkEditorComponent
   private correspondentService = inject(CorrespondentService)
   list = inject(DocumentListViewService)
   private documentService = inject(DocumentService)
+  private trashService = inject(TrashService)
   private modalService = inject(NgbModal)
   private openDocumentService = inject(OpenDocumentsService)
   private settings = inject(SettingsService)
   private toastService = inject(ToastService)
-  private trashService = inject(TrashService)
   private storagePathService = inject(StoragePathService)
   private customFieldService = inject(CustomFieldsService)
   public readonly permissionService = inject(PermissionsService)
@@ -299,11 +299,16 @@ export class BulkEditorComponent
       .bulkEdit(overrideSelection ?? this.getSelectionQuery(), method, args)
       .pipe(first())
       .subscribe({
-        next: () => this.handleOperationSuccess(modal),
+        next: () => this.handleOperationQueued(modal),
         error: (error) => this.handleOperationError(modal, error),
       })
   }
 
+  private handleOperationQueued(modal: NgbModalRef) {
+    modal?.close()
+    this.list.selectNone()
+  }
+
   private executeDocumentAction(
     modal: NgbModalRef,
     request: Observable<any>,
@@ -1088,14 +1093,7 @@ export class BulkEditorComponent
 
     dialog.selection = this.getSelectionQuery()
     dialog.selectionCount = this.getSelectionSize()
-    dialog.succeeded.subscribe((result) => {
+    dialog.succeeded.subscribe(() => this.handleOperationQueued(null))
-      this.toastService.showSuccess($localize`Custom fields updated.`)
-      this.list.reload()
-      this.list.reduceSelectionToFilter()
-      this.list.selected.forEach((id) => {
-        this.openDocumentService.refreshDocument(id)
-      })
-    })
     dialog.failed.subscribe((error) => {
       this.toastService.showError(
         $localize`Error updating custom fields.`,

src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.spec.ts

@@ -59,7 +59,7 @@ describe('CustomFieldsBulkEditDialogComponent', () => {
   it('should emit succeeded event and close modal on successful save', () => {
     const editSpy = jest
       .spyOn(documentService, 'bulkEdit')
-      .mockReturnValue(of('Success'))
+      .mockReturnValue(of({ result: 'OK', task_id: 'custom-fields-task' }))
     const successSpy = jest.spyOn(component.succeeded, 'emit')
 
     component.selection = [1, 2]

src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.ts

@@ -16,6 +16,7 @@ import { NumberComponent } from 'src/app/components/common/input/number/number.c
 import { SelectComponent } from 'src/app/components/common/input/select/select.component'
 import { TextComponent } from 'src/app/components/common/input/text/text.component'
 import { UrlComponent } from 'src/app/components/common/input/url/url.component'
+import { BulkOperationResponse } from 'src/app/data/bulk-operation'
 import { CustomField, CustomFieldDataType } from 'src/app/data/custom-field'
 import {
   DocumentSelectionQuery,
@@ -49,7 +50,7 @@ export class CustomFieldsBulkEditDialogComponent {
   CustomFieldDataType = CustomFieldDataType
 
   @Output()
-  succeeded = new EventEmitter()
+  succeeded = new EventEmitter<BulkOperationResponse>()
 
   @Output()
   failed = new EventEmitter()
@@ -103,6 +104,8 @@ export class CustomFieldsBulkEditDialogComponent {
   }
 
   public save() {
+    if (this.networkActive) return
+    this.networkActive = true
     this.documentService
       .bulkEdit(this.selection, 'modify_custom_fields', {
         add_custom_fields: this.form.value,
@@ -110,11 +113,12 @@ export class CustomFieldsBulkEditDialogComponent {
       })
       .pipe(first())
       .subscribe({
-        next: () => {
+        next: (response) => {
           this.activeModal.close()
-          this.succeeded.emit()
+          this.succeeded.emit(response)
         },
         error: (error) => {
+          this.networkActive = false
           this.failed.emit(error)
         },
       })

src-ui/src/app/data/bulk-operation.ts

@@ -0,0 +1,13 @@
+import { PaperlessTaskStatus } from './paperless-task'
+
+export interface BulkOperationResponse {
+  result: string
+  task_id: string
+}
+
+export interface BulkOperationStatus {
+  task_id: string
+  operation: string
+  status: PaperlessTaskStatus
+  error?: string
+}

src-ui/src/app/services/rest/document.service.spec.ts

@@ -248,6 +248,7 @@ describe(`DocumentService`, () => {
     )
     expect(req.request.method).toEqual('POST')
     expect(req.request.body).toEqual({
+      from_webui: true,
       documents: ids,
       method,
       parameters,
@@ -271,6 +272,7 @@ describe(`DocumentService`, () => {
     )
     expect(req.request.method).toEqual('POST')
     expect(req.request.body).toEqual({
+      from_webui: true,
       all: true,
       filters: { title__icontains: 'apple' },
       excluded_documents: [2, 3],

src-ui/src/app/services/rest/document.service.ts

@@ -14,6 +14,7 @@ import { DocumentSuggestions } from 'src/app/data/document-suggestions'
 import { FilterRule } from 'src/app/data/filter-rule'
 import { Results, SelectionData } from 'src/app/data/results'
 import { SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import { BulkOperationResponse } from '../../data/bulk-operation'
 import { queryParamsFromFilterRules } from '../../utils/query-params'
 import {
   PermissionAction,
@@ -337,11 +338,15 @@ export class DocumentService extends AbstractPaperlessService<Document> {
     method: DocumentBulkEditMethod,
     args: any
   ) {
-    return this.http.post(this.getResourceUrl(null, 'bulk_edit'), {
+    return this.http.post<BulkOperationResponse>(
-      ...selection,
+      this.getResourceUrl(null, 'bulk_edit'),
-      method: method,
+      {
-      parameters: args,
+        ...selection,
-    })
+        method: method,
+        from_webui: true,
+        parameters: args,
+      }
+    )
   }
 
   deleteDocuments(selection: DocumentSelectionQuery) {

src-ui/src/app/services/websocket-status.service.spec.ts

@@ -438,6 +438,20 @@ describe('ConsumerStatusService', () => {
     expect(updated).toBeTruthy()
   })
 
+  it('forwards bulk operation completion events', () => {
+    const received = jest.fn()
+    websocketStatusService.onBulkOperationStatus().subscribe(received)
+    const data = {
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: 'success',
+    }
+    websocketStatusService.connect()
+    server.send({ type: WebsocketStatusType.BULK_OPERATION_STATUS, data })
+    websocketStatusService.disconnect()
+    expect(received).toHaveBeenCalledWith(data)
+  })
+
   it('should ignore keep-alive heartbeat messages from the server', () => {
     let updated = false
     let deleted = false

src-ui/src/app/services/websocket-status.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, inject, signal } from '@angular/core'
 import { Subject } from 'rxjs'
 import { environment } from 'src/environments/environment'
+import { BulkOperationStatus } from '../data/bulk-operation'
 import { User } from '../data/user'
 import { WebsocketDocumentUpdatedMessage } from '../data/websocket-document-updated-message'
 import { WebsocketDocumentsDeletedMessage } from '../data/websocket-documents-deleted-message'
@@ -12,6 +13,7 @@ export enum WebsocketStatusType {
   DOCUMENTS_DELETED = 'documents_deleted',
   DOCUMENT_UPDATED = 'document_updated',
   HEARTBEAT = 'heartbeat',
+  BULK_OPERATION_STATUS = 'bulk_operation_status',
 }
 
 // see ProgressStatusOptions in src/documents/plugins/helpers.py
@@ -116,6 +118,13 @@ export class WebsocketStatusService {
   private readonly documentDeletedSubject = new Subject<boolean>()
   private readonly documentUpdatedSubject =
     new Subject<WebsocketDocumentUpdatedMessage>()
+  private readonly bulkOperationStatusSubject =
+    new Subject<BulkOperationStatus>()
+
+  onBulkOperationStatus() {
+    return this.bulkOperationStatusSubject.asObservable()
+  }
+
   private readonly connectionStatusSubject = new Subject<boolean>()
 
   private get(taskId: string, filename?: string) {
@@ -192,6 +201,7 @@ export class WebsocketStatusService {
           | WebsocketProgressMessage
           | WebsocketDocumentsDeletedMessage
           | WebsocketDocumentUpdatedMessage
+          | BulkOperationStatus
       } = JSON.parse(ev.data)
 
       switch (type) {
@@ -209,6 +219,12 @@ export class WebsocketStatusService {
           this.handleProgressUpdate(messageData as WebsocketProgressMessage)
           break
 
+        case WebsocketStatusType.BULK_OPERATION_STATUS:
+          this.bulkOperationStatusSubject.next(
+            messageData as BulkOperationStatus
+          )
+          break
+
         case WebsocketStatusType.HEARTBEAT:
           // keep-alive from the server, see paperless.consumers.StatusConsumer
           break

src/documents/bulk_edit.py

@@ -48,6 +48,7 @@ from documents.versioning import get_root_document
 from documents.versioning import versions_newest_first
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from collections.abc import Mapping
 
     from django.contrib.auth.models import User
@@ -131,9 +132,24 @@ def _scratch_path(name: str) -> Path:
     return Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / name
 
 
+def _finish_metadata_edit(
+    document_ids: list[int],
+    update_callback: Callable[[list[int]], None] | None,
+) -> None:
+    if update_callback is not None:
+        update_callback(document_ids)
+    else:
+        bulk_update_documents.apply_async(
+            kwargs={"document_ids": document_ids},
+            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
+        )
+
+
 def set_correspondent(
     doc_ids: list[int],
     correspondent: Correspondent,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
     if correspondent:
         correspondent = Correspondent.objects.only("pk").get(id=correspondent)
@@ -146,15 +162,17 @@ def set_correspondent(
     affected_docs = list(qs.values_list("pk", flat=True))
     qs.update(correspondent=correspondent)
 
-    bulk_update_documents.apply_async(
+    _finish_metadata_edit(affected_docs, update_callback)
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
 
     return "OK"
 
 
-def set_storage_path(doc_ids: list[int], storage_path: StoragePath) -> Literal["OK"]:
+def set_storage_path(
+    doc_ids: list[int],
+    storage_path: StoragePath,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
     if storage_path:
         storage_path = StoragePath.objects.only("pk").get(id=storage_path)
 
@@ -168,15 +186,17 @@ def set_storage_path(doc_ids: list[int], storage_path: StoragePath) -> Literal["
     affected_docs = list(qs.values_list("pk", flat=True))
     qs.update(storage_path=storage_path)
 
-    bulk_update_documents.apply_async(
+    _finish_metadata_edit(affected_docs, update_callback)
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
 
     return "OK"
 
 
-def set_document_type(doc_ids: list[int], document_type: DocumentType) -> Literal["OK"]:
+def set_document_type(
+    doc_ids: list[int],
+    document_type: DocumentType,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
     if document_type:
         document_type = DocumentType.objects.only("pk").get(id=document_type)
 
@@ -188,15 +208,17 @@ def set_document_type(doc_ids: list[int], document_type: DocumentType) -> Litera
     affected_docs = list(qs.values_list("pk", flat=True))
     qs.update(document_type=document_type)
 
-    bulk_update_documents.apply_async(
+    _finish_metadata_edit(affected_docs, update_callback)
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
 
     return "OK"
 
 
-def add_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
+def add_tag(
+    doc_ids: list[int],
+    tag: int,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
     tag_obj = Tag.objects.get(pk=tag)
     tags_to_add = [tag_obj, *tag_obj.get_ancestors()]
 
@@ -217,15 +239,17 @@ def add_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
         DocumentTagRelationship.objects.bulk_create(to_create)
 
     if affected_docs:
-        bulk_update_documents.apply_async(
+        _finish_metadata_edit(list(affected_docs), update_callback)
-            kwargs={"document_ids": list(affected_docs)},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
 
     return "OK"
 
 
-def remove_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
+def remove_tag(
+    doc_ids: list[int],
+    tag: int,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
     tag_obj = Tag.objects.get(pk=tag)
     tag_ids = [tag_obj.id, *tag_obj.get_descendants_pks()]
 
@@ -238,10 +262,7 @@ def remove_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
     qs.delete()
 
     if affected_docs:
-        bulk_update_documents.apply_async(
+        _finish_metadata_edit(affected_docs, update_callback)
-            kwargs={"document_ids": affected_docs},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
 
     return "OK"
 
@@ -250,6 +271,8 @@ def modify_tags(
     doc_ids: list[int],
     add_tags: list[int],
     remove_tags: list[int],
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
     qs = Document.objects.filter(id__in=doc_ids).only("pk")
     affected_docs = list(qs.values_list("pk", flat=True))
@@ -299,10 +322,7 @@ def modify_tags(
                 )
 
     if affected_docs:
-        bulk_update_documents.apply_async(
+        _finish_metadata_edit(affected_docs, update_callback)
-            kwargs={"document_ids": affected_docs},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
 
     return "OK"
 
@@ -311,6 +331,8 @@ def modify_custom_fields(
     doc_ids: list[int],
     add_custom_fields: list[int] | dict,
     remove_custom_fields: list[int],
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
     qs = Document.objects.filter(id__in=doc_ids).only("pk")
     affected_docs = list(qs.values_list("pk", flat=True))
@@ -373,10 +395,7 @@ def modify_custom_fields(
         field_id__in=remove_custom_fields,
     ).hard_delete()
 
-    bulk_update_documents.apply_async(
+    _finish_metadata_edit(affected_docs, update_callback)
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
 
     return "OK"
 
@@ -457,6 +476,7 @@ def set_permissions(
     *,
     owner: User | None = None,
     merge: bool = False,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
     qs = Document.objects.filter(id__in=doc_ids).select_related("owner")
 
@@ -474,10 +494,7 @@ def set_permissions(
         merge=merge,
     )
 
-    bulk_update_documents.apply_async(
+    _finish_metadata_edit(affected_docs, update_callback)
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
 
     return "OK"
 

src/documents/bulk_operations.py

@@ -0,0 +1,107 @@
+"""Queued document metadata edits."""
+
+from typing import Any
+
+from auditlog.context import set_actor
+from celery import shared_task
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.db import transaction
+from rest_framework.exceptions import PermissionDenied
+
+from documents import bulk_edit
+from documents.models import CustomFieldInstance
+from documents.models import Document
+from documents.permissions import has_document_operation_permissions
+from documents.tasks import bulk_update_documents
+
+MODIFIED_FIELDS = {
+    "set_correspondent": "correspondent",
+    "set_document_type": "document_type",
+    "set_storage_path": "storage_path",
+    "add_tag": "tags",
+    "remove_tag": "tags",
+    "modify_tags": "tags",
+    "modify_custom_fields": "custom_fields",
+    "set_permissions": None,
+}
+QUEUED_OPERATIONS = frozenset(MODIFIED_FIELDS)
+
+
+def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
+    """
+    Returns each document's current value of field, for the audit log.
+
+    Tags and custom fields are one row per value, so they are gathered
+    into a sorted list of pks per document (empty when there are none).
+    Reading them through Document.values() instead would join those rows
+    and return one arbitrary value per document.
+    """
+    if field == "tags":
+        rows = (
+            Document.tags.through.objects.filter(document_id__in=doc_ids)
+            .order_by("tag_id")
+            .values_list("document_id", "tag_id")
+        )
+    elif field == "custom_fields":
+        rows = (
+            CustomFieldInstance.objects.filter(document_id__in=doc_ids)
+            .order_by("pk")
+            .values_list("document_id", "pk")
+        )
+    else:
+        return dict(
+            Document.objects.filter(pk__in=doc_ids).values_list("pk", field),
+        )
+
+    values: dict[int, list[int]] = {doc_id: [] for doc_id in doc_ids}
+    for doc_id, pk in rows:
+        values[doc_id].append(pk)
+    return values
+
+
+@shared_task
+def execute_bulk_operation(
+    operation: str,
+    document_ids: list[int],
+    parameters: dict,
+    user_id: int,
+) -> dict:
+    if operation not in QUEUED_OPERATIONS:
+        raise ValueError(f"Unsupported bulk operation: {operation}")
+    method = getattr(bulk_edit, operation)
+    user = User.objects.get(pk=user_id)
+    if not has_document_operation_permissions(
+        user=user,
+        documents=document_ids,
+        method=method,
+        parameters=parameters,
+    ):
+        raise PermissionDenied("Insufficient permissions to execute bulk operation")
+
+    affected_ids: list[int] = []
+    modified_field = MODIFIED_FIELDS[operation]
+    # Keep metadata writes and their audit entries together. Follow-up processing
+    # runs after commit, within this same tracked task.
+    with set_actor(user), transaction.atomic():
+        old_values = (
+            snapshot_field(document_ids, modified_field)
+            if settings.AUDIT_LOG_ENABLED and modified_field
+            else {}
+        )
+        method(document_ids, **parameters, update_callback=affected_ids.extend)
+        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}"},
+                )
+    if affected_ids:
+        bulk_update_documents(list(dict.fromkeys(affected_ids)))
+    return {"operation": operation, "document_count": len(document_ids)}

src/documents/permissions.py

@@ -730,3 +730,100 @@ class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
         "OPTIONS": ["documents.change_paperlesstask"],
         "POST": ["documents.change_paperlesstask"],
     }
+
+
+def has_document_operation_permissions(
+    *,
+    user: User,
+    documents: list[int],
+    method,
+    parameters: dict[str, Any],
+) -> bool:
+    from documents import bulk_edit
+
+    if not user.is_active:
+        return False
+    if user.is_superuser:
+        return True
+
+    root_docs = {
+        get_root_document(doc)
+        for doc in Document.objects.select_related(
+            "owner",
+            "root_document__owner",
+        ).filter(pk__in=documents)
+    }
+    user_is_owner_of_all_documents = all(
+        (doc.owner == user or doc.owner is None) for doc in root_docs
+    )
+
+    # check global and object permissions for all documents
+    has_perms = (
+        user.has_perm(
+            "documents.change_document",
+        )
+        and not Document.global_objects.filter(pk__in=documents)
+        .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
+        .exists()
+    )
+
+    # check ownership for methods that change original document
+    if (
+        (
+            has_perms
+            and method
+            in [
+                bulk_edit.set_permissions,
+                bulk_edit.delete,
+                bulk_edit.rotate,
+                bulk_edit.delete_pages,
+                bulk_edit.edit_pdf,
+                bulk_edit.merge_as_versions,
+                bulk_edit.remove_password,
+            ]
+        )
+        or (
+            method in [bulk_edit.merge, bulk_edit.split]
+            and parameters.get("delete_originals")
+        )
+        or (method == bulk_edit.edit_pdf and parameters.get("update_document"))
+    ):
+        has_perms = has_perms and user_is_owner_of_all_documents
+
+    # check global add permissions for methods that create documents
+    if (
+        has_perms
+        and (
+            method in [bulk_edit.split, bulk_edit.merge]
+            or (
+                method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
+                and not parameters.get("update_document")
+            )
+        )
+        and not user.has_perm("documents.add_document")
+    ):
+        has_perms = False
+
+    # check global delete permissions for methods that delete documents
+    if (
+        has_perms
+        and (
+            method == bulk_edit.delete
+            # Sources stop being documents of their own, and removing one
+            # again afterwards needs delete_document
+            or method == bulk_edit.merge_as_versions
+            or (
+                method in [bulk_edit.merge, bulk_edit.split]
+                and parameters.get("delete_originals")
+            )
+            or (
+                method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
+                and parameters.get("delete_original")
+                and not parameters.get("update_document")
+            )
+        )
+        and not user.has_perm("documents.delete_document")
+    ):
+        has_perms = False
+
+    return has_perms

src/documents/signals/handlers.py

@@ -8,6 +8,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
 
+from asgiref.sync import async_to_sync
 from celery import shared_task
 from celery.signals import before_task_publish
 from celery.signals import task_failure
@@ -16,6 +17,7 @@ from celery.signals import task_prerun
 from celery.signals import task_revoked
 from celery.signals import worker_process_init
 from celery.signals import worker_process_shutdown
+from channels.layers import get_channel_layer
 from django.conf import settings
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import User
@@ -1064,6 +1066,7 @@ TRACKED_TASKS: dict[str, PaperlessTask.TaskType] = {
     "documents.tasks.build_share_link_bundle": PaperlessTask.TaskType.BUILD_SHARE_LINK,
     "documents.bulk_edit.delete": PaperlessTask.TaskType.BULK_DELETE,
     "documents.tasks.apply_ai_suggestions": PaperlessTask.TaskType.APPLY_AI_SUGGESTIONS,
+    "documents.bulk_operations.execute_bulk_operation": PaperlessTask.TaskType.BULK_UPDATE,
 }
 
 _CELERY_STATE_TO_STATUS: dict[str, PaperlessTask.Status] = {
@@ -1081,8 +1084,8 @@ def _extract_input_data(
 
     For consume_file tasks this includes the filename, MIME type, and any
     non-null overrides from the DocumentMetadataOverrides object.  For
-    mail_fetch tasks it captures the account_ids list.  All other task
+    mail_fetch tasks it captures the account_ids list. Bulk operations store
-    types store no input data and return {}.
+    the operation name and document count, without copying the full selection.
     """
     if task_type == PaperlessTask.TaskType.CONSUME_FILE:
         input_doc = task_kwargs.get("input_doc")
@@ -1111,6 +1114,12 @@ def _extract_input_data(
                 data["overrides"] = override_dict
         return data
 
+    if "operation" in task_kwargs and "user_id" in task_kwargs:
+        return {
+            "operation": task_kwargs["operation"],
+            "document_count": len(task_kwargs.get("document_ids", [])),
+        }
+
     if task_type == PaperlessTask.TaskType.MAIL_FETCH:
         account_ids = task_kwargs.get("account_ids")
         if account_ids is not None:
@@ -1147,7 +1156,9 @@ def _extract_owner_id(
     task_type: PaperlessTask.TaskType,
     task_kwargs: dict,
 ) -> int | None:
-    """Return the owner_id from consume_file overrides, or None for all other task types."""
+    """Use the initiator for bulk operations and the document owner for consumption."""
+    if "operation" in task_kwargs:
+        return task_kwargs.get("user_id")
     if task_type != PaperlessTask.TaskType.CONSUME_FILE:
         return None
     overrides = task_kwargs.get("overrides")
@@ -1228,6 +1239,29 @@ def task_prerun_handler(sender=None, task_id=None, task=None, **kwargs) -> None:
         logger.exception("Setting PaperlessTask started failed")
 
 
+def _send_bulk_operation_status(task_id: str) -> None:
+    """Notify the initiator, task records still happen."""
+    try:
+        task = PaperlessTask.objects.filter(task_id=task_id).first()
+        if task is None or task.owner_id is None or "operation" not in task.input_data:
+            return
+        async_to_sync(get_channel_layer().group_send)(
+            f"bulk_operations_{task.owner_id}",
+            {
+                "type": "bulk_operation_status",
+                "data": {
+                    "task_id": task.task_id,
+                    "status": task.status,
+                    "operation": task.input_data["operation"],
+                    "error": (task.result_data or {}).get("error_message", ""),
+                },
+            },
+        )
+    except Exception:
+        # Just a notification failure
+        logger.exception("Sending bulk operation status failed")
+
+
 @task_postrun.connect
 def task_postrun_handler(
     sender=None,
@@ -1285,6 +1319,7 @@ def task_postrun_handler(
                 changed_fields.append("status")
 
         task_instance.save(update_fields=changed_fields)
+        _send_bulk_operation_status(task_id)
     except Exception:  # pragma: no cover
         logger.exception("Updating PaperlessTask failed")
 
@@ -1350,6 +1385,7 @@ def task_failure_handler(
                     date_started - date_created
                 ).total_seconds()
             task_qs.update(**update_fields)
+            _send_bulk_operation_status(task_id)
     except Exception:  # pragma: no cover
         logger.exception("Updating PaperlessTask on failure failed")
 
@@ -1393,6 +1429,7 @@ def task_revoked_handler(
                 "reason": (reason + " Check worker logs for details."),
             },
         )
+        _send_bulk_operation_status(task_id)
     except Exception:  # pragma: no cover
         logger.exception("Updating PaperlessTask on revocation failed")
 

src/documents/tests/test_api_bulk_edit.py

@@ -7,6 +7,7 @@ from django.test import override_settings
 from rest_framework import status
 from rest_framework.test import APITestCase
 
+from documents.bulk_operations import execute_bulk_operation
 from documents.models import Correspondent
 from documents.models import CustomField
 from documents.models import CustomFieldInstance
@@ -29,9 +30,18 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.user = user
         self.client.force_authenticate(user=user)
 
-        patcher = mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+        patcher = mock.patch("documents.bulk_operations.bulk_update_documents")
         self.async_task = patcher.start()
         self.addCleanup(patcher.stop)
+        # Exercise accepted requests and worker execution together in these
+        # operation/validation tests. Queue-only behavior is tested separately.
+        enqueue = mock.patch(
+            "documents.views.execute_bulk_operation.apply_async",
+            side_effect=lambda *, kwargs, headers: self.run_operation(kwargs),
+        )
+        enqueue.start()
+        self.addCleanup(enqueue.stop)
+
         self.c1 = Correspondent.objects.create(name="c1")
         self.c2 = Correspondent.objects.create(name="c2")
         self.dt1 = DocumentType.objects.create(name="dt1")
@@ -61,11 +71,16 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.cf1 = CustomField.objects.create(name="cf1", data_type="string")
         self.cf2 = CustomField.objects.create(name="cf2", data_type="string")
 
+    @staticmethod
+    def run_operation(kwargs):
+        execute_bulk_operation(**kwargs)
+        return mock.Mock(id="test-bulk-operation")
+
     def setup_mock(self, m, method_name, return_value="OK") -> None:
         m.return_value = return_value
         m.__name__ = method_name
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_set_correspondent(self, bulk_update_task_mock) -> None:
         self.assertNotEqual(self.doc1.correspondent, self.c1)
         response = self.client.post(
@@ -84,11 +99,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(self.doc1.correspondent, self.c1)
         bulk_update_task_mock.assert_called_once()
         self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_unset_correspondent(self, bulk_update_task_mock) -> None:
         self.doc1.correspondent = self.c1
         self.doc1.save()
@@ -110,7 +125,7 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.doc1.refresh_from_db()
         self.assertIsNone(self.doc1.correspondent)
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_set_type(self, bulk_update_task_mock) -> None:
         self.assertNotEqual(self.doc1.document_type, self.dt1)
         response = self.client.post(
@@ -129,11 +144,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(self.doc1.document_type, self.dt1)
         bulk_update_task_mock.assert_called_once()
         self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_unset_type(self, bulk_update_task_mock) -> None:
         self.doc1.document_type = self.dt1
         self.doc1.save()
@@ -154,11 +169,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertIsNone(self.doc1.document_type)
         bulk_update_task_mock.assert_called_once()
         self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_add_tag(self, bulk_update_task_mock) -> None:
         self.assertFalse(self.doc1.tags.filter(pk=self.t1.pk).exists())
 
@@ -180,11 +195,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
 
         bulk_update_task_mock.assert_called_once()
         self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
             [self.doc1.pk],
         )
 
-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
     def test_api_remove_tag(self, bulk_update_task_mock) -> None:
         self.doc1.tags.add(self.t1)
 
@@ -223,6 +238,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
         self.assertEqual(kwargs["add_tags"], [self.t1.id])
         self.assertEqual(kwargs["remove_tags"], [self.t2.id])
@@ -277,6 +295,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
         self.assertEqual(kwargs["add_custom_fields"], [self.cf1.id])
         self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id])
@@ -345,6 +366,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
         self.assertEqual(kwargs["add_custom_fields"], {self.cf1.id: "foo"})
         self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id])
@@ -506,6 +530,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -520,6 +547,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -544,6 +574,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(len(kwargs), 0)
 
@@ -558,6 +591,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs, {"remote_ocr": False})
 
@@ -580,6 +616,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs, {"remote_ocr": True})
 
@@ -609,6 +648,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
 
         self.assertListEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
@@ -639,6 +681,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
 
         self.assertListEqual(args[0], [self.doc1.id])
         self.assertEqual(kwargs["storage_path"], None)
@@ -761,6 +806,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -784,6 +832,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc1.id, self.doc3.id, self.doc5.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -820,6 +871,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [first_duplicate.id, second_duplicate.id])
         self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -856,6 +910,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
                 self.assertEqual(response.status_code, status.HTTP_200_OK)
                 get_backend.return_value.search_ids.assert_called_once()
                 args, kwargs = m.call_args
+                kwargs = {
+                    key: value
+                    for key, value in kwargs.items()
+                    if key != "update_callback"
+                }
                 self.assertEqual(args[0], [self.doc2.id])
                 self.assertEqual(kwargs["storage_path"], self.sp1.id)
 
@@ -1232,6 +1291,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
 
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
         self.assertEqual(len(kwargs["set_permissions"]["view"]["users"]), 2)
 
@@ -1644,6 +1706,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
         self.assertEqual(kwargs["degrees"], 90)
         self.assertEqual(kwargs["source_mode"], "latest_version")
@@ -1930,6 +1995,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
         self.assertEqual(kwargs["metadata_document_id"], self.doc3.id)
         self.assertEqual(kwargs["source_mode"], "latest_version")
@@ -2137,6 +2205,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["operations"], [{"page": 1}])
         self.assertEqual(kwargs["source_mode"], "explicit_selection")
@@ -2482,6 +2553,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         m.assert_called_once()
         args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
         self.assertCountEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["password"], "secret")
         self.assertTrue(kwargs["update_document"])

src/documents/tests/test_bulk_operations.py

@@ -0,0 +1,281 @@
+"""Request acceptance, worker execution, and task tracking."""
+
+import uuid
+from unittest import mock
+
+import pytest
+from auditlog.models import LogEntry
+from django.contrib.auth.models import User
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.test import APIClient
+
+from documents.bulk_operations import execute_bulk_operation
+from documents.models import Correspondent
+from documents.models import Document
+from documents.models import PaperlessTask
+from documents.signals.handlers import before_task_publish_handler
+from documents.signals.handlers import task_failure_handler
+from documents.signals.handlers import task_postrun_handler
+from paperless_testing.factories import UserFactory
+
+
+@pytest.fixture
+def bulk_setup(paperless_dirs):
+    user = UserFactory(superuser=True)
+    client = APIClient()
+    client.force_authenticate(user)
+    doc = Document.objects.create(title="Example", checksum="bulk-op-example")
+    correspondent = Correspondent.objects.create(name="Example correspondent")
+    return user, client, doc, correspondent
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "endpoint,operation,parameters",
+    [
+        ("bulk_edit", "set_correspondent", {"correspondent": None}),
+    ],
+)
+def test_request_only_queues(bulk_setup, endpoint, operation, parameters):
+    user, client, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    data = {"documents": [doc.pk], "from_webui": True}
+    if endpoint == "bulk_edit":
+        data.update(method=operation, parameters=parameters)
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = task_id
+        response = client.post(f"/api/documents/{endpoint}/", data, format="json")
+    assert response.status_code == 200
+    assert response.data == {"result": "OK", "task_id": task_id}
+    assert Document.objects.filter(pk=doc.pk).exists()
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": operation,
+        "document_ids": [doc.pk],
+        "parameters": parameters,
+        "user_id": user.pk,
+    }
+    assert enqueue.call_args.kwargs["headers"]["trigger_source"] == "web_ui"
+
+
+@pytest.mark.django_db
+def test_metadata_task_includes_audit_and_followup(bulk_setup, settings):
+    user, _, doc, correspondent = bulk_setup
+    settings.AUDIT_LOG_ENABLED = True
+    with (
+        mock.patch("documents.bulk_operations.bulk_update_documents") as followup,
+        mock.patch(
+            "documents.bulk_edit.bulk_update_documents.apply_async",
+        ) as extra_task,
+    ):
+
+        def check_followup(ids):
+            doc.refresh_from_db()
+            assert doc.correspondent_id == correspondent.pk
+            assert LogEntry.objects.filter(object_pk=str(doc.pk), actor=user).exists()
+            assert ids == [doc.pk]
+
+        followup.side_effect = check_followup
+        result = execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    followup.assert_called_once()
+    extra_task.assert_not_called()
+    assert result["document_count"] == 1
+
+
+@pytest.mark.django_db
+def test_permissions_rechecked_on_worker(bulk_setup):
+    user, _, doc, correspondent = bulk_setup
+    User.objects.filter(pk=user.pk).update(is_superuser=False)
+    with pytest.raises(PermissionDenied):
+        execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+def test_queue_failure_does_not_change_documents(bulk_setup):
+    _, client, doc, correspondent = bulk_setup
+    with mock.patch(
+        "documents.views.execute_bulk_operation.apply_async",
+        side_effect=RuntimeError("broker unavailable"),
+    ):
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": [doc.pk],
+                "method": "set_correspondent",
+                "parameters": {"correspondent": correspondent.pk},
+            },
+            format="json",
+        )
+    assert response.status_code == 400
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("failed", [False, True])
+def test_tracked_outcome_records_initiator_and_result(bulk_setup, failed):
+    user, _, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": "documents.bulk_operations.execute_bulk_operation",
+            "id": task_id,
+        },
+        body=(
+            (),
+            {
+                "operation": "set_permissions",
+                "document_ids": [doc.pk],
+                "user_id": user.pk,
+            },
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id=task_id)
+    assert task.owner_id == user.pk
+    assert task.task_type == PaperlessTask.TaskType.BULK_UPDATE
+    if failed:
+        task_failure_handler(task_id=task_id, exception=ValueError("Could not update"))
+    else:
+        task_postrun_handler(
+            task_id=task_id,
+            state="SUCCESS",
+            retval={"document_count": 1},
+        )
+    task.refresh_from_db()
+    assert task.status == ("failure" if failed else "success")
+    if failed:
+        assert task.result_data["error_message"] == "Could not update"
+    else:
+        assert task.result_data == {"document_count": 1}
+
+
+@pytest.mark.django_db
+def test_large_filtered_selection_is_one_task(bulk_setup):
+    _, client, _, _ = bulk_setup
+    Document.objects.bulk_create(
+        [
+            Document(title="bulk-large-selection", checksum=f"large-{i}")
+            for i in range(10_000)
+        ],
+    )
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = str(uuid.uuid4())
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "all": True,
+                "filters": {"title__icontains": "bulk-large-selection"},
+                "method": "set_correspondent",
+                "parameters": {"correspondent": None},
+            },
+            format="json",
+        )
+    assert response.status_code == 200
+    enqueue.assert_called_once()
+    assert len(enqueue.call_args.kwargs["kwargs"]["document_ids"]) == 10_000
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("failed", [False, True])
+def test_tracked_outcome_notifies_only_initiator(bulk_setup, failed):
+    user, _, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": "documents.bulk_operations.execute_bulk_operation",
+            "id": task_id,
+        },
+        body=(
+            (),
+            {
+                "operation": "set_permissions",
+                "document_ids": [doc.pk],
+                "user_id": user.pk,
+            },
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id=task_id)
+    assert task.owner_id == user.pk
+    assert task.task_type == PaperlessTask.TaskType.BULK_UPDATE
+    with mock.patch("documents.signals.handlers.async_to_sync") as send:
+        if failed:
+            task_failure_handler(
+                task_id=task_id,
+                exception=ValueError("Could not update"),
+            )
+        else:
+            task_postrun_handler(
+                task_id=task_id,
+                state="SUCCESS",
+                retval={"document_count": 1},
+            )
+    group, event = send.return_value.call_args.args
+    assert group == f"bulk_operations_{user.pk}"
+    assert event["data"]["status"] == ("failure" if failed else "success")
+    assert event["data"]["task_id"] == task_id
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_completion_socket_is_private(anyio_backend):
+    from types import SimpleNamespace
+
+    from channels.layers import get_channel_layer
+    from channels.testing import WebsocketCommunicator
+
+    from paperless.consumers import StatusConsumer
+
+    own = WebsocketCommunicator(StatusConsumer.as_asgi(), "/ws/status/")
+    other = WebsocketCommunicator(StatusConsumer.as_asgi(), "/ws/status/")
+    own.scope["user"] = SimpleNamespace(id=1, is_authenticated=True)
+    other.scope["user"] = SimpleNamespace(id=2, is_authenticated=True)
+    assert (await own.connect())[0]
+    assert (await other.connect())[0]
+    try:
+        event = {
+            "type": "bulk_operation_status",
+            "data": {"task_id": "example", "status": "success"},
+        }
+        await get_channel_layer().group_send("bulk_operations_1", event)
+        assert await own.receive_json_from() == event
+        assert await other.receive_nothing(timeout=0.02)
+    finally:
+        await own.disconnect()
+        await other.disconnect()
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("endpoint", ["delete", "bulk_edit"])
+def test_trash_remains_synchronous(bulk_setup, endpoint):
+    _, client, doc, _ = bulk_setup
+    data = {"documents": [doc.pk]}
+    if endpoint == "bulk_edit":
+        data.update(method="delete", parameters={})
+    with (
+        mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue,
+        mock.patch("documents.search.get_backend"),
+    ):
+        response = client.post(
+            f"/api/documents/{endpoint}/",
+            data,
+            format="json",
+            HTTP_ACCEPT="application/json; version=9",
+        )
+    assert response.status_code == 200
+    assert response.data == {"result": "OK"}
+    assert not Document.objects.filter(pk=doc.pk).exists()
+    assert Document.deleted_objects.filter(pk=doc.pk).exists()
+    enqueue.assert_not_called()

src/documents/views.py

@@ -10,6 +10,7 @@ from collections.abc import Callable
 from datetime import datetime
 from datetime import timedelta
 from http import HTTPStatus
+from inspect import signature
 from pathlib import Path
 from time import mktime
 from time import sleep
@@ -104,6 +105,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 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
@@ -172,6 +176,7 @@ from documents.permissions import annotate_document_count_by_ids
 from documents.permissions import annotate_document_count_for_related_queryset
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_permission_grantees
+from documents.permissions import has_document_operation_permissions
 from documents.permissions import has_global_statistics_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
@@ -2958,98 +2963,29 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         "merge_as_versions",
     }
 
-    def _has_document_permissions(
+    def _has_document_permissions(self, **kwargs) -> bool:
-        self,
+        return has_document_operation_permissions(**kwargs)
-        *,
-        user: User,
-        documents: list[int],
-        method,
-        parameters: dict[str, Any],
-    ) -> bool:
-        if user.is_superuser:
-            return True
 
-        root_docs = {
+    def _queue_bulk_operation(self, method, documents, parameters, from_webui):
-            get_root_document(doc)
+        if "update_callback" in parameters:
-            for doc in Document.objects.select_related(
+            raise ValueError("Unsupported bulk operation parameter: update_callback")
-                "owner",
+        signature(method).bind(documents, **parameters)
-                "root_document__owner",
+        task = execute_bulk_operation.apply_async(
-            ).filter(pk__in=documents)
+            kwargs={
-        }
+                "operation": method.__name__,
-        user_is_owner_of_all_documents = all(
+                "document_ids": documents,
-            (doc.owner == user or doc.owner is None) for doc in root_docs
+                "parameters": parameters,
+                "user_id": self.request.user.pk,
+            },
+            headers={
+                "trigger_source": (
+                    PaperlessTask.TriggerSource.WEB_UI
+                    if from_webui
+                    else PaperlessTask.TriggerSource.API_UPLOAD
+                ),
+            },
         )
-
+        return Response({"result": "OK", "task_id": task.id})
-        # check global and object permissions for all documents
-        has_perms = (
-            user.has_perm(
-                "documents.change_document",
-            )
-            and not Document.global_objects.filter(pk__in=documents)
-            .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
-            .exists()
-        )
-
-        # check ownership for methods that change original document
-        if (
-            (
-                has_perms
-                and method
-                in [
-                    bulk_edit.set_permissions,
-                    bulk_edit.delete,
-                    bulk_edit.rotate,
-                    bulk_edit.delete_pages,
-                    bulk_edit.edit_pdf,
-                    bulk_edit.merge_as_versions,
-                    bulk_edit.remove_password,
-                ]
-            )
-            or (
-                method in [bulk_edit.merge, bulk_edit.split]
-                and parameters.get("delete_originals")
-            )
-            or (method == bulk_edit.edit_pdf and parameters.get("update_document"))
-        ):
-            has_perms = has_perms and user_is_owner_of_all_documents
-
-        # check global add permissions for methods that create documents
-        if (
-            has_perms
-            and (
-                method in [bulk_edit.split, bulk_edit.merge]
-                or (
-                    method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
-                    and not parameters.get("update_document")
-                )
-            )
-            and not user.has_perm("documents.add_document")
-        ):
-            has_perms = False
-
-        # check global delete permissions for methods that delete documents
-        if (
-            has_perms
-            and (
-                method == bulk_edit.delete
-                # Sources stop being documents of their own, and removing one
-                # again afterwards needs delete_document
-                or method == bulk_edit.merge_as_versions
-                or (
-                    method in [bulk_edit.merge, bulk_edit.split]
-                    and parameters.get("delete_originals")
-                )
-                or (
-                    method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
-                    and parameters.get("delete_original")
-                    and not parameters.get("update_document")
-                )
-            )
-            and not user.has_perm("documents.delete_document")
-        ):
-            has_perms = False
-
-        return has_perms
 
     def _execute_document_action(
         self,
@@ -3117,6 +3053,7 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
                 name="BulkEditDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(required=False),
                 },
             ),
         },
@@ -3146,37 +3083,7 @@ class BulkEditView(DocumentOperationPermissionMixin):
 
     serializer_class = BulkEditSerializer
 
-    @staticmethod
+    _snapshot_field = staticmethod(snapshot_field)
-    def _snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
-        """
-        Returns each document's current value of field, for the audit log.
-
-        Tags and custom fields are one row per value, so they are gathered
-        into a sorted list of pks per document (empty when there are none).
-        Reading them through Document.values() instead would join those rows
-        and return one arbitrary value per document.
-        """
-        if field == "tags":
-            rows = (
-                Document.tags.through.objects.filter(document_id__in=doc_ids)
-                .order_by("tag_id")
-                .values_list("document_id", "tag_id")
-            )
-        elif field == "custom_fields":
-            rows = (
-                CustomFieldInstance.objects.filter(document_id__in=doc_ids)
-                .order_by("pk")
-                .values_list("document_id", "pk")
-            )
-        else:
-            return dict(
-                Document.objects.filter(pk__in=doc_ids).values_list("pk", field),
-            )
-
-        values: dict[int, list[int]] = {doc_id: [] for doc_id in doc_ids}
-        for doc_id, pk in rows:
-            values[doc_id].append(pk)
-        return values
 
     def post(self, request, *args, **kwargs):
         request_method = request.data.get("method")
@@ -3222,6 +3129,13 @@ 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.MODIFIED_FIELD_BY_METHOD.get(method.__name__, None)
             if settings.AUDIT_LOG_ENABLED and modified_field:
                 old_values = self._snapshot_field(documents, modified_field)

src/paperless/consumers.py

@@ -45,12 +45,21 @@ class StatusConsumer(AsyncWebsocketConsumer):
             await self.close()
             return
         await self.channel_layer.group_add("status_updates", self.channel_name)
+        await self.channel_layer.group_add(
+            f"bulk_operations_{self.scope['user'].id}",
+            self.channel_name,
+        )
         await self.accept()
         self._start_heartbeat()
 
     async def disconnect(self, code: int) -> None:
         await self._stop_heartbeat()
         await self.channel_layer.group_discard("status_updates", self.channel_name)
+        if self._authenticated():
+            await self.channel_layer.group_discard(
+                f"bulk_operations_{self.scope['user'].id}",
+                self.channel_name,
+            )
 
     def _start_heartbeat(self) -> None:
         self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
@@ -84,3 +93,9 @@ class StatusConsumer(AsyncWebsocketConsumer):
             await self.close()
         elif await self._can_view(event["data"]):
             await self.send(json.dumps(event))
+
+    async def bulk_operation_status(self, event: dict) -> None:
+        if not self._authenticated():
+            await self.close()
+        else:
+            await self.send(json.dumps(event))

src/paperless/settings/__init__.py

@@ -691,7 +691,7 @@ logging.config.dictConfig(LOGGING)
 # https://docs.celeryq.dev/en/stable/userguide/configuration.html
 
 # Tasks outside tasks.py must be loaded even when Django skips checks
-CELERY_IMPORTS = ("documents.bulk_edit",)
+CELERY_IMPORTS = ("documents.bulk_edit", "documents.bulk_operations")
 
 CELERY_BROKER_URL = _CELERY_REDIS_URL
 CELERY_RESULT_BACKEND = _CELERY_REDIS_URL