chore(core): Move all event bus destination code to dedicated module (#24117)

packages/@n8n/backend-common/src/modules/__tests__/module-registry.test.ts

@@ -31,6 +31,7 @@ describe('eligibleModules', () => {
 			'dynamic-credentials',
 			'chat-hub',
 			'sso-oidc',
+			'log-streaming',
 		]);
 	});
 
@@ -48,6 +49,7 @@ describe('eligibleModules', () => {
 			'dynamic-credentials',
 			'chat-hub',
 			'sso-oidc',
+			'log-streaming',
 		]);
 	});
 

packages/@n8n/backend-common/src/modules/module-registry.ts

@@ -41,6 +41,7 @@ export class ModuleRegistry {
 		'dynamic-credentials',
 		'chat-hub',
 		'sso-oidc',
+		'log-streaming',
 	];
 
 	private readonly activeModules: string[] = [];

packages/@n8n/backend-common/src/modules/modules.config.ts

@@ -14,6 +14,7 @@ export const MODULE_NAMES = [
 	'dynamic-credentials',
 	'chat-hub',
 	'sso-oidc',
+	'log-streaming',
 ] as const;
 
 export type ModuleName = (typeof MODULE_NAMES)[number];

packages/@n8n/db/src/entities/index.ts

@@ -5,7 +5,6 @@ import { AuthIdentity } from './auth-identity';
 import { AuthProviderSyncHistory } from './auth-provider-sync-history';
 import { BinaryDataFile, SourceTypeSchema, type SourceType } from './binary-data-file';
 import { CredentialsEntity } from './credentials-entity';
-import { EventDestinations } from './event-destinations';
 import { ExecutionAnnotation } from './execution-annotation.ee';
 import { ExecutionData } from './execution-data';
 import { ExecutionEntity } from './execution-entity';
