Merge remote-tracking branch 'origin/master' into revert-eslint-prettier

packages/@n8n/benchmark/scenarios/binary-data/binary-data.script.js

@@ -3,8 +3,9 @@ import { check } from 'k6';
 
 const apiBaseUrl = __ENV.API_BASE_URL;
 
-const file = open(__ENV.SCRIPT_FILE_PATH, 'b');
+// This creates a 2MB file (16 * 128 * 1024 = 2 * 1024 * 1024 = 2MB)
-const filename = String(__ENV.SCRIPT_FILE_PATH).split('/').pop();
+const file = Array.from({ length: 128 * 1024 }, () => Math.random().toString().slice(2)).join('');
+const filename = 'test.bin';
 
 export default function () {
 	const data = {

packages/@n8n/benchmark/src/test-execution/k6-executor.ts

@@ -77,7 +77,6 @@ export function handleSummary(data) {
 			env: {
 				API_BASE_URL: this.opts.n8nApiBaseUrl,
 				K6_CLOUD_TOKEN: this.opts.k6ApiToken,
-				SCRIPT_FILE_PATH: augmentedTestScriptPath,
 			},
 			stdio: 'inherit',
 		})`${k6ExecutablePath} run ${flattedFlags} ${augmentedTestScriptPath}`;

packages/@n8n/storybook/package.json

@@ -3,19 +3,19 @@
   "private": true,
   "version": "0.0.1",
   "devDependencies": {
-    "@chromatic-com/storybook": "^1.5.0",
+    "@chromatic-com/storybook": "^2.0.2",
-    "@storybook/addon-a11y": "^8.1.4",
+    "@storybook/addon-a11y": "^8.3.1",
-    "@storybook/addon-actions": "^8.1.4",
+    "@storybook/addon-actions": "^8.3.1",
-    "@storybook/addon-docs": "^8.1.4",
+    "@storybook/addon-docs": "^8.3.1",
-    "@storybook/addon-essentials": "^8.1.4",
+    "@storybook/addon-essentials": "^8.3.1",
-    "@storybook/addon-interactions": "^8.1.4",
+    "@storybook/addon-interactions": "^8.3.1",
-    "@storybook/addon-links": "^8.1.4",
+    "@storybook/addon-links": "^8.3.1",
-    "@storybook/addon-themes": "^8.1.4",
+    "@storybook/addon-themes": "^8.3.1",
-    "@storybook/blocks": "^8.1.4",
+    "@storybook/blocks": "^8.3.1",
-    "@storybook/test": "^8.1.4",
+    "@storybook/test": "^8.3.1",
-    "@storybook/vue3": "^8.1.4",
+    "@storybook/vue3": "^8.3.1",
-    "@storybook/vue3-vite": "^8.1.4",
+    "@storybook/vue3-vite": "^8.3.1",
-    "chromatic": "^11.4.1",
+    "chromatic": "^11.10.2",
-    "storybook": "^8.1.4"
+    "storybook": "^8.3.1"
   }
 }

packages/cli/src/scaling/__tests__/publisher.service.test.ts

@@ -3,11 +3,11 @@ import { mock } from 'jest-mock-extended';
 
 import config from '@/config';
 import { generateNanoId } from '@/databases/utils/generators';
-import type { RedisClientService } from '@/services/redis/redis-client.service';
 import type {
 	RedisServiceCommandObject,
 	RedisServiceWorkerResponseObject,
-} from '@/services/redis/redis-service-commands';
+} from '@/scaling/redis/redis-service-commands';
+import type { RedisClientService } from '@/services/redis-client.service';
 
 import { Publisher } from '../pubsub/publisher.service';
 

packages/cli/src/scaling/__tests__/subscriber.service.test.ts

@@ -2,7 +2,7 @@ import type { Redis as SingleNodeClient } from 'ioredis';
 import { mock } from 'jest-mock-extended';
 
 import config from '@/config';
-import type { RedisClientService } from '@/services/redis/redis-client.service';
+import type { RedisClientService } from '@/services/redis-client.service';
 
 import { Subscriber } from '../pubsub/subscriber.service';
 

packages/cli/src/scaling/constants.ts

@@ -1,3 +1,7 @@
 export const QUEUE_NAME = 'jobs';
 
 export const JOB_TYPE_NAME = 'job';
+
+export const COMMAND_PUBSUB_CHANNEL = 'n8n.commands';
+
+export const WORKER_RESPONSE_PUBSUB_CHANNEL = 'n8n.worker-response';

packages/cli/src/scaling/pubsub/publisher.service.ts

@@ -3,11 +3,11 @@ import { Service } from 'typedi';
 
 import config from '@/config';
 import { Logger } from '@/logger';
-import { RedisClientService } from '@/services/redis/redis-client.service';
 import type {
 	RedisServiceCommandObject,
 	RedisServiceWorkerResponseObject,
-} from '@/services/redis/redis-service-commands';
+} from '@/scaling/redis/redis-service-commands';
+import { RedisClientService } from '@/services/redis-client.service';
 
 /**
  * Responsible for publishing messages into the pubsub channels used by scaling mode.

packages/cli/src/scaling/pubsub/pubsub.types.ts

@@ -1,7 +1,8 @@
-import type {
+import type { PushType, WorkerStatus } from '@n8n/api-types';
-	COMMAND_REDIS_CHANNEL,
+
-	WORKER_RESPONSE_REDIS_CHANNEL,
+import type { IWorkflowDb } from '@/interfaces';
-} from '@/services/redis/redis-constants';
+
+import type { COMMAND_PUBSUB_CHANNEL, WORKER_RESPONSE_PUBSUB_CHANNEL } from '../constants';
 
 /**
  * Pubsub channel used by scaling mode:
@@ -10,5 +11,90 @@ import type {
  * - `n8n.worker-response` for messages sent by workers in response to commands from main processes
  */
 export type ScalingPubSubChannel =
