chore(core): Extract log streaming log streaming standalone handling of destinations (#24197)

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

@@ -18,7 +18,6 @@ import { v4 as uuid } from 'uuid';
 
 import { ActiveWorkflowManager } from '@/active-workflow-manager';
 import { inE2ETests } from '@/constants';
-import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import type { FeatureReturnType } from '@/license';
 import { License } from '@/license';
 import { MfaService } from '@/mfa/mfa.service';
@@ -167,7 +166,6 @@ export class E2EController {
 		private readonly cacheService: CacheService,
 		private readonly push: Push,
 		private readonly passwordUtility: PasswordUtility,
-		private readonly eventBus: MessageEventBus,
 		private readonly userRepository: UserRepository,
 		private readonly frontendService: FrontendService,
 		private readonly executionsConfig: ExecutionsConfig,
@@ -282,9 +280,11 @@ export class E2EController {
 	}
 
 	private async resetLogStreaming() {
-		for (const id in this.eventBus.destinations) {
-			await this.eventBus.removeDestination(id, false);
-			await this.logStreamingDestinationsService.deleteDestinationFromDb(id);
+		const destinations = await this.logStreamingDestinationsService.findDestination();
+		for (const destination of destinations) {
+			if (destination.id) {
+				await this.logStreamingDestinationsService.removeDestination(destination.id, false);
+			}
 		}
 	}
 

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

@@ -1,16 +1,11 @@
 import { Logger } from '@n8n/backend-common';
 import { GlobalConfig } from '@n8n/config';
 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 { In, IsNull, Not } from '@n8n/typeorm';
 import EventEmitter from 'events';
 import uniqby from 'lodash/uniqBy';
-import type { MessageEventBusDestinationOptions } from 'n8n-workflow';
-
-import { License } from '@/license';
-import { Publisher } from '@/scaling/pubsub/publisher.service';
 
 import { ExecutionRecoveryService } from '../../executions/execution-recovery.service';
 import type { EventMessageTypes } from '../event-message-classes/';
@@ -23,10 +18,6 @@ import { EventMessageAudit } from '../event-message-classes/event-message-audit'
 import type { EventMessageConfirmSource } from '../event-message-classes/event-message-confirm';
 import type { EventMessageExecutionOptions } from '../event-message-classes/event-message-execution';
 import { EventMessageExecution } from '../event-message-classes/event-message-execution';
-import {
-	EventMessageGeneric,
-	eventMessageGenericDestinationTestEvent,
-} from '../event-message-classes/event-message-generic';
 import type { EventMessageNodeOptions } from '../event-message-classes/event-message-node';
 import { EventMessageNode } from '../event-message-classes/event-message-node';
 import type { EventMessageQueueOptions } from '../event-message-classes/event-message-queue';
@@ -50,17 +41,6 @@ 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
@@ -69,19 +49,13 @@ export class MessageEventBus extends EventEmitter {
 
 	logWriter: MessageEventBusLogWriter;
 
-	destinations: {
-		[key: string]: MessageEventBusDestinationType;
-	} = {};
-
 	private pushIntervalTimer: NodeJS.Timeout;
 
 	constructor(
 		private readonly logger: Logger,
 		private readonly executionRepository: ExecutionRepository,
 		private readonly workflowRepository: WorkflowRepository,
-		private readonly publisher: Publisher,
 		private readonly recoveryService: ExecutionRecoveryService,
-		private readonly license: License,
 		private readonly globalConfig: GlobalConfig,
 	) {
 		super();
@@ -226,36 +200,6 @@ export class MessageEventBus extends EventEmitter {
 		this.isInitialized = true;
 	}
 
-	async addDestination(destination: MessageEventBusDestinationType, notifyWorkers: boolean = true) {
-		await this.removeDestination(destination.getId(), false);
-		this.destinations[destination.getId()] = destination;
-		this.destinations[destination.getId()].startListening();
-		if (notifyWorkers) {
-			void this.publisher.publishCommand({ command: 'restart-event-bus' });
-		}
-		return destination;
-	}
-
-	async findDestination(id?: string): Promise<MessageEventBusDestinationOptions[]> {
-		let result: MessageEventBusDestinationOptions[];
-		if (id && Object.keys(this.destinations).includes(id)) {
-			result = [this.destinations[id].serialize()];
-		} else {
-			result = Object.keys(this.destinations).map((e) => this.destinations[e].serialize());
-		}
-		return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? ''));
-	}
-
-	async removeDestination(id: string, notifyWorkers: boolean = true) {
-		if (Object.keys(this.destinations).includes(id)) {
-			await this.destinations[id].close();
-			delete this.destinations[id];
-		}
-		if (notifyWorkers) {
-			void this.publisher.publishCommand({ command: 'restart-event-bus' });
-		}
-	}
-
 	private async trySendingUnsent(msgs?: EventMessageTypes[]) {
 		const unsentMessages = msgs ?? (await this.getEventsUnsent());
 		if (unsentMessages.length > 0) {
@@ -270,76 +214,30 @@ export class MessageEventBus extends EventEmitter {
 	async close() {
 		this.logger.debug('Shutting down event writer...');
 		await this.logWriter?.close();
-		for (const destinationName of Object.keys(this.destinations)) {
-			this.logger.debug(
-				`Shutting down event destination ${this.destinations[destinationName].getId()}...`,
-			);
-			await this.destinations[destinationName].close();
-		}
 		this.isInitialized = false;
 		this.logger.debug('EventBus shut down.');
 	}
 
-	@OnPubSubEvent('restart-event-bus')
-	async restart() {
-		await this.close();
-		await this.initialize({ skipRecoveryPass: true });
-	}
-
 	async send(msgs: EventMessageTypes | EventMessageTypes[]) {
 		if (!Array.isArray(msgs)) {
 			msgs = [msgs];
 		}
 		for (const msg of msgs) {
 			this.logWriter?.putMessage(msg);
-			// if there are no set up destinations, immediately mark the event as sent
-			if (!this.shouldSendMsg(msg)) {
-				this.confirmSent(msg, { id: '0', name: 'eventBus' });
-			}
 			await this.emitMessage(msg);
 		}
 	}
 
-	async testDestination(destinationId: string): Promise<boolean> {
-		const msg = new EventMessageGeneric({
-			eventName: eventMessageGenericDestinationTestEvent,
-		});
-		const destination = await this.findDestination(destinationId);
-		if (destination.length > 0) {
-			const sendResult = await this.destinations[destinationId].receiveFromEventBus({
-				msg,
-				confirmCallback: () => this.confirmSent(msg, { id: '0', name: 'eventBus' }),
-			});
-			return sendResult;
-		}
-		return false;
-	}
-
 	confirmSent(msg: EventMessageTypes, source?: EventMessageConfirmSource) {
 		this.logWriter?.confirmMessageSent(msg.id, source);
 	}
 
-	private hasAnyDestinationSubscribedToEvent(msg: EventMessageTypes): boolean {
-		for (const destinationName of Object.keys(this.destinations)) {
-			if (this.destinations[destinationName].hasSubscribedToEvent(msg)) {
-				return true;
-			}
-		}
-		return false;
-	}
-
 	private async emitMessage(msg: EventMessageTypes) {
 		this.emit('metrics.eventBus.event', msg);
 
 		// generic emit for external modules to capture events
 		// this is for internal use ONLY and not for use with custom destinations!
 		this.emitMessageWithCallback('message', msg);
-
-		if (this.shouldSendMsg(msg)) {
-			for (const destinationName of Object.keys(this.destinations)) {
-				this.emitMessageWithCallback(this.destinations[destinationName].getId(), msg);
-			}
-		}
 	}
 
 	private emitMessageWithCallback(eventName: string, msg: EventMessageTypes): boolean {
@@ -348,14 +246,6 @@ export class MessageEventBus extends EventEmitter {
 		return this.emit(eventName, msg, confirmCallback);
 	}
 
-	shouldSendMsg(msg: EventMessageTypes): boolean {
-		return (
-			this.license.isLogStreamingEnabled() &&
-			Object.keys(this.destinations).length > 0 &&
-			this.hasAnyDestinationSubscribedToEvent(msg)
-		);
-	}
-
 	async getEventsAll(): Promise<EventMessageTypes[]> {
 		const queryResult = await this.logWriter?.getMessagesAll();
 		const filtered = uniqby(queryResult, 'id');

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

@@ -22,7 +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());
+const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock());
 
 describe('workflow_success_total', () => {
 	test('support workflow id labels', async () => {

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

@@ -11,20 +11,15 @@ import type { INodeCredentials, MessageEventBusDestinationOptions } from 'n8n-wo
 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';
 
-export abstract class MessageEventBusDestination
-	implements MessageEventBusDestinationOptions, MessageEventBusDestinationType
-{
+export abstract class MessageEventBusDestination implements MessageEventBusDestinationOptions {
 	// 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;
@@ -83,44 +78,6 @@ export abstract class MessageEventBusDestination
 		this.logger.debug(`${this.__type}(${this.id}) event destination constructed`);
 	}
 
-	startListening() {
-		if (this.enabled) {
-			this.eventBusInstance.on(
-				this.getId(),
-				async (
-					msg: EventMessageTypes,
-					confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void,
-				) => {
-					try {
-						await this.circuitBreakerInstance.execute(async () => {
-							await this.receiveFromEventBus({ msg, confirmCallback });
-						});
-					} catch (error) {
-						this.logger.error(
-							`${this.__type}(${this.id}) event destination ${this.label} failed to send message`,
-							{ error },
-						);
-					}
-				},
-			);
-			this.logger.debug(`${this.id} listener started`);
-		}
-	}
-
-	stopListening() {
-		this.eventBusInstance.removeAllListeners(this.getId());
-	}
-
-	enable() {
-		this.enabled = true;
-		this.startListening();
-	}
-
-	disable() {
-		this.enabled = false;
-		this.stopListening();
-	}
-
 	getId() {
 		return this.id;
 	}
@@ -148,11 +105,20 @@ export abstract class MessageEventBusDestination
 
 	abstract receiveFromEventBus(emitterPayload: MessageWithCallback): Promise<boolean>;
 
+	/**
+	 * Send a message to this destination with circuit breaker protection
+	 */
+	async sendMessage(emitterPayload: MessageWithCallback): Promise<boolean> {
+		return await this.circuitBreakerInstance.execute(async () => {
+			return await this.receiveFromEventBus(emitterPayload);
+		});
+	}
+
 	toString() {
 		return JSON.stringify(this.serialize());
 	}
 
 	async close(): Promise<void> {
-		this.stopListening();
+		this.enabled = false;
 	}
 }

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

@@ -1,27 +1,51 @@
-import { Logger } from '@n8n/backend-common';
+import { LicenseState, Logger } from '@n8n/backend-common';
+import { OnPubSubEvent } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import type { DeleteResult } from '@n8n/typeorm';
+import type { MessageEventBusDestinationOptions } from 'n8n-workflow';
 
-import type { MessageEventBusDestinationType } from '@/eventbus/message-event-bus/message-event-bus';
+import type { EventMessageTypes } from '@/eventbus';
+import type { EventMessageConfirmSource } from '@/eventbus/event-message-classes/event-message-confirm';
+import {
+	EventMessageGeneric,
+	eventMessageGenericDestinationTestEvent,
+} from '@/eventbus/event-message-classes/event-message-generic';
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
+import { Publisher } from '@/scaling/pubsub/publisher.service';
 
 import { EventDestinationsRepository } from './database/repositories/event-destination.repository';
 import { messageEventBusDestinationFromDb } from './destinations/message-event-bus-destination-from-db';
+import { MessageEventBusDestination } from './destinations/message-event-bus-destination.ee';
 
 /**
- * Service to handle all database operations for log streaming destinations.
- * This ensures the message event bus doesn't directly access the database.
+ * Service to handle all log streaming destination operations including:
+ * - Database persistence
+ * - Destination lifecycle management
+ * - Event forwarding from event bus to destinations
  */
 @Service()
 export class LogStreamingDestinationService {
+	private destinations: { [key: string]: MessageEventBusDestination } = {};
+
+	private isListening = false;
+
+	private readonly messageHandler: (
+		msg: EventMessageTypes,
+		confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void,
+	) => Promise<void>;
+
 	constructor(
 		private readonly logger: Logger,
 		private readonly eventDestinationsRepository: EventDestinationsRepository,
 		private readonly eventBus: MessageEventBus,
-	) {}
+		private readonly licenseState: LicenseState,
+		private readonly publisher: Publisher,
+	) {
+		this.messageHandler = this.handleMessage.bind(this);
+	}
 
 	/**
-	 * Load all destinations from database and add them to the event bus
+	 * Load all destinations from database and add them to the local destinations map
 	 */
 	async loadDestinationsFromDb(): Promise<void> {
 		const savedEventDestinations = await this.eventDestinationsRepository.find({});
@@ -30,7 +54,8 @@ export class LogStreamingDestinationService {
 				try {
 					const destination = messageEventBusDestinationFromDb(this.eventBus, destinationData);
 					if (destination) {
-						await this.eventBus.addDestination(destination, false);
+						this.destinations[destination.getId()] = destination;
+						this.logger.debug(`Loaded destination ${destination.getId()} from database`);
 					}
 				} catch (error) {
 					this.logger.debug('Failed to load destination from database', { error });
@@ -42,7 +67,7 @@ export class LogStreamingDestinationService {
 	/**
 	 * Save a destination to the database
 	 */
-	async saveDestinationToDb(destination: MessageEventBusDestinationType) {
+	async saveDestinationToDb(destination: MessageEventBusDestination) {
 		const data = {
 			id: destination.getId(),
 			destination: destination.serialize(),
@@ -62,22 +87,179 @@ export class LogStreamingDestinationService {
 	}
 
 	/**
-	 * Add a destination to the event bus and save to database
+	 * Add a destination to the local map and save to database
 	 */
 	async addDestination(
-		destination: MessageEventBusDestinationType,
+		destination: MessageEventBusDestination,
 		notifyWorkers: boolean = true,
-	): Promise<MessageEventBusDestinationType> {
-		const addedDestination = await this.eventBus.addDestination(destination, notifyWorkers);
-		await this.saveDestinationToDb(addedDestination);
-		return addedDestination;
+	): Promise<MessageEventBusDestination> {
+		// Remove any existing destination with the same ID
+		await this.destinations[destination.getId()]?.close();
+
+		// Add to local destinations map
+		this.destinations[destination.getId()] = destination;
+		this.logger.debug(`Added destination ${destination.getId()}`);
+
+		// Save to database
+		await this.saveDestinationToDb(destination);
+
+		// Notify workers to reload destinations
+		if (notifyWorkers) {
+			void this.publisher.publishCommand({ command: 'restart-event-bus' });
+		}
+
+		return destination;
 	}
 
 	/**
-	 * Remove a destination from the event bus and delete from database
+	 * Remove a destination from the local map and delete from database
 	 */
 	async removeDestination(id: string, notifyWorkers: boolean = true): Promise<DeleteResult> {
-		await this.eventBus.removeDestination(id, notifyWorkers);
+		// Close and remove from local map
+		if (this.destinations[id]) {
+			await this.destinations[id].close();
+			delete this.destinations[id];
+			this.logger.debug(`Removed destination ${id}`);
+		}
+
+		// Notify workers to reload destinations
+		if (notifyWorkers) {
+			void this.publisher.publishCommand({ command: 'restart-event-bus' });
+		}
+
+		// Delete from database
 		return await this.deleteDestinationFromDb(id);
 	}
+
+	/**
+	 * Initialize the service by registering listener on event bus
+	 */
+	async initialize(): Promise<void> {
+		if (this.isListening) {
+			return;
+		}
+
+		this.logger.debug('Initializing log streaming destination service...');
+
+		// Register listener for all messages from event bus
+		this.eventBus.on('message', this.messageHandler);
+		this.isListening = true;
+
+		this.logger.debug(
+			`Log streaming destination service initialized with ${Object.keys(this.destinations).length} destinations`,
+		);
+	}
+
+	/**
+	 * Handle messages from event bus and forward to appropriate destinations
+	 */
+	private async handleMessage(
+		msg: EventMessageTypes,
+		confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void,
+	) {
+		// If there are no destinations that should receive this message, mark it as sent immediately
+		if (!this.shouldSendMsg(msg)) {
+			this.eventBus.confirmSent(msg, { id: '0', name: 'eventBus' });
+			return;
+		}
+
+		for (const destination of Object.values(this.destinations)) {
+			if (destination.hasSubscribedToEvent(msg)) {
+				try {
+					// Call destination with circuit breaker protection
+					await destination.sendMessage({ msg, confirmCallback });
+				} catch (error) {
+					this.logger.error(`Failed to send message to destination ${destination.getId()}`, {
+						error,
+					});
+				}
+			}
+		}
+	}
+
+	/**
+	 * Find destination(s) by ID
+	 */
+	async findDestination(id?: string): Promise<MessageEventBusDestinationOptions[]> {
+		const result: MessageEventBusDestinationOptions[] = id
+			? this.destinations[id]
+				? [this.destinations[id].serialize()]
+				: []
+			: Object.values(this.destinations).map((dest) => dest.serialize());
+		return result.sort((a, b) => (a.__type ?? '').localeCompare(b.__type ?? ''));
+	}
+
+	/**
+	 * Test a destination by sending a test message
+	 * Note: This bypasses the circuit breaker to allow testing even when circuit is open
+	 */
+	async testDestination(destinationId: string): Promise<boolean> {
+		const msg = new EventMessageGeneric({
+			eventName: eventMessageGenericDestinationTestEvent,
+		});
+		const destination = await this.findDestination(destinationId);
+		if (destination.length > 0) {
+			const sendResult = await this.destinations[destinationId].receiveFromEventBus({
+				msg,
+				confirmCallback: () => this.eventBus.confirmSent(msg, { id: '0', name: 'eventBus' }),
+			});
+			return sendResult;
+		}
+		return false;
+	}
+
+	/**
+	 * Check if any destination is subscribed to the given event
+	 */
+	private hasAnyDestinationSubscribedToEvent(msg: EventMessageTypes): boolean {
+		return Object.values(this.destinations).some((destination) =>
+			destination.hasSubscribedToEvent(msg),
+		);
+	}
+
+	/**
+	 * Determine if a message should be sent to destinations
+	 */
+	shouldSendMsg(msg: EventMessageTypes): boolean {
+		return (
+			this.licenseState.isLogStreamingLicensed() &&
+			Object.keys(this.destinations).length > 0 &&
+			this.hasAnyDestinationSubscribedToEvent(msg)
+		);
+	}
+
+	/**
+	 * Close all destinations
+	 */
+	async close(): Promise<void> {
+		this.logger.debug('Closing log streaming destination service...');
+
+		// Stop listening to event bus
+		if (this.isListening) {
+			this.eventBus.removeListener('message', this.messageHandler);
+			this.isListening = false;
+		}
+
+		// Close all destinations
+		for (const destination of Object.values(this.destinations)) {
+			this.logger.debug(`Closing destination ${destination.getId()}...`);
+			await destination.close();
+		}
+
+		// Clear destinations map
+		this.destinations = {};
+
+		this.logger.debug('Log streaming destination service closed');
+	}
+
+	/**
+	 * Handle restart event from pubsub
+	 */
+	@OnPubSubEvent('restart-event-bus')
+	async restart() {
+		this.logger.debug('Restarting log streaming destination service...');
+		await this.close();
+		await this.loadDestinationsFromDb();
+		await this.initialize();
+	}
 }

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

@@ -9,10 +9,7 @@ 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 { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 
 import {
 	isMessageEventBusDestinationSentryOptions,
@@ -23,6 +20,7 @@ import {
 	MessageEventBusDestinationSyslog,
 } from './destinations/message-event-bus-destination-syslog.ee';
 import { MessageEventBusDestinationWebhook } from './destinations/message-event-bus-destination-webhook.ee';
+import { MessageEventBusDestination } from './destinations/message-event-bus-destination.ee';
 import { LogStreamingDestinationService } from './log-streaming-destination.service';
 
 const isWithIdString = (candidate: unknown): candidate is { id: string } => {
@@ -64,9 +62,9 @@ export class EventBusController {
 	@GlobalScope('eventBusDestination:list')
 	async getDestination(req: express.Request): Promise<MessageEventBusDestinationOptions[]> {
 		if (isWithIdString(req.query)) {
-			return await this.eventBus.findDestination(req.query.id);
+			return await this.destinationService.findDestination(req.query.id);
 		} else {
-			return await this.eventBus.findDestination();
+			return await this.destinationService.findDestination();
 		}
 	}
 
@@ -74,7 +72,7 @@ export class EventBusController {
 	@Post('/destination')
 	@GlobalScope('eventBusDestination:create')
 	async postDestination(req: AuthenticatedRequest): Promise<any> {
-		let result: MessageEventBusDestinationType | undefined;
+		let result: MessageEventBusDestination | undefined;
 		if (isMessageEventBusDestinationOptions(req.body)) {
 			switch (req.body.__type) {
 				case MessageEventBusDestinationTypeNames.sentry:
@@ -119,7 +117,7 @@ export class EventBusController {
 	@GlobalScope('eventBusDestination:test')
 	async sendTestMessage(req: express.Request): Promise<boolean> {
 		if (isWithIdString(req.query)) {
-			return await this.eventBus.testDestination(req.query.id);
+			return await this.destinationService.testDestination(req.query.id);
 		}
 		return false;
 	}

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

@@ -22,6 +22,7 @@ export class LogStreamingModule implements ModuleInterface {
 		const { LogStreamingDestinationService } = await import('./log-streaming-destination.service');
 		const destinationService = Container.get(LogStreamingDestinationService);
 		await destinationService.loadDestinationsFromDb();
+		await destinationService.initialize();
 	}
 
 	async entities() {

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

@@ -23,6 +23,7 @@ import { ExecutionRecoveryService } from '@/executions/execution-recovery.servic
 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 { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
 
 import { createUser } from './shared/db/users';
@@ -70,6 +71,7 @@ const testSentryDestination: MessageEventBusDestinationSentryOptions = {
 };
 
 let eventBus: MessageEventBus;
+let destinationService: LogStreamingDestinationService;
 
 async function confirmIdInAll(id: string) {
 	const sent = await eventBus.getEventsAll();
@@ -96,6 +98,9 @@ beforeAll(async () => {
 
 	eventBus = Container.get(MessageEventBus);
 	await eventBus.initialize();
+
+	destinationService = Container.get(LogStreamingDestinationService);
+	await destinationService.initialize();
 });
 
 afterAll(async () => {
@@ -140,7 +145,7 @@ describe('GET /eventbus/destination', () => {
 
 		for (let index = 0; index < data.length; index++) {
 			const destination = data[index];
-			const foundDestinations = await eventBus.findDestination(destination.id);
+			const foundDestinations = await destinationService.findDestination(destination.id);
 			expect(Array.isArray(foundDestinations)).toBeTruthy();
 			expect(foundDestinations.length).toBe(1);
 			expect(foundDestinations[0].label).toBe(destination.label);
@@ -177,11 +182,11 @@ test('should anonymize audit message to syslog ', async () => {
 		id: uuid(),
 	});
 
-	const syslogDestination = eventBus.destinations[
+	const syslogDestination = destinationService['destinations'][
 		testSyslogDestination.id!
 	] as MessageEventBusDestinationSyslog;
 
-	syslogDestination.enable();
+	syslogDestination.enabled = true;
 
 	const mockedSyslogClientLog = jest.spyOn(syslogDestination.client, 'log');
 	mockedSyslogClientLog.mockImplementation(async (m, _options, _cb) => {
@@ -222,7 +227,7 @@ test('should anonymize audit message to syslog ', async () => {
 					await eventBus.getEventsAll();
 					await confirmIdInAll(testAuditMessage.id);
 					expect(mockedSyslogClientLog).toHaveBeenCalled();
-					syslogDestination.disable();
+					syslogDestination.enabled = false;
 					eventBus.logWriter.worker?.removeListener('message', handler006);
 					resolve(true);
 				}
@@ -237,11 +242,11 @@ test('should send message to webhook ', async () => {
 		id: uuid(),
 	});
 
-	const webhookDestination = eventBus.destinations[
+	const webhookDestination = destinationService['destinations'][
 		testWebhookDestination.id!
 	] as MessageEventBusDestinationWebhook;
 
-	webhookDestination.enable();
+	webhookDestination.enabled = true;
 
 	mockAxiosInstance.post.mockResolvedValue({ status: 200, data: { msg: 'OK' } });
 	mockAxiosInstance.request.mockResolvedValue({ status: 200, data: { msg: 'OK' } });
@@ -256,7 +261,7 @@ test('should send message to webhook ', async () => {
 				} else if (msg.command === 'confirmMessageSent') {
 					await confirmIdSent(testMessage.id);
 					expect(mockAxiosInstance.request).toHaveBeenCalled();
-					webhookDestination.disable();
+					webhookDestination.enabled = false;
 					eventBus.logWriter.worker?.removeListener('message', handler003);
 					resolve(true);
 				}
@@ -271,11 +276,11 @@ test('should send message to sentry ', async () => {
 		id: uuid(),
 	});
 
-	const sentryDestination = eventBus.destinations[
+	const sentryDestination = destinationService['destinations'][
 		testSentryDestination.id!
 	] as MessageEventBusDestinationSentry;
 
-	sentryDestination.enable();
+	sentryDestination.enabled = true;
 
 	const mockedSentryCaptureMessage = jest.spyOn(sentryDestination.sentryClient!, 'captureMessage');
 	mockedSentryCaptureMessage.mockImplementation((_m, _level, _hint, _scope) => {
@@ -296,7 +301,7 @@ test('should send message to sentry ', async () => {
 				} else if (msg.command === 'confirmMessageSent') {
 					await confirmIdSent(testMessage.id);
 					expect(mockedSentryCaptureMessage).toHaveBeenCalled();
-					sentryDestination.disable();
+					sentryDestination.enabled = false;
 					eventBus.logWriter.worker?.removeListener('message', handler004);
 					resolve(true);
 				}
@@ -306,7 +311,13 @@ test('should send message to sentry ', async () => {
 });
 
 test('DELETE /eventbus/destination delete all destinations by id', async () => {
-	const existingDestinationIds = [...Object.keys(eventBus.destinations)];
+	const existingDestinations = await destinationService.findDestination();
+	const existingDestinationIds = existingDestinations.reduce<string[]>((acc, d) => {
+		if (d.id) {
+			acc.push(d.id);
+		}
+		return acc;
+	}, []);
 
 	await Promise.all(
 		existingDestinationIds.map(async (id) => {
@@ -315,5 +326,6 @@ test('DELETE /eventbus/destination delete all destinations by id', async () => {
 		}),
 	);
 
-	expect(Object.keys(eventBus.destinations).length).toBe(0);
+	const remainingDestinations = await destinationService.findDestination();
+	expect(remainingDestinations.length).toBe(0);
 });

packages/cli/test/integration/eventbus/syslog-tls.test.ts

@@ -7,6 +7,7 @@ import type TestAgent from 'supertest/lib/agent';
 
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic';
+import { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service';
 
 import { TlsSyslogServer } from './tls-server';
 import { createUser } from '../shared/db/users';
@@ -17,12 +18,14 @@ jest.unmock('@/eventbus/message-event-bus/message-event-bus');
 const tlsServer = new TlsSyslogServer();
 let serverPort: number;
 let eventBus: MessageEventBus;
+let destinationService: LogStreamingDestinationService;
 let authOwnerAgent: TestAgent;
 let logger: Logger;
 
 const testServer = utils.setupTestServer({
 	endpointGroups: ['eventBus'],
 	enabledFeatures: ['feat:logStreaming'],
+	modules: ['log-streaming'],
 });
 
 afterAll(async () => {
@@ -41,6 +44,9 @@ beforeAll(async () => {
 	eventBus = Container.get(MessageEventBus);
 	logger = Container.get(Logger);
 	await eventBus.initialize();
+
+	destinationService = Container.get(LogStreamingDestinationService);
+	await destinationService.initialize();
 });
 
 describe('TLS Syslog E2E', () => {
@@ -51,8 +57,13 @@ describe('TLS Syslog E2E', () => {
 		tlsServer.clearMessages();
 	});
 
-	afterEach(() => {
-		eventBus.destinations = {};
+	afterEach(async () => {
+		const destinations = await destinationService.findDestination();
+		for (const destination of destinations) {
+			if (destination.id) {
+				await destinationService.removeDestination(destination.id, false);
+			}
+		}
 	});
 
 	test('should send message over real TLS connection', async () => {