@@ -35,7 +34,6 @@ import { WorkflowStatistics } from './workflow-statistics';
 import { WorkflowTagMapping } from './workflow-tag-mapping';
 
 export {
-	EventDestinations,
 	InvalidAuthToken,
 	ProcessedData,
 	Settings,
@@ -75,7 +73,6 @@ export {
 };
 
 export const entities = {
-	EventDestinations,
 	InvalidAuthToken,
 	ProcessedData,
 	Settings,

packages/@n8n/db/src/repositories/index.ts

@@ -9,7 +9,6 @@ export { ExecutionAnnotationRepository } from './execution-annotation.repository
 export { ExecutionDataRepository } from './execution-data.repository';
 export { ExecutionMetadataRepository } from './execution-metadata.repository';
 export { ExecutionRepository } from './execution.repository';
-export { EventDestinationsRepository } from './event-destinations.repository';
 export { FolderRepository } from './folder.repository';
 export { FolderTagMappingRepository } from './folder-tag-mapping.repository';
 export { ScopeRepository } from './scope.repository';

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

@@ -27,6 +27,7 @@ import { CacheService } from '@/services/cache/cache.service';
 import { FrontendService } from '@/services/frontend.service';
 import { PasswordUtility } from '@/services/password.utility';
 import { ExecutionsConfig } from '@n8n/config';
+import { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service';
 
 if (!inE2ETests) {
 	Container.get(Logger).error('E2E endpoints only allowed during E2E tests');
@@ -170,6 +171,7 @@ export class E2EController {
 		private readonly userRepository: UserRepository,
 		private readonly frontendService: FrontendService,
 		private readonly executionsConfig: ExecutionsConfig,
+		private readonly logStreamingDestinationsService: LogStreamingDestinationService,
 	) {
 		license.isLicensed = (feature: BooleanLicenseFeature) => this.enabledFeatures[feature] ?? false;
 
@@ -282,7 +284,7 @@ export class E2EController {
 	private async resetLogStreaming() {
 		for (const id in this.eventBus.destinations) {
 			await this.eventBus.removeDestination(id, false);
-			await this.eventBus.deleteDestination(id);
+			await this.logStreamingDestinationsService.deleteDestinationFromDb(id);
 		}
 	}
 

packages/cli/src/eventbus/message-event-bus/message-event-bus.ts

@@ -1,11 +1,9 @@
 import { Logger } from '@n8n/backend-common';
 import { GlobalConfig } from '@n8n/config';
-import { EventDestinationsRepository, ExecutionRepository, WorkflowRepository } from '@n8n/db';
+import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
 import { OnPubSubEvent } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 // eslint-disable-next-line n8n-local-rules/misplaced-n8n-typeorm-import
-import type { DeleteResult } from '@n8n/typeorm';
-// eslint-disable-next-line n8n-local-rules/misplaced-n8n-typeorm-import
 import { In, IsNull, Not } from '@n8n/typeorm';
 import EventEmitter from 'events';
 import uniqby from 'lodash/uniqBy';
@@ -37,8 +35,6 @@ import type { EventMessageRunnerOptions } from '../event-message-classes/event-m
 import { EventMessageRunner } from '../event-message-classes/event-message-runner';
 import type { EventMessageWorkflowOptions } from '../event-message-classes/event-message-workflow';
 import { EventMessageWorkflow } from '../event-message-classes/event-message-workflow';
-import { messageEventBusDestinationFromDb } from '../message-event-bus-destination/message-event-bus-destination-from-db';
-import type { MessageEventBusDestination } from '../message-event-bus-destination/message-event-bus-destination.ee';
 import { MessageEventBusLogWriter } from '../message-event-bus-writer/message-event-bus-log-writer';
 
 export type EventMessageReturnMode = 'sent' | 'unsent' | 'all' | 'unfinished';
@@ -54,6 +50,17 @@ export interface MessageEventBusInitializeOptions {
 	webhookProcessorId?: string;
 }
 
+// TODO: remove startListening and potential close method,
+// As it should be the responsibility of the event bus to listen and close destinations
+export interface MessageEventBusDestinationType {
+	getId(): string;
+	hasSubscribedToEvent(msg: EventMessageTypes): boolean;
+	receiveFromEventBus(messageWithCallback: MessageWithCallback): Promise<boolean>;
+	startListening(): void;
+	close(): Promise<void>;
+	serialize(): MessageEventBusDestinationOptions;
+}
+
 @Service()
 // TODO: Convert to TypedEventEmitter
 // eslint-disable-next-line n8n-local-rules/no-type-unsafe-event-emitter
@@ -63,7 +70,7 @@ export class MessageEventBus extends EventEmitter {
 	logWriter: MessageEventBusLogWriter;
 
 	destinations: {
-		[key: string]: MessageEventBusDestination;
+		[key: string]: MessageEventBusDestinationType;
 	} = {};
 
 	private pushIntervalTimer: NodeJS.Timeout;
@@ -71,7 +78,6 @@ export class MessageEventBus extends EventEmitter {
 	constructor(
 		private readonly logger: Logger,
 		private readonly executionRepository: ExecutionRepository,
-		private readonly eventDestinationsRepository: EventDestinationsRepository,
 		private readonly workflowRepository: WorkflowRepository,
 		private readonly publisher: Publisher,
 		private readonly recoveryService: ExecutionRecoveryService,
@@ -97,21 +103,6 @@ export class MessageEventBus extends EventEmitter {
 
 		this.logger.debug('Initializing event bus...');
 
-		const savedEventDestinations = await this.eventDestinationsRepository.find({});
-		if (savedEventDestinations.length > 0) {
-			for (const destinationData of savedEventDestinations) {
-				try {
-					const destination = messageEventBusDestinationFromDb(this, destinationData);
-					if (destination) {
-						await this.addDestination(destination, false);
-					}
-				} catch (error) {
-					// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-					if (error.message) this.logger.debug(error.message as string);
-				}
-			}
-		}
-
 		this.logger.debug('Initializing event writer');
 		if (options?.workerId || options?.webhookProcessorId) {
 			// only add 'worker' or 'webhook-processor' to log file name since the ID changes on every start and we
@@ -235,7 +226,7 @@ export class MessageEventBus extends EventEmitter {
 		this.isInitialized = true;
 	}
 
-	async addDestination(destination: MessageEventBusDestination, notifyWorkers: boolean = true) {
+	async addDestination(destination: MessageEventBusDestinationType, notifyWorkers: boolean = true) {
 		await this.removeDestination(destination.getId(), false);
 		this.destinations[destination.getId()] = destination;
 		this.destinations[destination.getId()].startListening();
@@ -265,12 +256,6 @@ export class MessageEventBus extends EventEmitter {
 		}
 	}
 
-	async deleteDestination(id: string): Promise<DeleteResult | undefined> {
-		return await this.eventDestinationsRepository.delete({
-			id,
-		});
-	}
-
 	private async trySendingUnsent(msgs?: EventMessageTypes[]) {
 		const unsentMessages = msgs ?? (await this.getEventsUnsent());
 		if (unsentMessages.length > 0) {

packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts

@@ -22,16 +22,7 @@ const eventService = mock<EventService>();
 const instanceSettings = mock<InstanceSettings>({ instanceType: 'main' });
 const workflowRepository = mock<WorkflowRepository>();
 const app = mock<express.Application>();
-const eventBus = new MessageEventBus(
-	mock(),
-	mock(),
-	mock(),
-	mock(),
-	mock(),
-	mock(),
-	mock(),
-	mock(),
-);
+const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock(), mock(), mock());
 
 describe('workflow_success_total', () => {
 	test('support workflow id labels', async () => {

packages/cli/src/modules/log-streaming.ee/database/entities/event-destination.entity.ts

@@ -1,8 +1,7 @@
+import { JsonColumn, WithTimestamps } from '@n8n/db';
 import { Entity, PrimaryColumn } from '@n8n/typeorm';
 import { MessageEventBusDestinationOptions } from 'n8n-workflow';
 
-import { JsonColumn, WithTimestamps } from './abstract-entity';
-
 @Entity({ name: 'event_destinations' })
 export class EventDestinations extends WithTimestamps {
 	@PrimaryColumn('uuid')

packages/cli/src/modules/log-streaming.ee/database/entities/index.ts

@@ -0,0 +1 @@
+export { EventDestinations } from './event-destination.entity';

packages/cli/src/modules/log-streaming.ee/database/repositories/event-destination.repository.ts

@@ -1,6 +1,5 @@
 import { Service } from '@n8n/di';
 import { DataSource, Repository } from '@n8n/typeorm';
-
 import { EventDestinations } from '../entities';
 
 @Service()

packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-from-db.ts

@@ -1,13 +1,14 @@
 import { Logger } from '@n8n/backend-common';
-import type { EventDestinations } from '@n8n/db';
 import { Container } from '@n8n/di';
 import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 
+import type { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
+
 import { MessageEventBusDestinationSentry } from './message-event-bus-destination-sentry.ee';
 import { MessageEventBusDestinationSyslog } from './message-event-bus-destination-syslog.ee';
 import { MessageEventBusDestinationWebhook } from './message-event-bus-destination-webhook.ee';
 import type { MessageEventBusDestination } from './message-event-bus-destination.ee';
-import type { MessageEventBus } from '../message-event-bus/message-event-bus';
+import type { EventDestinations } from '../database/entities';
 
 export function messageEventBusDestinationFromDb(
 	eventBusInstance: MessageEventBus,

packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee.ts

@@ -8,10 +8,13 @@ import type {
 } from 'n8n-workflow';
 
 import { N8N_VERSION } from '@/constants';
+import { eventMessageGenericDestinationTestEvent } from '@/eventbus/event-message-classes/event-message-generic';
+import type {
+	MessageEventBus,
+	MessageWithCallback,
+} from '@/eventbus/message-event-bus/message-event-bus';
 
 import { MessageEventBusDestination } from './message-event-bus-destination.ee';
-import { eventMessageGenericDestinationTestEvent } from '../event-message-classes/event-message-generic';
-import type { MessageEventBus, MessageWithCallback } from '../message-event-bus/message-event-bus';
 
 export const isMessageEventBusDestinationSentryOptions = (
 	candidate: unknown,
@@ -89,7 +92,7 @@ export class MessageEventBusDestinationSentry
 				sendResult = true;
 			}
 		} catch (error) {
-			if (error.message) this.logger.debug(error.message as string);
+			this.logger.debug(`Failed to send message to Sentry destination: ${this.label}`, { error });
 			throw error;
 		}
 		return sendResult;

packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee.ts

@@ -9,9 +9,13 @@ import type {
 } from 'n8n-workflow';
 import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 
+import { eventMessageGenericDestinationTestEvent } from '@/eventbus/event-message-classes/event-message-generic';
+import type {
+	MessageEventBus,
+	MessageWithCallback,
+} from '@/eventbus/message-event-bus/message-event-bus';
+
 import { MessageEventBusDestination } from './message-event-bus-destination.ee';
-import { eventMessageGenericDestinationTestEvent } from '../event-message-classes/event-message-generic';
-import type { MessageEventBus, MessageWithCallback } from '../message-event-bus/message-event-bus';
 export const isMessageEventBusDestinationSyslogOptions = (
 	candidate: unknown,
 ): candidate is MessageEventBusDestinationSyslogOptions => {

packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee.ts

@@ -1,5 +1,11 @@
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 
+import {
+	LOGSTREAMING_DEFAULT_MAX_FREE_SOCKETS,
+	LOGSTREAMING_DEFAULT_MAX_SOCKETS,
+	LOGSTREAMING_DEFAULT_MAX_TOTAL_SOCKETS,
+	LOGSTREAMING_DEFAULT_SOCKET_TIMEOUT_MS,
+} from '@n8n/constants';
 import { Container } from '@n8n/di';
 import axios from 'axios';
 import type { AxiosInstance, AxiosRequestConfig, CreateAxiosDefaults, Method } from 'axios';
@@ -16,16 +22,13 @@ import type {
 } from 'n8n-workflow';
 
 import { CredentialsHelper } from '@/credentials-helper';
+import { eventMessageGenericDestinationTestEvent } from '@/eventbus/event-message-classes/event-message-generic';
+import type {
+	MessageEventBus,
+	MessageWithCallback,
+} from '@/eventbus/message-event-bus/message-event-bus';
 
 import { MessageEventBusDestination } from './message-event-bus-destination.ee';
-import { eventMessageGenericDestinationTestEvent } from '../event-message-classes/event-message-generic';
-import type { MessageEventBus, MessageWithCallback } from '../message-event-bus/message-event-bus';
-import {
-	LOGSTREAMING_DEFAULT_MAX_FREE_SOCKETS,
-	LOGSTREAMING_DEFAULT_MAX_SOCKETS,
-	LOGSTREAMING_DEFAULT_MAX_TOTAL_SOCKETS,
-	LOGSTREAMING_DEFAULT_SOCKET_TIMEOUT_MS,
-} from '@n8n/constants';
 
 export const isMessageEventBusDestinationWebhookOptions = (
 	candidate: unknown,

packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts

@@ -6,21 +6,25 @@ import {
 	LOGSTREAMING_CB_DEFAULT_MAX_DURATION_MS,
 	LOGSTREAMING_CB_DEFAULT_MAX_FAILURES,
 } from '@n8n/constants';
-import { EventDestinationsRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
 import type { INodeCredentials, MessageEventBusDestinationOptions } from 'n8n-workflow';
 import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 import { v4 as uuid } from 'uuid';
 
+import type { EventMessageTypes } from '@/eventbus';
+import type { AbstractEventMessage } from '@/eventbus/event-message-classes/abstract-event-message';
+import type { EventMessageConfirmSource } from '@/eventbus/event-message-classes/event-message-confirm';
+import type {
+	MessageEventBus,
+	MessageEventBusDestinationType,
+	MessageWithCallback,
+} from '@/eventbus/message-event-bus/message-event-bus';
 import { License } from '@/license';
 import { CircuitBreaker } from '@/utils/circuit-breaker';
 
-import type { EventMessageTypes } from '../event-message-classes';
-import type { AbstractEventMessage } from '../event-message-classes/abstract-event-message';
-import type { EventMessageConfirmSource } from '../event-message-classes/event-message-confirm';
-import type { MessageEventBus, MessageWithCallback } from '../message-event-bus/message-event-bus';
-
-export abstract class MessageEventBusDestination implements MessageEventBusDestinationOptions {
+export abstract class MessageEventBusDestination
+	implements MessageEventBusDestinationOptions, MessageEventBusDestinationType
+{
 	// Since you can't have static abstract functions - this just serves as a reminder that you need to implement these. Please.
 	// static abstract deserialize(): MessageEventBusDestination | null;
 	readonly id: string;
@@ -131,27 +135,6 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 		return false;
 	}
 
-	async saveToDb() {
-		const data = {
-			id: this.getId(),
-			destination: this.serialize(),
-		};
-		const dbResult = await Container.get(EventDestinationsRepository).upsert(data, {
-			skipUpdateIfNoValuesChanged: true,
-			conflictPaths: ['id'],
-		});
-		return dbResult;
-	}
-
-	async deleteFromDb() {
-		return await MessageEventBusDestination.deleteFromDb(this.getId());
-	}
-
-	static async deleteFromDb(id: string) {
-		const dbResult = await Container.get(EventDestinationsRepository).delete({ id });
-		return dbResult;
-	}
-
 	serialize(): MessageEventBusDestinationOptions {
 		return {
 			__type: this.__type,
@@ -169,7 +152,7 @@ export abstract class MessageEventBusDestination implements MessageEventBusDesti
 		return JSON.stringify(this.serialize());
 	}
 
-	close(): void | Promise<void> {
+	async close(): Promise<void> {
 		this.stopListening();
 	}
 }

packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts

@@ -0,0 +1,83 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+import type { DeleteResult } from '@n8n/typeorm';
+
+import type { MessageEventBusDestinationType } from '@/eventbus/message-event-bus/message-event-bus';
+import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
+
+import { EventDestinationsRepository } from './database/repositories/event-destination.repository';
+import { messageEventBusDestinationFromDb } from './destinations/message-event-bus-destination-from-db';
+
+/**
+ * Service to handle all database operations for log streaming destinations.
+ * This ensures the message event bus doesn't directly access the database.
+ */
+@Service()
+export class LogStreamingDestinationService {
+	constructor(
+		private readonly logger: Logger,
+		private readonly eventDestinationsRepository: EventDestinationsRepository,
+		private readonly eventBus: MessageEventBus,
+	) {}
+
+	/**
+	 * Load all destinations from database and add them to the event bus
+	 */
+	async loadDestinationsFromDb(): Promise<void> {
+		const savedEventDestinations = await this.eventDestinationsRepository.find({});
+		if (savedEventDestinations.length > 0) {
+			for (const destinationData of savedEventDestinations) {
+				try {
+					const destination = messageEventBusDestinationFromDb(this.eventBus, destinationData);
+					if (destination) {
+						await this.eventBus.addDestination(destination, false);
+					}
+				} catch (error) {
+					this.logger.debug('Failed to load destination from database', { error });
+				}
+			}
+		}
+	}
+
+	/**
+	 * Save a destination to the database
+	 */
+	async saveDestinationToDb(destination: MessageEventBusDestinationType) {
+		const data = {
+			id: destination.getId(),
+			destination: destination.serialize(),
+		};
+		const dbResult = await this.eventDestinationsRepository.upsert(data, {
+			skipUpdateIfNoValuesChanged: true,
+			conflictPaths: ['id'],
+		});
+		return dbResult;
+	}
+
+	/**
+	 * Delete a destination from the database
+	 */
+	async deleteDestinationFromDb(id: string): Promise<DeleteResult> {
+		return await this.eventDestinationsRepository.delete({ id });
+	}
+
+	/**
+	 * Add a destination to the event bus and save to database
+	 */
+	async addDestination(
+		destination: MessageEventBusDestinationType,
+		notifyWorkers: boolean = true,
+	): Promise<MessageEventBusDestinationType> {
+		const addedDestination = await this.eventBus.addDestination(destination, notifyWorkers);
+		await this.saveDestinationToDb(addedDestination);
+		return addedDestination;
+	}
+
+	/**
+	 * Remove a destination from the event bus and delete from database
+	 */
+	async removeDestination(id: string, notifyWorkers: boolean = true): Promise<DeleteResult> {
+		await this.eventBus.removeDestination(id, notifyWorkers);
+		return await this.deleteDestinationFromDb(id);
+	}
+}

packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts

@@ -8,19 +8,22 @@ import type {
 import { MessageEventBusDestinationTypeNames } from 'n8n-workflow';
 
 import { BadRequestError } from '@/errors/response-errors/bad-request.error';
+import { eventNamesAll } from '@/eventbus/event-message-classes';
+import {
+	MessageEventBus,
+	MessageEventBusDestinationType,
+} from '@/eventbus/message-event-bus/message-event-bus';
 
-import { eventNamesAll } from './event-message-classes';
-import { MessageEventBus } from './message-event-bus/message-event-bus';
 import {
 	isMessageEventBusDestinationSentryOptions,
 	MessageEventBusDestinationSentry,
-} from './message-event-bus-destination/message-event-bus-destination-sentry.ee';
+} from './destinations/message-event-bus-destination-sentry.ee';
 import {
 	isMessageEventBusDestinationSyslogOptions,
 	MessageEventBusDestinationSyslog,
-} from './message-event-bus-destination/message-event-bus-destination-syslog.ee';
-import { MessageEventBusDestinationWebhook } from './message-event-bus-destination/message-event-bus-destination-webhook.ee';
-import type { MessageEventBusDestination } from './message-event-bus-destination/message-event-bus-destination.ee';
+} from './destinations/message-event-bus-destination-syslog.ee';
+import { MessageEventBusDestinationWebhook } from './destinations/message-event-bus-destination-webhook.ee';
+import { LogStreamingDestinationService } from './log-streaming-destination.service';
 
 const isWithIdString = (candidate: unknown): candidate is { id: string } => {
 	const o = candidate as { id: string };
@@ -46,7 +49,10 @@ const isMessageEventBusDestinationOptions = (
 
 @RestController('/eventbus')
 export class EventBusController {
-	constructor(private readonly eventBus: MessageEventBus) {}
+	constructor(
+		private readonly eventBus: MessageEventBus,
+		private readonly destinationService: LogStreamingDestinationService,
+	) {}
 
 	@Get('/eventnames')
 	async getEventNames(): Promise<string[]> {
@@ -68,26 +74,26 @@ export class EventBusController {
 	@Post('/destination')
 	@GlobalScope('eventBusDestination:create')
 	async postDestination(req: AuthenticatedRequest): Promise<any> {
-		let result: MessageEventBusDestination | undefined;
+		let result: MessageEventBusDestinationType | undefined;
 		if (isMessageEventBusDestinationOptions(req.body)) {
 			switch (req.body.__type) {
 				case MessageEventBusDestinationTypeNames.sentry:
 					if (isMessageEventBusDestinationSentryOptions(req.body)) {
-						result = await this.eventBus.addDestination(
+						result = await this.destinationService.addDestination(
 							new MessageEventBusDestinationSentry(this.eventBus, req.body),
 						);
 					}
 					break;
 				case MessageEventBusDestinationTypeNames.webhook:
 					if (isMessageEventBusDestinationWebhookOptions(req.body)) {
-						result = await this.eventBus.addDestination(
+						result = await this.destinationService.addDestination(
 							new MessageEventBusDestinationWebhook(this.eventBus, req.body),
 						);
 					}
 					break;
 				case MessageEventBusDestinationTypeNames.syslog:
 					if (isMessageEventBusDestinationSyslogOptions(req.body)) {
-						result = await this.eventBus.addDestination(
+						result = await this.destinationService.addDestination(
 							new MessageEventBusDestinationSyslog(this.eventBus, req.body),
 						);
 					}
@@ -98,7 +104,6 @@ export class EventBusController {
 					);
 			}
 			if (result) {
-				await result.saveToDb();
 				return {
 					...result.serialize(),
 					eventBusInstance: undefined,
@@ -124,8 +129,7 @@ export class EventBusController {
 	@GlobalScope('eventBusDestination:delete')
 	async deleteDestination(req: AuthenticatedRequest) {
 		if (isWithIdString(req.query)) {
-			await this.eventBus.removeDestination(req.query.id);
-			return await this.eventBus.deleteDestination(req.query.id);
+			await this.destinationService.removeDestination(req.query.id);
 		} else {
 			throw new BadRequestError('Query is missing id');
 		}

packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts

@@ -0,0 +1,31 @@
+import { LICENSE_FEATURES } from '@n8n/constants';
+import type { ModuleInterface } from '@n8n/decorators';
+import { BackendModule } from '@n8n/decorators';
+import { Container } from '@n8n/di';
+
+/**
+ * Log Streaming module provides enterprise-grade event logging
+ * to external destinations (Webhook, Syslog, Sentry).
+ *
+ * Requires 'feat:logStreaming' license feature.
+ * Runs on main, worker, and webhook process types.
+ */
+@BackendModule({
+	name: 'log-streaming',
+	licenseFlag: LICENSE_FEATURES.LOG_STREAMING,
+	instanceTypes: ['main', 'worker', 'webhook'],
+})
+export class LogStreamingModule implements ModuleInterface {
+	async init() {
+		await import('./log-streaming.controller');
+
+		const { LogStreamingDestinationService } = await import('./log-streaming-destination.service');
+		const destinationService = Container.get(LogStreamingDestinationService);
+		await destinationService.loadDestinationsFromDb();
+	}
+
+	async entities() {
+		const { EventDestinations } = await import('./database/entities/event-destination.entity');
+		return [EventDestinations];
+	}
+}

packages/cli/src/server.ts

@@ -54,7 +54,6 @@ import '@/controllers/user-settings.controller';
 import '@/controllers/workflow-statistics.controller';
 import '@/controllers/api-keys.controller';
 import '@/credentials/credentials.controller';
-import '@/eventbus/event-bus.controller';
 import '@/events/events.controller';
 import '@/executions/executions.controller';
 import '@/license/license.controller';

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

@@ -2,6 +2,7 @@ import { mockInstance } from '@n8n/backend-test-utils';
 import { GLOBAL_OWNER_ROLE, type User } from '@n8n/db';
 import { Container } from '@n8n/di';
 import axios from 'axios';
+import { mock } from 'jest-mock-extended';
 import type {
 	MessageEventBusDestinationSentryOptions,
 	MessageEventBusDestinationSyslogOptions,
@@ -14,16 +15,14 @@ import {
 } from 'n8n-workflow';
 import { v4 as uuid } from 'uuid';
 
-import { mock } from 'jest-mock-extended';
-
 import type { EventNamesTypes } from '@/eventbus/event-message-classes';
 import { EventMessageAudit } from '@/eventbus/event-message-classes/event-message-audit';
 import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic';
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
-import type { MessageEventBusDestinationSentry } from '@/eventbus/message-event-bus-destination/message-event-bus-destination-sentry.ee';
-import type { MessageEventBusDestinationSyslog } from '@/eventbus/message-event-bus-destination/message-event-bus-destination-syslog.ee';
-import type { MessageEventBusDestinationWebhook } from '@/eventbus/message-event-bus-destination/message-event-bus-destination-webhook.ee';
 import { ExecutionRecoveryService } from '@/executions/execution-recovery.service';
+import type { MessageEventBusDestinationSentry } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee';
+import type { MessageEventBusDestinationSyslog } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee';
+import type { MessageEventBusDestinationWebhook } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
 
 import { createUser } from './shared/db/users';
@@ -88,6 +87,7 @@ mockInstance(ExecutionRecoveryService);
 const testServer = utils.setupTestServer({
 	endpointGroups: ['eventBus'],
 	enabledFeatures: ['feat:logStreaming'],
+	modules: ['log-streaming'],
 });
 
 beforeAll(async () => {

packages/cli/test/integration/shared/types.ts

@@ -54,7 +54,8 @@ type ModuleName =
 	| 'community-packages'
 	| 'data-table'
 	| 'mcp'
-	| 'dynamic-credentials';
+	| 'dynamic-credentials'
+	| 'log-streaming';
 
 export interface SetupProps {
 	endpointGroups?: EndpointGroup[];

packages/cli/test/integration/shared/utils/test-server.ts

@@ -193,7 +193,7 @@ export const setupTestServer = ({
 					}
 
 					case 'eventBus':
-						await import('@/eventbus/event-bus.controller');
+						await import('@/modules/log-streaming.ee/log-streaming.controller');
 						break;
 
 					case 'auth':