-	| typeof COMMAND_REDIS_CHANNEL
+	| typeof COMMAND_PUBSUB_CHANNEL
-	| typeof WORKER_RESPONSE_REDIS_CHANNEL;
+	| typeof WORKER_RESPONSE_PUBSUB_CHANNEL;
+
+export type PubSubMessageMap = {
+	// #region Lifecycle
+
+	'reload-license': never;
+
+	'restart-event-bus': {
+		result: 'success' | 'error';
+		error?: string;
+	};
+
+	'reload-external-secrets-providers': {
+		result: 'success' | 'error';
+		error?: string;
+	};
+
+	'stop-worker': never;
+
+	// #endregion
+
+	// #region Community packages
+
+	'community-package-install': {
+		packageName: string;
+		packageVersion: string;
+	};
+
+	'community-package-update': {
+		packageName: string;
+		packageVersion: string;
+	};
+
+	'community-package-uninstall': {
+		packageName: string;
+		packageVersion: string;
+	};
+
+	// #endregion
+
+	// #region Worker view
+
+	'get-worker-id': never;
+
+	'get-worker-status': WorkerStatus;
+
+	// #endregion
+
+	// #region Multi-main setup
+
+	'add-webhooks-triggers-and-pollers': {
+		workflowId: string;
+	};
+
+	'remove-triggers-and-pollers': {
+		workflowId: string;
+	};
+
+	'display-workflow-activation': {
+		workflowId: string;
+	};
+
+	'display-workflow-deactivation': {
+		workflowId: string;
+	};
+
+	// currently 'workflow-failed-to-activate'
+	'display-workflow-activation-error': {
+		workflowId: string;
+		errorMessage: string;
+	};
+
+	'relay-execution-lifecycle-event': {
+		type: PushType;
+		args: Record<string, unknown>;
+		pushRef: string;
+	};
+
+	'clear-test-webhooks': {
+		webhookKey: string;
+		workflowEntity: IWorkflowDb;
+		pushRef: string;
+	};
+
+	// #endregion
+};

packages/cli/src/scaling/pubsub/subscriber.service.ts

@@ -3,7 +3,7 @@ import { Service } from 'typedi';
 
 import config from '@/config';
 import { Logger } from '@/logger';
-import { RedisClientService } from '@/services/redis/redis-client.service';
+import { RedisClientService } from '@/services/redis-client.service';
 
 import type { ScalingPubSubChannel } from './pubsub.types';
 

packages/cli/src/scaling/scaling.service.ts

@@ -24,7 +24,7 @@ import type {
 	JobStatus,
 	JobId,
 	QueueRecoveryContext,
-	PubSubMessage,
+	JobReport,
 } from './scaling.types';
 
 @Service()
