Instance commands and upgrade_migrations table (#19356)

# Introduction Now only using typeorm to generate migrations up and down statement We handle and maintain our own migration table history ## What's new Now all the instance commands will live within the same module and folder than the upgrade commands Sequentiality comes from the timestamp located in the filename Same sequentiality also applies to the workspace commands in the future, for the moment still expected a as code explicit declaration ( below screen is an example see below section ) <img width="1382" height="634" alt="image" src="https://github.com/user-attachments/assets/5610a246-4eae-485e-99f4-98fb89ad5ac8" /> ## Existing 1.21 migrations We won't start following this pattern in 1.21 yet at least not with the migration that has already been released as typeorm migrations in cloud production as they would rerun ## Small duplication Duplicating the legacy typeorm and instance commands run in the `run-instance-commands` to avoid any merge of interest for the moment ## Concurrency Not handling any run in parrallel of the upgrade for the moment
2026-04-07 10:55:17 +02:00
parent b23d2b4e73
commit 23874848a4
40 changed files with 1314 additions and 903 deletions
@@ -0,0 +1,145 @@
+import 'reflect-metadata';
+
+import { Test } from '@nestjs/testing';
+import { DiscoveryService } from '@nestjs/core';
+
+import { type MigrationInterface } from 'typeorm';
+
+import { RegisteredInstanceMigrationService } from 'src/engine/core-modules/upgrade/services/registered-instance-migration-registry.service';
+import { RegisteredInstanceMigration } from 'src/database/typeorm/core/decorators/registered-instance-migration.decorator';
+
+@RegisteredInstanceMigration('1.21.0', 1770000000000)
+class MigrationA1770000000000 implements MigrationInterface {
+  name = 'MigrationA1770000000000';
+
+  async up(): Promise<void> {}
+  async down(): Promise<void> {}
+}
+
+@RegisteredInstanceMigration('1.21.0', 1771000000000)
+class MigrationB1771000000000 implements MigrationInterface {
+  name = 'MigrationB1771000000000';
+
+  async up(): Promise<void> {}
+  async down(): Promise<void> {}
+}
+
+@RegisteredInstanceMigration('1.21.0', 1772000000000)
+class MigrationC1772000000000 implements MigrationInterface {
+  name = 'MigrationC1772000000000';
+
+  async up(): Promise<void> {}
+  async down(): Promise<void> {}
+}
+
+@RegisteredInstanceMigration('1.20.0', 1769000000000)
+class MigrationD1769000000000 implements MigrationInterface {
+  name = 'MigrationD1769000000000';
+
+  async up(): Promise<void> {}
+  async down(): Promise<void> {}
+}
+
+class UndecoratedMigration1768000000000 implements MigrationInterface {
+  name = 'UndecoratedMigration1768000000000';
+
+  async up(): Promise<void> {}
+  async down(): Promise<void> {}
+}
+
+const buildProviderWrapper = (migration: MigrationInterface) => ({
+  instance: migration,
+  metatype: migration.constructor,
+});
+
+const buildRegistryService = async (
+  migrations: MigrationInterface[],
+): Promise<RegisteredInstanceMigrationService> => {
+  const module = await Test.createTestingModule({
+    providers: [
+      RegisteredInstanceMigrationService,
+      {
+        provide: DiscoveryService,
+        useValue: {
+          getProviders: () => migrations.map(buildProviderWrapper),
+        },
+      },
+    ],
+  }).compile();
+
+  const service = module.get(RegisteredInstanceMigrationService);
+
+  service.onModuleInit();
+
+  return service;
+};
+
+describe('RegisteredInstanceMigrationService', () => {
+  it('should group migrations by version', async () => {
+    const service = await buildRegistryService([
+      new MigrationD1769000000000(),
+      new MigrationA1770000000000(),
+      new MigrationB1771000000000(),
+      new MigrationC1772000000000(),
+    ]);
+
+    const v120 = service.getInstanceCommandsForVersion('1.20.0');
+    const v121 = service.getInstanceCommandsForVersion('1.21.0');
+
+    expect(v120.map((m) => m.constructor.name)).toStrictEqual([
+      'MigrationD1769000000000',
+    ]);
+
+    expect(v121.map((m) => m.constructor.name)).toStrictEqual([
+      'MigrationA1770000000000',
+      'MigrationB1771000000000',
+      'MigrationC1772000000000',
+    ]);
+  });
+
+  it('should sort migrations by timestamp within a version bucket', async () => {
+    const service = await buildRegistryService([
+      new MigrationC1772000000000(),
+      new MigrationA1770000000000(),
+      new MigrationB1771000000000(),
+    ]);
+
+    const names = service
+      .getInstanceCommandsForVersion('1.21.0')
+      .map((m) => m.constructor.name);
+
+    expect(names).toStrictEqual([
+      'MigrationA1770000000000',
+      'MigrationB1771000000000',
+      'MigrationC1772000000000',
+    ]);
+  });
+
+  it('should skip undecorated migrations', async () => {
+    const service = await buildRegistryService([
+      new UndecoratedMigration1768000000000(),
+      new MigrationA1770000000000(),
+    ]);
+
+    const v121 = service.getInstanceCommandsForVersion('1.21.0');
+
+    expect(v121).toHaveLength(1);
+    expect(v121[0].constructor.name).toBe('MigrationA1770000000000');
+  });
+
+  it('should return empty array for version with no migrations', async () => {
+    const service = await buildRegistryService([]);
+
+    expect(service.getInstanceCommandsForVersion('1.19.0')).toStrictEqual([]);
+    expect(service.getInstanceCommandsForVersion('1.20.0')).toStrictEqual([]);
+    expect(service.getInstanceCommandsForVersion('1.21.0')).toStrictEqual([]);
+  });
+
+  it('should return empty array for unsupported version', async () => {
+    const service = await buildRegistryService([]);
+
+    expect(
+      service.getInstanceCommandsForVersion('99.0.0' as unknown as '1.21.0'),
+    ).toStrictEqual([]);
+  });
+});
@@ -0,0 +1,121 @@
+import { Injectable } from '@nestjs/common';
+import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+
+import {
+  DataSource,
+  MigrationInterface,
+  type QueryRunner,
+  Repository,
+} from 'typeorm';
+
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+import {
+  UpgradeMigrationEntity,
+  type UpgradeMigrationStatus,
+} from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
+
+export type RunSingleMigrationResult =
+  | { status: 'success' }
+  | { status: 'already-executed' }
+  | { status: 'failed'; error: unknown };
+
+@Injectable()
+export class InstanceUpgradeService {
+  constructor(
+    @InjectRepository(UpgradeMigrationEntity)
+    private readonly upgradeMigrationRepository: Repository<UpgradeMigrationEntity>,
+    @InjectDataSource()
+    private readonly dataSource: DataSource,
+    private readonly twentyConfigService: TwentyConfigService,
+  ) {}
+
+  async runSingleMigration(
+    migration: MigrationInterface,
+  ): Promise<RunSingleMigrationResult> {
+    const migrationName = migration.constructor.name;
+    const executedByVersion =
+      this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
+
+    const isAlreadyExecuted = await this.upgradeMigrationRepository.exists({
+      where: { name: migrationName, status: 'completed' },
+    });
+
+    if (isAlreadyExecuted) {
+      return { status: 'already-executed' };
+    }
+
+    const queryRunner = this.dataSource.createQueryRunner();
+
+    try {
+      await queryRunner.connect();
+      await queryRunner.startTransaction();
+
+      await migration.up(queryRunner);
+
+      await this.markAsCompleted({
+        queryRunner,
+        name: migrationName,
+        executedByVersion,
+      });
+
+      await queryRunner.commitTransaction();
+    } catch (error) {
+      if (queryRunner.isTransactionActive) {
+        await queryRunner.rollbackTransaction();
+      }
+
+      await this.markFailed({
+        name: migrationName,
+        executedByVersion,
+      });
+
+      return { status: 'failed', error };
+    } finally {
+      await queryRunner.release();
+    }
+
+    return { status: 'success' };
+  }
+
+  private async markAsCompleted({
+    queryRunner,
+    name,
+    executedByVersion,
+  }: {
+    queryRunner: QueryRunner;
+    name: string;
+    executedByVersion: string;
+  }): Promise<void> {
+    const repository = queryRunner.manager.getRepository(
+      UpgradeMigrationEntity,
+    );
+
+    const previousAttempts = await repository.count({ where: { name } });
+
+    await repository.save({
+      name,
+      status: 'completed' as UpgradeMigrationStatus,
+      attempt: previousAttempts + 1,
+      executedByVersion,
+    });
+  }
+
+  private async markFailed({
+    name,
+    executedByVersion,
+  }: {
+    name: string;
+    executedByVersion: string;
+  }): Promise<void> {
+    const previousAttempts = await this.upgradeMigrationRepository.count({
+      where: { name },
+    });
+
+    await this.upgradeMigrationRepository.save({
+      name,
+      status: 'failed' as UpgradeMigrationStatus,
+      attempt: previousAttempts + 1,
+      executedByVersion,
+    });
+  }
+}
@@ -0,0 +1,100 @@
+import { Injectable, Logger, type OnModuleInit } from '@nestjs/common';
+import { DiscoveryService } from '@nestjs/core';
+
+import { type MigrationInterface } from 'typeorm';
+
+import { getRegisteredInstanceMigrationMetadata } from 'src/database/typeorm/core/decorators/registered-instance-migration.decorator';
+import {
+  UPGRADE_COMMAND_SUPPORTED_VERSIONS,
+  type UpgradeCommandVersion,
+} from 'src/engine/constants/upgrade-command-supported-versions.constant';
+
+type TimestampedMigration = {
+  migration: MigrationInterface;
+  timestamp: number;
+};
+
+@Injectable()
+export class RegisteredInstanceMigrationService implements OnModuleInit {
+  private readonly logger = new Logger(RegisteredInstanceMigrationService.name);
+
+  private readonly migrationsByVersion = new Map<
+    UpgradeCommandVersion,
+    TimestampedMigration[]
+  >();
+
+  constructor(private readonly discoveryService: DiscoveryService) {}
+
+  onModuleInit(): void {
+    for (const version of UPGRADE_COMMAND_SUPPORTED_VERSIONS) {
+      this.migrationsByVersion.set(version, []);
+    }
+
+    const providers = this.discoveryService.getProviders();
+
+    for (const wrapper of providers) {
+      const { instance, metatype } = wrapper;
+
+      if (!instance || !metatype) {
+        continue;
+      }
+
+      const metadata = getRegisteredInstanceMigrationMetadata(metatype);
+
+      if (metadata === undefined) {
+        continue;
+      }
+
+      const bucket = this.migrationsByVersion.get(metadata.version);
+
+      if (!bucket) {
+        continue;
+      }
+
+      bucket.push({
+        migration: instance as MigrationInterface,
+        timestamp: metadata.timestamp,
+      });
+    }
+
+    for (const [, bucket] of this.migrationsByVersion) {
+      bucket.sort((entryA, entryB) => entryA.timestamp - entryB.timestamp);
+    }
+
+    for (const [version, bucket] of this.migrationsByVersion) {
+      if (bucket.length > 0) {
+        this.logger.log(
+          `Registered ${bucket.length} versioned migration(s) for ${version}: ${bucket.map((entry) => entry.migration.constructor.name).join(', ')}`,
+        );
+      }
+    }
+  }
+
+  getInstanceCommandsForVersion(
+    version: UpgradeCommandVersion,
+  ): MigrationInterface[] {
+    return (this.migrationsByVersion.get(version) ?? []).map(
+      (entry) => entry.migration,
+    );
+  }
+
+  getAllInstanceCommands(): {
+    version: UpgradeCommandVersion;
+    migration: MigrationInterface;
+  }[] {
+    const result: {
+      version: UpgradeCommandVersion;
+      migration: MigrationInterface;
+    }[] = [];
+
+    for (const version of UPGRADE_COMMAND_SUPPORTED_VERSIONS) {
+      const bucket = this.migrationsByVersion.get(version) ?? [];
+
+      for (const entry of bucket) {
+        result.push({ version, migration: entry.migration });
+      }
+    }
+
+    return result;
+  }
+}
@@ -0,0 +1,115 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { SemVer } from 'semver';
+import { assertUnreachable, isDefined } from 'twenty-shared/utils';
+import { Repository } from 'typeorm';
+
+import { type WorkspaceIteratorContext } from 'src/database/commands/command-runners/workspace-iterator.service';
+import {
+  type UpgradeCommandOptions,
+  type VersionCommands,
+} from 'src/database/commands/command-runners/upgrade.command-runner';
+import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import {
+  type CompareVersionMajorAndMinorReturnType,
+  compareVersionMajorAndMinor,
+} from 'src/utils/version/compare-version-minor-and-major';
+
+export type UpgradeWorkspaceArgs = {
+  iteratorContext: WorkspaceIteratorContext;
+  options: UpgradeCommandOptions;
+  fromWorkspaceVersion: SemVer;
+  currentAppVersion: SemVer;
+  workspaceCommands: VersionCommands;
+};
+
+@Injectable()
+export class WorkspaceUpgradeService {
+  private readonly logger = new Logger(WorkspaceUpgradeService.name);
+
+  constructor(
+    @InjectRepository(WorkspaceEntity)
+    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+  ) {}
+
+  async upgradeWorkspace({
+    iteratorContext,
+    options,
+    fromWorkspaceVersion,
+    currentAppVersion,
+    workspaceCommands,
+  }: UpgradeWorkspaceArgs): Promise<void> {
+    const { workspaceId, index, total } = iteratorContext;
+
+    this.logger.log(
+      `${options.dryRun ? '(dry run) ' : ''}Upgrading workspace ${workspaceId} from=${fromWorkspaceVersion} to=${currentAppVersion} ${index + 1}/${total}`,
+    );
+
+    const versionCompareResult =
+      await this.compareWorkspaceVersionToFromVersion(
+        workspaceId,
+        fromWorkspaceVersion,
+      );
+
+    switch (versionCompareResult) {
+      case 'lower': {
+        throw new Error(
+          `WORKSPACE_VERSION_MISMATCH Upgrade for workspace ${workspaceId} failed as its version is beneath fromWorkspaceVersion=${fromWorkspaceVersion.version}`,
+        );
+      }
+      case 'equal': {
+        for (const workspaceCommand of workspaceCommands) {
+          await workspaceCommand.runOnWorkspace({
+            options: options as RunOnWorkspaceArgs['options'],
+            workspaceId,
+            dataSource: iteratorContext.dataSource,
+            index,
+            total,
+          });
+        }
+
+        if (!options.dryRun) {
+          await this.workspaceRepository.update(
+            { id: workspaceId },
+            { version: currentAppVersion.version },
+          );
+        }
+
+        this.logger.log(`Upgrade for workspace ${workspaceId} completed.`);
+
+        return;
+      }
+      case 'higher': {
+        this.logger.log(
+          `Upgrade for workspace ${workspaceId} ignored as is already at a higher version.`,
+        );
+
+        return;
+      }
+      default: {
+        assertUnreachable(versionCompareResult);
+      }
+    }
+  }
+
+  private async compareWorkspaceVersionToFromVersion(
+    workspaceId: string,
+    fromWorkspaceVersion: SemVer,
+  ): Promise<CompareVersionMajorAndMinorReturnType> {
+    const workspace = await this.workspaceRepository.findOneByOrFail({
+      id: workspaceId,
+    });
+    const currentWorkspaceVersion = workspace.version;
+
+    if (!isDefined(currentWorkspaceVersion)) {
+      throw new Error(`WORKSPACE_VERSION_NOT_DEFINED workspace=${workspaceId}`);
+    }
+
+    return compareVersionMajorAndMinor(
+      currentWorkspaceVersion,
+      fromWorkspaceVersion.version,
+    );
+  }
+}
@@ -0,0 +1,31 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  PrimaryGeneratedColumn,
+  Unique,
+} from 'typeorm';
+
+export type UpgradeMigrationStatus = 'completed' | 'failed';
+
+@Entity({ name: 'upgradeMigration', schema: 'core' })
+@Unique('UQ_upgrade_migration_name_attempt', ['name', 'attempt'])
+export class UpgradeMigrationEntity {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ type: 'varchar', nullable: false })
+  name: string;
+
+  @Column({ type: 'varchar', nullable: false })
+  status: UpgradeMigrationStatus;
+
+  @Column({ type: 'integer', nullable: false, default: 1 })
+  attempt: number;
+
+  @Column({ type: 'varchar', nullable: false })
+  executedByVersion: string;
+
+  @CreateDateColumn({ type: 'timestamptz' })
+  createdAt: Date;
+}
@@ -0,0 +1,27 @@
+import { Module } from '@nestjs/common';
+import { DiscoveryModule } from '@nestjs/core';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+import { UpgradeMigrationEntity } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
+import { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
+import { RegisteredInstanceMigrationService } from 'src/engine/core-modules/upgrade/services/registered-instance-migration-registry.service';
+import { WorkspaceUpgradeService } from 'src/engine/core-modules/upgrade/services/workspace-upgrade.service';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+
+@Module({
+  imports: [
+    DiscoveryModule,
+    TypeOrmModule.forFeature([UpgradeMigrationEntity, WorkspaceEntity]),
+  ],
+  providers: [
+    InstanceUpgradeService,
+    WorkspaceUpgradeService,
+    RegisteredInstanceMigrationService,
+  ],
+  exports: [
+    InstanceUpgradeService,
+    WorkspaceUpgradeService,
+    RegisteredInstanceMigrationService,
+  ],
+})
+export class UpgradeModule {}