[command fix] Limit migration to workflow versions (#15145)

This migration updates the filter operand values of the workflowVersion
and workflowRuns in order to capitalize them as they should be (the
product works with both deprecated camel case and capitalized). But that
may involve thousands of workflowRuns!
Let's update the command to only update workflowVersion, and update the
cleanWorkflowRuns job to remove workflow runs that are more than 14 days
old.

This way after the command is run, all new workflow runs will have the
new value for the filter operand, and after fourteen days there will be
no trace of the workflow runs with the deprecated filter operand.

packages/twenty-server/src/database/commands/upgrade-version-command/1-8/1-8-migrate-workflow-step-filter-operand-value.ts

@@ -6,7 +6,7 @@ import {
   convertViewFilterOperandToCoreOperand,
   isDefined,
 } from 'twenty-shared/utils';
-import { In, Raw, Repository } from 'typeorm';
+import { Raw, Repository } from 'typeorm';
 
 import {
   ActiveOrSuspendedWorkspacesMigrationCommandRunner,
@@ -14,10 +14,6 @@ import {
 } from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import {
-  WorkflowRunState,
-  WorkflowRunWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
 import { isWorkflowFilterAction } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/guards/is-workflow-filter-action.guard';
 import { isWorkflowFindRecordsAction } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/guards/is-workflow-find-records-action.guard';
@@ -46,7 +42,6 @@ export class MigrateWorkflowStepFilterOperandValueCommand extends ActiveOrSuspen
       `[${index + 1}/${total}] Migrating workflow step filter operand values for workspace ${workspaceId}`,
     );
 
-    // workflowVersions
     const workflowVersionRepository =
       await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowVersionWorkspaceEntity>(
         workspaceId,
@@ -127,84 +122,5 @@ export class MigrateWorkflowStepFilterOperandValueCommand extends ActiveOrSuspen
         );
       }
     }
-
-    //workflowRuns
-    const workflowRunRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowRunWorkspaceEntity>(
-        workspaceId,
-        'workflowRun',
-        { shouldBypassPermissionChecks: true },
-      );
-
-    const workflowRunsToMigrate = await workflowRunRepository.find({
-      where: {
-        workflowVersionId: In(
-          workflowVersionsToMigrate.map(
-            (workflowVersion) => workflowVersion.id,
-          ),
-        ),
-      },
-    });
-
-    this.logger.log(
-      `Found ${workflowRunsToMigrate.length} workflowRuns to migrate`,
-    );
-
-    for (const workflowRun of workflowRunsToMigrate) {
-      let state: WorkflowRunState | null | undefined = workflowRun.state;
-
-      if (!isDefined(state)) {
-        continue;
-      }
-
-      let hasChanged = false;
-
-      for (const step of state.flow.steps) {
-        if (isWorkflowFindRecordsAction(step)) {
-          const filter = step.settings.input.filter;
-
-          for (const recordFilter of filter?.recordFilters ?? []) {
-            if (isString(recordFilter.operand)) {
-              const newOperand = convertViewFilterOperandToCoreOperand(
-                recordFilter.operand,
-              );
-
-              if (newOperand && newOperand !== recordFilter.operand) {
-                recordFilter.operand = newOperand;
-                hasChanged = true;
-              }
-            }
-          }
-        }
-        if (isWorkflowFilterAction(step)) {
-          for (const filter of step.settings.input.stepFilters ?? []) {
-            if (isString(filter.operand)) {
-              const newOperand = convertViewFilterOperandToCoreOperand(
-                filter.operand,
-              );
-
-              if (newOperand && newOperand !== filter.operand) {
-                filter.operand = newOperand;
-                hasChanged = true;
-              }
-            }
-          }
-        }
-      }
-
-      if (hasChanged) {
-        this.logger.log(
-          `${options.dryRun ? 'DRY RUN - Would be' : ''}Updating workflowRun ${workflowRun.id} in workspace ${workspaceId}`,
-        );
-
-        if (!options.dryRun) {
-          await workflowRunRepository.update({ id: workflowRun.id }, { state });
-        }
-
-        this.logger.log(
-          `${options.dryRun ? 'DRY RUN - Would have' : ''}Updated workflowRun ${workflowRun.id} in workspace ${workspaceId}`,
-        );
-      }
-    }
   }
 }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts

@@ -52,14 +52,17 @@ export class WorkflowCleanWorkflowRunsJob {
           WITH ranked_runs AS (
             SELECT id,
                    ROW_NUMBER() OVER (
-           PARTITION BY "workflowId"
-           ORDER BY "createdAt" DESC
-         ) AS rn
+                      PARTITION BY "workflowId"
+                      ORDER BY "createdAt" DESC
+                   ) AS rn,
+                   "createdAt"
             FROM ${schemaName}."workflowRun"
             WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}')
           )
-          SELECT id, rn FROM ranked_runs WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP};
-          `,
+          SELECT id, rn FROM ranked_runs
+          WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP}
+             OR "createdAt" < NOW() - INTERVAL '14 days';
+        `,
       );
 
       const workflowRunRepository =