diff --git a/packages/@n8n/backend-common/src/modules/__tests__/module-registry.test.ts b/packages/@n8n/backend-common/src/modules/__tests__/module-registry.test.ts index 31ce9955a14..ed7944eefc0 100644 --- a/packages/@n8n/backend-common/src/modules/__tests__/module-registry.test.ts +++ b/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', ]); }); diff --git a/packages/@n8n/backend-common/src/modules/module-registry.ts b/packages/@n8n/backend-common/src/modules/module-registry.ts index da6b996b991..8d59a7f2a00 100644 --- a/packages/@n8n/backend-common/src/modules/module-registry.ts +++ b/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[] = []; diff --git a/packages/@n8n/backend-common/src/modules/modules.config.ts b/packages/@n8n/backend-common/src/modules/modules.config.ts index a9876857e38..2570708ceb2 100644 --- a/packages/@n8n/backend-common/src/modules/modules.config.ts +++ b/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]; diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index 9015b149fa4..9faa85b11f8 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/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, diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index 437a972f81d..7a5ea1ec3d2 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/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'; diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index 54f88a0bd80..c0a30886d0c 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/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); } } diff --git a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts b/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts index a4ba70da628..3ec6572a00c 100644 --- a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts +++ b/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; + startListening(): void; + close(): Promise; + 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 { - return await this.eventDestinationsRepository.delete({ - id, - }); - } - private async trySendingUnsent(msgs?: EventMessageTypes[]) { const unsentMessages = msgs ?? (await this.getEventsUnsent()); if (unsentMessages.length > 0) { diff --git a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts index 48965fcceba..04eb6a20393 100644 --- a/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts +++ b/packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts @@ -22,16 +22,7 @@ const eventService = mock(); const instanceSettings = mock({ instanceType: 'main' }); const workflowRepository = mock(); const app = mock(); -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 () => { diff --git a/packages/@n8n/db/src/entities/event-destinations.ts b/packages/cli/src/modules/log-streaming.ee/database/entities/event-destination.entity.ts similarity index 83% rename from packages/@n8n/db/src/entities/event-destinations.ts rename to packages/cli/src/modules/log-streaming.ee/database/entities/event-destination.entity.ts index 42e1162e277..e451b1dbbde 100644 --- a/packages/@n8n/db/src/entities/event-destinations.ts +++ b/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') diff --git a/packages/cli/src/modules/log-streaming.ee/database/entities/index.ts b/packages/cli/src/modules/log-streaming.ee/database/entities/index.ts new file mode 100644 index 00000000000..26a826e8984 --- /dev/null +++ b/packages/cli/src/modules/log-streaming.ee/database/entities/index.ts @@ -0,0 +1 @@ +export { EventDestinations } from './event-destination.entity'; diff --git a/packages/@n8n/db/src/repositories/event-destinations.repository.ts b/packages/cli/src/modules/log-streaming.ee/database/repositories/event-destination.repository.ts similarity index 99% rename from packages/@n8n/db/src/repositories/event-destinations.repository.ts rename to packages/cli/src/modules/log-streaming.ee/database/repositories/event-destination.repository.ts index 00671a30f5a..72f40acaffb 100644 --- a/packages/@n8n/db/src/repositories/event-destinations.repository.ts +++ b/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() diff --git a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-from-db.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-from-db.ts similarity index 89% rename from packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-from-db.ts rename to packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-from-db.ts index 030bb9c948f..230cca1c777 100644 --- a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-from-db.ts +++ b/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, diff --git a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-sentry.ee.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee.ts similarity index 92% rename from packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-sentry.ee.ts rename to packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee.ts index f0f32cd22b4..982b9976842 100644 --- a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-sentry.ee.ts +++ b/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; diff --git a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-syslog.ee.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee.ts similarity index 95% rename from packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-syslog.ee.ts rename to packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee.ts index b53204ba8af..ba535083489 100644 --- a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-syslog.ee.ts +++ b/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 => { diff --git a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-webhook.ee.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee.ts similarity index 98% rename from packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-webhook.ee.ts rename to packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee.ts index da11f85c77b..8c99e1f6348 100644 --- a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination-webhook.ee.ts +++ b/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, diff --git a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination.ee.ts b/packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts similarity index 80% rename from packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination.ee.ts rename to packages/cli/src/modules/log-streaming.ee/destinations/message-event-bus-destination.ee.ts index 1c9e03e368b..cbd237b62cc 100644 --- a/packages/cli/src/eventbus/message-event-bus-destination/message-event-bus-destination.ee.ts +++ b/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 { + async close(): Promise { this.stopListening(); } } diff --git a/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming-destination.service.ts new file mode 100644 index 00000000000..e21174dcae4 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + await this.eventBus.removeDestination(id, notifyWorkers); + return await this.deleteDestinationFromDb(id); + } +} diff --git a/packages/cli/src/eventbus/event-bus.controller.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts similarity index 79% rename from packages/cli/src/eventbus/event-bus.controller.ts rename to packages/cli/src/modules/log-streaming.ee/log-streaming.controller.ts index 271680bdbfe..e50b409bddf 100644 --- a/packages/cli/src/eventbus/event-bus.controller.ts +++ b/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 { @@ -68,26 +74,26 @@ export class EventBusController { @Post('/destination') @GlobalScope('eventBusDestination:create') async postDestination(req: AuthenticatedRequest): Promise { - 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'); } diff --git a/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts b/packages/cli/src/modules/log-streaming.ee/log-streaming.module.ts new file mode 100644 index 00000000000..ddd13b38cb2 --- /dev/null +++ b/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]; + } +} diff --git a/packages/cli/src/server.ts b/packages/cli/src/server.ts index 7c5e9697d10..3f265782893 100644 --- a/packages/cli/src/server.ts +++ b/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'; diff --git a/packages/cli/test/integration/eventbus.ee.test.ts b/packages/cli/test/integration/eventbus.ee.test.ts index 68a797b649d..fb84340c636 100644 --- a/packages/cli/test/integration/eventbus.ee.test.ts +++ b/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 () => { diff --git a/packages/cli/test/integration/shared/types.ts b/packages/cli/test/integration/shared/types.ts index d5afbe47696..677f60243e0 100644 --- a/packages/cli/test/integration/shared/types.ts +++ b/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[]; diff --git a/packages/cli/test/integration/shared/utils/test-server.ts b/packages/cli/test/integration/shared/utils/test-server.ts index 0ab9d26d2be..9ab5a7cdbf8 100644 --- a/packages/cli/test/integration/shared/utils/test-server.ts +++ b/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':