fix(workflow): rebuild core workflowVersion rows in the 2-22 backfill (#22961)
## 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.
This commit is contained in:
-2
@@ -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,
|
||||
|
||||
+85
-15
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user