From 3dd70aad03dd97abc820e72e758fe0504c4f1039 Mon Sep 17 00:00:00 2001 From: Marie <51697796+ijreilly@users.noreply.github.com> Date: Thu, 16 Oct 2025 18:03:22 +0200 Subject: [PATCH] [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. --- ...rate-workflow-step-filter-operand-value.ts | 86 +------------------ .../workflow-clean-workflow-runs.cron.job.ts | 13 +-- 2 files changed, 9 insertions(+), 90 deletions(-) diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/1-8/1-8-migrate-workflow-step-filter-operand-value.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/1-8/1-8-migrate-workflow-step-filter-operand-value.ts index 35299258b2e..cdede0d9afa 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/1-8/1-8-migrate-workflow-step-filter-operand-value.ts +++ b/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( workspaceId, @@ -127,84 +122,5 @@ export class MigrateWorkflowStepFilterOperandValueCommand extends ActiveOrSuspen ); } } - - //workflowRuns - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - 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}`, - ); - } - } } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts index bee075dbd8b..75ea90ba346 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts +++ b/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 =