Implement cross version upgrade (#19559)

# Introduction
Refactoring the upgrade engine to handle cross version upgrade,
completely getting rid of the semver `version` at db and runtime level
It remains a visual a listing indicator for or CD process but also
during devenv in order to prepare next release
Will write a release process runbook documentation on how to handle
upgrade step patch, command insertion etc as it needs to be cascaded
across all the involved supported version

**The upgrade sequence model:**

The sequence is a flat, ordered array of upgrade steps
(`UpgradeStep[]`), built from the registry by chaining all versions in
order, each version contributing its fast-instance → slow-instance →
workspace commands sorted by timestamp. Version is metadata for logging,
not used in the algorithm.

**Segments:**

The sequence naturally splits into alternating segments of contiguous
instance steps and contiguous workspace steps. The runner processes
segments in order:

- **Instance segment:** Run sequentially from the instance cursor. Each
step runs once globally.
- **Workspace segment:** Each workspace independently walks from its own
cursor through the end of the segment. Workspaces are independent within
a segment — they can be at different positions.
- **Synchronization (workspace → instance):** The runner blocks before
entering an instance segment. All active/suspended workspaces must have
completed the last workspace step of the preceding workspace segment. If
any workspace failed, abort. This is the only explicit synchronization
point.
- Instance → workspace ordering is implicit — the runner processes
segments sequentially, so the instance segment naturally completes
before the workspace segment begins.


full docs
https://gist.github.com/prastoin/e62106d455fd72d6b6ebada8351e5492

## Version constants & type-level deprecation

Version management is split into three atomic constants:
`TWENTY_PREVIOUS_VERSIONS`, `TWENTY_CURRENT_VERSION`, and
`TWENTY_NEXT_VERSIONS`. Two derived constants compose them:
`CROSS_UPGRADE_SUPPORTED_VERSIONS` (previous + current — what the engine
runs) and `ALL_TWENTY_VERSIONS` (the full ordered tuple including next).
The registry service validates at module init that no version is
duplicated across constants and that at least one previous version
exists.

A `DeprecatedSinceVersion<RemoveAtVersion, T>` type utility resolves to
`T` while `TWENTY_CURRENT_VERSION` is below `RemoveAtVersion`, and to
`never` once it reaches it — turning deprecation into a compile-time
guarantee via `IndexOf` and `IsGreaterOrEqual` generics in
`twenty-shared`.

### `workspace.version` column deprecation

The column is replaced by cursor-based state inference from
`UpgradeMigration` records, but cannot be dropped in 1.22: workspaces
activated during 1.21 predate the cursor system and need their initial
cursor backfilled first (`backfillWorkspaceCreatedIn1_21_0Cursors`).
This backfill itself depends on a new `isInitial` column on
`UpgradeMigration`, bootstrapped via a targeted TypeORM migration before
the upgrade sequence runs.

Both functions and the entity field are typed with
`DeprecatedSinceVersion<'1.23.0', ...>`. When `TWENTY_CURRENT_VERSION`
reaches `1.23.0`, compile errors force their removal — and the
pre-declared `DropWorkspaceVersionColumnFastInstanceCommand` takes over
to drop the column.

## What's next
- ci cross version upgrade ( wip )
- banner asking to contact twenty administrator if workspace is outdated
- upgrade healthcheck cli 

## New unit/integ test pattern
Create a dedicated `createNestApp` that consumes a real database in
order not to have to mack any database interaction to the
`upgradeMigrations` allowing full coverage of the whole
`upgradeRunnerService.run` core logic
This commit is contained in:
Paul Rastoin
2026-04-13 11:42:27 +02:00
committed by GitHub
parent 7091561489
commit 21142d98fe
52 changed files with 2585 additions and 1359 deletions
+2 -1
View File
@@ -220,7 +220,8 @@
"options": {
"cwd": "packages/twenty-server",
"command": "node dist/command/command.js generate:instance-command"
}
},
"cache": false
},
"generate:integration-test": {
"executor": "nx:run-commands",
@@ -115,7 +115,10 @@ export class WorkspaceIteratorService {
}
private async fetchWorkspaceIds(
options: Omit<WorkspaceIteratorArgs, 'callback'>,
options: Pick<
WorkspaceIteratorArgs,
'activationStatuses' | 'startFromWorkspaceId' | 'workspaceCountLimit'
>,
): Promise<string[]> {
const activationStatuses =
options.activationStatuses ?? DEFAULT_ACTIVATION_STATUSES;
@@ -12,7 +12,6 @@ import { InstanceCommandProviderModule } from 'src/database/commands/upgrade-ver
import { UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/upgrade-version-command.module';
import { WorkspaceExportModule } from 'src/database/commands/workspace-export/workspace-export.module';
import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
import { CoreEngineVersionModule } from 'src/engine/core-engine-version/core-engine-version.module';
import { ApiKeyModule } from 'src/engine/core-modules/api-key/api-key.module';
import { GenerateApiKeyCommand } from 'src/engine/core-modules/api-key/commands/generate-api-key.command';
import { MarketplaceModule } from 'src/engine/core-modules/application/application-marketplace/marketplace.module';
@@ -74,7 +73,6 @@ import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/au
MarketplaceModule,
ApplicationUpgradeModule,
StaleRegistrationCleanupModule,
CoreEngineVersionModule,
WorkspaceVersionModule,
UpgradeModule,
],
@@ -6,7 +6,11 @@ import { Logger } from '@nestjs/common';
import { Command, CommandRunner, Option } from 'nest-commander';
import { InstanceCommandGenerationService } from 'src/database/commands/instance-command-generation.service';
import { UPGRADE_COMMAND_SUPPORTED_VERSIONS } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import {
TWENTY_ALL_VERSIONS,
type TwentyAllVersion,
} from 'src/engine/core-modules/upgrade/constants/twenty-all-versions.constant';
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
import { type InstanceCommandType } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
const UPGRADE_VERSION_COMMAND_DIR = path.resolve(
@@ -17,6 +21,7 @@ const UPGRADE_VERSION_COMMAND_DIR = path.resolve(
type GenerateInstanceCommandOptions = {
name: string;
type: InstanceCommandType;
version?: TwentyAllVersion;
};
@Command({
@@ -56,17 +61,30 @@ export class GenerateInstanceCommandCommand extends CommandRunner {
return value;
}
@Option({
flags: '--version <version>',
description: 'Target version (e.g. 1.23.0). Defaults to CURRENT_VERSION.',
})
parseVersion(value: string): TwentyAllVersion {
if (
!TWENTY_ALL_VERSIONS.includes(
value as (typeof TWENTY_ALL_VERSIONS)[number],
)
) {
throw new Error(
`Invalid version "${value}". Must be one of: ${TWENTY_ALL_VERSIONS.join(', ')}`,
);
}
return value as TwentyAllVersion;
}
async run(
_passedParams: string[],
options: GenerateInstanceCommandOptions,
): Promise<void> {
const migrationName = options.name;
const version = UPGRADE_COMMAND_SUPPORTED_VERSIONS.slice(-1)[0];
if (!version) {
throw new Error('No supported versions found');
}
const version = options.version ?? TWENTY_CURRENT_VERSION;
const commandType = options.type;
@@ -4,12 +4,12 @@ import { InjectDataSource } from '@nestjs/typeorm';
import { pascalCase } from 'twenty-shared/utils';
import { DataSource } from 'typeorm';
import { type UpgradeCommandVersion } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { type TwentyAllVersion } from 'src/engine/core-modules/upgrade/constants/twenty-all-versions.constant';
import { type InstanceCommandType } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
type GenerateInstanceCommandArgs = {
migrationName: string;
version: UpgradeCommandVersion;
version: TwentyAllVersion;
timestamp: number;
type?: InstanceCommandType;
};
@@ -5,9 +5,10 @@ import chalk from 'chalk';
import { Command, CommandRunner, Option } from 'nest-commander';
import { DataSource } from 'typeorm';
import { CoreEngineVersionService } from 'src/engine/core-engine-version/services/core-engine-version.service';
import { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
import { TWENTY_PREVIOUS_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-previous-versions.constant';
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
type RunInstanceCommandsOptions = {
@@ -15,6 +16,7 @@ type RunInstanceCommandsOptions = {
includeSlow?: boolean;
};
// TODO should be replaced by a specific call to the upgrade
@Command({
name: 'run-instance-commands',
description:
@@ -26,10 +28,10 @@ export class RunInstanceCommandsCommand extends CommandRunner {
constructor(
@InjectDataSource()
private readonly dataSource: DataSource,
private readonly coreEngineVersionService: CoreEngineVersionService,
private readonly workspaceVersionService: WorkspaceVersionService,
private readonly upgradeCommandRegistryService: UpgradeCommandRegistryService,
private readonly instanceUpgradeService: InstanceUpgradeService,
private readonly instanceUpgradeService: InstanceCommandRunnerService,
private readonly upgradeMigrationService: UpgradeMigrationService,
) {
super();
}
@@ -63,7 +65,7 @@ export class RunInstanceCommandsCommand extends CommandRunner {
for (const {
command,
name,
} of this.upgradeCommandRegistryService.getAllFastInstanceCommands()) {
} of this.upgradeCommandRegistryService.getCrossUpgradeSupportedFastInstanceCommands()) {
const result = await this.instanceUpgradeService.runFastInstanceCommand(
{
command,
@@ -83,7 +85,7 @@ export class RunInstanceCommandsCommand extends CommandRunner {
for (const {
command,
name,
} of this.upgradeCommandRegistryService.getAllSlowInstanceCommands()) {
} of this.upgradeCommandRegistryService.getCrossUpgradeSupportedSlowInstanceCommands()) {
const result =
await this.instanceUpgradeService.runSlowInstanceCommand({
command,
@@ -106,6 +108,52 @@ export class RunInstanceCommandsCommand extends CommandRunner {
}
}
private async checkWorkspaceVersionSafety(
options: RunInstanceCommandsOptions,
): Promise<void> {
if (options.force) {
this.logger.warn(
chalk.yellow('Skipping workspace version check (--force flag used)'),
);
return;
}
const activeWorkspaceIds =
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds();
if (activeWorkspaceIds.length === 0) {
return;
}
const previousVersion =
TWENTY_PREVIOUS_VERSIONS[TWENTY_PREVIOUS_VERSIONS.length - 1];
const lastWorkspaceCommand =
this.upgradeCommandRegistryService.getLastWorkspaceCommandForVersion(
previousVersion,
);
if (!lastWorkspaceCommand) {
return;
}
const allAtPreviousVersion =
await this.upgradeMigrationService.areAllWorkspacesAtCommand({
commandName: lastWorkspaceCommand.name,
workspaceIds: activeWorkspaceIds,
});
if (!allAtPreviousVersion) {
throw new Error(
'Unable to run instance commands. Some workspace(s) have not completed ' +
`the last workspace command for ${previousVersion} ("${lastWorkspaceCommand.name}").\n` +
'Please ensure all workspaces are upgraded to at least the previous version before running migrations.\n' +
'Use --force to bypass this check (not recommended).',
);
}
}
private async runLegacyPendingTypeOrmMigrations(): Promise<void> {
this.logger.log('Running legacy TypeORM migrations...');
@@ -121,39 +169,4 @@ export class RunInstanceCommandsCommand extends CommandRunner {
);
}
}
private async checkWorkspaceVersionSafety(
options: RunInstanceCommandsOptions,
): Promise<void> {
if (options.force) {
this.logger.warn(
chalk.yellow('Skipping workspace version check (--force flag used)'),
);
return;
}
const previousVersion = this.coreEngineVersionService.getPreviousVersion();
const workspacesBelow =
await this.workspaceVersionService.getWorkspacesBelowVersion(
previousVersion.version,
);
if (workspacesBelow.length > 0) {
for (const workspace of workspacesBelow) {
this.logger.error(
chalk.red(
`Workspace ${workspace.id} (${workspace.displayName}) is at version ${workspace.version ?? 'undefined'}, which is below the minimum required version.`,
),
);
}
throw new Error(
'Unable to run instance commands. Some workspace(s) are below the minimum required version.\n' +
'Please ensure all workspaces are on at least the previous minor version before running migrations.\n' +
'Use --force to bypass this check (not recommended).',
);
}
}
}
@@ -0,0 +1,21 @@
import { QueryRunner } from 'typeorm';
import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
import { FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface';
@RegisteredInstanceCommand('1.23.0', 1785000000000)
export class DropWorkspaceVersionColumnFastInstanceCommand
implements FastInstanceCommand
{
public async up(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(
`ALTER TABLE "core"."workspace" DROP COLUMN IF EXISTS "version"`,
);
}
public async down(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(
`ALTER TABLE "core"."workspace" ADD "version" character varying`,
);
}
}
@@ -0,0 +1,4 @@
import { Module } from '@nestjs/common';
@Module({})
export class V1_23_UpgradeVersionCommandModule {}
@@ -1,656 +0,0 @@
import { Test, type TestingModule } from '@nestjs/testing';
import {
eachTestingContextFilter,
type EachTestingContext,
} from 'twenty-shared/testing';
import { type DataSource, type QueryRunner } from 'typeorm';
import { type FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface';
import { type SlowInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/slow-instance-command.interface';
import { getDataSourceToken } from '@nestjs/typeorm';
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
import {
UpgradeCommand,
UpgradeCommandOptions,
} from 'src/database/commands/upgrade-version-command/upgrade.command';
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 { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { WorkspaceUpgradeService } from 'src/engine/core-modules/upgrade/services/workspace-upgrade.service';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
import { compareVersionMajorAndMinor } from 'src/utils/version/compare-version-minor-and-major';
const CURRENT_VERSION =
UPGRADE_COMMAND_SUPPORTED_VERSIONS[
UPGRADE_COMMAND_SUPPORTED_VERSIONS.length - 1
];
const PREVIOUS_VERSION =
UPGRADE_COMMAND_SUPPORTED_VERSIONS[
UPGRADE_COMMAND_SUPPORTED_VERSIONS.length - 2
];
type CommandRunnerValues = typeof UpgradeCommand;
const generateMockWorkspace = (overrides?: Partial<WorkspaceEntity>) =>
({
id: 'workspace-id',
version: PREVIOUS_VERSION,
createdAt: new Date(),
updatedAt: new Date(),
allowImpersonation: false,
isPublicInviteLinkEnabled: false,
displayName: 'Test Workspace',
domainName: 'test',
inviteHash: 'hash',
logo: null,
deletedAt: null,
activationStatus: 'active',
workspaceMembersCount: 1,
...overrides,
}) as WorkspaceEntity;
type BuildUpgradeCommandModuleArgs = {
workspaces: WorkspaceEntity[];
appVersion: string | null;
commandRunner: CommandRunnerValues;
migrations?: FastInstanceCommand[];
};
const buildUpgradeCommandModule = async ({
workspaces,
appVersion,
commandRunner,
migrations,
}: BuildUpgradeCommandModuleArgs) => {
const registryProvider = migrations
? {
provide: UpgradeCommandRegistryService,
useFactory: () => {
const fakeDiscoveryService = {
getProviders: () =>
migrations.map((migration) => ({
instance: migration,
metatype: migration.constructor,
})),
} as unknown as import('@nestjs/core').DiscoveryService;
const registry = new UpgradeCommandRegistryService(
fakeDiscoveryService,
);
registry.onModuleInit();
return registry;
},
}
: {
provide: UpgradeCommandRegistryService,
useValue: {
getBundleForVersion: jest.fn().mockReturnValue({
fastInstanceCommands: [],
slowInstanceCommands: [],
workspaceCommands: [],
}),
},
};
const module: TestingModule = await Test.createTestingModule({
providers: [
{
provide: getDataSourceToken(),
useValue: {
runMigrations: jest.fn().mockResolvedValue([]),
},
},
{
provide: commandRunner,
useFactory: (
coreEngineVersionService: CoreEngineVersionService,
workspaceVersionService: WorkspaceVersionService,
upgradeCommandRegistryService: UpgradeCommandRegistryService,
instanceUpgradeService: InstanceUpgradeService,
workspaceIteratorService: WorkspaceIteratorService,
workspaceUpgradeService: WorkspaceUpgradeService,
dataSource: DataSource,
) => {
return new commandRunner(
coreEngineVersionService,
workspaceVersionService,
upgradeCommandRegistryService,
instanceUpgradeService,
workspaceIteratorService,
workspaceUpgradeService,
dataSource,
);
},
inject: [
CoreEngineVersionService,
WorkspaceVersionService,
UpgradeCommandRegistryService,
InstanceUpgradeService,
WorkspaceIteratorService,
WorkspaceUpgradeService,
getDataSourceToken(),
],
},
{
provide: TwentyConfigService,
useValue: {
get: jest.fn().mockImplementation((key: keyof ConfigVariables) => {
switch (key) {
case 'APP_VERSION': {
return appVersion;
}
default: {
return;
}
}
}),
},
},
CoreEngineVersionService,
{
provide: WorkspaceVersionService,
useValue: {
hasActiveOrSuspendedWorkspaces: jest
.fn()
.mockResolvedValue(workspaces.length > 0),
getWorkspacesBelowVersion: jest
.fn()
.mockImplementation((version: string) => {
return workspaces.filter((workspace) => {
if (
workspace.version === null ||
workspace.version === undefined
) {
return true;
}
try {
return (
compareVersionMajorAndMinor(workspace.version, version) ===
'lower'
);
} catch {
return true;
}
});
}),
},
},
{
provide: InstanceUpgradeService,
useValue: {
runFastInstanceCommand: jest
.fn()
.mockResolvedValue({ status: 'success' }),
runSlowInstanceCommand: jest
.fn()
.mockResolvedValue({ status: 'success' }),
},
},
{
provide: WorkspaceUpgradeService,
useValue: {
upgradeWorkspace: jest.fn().mockResolvedValue(undefined),
},
},
registryProvider,
{
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();
return module;
};
describe('UpgradeCommandRunner', () => {
let upgradeCommandRunner: UpgradeCommand;
type BuildModuleAndSetupSpiesArgs = {
numberOfWorkspace?: number;
workspaceOverride?: Partial<WorkspaceEntity>;
workspaces?: WorkspaceEntity[];
appVersion?: string | null;
commandRunner?: CommandRunnerValues;
migrations?: FastInstanceCommand[];
};
const buildModuleAndSetupSpies = async ({
numberOfWorkspace = 1,
workspaceOverride,
workspaces,
commandRunner = UpgradeCommand,
appVersion = CURRENT_VERSION,
migrations,
}: BuildModuleAndSetupSpiesArgs) => {
const generatedWorkspaces = Array.from(
{ length: numberOfWorkspace },
(_v, index) =>
generateMockWorkspace({
id: `workspace_${index}`,
...workspaceOverride,
}),
);
const module = await buildUpgradeCommandModule({
commandRunner,
appVersion,
workspaces: [...generatedWorkspaces, ...(workspaces ?? [])],
migrations,
});
upgradeCommandRunner = module.get(commandRunner);
jest.spyOn(upgradeCommandRunner['logger'], 'log').mockImplementation();
jest.spyOn(upgradeCommandRunner['logger'], 'error').mockImplementation();
jest.spyOn(upgradeCommandRunner['logger'], 'warn').mockImplementation();
return module;
};
it('should delegate workspace upgrade to WorkspaceUpgradeService', async () => {
const module = await buildModuleAndSetupSpies({});
const workspaceUpgradeService = module.get(WorkspaceUpgradeService);
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run(passedParams, options);
expect(workspaceUpgradeService.upgradeWorkspace).toHaveBeenCalledTimes(1);
});
it('should call upgradeWorkspace for each workspace', async () => {
const numberOfWorkspace = 42;
const module = await buildModuleAndSetupSpies({
numberOfWorkspace,
});
const workspaceUpgradeService = module.get(WorkspaceUpgradeService);
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run(passedParams, options);
expect(workspaceUpgradeService.upgradeWorkspace).toHaveBeenCalledTimes(
numberOfWorkspace,
);
});
describe('Workspace upgrade should succeed ', () => {
const successfulTestUseCases: EachTestingContext<{
input: Omit<BuildModuleAndSetupSpiesArgs, 'numberOfWorkspace'>;
}>[] = [
{
title: 'even if workspace version and app version differ in patch',
context: {
input: {
appVersion: `v${CURRENT_VERSION}`,
workspaceOverride: {
version: `v${PREVIOUS_VERSION.replace('.0', '.12')}`,
},
},
},
},
{
title:
'even if workspace version and app version differ in patch and semantic',
context: {
input: {
appVersion: `v${CURRENT_VERSION}`,
workspaceOverride: {
version: PREVIOUS_VERSION.replace('.0', '.12'),
},
},
},
},
{
title: 'even if app version contains a patch value',
context: {
input: {
appVersion: CURRENT_VERSION.replace('.0', '.24'),
workspaceOverride: {
version: PREVIOUS_VERSION.replace('.0', '.12'),
},
},
},
},
];
it.each(eachTestingContextFilter(successfulTestUseCases))(
'$title',
async ({ context: { input } }) => {
const module = await buildModuleAndSetupSpies(input);
const workspaceUpgradeService = module.get(WorkspaceUpgradeService);
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run(passedParams, options);
expect(workspaceUpgradeService.upgradeWorkspace).toHaveBeenCalled();
},
);
});
it('should call runFastInstanceCommand for each current-version instance command', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class AddIndexToUsers1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(CURRENT_VERSION, 1771000000000)
class AddColumnToAccounts1771000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(PREVIOUS_VERSION, 1769000000000)
class DropLegacyTable1769000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
class UndecoratedMigration1768000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const addIndex = new AddIndexToUsers1770000000000();
const addColumn = new AddColumnToAccounts1771000000000();
const dropLegacy = new DropLegacyTable1769000000000();
const undecorated = new UndecoratedMigration1768000000000();
const module = await buildModuleAndSetupSpies({
migrations: [undecorated, dropLegacy, addIndex, addColumn],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
await upgradeCommandRunner.run([], {});
expect(instanceUpgradeService.runFastInstanceCommand).toHaveBeenCalledTimes(
2,
);
expect(
instanceUpgradeService.runFastInstanceCommand,
).toHaveBeenNthCalledWith(1, {
command: addIndex,
name: `${CURRENT_VERSION}_AddIndexToUsers1770000000000_1770000000000`,
});
expect(
instanceUpgradeService.runFastInstanceCommand,
).toHaveBeenNthCalledWith(2, {
command: addColumn,
name: `${CURRENT_VERSION}_AddColumnToAccounts1771000000000_1771000000000`,
});
});
it('should propagate errors from runFastInstanceCommand', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class FailingMigration1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
migrations: [new FailingMigration1770000000000()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(
instanceUpgradeService.runFastInstanceCommand as jest.Mock
).mockResolvedValue({
status: 'failed',
error: new Error('SQL error'),
});
await expect(upgradeCommandRunner.run([], {})).rejects.toThrow('SQL error');
});
describe('Workspace upgrade should fail', () => {
const failingTestUseCases: EachTestingContext<{
input: Omit<BuildModuleAndSetupSpiesArgs, 'numberOfWorkspace'>;
expectedErrorMessage: string;
}>[] = [
{
title: 'when workspace version is not equal to fromVersion',
context: {
input: {
workspaceOverride: {
version: '0.1.0',
},
},
expectedErrorMessage:
'Unable to run the upgrade command. Aborting the upgrade process.',
},
},
{
title: 'when workspace version is not defined',
context: {
input: {
workspaceOverride: {
version: null,
},
},
expectedErrorMessage:
'Unable to run the upgrade command. Aborting the upgrade process.',
},
},
{
title: 'when APP_VERSION is not defined',
context: {
input: {
appVersion: null,
},
expectedErrorMessage:
'APP_VERSION is not defined, please double check your env variables',
},
},
{
title: 'when previous version is not found',
context: {
input: {
appVersion: UPGRADE_COMMAND_SUPPORTED_VERSIONS[0],
},
expectedErrorMessage: `No previous version found for version ${UPGRADE_COMMAND_SUPPORTED_VERSIONS[0]}`,
},
},
];
it.each(eachTestingContextFilter(failingTestUseCases))(
'$title',
async ({ context: { input, expectedErrorMessage } }) => {
await buildModuleAndSetupSpies(input);
const passedParams: string[] = [];
const options = {};
await expect(
upgradeCommandRunner.run(passedParams, options),
).rejects.toThrow(expectedErrorMessage);
},
);
});
it('should call runSlowInstanceCommand for each current-version slow command', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigration1780000000000 implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const slowMigration = new SlowMigration1780000000000();
const module = await buildModuleAndSetupSpies({
migrations: [slowMigration],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
await upgradeCommandRunner.run([], {});
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledTimes(
1,
);
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledWith({
command: slowMigration,
name: `${CURRENT_VERSION}_SlowMigration1780000000000_1780000000000`,
skipDataMigration: false,
});
});
it('should run slow commands after fast commands but before workspace commands', async () => {
const executionOrder: string[] = [];
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class FastMigration1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigration1780000000000 implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
migrations: [
new FastMigration1770000000000(),
new SlowMigration1780000000000(),
],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(
instanceUpgradeService.runFastInstanceCommand as jest.Mock
).mockImplementation(async () => {
executionOrder.push('fast');
return { status: 'success' };
});
(
instanceUpgradeService.runSlowInstanceCommand as jest.Mock
).mockImplementation(async () => {
executionOrder.push('slow');
return { status: 'success' };
});
const workspaceIteratorService = module.get(WorkspaceIteratorService);
(workspaceIteratorService.iterate as jest.Mock).mockImplementation(
async () => {
executionOrder.push('workspace');
return { success: [], fail: [] };
},
);
await upgradeCommandRunner.run([], {});
expect(executionOrder).toStrictEqual(['fast', 'slow', 'workspace']);
});
it('should pass skipDataMigration: true on fresh install (no workspaces)', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigrationFreshInstall implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
numberOfWorkspace: 0,
migrations: [new SlowMigrationFreshInstall()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
await upgradeCommandRunner.run([], {});
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledWith({
command: expect.any(SlowMigrationFreshInstall),
name: `${CURRENT_VERSION}_SlowMigrationFreshInstall_1780000000000`,
skipDataMigration: true,
});
});
it('should propagate errors from runSlowInstanceCommand', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class FailingSlowMigration implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
migrations: [new FailingSlowMigration()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(
instanceUpgradeService.runSlowInstanceCommand as jest.Mock
).mockResolvedValue({
status: 'failed',
error: new Error('Data migration error'),
});
await expect(upgradeCommandRunner.run([], {})).rejects.toThrow(
'Data migration error',
);
});
});
@@ -5,9 +5,10 @@ import { MigrateMessagingCalendarToCoreFastInstanceCommand } from 'src/database/
import { AddEmailThreadWidgetTypeFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-21/1-21-instance-command-fast-1775200000000-add-email-thread-widget-type';
import { AddPermissionFlagRoleIdIndexFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-fast-1775749486425-add-permission-flag-role-id-index';
import { AddWorkspaceIdToIndirectEntitiesFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-fast-1775758621017-add-workspace-id-to-indirect-entities';
import { BackfillWorkspaceIdOnIndirectEntitiesSlowInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-slow-1775758621018-backfill-workspace-id-on-indirect-entities';
import { AddWorkspaceIdIndexesAndFksFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-fast-1775761294897-add-workspace-id-indexes-and-fks-to-indirect-entities';
import { DropObjectMetadataDataSourceFkFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-fast-1775804361516-drop-object-metadata-data-source-fk';
import { BackfillWorkspaceIdOnIndirectEntitiesSlowInstanceCommand } from 'src/database/commands/upgrade-version-command/1-22/1-22-instance-command-slow-1775758621018-backfill-workspace-id-on-indirect-entities';
import { DropWorkspaceVersionColumnFastInstanceCommand } from 'src/database/commands/upgrade-version-command/1-23/1-23-instance-command-fast-1785000000000-drop-workspace-version-column';
export const INSTANCE_COMMANDS = [
AddViewFieldGroupIdIndexOnViewFieldFastInstanceCommand,
@@ -18,4 +19,5 @@ export const INSTANCE_COMMANDS = [
BackfillWorkspaceIdOnIndirectEntitiesSlowInstanceCommand,
AddWorkspaceIdIndexesAndFksFastInstanceCommand,
DropObjectMetadataDataSourceFkFastInstanceCommand,
DropWorkspaceVersionColumnFastInstanceCommand,
];
@@ -3,8 +3,8 @@ import { Module } from '@nestjs/common';
import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module';
import { V1_21_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/1-21/1-21-upgrade-version-command.module';
import { V1_22_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/1-22/1-22-upgrade-version-command.module';
import { V1_23_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/1-23/1-23-upgrade-version-command.module';
import { UpgradeCommand } from 'src/database/commands/upgrade-version-command/upgrade.command';
import { CoreEngineVersionModule } from 'src/engine/core-engine-version/core-engine-version.module';
import { UpgradeModule } from 'src/engine/core-modules/upgrade/upgrade.module';
import { WorkspaceVersionModule } from 'src/engine/workspace-manager/workspace-version/workspace-version.module';
@@ -12,7 +12,7 @@ import { WorkspaceVersionModule } from 'src/engine/workspace-manager/workspace-v
imports: [
V1_21_UpgradeVersionCommandModule,
V1_22_UpgradeVersionCommandModule,
CoreEngineVersionModule,
V1_23_UpgradeVersionCommandModule,
UpgradeModule,
WorkspaceVersionModule,
WorkspaceIteratorModule,
@@ -2,25 +2,18 @@ import { InjectDataSource } from '@nestjs/typeorm';
import chalk from 'chalk';
import { Command, CommandRunner, Option } from 'nest-commander';
import { SemVer } from 'semver';
import { DataSource } from 'typeorm';
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.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 { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
import {
UpgradeCommandRegistryService,
type RegisteredWorkspaceCommand,
type VersionBundle,
} from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { WorkspaceUpgradeService } from 'src/engine/core-modules/upgrade/services/workspace-upgrade.service';
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { UpgradeSequenceReaderService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { UpgradeSequenceRunnerService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service';
import { RemovedSinceVersion } from 'src/engine/core-modules/upgrade/types/removed-since-version.type';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
import { isDefined } from 'twenty-shared/utils';
export type VersionCommands = RegisteredWorkspaceCommand[];
export type UpgradeCommandOptions = {
type RawUpgradeCommandOptions = {
workspaceId?: Set<string>;
startFromWorkspaceId?: string;
workspaceCountLimit?: number;
@@ -28,10 +21,12 @@ export type UpgradeCommandOptions = {
verbose?: boolean;
};
type VersionContext = VersionBundle & {
fromWorkspaceVersion: SemVer;
currentAppVersion: SemVer;
currentVersionMajorMinor: UpgradeCommandVersion;
export type ParsedUpgradeCommandOptions = {
workspaceIds?: string[];
startFromWorkspaceId?: string;
workspaceCountLimit?: number;
dryRun?: boolean;
verbose?: boolean;
};
@Command({
@@ -42,12 +37,11 @@ export class UpgradeCommand extends CommandRunner {
protected logger: CommandLogger;
constructor(
protected readonly coreEngineVersionService: CoreEngineVersionService,
protected readonly workspaceVersionService: WorkspaceVersionService,
protected readonly upgradeCommandRegistryService: UpgradeCommandRegistryService,
protected readonly instanceUpgradeService: InstanceUpgradeService,
protected readonly workspaceIteratorService: WorkspaceIteratorService,
protected readonly workspaceUpgradeService: WorkspaceUpgradeService,
protected readonly upgradeSequenceReaderService: UpgradeSequenceReaderService,
protected readonly upgradeSequenceRunnerService: UpgradeSequenceRunnerService,
protected readonly upgradeMigrationService: UpgradeMigrationService,
protected readonly workspaceVersionService: WorkspaceVersionService,
@InjectDataSource()
protected readonly dataSource: DataSource,
) {
@@ -122,7 +116,7 @@ export class UpgradeCommand extends CommandRunner {
override async run(
_passedParams: string[],
options: UpgradeCommandOptions,
options: RawUpgradeCommandOptions,
): Promise<void> {
if (options.verbose) {
this.logger = new CommandLogger({
@@ -132,93 +126,44 @@ export class UpgradeCommand extends CommandRunner {
}
try {
const versionContext = this.resolveVersionContext();
await this.runBootstrapMigrations();
await this.backfillWorkspaceCreatedIn1_21_0Cursors();
const sequence = this.upgradeSequenceReaderService.getUpgradeSequence();
this.logger.log(
chalk.blue(
[
'Initialized upgrade context with:',
`- currentVersion (migrating to): ${versionContext.currentAppVersion}`,
`- fromWorkspaceVersion: ${versionContext.fromWorkspaceVersion}`,
`- ${versionContext.fastInstanceCommands.length} fast instance commands (from registry)`,
`- ${versionContext.slowInstanceCommands.length} slow instance commands (from registry)`,
`- ${versionContext.workspaceCommands.length} workspace commands`,
'Initialized upgrade sequence:',
`- ${sequence.length} step(s)`,
...sequence.map(
(step, index) =>
` [${index}] ${step.kind}${step.name} (${step.version})`,
),
].join('\n '),
),
);
const workspacesBelowMinimumVersion =
await this.workspaceVersionService.getWorkspacesBelowVersion(
versionContext.fromWorkspaceVersion.version,
);
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.`,
);
}
await this.runLegacyPendingTypeOrmMigrations();
for (const { command, name } of versionContext.fastInstanceCommands) {
const result = await this.instanceUpgradeService.runFastInstanceCommand(
{
command,
name,
const { totalSuccesses, totalFailures } =
await this.upgradeSequenceRunnerService.run({
sequence,
options: {
...options,
workspaceIds: isDefined(options.workspaceId)
? Array.from(options.workspaceId)
: undefined,
},
);
if (result.status === 'failed') {
throw result.error;
}
}
const hasWorkspaces =
await this.workspaceVersionService.hasActiveOrSuspendedWorkspaces();
for (const { command, name } of versionContext.slowInstanceCommands) {
const result = await this.instanceUpgradeService.runSlowInstanceCommand(
{
command,
name,
skipDataMigration: !hasWorkspaces,
},
);
if (result.status === 'failed') {
throw result.error;
}
}
if (!hasWorkspaces) {
this.logger.log(
chalk.blue(
'Fresh installation detected, skipping workspace commands',
),
);
return;
}
const iteratorReport = await this.runWorkspaceCommands(
options,
versionContext,
);
});
this.logger.log(
chalk.blue(
`Upgrade summary: ${iteratorReport.success.length} succeeded, ${iteratorReport.fail.length} failed`,
`Upgrade summary: ${totalSuccesses} workspace(s) succeeded, ${totalFailures} workspace(s) failed`,
),
);
if (iteratorReport.fail.length > 0) {
if (totalFailures > 0) {
throw new Error(
`Upgrade completed with ${iteratorReport.fail.length} workspace failure(s)`,
`Upgrade completed with ${totalFailures} workspace failure(s)`,
);
}
} catch (error) {
@@ -227,70 +172,110 @@ Please roll back to that version and run the upgrade command again.`,
}
}
private async runLegacyPendingTypeOrmMigrations(): Promise<void> {
this.logger.log('Running legacy TypeORM migrations...');
// Workspaces created during 1.21 were activated before the cursor-based
// upgrade system existed. They have no upgradeMigration record yet.
// Stamp them with the last 1.21 workspace command as their initial cursor.
private async backfillWorkspaceCreatedIn1_21_0Cursors(): RemovedSinceVersion<
'1.23.0',
Promise<void>
> {
const allWorkspaceIds =
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds();
const migrations = await this.dataSource.runMigrations({
transaction: 'each',
});
if (allWorkspaceIds.length === 0) {
return;
}
if (migrations.length === 0) {
this.logger.log('No pending legacy migrations');
} else {
this.logger.log(
`Executed ${migrations.length} legacy migration(s): ${migrations.map((migration) => migration.name).join(', ')}`,
const existingCursorWorkspaceIds: { workspaceId: string }[] =
await this.dataSource.query(
`SELECT DISTINCT "workspaceId" FROM "core"."upgradeMigration" WHERE "workspaceId" IS NOT NULL`,
);
const existingCursorSet = new Set(
existingCursorWorkspaceIds.map((row) => row.workspaceId),
);
const workspacesWithoutCursor = allWorkspaceIds.filter(
(workspaceId) => !existingCursorSet.has(workspaceId),
);
if (workspacesWithoutCursor.length === 0) {
return;
}
const lastWorkspaceCommand =
this.upgradeCommandRegistryService.getLastWorkspaceCommandForVersion(
'1.21.0',
);
if (!lastWorkspaceCommand) {
throw new Error(
`Cannot backfill workspace cursors: no workspace commands found for version 1.21.0`,
);
}
this.logger.log(
chalk.blue(
`Backfilling initial cursor for ${workspacesWithoutCursor.length} workspace(s) → "${lastWorkspaceCommand.name}"`,
),
);
for (const workspaceId of workspacesWithoutCursor) {
await this.upgradeMigrationService.markAsInitial({
name: lastWorkspaceCommand.name,
workspaceId,
executedByVersion: '1.21.0',
});
}
}
private resolveVersionContext(): VersionContext {
const currentAppVersion = this.coreEngineVersionService.getCurrentVersion();
const currentVersionMajorMinor =
`${currentAppVersion.major}.${currentAppVersion.minor}.0` as UpgradeCommandVersion;
// Schema changes required by the upgrade engine itself (e.g. new columns
// on upgradeMigration) must be applied before the sequence runs.
private async runBootstrapMigrations(): RemovedSinceVersion<
'1.23.0',
Promise<void>
> {
const BOOTSTRAP_MIGRATION = 'AddIsInitialToUpgradeMigration1775909335324';
const fromWorkspaceVersion =
this.coreEngineVersionService.getPreviousVersion();
const alreadyExecuted = await this.dataSource.query(
`SELECT 1 FROM "core"."_typeorm_migrations" WHERE "name" = $1`,
[BOOTSTRAP_MIGRATION],
);
const { fastInstanceCommands, slowInstanceCommands, workspaceCommands } =
this.upgradeCommandRegistryService.getBundleForVersion(
currentVersionMajorMinor,
if (alreadyExecuted.length > 0) {
return;
}
const migration = this.dataSource.migrations.find(
(migration) => migration.name === BOOTSTRAP_MIGRATION,
);
if (!migration) {
throw new Error(
`Bootstrap migration "${BOOTSTRAP_MIGRATION}" not found in registered migrations`,
);
}
const queryRunner = this.dataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
await migration.up(queryRunner);
await queryRunner.query(
`INSERT INTO "core"."_typeorm_migrations" ("timestamp", "name") VALUES ($1, $2)`,
[1775909335324, BOOTSTRAP_MIGRATION],
);
return {
fromWorkspaceVersion,
currentAppVersion,
currentVersionMajorMinor,
fastInstanceCommands,
slowInstanceCommands,
workspaceCommands,
};
}
await queryRunner.commitTransaction();
} catch (error) {
await queryRunner.rollbackTransaction();
private async runWorkspaceCommands(
options: UpgradeCommandOptions,
{
currentAppVersion,
fromWorkspaceVersion,
workspaceCommands,
}: VersionContext,
) {
return 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.workspaceUpgradeService.upgradeWorkspace({
iteratorContext: context,
options,
fromWorkspaceVersion,
currentAppVersion,
workspaceCommands,
});
},
});
throw error;
} finally {
await queryRunner.release();
}
}
}
@@ -0,0 +1,17 @@
import { MigrationInterface, QueryRunner } from 'typeorm';
export class AddIsInitialToUpgradeMigration1775909335324
implements MigrationInterface
{
public async up(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(
`ALTER TABLE "core"."upgradeMigration" ADD "isInitial" boolean NOT NULL DEFAULT false`,
);
}
public async down(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(
`ALTER TABLE "core"."upgradeMigration" DROP COLUMN "isInitial"`,
);
}
}
@@ -1,12 +0,0 @@
// Keep at least two entries: the current version and the one before it.
// getPreviousVersion() looks up the entry just below the current version
// to determine the minimum workspace version eligible for upgrade.
// Removing the previous version would cause the upgrade command to fail.
export const UPGRADE_COMMAND_SUPPORTED_VERSIONS = [
'1.20.0',
'1.21.0',
'1.22.0',
] as const;
export type UpgradeCommandVersion =
(typeof UPGRADE_COMMAND_SUPPORTED_VERSIONS)[number];
@@ -1,9 +0,0 @@
import { Module } from '@nestjs/common';
import { CoreEngineVersionService } from 'src/engine/core-engine-version/services/core-engine-version.service';
@Module({
providers: [CoreEngineVersionService],
exports: [CoreEngineVersionService],
})
export class CoreEngineVersionModule {}
@@ -1,47 +0,0 @@
import { Injectable } from '@nestjs/common';
import { SemVer } from 'semver';
import { isDefined } from 'twenty-shared/utils';
import { UPGRADE_COMMAND_SUPPORTED_VERSIONS } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { getPreviousVersion } from 'src/utils/version/get-previous-version';
@Injectable()
export class CoreEngineVersionService {
constructor(private readonly twentyConfigService: TwentyConfigService) {}
getCurrentVersion(): SemVer {
const appVersion = this.twentyConfigService.get('APP_VERSION');
if (!isDefined(appVersion)) {
throw new Error(
'APP_VERSION is not defined, please double check your env variables',
);
}
try {
return new SemVer(appVersion);
} catch {
throw new Error(`APP_VERSION is not a valid semver: "${appVersion}"`);
}
}
getPreviousVersion(): SemVer {
const currentAppVersion = this.getCurrentVersion();
const currentVersionMajorMinor = `${currentAppVersion.major}.${currentAppVersion.minor}.0`;
const previousVersion = getPreviousVersion({
currentVersion: currentVersionMajorMinor,
versions: [...UPGRADE_COMMAND_SUPPORTED_VERSIONS],
});
if (!isDefined(previousVersion)) {
throw new Error(
`No previous version found for version ${currentAppVersion}. Available versions: ${UPGRADE_COMMAND_SUPPORTED_VERSIONS.join(', ')}`,
);
}
return previousVersion;
}
}
@@ -0,0 +1,11 @@
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
import { TWENTY_NEXT_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-next-versions.constant';
import { TWENTY_PREVIOUS_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-previous-versions.constant';
export const TWENTY_ALL_VERSIONS = [
...TWENTY_PREVIOUS_VERSIONS,
TWENTY_CURRENT_VERSION,
...TWENTY_NEXT_VERSIONS,
] as const;
export type TwentyAllVersion = (typeof TWENTY_ALL_VERSIONS)[number];
@@ -0,0 +1,10 @@
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
import { TWENTY_PREVIOUS_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-previous-versions.constant';
export const TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS = [
...TWENTY_PREVIOUS_VERSIONS,
TWENTY_CURRENT_VERSION,
] as const;
export type TwentyCrossUpgradeSupportedVersion =
(typeof TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS)[number];
@@ -0,0 +1 @@
export const TWENTY_CURRENT_VERSION = '1.22.0' as const;
@@ -0,0 +1 @@
export const TWENTY_NEXT_VERSIONS = ['1.23.0'] as const;
@@ -0,0 +1 @@
export const TWENTY_PREVIOUS_VERSIONS = ['1.20.0', '1.21.0'] as const;
@@ -2,24 +2,24 @@ import 'reflect-metadata';
import { Injectable } from '@nestjs/common';
import { type UpgradeCommandVersion } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { type TwentyAllVersion } from 'src/engine/core-modules/upgrade/constants/twenty-all-versions.constant';
export type InstanceCommandType = 'fast' | 'slow';
export type RegisteredInstanceCommandMetadata = {
version: UpgradeCommandVersion;
version: TwentyAllVersion;
timestamp: number;
type: InstanceCommandType;
};
const REGISTERED_INSTANCE_COMMAND_KEY = 'REGISTERED_INSTANCE_COMMAND';
// When dropping a version from UPGRADE_COMMAND_SUPPORTED_VERSIONS, also
// When dropping a version from TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS, also
// remove the @RegisteredInstanceCommand decorator from its associated
// command files.
export const RegisteredInstanceCommand =
(
version: UpgradeCommandVersion,
version: TwentyAllVersion,
timestamp: number,
options?: { type: 'slow' },
): ClassDecorator =>
@@ -1,16 +1,19 @@
import 'reflect-metadata';
import { type UpgradeCommandVersion } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { type TwentyCrossUpgradeSupportedVersion } from 'src/engine/core-modules/upgrade/constants/twenty-cross-upgrade-supported-version.constant';
export type RegisteredWorkspaceCommandMetadata = {
version: UpgradeCommandVersion;
version: TwentyCrossUpgradeSupportedVersion;
timestamp: number;
};
const REGISTERED_WORKSPACE_COMMAND_KEY = 'REGISTERED_WORKSPACE_COMMAND';
export const RegisteredWorkspaceCommand =
(version: UpgradeCommandVersion, timestamp: number): ClassDecorator =>
(
version: TwentyCrossUpgradeSupportedVersion,
timestamp: number,
): ClassDecorator =>
(target) => {
Reflect.defineMetadata(
REGISTERED_WORKSPACE_COMMAND_KEY,
@@ -0,0 +1,369 @@
import {
type UpgradeStep,
type WorkspaceUpgradeStep,
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import {
type IntegrationTestContext,
createUpgradeSequenceRunnerIntegrationTestModule,
DEFAULT_OPTIONS,
makeFastInstance,
makeSlowInstance,
makeStep,
makeWorkspace,
resetSeedSequenceCounter,
seedMigration,
setMockActiveWorkspaceIds,
testGetLatestMigrationForCommand,
WS_1,
WS_2,
} from './utils/upgrade-sequence-runner-integration-test.util';
const makeFailingFastInstance = (name: string, error: Error): UpgradeStep =>
({
...makeStep('fast-instance', name),
command: {
up: async () => {
throw error;
},
down: async () => {},
},
}) as unknown as UpgradeStep;
const makeFailingWorkspace = (
name: string,
error: Error,
): WorkspaceUpgradeStep =>
({
...makeStep('workspace', name),
command: {
runOnWorkspace: async () => {
throw error;
},
},
}) as unknown as WorkspaceUpgradeStep;
const makeWorkspaceFailingForIds = (
name: string,
failingWorkspaceIds: Set<string>,
error: Error,
): WorkspaceUpgradeStep =>
({
...makeStep('workspace', name),
command: {
runOnWorkspace: async ({ workspaceId }: { workspaceId: string }) => {
if (failingWorkspaceIds.has(workspaceId)) {
throw error;
}
},
},
}) as unknown as WorkspaceUpgradeStep;
describe('UpgradeSequenceRunnerService — failing sequence (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 throw when no migration history exists', async () => {
const sequence = [makeFastInstance('Ic1')];
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow(
'No upgrade migration found — the database may not have been initialized',
);
});
it('should throw when cursor command is not found in the sequence', async () => {
const sequence = [makeFastInstance('Ic1'), makeFastInstance('Ic2')];
await seedMigration(context.dataSource, {
name: 'RemovedCommand',
status: 'completed',
});
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow('Step "RemovedCommand" not found in upgrade sequence');
});
it('should throw when workspace cursors are outside the current slice', async () => {
const sequence = [
makeWorkspace('Wc1'),
makeWorkspace('Wc2'),
makeFastInstance('Ic1'),
makeWorkspace('Wc3'),
makeWorkspace('Wc4'),
];
setMockActiveWorkspaceIds([WS_1, WS_2]);
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc2',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_2,
});
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc3',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc4',
status: 'completed',
workspaceId: WS_1,
});
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow('workspaces are not aligned');
});
it('should throw when an active workspace has no migration history', async () => {
const sequence = [makeFastInstance('Ic1'), makeWorkspace('Wc1')];
setMockActiveWorkspaceIds([WS_1, WS_2]);
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow('No upgrade migration found for workspace(s)');
});
it('should record failure in DB when a fast instance command fails', async () => {
const error = new Error('fast command exploded');
const sequence = [
makeFastInstance('Ic1'),
makeFailingFastInstance('Ic2', error),
];
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow('fast command exploded');
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'failed' }),
);
});
it('should record failure in DB when a slow instance command fails', async () => {
const error = new Error('slow data migration exploded');
const sequence = [
makeFastInstance('Ic1'),
{
...makeSlowInstance('Ic2'),
command: {
up: async () => {},
down: async () => {},
runDataMigration: async () => {
throw error;
},
},
} as unknown as UpgradeStep,
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await expect(
context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
}),
).rejects.toThrow('slow data migration exploded');
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'failed' }),
);
});
it('should abort and report failures when workspace commands fail, without running subsequent instance steps', async () => {
const error = new Error('workspace command exploded');
const sequence = [
makeFailingWorkspace('Wc1', error),
makeFastInstance('Ic1'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'failed',
workspaceId: WS_1,
});
const report = await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(report.totalFailures).toBe(1);
expect(report.totalSuccesses).toBe(0);
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic1',
});
expect(ic1).toBeNull();
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc1',
workspaceId: WS_1,
});
expect(wc1).toEqual(
expect.objectContaining({ status: 'failed', attempt: 2 }),
);
});
it('should abort at workspace failure in a multi-segment sequence with two workspaces starting aligned', async () => {
const error = new Error('Wc2 exploded for WS_2');
const sequence = [
makeWorkspace('Wc0'),
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeWorkspaceFailingForIds('Wc2', new Set([WS_2]), error),
makeFastInstance('Ic2'),
makeWorkspace('Wc3'),
];
setMockActiveWorkspaceIds([WS_1, WS_2]);
await seedMigration(context.dataSource, {
name: 'Wc0',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc0',
status: 'completed',
workspaceId: WS_2,
});
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_2,
});
const report = await context.runner.run({
sequence,
options: {
...DEFAULT_OPTIONS,
workspaceIds: [WS_1, WS_2],
},
});
expect(report.totalSuccesses).toBe(1);
expect(report.totalFailures).toBe(1);
// WS_1 succeeded Wc2
const ws1Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
expect(ws1Wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
// WS_2 failed Wc2
const ws2Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_2,
});
expect(ws2Wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'failed' }),
);
// Ic2 never ran — runner aborted at the workspace segment failure
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
expect(ic2).toBeNull();
// Wc3 never ran either
const ws1Wc3 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc3',
workspaceId: WS_1,
});
expect(ws1Wc3).toBeNull();
});
});
@@ -0,0 +1,466 @@
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
import {
type IntegrationTestContext,
createUpgradeSequenceRunnerIntegrationTestModule,
DEFAULT_OPTIONS,
makeFastInstance,
makeSlowInstance,
makeWorkspace,
resetSeedSequenceCounter,
seedMigration,
setMockActiveWorkspaceIds,
testGetLatestMigrationForCommand,
WS_1,
WS_2,
} from './utils/upgrade-sequence-runner-integration-test.util';
describe('UpgradeSequenceRunnerService — execution (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 return zero counts for an empty sequence', async () => {
const report = await context.runner.run({
sequence: [],
options: DEFAULT_OPTIONS,
});
expect(report).toEqual({ totalSuccesses: 0, totalFailures: 0 });
});
it('should resume from a completed instance command and run remaining steps', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeFastInstance('Ic2'),
makeSlowInstance('Ic3'),
];
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Ic2',
status: 'completed',
});
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic1',
});
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
const ic3 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic3',
});
expect(ic1).toEqual(
expect.objectContaining({ status: 'completed', attempt: 1 }),
);
expect(ic2).toEqual(
expect.objectContaining({ status: 'completed', attempt: 1 }),
);
expect(ic3).toEqual(expect.objectContaining({ status: 'completed' }));
});
it('should retry a failed instance command', async () => {
const sequence = [makeFastInstance('Ic1'), makeFastInstance('Ic2')];
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Ic2',
status: 'failed',
});
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'completed', attempt: 2 }),
);
});
it('should resume workspace commands from per-workspace cursors', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeWorkspace('Wc2'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
expect(wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
});
it('should enforce workspace sync barrier before instance step', async () => {
const sequence = [makeWorkspace('Wc1'), makeFastInstance('Ic1')];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic1',
});
expect(ic1).toEqual(
expect.objectContaining({ name: 'Ic1', status: 'completed' }),
);
});
it('should skip data migration for slow instance commands when no workspaces exist', async () => {
const sequence = [makeFastInstance('Ic1'), makeSlowInstance('Ic2')];
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
const instanceCommandRunnerService = context.module.get(
InstanceCommandRunnerService,
);
const spy = jest.spyOn(
instanceCommandRunnerService,
'runSlowInstanceCommand',
);
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(spy).toHaveBeenCalledWith(
expect.objectContaining({ skipDataMigration: true }),
);
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
);
});
it('should run data migration for slow instance commands when workspaces exist', async () => {
const sequence = [makeFastInstance('Ic0'), makeSlowInstance('Ic1')];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic0',
status: 'completed',
});
const instanceCommandRunnerService = context.module.get(
InstanceCommandRunnerService,
);
const spy = jest.spyOn(
instanceCommandRunnerService,
'runSlowInstanceCommand',
);
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(spy).toHaveBeenCalledWith(
expect.objectContaining({ skipDataMigration: false }),
);
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic1',
});
expect(ic1).toEqual(
expect.objectContaining({ name: 'Ic1', status: 'completed' }),
);
});
it('should run workspace commands for multiple workspaces successfully', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeWorkspace('Wc2'),
];
setMockActiveWorkspaceIds([WS_1, WS_2]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_2,
});
const report = await context.runner.run({
sequence,
options: {
...DEFAULT_OPTIONS,
workspaceIds: [WS_1, WS_2],
},
});
expect(report).toEqual({ totalSuccesses: 2, totalFailures: 0 });
const ws1Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
const ws2Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_2,
});
expect(ws1Wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
expect(ws2Wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
});
it('should execute the full sequence from the initial cursor on a fresh run', async () => {
const sequence = [
makeWorkspace('Wc0'),
makeFastInstance('Ic1'),
makeFastInstance('Ic2'),
makeWorkspace('Wc1'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Wc0',
status: 'completed',
workspaceId: WS_1,
});
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc1',
workspaceId: WS_1,
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
);
expect(wc1).toEqual(
expect.objectContaining({ name: 'Wc1', status: 'completed' }),
);
});
it('should retry a failed workspace command', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeWorkspace('Wc2'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'failed',
workspaceId: WS_1,
});
const report = await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(report.totalFailures).toBe(0);
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc1',
workspaceId: WS_1,
});
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
expect(wc1).toEqual(
expect.objectContaining({
name: 'Wc1',
status: 'completed',
attempt: 2,
}),
);
expect(wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
});
it('should traverse a multi-segment sequence with sync barriers', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeFastInstance('Ic2'),
makeWorkspace('Wc2'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
const report = await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(report.totalFailures).toBe(0);
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Ic2',
});
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
expect(ic2).toEqual(
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
);
expect(wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
});
it('should ignore migration records from inactive workspaces when resolving the global cursor', async () => {
const sequence = [
makeFastInstance('Ic1'),
makeWorkspace('Wc1'),
makeWorkspace('Wc2'),
];
setMockActiveWorkspaceIds([WS_1]);
await seedMigration(context.dataSource, {
name: 'Ic1',
status: 'completed',
});
await seedMigration(context.dataSource, {
name: 'Wc1',
status: 'completed',
workspaceId: WS_1,
});
// WS_2 is inactive — its record is more recent (seeded later)
// but should not influence the global cursor
await seedMigration(context.dataSource, {
name: 'Wc2',
status: 'completed',
workspaceId: WS_2,
});
const report = await context.runner.run({
sequence,
options: DEFAULT_OPTIONS,
});
expect(report.totalFailures).toBe(0);
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
name: 'Wc2',
workspaceId: WS_1,
});
expect(wc2).toEqual(
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
);
});
});
@@ -95,6 +95,7 @@ describe('UpgradeCommandRegistryService', () => {
new MigrationA1770000000000(),
new MigrationB1771000000000(),
new MigrationC1772000000000(),
new WorkspaceCommandA(),
]);
const v120 = service.getBundleForVersion('1.20.0');
@@ -118,6 +119,7 @@ describe('UpgradeCommandRegistryService', () => {
new MigrationC1772000000000(),
new MigrationA1770000000000(),
new MigrationB1771000000000(),
new WorkspaceCommandA(),
]);
const names = service
@@ -135,6 +137,7 @@ describe('UpgradeCommandRegistryService', () => {
const service = await buildRegistryService([
new UndecoratedMigration1768000000000(),
new MigrationA1770000000000(),
new WorkspaceCommandA(),
]);
const v121 = service.getBundleForVersion('1.21.0');
@@ -157,6 +160,10 @@ describe('UpgradeCommandRegistryService', () => {
expect(v121.workspaceCommands).toStrictEqual([]);
});
it('should not throw when no commands are discovered (empty bundle)', async () => {
await expect(buildRegistryService([])).resolves.toBeDefined();
});
it('should return empty array for unsupported version', async () => {
const service = await buildRegistryService([]);
@@ -255,9 +262,10 @@ describe('UpgradeCommandRegistryService', () => {
new MigrationD1769000000000(),
new MigrationA1770000000000(),
new MigrationB1771000000000(),
new WorkspaceCommandA(),
]);
const allCommands = service.getAllFastInstanceCommands();
const allCommands = service.getCrossUpgradeSupportedFastInstanceCommands();
expect(allCommands.map((entry) => entry.name)).toStrictEqual([
'1.20.0_MigrationD1769000000000_1769000000000',
@@ -267,10 +275,12 @@ describe('UpgradeCommandRegistryService', () => {
]);
});
it('should return empty array from getAllFastInstanceCommands when no commands registered', async () => {
it('should return empty array from getCrossUpgradeSupportedFastInstanceCommands when no commands registered', async () => {
const service = await buildRegistryService([]);
expect(service.getAllFastInstanceCommands()).toStrictEqual([]);
expect(
service.getCrossUpgradeSupportedFastInstanceCommands(),
).toStrictEqual([]);
});
it('should allow same class name with different timestamps across kinds', async () => {
@@ -316,6 +326,7 @@ describe('UpgradeCommandRegistryService', () => {
const service = await buildRegistryService([
new SlowMigrationB1780000000000(),
new SlowMigrationA1779000000000(),
new WorkspaceCommandA(),
]);
const { slowInstanceCommands } = service.getBundleForVersion('1.21.0');
@@ -341,6 +352,7 @@ describe('UpgradeCommandRegistryService', () => {
const service = await buildRegistryService([
new MigrationA1770000000000(),
new SlowMigration1780000000000(),
new WorkspaceCommandA(),
]);
const bucket = service.getBundleForVersion('1.21.0');
@@ -391,6 +403,7 @@ describe('UpgradeCommandRegistryService', () => {
const service = await buildRegistryService([
new MigrationA1770000000000(),
new SlowMigrationSameTimestamp(),
new WorkspaceCommandA(),
]);
const bucket = service.getBundleForVersion('1.21.0');
@@ -421,9 +434,11 @@ describe('UpgradeCommandRegistryService', () => {
const service = await buildRegistryService([
new SlowMigration1780000000000(),
new SlowMigration1768000000000(),
new WorkspaceCommandA(),
]);
const allSlowCommands = service.getAllSlowInstanceCommands();
const allSlowCommands =
service.getCrossUpgradeSupportedSlowInstanceCommands();
expect(allSlowCommands.map((entry) => entry.name)).toStrictEqual([
'1.20.0_SlowMigration1768000000000_1768000000000',
@@ -0,0 +1,275 @@
import { Test, type TestingModule } from '@nestjs/testing';
import { getDataSourceToken, getRepositoryToken } from '@nestjs/typeorm';
import { config } from 'dotenv';
import { DataSource, type Repository } from 'typeorm';
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import {
UpgradeSequenceReaderService,
type UpgradeStep,
type WorkspaceUpgradeStep,
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { UpgradeSequenceRunnerService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service';
import { WorkspaceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/workspace-command-runner.service';
import { UpgradeMigrationEntity } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
import {
SEED_APPLE_WORKSPACE_ID,
SEED_YCOMBINATOR_WORKSPACE_ID,
} from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
jest.useRealTimers();
config({
path: process.env.NODE_ENV === 'test' ? '.env.test' : '.env',
override: true,
});
export const WS_1 = SEED_APPLE_WORKSPACE_ID;
export const WS_2 = SEED_YCOMBINATOR_WORKSPACE_ID;
const EXECUTED_BY_VERSION = '42.42.42';
const noopAsync = async () => {};
export const makeStep = (
kind: UpgradeStep['kind'],
name: string,
): UpgradeStep => {
const command =
kind === 'workspace'
? { runOnWorkspace: noopAsync }
: kind === 'slow-instance'
? { up: noopAsync, down: noopAsync, runDataMigration: noopAsync }
: { up: noopAsync, down: noopAsync };
return {
kind,
name,
command,
version: '1.21.0',
timestamp: 0,
} as unknown as UpgradeStep;
};
export const makeFastInstance = (name: string) =>
makeStep('fast-instance', name);
export const makeSlowInstance = (name: string) =>
makeStep('slow-instance', name);
export const makeWorkspace = (name: string) =>
makeStep('workspace', name) as WorkspaceUpgradeStep;
let mockActiveWorkspaceIds: string[] = [];
export const setMockActiveWorkspaceIds = (ids: string[]) => {
mockActiveWorkspaceIds = ids;
};
export const DEFAULT_OPTIONS = {
workspaceIds: undefined,
startFromWorkspaceId: undefined,
workspaceCountLimit: undefined,
dryRun: false,
verbose: false,
};
type IntegrationTestModule = Awaited<
ReturnType<typeof createUpgradeSequenceRunnerIntegrationTestModule>
>;
export type IntegrationTestContext = {
[K in keyof IntegrationTestModule]: IntegrationTestModule[K];
};
export const createUpgradeSequenceRunnerIntegrationTestModule = async () => {
const dataSource = new DataSource({
type: 'postgres',
url: process.env.PG_DATABASE_URL,
schema: 'core',
entities: [
'src/engine/core-modules/**/*.entity.ts',
'src/engine/metadata-modules/**/*.entity.ts',
],
synchronize: false,
});
await dataSource.initialize();
const migrationRepo: Repository<UpgradeMigrationEntity> =
dataSource.getRepository(UpgradeMigrationEntity);
const module: TestingModule = await Test.createTestingModule({
providers: [
{
provide: getRepositoryToken(UpgradeMigrationEntity),
useValue: migrationRepo,
},
{
provide: getDataSourceToken(),
useValue: dataSource,
},
{
provide: TwentyConfigService,
useValue: {
get: (key: string) =>
key === 'APP_VERSION' ? EXECUTED_BY_VERSION : undefined,
},
},
UpgradeMigrationService,
{
provide: WorkspaceVersionService,
useValue: {
getActiveOrSuspendedWorkspaceIds: jest
.fn()
.mockImplementation(async () => mockActiveWorkspaceIds),
hasActiveOrSuspendedWorkspaces: jest
.fn()
.mockImplementation(async () => mockActiveWorkspaceIds.length > 0),
},
},
{
provide: UpgradeSequenceReaderService,
useFactory: () => new UpgradeSequenceReaderService({} as any),
},
InstanceCommandRunnerService,
WorkspaceCommandRunnerService,
{
provide: WorkspaceIteratorService,
useValue: {
iterate: jest.fn().mockImplementation(async (args: any) => {
const { callback, workspaceIds } = args;
const ids = workspaceIds ?? [WS_1];
const report = { fail: [] as any[], success: [] as any[] };
for (const [index, workspaceId] of ids.entries()) {
try {
await callback({
workspaceId,
index,
total: ids.length,
dataSource,
});
report.success.push({ workspaceId });
} catch (error) {
report.fail.push({ error, workspaceId });
}
}
return report;
}),
},
},
UpgradeSequenceRunnerService,
],
}).compile();
const runner = module.get(UpgradeSequenceRunnerService);
jest.spyOn(runner['logger'], 'log').mockImplementation();
jest.spyOn(runner['logger'], 'error').mockImplementation();
jest.spyOn(runner['logger'], 'warn').mockImplementation();
const instanceCommandRunnerService = module.get(InstanceCommandRunnerService);
jest
.spyOn(instanceCommandRunnerService['logger'], 'log')
.mockImplementation();
jest
.spyOn(instanceCommandRunnerService['logger'], 'error')
.mockImplementation();
const workspaceCommandRunnerService = module.get(
WorkspaceCommandRunnerService,
);
jest
.spyOn(workspaceCommandRunnerService['logger'], 'log')
.mockImplementation();
jest
.spyOn(workspaceCommandRunnerService['logger'], 'error')
.mockImplementation();
return {
module,
dataSource,
runner,
};
};
let seedSequenceCounter = 0;
export const resetSeedSequenceCounter = () => {
seedSequenceCounter = 0;
};
export const seedMigration = async (
dataSource: DataSource,
{
name,
status,
workspaceId = null,
attempt = 1,
}: {
name: string;
status: 'completed' | 'failed';
workspaceId?: string | null;
attempt?: number;
},
) => {
const createdAt = new Date(
Date.now() + seedSequenceCounter * 1000,
).toISOString();
seedSequenceCounter++;
await dataSource.query(
`INSERT INTO core."upgradeMigration" (name, status, attempt, "executedByVersion", "workspaceId", "createdAt")
VALUES ($1, $2, $3, $4, $5, $6)`,
[name, status, attempt, EXECUTED_BY_VERSION, workspaceId, createdAt],
);
};
export const testCountMigrationsForCommand = async (
dataSource: DataSource,
{
name,
workspaceId = null,
}: {
name: string;
workspaceId?: string | null;
},
): Promise<number> => {
const rows = await dataSource.query(
`SELECT COUNT(*)::int AS count FROM core."upgradeMigration"
WHERE name = $1 AND ($2::uuid IS NULL AND "workspaceId" IS NULL OR "workspaceId" = $2)`,
[name, workspaceId],
);
return rows[0].count;
};
export const testGetLatestMigrationForCommand = async (
dataSource: DataSource,
{
name,
workspaceId = null,
}: {
name: string;
workspaceId?: string | null;
},
): Promise<{ name: string; status: string; attempt: number } | null> => {
const rows = await dataSource.query(
`SELECT name, status, attempt FROM core."upgradeMigration"
WHERE name = $1 AND ($2::uuid IS NULL AND "workspaceId" IS NULL OR "workspaceId" = $2)
ORDER BY attempt DESC LIMIT 1`,
[name, workspaceId],
);
return rows.length > 0 ? rows[0] : null;
};
@@ -14,8 +14,8 @@ type RunSingleMigrationResult =
| { status: 'failed'; error: unknown };
@Injectable()
export class InstanceUpgradeService {
private readonly logger = new Logger(InstanceUpgradeService.name);
export class InstanceCommandRunnerService {
private readonly logger = new Logger(InstanceCommandRunnerService.name);
constructor(
@InjectDataSource()
@@ -3,14 +3,18 @@ import { DiscoveryService } from '@nestjs/core';
import { type ActiveOrSuspendedWorkspaceCommandRunner } from 'src/database/commands/command-runners/active-or-suspended-workspace.command-runner';
import { type WorkspaceCommandRunner } from 'src/database/commands/command-runners/workspace.command-runner';
import {
TWENTY_ALL_VERSIONS,
type TwentyAllVersion,
} from 'src/engine/core-modules/upgrade/constants/twenty-all-versions.constant';
import { TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-cross-upgrade-supported-version.constant';
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
import { TWENTY_NEXT_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-next-versions.constant';
import { TWENTY_PREVIOUS_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-previous-versions.constant';
import { getRegisteredInstanceCommandMetadata } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
import { getRegisteredWorkspaceCommandMetadata } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator';
import { type FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface';
import { type SlowInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/slow-instance-command.interface';
import {
UPGRADE_COMMAND_SUPPORTED_VERSIONS,
type UpgradeCommandVersion,
} from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { isDefined } from 'twenty-shared/utils';
type WorkspaceCommand =
@@ -20,22 +24,25 @@ type WorkspaceCommand =
export type RegisteredFastInstanceCommand = {
name: string;
command: FastInstanceCommand;
version: TwentyAllVersion;
timestamp: number;
};
export type RegisteredSlowInstanceCommand = {
name: string;
command: SlowInstanceCommand;
version: TwentyAllVersion;
timestamp: number;
};
export type RegisteredWorkspaceCommand = {
name: string;
command: WorkspaceCommand;
version: TwentyAllVersion;
timestamp: number;
};
export type VersionBundle = {
type VersionBundle = {
fastInstanceCommands: RegisteredFastInstanceCommand[];
slowInstanceCommands: RegisteredSlowInstanceCommand[];
workspaceCommands: RegisteredWorkspaceCommand[];
@@ -52,14 +59,14 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
private readonly logger = new Logger(UpgradeCommandRegistryService.name);
private readonly bundlesByVersion = new Map<
UpgradeCommandVersion,
TwentyAllVersion,
VersionBundle
>();
constructor(private readonly discoveryService: DiscoveryService) {}
onModuleInit(): void {
for (const version of UPGRADE_COMMAND_SUPPORTED_VERSIONS) {
for (const version of TWENTY_ALL_VERSIONS) {
this.bundlesByVersion.set(version, {
fastInstanceCommands: [],
slowInstanceCommands: [],
@@ -84,27 +91,30 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
instanceCommandMetadata.version,
);
if (isDefined(bundle)) {
const entry = {
name: this.computeCommandName(
instanceCommandMetadata.version,
(instance as FastInstanceCommand).constructor.name,
instanceCommandMetadata.timestamp,
),
timestamp: instanceCommandMetadata.timestamp,
};
if (!isDefined(bundle)) {
continue;
}
if (instanceCommandMetadata.type === 'slow') {
bundle.slowInstanceCommands.push({
...entry,
command: instance as SlowInstanceCommand,
});
} else {
bundle.fastInstanceCommands.push({
...entry,
command: instance as FastInstanceCommand,
});
}
const entry = {
name: this.computeCommandName(
instanceCommandMetadata.version,
(instance as FastInstanceCommand).constructor.name,
instanceCommandMetadata.timestamp,
),
version: instanceCommandMetadata.version,
timestamp: instanceCommandMetadata.timestamp,
};
if (instanceCommandMetadata.type === 'slow') {
bundle.slowInstanceCommands.push({
...entry,
command: instance as SlowInstanceCommand,
});
} else {
bundle.fastInstanceCommands.push({
...entry,
command: instance as FastInstanceCommand,
});
}
continue;
@@ -118,17 +128,20 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
workspaceCommandMetadata.version,
);
if (isDefined(bundle)) {
bundle.workspaceCommands.push({
name: this.computeCommandName(
workspaceCommandMetadata.version,
(instance as WorkspaceCommand).constructor.name,
workspaceCommandMetadata.timestamp,
),
command: instance as WorkspaceCommand,
timestamp: workspaceCommandMetadata.timestamp,
});
if (!isDefined(bundle)) {
continue;
}
bundle.workspaceCommands.push({
name: this.computeCommandName(
workspaceCommandMetadata.version,
(instance as WorkspaceCommand).constructor.name,
workspaceCommandMetadata.timestamp,
),
command: instance as WorkspaceCommand,
version: workspaceCommandMetadata.version,
timestamp: workspaceCommandMetadata.timestamp,
});
}
}
@@ -144,7 +157,10 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
);
}
this.validateNoVersionDuplicatesAcrossConstants();
this.validatePreviousVersionsNotEmpty();
this.validateNoDuplicates();
this.validateAtLeastOneVersionBundleHasWorkspaceCommands();
for (const [version, bundle] of this.bundlesByVersion) {
const totalCount =
@@ -160,24 +176,32 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
}
}
getBundleForVersion(version: UpgradeCommandVersion): VersionBundle {
getBundleForVersion(version: TwentyAllVersion): VersionBundle {
return this.bundlesByVersion.get(version) ?? buildEmptyVersionBundle();
}
getAllFastInstanceCommands(): RegisteredFastInstanceCommand[] {
return UPGRADE_COMMAND_SUPPORTED_VERSIONS.flatMap(
getLastWorkspaceCommandForVersion(
version: TwentyAllVersion,
): RegisteredWorkspaceCommand | undefined {
const bundle = this.getBundleForVersion(version);
return bundle.workspaceCommands[bundle.workspaceCommands.length - 1];
}
getCrossUpgradeSupportedFastInstanceCommands(): RegisteredFastInstanceCommand[] {
return TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS.flatMap(
(version) => this.getBundleForVersion(version).fastInstanceCommands,
);
}
getAllSlowInstanceCommands(): RegisteredSlowInstanceCommand[] {
return UPGRADE_COMMAND_SUPPORTED_VERSIONS.flatMap(
getCrossUpgradeSupportedSlowInstanceCommands(): RegisteredSlowInstanceCommand[] {
return TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS.flatMap(
(version) => this.getBundleForVersion(version).slowInstanceCommands,
);
}
private computeCommandName(
version: UpgradeCommandVersion,
version: TwentyAllVersion,
className: string,
timestamp: number,
): string {
@@ -222,8 +246,43 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
}
}
private validateAtLeastOneVersionBundleHasWorkspaceCommands(): void {
let totalCommandCount = 0;
let hasWorkspaceCommands = false;
for (const version of TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS) {
const bundle = this.getBundleForVersion(version);
totalCommandCount +=
bundle.fastInstanceCommands.length +
bundle.slowInstanceCommands.length +
bundle.workspaceCommands.length;
if (bundle.workspaceCommands.length > 0) {
hasWorkspaceCommands = true;
}
}
// UpgradeModule is loaded in the worker transitively via WorkspaceModule,
// but no command modules are imported — zero providers are discovered.
// TODO: split WorkspaceModule so the worker doesn't pull in UpgradeModule
if (totalCommandCount === 0) {
this.logger.warn(
'No upgrade commands discovered — skipping workspace command validation',
);
return;
}
if (!hasWorkspaceCommands) {
throw new Error(
'Upgrade sequence must contain at least one workspace command',
);
}
}
private validateNoTimestampDuplicatesWithinKind(
version: UpgradeCommandVersion,
version: TwentyAllVersion,
kind: 'fast-instance' | 'slow-instance' | 'workspace',
entries:
| RegisteredFastInstanceCommand[]
@@ -242,4 +301,32 @@ export class UpgradeCommandRegistryService implements OnModuleInit {
seenTimestamps.add(entry.timestamp);
}
}
private validateNoVersionDuplicatesAcrossConstants(): void {
const allVersions = [
...TWENTY_PREVIOUS_VERSIONS,
TWENTY_CURRENT_VERSION,
...TWENTY_NEXT_VERSIONS,
];
const uniqueVersions = new Set(allVersions);
if (uniqueVersions.size !== allVersions.length) {
const duplicates = allVersions.filter(
(version, index) => allVersions.indexOf(version) !== index,
);
throw new Error(
`Duplicate version(s) across TWENTY_PREVIOUS_VERSIONS, TWENTY_CURRENT_VERSION, and TWENTY_NEXT_VERSIONS: ${duplicates.join(', ')}`,
);
}
}
private validatePreviousVersionsNotEmpty(): void {
if ((TWENTY_PREVIOUS_VERSIONS as readonly string[]).length === 0) {
throw new Error(
'TWENTY_PREVIOUS_VERSIONS must contain at least one version before TWENTY_CURRENT_VERSION',
);
}
}
}
@@ -2,9 +2,12 @@ import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { isDefined } from 'twenty-shared/utils';
import { IsNull, type QueryRunner, Repository } from 'typeorm';
import { In, IsNull, type QueryRunner, Repository } from 'typeorm';
import { UpgradeMigrationEntity } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
import {
UpgradeMigrationEntity,
type UpgradeMigrationStatus,
} from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
import { formatUpgradeErrorForStorage } from 'src/engine/core-modules/upgrade/utils/format-upgrade-error-for-storage.util';
@Injectable()
@@ -89,4 +92,161 @@ export class UpgradeMigrationService {
errorMessage: formatUpgradeErrorForStorage(error),
});
}
async markAsInitial({
name,
workspaceId,
executedByVersion,
queryRunner,
}: {
name: string;
workspaceId: string;
executedByVersion: string;
queryRunner?: QueryRunner;
}): Promise<void> {
const repository = queryRunner
? queryRunner.manager.getRepository(UpgradeMigrationEntity)
: this.upgradeMigrationRepository;
await repository.save({
name,
status: 'completed',
isInitial: true,
attempt: 1,
executedByVersion,
workspaceId,
});
}
// Returns the most recently attempted command (by createdAt)
// across instance and active-workspace scopes, with its status.
// Workspace-scoped records from inactive/deleted workspaces are
// excluded so they cannot incorrectly influence the global cursor.
async getLastAttemptedCommandNameOrThrow(
allActiveOrSuspendedWorkspaceIds: string[],
): Promise<{
name: string;
status: UpgradeMigrationStatus;
}> {
const queryBuilder = this.upgradeMigrationRepository
.createQueryBuilder('migration')
.select(['migration.name', 'migration.status'])
.andWhere(
`migration.attempt = (
SELECT MAX(sub.attempt)
FROM core."upgradeMigration" sub
WHERE sub.name = migration.name
AND (
(sub."workspaceId" IS NULL AND migration."workspaceId" IS NULL)
OR sub."workspaceId" = migration."workspaceId"
)
)`,
);
if (allActiveOrSuspendedWorkspaceIds.length > 0) {
queryBuilder.andWhere(
'(migration."workspaceId" IS NULL OR migration."workspaceId" IN (:...allActiveOrSuspendedWorkspaceIds))',
{ allActiveOrSuspendedWorkspaceIds },
);
} else {
queryBuilder.andWhere('migration."workspaceId" IS NULL');
}
const migration = await queryBuilder
.orderBy('migration.createdAt', 'DESC')
.getOne();
if (!migration) {
throw new Error(
'No upgrade migration found — the database may not have been initialized',
);
}
return { name: migration.name, status: migration.status };
}
async getWorkspaceLastAttemptedCommandNameOrThrow(
workspaceIds: string[],
): Promise<Map<string, { name: string; status: UpgradeMigrationStatus }>> {
if (workspaceIds.length === 0) {
return new Map();
}
const results = await this.upgradeMigrationRepository
.createQueryBuilder('migration')
.select('migration.workspaceId', 'workspaceId')
.addSelect('migration.name', 'name')
.addSelect('migration.status', 'status')
.where({
workspaceId: In(workspaceIds),
})
.andWhere(
`migration.attempt = (
SELECT MAX(sub.attempt)
FROM core."upgradeMigration" sub
WHERE sub.name = migration.name
AND sub."workspaceId" = migration."workspaceId"
)`,
)
.orderBy('migration.workspaceId')
.addOrderBy('migration.createdAt', 'DESC')
.distinctOn(['migration.workspaceId'])
.getRawMany<{
workspaceId: string;
name: string;
status: UpgradeMigrationStatus;
}>();
const cursors = new Map<
string,
{ name: string; status: UpgradeMigrationStatus }
>();
for (const row of results) {
cursors.set(row.workspaceId, { name: row.name, status: row.status });
}
const missingWorkspaceIds = workspaceIds.filter(
(workspaceId) => !cursors.has(workspaceId),
);
if (missingWorkspaceIds.length > 0) {
throw new Error(
`No upgrade migration found for workspace(s): ${missingWorkspaceIds.join(', ')}`,
);
}
return cursors;
}
async areAllWorkspacesAtCommand({
commandName,
workspaceIds,
}: {
commandName: string;
workspaceIds: string[];
}): Promise<boolean> {
if (workspaceIds.length === 0) {
return true;
}
const completedCount = await this.upgradeMigrationRepository
.createQueryBuilder('migration')
.where({
name: commandName,
status: 'completed',
workspaceId: In(workspaceIds),
})
.andWhere(
`migration.attempt = (
SELECT MAX(sub.attempt)
FROM core."upgradeMigration" sub
WHERE sub.name = migration.name
AND sub."workspaceId" = migration."workspaceId"
)`,
)
.getCount();
return completedCount === workspaceIds.length;
}
}
@@ -0,0 +1,171 @@
import { Injectable } from '@nestjs/common';
import { TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS } from 'src/engine/core-modules/upgrade/constants/twenty-cross-upgrade-supported-version.constant';
import {
type RegisteredFastInstanceCommand,
type RegisteredSlowInstanceCommand,
type RegisteredWorkspaceCommand,
UpgradeCommandRegistryService,
} from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
export type FastInstanceUpgradeStep = {
kind: 'fast-instance';
} & RegisteredFastInstanceCommand;
export type SlowInstanceUpgradeStep = {
kind: 'slow-instance';
} & RegisteredSlowInstanceCommand;
export type InstanceUpgradeStep =
| FastInstanceUpgradeStep
| SlowInstanceUpgradeStep;
export type WorkspaceUpgradeStep = {
kind: 'workspace';
} & RegisteredWorkspaceCommand;
export type UpgradeStep = InstanceUpgradeStep | WorkspaceUpgradeStep;
@Injectable()
export class UpgradeSequenceReaderService {
constructor(
private readonly upgradeCommandRegistryService: UpgradeCommandRegistryService,
) {}
getUpgradeSequence(): UpgradeStep[] {
const sequence: UpgradeStep[] = [];
for (const version of TWENTY_CROSS_UPGRADE_SUPPORTED_VERSIONS) {
const bundle =
this.upgradeCommandRegistryService.getBundleForVersion(version);
for (const command of bundle.fastInstanceCommands) {
sequence.push({ kind: 'fast-instance', ...command });
}
for (const command of bundle.slowInstanceCommands) {
sequence.push({ kind: 'slow-instance', ...command });
}
for (const command of bundle.workspaceCommands) {
sequence.push({ kind: 'workspace', ...command });
}
}
return sequence;
}
locateStepInSequenceOrThrow({
sequence,
stepName,
}: {
sequence: UpgradeStep[];
stepName: string;
}): number {
const cursor = sequence.findIndex((step) => step.name === stepName);
if (cursor === -1) {
throw new Error(`Step "${stepName}" not found in upgrade sequence`);
}
return cursor;
}
getWorkspaceCommandsSliceBounds({
sequence,
workspaceCommand,
}: {
sequence: UpgradeStep[];
workspaceCommand: WorkspaceUpgradeStep;
}): { startCursor: number; endCursor: number } {
const workspaceCommandCursor = this.locateStepInSequenceOrThrow({
sequence,
stepName: workspaceCommand.name,
});
let startCursor = workspaceCommandCursor;
while (startCursor > 0 && sequence[startCursor - 1].kind === 'workspace') {
startCursor--;
}
let endCursor = workspaceCommandCursor;
while (
endCursor < sequence.length - 1 &&
sequence[endCursor + 1].kind === 'workspace'
) {
endCursor++;
}
return { startCursor, endCursor };
}
collectContiguousWorkspaceSteps({
sequence,
fromWorkspaceCommand,
}: {
sequence: UpgradeStep[];
fromWorkspaceCommand: WorkspaceUpgradeStep;
}): WorkspaceUpgradeStep[] {
const fromCursor = this.locateStepInSequenceOrThrow({
sequence,
stepName: fromWorkspaceCommand.name,
});
const slice: WorkspaceUpgradeStep[] = [];
for (let cursor = fromCursor; cursor < sequence.length; cursor++) {
const step = sequence[cursor];
if (step.kind !== 'workspace') {
break;
}
slice.push(step);
}
return slice;
}
// Returns workspace commands that still need to run, based on the
// workspace's cursor position. If the cursor points to a command from
// a previous slice (not found in the current one), the entire slice
// is pending — this happens when a workspace enters a new slice for
// the first time after a sync barrier.
getPendingWorkspaceCommands({
workspaceCommands,
workspaceCursor,
}: {
workspaceCommands: WorkspaceUpgradeStep[];
workspaceCursor: { name: string; status: 'completed' | 'failed' };
}): WorkspaceUpgradeStep[] {
const cursorIndex = workspaceCommands.findIndex(
(command) => command.name === workspaceCursor.name,
);
if (cursorIndex === -1) {
return workspaceCommands;
}
return workspaceCursor.status === 'completed'
? workspaceCommands.slice(cursorIndex + 1)
: workspaceCommands.slice(cursorIndex);
}
getLastWorkspaceCommand(): RegisteredWorkspaceCommand {
const sequence = this.getUpgradeSequence();
for (let index = sequence.length - 1; index >= 0; index--) {
const step = sequence[index];
if (step.kind === 'workspace') {
return step;
}
}
throw new Error(
'No workspace commands found in upgrade sequence — this should have been caught at startup',
);
}
}
@@ -0,0 +1,294 @@
import { Injectable, Logger } from '@nestjs/common';
import {
type WorkspaceIteratorReport,
WorkspaceIteratorService,
} from 'src/database/commands/command-runners/workspace-iterator.service';
import { type ParsedUpgradeCommandOptions } from 'src/database/commands/upgrade-version-command/upgrade.command';
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import {
type InstanceUpgradeStep,
type UpgradeStep,
type WorkspaceUpgradeStep,
UpgradeSequenceReaderService,
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { WorkspaceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/workspace-command-runner.service';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
import { assertUnreachable, isDefined } from 'twenty-shared/utils';
export type UpgradeSequenceRunnerReport = {
totalSuccesses: number;
totalFailures: number;
};
@Injectable()
export class UpgradeSequenceRunnerService {
private readonly logger = new Logger(UpgradeSequenceRunnerService.name);
constructor(
private readonly upgradeMigrationService: UpgradeMigrationService,
private readonly instanceCommandRunnerService: InstanceCommandRunnerService,
private readonly workspaceCommandRunnerService: WorkspaceCommandRunnerService,
private readonly upgradeSequenceReaderService: UpgradeSequenceReaderService,
private readonly workspaceIteratorService: WorkspaceIteratorService,
private readonly workspaceVersionService: WorkspaceVersionService,
) {}
async run({
sequence,
options,
}: {
sequence: UpgradeStep[];
options: ParsedUpgradeCommandOptions;
}): Promise<UpgradeSequenceRunnerReport> {
if (sequence.length === 0) {
return { totalSuccesses: 0, totalFailures: 0 };
}
const allActiveOrSuspendedWorkspaceIds =
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds();
const startCursor = await this.resolveStartCursor({
sequence,
allActiveOrSuspendedWorkspaceIds,
});
let totalSuccesses = 0;
let totalFailures = 0;
let cursor = startCursor;
while (cursor < sequence.length) {
const step = sequence[cursor];
if (step.kind === 'fast-instance' || step.kind === 'slow-instance') {
const previousStep = cursor > 0 ? sequence[cursor - 1] : undefined;
if (previousStep?.kind === 'workspace') {
await this.enforceWorkspaceSyncBarrier({
previousWorkspaceStep: previousStep,
allActiveOrSuspendedWorkspaceIds,
});
}
await this.runInstanceStep({
instanceStep: step,
skipDataMigration: allActiveOrSuspendedWorkspaceIds.length === 0,
});
cursor++;
continue;
}
const contiguousWorkspaceSteps =
this.upgradeSequenceReaderService.collectContiguousWorkspaceSteps({
sequence,
fromWorkspaceCommand: step,
});
const report = await this.resumeWorkspaceCommandsFromCursors({
contiguousWorkspaceSteps,
allActiveOrSuspendedWorkspaceIds,
options,
});
totalSuccesses += report.success.length;
totalFailures += report.fail.length;
if (report.fail.length > 0) {
this.logger.error(
`Workspace steps ended with ${report.fail.length} failure(s). ` +
'Aborting — cannot proceed to next instance step.',
);
return { totalSuccesses, totalFailures };
}
cursor += contiguousWorkspaceSteps.length;
}
return { totalSuccesses, totalFailures };
}
private async resolveStartCursor({
sequence,
allActiveOrSuspendedWorkspaceIds,
}: {
sequence: UpgradeStep[];
allActiveOrSuspendedWorkspaceIds: string[];
}): Promise<number> {
const lastAttempted =
await this.upgradeMigrationService.getLastAttemptedCommandNameOrThrow(
allActiveOrSuspendedWorkspaceIds,
);
const lastAttemptedCursor =
this.upgradeSequenceReaderService.locateStepInSequenceOrThrow({
sequence,
stepName: lastAttempted.name,
});
const lastAttemptedStep = sequence[lastAttemptedCursor];
switch (lastAttemptedStep.kind) {
case 'fast-instance':
case 'slow-instance': {
return lastAttempted.status === 'completed'
? lastAttemptedCursor + 1
: lastAttemptedCursor;
}
case 'workspace': {
const workspaceSliceBounds =
this.upgradeSequenceReaderService.getWorkspaceCommandsSliceBounds({
sequence,
workspaceCommand: lastAttemptedStep,
});
await this.validateWorkspaceCursorsAreInSameWorkspaceStepsSlice({
sequence,
allActiveOrSuspendedWorkspaceIds,
workspaceSliceBounds,
});
return workspaceSliceBounds.startCursor;
}
default:
assertUnreachable(lastAttemptedStep);
}
}
private async validateWorkspaceCursorsAreInSameWorkspaceStepsSlice({
allActiveOrSuspendedWorkspaceIds,
sequence,
workspaceSliceBounds: { startCursor, endCursor },
}: {
sequence: UpgradeStep[];
allActiveOrSuspendedWorkspaceIds: string[];
workspaceSliceBounds: { startCursor: number; endCursor: number };
}): Promise<void> {
const workspaceCursors =
await this.upgradeMigrationService.getWorkspaceLastAttemptedCommandNameOrThrow(
allActiveOrSuspendedWorkspaceIds,
);
for (const [workspaceId, workspaceCursor] of workspaceCursors) {
const cursor =
this.upgradeSequenceReaderService.locateStepInSequenceOrThrow({
sequence,
stepName: workspaceCursor.name,
});
if (cursor < startCursor || cursor > endCursor) {
throw new Error(
`Workspace ${workspaceId} cursor "${workspaceCursor.name}" is outside the ` +
`current workspace slice [${startCursor}..${endCursor}] — ` +
'workspaces are not aligned',
);
}
}
}
private async runInstanceStep({
instanceStep,
skipDataMigration,
}: {
instanceStep: InstanceUpgradeStep;
skipDataMigration: boolean;
}): Promise<void> {
switch (instanceStep.kind) {
case 'fast-instance': {
const result =
await this.instanceCommandRunnerService.runFastInstanceCommand({
command: instanceStep.command,
name: instanceStep.name,
});
if (result.status === 'failed') {
throw result.error;
}
return;
}
case 'slow-instance': {
const result =
await this.instanceCommandRunnerService.runSlowInstanceCommand({
command: instanceStep.command,
name: instanceStep.name,
skipDataMigration,
});
if (result.status === 'failed') {
throw result.error;
}
return;
}
default:
assertUnreachable(instanceStep);
}
}
private async resumeWorkspaceCommandsFromCursors({
contiguousWorkspaceSteps,
allActiveOrSuspendedWorkspaceIds,
options,
}: {
contiguousWorkspaceSteps: WorkspaceUpgradeStep[];
allActiveOrSuspendedWorkspaceIds: string[];
options: ParsedUpgradeCommandOptions;
}): Promise<WorkspaceIteratorReport> {
const workspaceCursors =
await this.upgradeMigrationService.getWorkspaceLastAttemptedCommandNameOrThrow(
allActiveOrSuspendedWorkspaceIds,
);
return this.workspaceIteratorService.iterate({
workspaceIds:
isDefined(options.workspaceIds) && options.workspaceIds.length > 0
? options.workspaceIds
: allActiveOrSuspendedWorkspaceIds,
startFromWorkspaceId: options.startFromWorkspaceId,
workspaceCountLimit: options.workspaceCountLimit,
dryRun: options.dryRun,
callback: async (context) => {
const workspaceCursor = workspaceCursors.get(context.workspaceId);
if (!workspaceCursor) {
throw new Error(
`No upgrade migration found for workspace ${context.workspaceId}. This should never occur.`,
);
}
const pendingCommands =
this.upgradeSequenceReaderService.getPendingWorkspaceCommands({
workspaceCommands: contiguousWorkspaceSteps,
workspaceCursor,
});
await this.workspaceCommandRunnerService.runWorkspaceCommands({
iteratorContext: context,
options,
workspaceCommands: pendingCommands,
});
},
});
}
private async enforceWorkspaceSyncBarrier({
previousWorkspaceStep,
allActiveOrSuspendedWorkspaceIds,
}: {
previousWorkspaceStep: WorkspaceUpgradeStep;
allActiveOrSuspendedWorkspaceIds: string[];
}): Promise<void> {
const allWorkspacesReady =
await this.upgradeMigrationService.areAllWorkspacesAtCommand({
commandName: previousWorkspaceStep.name,
workspaceIds: allActiveOrSuspendedWorkspaceIds,
});
if (!allWorkspacesReady) {
throw new Error(
'Cannot run instance step: not all workspaces have completed ' +
`"${previousWorkspaceStep.name}"`,
);
}
}
}
@@ -0,0 +1,100 @@
import { Injectable, Logger } from '@nestjs/common';
import { type WorkspaceIteratorContext } from 'src/database/commands/command-runners/workspace-iterator.service';
import { type ParsedUpgradeCommandOptions } from 'src/database/commands/upgrade-version-command/upgrade.command';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { type RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
type WorkspaceCommandEntry = Pick<
RegisteredWorkspaceCommand,
'name' | 'command'
>;
export type RunWorkspaceCommandsArgs = {
iteratorContext: WorkspaceIteratorContext;
options: ParsedUpgradeCommandOptions;
workspaceCommands: WorkspaceCommandEntry[];
};
@Injectable()
export class WorkspaceCommandRunnerService {
private readonly logger = new Logger(WorkspaceCommandRunnerService.name);
constructor(
private readonly twentyConfigService: TwentyConfigService,
private readonly upgradeMigrationService: UpgradeMigrationService,
) {}
async runWorkspaceCommands({
iteratorContext,
options,
workspaceCommands,
}: RunWorkspaceCommandsArgs): Promise<void> {
const { workspaceId, index, total } = iteratorContext;
this.logger.log(
`${options.dryRun ? '(dry run) ' : ''}Upgrading workspace ${workspaceId} ${index + 1}/${total}`,
);
const executedByVersion =
this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
for (const workspaceCommandEntry of workspaceCommands) {
await this.runSingleWorkspaceCommandOrThrow({
workspaceCommandEntry,
workspaceId,
executedByVersion,
options,
iteratorContext,
});
}
this.logger.log(`Upgrade for workspace ${workspaceId} completed.`);
}
private async runSingleWorkspaceCommandOrThrow({
workspaceCommandEntry,
workspaceId,
executedByVersion,
options,
iteratorContext,
}: {
workspaceCommandEntry: WorkspaceCommandEntry;
workspaceId: string;
executedByVersion: string;
options: ParsedUpgradeCommandOptions;
iteratorContext: WorkspaceIteratorContext;
}): Promise<void> {
const { name, command: workspaceCommand } = workspaceCommandEntry;
try {
await workspaceCommand.runOnWorkspace({
options,
workspaceId,
dataSource: iteratorContext.dataSource,
index: iteratorContext.index,
total: iteratorContext.total,
});
if (!options.dryRun) {
await this.upgradeMigrationService.markAsCompleted({
name,
workspaceId,
executedByVersion,
});
}
} catch (error) {
if (!options.dryRun) {
await this.upgradeMigrationService.markAsFailed({
name,
workspaceId,
executedByVersion,
error,
});
}
throw error;
}
}
}
@@ -1,181 +0,0 @@
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/upgrade-version-command/upgrade.command';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
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>,
private readonly twentyConfigService: TwentyConfigService,
private readonly upgradeMigrationService: UpgradeMigrationService,
) {}
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': {
const executedByVersion =
this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
for (const workspaceCommandEntry of workspaceCommands) {
await this.runSingleWorkspaceCommandOrThrow({
workspaceCommandEntry,
workspaceId,
executedByVersion,
options,
iteratorContext,
});
}
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,
);
}
private async runSingleWorkspaceCommandOrThrow({
workspaceCommandEntry,
workspaceId,
executedByVersion,
options,
iteratorContext,
}: {
workspaceCommandEntry: RegisteredWorkspaceCommand;
workspaceId: string;
executedByVersion: string;
options: UpgradeCommandOptions;
iteratorContext: WorkspaceIteratorContext;
}): Promise<void> {
const { name, command: workspaceCommand } = workspaceCommandEntry;
const isAlreadyCompleted =
await this.upgradeMigrationService.isLastAttemptCompleted({
name,
workspaceId,
});
if (isAlreadyCompleted) {
this.logger.log(
`Workspace command ${name} already completed for workspace ${workspaceId}, skipping`,
);
return;
}
try {
await workspaceCommand.runOnWorkspace({
options,
workspaceId,
dataSource: iteratorContext.dataSource,
index: iteratorContext.index,
total: iteratorContext.total,
});
if (!options.dryRun) {
await this.upgradeMigrationService.markAsCompleted({
name,
workspaceId,
executedByVersion,
});
}
} catch (error) {
if (!options.dryRun) {
await this.upgradeMigrationService.markAsFailed({
name,
workspaceId,
executedByVersion,
error,
});
}
throw error;
}
}
}
@@ -0,0 +1,14 @@
import {
TWENTY_ALL_VERSIONS,
TwentyAllVersion,
} from 'src/engine/core-modules/upgrade/constants/twenty-all-versions.constant';
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
import { IndexOf, IsGreaterOrEqual } from 'twenty-shared/types';
export type RemovedSinceVersion<RemoveAtVersion extends TwentyAllVersion, T> =
IsGreaterOrEqual<
IndexOf<typeof TWENTY_CURRENT_VERSION, typeof TWENTY_ALL_VERSIONS>,
IndexOf<RemoveAtVersion, typeof TWENTY_ALL_VERSIONS>
> extends true
? never
: T;
@@ -41,6 +41,9 @@ export class UpgradeMigrationEntity {
@Column({ type: 'text', nullable: true })
errorMessage: string | null;
@Column({ type: 'boolean', nullable: false, default: false })
isInitial: boolean;
@ManyToOne(() => WorkspaceEntity, { onDelete: 'CASCADE', nullable: true })
@JoinColumn({ name: 'workspaceId' })
workspace: Relation<WorkspaceEntity> | null;
@@ -2,29 +2,39 @@ import { Module } from '@nestjs/common';
import { DiscoveryModule } from '@nestjs/core';
import { TypeOrmModule } from '@nestjs/typeorm';
import { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module';
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { WorkspaceUpgradeService } from 'src/engine/core-modules/upgrade/services/workspace-upgrade.service';
import { UpgradeSequenceReaderService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { UpgradeSequenceRunnerService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service';
import { WorkspaceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/workspace-command-runner.service';
import { UpgradeMigrationEntity } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { WorkspaceVersionModule } from 'src/engine/workspace-manager/workspace-version/workspace-version.module';
@Module({
imports: [
DiscoveryModule,
WorkspaceIteratorModule,
WorkspaceVersionModule,
TypeOrmModule.forFeature([UpgradeMigrationEntity, WorkspaceEntity]),
],
providers: [
UpgradeMigrationService,
InstanceUpgradeService,
WorkspaceUpgradeService,
InstanceCommandRunnerService,
WorkspaceCommandRunnerService,
UpgradeCommandRegistryService,
UpgradeSequenceReaderService,
UpgradeSequenceRunnerService,
],
exports: [
UpgradeMigrationService,
InstanceUpgradeService,
WorkspaceUpgradeService,
InstanceCommandRunnerService,
WorkspaceCommandRunnerService,
UpgradeCommandRegistryService,
UpgradeSequenceReaderService,
UpgradeSequenceRunnerService,
],
})
export class UpgradeModule {}
@@ -28,8 +28,10 @@ import { WorkspaceService } from 'src/engine/core-modules/workspace/services/wor
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { createEmptyAllFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/constant/create-empty-all-flat-entity-maps.constant';
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
import { ObjectMetadataService } from 'src/engine/metadata-modules/object-metadata/object-metadata.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { UpgradeSequenceReaderService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { AiModelRegistryService } from 'src/engine/metadata-modules/ai/ai-models/services/ai-model-registry.service';
import { ObjectMetadataService } from 'src/engine/metadata-modules/object-metadata/object-metadata.service';
import { PermissionsService } from 'src/engine/metadata-modules/permissions/permissions.service';
import { CoreEntityCacheService } from 'src/engine/core-entity-cache/services/core-entity-cache.service';
import { WorkspaceCacheStorageService } from 'src/engine/workspace-cache-storage/workspace-cache-storage.service';
@@ -122,6 +124,8 @@ describe('WorkspaceService', () => {
FileCorePictureService,
AiModelRegistryService,
PrefillLogicFunctionService,
UpgradeMigrationService,
UpgradeSequenceReaderService,
].map((service) => ({
provide: service,
useValue: {},
@@ -10,7 +10,9 @@ import { assertIsDefinedOrThrow, isDefined } from 'twenty-shared/utils';
import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
import { DataSource, QueryRunner, Repository } from 'typeorm';
import { CoreEntityCacheService } from 'src/engine/core-entity-cache/services/core-entity-cache.service';
import { ApiKeyEntity } from 'src/engine/core-modules/api-key/api-key.entity';
import { type AuthContextUser } from 'src/engine/core-modules/auth/types/auth-context.type';
import { BillingSubscriptionService } from 'src/engine/core-modules/billing/services/billing-subscription.service';
import { BillingService } from 'src/engine/core-modules/billing/services/billing.service';
import { DnsManagerService } from 'src/engine/core-modules/dns-manager/services/dns-manager.service';
@@ -27,9 +29,10 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { UpgradeSequenceReaderService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
import { UserWorkspaceService } from 'src/engine/core-modules/user-workspace/user-workspace.service';
import { type AuthContextUser } from 'src/engine/core-modules/auth/types/auth-context.type';
import { UserEntity } from 'src/engine/core-modules/user/user.entity';
import { type ActivateWorkspaceInput } from 'src/engine/core-modules/workspace/dtos/activate-workspace-input';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
@@ -38,7 +41,6 @@ import {
WorkspaceExceptionCode,
WorkspaceNotFoundDefaultError,
} from 'src/engine/core-modules/workspace/workspace.exception';
import { CoreEntityCacheService } from 'src/engine/core-entity-cache/services/core-entity-cache.service';
import { AiModelRegistryService } from 'src/engine/metadata-modules/ai/ai-models/services/ai-model-registry.service';
import { isModelAllowedByWorkspace } from 'src/engine/metadata-modules/ai/ai-models/utils/is-model-allowed.util';
import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
@@ -54,17 +56,16 @@ import { PermissionsService } from 'src/engine/metadata-modules/permissions/perm
import { WorkspaceCacheStorageService } from 'src/engine/workspace-cache-storage/workspace-cache-storage.service';
import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
import { PrefillLogicFunctionService } from 'src/engine/workspace-manager/standard-objects-prefill-data/services/prefill-logic-function.service';
import { prefillCompanies } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-companies.util';
import { prefillDashboards } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-dashboards.util';
import { prefillOpportunities } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-opportunities.util';
import { prefillPeople } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-people.util';
import { prefillWorkflowCommandMenuItems } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-workflow-command-menu-items.util';
import { getCreateCompanyWhenAddingNewPersonCodeStepLogicFunctionDefinitions } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-workflow-code-step-logic-functions.util';
import { prefillWorkflowCommandMenuItems } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-workflow-command-menu-items.util';
import { prefillWorkflows } from 'src/engine/workspace-manager/standard-objects-prefill-data/utils/prefill-workflows.util';
import { PrefillLogicFunctionService } from 'src/engine/workspace-manager/standard-objects-prefill-data/services/prefill-logic-function.service';
import { WorkspaceManagerService } from 'src/engine/workspace-manager/workspace-manager.service';
import { DEFAULT_FEATURE_FLAGS } from 'src/engine/workspace-manager/workspace-migration/constant/default-feature-flags';
import { extractVersionMajorMinorPatch } from 'src/utils/version/extract-version-major-minor-patch';
@Injectable()
// oxlint-disable-next-line twenty/inject-workspace-repository
@@ -126,6 +127,8 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
@InjectDataSource()
private readonly coreDataSource: DataSource,
private readonly coreEntityCacheService: CoreEntityCacheService,
private readonly upgradeMigrationService: UpgradeMigrationService,
private readonly upgradeSequenceReaderService: UpgradeSequenceReaderService,
) {
super(workspaceRepository);
}
@@ -354,12 +357,9 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
schemaName: getWorkspaceSchemaName(workspace.id),
});
const appVersion = this.twentyConfigService.get('APP_VERSION');
await this.workspaceRepository.update(workspace.id, {
await this.activateAndInitializeUpgradeState({
workspaceId: workspace.id,
displayName: data.displayName,
activationStatus: WorkspaceActivationStatus.ACTIVE,
version: extractVersionMajorMinorPatch(appVersion),
});
await this.coreEntityCacheService.invalidate(
@@ -372,6 +372,46 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
});
}
private async activateAndInitializeUpgradeState({
displayName,
workspaceId,
}: {
workspaceId: string;
displayName: string;
}): Promise<void> {
const lastWorkspaceCommand =
this.upgradeSequenceReaderService.getLastWorkspaceCommand();
const executedByVersion =
this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
const queryRunner = this.coreDataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
await queryRunner.manager.update(WorkspaceEntity, workspaceId, {
displayName,
activationStatus: WorkspaceActivationStatus.ACTIVE,
});
await this.upgradeMigrationService.markAsInitial({
name: lastWorkspaceCommand.name,
workspaceId,
executedByVersion,
queryRunner,
});
await queryRunner.commitTransaction();
} catch (error) {
await queryRunner.rollbackTransaction();
throw error;
} finally {
await queryRunner.release();
}
}
async deleteWorkspace(id: string, softDelete = false) {
const workspace = await this.workspaceRepository.findOne({
where: { id },
@@ -1,5 +1,5 @@
import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { type FlatWorkspace } from 'src/engine/core-modules/workspace/types/flat-workspace.type';
import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
export const fromWorkspaceEntityToFlat = (
entity: WorkspaceEntity,
@@ -28,7 +28,6 @@ export const fromWorkspaceEntityToFlat = (
isCustomDomainEnabled: entity.isCustomDomainEnabled,
editableProfileFields: entity.editableProfileFields,
defaultRoleId: entity.defaultRoleId,
version: entity.version,
fastModel: entity.fastModel,
smartModel: entity.smartModel,
aiAdditionalInstructions: entity.aiAdditionalInstructions,
@@ -40,4 +39,5 @@ export const fromWorkspaceEntityToFlat = (
updatedAt: entity.updatedAt.toISOString(),
deletedAt: entity.deletedAt?.toISOString(),
suspendedAt: entity.suspendedAt?.toISOString() ?? null,
version: entity.version ?? null,
});
@@ -32,12 +32,9 @@ import { KeyValuePairEntity } from 'src/engine/core-modules/key-value-pair/key-v
import { PostgresCredentialsEntity } from 'src/engine/core-modules/postgres-credentials/postgres-credentials.entity';
import { PublicDomainEntity } from 'src/engine/core-modules/public-domain/public-domain.entity';
import { WorkspaceSSOIdentityProviderEntity } from 'src/engine/core-modules/sso/workspace-sso-identity-provider.entity';
import { RemovedSinceVersion } from 'src/engine/core-modules/upgrade/types/removed-since-version.type';
import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
import { AgentEntity } from 'src/engine/metadata-modules/ai/ai-agent/entities/agent.entity';
import {
AUTO_SELECT_FAST_MODEL_ID,
AUTO_SELECT_SMART_MODEL_ID,
} from 'twenty-shared/constants';
import { type ModelId } from 'src/engine/metadata-modules/ai/ai-models/types/model-id.type';
import { RoleDTO } from 'src/engine/metadata-modules/role/dtos/role.dto';
import { ViewFieldDTO } from 'src/engine/metadata-modules/view-field/dtos/view-field.dto';
@@ -53,6 +50,10 @@ import { ViewSortEntity } from 'src/engine/metadata-modules/view-sort/entities/v
import { ViewDTO } from 'src/engine/metadata-modules/view/dtos/view.dto';
import { ViewEntity } from 'src/engine/metadata-modules/view/entities/view.entity';
import { WebhookEntity } from 'src/engine/metadata-modules/webhook/entities/webhook.entity';
import {
AUTO_SELECT_FAST_MODEL_ID,
AUTO_SELECT_SMART_MODEL_ID,
} from 'twenty-shared/constants';
registerEnumType(WorkspaceActivationStatus, {
name: 'WorkspaceActivationStatus',
@@ -296,7 +297,7 @@ export class WorkspaceEntity {
@Field(() => String, { nullable: true })
@Column({ type: 'varchar', nullable: true })
version: string | null;
version: RemovedSinceVersion<'1.23.0', string | null>;
@Field(() => String, { nullable: false })
@Column({
@@ -25,6 +25,7 @@ import { UserWorkspaceModule } from 'src/engine/core-modules/user-workspace/user
import { UserEntity } from 'src/engine/core-modules/user/user.entity';
import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/workspace/crons/commands/check-custom-domain-valid-records.cron.command';
import { CheckCustomDomainValidRecordsCronJob } from 'src/engine/core-modules/workspace/crons/jobs/check-custom-domain-valid-records.cron.job';
import { UpgradeModule } from 'src/engine/core-modules/upgrade/upgrade.module';
import { CoreEntityCacheModule } from 'src/engine/core-entity-cache/core-entity-cache.module';
import { WorkspaceEntityCacheProviderService } from 'src/engine/core-modules/workspace/services/workspace-entity-cache-provider.service';
import { WorkspaceService } from 'src/engine/core-modules/workspace/services/workspace.service';
@@ -83,6 +84,7 @@ import { StandardObjectsPrefillModule } from 'src/engine/workspace-manager/stand
EnterpriseModule,
StandardObjectsPrefillModule,
CoreEntityCacheModule,
UpgradeModule,
],
services: [WorkspaceService],
resolvers: workspaceAutoResolverOpts,
@@ -10,7 +10,6 @@ export const WORKSPACE_FIELDS_TO_SEED = [
'logo',
'activationStatus',
'isTwoFactorAuthenticationEnforced',
'version',
'workspaceCustomApplicationId',
] as const satisfies (keyof WorkspaceEntity)[];
@@ -48,5 +47,5 @@ export const SEEDER_CREATE_WORKSPACE_INPUT = {
},
} as const satisfies Record<
SeededWorkspacesIds,
Omit<CreateWorkspaceInput, 'version' | 'workspaceCustomApplicationId'>
Omit<CreateWorkspaceInput, 'workspaceCustomApplicationId'>
>;
@@ -1,106 +0,0 @@
import { type DataSource } from 'typeorm';
import { v4 } from 'uuid';
import { type ApplicationService } from 'src/engine/core-modules/application/application.service';
import { seedBillingCustomers } from 'src/engine/workspace-manager/dev-seeder/core/billing/utils/seed-billing-customers.util';
import { seedBillingSubscriptions } from 'src/engine/workspace-manager/dev-seeder/core/billing/utils/seed-billing-subscriptions.util';
import {
type SeededWorkspacesIds,
SEEDER_CREATE_WORKSPACE_INPUT,
} from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant';
import { seedAgents } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-agents.util';
import { seedApiKeys } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-api-keys.util';
import { seedFeatureFlags } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-feature-flags.util';
import { seedMetadataEntities } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-metadata-entities.util';
import { seedServerId } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-server-id.util';
import { seedUserWorkspaces } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-user-workspaces.util';
import { seedUsers } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-users.util';
import { createWorkspace } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-workspace.util';
import { extractVersionMajorMinorPatch } from 'src/utils/version/extract-version-major-minor-patch';
type SeedCoreSchemaArgs = {
dataSource: DataSource;
workspaceId: SeededWorkspacesIds;
appVersion: string | undefined;
applicationService: ApplicationService;
seedBilling?: boolean;
seedFeatureFlags?: boolean;
};
export const seedCoreSchema = async ({
appVersion,
dataSource,
workspaceId,
applicationService,
seedBilling = true,
seedFeatureFlags: shouldSeedFeatureFlags = true,
}: SeedCoreSchemaArgs) => {
const schemaName = 'core';
const createWorkspaceStaticInput = SEEDER_CREATE_WORKSPACE_INPUT[workspaceId];
const version = extractVersionMajorMinorPatch(appVersion);
const queryRunner = dataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
const workspaceCustomApplicationId = v4();
await createWorkspace({
queryRunner,
schemaName,
createWorkspaceInput: {
...createWorkspaceStaticInput,
version,
workspaceCustomApplicationId,
},
});
await applicationService.createWorkspaceCustomApplication(
{
workspaceId,
applicationId: workspaceCustomApplicationId,
workspaceDisplayName: createWorkspaceStaticInput.displayName,
},
queryRunner,
);
await seedServerId({ queryRunner, schemaName });
await seedUsers({ queryRunner, schemaName });
await seedUserWorkspaces({ queryRunner, schemaName, workspaceId });
await applicationService.createTwentyStandardApplication(
{
workspaceId,
skipCacheInvalidation: true,
},
queryRunner,
);
await seedAgents({ queryRunner, schemaName, workspaceId });
await seedApiKeys({ queryRunner, schemaName, workspaceId });
if (shouldSeedFeatureFlags) {
await seedFeatureFlags({ queryRunner, schemaName, workspaceId });
}
if (seedBilling) {
await seedBillingCustomers({ queryRunner, schemaName, workspaceId });
await seedBillingSubscriptions({ queryRunner, schemaName, workspaceId });
}
await seedMetadataEntities({ queryRunner, schemaName, workspaceId });
await queryRunner.commitTransaction();
} catch (error) {
await queryRunner.rollbackTransaction();
throw error;
} finally {
await queryRunner.release();
}
};
@@ -17,6 +17,7 @@ import { ObjectPermissionModule } from 'src/engine/metadata-modules/object-permi
import { RoleTargetModule } from 'src/engine/metadata-modules/role-target/role-target.module';
import { RoleModule } from 'src/engine/metadata-modules/role/role.module';
import { UserRoleModule } from 'src/engine/metadata-modules/user-role/user-role.module';
import { UpgradeModule } from 'src/engine/core-modules/upgrade/upgrade.module';
import { WorkspaceCacheStorageModule } from 'src/engine/workspace-cache-storage/workspace-cache-storage.module';
import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
@@ -52,6 +53,7 @@ import { WorkspaceMigrationModule } from 'src/engine/workspace-manager/workspace
WorkspaceMigrationModule,
TwentyStandardApplicationModule,
SdkClientModule,
UpgradeModule,
],
exports: [DevSeederService],
providers: [
@@ -3,11 +3,14 @@ import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
import { ALL_METADATA_NAME } from 'twenty-shared/metadata';
import { DataSource, Repository } from 'typeorm';
import { v4 } from 'uuid';
import { ApplicationRegistrationService } from 'src/engine/core-modules/application/application-registration/application-registration.service';
import { ApplicationService } from 'src/engine/core-modules/application/application.service';
import { SdkClientGenerationService } from 'src/engine/core-modules/sdk-client/sdk-client-generation.service';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
import { UpgradeSequenceReaderService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { getMetadataFlatEntityMapsKey } from 'src/engine/metadata-modules/flat-entity/utils/get-metadata-flat-entity-maps-key.util';
import { getMetadataRelatedMetadataNames } from 'src/engine/metadata-modules/flat-entity/utils/get-metadata-related-metadata-names.util';
@@ -15,9 +18,21 @@ import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadat
import { WorkspaceCacheStorageService } from 'src/engine/workspace-cache-storage/workspace-cache-storage.service';
import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
import { SeededWorkspacesIds } from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant';
import {
type SeededWorkspacesIds,
SEEDER_CREATE_WORKSPACE_INPUT,
} from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant';
import { seedBillingCustomers } from 'src/engine/workspace-manager/dev-seeder/core/billing/utils/seed-billing-customers.util';
import { seedBillingSubscriptions } from 'src/engine/workspace-manager/dev-seeder/core/billing/utils/seed-billing-subscriptions.util';
import { DevSeederPermissionsService } from 'src/engine/workspace-manager/dev-seeder/core/services/dev-seeder-permissions.service';
import { seedCoreSchema } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-core-schema.util';
import { seedAgents } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-agents.util';
import { seedApiKeys } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-api-keys.util';
import { seedFeatureFlags } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-feature-flags.util';
import { seedMetadataEntities } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-metadata-entities.util';
import { seedServerId } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-server-id.util';
import { seedUserWorkspaces } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-user-workspaces.util';
import { seedUsers } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-users.util';
import { createWorkspace } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-workspace.util';
import { seedPageLayoutTabs } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-page-layout-tabs.util';
import { seedPageLayoutWidgets } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-page-layout-widgets.util';
import { seedPageLayouts } from 'src/engine/workspace-manager/dev-seeder/core/utils/seed-page-layouts.util';
@@ -41,6 +56,8 @@ export class DevSeederService {
private readonly applicationRegistrationService: ApplicationRegistrationService,
private readonly workspaceCacheService: WorkspaceCacheService,
private readonly sdkClientGenerationService: SdkClientGenerationService,
private readonly upgradeMigrationService: UpgradeMigrationService,
private readonly upgradeSequenceReaderService: UpgradeSequenceReaderService,
@InjectDataSource()
private readonly coreDataSource: DataSource,
@InjectRepository(WorkspaceEntity)
@@ -53,14 +70,16 @@ export class DevSeederService {
): Promise<void> {
const light = options?.light ?? false;
const isBillingEnabled = this.twentyConfigService.get('IS_BILLING_ENABLED');
const appVersion = this.twentyConfigService.get('APP_VERSION');
const appVersion = this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
await seedCoreSchema({
dataSource: this.coreDataSource,
const lastWorkspaceCommand =
this.upgradeSequenceReaderService.getLastWorkspaceCommand();
await this.seedCoreSchema({
workspaceId,
applicationService: this.applicationService,
seedBilling: isBillingEnabled,
appVersion,
lastUpgradeStepName: lastWorkspaceCommand.name,
});
await this.applicationRegistrationService.createCliRegistrationIfNotExists();
@@ -171,4 +190,87 @@ export class DevSeederService {
await this.workspaceCacheStorageService.flush(workspaceId, undefined);
}
private async seedCoreSchema({
workspaceId,
appVersion,
lastUpgradeStepName,
seedBilling = true,
}: {
workspaceId: SeededWorkspacesIds;
appVersion: string;
lastUpgradeStepName: string;
seedBilling?: boolean;
}): Promise<void> {
const schemaName = 'core';
const createWorkspaceStaticInput =
SEEDER_CREATE_WORKSPACE_INPUT[workspaceId];
const queryRunner = this.coreDataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
const workspaceCustomApplicationId = v4();
await createWorkspace({
queryRunner,
schemaName,
createWorkspaceInput: {
...createWorkspaceStaticInput,
workspaceCustomApplicationId,
},
});
await this.applicationService.createWorkspaceCustomApplication(
{
workspaceId,
applicationId: workspaceCustomApplicationId,
workspaceDisplayName: createWorkspaceStaticInput.displayName,
},
queryRunner,
);
await seedServerId({ queryRunner, schemaName });
await seedUsers({ queryRunner, schemaName });
await seedUserWorkspaces({ queryRunner, schemaName, workspaceId });
await this.applicationService.createTwentyStandardApplication(
{
workspaceId,
skipCacheInvalidation: true,
},
queryRunner,
);
await seedAgents({ queryRunner, schemaName, workspaceId });
await seedApiKeys({ queryRunner, schemaName, workspaceId });
await seedFeatureFlags({ queryRunner, schemaName, workspaceId });
if (seedBilling) {
await seedBillingCustomers({ queryRunner, schemaName, workspaceId });
await seedBillingSubscriptions({
queryRunner,
schemaName,
workspaceId,
});
}
await seedMetadataEntities({ queryRunner, schemaName, workspaceId });
await this.upgradeMigrationService.markAsInitial({
name: lastUpgradeStepName,
workspaceId,
executedByVersion: appVersion,
queryRunner,
});
await queryRunner.commitTransaction();
} catch (error) {
await queryRunner.rollbackTransaction();
throw error;
} finally {
await queryRunner.release();
}
}
}
@@ -1,17 +1,13 @@
import { Injectable, Logger } from '@nestjs/common';
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { isDefined } from 'twenty-shared/utils';
import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
import { In, Repository } from 'typeorm';
import { In, MoreThanOrEqual, Repository } from 'typeorm';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { compareVersionMajorAndMinor } from 'src/utils/version/compare-version-minor-and-major';
@Injectable()
export class WorkspaceVersionService {
private readonly logger = new Logger(WorkspaceVersionService.name);
constructor(
@InjectRepository(WorkspaceEntity)
private readonly workspaceRepository: Repository<WorkspaceEntity>,
@@ -28,56 +24,28 @@ export class WorkspaceVersionService {
});
}
async getWorkspacesBelowVersion(
version: string,
): Promise<Pick<WorkspaceEntity, 'id' | 'displayName' | 'version'>[]> {
const allActiveOrSuspendedWorkspaces =
await this.loadActiveOrSuspendedWorkspaces();
if (allActiveOrSuspendedWorkspaces.length === 0) {
this.logger.log(
'No workspaces found. Running migrations for fresh installation.',
);
return [];
}
return allActiveOrSuspendedWorkspaces.filter((workspace) => {
if (!isDefined(workspace.version)) {
return true;
}
try {
const versionCompareResult = compareVersionMajorAndMinor(
workspace.version,
version,
);
return versionCompareResult === 'lower';
} catch (error) {
this.logger.error(
`Error checking workspace ${workspace.id} version: ${error.message}`,
);
return true;
}
});
}
private async loadActiveOrSuspendedWorkspaces(): Promise<
Pick<WorkspaceEntity, 'id' | 'version' | 'displayName'>[]
> {
return this.workspaceRepository.find({
select: ['id', 'version', 'displayName'],
async getActiveOrSuspendedWorkspaceIds({
startFromWorkspaceId,
workspaceCountLimit,
}: {
startFromWorkspaceId?: string;
workspaceCountLimit?: number;
} = {}): Promise<string[]> {
const workspaces = await this.workspaceRepository.find({
select: ['id'],
where: {
activationStatus: In([
WorkspaceActivationStatus.ACTIVE,
WorkspaceActivationStatus.SUSPENDED,
]),
...(startFromWorkspaceId
? { id: MoreThanOrEqual(startFromWorkspaceId) }
: {}),
},
order: {
id: 'ASC',
},
order: { id: 'ASC' },
take: workspaceCountLimit,
});
return workspaces.map((workspace) => workspace.id);
}
}
@@ -0,0 +1,10 @@
// Finds the index of T in a readonly string tuple.
export type IndexOf<
T extends string,
Arr extends readonly string[],
Acc extends unknown[] = [],
> = Arr extends readonly [infer Head, ...infer Tail extends readonly string[]]
? Head extends T
? Acc['length']
: IndexOf<T, Tail, [...Acc, unknown]>
: never;
@@ -0,0 +1,12 @@
// Returns true if A >= B using tuple-length counting.
export type IsGreaterOrEqual<
A extends number,
B extends number,
Acc extends unknown[] = [],
> = Acc['length'] extends A
? Acc['length'] extends B
? true
: false
: Acc['length'] extends B
? true
: IsGreaterOrEqual<A, B, [...Acc, unknown]>;
@@ -0,0 +1,22 @@
import { type Equal, type Expect } from '@/testing';
import { type IndexOf } from '@/types/IndexOf.type';
type Versions = readonly ['1.20.0', '1.21.0', '1.22.0', '1.23.0'];
// oxlint-disable-next-line unused-imports/no-unused-vars
type Assertions = [
Expect<Equal<IndexOf<'1.20.0', Versions>, 0>>,
Expect<Equal<IndexOf<'1.21.0', Versions>, 1>>,
Expect<Equal<IndexOf<'1.22.0', Versions>, 2>>,
Expect<Equal<IndexOf<'1.23.0', Versions>, 3>>,
// Not found resolves to never
Expect<Equal<IndexOf<'1.99.0', Versions>, never>>,
// Single element tuple
Expect<Equal<IndexOf<'a', readonly ['a']>, 0>>,
// Empty tuple
Expect<Equal<IndexOf<'a', readonly []>, never>>,
];
@@ -0,0 +1,20 @@
import { type Equal, type Expect } from '@/testing';
import { type IsGreaterOrEqual } from '@/types/IsGreaterOrEqual.type';
// oxlint-disable-next-line unused-imports/no-unused-vars
type Assertions = [
// Equal values
Expect<Equal<IsGreaterOrEqual<0, 0>, true>>,
Expect<Equal<IsGreaterOrEqual<3, 3>, true>>,
// A > B
Expect<Equal<IsGreaterOrEqual<3, 0>, true>>,
Expect<Equal<IsGreaterOrEqual<3, 2>, true>>,
Expect<Equal<IsGreaterOrEqual<1, 0>, true>>,
// A < B
Expect<Equal<IsGreaterOrEqual<0, 1>, false>>,
Expect<Equal<IsGreaterOrEqual<0, 3>, false>>,
Expect<Equal<IsGreaterOrEqual<2, 3>, false>>,
];
@@ -124,9 +124,11 @@ export { FirstDayOfTheWeek } from './FirstDayOfTheWeek';
export type { FormatRecordSerializedRelationProperties } from './FormatRecordSerializedRelationProperties.type';
export type { FromTo } from './FromToType';
export { HTTPMethod } from './HttpMethod';
export type { IndexOf } from './IndexOf.type';
export type { IsEmptyObject } from './IsEmptyObject.type';
export type { IsEmptyRecord } from './IsEmptyRecord.type';
export type { IsExactly } from './IsExactly';
export type { IsGreaterOrEqual } from './IsGreaterOrEqual.type';
export type { IsNever } from './IsNever.type';
export type { IsSerializedRelation } from './IsSerializedRelation.type';
export type { LogicFunctionEvent } from './LogicFunctionEvent';