diff --git a/packages/twenty-server/src/database/commands/database-command.module.ts b/packages/twenty-server/src/database/commands/database-command.module.ts index e66104a6af..3b57304240 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -17,6 +17,7 @@ import { WorkspaceExportModule } from 'src/database/commands/workspace-export/wo import { TypeORMModule } from 'src/database/typeorm/typeorm.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 { ApplicationInstallModule } from 'src/engine/core-modules/application/application-install/application-install.module'; import { MarketplaceModule } from 'src/engine/core-modules/application/application-marketplace/marketplace.module'; import { StaleRegistrationCleanupModule } from 'src/engine/core-modules/application/application-oauth/stale-registration-cleanup/stale-registration-cleanup.module'; import { ApplicationUpgradeModule } from 'src/engine/core-modules/application/application-upgrade/application-upgrade.module'; @@ -88,6 +89,7 @@ import { WorkflowCoreConsistencyModule } from 'src/modules/workflow/workflow-cor EnterpriseModule, TwentyConfigModule, MarketplaceModule, + ApplicationInstallModule, ApplicationUpgradeModule, StaleRegistrationCleanupModule, PreInstalledAppsModule, diff --git a/packages/twenty-server/src/database/commands/utils/parse-bounded-positive-integer.util.ts b/packages/twenty-server/src/database/commands/utils/parse-bounded-positive-integer.util.ts new file mode 100644 index 0000000000..4311cf6de4 --- /dev/null +++ b/packages/twenty-server/src/database/commands/utils/parse-bounded-positive-integer.util.ts @@ -0,0 +1,19 @@ +export const parseBoundedPositiveInteger = ( + value: string, + optionName: string, + maximum: number, +): number => { + const parsedValue = Number(value); + + if (!Number.isInteger(parsedValue) || parsedValue < 1) { + throw new Error( + `Invalid ${optionName} "${value}". Expected a positive integer`, + ); + } + + if (parsedValue > maximum) { + throw new Error(`Invalid ${optionName} "${value}". Maximum is ${maximum}`); + } + + return parsedValue; +}; diff --git a/packages/twenty-server/src/engine/core-modules/application/application-install/application-install.module.ts b/packages/twenty-server/src/engine/core-modules/application/application-install/application-install.module.ts index 98186a4763..793c81c6cc 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-install/application-install.module.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-install/application-install.module.ts @@ -1,6 +1,7 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; +import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module'; import { CacheLockModule } from 'src/engine/core-modules/cache-lock/cache-lock.module'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; @@ -11,6 +12,8 @@ import { ApplicationPackageModule } from 'src/engine/core-modules/application/ap import { MarketplaceModule } from 'src/engine/core-modules/application/application-marketplace/marketplace.module'; import { ApplicationInstallResolver } from 'src/engine/core-modules/application/application-install/application-install.resolver'; import { ApplicationInstallService } from 'src/engine/core-modules/application/application-install/application-install.service'; +import { InstallApplicationCommand } from 'src/engine/core-modules/application/application-install/commands/install-application.command'; +import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; import { FileStorageModule } from 'src/engine/core-modules/file-storage/file-storage.module'; import { LogicFunctionModule } from 'src/engine/core-modules/logic-function/logic-function.module'; import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; @@ -20,7 +23,10 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache @Module({ imports: [ - TypeOrmModule.forFeature([ApplicationRegistrationEntity]), + TypeOrmModule.forFeature([ + ApplicationEntity, + ApplicationRegistrationEntity, + ]), ApplicationModule, ApplicationRegistrationModule, ApplicationManifestModule, @@ -34,8 +40,13 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache PermissionsModule, FileStorageModule, WorkspaceCacheModule, + WorkspaceIteratorModule, + ], + providers: [ + ApplicationInstallResolver, + ApplicationInstallService, + InstallApplicationCommand, ], - providers: [ApplicationInstallResolver, ApplicationInstallService], exports: [ApplicationInstallService], }) export class ApplicationInstallModule {} diff --git a/packages/twenty-server/src/engine/core-modules/application/application-install/commands/install-application.command.ts b/packages/twenty-server/src/engine/core-modules/application/application-install/commands/install-application.command.ts new file mode 100644 index 0000000000..d250df6208 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/application/application-install/commands/install-application.command.ts @@ -0,0 +1,281 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +import chalk from 'chalk'; +import { Command, CommandRunner, Option } from 'nest-commander'; +import { isDefined } from 'twenty-shared/utils'; +import { In, Repository } from 'typeorm'; + +import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service'; +import { CommandLogger } from 'src/database/commands/logger'; +import { askCommandConfirmation } from 'src/database/commands/utils/ask-command-confirmation.util'; +import { parseBoundedPositiveInteger } from 'src/database/commands/utils/parse-bounded-positive-integer.util'; +import { ApplicationInstallService } from 'src/engine/core-modules/application/application-install/application-install.service'; +import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; +import { ApplicationRegistrationSourceType } from 'src/engine/core-modules/application/application-registration/enums/application-registration-source-type.enum'; +import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; + +type InstallApplicationCommandOptions = { + applicationRegistrationUniversalIdentifier: string; + workspaceId?: Set; + workspaceCountLimit?: number; + dryRun?: boolean; + yes?: boolean; +}; + +const MAX_WORKSPACE_COUNT_LIMIT = 50; + +@Command({ + name: 'application:install', + description: + 'Install an application on every workspace that does not have it yet. Workspaces where it is already installed are left untouched, use application:upgrade for those', +}) +export class InstallApplicationCommand extends CommandRunner { + protected logger: CommandLogger; + + constructor( + @InjectRepository(ApplicationRegistrationEntity) + private readonly applicationRegistrationRepository: Repository, + @InjectRepository(ApplicationEntity) + private readonly applicationRepository: Repository, + private readonly applicationInstallService: ApplicationInstallService, + private readonly workspaceIteratorService: WorkspaceIteratorService, + ) { + super(); + this.logger = new CommandLogger({ + verbose: false, + constructorName: this.constructor.name, + }); + } + + @Option({ + flags: + '-u, --application-registration-universal-identifier ', + description: 'Application registration universal identifier', + required: true, + }) + parseApplicationRegistrationUniversalIdentifier(value: string): string { + return value; + } + + @Option({ + flags: '-w, --workspace-id ', + description: + 'Only install on the given workspace id. Can be repeated to target several workspaces. Targets all provisioned workspaces if not provided.', + required: false, + }) + parseWorkspaceId(value: string, previous?: Set): Set { + const accumulator = previous ?? new Set(); + + accumulator.add(value); + + return accumulator; + } + + @Option({ + flags: '--workspace-count-limit ', + description: `Limit the number of workspaces to iterate over (max ${MAX_WORKSPACE_COUNT_LIMIT})`, + required: false, + }) + parseWorkspaceCountLimit(value: string): number { + return parseBoundedPositiveInteger( + value, + 'workspace count limit', + MAX_WORKSPACE_COUNT_LIMIT, + ); + } + + @Option({ + flags: '-d, --dry-run', + description: + 'List the workspaces that would be installed without installing', + required: false, + }) + parseDryRun(): boolean { + return true; + } + + @Option({ + flags: '-y, --yes', + description: 'Skip the confirmation prompt (for non-interactive usage)', + required: false, + }) + parseYes(): boolean { + return true; + } + + override async run( + _passedParams: string[], + options: InstallApplicationCommandOptions, + ): Promise { + const registration = await this.applicationRegistrationRepository.findOne({ + where: { + universalIdentifier: options.applicationRegistrationUniversalIdentifier, + }, + }); + + if (!isDefined(registration)) { + throw new Error( + `Application registration with universal identifier ${options.applicationRegistrationUniversalIdentifier} not found`, + ); + } + + if ( + registration.sourceType === ApplicationRegistrationSourceType.LOCAL || + registration.sourceType === ApplicationRegistrationSourceType.OAUTH_ONLY + ) { + throw new Error( + `Cannot install application ${registration.universalIdentifier}: applications with source type ${registration.sourceType} have no code artifacts to install`, + ); + } + + const targetVersion = registration.latestAvailableVersion; + const versionLabel = targetVersion ?? 'latest available'; + + const isDryRun = options.dryRun ?? false; + + const requestedWorkspaceIds = isDefined(options.workspaceId) + ? Array.from(options.workspaceId) + : undefined; + + // Explicit ids bypass the iterator's own workspace selection, so the count + // limit and the already-installed filter are applied here instead. + const alreadyInstalledRequestedWorkspaceIds = isDefined( + requestedWorkspaceIds, + ) + ? await this.findAlreadyInstalledWorkspaceIds({ + universalIdentifier: registration.universalIdentifier, + workspaceIds: requestedWorkspaceIds, + }) + : undefined; + + const workspaceIdsToIterate = isDefined(requestedWorkspaceIds) + ? requestedWorkspaceIds + .filter( + (workspaceId) => + !alreadyInstalledRequestedWorkspaceIds?.has(workspaceId), + ) + .slice(0, options.workspaceCountLimit) + : undefined; + + if ( + isDefined(workspaceIdsToIterate) && + workspaceIdsToIterate.length === 0 + ) { + this.logger.log( + `No workspace to install, every targeted workspace already has "${registration.name}" installed`, + ); + + return; + } + + if (!isDryRun && !(options.yes ?? false)) { + const isConfirmed = await askCommandConfirmation( + `Confirm installing application ${registration.universalIdentifier} version ${versionLabel} on ${this.describeConfirmationTarget({ workspaceIdsToIterate, workspaceCountLimit: options.workspaceCountLimit })}`, + ); + + if (!isConfirmed) { + this.logger.log('Aborted, no installation performed'); + + return; + } + } + + let skippedWorkspaceCount = 0; + + const prefilteredWorkspaceCount = + alreadyInstalledRequestedWorkspaceIds?.size ?? 0; + + const report = await this.workspaceIteratorService.iterate({ + workspaceIds: workspaceIdsToIterate, + workspaceCountLimit: isDefined(workspaceIdsToIterate) + ? undefined + : options.workspaceCountLimit, + dryRun: isDryRun, + callback: async ({ workspaceId }) => { + if ( + await this.isApplicationInstalled({ + universalIdentifier: registration.universalIdentifier, + workspaceId, + }) + ) { + skippedWorkspaceCount += 1; + + this.logger.log( + `Skipping workspace ${workspaceId}: "${registration.name}" is already installed, run application:upgrade to update it`, + ); + + return; + } + + if (isDryRun) { + this.logger.log( + `[DRY RUN] Would install "${registration.name}" (${registration.universalIdentifier}) version ${versionLabel} on workspace ${workspaceId}`, + ); + + return; + } + + await this.applicationInstallService.installApplication({ + appRegistrationId: registration.id, + version: targetVersion ?? undefined, + workspaceId, + }); + }, + }); + + this.logger.log( + `${isDryRun ? '[DRY RUN] ' : ''}Installed on ${report.success.length - skippedWorkspaceCount} workspace(s), skipped ${skippedWorkspaceCount + prefilteredWorkspaceCount} already installed, ${report.fail.length} failed`, + ); + + this.logger.log(chalk.blue('Command completed!')); + } + + private describeConfirmationTarget({ + workspaceIdsToIterate, + workspaceCountLimit, + }: { + workspaceIdsToIterate?: string[]; + workspaceCountLimit?: number; + }): string { + if (isDefined(workspaceIdsToIterate)) { + return `workspace(s) ${workspaceIdsToIterate.join(', ')}`; + } + + return isDefined(workspaceCountLimit) + ? `up to ${workspaceCountLimit} provisioned workspace(s) that do not have it yet` + : 'every provisioned workspace that does not have it yet'; + } + + private async findAlreadyInstalledWorkspaceIds({ + universalIdentifier, + workspaceIds, + }: { + universalIdentifier: string; + workspaceIds: string[]; + }): Promise> { + const existingApplications = await this.applicationRepository.find({ + select: ['workspaceId'], + where: { universalIdentifier, workspaceId: In(workspaceIds) }, + }); + + return new Set( + existingApplications.map((application) => application.workspaceId), + ); + } + + // Matches on the universal identifier, the same identity + // ApplicationInstallService uses to decide between a fresh install and a + // version upgrade, so a row with a stale registration id is not mistaken + // for a missing installation. + private async isApplicationInstalled({ + universalIdentifier, + workspaceId, + }: { + universalIdentifier: string; + workspaceId: string; + }): Promise { + return this.applicationRepository.exists({ + where: { universalIdentifier, workspaceId }, + }); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.module.ts b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.module.ts index 2b71d52c26..253ce27ef7 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.module.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.module.ts @@ -1,6 +1,7 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; +import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module'; import { ApplicationInstallModule } from 'src/engine/core-modules/application/application-install/application-install.module'; import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; @@ -25,6 +26,7 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi FeatureFlagModule, PermissionsModule, TwentyConfigModule, + WorkspaceIteratorModule, ], providers: [ ApplicationUpgradeService, diff --git a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.service.ts b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.service.ts index b542f62465..e8dde87460 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.service.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/application-upgrade.service.ts @@ -6,6 +6,10 @@ import { isDefined, isNonEmptyArray } from 'twenty-shared/utils'; import { In, Repository } from 'typeorm'; import { z } from 'zod'; +import { + WorkspaceIteratorService, + type WorkspaceIteratorReport, +} from 'src/database/commands/command-runners/workspace-iterator.service'; import { ApplicationInstallService } from 'src/engine/core-modules/application/application-install/application-install.service'; import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; @@ -21,8 +25,6 @@ const npmPackageMetadataSchema = z.object({ version: z.string(), }); -const UPGRADE_APPLICATIONS_DEFAULT_BATCH_SIZE = 5; - @Injectable() export class ApplicationUpgradeService { private readonly logger = new Logger(ApplicationUpgradeService.name); @@ -35,6 +37,7 @@ export class ApplicationUpgradeService { private readonly applicationInstallService: ApplicationInstallService, private readonly applicationRegistrationService: ApplicationRegistrationService, private readonly twentyConfigService: TwentyConfigService, + private readonly workspaceIteratorService: WorkspaceIteratorService, ) {} async checkForUpdates( @@ -168,54 +171,38 @@ export class ApplicationUpgradeService { appRegistration, targetVersion, applications, - batchSize = UPGRADE_APPLICATIONS_DEFAULT_BATCH_SIZE, }: { appRegistration: ApplicationRegistrationEntity; targetVersion: string; applications: ApplicationEntity[]; - batchSize?: number; - }): Promise { - const sanitizedBatchSize = Math.max(1, Math.floor(batchSize)); - - for ( - let batchStart = 0; - batchStart < applications.length; - batchStart += sanitizedBatchSize - ) { - const batch = applications.slice( - batchStart, - batchStart + sanitizedBatchSize, - ); - - await Promise.all( - batch.map(async (application) => { - try { - await this.upgradeApplicationToVersion({ - appRegistration, - targetVersion, - workspaceId: application.workspaceId, - }); - } catch (error) { - this.logger.error( - `Failed to upgrade application ${application.id} to version ${targetVersion} in workspace ${application.workspaceId}`, - error, - ); - } - }), - ); + }): Promise { + // An empty workspace id list makes the iterator fall back to every + // provisioned workspace, which would upgrade workspaces that were + // filtered out. + if (!isNonEmptyArray(applications)) { + return { success: [], fail: [] }; } + + return this.workspaceIteratorService.iterate({ + workspaceIds: applications.map((application) => application.workspaceId), + callback: async ({ workspaceId }) => { + await this.upgradeApplicationToVersion({ + appRegistration, + targetVersion, + workspaceId, + }); + }, + }); } async upgradeAllApplications({ applicationRegistrationId, onlyAutoUpgrade = false, - batchSize = UPGRADE_APPLICATIONS_DEFAULT_BATCH_SIZE, workspaceIds, workspaceCountLimit, }: { applicationRegistrationId: string; onlyAutoUpgrade?: boolean; - batchSize?: number; workspaceIds?: string[]; workspaceCountLimit?: number; }): Promise { @@ -235,7 +222,6 @@ export class ApplicationUpgradeService { appRegistration, targetVersion, applications: applicationsToUpgrade, - batchSize, }); } diff --git a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/commands/upgrade-application.command.ts b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/commands/upgrade-application.command.ts index 07c777bedf..e26bce931b 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-upgrade/commands/upgrade-application.command.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-upgrade/commands/upgrade-application.command.ts @@ -7,41 +7,20 @@ import { Repository } from 'typeorm'; import { CommandLogger } from 'src/database/commands/logger'; import { askCommandConfirmation } from 'src/database/commands/utils/ask-command-confirmation.util'; +import { parseBoundedPositiveInteger } from 'src/database/commands/utils/parse-bounded-positive-integer.util'; import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; import { ApplicationUpgradeService } from 'src/engine/core-modules/application/application-upgrade/application-upgrade.service'; type UpgradeApplicationCommandOptions = { applicationRegistrationUniversalIdentifier: string; - batchSize?: number; workspaceId?: Set; workspaceCountLimit?: number; dryRun?: boolean; yes?: boolean; }; -const MAX_BATCH_SIZE = 50; const MAX_WORKSPACE_COUNT_LIMIT = 50; -const parseBoundedPositiveInteger = ( - value: string, - optionName: string, - maximum: number, -): number => { - const parsedValue = Number(value); - - if (!Number.isInteger(parsedValue) || parsedValue < 1) { - throw new Error( - `Invalid ${optionName} "${value}". Expected a positive integer`, - ); - } - - if (parsedValue > maximum) { - throw new Error(`Invalid ${optionName} "${value}". Maximum is ${maximum}`); - } - - return parsedValue; -}; - @Command({ name: 'application:upgrade', description: @@ -72,15 +51,6 @@ export class UpgradeApplicationCommand extends CommandRunner { return value; } - @Option({ - flags: '-b, --batch-size ', - description: `Number of workspaces upgraded in parallel (defaults to 5, max ${MAX_BATCH_SIZE})`, - required: false, - }) - parseBatchSize(value: string): number { - return parseBoundedPositiveInteger(value, 'batch size', MAX_BATCH_SIZE); - } - @Option({ flags: '-w, --workspace-id ', description: @@ -208,13 +178,16 @@ export class UpgradeApplicationCommand extends CommandRunner { // Runs on the exact set shown at confirmation time, so installations // created or versions published while the operator answered are excluded. - await this.applicationUpgradeService.upgradeApplications({ + const report = await this.applicationUpgradeService.upgradeApplications({ appRegistration, targetVersion, applications: applicationsToUpgrade, - batchSize: options.batchSize, }); + this.logger.log( + `Upgraded ${report.success.length} workspace(s), ${report.fail.length} failed`, + ); + this.logger.log(chalk.blue('Command completed!')); } } diff --git a/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job-constants.ts b/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job-constants.ts index 1b3afb47f6..1846a4b0cb 100644 --- a/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job-constants.ts +++ b/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job-constants.ts @@ -3,5 +3,4 @@ export const UPGRADE_APPLICATIONS_JOB_NAME = 'UpgradeApplicationsJob'; export type UpgradeApplicationsJobData = { applicationRegistrationId: string; onlyAutoUpgrade: boolean; - batchSize?: number; }; diff --git a/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job.ts b/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job.ts index 161dda47d2..0e208071d7 100644 --- a/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job.ts +++ b/packages/twenty-server/src/engine/core-modules/application/jobs/upgrade-applications.job.ts @@ -18,7 +18,6 @@ export class UpgradeApplicationsJob { await this.applicationUpgradeService.upgradeAllApplications({ applicationRegistrationId: data.applicationRegistrationId, onlyAutoUpgrade: data.onlyAutoUpgrade, - batchSize: data.batchSize, }); } }