From dce9ff0eb95d3c3fd40187d36da6aa919c641a2c Mon Sep 17 00:00:00 2001 From: Sandra Zollner Date: Wed, 26 Aug 2026 09:17:25 +0000 Subject: [PATCH] refactor(core): Unify the review auto-close into one query and close transition (no-changelog) (#37056) Co-authored-by: Cursor --- .../api-types/src/workflow-review-activity.ts | 11 +- ...workflow-review-request.repository.test.ts | 77 +++++ .../workflow-review-activity.repository.ts | 3 +- .../workflow-review-request.repository.ts | 93 +++--- .../workflow-review-lifecycle.service.test.ts | 100 ++++--- .../workflow-review-lifecycle.service.ts | 283 ++++++++---------- 6 files changed, 299 insertions(+), 268 deletions(-) diff --git a/packages/@n8n/api-types/src/workflow-review-activity.ts b/packages/@n8n/api-types/src/workflow-review-activity.ts index 501ccb01a16..bef8b4ad0ea 100644 --- a/packages/@n8n/api-types/src/workflow-review-activity.ts +++ b/packages/@n8n/api-types/src/workflow-review-activity.ts @@ -3,6 +3,11 @@ import { z } from 'zod'; import type { Iso8601DateTimeString } from './datetime'; import type { WorkflowReviewEligibleReviewer } from './workflow-review-eligible-reviewer'; +export type WorkflowReviewWorkflowCauseActivityType = + | 'workflow.archived' + | 'workflow.deleted' + | 'workflow.moved'; + /** * Feed entry kinds, named `.`. Constrained only here and at the write sites — the * database intentionally has no CHECK on `type`, so growing this union needs no migration. @@ -16,9 +21,7 @@ export type WorkflowReviewActivityType = | 'review.changes_requested' | 'review.version_updated' | 'review.approved' - | 'workflow.archived' - | 'workflow.deleted' - | 'workflow.moved' + | WorkflowReviewWorkflowCauseActivityType | 'workflow.published' /** Closed without an approval. An approval writes `review.approved` instead, never both. */ | 'review.closed'; @@ -146,7 +149,7 @@ export type WorkflowReviewActivityEntry = data: WorkflowReviewClosedActivityData | null; }) | (WorkflowReviewActivityBase & { - type: 'workflow.archived' | 'workflow.deleted' | 'workflow.moved'; + type: WorkflowReviewWorkflowCauseActivityType; data: WorkflowReviewWorkflowCauseActivityData | null; }) | (WorkflowReviewActivityBase & { diff --git a/packages/@n8n/db/src/repositories/__tests__/workflow-review-request.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/workflow-review-request.repository.test.ts index 07a14acc696..f9f5cb3362d 100644 --- a/packages/@n8n/db/src/repositories/__tests__/workflow-review-request.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/workflow-review-request.repository.test.ts @@ -371,6 +371,83 @@ describe('WorkflowReviewRequestRepository', () => { }); }); + describe('findUnreviewableOpenRequestIds', () => { + let queryBuilder: Mocked>; + + beforeEach(() => { + queryBuilder = mock>(); + queryBuilder.select.mockReturnThis(); + queryBuilder.addSelect.mockReturnThis(); + queryBuilder.leftJoin.mockReturnThis(); + queryBuilder.where.mockReturnThis(); + queryBuilder.andWhere.mockReturnThis(); + queryBuilder.getRawMany.mockResolvedValue([]); + (entityManager.createQueryBuilder as Mock).mockReturnValue(queryBuilder); + }); + + it('returns a request only when none of its linked workflows is reviewable', async () => { + queryBuilder.getRawMany.mockResolvedValue([ + // A missing owner row is broken data, not evidence that the workflow moved. + { + requestId: 'req-no-owner', + requestProjectId: 'proj-1', + linkedWorkflowId: 'wf-1', + isArchived: false, + owningProjectId: null, + }, + { + requestId: 'req-orphan', + requestProjectId: 'proj-1', + linkedWorkflowId: null, + isArchived: null, + owningProjectId: null, + }, + { + requestId: 'req-mixed', + requestProjectId: 'proj-1', + linkedWorkflowId: 'wf-2', + isArchived: true, + owningProjectId: 'proj-1', + }, + { + requestId: 'req-mixed', + requestProjectId: 'proj-1', + linkedWorkflowId: 'wf-3', + isArchived: false, + owningProjectId: 'proj-1', + }, + ]); + + expect(await repo.findUnreviewableOpenRequestIds({})).toEqual(['req-orphan']); + expect(queryBuilder.andWhere).not.toHaveBeenCalled(); + + await repo.findUnreviewableOpenRequestIds({}, ['req-1', 'req-2']); + expect(queryBuilder.andWhere).toHaveBeenCalledWith('review.id IN (:...candidateRequestIds)', { + candidateRequestIds: ['req-1', 'req-2'], + }); + + (entityManager.createQueryBuilder as Mock).mockClear(); + expect(await repo.findUnreviewableOpenRequestIds({}, [])).toEqual([]); + expect(entityManager.createQueryBuilder).not.toHaveBeenCalled(); + }); + }); + + describe('closeRequests', () => { + it('bulk-closes the given requests, clearing the closing user and bumping updatedAt', async () => { + await repo.closeRequests(['req-1', 'req-2'], {}); + + expect(entityManager.update).toHaveBeenCalledWith(WorkflowReviewRequest, ['req-1', 'req-2'], { + state: 'closed', + closedById: null, + updatedAt: expect.any(Date), + }); + + entityManager.update.mockClear(); + await repo.closeRequests([], {}); + expect(entityManager.update).not.toHaveBeenCalled(); + }); + }); + describe('findById', () => { it("reads through the context's transaction manager", async () => { const transactionManager = mock(); diff --git a/packages/@n8n/db/src/repositories/workflow-review-activity.repository.ts b/packages/@n8n/db/src/repositories/workflow-review-activity.repository.ts index 10bdf2e25ee..67c44134d0e 100644 --- a/packages/@n8n/db/src/repositories/workflow-review-activity.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-review-activity.repository.ts @@ -4,6 +4,7 @@ import type { WorkflowReviewOpenedActivityData, WorkflowReviewVersionUpdatedActivityData, WorkflowReviewWorkflowCauseActivityData, + WorkflowReviewWorkflowCauseActivityType, WorkflowReviewWorkflowPublishedActivityData, } from '@n8n/api-types'; import { Service } from '@n8n/di'; @@ -35,7 +36,7 @@ export type WorkflowReviewActivityPayload = | { type: 'review.version_updated'; data: WorkflowReviewVersionUpdatedActivityData } | { type: 'review.closed'; data: WorkflowReviewClosedActivityData } | { - type: 'workflow.archived' | 'workflow.deleted' | 'workflow.moved'; + type: WorkflowReviewWorkflowCauseActivityType; data: WorkflowReviewWorkflowCauseActivityData; } | { type: 'workflow.published'; data: WorkflowReviewWorkflowPublishedActivityData }; diff --git a/packages/@n8n/db/src/repositories/workflow-review-request.repository.ts b/packages/@n8n/db/src/repositories/workflow-review-request.repository.ts index 0a38851a8b9..ef2ea0db23a 100644 --- a/packages/@n8n/db/src/repositories/workflow-review-request.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-review-request.repository.ts @@ -173,24 +173,29 @@ export class WorkflowReviewRequestRepository extends BaseRepository { - const openState: WorkflowReviewRequestState = 'open'; - const closedState: WorkflowReviewRequestState = 'closed'; - const ownerRole: WorkflowSharingRole = 'workflow:owner'; - const manager = this.managerFor(ctx); + async findUnreviewableOpenRequestIds( + ctx: OperationContext, + candidateRequestIds?: string[], + ): Promise { + if (candidateRequestIds?.length === 0) return []; - const rows = await manager + const openState: WorkflowReviewRequestState = 'open'; + const ownerRole: WorkflowSharingRole = 'workflow:owner'; + + const qb = this.managerFor(ctx) .createQueryBuilder(WorkflowReviewRequest, 'review') .select('review.id', 'requestId') .addSelect('review.projectId', 'requestProjectId') @@ -207,11 +212,15 @@ export class WorkflowReviewRequestRepository extends BaseRepository(); + .where('review.state = :openState', { openState }); - // Same close policy as the per-mutation path: one reviewable workflow keeps the - // request open, however many of its siblings are gone. + if (candidateRequestIds !== undefined) { + qb.andWhere('review.id IN (:...candidateRequestIds)', { candidateRequestIds }); + } + + const rows = await qb.getRawMany(); + + // One reviewable workflow keeps the request open, however many of its siblings are gone. const closableRequestIds = new Set(); const requestIdsWithReviewableWorkflow = new Set(); for (const row of rows) { @@ -225,54 +234,26 @@ export class WorkflowReviewRequestRepository extends BaseRepository { - const ownerRole: WorkflowSharingRole = 'workflow:owner'; + async closeRequests(requestIds: string[], ctx: OperationContext): Promise { + if (requestIds.length === 0) return; - const qb = this.managerFor(ctx) - .createQueryBuilder(WorkflowReviewRequest, 'review') - .select('1') - .innerJoin(WorkflowReviewRequestWorkflow, 'link', 'link.workflowReviewRequestId = review.id') - .innerJoin(WorkflowEntity, 'workflow', 'workflow.id = link.workflowId') - // Left join, like the sweep: a workflow with no owner row is a broken row, not - // a move, and {@link isReviewable} keeps it reviewable — dropping it here would - // let a targeted close take a request the sweep would leave open. - .leftJoin( - SharedWorkflow, - 'shared', - 'shared.workflowId = link.workflowId AND shared.role = :ownerRole', - { ownerRole }, - ) - .where('review.id = :requestId', { requestId }) - .andWhere('workflow.isArchived = :isArchived', { isArchived: false }) - .andWhere('(shared.projectId IS NULL OR shared.projectId = review.projectId)') - .limit(1); - - if (excludedWorkflowIds.length > 0) { - qb.andWhere('link.workflowId NOT IN (:...excludedWorkflowIds)', { excludedWorkflowIds }); - } - - return (await qb.getRawOne()) !== undefined; + const closedState: WorkflowReviewRequestState = 'closed'; + await this.managerFor(ctx).update(WorkflowReviewRequest, requestIds, { + state: closedState, + closedById: null, + updatedAt: new Date(), + }); } async findById(id: string, ctx: OperationContext): Promise { diff --git a/packages/cli/src/modules/workflow-reviews.ee/__tests__/workflow-review-lifecycle.service.test.ts b/packages/cli/src/modules/workflow-reviews.ee/__tests__/workflow-review-lifecycle.service.test.ts index 45b559dca57..031e4b3d988 100644 --- a/packages/cli/src/modules/workflow-reviews.ee/__tests__/workflow-review-lifecycle.service.test.ts +++ b/packages/cli/src/modules/workflow-reviews.ee/__tests__/workflow-review-lifecycle.service.test.ts @@ -52,17 +52,17 @@ describe('WorkflowReviewLifecycleService', () => { eventService, ); dbLockService.withLockContext.mockImplementation(async (_id, fn) => await fn(ctx)); - requestRepository.saveRequest.mockImplementation(async (request) => request); - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue([]); - // Single-workflow default: nothing reviewable remains, so the policy closes. - requestRepository.hasReviewableWorkflowOutside.mockResolvedValue(false); + requestRepository.closeRequests.mockResolvedValue(undefined); + requestRepository.findUnreviewableOpenRequestIds.mockImplementation( + async (_ctx, candidateRequestIds) => candidateRequestIds ?? [], + ); activityRepository.createActivity.mockResolvedValue(mock()); collaborationService.broadcastWorkflowReviewStateChanged.mockResolvedValue(undefined); }); describe('archive', () => { it('records the cause entry and the close entry together, in the lock transaction', async () => { - const request = openRequest({ decision: 'changes_requested', updatedById: 'user-2' }); + const request = openRequest(); requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([ { request, links: [{ workflowId: 'wf-1', workflowVersionId: 'wfv-1' }] }, ]); @@ -92,11 +92,9 @@ describe('WorkflowReviewLifecycleService', () => { }, ctx, ); - expect(requestRepository.saveRequest).toHaveBeenCalledExactlyOnceWith(request, ctx); - expect(request.state).toBe('closed'); - expect(request.decision).toBe('changes_requested'); - expect(request.closedById).toBeNull(); - expect(request.updatedById).toBe('user-2'); + // The close policy is evaluated for the linked request, then it is bulk-closed by id. + expect(requestRepository.findUnreviewableOpenRequestIds).toHaveBeenCalledWith(ctx, ['req-1']); + expect(requestRepository.closeRequests).toHaveBeenCalledWith(['req-1'], ctx); expect( collaborationService.broadcastWorkflowReviewStateChanged, ).toHaveBeenCalledExactlyOnceWith('wf-1'); @@ -131,26 +129,21 @@ describe('WorkflowReviewLifecycleService', () => { }); it('leaves the request open while a reviewable workflow remains outside the affected set', async () => { - const request = openRequest(); requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([ - { request, links: [{ workflowId: 'wf-1', workflowVersionId: 'wfv-1' }] }, + { request: openRequest(), links: [{ workflowId: 'wf-1', workflowVersionId: 'wfv-1' }] }, ]); - requestRepository.hasReviewableWorkflowOutside.mockResolvedValue(true); + // The request still covers something reviewable, so the policy closes nothing. + requestRepository.findUnreviewableOpenRequestIds.mockResolvedValue([]); await service.afterWorkflowArchived('wf-1', 'user-9'); - expect(requestRepository.hasReviewableWorkflowOutside).toHaveBeenCalledWith( - 'req-1', - ['wf-1'], - ctx, - ); + expect(requestRepository.findUnreviewableOpenRequestIds).toHaveBeenCalledWith(ctx, ['req-1']); // The cause entry is still recorded; only the close is withheld. expect(activityRepository.createActivity).toHaveBeenCalledExactlyOnceWith( expect.objectContaining({ type: 'workflow.archived' }), ctx, ); - expect(requestRepository.saveRequest).not.toHaveBeenCalled(); - expect(request.state).toBe('open'); + expect(requestRepository.closeRequests).toHaveBeenCalledWith([], ctx); expect(eventService.emit).not.toHaveBeenCalled(); }); @@ -160,7 +153,6 @@ describe('WorkflowReviewLifecycleService', () => { await service.afterWorkflowArchived('wf-1', 'user-9'); expect(activityRepository.createActivity).not.toHaveBeenCalled(); - expect(requestRepository.saveRequest).not.toHaveBeenCalled(); expect(collaborationService.broadcastWorkflowReviewStateChanged).not.toHaveBeenCalled(); expect(eventService.emit).not.toHaveBeenCalled(); expect(logger.error).not.toHaveBeenCalled(); @@ -181,7 +173,10 @@ describe('WorkflowReviewLifecycleService', () => { requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([ { request: openRequest(), links: [{ workflowId: 'wf-1', workflowVersionId: 'wfv-1' }] }, ]); - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue(['req-9']); + // Sweep (no candidate ids) strands req-9 too. + requestRepository.findUnreviewableOpenRequestIds.mockImplementation( + async (_ctx, ids) => ids ?? ['req-9'], + ); eventService.emit.mockImplementation(() => { throw new Error('listener down'); }); @@ -189,7 +184,8 @@ describe('WorkflowReviewLifecycleService', () => { await expect(service.afterWorkflowArchived('wf-1', 'user-9')).resolves.toBeUndefined(); expect(logger.error).toHaveBeenCalled(); - expect(requestRepository.closeUnreviewableOpenRequests).toHaveBeenCalledExactlyOnceWith(ctx); + // The sweep still closed what the mutation stranded, after the targeted close. + expect(requestRepository.closeRequests).toHaveBeenCalledWith(['req-9'], ctx); }); }); @@ -223,8 +219,8 @@ describe('WorkflowReviewLifecycleService', () => { }), ctx, ); - expect(first.state).toBe('closed'); - expect(second.state).toBe('closed'); + // Both requests are evaluated together and closed in one bulk update. + expect(requestRepository.closeRequests).toHaveBeenCalledWith(['req-1', 'req-2'], ctx); expect(collaborationService.broadcastWorkflowReviewStateChanged).toHaveBeenCalledTimes(3); expect(collaborationService.broadcastWorkflowReviewStateChanged).toHaveBeenCalledWith('wf-1'); expect(collaborationService.broadcastWorkflowReviewStateChanged).toHaveBeenCalledWith('wf-2'); @@ -263,7 +259,7 @@ describe('WorkflowReviewLifecycleService', () => { expect(requestRepository.findOpenRequestsForWorkflows).toHaveBeenCalledWith(['wf-1'], {}); expect(activityRepository.createActivity).not.toHaveBeenCalled(); - expect(requestRepository.saveRequest).not.toHaveBeenCalled(); + expect(requestRepository.closeRequests).not.toHaveBeenCalled(); expect(dbLockService.withLockContext).not.toHaveBeenCalled(); }); @@ -300,7 +296,7 @@ describe('WorkflowReviewLifecycleService', () => { expect.objectContaining({ type: 'review.closed' }), ctx, ); - expect(request.state).toBe('closed'); + expect(requestRepository.closeRequests).toHaveBeenCalledWith(['req-1'], ctx); expect( collaborationService.broadcastWorkflowReviewStateChanged, ).toHaveBeenCalledExactlyOnceWith('wf-1'); @@ -335,13 +331,8 @@ describe('WorkflowReviewLifecycleService', () => { }), ctx, ); - // The whole batch is the affected set, so a deleted batch-mate cannot pass - // for a still-reviewable workflow. - expect(requestRepository.hasReviewableWorkflowOutside).toHaveBeenCalledExactlyOnceWith( - 'req-1', - ['wf-1', 'wf-2'], - ctx, - ); + // The request is evaluated once by its id against current state, so it closes exactly once. + expect(requestRepository.findUnreviewableOpenRequestIds).toHaveBeenCalledWith(ctx, ['req-1']); expect( activityRepository.createActivity.mock.calls.filter( ([input]) => input.type === 'review.closed', @@ -386,7 +377,7 @@ describe('WorkflowReviewLifecycleService', () => { }); it('degrades to the sweep when nothing was captured', async () => { - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue(['req-9']); + requestRepository.findUnreviewableOpenRequestIds.mockResolvedValue(['req-9']); await service.afterWorkflowsDeleted(['wf-1']); @@ -404,7 +395,7 @@ describe('WorkflowReviewLifecycleService', () => { // The delete already committed, so there is nothing left to abort. it('swallows repository errors after a delete', async () => { - requestRepository.closeUnreviewableOpenRequests.mockRejectedValue(new Error('db down')); + requestRepository.findUnreviewableOpenRequestIds.mockRejectedValue(new Error('db down')); await expect(service.afterWorkflowsDeleted(['wf-1', 'wf-2'])).resolves.toBeUndefined(); @@ -479,15 +470,14 @@ describe('WorkflowReviewLifecycleService', () => { describe('reconciliation sweep', () => { it('closes the requests the mutation stranded and explains each of them', async () => { requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([]); - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue(['req-9', 'req-10']); + // Global sweep (no candidate ids) strands req-9 and req-10. + requestRepository.findUnreviewableOpenRequestIds.mockImplementation( + async (_ctx, ids) => ids ?? ['req-9', 'req-10'], + ); await service.afterWorkflowArchived('wf-1', 'user-9'); - // Same lock and same transaction as every other close path: the sweep updates the - // requests it selected by id, so two racing sweeps would otherwise explain the same - // close twice. The sweep is global, so the batch never reaches the query; it is log - // context only. - expect(requestRepository.closeUnreviewableOpenRequests).toHaveBeenCalledExactlyOnceWith(ctx); + expect(requestRepository.closeRequests).toHaveBeenCalledWith(['req-9', 'req-10'], ctx); for (const requestId of ['req-9', 'req-10']) { expect(activityRepository.createActivity).toHaveBeenCalledWith( { @@ -498,6 +488,7 @@ describe('WorkflowReviewLifecycleService', () => { }, ctx, ); + // The sweep is the backstop: it recovers neither the trigger nor an actor. expect(eventService.emit).toHaveBeenCalledWith('workflow-review-closed', { workflowReviewRequestId: requestId, cause: { trigger: 'unknown', actorKind: 'system', userId: null }, @@ -512,8 +503,7 @@ describe('WorkflowReviewLifecycleService', () => { }); it('stays quiet when nothing is left unreviewable', async () => { - requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([]); - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue([]); + requestRepository.findUnreviewableOpenRequestIds.mockResolvedValue([]); await service.afterWorkflowsDeleted(['wf-1']); @@ -524,7 +514,7 @@ describe('WorkflowReviewLifecycleService', () => { // The close and its explanation share one transaction, so an unwritable entry rolls the // close back and the next sweep picks the review up again. it('leaves a review it cannot explain to the next sweep', async () => { - requestRepository.closeUnreviewableOpenRequests.mockResolvedValue(['req-9']); + requestRepository.findUnreviewableOpenRequestIds.mockResolvedValue(['req-9']); activityRepository.createActivity.mockRejectedValue(new Error('db down')); await expect(service.afterWorkflowsDeleted(['wf-1'])).resolves.toBeUndefined(); @@ -539,22 +529,34 @@ describe('WorkflowReviewLifecycleService', () => { // there too — the targeted close is done with the review by then. it('runs after the targeted close on archive, and again on transfer', async () => { requestRepository.findOpenRequestsForWorkflows.mockResolvedValue([]); + requestRepository.findUnreviewableOpenRequestIds.mockImplementation( + async (_ctx, ids) => ids ?? ['req-9'], + ); await service.afterWorkflowArchived('wf-1', 'user-9'); - expect(requestRepository.closeUnreviewableOpenRequests).toHaveBeenCalledExactlyOnceWith(ctx); + expect(eventService.emit).toHaveBeenCalledExactlyOnceWith('workflow-review-closed', { + workflowReviewRequestId: 'req-9', + cause: { trigger: 'unknown', actorKind: 'system', userId: null }, + }); await service.afterWorkflowsTransferred(['wf-2'], 'user-9'); - expect(requestRepository.closeUnreviewableOpenRequests).toHaveBeenCalledTimes(2); + expect(eventService.emit).toHaveBeenCalledTimes(2); }); // A close that rolled back is exactly what the sweep is there to repair, so a throwing // targeted close must not skip it. it('still runs when the targeted close on archive failed', async () => { requestRepository.findOpenRequestsForWorkflows.mockRejectedValue(new Error('db down')); + requestRepository.findUnreviewableOpenRequestIds.mockImplementation( + async (_ctx, ids) => ids ?? ['req-9'], + ); await expect(service.afterWorkflowArchived('wf-1', 'user-9')).resolves.toBeUndefined(); - expect(requestRepository.closeUnreviewableOpenRequests).toHaveBeenCalledExactlyOnceWith(ctx); + expect(eventService.emit).toHaveBeenCalledWith('workflow-review-closed', { + workflowReviewRequestId: 'req-9', + cause: { trigger: 'unknown', actorKind: 'system', userId: null }, + }); }); // The pre-delete hook only captures; reconciliation waits for the delete to commit. @@ -563,7 +565,7 @@ describe('WorkflowReviewLifecycleService', () => { await service.beforeWorkflowDeleted('wf-1', 'user-9'); - expect(requestRepository.closeUnreviewableOpenRequests).not.toHaveBeenCalled(); + expect(requestRepository.findUnreviewableOpenRequestIds).not.toHaveBeenCalled(); }); }); diff --git a/packages/cli/src/modules/workflow-reviews.ee/workflow-review-lifecycle.service.ts b/packages/cli/src/modules/workflow-reviews.ee/workflow-review-lifecycle.service.ts index e44bbcba712..b9607f397d9 100644 --- a/packages/cli/src/modules/workflow-reviews.ee/workflow-review-lifecycle.service.ts +++ b/packages/cli/src/modules/workflow-reviews.ee/workflow-review-lifecycle.service.ts @@ -1,4 +1,5 @@ -import type { OperationContext, WorkflowReviewRequest } from '@n8n/db'; +import type { WorkflowReviewWorkflowCauseActivityType } from '@n8n/api-types'; +import type { OperationContext } from '@n8n/db'; import { Logger } from '@n8n/backend-common'; import { DbLock, @@ -20,6 +21,18 @@ type PendingDeleteCapture = { requestIds: string[]; }; +/** A request a cause-recording pass closed, with the actor to attribute the close telemetry to. */ +type ClosedRequest = { requestId: string; actorKind: 'user' | 'system'; userId: string | null }; + +const CLOSE_TRIGGER_BY_ACTIVITY_TYPE = { + 'workflow.archived': 'workflow-archived', + 'workflow.moved': 'workflow-moved', + 'workflow.deleted': 'workflow-deleted', +} as const satisfies Record< + WorkflowReviewWorkflowCauseActivityType, + 'workflow-archived' | 'workflow-moved' | 'workflow-deleted' +>; + /** * A delete that never completes leaves its capture behind; the map is bounded so those * leftovers cannot accumulate forever. Far above any real burst of parallel deletes. @@ -183,31 +196,10 @@ export class WorkflowReviewLifecycleService implements WorkflowMutationHooks { */ private async reconcileUnreviewableOpenRequests(workflowIds: string[]): Promise { try { + // No candidate ids: every open request is evaluated, not just those this mutation touched. const closedRequestIds = await this.dbLockService.withLockContext( DbLock.WORKFLOW_REVIEW_MUTATION, - async (ctx) => { - const requestIds = - await this.workflowReviewRequestRepository.closeUnreviewableOpenRequests(ctx); - - // Under the same lock and in the same transaction as every other close path: the - // sweep selects the requests and then updates them by id, so two sweeps racing - // across that gap would both explain the same close and both overwrite whatever a - // concurrent approval wrote. A failed entry rolls the close back and the next - // sweep closes it again, which is what the sweep is for. - for (const requestId of requestIds) { - await this.activityRepository.createActivity( - { - workflowReviewRequestId: requestId, - type: 'review.closed', - data: { reason: 'no-reviewable-workflows' }, - createdById: null, - }, - ctx, - ); - } - - return requestIds; - }, + async (ctx) => await this.closeUnreviewable(ctx), ); if (closedRequestIds.length === 0) return; @@ -236,7 +228,8 @@ export class WorkflowReviewLifecycleService implements WorkflowMutationHooks { /** * Archive/move path: for every open request linked to an affected workflow, write one - * cause entry per affected link, then close the request iff the close policy fires. + * cause entry per affected link, then close the requests the close policy leaves with + * nothing reviewable. */ private async recordCauseEventsAndApplyClosePolicy( workflowIds: string[], @@ -245,185 +238,159 @@ export class WorkflowReviewLifecycleService implements WorkflowMutationHooks { ): Promise { const actorKind = userId === null ? 'system' : 'user'; - try { - const { affectedWorkflowIds, closedRequestIds } = await this.dbLockService.withLockContext( - DbLock.WORKFLOW_REVIEW_MUTATION, - async (ctx) => { - // Fetched under the lock so the close can't race a concurrent - // decide/version sync on the same request. - const openRequests = - await this.workflowReviewRequestRepository.findOpenRequestsForWorkflows( - workflowIds, - ctx, - ); - - const affected = new Set(); - const closedRequestIds: string[] = []; - for (const { request, links } of openRequests) { - for (const { workflowId: linkedWorkflowId } of links) { - await this.activityRepository.createActivity( - { - workflowReviewRequestId: request.id, - type, - data: { workflowId: linkedWorkflowId, actorKind }, - createdById: userId, - }, - ctx, - ); - affected.add(linkedWorkflowId); - } - - if (await this.applyClosePolicy(request, workflowIds, ctx)) { - closedRequestIds.push(request.id); - } - } - - return { affectedWorkflowIds: [...affected], closedRequestIds }; - }, + await this.recordCauseEventsAndClose(type, workflowIds, async (ctx) => { + // Under the lock, so the close can't race a concurrent decide/version sync. + const openRequests = await this.workflowReviewRequestRepository.findOpenRequestsForWorkflows( + workflowIds, + ctx, ); - // Ahead of the affected-ids guard below: the close is what is being reported. - for (const requestId of closedRequestIds) { - this.eventService.emit('workflow-review-closed', { - workflowReviewRequestId: requestId, - cause: { - trigger: type === 'workflow.archived' ? 'workflow-archived' : 'workflow-moved', - actorKind, - userId, - }, - }); + const affected = new Set(); + const candidateRequestIds: string[] = []; + for (const { request, links } of openRequests) { + for (const { workflowId: linkedWorkflowId } of links) { + await this.activityRepository.createActivity( + { + workflowReviewRequestId: request.id, + type, + data: { workflowId: linkedWorkflowId, actorKind }, + createdById: userId, + }, + ctx, + ); + affected.add(linkedWorkflowId); + } + candidateRequestIds.push(request.id); } - if (affectedWorkflowIds.length === 0) return; + const closedRequestIds = await this.closeUnreviewable(ctx, candidateRequestIds); - this.logger.info('Recorded workflow review cause event(s)', { - type, - workflowIds: affectedWorkflowIds, - }); - - this.broadcastReviewStateChanged(affectedWorkflowIds); - } catch (error) { - // The mutation has already committed — this hook observes it, so it never - // rethrows. A rolled-back close leaves the review open until the - // reconciliation sweep closes it again. - this.logger.error('Failed to record workflow review cause event(s)', { - type, - workflowIds, - error, - }); - } + return { + affectedWorkflowIds: [...affected], + closedRequests: closedRequestIds.map((requestId) => ({ requestId, actorKind, userId })), + }; + }); } /** * Delete path: consumes what `beforeWorkflowDeleted` captured, now that the delete has - * committed and the truth can be written. Batch-correct by construction: the whole batch - * arrives in one call, so the close policy never mistakes a deleted batch-mate for a - * still-reviewable workflow. + * committed and the truth can be written. Batch-correct by construction — every deleted + * workflow's rows are already gone, so the close policy cannot mistake a deleted batch-mate + * for a still-reviewable workflow. */ private async recordCapturedDeletions( capturesByRequestId: Map, batchWorkflowIds: string[], + ): Promise { + await this.recordCauseEventsAndClose('workflow.deleted', batchWorkflowIds, async (ctx) => { + const affected = new Set(); + const candidateRequestIds: string[] = []; + const actorByRequestId = new Map(); + for (const [requestId, capture] of capturesByRequestId) { + const request = await this.workflowReviewRequestRepository.findById(requestId, ctx); + if (!request || request.state !== 'open') continue; + + const actorKind = capture.userId === null ? 'system' : 'user'; + + for (const workflowId of capture.workflowIds) { + await this.activityRepository.createActivity( + { + workflowReviewRequestId: requestId, + type: 'workflow.deleted', + data: { workflowId, actorKind }, + createdById: capture.userId, + }, + ctx, + ); + affected.add(workflowId); + } + + candidateRequestIds.push(requestId); + actorByRequestId.set(requestId, { requestId, actorKind, userId: capture.userId }); + } + + const closedRequestIds = await this.closeUnreviewable(ctx, candidateRequestIds); + + return { + affectedWorkflowIds: [...affected], + closedRequests: closedRequestIds.map((requestId) => actorByRequestId.get(requestId)!), + }; + }); + } + + private async recordCauseEventsAndClose( + type: WorkflowReviewWorkflowCauseActivityType, + logWorkflowIds: string[], + gather: ( + ctx: OperationContext, + ) => Promise<{ affectedWorkflowIds: string[]; closedRequests: ClosedRequest[] }>, ): Promise { try { const { affectedWorkflowIds, closedRequests } = await this.dbLockService.withLockContext( DbLock.WORKFLOW_REVIEW_MUTATION, - async (ctx) => { - const affected = new Set(); - const closed: Array<{ - requestId: string; - actorKind: 'user' | 'system'; - userId: string | null; - }> = []; - for (const [requestId, capture] of capturesByRequestId) { - // Cause events record into open reviews; one that closed since the - // capture (e.g. approved meanwhile) gets nothing. - const request = await this.workflowReviewRequestRepository.findById(requestId, ctx); - if (!request || request.state !== 'open') continue; - - const actorKind = capture.userId === null ? 'system' : 'user'; - - for (const workflowId of capture.workflowIds) { - await this.activityRepository.createActivity( - { - workflowReviewRequestId: requestId, - type: 'workflow.deleted', - data: { workflowId, actorKind }, - createdById: capture.userId, - }, - ctx, - ); - affected.add(workflowId); - } - - if (await this.applyClosePolicy(request, batchWorkflowIds, ctx)) { - closed.push({ requestId, actorKind, userId: capture.userId }); - } - } - - return { affectedWorkflowIds: [...affected], closedRequests: closed }; - }, + gather, ); + const trigger = CLOSE_TRIGGER_BY_ACTIVITY_TYPE[type]; // Ahead of the affected-ids guard below: the close is what is being reported. for (const { requestId, actorKind, userId } of closedRequests) { this.eventService.emit('workflow-review-closed', { workflowReviewRequestId: requestId, - cause: { trigger: 'workflow-deleted', actorKind, userId }, + cause: { trigger, actorKind, userId }, }); } if (affectedWorkflowIds.length === 0) return; this.logger.info('Recorded workflow review cause event(s)', { - type: 'workflow.deleted', + type, workflowIds: affectedWorkflowIds, }); this.broadcastReviewStateChanged(affectedWorkflowIds); } catch (error) { - // The delete has committed; the sweep that follows closes what this pass - // missed, with `review.closed` alone. this.logger.error('Failed to record workflow review cause event(s)', { - type: 'workflow.deleted', - workflowIds: batchWorkflowIds, + type, + workflowIds: logWorkflowIds, error, }); } } /** - * Closes the request iff no linked workflow outside the affected set is reviewable, and - * reports whether it did. In the cause entries' transaction on purpose: a review closed - * without an explanation is worse than a close that rolls back and waits for the sweep. + * The one close transition, shared by the targeted paths and the reconciliation sweep: closes + * every candidate the close policy leaves with no reviewable workflow, appending a + * `review.closed` entry for each in the caller's transaction. + * + * Select-then-close-by-id runs under the caller's lock: a close that races a concurrent + * approval would otherwise both explain the same close and overwrite what the approval wrote. + * A `review.closed` write that fails rolls the whole close back, and the next sweep repairs it. */ - private async applyClosePolicy( - request: WorkflowReviewRequest, - affectedWorkflowIds: string[], + private async closeUnreviewable( ctx: OperationContext, - ): Promise { - const staysOpen = await this.workflowReviewRequestRepository.hasReviewableWorkflowOutside( - request.id, - affectedWorkflowIds, - ctx, - ); - if (staysOpen) return false; + candidateRequestIds?: string[], + ): Promise { + const closableRequestIds = + await this.workflowReviewRequestRepository.findUnreviewableOpenRequestIds( + ctx, + candidateRequestIds, + ); - request.state = 'closed'; - // A system close has no closing user; the decision stays as-is. - request.closedById = null; - await this.workflowReviewRequestRepository.saveRequest(request, ctx); + await this.workflowReviewRequestRepository.closeRequests(closableRequestIds, ctx); - await this.activityRepository.createActivity( - { - workflowReviewRequestId: request.id, - type: 'review.closed', - data: { reason: 'no-reviewable-workflows' }, - createdById: null, - }, - ctx, - ); + for (const requestId of closableRequestIds) { + await this.activityRepository.createActivity( + { + workflowReviewRequestId: requestId, + type: 'review.closed', + data: { reason: 'no-reviewable-workflows' }, + createdById: null, + }, + ctx, + ); + } - return true; + return closableRequestIds; } private broadcastReviewStateChanged(workflowIds: string[]): void {