diff --git a/packages/cli/src/InternalHooks.ts b/packages/cli/src/InternalHooks.ts index ea36dc04b8..57e7b890fa 100644 --- a/packages/cli/src/InternalHooks.ts +++ b/packages/cli/src/InternalHooks.ts @@ -25,11 +25,11 @@ import type { import { Telemetry } from '@/telemetry'; import type { AuthProviderType } from '@db/entities/AuthIdentity'; import { RoleService } from './role/role.service'; -import { eventBus } from './eventbus'; import type { User } from '@db/entities/User'; import { N8N_VERSION } from '@/constants'; import * as Db from '@/Db'; import { NodeTypes } from './NodeTypes'; +import { MessageEventBus } from '@/eventbus'; function userToPayload(user: User): { userId: string; @@ -51,7 +51,11 @@ function userToPayload(user: User): { export class InternalHooks implements IInternalHooksClass { private instanceId: string; - constructor(private telemetry: Telemetry, private nodeTypes: NodeTypes) {} + constructor( + private telemetry: Telemetry, + private nodeTypes: NodeTypes, + private eventBus: MessageEventBus, + ) {} async init(instanceId: string) { this.instanceId = instanceId; @@ -111,7 +115,7 @@ export class InternalHooks implements IInternalHooksClass { async onWorkflowCreated(user: User, workflow: IWorkflowBase, publicApi: boolean): Promise { const { nodeGraph } = TelemetryHelpers.generateNodesGraph(workflow, this.nodeTypes); void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.workflow.created', payload: { ...userToPayload(user), @@ -130,7 +134,7 @@ export class InternalHooks implements IInternalHooksClass { async onWorkflowDeleted(user: User, workflowId: string, publicApi: boolean): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.workflow.deleted', payload: { ...userToPayload(user), @@ -162,7 +166,7 @@ export class InternalHooks implements IInternalHooksClass { } void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.workflow.updated', payload: { ...userToPayload(user), @@ -194,7 +198,7 @@ export class InternalHooks implements IInternalHooksClass { nodeName: string, ): Promise { const nodeInWorkflow = workflow.nodes.find((node) => node.name === nodeName); - void eventBus.sendNodeEvent({ + void this.eventBus.sendNodeEvent({ eventName: 'n8n.node.started', payload: { executionId, @@ -212,7 +216,7 @@ export class InternalHooks implements IInternalHooksClass { nodeName: string, ): Promise { const nodeInWorkflow = workflow.nodes.find((node) => node.name === nodeName); - void eventBus.sendNodeEvent({ + void this.eventBus.sendNodeEvent({ eventName: 'n8n.node.finished', payload: { executionId, @@ -230,7 +234,7 @@ export class InternalHooks implements IInternalHooksClass { ): Promise { void Promise.all([ Db.collections.Execution.update(executionId, { status: 'running' }), - eventBus.sendWorkflowEvent({ + this.eventBus.sendWorkflowEvent({ eventName: 'n8n.workflow.started', payload: { executionId, @@ -249,7 +253,7 @@ export class InternalHooks implements IInternalHooksClass { workflowData?: IWorkflowBase, ): Promise { void Promise.all([ - eventBus.sendWorkflowEvent({ + this.eventBus.sendWorkflowEvent({ eventName: 'n8n.workflow.crashed', payload: { executionId, @@ -410,7 +414,7 @@ export class InternalHooks implements IInternalHooksClass { promises.push( properties.success - ? eventBus.sendWorkflowEvent({ + ? this.eventBus.sendWorkflowEvent({ eventName: 'n8n.workflow.success', payload: { executionId, @@ -421,7 +425,7 @@ export class InternalHooks implements IInternalHooksClass { workflowName: workflow.name, }, }) - : eventBus.sendWorkflowEvent({ + : this.eventBus.sendWorkflowEvent({ eventName: 'n8n.workflow.failed', payload: { executionId, @@ -469,7 +473,7 @@ export class InternalHooks implements IInternalHooksClass { publicApi: boolean; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.deleted', payload: { ...userToPayload(userDeletionData.user), @@ -490,7 +494,7 @@ export class InternalHooks implements IInternalHooksClass { email_sent: boolean; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.invited', payload: { ...userToPayload(userInviteData.user), @@ -512,7 +516,7 @@ export class InternalHooks implements IInternalHooksClass { public_api: boolean; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.reinvited', payload: { ...userToPayload(userReinviteData.user), @@ -571,7 +575,7 @@ export class InternalHooks implements IInternalHooksClass { async onUserUpdate(userUpdateData: { user: User; fields_changed: string[] }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.updated', payload: { ...userToPayload(userUpdateData.user), @@ -590,7 +594,7 @@ export class InternalHooks implements IInternalHooksClass { invitee: User; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.invitation.accepted', payload: { invitee: { @@ -609,7 +613,7 @@ export class InternalHooks implements IInternalHooksClass { async onUserPasswordResetEmailClick(userPasswordResetData: { user: User }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.reset', payload: { ...userToPayload(userPasswordResetData.user), @@ -643,7 +647,7 @@ export class InternalHooks implements IInternalHooksClass { async onApiKeyDeleted(apiKeyDeletedData: { user: User; public_api: boolean }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.api.deleted', payload: { ...userToPayload(apiKeyDeletedData.user), @@ -658,7 +662,7 @@ export class InternalHooks implements IInternalHooksClass { async onApiKeyCreated(apiKeyCreatedData: { user: User; public_api: boolean }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.api.created', payload: { ...userToPayload(apiKeyCreatedData.user), @@ -673,7 +677,7 @@ export class InternalHooks implements IInternalHooksClass { async onUserPasswordResetRequestClick(userPasswordResetData: { user: User }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.reset.requested', payload: { ...userToPayload(userPasswordResetData.user), @@ -697,7 +701,7 @@ export class InternalHooks implements IInternalHooksClass { }, ): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.signedup', payload: { ...userToPayload(user), @@ -716,7 +720,7 @@ export class InternalHooks implements IInternalHooksClass { public_api: boolean; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.email.failed', payload: { messageType: failedEmailData.message_type, @@ -741,7 +745,7 @@ export class InternalHooks implements IInternalHooksClass { public_api: boolean; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.credentials.created', payload: { ...userToPayload(userCreatedCredentialsData.user), @@ -769,7 +773,7 @@ export class InternalHooks implements IInternalHooksClass { sharees_removed: number | null; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.user.credentials.shared', payload: { ...userToPayload(userSharedCredentialsData.user), @@ -809,7 +813,7 @@ export class InternalHooks implements IInternalHooksClass { failure_reason?: string; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.package.installed', payload: { ...userToPayload(installationData.user), @@ -847,7 +851,7 @@ export class InternalHooks implements IInternalHooksClass { package_author_email?: string; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.package.updated', payload: { ...userToPayload(updateData.user), @@ -880,7 +884,7 @@ export class InternalHooks implements IInternalHooksClass { package_author_email?: string; }): Promise { void Promise.all([ - eventBus.sendAuditEvent({ + this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.package.deleted', payload: { ...userToPayload(deleteData.user), diff --git a/packages/cli/src/Server.ts b/packages/cli/src/Server.ts index 5bffa178c7..256b129f4b 100644 --- a/packages/cli/src/Server.ts +++ b/packages/cli/src/Server.ts @@ -146,7 +146,6 @@ import { configureMetrics } from './metrics'; import { setupBasicAuth } from './middlewares/basicAuth'; import { setupExternalJWTAuth } from './middlewares/externalJWTAuth'; import { PostHogClient } from './posthog'; -import { eventBus } from './eventbus'; import { Container } from 'typedi'; import { InternalHooks } from './InternalHooks'; import { @@ -157,6 +156,7 @@ import { getSamlLoginLabel, isSamlLoginEnabled, isSamlLicensed } from './sso/sam import { SamlController } from './sso/saml/routes/saml.controller.ee'; import { SamlService } from './sso/saml/saml.service.ee'; import { LdapManager } from './Ldap/LdapManager.ee'; +import { MessageEventBus } from '@/eventbus'; const exec = promisify(callbackExec); @@ -365,7 +365,7 @@ class Server extends AbstractServer { return this.frontendSettings; } - private registerControllers(ignoredEndpoints: Readonly) { + private async registerControllers(ignoredEndpoints: Readonly) { const { app, externalHooks, activeWorkflowRunner, nodeTypes } = this; const repositories = Db.collections; setupAuthMiddlewares(app, ignoredEndpoints, this.restEndpoint, repositories.User); @@ -376,8 +376,11 @@ class Server extends AbstractServer { const postHog = this.postHog; const samlService = Container.get(SamlService); + const eventBus = Container.get(MessageEventBus); + await eventBus.initialize(); + const controllers: object[] = [ - new EventBusController(), + new EventBusController(eventBus), new AuthController({ config, internalHooks, repositories, logger, postHog }), new OwnerController({ config, internalHooks, repositories, logger }), new MeController({ externalHooks, internalHooks, repositories, logger }), @@ -497,7 +500,7 @@ class Server extends AbstractServer { await handleLdapInit(); - this.registerControllers(ignoredEndpoints); + await this.registerControllers(ignoredEndpoints); this.app.use(`/${this.restEndpoint}/credentials`, credentialsController); @@ -1223,14 +1226,6 @@ class Server extends AbstractServer { ), ); - // ---------------------------------------- - // EventBus Setup - // ---------------------------------------- - - if (!eventBus.isInitialized) { - await eventBus.initialize(); - } - // ---------------------------------------- // Webhooks // ---------------------------------------- diff --git a/packages/cli/src/WorkflowRunner.ts b/packages/cli/src/WorkflowRunner.ts index 16683b57c9..ab7877fff3 100644 --- a/packages/cli/src/WorkflowRunner.ts +++ b/packages/cli/src/WorkflowRunner.ts @@ -55,7 +55,7 @@ import { generateFailedExecutionFromError } from '@/WorkflowHelpers'; import { initErrorHandling } from '@/ErrorReporting'; import { PermissionChecker } from '@/UserManagement/PermissionChecker'; import { Push } from '@/push'; -import { eventBus } from './eventbus'; +import { MessageEventBus } from '@/eventbus'; import { recoverExecutionDataFromEventLogMessages } from './eventbus/MessageEventBus/recoverEvents'; import { Container } from 'typedi'; import { InternalHooks } from './InternalHooks'; @@ -67,9 +67,12 @@ export class WorkflowRunner { jobQueue: JobQueue; + eventBus: MessageEventBus; + constructor() { this.push = Container.get(Push); this.activeExecutions = Container.get(ActiveExecutions); + this.eventBus = Container.get(MessageEventBus); } /** @@ -116,7 +119,7 @@ export class WorkflowRunner { // does contain those messages. try { // Search for messages for this executionId in event logs - const eventLogMessages = await eventBus.getEventsByExecutionId(executionId); + const eventLogMessages = await this.eventBus.getEventsByExecutionId(executionId); // Attempt to recover more better runData from these messages (but don't update the execution db entry yet) if (eventLogMessages.length > 0) { const eventLogExecutionData = await recoverExecutionDataFromEventLogMessages( diff --git a/packages/cli/src/api/e2e.api.ts b/packages/cli/src/api/e2e.api.ts index befd913114..02c3fcf229 100644 --- a/packages/cli/src/api/e2e.api.ts +++ b/packages/cli/src/api/e2e.api.ts @@ -4,6 +4,7 @@ /* eslint-disable @typescript-eslint/no-unsafe-argument */ /* eslint-disable @typescript-eslint/no-unsafe-assignment */ /* eslint-disable @typescript-eslint/naming-convention */ +import { Container } from 'typedi'; import { Router } from 'express'; import bodyParser from 'body-parser'; import { v4 as uuid } from 'uuid'; @@ -11,7 +12,7 @@ import config from '@/config'; import * as Db from '@/Db'; import type { Role } from '@db/entities/Role'; import { hashPassword } from '@/UserManagement/UserManagementHelper'; -import { eventBus } from '@/eventbus/MessageEventBus/MessageEventBus'; +import { MessageEventBus } from '@/eventbus'; if (process.env.E2E_TESTS !== 'true') { console.error('E2E endpoints only allowed during E2E tests'); @@ -79,6 +80,7 @@ const setupUserManagement = async () => { const resetLogStreaming = async () => { config.set('enterprise.features.logStreaming', false); + const eventBus = Container.get(MessageEventBus); for (const id in eventBus.destinations) { await eventBus.removeDestination(id); } diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index db53900de2..d8969a416c 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -25,10 +25,10 @@ import * as Server from '@/Server'; import { TestWebhooks } from '@/TestWebhooks'; import { getAllInstalledPackages } from '@/CommunityNodes/packageModel'; import { EDITOR_UI_DIST_DIR, GENERATED_STATIC_DIR } from '@/constants'; -import { eventBus } from '@/eventbus'; import { BaseCommand } from './BaseCommand'; import { InternalHooks } from '@/InternalHooks'; import { License } from '@/License'; +import { MessageEventBus } from '@/eventbus'; // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-var-requires const open = require('open'); @@ -133,7 +133,7 @@ export class Start extends BaseCommand { } //finally shut down Event Bus - await eventBus.close(); + await Container.get(MessageEventBus).close(); } catch (error) { await this.exitWithCrash('There was an error shutting down n8n.', error); } diff --git a/packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts b/packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts index 5d63f57b26..46b88cd37d 100644 --- a/packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts +++ b/packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts @@ -1,3 +1,4 @@ +import { Service } from 'typedi'; import { LoggerProxy } from 'n8n-workflow'; import type { MessageEventBusDestinationOptions } from 'n8n-workflow'; import type { DeleteResult } from 'typeorm'; @@ -37,12 +38,9 @@ export interface MessageWithCallback { confirmCallback: (message: EventMessageTypes, src: EventMessageConfirmSource) => void; } +@Service() export class MessageEventBus extends EventEmitter { - private static instance: MessageEventBus; - - isInitialized: boolean; - - logWriter: MessageEventBusLogWriter; + private isInitialized = false; destinations: { [key: string]: MessageEventBusDestination; @@ -50,16 +48,8 @@ export class MessageEventBus extends EventEmitter { private pushIntervalTimer: NodeJS.Timer; - constructor() { + constructor(private logWriter: MessageEventBusLogWriter) { super(); - this.isInitialized = false; - } - - static getInstance(): MessageEventBus { - if (!MessageEventBus.instance) { - MessageEventBus.instance = new MessageEventBus(); - } - return MessageEventBus.instance; } /** @@ -93,7 +83,7 @@ export class MessageEventBus extends EventEmitter { } LoggerProxy.debug('Initializing event writer'); - this.logWriter = await MessageEventBusLogWriter.getInstance(); + await this.logWriter.startThread(); // unsent event check: // - find unsent messages in current event log(s) @@ -102,9 +92,9 @@ export class MessageEventBus extends EventEmitter { LoggerProxy.debug('Checking for unsent event messages'); const unsentAndUnfinished = await this.getUnsentAndUnfinishedExecutions(); LoggerProxy.debug( - `Start logging into ${this.logWriter?.getLogFileName() ?? 'unknown filename'} `, + `Start logging into ${this.logWriter.getLogFileName() ?? 'unknown filename'} `, ); - this.logWriter?.startLogging(); + this.logWriter.startLogging(); await this.send(unsentAndUnfinished.unsentMessages); if (Object.keys(unsentAndUnfinished.unfinishedExecutions).length > 0) { @@ -171,7 +161,7 @@ export class MessageEventBus extends EventEmitter { async close() { LoggerProxy.debug('Shutting down event writer...'); - await this.logWriter?.close(); + await this.logWriter.close(); for (const destinationName of Object.keys(this.destinations)) { LoggerProxy.debug( `Shutting down event destination ${this.destinations[destinationName].getId()}...`, @@ -186,7 +176,7 @@ export class MessageEventBus extends EventEmitter { msgs = [msgs]; } for (const msg of msgs) { - this.logWriter?.putMessage(msg); + 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' }); @@ -211,7 +201,7 @@ export class MessageEventBus extends EventEmitter { } confirmSent(msg: EventMessageTypes, source?: EventMessageConfirmSource) { - this.logWriter?.confirmMessageSent(msg.id, source); + this.logWriter.confirmMessageSent(msg.id, source); } private hasAnyDestinationSubscribedToEvent(msg: EventMessageTypes): boolean { @@ -256,7 +246,7 @@ export class MessageEventBus extends EventEmitter { async getEventsFailed(amount = 5): Promise { const result: FailedEventSummary[] = []; try { - const queryResult = await this.logWriter?.getMessagesAll(); + const queryResult = await this.logWriter.getMessagesAll(); const uniques = uniqby(queryResult, 'id'); const filteredExecutionIds = uniques .filter((e) => @@ -296,25 +286,25 @@ export class MessageEventBus extends EventEmitter { } async getEventsAll(): Promise { - const queryResult = await this.logWriter?.getMessagesAll(); + const queryResult = await this.logWriter.getMessagesAll(); const filtered = uniqby(queryResult, 'id'); return filtered; } async getEventsSent(): Promise { - const queryResult = await this.logWriter?.getMessagesSent(); + const queryResult = await this.logWriter.getMessagesSent(); const filtered = uniqby(queryResult, 'id'); return filtered; } async getEventsUnsent(): Promise { - const queryResult = await this.logWriter?.getMessagesUnsent(); + const queryResult = await this.logWriter.getMessagesUnsent(); const filtered = uniqby(queryResult, 'id'); return filtered; } async getUnfinishedExecutions(): Promise> { - const queryResult = await this.logWriter?.getUnfinishedExecutions(); + const queryResult = await this.logWriter.getUnfinishedExecutions(); return queryResult; } @@ -322,7 +312,7 @@ export class MessageEventBus extends EventEmitter { unsentMessages: EventMessageTypes[]; unfinishedExecutions: Record; }> { - const queryResult = await this.logWriter?.getUnsentAndUnfinishedExecutions(); + const queryResult = await this.logWriter.getUnsentAndUnfinishedExecutions(); return queryResult; } @@ -336,7 +326,7 @@ export class MessageEventBus extends EventEmitter { executionId: string, logHistory?: number, ): Promise { - const result = await this.logWriter?.getMessagesByExecutionId(executionId, logHistory); + const result = await this.logWriter.getMessagesByExecutionId(executionId, logHistory); return result; } /** @@ -355,5 +345,3 @@ export class MessageEventBus extends EventEmitter { await this.send(new EventMessageNode(options)); } } - -export const eventBus = MessageEventBus.getInstance(); diff --git a/packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts b/packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts index af95df2e88..766417eee3 100644 --- a/packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts +++ b/packages/cli/src/eventbus/MessageEventBus/recoverEvents.ts @@ -1,3 +1,4 @@ +import { Container } from 'typedi'; import { parse, stringify } from 'flatted'; import type { IRun, IRunExecutionData, ITaskData } from 'n8n-workflow'; import { NodeOperationError, WorkflowOperationError } from 'n8n-workflow'; @@ -5,12 +6,11 @@ import * as Db from '@/Db'; import type { EventMessageTypes, EventNamesTypes } from '../EventMessageClasses'; import type { DateTime } from 'luxon'; import { Push } from '@/push'; -import type { IPushDataExecutionRecovered } from '../../Interfaces'; -import { workflowExecutionCompleted } from '../../events/WorkflowStatistics'; -import { eventBus } from './MessageEventBus'; -import { Container } from 'typedi'; +import type { IPushDataExecutionRecovered } from '@/Interfaces'; +import { workflowExecutionCompleted } from '@/events/WorkflowStatistics'; import { InternalHooks } from '@/InternalHooks'; import { getWorkflowHooksMain } from '@/WorkflowExecuteAdditionalData'; +import { MessageEventBus } from './MessageEventBus'; export async function recoverExecutionDataFromEventLogMessages( executionId: string, @@ -201,7 +201,7 @@ export async function recoverExecutionDataFromEventLogMessages( await workflowExecutionCompleted(executionEntry.workflowData, iRunData); // wait for UI to be back up and send the execution data - eventBus.once('editorUiConnected', function handleUiBackUp() { + Container.get(MessageEventBus).once('editorUiConnected', function handleUiBackUp() { // add a small timeout to make sure the UI is back up setTimeout(() => { Container.get(Push).send('executionRecovered', { diff --git a/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriter.ts b/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriter.ts index 87293f6149..207509272a 100644 --- a/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriter.ts +++ b/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriter.ts @@ -1,14 +1,17 @@ /* eslint-disable @typescript-eslint/no-unsafe-member-access */ /* eslint-disable @typescript-eslint/no-explicit-any */ -import { isEventMessageOptions } from '../EventMessageClasses/AbstractEventMessage'; -import { UserSettings } from 'n8n-core'; +import { Service } from 'typedi'; +import { once as eventOnce } from 'events'; import path, { parse } from 'path'; import { Worker } from 'worker_threads'; import { createReadStream, existsSync, rmSync } from 'fs'; import readline from 'readline'; +import { UserSettings } from 'n8n-core'; import { jsonParse, LoggerProxy } from 'n8n-workflow'; import remove from 'lodash.remove'; import config from '@/config'; +import { inTest } from '@/constants'; +import { isEventMessageOptions } from '../EventMessageClasses/AbstractEventMessage'; import { getEventMessageObjectByType } from '../EventMessageClasses/Helpers'; import type { EventMessageReturnMode } from '../MessageEventBus/MessageEventBus'; import type { EventMessageTypes } from '../EventMessageClasses'; @@ -17,15 +20,6 @@ import { EventMessageConfirm, isEventMessageConfirm, } from '../EventMessageClasses/EventMessageConfirm'; -import { once as eventOnce } from 'events'; -import { inTest } from '../../constants'; - -interface MessageEventBusLogWriterConstructorOptions { - logBaseName?: string; - logBasePath?: string; - keepNumberOfFiles?: number; - maxFileSizeInKB?: number; -} export interface MessageEventBusLogWriterOptions { logFullBasePath: string; @@ -42,41 +36,23 @@ interface ReadMessagesFromLogFileResult { /** * MessageEventBusWriter for Files */ +@Service() export class MessageEventBusLogWriter { - private static instance: MessageEventBusLogWriter; - - static options: Required; + private options: Required; private _worker: Worker | undefined; - public get worker(): Worker | undefined { - return this._worker; + constructor() { + const { keepLogCount, logBaseName, maxFileSizeInKB } = config.get('eventBus.logWriter'); + this.options = { + logFullBasePath: path.join(UserSettings.getUserN8nFolderPath(), logBaseName), + keepNumberOfFiles: keepLogCount, + maxFileSizeInKB, + }; } - /** - * Instantiates the Writer and the corresponding worker thread. - * To actually start logging, call startLogging() function on the instance. - * - * **Note** that starting to log will archive existing logs, so handle unsent events first before calling startLogging() - */ - static async getInstance( - options?: MessageEventBusLogWriterConstructorOptions, - ): Promise { - if (!MessageEventBusLogWriter.instance) { - MessageEventBusLogWriter.instance = new MessageEventBusLogWriter(); - MessageEventBusLogWriter.options = { - logFullBasePath: path.join( - options?.logBasePath ?? UserSettings.getUserN8nFolderPath(), - options?.logBaseName ?? config.getEnv('eventBus.logWriter.logBaseName'), - ), - keepNumberOfFiles: - options?.keepNumberOfFiles ?? config.getEnv('eventBus.logWriter.keepLogCount'), - maxFileSizeInKB: - options?.maxFileSizeInKB ?? config.getEnv('eventBus.logWriter.maxFileSizeInKB'), - }; - await MessageEventBusLogWriter.instance.startThread(); - } - return MessageEventBusLogWriter.instance; + public get worker(): Worker | undefined { + return this._worker; } /** @@ -98,13 +74,19 @@ export class MessageEventBusLogWriter { } } - private async startThread() { + /** + * Instantiates the Writer and the corresponding worker thread. + * To actually start logging, call startLogging() function on the instance. + * + * **Note** that starting to log will archive existing logs, so handle unsent events first before calling startLogging() + */ + async startThread() { if (this.worker) { await this.close(); } - await MessageEventBusLogWriter.instance.spawnThread(); + await this.spawnThread(); if (this.worker) { - this.worker.postMessage({ command: 'initialize', data: MessageEventBusLogWriter.options }); + this.worker.postMessage({ command: 'initialize', data: this.options }); } } @@ -120,7 +102,7 @@ export class MessageEventBusLogWriter { if (this.worker) { this.worker.on('messageerror', async (error) => { LoggerProxy.error('Event Bus Log Writer thread error, attempting to restart...', error); - await MessageEventBusLogWriter.instance.startThread(); + await this.startThread(); }); return true; } @@ -235,14 +217,14 @@ export class MessageEventBusLogWriter { getLogFileName(counter?: number): string { if (counter) { - return `${MessageEventBusLogWriter.options.logFullBasePath}-${counter}.log`; + return `${this.options.logFullBasePath}-${counter}.log`; } else { - return `${MessageEventBusLogWriter.options.logFullBasePath}.log`; + return `${this.options.logFullBasePath}.log`; } } cleanAllLogs() { - for (let i = 0; i <= MessageEventBusLogWriter.options.keepNumberOfFiles; i++) { + for (let i = 0; i <= this.options.keepNumberOfFiles; i++) { if (existsSync(this.getLogFileName(i))) { rmSync(this.getLogFileName(i)); } diff --git a/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriterWorker.ts b/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriterWorker.ts index 2672a3b987..655022b900 100644 --- a/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriterWorker.ts +++ b/packages/cli/src/eventbus/MessageEventBusWriter/MessageEventBusLogWriterWorker.ts @@ -90,11 +90,12 @@ if (!isMainThread) { clearInterval(fileStatTimer); break; case 'initialize': - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + const { logFullBasePath, keepNumberOfFiles, maxFileSizeInKB } = + data as MessageEventBusLogWriterOptions; const settings: MessageEventBusLogWriterOptions = { - logFullBasePath: (data as MessageEventBusLogWriterOptions).logFullBasePath ?? '', - keepNumberOfFiles: (data as MessageEventBusLogWriterOptions).keepNumberOfFiles ?? 10, - maxFileSizeInKB: (data as MessageEventBusLogWriterOptions).maxFileSizeInKB ?? 102400, + logFullBasePath: logFullBasePath ?? '', + keepNumberOfFiles: keepNumberOfFiles ?? 3, + maxFileSizeInKB: maxFileSizeInKB ?? 1024, }; setLogFileBasePath(settings.logFullBasePath); setKeepFiles(settings.keepNumberOfFiles); diff --git a/packages/cli/src/eventbus/eventBus.controller.ts b/packages/cli/src/eventbus/eventBus.controller.ts index bc18472d4c..b493828e95 100644 --- a/packages/cli/src/eventbus/eventBus.controller.ts +++ b/packages/cli/src/eventbus/eventBus.controller.ts @@ -6,8 +6,8 @@ import { isEventMessageOptions } from './EventMessageClasses/AbstractEventMessag import { EventMessageGeneric } from './EventMessageClasses/EventMessageGeneric'; import type { EventMessageWorkflowOptions } from './EventMessageClasses/EventMessageWorkflow'; import { EventMessageWorkflow } from './EventMessageClasses/EventMessageWorkflow'; +import { MessageEventBus } from './MessageEventBus/MessageEventBus'; import type { EventMessageReturnMode } from './MessageEventBus/MessageEventBus'; -import { eventBus } from './MessageEventBus/MessageEventBus'; import { isMessageEventBusDestinationSentryOptions, MessageEventBusDestinationSentry, @@ -76,6 +76,8 @@ const isMessageEventBusDestinationOptions = ( @RestController('/eventbus') export class EventBusController { + constructor(private eventBus: MessageEventBus) {} + // ---------------------------------------- // Events // ---------------------------------------- @@ -86,24 +88,24 @@ export class EventBusController { if (isWithQueryString(req.query)) { switch (req.query.query as EventMessageReturnMode) { case 'sent': - return eventBus.getEventsSent(); + return this.eventBus.getEventsSent(); case 'unsent': - return eventBus.getEventsUnsent(); + return this.eventBus.getEventsUnsent(); case 'unfinished': - return eventBus.getUnfinishedExecutions(); + return this.eventBus.getUnfinishedExecutions(); case 'all': default: - return eventBus.getEventsAll(); + return this.eventBus.getEventsAll(); } } else { - return eventBus.getEventsAll(); + return this.eventBus.getEventsAll(); } } @Get('/failed') async getFailedEvents(req: express.Request): Promise { const amount = parseInt(req.query?.amount as string) ?? 5; - return eventBus.getEventsFailed(amount); + return this.eventBus.getEventsFailed(amount); } @Get('/execution/:id') @@ -113,7 +115,7 @@ export class EventBusController { if (req.query?.logHistory) { logHistory = parseInt(req.query.logHistory as string, 10); } - return eventBus.getEventsByExecutionId(req.params.id, logHistory); + return this.eventBus.getEventsByExecutionId(req.params.id, logHistory); } return; } @@ -124,7 +126,7 @@ export class EventBusController { if (req.params?.id) { const logHistory = parseInt(req.query.logHistory as string, 10) || undefined; const applyToDb = req.query.applyToDb !== undefined ? !!req.query.applyToDb : true; - const messages = await eventBus.getEventsByExecutionId(id, logHistory); + const messages = await this.eventBus.getEventsByExecutionId(id, logHistory); if (messages.length > 0) { return recoverExecutionDataFromEventLogMessages(id, messages, applyToDb); } @@ -150,7 +152,7 @@ export class EventBusController { default: msg = new EventMessageGeneric(req.body); } - await eventBus.send(msg); + await this.eventBus.send(msg); } else { throw new BadRequestError( 'Body is not a serialized EventMessage or eventName does not match format {namespace}.{domain}.{event}', @@ -166,9 +168,9 @@ export class EventBusController { @Get('/destination') async getDestination(req: express.Request): Promise { if (isWithIdString(req.query)) { - return eventBus.findDestination(req.query.id); + return this.eventBus.findDestination(req.query.id); } else { - return eventBus.findDestination(); + return this.eventBus.findDestination(); } } @@ -183,22 +185,22 @@ export class EventBusController { switch (req.body.__type) { case MessageEventBusDestinationTypeNames.sentry: if (isMessageEventBusDestinationSentryOptions(req.body)) { - result = await eventBus.addDestination( - new MessageEventBusDestinationSentry(eventBus, req.body), + result = await this.eventBus.addDestination( + new MessageEventBusDestinationSentry(this.eventBus, req.body), ); } break; case MessageEventBusDestinationTypeNames.webhook: if (isMessageEventBusDestinationWebhookOptions(req.body)) { - result = await eventBus.addDestination( - new MessageEventBusDestinationWebhook(eventBus, req.body), + result = await this.eventBus.addDestination( + new MessageEventBusDestinationWebhook(this.eventBus, req.body), ); } break; case MessageEventBusDestinationTypeNames.syslog: if (isMessageEventBusDestinationSyslogOptions(req.body)) { - result = await eventBus.addDestination( - new MessageEventBusDestinationSyslog(eventBus, req.body), + result = await this.eventBus.addDestination( + new MessageEventBusDestinationSyslog(this.eventBus, req.body), ); } break; @@ -223,7 +225,7 @@ export class EventBusController { @Get('/testmessage') async sendTestMessage(req: express.Request): Promise { if (isWithIdString(req.query)) { - return eventBus.testDestination(req.query.id); + return this.eventBus.testDestination(req.query.id); } return false; } @@ -234,7 +236,7 @@ export class EventBusController { throw new ResponseHelper.UnauthorizedError('Invalid request'); } if (isWithIdString(req.query)) { - return eventBus.removeDestination(req.query.id); + return this.eventBus.removeDestination(req.query.id); } else { throw new BadRequestError('Query is missing id'); } diff --git a/packages/cli/src/eventbus/index.ts b/packages/cli/src/eventbus/index.ts index 1b3b48d7af..563705976d 100644 --- a/packages/cli/src/eventbus/index.ts +++ b/packages/cli/src/eventbus/index.ts @@ -1 +1 @@ -export { eventBus } from './MessageEventBus/MessageEventBus'; +export { MessageEventBus } from './MessageEventBus/MessageEventBus'; diff --git a/packages/cli/src/push/abstract.push.ts b/packages/cli/src/push/abstract.push.ts index a9022bdc5c..c789cd7a52 100644 --- a/packages/cli/src/push/abstract.push.ts +++ b/packages/cli/src/push/abstract.push.ts @@ -1,17 +1,24 @@ +import { Container } from 'typedi'; import { jsonStringify, LoggerProxy as Logger } from 'n8n-workflow'; import type { IPushDataType } from '@/Interfaces'; -import { eventBus } from '../eventbus'; +import { MessageEventBus } from '@/eventbus'; export abstract class AbstractPush { protected connections: Record = {}; + protected eventBus: MessageEventBus; + + constructor() { + this.eventBus = Container.get(MessageEventBus); + } + protected abstract close(connection: T): void; protected abstract sendToOne(connection: T, data: string): void; protected add(sessionId: string, connection: T): void { const { connections } = this; Logger.debug('Add editor-UI session', { sessionId }); - eventBus.emit('editorUiConnected', sessionId); + this.eventBus.emit('editorUiConnected', sessionId); const existingConnection = connections[sessionId]; if (existingConnection) { diff --git a/packages/cli/test/integration/eventbus.test.ts b/packages/cli/test/integration/eventbus.test.ts index d456f86799..8ad83f0fc3 100644 --- a/packages/cli/test/integration/eventbus.test.ts +++ b/packages/cli/test/integration/eventbus.test.ts @@ -1,3 +1,4 @@ +import { Container } from 'typedi'; import express from 'express'; import config from '@/config'; import axios from 'axios'; @@ -16,7 +17,7 @@ import { MessageEventBusDestinationSyslogOptions, MessageEventBusDestinationWebhookOptions, } from 'n8n-workflow'; -import { eventBus } from '@/eventbus'; +import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus'; import { EventMessageGeneric } from '@/eventbus/EventMessageClasses/EventMessageGeneric'; import { MessageEventBusDestinationSyslog } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationSyslog.ee'; import { MessageEventBusDestinationWebhook } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationWebhook.ee'; @@ -64,6 +65,8 @@ const testSentryDestination: MessageEventBusDestinationSentryOptions = { subscribedEvents: ['n8n.test.message', 'n8n.audit.user.updated'], }; +const eventBus = Container.get(MessageEventBus); + async function confirmIdInAll(id: string) { const sent = await eventBus.getEventsAll(); expect(sent.length).toBeGreaterThan(0); diff --git a/packages/cli/test/integration/shared/utils.ts b/packages/cli/test/integration/shared/utils.ts index e8bf676ba7..c3f77fd9b4 100644 --- a/packages/cli/test/integration/shared/utils.ts +++ b/packages/cli/test/integration/shared/utils.ts @@ -81,6 +81,7 @@ import { setSamlLoginEnabled } from '@/sso/saml/samlHelpers'; import { SamlService } from '@/sso/saml/saml.service.ee'; import { SamlController } from '@/sso/saml/routes/saml.controller.ee'; import { EventBusController } from '@/eventbus/eventBus.controller'; +import { MessageEventBus } from '@/eventbus'; export const mockInstance = ( ctor: new (...args: any[]) => T, @@ -176,7 +177,8 @@ export async function initTestServer({ for (const group of functionEndpoints) { switch (group) { case 'eventBus': - registerController(testServer.app, config, new EventBusController()); + const eventBus = Container.get(MessageEventBus); + registerController(testServer.app, config, new EventBusController(eventBus)); break; case 'auth': registerController( diff --git a/packages/cli/test/unit/ActiveWorkflowRunner.test.ts b/packages/cli/test/unit/ActiveWorkflowRunner.test.ts index e3290446e0..052cbf880a 100644 --- a/packages/cli/test/unit/ActiveWorkflowRunner.test.ts +++ b/packages/cli/test/unit/ActiveWorkflowRunner.test.ts @@ -29,6 +29,7 @@ import { mockInstance } from '../integration/shared/utils'; import { Push } from '@/push'; import { ActiveExecutions } from '@/ActiveExecutions'; import { NodeTypes } from '@/NodeTypes'; +import { MessageEventBus } from '@/eventbus'; /** * TODO: @@ -155,6 +156,7 @@ describe('ActiveWorkflowRunner', () => { }; Container.set(LoadNodesAndCredentials, nodesAndCredentials); mockInstance(Push); + mockInstance(MessageEventBus); }); beforeEach(() => {