feat(deployment): add concurrency for top up deployment

apps/api/src/billing/services/provider-cleanup/provider-cleanup.service.ts

@@ -4,17 +4,12 @@ import { singleton } from "tsyringe";
 import { BillingConfig, InjectBillingConfig } from "@src/billing/providers";
 import { UserWalletOutput, UserWalletRepository } from "@src/billing/repositories";
 import { ManagedUserWalletService, RpcMessageService } from "@src/billing/services";
+import { ProviderCleanupParams } from "@src/billing/types/provider-cleanup";
 import { ErrorService } from "@src/core/services/error/error.service";
 import { ProviderCleanupSummarizer } from "@src/deployment/lib/provider-cleanup-summarizer/provider-cleanup-summarizer";
 import { DeploymentRepository } from "@src/deployment/repositories/deployment/deployment.repository";
 import { TxSignerService } from "../tx-signer/tx-signer.service";
 
-export interface ProviderCleanupParams {
-  concurrency: number;
-  provider: string;
-  dryRun: boolean;
-}
-
 @singleton()
 export class ProviderCleanupService {
   private readonly logger = LoggerService.forContext(ProviderCleanupService.name);

apps/api/src/billing/types/provider-cleanup.ts

@@ -0,0 +1,5 @@
+import { ConcurrencyOptions, DryRunOptions } from "@src/core/types/console";
+
+export interface ProviderCleanupParams extends DryRunOptions, ConcurrencyOptions {
+  provider: string;
+}

apps/api/src/console.ts

@@ -33,10 +33,11 @@ program
 program
   .command("top-up-deployments")
   .description("Refill deployments with auto top up enabled")
+  .option("-c, --concurrency <number>", "How many wallets is processed concurrently", value => z.number({ coerce: true }).optional().default(10).parse(value))
   .option("-d, --dry-run", "Dry run the top up deployments", false)
   .action(async (options, command) => {
     await executeCliHandler(command.name(), async () => {
-      await container.resolve(TopUpDeploymentsController).topUpDeployments({ dryRun: options.dryRun });
+      await container.resolve(TopUpDeploymentsController).topUpDeployments(options);
     });
   });
 

apps/api/src/core/types/console.ts

@@ -1,3 +1,8 @@
 export interface DryRunOptions {
   dryRun: boolean;
 }
+
+export interface ConcurrencyOptions {
+  concurrency?: number;
+}
+

apps/api/src/deployment/controllers/deployment/deployment.controller.ts

@@ -1,12 +1,10 @@
 import { singleton } from "tsyringe";
 
-import {
-  CleanUpStaleDeploymentsParams,
-  StaleManagedDeploymentsCleanerService
-} from "@src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service";
+import { StaleManagedDeploymentsCleanerService } from "@src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service";
 import { TopUpCustodialDeploymentsService } from "@src/deployment/services/top-up-custodial-deployments/top-up-custodial-deployments.service";
 import { TopUpManagedDeploymentsService } from "@src/deployment/services/top-up-managed-deployments/top-up-managed-deployments.service";
 import { TopUpDeploymentsOptions } from "@src/deployment/types/deployments-refiller";
+import { CleanUpStaleDeploymentsParams } from "@src/deployment/types/state-deployments";
 
 @singleton()
 export class TopUpDeploymentsController {

apps/api/src/deployment/controllers/provider/provider.controller.ts

@@ -1,6 +1,7 @@
 import { singleton } from "tsyringe";
 
-import { ProviderCleanupParams, ProviderCleanupService } from "@src/billing/services/provider-cleanup/provider-cleanup.service";
+import { ProviderCleanupService } from "@src/billing/services/provider-cleanup/provider-cleanup.service";
+import { ProviderCleanupParams } from "@src/billing/types/provider-cleanup";
 import { TrialProvidersService } from "@src/deployment/services/trial-providers/trial-providers.service";
 
 @singleton()

apps/api/src/deployment/services/stale-managed-deployments-cleaner/stale-managed-deployments-cleaner.service.ts

@@ -9,12 +9,9 @@ import { TxSignerService } from "@src/billing/services/tx-signer/tx-signer.servi
 import { BlockRepository } from "@src/chain/repositories/block.repository";
 import { ErrorService } from "@src/core/services/error/error.service";
 import { DeploymentRepository } from "@src/deployment/repositories/deployment/deployment.repository";
+import { CleanUpStaleDeploymentsParams } from "@src/deployment/types/state-deployments";
 import { averageBlockTime } from "@src/utils/constants";
 
-export interface CleanUpStaleDeploymentsParams {
-  concurrency: number;
-}
-
 @singleton()
 export class StaleManagedDeploymentsCleanerService {
   private readonly logger = LoggerService.forContext(StaleManagedDeploymentsCleanerService.name);

apps/api/src/deployment/services/top-up-managed-deployments/top-up-managed-deployments.service.ts

@@ -15,8 +15,6 @@ import { DeploymentsRefiller, TopUpDeploymentsOptions } from "@src/deployment/ty
 
 @singleton()
 export class TopUpManagedDeploymentsService implements DeploymentsRefiller {
-  private readonly CONCURRENCY = 10;
-
   private readonly logger = LoggerService.forContext(TopUpManagedDeploymentsService.name);
 
   constructor(
@@ -35,7 +33,7 @@ export class TopUpManagedDeploymentsService implements DeploymentsRefiller {
     const summary = new TopUpSummarizer();
     summary.set("startBlockHeight", await this.blockHttpService.getCurrentHeight());
 
-    await this.userWalletRepository.paginate({ limit: this.CONCURRENCY }, async wallets => {
+    await this.userWalletRepository.paginate({ limit: options.concurrency || 10 }, async wallets => {
       await Promise.all(
         wallets.map(async wallet => {
           await this.errorService.execWithErrorHandler(

apps/api/src/deployment/types/deployments-refiller.ts

@@ -1,6 +1,6 @@
-import { DryRunOptions } from "@src/core/types/console";
+import { ConcurrencyOptions, DryRunOptions } from "@src/core/types/console";
 
-export interface TopUpDeploymentsOptions extends DryRunOptions {}
+export interface TopUpDeploymentsOptions extends DryRunOptions, ConcurrencyOptions {}
 
 export interface DeploymentsRefiller {
   topUpDeployments(options: TopUpDeploymentsOptions): Promise<void>;

apps/api/src/deployment/types/state-deployments.ts

@@ -0,0 +1,3 @@
+import { ConcurrencyOptions } from "@src/core/types/console";
+
+export interface CleanUpStaleDeploymentsParams extends ConcurrencyOptions {}