fix(core): Fixes issue with workflow lastUpdated field

packages/cli/src/Server.ts

@@ -160,6 +160,7 @@ import { setupErrorMiddleware } from '@/ErrorReporting';
 import { getLicense } from '@/License';
 import { licenseController } from './license/license.controller';
 import { corsMiddleware } from './middlewares/cors';
+import { initEvents } from './events';
 
 require('body-parser-xml')(bodyParser);
 
@@ -1690,6 +1691,8 @@ export async function start(): Promise<void> {
 		}
 
 		await app.externalHooks.run('n8n.ready', [app, config]);
+		// Set up event handling
+		initEvents();
 		const cpus = os.cpus();
 		const binaryDataConfig = config.getEnv('binaryDataManager');
 		const diagnosticInfo: IDiagnosticInfo = {

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

@@ -169,15 +169,17 @@ workflowStatsController.get(
 		// 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 };
-		}
+		// Get the flag
+		const stats = await Db.collections.WorkflowStatistics.findOne({
+			select: ['latestEvent'],
+			where: {
+				workflowId,
+				name: StatisticsNames.dataLoaded,
+			},
+		});
 
 		const data: IWorkflowStatisticsDataLoaded = {
-			dataLoaded: workflow.dataLoaded,
+			dataLoaded: stats ? true : false,
 		};
 
 		return data;

packages/cli/src/commands/execute.ts

@@ -22,6 +22,7 @@ import { getLogger } from '@/Logger';
 import config from '@/config';
 import { getInstanceOwner } from '@/UserManagement/UserManagementHelper';
 import { findCliWorkflowStart } from '@/utils';
+import { initEvents } from '@/events';
 
 export class Execute extends Command {
 	static description = '\nExecutes a given workflow';
@@ -48,6 +49,9 @@ export class Execute extends Command {
 		const binaryDataConfig = config.getEnv('binaryDataManager');
 		await BinaryDataManager.init(binaryDataConfig, true);
 
+		// Add event handlers
+		initEvents();
+
 		// eslint-disable-next-line @typescript-eslint/no-shadow
 		const { flags } = this.parse(Execute);
 

packages/cli/src/commands/executeBatch.ts

@@ -35,6 +35,7 @@ import config from '@/config';
 import { User } from '@db/entities/User';
 import { getInstanceOwner } from '@/UserManagement/UserManagementHelper';
 import { findCliWorkflowStart } from '@/utils';
+import { initEvents } from '@/events';
 
 const re = /\d+/;
 
@@ -198,6 +199,9 @@ export class ExecuteBatch extends Command {
 		// eslint-disable-next-line @typescript-eslint/no-shadow
 		const { flags } = this.parse(ExecuteBatch);
 
+		// Add event handlers
+		initEvents();
+
 		ExecuteBatch.debug = flags.debug;
 		ExecuteBatch.concurrency = flags.concurrency || 1;
 

packages/cli/src/databases/entities/WorkflowEntity.ts

@@ -86,9 +86,6 @@ export class WorkflowEntity extends AbstractEntity implements IWorkflowDb {
 	@JoinColumn({ referencedColumnName: 'workflow' })
 	statistics: WorkflowStatistics[];
 
-	@Column({ default: false })
-	dataLoaded: boolean;
-
 	@Column({
 		type: config.getEnv('database.type') === 'sqlite' ? 'text' : 'json',
 		nullable: true,

packages/cli/src/databases/entities/WorkflowStatistics.ts

@@ -8,6 +8,7 @@ export enum StatisticsNames {
 	productionError = 'production_error',
 	manualSuccess = 'manual_success',
 	manualError = 'manual_error',
+	dataLoaded = 'data_loaded',
 }
 
 @Entity()

packages/cli/src/databases/migrations/mysqldb/1671726148420-RemoveWorkflowDataLoadedFlag.ts

@@ -0,0 +1,66 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+import { logMigrationEnd, logMigrationStart } from '@db/utils/migrationHelpers';
+import config from '@/config';
+import { StatisticsNames } from '@/databases/entities/WorkflowStatistics';
+
+export class RemoveWorkflowDataLoadedFlag1671726148420 implements MigrationInterface {
+	name = 'RemoveWorkflowDataLoadedFlag1671726148420';
+
+	async up(queryRunner: QueryRunner) {
+		logMigrationStart(this.name);
+		const tablePrefix = config.getEnv('database.tablePrefix');
+
+		// If any existing workflow has dataLoaded set to true, insert the relevant information to the statistics table
+		const workflowIds: Array<{ id: number; dataLoaded: boolean }> = await queryRunner.query(`
+			SELECT id, dataLoaded
+			FROM ${tablePrefix}workflow_entity
+		`);
+
+		workflowIds.map(({ id, dataLoaded }) => {
+			if (dataLoaded) {
+				const [insertQuery, insertParams] = queryRunner.connection.driver.escapeQueryWithParameters(
+					`
+					INSERT INTO ${tablePrefix}workflow_statistics (workflowId, name, count, latestEvent) VALUES
+					(:id, :name, 1, CURRENT_TIMESTAMP(3))
+					`,
+					{ id, name: StatisticsNames.dataLoaded },
+					{},
+				);
+
+				return queryRunner.query(insertQuery, insertParams);
+			}
+			return undefined;
+		});
+
+		await queryRunner.query(
+			`ALTER TABLE ${tablePrefix}workflow_entity DROP COLUMN dataLoaded`,
+		);
+
+		logMigrationEnd(this.name);
+	}
+
+	async down(queryRunner: QueryRunner) {
+		const tablePrefix = config.getEnv('database.tablePrefix');
+
+		await queryRunner.query(
+			`ALTER TABLE ${tablePrefix}workflow_entity ADD COLUMN dataLoaded BOOLEAN DEFAULT false`,
+		);
+
+		// Search through statistics for any workflows that have the dataLoaded stat
+		const workflowsIds: Array<{ workflowId: string }> = await queryRunner.query(`
+			SELECT workflowId
+			FROM ${tablePrefix}workflow_statistics
+			WHERE name = '${StatisticsNames.dataLoaded}'
+		`);
+		workflowsIds.map(({ workflowId }) => {
+			return queryRunner.query(`
+				UPDATE ${tablePrefix}workflow_entity
+				SET dataLoaded = true
+				WHERE id = '${workflowId}'`);
+		});
+
+		await queryRunner.query(
+			`DELETE FROM ${tablePrefix}workflow_statistics WHERE name = '${StatisticsNames.dataLoaded}'`,
+		);
+	}
+}

packages/cli/src/databases/migrations/mysqldb/index.ts

@@ -26,6 +26,7 @@ import { CreateCredentialUsageTable1665484192213 } from './1665484192213-CreateC
 import { RemoveCredentialUsageTable1665754637026 } from './1665754637026-RemoveCredentialUsageTable';
 import { AddWorkflowVersionIdColumn1669739707125 } from './1669739707125-AddWorkflowVersionIdColumn';
 import { AddTriggerCountColumn1669823906994 } from './1669823906994-AddTriggerCountColumn';
+import { RemoveWorkflowDataLoadedFlag1671726148420 } from './1671726148420-RemoveWorkflowDataLoadedFlag';
 
 export const mysqlMigrations = [
 	InitialMigration1588157391238,
@@ -56,4 +57,5 @@ export const mysqlMigrations = [
 	AddWorkflowVersionIdColumn1669739707125,
 	WorkflowStatistics1664196174002,
 	AddTriggerCountColumn1669823906994,
+	RemoveWorkflowDataLoadedFlag1671726148420,
 ];

packages/cli/src/databases/migrations/postgresdb/1671726148421-RemoveWorkflowDataLoadedFlag.ts

@@ -0,0 +1,64 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+import { getTablePrefix, logMigrationEnd, logMigrationStart } from '@db/utils/migrationHelpers';
+import config from '@/config';
+import { StatisticsNames } from '@/databases/entities/WorkflowStatistics';
+
+export class RemoveWorkflowDataLoadedFlag1671726148421 implements MigrationInterface {
+	name = 'RemoveWorkflowDataLoadedFlag1671726148421';
+
+	async up(queryRunner: QueryRunner) {
+		logMigrationStart(this.name);
+		const tablePrefix = getTablePrefix();
+
+		// If any existing workflow has dataLoaded set to true, insert the relevant information to the statistics table
+		const workflowIds: Array<{ id: number; dataLoaded: boolean }> = await queryRunner.query(`
+			SELECT id, "dataLoaded"
+			FROM ${tablePrefix}workflow_entity
+		`);
+
+		workflowIds.map(({ id, dataLoaded }) => {
+			if (dataLoaded) {
+				const [insertQuery, insertParams] = queryRunner.connection.driver.escapeQueryWithParameters(
+					`
+					INSERT INTO ${tablePrefix}workflow_statistics ("workflowId", name, count, "latestEvent") VALUES
+					(:id, :name, 1, CURRENT_TIMESTAMP(3))
+					`,
+					{ id, name: StatisticsNames.dataLoaded },
+					{},
+				);
+
+				return queryRunner.query(insertQuery, insertParams);
+			}
+			return undefined;
+		});
+
+		await queryRunner.query(`ALTER TABLE ${tablePrefix}workflow_entity DROP COLUMN "dataLoaded"`);
+
+		logMigrationEnd(this.name);
+	}
+
+	async down(queryRunner: QueryRunner) {
+		const tablePrefix = getTablePrefix();
+
+		await queryRunner.query(
+			`ALTER TABLE ${tablePrefix}workflow_entity ADD COLUMN "dataLoaded" BOOLEAN DEFAULT false`,
+		);
+
+		// Search through statistics for any workflows that have the dataLoaded stat
+		const workflowsIds: Array<{ workflowId: string }> = await queryRunner.query(`
+			SELECT "workflowId"
+			FROM ${tablePrefix}workflow_statistics
+			WHERE name = '${StatisticsNames.dataLoaded}'
+		`);
+		workflowsIds.map(({ workflowId }) => {
+			return queryRunner.query(`
+				UPDATE ${tablePrefix}workflow_entity
+				SET "dataLoaded" = true
+				WHERE id = '${workflowId}'`);
+		});
+
+		await queryRunner.query(
+			`DELETE FROM ${tablePrefix}workflow_statistics WHERE name = '${StatisticsNames.dataLoaded}'`,
+		);
+	}
+}

packages/cli/src/databases/migrations/postgresdb/index.ts

@@ -24,6 +24,7 @@ import { CreateCredentialUsageTable1665484192212 } from './1665484192212-CreateC
 import { RemoveCredentialUsageTable1665754637025 } from './1665754637025-RemoveCredentialUsageTable';
 import { AddWorkflowVersionIdColumn1669739707126 } from './1669739707126-AddWorkflowVersionIdColumn';
 import { AddTriggerCountColumn1669823906995 } from './1669823906995-AddTriggerCountColumn';
+import { RemoveWorkflowDataLoadedFlag1671726148421 } from './1671726148421-RemoveWorkflowDataLoadedFlag';
 
 export const postgresMigrations = [
 	InitialMigration1587669153312,
@@ -52,4 +53,5 @@ export const postgresMigrations = [
 	AddWorkflowVersionIdColumn1669739707126,
 	WorkflowStatistics1664196174001,
 	AddTriggerCountColumn1669823906995,
+	RemoveWorkflowDataLoadedFlag1671726148421,
 ];

packages/cli/src/databases/migrations/sqlite/1671726148419-RemoveWorkflowDataLoadedFlag.ts

@@ -0,0 +1,67 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+import { logMigrationEnd, logMigrationStart } from '@db/utils/migrationHelpers';
+import config from '@/config';
+import { v4 as uuidv4 } from 'uuid';
+import { StatisticsNames } from '@/databases/entities/WorkflowStatistics';
+
+export class RemoveWorkflowDataLoadedFlag1671726148419 implements MigrationInterface {
+	name = 'RemoveWorkflowDataLoadedFlag1671726148419';
+
+	async up(queryRunner: QueryRunner) {
+		logMigrationStart(this.name);
+		const tablePrefix = config.getEnv('database.tablePrefix');
+
+		// If any existing workflow has dataLoaded set to true, insert the relevant information to the statistics table
+		const workflowIds: Array<{ id: number; dataLoaded: boolean }> = await queryRunner.query(`
+			SELECT id, dataLoaded
+			FROM "${tablePrefix}workflow_entity"
+		`);
+
+		workflowIds.map(({ id, dataLoaded }) => {
+			if (dataLoaded) {
+				const [insertQuery, insertParams] = queryRunner.connection.driver.escapeQueryWithParameters(
+					`
+					INSERT INTO "${tablePrefix}workflow_statistics" (workflowId, name, count, latestEvent) VALUES
+					(:id, :name, 1, STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW'))
+					`,
+					{ id, name: StatisticsNames.dataLoaded },
+					{},
+				);
+
+				return queryRunner.query(insertQuery, insertParams);
+			}
+			return undefined;
+		});
+
+		await queryRunner.query(
+			`ALTER TABLE \`${tablePrefix}workflow_entity\` DROP COLUMN "dataLoaded"`,
+		);
+
+		logMigrationEnd(this.name);
+	}
+
+	async down(queryRunner: QueryRunner) {
+		const tablePrefix = config.getEnv('database.tablePrefix');
+
+		await queryRunner.query(
+			`ALTER TABLE \`${tablePrefix}workflow_entity\` ADD COLUMN "dataLoaded" BOOLEAN DEFAULT false`,
+		);
+
+		// Search through statistics for any workflows that have the dataLoaded stat
+		const workflowsIds: Array<{ workflowId: string }> = await queryRunner.query(`
+			SELECT workflowId
+			FROM "${tablePrefix}workflow_statistics"
+			WHERE name = '${StatisticsNames.dataLoaded}'
+		`);
+		workflowsIds.map(({ workflowId }) => {
+			return queryRunner.query(`
+				UPDATE "${tablePrefix}workflow_entity"
+				SET dataLoaded = true
+				WHERE id = '${workflowId}'`);
+		});
+
+		await queryRunner.query(
+			`DELETE FROM "${tablePrefix}workflow_statistics" WHERE name = '${StatisticsNames.dataLoaded}'`,
+		);
+	}
+}

packages/cli/src/databases/migrations/sqlite/index.ts

@@ -23,6 +23,7 @@ import { CreateCredentialUsageTable1665484192211 } from './1665484192211-CreateC
 import { RemoveCredentialUsageTable1665754637024 } from './1665754637024-RemoveCredentialUsageTable';
 import { AddWorkflowVersionIdColumn1669739707124 } from './1669739707124-AddWorkflowVersionIdColumn';
 import { AddTriggerCountColumn1669823906993 } from './1669823906993-AddTriggerCountColumn';
+import { RemoveWorkflowDataLoadedFlag1671726148419 } from './1671726148419-RemoveWorkflowDataLoadedFlag';
 
 const sqliteMigrations = [
 	InitialMigration1588102412422,
@@ -50,6 +51,7 @@ const sqliteMigrations = [
 	AddWorkflowVersionIdColumn1669739707124,
 	AddTriggerCountColumn1669823906993,
 	WorkflowStatistics1664196174000,
+	RemoveWorkflowDataLoadedFlag1671726148419,
 ];
 
 export { sqliteMigrations };

packages/cli/src/events/WorkflowStatistics.ts

@@ -56,16 +56,20 @@ export async function workflowExecutionCompleted(
 }
 
 export async function nodeFetchedData(workflowId: string, node: INode): Promise<void> {
-	// Update only if necessary
-	const response = await Db.collections.Workflow.update(
-		{ id: workflowId, dataLoaded: false },
-		{ dataLoaded: true },
-	);
-
-	// If response.affected is 1 then we know this was the first time data was loaded into the workflow; do posthog event here
-	if (!response.affected) return;
+	// Try to insert the data loaded statistic
+	try {
+		await Db.collections.WorkflowStatistics.insert({
+			workflowId,
+			name: StatisticsNames.dataLoaded,
+			count: 1,
+			latestEvent: new Date(),
+		});
+	} catch (error) {
+		// If this fails, it'll be a duplicate key failure. It is safe to return from here
+		return;
+	}
 
-	// Compile the metrics
+	// Compile the metrics since this was a new data loaded event
 	const owner = await getWorkflowOwner(workflowId);
 	let metrics = {
 		user_id: owner.id,

packages/cli/src/events/index.ts

@@ -1,7 +1,12 @@
 import { eventEmitter } from 'n8n-core';
 import { nodeFetchedData, workflowExecutionCompleted } from './WorkflowStatistics';
 
-// Check for undefined as during testing these functions end up undefined for some reason
-if (nodeFetchedData) eventEmitter.on(eventEmitter.types.nodeFetchedData, nodeFetchedData);
-if (workflowExecutionCompleted)
-	eventEmitter.on(eventEmitter.types.workflowExecutionCompleted, workflowExecutionCompleted);
+export function initEvents() {
+	// Check for undefined as during testing these functions end up undefined for some reason
+	if (nodeFetchedData) {
+		eventEmitter.on(eventEmitter.types.nodeFetchedData, nodeFetchedData);
+	}
+	if (workflowExecutionCompleted) {
+		eventEmitter.on(eventEmitter.types.workflowExecutionCompleted, workflowExecutionCompleted);
+	}
+}

packages/cli/src/index.ts

@@ -25,8 +25,6 @@ import * as WebhookServer from './WebhookServer';
 import * as WorkflowExecuteAdditionalData from './WorkflowExecuteAdditionalData';
 import * as WorkflowHelpers from './WorkflowHelpers';
 
-import './events';
-
 export {
 	ActiveExecutions,
 	ActiveWorkflowRunner,