Refactor upgrade devx to allow configuring workspaces status to pass over (#16066)

# Introduction
We need to be able to create custom workspace application on all
workspaces, even pending and ongoing etc
Right now the upgrade devx only allows and expect active or suspended
workspace to be passed to runOnWorkspace.

## WorkspacesMigrationRunner
Created an intermediate class `WorkspacesMigrationRunner` that expect an
array `WorkspaceStatus` to be fetched for the current command to be run
on
The `ActiveOrSuspendedCommandRunner` statically passes both `SUSPENDED`
and `ACTIVE`, whereas the create workspace custom application passed all
the enum values

## DataSource
Workspace that are not fully init don't have a `workspace_schema` so
they don't have `dataSource`
Made a not very elegant check to see if current workspace we're about to
create dataSource on has one historically
Which means that dataSource is now optional, it had only one impact on
an existing command and the desired devx will become consuming existing
services that do not expect dataSource ( or at least yet )

packages/twenty-server/src/database/commands/command-runners/__tests__/upgrade.command-runner.spec.ts

@@ -11,6 +11,7 @@ import { UpgradeCommandRunner } from 'src/database/commands/command-runners/upgr
 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 { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import { SyncWorkspaceMetadataCommand } from 'src/engine/workspace-manager/workspace-sync-metadata/commands/sync-workspace-metadata.command';
 
@@ -72,9 +73,37 @@ const buildUpgradeCommandModule = async ({
   appVersion,
   commandRunner,
 }: BuildUpgradeCommandModuleArgs) => {
+  const mockDataSourceService = {
+    getLastDataSourceMetadataFromWorkspaceId: jest.fn(),
+  };
+
   const module: TestingModule = await Test.createTestingModule({
     providers: [
-      commandRunner,
+      {
+        provide: commandRunner,
+        useFactory: (
+          workspaceRepository: Repository<WorkspaceEntity>,
+          twentyConfigService: TwentyConfigService,
+          twentyORMGlobalManager: TwentyORMGlobalManager,
+          dataSourceService: DataSourceService,
+          syncWorkspaceMetadataCommand: SyncWorkspaceMetadataCommand,
+        ) => {
+          return new commandRunner(
+            workspaceRepository,
+            twentyConfigService,
+            twentyORMGlobalManager,
+            dataSourceService,
+            syncWorkspaceMetadataCommand,
+          );
+        },
+        inject: [
+          getRepositoryToken(WorkspaceEntity),
+          TwentyConfigService,
+          TwentyORMGlobalManager,
+          DataSourceService,
+          SyncWorkspaceMetadataCommand,
+        ],
+      },
       {
         provide: getRepositoryToken(WorkspaceEntity),
         useValue: {
@@ -110,6 +139,10 @@ const buildUpgradeCommandModule = async ({
           getDataSourceForWorkspace: jest.fn(),
         },
       },
+      {
+        provide: DataSourceService,
+        useValue: mockDataSourceService,
+      },
       {
         provide: SyncWorkspaceMetadataCommand,
         useValue: {

packages/twenty-server/src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner.ts

@@ -1,180 +1,29 @@
-import chalk from 'chalk';
-import { Option } from 'nest-commander';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { In, MoreThanOrEqual, type Repository } from 'typeorm';
+import { 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 WorkspaceDataSource } from 'src/engine/twenty-orm/datasource/workspace.datasource';
+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 TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 
-export type ActiveOrSuspendedWorkspacesMigrationCommandOptions = {
-  workspaceIds: string[];
-  startFromWorkspaceId?: string;
-  workspaceCountLimit?: number;
-  dryRun?: boolean;
-  verbose?: boolean;
-};
-
-export type RunOnWorkspaceArgs = {
-  options: ActiveOrSuspendedWorkspacesMigrationCommandOptions;
-  workspaceId: string;
-  dataSource: WorkspaceDataSource;
-  index: number;
-  total: number;
-};
-
-export type WorkspaceMigrationReport = {
-  fail: {
-    workspaceId: string;
-    error: Error;
-  }[];
-  success: {
-    workspaceId: string;
-  }[];
-};
+export type ActiveOrSuspendedWorkspacesMigrationCommandOptions =
+  WorkspacesMigrationCommandOptions;
 
 export abstract class ActiveOrSuspendedWorkspacesMigrationCommandRunner<
   Options extends
     ActiveOrSuspendedWorkspacesMigrationCommandOptions = ActiveOrSuspendedWorkspacesMigrationCommandOptions,
-> extends MigrationCommandRunner {
-  private workspaceIds: Set<string> = new Set();
-  private startFromWorkspaceId: string | undefined;
-  private workspaceCountLimit: number | undefined;
-  public migrationReport: WorkspaceMigrationReport = {
-    fail: [],
-    success: [],
-  };
-
+> extends WorkspacesMigrationCommandRunner<Options> {
   constructor(
     protected readonly workspaceRepository: Repository<WorkspaceEntity>,
     protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    protected readonly dataSourceService: DataSourceService,
   ) {
-    super();
+    super(workspaceRepository, twentyORMGlobalManager, dataSourceService, [
+      WorkspaceActivationStatus.ACTIVE,
+      WorkspaceActivationStatus.SUSPENDED,
+    ]);
   }
-
-  @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 active workspaces if not provided.',
-    required: false,
-  })
-  parseWorkspaceId(val: string): Set<string> {
-    this.workspaceIds.add(val);
-
-    return this.workspaceIds;
-  }
-
-  protected async fetchActiveWorkspaceIds(): Promise<string[]> {
-    const activeWorkspaces = await this.workspaceRepository.find({
-      select: ['id'],
-      where: {
-        activationStatus: In([
-          WorkspaceActivationStatus.ACTIVE,
-          WorkspaceActivationStatus.SUSPENDED,
-        ]),
-        ...(this.startFromWorkspaceId
-          ? { id: MoreThanOrEqual(this.startFromWorkspaceId) }
-          : {}),
-      },
-      order: {
-        id: 'ASC',
-      },
-      take: this.workspaceCountLimit,
-    });
-
-    return activeWorkspaces.map((workspace) => workspace.id);
-  }
-
-  override async runMigrationCommand(
-    _passedParams: string[],
-    options: Options,
-  ) {
-    const activeWorkspaceIds =
-      this.workspaceIds.size > 0
-        ? Array.from(this.workspaceIds)
-        : await this.fetchActiveWorkspaceIds();
-
-    if (options.dryRun) {
-      this.logger.log(chalk.yellow('Dry run mode: No changes will be applied'));
-    }
-
-    for (const [index, workspaceId] of activeWorkspaceIds.entries()) {
-      this.logger.log(
-        `Running command on workspace ${workspaceId} ${index + 1}/${activeWorkspaceIds.length}`,
-      );
-
-      try {
-        const dataSource =
-          await this.twentyORMGlobalManager.getDataSourceForWorkspace({
-            workspaceId,
-          });
-
-        await this.runOnWorkspace({
-          options,
-          workspaceId,
-          dataSource,
-          index: index,
-          total: activeWorkspaceIds.length,
-        });
-        this.migrationReport.success.push({
-          workspaceId,
-        });
-      } catch (error) {
-        this.migrationReport.fail.push({
-          error,
-          workspaceId,
-        });
-        this.logger.warn(
-          chalk.red(`Error in workspace ${workspaceId}: ${error.message}`),
-        );
-      }
-
-      try {
-        await this.twentyORMGlobalManager.destroyDataSourceForWorkspace(
-          workspaceId,
-        );
-      } catch (error) {
-        this.logger.error(error);
-      }
-    }
-
-    this.migrationReport.fail.forEach(({ error, workspaceId }) =>
-      this.logger.error(`Error in workspace ${workspaceId}: ${error.message}`),
-    );
-  }
-
-  public abstract runOnWorkspace(args: RunOnWorkspaceArgs): Promise<void>;
 }

packages/twenty-server/src/database/commands/command-runners/upgrade.command-runner.ts

@@ -12,10 +12,14 @@ import { In, Repository } from 'typeorm';
 import {
   ActiveOrSuspendedWorkspacesMigrationCommandOptions,
   ActiveOrSuspendedWorkspacesMigrationCommandRunner,
-  type RunOnWorkspaceArgs,
 } from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
+import {
+  RunOnWorkspaceArgs,
+  WorkspacesMigrationCommandRunner,
+} from 'src/database/commands/command-runners/workspaces-migration.command-runner';
 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 { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import { SyncWorkspaceMetadataCommand } from 'src/engine/workspace-manager/workspace-sync-metadata/commands/sync-workspace-metadata.command';
 import {
@@ -25,8 +29,14 @@ import {
 import { getPreviousVersion } from 'src/utils/version/get-previous-version';
 
 export type VersionCommands = {
-  beforeSyncMetadata: ActiveOrSuspendedWorkspacesMigrationCommandRunner[];
-  afterSyncMetadata: ActiveOrSuspendedWorkspacesMigrationCommandRunner[];
+  beforeSyncMetadata: (
+    | WorkspacesMigrationCommandRunner
+    | ActiveOrSuspendedWorkspacesMigrationCommandRunner
+  )[];
+  afterSyncMetadata: (
+    | WorkspacesMigrationCommandRunner
+    | ActiveOrSuspendedWorkspacesMigrationCommandRunner
+  )[];
 };
 export type AllCommands = Record<string, VersionCommands>;
 const execPromise = promisify(exec);
@@ -43,9 +53,10 @@ export abstract class UpgradeCommandRunner extends ActiveOrSuspendedWorkspacesMi
     protected readonly workspaceRepository: Repository<WorkspaceEntity>,
     protected readonly twentyConfigService: TwentyConfigService,
     protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    protected readonly dataSourceService: DataSourceService,
     protected readonly syncWorkspaceMetadataCommand: SyncWorkspaceMetadataCommand,
   ) {
-    super(workspaceRepository, twentyORMGlobalManager);
+    super(workspaceRepository, twentyORMGlobalManager, dataSourceService);
   }
 
   private async loadActiveOrSuspendedWorkspace() {

packages/twenty-server/src/database/commands/command-runners/workspaces-migration.command-runner.ts

@@ -0,0 +1,187 @@
+import chalk from 'chalk';
+import { Option } from 'nest-commander';
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { In, MoreThanOrEqual, type Repository } from 'typeorm';
+import { isDefined } from 'twenty-shared/utils';
+
+import { MigrationCommandRunner } from 'src/database/commands/command-runners/migration.command-runner';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
+import { type WorkspaceDataSource } from 'src/engine/twenty-orm/datasource/workspace.datasource';
+import { type TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+
+export type WorkspacesMigrationCommandOptions = {
+  workspaceIds: string[];
+  startFromWorkspaceId?: string;
+  workspaceCountLimit?: number;
+  dryRun?: boolean;
+  verbose?: boolean;
+};
+
+export type RunOnWorkspaceArgs = {
+  options: WorkspacesMigrationCommandOptions;
+  workspaceId: string;
+  dataSource?: WorkspaceDataSource;
+  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 twentyORMGlobalManager: TwentyORMGlobalManager,
+    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(
+        `Running command on workspace ${workspaceId} ${index + 1}/${workspaceIdsToProcess.length}`,
+      );
+
+      try {
+        const workspaceHasDataSource =
+          await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceId(
+            workspaceId,
+          );
+
+        const dataSource = isDefined(workspaceHasDataSource)
+          ? await this.twentyORMGlobalManager.getDataSourceForWorkspace({
+              workspaceId,
+            })
+          : undefined;
+
+        await this.runOnWorkspace({
+          options,
+          workspaceId,
+          dataSource,
+          index: index,
+          total: workspaceIdsToProcess.length,
+        });
+        this.migrationReport.success.push({
+          workspaceId,
+        });
+      } catch (error) {
+        this.migrationReport.fail.push({
+          error,
+          workspaceId,
+        });
+        this.logger.warn(
+          chalk.red(`Error in workspace ${workspaceId}: ${error.message}`),
+        );
+      }
+
+      try {
+        await this.twentyORMGlobalManager.destroyDataSourceForWorkspace(
+          workspaceId,
+        );
+      } catch (error) {
+        this.logger.error(error);
+      }
+    }
+
+    this.migrationReport.fail.forEach(({ error, workspaceId }) =>
+      this.logger.error(`Error in workspace ${workspaceId}: ${error.message}`),
+    );
+  }
+
+  public abstract runOnWorkspace(args: RunOnWorkspaceArgs): Promise<void>;
+}