feat(deployment): implements custodial wallet balances collection for…

… top up

Also assembles rough service interface for further development

refs #395

apps/api/src/billing/providers/http-sdk.provider.ts

@@ -1,6 +1,8 @@
-import { AllowanceHttpService } from "@akashnetwork/http-sdk";
+import { AllowanceHttpService, BalanceHttpService } from "@akashnetwork/http-sdk";
 import { container } from "tsyringe";
 
 import { apiNodeUrl } from "@src/utils/constants";
 
-container.register(AllowanceHttpService, { useValue: new AllowanceHttpService({ baseURL: apiNodeUrl }) });
+const SERVICES = [BalanceHttpService, AllowanceHttpService];
+
+SERVICES.forEach(Service => container.register(Service, { useValue: new Service({ baseURL: apiNodeUrl }) }));

apps/api/src/billing/providers/wallet.provider.ts

@@ -7,10 +7,10 @@ import { MasterWalletType } from "@src/billing/types/wallet.type";
 export const MANAGED_MASTER_WALLET = "MANAGED_MASTER_WALLET";
 container.register(MANAGED_MASTER_WALLET, { useFactory: () => new MasterWalletService(config.MASTER_WALLET_MNEMONIC) });
 
-export const UAKT_TOP_UP_MASTER_WALLET = "TOP_UP_UAKT_MASTER_WALLET";
+export const UAKT_TOP_UP_MASTER_WALLET = "UAKT_TOP_UP_MASTER_WALLET";
 container.register(UAKT_TOP_UP_MASTER_WALLET, { useFactory: () => new MasterWalletService(config.UAKT_TOP_UP_MASTER_WALLET_MNEMONIC) });
 
-export const USDC_TOP_UP_MASTER_WALLET = "TOP_UP_USDC_MASTER_WALLET";
+export const USDC_TOP_UP_MASTER_WALLET = "USDC_TOP_UP_MASTER_WALLET";
 container.register(USDC_TOP_UP_MASTER_WALLET, { useFactory: () => new MasterWalletService(config.USDC_TOP_UP_MASTER_WALLET_MNEMONIC) });
 
 export const InjectWallet = (walletType: MasterWalletType) => inject(`${walletType}_MASTER_WALLET`);

apps/api/src/billing/services/balances/balances.service.ts

@@ -45,11 +45,11 @@ export class BalancesService {
   }
 
   async getFreshLimits(userWallet: UserWalletOutput): Promise<{ fee: number; deployment: number }> {
-    const [fee, deployment] = await Promise.all([this.calculateFeeLimit(userWallet), this.calculateDeploymentLimit(userWallet)]);
+    const [fee, deployment] = await Promise.all([this.retrieveAndCalcFeeLimit(userWallet), this.retrieveAndCalcDeploymentLimit(userWallet)]);
     return { fee, deployment };
   }
 
-  private async calculateFeeLimit(userWallet: UserWalletOutput): Promise<number> {
+  private async retrieveAndCalcFeeLimit(userWallet: UserWalletOutput): Promise<number> {
     const feeAllowance = await this.allowanceHttpService.getFeeAllowancesForGrantee(userWallet.address);
     const masterWalletAddress = await this.masterWalletService.getFirstAddress();
 
@@ -68,7 +68,7 @@ export class BalancesService {
     }, 0);
   }
 