@@ -46,7 +46,7 @@ export class ScalingService {
 
 	async setupQueue() {
 		const { default: BullQueue } = await import('bull');
-		const { RedisClientService } = await import('@/services/redis/redis-client.service');
+		const { RedisClientService } = await import('@/services/redis-client.service');
 		const service = Container.get(RedisClientService);
 
 		const bullPrefix = this.globalConfig.queue.bull.prefix;
@@ -265,7 +265,7 @@ export class ScalingService {
 		}
 	}
 
-	private isPubSubMessage(candidate: unknown): candidate is PubSubMessage {
+	private isPubSubMessage(candidate: unknown): candidate is JobReport {
 		return typeof candidate === 'object' && candidate !== null && 'kind' in candidate;
 	}
 

packages/cli/src/scaling/scaling.types.ts

@@ -23,11 +23,11 @@ export type JobStatus = Bull.JobStatus;
 
 export type JobOptions = Bull.JobOptions;
 
-export type PubSubMessage = MessageToMain | MessageToWorker;
+export type JobReport = JobReportToMain | JobReportToWorker;
 
-type MessageToMain = RespondToWebhookMessage;
+type JobReportToMain = RespondToWebhookMessage;
 
-type MessageToWorker = AbortJobMessage;
+type JobReportToWorker = AbortJobMessage;
 
 type RespondToWebhookMessage = {
 	kind: 'respond-to-webhook';

packages/cli/src/services/__tests__/orchestration.service.test.ts

@@ -9,13 +9,13 @@ import config from '@/config';
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import { ExternalSecretsManager } from '@/external-secrets/external-secrets-manager.ee';
 import { Push } from '@/push';
+import type { RedisServiceWorkerResponseObject } from '@/scaling/redis/redis-service-commands';
 import * as helpers from '@/services/orchestration/helpers';
 import { handleCommandMessageMain } from '@/services/orchestration/main/handle-command-message-main';
 import { handleWorkerResponseMessageMain } from '@/services/orchestration/main/handle-worker-response-message-main';
 import { OrchestrationHandlerMainService } from '@/services/orchestration/main/orchestration.handler.main.service';
 import { OrchestrationService } from '@/services/orchestration.service';
-import { RedisClientService } from '@/services/redis/redis-client.service';
+import { RedisClientService } from '@/services/redis-client.service';
-import type { RedisServiceWorkerResponseObject } from '@/services/redis/redis-service-commands';
 import { mockInstance } from '@test/mocking';
 
 import type { MainResponseReceivedHandlerOptions } from '../orchestration/main/types';

packages/cli/src/services/cache/cache.service.ts

@@ -36,7 +36,7 @@ export class CacheService extends TypedEmitter<CacheEvents> {
 		const useRedis = backend === 'redis' || (backend === 'auto' && mode === 'queue');
 
 		if (useRedis) {
-			const { RedisClientService } = await import('../redis/redis-client.service');
+			const { RedisClientService } = await import('../redis-client.service');
 			const redisClientService = Container.get(RedisClientService);
 
 			const prefixBase = config.getEnv('redis.prefix');

packages/cli/src/services/orchestration.service.ts

@@ -8,7 +8,10 @@ import type { Publisher } from '@/scaling/pubsub/publisher.service';
 import type { Subscriber } from '@/scaling/pubsub/subscriber.service';
 
 import { MultiMainSetup } from './orchestration/main/multi-main-setup.ee';
-import type { RedisServiceBaseCommand, RedisServiceCommand } from './redis/redis-service-commands';
+import type {
+	RedisServiceBaseCommand,
+	RedisServiceCommand,
+} from '../scaling/redis/redis-service-commands';
 
 @Service()
 export class OrchestrationService {

packages/cli/src/services/orchestration/helpers.ts

@@ -3,9 +3,9 @@ import os from 'node:os';
 import { Container } from 'typedi';
 
 import { Logger } from '@/logger';
+import { COMMAND_PUBSUB_CHANNEL } from '@/scaling/constants';
 
-import { COMMAND_REDIS_CHANNEL } from '../redis/redis-constants';
+import type { RedisServiceCommandObject } from '../../scaling/redis/redis-service-commands';
-import type { RedisServiceCommandObject } from '../redis/redis-service-commands';
 
 export interface RedisServiceCommandLastReceived {
 	[date: string]: Date;
@@ -18,7 +18,7 @@ export function messageToRedisServiceCommandObject(messageString: string) {
 		message = jsonParse<RedisServiceCommandObject>(messageString);
 	} catch {
 		Container.get(Logger).debug(
-			`Received invalid message via channel ${COMMAND_REDIS_CHANNEL}: "${messageString}"`,
+			`Received invalid message via channel ${COMMAND_PUBSUB_CHANNEL}: "${messageString}"`,
 		);
 		return;
 	}

packages/cli/src/services/orchestration/main/handle-worker-response-message-main.ts

@@ -3,11 +3,11 @@ import { jsonParse } from 'n8n-workflow';
 import Container from 'typedi';
 
 import { Logger } from '@/logger';
-import { WORKER_RESPONSE_REDIS_CHANNEL } from '@/services/redis/redis-constants';
+import { WORKER_RESPONSE_PUBSUB_CHANNEL } from '@/scaling/constants';
 
 import type { MainResponseReceivedHandlerOptions } from './types';
 import { Push } from '../../../push';
-import type { RedisServiceWorkerResponseObject } from '../../redis/redis-service-commands';
+import type { RedisServiceWorkerResponseObject } from '../../../scaling/redis/redis-service-commands';
 
 export async function handleWorkerResponseMessageMain(
 	messageString: string,
@@ -19,7 +19,7 @@ export async function handleWorkerResponseMessageMain(
 
 	if (!workerResponse) {
 		Container.get(Logger).debug(
-			`Received invalid message via channel ${WORKER_RESPONSE_REDIS_CHANNEL}: "${messageString}"`,
+			`Received invalid message via channel ${WORKER_RESPONSE_PUBSUB_CHANNEL}: "${messageString}"`,
 		);
 		return;
 	}

packages/cli/src/services/orchestration/main/multi-main-setup.ee.ts

@@ -6,7 +6,7 @@ import config from '@/config';
 import { TIME } from '@/constants';
 import { Logger } from '@/logger';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
-import { RedisClientService } from '@/services/redis/redis-client.service';
+import { RedisClientService } from '@/services/redis-client.service';
 import { TypedEmitter } from '@/typed-emitter';
 
 type MultiMainEvents = {

packages/cli/src/services/orchestration/main/orchestration.handler.main.service.ts

@@ -1,12 +1,12 @@
 import { Service } from 'typedi';
 
+import { COMMAND_PUBSUB_CHANNEL, WORKER_RESPONSE_PUBSUB_CHANNEL } from '@/scaling/constants';
 import { Subscriber } from '@/scaling/pubsub/subscriber.service';
 
 import { handleCommandMessageMain } from './handle-command-message-main';
 import { handleWorkerResponseMessageMain } from './handle-worker-response-message-main';
 import type { MainResponseReceivedHandlerOptions } from './types';
 import { OrchestrationHandlerService } from '../../orchestration.handler.base.service';
-import { COMMAND_REDIS_CHANNEL, WORKER_RESPONSE_REDIS_CHANNEL } from '../../redis/redis-constants';
 
 @Service()
 export class OrchestrationHandlerMainService extends OrchestrationHandlerService {
@@ -19,9 +19,9 @@ export class OrchestrationHandlerMainService extends OrchestrationHandlerService
 		await this.subscriber.subscribe('n8n.worker-response');
 
 		this.subscriber.addMessageHandler(async (channel: string, messageString: string) => {
-			if (channel === WORKER_RESPONSE_REDIS_CHANNEL) {
+			if (channel === WORKER_RESPONSE_PUBSUB_CHANNEL) {
 				await handleWorkerResponseMessageMain(messageString, options);
-			} else if (channel === COMMAND_REDIS_CHANNEL) {
+			} else if (channel === COMMAND_PUBSUB_CHANNEL) {
 				await handleCommandMessageMain(messageString);
 			}
 		});

packages/cli/src/services/orchestration/pubsub/pubsub-message-map.ts

@@ -1,88 +0,0 @@
-import type { PushType, WorkerStatus } from '@n8n/api-types';
-
-import type { IWorkflowDb } from '@/interfaces';
-
-export type PubSubMessageMap = {
-	// #region Lifecycle
-
-	'reload-license': never;
-
-	'restart-event-bus': {
-		result: 'success' | 'error';
-		error?: string;
-	};
-
-	'reload-external-secrets-providers': {
-		result: 'success' | 'error';
-		error?: string;
-	};
-
-	'stop-worker': never;
-
-	// #endregion
-
-	// #region Community packages
-
-	'community-package-install': {
-		packageName: string;
-		packageVersion: string;
-	};
-
-	'community-package-update': {
-		packageName: string;
-		packageVersion: string;
-	};
-
-	'community-package-uninstall': {
-		packageName: string;
-		packageVersion: string;
-	};
-
-	// #endregion
-
-	// #region Worker view
-
-	'get-worker-id': never;
-
-	'get-worker-status': WorkerStatus;
-
-	// #endregion
-
-	// #region Multi-main setup
-
-	'add-webhooks-triggers-and-pollers': {
-		workflowId: string;
-	};
-
-	'remove-triggers-and-pollers': {
-		workflowId: string;
-	};
-
-	'display-workflow-activation': {
-		workflowId: string;
-	};
-
-	'display-workflow-deactivation': {
-		workflowId: string;
-	};
-
-	// currently 'workflow-failed-to-activate'
-	'display-workflow-activation-error': {
-		workflowId: string;
-		errorMessage: string;
-	};
-
-	'relay-execution-lifecycle-event': {
-		type: PushType;
-		args: Record<string, unknown>;
-		pushRef: string;
-	};
-
-	'clear-test-webhooks': {
-		webhookKey: string;
-		workflowEntity: IWorkflowDb;
-		pushRef: string;
-	};
-
-	// #endregion
-};

packages/cli/src/services/orchestration/webhook/orchestration.handler.webhook.service.ts

@@ -1,10 +1,10 @@
 import { Service } from 'typedi';
 
+import { COMMAND_PUBSUB_CHANNEL } from '@/scaling/constants';
 import { Subscriber } from '@/scaling/pubsub/subscriber.service';
 
 import { handleCommandMessageWebhook } from './handle-command-message-webhook';
 import { OrchestrationHandlerService } from '../../orchestration.handler.base.service';
-import { COMMAND_REDIS_CHANNEL } from '../../redis/redis-constants';
 
 @Service()
 export class OrchestrationHandlerWebhookService extends OrchestrationHandlerService {
@@ -16,7 +16,7 @@ export class OrchestrationHandlerWebhookService extends OrchestrationHandlerServ
 		await this.subscriber.subscribe('n8n.commands');
 
 		this.subscriber.addMessageHandler(async (channel: string, messageString: string) => {
-			if (channel === COMMAND_REDIS_CHANNEL) {
+			if (channel === COMMAND_PUBSUB_CHANNEL) {
 				await handleCommandMessageWebhook(messageString);
 			}
 		});

packages/cli/src/services/orchestration/worker/handle-command-message-worker.ts

@@ -7,9 +7,9 @@ import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'
 import { ExternalSecretsManager } from '@/external-secrets/external-secrets-manager.ee';
 import { License } from '@/license';
 import { Logger } from '@/logger';
+import { COMMAND_PUBSUB_CHANNEL } from '@/scaling/constants';
+import type { RedisServiceCommandObject } from '@/scaling/redis/redis-service-commands';
 import { CommunityPackagesService } from '@/services/community-packages.service';
-import { COMMAND_REDIS_CHANNEL } from '@/services/redis/redis-constants';
-import type { RedisServiceCommandObject } from '@/services/redis/redis-service-commands';
 
 import type { WorkerCommandReceivedHandlerOptions } from './types';
 import { debounceMessageReceiver, getOsCpuString } from '../helpers';
@@ -17,7 +17,7 @@ import { debounceMessageReceiver, getOsCpuString } from '../helpers';
 export function getWorkerCommandReceivedHandler(options: WorkerCommandReceivedHandlerOptions) {
 	// eslint-disable-next-line complexity
 	return async (channel: string, messageString: string) => {
-		if (channel === COMMAND_REDIS_CHANNEL) {
+		if (channel === COMMAND_PUBSUB_CHANNEL) {
 			if (!messageString) return;
 			const logger = Container.get(Logger);
 			let message: RedisServiceCommandObject;
@@ -25,7 +25,7 @@ export function getWorkerCommandReceivedHandler(options: WorkerCommandReceivedHa
 				message = jsonParse<RedisServiceCommandObject>(messageString);
 			} catch {
 				logger.debug(
-					`Received invalid message via channel ${COMMAND_REDIS_CHANNEL}: "${messageString}"`,
+					`Received invalid message via channel ${COMMAND_PUBSUB_CHANNEL}: "${messageString}"`,
 				);
 				return;
 			}
@@ -145,7 +145,7 @@ export function getWorkerCommandReceivedHandler(options: WorkerCommandReceivedHa
 						}
 
 						logger.debug(
-							`Received unknown command via channel ${COMMAND_REDIS_CHANNEL}: "${message.command}"`,
+							`Received unknown command via channel ${COMMAND_PUBSUB_CHANNEL}: "${message.command}"`,
 						);
 						break;
 				}

packages/cli/src/services/orchestration/worker/types.ts

@@ -1,5 +1,4 @@
 import type { RunningJobSummary } from '@n8n/api-types';
-import type { ExecutionStatus, WorkflowExecuteMode } from 'n8n-workflow';
 
 import type { Publisher } from '@/scaling/pubsub/publisher.service';
 
@@ -9,14 +8,3 @@ export interface WorkerCommandReceivedHandlerOptions {
 	getRunningJobIds: () => Array<string | number>;
 	getRunningJobsSummary: () => RunningJobSummary[];
 }
-
-export interface WorkerJobStatusSummary {
-	jobId: string;
-	executionId: string;
-	retryOf?: string;
-	startedAt: Date;
-	mode: WorkflowExecuteMode;
-	workflowName: string;
-	workflowId: string;
-	status: ExecutionStatus;
-}

packages/cli/src/services/redis-client.service.ts

@@ -5,7 +5,7 @@ import { Service } from 'typedi';
 
 import { Logger } from '@/logger';
 
-import type { RedisClientType } from './redis.types';
+import type { RedisClientType } from '../scaling/redis/redis.types';
 
 @Service()
 export class RedisClientService {

packages/cli/src/services/redis/redis-constants.ts

@@ -1,2 +0,0 @@
-export const COMMAND_REDIS_CHANNEL = 'n8n.commands';
-export const WORKER_RESPONSE_REDIS_CHANNEL = 'n8n.worker-response';

packages/cli/src/services/redis/redis-service-base-classes.ts

@@ -1,70 +0,0 @@
-import type Redis from 'ioredis';
-import type { Cluster } from 'ioredis';
-import { Service } from 'typedi';
-
-import config from '@/config';
-import { Logger } from '@/logger';
-
-import { RedisClientService } from './redis-client.service';
-import type { RedisClientType } from './redis.types';
-
-export type RedisServiceMessageHandler =
-	| ((channel: string, message: string) => void)
-	| ((stream: string, id: string, message: string[]) => void);
-
-@Service()
-class RedisServiceBase {
-	redisClient: Redis | Cluster | undefined;
-
-	isInitialized = false;
-
-	constructor(
-		protected readonly logger: Logger,
-		private readonly redisClientService: RedisClientService,
-	) {}
-
-	async init(type: RedisClientType): Promise<void> {
-		if (this.redisClient && this.isInitialized) {
-			return;
-		}
-		this.redisClient = this.redisClientService.createClient({ type });
-
-		this.redisClient.on('error', (error) => {
-			if (!String(error).includes('ECONNREFUSED')) {
-				this.logger.warn('Error with Redis: ', error);
-			}
-		});
-
-		this.isInitialized = true;
-	}
-
-	async destroy(): Promise<void> {
-		if (!this.redisClient) {
-			return;
-		}
-		await this.redisClient.quit();
-		this.isInitialized = false;
-		this.redisClient = undefined;
-	}
-}
-
-export abstract class RedisServiceBaseSender extends RedisServiceBase {
-	senderId: string;
-
-	async init(type: RedisClientType): Promise<void> {
-		await super.init(type);
-		this.senderId = config.get('redis.queueModeId');
-	}
-}
-
-export abstract class RedisServiceBaseReceiver extends RedisServiceBase {
-	messageHandlers: Map<string, RedisServiceMessageHandler> = new Map();
-
-	addMessageHandler(handlerName: string, handler: RedisServiceMessageHandler): void {
-		this.messageHandlers.set(handlerName, handler);
-	}
-
-	removeMessageHandler(handlerName: string): void {
-		this.messageHandlers.delete(handlerName);
-	}
-}

packages/editor-ui/src/components/RunDataJson.vue

@@ -14,6 +14,7 @@ import { nonExistingJsonPath } from '@/constants';
 import { useExternalHooks } from '@/composables/useExternalHooks';
 import TextWithHighlights from './TextWithHighlights.vue';
 import { useTelemetry } from '@/composables/useTelemetry';
+import { useElementSize } from '@vueuse/core';
 
 const LazyRunDataJsonActions = defineAsyncComponent(
 	async () => await import('@/components/RunDataJsonActions.vue'),
@@ -45,6 +46,9 @@ const telemetry = useTelemetry();
 const selectedJsonPath = ref(nonExistingJsonPath);
 const draggingPath = ref<null | string>(null);
 const displayMode = ref('json');
+const jsonDataContainer = ref(null);
+
+const { height } = useElementSize(jsonDataContainer);
 
 const jsonData = computed(() => executionDataToJson(props.inputData));
 
@@ -110,7 +114,7 @@ const getListItemName = (path: string) => {
 </script>
 
 <template>
-	<div :class="[$style.jsonDisplay, { [$style.highlight]: highlight }]">
+	<div ref="jsonDataContainer" :class="[$style.jsonDisplay, { [$style.highlight]: highlight }]">
 		<Suspense>
 			<LazyRunDataJsonActions
 				v-if="!editMode.enabled"
@@ -141,6 +145,8 @@ const getListItemName = (path: string) => {
 				root-path=""
 				selectable-type="single"
 				class="json-data"
+				:virtual="true"
+				:height="height"
 				@update:selected-value="selectedJsonPath = $event"
 			>
 				<template #renderNodeKey="{ node }">
@@ -192,11 +198,10 @@ const getListItemName = (path: string) => {
 	left: 0;
 	padding-left: var(--spacing-s);
 	right: 0;
-	overflow-y: auto;
+	overflow-y: hidden;
 	line-height: 1.5;
 	word-break: normal;
 	height: 100%;
-	padding-bottom: var(--spacing-3xl);
 
 	&:hover {
 		/* Shows .actionsGroup element from <run-data-json-actions /> child component */
@@ -299,4 +304,8 @@ const getListItemName = (path: string) => {
 .vjs-tree .vjs-tree__content.has-line {
 	border-left: 1px dotted var(--color-json-line);
 }
+
+.vjs-tree .vjs-tree-list-holder-inner {
+	padding-bottom: var(--spacing-3xl);
+}
 </style>

packages/editor-ui/src/components/__tests__/RunDataJson.test.ts

@@ -2,6 +2,22 @@ import { createTestingPinia } from '@pinia/testing';
 import { screen, cleanup } from '@testing-library/vue';
 import RunDataJson from '@/components/RunDataJson.vue';
 import { createComponentRenderer } from '@/__tests__/render';
+import { useElementSize } from '@vueuse/core'; // Import the composable to mock
+
+vi.mock('@vueuse/core', async () => {
+	// eslint-disable-next-line @typescript-eslint/consistent-type-imports
+	const originalModule = await vi.importActual<typeof import('@vueuse/core')>('@vueuse/core');
+
+	return {
+		...originalModule, // Keep all original exports
+		useElementSize: vi.fn(), // Mock useElementSize
+	};
+});
+
+(useElementSize as jest.Mock).mockReturnValue({
+	height: 500, // Mocked height value
+	width: 300, // Mocked width value
+});
 
 const renderComponent = createComponentRenderer(RunDataJson, {
 	props: {
@@ -34,18 +50,6 @@ const renderComponent = createComponentRenderer(RunDataJson, {
 			disabled: false,
 		},
 	},
-	global: {
-		mocks: {
-			$locale: {
-				baseText() {
-					return '';
-				},
-			},
-			$store: {
-				getters: {},
-			},
-		},
-	},
 });
 
 describe('RunDataJson.vue', () => {

packages/editor-ui/src/stores/assistant.store.ts

@@ -356,9 +356,9 @@ export const useAssistantStore = defineStore(STORES.ASSISTANT, () => {
 		resetAssistantChat();
 		chatSessionTask.value = credentialType ? 'credentials' : 'support';
 		chatSessionCredType.value = credentialType;
-		chatWindowOpen.value = true;
 		addUserMessage(userMessage, id);
 		addLoadingAssistantMessage(locale.baseText('aiAssistant.thinkingSteps.thinking'));
+		openChat();
 		streaming.value = true;
 
 		let payload: ChatRequest.InitSupportChat | ChatRequest.InitCredHelp = {

packages/editor-ui/src/stores/logStreaming.store.ts

@@ -7,8 +7,9 @@ import {
 	hasDestinationId,
 	saveDestinationToDb,
 	sendTestMessageToDestination,
-} from '../api/eventbus.ee';
+} from '@/api/eventbus.ee';
 import { useRootStore } from './root.store';
+import { ref } from 'vue';
 
 export interface EventSelectionItem {
 	selected: boolean;
@@ -32,221 +33,242 @@ export interface DestinationSettingsStore {
 	[key: string]: DestinationStoreItem;
 }
 
-export const useLogStreamingStore = defineStore('logStreaming', {
+export const useLogStreamingStore = defineStore('logStreaming', () => {
-	state: () => ({
+	const items = ref<DestinationSettingsStore>({});
-		items: {} as DestinationSettingsStore,
+	const eventNames = ref(new Set<string>());
-		eventNames: new Set<string>(),
+
-	}),
+	const rootStore = useRootStore();
-	getters: {},
+
-	actions: {
+	const addDestination = (destination: MessageEventBusDestinationOptions) => {
-		addDestination(destination: MessageEventBusDestinationOptions) {
+		if (destination.id && items.value[destination.id]) {
-			if (destination.id && this.items[destination.id]) {
+			items.value[destination.id].destination = destination;
-				this.items[destination.id].destination = destination;
+		} else {
-			} else {
+			setSelectionAndBuildItems(destination);
-				this.setSelectionAndBuildItems(destination);
+		}
+	};
+
+	const setSelectionAndBuildItems = (destination: MessageEventBusDestinationOptions) => {
+		if (destination.id) {
+			if (!items.value[destination.id]) {
+				items.value[destination.id] = {
+					destination,
+					selectedEvents: new Set<string>(),
+					eventGroups: [],
+					isNew: false,
+				} as DestinationStoreItem;
 			}
-		},
+			items.value[destination.id]?.selectedEvents?.clear();
-		getDestination(destinationId: string): MessageEventBusDestinationOptions | undefined {
+			if (destination.subscribedEvents) {
-			if (this.items[destinationId]) {
+				for (const eventName of destination.subscribedEvents) {
-				return this.items[destinationId].destination;
+					items.value[destination.id]?.selectedEvents?.add(eventName);
-			} else {
+				}
-				return;
 			}
-		},
+			items.value[destination.id].eventGroups = eventGroupsFromStringList(
-		getAllDestinations(): MessageEventBusDestinationOptions[] {
+				eventNames.value,
-			const destinations: MessageEventBusDestinationOptions[] = [];
+				items.value[destination.id]?.selectedEvents,
-			for (const key of Object.keys(this.items)) {
+			);
-				destinations.push(this.items[key].destination);
+		}
-			}
+	};
-			return destinations;
+
-		},
+	const getDestination = (destinationId: string) => {
-		updateDestination(destination: MessageEventBusDestinationOptions) {
+		if (items.value[destinationId]) {
-			if (destination.id && this.items[destination.id]) {
+			return items.value[destinationId].destination;
-				this.$patch((state) => {
+		} else {
-					if (destination.id && this.items[destination.id]) {
+			return;
-						state.items[destination.id].destination = destination;
+		}
-					}
+	};
-					// to trigger refresh
+
-					state.items = { ...state.items };
+	const getAllDestinations = () => {
-				});
+		const destinations: MessageEventBusDestinationOptions[] = [];
-			}
+		for (const key of Object.keys(items)) {
-		},
+			destinations.push(items.value[key].destination);
-		removeDestination(destinationId: string) {
+		}
-			if (!destinationId) return;
+		return destinations;
-			delete this.items[destinationId];
+	};
-			if (this.items[destinationId]) {
+
-				this.$patch({
+	const clearDestinations = () => {
-					items: {
+		items.value = {};
-						...this.items,
+	};
-					},
+
-				});
+	const addEventName = (name: string) => {
-			}
+		eventNames.value.add(name);
-		},
+	};
-		clearDestinations() {
+
-			this.items = {};
+	const removeEventName = (name: string) => {
-		},
+		eventNames.value.delete(name);
-		addEventName(name: string) {
+	};
-			this.eventNames.add(name);
+
-		},
+	const clearEventNames = () => {
-		removeEventName(name: string) {
+		eventNames.value.clear();
-			this.eventNames.delete(name);
+	};
-		},
+
-		clearEventNames() {
+	const addSelectedEvent = (id: string, name: string) => {
-			this.eventNames.clear();
+		items.value[id]?.selectedEvents?.add(name);
-		},
+		setSelectedInGroup(id, name, true);
-		addSelectedEvent(id: string, name: string) {
+	};
-			this.items[id]?.selectedEvents?.add(name);
+
-			this.setSelectedInGroup(id, name, true);
+	const removeSelectedEvent = (id: string, name: string) => {
-		},
+		items.value[id]?.selectedEvents?.delete(name);
-		removeSelectedEvent(id: string, name: string) {
+		setSelectedInGroup(id, name, false);
-			this.items[id]?.selectedEvents?.delete(name);
+	};
-			this.setSelectedInGroup(id, name, false);
+
-		},
+	const setSelectedInGroup = (destinationId: string, name: string, isSelected: boolean) => {
-		getSelectedEvents(destinationId: string): string[] {
+		if (items.value[destinationId]) {
-			const selectedEvents: string[] = [];
+			const groupName = eventGroupFromEventName(name);
-			if (this.items[destinationId]) {
+			const groupIndex = items.value[destinationId].eventGroups.findIndex(
-				for (const group of this.items[destinationId].eventGroups) {
+				(e) => e.name === groupName,
-					if (group.selected) {
+			);
-						selectedEvents.push(group.name);
+
-					}
+			if (groupIndex > -1) {
-					for (const event of group.children) {
+				if (groupName === name) {
-						if (event.selected) {
+					items.value[destinationId].eventGroups[groupIndex].selected = isSelected;
-							selectedEvents.push(event.name);
+				} else {
+					const eventIndex = items.value[destinationId].eventGroups[groupIndex].children.findIndex(
+						(e) => e.name === name,
+					);
+					if (eventIndex > -1) {
+						items.value[destinationId].eventGroups[groupIndex].children[eventIndex].selected =
+							isSelected;
+						if (isSelected) {
+							items.value[destinationId].eventGroups[groupIndex].indeterminate = isSelected;
+						} else {
+							let anySelected = false;
+							for (
+								let i = 0;
+								i < items.value[destinationId].eventGroups[groupIndex].children.length;
+								i++
+							) {
+								anySelected =
+									anySelected ||
+									items.value[destinationId].eventGroups[groupIndex].children[i].selected;
+							}
+							items.value[destinationId].eventGroups[groupIndex].indeterminate = anySelected;
 						}
 					}
 				}
 			}
-			return selectedEvents;
+		}
-		},
+	};
-		setSelectedInGroup(destinationId: string, name: string, isSelected: boolean) {
+
-			if (this.items[destinationId]) {
+	const removeDestinationItemTree = (id: string) => {
-				const groupName = eventGroupFromEventName(name);
+		delete items.value[id];
-				const groupIndex = this.items[destinationId].eventGroups.findIndex(
+	};
-					(e) => e.name === groupName,
+
-				);
+	const updateDestination = (destination: MessageEventBusDestinationOptions) => {
-				if (groupIndex > -1) {
+		if (destination.id && items.value[destination.id]) {
-					if (groupName === name) {
+			items.value[destination.id].destination = destination;
-						this.$patch((state) => {
+		}
-							state.items[destinationId].eventGroups[groupIndex].selected = isSelected;
+	};
-						});
+
-					} else {
+	const removeDestination = (destinationId: string) => {
-						const eventIndex = this.items[destinationId].eventGroups[groupIndex].children.findIndex(
+		if (!destinationId) return;
-							(e) => e.name === name,
+		delete items.value[destinationId];
-						);
+	};
-						if (eventIndex > -1) {
+
-							this.$patch((state) => {
+	const getSelectedEvents = (destinationId: string): string[] => {
-								state.items[destinationId].eventGroups[groupIndex].children[eventIndex].selected =
+		const selectedEvents: string[] = [];
-									isSelected;
+		if (items.value[destinationId]) {
-								if (isSelected) {
+			for (const group of items.value[destinationId].eventGroups) {
-									state.items[destinationId].eventGroups[groupIndex].indeterminate = isSelected;
+				if (group.selected) {
-								} else {
+					selectedEvents.push(group.name);
-									let anySelected = false;
+				}
-									for (
+				for (const event of group.children) {
-										let i = 0;
+					if (event.selected) {
-										i < state.items[destinationId].eventGroups[groupIndex].children.length;
+						selectedEvents.push(event.name);
-										i++
-									) {
-										anySelected =
-											anySelected ||
-											state.items[destinationId].eventGroups[groupIndex].children[i].selected;
-									}
-									state.items[destinationId].eventGroups[groupIndex].indeterminate = anySelected;
-								}
-							});
-						}
 					}
 				}
 			}
-		},
+		}
-		removeDestinationItemTree(id: string) {
+		return selectedEvents;
-			delete this.items[id];
+	};
-		},
-		clearDestinationItemTrees() {
-			this.items = {} as DestinationSettingsStore;
-		},
-		setSelectionAndBuildItems(destination: MessageEventBusDestinationOptions) {
-			if (destination.id) {
-				if (!this.items[destination.id]) {
-					this.items[destination.id] = {
-						destination,
-						selectedEvents: new Set<string>(),
-						eventGroups: [],
-						isNew: false,
-					} as DestinationStoreItem;
-				}
-				this.items[destination.id]?.selectedEvents?.clear();
-				if (destination.subscribedEvents) {
-					for (const eventName of destination.subscribedEvents) {
-						this.items[destination.id]?.selectedEvents?.add(eventName);
-					}
-				}
-				this.items[destination.id].eventGroups = eventGroupsFromStringList(
-					this.eventNames,
-					this.items[destination.id]?.selectedEvents,
-				);
-			}
-		},
-		async saveDestination(destination: MessageEventBusDestinationOptions): Promise<boolean> {
-			if (!hasDestinationId(destination)) {
-				return false;
-			}
 
-			const rootStore = useRootStore();
+	const saveDestination = async (
-			const selectedEvents = this.getSelectedEvents(destination.id);
+		destination: MessageEventBusDestinationOptions,
-			try {
+	): Promise<boolean> => {
-				await saveDestinationToDb(rootStore.restApiContext, destination, selectedEvents);
+		if (!hasDestinationId(destination)) {
-				this.updateDestination(destination);
+			return false;
-				return true;
+		}
-			} catch (e) {
-				return false;
-			}
-		},
-		async sendTestMessage(destination: MessageEventBusDestinationOptions): Promise<boolean> {
-			if (!hasDestinationId(destination)) {
-				return false;
-			}
 
-			const rootStore = useRootStore();
+		const selectedEvents = getSelectedEvents(destination.id);
-			const testResult = await sendTestMessageToDestination(rootStore.restApiContext, destination);
+		try {
-			return testResult;
+			await saveDestinationToDb(rootStore.restApiContext, destination, selectedEvents);
-		},
+			updateDestination(destination);
-		async fetchEventNames(): Promise<string[]> {
+			return true;
-			const rootStore = useRootStore();
+		} catch (e) {
-			return await getEventNamesFromBackend(rootStore.restApiContext);
+			return false;
-		},
+		}
-		async fetchDestinations(): Promise<MessageEventBusDestinationOptions[]> {
+	};
-			const rootStore = useRootStore();
+
-			return await getDestinationsFromBackend(rootStore.restApiContext);
+	const sendTestMessage = async (
-		},
+		destination: MessageEventBusDestinationOptions,
-		async deleteDestination(destinationId: string) {
+	): Promise<boolean> => {
-			const rootStore = useRootStore();
+		if (!hasDestinationId(destination)) {
-			await deleteDestinationFromDb(rootStore.restApiContext, destinationId);
+			return false;
-			this.removeDestination(destinationId);
+		}
-		},
+
-	},
+		const testResult = await sendTestMessageToDestination(rootStore.restApiContext, destination);
+		return testResult;
+	};
+
+	const fetchEventNames = async () => {
+		return await getEventNamesFromBackend(rootStore.restApiContext);
+	};
+
+	const fetchDestinations = async (): Promise<MessageEventBusDestinationOptions[]> => {
+		return await getDestinationsFromBackend(rootStore.restApiContext);
+	};
+
+	const deleteDestination = async (destinationId: string) => {
+		await deleteDestinationFromDb(rootStore.restApiContext, destinationId);
+		removeDestination(destinationId);
+	};
+
+	return {
+		addDestination,
+		setSelectionAndBuildItems,
+		getDestination,
+		getAllDestinations,
+		clearDestinations,
+		addEventName,
+		removeEventName,
+		clearEventNames,
+		addSelectedEvent,
+		removeSelectedEvent,
+		setSelectedInGroup,
+		removeDestinationItemTree,
+		updateDestination,
+		removeDestination,
+		getSelectedEvents,
+		saveDestination,
+		sendTestMessage,
+		fetchEventNames,
+		fetchDestinations,
+		deleteDestination,
+		items,
+	};
 });
 
-export function eventGroupFromEventName(eventName: string): string | undefined {
+export const eventGroupFromEventName = (eventName: string): string | undefined => {
 	const matches = eventName.match(/^[\w\s]+\.[\w\s]+/);
 	if (matches && matches?.length > 0) {
 		return matches[0];
 	}
 	return undefined;
-}
+};
 
-function prettifyEventName(label: string, group = ''): string {
+const prettifyEventName = (label: string, group = ''): string => {
 	label = label.replace(group + '.', '');
 	if (label.length > 0) {
 		label = label[0].toUpperCase() + label.substring(1);
 		label = label.replaceAll('.', ' ');
 	}
 	return label;
-}
+};
 
-export function eventGroupsFromStringList(
+export const eventGroupsFromStringList = (
 	dottedList: Set<string>,
 	selectionList: Set<string> = new Set(),
-) {
+) => {
 	const result = [] as EventSelectionGroup[];
 	const eventNameArray = Array.from(dottedList.values());
 
@@ -287,4 +309,4 @@ export function eventGroupsFromStringList(
 		result.push(collection);
 	}
 	return result;
-}
+};

packages/editor-ui/src/views/SettingsLogStreamingView.vue

@@ -95,7 +95,7 @@ export default defineComponent({
 		},
 		async getDestinationDataFromBackend(): Promise<void> {
 			this.logStreamingStore.clearEventNames();
-			this.logStreamingStore.clearDestinationItemTrees();
+			this.logStreamingStore.clearDestinations();
 			this.allDestinations = [];
 			const eventNamesData = await this.logStreamingStore.fetchEventNames();
 			if (eventNamesData) {