diff --git a/src-ui/src/app/components/admin/tasks/tasks.component.spec.ts b/src-ui/src/app/components/admin/tasks/tasks.component.spec.ts index 20a8de81e8..887e690982 100644 --- a/src-ui/src/app/components/admin/tasks/tasks.component.spec.ts +++ b/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 let component: TasksComponent let fixture: ComponentFixture 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: { @@ -244,6 +262,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() diff --git a/src-ui/src/app/components/admin/tasks/tasks.component.ts b/src-ui/src/app/components/admin/tasks/tasks.component.ts index c354a986ac..ad8793ef85 100644 --- a/src-ui/src/app/components/admin/tasks/tasks.component.ts +++ b/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) @@ -630,6 +644,16 @@ export class TasksComponent return false } + if (!this._filterText.length) { + return true + } + + const query = this._filterText.toLowerCase() + + if (this.filterTargetID === TaskFilterTargetID.TaskId) { + return task.task_id.toLowerCase() === query + } + // Text filtering now handled API-side return true } @@ -691,9 +715,15 @@ export class TasksComponent } if (this._filterText.length) { - params[ - this.filterTargetID === TaskFilterTargetID.Name ? 'name' : 'result' - ] = this._filterText + const parameter = { + [TaskFilterTargetID.Name]: 'name', + [TaskFilterTargetID.Result]: 'result', + [TaskFilterTargetID.TaskId]: 'task_id', + }[this.filterTargetID] + params[parameter] = this._filterText + if (this.filterTargetID === TaskFilterTargetID.TaskId) { + delete params.acknowledged + } } return params diff --git a/src-ui/src/app/components/app-frame/app-frame.component.spec.ts b/src-ui/src/app/components/app-frame/app-frame.component.spec.ts index 2b4c3d63a7..c4bbd2c2d8 100644 --- a/src-ui/src/app/components/app-frame/app-frame.component.spec.ts +++ b/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 let tasksService: TasksService beforeEach(async () => { @@ -182,6 +186,10 @@ describe('AppFrameComponent', () => { savedViewSpy = jest.spyOn(savedViewService, 'reload') maybeRefreshSpy = jest.spyOn(savedViewService, 'maybeRefreshDocumentCounts') + bulkEvents = new Subject() + jest + .spyOn(TestBed.inject(WebsocketStatusService), 'onBulkOperationStatus') + .mockReturnValue(bulkEvents.asObservable()) fixture = TestBed.createComponent(AppFrameComponent) component = fixture.componentInstance @@ -190,6 +198,78 @@ 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(reload).toHaveBeenCalledWith(null, false) + 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: + status === PaperlessTaskStatus.Revoked + ? 'Task expired before execution.' + : 'Could not update', + }) + } + expect(reload).toHaveBeenCalledTimes(2) + expect(error).toHaveBeenCalledWith(expect.any(String), 'Could not update') + expect(error).toHaveBeenCalledWith( + 'Bulk operation revoked.', + 'Task expired before execution.' + ) + expect(success).not.toHaveBeenCalled() + }) + it('should initialize the saved view service', () => { expect(savedViewSpy).toHaveBeenCalled() }) diff --git a/src-ui/src/app/components/app-frame/app-frame.component.ts b/src-ui/src/app/components/app-frame/app-frame.component.ts index 54caff0582..beae9bb4f6 100644 --- a/src-ui/src/app/components/app-frame/app-frame.component.ts +++ b/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,46 @@ 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(null, false) + 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 if (status === PaperlessTaskStatus.Revoked) { + this.toastService.showError($localize`Bulk operation revoked.`, error) + } else { + this.toastService.showError( + $localize`Bulk operation failed. Some changes may already have been applied.`, + error + ) + } + }) + this.lastScrollY = window.scrollY this.detectClassicScrollbars() diff --git a/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.spec.ts b/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.spec.ts index 97be61b070..34540ac687 100644 --- a/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.spec.ts +++ b/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,47 @@ describe('BulkEditorComponent', () => { ) }) + it.each([true, false])( + 'shows queued feedback and gates the task link on permission (%s)', + (canViewTasks) => { + jest + .spyOn(permissionsService, 'currentUserCan') + .mockReturnValue(canViewTasks) + const queued = jest.spyOn(toastService, 'show') + 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) + const notification = queued.mock.calls.at(-1)[0] + expect(notification.content).toBe('Bulk operation queued.') + if (canViewTasks) { + expect(notification.actionName).toBe('View task') + notification.action() + expect(router.navigate).toHaveBeenCalledWith(['/tasks'], { + queryParams: { task_id: 'queued-operation' }, + }) + } else { + expect(notification.action).toBeUndefined() + expect(notification.actionName).toBeUndefined() + } + 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 +1619,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 +1629,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 +1935,18 @@ describe('BulkEditorComponent', () => { expect(toastServiceShowErrorSpy).toHaveBeenCalled() expect(listReloadSpy).not.toHaveBeenCalled() - modal.componentInstance.succeeded.emit() - expect(toastServiceShowInfoSpy).toHaveBeenCalled() - expect(listReloadSpy).toHaveBeenCalled() - httpTestingController.match( + modal.componentInstance.succeeded.emit({ + result: 'OK', + task_id: 'custom-fields-task', + }) + 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', () => { diff --git a/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts b/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts index 067138b3a2..94f742366d 100644 --- a/src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts +++ b/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,33 @@ export class BulkEditorComponent .bulkEdit(overrideSelection ?? this.getSelectionQuery(), method, args) .pipe(first()) .subscribe({ - next: () => this.handleOperationSuccess(modal), + next: ({ task_id }) => this.handleOperationQueued(modal, task_id), error: (error) => this.handleOperationError(modal, error), }) } + private handleOperationQueued(modal: NgbModalRef, taskId: string) { + modal?.close() + this.list.selectNone() + const canViewTasks = this.permissionService.currentUserCan( + PermissionAction.View, + PermissionType.PaperlessTask + ) + this.toastService.show({ + content: $localize`Bulk operation queued.`, + delay: 5000, + ...(canViewTasks + ? { + actionName: $localize`View task`, + action: () => + this.router.navigate(['/tasks'], { + queryParams: { task_id: taskId }, + }), + } + : {}), + }) + } + private executeDocumentAction( modal: NgbModalRef, request: Observable, @@ -1088,14 +1110,9 @@ export class BulkEditorComponent dialog.selection = this.getSelectionQuery() dialog.selectionCount = this.getSelectionSize() - dialog.succeeded.subscribe((result) => { - this.toastService.showSuccess($localize`Custom fields updated.`) - this.list.reload() - this.list.reduceSelectionToFilter() - this.list.selected.forEach((id) => { - this.openDocumentService.refreshDocument(id) - }) - }) + dialog.succeeded.subscribe(({ task_id }) => + this.handleOperationQueued(null, task_id) + ) dialog.failed.subscribe((error) => { this.toastService.showError( $localize`Error updating custom fields.`, diff --git a/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.spec.ts b/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.spec.ts index 01865e5311..064961058d 100644 --- a/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.spec.ts +++ b/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] diff --git a/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.ts b/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.ts index e4eda79c89..67ce981417 100644 --- a/src-ui/src/app/components/document-list/bulk-editor/custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component.ts +++ b/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() @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) }, }) diff --git a/src-ui/src/app/data/bulk-operation.ts b/src-ui/src/app/data/bulk-operation.ts new file mode 100644 index 0000000000..4c569a86c5 --- /dev/null +++ b/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 +} diff --git a/src-ui/src/app/services/document-list-view.service.spec.ts b/src-ui/src/app/services/document-list-view.service.spec.ts index d5ef7e7b4b..aa9e80b55f 100644 --- a/src-ui/src/app/services/document-list-view.service.spec.ts +++ b/src-ui/src/app/services/document-list-view.service.spec.ts @@ -124,6 +124,42 @@ describe('DocumentListViewService', () => { }) }) + it.each([false, true])( + 'preserves the current route on a background refresh (saved view: %s)', + (savedView) => { + if (savedView) documentListViewService.activateSavedView(view) + const navigate = jest.mocked(router.navigate) + navigate.mockClear() + documentListViewService.reload(null, false) + httpTestingController + .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`) + .flush(full_results) + expect(documentListViewService.documents).toEqual(full_results.results) + expect(navigate).not.toHaveBeenCalled() + } + ) + + it('preserves the route when a background refresh falls back to the first page', () => { + jest + .spyOn(documentListViewService, 'saveDocumentListView') + .mockImplementation(() => {}) + documentListViewService.currentPage = 2 + httpTestingController + .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`) + .flush(full_results) + const navigate = jest.mocked(router.navigate) + navigate.mockClear() + documentListViewService.reload(null, false) + httpTestingController + .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`) + .flush('Not found', { status: 404, statusText: 'Not found' }) + httpTestingController + .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`) + .flush(full_results) + expect(documentListViewService.currentPage).toBe(1) + expect(navigate).not.toHaveBeenCalled() + }) + it('should reload the list', () => { expect(documentListViewService.currentPage).toEqual(1) documentListViewService.reload() diff --git a/src-ui/src/app/services/document-list-view.service.ts b/src-ui/src/app/services/document-list-view.service.ts index 5a3a60c5b0..1947f62c98 100644 --- a/src-ui/src/app/services/document-list-view.service.ts +++ b/src-ui/src/app/services/document-list-view.service.ts @@ -380,7 +380,7 @@ export class DocumentListViewService { queryParams: paramsFromViewState(activeListViewState), replaceUrl: !this.router.routerState.snapshot.url.includes('?'), // in case navigating from params-less /documents }) - } else if (this._activeSavedViewId) { + } else if (updateQueryParams && this._activeSavedViewId) { this.router.navigate([], { queryParams: paramsFromViewState(activeListViewState, true), queryParamsHandling: 'merge', @@ -399,7 +399,7 @@ export class DocumentListViewService { // this happens when applying a filter: the current page might not be available anymore due to the reduced result set. activeListViewState.currentPage = 1 this.markChanged() - this.reload() + this.reload(onFinish, updateQueryParams) } else if ( activeListViewState.sortField.indexOf('custom_field') === 0 && !this.settings diff --git a/src-ui/src/app/services/rest/document.service.spec.ts b/src-ui/src/app/services/rest/document.service.spec.ts index 4250d8e84f..eaa5d60f49 100644 --- a/src-ui/src/app/services/rest/document.service.spec.ts +++ b/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], diff --git a/src-ui/src/app/services/rest/document.service.ts b/src-ui/src/app/services/rest/document.service.ts index 3272d9695e..ef7a171254 100644 --- a/src-ui/src/app/services/rest/document.service.ts +++ b/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 { method: DocumentBulkEditMethod, args: any ) { - return this.http.post(this.getResourceUrl(null, 'bulk_edit'), { - ...selection, - method: method, - parameters: args, - }) + return this.http.post( + this.getResourceUrl(null, 'bulk_edit'), + { + ...selection, + method: method, + from_webui: true, + parameters: args, + } + ) } deleteDocuments(selection: DocumentSelectionQuery) { diff --git a/src-ui/src/app/services/websocket-status.service.spec.ts b/src-ui/src/app/services/websocket-status.service.spec.ts index 7e1bfd424e..9366eee994 100644 --- a/src-ui/src/app/services/websocket-status.service.spec.ts +++ b/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 diff --git a/src-ui/src/app/services/websocket-status.service.ts b/src-ui/src/app/services/websocket-status.service.ts index d3f90e6a50..e26b849fe6 100644 --- a/src-ui/src/app/services/websocket-status.service.ts +++ b/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() private readonly documentUpdatedSubject = new Subject() + private readonly bulkOperationStatusSubject = + new Subject() + + onBulkOperationStatus() { + return this.bulkOperationStatusSubject.asObservable() + } + private readonly connectionStatusSubject = new Subject() 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 diff --git a/src/documents/bulk_edit.py b/src/documents/bulk_edit.py index 6a91f4319c..08022df584 100644 --- a/src/documents/bulk_edit.py +++ b/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( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) 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( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) 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( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) 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( - kwargs={"document_ids": list(affected_docs)}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(list(affected_docs), update_callback) 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( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) 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)) @@ -293,10 +316,7 @@ def modify_tags( ) if affected_docs: - bulk_update_documents.apply_async( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) return "OK" @@ -305,6 +325,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)) @@ -367,10 +389,7 @@ def modify_custom_fields( field_id__in=remove_custom_fields, ).hard_delete() - bulk_update_documents.apply_async( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) return "OK" @@ -451,6 +470,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") @@ -468,10 +488,7 @@ def set_permissions( merge=merge, ) - bulk_update_documents.apply_async( - kwargs={"document_ids": affected_docs}, - headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM}, - ) + _finish_metadata_edit(affected_docs, update_callback) return "OK" diff --git a/src/documents/bulk_operations.py b/src/documents/bulk_operations.py new file mode 100644 index 0000000000..eee7f8b077 --- /dev/null +++ b/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)} diff --git a/src/documents/permissions.py b/src/documents/permissions.py index 9a48e3a809..196b145c5c 100644 --- a/src/documents/permissions.py +++ b/src/documents/permissions.py @@ -739,3 +739,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 diff --git a/src/documents/signals/handlers.py b/src/documents/signals/handlers.py index 9c20e97f21..9b726fdfaf 100644 --- a/src/documents/signals/handlers.py +++ b/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 @@ -1051,6 +1053,8 @@ def run_workflows( # Task tracking -- Celery signal handlers # --------------------------------------------------------------------------- +BULK_OPERATION_TASK_NAME = "documents.bulk_operations.execute_bulk_operation" + TRACKED_TASKS: dict[str, PaperlessTask.TaskType] = { "documents.tasks.consume_file": PaperlessTask.TaskType.CONSUME_FILE, "documents.tasks.train_classifier": PaperlessTask.TaskType.TRAIN_CLASSIFIER, @@ -1064,6 +1068,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, + BULK_OPERATION_TASK_NAME: PaperlessTask.TaskType.BULK_UPDATE, } _CELERY_STATE_TO_STATUS: dict[str, PaperlessTask.Status] = { @@ -1081,8 +1086,7 @@ 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 - types store no input data and return {}. + mail_fetch tasks it captures the account_ids list. """ if task_type == PaperlessTask.TaskType.CONSUME_FILE: input_doc = task_kwargs.get("input_doc") @@ -1147,7 +1151,7 @@ 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.""" + """Return the document owner for consumption tasks.""" if task_type != PaperlessTask.TaskType.CONSUME_FILE: return None overrides = task_kwargs.get("overrides") @@ -1186,9 +1190,16 @@ def before_task_publish_handler( _, task_kwargs, _ = body task_id = headers["id"] - input_data = _extract_input_data(task_type, task_kwargs) + if task_name == BULK_OPERATION_TASK_NAME: + input_data = { + "operation": task_kwargs["operation"], + "document_count": len(task_kwargs["document_ids"]), + } + owner_id = task_kwargs["user_id"] + else: + input_data = _extract_input_data(task_type, task_kwargs) + owner_id = _extract_owner_id(task_type, task_kwargs) trigger_source = _determine_trigger_source(headers) - owner_id = _extract_owner_id(task_type, task_kwargs) # A retried task is republished with the same task_id, so this fires # again for it; get_or_create keeps the original PENDING record @@ -1228,6 +1239,30 @@ 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: PaperlessTask | None) -> None: + """Send the recorded outcome to the initiator.""" + try: + if task is None or task.owner_id is None: + return + result_data = task.result_data or {} + 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": result_data.get("error_message") + or result_data.get("reason", ""), + }, + }, + ) + except Exception: + # Just a notification failure + logger.exception("Sending bulk operation status failed") + + @task_postrun.connect def task_postrun_handler( sender=None, @@ -1285,6 +1320,8 @@ def task_postrun_handler( changed_fields.append("status") task_instance.save(update_fields=changed_fields) + if task and task.name == BULK_OPERATION_TASK_NAME: + _send_bulk_operation_status(task_instance) except Exception: # pragma: no cover logger.exception("Updating PaperlessTask failed") @@ -1350,6 +1387,8 @@ def task_failure_handler( date_started - date_created ).total_seconds() task_qs.update(**update_fields) + if sender and sender.name == BULK_OPERATION_TASK_NAME: + _send_bulk_operation_status(task_qs.first()) except Exception: # pragma: no cover logger.exception("Updating PaperlessTask on failure failed") @@ -1386,13 +1425,16 @@ def task_revoked_handler( reason = "Task expired before execution." elif terminated: reason = "Task terminated during execution." - PaperlessTask.objects.filter(task_id=task_id).update( + task_qs = PaperlessTask.objects.filter(task_id=task_id) + task_qs.update( status=PaperlessTask.Status.REVOKED, date_done=timezone.now(), result_data={ "reason": (reason + " Check worker logs for details."), }, ) + if sender and sender.name == BULK_OPERATION_TASK_NAME: + _send_bulk_operation_status(task_qs.first()) except Exception: # pragma: no cover logger.exception("Updating PaperlessTask on revocation failed") diff --git a/src/documents/tests/test_api_bulk_edit.py b/src/documents/tests/test_api_bulk_edit.py index 7f0cec816a..5f0272aeb8 100644 --- a/src/documents/tests/test_api_bulk_edit.py +++ b/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"]) diff --git a/src/documents/tests/test_api_search_query_length.py b/src/documents/tests/test_api_search_query_length.py index 77dca7bdb0..9e6b358c12 100644 --- a/src/documents/tests/test_api_search_query_length.py +++ b/src/documents/tests/test_api_search_query_length.py @@ -149,10 +149,10 @@ class TestPostSelectionPathsEnforceTheCap: assert str(_MAX_QUERY_LENGTH + 1) in message parse_spy.assert_not_called() - @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async") + @mock.patch("documents.views.execute_bulk_operation.apply_async") def test_bulk_edit_query_at_exactly_the_cap_is_accepted( self, - bulk_update_task_mock: mock.MagicMock, + enqueue_mock: mock.MagicMock, admin_client: APIClient, searchable_document: Document, ) -> None: @@ -163,12 +163,12 @@ class TestPostSelectionPathsEnforceTheCap: - Its `filters.query` is exactly `_MAX_QUERY_LENGTH` characters long THEN: - - The cap check accepts it and the request reaches the real - bulk-edit method (its Celery dispatch is mocked out here, - same as every other bulk-edit test, since nothing here is - testing that method itself) + - The cap check accepts it and queues the metadata operation. + Mock the executor dispatch so this exercises query validation + without requiring a broker. """ query = "a" * _MAX_QUERY_LENGTH + enqueue_mock.return_value.id = "query-cap-task" response = admin_client.post( "/api/documents/bulk_edit/", @@ -183,6 +183,11 @@ class TestPostSelectionPathsEnforceTheCap: ) assert response.status_code == status.HTTP_200_OK + assert response.data == {"result": "OK", "task_id": "query-cap-task"} + enqueue_mock.assert_called_once() + assert ( + enqueue_mock.call_args.kwargs["kwargs"]["operation"] == "set_document_type" + ) def test_bulk_download_query_one_over_the_cap_is_a_400( self, diff --git a/src/documents/tests/test_bulk_operations.py b/src/documents/tests/test_bulk_operations.py new file mode 100644 index 0000000000..88f12edb22 --- /dev/null +++ b/src/documents/tests/test_bulk_operations.py @@ -0,0 +1,307 @@ +"""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 documents.signals.handlers import task_revoked_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( + ("outcome", "expected_status", "expected_error"), + [ + ("success", "success", ""), + ("failure", "failure", "Could not update"), + ( + "expired", + "revoked", + "Task expired before execution. Check worker logs for details.", + ), + ], +) +def test_tracked_outcome_notifies_only_initiator( + bulk_setup, + outcome, + expected_status, + expected_error, +): + user, _, doc, _ = bulk_setup + task_id = str(uuid.uuid4()) + before_task_publish_handler( + headers={ + "task": execute_bulk_operation.name, + "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 outcome == "expired": + task_revoked_handler( + sender=execute_bulk_operation, + request=mock.Mock(id=task_id), + expired=True, + ) + elif outcome == "failure": + task_failure_handler( + sender=execute_bulk_operation, + task_id=task_id, + exception=ValueError("Could not update"), + ) + else: + task_postrun_handler( + task=execute_bulk_operation, + 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"] == expected_status + assert event["data"]["error"] == expected_error + 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() diff --git a/src/documents/views.py b/src/documents/views.py index 1c1e90065f..01db0a1789 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -11,6 +11,7 @@ from collections.abc import Callable from datetime import datetime from datetime import timedelta from http import HTTPStatus +from inspect import signature from itertools import islice from pathlib import Path from time import mktime @@ -106,6 +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 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 @@ -173,6 +177,7 @@ from documents.permissions import ViewDocumentsPermissions 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 @@ -2982,98 +2987,26 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): "merge_as_versions", } - def _has_document_permissions( - self, - *, - user: User, - documents: list[int], - method, - parameters: dict[str, Any], - ) -> bool: - 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 + 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) + task = execute_bulk_operation.apply_async( + kwargs={ + "operation": method.__name__, + "document_ids": documents, + "parameters": parameters, + "user_id": self.request.user.pk, + }, + headers={ + "trigger_source": ( + PaperlessTask.TriggerSource.WEB_UI + if from_webui + else PaperlessTask.TriggerSource.API_UPLOAD + ), + }, ) - - # 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 + return Response({"result": "OK", "task_id": task.id}) def _execute_document_action( self, @@ -3110,7 +3043,7 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): else PaperlessTask.TriggerSource.API_UPLOAD ) - if not self._has_document_permissions( + if not has_document_operation_permissions( user=user, documents=documents, method=method, @@ -3141,67 +3074,20 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): name="BulkEditDocumentsResult", fields={ "result": serializers.CharField(), + "task_id": serializers.CharField(required=False), }, ), }, ), ) class BulkEditView(DocumentOperationPermissionMixin): - MODIFIED_FIELD_BY_METHOD = { - "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, + LEGACY_MODIFIED_FIELDS = { "delete": "deleted_at", - # These operations create new documents/versions no longer altering - # fields on the selected document in place - "rotate": None, - "delete_pages": None, - "split": None, - "merge": None, - "edit_pdf": None, "reprocess": "checksum", - "remove_password": None, } serializer_class = BulkEditSerializer - @staticmethod - 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") api_version = int(request.version or settings.REST_FRAMEWORK["DEFAULT_VERSION"]) @@ -3237,7 +3123,7 @@ class BulkEditView(DocumentOperationPermissionMixin): if from_webui else PaperlessTask.TriggerSource.API_UPLOAD ) - if not self._has_document_permissions( + if not has_document_operation_permissions( user=user, documents=documents, method=method, @@ -3246,14 +3132,21 @@ class BulkEditView(DocumentOperationPermissionMixin): return HttpResponseForbidden("Insufficient permissions") try: - modified_field = self.MODIFIED_FIELD_BY_METHOD.get(method.__name__, None) + 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 = self._snapshot_field(documents, modified_field) + old_values = snapshot_field(documents, modified_field) result = method(documents, **parameters) if settings.AUDIT_LOG_ENABLED and modified_field: - new_values = self._snapshot_field(documents, modified_field) + new_values = snapshot_field(documents, modified_field) for doc in Document.objects.filter(pk__in=documents): LogEntry.objects.log_create( instance=doc, diff --git a/src/paperless/consumers.py b/src/paperless/consumers.py index 5f9a8891ed..63c78c5b2b 100644 --- a/src/paperless/consumers.py +++ b/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)) diff --git a/src/paperless/settings/__init__.py b/src/paperless/settings/__init__.py index 4a53416771..5b0b62ced8 100644 --- a/src/paperless/settings/__init__.py +++ b/src/paperless/settings/__init__.py @@ -701,7 +701,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