mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-19 16:55:02 +02:00
852 lines
25 KiB
TypeScript
852 lines
25 KiB
TypeScript
import type { DeleteExecutionsDto } from '@n8n/api-types';
|
|
import { ExecutionRedactionQueryDtoSchema } from '@n8n/api-types';
|
|
import { Logger } from '@n8n/backend-common';
|
|
import { GlobalConfig } from '@n8n/config';
|
|
import type {
|
|
CreateExecutionPayload,
|
|
ExecutionSummaries,
|
|
IExecutionBase,
|
|
IExecutionResponse,
|
|
IGetExecutionsQueryFilter,
|
|
User,
|
|
} from '@n8n/db';
|
|
import {
|
|
AnnotationTagMappingRepository,
|
|
ExecutionAnnotationRepository,
|
|
ExecutionRepository,
|
|
In,
|
|
WorkflowHistoryRepository,
|
|
WorkflowRepository,
|
|
} from '@n8n/db';
|
|
import { Service } from '@n8n/di';
|
|
import { QueryFailedError } from '@n8n/typeorm';
|
|
import { ensureError } from '@n8n/utils/errors/ensure-error';
|
|
import { stringify } from 'flatted';
|
|
import { validate as jsonSchemaValidate } from 'jsonschema';
|
|
import type {
|
|
ExecutionError,
|
|
ExecutionStatus,
|
|
INode,
|
|
IWorkflowBase,
|
|
IWorkflowExecutionDataProcess,
|
|
WorkflowExecuteMode,
|
|
} from 'n8n-workflow';
|
|
import {
|
|
ManualExecutionCancelledError,
|
|
UnexpectedError,
|
|
UserError,
|
|
Workflow,
|
|
WorkflowOperationError,
|
|
createEmptyRunExecutionData,
|
|
createErrorExecutionData,
|
|
} from 'n8n-workflow';
|
|
|
|
import { ActiveExecutions } from '@/active-executions';
|
|
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
|
|
import { AbortedExecutionRetryError } from '@/errors/aborted-execution-retry.error';
|
|
import { MissingExecutionStopError } from '@/errors/missing-execution-stop.error';
|
|
import { QueuedExecutionRetryError } from '@/errors/queued-execution-retry.error';
|
|
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
|
|
import { ConflictError } from '@/errors/response-errors/conflict.error';
|
|
import { InternalServerError } from '@/errors/response-errors/internal-server.error';
|
|
import { NotFoundError } from '@/errors/response-errors/not-found.error';
|
|
import { EventService } from '@/events/event.service';
|
|
import type { IExecutionFlattedResponse } from '@/interfaces';
|
|
import { License } from '@/license';
|
|
import { NodeTypes } from '@/node-types';
|
|
import { ExecutionStopService } from '@/scaling/execution-stop.service';
|
|
import { OwnershipService } from '@/services/ownership.service';
|
|
import { WaitTracker } from '@/wait-tracker';
|
|
import { WorkflowRunner } from '@/workflow-runner';
|
|
import { getWorkflowProjectDetailsSafe } from '@/workflows/utils';
|
|
|
|
import { EngineV2ExecutionReader } from './engine-v2-execution-reader.service';
|
|
import { MissingExecutionDataError } from './execution-data/missing-execution-data.error';
|
|
import { isExecutionIdV2 } from './execution-id';
|
|
import { ExecutionPersistence } from './execution-persistence';
|
|
import { ExecutionRedactionServiceProxy } from './execution-redaction-proxy.service';
|
|
import type { ExecutionRequest, StopResult } from './execution.types';
|
|
|
|
export const schemaGetExecutionsQueryFilter = {
|
|
$id: '/IGetExecutionsQueryFilter',
|
|
type: 'object',
|
|
properties: {
|
|
mode: { type: 'string' },
|
|
status: {
|
|
type: 'array',
|
|
items: { type: 'string' },
|
|
},
|
|
workflowId: { anyOf: [{ type: 'integer' }, { type: 'string' }] },
|
|
metadata: { type: 'array', items: { $ref: '#/$defs/metadata' } },
|
|
startedAfter: { type: 'date-time' },
|
|
startedBefore: { type: 'date-time' },
|
|
annotationTags: { type: 'array', items: { type: 'string' } },
|
|
vote: { type: 'string' },
|
|
projectId: { type: 'string' },
|
|
workflowVersionId: { type: 'string' },
|
|
},
|
|
$defs: {
|
|
metadata: {
|
|
type: 'object',
|
|
required: ['key', 'value'],
|
|
properties: {
|
|
key: {
|
|
type: 'string',
|
|
},
|
|
value: { type: 'string' },
|
|
exactMatch: {
|
|
type: 'boolean',
|
|
default: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
};
|
|
|
|
export const allowedExecutionsQueryFilterFields = Object.keys(
|
|
schemaGetExecutionsQueryFilter.properties,
|
|
);
|
|
|
|
@Service()
|
|
export class ExecutionService {
|
|
constructor(
|
|
private readonly globalConfig: GlobalConfig,
|
|
private readonly logger: Logger,
|
|
private readonly activeExecutions: ActiveExecutions,
|
|
private readonly executionAnnotationRepository: ExecutionAnnotationRepository,
|
|
private readonly annotationTagMappingRepository: AnnotationTagMappingRepository,
|
|
private readonly executionRepository: ExecutionRepository,
|
|
private readonly executionPersistence: ExecutionPersistence,
|
|
private readonly workflowHistoryRepository: WorkflowHistoryRepository,
|
|
private readonly workflowRepository: WorkflowRepository,
|
|
private readonly nodeTypes: NodeTypes,
|
|
private readonly waitTracker: WaitTracker,
|
|
private readonly workflowRunner: WorkflowRunner,
|
|
private readonly concurrencyControl: ConcurrencyControlService,
|
|
private readonly license: License,
|
|
private readonly eventService: EventService,
|
|
private readonly executionRedactionServiceProxy: ExecutionRedactionServiceProxy,
|
|
private readonly executionStopService: ExecutionStopService,
|
|
private readonly ownershipService: OwnershipService,
|
|
private readonly engineV2ExecutionReader: EngineV2ExecutionReader,
|
|
) {}
|
|
|
|
/**
|
|
* Editor/internal GET: load an execution for display, apply redaction, and
|
|
* return flatted `data` (`IExecutionFlattedResponse`).
|
|
*
|
|
* Prefer this for the private executions API. For a domain entity with
|
|
* caller-controlled options, use {@link findOneInWorkflows}.
|
|
*/
|
|
async findOne(
|
|
req: ExecutionRequest.GetOne | ExecutionRequest.Update,
|
|
sharedWorkflowIds: string[],
|
|
): Promise<IExecutionFlattedResponse | undefined> {
|
|
if (!sharedWorkflowIds.length) return undefined;
|
|
|
|
const { id: executionId } = req.params;
|
|
let execution: IExecutionResponse | IExecutionBase | undefined;
|
|
// A v2 execution has no control-plane row.
|
|
if (isExecutionIdV2(executionId)) {
|
|
execution = await this.engineV2ExecutionReader.findOne(executionId, sharedWorkflowIds);
|
|
} else {
|
|
try {
|
|
execution = await this.executionPersistence.findOneInWorkflows(
|
|
executionId,
|
|
sharedWorkflowIds,
|
|
{ maxDataSizeBytes: this.globalConfig.executions.maxDisplaySize },
|
|
);
|
|
} catch (error) {
|
|
if (error instanceof MissingExecutionDataError) {
|
|
throw new NotFoundError(
|
|
'Data for this execution is unavailable. It may have already been deleted based on your data retention settings.',
|
|
);
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
if (!execution) {
|
|
this.logger.info('Attempt to read execution was blocked due to insufficient permissions', {
|
|
userId: req.user.id,
|
|
executionId,
|
|
});
|
|
return undefined;
|
|
}
|
|
|
|
if (!('data' in execution)) {
|
|
throw new UnexpectedError('Expected execution data for display read');
|
|
}
|
|
|
|
let redactExecutionData: boolean | undefined;
|
|
const redactQuery = ExecutionRedactionQueryDtoSchema.safeParse(req.query);
|
|
if (redactQuery.success) {
|
|
redactExecutionData = redactQuery.data.redactExecutionData;
|
|
}
|
|
|
|
const processedExecution = await this.executionRedactionServiceProxy.processExecution(
|
|
execution,
|
|
{
|
|
user: req.user,
|
|
redactExecutionData,
|
|
ipAddress: req.ip ?? '',
|
|
userAgent: req.headers['user-agent'] ?? '',
|
|
},
|
|
);
|
|
|
|
return {
|
|
...execution,
|
|
data: stringify(processedExecution.data),
|
|
dataTooLargeToDisplay: execution.dataTooLargeToDisplay,
|
|
};
|
|
}
|
|
|
|
async getLastSuccessfulExecution(
|
|
workflowId: string,
|
|
user: User,
|
|
redactExecutionData?: boolean,
|
|
): Promise<IExecutionResponse | undefined> {
|
|
const executions = await this.executionPersistence.findMultipleExecutions(
|
|
{
|
|
select: ['id', 'mode', 'startedAt', 'stoppedAt', 'workflowId', 'jsonSizeBytes'],
|
|
where: {
|
|
workflowId,
|
|
status: 'success',
|
|
},
|
|
order: { id: 'DESC' },
|
|
take: 1,
|
|
},
|
|
{
|
|
includeData: true,
|
|
unflattenData: true,
|
|
maxDataSizeBytes: this.globalConfig.executions.maxDisplaySize,
|
|
},
|
|
);
|
|
|
|
const execution = executions[0];
|
|
if (!execution) return undefined;
|
|
|
|
await this.executionRedactionServiceProxy.processExecution(execution, {
|
|
user,
|
|
redactExecutionData,
|
|
});
|
|
return execution;
|
|
}
|
|
|
|
async retry({
|
|
executionId,
|
|
options = {},
|
|
sharedWorkflowIds,
|
|
user,
|
|
}: {
|
|
executionId: string;
|
|
options?: { loadWorkflow?: boolean; redactExecutionData?: boolean };
|
|
sharedWorkflowIds: string[];
|
|
user: User;
|
|
}): Promise<Omit<IExecutionResponse, 'createdAt'>> {
|
|
const execution = await this.executionPersistence.findWithUnflattenedData(
|
|
executionId,
|
|
sharedWorkflowIds,
|
|
);
|
|
|
|
if (!execution) {
|
|
this.logger.info(
|
|
'Attempt to retry an execution was blocked due to insufficient permissions',
|
|
{
|
|
userId: user.id,
|
|
executionId,
|
|
},
|
|
);
|
|
throw new NotFoundError(`The execution with the ID "${executionId}" does not exist.`);
|
|
}
|
|
|
|
if (execution.status === 'new') throw new QueuedExecutionRetryError();
|
|
|
|
if (!execution.data.executionData) throw new AbortedExecutionRetryError();
|
|
|
|
if (execution.finished) {
|
|
throw new ConflictError('The execution succeeded, so it cannot be retried.');
|
|
}
|
|
|
|
const executionMode = 'retry';
|
|
|
|
execution.workflowData.active = false;
|
|
execution.workflowData.activeVersionId = null;
|
|
|
|
// Start the workflow
|
|
const data: IWorkflowExecutionDataProcess = {
|
|
executionMode,
|
|
executionData: execution.data,
|
|
retryOf: executionId,
|
|
workflowData: execution.workflowData,
|
|
userId: user.id,
|
|
};
|
|
|
|
const { lastNodeExecuted } = data.executionData!.resultData;
|
|
|
|
if (lastNodeExecuted) {
|
|
// Remove the old error and the data of the last run of the node that it can be replaced
|
|
delete data.executionData!.resultData.error;
|
|
const nodeRunData = data.executionData!.resultData.runData?.[lastNodeExecuted];
|
|
if (
|
|
nodeRunData &&
|
|
nodeRunData.length > 0 &&
|
|
nodeRunData[nodeRunData.length - 1].error !== undefined
|
|
) {
|
|
// Remove results only if it is an error.
|
|
// If we are retrying due to a crash, the information is simply success info from last node
|
|
nodeRunData.pop();
|
|
// Stack will determine what to run next
|
|
}
|
|
}
|
|
|
|
if (options.loadWorkflow) {
|
|
// Loads the currently saved workflow to execute instead of the
|
|
// one saved at the time of the execution.
|
|
const workflowId = execution.workflowData.id;
|
|
const workflowData = (await this.workflowRepository.findOneBy({
|
|
id: workflowId,
|
|
})) as IWorkflowBase;
|
|
|
|
if (workflowData === undefined) {
|
|
throw new UserError(
|
|
'Workflow could not be found and so the data not be loaded for the retry.',
|
|
{ extra: { workflowId } },
|
|
);
|
|
}
|
|
|
|
data.workflowData = workflowData;
|
|
|
|
const workflowInstance = new Workflow({
|
|
id: workflowData.id,
|
|
name: workflowData.name,
|
|
nodes: workflowData.nodes,
|
|
connections: workflowData.connections,
|
|
active: false,
|
|
nodeTypes: this.nodeTypes,
|
|
staticData: undefined,
|
|
settings: workflowData.settings,
|
|
});
|
|
|
|
// Replace all of the nodes in the execution stack with the ones of the new workflow
|
|
for (const stack of data.executionData!.executionData!.nodeExecutionStack) {
|
|
// Find the data of the last executed node in the new workflow
|
|
const node = workflowInstance.getNode(stack.node.name);
|
|
if (node === null) {
|
|
this.logger.error('Failed to retry an execution because a node could not be found', {
|
|
userId: user.id,
|
|
executionId,
|
|
nodeName: stack.node.name,
|
|
});
|
|
throw new WorkflowOperationError(
|
|
`Could not find the node "${stack.node.name}" in workflow. It probably got deleted or renamed. Without it the workflow can sadly not be retried.`,
|
|
);
|
|
}
|
|
|
|
// Replace the node data in the stack that it really uses the current data
|
|
stack.node = node;
|
|
}
|
|
}
|
|
|
|
const retriedExecutionId = await this.workflowRunner.run(data);
|
|
|
|
const executionData = await this.activeExecutions.getPostExecutePromise(retriedExecutionId);
|
|
|
|
if (!executionData) {
|
|
throw new UnexpectedError('The retry did not start for an unknown reason.');
|
|
}
|
|
|
|
const { projectId, projectName } = await getWorkflowProjectDetailsSafe(
|
|
this.ownershipService,
|
|
execution.workflowId,
|
|
);
|
|
|
|
this.eventService.emit('workflow-executed', {
|
|
user: {
|
|
id: user.id,
|
|
email: user.email,
|
|
firstName: user.firstName,
|
|
lastName: user.lastName,
|
|
role: user.role,
|
|
},
|
|
workflowId: execution.workflowId,
|
|
workflowName: execution.workflowData.name,
|
|
executionId: retriedExecutionId,
|
|
projectId,
|
|
projectName,
|
|
source: 'user-retry',
|
|
});
|
|
|
|
const response: Omit<IExecutionResponse, 'createdAt'> = {
|
|
id: retriedExecutionId,
|
|
mode: executionData.mode,
|
|
startedAt: executionData.startedAt,
|
|
workflowId: execution.workflowId,
|
|
finished: executionData.finished ?? false,
|
|
retryOf: executionId,
|
|
status: executionData.status,
|
|
waitTill: executionData.waitTill,
|
|
data: executionData.data,
|
|
workflowData: execution.workflowData,
|
|
customData: execution.customData,
|
|
annotation: execution.annotation,
|
|
storedAt: execution.storedAt,
|
|
};
|
|
|
|
await this.executionRedactionServiceProxy.processExecution(response, {
|
|
user,
|
|
redactExecutionData: options.redactExecutionData,
|
|
});
|
|
|
|
return response;
|
|
}
|
|
|
|
async delete(user: User, payload: DeleteExecutionsDto, sharedWorkflowIds: string[]) {
|
|
const { deleteBefore, ids, filters: requestFiltersRaw } = payload;
|
|
|
|
let requestFilters: IGetExecutionsQueryFilter | undefined;
|
|
if (requestFiltersRaw) {
|
|
try {
|
|
Object.keys(requestFiltersRaw).map((key) => {
|
|
if (!allowedExecutionsQueryFilterFields.includes(key)) delete requestFiltersRaw[key];
|
|
});
|
|
if (jsonSchemaValidate(requestFiltersRaw, schemaGetExecutionsQueryFilter).valid) {
|
|
requestFilters = requestFiltersRaw;
|
|
}
|
|
} catch (error) {
|
|
throw new InternalServerError('Parameter "filter" contained invalid JSON string.', error);
|
|
}
|
|
}
|
|
|
|
if (requestFilters?.metadata && !this.license.isAdvancedExecutionFiltersEnabled()) {
|
|
delete requestFilters.metadata;
|
|
}
|
|
|
|
await this.executionPersistence.hardDeleteBy({
|
|
filters: requestFilters,
|
|
accessibleWorkflowIds: sharedWorkflowIds,
|
|
deleteConditions: { deleteBefore, ids },
|
|
});
|
|
|
|
this.eventService.emit('execution-deleted', {
|
|
user: {
|
|
id: user.id,
|
|
email: user.email,
|
|
firstName: user.firstName,
|
|
lastName: user.lastName,
|
|
role: user.role,
|
|
},
|
|
executionIds: ids ?? [],
|
|
deleteBefore,
|
|
});
|
|
}
|
|
|
|
async createErrorExecution(
|
|
error: ExecutionError,
|
|
node: INode,
|
|
workflowData: IWorkflowBase,
|
|
workflow: Workflow,
|
|
mode: WorkflowExecuteMode,
|
|
) {
|
|
const saveDataErrorExecutionDisabled =
|
|
workflowData?.settings?.saveDataErrorExecution === 'none';
|
|
|
|
if (saveDataErrorExecutionDisabled) return;
|
|
|
|
const executionData = createErrorExecutionData(node, error);
|
|
|
|
const fullExecutionData: CreateExecutionPayload = {
|
|
data: executionData,
|
|
mode,
|
|
finished: false,
|
|
workflowData,
|
|
workflowId: workflow.id,
|
|
stoppedAt: new Date(),
|
|
status: 'error',
|
|
};
|
|
|
|
await this.executionPersistence.create(fullExecutionData);
|
|
}
|
|
|
|
// ----------------------------------
|
|
// new API
|
|
// ----------------------------------
|
|
|
|
/**
|
|
* @returns
|
|
* - the number of concurrent executions
|
|
* - `-1` if the count is not applicable (e.g. in 'queue' mode or if concurrency control is disabled)
|
|
*
|
|
* In 'queue' mode, concurrency control is applied per worker, so returning a global count of concurrent executions
|
|
* would not be meaningful or helpful.
|
|
*/
|
|
async getConcurrentExecutionsCount() {
|
|
if (!this.isConcurrentExecutionsCountSupported()) {
|
|
return -1;
|
|
}
|
|
|
|
return await this.executionRepository.getConcurrentExecutionsCount();
|
|
}
|
|
|
|
private isConcurrentExecutionsCountSupported(): boolean {
|
|
const isConcurrencyEnabled = this.globalConfig.executions.concurrency.productionLimit !== -1;
|
|
const isInRegularMode = this.globalConfig.executions.mode === 'regular';
|
|
|
|
if (!isConcurrencyEnabled || !isInRegularMode) {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* All executions still enqueued (`new`), plus the ids of those whose data could not be
|
|
* read - those can never run, so the caller has to take them out of `new` itself.
|
|
*/
|
|
async findAllEnqueuedExecutions() {
|
|
return await this.executionPersistence.findMultipleExecutionsWithUnreadable({
|
|
select: ['id', 'mode'],
|
|
where: { status: 'new' },
|
|
order: { id: 'ASC' },
|
|
});
|
|
}
|
|
|
|
async stop(executionId: string, sharedWorkflowIds: string[]): Promise<StopResult> {
|
|
const execution = await this.executionPersistence.findWithUnflattenedData(
|
|
executionId,
|
|
sharedWorkflowIds,
|
|
);
|
|
|
|
if (!execution) {
|
|
this.logger.info(
|
|
`Unable to stop execution "${executionId}" as it was not found or not accessible`,
|
|
{
|
|
executionId,
|
|
},
|
|
);
|
|
|
|
throw new MissingExecutionStopError(executionId);
|
|
}
|
|
|
|
this.assertStoppable(execution);
|
|
|
|
const { mode, startedAt, stoppedAt, finished, status } =
|
|
this.globalConfig.executions.mode === 'regular'
|
|
? await this.stopInRegularMode(execution)
|
|
: await this.stopInScalingMode(execution);
|
|
|
|
return {
|
|
mode,
|
|
startedAt: new Date(startedAt),
|
|
stoppedAt: stoppedAt ? new Date(stoppedAt) : undefined,
|
|
finished,
|
|
status,
|
|
};
|
|
}
|
|
|
|
async stopMany(query: ExecutionSummaries.StopExecutionFilterQuery, sharedWorkflowIds: string[]) {
|
|
const executions = await this.executionRepository.findByStopExecutionsFilter(query);
|
|
let stopped = 0;
|
|
for (const { id } of executions) {
|
|
try {
|
|
await this.stop(id, sharedWorkflowIds);
|
|
this.logger.debug(`Stopped execution ${id}`);
|
|
stopped++;
|
|
} catch (e) {
|
|
// the throwing code already logs the failure otherwise
|
|
if (!(e instanceof MissingExecutionStopError)) {
|
|
this.logger.warn(
|
|
`Unexpected error while attempting to stop execution ${id}: ${ensureError(e).message}`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
return stopped;
|
|
}
|
|
|
|
private assertStoppable(execution: IExecutionResponse) {
|
|
const STOPPABLE_STATUSES: ExecutionStatus[] = ['new', 'unknown', 'waiting', 'running'];
|
|
|
|
if (!STOPPABLE_STATUSES.includes(execution.status)) {
|
|
throw new WorkflowOperationError(
|
|
`Only running or waiting executions can be stopped and ${execution.id} is currently ${execution.status}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
private async stopInRegularMode(execution: IExecutionResponse) {
|
|
if (this.concurrencyControl.has(execution.id)) {
|
|
this.concurrencyControl.remove({ mode: execution.mode, executionId: execution.id });
|
|
return await this.executionRepository.stopBeforeRun(execution);
|
|
}
|
|
|
|
if (this.activeExecutions.has(execution.id)) {
|
|
this.activeExecutions.stopExecution(
|
|
execution.id,
|
|
new ManualExecutionCancelledError(execution.id),
|
|
);
|
|
}
|
|
|
|
if (this.waitTracker.has(execution.id)) {
|
|
this.waitTracker.stopExecution(execution.id);
|
|
}
|
|
|
|
return await this.stopDuringRun(execution);
|
|
}
|
|
|
|
private async stopInScalingMode(execution: IExecutionResponse) {
|
|
if (this.activeExecutions.has(execution.id)) {
|
|
this.activeExecutions.stopExecution(
|
|
execution.id,
|
|
new ManualExecutionCancelledError(execution.id),
|
|
);
|
|
}
|
|
|
|
if (this.waitTracker.has(execution.id)) {
|
|
this.waitTracker.stopExecution(execution.id);
|
|
}
|
|
|
|
// Broadcast a stop to whichever worker is running this execution; it cancels the execution
|
|
// from its own ActiveExecutions, and workers not running it ignore the command. This is the
|
|
// only way to reach a subworkflow execution, which runs inline in the parent's worker process
|
|
// and has no Bull job to abort. For a top-level execution this is redundant with the
|
|
// abort-job triggered above via ActiveExecutions, but both paths are idempotent.
|
|
await this.executionStopService.requestStop(execution.id);
|
|
|
|
return await this.stopDuringRun(execution);
|
|
}
|
|
|
|
private async stopDuringRun(execution: IExecutionResponse) {
|
|
const error = new ManualExecutionCancelledError(execution.id);
|
|
|
|
execution.data = execution.data ?? createEmptyRunExecutionData();
|
|
execution.data.resultData.error = {
|
|
...error,
|
|
message: error.message,
|
|
stack: error.stack,
|
|
};
|
|
execution.stoppedAt = new Date();
|
|
execution.waitTill = null;
|
|
execution.status = 'canceled';
|
|
|
|
await this.executionPersistence.updateExistingExecution(execution.id, execution);
|
|
|
|
return execution;
|
|
}
|
|
|
|
async annotate(
|
|
executionId: string,
|
|
updateData: ExecutionRequest.ExecutionUpdatePayload,
|
|
sharedWorkflowIds: string[],
|
|
) {
|
|
// Check if user can access the execution
|
|
const execution = await this.executionRepository.findIfAccessible(
|
|
executionId,
|
|
sharedWorkflowIds,
|
|
);
|
|
|
|
if (!execution) {
|
|
this.logger.info('Attempt to read execution was blocked due to insufficient permissions', {
|
|
executionId,
|
|
});
|
|
|
|
throw new NotFoundError('Execution not found');
|
|
}
|
|
|
|
// Create or update execution annotation
|
|
await this.executionAnnotationRepository.upsert(
|
|
{ execution: { id: executionId }, vote: updateData.vote },
|
|
['execution'],
|
|
);
|
|
|
|
// Upsert behavior differs for Postgres and sqlite,
|
|
// so we need to fetch the annotation to get the ID
|
|
const annotation = await this.executionAnnotationRepository.findOneOrFail({
|
|
where: {
|
|
execution: { id: executionId },
|
|
},
|
|
});
|
|
|
|
if (updateData.tags) {
|
|
await this.annotationTagMappingRepository.overwriteTags(annotation.id, updateData.tags);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Load one execution scoped to `workflowIds` as a domain entity.
|
|
* Options control data/annotation inclusion; no redaction or flatting.
|
|
*
|
|
* Prefer this for public API and service-to-service loads. For the editor
|
|
* flatted response, use {@link findOne}.
|
|
*/
|
|
async findOneInWorkflows(
|
|
executionId: string,
|
|
workflowIds: string[],
|
|
options?: {
|
|
includeData?: boolean;
|
|
includeAnnotation?: boolean;
|
|
maxDataSizeBytes?: number;
|
|
},
|
|
) {
|
|
return await this.executionPersistence.findOneInWorkflows(executionId, workflowIds, options);
|
|
}
|
|
|
|
async findManyAndCount(
|
|
workflowIds: string[],
|
|
options: {
|
|
limit: number;
|
|
includeData?: boolean;
|
|
lastId?: string;
|
|
status?: ExecutionStatus;
|
|
excludeRunning?: boolean;
|
|
maxDataSizeBytes?: number;
|
|
startedAfter?: string;
|
|
startedBefore?: string;
|
|
},
|
|
): Promise<{ executions: IExecutionBase[]; count: number }> {
|
|
const excludedExecutionsIds = options.excludeRunning
|
|
? this.activeExecutions
|
|
.getActiveExecutions()
|
|
.filter(({ status }) => status === 'running')
|
|
.map(({ id }) => id)
|
|
: undefined;
|
|
|
|
const listOptions = {
|
|
limit: options.limit,
|
|
includeData: options.includeData,
|
|
lastId: options.lastId,
|
|
status: options.status,
|
|
excludedExecutionsIds,
|
|
startedAfter: options.startedAfter,
|
|
startedBefore: options.startedBefore,
|
|
};
|
|
|
|
const executions = await this.executionPersistence.findManyInWorkflows(
|
|
workflowIds,
|
|
listOptions,
|
|
options.maxDataSizeBytes,
|
|
);
|
|
|
|
const newLastId = executions.length === 0 ? '0' : executions.at(-1)!.id;
|
|
const count = await this.executionRepository.countInWorkflows(workflowIds, {
|
|
...listOptions,
|
|
lastId: newLastId,
|
|
});
|
|
|
|
return { executions, count };
|
|
}
|
|
|
|
async deleteOne(executionId: string, sharedWorkflowIds: string[]) {
|
|
const execution = await this.findOneInWorkflows(executionId, sharedWorkflowIds, {
|
|
includeData: false,
|
|
includeAnnotation: false,
|
|
});
|
|
|
|
if (!execution) {
|
|
throw new NotFoundError('Not Found');
|
|
}
|
|
|
|
if (execution.status === 'running') {
|
|
throw new BadRequestError('Cannot delete a running execution');
|
|
}
|
|
|
|
if (execution.status === 'new') {
|
|
this.concurrencyControl.remove({
|
|
executionId: execution.id,
|
|
mode: execution.mode,
|
|
});
|
|
}
|
|
|
|
await this.executionPersistence.hardDelete({
|
|
workflowId: execution.workflowId,
|
|
executionId: execution.id,
|
|
storedAt: execution.storedAt,
|
|
});
|
|
|
|
return execution;
|
|
}
|
|
|
|
async getExecutionTags(executionId: string, sharedWorkflowIds: string[]) {
|
|
const execution = await this.findOneInWorkflows(executionId, sharedWorkflowIds, {
|
|
includeData: false,
|
|
includeAnnotation: false,
|
|
});
|
|
|
|
if (!execution) {
|
|
throw new NotFoundError('Not Found');
|
|
}
|
|
|
|
const annotation = await this.executionAnnotationRepository.findOne({
|
|
where: { execution: { id: executionId } },
|
|
relations: ['tags'],
|
|
});
|
|
|
|
return (annotation?.tags ?? []).map(({ id, name, createdAt, updatedAt }) => ({
|
|
id,
|
|
name,
|
|
createdAt,
|
|
updatedAt,
|
|
}));
|
|
}
|
|
|
|
async updateExecutionTags(executionId: string, tagIds: string[], sharedWorkflowIds: string[]) {
|
|
const execution = await this.findOneInWorkflows(executionId, sharedWorkflowIds, {
|
|
includeData: false,
|
|
includeAnnotation: false,
|
|
});
|
|
|
|
if (!execution) {
|
|
throw new NotFoundError('Not Found');
|
|
}
|
|
|
|
await this.executionAnnotationRepository.upsert({ execution: { id: executionId } }, [
|
|
'execution',
|
|
]);
|
|
|
|
const annotation = await this.executionAnnotationRepository.findOneOrFail({
|
|
where: { execution: { id: executionId } },
|
|
});
|
|
|
|
try {
|
|
await this.annotationTagMappingRepository.overwriteTags(annotation.id, tagIds);
|
|
} catch (error) {
|
|
if (error instanceof QueryFailedError) {
|
|
throw new NotFoundError('Some tags not found');
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
const updatedAnnotation = await this.executionAnnotationRepository.findOneOrFail({
|
|
where: { execution: { id: executionId } },
|
|
relations: ['tags'],
|
|
});
|
|
|
|
return (updatedAnnotation.tags ?? []).map(({ id, name, createdAt, updatedAt }) => ({
|
|
id,
|
|
name,
|
|
createdAt,
|
|
updatedAt,
|
|
}));
|
|
}
|
|
|
|
async getExecutedVersions(
|
|
workflowId: string,
|
|
): Promise<Array<{ versionId: string; name: string | null; createdAt: Date }>> {
|
|
const versionIds = await this.executionRepository.getDistinctVersionIds(workflowId);
|
|
if (versionIds.length === 0) return [];
|
|
|
|
const versions = await this.workflowHistoryRepository.find({
|
|
where: { workflowId, versionId: In(versionIds) },
|
|
select: ['versionId', 'name', 'createdAt'],
|
|
order: { createdAt: 'DESC' },
|
|
});
|
|
|
|
return versions.map((v) => ({
|
|
versionId: v.versionId,
|
|
name: v.name,
|
|
createdAt: v.createdAt,
|
|
}));
|
|
}
|
|
}
|