feat(core): Workflow Execution Statistics

packages/cli/package.json

@@ -165,7 +165,7 @@
     "passport-jwt": "^4.0.0",
     "pg": "^8.3.0",
     "picocolors": "^1.0.0",
-    "posthog-node": "^1.3.0",
+    "posthog-node": "^2.2.2",
     "prom-client": "^13.1.0",
     "psl": "^1.8.0",
     "replacestream": "^4.0.3",

packages/cli/src/Db.ts

@@ -178,6 +178,7 @@ export async function init(
 	collections.Settings = linkRepository(entities.Settings);
 	collections.InstalledPackages = linkRepository(entities.InstalledPackages);
 	collections.InstalledNodes = linkRepository(entities.InstalledNodes);
+	collections.WorkflowStatistics = linkRepository(entities.WorkflowStatistics);
 
 	isInitialized = true;
 

packages/cli/src/Interfaces.ts

@@ -39,6 +39,7 @@ import type { SharedWorkflow } from '@db/entities/SharedWorkflow';
 import type { TagEntity } from '@db/entities/TagEntity';
 import type { User } from '@db/entities/User';
 import type { WorkflowEntity } from '@db/entities/WorkflowEntity';
+import type { WorkflowStatistics } from '@db/entities/WorkflowStatistics';
 
 export interface IActivationError {
 	time: number;
@@ -79,6 +80,7 @@ export interface IDatabaseCollections {
 	Settings: Repository<Settings>;
 	InstalledPackages: Repository<InstalledPackages>;
 	InstalledNodes: Repository<InstalledNodes>;
+	WorkflowStatistics: Repository<WorkflowStatistics>;
 }
 
 export interface IWebhookDb {
@@ -691,6 +693,24 @@ export interface IWorkflowExecuteProcess {
 	workflowExecute: WorkflowExecute;
 }
 
+export interface IWorkflowStatisticsCounts {
+	productionSuccess: number;
+	productionError: number;
+	manualSuccess: number;
+	manualError: number;
+}
+
+export interface IWorkflowStatisticsDataLoaded {
+	dataLoaded: boolean;
+}
+
+export interface IWorkflowStatisticsTimestamps {
+	productionSuccess: Date | null;
+	productionError: Date | null;
+	manualSuccess: Date | null;
+	manualError: Date | null;
+}
+
 export type WhereClause = Record<string, { [key: string]: string | FindOperator<unknown> }>;
 
 // ----------------------------------

packages/cli/src/InternalHooks.ts

@@ -477,4 +477,25 @@ export class InternalHooksClass implements IInternalHooksClass {
 	}): Promise<void> {
 		return this.telemetry.track('cnr package deleted', updateData);
 	}
+
+	/**
+	 * Execution Statistics
+	 */
+	async onFirstProductionWorkflowSuccess(data: {
+		user_id: string;
+		workflow_id: string | number;
+	}): Promise<void> {
+		return this.telemetry.track('Workflow first prod success', data, { withPostHog: true });
+	}
+
+	async onFirstWorkflowDataLoad(data: {
+		user_id: string;
+		workflow_id: string | number;
+		node_type: string;
+		node_id: string;
+		credential_type?: string;
+		credential_id?: string;
+	}): Promise<void> {
+		return this.telemetry.track('Workflow first data fetched', data, { withPostHog: true });
+	}
 }

packages/cli/src/Server.ts

@@ -112,6 +112,7 @@ import { resolveJwt } from '@/UserManagement/auth/jwt';
 import { executionsController } from '@/executions/executions.controller';
 import { nodeTypesController } from '@/api/nodeTypes.api';
 import { tagsController } from '@/api/tags.api';
+import { workflowStatsController } from '@/api/workflowStats.api';
 import { loadPublicApiVersions } from '@/PublicApi';
 import {
 	getInstanceBaseUrl,
@@ -806,6 +807,11 @@ class App {
 		// ----------------------------------------
 		this.app.use(`/${this.restEndpoint}/workflows`, workflowsController);
 
+		// ----------------------------------------
+		// Workflow Statistics
+		// ----------------------------------------
+		this.app.use(`/${this.restEndpoint}/workflow-stats`, workflowStatsController);
+
 		// ----------------------------------------
 		// Tags
 		// ----------------------------------------

packages/cli/src/WebhookHelpers.ts

@@ -16,7 +16,7 @@
 import express from 'express';
 import get from 'lodash.get';
 
-import { BINARY_ENCODING, BinaryDataManager, NodeExecuteFunctions } from 'n8n-core';
+import { BINARY_ENCODING, BinaryDataManager, NodeExecuteFunctions, eventEmitter } from 'n8n-core';
 
 import {
 	createDeferredPromise,
@@ -233,6 +233,7 @@ export async function executeWebhook(
 				NodeExecuteFunctions,
 				executionMode,
 			);
+			eventEmitter.emit(eventEmitter.types.nodeFetchedData, workflow.id, workflowStartNode);
 		} catch (err) {
 			// Send error response to webhook caller
 			const errorMessage = 'Workflow Webhook Error: Workflow could not be started!';

packages/cli/src/WorkflowExecuteAdditionalData.ts

@@ -15,12 +15,13 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 /* eslint-disable func-names */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
-import { BinaryDataManager, UserSettings, WorkflowExecute } from 'n8n-core';
+import { BinaryDataManager, eventEmitter, UserSettings, WorkflowExecute } from 'n8n-core';
 
 import {
 	IDataObject,
 	IExecuteData,
 	IExecuteWorkflowInfo,
+	INode,
 	INodeExecutionData,
 	INodeParameters,
 	IRun,
@@ -648,9 +649,20 @@ function hookFunctionsSave(parentProcessMode?: string): IWorkflowExecuteHooks {
 							this.retryOf,
 						);
 					}
+				} finally {
+					eventEmitter.emit(
+						eventEmitter.types.workflowExecutionCompleted,
+						this.workflowData,
+						fullRunData,
+					);
 				}
 			},
 		],
+		nodeFetchedData: [
+			async (workflowId: string, node: INode) => {
+				eventEmitter.emit(eventEmitter.types.nodeFetchedData, workflowId, node);
+			},
+		],
 	};
 }
 
