fix(core): Disallow orphan executions

packages/cli/src/databases/migrations/mysqldb/1693554410387-DisallowOrphanExecutions.ts

@@ -0,0 +1,26 @@
+import type { MigrationContext, ReversibleMigration } from '@db/types';
+
+export class DisallowOrphanExecutions1693554410387 implements ReversibleMigration {
+	/**
+	 * Ensure all executions point to a workflow.
+	 */
+	async up({ queryRunner, escape }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+		const workflowId = escape.columnName('workflowId');
+
+		await queryRunner.query(`DELETE FROM ${executionEntity} WHERE ${workflowId} IS NULL;`);
+		await queryRunner.query(
+			`ALTER TABLE ${executionEntity} MODIFY ${workflowId} VARCHAR(36) NOT NULL;`,
+		);
+	}
+
+	/**
+	 * Reversal excludes restoring deleted rows.
+	 */
+	async down({ queryRunner, escape }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+		const workflowId = escape.columnName('workflowId');
+
+		await queryRunner.query(`ALTER TABLE ${executionEntity} MODIFY ${workflowId} VARCHAR(36);`);
+	}
+}

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

@@ -46,6 +46,7 @@ import { RemoveResetPasswordColumns1690000000030 } from '../common/1690000000030
 import { CreateWorkflowNameIndex1691088862123 } from '../common/1691088862123-CreateWorkflowNameIndex';
 import { AddMfaColumns1690000000030 } from './../common/1690000000040-AddMfaColumns';
 import { CreateWorkflowHistoryTable1692967111175 } from '../common/1692967111175-CreateWorkflowHistoryTable';
