From f8b7ecf680eab20d6b7773093ebaf155760b16b5 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Thu, 16 Jul 2026 18:12:50 +0200 Subject: [PATCH] fix(workflow): rebuild core workflowVersion rows in the 2-22 backfill (#22961) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem Syncing workflowVersion to core fails with `duplicate key value violates unique constraint "IDX_WORKFLOW_VERSION_ONE_ACTIVE_PER_WORKFLOW"`. The sync's `INSERT ... ON CONFLICT ("id")` only dedupes the primary key — it can't dedupe `IDX_WORKFLOW_VERSION_ONE_ACTIVE_PER_WORKFLOW` (`(workspaceId, workflowId) WHERE status='ACTIVE'`). When a leftover ACTIVE core row exists for a `(workspaceId, workflowId)` with a stale id (accumulated across the sync's earlier id schemes), inserting the new active row collides, and the per-id purge misses it. ## Fix Rewrite the 2-22 `backfill-workflow-version-core-links` command as a **full rebuild**. Per workspace, in one raw-SQL transaction (core and workspace schemas are the same database): 1. `DELETE` all `core.workflowVersion` for the workspace — clears every stale/leftover row. 2. Insert a fresh own-id core row for **every** workspace version. 3. `UPDATE` `coreWorkflowVersionId` on **every** workspace record to its new core id. Because it wipes first and re-links all records, leftover ACTIVE rows can't collide and no record is left pointing at a deleted core row — so the dual-write's `linked → update` path stays correct afterward. ## Test (local) Fresh reset, and a reproduced dirty state (leftover ACTIVE core row with a stale id + a stale link + an unlinked record): - rebuild runs with no `ONE_ACTIVE` / duplicate-key error, - leftover wiped, stale link replaced, every record re-linked to a fresh own-id row, - 0 dangling/unlinked, no duplicate core rows, exactly one ACTIVE core version per workflow. Typecheck + lint + oxfmt clean. ## Note The 2-20 `backfill-workflow-version-to-core` can still log per-workspace conflicts on already-dirty instances, but they're non-fatal (the iterator continues) and this rebuild corrects the end state. The dual-write (`upsertToCore`) is unchanged and works on the clean data this produces. --- .../2-22-upgrade-version-command.module.ts | 2 - ...ill-workflow-version-core-links.command.ts | 100 +++++++++++++++--- 2 files changed, 85 insertions(+), 17 deletions(-) diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-upgrade-version-command.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-upgrade-version-command.module.ts index ef300ef6d1..8f96cf7c35 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-upgrade-version-command.module.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-upgrade-version-command.module.ts @@ -8,7 +8,6 @@ import { BackfillWorkflowVersionCoreLinksCommand } from 'src/database/commands/u import { ApplicationModule } from 'src/engine/core-modules/application/application.module'; import { FilesFieldModule } from 'src/engine/core-modules/file/files-field/files-field.module'; import { SecureHttpClientModule } from 'src/engine/core-modules/secure-http-client/secure-http-client.module'; -import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module'; import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module'; import { WorkspaceMigrationModule } from 'src/engine/workspace-manager/workspace-migration/workspace-migration.module'; @@ -17,7 +16,6 @@ import { WorkspaceMigrationModule } from 'src/engine/workspace-manager/workspace ApplicationModule, FilesFieldModule, SecureHttpClientModule, - WorkflowVersionCoreModule, WorkspaceCacheModule, WorkspaceMigrationModule, WorkspaceIteratorModule, diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-workspace-command-1784193207000-backfill-workflow-version-core-links.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-workspace-command-1784193207000-backfill-workflow-version-core-links.command.ts index f1091ae2a6..8f497f1798 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-workspace-command-1784193207000-backfill-workflow-version-core-links.command.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-22/2-22-workspace-command-1784193207000-backfill-workflow-version-core-links.command.ts @@ -1,28 +1,31 @@ import { Command } from 'nest-commander'; import { isDefined } from 'twenty-shared/utils'; import { EntityMetadataNotFoundError } from 'typeorm/error/EntityMetadataNotFoundError'; +import { v4 as uuidv4 } from 'uuid'; import { ProvisionedWorkspaceCommandRunner } from 'src/database/commands/command-runners/provisioned-workspace.command-runner'; import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service'; import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner'; import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator'; -import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service'; +import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity'; -// One-time migration to the soft-ref model after coreWorkflowVersionId is -// provisioned. Re-runs the sync: upsertToCore purges each legacy shared-id -// core row (id === record id) and rebuilds a deterministic own-id row, then -// links the workspace record. +// Full rebuild of the core workflowVersion rows for a workspace once +// coreWorkflowVersionId is provisioned. Per workspace, in one transaction: +// wipe every core row (clearing all pre-soft-ref / stale-id leftovers), insert +// a fresh own-id row for every workspace version, and write that id back onto +// the workspace record. Because it re-links every record, no version is left +// pointing at a deleted core row, so the dual-write's update path stays correct. @RegisteredWorkspaceCommand('2.22.0', 1784193207000) @Command({ name: 'upgrade:2-22:backfill-workflow-version-core-links', description: - 'Re-run the workflowVersion core sync so workspaces newly provisioned with coreWorkflowVersionId get linked', + 'Rebuild core workflowVersion rows for each workspace and link every workspace record via coreWorkflowVersionId', }) export class BackfillWorkflowVersionCoreLinksCommand extends ProvisionedWorkspaceCommandRunner { constructor( protected readonly workspaceIteratorService: WorkspaceIteratorService, - private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService, + private readonly workspaceCacheService: WorkspaceCacheService, ) { super(workspaceIteratorService); } @@ -64,19 +67,86 @@ export class BackfillWorkflowVersionCoreLinksCommand extends ProvisionedWorkspac if (options.dryRun === true) { this.logger.log( - `[DRY RUN] Would re-sync ${workspaceWorkflowVersions.length} workflowVersion row(s) for workspace ${workspaceId}`, + `[DRY RUN] Would rebuild ${workspaceWorkflowVersions.length} core workflowVersion row(s) for workspace ${workspaceId}`, ); return; } - await this.workflowVersionCoreSyncService.upsertToCore( - workspaceId, - workspaceWorkflowVersions, - ); + const queryRunner = dataSource.createQueryRunner(); - this.logger.log( - `Linked ${workspaceWorkflowVersions.length} workflowVersion row(s) for workspace ${workspaceId}`, - ); + await queryRunner.connect(); + + try { + const [workspace] = await queryRunner.query( + `SELECT "databaseSchema", "workspaceCustomApplicationId" FROM core."workspace" WHERE id = $1`, + [workspaceId], + ); + + if (!isDefined(workspace?.workspaceCustomApplicationId)) { + throw new Error( + `Workspace custom application not found for workspace ${workspaceId}`, + ); + } + + const schema: string = workspace.databaseSchema; + const applicationId: string = workspace.workspaceCustomApplicationId; + + await queryRunner.startTransaction(); + + try { + await queryRunner.query( + `DELETE FROM core."workflowVersion" WHERE "workspaceId" = $1`, + [workspaceId], + ); + + for (const workflowVersion of workspaceWorkflowVersions) { + const coreWorkflowVersionId = uuidv4(); + + await queryRunner.query( + `INSERT INTO core."workflowVersion" + (id, "workspaceId", "universalIdentifier", "applicationId", triggers, steps, status, "workflowId") + VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7, $8)`, + [ + coreWorkflowVersionId, + workspaceId, + uuidv4(), + applicationId, + isDefined(workflowVersion.trigger) + ? JSON.stringify([workflowVersion.trigger]) + : null, + isDefined(workflowVersion.steps) + ? JSON.stringify(workflowVersion.steps) + : null, + workflowVersion.status, + workflowVersion.workflowId, + ], + ); + + await queryRunner.query( + `UPDATE "${schema}"."workflowVersion" SET "coreWorkflowVersionId" = $1 WHERE id = $2`, + [coreWorkflowVersionId, workflowVersion.id], + ); + } + + await queryRunner.commitTransaction(); + } catch (error) { + await queryRunner.rollbackTransaction(); + + throw error; + } + + // The old sync path invalidated workflowAutomatedTriggerMaps; the raw-SQL + // rebuild bypasses it, so recompute it from the fresh core rows. + await this.workspaceCacheService.invalidateAndRecompute(workspaceId, [ + 'workflowAutomatedTriggerMaps', + ]); + + this.logger.log( + `Rebuilt ${workspaceWorkflowVersions.length} core workflowVersion row(s) for workspace ${workspaceId}`, + ); + } finally { + await queryRunner.release(); + } } }