feat(server): add application:install command (#23430)

Adds `application:install` to install an application on workspaces that do not have it yet, and moves both application commands onto `WorkspaceIteratorService`. ### Behavior - Iterates provisioned workspaces through `WorkspaceIteratorService` (workspace id resolution, workspace context, per-workspace success/fail report), or the ones passed with `-w`. - Workspaces where the application is already installed are skipped, with a log line pointing at `application:upgrade`. This command never upgrades. - Installed workspaces are detected by `universalIdentifier`, the same identity `ApplicationInstallService` uses to tell a fresh install from a version upgrade, checked per workspace on the `(universalIdentifier, workspaceId)` unique index. - `--workspace-count-limit` caps both the iterator's own selection and an explicitly targeted `-w` list. - Fails fast for `LOCAL` and `OAUTH_ONLY` registrations, which have no code artifacts to install. - Per-workspace failures are collected in the iterator report and never abort the run; the command ends with an installed / skipped / failed summary. ### Options | Flag | Description | | --- | --- | | `-u, --application-registration-universal-identifier` | Application registration universal identifier (required) | | `-w, --workspace-id` | Target a specific workspace, repeatable | | `--workspace-count-limit` | Cap the number of workspaces to iterate over (max 50) | | `-d, --dry-run` | Print the workspaces that would be installed without installing | | `-y, --yes` | Skip the confirmation prompt | ### Example ``` yarn command:prod application:install -u UNIVERSAL_IDENTIFIER --dry-run ``` ### Changes to application:upgrade - `ApplicationUpgradeService.upgradeApplications` iterates through `WorkspaceIteratorService` and returns its report, replacing the hand-rolled parallel batching. - `--batch-size` dropped from the command, and `batchSize` dropped from the service and from `UpgradeApplicationsJobData`, since `iterate()` is sequential. - `parseBoundedPositiveInteger` moved to `src/database/commands/utils/` and is shared by both commands. ### Files - `application-install/commands/install-application.command.ts` (new) - `src/database/commands/utils/parse-bounded-positive-integer.util.ts` (new) - `application-upgrade/application-upgrade.service.ts`, `application-upgrade/commands/upgrade-application.command.ts`, `jobs/upgrade-applications.job*`: iterator instead of batching - `application-install.module.ts` / `application-upgrade.module.ts`: register the command, wire `WorkspaceIteratorModule` - `database-command.module.ts`: import `ApplicationInstallModule` so the command is discovered by the CLI ### Testing - `npx jest src/engine/core-modules/application` (32 suites, 181 tests passing) - `npx nx typecheck twenty-server` - `oxlint --type-aware` and `oxfmt` on the changed files
2026-07-28 17:02:32 +02:00
parent 942755d0dd
commit f5da810e59
9 changed files with 345 additions and 73 deletions
@@ -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 {}
@@ -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<string>;
+  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<ApplicationRegistrationEntity>,
+    @InjectRepository(ApplicationEntity)
+    private readonly applicationRepository: Repository<ApplicationEntity>,
+    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 <application_registration_universal_identifier>',
+    description: 'Application registration universal identifier',
+    required: true,
+  })
+  parseApplicationRegistrationUniversalIdentifier(value: string): string {
+    return value;
+  }
+
+  @Option({
+    flags: '-w, --workspace-id <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<string>): Set<string> {
+    const accumulator = previous ?? new Set<string>();
+
+    accumulator.add(value);
+
+    return accumulator;
+  }
+
+  @Option({
+    flags: '--workspace-count-limit <count>',
+    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<void> {
+    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<Set<string>> {
+    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<boolean> {
+    return this.applicationRepository.exists({
+      where: { universalIdentifier, workspaceId },
+    });
+  }
+}
@@ -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,
@@ -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<void> {
-    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<WorkspaceIteratorReport> {
+    // 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<void> {
@@ -235,7 +222,6 @@ export class ApplicationUpgradeService {
      appRegistration,
      targetVersion,
      applications: applicationsToUpgrade,
-      batchSize,
    });
  }

@@ -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<string>;
  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 <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 <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!'));
  }
 }
@@ -3,5 +3,4 @@ export const UPGRADE_APPLICATIONS_JOB_NAME = 'UpgradeApplicationsJob';
 export type UpgradeApplicationsJobData = {
  applicationRegistrationId: string;
  onlyAutoUpgrade: boolean;
-  batchSize?: number;
 };
@@ -18,7 +18,6 @@ export class UpgradeApplicationsJob {
    await this.applicationUpgradeService.upgradeAllApplications({
      applicationRegistrationId: data.applicationRegistrationId,
      onlyAutoUpgrade: data.onlyAutoUpgrade,
-      batchSize: data.batchSize,
    });
  }
 }