github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-19 16:55:02 +02:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity Graph

optimize requests

Browse Source
This commit is contained in:
yehorkardash 2026-08-12 11:30:06 +02:00
parent 0846e2bdbb
commit 86e4162ce1
No known key found for this signature in database
4 changed files with 196 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
packages/cli/src/workflows/__tests__/workflow-bulk-action.service.test.ts
View File

@ -4,6 +4,7 @@ import type {
WorkflowEntity,
WorkflowPublishedVersionRepository,
} from '@n8n/db';
import type { Scope } from '@n8n/permissions';
import { mock } from 'vitest-mock-extended';
import type { CollaborationService } from '@/collaboration/collaboration.service';
@ -11,7 +12,10 @@ import type { CredentialsFinderService } from '@/credentials/credentials-finder.
import { UnprocessableRequestError } from '@/errors/response-errors/unprocessable.error';
import type { ProjectService } from '@/services/project.service.ee';
import { WorkflowBulkActionService } from '@/workflows/workflow-bulk-action.service';
import type { WorkflowFinderService } from '@/workflows/workflow-finder.service';
import type {
AuthorizedWorkflow,
WorkflowFinderService,
} from '@/workflows/workflow-finder.service';
import type { WorkflowService } from '@/workflows/workflow.service';
import type { EnterpriseWorkflowService } from '@/workflows/workflow.service.ee';
@ -26,6 +30,8 @@ describe('WorkflowBulkActionService', () => {
const workflowPublishedVersionRepository = mock<WorkflowPublishedVersionRepository>();
const credentialsFinderService = mock<CredentialsFinderService>();
let service: WorkflowBulkActionService;
const authorize = <S extends Scope>(workflow: WorkflowEntity, scope: S): AuthorizedWorkflow<S> =>
({ workflow, scope, userId: user.id }) as AuthorizedWorkflow<S>;
beforeEach(() => {
vi.clearAllMocks();
@ -45,24 +51,26 @@ describe('WorkflowBulkActionService', () => {
});
it('rejects the whole request when one workflow is inaccessible', async () => {
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue([
mock<WorkflowEntity>({ id: 'workflow-1', isArchived: false }),
workflowFinderService.findAuthorizedWorkflowsByIdsForUser.mockResolvedValue([
authorize(mock<WorkflowEntity>({ id: 'workflow-1', isArchived: false }), 'workflow:delete'),
]);
await expect(service.archive(user, ['workflow-1', 'workflow-2'])).rejects.toBeInstanceOf(
UnprocessableRequestError,
);
expect(workflowService.archive).not.toHaveBeenCalled();
expect(workflowService.archiveAuthorized).not.toHaveBeenCalled();
});
it('finishes the active parallel batch and stops before the next batch on failure', async () => {
const workflows = Array.from({ length: 6 }, (_, index) => `workflow-${index + 1}`).map((id) =>
mock<WorkflowEntity>({ id, isArchived: false }),
);
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue(workflows);
workflowService.archive.mockImplementation(async (_user, workflowId) => {
if (workflowId === 'workflow-2') throw new Error('hook rejected');
return workflows.find(({ id }) => id === workflowId);
workflowFinderService.findAuthorizedWorkflowsByIdsForUser.mockResolvedValue(
workflows.map((workflow) => authorize(workflow, 'workflow:delete')),
);
workflowService.archiveAuthorized.mockImplementation(async (_user, authorized) => {
if (authorized.workflow.id === 'workflow-2') throw new Error('hook rejected');
return authorized.workflow;
});
await expect(
@ -86,21 +94,29 @@ describe('WorkflowBulkActionService', () => {
{ workflowId: 'workflow-6', status: 'notAttempted' },
],
});
expect(workflowService.archive).toHaveBeenCalledTimes(5);
expect(workflowService.archiveAuthorized).toHaveBeenCalledTimes(5);
expect(workflowService.archive).not.toHaveBeenCalled();
expect(workflowFinderService.findAuthorizedWorkflowsByIdsForUser).toHaveBeenCalledTimes(1);
});
it('rejects permanent deletion unless every workflow is archived and fully unpublished', async () => {
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue([
mock<WorkflowEntity>({
id: 'not-archived',
isArchived: false,
activeVersionId: null,
}),
mock<WorkflowEntity>({
id: 'pending-unpublish',
isArchived: true,
activeVersionId: null,
}),
workflowFinderService.findAuthorizedWorkflowsByIdsForUser.mockResolvedValue([
authorize(
mock<WorkflowEntity>({
id: 'not-archived',
isArchived: false,
activeVersionId: null,
}),
'workflow:delete',
),
authorize(
mock<WorkflowEntity>({
id: 'pending-unpublish',
isArchived: true,
activeVersionId: null,
}),
'workflow:delete',
),
]);
workflowPublishedVersionRepository.getWorkflowIdsWithPublishedVersion.mockResolvedValue(
new Set(['pending-unpublish']),
@ -119,18 +135,23 @@ describe('WorkflowBulkActionService', () => {
},
},
);
expect(workflowService.deleteAuthorized).not.toHaveBeenCalled();
expect(workflowService.delete).not.toHaveBeenCalled();
});
it('reports already-unpublished workflows as unchanged', async () => {
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue([
mock<WorkflowEntity>({ id: 'workflow-1', activeVersionId: null }),
workflowFinderService.findAuthorizedWorkflowsByIdsForUser.mockResolvedValue([
authorize(
mock<WorkflowEntity>({ id: 'workflow-1', activeVersionId: null }),
'workflow:unpublish',
),
]);
await expect(service.unpublish(user, ['workflow-1'])).resolves.toEqual({
status: 'completed',
results: [{ workflowId: 'workflow-1', status: 'unchanged' }],
});
expect(workflowService.deactivateAuthorized).not.toHaveBeenCalled();
expect(workflowService.deactivateWorkflow).not.toHaveBeenCalled();
});
});

87
packages/cli/src/workflows/workflow-bulk-action.service.ts
View File

@ -7,11 +7,10 @@ import { ensureError } from '@n8n/utils/errors/ensure-error';
import { CollaborationService } from '@/collaboration/collaboration.service';
import { CredentialsFinderService } from '@/credentials/credentials-finder.service';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import { UnprocessableRequestError } from '@/errors/response-errors/unprocessable.error';
import { ProjectService } from '@/services/project.service.ee';
import { WorkflowFinderService } from './workflow-finder.service';
import { type AuthorizedWorkflow, WorkflowFinderService } from './workflow-finder.service';
import { WorkflowService } from './workflow.service';
import { EnterpriseWorkflowService } from './workflow.service.ee';
@ -26,13 +25,10 @@ type PreflightOptions = {
clientId?: string;
includeParentFolder?: boolean;
includeShared?: boolean;
includeActiveVersion?: boolean;
validate?: (workflow: WorkflowEntity) => PreflightIssue | undefined;
};
type ExecuteItem = (
workflow: WorkflowEntity,
) => Promise<Omit<BulkWorkflowActionResultItem, 'workflowId'>>;
const LIFECYCLE_BATCH_SIZE = 5;
@Service()
@ -53,7 +49,7 @@ export class WorkflowBulkActionService {
workflowIds: string[],
clientId?: string,
): Promise<BulkWorkflowActionResult> {
const workflows = await this.preflight(user, workflowIds, ['workflow:delete'], {
const workflows = await this.preflight(user, workflowIds, 'workflow:delete', {
action: 'archive',
clientId,
validate: (workflow) =>
@ -66,11 +62,9 @@ export class WorkflowBulkActionService {
: undefined,
});
return await this.execute(workflows, async (workflow) => {
const archived = await this.workflowService.archive(user, workflow.id);
if (!archived) throw new NotFoundError('Workflow is no longer accessible.');
await this.collaborationService.broadcastWorkflowUpdate(workflow.id, user.id);
return await this.execute(workflows, async (authorized) => {
const archived = await this.workflowService.archiveAuthorized(user, authorized);
await this.collaborationService.broadcastWorkflowUpdate(archived.id, user.id);
return { status: 'completed' };
});
}
@ -83,7 +77,7 @@ export class WorkflowBulkActionService {
const uniqueIds = this.uniqueIds(workflowIds);
const pendingPublishedIds =
await this.workflowPublishedVersionRepository.getWorkflowIdsWithPublishedVersion(uniqueIds);
const workflows = await this.preflight(user, uniqueIds, ['workflow:delete'], {
const workflows = await this.preflight(user, uniqueIds, 'workflow:delete', {
action: 'delete',
clientId,
validate: (workflow) => {
@ -112,9 +106,8 @@ export class WorkflowBulkActionService {
},
});
return await this.execute(workflows, async (workflow) => {
const deleted = await this.workflowService.delete(user, workflow.id);
if (!deleted) throw new NotFoundError('Workflow is no longer accessible.');
return await this.execute(workflows, async (authorized) => {
await this.workflowService.deleteAuthorized(user, authorized);
return { status: 'completed' };
});
}
@ -124,15 +117,16 @@ export class WorkflowBulkActionService {
workflowIds: string[],
clientId?: string,
): Promise<BulkWorkflowActionResult> {
const workflows = await this.preflight(user, workflowIds, ['workflow:unpublish'], {
const workflows = await this.preflight(user, workflowIds, 'workflow:unpublish', {
action: 'unpublish',
clientId,
includeActiveVersion: true,
});
return await this.execute(workflows, async (workflow) => {
if (workflow.activeVersionId === null) return { status: 'unchanged' };
return await this.execute(workflows, async (authorized) => {
if (authorized.workflow.activeVersionId === null) return { status: 'unchanged' };
await this.workflowService.deactivateWorkflow(user, workflow.id);
const workflow = await this.workflowService.deactivateAuthorized(user, authorized);
await this.collaborationService.broadcastWorkflowUpdate(workflow.id, user.id);
return { status: 'completed' };
});
@ -176,7 +170,7 @@ export class WorkflowBulkActionService {
}
}
const workflows = await this.preflight(user, workflowIds, ['workflow:move'], {
const workflows = await this.preflight(user, workflowIds, 'workflow:move', {
action: 'transfer',
clientId,
includeParentFolder: true,
@ -211,7 +205,7 @@ export class WorkflowBulkActionService {
(id) => !shareableCredentialIdSet.has(id),
);
const result = await this.execute(workflows, async (workflow) => {
const result = await this.execute(workflows, async ({ workflow }) => {
const activationError = await this.enterpriseWorkflowService.transferWorkflow(
user,
workflow.id,
@ -232,28 +226,27 @@ export class WorkflowBulkActionService {
return { ...result, unsharedCredentialIds };
}
private async preflight(
private async preflight<S extends Scope>(
user: User,
workflowIds: string[],
scopes: Scope[],
scope: S,
options: PreflightOptions,
): Promise<WorkflowEntity[]> {
): Promise<Array<AuthorizedWorkflow<S>>> {
const uniqueIds = this.uniqueIds(workflowIds);
const accessibleWorkflows = await this.workflowFinderService.findWorkflowsByIdsForUser(
uniqueIds,
user,
scopes,
{
const authorizedWorkflows =
await this.workflowFinderService.findAuthorizedWorkflowsByIdsForUser(uniqueIds, user, scope, {
includeParentFolder: options.includeParentFolder,
includeShared: options.includeShared,
},
includeActiveVersion: options.includeActiveVersion,
});
const workflowById = new Map(
authorizedWorkflows.map((authorized) => [authorized.workflow.id, authorized]),
);
const workflowById = new Map(accessibleWorkflows.map((workflow) => [workflow.id, workflow]));
const issues: PreflightIssue[] = [];
for (const workflowId of uniqueIds) {
const workflow = workflowById.get(workflowId);
if (!workflow) {
const authorized = workflowById.get(workflowId);
if (!authorized) {
issues.push({
workflowId,
reason: 'notFoundOrForbidden',
@ -261,6 +254,7 @@ export class WorkflowBulkActionService {
});
continue;
}
const { workflow } = authorized;
try {
await this.collaborationService.validateWriteLock(
@ -284,24 +278,31 @@ export class WorkflowBulkActionService {
if (issues.length > 0) throw this.preflightError(issues);
return uniqueIds.map((id) => workflowById.get(id)).filter((workflow) => workflow !== undefined);
return uniqueIds
.map((id) => workflowById.get(id))
.filter((authorized) => authorized !== undefined);
}
private async execute(
workflows: WorkflowEntity[],
executeItem: ExecuteItem,
private async execute<S extends Scope>(
workflows: Array<AuthorizedWorkflow<S>>,
executeItem: (
authorized: AuthorizedWorkflow<S>,
) => Promise<Omit<BulkWorkflowActionResultItem, 'workflowId'>>,
): Promise<BulkWorkflowActionResult> {
const results: BulkWorkflowActionResultItem[] = [];
for (let start = 0; start < workflows.length; start += LIFECYCLE_BATCH_SIZE) {
const batch = workflows.slice(start, start + LIFECYCLE_BATCH_SIZE);
const batchResults = await Promise.all(
batch.map(async (workflow): Promise<BulkWorkflowActionResultItem> => {
batch.map(async (authorized): Promise<BulkWorkflowActionResultItem> => {
try {
return { workflowId: workflow.id, ...(await executeItem(workflow)) };
return {
workflowId: authorized.workflow.id,
...(await executeItem(authorized)),
};
} catch (error) {
return {
workflowId: workflow.id,
workflowId: authorized.workflow.id,
status: 'failed',
reason: 'runtimeFailure',
message: ensureError(error).message,
@ -313,8 +314,8 @@ export class WorkflowBulkActionService {
if (batchResults.some(({ status }) => status === 'failed')) {
results.push(
...workflows.slice(start + batch.length).map(({ id }) => ({
workflowId: id,
...workflows.slice(start + batch.length).map(({ workflow }) => ({
workflowId: workflow.id,
status: 'notAttempted' as const,
})),
);

31
packages/cli/src/workflows/workflow-finder.service.ts
View File

@ -8,6 +8,15 @@ import { In, IsNull } from '@n8n/typeorm';
import { userHasScopes } from '@/permissions.ee/check-access';
import { RoleService } from '@/services/role.service';
const authorizedWorkflowBrand = Symbol('authorizedWorkflow');
export type AuthorizedWorkflow<S extends Scope> = {
workflow: WorkflowEntity;
scope: S;
userId: string;
readonly [authorizedWorkflowBrand]: true;
};
export type FindWorkflowsForUserOptions = {
filters?: {
name?: string;
@ -185,6 +194,7 @@ export class WorkflowFinderService {
includeParentFolder?: boolean;
includeTags?: boolean;
includeShared?: boolean;
includeActiveVersion?: boolean;
} = {},
): Promise<WorkflowEntity[]> {
if (workflowIds.length === 0) return [];
@ -197,6 +207,7 @@ export class WorkflowFinderService {
parentFolder: options.includeParentFolder,
tags: options.includeTags,
shared: options.includeShared ? { project: true } : false,
activeVersion: options.includeActiveVersion,
},
},
});
@ -213,6 +224,26 @@ export class WorkflowFinderService {
return workflows;
}
async findAuthorizedWorkflowsByIdsForUser<S extends Scope>(
workflowIds: string[],
user: User,
scope: S,
options: {
includeParentFolder?: boolean;
includeTags?: boolean;
includeShared?: boolean;
includeActiveVersion?: boolean;
} = {},
): Promise<Array<AuthorizedWorkflow<S>>> {
const workflows = await this.findWorkflowsByIdsForUser(workflowIds, user, [scope], options);
return workflows.map((workflow) => ({
workflow,
scope,
userId: user.id,
[authorizedWorkflowBrand]: true,
}));
}
async findWorkflowIdsByFolder(folderIds: string[]): Promise<Map<string, string[]>> {
if (folderIds.length === 0) return new Map();

81
packages/cli/src/workflows/workflow.service.ts
View File

@ -29,7 +29,7 @@ import { v4 as uuid } from 'uuid';
import { WorkflowPublicationNotifier } from './publication/workflow-publication-notifier';
import { getErrorDescription, getErrorNodeId, getRequiredRedactionScopes } from './utils';
import { WorkflowFinderService } from './workflow-finder.service';
import { type AuthorizedWorkflow, WorkflowFinderService } from './workflow-finder.service';
import { WorkflowHistoryService } from './workflow-history/workflow-history.service';
import { WorkflowMutationHooksProxy } from './workflow-mutation-hooks-proxy.service';
import { WorkflowPublishGuardProxy } from './workflow-publish-guard-proxy.service';
@ -992,8 +992,6 @@ export class WorkflowService {
source?: WorkflowActionSource;
},
): Promise<WorkflowEntity> {
const source = options?.source ?? 'ui';
const publicApi = source === 'api';
const workflow = await this.workflowFinderService.findWorkflowForUser(
workflowId,
user,
@ -1011,6 +1009,32 @@ export class WorkflowService {
);
}
return await this.deactivateWorkflowEntity(user, workflow, options);
}
async deactivateAuthorized(
user: User,
authorized: AuthorizedWorkflow<'workflow:unpublish'>,
options?: {
expectedChecksum?: string;
source?: WorkflowActionSource;
},
): Promise<WorkflowEntity> {
this.assertAuthorizedWorkflow(user, authorized, 'workflow:unpublish');
return await this.deactivateWorkflowEntity(user, authorized.workflow, options);
}
private async deactivateWorkflowEntity(
user: User,
workflow: WorkflowEntity,
options?: {
expectedChecksum?: string;
source?: WorkflowActionSource;
},
): Promise<WorkflowEntity> {
const source = options?.source ?? 'ui';
const publicApi = source === 'api';
const workflowId = workflow.id;
const deactivatedVersionId = workflow.activeVersionId;
if (deactivatedVersionId === null) {
return workflow;
@ -1154,6 +1178,28 @@ export class WorkflowService {
return;
}
return await this.deleteWorkflowEntity(user, workflow, force);
}
async deleteAuthorized(
user: User,
authorized: AuthorizedWorkflow<'workflow:delete'>,
force = false,
): Promise<WorkflowEntity> {
this.assertAuthorizedWorkflow(user, authorized, 'workflow:delete');
await this.externalHooks.run('workflow.delete', [
authorized.workflow.id,
toWorkflowLifecycleHookActor(user),
]);
return await this.deleteWorkflowEntity(user, authorized.workflow, force);
}
private async deleteWorkflowEntity(
user: User,
workflow: WorkflowEntity,
force: boolean,
): Promise<WorkflowEntity> {
const workflowId = workflow.id;
if (this.globalConfig.workflows.useWorkflowPublicationService) {
if (workflow.activeVersionId !== null) {
throw new ConflictError(
@ -1222,6 +1268,24 @@ export class WorkflowService {
return;
}
return await this.archiveWorkflowEntity(user, workflow, options);
}
async archiveAuthorized(
user: User,
authorized: AuthorizedWorkflow<'workflow:delete'>,
options?: { skipArchived?: boolean; expectedChecksum?: string; publicApi?: boolean },
): Promise<WorkflowEntity> {
this.assertAuthorizedWorkflow(user, authorized, 'workflow:delete');
return await this.archiveWorkflowEntity(user, authorized.workflow, options);
}
private async archiveWorkflowEntity(
user: User,
workflow: WorkflowEntity,
options?: { skipArchived?: boolean; expectedChecksum?: string; publicApi?: boolean },
): Promise<WorkflowEntity> {
const workflowId = workflow.id;
if (workflow.isArchived) {
if (options?.skipArchived) {
return workflow;
@ -1638,6 +1702,17 @@ export class WorkflowService {
this.workflowPublicationNotifier.requestDrain();
}
private assertAuthorizedWorkflow<S extends Scope>(
user: User,
authorized: AuthorizedWorkflow<S>,
requiredScope: S,
): void {
assert(
authorized.userId === user.id && authorized.scope === requiredScope,
'Authorized workflow does not match the requested actor and scope',
);
}
/**
* Replace all tag mappings on a workflow. Missing tag IDs surface as NotFoundError.
*/