Merge branch 'master' into ai-assistant-node-data-improvements

* master:
  fix(core): Restore queue listeners for `webhook` process (#10781)
  refactor(benchmark): Rename scaling mode setups (#10783)
  refactor(core): Include AI events in log streaming relay (#10768)
  ci: Upgrade `validate-n8n-pull-request-title` to v2.2.0 (#10784)
  ci: Upgrade express to address CVE-2024-43796 (no-changelog) (#10785)
  fix(core): Fix the type for InstalledNodes.latestVersion (no-changelog) (#10782)
  fix(benchmark): Fix scaling mode setup (#10780)

.github/pull_request_title_conventions.md

@@ -13,7 +13,7 @@ A PR title consists of these elements:
   │       │
   │       └─⫸ Scope: API | benchmark | core | editor | * Node
   │
-  └─⫸ Type: build | ci | docs | feat | fix | perf | refactor | test
+  └─⫸ Type: build | ci | chore | docs | feat | fix | perf | refactor | test
 ```
 
 - PR title
@@ -41,6 +41,7 @@ Must be one of the following:
 | `refactor` | A behavior-neutral code change that neither fixes a bug nor adds a feature | ❌ |
 | `build` | Changes that affect the build system or external dependencies (TypeScript, Jest, pnpm, etc.) | ❌ |
 | `ci` | Changes to CI configuration files and scripts (e.g. Github actions) | ❌ |
+| `chore` | Routine tasks, maintenance, and minor updates not covered by other types | ❌ |
 
 > BREAKING CHANGES (see Footer section below), will **always** appear in the changelog unless suffixed with `no-changelog`.
 

.github/workflows/check-pr-title.yml

@@ -28,6 +28,6 @@ jobs:
 
       - name: Validate PR title
         id: validate_pr_title
-        uses: n8n-io/validate-n8n-pull-request-title@v2.1.0
+        uses: n8n-io/validate-n8n-pull-request-title@v2.2.0
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

packages/@n8n/benchmark/scripts/n8nSetups/postgres/setup.mjs

@@ -4,8 +4,7 @@ import path from 'path';
 import { fs } from 'zx';
 
 /**
- * Creates the needed directories for the queue setup so their
- * permissions get set correctly.
+ * Creates the needed directories so the permissions get set correctly.
  */
 export function setup({ runDir }) {
 	const neededDirs = ['n8n', 'postgres'];

packages/@n8n/benchmark/scripts/n8nSetups/scaling/docker-compose.yml → packages/@n8n/benchmark/scripts/n8nSetups/scaling-multi-main/docker-compose.yml

@@ -154,7 +154,7 @@ services:
       - DB_POSTGRESDB_PASSWORD=password
 
     volumes:
-      - ${RUN_DIR}/n8n-main:/n8n
+      - ${RUN_DIR}/n8n-main1:/n8n
     depends_on:
       n8n_worker1:
         condition: service_healthy

packages/@n8n/benchmark/scripts/n8nSetups/scaling/setup.mjs → packages/@n8n/benchmark/scripts/n8nSetups/scaling-multi-main/setup.mjs

@@ -4,11 +4,10 @@ import path from 'path';
 import { fs } from 'zx';
 
 /**
- * Creates the needed directories for the queue setup so their
- * permissions get set correctly.
+ * Creates the needed directories so the permissions get set correctly.
  */
 export function setup({ runDir }) {
-	const neededDirs = ['n8n-worker1', 'n8n-worker2', 'n8n-main1', 'n8n_main2', 'postgres'];
+	const neededDirs = ['n8n-worker1', 'n8n-worker2', 'n8n-main1', 'n8n-main2', 'postgres'];
 
 	for (const dir of neededDirs) {
 		fs.ensureDirSync(path.join(runDir, dir));

packages/@n8n/benchmark/scripts/n8nSetups/queue/setup.mjs → packages/@n8n/benchmark/scripts/n8nSetups/scaling-single-main/setup.mjs

@@ -4,8 +4,7 @@ import path from 'path';
 import { fs } from 'zx';
 
 /**
- * Creates the needed directories for the queue setup so their
- * permissions get set correctly.
+ * Creates the needed directories so the permissions get set correctly.
  */
 export function setup({ runDir }) {
 	const neededDirs = ['n8n-worker1', 'n8n-worker2', 'n8n-main', 'postgres'];

packages/@n8n/benchmark/scripts/n8nSetups/sqlite-legacy/setup.mjs

@@ -4,8 +4,7 @@ import path from 'path';
 import { fs } from 'zx';
 
 /**
- * Creates the needed directories for the queue setup so their
- * permissions get set correctly.
+ * Creates the needed directories so the permissions get set correctly.
  */
 export function setup({ runDir }) {
 	const neededDirs = ['n8n'];

packages/@n8n/benchmark/scripts/n8nSetups/sqlite/setup.mjs

@@ -4,8 +4,7 @@ import path from 'path';
 import { fs } from 'zx';
 
 /**
- * Creates the needed directories for the queue setup so their
- * permissions get set correctly.
+ * Creates the needed directories so the permissions get set correctly.
  */
 export function setup({ runDir }) {
 	const neededDirs = ['n8n'];

packages/@n8n/benchmark/scripts/runForN8nSetup.mjs

@@ -37,9 +37,9 @@ async function main() {
 	const duration = argv.duration;
 
 	const hasN8nLicense = !!n8nLicenseCert || !!n8nLicenseActivationKey;
-	if (n8nSetupToUse === 'scaling' && !hasN8nLicense) {
+	if (n8nSetupToUse === 'scaling-multi-main' && !hasN8nLicense) {
 		console.error(
-			'n8n license is required to run the scaling setup. Please provide N8N_LICENSE_CERT or N8N_LICENSE_ACTIVATION_KEY',
+			'n8n license is required to run the multi-main scaling setup. Please provide N8N_LICENSE_CERT or N8N_LICENSE_ACTIVATION_KEY (and N8N_LICENSE_TENANT_ID if needed)',
 		);
 		process.exit(1);
 	}

packages/@n8n/nodes-langchain/nodes/llms/N8nLlmTracing.ts

@@ -138,7 +138,7 @@ export class N8nLlmTracing extends BaseCallbackHandler {
 		this.executionFunctions.addOutputData(this.connectionType, runDetails.index, [
 			[{ json: { ...response } }],
 		]);
-		void logAiEvent(this.executionFunctions, 'n8n.ai.llm.generated', {
+		void logAiEvent(this.executionFunctions, 'ai-llm-generated-output', {
 			messages: parsedMessages,
 			options: runDetails.options,
 			response,
@@ -186,7 +186,7 @@ export class N8nLlmTracing extends BaseCallbackHandler {
 			});
 		}
 
-		void logAiEvent(this.executionFunctions, 'n8n.ai.llm.error', {
+		void logAiEvent(this.executionFunctions, 'ai-llm-errored', {
 			error: Object.keys(error).length === 0 ? error.toString() : error,
 			runId,
 			parentRunId,

packages/@n8n/nodes-langchain/nodes/vector_store/shared/createVectorStoreNode.ts

@@ -280,7 +280,7 @@ export const createVectorStoreNode = (args: VectorStoreNodeConstructorArgs) =>
 					});
 
 					resultData.push(...serializedDocs);
-					void logAiEvent(this, 'n8n.ai.vector.store.searched', { query: prompt });
+					void logAiEvent(this, 'ai-vector-store-searched', { query: prompt });
 				}
 
 				return [resultData];
@@ -307,7 +307,7 @@ export const createVectorStoreNode = (args: VectorStoreNodeConstructorArgs) =>
 					try {
 						await args.populateVectorStore(this, embeddings, processedDocuments, itemIndex);
 
-						void logAiEvent(this, 'n8n.ai.vector.store.populated');
+						void logAiEvent(this, 'ai-vector-store-populated');
 					} catch (error) {
 						throw error;
 					}
@@ -361,7 +361,7 @@ export const createVectorStoreNode = (args: VectorStoreNodeConstructorArgs) =>
 							ids: [documentId],
 						});
 
-						void logAiEvent(this, 'n8n.ai.vector.store.updated');
+						void logAiEvent(this, 'ai-vector-store-updated');
 					} catch (error) {
 						throw error;
 					}

packages/@n8n/nodes-langchain/utils/helpers.ts

@@ -1,10 +1,5 @@
 import { NodeConnectionType, NodeOperationError, jsonStringify } from 'n8n-workflow';
-import type {
-	EventNamesAiNodesType,
-	IDataObject,
-	IExecuteFunctions,
-	IWebhookFunctions,
-} from 'n8n-workflow';
+import type { AiEvent, IDataObject, IExecuteFunctions, IWebhookFunctions } from 'n8n-workflow';
 import type { BaseChatModel } from '@langchain/core/language_models/chat_models';
 import type { BaseOutputParser } from '@langchain/core/output_parsers';
 import type { BaseMessage } from '@langchain/core/messages';
@@ -155,7 +150,7 @@ export function getSessionId(
 
 export async function logAiEvent(
 	executeFunctions: IExecuteFunctions,
-	event: EventNamesAiNodesType,
+	event: AiEvent,
 	data?: IDataObject,
 ) {
 	try {

packages/@n8n/nodes-langchain/utils/logWrapper.ts

@@ -196,7 +196,7 @@ export function logWrapper(
 						const payload = { action: 'getMessages', response };
 						executeFunctions.addOutputData(connectionType, index, [[{ json: payload }]]);
 
-						void logAiEvent(executeFunctions, 'n8n.ai.memory.get.messages', { response });
+						void logAiEvent(executeFunctions, 'ai-messages-retrieved-from-memory', { response });
 						return response;
 					};
 				} else if (prop === 'addMessage' && 'addMessage' in target) {
@@ -213,7 +213,7 @@ export function logWrapper(
 							arguments: [message],
 						});
 
-						void logAiEvent(executeFunctions, 'n8n.ai.memory.added.message', { message });
+						void logAiEvent(executeFunctions, 'ai-message-added-to-memory', { message });
 						executeFunctions.addOutputData(connectionType, index, [[{ json: payload }]]);
 					};
 				}
@@ -238,13 +238,13 @@ export function logWrapper(
 								arguments: [stringifiedText],
 							})) as object;
 
-							void logAiEvent(executeFunctions, 'n8n.ai.output.parser.parsed', { text, response });
+							void logAiEvent(executeFunctions, 'ai-output-parsed', { text, response });
 							executeFunctions.addOutputData(connectionType, index, [
 								[{ json: { action: 'parse', response } }],
 							]);
 							return response;
 						} catch (error) {
-							void logAiEvent(executeFunctions, 'n8n.ai.output.parser.parsed', {
+							void logAiEvent(executeFunctions, 'ai-output-parsed', {
 								text,
 								response: error.message ?? error,
 							});
@@ -277,7 +277,7 @@ export function logWrapper(
 							arguments: [query, config],
 						})) as Array<Document<Record<string, any>>>;
 
-						void logAiEvent(executeFunctions, 'n8n.ai.retriever.get.relevant.documents', { query });
+						void logAiEvent(executeFunctions, 'ai-documents-retrieved', { query });
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 						return response;
 					};
@@ -302,7 +302,7 @@ export function logWrapper(
 							arguments: [documents],
 						})) as number[][];
 
-						void logAiEvent(executeFunctions, 'n8n.ai.embeddings.embedded.document');
+						void logAiEvent(executeFunctions, 'ai-document-embedded');
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 						return response;
 					};
@@ -322,7 +322,7 @@ export function logWrapper(
 							method: target[prop],
 							arguments: [query],
 						})) as number[];
-						void logAiEvent(executeFunctions, 'n8n.ai.embeddings.embedded.query');
+						void logAiEvent(executeFunctions, 'ai-query-embedded');
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 						return response;
 					};
@@ -367,7 +367,7 @@ export function logWrapper(
 							arguments: [item, itemIndex],
 						})) as number[];
 
-						void logAiEvent(executeFunctions, 'n8n.ai.document.processed');
+						void logAiEvent(executeFunctions, 'ai-document-processed');
 						executeFunctions.addOutputData(connectionType, index, [
 							[{ json: { response }, pairedItem: { item: itemIndex } }],
 						]);
@@ -393,7 +393,7 @@ export function logWrapper(
 							arguments: [text],
 						})) as string[];
 
-						void logAiEvent(executeFunctions, 'n8n.ai.text.splitter.split');
+						void logAiEvent(executeFunctions, 'ai-text-split');
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 						return response;
 					};
@@ -417,7 +417,7 @@ export function logWrapper(
 							arguments: [query],
 						})) as string;
 
-						void logAiEvent(executeFunctions, 'n8n.ai.tool.called', { query, response });
+						void logAiEvent(executeFunctions, 'ai-tool-called', { query, response });
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 						return response;
 					};
@@ -447,7 +447,7 @@ export function logWrapper(
 							arguments: [query, k, filter, _callbacks],
 						})) as Array<Document<Record<string, any>>>;
 
-						void logAiEvent(executeFunctions, 'n8n.ai.vector.store.searched', { query });
+						void logAiEvent(executeFunctions, 'ai-vector-store-searched', { query });
 						executeFunctions.addOutputData(connectionType, index, [[{ json: { response } }]]);
 
 						return response;

packages/cli/package.json

@@ -112,7 +112,7 @@
     "csrf": "3.1.0",
     "curlconverter": "3.21.0",
     "dotenv": "8.6.0",
-    "express": "4.19.2",
+    "express": "4.21.0",
     "express-async-errors": "3.1.1",
     "express-handlebars": "7.1.2",
     "express-openapi-validator": "5.3.3",

packages/cli/src/__tests__/workflow-execute-additional-data.test.ts

@@ -1,26 +1,25 @@
 import { VariablesService } from '@/environments/variables/variables.service.ee';
 import { mockInstance } from '@test/mocking';
-import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import { getBase } from '@/workflow-execute-additional-data';
 import Container from 'typedi';
 import { CredentialsHelper } from '@/credentials-helper';
 import { SecretsHelper } from '@/secrets-helpers';
+import { EventService } from '@/events/event.service';
 
 describe('WorkflowExecuteAdditionalData', () => {
-	const messageEventBus = mockInstance(MessageEventBus);
 	const variablesService = mockInstance(VariablesService);
 	variablesService.getAllCached.mockResolvedValue([]);
 	const credentialsHelper = mockInstance(CredentialsHelper);
 	const secretsHelper = mockInstance(SecretsHelper);
-	Container.set(MessageEventBus, messageEventBus);
+	const eventService = mockInstance(EventService);
 	Container.set(VariablesService, variablesService);
 	Container.set(CredentialsHelper, credentialsHelper);
 	Container.set(SecretsHelper, secretsHelper);
 
 	test('logAiEvent should call MessageEventBus', async () => {
 		const additionalData = await getBase('user-id');
 
-		const eventName = 'n8n.ai.memory.get.messages';
+		const eventName = 'ai-messages-retrieved-from-memory';
 		const payload = {
 			msg: 'test message',
 			executionId: '123',
@@ -30,12 +29,9 @@ describe('WorkflowExecuteAdditionalData', () => {
 			nodeType: 'n8n-memory',
 		};
 
-		await additionalData.logAiEvent(eventName, payload);
+		additionalData.logAiEvent(eventName, payload);
 
-		expect(messageEventBus.sendAiNodeEvent).toHaveBeenCalledTimes(1);
-		expect(messageEventBus.sendAiNodeEvent).toHaveBeenCalledWith({
-			eventName,
-			payload,
-		});
+		expect(eventService.emit).toHaveBeenCalledTimes(1);
+		expect(eventService.emit).toHaveBeenCalledWith(eventName, payload);
 	});
 });

packages/cli/src/databases/entities/installed-nodes.ts

@@ -10,7 +10,7 @@ export class InstalledNodes {
 	type: string;
 
 	@Column()
-	latestVersion: string;
+	latestVersion: number;
 
 	@ManyToOne('InstalledPackages', 'installedNodes')
 	@JoinColumn({ name: 'package', referencedColumnName: 'packageName' })

packages/cli/src/databases/repositories/installed-packages.repository.ts

@@ -35,7 +35,7 @@ export class InstalledPackagesRepository extends Repository<InstalledPackages> {
 				const installedNode = this.installedNodesRepository.create({
 					name: nodeTypes[loadedNode.name].type.description.displayName,
 					type: loadedNode.name,
-					latestVersion: loadedNode.version.toString(),
+					latestVersion: loadedNode.version,
 					package: { packageName },
 				});
 

packages/cli/src/eventbus/event-message-classes/event-message-ai-node.ts

@@ -1,5 +1,6 @@
 import { AbstractEventMessage, isEventMessageOptionsWithType } from './abstract-event-message';
-import type { EventNamesAiNodesType, JsonObject } from 'n8n-workflow';
+import type { JsonObject } from 'n8n-workflow';
+import type { EventNamesAiNodesType } from '.';
 import { EventMessageTypeNames } from 'n8n-workflow';
 import type { AbstractEventMessageOptions } from './abstract-event-message-options';
 import type { AbstractEventPayload } from './abstract-event-payload';

packages/cli/src/eventbus/event-message-classes/index.ts

@@ -4,7 +4,25 @@ import type { EventMessageExecution } from './event-message-execution';
 import type { EventMessageGeneric } from './event-message-generic';
 import type { EventMessageNode } from './event-message-node';
 import type { EventMessageWorkflow } from './event-message-workflow';
-import { eventNamesAiNodes, type EventNamesAiNodesType } from 'n8n-workflow';
+
+export const eventNamesAiNodes = [
+	'n8n.ai.memory.get.messages',
+	'n8n.ai.memory.added.message',
+	'n8n.ai.output.parser.parsed',
+	'n8n.ai.retriever.get.relevant.documents',
+	'n8n.ai.embeddings.embedded.document',
+	'n8n.ai.embeddings.embedded.query',
+	'n8n.ai.document.processed',
+	'n8n.ai.text.splitter.split',
+	'n8n.ai.tool.called',
+	'n8n.ai.vector.store.searched',
+	'n8n.ai.llm.generated',
+	'n8n.ai.llm.error',
+	'n8n.ai.vector.store.populated',
+	'n8n.ai.vector.store.updated',
+] as const;
+
+export type EventNamesAiNodesType = (typeof eventNamesAiNodes)[number];
 
 export const eventNamesWorkflow = [
 	'n8n.workflow.started',