n8n/packages/cli/src/commands/execute.ts

import { WorkflowRepository } from '@n8n/db';
import { Command } from '@n8n/decorators';
import { Container } from '@n8n/di';
import type { IWorkflowBase, IWorkflowExecutionDataProcess } from 'n8n-workflow';
import { ExecutionBaseError, UnexpectedError, UserError } from 'n8n-workflow';
import { z } from 'zod';

import { ActiveExecutions } from '@/active-executions';
import { EventService } from '@/events/event.service';
import { OwnershipService } from '@/services/ownership.service';
import { findCliWorkflowStart, isWorkflowIdValid } from '@/utils';
import { WorkflowRunner } from '@/workflow-runner';

import { BaseCommand } from './base-command';

const flagsSchema = z.object({
	id: z.string().describe('id of the workflow to execute').optional(),
	rawOutput: z.boolean().describe('Outputs only JSON data, with no other text').optional(),
	/**@deprecated */
	file: z.string().describe('DEPRECATED: Please use --id instead').optional(),
});

@Command({
	name: 'execute',
	description: 'Executes a given workflow',
	examples: ['--id=5'],
	flagsSchema,
})
export class Execute extends BaseCommand<z.infer<typeof flagsSchema>> {
	override needsCommunityPackages = true;

	override needsTaskRunner = true;

	async init() {
		await super.init();
		await this.initLicense();
		await this.initCommunityPackages();
		await this.initBinaryDataService();
		await this.initDataDeduplicationService();
		await this.initExternalHooks();
	}

	async run() {
		const { flags } = this;

		if (!flags.id) {
			this.logger.info('"--id" has to be set!');
			return;
		}

		if (flags.file) {
			throw new UserError(
				'The --file flag is no longer supported. Please first import the workflow and then execute it using the --id flag.',
				{ level: 'warning' },
			);
		}

		let workflowId: string | undefined;
		let workflowData: IWorkflowBase | null = null;

		if (flags.id) {
			// Id of workflow is given
			workflowId = flags.id;
			workflowData = await Container.get(WorkflowRepository).findOneBy({ id: workflowId });
			if (workflowData === null) {
				this.logger.info(`The workflow with the id "${workflowId}" does not exist.`);
				process.exit(1);
			}
		}

		if (!workflowData) {
			throw new UnexpectedError('Failed to retrieve workflow data for requested workflow');
		}

		if (!isWorkflowIdValid(workflowId)) {
			workflowId = undefined;
		}

		const startingNode = findCliWorkflowStart(workflowData.nodes);

		const user = await Container.get(OwnershipService).getInstanceOwner();
		const runData: IWorkflowExecutionDataProcess = {
			executionMode: 'cli',
			startNodes: [{ name: startingNode.name, sourceData: null }],
			workflowData,
			userId: user.id,
		};

		const workflowRunner = Container.get(WorkflowRunner);

		if (this.globalConfig.executions.mode === 'queue') {
			this.logger.warn(
				'CLI command `execute` does not support queue mode. Falling back to regular mode.',
			);
			this.globalConfig.executions.mode = 'regular';
		}

		const executionId = await workflowRunner.run(runData);

		Container.get(EventService).emit('workflow-executed', {
			user: { id: user.id },
			workflowId: workflowData.id,
			workflowName: workflowData.name,
			executionId,
			source: 'cli',
		});

		const activeExecutions = Container.get(ActiveExecutions);
		const data = await activeExecutions.getPostExecutePromise(executionId);

		if (data === undefined) {
			throw new UnexpectedError('Workflow did not return any data');
		}

		if (data.data.resultData.error) {
			this.logger.info('Execution was NOT successful. See log message for details.');
			this.logger.info('Execution error:');
			this.logger.info('====================================');
			this.logger.info(JSON.stringify(data, null, 2));

			const { error } = data.data.resultData;
			// eslint-disable-next-line @typescript-eslint/only-throw-error
			throw {
				...error,
				stack: error.stack,
			};
		}
		if (flags.rawOutput === undefined) {
			this.log('Execution was successful:');
			this.log('====================================');
		}
		this.log(JSON.stringify(data, null, 2));
	}

	async catch(error: Error) {
		this.logger.error('Error executing workflow. See log messages for details.');
		this.logger.error('\nExecution error:');
		this.logger.info('====================================');
		this.logger.error(error.message);
		if (error instanceof ExecutionBaseError) this.logger.error(error.description!);
		this.logger.error(error.stack!);
	}
}