refactor(core): Centralize scaling mode (no-changelog)

packages/cli/src/Interfaces.ts

@@ -42,7 +42,7 @@ import type { WorkflowRepository } from '@db/repositories/workflow.repository';
 import type { ExternalHooks } from './ExternalHooks';
 import type { LICENSE_FEATURES, LICENSE_QUOTAS } from './constants';
 import type { WorkflowWithSharingsAndCredentials } from './workflows/workflows.types';
-import type { WorkerJobStatusSummary } from './services/orchestration/worker/types';
+import type { RunningJobSummary } from './scaling/types';
 import type { Scope } from '@n8n/permissions';
 
 export interface ICredentialsTypeData {
@@ -420,7 +420,7 @@ export interface IPushDataWorkerStatusMessage {
 
 export interface IPushDataWorkerStatusPayload {
 	workerId: string;
-	runningJobsSummary: WorkerJobStatusSummary[];
+	runningJobsSummary: RunningJobSummary[];
 	freeMem: number;
 	totalMem: number;
 	uptime: number;

packages/cli/src/Server.ts

@@ -211,8 +211,8 @@ export class Server extends AbstractServer {
 		setupPushHandler(restEndpoint, app);
 
 		if (config.getEnv('executions.mode') === 'queue') {
-			const { Queue } = await import('@/Queue');
-			await Container.get(Queue).init();
+			const { ScalingService } = await import('@/scaling/scaling.service');
+			await Container.get(ScalingService).setupQueue();
 		}
 
 		await handleMfaDisable();

packages/cli/src/WorkflowRunner.ts

@@ -28,8 +28,8 @@ import { ExecutionRepository } from '@db/repositories/execution.repository';
 import { ExternalHooks } from '@/ExternalHooks';
 import type { IExecutionResponse, IWorkflowExecutionDataProcess } from '@/Interfaces';
 import { NodeTypes } from '@/NodeTypes';
-import type { Job, JobData, JobResponse } from '@/Queue';
-import { Queue } from '@/Queue';
+import type { Job, JobData, JobResult } from '@/scaling/types';
+import { ScalingService } from '@/scaling/scaling.service';
 import * as WorkflowHelpers from '@/WorkflowHelpers';
 import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
 import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
@@ -40,7 +40,7 @@ import { EventService } from './events/event.service';
 
 @Service()
 export class WorkflowRunner {
-	private jobQueue: Queue;
+	private readonly scalingService: ScalingService;
 
 	private executionsMode = config.getEnv('executions.mode');
 
@@ -55,7 +55,7 @@ export class WorkflowRunner {
 		private readonly eventService: EventService,
 	) {
 		if (this.executionsMode === 'queue') {
-			this.jobQueue = Container.get(Queue);
+			this.scalingService = Container.get(ScalingService);
 		}
 	}
 
@@ -375,9 +375,7 @@ export class WorkflowRunner {
 		let job: Job;
 		let hooks: WorkflowHooks;
 		try {
-			job = await this.jobQueue.add(jobData, jobOptions);
-
-			this.logger.info(`Started with job ID: ${job.id.toString()} (Execution ID: ${executionId})`);
+			job = await this.scalingService.addJob(jobData, jobOptions);
 
 			hooks = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerMain(
 				data.executionMode,
@@ -406,8 +404,7 @@ export class WorkflowRunner {
 			async (resolve, reject, onCancel) => {
 				onCancel.shouldReject = false;
 				onCancel(async () => {
-					const queue = Container.get(Queue);
-					await queue.stopJob(job);
+					await Container.get(ScalingService).stopJob(job);
 
 					// We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
 					// "workflowExecuteAfter" which we require.
@@ -424,11 +421,11 @@ export class WorkflowRunner {
 					reject(error);
 				});
 
-				const jobData: Promise<JobResponse> = job.finished();
+				const jobData: Promise<JobResult> = job.finished();
 
 				const queueRecoveryInterval = config.getEnv('queue.bull.queueRecoveryInterval');
 
-				const racingPromises: Array<Promise<JobResponse>> = [jobData];
+				const racingPromises: Array<Promise<JobResult>> = [jobData];
 
 				let clearWatchdogInterval;
 				if (queueRecoveryInterval > 0) {
@@ -446,9 +443,9 @@ export class WorkflowRunner {
 					 ************************************************ */
 					let watchDogInterval: NodeJS.Timeout | undefined;
 
-					const watchDog: Promise<JobResponse> = new Promise((res) => {
+					const watchDog: Promise<JobResult> = new Promise((res) => {
 						watchDogInterval = setInterval(async () => {
-							const currentJob = await this.jobQueue.getJob(job.id);
+							const currentJob = await this.scalingService.getJob(job.id);
 							// When null means job is finished (not found in queue)
 							if (currentJob === null) {
 								// Mimic worker's success message

packages/cli/src/commands/webhook.ts

@@ -4,8 +4,8 @@ import { ApplicationError } from 'n8n-workflow';
 
 import config from '@/config';
 import { ActiveExecutions } from '@/ActiveExecutions';
+import { ScalingService } from '@/scaling/scaling.service';
 import { WebhookServer } from '@/webhooks/WebhookServer';
-import { Queue } from '@/Queue';
 import { BaseCommand } from './BaseCommand';
 
 import { OrchestrationWebhookService } from '@/services/orchestration/webhook/orchestration.webhook.service';
@@ -96,7 +96,7 @@ export class Webhook extends BaseCommand {
 			);
 		}
 
-		await Container.get(Queue).init();
+		await Container.get(ScalingService).setupQueue();
 		await this.server.start();
 		this.logger.debug(`Webhook listener ID: ${this.server.uniqueInstanceId}`);
 		this.logger.info('Webhook listener waiting for requests.');