-  async calculateDeploymentLimit(userWallet: UserWalletOutput): Promise<number> {
+  async retrieveAndCalcDeploymentLimit(userWallet: Pick<UserWalletOutput, "address">): Promise<number> {
     const deploymentAllowance = await this.allowanceHttpService.getDeploymentAllowancesForGrantee(userWallet.address);
     const masterWalletAddress = await this.masterWalletService.getFirstAddress();
 

apps/api/src/billing/services/refill/refill.service.ts

@@ -55,7 +55,7 @@ export class RefillService {
     let currentLimit: number = 0;
 
     if (userWallet) {
-      currentLimit = await this.balancesService.calculateDeploymentLimit(userWallet);
+      currentLimit = await this.balancesService.retrieveAndCalcDeploymentLimit(userWallet);
     } else {
       userWallet = await this.walletInitializerService.initialize(userId);
     }

apps/api/src/console.ts

@@ -8,6 +8,7 @@ import { container } from "tsyringe";
 
 import { WalletController } from "@src/billing/controllers/wallet/wallet.controller";
 import { LoggerService } from "@src/core";
+import { chainDb } from "@src/db/dbConnection";
 import { TopUpDeploymentsController } from "@src/deployment/controllers/deployment/deployment.controller";
 
 const program = new Command();
@@ -40,10 +41,12 @@ async function executeCliHandler(name: string, handler: () => Promise<void>) {
     // eslint-disable-next-line @typescript-eslint/no-var-requires
     const { migratePG, closeConnections } = await require("./core/providers/postgres.provider");
     await migratePG();
+    await chainDb.authenticate();
 
     await handler();
 
     await closeConnections();
+    await chainDb.close();
     logger.info({ event: "COMMAND_END", name });
   });
 }

apps/api/src/core/config/env.config.ts

@@ -8,8 +8,7 @@ const envSchema = z.object({
   FLUENTD_HOST: z.string().optional(),
   FLUENTD_PORT: z.number({ coerce: true }).optional().default(24224),
   NODE_ENV: z.enum(["development", "production", "test"]).optional().default("development"),
-  // TODO: make required once billing is in prod
-  POSTGRES_DB_URI: z.string().optional(),
+  POSTGRES_DB_URI: z.string(),
   POSTGRES_MAX_CONNECTIONS: z.number({ coerce: true }).optional().default(20),
   DRIZZLE_MIGRATIONS_FOLDER: z.string().optional().default("./drizzle"),
   DEPLOYMENT_ENV: z.string().optional().default("production"),

apps/api/src/core/repositories/base.repository.ts

@@ -1,5 +1,5 @@
 import { AnyAbility } from "@casl/ability";
-import { and, eq } from "drizzle-orm";
+import { and, DBQueryConfig, eq } from "drizzle-orm";
 import { PgTableWithColumns } from "drizzle-orm/pg-core/table";
 import { SQL } from "drizzle-orm/sql/sql";
 import first from "lodash/first";
@@ -78,12 +78,16 @@ export abstract class BaseRepository<
     return this.toOutput(first(items));
   }
 
-  async find(query?: Partial<Output>) {
-    return this.toOutputList(
-      await this.queryCursor.findMany({
-        where: this.queryToWhere(query)
-      })
-    );
+  async find(query?: Partial<Output>, select?: Array<keyof Output>) {
+    const params: DBQueryConfig<"many", true> = {
+      where: this.queryToWhere(query)
+    };
+
+    if (select) {
+      params.columns = select.reduce((acc, field) => ({ ...acc, [field]: true }), {});
+    }
+
+    return this.toOutputList(await this.queryCursor.findMany(params));
   }
 
   async updateById(id: Output["id"], payload: Partial<Input>, options?: MutationOptions): Promise<Output>;

apps/api/src/core/services/postgres-logger/postgres-logger.service.ts

@@ -5,6 +5,7 @@ import { LoggerService } from "@src/core/services/logger/logger.service";
 
 interface PostgresLoggerServiceOptions {
   orm?: "drizzle" | "sequelize";
+  database?: string;
   useFormat?: boolean;
 }
 
@@ -17,7 +18,7 @@ export class PostgresLoggerService implements LogWriter {
 
   constructor(options?: PostgresLoggerServiceOptions) {
     const orm = options?.orm || "drizzle";
-    this.logger = new LoggerService({ context: "POSTGRES", orm });
+    this.logger = new LoggerService({ context: "POSTGRES", orm, database: options?.database });
     this.isDrizzle = orm === "drizzle";
     this.useFormat = options?.useFormat || false;
   }

apps/api/src/deployment/repositories/lease/lease.repository.ts

@@ -0,0 +1,32 @@
+import { Lease } from "@akashnetwork/database/dbSchemas/akash";
+import { col, fn, Op } from "sequelize";
+import { singleton } from "tsyringe";
+
+interface DrainingLeasesOptions {
+  closureHeight: number;
+  owner: string;
+}
+
+export interface DrainingDeploymentOutput {
+  dseq: number;
+  denom: string;
+  blockRate: number;
+}
+
+@singleton()
+export class LeaseRepository {
+  async findDrainingLeases(options: DrainingLeasesOptions): Promise<DrainingDeploymentOutput[]> {
+    return (await Lease.findAll({
+      where: {
+        closedHeight: null,
+        owner: options.owner,
+        predictedClosedHeight: {
+          [Op.lte]: options.closureHeight
+        }
+      },
+      attributes: ["dseq", "denom", [fn("sum", col("price")), "blockRate"]],
+      group: ["dseq", "denom"],
+      plain: true
+    })) as unknown as DrainingDeploymentOutput[];
+  }
+}

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

@@ -1,12 +1,135 @@
+import { AllowanceHttpService, BalanceHttpService, DeploymentAllowance } from "@akashnetwork/http-sdk";
+import { PromisePool } from "@supercharge/promise-pool";
 import { singleton } from "tsyringe";
 
+import { InjectWallet } from "@src/billing/providers/wallet.provider";
+import { UserWalletOutput, UserWalletRepository } from "@src/billing/repositories";
+import { MasterWalletService } from "@src/billing/services";
 import { LoggerService } from "@src/core";
+import { DrainingDeploymentOutput, LeaseRepository } from "@src/deployment/repositories/lease/lease.repository";
+
+interface Balances {
+  denom: string;
+  feesLimit: number;
+  deploymentLimit: number;
+  balance: number;
+  isManaged: boolean;
+}
 
 @singleton()
 export class TopUpDeploymentsService {
+  private readonly CONCURRENCY = 10;
+
   private readonly logger = new LoggerService({ context: TopUpDeploymentsService.name });
 
+  constructor(
+    private readonly userWalletRepository: UserWalletRepository,
+    private readonly allowanceHttpService: AllowanceHttpService,
+    private readonly balanceHttpService: BalanceHttpService,
+    @InjectWallet("MANAGED") private readonly managedMasterWalletService: MasterWalletService,
+    @InjectWallet("UAKT_TOP_UP") private readonly uaktMasterWalletService: MasterWalletService,
+    @InjectWallet("USDC_TOP_UP") private readonly usdtMasterWalletService: MasterWalletService,
+    private readonly leaseRepository: LeaseRepository
+  ) {}
+
   async topUpDeployments() {
-    this.logger.warn("Top up deployments not implemented");
+    const wallets = [this.uaktMasterWalletService, this.usdtMasterWalletService];
+
+    const topUpAllManagedDeployments = wallets.map(async wallet => {
+      const address = await wallet.getFirstAddress();
+      await this.allowanceHttpService.paginateDeploymentGrantsForGrantee(address, async grants => {
+        await PromisePool.withConcurrency(this.CONCURRENCY)
+          .for(grants)
+          .process(async grant => this.topUpForGrant(grant));
+      });
+    });
+    await Promise.all(topUpAllManagedDeployments);
+
+    await this.paginateManagedWallets(async userWallets => {
+      await Promise.all(userWallets.map(async userWallet => this.topUpForManagedWallet(userWallet)));
+    });
+  }
+
+  private async topUpForGrant(grant: DeploymentAllowance) {
+    const balances = await this.collectCustodialWalletBalances(grant);
+    const owner = grant.granter;
+    this.logger.debug({ event: "BALANCES_COLLECTED", granter: owner, grantee: grant.grantee, balances });
+
+    const drainingDeployments = await this.retrieveDrainingDeployments(owner);
+
+    drainingDeployments.map(async deployment => {
+      const topUpAmount = await this.calculateTopUpAmount(deployment);
+      this.validateTopUpAmount(topUpAmount, balances);
+    });
+  }
+
+  private async collectCustodialWalletBalances(grant: DeploymentAllowance): Promise<Balances> {
+    const denom = grant.authorization.spend_limit.denom;
+    const deploymentLimit = parseFloat(grant.authorization.spend_limit.amount);
+
+    const feesAllowance = await this.allowanceHttpService.getFeeAllowanceForGranterAndGrantee(grant.granter, grant.grantee);
+    const feesSpendLimit = feesAllowance.allowance.spend_limit.find(limit => limit.denom === denom);
+    const feesLimit = feesSpendLimit ? parseFloat(feesSpendLimit.amount) : 0;
+
+    const { amount } = await this.balanceHttpService.getBalance(grant.granter, "uakt");
+    const balance = parseFloat(amount);
+
+    return {
+      denom,
+      feesLimit: feesLimit,
+      deploymentLimit,
+      balance,
+      isManaged: false
+    };
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  private async paginateManagedWallets(cb: (page: UserWalletOutput[]) => Promise<void>) {
+    this.logger.debug({ event: "PAGINATING_MANAGED_WALLETS", warning: "Not implemented yet" });
+  }
+
+  private async topUpForManagedWallet(userWallet: UserWalletOutput) {
+    const balances = await this.collectManagedWalletBalances(userWallet);
+    this.logger.debug({ event: "BALANCES_COLLECTED", wallet: userWallet, balances });
+
+    const drainingDeployments = await this.retrieveDrainingDeployments(userWallet.address);
+
+    drainingDeployments.map(async deployment => {
+      const topUpAmount = await this.calculateTopUpAmount(deployment);
+      this.validateTopUpAmount(topUpAmount, balances);
+    });
+  }
+
+  private async collectManagedWalletBalances(userWallet: UserWalletOutput): Promise<Balances> {
+    this.logger.debug({ event: "CALCULATING_MANAGE_WALLET_BALANCES", userWallet, warning: "Not implemented yet" });
+    return {
+      denom: "usdc",
+      feesLimit: 0,
+      deploymentLimit: 0,
+      balance: 0,
+      isManaged: true
+    };
+  }
+
+  private async retrieveDrainingDeployments(owner: string): Promise<DrainingDeploymentOutput[]> {
+    this.logger.debug({ event: "RETRIEVING_DRAINING_DEPLOYMENTS", owner, warning: "Not implemented yet" });
+    return [];
+  }
+
+  private async calculateTopUpAmount(deployment: DrainingDeploymentOutput): Promise<number> {
+    this.logger.debug({ event: "CALCULATING_TOP_UP_AMOUNT", deployment, warning: "Not implemented yet" });
+    return 0;
+  }
+
+  private validateTopUpAmount(amount: number, balances: Balances) {
+    this.logger.debug({ event: "VALIDATING_TOP_UP_AMOUNT", amount, balances, warning: "Not implemented yet" });
+  }
+
+  private async topUpCustodialDeployment() {
+    this.logger.debug({ event: "TOPPING_UP_CUSTODIAL_DEPLOYMENT", warning: "Not implemented yet" });
+  }
+
+  private async topUpManagedDeployment() {
+    this.logger.debug({ event: "TOPPING_UP_MANAGED_DEPLOYMENT", warning: "Not implemented yet" });
   }
 }

packages/http-sdk/src/allowance/allowance-http.service.ts

@@ -1,11 +1,7 @@
 import type { AxiosRequestConfig } from "axios";
 
 import { HttpService } from "../http/http.service";
-
-type Denom =
-  | "uakt"
-  | "ibc/170C677610AC31DF0904FFE09CD3B5C657492170E7E52372E48756B71E56F2F1"
-  | "ibc/12C6A0C374171B595A0A9E18B83FA09D295FB1F2D8C6DAA3AC28683471752D84";
+import type { Denom } from "../types/denom.type";
 
 type SpendLimit = {
   denom: Denom;
@@ -22,7 +18,7 @@ interface FeeAllowance {
   };
 }
 
-interface DeploymentAllowance {
+export interface DeploymentAllowance {
   granter: string;
   grantee: string;
   authorization: {
@@ -32,12 +28,19 @@ interface DeploymentAllowance {
   };
 }
 
-interface FeeAllowanceResponse {
+interface FeeAllowanceListResponse {
   allowances: FeeAllowance[];
 }
 
+interface FeeAllowanceResponse {
+  allowance: FeeAllowance;
+}
+
 interface DeploymentAllowanceResponse {
   grants: DeploymentAllowance[];
+  pagination: {
+    next_key: string | null;
+  };
 }
 
 export class AllowanceHttpService extends HttpService {
@@ -46,12 +49,35 @@ export class AllowanceHttpService extends HttpService {
   }
 
   async getFeeAllowancesForGrantee(address: string) {
-    const allowances = this.extractData(await this.get<FeeAllowanceResponse>(`cosmos/feegrant/v1beta1/allowances/${address}`));
+    const allowances = this.extractData(await this.get<FeeAllowanceListResponse>(`cosmos/feegrant/v1beta1/allowances/${address}`));
     return allowances.allowances;
   }
 
+  async getFeeAllowanceForGranterAndGrantee(granter: string, grantee: string) {
+    const allowances = this.extractData(await this.get<FeeAllowanceResponse>(`cosmos/feegrant/v1beta1/allowance/${granter}/${grantee}`));
+    return allowances.allowance;
+  }
+
   async getDeploymentAllowancesForGrantee(address: string) {
     const allowances = this.extractData(await this.get<DeploymentAllowanceResponse>(`cosmos/authz/v1beta1/grants/grantee/${address}`));
     return allowances.grants;
   }
+
+  async paginateDeploymentGrantsForGrantee(address: string, cb: (page: DeploymentAllowanceResponse["grants"]) => Promise<void>) {
+    let nextPageKey: string | null;
+
+    do {
+      const response = this.extractData(
+        await this.get<DeploymentAllowanceResponse>(
+          `cosmos/authz/v1beta1/grants/grantee/${address}`,
+          nextPageKey && {
+            params: { "pagination.key": nextPageKey }
+          }
+        )
+      );
+      nextPageKey = response.pagination.next_key;
+
+      await cb(response.grants);
+    } while (nextPageKey);
+  }
 }