refactor(core): Decouple lifecycle events from internal hooks (no-changelog) (#10305)

packages/cli/src/InternalHooks.ts

@@ -1,7 +1,6 @@
 import { Service } from 'typedi';
 import type { ITelemetryTrackProperties } from 'n8n-workflow';
 import type { User } from '@db/entities/User';
-import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
 import { Telemetry } from '@/telemetry';
 import { MessageEventBus } from './eventbus/MessageEventBus/MessageEventBus';
 
@@ -14,28 +13,16 @@ import { MessageEventBus } from './eventbus/MessageEventBus/MessageEventBus';
 export class InternalHooks {
 	constructor(
 		private readonly telemetry: Telemetry,
-		workflowStatisticsService: WorkflowStatisticsService,
 		// Can't use @ts-expect-error because only dev time tsconfig considers this as an error, but not build time
 		// eslint-disable-next-line @typescript-eslint/ban-ts-comment
 		// @ts-ignore - needed until we decouple telemetry
 		private readonly _eventBus: MessageEventBus, // needed until we decouple telemetry
-	) {
-		workflowStatisticsService.on('telemetry.onFirstProductionWorkflowSuccess', (metrics) =>
-			this.onFirstProductionWorkflowSuccess(metrics),
-		);
-		workflowStatisticsService.on('telemetry.onFirstWorkflowDataLoad', (metrics) =>
-			this.onFirstWorkflowDataLoad(metrics),
-		);
-	}
+	) {}
 
 	async init() {
 		await this.telemetry.init();
 	}
 
-	onFrontendSettingsAPI(pushRef?: string): void {
-		this.telemetry.track('Session started', { session_id: pushRef });
-	}
-
 	onWorkflowSharingUpdate(workflowId: string, userId: string, userList: string[]) {
 		const properties: ITelemetryTrackProperties = {
 			workflow_id: workflowId,
@@ -46,14 +33,6 @@ export class InternalHooks {
 		this.telemetry.track('User updated workflow sharing', properties);
 	}
 
-	async onN8nStop(): Promise<void> {
-		const timeoutPromise = new Promise<void>((resolve) => {
-			setTimeout(resolve, 3000);
-		});
-
-		return await Promise.race([timeoutPromise, this.telemetry.trackN8nStop()]);
-	}
-
 	onUserInviteEmailClick(userInviteClickData: { inviter: User; invitee: User }) {
 		this.telemetry.track('User clicked invite link from email', {
 			user_id: userInviteClickData.invitee.id,
@@ -85,10 +64,6 @@ export class InternalHooks {
 		});
 	}
 
-	onInstanceOwnerSetup(instanceOwnerSetupData: { user_id: string }) {
-		this.telemetry.track('Owner finished instance setup', instanceOwnerSetupData);
-	}
-
 	onEmailFailed(failedEmailData: {
 		user: User;
 		message_type:
@@ -103,22 +78,4 @@ export class InternalHooks {
 			user_id: failedEmailData.user.id,
 		});
 	}
-
-	/*
-	 * Execution Statistics
-	 */
-	onFirstProductionWorkflowSuccess(data: { user_id: string; workflow_id: string }) {
-		this.telemetry.track('Workflow first prod success', data);
-	}
-
-	onFirstWorkflowDataLoad(data: {
-		user_id: string;
-		workflow_id: string;
-		node_type: string;
-		node_id: string;
-		credential_type?: string;
-		credential_id?: string;
-	}) {
-		this.telemetry.track('Workflow first data fetched', data);
-	}
 }

packages/cli/src/commands/start.ts

@@ -16,7 +16,6 @@ import { ActiveWorkflowManager } from '@/ActiveWorkflowManager';
 import { Server } from '@/Server';
 import { EDITOR_UI_DIST_DIR, LICENSE_FEATURES } from '@/constants';
 import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
-import { InternalHooks } from '@/InternalHooks';
 import { License } from '@/License';
 import { OrchestrationService } from '@/services/orchestration.service';
 import { OrchestrationHandlerMainService } from '@/services/orchestration/main/orchestration.handler.main.service';
@@ -110,7 +109,7 @@ export class Start extends BaseCommand {
 				await Container.get(OrchestrationService).shutdown();
 			}
 
-			await Container.get(InternalHooks).onN8nStop();
+			Container.get(EventService).emit('instance-stopped');
 
 			await Container.get(ActiveExecutions).shutdown();
 

packages/cli/src/controllers/__tests__/owner.controller.test.ts

@@ -10,7 +10,6 @@ import type { User } from '@db/entities/User';
 import type { SettingsRepository } from '@db/repositories/settings.repository';
 import type { UserRepository } from '@db/repositories/user.repository';
 import { BadRequestError } from '@/errors/response-errors/bad-request.error';
-import type { InternalHooks } from '@/InternalHooks';
 import { License } from '@/License';
 import type { OwnerRequest } from '@/requests';
 import type { UserService } from '@/services/user.service';
@@ -21,7 +20,6 @@ import { badPasswords } from '@test/testData';
 
 describe('OwnerController', () => {
 	const configGetSpy = jest.spyOn(config, 'getEnv');
-	const internalHooks = mock<InternalHooks>();
 	const authService = mock<AuthService>();
 	const userService = mock<UserService>();
 	const userRepository = mock<UserRepository>();
@@ -29,7 +27,7 @@ describe('OwnerController', () => {
 	mockInstance(License).isWithinUsersLimit.mockReturnValue(true);
 	const controller = new OwnerController(
 		mock(),
-		internalHooks,
+		mock(),
 		settingsRepository,
 		authService,
 		userService,

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

@@ -13,13 +13,13 @@ import { PostHogClient } from '@/posthog';
 import { UserService } from '@/services/user.service';
 import { Logger } from '@/Logger';
 import { BadRequestError } from '@/errors/response-errors/bad-request.error';
-import { InternalHooks } from '@/InternalHooks';
+import { EventService } from '@/events/event.service';
 
 @RestController('/owner')
 export class OwnerController {
 	constructor(
 		private readonly logger: Logger,
-		private readonly internalHooks: InternalHooks,
+		private readonly eventService: EventService,
 		private readonly settingsRepository: SettingsRepository,
 		private readonly authService: AuthService,
 		private readonly userService: UserService,
@@ -85,7 +85,7 @@ export class OwnerController {
 
 		this.authService.issueCookie(res, owner, req.browserId);
 
-		this.internalHooks.onInstanceOwnerSetup({ user_id: owner.id });
+		this.eventService.emit('instance-owner-setup', { userId: owner.id });
 
 		return await this.userService.toPublic(owner, { posthog: this.postHog, withScopes: true });
 	}

packages/cli/src/events/relay-event-map.ts

@@ -13,10 +13,35 @@ export type UserLike = {
 };
 
 export type RelayEventMap = {
-	// #region Server
+	// #region Lifecycle
 
 	'server-started': {};
 
+	'session-started': {
+		pushRef?: string;
+	};
+
+	'instance-stopped': {};
+
+	'instance-owner-setup': {
+		userId: string;
+	};
+
+	'first-production-workflow-succeeded': {
+		projectId: string;
+		workflowId: string;
+		userId: string;
+	};
+
+	'first-workflow-data-loaded': {
+		userId: string;
+		workflowId: string;
+		nodeType: string;
+		nodeId: string;
+		credentialType?: string;
+		credentialId?: string;
+	};
+
 	// #endregion
 
 	// #region Workflow

packages/cli/src/events/telemetry-event-relay.ts

@@ -73,6 +73,12 @@ export class TelemetryEventRelay extends EventRelay {
 			'workflow-deleted': (event) => this.workflowDeleted(event),
 			'workflow-saved': async (event) => await this.workflowSaved(event),
 			'server-started': async () => await this.serverStarted(),
+			'session-started': (event) => this.sessionStarted(event),
+			'instance-stopped': () => this.instanceStopped(),
+			'instance-owner-setup': async (event) => await this.instanceOwnerSetup(event),
+			'first-production-workflow-succeeded': (event) =>
+				this.firstProductionWorkflowSucceeded(event),
+			'first-workflow-data-loaded': (event) => this.firstWorkflowDataLoaded(event),
 			'workflow-post-execute': async (event) => await this.workflowPostExecute(event),
 			'user-changed-role': (event) => this.userChangedRole(event),
 			'user-retrieved-user': (event) => this.userRetrievedUser(event),
@@ -687,7 +693,7 @@ export class TelemetryEventRelay extends EventRelay {
 
 	// #endregion
 
-	// #region Server
+	// #region Lifecycle
 
 	private async serverStarted() {
 		const cpus = os.cpus();
@@ -753,6 +759,48 @@ export class TelemetryEventRelay extends EventRelay {
 		});
 	}
 
+	private sessionStarted({ pushRef }: RelayEventMap['session-started']) {
+		this.telemetry.track('Session started', { session_id: pushRef });
+	}
+
+	private instanceStopped() {
+		this.telemetry.track('User instance stopped');
+	}
+
+	private async instanceOwnerSetup({ userId }: RelayEventMap['instance-owner-setup']) {
+		this.telemetry.track('Owner finished instance setup', { user_id: userId });
+	}
+
+	private firstProductionWorkflowSucceeded({
+		projectId,
+		workflowId,
+		userId,
+	}: RelayEventMap['first-production-workflow-succeeded']) {
+		this.telemetry.track('Workflow first prod success', {
+			project_id: projectId,
+			workflow_id: workflowId,
+			user_id: userId,
+		});
+	}
+
+	private firstWorkflowDataLoaded({
+		userId,
+		workflowId,
+		nodeType,
+		nodeId,
+		credentialType,
+		credentialId,
+	}: RelayEventMap['first-workflow-data-loaded']) {
+		this.telemetry.track('Workflow first data fetched', {
+			user_id: userId,
+			workflow_id: workflowId,
+			node_type: nodeType,
+			node_id: nodeId,
+			credential_type: credentialType,
+			credential_id: credentialId,
+		});
+	}
+
 	// #endregion
 
 	// #region User

packages/cli/src/services/__tests__/workflow-statistics.service.test.ts

@@ -19,6 +19,7 @@ import { UserService } from '@/services/user.service';
 import { OwnershipService } from '@/services/ownership.service';
 import { mockInstance } from '@test/mocking';
 import type { Project } from '@/databases/entities/Project';
+import type { EventService } from '@/events/event.service';
 
 describe('WorkflowStatisticsService', () => {
 	const fakeUser = mock<User>({ id: 'abcde-fghij' });
@@ -44,21 +45,15 @@ describe('WorkflowStatisticsService', () => {
 	mocked(ownershipService.getProjectOwnerCached).mockResolvedValue(fakeUser);
 	const updateSettingsMock = jest.spyOn(userService, 'updateSettings').mockImplementation();
 
+	const eventService = mock<EventService>();
 	const workflowStatisticsService = new WorkflowStatisticsService(
 		mock(),
 		new WorkflowStatisticsRepository(dataSource, globalConfig),
 		ownershipService,
 		userService,
+		eventService,
 	);
 
-	const onFirstProductionWorkflowSuccess = jest.fn();
-	const onFirstWorkflowDataLoad = jest.fn();
-	workflowStatisticsService.on(
-		'telemetry.onFirstProductionWorkflowSuccess',
-		onFirstProductionWorkflowSuccess,
-	);
-	workflowStatisticsService.on('telemetry.onFirstWorkflowDataLoad', onFirstWorkflowDataLoad);
-
 	beforeEach(() => {
 		jest.clearAllMocks();
 	});
@@ -97,11 +92,10 @@ describe('WorkflowStatisticsService', () => {
 
 			await workflowStatisticsService.workflowExecutionCompleted(workflow, runData);
 			expect(updateSettingsMock).toHaveBeenCalledTimes(1);
-			expect(onFirstProductionWorkflowSuccess).toBeCalledTimes(1);
-			expect(onFirstProductionWorkflowSuccess).toHaveBeenNthCalledWith(1, {
-				project_id: fakeProject.id,
-				user_id: fakeUser.id,
-				workflow_id: workflow.id,
+			expect(eventService.emit).toHaveBeenCalledWith('first-production-workflow-succeeded', {
+				projectId: fakeProject.id,
+				workflowId: workflow.id,
+				userId: fakeUser.id,
 			});
 		});
 
@@ -124,7 +118,7 @@ describe('WorkflowStatisticsService', () => {
 				startedAt: new Date(),
 			};
 			await workflowStatisticsService.workflowExecutionCompleted(workflow, runData);
-			expect(onFirstProductionWorkflowSuccess).toBeCalledTimes(0);
+			expect(eventService.emit).not.toHaveBeenCalled();
 		});
 
 		test('should not send metrics for updated entries', async () => {
@@ -147,7 +141,7 @@ describe('WorkflowStatisticsService', () => {
 			};
 			mockDBCall(2);
 			await workflowStatisticsService.workflowExecutionCompleted(workflow, runData);
-			expect(onFirstProductionWorkflowSuccess).toBeCalledTimes(0);
+			expect(eventService.emit).not.toHaveBeenCalled();
 		});
 	});
 
@@ -164,13 +158,12 @@ describe('WorkflowStatisticsService', () => {
 				parameters: {},
 			};
 			await workflowStatisticsService.nodeFetchedData(workflowId, node);
-			expect(onFirstWorkflowDataLoad).toBeCalledTimes(1);
-			expect(onFirstWorkflowDataLoad).toHaveBeenNthCalledWith(1, {
-				user_id: fakeUser.id,
-				project_id: fakeProject.id,
-				workflow_id: workflowId,
-				node_type: node.type,
-				node_id: node.id,
+			expect(eventService.emit).toHaveBeenCalledWith('first-workflow-data-loaded', {
+				userId: fakeUser.id,
+				project: fakeProject.id,
+				workflowId,
+				nodeType: node.type,
+				nodeId: node.id,
 			});
 		});
 
@@ -192,15 +185,14 @@ describe('WorkflowStatisticsService', () => {
 				},
 			};
 			await workflowStatisticsService.nodeFetchedData(workflowId, node);
-			expect(onFirstWorkflowDataLoad).toBeCalledTimes(1);
-			expect(onFirstWorkflowDataLoad).toHaveBeenNthCalledWith(1, {
-				user_id: fakeUser.id,
-				project_id: fakeProject.id,
-				workflow_id: workflowId,
-				node_type: node.type,
-				node_id: node.id,
-				credential_type: 'testCredentials',
-				credential_id: node.credentials.testCredentials.id,
+			expect(eventService.emit).toHaveBeenCalledWith('first-workflow-data-loaded', {
+				userId: fakeUser.id,
+				project: fakeProject.id,
+				workflowId,
+				nodeType: node.type,
+				nodeId: node.id,
+				credentialType: 'testCredentials',
+				credentialId: node.credentials.testCredentials.id,
 			});
 		});
 
@@ -217,7 +209,7 @@ describe('WorkflowStatisticsService', () => {
 				parameters: {},
 			};
 			await workflowStatisticsService.nodeFetchedData(workflowId, node);
-			expect(onFirstWorkflowDataLoad).toBeCalledTimes(0);
+			expect(eventService.emit).not.toHaveBeenCalled();
 		});
 	});
 });

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

@@ -31,7 +31,7 @@ import { UserManagementMailer } from '@/UserManagement/email';
 import type { CommunityPackagesService } from '@/services/communityPackages.service';
 import { Logger } from '@/Logger';
 import { UrlService } from './url.service';
-import { InternalHooks } from '@/InternalHooks';
+import { EventService } from '@/events/event.service';
 import { isApiEnabled } from '@/PublicApi';
 
 @Service()
@@ -50,7 +50,7 @@ export class FrontendService {
 		private readonly mailer: UserManagementMailer,
 		private readonly instanceSettings: InstanceSettings,
 		private readonly urlService: UrlService,
-		private readonly internalHooks: InternalHooks,
+		private readonly eventService: EventService,
 	) {
 		loadNodesAndCredentials.addPostProcessor(async () => await this.generateTypes());
 		void this.generateTypes();
@@ -244,7 +244,7 @@ export class FrontendService {
 	}
 
 	getSettings(pushRef?: string): IN8nUISettings {
-		this.internalHooks.onFrontendSettingsAPI(pushRef);
+		this.eventService.emit('session-started', { pushRef });
 
 		const restEndpoint = this.globalConfig.endpoints.rest;
 

packages/cli/src/services/workflow-statistics.service.ts

@@ -6,6 +6,7 @@ import { UserService } from '@/services/user.service';
 import { Logger } from '@/Logger';
 import { OwnershipService } from './ownership.service';
 import { TypedEmitter } from '@/TypedEmitter';
+import { EventService } from '@/events/event.service';
 
 type WorkflowStatisticsEvents = {
 	nodeFetchedData: { workflowId: string; node: INode };
@@ -31,6 +32,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 		private readonly repository: WorkflowStatisticsRepository,
 		private readonly ownershipService: OwnershipService,
 		private readonly userService: UserService,
+		private readonly eventService: EventService,
 	) {
 		super({ captureRejections: true });
 		if ('SKIP_STATISTICS_EVENTS' in process.env) return;
@@ -72,12 +74,6 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 				if (project.type === 'personal') {
 					const owner = await this.ownershipService.getProjectOwnerCached(project.id);
 
-					const metrics = {
-						project_id: project.id,
-						workflow_id: workflowId,
-						user_id: owner!.id,
-					};
-
 					if (owner && !owner.settings?.userActivated) {
 						await this.userService.updateSettings(owner.id, {
 							firstSuccessfulWorkflowId: workflowId,
@@ -86,8 +82,11 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 						});
 					}
 
-					// Send the metrics
-					this.emit('telemetry.onFirstProductionWorkflowSuccess', metrics);
+					this.eventService.emit('first-production-workflow-succeeded', {
+						projectId: project.id,
+						workflowId,
+						userId: owner!.id,
+					});
 				}
 			}
 		} catch (error) {
@@ -109,24 +108,23 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 		const owner = await this.ownershipService.getProjectOwnerCached(project.id);
 
 		let metrics = {
-			user_id: owner!.id,
-			project_id: project.id,
-			workflow_id: workflowId,
-			node_type: node.type,
-			node_id: node.id,
+			userId: owner!.id,
+			project: project.id,
+			workflowId,
+			nodeType: node.type,
+			nodeId: node.id,
 		};
 
 		// This is probably naive but I can't see a way for a node to have multiple credentials attached so..
 		if (node.credentials) {
 			Object.entries(node.credentials).forEach(([credName, credDetails]) => {
 				metrics = Object.assign(metrics, {
-					credential_type: credName,
-					credential_id: credDetails.id,
+					credentialType: credName,
+					credentialId: credDetails.id,
 				});
 			});
 		}
 
-		// Send metrics to posthog
-		this.emit('telemetry.onFirstWorkflowDataLoad', metrics);
+		this.eventService.emit('first-workflow-data-loaded', metrics);
 	}
 }

packages/cli/src/telemetry/__tests__/telemetry.test.ts

@@ -34,7 +34,7 @@ describe('Telemetry', () => {
 		jest.clearAllTimers();
 		jest.useRealTimers();
 		startPulseSpy.mockRestore();
-		await telemetry.trackN8nStop();
+		await telemetry.stopTracking();
 	});
 
 	beforeEach(async () => {
@@ -49,14 +49,7 @@ describe('Telemetry', () => {
 	});
 
 	afterEach(async () => {
-		await telemetry.trackN8nStop();
-	});
-
-	describe('trackN8nStop', () => {
-		test('should call track method', async () => {
-			await telemetry.trackN8nStop();
-			expect(spyTrack).toHaveBeenCalledTimes(1);
-		});
+		await telemetry.stopTracking();
 	});
 
 	describe('trackWorkflowExecution', () => {

packages/cli/src/telemetry/index.ts

@@ -9,12 +9,13 @@ import config from '@/config';
 import type { IExecutionTrackProperties } from '@/Interfaces';
 import { Logger } from '@/Logger';
 import { License } from '@/License';
-import { N8N_VERSION } from '@/constants';
+import { LOWEST_SHUTDOWN_PRIORITY, N8N_VERSION } from '@/constants';
 import { WorkflowRepository } from '@db/repositories/workflow.repository';
 import { SourceControlPreferencesService } from '../environments/sourceControl/sourceControlPreferences.service.ee';
 import { UserRepository } from '@db/repositories/user.repository';
 import { ProjectRepository } from '@/databases/repositories/project.repository';
 import { ProjectRelationRepository } from '@/databases/repositories/projectRelation.repository';
+import { OnShutdown } from '@/decorators/OnShutdown';
 
 type ExecutionTrackDataKey = 'manual_error' | 'manual_success' | 'prod_error' | 'prod_success';
 
@@ -167,11 +168,10 @@ export class Telemetry {
 		}
 	}
 
-	async trackN8nStop(): Promise<void> {
+	@OnShutdown(LOWEST_SHUTDOWN_PRIORITY)
+	async stopTracking(): Promise<void> {
 		clearInterval(this.pulseIntervalReference);
 
-		this.track('User instance stopped');
-
 		await Promise.all([this.postHog.stop(), this.rudderStack?.flush()]);
 	}