feat(workflow): repair orphan core workflow versions via upgrade command (#23739)

## Context Part of the workflow + workflowVersion in core migration. Before enabling `IS_WORKFLOW_VERSION_IN_CORE_ENABLED` on real workspaces, legacy drift between the workspace source-of-truth and the core mirror must be repaired. The dangerous category is orphan **ACTIVE** core `workflowVersion` rows: once dispatch reads core (later step), they become phantom triggers. ## The problem Some workspaces carry orphan `core."workflowVersion"` rows: core versions that no workspace version references via `coreWorkflowVersionId`. On one production workspace this was 57 rows, 5 of them ACTIVE `DATABASE_EVENT`. Root cause is pre-2.25 residue: - The v2.22 `backfill-workflow-version-core-links` command minted a core row per active workspace version, copying `status` verbatim. - The workflow delete/destroy cascade did not clean core until #23356 (first in v2.25.0): there was no `deleteCoreVersionsByWorkflowIds` and the deactivate-on-delete status flip was not mirrored. - Workflows deleted then destroyed in that window left their core rows behind, still ACTIVE, with every workspace referrer gone. Current code (>= v2.25) cleans core transactionally on delete/destroy, so this cannot recur. This command clears the historical residue. ## What this does A `@RegisteredWorkspaceCommand('2.28.0')` that, per provisioned workspace: - Deletes `core."workflowVersion"` rows with no workspace referrer, then `invalidateAndRecompute`s the automated trigger map. - `NOT EXISTS` does not filter `deletedAt`, so a soft-deleted (restorable) workspace version still protects its core row. - Scoped to `applicationId = workspaceCustomApplicationId OR NULL`, so future app-owned core versions are never touched. - Supports `--dry-run` and is idempotent (re-run is a no-op). ## Testing Run through the real upgrade harness on a dev instance: - Injected 2 synthetic orphans (1 ACTIVE `DATABASE_EVENT`, 1 ARCHIVED). `--dry-run` reported `Would delete 2 (1 ACTIVE)`; real run reported `Deleted 2 (1 ACTIVE)`. - The 5 legit linked core versions were untouched (referrer guard verified). Orphans remaining: 0. - Re-run logged `No orphan core workflowVersion rows` (idempotent). - typecheck, oxlint, oxfmt all clean.  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/23739?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-08-04 16:55:09 +02:00
parent 1ebcbdda42
commit 0379b537dd
4 changed files with 286 additions and 0 deletions
@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+
+import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module';
+import { RepairOrphanCoreWorkflowVersionsCommand } from 'src/database/commands/upgrade-version-command/2-28/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command';
+import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
+
+@Module({
+  imports: [WorkspaceCacheModule, WorkspaceIteratorModule],
+  providers: [RepairOrphanCoreWorkflowVersionsCommand],
+})
+export class V2_28_UpgradeVersionCommandModule {}
@@ -0,0 +1,123 @@
+import { Command } from 'nest-commander';
+import { isDefined } from 'twenty-shared/utils';
+import { EntityMetadataNotFoundError } from 'typeorm/error/EntityMetadataNotFoundError';
+
+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 { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
+import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
+import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+
+// Delete orphan core.workflowVersion rows left by delete/destroy paths that
+// predated the transactional core cleanup (< 2.25). An orphan is a core row no
+// workspace version references via coreWorkflowVersionId. The NOT EXISTS check
+// does not filter deletedAt, so a soft-deleted (restorable) workspace version
+// still protects its core row, and it is evaluated at delete time so a version
+// created concurrently keeps its core row.
+@RegisteredWorkspaceCommand('2.28.0', 1785600000000)
+@Command({
+  name: 'upgrade:2-28:repair-orphan-core-workflow-versions',
+  description:
+    'Delete orphan core workflowVersion rows with no workspace referrer',
+})
+export class RepairOrphanCoreWorkflowVersionsCommand extends ProvisionedWorkspaceCommandRunner {
+  constructor(
+    protected readonly workspaceIteratorService: WorkspaceIteratorService,
+    private readonly workspaceCacheService: WorkspaceCacheService,
+  ) {
+    super(workspaceIteratorService);
+  }
+
+  override async runOnWorkspace({
+    workspaceId,
+    options,
+    dataSource,
+  }: RunOnWorkspaceArgs): Promise<void> {
+    if (!isDefined(dataSource)) {
+      return;
+    }
+
+    try {
+      // Resolve the workspace workflowVersion object; EntityMetadataNotFoundError
+      // means it was never provisioned, so there is nothing to repair and the
+      // orphan query below would fail to resolve its schema table. Skip cleanly
+      // like the sibling backfill commands rather than aborting the upgrade.
+      await dataSource
+        .getRepository<WorkflowVersionWorkspaceEntity>('workflowVersion', {
+          shouldBypassPermissionChecks: true,
+        })
+        .count();
+    } catch (error) {
+      if (error instanceof EntityMetadataNotFoundError) {
+        return;
+      }
+
+      throw error;
+    }
+
+    const schema = getWorkspaceSchemaName(workspaceId);
+
+    const orphanClause = `
+      FROM core."workflowVersion" c
+      WHERE c."workspaceId" = $1
+        AND NOT EXISTS (
+          SELECT 1
+          FROM "${schema}"."workflowVersion" wf
+          WHERE wf."coreWorkflowVersionId" = c.id
+        )`;
+
+    const queryRunner = dataSource.createQueryRunner();
+
+    await queryRunner.connect();
+
+    try {
+      const [counts] = await queryRunner.query(
+        `SELECT count(*)::int AS total,
+                count(*) FILTER (WHERE c.status = 'ACTIVE')::int AS active
+         ${orphanClause}`,
+        [workspaceId],
+      );
+
+      if (counts.total === 0) {
+        return;
+      }
+
+      if (options.dryRun) {
+        this.logger.log(
+          `[DRY RUN] Would delete ${counts.total} orphan core workflowVersion row(s) (${counts.active} ACTIVE) for workspace ${workspaceId}`,
+        );
+
+        return;
+      }
+
+      await queryRunner.startTransaction();
+
+      try {
+        await queryRunner.query(`DELETE ${orphanClause}`, [workspaceId]);
+
+        await queryRunner.commitTransaction();
+      } catch (error) {
+        await queryRunner.rollbackTransaction();
+
+        throw error;
+      }
+
+      this.logger.log(
+        `Deleted ${counts.total} orphan core workflowVersion row(s) (${counts.active} ACTIVE) for workspace ${workspaceId}`,
+      );
+
+      // The raw delete bypasses the sync path that normally invalidates the
+      // automated trigger map, which is built from ACTIVE core versions, so
+      // recompute it only when an ACTIVE orphan was removed.
+      if (counts.active > 0) {
+        await this.workspaceCacheService.invalidateAndRecompute(workspaceId, [
+          'workflowAutomatedTriggerMaps',
+        ]);
+      }
+    } finally {
+      await queryRunner.release();
+    }
+  }
+}
@@ -0,0 +1,150 @@
+import { type DataSource, type QueryRunner } from 'typeorm';
+import { EntityMetadataNotFoundError } from 'typeorm/error/EntityMetadataNotFoundError';
+
+import { type WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
+import { RepairOrphanCoreWorkflowVersionsCommand } from 'src/database/commands/upgrade-version-command/2-28/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command';
+import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
+
+const WORKSPACE_ID = '20202020-0000-0000-0000-000000000001';
+
+const isDelete = (sql: unknown) =>
+  typeof sql === 'string' && sql.trimStart().startsWith('DELETE');
+
+const setup = ({
+  total = 0,
+  active = 0,
+  objectMissing = false,
+  deleteFails = false,
+}: {
+  total?: number;
+  active?: number;
+  objectMissing?: boolean;
+  deleteFails?: boolean;
+} = {}) => {
+  const query = jest.fn(async (...args: unknown[]) => {
+    const sql = args[0] as string;
+
+    if (isDelete(sql)) {
+      if (deleteFails) throw new Error('boom');
+
+      return [];
+    }
+    if (sql.includes('count(*)')) {
+      return [{ total, active }];
+    }
+
+    return [];
+  });
+
+  const startTransaction = jest.fn();
+  const commitTransaction = jest.fn();
+  const rollbackTransaction = jest.fn();
+  const queryRunner = {
+    connect: jest.fn(),
+    startTransaction,
+    commitTransaction,
+    rollbackTransaction,
+    release: jest.fn(),
+    query,
+  } as unknown as QueryRunner;
+
+  const count = objectMissing
+    ? jest
+        .fn()
+        .mockRejectedValue(new EntityMetadataNotFoundError('workflowVersion'))
+    : jest.fn().mockResolvedValue(0);
+
+  const dataSource = {
+    getRepository: () => ({ count }),
+    createQueryRunner: () => queryRunner,
+  } as unknown as DataSource;
+
+  const recompute = jest.fn();
+  const command = new RepairOrphanCoreWorkflowVersionsCommand(
+    {} as WorkspaceIteratorService,
+    { invalidateAndRecompute: recompute } as unknown as WorkspaceCacheService,
+  );
+
+  const run = (dryRun = false) =>
+    command.runOnWorkspace({
+      workspaceId: WORKSPACE_ID,
+      dataSource: dataSource as never,
+      options: { dryRun },
+      index: 0,
+      total: 1,
+    });
+
+  return { run, query, startTransaction, rollbackTransaction, recompute };
+};
+
+describe('RepairOrphanCoreWorkflowVersionsCommand', () => {
+  beforeEach(() => jest.clearAllMocks());
+
+  it('deletes orphans with an atomic NOT EXISTS and recomputes the trigger map', async () => {
+    const { run, query, recompute } = setup({ total: 2, active: 1 });
+
+    await run();
+
+    const deleteCall = query.mock.calls.find((call) => isDelete(call[0]));
+
+    expect(deleteCall?.[0]).toContain('NOT EXISTS');
+    expect(deleteCall?.[0]).toContain('coreWorkflowVersionId');
+    expect(deleteCall?.[1]).toEqual([WORKSPACE_ID]);
+    expect(recompute).toHaveBeenCalledWith(WORKSPACE_ID, [
+      'workflowAutomatedTriggerMaps',
+    ]);
+  });
+
+  it('does not delete or recompute on a dry run', async () => {
+    const { run, query, startTransaction, recompute } = setup({
+      total: 2,
+      active: 1,
+    });
+
+    await run(true);
+
+    expect(query.mock.calls.some((call) => isDelete(call[0]))).toBe(false);
+    expect(startTransaction).not.toHaveBeenCalled();
+    expect(recompute).not.toHaveBeenCalled();
+  });
+
+  it('is a no-op when there are no orphans', async () => {
+    const { run, query, recompute } = setup({ total: 0 });
+
+    await run();
+
+    expect(query.mock.calls.some((call) => isDelete(call[0]))).toBe(false);
+    expect(recompute).not.toHaveBeenCalled();
+  });
+
+  it('does not recompute when only non-active orphans are removed', async () => {
+    const { run, query, recompute } = setup({ total: 2, active: 0 });
+
+    await run();
+
+    expect(query.mock.calls.some((call) => isDelete(call[0]))).toBe(true);
+    expect(recompute).not.toHaveBeenCalled();
+  });
+
+  it('rolls back and does not recompute when the delete fails', async () => {
+    const { run, rollbackTransaction, recompute } = setup({
+      total: 2,
+      active: 1,
+      deleteFails: true,
+    });
+
+    await expect(run()).rejects.toThrow('boom');
+
+    expect(rollbackTransaction).toHaveBeenCalledTimes(1);
+    expect(recompute).not.toHaveBeenCalled();
+  });
+
+  it('skips a workspace that never provisioned the workflowVersion object', async () => {
+    const { run, query, recompute } = setup({ total: 2, objectMissing: true });
+
+    await run();
+
+    expect(query.mock.calls.some((call) => isDelete(call[0]))).toBe(false);
+    expect(recompute).not.toHaveBeenCalled();
+  });
+});
@@ -27,6 +27,7 @@ import { V2_23_UpgradeVersionCommandModule } from 'src/database/commands/upgrade
 import { V2_25_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-25/2-25-upgrade-version-command.module';
 import { V2_26_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-26/2-26-upgrade-version-command.module';
 import { V2_27_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-27/2-27-upgrade-version-command.module';
+import { V2_28_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-28/2-28-upgrade-version-command.module';

@Module({
  imports: [
@@ -57,6 +58,7 @@ import { V2_27_UpgradeVersionCommandModule } from 'src/database/commands/upgrade
    V2_25_UpgradeVersionCommandModule,
    V2_26_UpgradeVersionCommandModule,
    V2_27_UpgradeVersionCommandModule,
+    V2_28_UpgradeVersionCommandModule,
  ],
 })
 export class WorkspaceCommandProviderModule {}