n8n/packages/cli/src/scheduling/schedule-trigger-node/schedule-trigger-task-handler.ts

import { Logger } from '@n8n/backend-common';
import { EventService, OwnershipService } from '@n8n/backend-services';
import { GlobalConfig } from '@n8n/config';
import { ExecutionRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler';
import { ErrorReporter } from 'n8n-core';
import { CRON_NODE_TYPE, UnexpectedError } from 'n8n-workflow';

import { DuplicateExecutionError } from '@/errors/duplicate-execution.error';
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
import { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory';
import { getWorkflowProjectDetailsSafe } from '@/workflows/utils';
import { WorkflowExecutionService } from '@/workflows/workflow-execution.service';
import { WorkflowPublisherService } from '@/workflows/workflow-publisher.service';

import { resolveTaskTriggerNode } from '../resolve-task-trigger-node';
import {
	buildScheduleTriggerItem,
	isScheduleTriggerTaskPayload,
	SCHEDULE_TRIGGER_TASK_TYPE,
	scheduleTriggerDeduplicationKey,
	type ScheduleTriggerTaskPayload,
} from './schedule-trigger-task';

/**
 * Turns a due schedule-trigger occurrence into a real workflow execution.
 *
 * The handoff is the whole job: build the trigger item, create the execution
 * with the occurrence-derived dedup key, and return.
 */
@Service()
export class ScheduleTriggerTaskHandler implements TaskHandler {
	readonly taskType = SCHEDULE_TRIGGER_TASK_TYPE;

	constructor(
		private logger: Logger,
		private readonly errorReporter: ErrorReporter,
		private readonly globalConfig: GlobalConfig,
		private readonly executionRepository: ExecutionRepository,
		private readonly eventService: EventService,
		private readonly triggerExecutionContextFactory: TriggerExecutionContextFactory,
		private readonly workflowExecutionService: WorkflowExecutionService,
		private readonly ownershipService: OwnershipService,
		private readonly workflowPublisherService: WorkflowPublisherService,
	) {
		this.logger = this.logger.scoped('scheduler');
	}

	async execute(task: ClaimedTask, report: DispatchReporter): Promise<DispatchDecision> {
		const { workflowId, nodeId } = this.parsePayload(task);
		const workflowData =
			await this.triggerExecutionContextFactory.findPublishedWorkflowData(workflowId);

		if (workflowData === null) {
			this.logger.debug('Workflow has no published version. Skipping the occurrence', {
				taskId: task.id,
				jobId: task.jobId,
				workflowId,
				nodeId,
			});
			return report.notDispatched();
		}
		const node = resolveTaskTriggerNode(
			workflowData,
			nodeId,
			task,
			'Schedule-trigger task points to a node that is missing or disabled in the published workflow',
		);

		const deduplicationKey = scheduleTriggerDeduplicationKey(task);
		// `''`/`'DEFAULT'` are the instance-default sentinels, not Moment zones:
		// resolve them to the instance timezone so the item carries valid metadata.
		const settingsTimezone = workflowData.settings?.timezone;
		const timezone =
			settingsTimezone && settingsTimezone !== 'DEFAULT'
				? settingsTimezone
				: this.globalConfig.generic.timezone;
		const item =
			node.type === CRON_NODE_TYPE
				? { json: {} }
				: buildScheduleTriggerItem(task.scheduledFor, timezone);

		const additionalData = await WorkflowExecuteAdditionalData.getBase({
			workflowId,
			workflowSettings: workflowData.settings,
			// A schedule has nobody to be, so the run is attributed to the publisher
			// of the version it runs. That is the mapping's `versionId`, not the
			// workflow row's `activeVersionId`: publication updates the row first and
			// the mapping after, so mid-publication the row already points at a
			// version whose nodes are not the ones below.
			userId: await this.workflowPublisherService.findPublisherUserId(
				workflowId,
				workflowData.versionId,
			),
		});

		try {
			// TODO(CAT-4078): an engine v2 run writes no execution row, so the unique
			// index that turns a redelivered occurrence into a `DuplicateExecutionError`
			// never applies and the redelivery starts a second run.
			const executionId = await this.workflowExecutionService.runWorkflow(
				workflowData,
				node,
				[[item]],
				additionalData,
				'trigger',
				/* responsePromise= */ undefined,
				deduplicationKey,
			);

			const decision = report.dispatched();

			const { projectId, projectName } = await getWorkflowProjectDetailsSafe(
				this.ownershipService,
				workflowData.id,
			);

			this.eventService.emit('workflow-executed', {
				workflowId,
				workflowName: workflowData.name,
				executionId,
				projectId,
				projectName,
				source: 'trigger',
			});

			this.logger.debug('Handed off scheduled occurrence to a new execution', {
				taskId: task.id,
				jobId: task.jobId,
				workflowId,
				executionId,
				deduplicationKey,
			});

			return decision;
		} catch (error) {
			if (!(error instanceof DuplicateExecutionError)) {
				throw error;
			}
			// The effect already exists from a prior delivery; this occurrence hands off
			// nothing new, so it reports no dispatch and the executor writes no marker.
			await this.recordExistingHandoff(task, error);
			return report.notDispatched();
		}
	}

	private parsePayload(task: ClaimedTask): ScheduleTriggerTaskPayload {
		if (!isScheduleTriggerTaskPayload(task.payload)) {
			throw new UnexpectedError('Schedule-trigger task payload is missing workflowId or nodeId', {
				extra: { taskId: task.id, jobId: task.jobId },
			});
		}
		return task.payload;
	}

	/**
	 * The redelivery backstop: an execution already exists under this key, so
	 * find it, record the pre-existing handoff, and let the task complete. The
	 * logged status makes the rare stuck row (inserted as `new`, never started
	 * because the instance died mid-handoff) observable.
	 */
	private async recordExistingHandoff(
		task: ClaimedTask,
		error: DuplicateExecutionError,
	): Promise<void> {
		const { deduplicationKey } = error;
		const existing = await this.executionRepository.findOne({
			where: { deduplicationKey },
			select: ['id', 'status'],
		});

		const context = {
			taskId: task.id,
			jobId: task.jobId,
			attempts: task.attempts,
			deduplicationKey,
			executionId: existing?.id,
			executionStatus: existing?.status,
		};
		this.logger.warn(
			'Occurrence redelivered after a previously recorded execution; skipping',
			context,
		);
		this.errorReporter.warn(error, { extra: context, shouldBeLogged: false });
	}
}