diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-upgrade-version-command.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-upgrade-version-command.module.ts new file mode 100644 index 0000000000..3fb69ad5f5 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-upgrade-version-command.module.ts @@ -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 {} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.ts new file mode 100644 index 0000000000..f8ce17dfa8 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.ts @@ -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 { + 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('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(); + } + } +} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/__tests__/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.spec.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/__tests__/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.spec.ts new file mode 100644 index 0000000000..9b7e0ce9f1 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-28/__tests__/2-28-workspace-command-1785600000000-repair-orphan-core-workflow-versions.command.spec.ts @@ -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(); + }); +}); diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts index 3d742e4da3..cbacabac19 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts @@ -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 {}