@@ -742,9 +754,20 @@ function hookFunctionsSaveWorker(): IWorkflowExecuteHooks {
 						this.executionId,
 						this.retryOf,
 					);
+				} finally {
+					eventEmitter.emit(
+						eventEmitter.types.workflowExecutionCompleted,
+						this.workflowData,
+						fullRunData,
+					);
 				}
 			},
 		],
+		nodeFetchedData: [
+			async (workflowId: string, node: INode) => {
+				eventEmitter.emit(eventEmitter.types.nodeFetchedData, workflowId, node);
+			},
+		],
 	};
 }
 

packages/cli/src/WorkflowRunnerProcess.ts

@@ -16,6 +16,7 @@ import {
 	IExecuteResponsePromiseData,
 	IExecuteWorkflowInfo,
 	ILogger,
+	INode,
 	INodeExecutionData,
 	IRun,
 	ITaskData,
@@ -396,6 +397,11 @@ class WorkflowRunnerProcess {
 					await this.sendHookToParentProcess('workflowExecuteAfter', [fullRunData, newStaticData]);
 				},
 			],
+			nodeFetchedData: [
+				async (workflowId: string, node: INode) => {
+					await this.sendHookToParentProcess('nodeFetchedData', [workflowId, node]);
+				},
+			],
 		};
 
 		const preExecuteFunctions = WorkflowExecuteAdditionalData.hookFunctionsPreExecute();

packages/cli/src/api/workflowStats.api.ts