+import { DisallowOrphanExecutions1693554410387 } from './1693554410387-DisallowOrphanExecutions';
 
 export const mysqlMigrations: Migration[] = [
 	InitialMigration1588157391238,
@@ -95,4 +96,5 @@ export const mysqlMigrations: Migration[] = [
 	CreateWorkflowNameIndex1691088862123,
 	AddMfaColumns1690000000030,
 	CreateWorkflowHistoryTable1692967111175,
+	DisallowOrphanExecutions1693554410387,
 ];

packages/cli/src/databases/migrations/postgresdb/1693554410387-DisallowOrphanExecutions.ts

@@ -0,0 +1,28 @@
+import type { MigrationContext, ReversibleMigration } from '@db/types';
+
+export class DisallowOrphanExecutions1693554410387 implements ReversibleMigration {
+	/**
+	 * Ensure all executions point to a workflow.
+	 */
+	async up({ queryRunner, escape }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+		const workflowId = escape.columnName('workflowId');
+
+		await queryRunner.query(`DELETE FROM ${executionEntity} WHERE ${workflowId} IS NULL;`);
+		await queryRunner.query(
+			`ALTER TABLE ${executionEntity} ALTER COLUMN ${workflowId} SET NOT NULL;`,
+		);
+	}
+
+	/**
+	 * Reversal excludes restoring deleted rows.
+	 */
+	async down({ queryRunner, escape }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+		const workflowId = escape.columnName('workflowId');
+
+		await queryRunner.query(
+			`ALTER TABLE ${executionEntity} ALTER COLUMN ${workflowId} DROP NOT NULL;`,
+		);
+	}
+}

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

@@ -44,6 +44,7 @@ import { AddMissingPrimaryKeyOnExecutionData1690787606731 } from './169078760673
 import { CreateWorkflowNameIndex1691088862123 } from '../common/1691088862123-CreateWorkflowNameIndex';
 import { AddMfaColumns1690000000030 } from './../common/1690000000040-AddMfaColumns';
 import { CreateWorkflowHistoryTable1692967111175 } from '../common/1692967111175-CreateWorkflowHistoryTable';
+import { DisallowOrphanExecutions1693554410387 } from './1693554410387-DisallowOrphanExecutions';
 
 export const postgresMigrations: Migration[] = [
 	InitialMigration1587669153312,
@@ -91,4 +92,5 @@ export const postgresMigrations: Migration[] = [
 	CreateWorkflowNameIndex1691088862123,
 	AddMfaColumns1690000000030,
 	CreateWorkflowHistoryTable1692967111175,
+	DisallowOrphanExecutions1693554410387,
 ];

packages/cli/src/databases/migrations/sqlite/1693554410387-DisallowOrphanExecutions.ts

@@ -0,0 +1,77 @@
+import type { MigrationContext, ReversibleMigration } from '@db/types';
+
+export class DisallowOrphanExecutions1693554410387 implements ReversibleMigration {
+	transaction = false as const;
+
+	/**
+	 * Ensure all executions point to a workflow. Recreate table because sqlite
+	 * does not support modifying column.
+	 */
+	async up({ escape, schemaBuilder: { createTable, column }, runQuery }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+		const workflowId = escape.columnName('workflowId');
+
+		await runQuery(`DELETE FROM ${executionEntity} WHERE ${workflowId} IS NULL;`);
+
+		await createTable('temp')
+			.withColumns(
+				column('id').int.primary.autoGenerate.notNull,
+				column('workflowId').varchar(36).notNull, // only change
+				column('finished').bool.notNull,
+				column('mode').varchar().notNull,
+				column('retryOf').varchar(),
+				column('retrySuccessId').varchar(),
+				column('startedAt').timestamp().notNull,
+				column('stoppedAt').timestamp(),
+				column('waitTill').timestamp(),
+				column('status').varchar(),
+			)
+			.withForeignKey('workflowId', {
+				tableName: 'workflow_entity',
+				columnName: 'id',
+				onDelete: 'CASCADE',
+			});
+
+		await runQuery(
+			`INSERT INTO "temp" ("id", "workflowId", "finished", "mode", "retryOf", "retrySuccessId", "startedAt", "stoppedAt", "waitTill", "status") SELECT "id", "workflowId", "finished", "mode", "retryOf", "retrySuccessId", "startedAt", "stoppedAt", "waitTill", "status" FROM ${executionEntity};`,
+		);
+
+		await runQuery(`DROP TABLE ${executionEntity};`);
+
+		await runQuery(`ALTER TABLE "temp" RENAME TO ${executionEntity};`);
+	}
+
+	/**
+	 * Reversal excludes restoring deleted rows.
+	 */
+	async down({ escape, schemaBuilder: { createTable, column }, runQuery }: MigrationContext) {
+		const executionEntity = escape.tableName('execution_entity');
+
+		await createTable('temp')
+			.withColumns(
+				column('id').int.primary.autoGenerate.notNull,
+				column('workflowId').varchar(36), // only change
+				column('finished').bool.notNull,
+				column('mode').varchar().notNull,
+				column('retryOf').varchar(),
+				column('retrySuccessId').varchar(),
+				column('startedAt').timestamp().notNull,
+				column('stoppedAt').timestamp(),
+				column('waitTill').timestamp(),
+				column('status').varchar(),
+			)
+			.withForeignKey('workflowId', {
+				tableName: 'workflow_entity',
+				columnName: 'id',
+				onDelete: 'CASCADE',
+			});
+
+		await runQuery(
+			`INSERT INTO "temp" ("id", "workflowId", "finished", "mode", "retryOf", "retrySuccessId", "startedAt", "stoppedAt", "waitTill", "status") SELECT "id", "workflowId", "finished", "mode", "retryOf", "retrySuccessId", "startedAt", "stoppedAt", "waitTill", "status" FROM ${executionEntity};`,
+		);
+
+		await runQuery(`DROP TABLE ${executionEntity};`);
+
+		await runQuery(`ALTER TABLE "temp" RENAME TO ${executionEntity};`);
+	}
+}

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

@@ -43,6 +43,7 @@ import { RemoveResetPasswordColumns1690000000030 } from './1690000000030-RemoveR
 import { CreateWorkflowNameIndex1691088862123 } from '../common/1691088862123-CreateWorkflowNameIndex';
 import { AddMfaColumns1690000000030 } from './1690000000040-AddMfaColumns';
 import { CreateWorkflowHistoryTable1692967111175 } from '../common/1692967111175-CreateWorkflowHistoryTable';
+import { DisallowOrphanExecutions1693554410387 } from './1693554410387-DisallowOrphanExecutions';
 
 const sqliteMigrations: Migration[] = [
 	InitialMigration1588102412422,
@@ -89,6 +90,7 @@ const sqliteMigrations: Migration[] = [
 	CreateWorkflowNameIndex1691088862123,
 	AddMfaColumns1690000000030,
 	CreateWorkflowHistoryTable1692967111175,
+	DisallowOrphanExecutions1693554410387,
 ];
 
 export { sqliteMigrations };