mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-19 16:55:02 +02:00
add bulk actions endpoints for workflows
This commit is contained in:
parent
e4f0c33584
commit
0846e2bdbb
|
|
@ -130,6 +130,15 @@ export { TransferWorkflowBodyDto } from './workflows/transfer.dto';
|
|||
export { ActivateWorkflowDto } from './workflows/activate-workflow.dto';
|
||||
export { DeactivateWorkflowDto } from './workflows/deactivate-workflow.dto';
|
||||
export { ArchiveWorkflowDto } from './workflows/archive-workflow.dto';
|
||||
export {
|
||||
BulkArchiveWorkflowsDto,
|
||||
BulkDeleteWorkflowsDto,
|
||||
BulkTransferWorkflowsDto,
|
||||
BulkUnpublishWorkflowsDto,
|
||||
type BulkWorkflowActionItemStatus,
|
||||
type BulkWorkflowActionResult,
|
||||
type BulkWorkflowActionResultItem,
|
||||
} from './workflows/bulk-workflow-action.dto';
|
||||
export { GetResourceDependencyCountsDto } from './workflows/get-resource-dependency-counts.dto';
|
||||
export { GetResourceDependenciesDto } from './workflows/get-resource-dependencies.dto';
|
||||
export { GetWorkflowQueryDto } from './workflows/get-workflow-query.dto';
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
import {
|
||||
BulkArchiveWorkflowsDto,
|
||||
BulkDeleteWorkflowsDto,
|
||||
BulkTransferWorkflowsDto,
|
||||
BulkUnpublishWorkflowsDto,
|
||||
} from '../bulk-workflow-action.dto';
|
||||
|
||||
const idDtos = [BulkArchiveWorkflowsDto, BulkDeleteWorkflowsDto, BulkUnpublishWorkflowsDto];
|
||||
|
||||
describe('bulk workflow action DTOs', () => {
|
||||
it.each(idDtos)('accepts between 1 and 100 workflow IDs', (Dto) => {
|
||||
expect(Dto.safeParse({ workflowIds: ['workflow-1'] }).success).toBe(true);
|
||||
expect(
|
||||
Dto.safeParse({ workflowIds: Array.from({ length: 100 }, (_, index) => `${index}`) }).success,
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it.each(idDtos)('rejects missing, empty, and oversized workflow ID lists', (Dto) => {
|
||||
expect(Dto.safeParse({}).success).toBe(false);
|
||||
expect(Dto.safeParse({ workflowIds: [] }).success).toBe(false);
|
||||
expect(Dto.safeParse({ workflowIds: [''] }).success).toBe(false);
|
||||
expect(
|
||||
Dto.safeParse({ workflowIds: Array.from({ length: 101 }, (_, index) => `${index}`) }).success,
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('validates bulk transfer fields', () => {
|
||||
expect(
|
||||
BulkTransferWorkflowsDto.safeParse({
|
||||
workflowIds: ['workflow-1'],
|
||||
destinationProjectId: 'project-1',
|
||||
destinationParentFolderId: 'folder-1',
|
||||
shareCredentials: ['credential-1'],
|
||||
}).success,
|
||||
).toBe(true);
|
||||
expect(
|
||||
BulkTransferWorkflowsDto.safeParse({
|
||||
workflowIds: ['workflow-1'],
|
||||
destinationProjectId: '',
|
||||
}).success,
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
import { z } from 'zod';
|
||||
|
||||
import { Z } from '../../zod-class';
|
||||
|
||||
const workflowIdsSchema = z.array(z.string().min(1)).min(1).max(100);
|
||||
|
||||
export class BulkArchiveWorkflowsDto extends Z.class({
|
||||
workflowIds: workflowIdsSchema,
|
||||
}) {}
|
||||
|
||||
export class BulkDeleteWorkflowsDto extends Z.class({
|
||||
workflowIds: workflowIdsSchema,
|
||||
}) {}
|
||||
|
||||
export class BulkUnpublishWorkflowsDto extends Z.class({
|
||||
workflowIds: workflowIdsSchema,
|
||||
}) {}
|
||||
|
||||
export class BulkTransferWorkflowsDto extends Z.class({
|
||||
workflowIds: workflowIdsSchema,
|
||||
destinationProjectId: z.string().min(1),
|
||||
destinationParentFolderId: z.string().min(1).optional(),
|
||||
shareCredentials: z.array(z.string().min(1)).optional(),
|
||||
}) {}
|
||||
|
||||
export const bulkWorkflowActionItemStatusSchema = z.enum([
|
||||
'completed',
|
||||
'unchanged',
|
||||
'failed',
|
||||
'notAttempted',
|
||||
]);
|
||||
|
||||
export type BulkWorkflowActionItemStatus = z.infer<typeof bulkWorkflowActionItemStatusSchema>;
|
||||
|
||||
export type BulkWorkflowActionResultItem = {
|
||||
workflowId: string;
|
||||
status: BulkWorkflowActionItemStatus;
|
||||
reason?: string;
|
||||
message?: string;
|
||||
};
|
||||
|
||||
export type BulkWorkflowActionResult = {
|
||||
status: 'completed' | 'partial';
|
||||
results: BulkWorkflowActionResultItem[];
|
||||
unsharedCredentialIds?: string[];
|
||||
};
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
import { Service } from '@n8n/di';
|
||||
import { DataSource, Repository } from '@n8n/typeorm';
|
||||
import { DataSource, In, Repository } from '@n8n/typeorm';
|
||||
import type { IWorkflowBase } from 'n8n-workflow';
|
||||
|
||||
import { WorkflowPublishedVersion } from '../entities';
|
||||
|
|
@ -48,6 +48,17 @@ export class WorkflowPublishedVersionRepository extends Repository<WorkflowPubli
|
|||
return record?.publishedVersionId ?? null;
|
||||
}
|
||||
|
||||
async getWorkflowIdsWithPublishedVersion(workflowIds: string[]): Promise<Set<string>> {
|
||||
if (workflowIds.length === 0) return new Set();
|
||||
|
||||
const records = await this.find({
|
||||
where: { workflowId: In(workflowIds) },
|
||||
select: ['workflowId'],
|
||||
});
|
||||
|
||||
return new Set(records.map(({ workflowId }) => workflowId));
|
||||
}
|
||||
|
||||
/**
|
||||
* Loads the published version record with the related workflow entity
|
||||
* (including shared/project relations) and the workflow history version
|
||||
|
|
|
|||
|
|
@ -0,0 +1,136 @@
|
|||
import type {
|
||||
FolderRepository,
|
||||
User,
|
||||
WorkflowEntity,
|
||||
WorkflowPublishedVersionRepository,
|
||||
} from '@n8n/db';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { CollaborationService } from '@/collaboration/collaboration.service';
|
||||
import type { CredentialsFinderService } from '@/credentials/credentials-finder.service';
|
||||
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 { WorkflowService } from '@/workflows/workflow.service';
|
||||
import type { EnterpriseWorkflowService } from '@/workflows/workflow.service.ee';
|
||||
|
||||
describe('WorkflowBulkActionService', () => {
|
||||
const user = mock<User>({ id: 'user-1' });
|
||||
const workflowFinderService = mock<WorkflowFinderService>();
|
||||
const workflowService = mock<WorkflowService>();
|
||||
const enterpriseWorkflowService = mock<EnterpriseWorkflowService>();
|
||||
const collaborationService = mock<CollaborationService>();
|
||||
const projectService = mock<ProjectService>();
|
||||
const folderRepository = mock<FolderRepository>();
|
||||
const workflowPublishedVersionRepository = mock<WorkflowPublishedVersionRepository>();
|
||||
const credentialsFinderService = mock<CredentialsFinderService>();
|
||||
let service: WorkflowBulkActionService;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
workflowPublishedVersionRepository.getWorkflowIdsWithPublishedVersion.mockResolvedValue(
|
||||
new Set(),
|
||||
);
|
||||
service = new WorkflowBulkActionService(
|
||||
workflowFinderService,
|
||||
workflowService,
|
||||
enterpriseWorkflowService,
|
||||
collaborationService,
|
||||
projectService,
|
||||
folderRepository,
|
||||
workflowPublishedVersionRepository,
|
||||
credentialsFinderService,
|
||||
);
|
||||
});
|
||||
|
||||
it('rejects the whole request when one workflow is inaccessible', async () => {
|
||||
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue([
|
||||
mock<WorkflowEntity>({ id: 'workflow-1', isArchived: false }),
|
||||
]);
|
||||
|
||||
await expect(service.archive(user, ['workflow-1', 'workflow-2'])).rejects.toBeInstanceOf(
|
||||
UnprocessableRequestError,
|
||||
);
|
||||
expect(workflowService.archive).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);
|
||||
});
|
||||
|
||||
await expect(
|
||||
service.archive(
|
||||
user,
|
||||
workflows.map(({ id }) => id),
|
||||
),
|
||||
).resolves.toEqual({
|
||||
status: 'partial',
|
||||
results: [
|
||||
{ workflowId: 'workflow-1', status: 'completed' },
|
||||
{
|
||||
workflowId: 'workflow-2',
|
||||
status: 'failed',
|
||||
reason: 'runtimeFailure',
|
||||
message: 'hook rejected',
|
||||
},
|
||||
{ workflowId: 'workflow-3', status: 'completed' },
|
||||
{ workflowId: 'workflow-4', status: 'completed' },
|
||||
{ workflowId: 'workflow-5', status: 'completed' },
|
||||
{ workflowId: 'workflow-6', status: 'notAttempted' },
|
||||
],
|
||||
});
|
||||
expect(workflowService.archive).toHaveBeenCalledTimes(5);
|
||||
});
|
||||
|
||||
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,
|
||||
}),
|
||||
]);
|
||||
workflowPublishedVersionRepository.getWorkflowIdsWithPublishedVersion.mockResolvedValue(
|
||||
new Set(['pending-unpublish']),
|
||||
);
|
||||
|
||||
await expect(service.delete(user, ['not-archived', 'pending-unpublish'])).rejects.toMatchObject(
|
||||
{
|
||||
meta: {
|
||||
issues: [
|
||||
expect.objectContaining({ workflowId: 'not-archived', reason: 'notArchived' }),
|
||||
expect.objectContaining({
|
||||
workflowId: 'pending-unpublish',
|
||||
reason: 'unpublishPending',
|
||||
}),
|
||||
],
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(workflowService.delete).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('reports already-unpublished workflows as unchanged', async () => {
|
||||
workflowFinderService.findWorkflowsByIdsForUser.mockResolvedValue([
|
||||
mock<WorkflowEntity>({ id: 'workflow-1', activeVersionId: null }),
|
||||
]);
|
||||
|
||||
await expect(service.unpublish(user, ['workflow-1'])).resolves.toEqual({
|
||||
status: 'completed',
|
||||
results: [{ workflowId: 'workflow-1', status: 'unchanged' }],
|
||||
});
|
||||
expect(workflowService.deactivateWorkflow).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
349
packages/cli/src/workflows/workflow-bulk-action.service.ts
Normal file
349
packages/cli/src/workflows/workflow-bulk-action.service.ts
Normal file
|
|
@ -0,0 +1,349 @@
|
|||
import type { BulkWorkflowActionResult, BulkWorkflowActionResultItem } from '@n8n/api-types';
|
||||
import type { User, WorkflowEntity } from '@n8n/db';
|
||||
import { FolderRepository, WorkflowPublishedVersionRepository } from '@n8n/db';
|
||||
import { Service } from '@n8n/di';
|
||||
import { hasGlobalScope, type Scope } from '@n8n/permissions';
|
||||
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 { WorkflowService } from './workflow.service';
|
||||
import { EnterpriseWorkflowService } from './workflow.service.ee';
|
||||
|
||||
type PreflightIssue = {
|
||||
workflowId?: string;
|
||||
reason: string;
|
||||
message: string;
|
||||
};
|
||||
|
||||
type PreflightOptions = {
|
||||
action: string;
|
||||
clientId?: string;
|
||||
includeParentFolder?: boolean;
|
||||
includeShared?: boolean;
|
||||
validate?: (workflow: WorkflowEntity) => PreflightIssue | undefined;
|
||||
};
|
||||
|
||||
type ExecuteItem = (
|
||||
workflow: WorkflowEntity,
|
||||
) => Promise<Omit<BulkWorkflowActionResultItem, 'workflowId'>>;
|
||||
|
||||
const LIFECYCLE_BATCH_SIZE = 5;
|
||||
|
||||
@Service()
|
||||
export class WorkflowBulkActionService {
|
||||
constructor(
|
||||
private readonly workflowFinderService: WorkflowFinderService,
|
||||
private readonly workflowService: WorkflowService,
|
||||
private readonly enterpriseWorkflowService: EnterpriseWorkflowService,
|
||||
private readonly collaborationService: CollaborationService,
|
||||
private readonly projectService: ProjectService,
|
||||
private readonly folderRepository: FolderRepository,
|
||||
private readonly workflowPublishedVersionRepository: WorkflowPublishedVersionRepository,
|
||||
private readonly credentialsFinderService: CredentialsFinderService,
|
||||
) {}
|
||||
|
||||
async archive(
|
||||
user: User,
|
||||
workflowIds: string[],
|
||||
clientId?: string,
|
||||
): Promise<BulkWorkflowActionResult> {
|
||||
const workflows = await this.preflight(user, workflowIds, ['workflow:delete'], {
|
||||
action: 'archive',
|
||||
clientId,
|
||||
validate: (workflow) =>
|
||||
workflow.isArchived
|
||||
? {
|
||||
workflowId: workflow.id,
|
||||
reason: 'alreadyArchived',
|
||||
message: 'Workflow is already archived.',
|
||||
}
|
||||
: 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 { status: 'completed' };
|
||||
});
|
||||
}
|
||||
|
||||
async delete(
|
||||
user: User,
|
||||
workflowIds: string[],
|
||||
clientId?: string,
|
||||
): Promise<BulkWorkflowActionResult> {
|
||||
const uniqueIds = this.uniqueIds(workflowIds);
|
||||
const pendingPublishedIds =
|
||||
await this.workflowPublishedVersionRepository.getWorkflowIdsWithPublishedVersion(uniqueIds);
|
||||
const workflows = await this.preflight(user, uniqueIds, ['workflow:delete'], {
|
||||
action: 'delete',
|
||||
clientId,
|
||||
validate: (workflow) => {
|
||||
if (!workflow.isArchived) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
reason: 'notArchived',
|
||||
message: 'Workflow must be archived before it can be deleted.',
|
||||
};
|
||||
}
|
||||
if (workflow.activeVersionId !== null) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
reason: 'published',
|
||||
message: 'Cannot delete a published workflow. Unpublish it before deleting.',
|
||||
};
|
||||
}
|
||||
if (pendingPublishedIds.has(workflow.id)) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
reason: 'unpublishPending',
|
||||
message: 'Workflow is still being unpublished.',
|
||||
};
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
});
|
||||
|
||||
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 { status: 'completed' };
|
||||
});
|
||||
}
|
||||
|
||||
async unpublish(
|
||||
user: User,
|
||||
workflowIds: string[],
|
||||
clientId?: string,
|
||||
): Promise<BulkWorkflowActionResult> {
|
||||
const workflows = await this.preflight(user, workflowIds, ['workflow:unpublish'], {
|
||||
action: 'unpublish',
|
||||
clientId,
|
||||
});
|
||||
|
||||
return await this.execute(workflows, async (workflow) => {
|
||||
if (workflow.activeVersionId === null) return { status: 'unchanged' };
|
||||
|
||||
await this.workflowService.deactivateWorkflow(user, workflow.id);
|
||||
await this.collaborationService.broadcastWorkflowUpdate(workflow.id, user.id);
|
||||
return { status: 'completed' };
|
||||
});
|
||||
}
|
||||
|
||||
async transfer(
|
||||
user: User,
|
||||
workflowIds: string[],
|
||||
destinationProjectId: string,
|
||||
shareCredentials: string[] = [],
|
||||
destinationParentFolderId?: string,
|
||||
clientId?: string,
|
||||
): Promise<BulkWorkflowActionResult> {
|
||||
const destinationProject = await this.projectService.getProjectWithScope(
|
||||
user,
|
||||
destinationProjectId,
|
||||
['workflow:create'],
|
||||
);
|
||||
if (!destinationProject) {
|
||||
throw this.preflightError([
|
||||
{
|
||||
reason: 'destinationNotFoundOrForbidden',
|
||||
message: 'The destination project does not exist or is not accessible.',
|
||||
},
|
||||
]);
|
||||
}
|
||||
|
||||
if (destinationParentFolderId) {
|
||||
try {
|
||||
await this.folderRepository.findOneOrFailFolderInProject(
|
||||
destinationParentFolderId,
|
||||
destinationProjectId,
|
||||
);
|
||||
} catch {
|
||||
throw this.preflightError([
|
||||
{
|
||||
reason: 'destinationFolderNotFound',
|
||||
message: 'The destination folder does not exist in the destination project.',
|
||||
},
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
const workflows = await this.preflight(user, workflowIds, ['workflow:move'], {
|
||||
action: 'transfer',
|
||||
clientId,
|
||||
includeParentFolder: true,
|
||||
includeShared: true,
|
||||
validate: (workflow) => {
|
||||
const ownerSharing = workflow.shared.find((sharing) => sharing.role === 'workflow:owner');
|
||||
if (!ownerSharing) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
reason: 'ownerMissing',
|
||||
message: 'Workflow does not have an owner project.',
|
||||
};
|
||||
}
|
||||
|
||||
if (
|
||||
ownerSharing.project.id === destinationProjectId &&
|
||||
workflow.parentFolder?.id === destinationParentFolderId
|
||||
) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
reason: 'sameDestination',
|
||||
message: 'Workflow already belongs to the requested destination.',
|
||||
};
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
});
|
||||
|
||||
const shareableCredentialIds = await this.getShareableCredentialIds(user, shareCredentials);
|
||||
const shareableCredentialIdSet = new Set(shareableCredentialIds);
|
||||
const unsharedCredentialIds = this.uniqueIds(shareCredentials).filter(
|
||||
(id) => !shareableCredentialIdSet.has(id),
|
||||
);
|
||||
|
||||
const result = await this.execute(workflows, async (workflow) => {
|
||||
const activationError = await this.enterpriseWorkflowService.transferWorkflow(
|
||||
user,
|
||||
workflow.id,
|
||||
destinationProjectId,
|
||||
shareableCredentialIds,
|
||||
destinationParentFolderId,
|
||||
);
|
||||
|
||||
return activationError
|
||||
? {
|
||||
status: 'completed',
|
||||
reason: 'reactivationFailed',
|
||||
message: 'Workflow was moved but could not be reactivated.',
|
||||
}
|
||||
: { status: 'completed' };
|
||||
});
|
||||
|
||||
return { ...result, unsharedCredentialIds };
|
||||
}
|
||||
|
||||
private async preflight(
|
||||
user: User,
|
||||
workflowIds: string[],
|
||||
scopes: Scope[],
|
||||
options: PreflightOptions,
|
||||
): Promise<WorkflowEntity[]> {
|
||||
const uniqueIds = this.uniqueIds(workflowIds);
|
||||
const accessibleWorkflows = await this.workflowFinderService.findWorkflowsByIdsForUser(
|
||||
uniqueIds,
|
||||
user,
|
||||
scopes,
|
||||
{
|
||||
includeParentFolder: options.includeParentFolder,
|
||||
includeShared: options.includeShared,
|
||||
},
|
||||
);
|
||||
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) {
|
||||
issues.push({
|
||||
workflowId,
|
||||
reason: 'notFoundOrForbidden',
|
||||
message: 'Workflow does not exist or is not accessible.',
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.collaborationService.validateWriteLock(
|
||||
user.id,
|
||||
options.clientId,
|
||||
workflowId,
|
||||
options.action,
|
||||
);
|
||||
} catch (error) {
|
||||
issues.push({
|
||||
workflowId,
|
||||
reason: 'locked',
|
||||
message: ensureError(error).message,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const issue = options.validate?.(workflow);
|
||||
if (issue) issues.push(issue);
|
||||
}
|
||||
|
||||
if (issues.length > 0) throw this.preflightError(issues);
|
||||
|
||||
return uniqueIds.map((id) => workflowById.get(id)).filter((workflow) => workflow !== undefined);
|
||||
}
|
||||
|
||||
private async execute(
|
||||
workflows: WorkflowEntity[],
|
||||
executeItem: ExecuteItem,
|
||||
): 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> => {
|
||||
try {
|
||||
return { workflowId: workflow.id, ...(await executeItem(workflow)) };
|
||||
} catch (error) {
|
||||
return {
|
||||
workflowId: workflow.id,
|
||||
status: 'failed',
|
||||
reason: 'runtimeFailure',
|
||||
message: ensureError(error).message,
|
||||
};
|
||||
}
|
||||
}),
|
||||
);
|
||||
results.push(...batchResults);
|
||||
|
||||
if (batchResults.some(({ status }) => status === 'failed')) {
|
||||
results.push(
|
||||
...workflows.slice(start + batch.length).map(({ id }) => ({
|
||||
workflowId: id,
|
||||
status: 'notAttempted' as const,
|
||||
})),
|
||||
);
|
||||
return { status: 'partial', results };
|
||||
}
|
||||
}
|
||||
|
||||
return { status: 'completed', results };
|
||||
}
|
||||
|
||||
private async getShareableCredentialIds(user: User, credentialIds: string[]): Promise<string[]> {
|
||||
const uniqueIds = this.uniqueIds(credentialIds);
|
||||
if (hasGlobalScope(user, ['credential:share'], { mode: 'allOf' })) return uniqueIds;
|
||||
|
||||
const accessibleIds = new Set(
|
||||
await this.credentialsFinderService.getCredentialIdsByUserAndRole([user.id], {
|
||||
scopes: ['credential:share'],
|
||||
}),
|
||||
);
|
||||
return uniqueIds.filter((id) => accessibleIds.has(id));
|
||||
}
|
||||
|
||||
private uniqueIds(ids: string[]): string[] {
|
||||
return [...new Set(ids)];
|
||||
}
|
||||
|
||||
private preflightError(issues: PreflightIssue[]): UnprocessableRequestError {
|
||||
return new UnprocessableRequestError('Bulk workflow action preflight failed', undefined, {
|
||||
issues,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -181,7 +181,11 @@ export class WorkflowFinderService {
|
|||
workflowIds: string[],
|
||||
user: User,
|
||||
scopes: Scope[],
|
||||
options: { includeParentFolder?: boolean; includeTags?: boolean } = {},
|
||||
options: {
|
||||
includeParentFolder?: boolean;
|
||||
includeTags?: boolean;
|
||||
includeShared?: boolean;
|
||||
} = {},
|
||||
): Promise<WorkflowEntity[]> {
|
||||
if (workflowIds.length === 0) return [];
|
||||
|
||||
|
|
@ -189,7 +193,11 @@ export class WorkflowFinderService {
|
|||
const sharedWorkflows = await this.sharedWorkflowRepository.find({
|
||||
where: { ...where, workflowId: In(workflowIds) },
|
||||
relations: {
|
||||
workflow: { parentFolder: options.includeParentFolder, tags: options.includeTags },
|
||||
workflow: {
|
||||
parentFolder: options.includeParentFolder,
|
||||
tags: options.includeTags,
|
||||
shared: options.includeShared ? { project: true } : false,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
import {
|
||||
ActivateWorkflowDto,
|
||||
ArchiveWorkflowDto,
|
||||
BulkArchiveWorkflowsDto,
|
||||
BulkDeleteWorkflowsDto,
|
||||
BulkTransferWorkflowsDto,
|
||||
BulkUnpublishWorkflowsDto,
|
||||
CreateWorkflowDto,
|
||||
DeactivateWorkflowDto,
|
||||
ExecutionRedactionQueryDtoSchema,
|
||||
|
|
@ -60,6 +64,7 @@ import { getWorkflowProjectDetailsSafe } from '@/workflows/utils';
|
|||
|
||||
import { CollaborationService } from '../collaboration/collaboration.service';
|
||||
import { WorkflowPublicationStatusService } from './publication/workflow-publication-status.service';
|
||||
import { WorkflowBulkActionService } from './workflow-bulk-action.service';
|
||||
import { WorkflowCreationService } from './workflow-creation.service';
|
||||
import { createWorkflowEntityFromPayload } from './workflow-entity-mapper';
|
||||
import { WorkflowExecutionService } from './workflow-execution.service';
|
||||
|
|
@ -94,6 +99,7 @@ export class WorkflowsController {
|
|||
private readonly outboundHttp: OutboundHttp,
|
||||
private readonly workflowPublicationStatusService: WorkflowPublicationStatusService,
|
||||
private readonly ownershipService: OwnershipService,
|
||||
private readonly workflowBulkActionService: WorkflowBulkActionService,
|
||||
) {}
|
||||
|
||||
@Post('/')
|
||||
|
|
@ -347,6 +353,59 @@ export class WorkflowsController {
|
|||
return writeLock;
|
||||
}
|
||||
|
||||
// ProjectScope cannot authorize IDs from several projects in a request body.
|
||||
// WorkflowBulkActionService verifies every requested workflow before mutating any of them.
|
||||
@Post('/bulk/archive')
|
||||
async bulkArchive(
|
||||
req: AuthenticatedRequest,
|
||||
_res: Response,
|
||||
@Body body: BulkArchiveWorkflowsDto,
|
||||
) {
|
||||
return await this.workflowBulkActionService.archive(
|
||||
req.user,
|
||||
body.workflowIds,
|
||||
req.headers['push-ref'] as string | undefined,
|
||||
);
|
||||
}
|
||||
|
||||
@Post('/bulk/delete')
|
||||
async bulkDelete(req: AuthenticatedRequest, _res: Response, @Body body: BulkDeleteWorkflowsDto) {
|
||||
return await this.workflowBulkActionService.delete(
|
||||
req.user,
|
||||
body.workflowIds,
|
||||
req.headers['push-ref'] as string | undefined,
|
||||
);
|
||||
}
|
||||
|
||||
@Post('/bulk/unpublish')
|
||||
async bulkUnpublish(
|
||||
req: AuthenticatedRequest,
|
||||
_res: Response,
|
||||
@Body body: BulkUnpublishWorkflowsDto,
|
||||
) {
|
||||
return await this.workflowBulkActionService.unpublish(
|
||||
req.user,
|
||||
body.workflowIds,
|
||||
req.headers['push-ref'] as string | undefined,
|
||||
);
|
||||
}
|
||||
|
||||
@Post('/bulk/transfer')
|
||||
async bulkTransfer(
|
||||
req: AuthenticatedRequest,
|
||||
_res: Response,
|
||||
@Body body: BulkTransferWorkflowsDto,
|
||||
) {
|
||||
return await this.workflowBulkActionService.transfer(
|
||||
req.user,
|
||||
body.workflowIds,
|
||||
body.destinationProjectId,
|
||||
body.shareCredentials,
|
||||
body.destinationParentFolderId,
|
||||
req.headers['push-ref'] as string | undefined,
|
||||
);
|
||||
}
|
||||
|
||||
@Delete('/:workflowId')
|
||||
@ProjectScope('workflow:delete')
|
||||
async delete(req: AuthenticatedRequest, _res: Response, @Param('workflowId') workflowId: string) {
|
||||
|
|
|
|||
|
|
@ -4873,6 +4873,107 @@ describe('POST /workflows/:workflowId/run', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('workflow bulk actions', () => {
|
||||
test('archives workflows from different projects', async () => {
|
||||
const teamProject = await createTeamProject('Bulk source', owner);
|
||||
const workflows = await Promise.all([
|
||||
createWorkflow({}, owner),
|
||||
createWorkflow({}, teamProject),
|
||||
]);
|
||||
|
||||
const response = await authOwnerAgent
|
||||
.post('/workflows/bulk/archive')
|
||||
.send({ workflowIds: workflows.map(({ id }) => id) })
|
||||
.expect(200);
|
||||
|
||||
expect(response.body.data).toEqual({
|
||||
status: 'completed',
|
||||
results: workflows.map(({ id }) => ({ workflowId: id, status: 'completed' })),
|
||||
});
|
||||
await expect(workflowRepository.findByIds(workflows.map(({ id }) => id))).resolves.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ id: workflows[0].id, isArchived: true }),
|
||||
expect.objectContaining({ id: workflows[1].id, isArchived: true }),
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
test('does not mutate any workflow when strict preflight fails', async () => {
|
||||
const accessibleWorkflow = await createWorkflow({}, owner);
|
||||
const inaccessibleWorkflow = await createWorkflow({}, member);
|
||||
|
||||
const response = await authMemberAgent
|
||||
.post('/workflows/bulk/archive')
|
||||
.send({ workflowIds: [inaccessibleWorkflow.id, accessibleWorkflow.id] })
|
||||
.expect(422);
|
||||
|
||||
expect(response.body.meta.issues).toEqual([
|
||||
expect.objectContaining({
|
||||
workflowId: accessibleWorkflow.id,
|
||||
reason: 'notFoundOrForbidden',
|
||||
}),
|
||||
]);
|
||||
await expect(workflowRepository.findById(inaccessibleWorkflow.id)).resolves.toMatchObject({
|
||||
isArchived: false,
|
||||
});
|
||||
});
|
||||
|
||||
test('unpublishes active workflows and reports inactive workflows as unchanged', async () => {
|
||||
const activeWorkflow = await createActiveWorkflow({}, owner);
|
||||
const inactiveWorkflow = await createWorkflow({}, owner);
|
||||
|
||||
const response = await authOwnerAgent
|
||||
.post('/workflows/bulk/unpublish')
|
||||
.send({ workflowIds: [activeWorkflow.id, inactiveWorkflow.id] })
|
||||
.expect(200);
|
||||
|
||||
expect(response.body.data).toEqual({
|
||||
status: 'completed',
|
||||
results: [
|
||||
{ workflowId: activeWorkflow.id, status: 'completed' },
|
||||
{ workflowId: inactiveWorkflow.id, status: 'unchanged' },
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
test('permanently deletes archived workflows', async () => {
|
||||
const workflows = await Promise.all([
|
||||
createWorkflow({ isArchived: true }, owner),
|
||||
createWorkflow({ isArchived: true }, owner),
|
||||
]);
|
||||
|
||||
const response = await authOwnerAgent
|
||||
.post('/workflows/bulk/delete')
|
||||
.send({ workflowIds: workflows.map(({ id }) => id) })
|
||||
.expect(200);
|
||||
|
||||
expect(response.body.data.status).toBe('completed');
|
||||
await expect(workflowRepository.findByIds(workflows.map(({ id }) => id))).resolves.toHaveLength(
|
||||
0,
|
||||
);
|
||||
});
|
||||
|
||||
test('transfers workflows to one destination project', async () => {
|
||||
const destinationProject = await createTeamProject('Bulk destination', owner);
|
||||
const workflows = await Promise.all([createWorkflow({}, owner), createWorkflow({}, member)]);
|
||||
|
||||
const response = await authOwnerAgent
|
||||
.post('/workflows/bulk/transfer')
|
||||
.send({
|
||||
workflowIds: workflows.map(({ id }) => id),
|
||||
destinationProjectId: destinationProject.id,
|
||||
})
|
||||
.expect(200);
|
||||
|
||||
expect(response.body.data.status).toBe('completed');
|
||||
const sharings = await Container.get(SharedWorkflowRepository).findByWorkflowIds(
|
||||
workflows.map(({ id }) => id),
|
||||
);
|
||||
expect(sharings).toHaveLength(2);
|
||||
expect(sharings.every(({ projectId }) => projectId === destinationProject.id)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('POST /workflows/:workflowId/archive', () => {
|
||||
test('should archive workflow', async () => {
|
||||
const workflow = await createWorkflow({}, owner);
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user