@@ -0,0 +1,185 @@
+import { User } from '@/databases/entities/User';
+import { whereClause } from '@/UserManagement/UserManagementHelper';
+import express from 'express';
+import { LoggerProxy } from 'n8n-workflow';
+import {
+	Db,
+	IWorkflowStatisticsCounts,
+	IWorkflowStatisticsDataLoaded,
+	IWorkflowStatisticsTimestamps,
+	ResponseHelper,
+} from '..';
+import { StatisticsNames } from '../databases/entities/WorkflowStatistics';
+import { getLogger } from '../Logger';
+import { ExecutionRequest } from '../requests';
+
+export const workflowStatsController = express.Router();
+
+// Helper function that validates the ID, return a flag stating whether the request is allowed
+async function checkWorkflowId(workflowId: string, user: User): Promise<boolean> {
+	// Check permissions
+	const shared = await Db.collections.SharedWorkflow.findOne({
+		relations: ['workflow'],
+		where: whereClause({
+			user,
+			entityType: 'workflow',
+			entityId: workflowId,
+		}),
+	});
+
+	if (!shared) {
+		LoggerProxy.info('User attempted to read a workflow without permissions', {
+			workflowId,
+			userId: user.id,
+		});
+		return false;
+	}
+	return true;
+}
+
+/**
+ * Initialise Logger if needed
+ */
+workflowStatsController.use((req, res, next) => {
+	try {
+		LoggerProxy.getInstance();
+	} catch (error) {
+		LoggerProxy.init(getLogger());
+	}
+
+	next();
+});
+
+/**
+ * Check that the workflow ID is valid and allowed to be read by the user
+ */
+workflowStatsController.use(async (req: ExecutionRequest.Get, res, next) => {
+	const allowed = await checkWorkflowId(req.params.id, req.user);
+	if (allowed) {
+		next();
+	} else {
+		// Otherwise, make and return an error
+		const response = new ResponseHelper.NotFoundError(`Workflow ${req.params.id} does not exist.`);
+		next(response);
+	}
+});
+
+/**
+ * GET /workflow-stats/:id/counts/
+ */
+workflowStatsController.get(
+	'/:id/counts/',
+	ResponseHelper.send(async (req: ExecutionRequest.Get): Promise<IWorkflowStatisticsCounts> => {
+		// Get counts from DB
+		const workflowId = req.params.id;
+
+		// Find the stats for this workflow
+		const stats = await Db.collections.WorkflowStatistics.find({
+			select: ['count', 'name'],
+			where: {
+				workflowId,
+			},
+		});
+
+		const data: IWorkflowStatisticsCounts = {
+			productionSuccess: 0,
+			productionError: 0,
+			manualSuccess: 0,
+			manualError: 0,
+		};
+
+		// There will be a maximum of 4 stats (currently)
+		stats.forEach(({ count, name }) => {
+			switch (name) {
+				case StatisticsNames.manualError:
+					data.manualError = count;
+					break;
+
+				case StatisticsNames.manualSuccess:
+					data.manualSuccess = count;
+					break;
+
+				case StatisticsNames.productionError:
+					data.productionError = count;
+					break;
+
+				case StatisticsNames.productionSuccess:
+					data.productionSuccess = count;
+			}
+		});
+
+		return data;
+	}),
+);
+
+/**
+ * GET /workflow-stats/:id/times/
+ */
+workflowStatsController.get(
+	'/:id/times/',
+	ResponseHelper.send(async (req: ExecutionRequest.Get): Promise<IWorkflowStatisticsTimestamps> => {
+		// Get times from DB
+		const workflowId = req.params.id;
+
+		// Find the stats for this workflow
+		const stats = await Db.collections.WorkflowStatistics.find({
+			select: ['latestEvent', 'name'],
+			where: {
+				workflowId,
+			},
+		});
+
+		const data: IWorkflowStatisticsTimestamps = {
+			productionSuccess: null,
+			productionError: null,
+			manualSuccess: null,
+			manualError: null,
+		};
+
+		// There will be a maximum of 4 stats (currently)
+		stats.forEach(({ latestEvent, name }) => {
+			switch (name) {
+				case StatisticsNames.manualError:
+					data.manualError = latestEvent;
+					break;
+
+				case StatisticsNames.manualSuccess:
+					data.manualSuccess = latestEvent;
+					break;
+
+				case StatisticsNames.productionError:
+					data.productionError = latestEvent;
+					break;
+
+				case StatisticsNames.productionSuccess:
+					data.productionSuccess = latestEvent;
+			}
+		});
+
+		return data;
+	}),
+);
+
+/**
+ * GET /workflow-stats/:id/data-loaded/
+ */
+workflowStatsController.get(
+	'/:id/data-loaded/',
+	ResponseHelper.send(async (req: ExecutionRequest.Get): Promise<IWorkflowStatisticsDataLoaded> => {
+		// Get flag
+		const workflowId = req.params.id;
+
+		// Get the corresponding workflow
+		const workflow = await Db.collections.Workflow.findOne(workflowId);
+		// It will be valid if we reach this point, this is just for TS
+		if (!workflow) {
+			return { dataLoaded: false };
+		}
+
+		const data: IWorkflowStatisticsDataLoaded = {
+			dataLoaded: workflow.dataLoaded,
+		};
+
+		return data;
+	}),
+);