diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts index e9a1335b01..0cc931af78 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts @@ -113,6 +113,15 @@ export class UpgradeCommand extends CommandRunner { }); } + if ( + isDefined(options.workspaceId) && + isDefined(options.startFromWorkspaceId) + ) { + throw new Error( + 'Cannot use --start-from-workspace-id together with -w/--workspace-id', + ); + } + try { const sequence = this.upgradeSequenceReaderService.getUpgradeSequence(); diff --git a/packages/twenty-server/src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service.ts b/packages/twenty-server/src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service.ts index 4f3f099a82..0fd848f6d5 100644 --- a/packages/twenty-server/src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service.ts +++ b/packages/twenty-server/src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service.ts @@ -69,12 +69,14 @@ export class UpgradeSequenceRunnerService { if (step.kind === 'fast-instance' || step.kind === 'slow-instance') { if ( - isDefined(options.workspaceIds) && - options.workspaceIds.length > 0 + (isDefined(options.workspaceIds) && + options.workspaceIds.length > 0) || + isDefined(options.startFromWorkspaceId) || + isDefined(options.workspaceCountLimit) ) { this.logger.log( `Stopping before instance step "${step.name}": ` + - 'upgrade was run with workspace filter (-w). ' + + 'upgrade was run with a workspace filter (-w, --start-from-workspace-id, or --workspace-count-limit). ' + 'Instance commands require all workspaces to be aligned.', ); @@ -303,15 +305,13 @@ export class UpgradeSequenceRunnerService { allActiveOrSuspendedWorkspaceIds: string[]; options: ParsedUpgradeCommandOptions; }): Promise { - const workspaceIds = - isDefined(options.workspaceIds) && options.workspaceIds.length > 0 - ? options.workspaceIds - : allActiveOrSuspendedWorkspaceIds; + const workspaceIds = this.deriveWorkspaceIdsToProcess({ + allActiveOrSuspendedWorkspaceIds, + options, + }); return this.workspaceIteratorService.iterate({ workspaceIds, - startFromWorkspaceId: options.startFromWorkspaceId, - workspaceCountLimit: options.workspaceCountLimit, dryRun: options.dryRun, callback: async (context) => { const workspaceCursor = workspaceCursors.get(context.workspaceId); @@ -337,6 +337,32 @@ export class UpgradeSequenceRunnerService { }); } + private deriveWorkspaceIdsToProcess({ + allActiveOrSuspendedWorkspaceIds, + options, + }: { + allActiveOrSuspendedWorkspaceIds: string[]; + options: ParsedUpgradeCommandOptions; + }): string[] { + if (isDefined(options.workspaceIds) && options.workspaceIds.length > 0) { + return options.workspaceIds; + } + + let workspaceIds = allActiveOrSuspendedWorkspaceIds; + + if (isDefined(options.startFromWorkspaceId)) { + workspaceIds = workspaceIds.filter( + (id) => id >= options.startFromWorkspaceId!, + ); + } + + if (isDefined(options.workspaceCountLimit)) { + workspaceIds = workspaceIds.slice(0, options.workspaceCountLimit); + } + + return workspaceIds; + } + private enforceWorkspacesCompletedPreviousWorkspaceSegment({ sequence, previousWorkspaceStep, diff --git a/packages/twenty-server/test/integration/upgrade/suites/sequence-runner/start-from-workspace-id.integration-spec.ts b/packages/twenty-server/test/integration/upgrade/suites/sequence-runner/start-from-workspace-id.integration-spec.ts new file mode 100644 index 0000000000..6aa15c61a3 --- /dev/null +++ b/packages/twenty-server/test/integration/upgrade/suites/sequence-runner/start-from-workspace-id.integration-spec.ts @@ -0,0 +1,191 @@ +import { + type IntegrationTestContext, + createUpgradeSequenceRunnerIntegrationTestModule, + DEFAULT_OPTIONS, + makeFastInstance, + makeWorkspace, + migrationRecordToKey, + resetSeedSequenceCounter, + seedInstanceMigration, + setMockActiveWorkspaceIds, + testGetExecutedMigrationsInOrder, + WS_1, + WS_2, + WS_3, + WS_4, +} from 'test/integration/upgrade/utils/upgrade-sequence-runner-integration-test.util'; + +// Sorted ASC: WS_1 < WS_2 < WS_3 < WS_4 + +describe('UpgradeSequenceRunnerService — startFromWorkspaceId (integration)', () => { + let context: IntegrationTestContext; + + beforeAll(async () => { + context = await createUpgradeSequenceRunnerIntegrationTestModule(); + }, 30000); + + afterAll(async () => { + await context.dataSource.query('DELETE FROM core."upgradeMigration"'); + await context.module?.close(); + await context.dataSource?.destroy(); + }, 15000); + + beforeEach(async () => { + await context.dataSource.query('DELETE FROM core."upgradeMigration"'); + resetSeedSequenceCounter(); + setMockActiveWorkspaceIds([]); + jest.restoreAllMocks(); + }); + + it('should only process workspaces whose id >= startFromWorkspaceId', async () => { + const sequence = [ + makeFastInstance('Ic1'), + makeWorkspace('Wc1'), + makeWorkspace('Wc2'), + makeFastInstance('Ic2'), + ]; + + setMockActiveWorkspaceIds([WS_1, WS_2, WS_3]); + + await seedInstanceMigration(context.dataSource, { + name: 'Ic1', + status: 'completed', + workspaceIds: [WS_1, WS_2, WS_3], + }); + + const report = await context.runner.run({ + sequence, + options: { + ...DEFAULT_OPTIONS, + startFromWorkspaceId: WS_2, + }, + }); + + expect(report.totalFailures).toBe(0); + expect(report.totalSuccesses).toBe(2); + + const executed = await testGetExecutedMigrationsInOrder(context.dataSource); + + expect(executed.map(migrationRecordToKey)).toStrictEqual([ + // Seeds + 'Ic1:instance:completed:1', + `Ic1:${WS_1}:completed:1`, + `Ic1:${WS_2}:completed:1`, + `Ic1:${WS_3}:completed:1`, + + // Only WS_2 and WS_3 are processed (WS_1 skipped) + `Wc1:${WS_2}:completed:1`, + `Wc2:${WS_2}:completed:1`, + `Wc1:${WS_3}:completed:1`, + `Wc2:${WS_3}:completed:1`, + ]); + }); + + it('should respect workspaceCountLimit together with startFromWorkspaceId', async () => { + const sequence = [makeFastInstance('Ic1'), makeWorkspace('Wc1')]; + + setMockActiveWorkspaceIds([WS_1, WS_2, WS_3, WS_4]); + + await seedInstanceMigration(context.dataSource, { + name: 'Ic1', + status: 'completed', + workspaceIds: [WS_1, WS_2, WS_3, WS_4], + }); + + const report = await context.runner.run({ + sequence, + options: { + ...DEFAULT_OPTIONS, + startFromWorkspaceId: WS_2, + workspaceCountLimit: 1, + }, + }); + + expect(report.totalFailures).toBe(0); + expect(report.totalSuccesses).toBe(1); + + const executed = await testGetExecutedMigrationsInOrder(context.dataSource); + + expect(executed.map(migrationRecordToKey)).toStrictEqual([ + // Seeds + 'Ic1:instance:completed:1', + `Ic1:${WS_1}:completed:1`, + `Ic1:${WS_2}:completed:1`, + `Ic1:${WS_3}:completed:1`, + `Ic1:${WS_4}:completed:1`, + + // Only WS_2 processed (first after startFrom, limited to 1) + `Wc1:${WS_2}:completed:1`, + ]); + }); + + it('should stop before instance step when startFromWorkspaceId is set', async () => { + const sequence = [ + makeFastInstance('Ic1'), + makeWorkspace('Wc1'), + makeWorkspace('Wc2'), + makeFastInstance('Ic2'), + makeWorkspace('Wc3'), + ]; + + setMockActiveWorkspaceIds([WS_1, WS_2, WS_3]); + + await seedInstanceMigration(context.dataSource, { + name: 'Ic1', + status: 'completed', + workspaceIds: [WS_1, WS_2, WS_3], + }); + + const report = await context.runner.run({ + sequence, + options: { + ...DEFAULT_OPTIONS, + startFromWorkspaceId: WS_2, + }, + }); + + expect(report.totalFailures).toBe(0); + expect(report.totalSuccesses).toBe(2); + + const executed = await testGetExecutedMigrationsInOrder(context.dataSource); + + expect(executed.map(migrationRecordToKey)).toStrictEqual([ + // Seeds + 'Ic1:instance:completed:1', + `Ic1:${WS_1}:completed:1`, + `Ic1:${WS_2}:completed:1`, + `Ic1:${WS_3}:completed:1`, + + // Workspace segment: only WS_2 and WS_3 run + `Wc1:${WS_2}:completed:1`, + `Wc2:${WS_2}:completed:1`, + `Wc1:${WS_3}:completed:1`, + `Wc2:${WS_3}:completed:1`, + + // Ic2 and Wc3 never reached — runner stopped at instance step boundary + ]); + }); + + it('should process no workspaces when startFromWorkspaceId is greater than all ids', async () => { + const sequence = [makeFastInstance('Ic1'), makeWorkspace('Wc1')]; + + setMockActiveWorkspaceIds([WS_1, WS_2]); + + await seedInstanceMigration(context.dataSource, { + name: 'Ic1', + status: 'completed', + workspaceIds: [WS_1, WS_2], + }); + + const report = await context.runner.run({ + sequence, + options: { + ...DEFAULT_OPTIONS, + startFromWorkspaceId: 'ffffffff-ffff-ffff-ffff-ffffffffffff', + }, + }); + + expect(report.totalFailures).toBe(0); + expect(report.totalSuccesses).toBe(0); + }); +});