refactor(core): Continue breaking dependency cycles (no-changelog) (#8545)

packages/cli/src/CredentialsHelper.ts

@@ -2,64 +2,43 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 /* eslint-disable @typescript-eslint/no-unsafe-return */
-/* eslint-disable @typescript-eslint/no-unsafe-call */
 import { Service } from 'typedi';
 import { Credentials, NodeExecuteFunctions } from 'n8n-core';
-import get from 'lodash/get';
 
 import type {
 	ICredentialDataDecryptedObject,
-	ICredentialsDecrypted,
 	ICredentialsExpressionResolveValues,
-	ICredentialTestFunction,
-	ICredentialTestRequestData,
 	IHttpRequestOptions,
 	INode,
 	INodeCredentialsDetails,
-	INodeCredentialTestResult,
-	INodeExecutionData,
 	INodeParameters,
 	INodeProperties,
 	INodeType,
 	IVersionedNodeType,
 	IRequestOptionsSimplified,
-	IRunExecutionData,
 	IWorkflowDataProxyAdditionalKeys,
 	WorkflowExecuteMode,
-	ITaskDataConnections,
 	IHttpRequestHelper,
 	INodeTypeData,
 	INodeTypes,
 	IWorkflowExecuteAdditionalData,
-	ICredentialTestFunctions,
 	IExecuteData,
 } from 'n8n-workflow';
-import {
+import { ICredentialsHelper, NodeHelpers, Workflow, ApplicationError } from 'n8n-workflow';
-	ICredentialsHelper,
-	VersionedNodeType,
-	NodeHelpers,
-	RoutingNode,
-	Workflow,
-	ErrorReporterProxy as ErrorReporter,
-	ApplicationError,
-} from 'n8n-workflow';
 
 import type { ICredentialsDb } from '@/Interfaces';
-import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
+
-import type { User } from '@db/entities/User';
 import type { CredentialsEntity } from '@db/entities/CredentialsEntity';
 import { NodeTypes } from '@/NodeTypes';
 import { CredentialTypes } from '@/CredentialTypes';
 import { CredentialsOverwrites } from '@/CredentialsOverwrites';
 import { RESPONSE_ERROR_MESSAGES } from './constants';
-import { isObjectLiteral } from './utils';
+
 import { Logger } from '@/Logger';
 import { CredentialsRepository } from '@db/repositories/credentials.repository';
 import { SharedCredentialsRepository } from '@db/repositories/sharedCredentials.repository';
 import { CredentialNotFoundError } from './errors/credential-not-found.error';
 
-const { OAUTH2_CREDENTIAL_TEST_SUCCEEDED, OAUTH2_CREDENTIAL_TEST_FAILED } = RESPONSE_ERROR_MESSAGES;
-
 const mockNode = {
 	name: '',
 	typeVersion: 1,
@@ -475,310 +454,6 @@ export class CredentialsHelper extends ICredentialsHelper {
 		await this.credentialsRepository.update(findQuery, newCredentialsData);
 	}
 
-	private static hasAccessToken(credentialsDecrypted: ICredentialsDecrypted) {
-		const oauthTokenData = credentialsDecrypted?.data?.oauthTokenData;
-
-		if (!isObjectLiteral(oauthTokenData)) return false;
-
-		return 'access_token' in oauthTokenData;
-	}
-
-	private getCredentialTestFunction(
-		credentialType: string,
-	): ICredentialTestFunction | ICredentialTestRequestData | undefined {
-		// Check if test is defined on credentials
-		const type = this.credentialTypes.getByName(credentialType);
-		if (type.test) {
-			return {
-				testRequest: type.test,
-			};
-		}
-
-		const supportedNodes = this.credentialTypes.getSupportedNodes(credentialType);
-		for (const nodeName of supportedNodes) {
-			const node = this.nodeTypes.getByName(nodeName);
-
-			// Always set to an array even if node is not versioned to not having
-			// to duplicate the logic
-			const allNodeTypes: INodeType[] = [];
-			if (node instanceof VersionedNodeType) {
-				// Node is versioned
-				allNodeTypes.push(...Object.values(node.nodeVersions));
-			} else {
-				// Node is not versioned
-				allNodeTypes.push(node as INodeType);
-			}
-
-			// Check each of the node versions for credential tests
-			for (const nodeType of allNodeTypes) {
-				// Check each of teh credentials
-				for (const { name, testedBy } of nodeType.description.credentials ?? []) {
-					if (
-						name === credentialType &&
-						this.credentialTypes.getParentTypes(name).includes('oAuth2Api')
-					) {
-						return async function oauth2CredTest(
-							this: ICredentialTestFunctions,
-							cred: ICredentialsDecrypted,
-						): Promise<INodeCredentialTestResult> {
-							return CredentialsHelper.hasAccessToken(cred)
-								? {
-										status: 'OK',
-										message: OAUTH2_CREDENTIAL_TEST_SUCCEEDED,
-								  }
-								: {
-										status: 'Error',
-										message: OAUTH2_CREDENTIAL_TEST_FAILED,
-								  };
-						};
-					}
-
-					if (name === credentialType && !!testedBy) {
-						if (typeof testedBy === 'string') {
-							if (node instanceof VersionedNodeType) {
-								// The node is versioned. So check all versions for test function
-								// starting with the latest
-								const versions = Object.keys(node.nodeVersions).sort().reverse();
-								for (const version of versions) {
-									const versionedNode = node.nodeVersions[parseInt(version, 10)];
-									const credentialTest = versionedNode.methods?.credentialTest;
-									if (credentialTest && testedBy in credentialTest) {
-										return credentialTest[testedBy];
-									}
-								}
-							}
-							// Test is defined as string which links to a function
-							return (node as unknown as INodeType).methods?.credentialTest![testedBy];
-						}
-
-						// Test is defined as JSON with a definition for the request to make
-						return {
-							nodeType,
-							testRequest: testedBy,
-						};
-					}
-				}
-			}
-		}
-
-		return undefined;
-	}
-
-	async testCredentials(
-		user: User,
-		credentialType: string,
-		credentialsDecrypted: ICredentialsDecrypted,
-	): Promise<INodeCredentialTestResult> {
-		const credentialTestFunction = this.getCredentialTestFunction(credentialType);
-		if (credentialTestFunction === undefined) {
-			return {
-				status: 'Error',
-				message: 'No testing function found for this credential.',
-			};
-		}
-
-		if (credentialsDecrypted.data) {
-			try {
-				const additionalData = await WorkflowExecuteAdditionalData.getBase(user.id);
-				credentialsDecrypted.data = this.applyDefaultsAndOverwrites(
-					additionalData,
-					credentialsDecrypted.data,
-					credentialType,
-					'internal' as WorkflowExecuteMode,
-					undefined,
-					undefined,
-					user.hasGlobalScope('externalSecret:use'),
-				);
-			} catch (error) {
-				this.logger.debug('Credential test failed', error);
-				return {
-					status: 'Error',
-					message: error.message.toString(),
-				};
-			}
-		}
-
-		if (typeof credentialTestFunction === 'function') {
-			// The credentials get tested via a function that is defined on the node
-			const credentialTestFunctions = NodeExecuteFunctions.getCredentialTestFunctions();
-
-			return credentialTestFunction.call(credentialTestFunctions, credentialsDecrypted);
-		}
-
-		// Credentials get tested via request instructions
-
-		// TODO: Temp workflows get created at multiple locations (for example also LoadNodeParameterOptions),
-		//       check if some of them are identical enough that it can be combined
-
-		let nodeType: INodeType;
-		if (credentialTestFunction.nodeType) {
-			nodeType = credentialTestFunction.nodeType;
-		} else {
-			nodeType = this.nodeTypes.getByNameAndVersion('n8n-nodes-base.noOp');
-		}
-
-		const node: INode = {
-			id: 'temp',
-			parameters: {},
-			name: 'Temp-Node',
-			type: nodeType.description.name,
-			typeVersion: Array.isArray(nodeType.description.version)
-				? nodeType.description.version.slice(-1)[0]
-				: nodeType.description.version,
-			position: [0, 0],
-			credentials: {
-				[credentialType]: {
-					id: credentialsDecrypted.id,
-					name: credentialsDecrypted.name,
-				},
-			},
-		};
-
-		const workflowData = {
-			nodes: [node],
-			connections: {},
-		};
-
-		const nodeTypeCopy: INodeType = {
-			description: {
-				...nodeType.description,
-				credentials: [
-					{
-						name: credentialType,
-						required: true,
-					},
-				],
-				properties: [
-					{
-						displayName: 'Temp',
-						name: 'temp',
-						type: 'string',
-						routing: {
-							request: credentialTestFunction.testRequest.request,
-						},
-						default: '',
-					},
-				],
-			},
-		};
-
-		mockNodesData[nodeTypeCopy.description.name] = {
-			sourcePath: '',
-			type: nodeTypeCopy,
-		};
-
-		const workflow = new Workflow({
-			nodes: workflowData.nodes,
-			connections: workflowData.connections,
-			active: false,
-			nodeTypes: mockNodeTypes,
-		});
-
-		const mode = 'internal';
-		const runIndex = 0;
-		const inputData: ITaskDataConnections = {
-			main: [[{ json: {} }]],
-		};
-		const connectionInputData: INodeExecutionData[] = [];
-		const runExecutionData: IRunExecutionData = {
-			resultData: {
-				runData: {},
-			},
-		};
-
-		const additionalData = await WorkflowExecuteAdditionalData.getBase(user.id, node.parameters);
-
-		const routingNode = new RoutingNode(
-			workflow,
-			node,
-			connectionInputData,
-			runExecutionData ?? null,
-			additionalData,
-			mode,
-		);
-
-		let response: INodeExecutionData[][] | null | undefined;
-
-		try {
-			response = await routingNode.runNode(
-				inputData,
-				runIndex,
-				nodeTypeCopy,
-				{ node, data: {}, source: null },
-				NodeExecuteFunctions,
-				credentialsDecrypted,
-			);
-		} catch (error) {
-			ErrorReporter.error(error);
-			// Do not fail any requests to allow custom error messages and
-			// make logic easier
-			if (error.cause?.response) {
-				const errorResponseData = {
-					statusCode: error.cause.response.status,
-					statusMessage: error.cause.response.statusText,
-				};
-				if (credentialTestFunction.testRequest.rules) {
-					// Special testing rules are defined so check all in order
-					for (const rule of credentialTestFunction.testRequest.rules) {
-						if (rule.type === 'responseCode') {
-							if (errorResponseData.statusCode === rule.properties.value) {
-								return {
-									status: 'Error',
-									message: rule.properties.message,
-								};
-							}
-						}
-					}
-				}
-
-				if (errorResponseData.statusCode < 199 || errorResponseData.statusCode > 299) {
-					// All requests with response codes that are not 2xx are treated by default as failed
-					return {
-						status: 'Error',
-						message:
-							errorResponseData.statusMessage ||
-							`Received HTTP status code: ${errorResponseData.statusCode}`,
-					};
-				}
-			} else if (error.cause?.code) {
-				return {
-					status: 'Error',
-					message: error.cause.code,
-				};
-			}
-			this.logger.debug('Credential test failed', error);
-			return {
-				status: 'Error',
-				message: error.message.toString(),
-			};
-		} finally {
-			delete mockNodesData[nodeTypeCopy.description.name];
-		}
-
-		if (
-			credentialTestFunction.testRequest.rules &&
-			Array.isArray(credentialTestFunction.testRequest.rules)
-		) {
-			// Special testing rules are defined so check all in order
-			for (const rule of credentialTestFunction.testRequest.rules) {
-				if (rule.type === 'responseSuccessBody') {
-					const responseData = response![0][0].json;
-					if (get(responseData, rule.properties.key) === rule.properties.value) {
-						return {
-							status: 'Error',
-							message: rule.properties.message,
-						};
-					}
-				}
-			}
-		}
-
-		return {
-			status: 'OK',
-			message: 'Connection successful!',
-		};
-	}
-
 	async credentialOwnedByOwner(nodeCredential: INodeCredentialsDetails): Promise<boolean> {
 		if (!nodeCredential.id) {
 			return false;

packages/cli/src/InternalHooks.ts

@@ -21,7 +21,8 @@ import type { GlobalRole, User } from '@db/entities/User';
 import type { ExecutionMetadata } from '@db/entities/ExecutionMetadata';
 import { SharedWorkflowRepository } from '@db/repositories/sharedWorkflow.repository';
 import { WorkflowRepository } from '@db/repositories/workflow.repository';
-import { MessageEventBus, type EventPayloadWorkflow } from '@/eventbus';
+import type { EventPayloadWorkflow } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { determineFinalExecutionStatus } from '@/executionLifecycleHooks/shared/sharedHookFunctions';
 import type {
 	ITelemetryUserDeletionData,

packages/cli/src/Server.ts

@@ -66,7 +66,7 @@ import { setupAuthMiddlewares } from './middlewares';
 import { isLdapEnabled } from './Ldap/helpers';
 import { AbstractServer } from './AbstractServer';
 import { PostHogClient } from './posthog';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { InternalHooks } from './InternalHooks';
 import { SamlController } from './sso/saml/routes/saml.controller.ee';
 import { SamlService } from './sso/saml/saml.service.ee';

packages/cli/src/WorkflowRunner.ts

@@ -25,7 +25,7 @@ import PCancelable from 'p-cancelable';
 import { ActiveExecutions } from '@/ActiveExecutions';
 import config from '@/config';
 import { ExecutionRepository } from '@db/repositories/execution.repository';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { ExecutionDataRecoveryService } from '@/eventbus/executionDataRecovery.service';
 import { ExternalHooks } from '@/ExternalHooks';
 import type { IExecutionResponse, IWorkflowExecutionDataProcess } from '@/Interfaces';

packages/cli/src/commands/start.ts

@@ -15,7 +15,7 @@ import { ActiveExecutions } from '@/ActiveExecutions';
 import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
 import { Server } from '@/Server';
 import { EDITOR_UI_DIST_DIR, LICENSE_FEATURES } from '@/constants';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { InternalHooks } from '@/InternalHooks';
 import { License } from '@/License';
 import { OrchestrationService } from '@/services/orchestration.service';

packages/cli/src/commands/worker.ts

@@ -29,7 +29,7 @@ import { OwnershipService } from '@/services/ownership.service';
 import type { ICredentialsOverwrite } from '@/Interfaces';
 import { CredentialsOverwrites } from '@/CredentialsOverwrites';
 import { rawBodyReader, bodyParser } from '@/middlewares';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import type { RedisServicePubSubSubscriber } from '@/services/redis/RedisServicePubSubSubscriber';
 import { EventMessageGeneric } from '@/eventbus/EventMessageClasses/EventMessageGeneric';
 import { OrchestrationHandlerWorkerService } from '@/services/orchestration/worker/orchestration.handler.worker.service';

packages/cli/src/controllers/e2e.controller.ts

@@ -4,7 +4,7 @@ import config from '@/config';
 import { SettingsRepository } from '@db/repositories/settings.repository';
 import { UserRepository } from '@db/repositories/user.repository';
 import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { License } from '@/License';
 import { LICENSE_FEATURES, inE2ETests } from '@/constants';
 import { NoAuthRequired, Patch, Post, RestController } from '@/decorators';

packages/cli/src/credentials/credentials.service.ts

@@ -10,7 +10,7 @@ import type { FindOptionsWhere } from 'typeorm';
 import type { Scope } from '@n8n/permissions';
 import * as Db from '@/Db';
 import type { ICredentialsDb } from '@/Interfaces';
-import { CredentialsHelper, createCredentialsFromCredentialsEntity } from '@/CredentialsHelper';
+import { createCredentialsFromCredentialsEntity } from '@/CredentialsHelper';
 import { CREDENTIAL_BLANKING_VALUE } from '@/constants';
 import { CredentialsEntity } from '@db/entities/CredentialsEntity';
 import { SharedCredentials } from '@db/entities/SharedCredentials';
@@ -24,6 +24,7 @@ import { Logger } from '@/Logger';
 import { CredentialsRepository } from '@db/repositories/credentials.repository';
 import { SharedCredentialsRepository } from '@db/repositories/sharedCredentials.repository';
 import { Service } from 'typedi';
+import { CredentialsTester } from '@/services/credentials-tester.service';
 
 export type CredentialsGetSharedOptions =
 	| { allowGlobalScope: true; globalScope: Scope }
@@ -36,7 +37,7 @@ export class CredentialsService {
 		private readonly sharedCredentialsRepository: SharedCredentialsRepository,
 		private readonly ownershipService: OwnershipService,
 		private readonly logger: Logger,
-		private readonly credenntialsHelper: CredentialsHelper,
+		private readonly credentialsTester: CredentialsTester,
 		private readonly externalHooks: ExternalHooks,
 		private readonly credentialTypes: CredentialTypes,
 	) {}
@@ -218,7 +219,7 @@ export class CredentialsService {
 	}
 
 	async test(user: User, credentials: ICredentialsDecrypted) {
-		return await this.credenntialsHelper.testCredentials(user, credentials.type, credentials);
+		return await this.credentialsTester.testCredentials(user, credentials.type, credentials);
 	}
 
 	// Take data and replace all sensitive values with a sentinel value.

packages/cli/src/eventbus/index.ts

@@ -1,4 +1,3 @@
-export { MessageEventBus } from './MessageEventBus/MessageEventBus';
 export { EventMessageTypes } from './EventMessageClasses';
 export { EventPayloadWorkflow } from './EventMessageClasses/EventMessageWorkflow';
 export { METRICS_EVENT_NAME, getLabelsForEvent } from './MessageEventBusDestination/Helpers.ee';

packages/cli/src/services/credentials-tester.service.ts

@@ -0,0 +1,382 @@
+/* eslint-disable @typescript-eslint/no-unsafe-argument */
+/* eslint-disable @typescript-eslint/no-unsafe-member-access */
+/* eslint-disable @typescript-eslint/no-unsafe-assignment */
+/* eslint-disable @typescript-eslint/no-unsafe-return */
+/* eslint-disable @typescript-eslint/no-unsafe-call */
+import { Service } from 'typedi';
+import { NodeExecuteFunctions } from 'n8n-core';
+import get from 'lodash/get';
+
+import type {
+	ICredentialsDecrypted,
+	ICredentialTestFunction,
+	ICredentialTestRequestData,
+	INode,
+	INodeCredentialTestResult,
+	INodeExecutionData,
+	INodeProperties,
+	INodeType,
+	IVersionedNodeType,
+	IRunExecutionData,
+	WorkflowExecuteMode,
+	ITaskDataConnections,
+	INodeTypeData,
+	INodeTypes,
+	ICredentialTestFunctions,
+} from 'n8n-workflow';
+import {
+	VersionedNodeType,
+	NodeHelpers,
+	RoutingNode,
+	Workflow,
+	ErrorReporterProxy as ErrorReporter,
+	ApplicationError,
+} from 'n8n-workflow';
+
+import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
+import type { User } from '@db/entities/User';
+import { NodeTypes } from '@/NodeTypes';
+import { CredentialTypes } from '@/CredentialTypes';
+import { RESPONSE_ERROR_MESSAGES } from '../constants';
+import { isObjectLiteral } from '../utils';
+import { Logger } from '@/Logger';
+import { CredentialsHelper } from '../CredentialsHelper';
+
+const { OAUTH2_CREDENTIAL_TEST_SUCCEEDED, OAUTH2_CREDENTIAL_TEST_FAILED } = RESPONSE_ERROR_MESSAGES;
+
+const mockNodesData: INodeTypeData = {
+	mock: {
+		sourcePath: '',
+		type: {
+			description: { properties: [] as INodeProperties[] },
+		} as INodeType,
+	},
+};
+
+const mockNodeTypes: INodeTypes = {
+	getByName(nodeType: string): INodeType | IVersionedNodeType {
+		return mockNodesData[nodeType]?.type;
+	},
+	getByNameAndVersion(nodeType: string, version?: number): INodeType {
+		if (!mockNodesData[nodeType]) {
+			throw new ApplicationError(RESPONSE_ERROR_MESSAGES.NO_NODE, {
+				tags: { nodeType },
+			});
+		}
+		return NodeHelpers.getVersionedNodeType(mockNodesData[nodeType].type, version);
+	},
+};
+
+@Service()
+export class CredentialsTester {
+	constructor(
+		private readonly logger: Logger,
+		private readonly credentialTypes: CredentialTypes,
+		private readonly nodeTypes: NodeTypes,
+		private readonly credentialsHelper: CredentialsHelper,
+	) {}
+
+	private static hasAccessToken(credentialsDecrypted: ICredentialsDecrypted) {
+		const oauthTokenData = credentialsDecrypted?.data?.oauthTokenData;
+
+		if (!isObjectLiteral(oauthTokenData)) return false;
+
+		return 'access_token' in oauthTokenData;
+	}
+
+	private getCredentialTestFunction(
+		credentialType: string,
+	): ICredentialTestFunction | ICredentialTestRequestData | undefined {
+		// Check if test is defined on credentials
+		const type = this.credentialTypes.getByName(credentialType);
+		if (type.test) {
+			return {
+				testRequest: type.test,
+			};
+		}
+
+		const supportedNodes = this.credentialTypes.getSupportedNodes(credentialType);
+		for (const nodeName of supportedNodes) {
+			const node = this.nodeTypes.getByName(nodeName);
+
+			// Always set to an array even if node is not versioned to not having
+			// to duplicate the logic
+			const allNodeTypes: INodeType[] = [];
+			if (node instanceof VersionedNodeType) {
+				// Node is versioned
+				allNodeTypes.push(...Object.values(node.nodeVersions));
+			} else {
+				// Node is not versioned
+				allNodeTypes.push(node as INodeType);
+			}
+
+			// Check each of the node versions for credential tests
+			for (const nodeType of allNodeTypes) {
+				// Check each of teh credentials
+				for (const { name, testedBy } of nodeType.description.credentials ?? []) {
+					if (
+						name === credentialType &&
+						this.credentialTypes.getParentTypes(name).includes('oAuth2Api')
+					) {
+						return async function oauth2CredTest(
+							this: ICredentialTestFunctions,
+							cred: ICredentialsDecrypted,
+						): Promise<INodeCredentialTestResult> {
+							return CredentialsTester.hasAccessToken(cred)
+								? {
+										status: 'OK',
+										message: OAUTH2_CREDENTIAL_TEST_SUCCEEDED,
+								  }
+								: {
+										status: 'Error',
+										message: OAUTH2_CREDENTIAL_TEST_FAILED,
+								  };
+						};
+					}
+
+					if (name === credentialType && !!testedBy) {
+						if (typeof testedBy === 'string') {
+							if (node instanceof VersionedNodeType) {
+								// The node is versioned. So check all versions for test function
+								// starting with the latest
+								const versions = Object.keys(node.nodeVersions).sort().reverse();
+								for (const version of versions) {
+									const versionedNode = node.nodeVersions[parseInt(version, 10)];
+									const credentialTest = versionedNode.methods?.credentialTest;
+									if (credentialTest && testedBy in credentialTest) {
+										return credentialTest[testedBy];
+									}
+								}
+							}
+							// Test is defined as string which links to a function
+							return (node as unknown as INodeType).methods?.credentialTest![testedBy];
+						}
+
+						// Test is defined as JSON with a definition for the request to make
+						return {
+							nodeType,
+							testRequest: testedBy,
+						};
+					}
+				}
+			}
+		}
+
+		return undefined;
+	}
+
+	async testCredentials(
+		user: User,
+		credentialType: string,
+		credentialsDecrypted: ICredentialsDecrypted,
+	): Promise<INodeCredentialTestResult> {
+		const credentialTestFunction = this.getCredentialTestFunction(credentialType);
+		if (credentialTestFunction === undefined) {
+			return {
+				status: 'Error',
+				message: 'No testing function found for this credential.',
+			};
+		}
+
+		if (credentialsDecrypted.data) {
+			try {
+				const additionalData = await WorkflowExecuteAdditionalData.getBase(user.id);
+				credentialsDecrypted.data = this.credentialsHelper.applyDefaultsAndOverwrites(
+					additionalData,
+					credentialsDecrypted.data,
+					credentialType,
+					'internal' as WorkflowExecuteMode,
+					undefined,
+					undefined,
+					user.hasGlobalScope('externalSecret:use'),
+				);
+			} catch (error) {
+				this.logger.debug('Credential test failed', error);
+				return {
+					status: 'Error',
+					message: error.message.toString(),
+				};
+			}
+		}
+
+		if (typeof credentialTestFunction === 'function') {
+			// The credentials get tested via a function that is defined on the node
+			const credentialTestFunctions = NodeExecuteFunctions.getCredentialTestFunctions();
+
+			return credentialTestFunction.call(credentialTestFunctions, credentialsDecrypted);
+		}
+
+		// Credentials get tested via request instructions
+
+		// TODO: Temp workflows get created at multiple locations (for example also LoadNodeParameterOptions),
+		//       check if some of them are identical enough that it can be combined
+
+		let nodeType: INodeType;
+		if (credentialTestFunction.nodeType) {
+			nodeType = credentialTestFunction.nodeType;
+		} else {
+			nodeType = this.nodeTypes.getByNameAndVersion('n8n-nodes-base.noOp');
+		}
+
+		const node: INode = {
+			id: 'temp',
+			parameters: {},
+			name: 'Temp-Node',
+			type: nodeType.description.name,
+			typeVersion: Array.isArray(nodeType.description.version)
+				? nodeType.description.version.slice(-1)[0]
+				: nodeType.description.version,
+			position: [0, 0],
+			credentials: {
+				[credentialType]: {
+					id: credentialsDecrypted.id,
+					name: credentialsDecrypted.name,
+				},
+			},
+		};
+
+		const workflowData = {
+			nodes: [node],
+			connections: {},
+		};
+
+		const nodeTypeCopy: INodeType = {
+			description: {
+				...nodeType.description,
+				credentials: [
+					{
+						name: credentialType,
+						required: true,
+					},
+				],
+				properties: [
+					{
+						displayName: 'Temp',
+						name: 'temp',
+						type: 'string',
+						routing: {
+							request: credentialTestFunction.testRequest.request,
+						},
+						default: '',
+					},
+				],
+			},
+		};
+
+		mockNodesData[nodeTypeCopy.description.name] = {
+			sourcePath: '',
+			type: nodeTypeCopy,
+		};
+
+		const workflow = new Workflow({
+			nodes: workflowData.nodes,
+			connections: workflowData.connections,
+			active: false,
+			nodeTypes: mockNodeTypes,
+		});
+
+		const mode = 'internal';
+		const runIndex = 0;
+		const inputData: ITaskDataConnections = {
+			main: [[{ json: {} }]],
+		};
+		const connectionInputData: INodeExecutionData[] = [];
+		const runExecutionData: IRunExecutionData = {
+			resultData: {
+				runData: {},
+			},
+		};
+
+		const additionalData = await WorkflowExecuteAdditionalData.getBase(user.id, node.parameters);
+
+		const routingNode = new RoutingNode(
+			workflow,
+			node,
+			connectionInputData,
+			runExecutionData ?? null,
+			additionalData,
+			mode,
+		);
+
+		let response: INodeExecutionData[][] | null | undefined;
+
+		try {
+			response = await routingNode.runNode(
+				inputData,
+				runIndex,
+				nodeTypeCopy,
+				{ node, data: {}, source: null },
+				NodeExecuteFunctions,
+				credentialsDecrypted,
+			);
+		} catch (error) {
+			ErrorReporter.error(error);
+			// Do not fail any requests to allow custom error messages and
+			// make logic easier
+			if (error.cause?.response) {
+				const errorResponseData = {
+					statusCode: error.cause.response.status,
+					statusMessage: error.cause.response.statusText,
+				};
+				if (credentialTestFunction.testRequest.rules) {
+					// Special testing rules are defined so check all in order
+					for (const rule of credentialTestFunction.testRequest.rules) {
+						if (rule.type === 'responseCode') {
+							if (errorResponseData.statusCode === rule.properties.value) {
+								return {
+									status: 'Error',
+									message: rule.properties.message,
+								};
+							}
+						}
+					}
+				}
+
+				if (errorResponseData.statusCode < 199 || errorResponseData.statusCode > 299) {
+					// All requests with response codes that are not 2xx are treated by default as failed
+					return {
+						status: 'Error',
+						message:
+							errorResponseData.statusMessage ||
+							`Received HTTP status code: ${errorResponseData.statusCode}`,
+					};
+				}
+			} else if (error.cause?.code) {
+				return {
+					status: 'Error',
+					message: error.cause.code,
+				};
+			}
+			this.logger.debug('Credential test failed', error);
+			return {
+				status: 'Error',
+				message: error.message.toString(),
+			};
+		} finally {
+			delete mockNodesData[nodeTypeCopy.description.name];
+		}
+
+		if (
+			credentialTestFunction.testRequest.rules &&
+			Array.isArray(credentialTestFunction.testRequest.rules)
+		) {
+			// Special testing rules are defined so check all in order
+			for (const rule of credentialTestFunction.testRequest.rules) {
+				if (rule.type === 'responseSuccessBody') {
+					const responseData = response![0][0].json;
+					if (get(responseData, rule.properties.key) === rule.properties.value) {
+						return {
+							status: 'Error',
+							message: rule.properties.message,
+						};
+					}
+				}
+			}
+		}
+
+		return {
+			status: 'OK',
+			message: 'Connection successful!',
+		};
+	}
+}

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

@@ -8,12 +8,8 @@ import { Service } from 'typedi';
 import EventEmitter from 'events';
 
 import { CacheService } from '@/services/cache/cache.service';
-import {
+import { METRICS_EVENT_NAME, getLabelsForEvent, type EventMessageTypes } from '@/eventbus';
-	MessageEventBus,
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
-	METRICS_EVENT_NAME,
-	getLabelsForEvent,
-	type EventMessageTypes,
-} from '@/eventbus';
 import { Logger } from '@/Logger';
 
 @Service()

packages/cli/test/integration/eventbus.ee.test.ts

@@ -16,7 +16,7 @@ import {
 } from 'n8n-workflow';
 
 import type { User } from '@db/entities/User';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { EventMessageGeneric } from '@/eventbus/EventMessageClasses/EventMessageGeneric';
 import type { MessageEventBusDestinationSyslog } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationSyslog.ee';
 import type { MessageEventBusDestinationWebhook } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationWebhook.ee';

packages/cli/test/integration/eventbus.test.ts

@@ -1,7 +1,7 @@
 import type { SuperAgentTest } from 'supertest';
 
 import type { User } from '@db/entities/User';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { ExecutionDataRecoveryService } from '@/eventbus/executionDataRecovery.service';
 
 import * as utils from './shared/utils/';

packages/cli/test/integration/workflows/workflow.service.test.ts

@@ -3,7 +3,7 @@ import { mock } from 'jest-mock-extended';
 import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
 import { SharedWorkflowRepository } from '@db/repositories/sharedWorkflow.repository';
 import { WorkflowRepository } from '@db/repositories/workflow.repository';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { Telemetry } from '@/telemetry';
 import { OrchestrationService } from '@/services/orchestration.service';
 import { WorkflowService } from '@/workflows/workflow.service';

packages/cli/test/unit/services/orchestration.service.test.ts

@@ -2,7 +2,7 @@ import Container from 'typedi';
 import config from '@/config';
 import { OrchestrationService } from '@/services/orchestration.service';
 import type { RedisServiceWorkerResponseObject } from '@/services/redis/RedisServiceCommands';
-import { MessageEventBus } from '@/eventbus';
+import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
 import { RedisService } from '@/services/redis.service';
 import { handleWorkerResponseMessageMain } from '@/services/orchestration/main/handleWorkerResponseMessageMain';
 import { handleCommandMessageMain } from '@/services/orchestration/main/handleCommandMessageMain';