Cleaning UpgradeCommand code flow (#19241)

# Introduction Currently preparing the `UpgradeCommand` refactor, in this way started by cleaning up the existing avoiding unecessary dependencies to others services allowing easier readability and concern centralization for upcoming refactor The UpgradeCommand was extending up to five classes, overriding abstracted class and so on. It was also cascade injecting 3 services Introducing the `WorkspaceIteratorService` that centralize the commands set to run over a single workspace logic shared between both atomic upgrade command call and global upgradeCommand ## Tradeoff Duplicated `@Option` between both `UpgradeCommandRunner` and `WorkspaceMigrationRunner` ## Before ``` UpgradeCommand └─ extends UpgradeCommandRunner └─ extends ActiveOrSuspendedWorkspacesMigrationCommandRunner └─ extends WorkspacesMigrationCommandRunner (owns workspace iteration loop + ORM deps) └─ extends MigrationCommandRunner (dry-run, verbose, error handling) └─ extends CommandRunner (nest-commander) ``` ## Now ``` UpgradeCommand └─ extends UpgradeCommandRunner └─ extends CommandRunner (nest-commander) uses ─► Services (via composition) ``` ## Logging management At the moment all services are logging, in the best of the world only the runners should be doing so
2026-04-02 15:24:55 +02:00
parent 6ae5900ac9
commit cd23a2bc80
43 changed files with 816 additions and 814 deletions
@@ -11,14 +11,13 @@ import {
  UpgradeCommandRunner,
  type AllCommands,
 } from 'src/database/commands/command-runners/upgrade.command-runner';
+import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
 import { CoreMigrationRunnerService } from 'src/database/commands/core-migration-runner/services/core-migration-runner.service';
 import { UPGRADE_COMMAND_SUPPORTED_VERSIONS } from 'src/engine/constants/upgrade-command-supported-versions.constant';
 import { CoreEngineVersionService } from 'src/engine/core-engine-version/services/core-engine-version.service';
 import { type ConfigVariables } from 'src/engine/core-modules/twenty-config/config-variables';
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
-import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
-import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';

 const CURRENT_VERSION =
@@ -66,41 +65,31 @@ const buildUpgradeCommandModule = async ({
  appVersion,
  commandRunner,
 }: BuildUpgradeCommandModuleArgs) => {
-  const mockDataSourceService = {
-    getLastDataSourceMetadataFromWorkspaceId: jest.fn(),
-  };
-
  const module: TestingModule = await Test.createTestingModule({
    providers: [
      {
        provide: commandRunner,
        useFactory: (
          workspaceRepository: Repository<WorkspaceEntity>,
-          twentyConfigService: TwentyConfigService,
-          globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
-          dataSourceService: DataSourceService,
          coreEngineVersionService: CoreEngineVersionService,
          workspaceVersionService: WorkspaceVersionService,
          coreMigrationRunnerService: CoreMigrationRunnerService,
+          workspaceIteratorService: WorkspaceIteratorService,
        ) => {
          return new commandRunner(
            workspaceRepository,
-            twentyConfigService,
-            globalWorkspaceOrmManager,
-            dataSourceService,
            coreEngineVersionService,
            workspaceVersionService,
            coreMigrationRunnerService,
+            workspaceIteratorService,
          );
        },
        inject: [
          getRepositoryToken(WorkspaceEntity),
-          TwentyConfigService,
-          GlobalWorkspaceOrmManager,
-          DataSourceService,
          CoreEngineVersionService,
          WorkspaceVersionService,
          CoreMigrationRunnerService,
+          WorkspaceIteratorService,
        ],
      },
      {
@@ -131,27 +120,43 @@ const buildUpgradeCommandModule = async ({
          }),
        },
      },
-      {
-        provide: GlobalWorkspaceOrmManager,
-        useValue: {
-          connect: jest.fn(),
-          destroyDataSourceForWorkspace: jest.fn(),
-          getDataSourceForWorkspace: jest.fn(),
-          executeInWorkspaceContext: jest
-            .fn()
-            .mockImplementation((fn: () => any, _authContext?: any) => fn()),
-        },
-      },
-      {
-        provide: DataSourceService,
-        useValue: mockDataSourceService,
-      },
      CoreEngineVersionService,
      WorkspaceVersionService,
      {
        provide: CoreMigrationRunnerService,
        useValue: { run: jest.fn().mockResolvedValue(undefined) },
      },
+      {
+        provide: WorkspaceIteratorService,
+        useValue: {
+          iterate: jest.fn().mockImplementation(async (args: any) => {
+            const { callback, ...options } = args;
+            const workspaceIds =
+              options.workspaceIds ??
+              workspaces.map((workspace) => workspace.id);
+
+            const report = {
+              fail: [] as any[],
+              success: [] as any[],
+            };
+
+            for (const [index, workspaceId] of workspaceIds.entries()) {
+              try {
+                await callback({
+                  workspaceId,
+                  index,
+                  total: workspaceIds.length,
+                });
+                report.success.push({ workspaceId });
+              } catch (error) {
+                report.fail.push({ error, workspaceId });
+              }
+            }
+
+            return report;
+          }),
+        },
+      },
    ],
  }).compile();

@@ -161,7 +166,6 @@ const buildUpgradeCommandModule = async ({
 describe('UpgradeCommandRunner', () => {
  let upgradeCommandRunner: BasicUpgradeCommandRunner;
  let workspaceRepository: Repository<WorkspaceEntity>;
-  let coreMigrationRunnerService: CoreMigrationRunnerService;

  type BuildModuleAndSetupSpiesArgs = {
    numberOfWorkspace?: number;
@@ -197,10 +201,6 @@ describe('UpgradeCommandRunner', () => {
    jest.spyOn(upgradeCommandRunner['logger'], 'error').mockImplementation();
    jest.spyOn(upgradeCommandRunner['logger'], 'warn').mockImplementation();

-    jest.spyOn(upgradeCommandRunner, 'runOnWorkspace');
-
-    coreMigrationRunnerService = module.get(CoreMigrationRunnerService);
-
    workspaceRepository = module.get<Repository<WorkspaceEntity>>(
      getRepositoryToken(WorkspaceEntity),
    );
@@ -223,16 +223,6 @@ describe('UpgradeCommandRunner', () => {
    // @ts-expect-error legacy noImplicitAny
    await upgradeCommandRunner.run(passedParams, options);

-    const { fail: failReport, success: successReport } =
-      upgradeCommandRunner.migrationReport;
-
-    expect(successReport.length).toBe(1);
-    expect(failReport.length).toBe(0);
-
-    [upgradeCommandRunner.runOnWorkspace].forEach((fn) =>
-      expect(fn).toHaveBeenCalledTimes(1),
-    );
-
    [workspaceRepository.update].forEach((fn) =>
      expect(fn).not.toHaveBeenCalled(),
    );
@@ -251,16 +241,12 @@ describe('UpgradeCommandRunner', () => {
    // @ts-expect-error legacy noImplicitAny
    await upgradeCommandRunner.run(passedParams, options);

-    [upgradeCommandRunner.runOnWorkspace].forEach((fn) =>
-      expect(fn).toHaveBeenCalledTimes(numberOfWorkspace),
-    );
    expect(workspaceRepository.update).toHaveBeenNthCalledWith(
      numberOfWorkspace,
      { id: expect.any(String) },
      { version: CURRENT_VERSION },
    );
-    expect(upgradeCommandRunner.migrationReport.success.length).toBe(42);
-    expect(upgradeCommandRunner.migrationReport.fail.length).toBe(0);
+    expect(workspaceRepository.update).toHaveBeenCalledTimes(numberOfWorkspace);
  });

  describe('Workspace upgrade should succeed ', () => {
@@ -315,15 +301,10 @@ describe('UpgradeCommandRunner', () => {
        // @ts-expect-error legacy noImplicitAny
        await upgradeCommandRunner.run(passedParams, options);

-        const { fail: failReport, success: successReport } =
-          upgradeCommandRunner.migrationReport;
-
-        expect(failReport.length).toBe(0);
-        expect(successReport.length).toBe(1);
-        expect(coreMigrationRunnerService.run).toHaveBeenCalledTimes(1);
-        const { workspaceId } = successReport[0];
-
-        expect(workspaceId).toBe('workspace_0');
+        expect(workspaceRepository.update).toHaveBeenCalledWith(
+          { id: 'workspace_0' },
+          { version: expect.any(String) },
+        );
      },
    );
  });
@@ -331,10 +312,7 @@ describe('UpgradeCommandRunner', () => {
  describe('Workspace upgrade should fail', () => {
    const failingTestUseCases: EachTestingContext<{
      input: Omit<BuildModuleAndSetupSpiesArgs, 'numberOfWorkspace'>;
-      output?: {
-        failReportWorkspaceId: string;
-        expectedErrorMessage: string;
-      };
+      expectedErrorMessage: string;
    }>[] = [
      {
        title: 'when workspace version is not equal to fromVersion',
@@ -344,10 +322,8 @@ describe('UpgradeCommandRunner', () => {
              version: '0.1.0',
            },
          },
-          output: {
-            failReportWorkspaceId: 'workspace_0',
-            expectedErrorMessage: `Unable to run the upgrade command. Aborting the upgrade process.\nPlease ensure that all workspaces are on at least the previous minor version (${PREVIOUS_VERSION}).\nIf any workspaces are not on the previous minor version, roll back to that version and run the upgrade command again.`,
-          },
+          expectedErrorMessage:
+            'Unable to run the upgrade command. Aborting the upgrade process.',
        },
      },
      {
@@ -358,10 +334,8 @@ describe('UpgradeCommandRunner', () => {
              version: null,
            },
          },
-          output: {
-            failReportWorkspaceId: 'workspace_0',
-            expectedErrorMessage: `Unable to run the upgrade command. Aborting the upgrade process.\nPlease ensure that all workspaces are on at least the previous minor version (${PREVIOUS_VERSION}).\nIf any workspaces are not on the previous minor version, roll back to that version and run the upgrade command again.`,
-          },
+          expectedErrorMessage:
+            'Unable to run the upgrade command. Aborting the upgrade process.',
        },
      },
      {
@@ -370,11 +344,8 @@ describe('UpgradeCommandRunner', () => {
          input: {
            appVersion: null,
          },
-          output: {
-            failReportWorkspaceId: 'global',
-            expectedErrorMessage:
-              'APP_VERSION is not defined, please double check your env variables',
-          },
+          expectedErrorMessage:
+            'APP_VERSION is not defined, please double check your env variables',
        },
      },
      {
@@ -383,11 +354,8 @@ describe('UpgradeCommandRunner', () => {
          input: {
            appVersion: '42.0.0',
          },
-          output: {
-            failReportWorkspaceId: 'global',
-            expectedErrorMessage:
-              'No command found for version 42.0.0. Please check the commands record.',
-          },
+          expectedErrorMessage:
+            'No command found for version 42.0.0. Please check the commands record.',
        },
      },
      {
@@ -396,33 +364,22 @@ describe('UpgradeCommandRunner', () => {
          input: {
            appVersion: UPGRADE_COMMAND_SUPPORTED_VERSIONS[0],
          },
-          output: {
-            failReportWorkspaceId: 'global',
-            expectedErrorMessage: `No previous version found for version ${UPGRADE_COMMAND_SUPPORTED_VERSIONS[0]}. Available versions: ${UPGRADE_COMMAND_SUPPORTED_VERSIONS.join(', ')}`,
-          },
+          expectedErrorMessage: `No previous version found for version ${UPGRADE_COMMAND_SUPPORTED_VERSIONS[0]}`,
        },
      },
    ];

    it.each(eachTestingContextFilter(failingTestUseCases))(
      '$title',
-      async ({ context: { input, output } }) => {
+      async ({ context: { input, expectedErrorMessage } }) => {
        await buildModuleAndSetupSpies(input);

        const passedParams: string[] = [];
        const options = {};

-        await upgradeCommandRunner.run(passedParams, options);
-
-        const { fail: failReport, success: successReport } =
-          upgradeCommandRunner.migrationReport;
-
-        expect(successReport.length).toBe(0);
-        expect(failReport.length).toBe(1);
-        const { workspaceId, error } = failReport[0];
-
-        expect(workspaceId).toBe(output?.failReportWorkspaceId ?? 'global');
-        expect(error).toEqual(new Error(output?.expectedErrorMessage ?? ''));
+        await expect(
+          upgradeCommandRunner.run(passedParams, options),
+        ).rejects.toThrow(expectedErrorMessage);
      },
    );
  });
@@ -0,0 +1,23 @@
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+
+import {
+  WorkspaceCommandRunner,
+  type WorkspaceCommandOptions,
+} from 'src/database/commands/command-runners/workspace.command-runner';
+import { type WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
+
+export type ActiveOrSuspendedWorkspaceCommandOptions = WorkspaceCommandOptions;
+
+export abstract class ActiveOrSuspendedWorkspaceCommandRunner<
+  Options extends
+    ActiveOrSuspendedWorkspaceCommandOptions = ActiveOrSuspendedWorkspaceCommandOptions,
+> extends WorkspaceCommandRunner<Options> {
+  constructor(
+    protected readonly workspaceIteratorService: WorkspaceIteratorService,
+  ) {
+    super(workspaceIteratorService, [
+      WorkspaceActivationStatus.ACTIVE,
+      WorkspaceActivationStatus.SUSPENDED,
+    ]);
+  }
+}
@@ -1,29 +0,0 @@
-import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { type Repository } from 'typeorm';
-
-import {
-  WorkspacesMigrationCommandRunner,
-  type WorkspacesMigrationCommandOptions,
-} from 'src/database/commands/command-runners/workspaces-migration.command-runner';
-import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
-import { type DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
-import { type GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
-
-export type ActiveOrSuspendedWorkspacesMigrationCommandOptions =
-  WorkspacesMigrationCommandOptions;
-
-export abstract class ActiveOrSuspendedWorkspacesMigrationCommandRunner<
-  Options extends
-    ActiveOrSuspendedWorkspacesMigrationCommandOptions = ActiveOrSuspendedWorkspacesMigrationCommandOptions,
-> extends WorkspacesMigrationCommandRunner<Options> {
-  constructor(
-    protected readonly workspaceRepository: Repository<WorkspaceEntity>,
-    protected readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
-    protected readonly dataSourceService: DataSourceService,
-  ) {
-    super(workspaceRepository, globalWorkspaceOrmManager, dataSourceService, [
-      WorkspaceActivationStatus.ACTIVE,
-      WorkspaceActivationStatus.SUSPENDED,
-    ]);
-  }
-}
@@ -52,11 +52,11 @@ export abstract class MigrationCommandRunner extends CommandRunner {

    try {
      await this.runMigrationCommand(passedParams, options);
+
+      this.logger.log(chalk.blue('Command completed!'));
    } catch (error) {
      this.logger.error(chalk.red(`Command failed`));
      throw error;
-    } finally {
-      this.logger.log(chalk.blue('Command completed!'));
    }
  }

@@ -1,104 +1,143 @@
 import { InjectRepository } from '@nestjs/typeorm';

 import chalk from 'chalk';
+import { CommandRunner, Option } from 'nest-commander';
 import { SemVer } from 'semver';
-import { isDefined } from 'twenty-shared/utils';
+import { assertUnreachable, isDefined } from 'twenty-shared/utils';
 import { Repository } from 'typeorm';

 import {
-  ActiveOrSuspendedWorkspacesMigrationCommandOptions,
-  ActiveOrSuspendedWorkspacesMigrationCommandRunner,
-} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
+  type RunOnWorkspaceArgs,
+  WorkspaceCommandRunner,
+} from 'src/database/commands/command-runners/workspace.command-runner';
 import {
-  RunOnWorkspaceArgs,
-  WorkspacesMigrationCommandRunner,
-} from 'src/database/commands/command-runners/workspaces-migration.command-runner';
+  type WorkspaceIteratorContext,
+  WorkspaceIteratorService,
+} from 'src/database/commands/command-runners/workspace-iterator.service';
 import { CoreMigrationRunnerService } from 'src/database/commands/core-migration-runner/services/core-migration-runner.service';
+import { CommandLogger } from 'src/database/commands/logger';
 import { type UpgradeCommandVersion } from 'src/engine/constants/upgrade-command-supported-versions.constant';
 import { CoreEngineVersionService } from 'src/engine/core-engine-version/services/core-engine-version.service';
-import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
-import { type DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
-import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
 import {
  type CompareVersionMajorAndMinorReturnType,
  compareVersionMajorAndMinor,
 } from 'src/utils/version/compare-version-minor-and-major';

-export type VersionCommands = (
-  | WorkspacesMigrationCommandRunner
-  | ActiveOrSuspendedWorkspacesMigrationCommandRunner
-)[];
+export type VersionCommands = WorkspaceCommandRunner[];
 export type AllCommands = Record<UpgradeCommandVersion, VersionCommands>;

-export abstract class UpgradeCommandRunner extends ActiveOrSuspendedWorkspacesMigrationCommandRunner {
-  private fromWorkspaceVersion: SemVer;
-  private currentAppVersion: SemVer;
+export type UpgradeCommandOptions = {
+  workspaceId?: Set<string>;
+  startFromWorkspaceId?: string;
+  workspaceCountLimit?: number;
+  dryRun?: boolean;
+  verbose?: boolean;
+};
+
+type VersionContext = {
+  fromWorkspaceVersion: SemVer;
+  currentAppVersion: SemVer;
+  commands: VersionCommands;
+};
+
+export abstract class UpgradeCommandRunner extends CommandRunner {
+  protected logger: CommandLogger;
+
  public abstract allCommands: AllCommands;
-  public commands: VersionCommands;
-  public readonly VALIDATE_WORKSPACE_VERSION_FEATURE_FLAG?: true;

  constructor(
    @InjectRepository(WorkspaceEntity)
    protected readonly workspaceRepository: Repository<WorkspaceEntity>,
-    protected readonly twentyConfigService: TwentyConfigService,
-    protected readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
-    protected readonly dataSourceService: DataSourceService,
    protected readonly coreEngineVersionService: CoreEngineVersionService,
    protected readonly workspaceVersionService: WorkspaceVersionService,
    protected readonly coreMigrationRunnerService: CoreMigrationRunnerService,
+    protected readonly workspaceIteratorService: WorkspaceIteratorService,
  ) {
-    super(workspaceRepository, globalWorkspaceOrmManager, dataSourceService);
+    super();
+    this.logger = new CommandLogger({
+      verbose: false,
+      constructorName: this.constructor.name,
+    });
  }

-  private setUpgradeContextVersionsAndCommandsForCurrentAppVersion() {
-    const upgradeContextIsAlreadyDefined = [
-      this.currentAppVersion,
-      this.commands,
-      this.fromWorkspaceVersion,
-    ].every(isDefined);
-
-    if (upgradeContextIsAlreadyDefined) {
-      return;
-    }
-
-    const currentAppVersion = this.coreEngineVersionService.getCurrentVersion();
-    const currentVersionMajorMinor =
-      `${currentAppVersion.major}.${currentAppVersion.minor}.0` as UpgradeCommandVersion;
-    const currentCommands = this.allCommands[currentVersionMajorMinor];
-
-    if (!isDefined(currentCommands)) {
-      throw new Error(
-        `No command found for version ${currentAppVersion}. Please check the commands record.`,
-      );
-    }
-
-    const previousVersion = this.coreEngineVersionService.getPreviousVersion();
-
-    this.commands = currentCommands;
-    this.fromWorkspaceVersion = previousVersion;
-    this.currentAppVersion = currentAppVersion;
-
-    const message = [
-      'Initialized upgrade context with:',
-      `- currentVersion (migrating to): ${currentAppVersion}`,
-      `- fromWorkspaceVersion: ${previousVersion}`,
-      `- ${this.commands.length} commands`,
-    ];
-
-    this.logger.log(chalk.blue(message.join('\n   ')));
+  @Option({
+    flags: '-d, --dry-run',
+    description: 'Simulate the command without making actual changes',
+    required: false,
+  })
+  parseDryRun(): boolean {
+    return true;
  }

-  override async runMigrationCommand(
-    passedParams: string[],
-    options: ActiveOrSuspendedWorkspacesMigrationCommandOptions,
+  @Option({
+    flags: '-v, --verbose',
+    description: 'Verbose output',
+    required: false,
+  })
+  parseVerbose(): boolean {
+    return true;
+  }
+
+  @Option({
+    flags: '-w, --workspace-id [workspace_id]',
+    description:
+      'workspace id. Command runs on all active/suspended workspaces if not provided.',
+    required: false,
+  })
+  parseWorkspaceId(val: string, previous?: Set<string>): Set<string> {
+    const accumulator = previous ?? new Set<string>();
+
+    accumulator.add(val);
+
+    return accumulator;
+  }
+
+  @Option({
+    flags: '--start-from-workspace-id [workspace_id]',
+    description:
+      'Start from a specific workspace id. Workspaces are processed in ascending order of id.',
+    required: false,
+  })
+  parseStartFromWorkspaceId(val: string): string {
+    return val;
+  }
+
+  @Option({
+    flags: '--workspace-count-limit [count]',
+    description:
+      'Limit the number of workspaces to process. Workspaces are processed in ascending order of id.',
+    required: false,
+  })
+  parseWorkspaceCountLimit(val: string): number {
+    const limit = parseInt(val);
+
+    if (isNaN(limit)) {
+      throw new Error('Workspace count limit must be a number');
+    }
+
+    if (limit <= 0) {
+      throw new Error('Workspace count limit must be greater than 0');
+    }
+
+    return limit;
+  }
+
+  override async run(
+    _passedParams: string[],
+    options: UpgradeCommandOptions,
  ): Promise<void> {
-    try {
-      this.setUpgradeContextVersionsAndCommandsForCurrentAppVersion();
+    if (options.verbose) {
+      this.logger = new CommandLogger({
+        verbose: true,
+        constructorName: this.constructor.name,
+      });
+    }
+
+    try {
+      const versionContext = this.resolveVersionContext();

-      // On fresh installs there are no workspaces yet, so skip the
-      // per-workspace upgrade loop (core migrations already ran above).
      const hasWorkspaces =
        await this.workspaceVersionService.hasActiveOrSuspendedWorkspaces();

@@ -110,75 +149,129 @@ export abstract class UpgradeCommandRunner extends ActiveOrSuspendedWorkspacesMi
        return;
      }

-      const workspacesThatAreBelowFromWorkspaceVersion =
+      const workspacesBelowMinimumVersion =
        await this.workspaceVersionService.getWorkspacesBelowVersion(
-          this.fromWorkspaceVersion.version,
+          versionContext.fromWorkspaceVersion.version,
        );

-      if (workspacesThatAreBelowFromWorkspaceVersion.length > 0) {
-        this.migrationReport.fail.push(
-          ...workspacesThatAreBelowFromWorkspaceVersion.map((workspace) => ({
-            error: new Error(
-              `Unable to run the upgrade command. Aborting the upgrade process.
-Please ensure that all workspaces are on at least the previous minor version (${this.fromWorkspaceVersion.version}).
-If any workspaces are not on the previous minor version, roll back to that version and run the upgrade command again.`,
-            ),
-            workspaceId: workspace.id,
-          })),
+      if (workspacesBelowMinimumVersion.length > 0) {
+        const ineligibleIds = workspacesBelowMinimumVersion
+          .map((workspace) => workspace.id)
+          .join(', ');
+
+        throw new Error(
+          `Unable to run the upgrade command. Aborting the upgrade process.
+Workspaces below minimum version (${versionContext.fromWorkspaceVersion.version}): ${ineligibleIds}.
+Please roll back to that version and run the upgrade command again.`,
        );
      }
-    } catch (error) {
-      this.migrationReport.fail.push({
-        error,
-        workspaceId: 'global',
-      });
-    }

-    if (this.migrationReport.fail.length > 0) {
-      this.migrationReport.fail.forEach(({ error, workspaceId }) =>
+      await this.coreMigrationRunnerService.run();
+
+      const iteratorReport = await this.workspaceIteratorService.iterate({
+        workspaceIds:
+          options.workspaceId && options.workspaceId.size > 0
+            ? Array.from(options.workspaceId)
+            : undefined,
+        startFromWorkspaceId: options.startFromWorkspaceId,
+        workspaceCountLimit: options.workspaceCountLimit,
+        dryRun: options.dryRun,
+        callback: async (context) => {
+          await this.runOnWorkspace(context, options, versionContext);
+        },
+      });
+
+      if (iteratorReport.fail.length > 0) {
        this.logger.error(
-          `Error in workspace ${workspaceId}: ${error.message}`,
+          chalk.red(
+            `Upgrade completed with ${iteratorReport.fail.length} workspace failure(s)`,
+          ),
+        );
+      }
+
+      this.logger.log(
+        chalk.blue(
+          `Upgrade summary: ${iteratorReport.success.length} succeeded, ${iteratorReport.fail.length} failed`,
        ),
      );
-
-      return;
+      this.logger.log(chalk.blue('Command completed!'));
+    } catch (error) {
+      this.logger.error(chalk.red(`Upgrade failed: ${error.message}`));
+      throw error;
    }
-
-    await this.coreMigrationRunnerService.run();
-    await super.runMigrationCommand(passedParams, options);
  }

-  override async runOnWorkspace(args: RunOnWorkspaceArgs): Promise<void> {
-    this.setUpgradeContextVersionsAndCommandsForCurrentAppVersion();
+  private resolveVersionContext(): VersionContext {
+    const currentAppVersion = this.coreEngineVersionService.getCurrentVersion();
+    const currentVersionMajorMinor =
+      `${currentAppVersion.major}.${currentAppVersion.minor}.0` as UpgradeCommandVersion;
+    const commands = this.allCommands[currentVersionMajorMinor];

-    const { workspaceId, index, total, options } = args;
+    if (!isDefined(commands)) {
+      throw new Error(
+        `No command found for version ${currentAppVersion}. Please check the commands record.`,
+      );
+    }
+
+    const fromWorkspaceVersion =
+      this.coreEngineVersionService.getPreviousVersion();

    this.logger.log(
      chalk.blue(
-        `${options.dryRun ? '(dry run) ' : ''}Upgrading workspace ${workspaceId} from=${this.fromWorkspaceVersion} to=${this.currentAppVersion} ${index + 1}/${total}`,
+        [
+          'Initialized upgrade context with:',
+          `- currentVersion (migrating to): ${currentAppVersion}`,
+          `- fromWorkspaceVersion: ${fromWorkspaceVersion}`,
+          `- ${commands.length} commands`,
+        ].join('\n   '),
      ),
    );

-    const workspaceVersionCompareResult =
-      await this.retrieveWorkspaceVersionAndCompareToWorkspaceFromVersion(
+    return { fromWorkspaceVersion, currentAppVersion, commands };
+  }
+
+  private async runOnWorkspace(
+    iteratorContext: WorkspaceIteratorContext,
+    options: UpgradeCommandOptions,
+    versionContext: VersionContext,
+  ): Promise<void> {
+    const { workspaceId, index, total } = iteratorContext;
+    const { fromWorkspaceVersion, currentAppVersion, commands } =
+      versionContext;
+
+    this.logger.log(
+      chalk.blue(
+        `${options.dryRun ? '(dry run) ' : ''}Upgrading workspace ${workspaceId} from=${fromWorkspaceVersion} to=${currentAppVersion} ${index + 1}/${total}`,
+      ),
+    );
+
+    const versionCompareResult =
+      await this.compareWorkspaceVersionToFromVersion(
        workspaceId,
+        fromWorkspaceVersion,
      );

-    switch (workspaceVersionCompareResult) {
+    switch (versionCompareResult) {
      case 'lower': {
        throw new Error(
-          `WORKSPACE_VERSION_MISSMATCH Upgrade for workspace ${workspaceId} failed as its version is beneath fromWorkspaceVersion=${this.fromWorkspaceVersion.version}`,
+          `WORKSPACE_VERSION_MISSMATCH Upgrade for workspace ${workspaceId} failed as its version is beneath fromWorkspaceVersion=${fromWorkspaceVersion.version}`,
        );
      }
      case 'equal': {
-        for (const command of this.commands) {
-          await command.runOnWorkspace(args);
+        for (const command of commands) {
+          await command.runOnWorkspace({
+            options: options as RunOnWorkspaceArgs['options'],
+            workspaceId,
+            dataSource: iteratorContext.dataSource,
+            index,
+            total,
+          });
        }

        if (!options.dryRun) {
          await this.workspaceRepository.update(
            { id: workspaceId },
-            { version: this.currentAppVersion.version },
+            { version: currentAppVersion.version },
          );
        }

@@ -198,15 +291,14 @@ If any workspaces are not on the previous minor version, roll back to that versi
        return;
      }
      default: {
-        throw new Error(
-          `Should never occur, encountered unexpected value from retrieveWorkspaceVersionAndCompareToWorkspaceFromVersion ${workspaceVersionCompareResult}`,
-        );
+        assertUnreachable(versionCompareResult);
      }
    }
  }

-  private async retrieveWorkspaceVersionAndCompareToWorkspaceFromVersion(
+  private async compareWorkspaceVersionToFromVersion(
    workspaceId: string,
+    fromWorkspaceVersion: SemVer,
  ): Promise<CompareVersionMajorAndMinorReturnType> {
    const workspace = await this.workspaceRepository.findOneByOrFail({
      id: workspaceId,
@@ -219,7 +311,7 @@ If any workspaces are not on the previous minor version, roll back to that versi

    return compareVersionMajorAndMinor(
      currentWorkspaceVersion,
-      this.fromWorkspaceVersion.version,
+      fromWorkspaceVersion.version,
    );
  }
 }
@@ -0,0 +1,13 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([WorkspaceEntity]), DataSourceModule],
+  providers: [WorkspaceIteratorService],
+  exports: [WorkspaceIteratorService],
+})
+export class WorkspaceIteratorModule {}
@@ -0,0 +1,139 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import chalk from 'chalk';
+import { isDefined } from 'twenty-shared/utils';
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { In, MoreThanOrEqual, Repository } from 'typeorm';
+
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
+import { GlobalWorkspaceDataSource } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource';
+import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
+import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
+
+export type WorkspaceIteratorArgs = {
+  workspaceIds?: string[];
+  activationStatuses?: WorkspaceActivationStatus[];
+  startFromWorkspaceId?: string;
+  workspaceCountLimit?: number;
+  dryRun?: boolean;
+  callback: (context: WorkspaceIteratorContext) => Promise<void>;
+};
+
+export type WorkspaceIteratorContext = {
+  workspaceId: string;
+  dataSource?: GlobalWorkspaceDataSource;
+  index: number;
+  total: number;
+};
+
+export type WorkspaceIteratorReport = {
+  fail: {
+    workspaceId: string;
+    error: Error;
+  }[];
+  success: {
+    workspaceId: string;
+  }[];
+};
+
+const DEFAULT_ACTIVATION_STATUSES = [
+  WorkspaceActivationStatus.ACTIVE,
+  WorkspaceActivationStatus.SUSPENDED,
+];
+
+@Injectable()
+export class WorkspaceIteratorService {
+  private readonly logger = new Logger(WorkspaceIteratorService.name);
+
+  constructor(
+    @InjectRepository(WorkspaceEntity)
+    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly dataSourceService: DataSourceService,
+  ) {}
+
+  async iterate(args: WorkspaceIteratorArgs): Promise<WorkspaceIteratorReport> {
+    const { callback, ...options } = args;
+
+    const report: WorkspaceIteratorReport = {
+      fail: [],
+      success: [],
+    };
+
+    const workspaceIdsToProcess =
+      options.workspaceIds && options.workspaceIds.length > 0
+        ? options.workspaceIds
+        : await this.fetchWorkspaceIds(options);
+
+    if (options.dryRun) {
+      this.logger.log(chalk.yellow('Dry run mode: No changes will be applied'));
+    }
+
+    for (const [index, workspaceId] of workspaceIdsToProcess.entries()) {
+      this.logger.log(
+        `Running on workspace ${workspaceId} ${index + 1}/${workspaceIdsToProcess.length}`,
+      );
+
+      try {
+        const authContext = buildSystemAuthContext(workspaceId);
+
+        await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+          async () => {
+            const workspaceHasDataSource =
+              await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceId(
+                workspaceId,
+              );
+
+            const dataSource = isDefined(workspaceHasDataSource)
+              ? await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource()
+              : undefined;
+
+            await callback({
+              workspaceId,
+              dataSource,
+              index,
+              total: workspaceIdsToProcess.length,
+            });
+          },
+          authContext,
+        );
+
+        report.success.push({ workspaceId });
+      } catch (error: unknown) {
+        report.fail.push({ error: error as Error, workspaceId });
+      }
+    }
+
+    report.fail.forEach(({ error, workspaceId }) =>
+      this.logger.error(
+        `Error in workspace ${workspaceId}: ${error.message}`,
+        error.stack,
+      ),
+    );
+
+    return report;
+  }
+
+  private async fetchWorkspaceIds(
+    options: Omit<WorkspaceIteratorArgs, 'callback'>,
+  ): Promise<string[]> {
+    const activationStatuses =
+      options.activationStatuses ?? DEFAULT_ACTIVATION_STATUSES;
+
+    const workspaces = await this.workspaceRepository.find({
+      select: ['id'],
+      where: {
+        activationStatus: In(activationStatuses),
+        ...(options.startFromWorkspaceId
+          ? { id: MoreThanOrEqual(options.startFromWorkspaceId) }
+          : {}),
+      },
+      order: { id: 'ASC' },
+      take: options.workspaceCountLimit,
+    });
+
+    return workspaces.map((workspace) => workspace.id);
+  }
+}
@@ -0,0 +1,140 @@
+import chalk from 'chalk';
+import { CommandRunner, Option } from 'nest-commander';
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+
+import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
+import { CommandLogger } from 'src/database/commands/logger';
+import { GlobalWorkspaceDataSource } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource';
+
+export type WorkspaceCommandOptions = {
+  workspaceId?: Set<string>;
+  startFromWorkspaceId?: string;
+  workspaceCountLimit?: number;
+  dryRun?: boolean;
+  verbose?: boolean;
+};
+
+export type RunOnWorkspaceArgs = {
+  options: WorkspaceCommandOptions;
+  workspaceId: string;
+  dataSource?: GlobalWorkspaceDataSource;
+  index: number;
+  total: number;
+};
+
+export abstract class WorkspaceCommandRunner<
+  Options extends WorkspaceCommandOptions = WorkspaceCommandOptions,
+> extends CommandRunner {
+  protected logger: CommandLogger;
+
+  constructor(
+    protected readonly workspaceIteratorService: WorkspaceIteratorService,
+    protected readonly activationStatuses: WorkspaceActivationStatus[],
+  ) {
+    super();
+    this.logger = new CommandLogger({
+      verbose: false,
+      constructorName: this.constructor.name,
+    });
+  }
+
+  @Option({
+    flags: '-d, --dry-run',
+    description: 'Simulate the command without making actual changes',
+    required: false,
+  })
+  parseDryRun(): boolean {
+    return true;
+  }
+
+  @Option({
+    flags: '-v, --verbose',
+    description: 'Verbose output',
+    required: false,
+  })
+  parseVerbose(): boolean {
+    return true;
+  }
+
+  @Option({
+    flags: '--start-from-workspace-id [workspace_id]',
+    description:
+      'Start from a specific workspace id. Workspaces are processed in ascending order of id.',
+    required: false,
+  })
+  parseStartFromWorkspaceId(val: string): string {
+    return val;
+  }
+
+  @Option({
+    flags: '--workspace-count-limit [count]',
+    description:
+      'Limit the number of workspaces to process. Workspaces are processed in ascending order of id.',
+    required: false,
+  })
+  parseWorkspaceCountLimit(val: string): number {
+    const limit = parseInt(val);
+
+    if (isNaN(limit)) {
+      throw new Error('Workspace count limit must be a number');
+    }
+
+    if (limit <= 0) {
+      throw new Error('Workspace count limit must be greater than 0');
+    }
+
+    return limit;
+  }
+
+  @Option({
+    flags: '-w, --workspace-id [workspace_id]',
+    description:
+      'workspace id. Command runs on all workspaces matching the activation statuses if not provided.',
+    required: false,
+  })
+  parseWorkspaceId(val: string, previous?: Set<string>): Set<string> {
+    const accumulator = previous ?? new Set<string>();
+
+    accumulator.add(val);
+
+    return accumulator;
+  }
+
+  override async run(_passedParams: string[], options: Options): Promise<void> {
+    if (options.verbose) {
+      this.logger = new CommandLogger({
+        verbose: true,
+        constructorName: this.constructor.name,
+      });
+    }
+
+    try {
+      await this.workspaceIteratorService.iterate({
+        workspaceIds:
+          options.workspaceId && options.workspaceId.size > 0
+            ? Array.from(options.workspaceId)
+            : undefined,
+        activationStatuses: this.activationStatuses,
+        startFromWorkspaceId: options.startFromWorkspaceId,
+        workspaceCountLimit: options.workspaceCountLimit,
+        dryRun: options.dryRun,
+        callback: async (context) => {
+          await this.runOnWorkspace({
+            options,
+            workspaceId: context.workspaceId,
+            dataSource: context.dataSource,
+            index: context.index,
+            total: context.total,
+          });
+        },
+      });
+
+      this.logger.log(chalk.blue('Command completed!'));
+    } catch (error) {
+      this.logger.error(chalk.red(`Command failed`));
+      throw error;
+    }
+  }
+
+  public abstract runOnWorkspace(args: RunOnWorkspaceArgs): Promise<void>;
+}
@@ -1,189 +0,0 @@
-import chalk from 'chalk';
-import { Option } from 'nest-commander';
-import { isDefined } from 'twenty-shared/utils';
-import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { In, MoreThanOrEqual, type Repository } from 'typeorm';
-
-import { MigrationCommandRunner } from 'src/database/commands/command-runners/migration.command-runner';
-import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
-import { type DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
-import { GlobalWorkspaceDataSource } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource';
-import { type GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
-import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
-
-export type WorkspacesMigrationCommandOptions = {
-  workspaceIds: string[];
-  startFromWorkspaceId?: string;
-  workspaceCountLimit?: number;
-  dryRun?: boolean;
-  verbose?: boolean;
-};
-
-export type RunOnWorkspaceArgs = {
-  options: WorkspacesMigrationCommandOptions;
-  workspaceId: string;
-  dataSource?: GlobalWorkspaceDataSource;
-  index: number;
-  total: number;
-};
-
-export type WorkspaceMigrationReport = {
-  fail: {
-    workspaceId: string;
-    error: Error;
-  }[];
-  success: {
-    workspaceId: string;
-  }[];
-};
-
-export abstract class WorkspacesMigrationCommandRunner<
-  Options extends
-    WorkspacesMigrationCommandOptions = WorkspacesMigrationCommandOptions,
-> extends MigrationCommandRunner {
-  protected workspaceIds: Set<string> = new Set();
-  private startFromWorkspaceId: string | undefined;
-  private workspaceCountLimit: number | undefined;
-  public migrationReport: WorkspaceMigrationReport = {
-    fail: [],
-    success: [],
-  };
-
-  constructor(
-    protected readonly workspaceRepository: Repository<WorkspaceEntity>,
-    protected readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
-    protected readonly dataSourceService: DataSourceService,
-    protected readonly activationStatuses: WorkspaceActivationStatus[],
-  ) {
-    super();
-  }
-
-  @Option({
-    flags: '--start-from-workspace-id [workspace_id]',
-    description:
-      'Start from a specific workspace id. Workspaces are processed in ascending order of id.',
-    required: false,
-  })
-  parseStartFromWorkspaceId(val: string): string {
-    this.startFromWorkspaceId = val;
-
-    return val;
-  }
-
-  @Option({
-    flags: '--workspace-count-limit [count]',
-    description:
-      'Limit the number of workspaces to process. Workspaces are processed in ascending order of id.',
-    required: false,
-  })
-  parseWorkspaceCountLimit(val: string): number {
-    this.workspaceCountLimit = parseInt(val);
-
-    if (isNaN(this.workspaceCountLimit)) {
-      throw new Error('Workspace count limit must be a number');
-    }
-
-    if (this.workspaceCountLimit <= 0) {
-      throw new Error('Workspace count limit must be greater than 0');
-    }
-
-    return this.workspaceCountLimit;
-  }
-
-  @Option({
-    flags: '-w, --workspace-id [workspace_id]',
-    description:
-      'workspace id. Command runs on all workspaces matching the activation statuses if not provided.',
-    required: false,
-  })
-  parseWorkspaceId(val: string): Set<string> {
-    this.workspaceIds.add(val);
-
-    return this.workspaceIds;
-  }
-
-  protected async fetchWorkspaceIds(): Promise<string[]> {
-    const workspaces = await this.workspaceRepository.find({
-      select: ['id'],
-      where: {
-        activationStatus: In(this.activationStatuses),
-        ...(this.startFromWorkspaceId
-          ? { id: MoreThanOrEqual(this.startFromWorkspaceId) }
-          : {}),
-      },
-      order: {
-        id: 'ASC',
-      },
-      take: this.workspaceCountLimit,
-    });
-
-    return workspaces.map((workspace) => workspace.id);
-  }
-
-  override async runMigrationCommand(
-    _passedParams: string[],
-    options: Options,
-  ) {
-    const workspaceIdsToProcess =
-      this.workspaceIds.size > 0
-        ? Array.from(this.workspaceIds)
-        : await this.fetchWorkspaceIds();
-
-    if (options.dryRun) {
-      this.logger.log(chalk.yellow('Dry run mode: No changes will be applied'));
-    }
-
-    for (const [index, workspaceId] of workspaceIdsToProcess.entries()) {
-      this.logger.log(
-        `Upgrading workspace ${workspaceId} ${index + 1}/${workspaceIdsToProcess.length}`,
-      );
-
-      try {
-        const authContext = buildSystemAuthContext(workspaceId);
-
-        await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
-          async () => {
-            const workspaceHasDataSource =
-              await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceId(
-                workspaceId,
-              );
-
-            const dataSource = isDefined(workspaceHasDataSource)
-              ? await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource()
-              : undefined;
-
-            await this.runOnWorkspace({
-              options,
-              workspaceId,
-              dataSource,
-              index: index,
-              total: workspaceIdsToProcess.length,
-            });
-          },
-          authContext,
-        );
-
-        this.migrationReport.success.push({
-          workspaceId,
-        });
-      } catch (error) {
-        this.migrationReport.fail.push({
-          error,
-          workspaceId,
-        });
-        this.logger.warn(
-          chalk.red(`Error in workspace ${workspaceId}: ${error.message}`),
-        );
-      }
-    }
-
-    this.migrationReport.fail.forEach(({ error, workspaceId }) =>
-      this.logger.error(
-        `Error in workspace ${workspaceId}: ${error.message}`,
-        error.stack,
-      ),
-    );
-  }
-
-  public abstract runOnWorkspace(args: RunOnWorkspaceArgs): Promise<void>;
-}