Liquidity Manager

jest.config.ts

@@ -8,4 +8,7 @@ module.exports = {
   collectCoverageFrom: ['**/*.(t|j)s'],
   coverageDirectory: '../coverage',
   testEnvironment: 'node',
+  moduleNameMapper: {
+    '^@/(.*)$': '<rootDir>//$1',
+  },
 }

package.json

@@ -25,9 +25,10 @@
   },
   "dependencies": {
     "@aws-sdk/client-secrets-manager": "^3.592.0",
-    "@eco-foundation/routes": "^0.0.508-beta",
+    "@eco-foundation/routes": "~0.0.711-beta",
     "@launchdarkly/node-server-sdk": "^9.7.1",
     "@liaoliaots/nestjs-redis-health": "^9.0.4",
+    "@lifi/sdk": "^3.4.1",
     "@nestjs/bullmq": "^10.1.1",
     "@nestjs/common": "^10.0.0",
     "@nestjs/config": "^3.2.2",
@@ -48,6 +49,7 @@
     "redlock": "^5.0.0-beta.2",
     "reflect-metadata": "^0.2.0",
     "rxjs": "^7.8.1",
+    "table": "^6.8.2",
     "viem": "^2.21.32"
   },
   "devDependencies": {

src/app.module.ts

@@ -14,6 +14,7 @@ import { HealthModule } from './health/health.module'
 import { ProcessorModule } from './bullmq/processors/processor.module'
 import { SolverModule } from './solver/solver.module'
 import { FlagsModule } from './flags/flags.module'
+import { LiquidityManagerModule } from '@/liquidity-manager/liquidity-manager.module'
 
 @Module({
   imports: [
@@ -41,6 +42,7 @@ import { FlagsModule } from './flags/flags.module'
     }),
     ProverModule,
     SolverModule,
+    LiquidityManagerModule,
     ...getPino(),
   ],
   controllers: [],

src/balance/balance.service.ts

@@ -1,14 +1,13 @@
 import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'
+import { groupBy, zipWith } from 'lodash'
 import { EcoConfigService } from '../eco-configs/eco-config.service'
-import { Solver } from '../eco-configs/eco-config.types'
 import { getDestinationNetworkAddressKey } from '../common/utils/strings'
 import { EcoLogMessage } from '../common/logging/eco-log-message'
-import { erc20Abi, Hex } from 'viem'
+import { erc20Abi, Hex, MulticallParameters, MulticallReturnType } from 'viem'
 import { ViemEventLog } from '../common/events/viem'
 import { decodeTransferLog, isSupportedTokenType } from '../contracts'
 import { KernelAccountClientService } from '../transaction/smart-wallets/kernel/kernel-account-client.service'
-
-type TokenBalance = { decimals: bigint; balance: bigint }
+import { TokenBalance, TokenConfig } from '@/balance/types'
 
 /**
  * Service class for getting configs for the app
@@ -25,13 +24,8 @@ export class BalanceService implements OnApplicationBootstrap {
   ) {}
 
   async onApplicationBootstrap() {
-    //iterate over all solvers
-    await Promise.all(
-      Object.entries(this.ecoConfig.getSolvers()).map(async (entry) => {
-        const [, solver] = entry
-        await this.loadTokenBalances(solver)
-      }),
-    )
+    // iterate over all tokens
+    await Promise.all(this.getTokens().map((token) => this.loadTokenBalance(token)))
   }
 
   /**
@@ -69,20 +63,104 @@ export class BalanceService implements OnApplicationBootstrap {
     }
   }
 
+  getTokens(): TokenConfig[] {
+    return Object.values(this.ecoConfig.getSolvers()).flatMap((solver) => {
+      return Object.entries(solver.targets)
+        .filter(([, targetContract]) => isSupportedTokenType(targetContract.contractType))
+        .map(([tokenAddress, targetContract]) => ({
+          address: tokenAddress as Hex,
+          chainId: solver.chainID,
+          type: targetContract.contractType,
+          minBalance: targetContract.minBalance,
+          targetBalance: targetContract.targetBalance,
+        }))
+    })
+  }
+
+  async fetchTokenBalances(
+    chainID: number,
+    tokenAddresses: Hex[],
+  ): Promise<Record<Hex, TokenBalance>> {
+    const client = await this.kernelAccountClientService.getClient(chainID)
+    const walletAddress = client.kernelAccount.address
+
+    this.logger.debug(
+      EcoLogMessage.fromDefault({
+        message: `fetchTokenBalances`,
+        properties: {
+          chainID,
+          tokenAddresses,
+          walletAddress,
+        },
+      }),
+    )
+
+    const results = (await client.multicall({
+      contracts: tokenAddresses.flatMap((tokenAddress): MulticallParameters['contracts'] => [
+        {
+          abi: erc20Abi,
+          address: tokenAddress,
+          functionName: 'balanceOf',
+          args: [walletAddress],
+        },
+        {
+          abi: erc20Abi,
+          address: tokenAddress,
+          functionName: 'decimals',
+        },
+      ]),
+      allowFailure: false,
+    })) as MulticallReturnType
+
+    const result: Record<Hex, TokenBalance> = {}
+
+    tokenAddresses.forEach((tokenAddress, index) => {
+      const [balance = 0n, decimals = 0] = [results[index * 2], results[index * 2 + 1]]
+      result[tokenAddress] = {
+        address: tokenAddress,
+        balance: balance as bigint,
+        decimals: decimals as number,
+      }
+    })
+
+    return result
+  }
+
+  async fetchTokenBalance(chainID: number, tokenAddress: Hex): Promise<TokenBalance> {
+    const result = await this.fetchTokenBalances(chainID, [tokenAddress])
+    return result[tokenAddress]
+  }
+
+  async getAllTokenData() {
+    const tokens = this.getTokens()
+    const tokensByChainId = groupBy(tokens, 'chainId')
+    const chainIds = Object.keys(tokensByChainId)
+
+    const balancesPerChainIdPromise = chainIds.map(async (chainId) => {
+      const configs = tokensByChainId[chainId]
+      const tokenAddresses = configs.map((token) => token.address)
+      const balances = await this.fetchTokenBalances(parseInt(chainId), tokenAddresses)
+      return zipWith(configs, Object.values(balances), (config, balance) => ({
+        config,
+        balance,
+        chainId: parseInt(chainId),
+      }))
+    })
+
+    return Promise.all(balancesPerChainIdPromise).then((result) => result.flat())
+  }
+
   /**
    * Loads the token balance of the solver
    * @returns
    */
-  private async loadTokenBalances(solver: Solver) {
-    await Promise.all(
-      Object.entries(solver.targets).map(async (target) => {
-        const [tokenAddress, targetContract] = target
-        if (isSupportedTokenType(targetContract.contractType)) {
-          //load the balance in the local mapping
-          return await this.loadERC20TokenBalance(solver.chainID, tokenAddress as Hex)
-        }
-      }),
-    )
+  private async loadTokenBalance(token: TokenConfig) {
+    switch (token.type) {
+      case 'erc20':
+        return this.loadERC20TokenBalance(token.chainId, token.address)
+      default:
+        throw new Error('Unsupported token type')
+    }
   }
 
   private async loadERC20TokenBalance(
@@ -91,27 +169,8 @@ export class BalanceService implements OnApplicationBootstrap {
   ): Promise<TokenBalance | undefined> {
     const key = getDestinationNetworkAddressKey(chainID, tokenAddress)
     if (!this.tokenBalances.has(key)) {
-      const client = await this.kernelAccountClientService.getClient(chainID)
-      const erc20 = {
-        address: tokenAddress,
-        abi: erc20Abi,
-      }
-
-      const [{ result: balance }, { result: decimals }] = await client.multicall({
-        contracts: [
-          {
-            ...erc20,
-            functionName: 'balanceOf',
-            args: [client.kernelAccount.address],
-          },
-          {
-            ...erc20,
-            functionName: 'decimals',
-          },
-        ],
-      })
-
-      this.tokenBalances.set(key, { balance: balance ?? 0n, decimals: BigInt(decimals ?? 0) })
+      const tokenBalance = await this.fetchTokenBalance(chainID, tokenAddress)
+      this.tokenBalances.set(key, tokenBalance)
     }
     return this.tokenBalances.get(key)
   }

src/balance/types.ts

@@ -1,4 +1,16 @@
+import { Hex } from 'viem'
+import { TargetContractType } from '@/eco-configs/eco-config.types'
+
+export type TokenConfig = {
+  address: Hex
+  chainId: number
+  minBalance: number
+  targetBalance: number
+  type: TargetContractType
+}
+
 export type TokenBalance = {
   address: string
+  decimals: number
   balance: bigint
 }

src/bullmq/bullmq.helper.ts

@@ -1,20 +1,41 @@
-import { BullModule } from '@nestjs/bullmq'
+import { BullModule, RegisterQueueOptions } from '@nestjs/bullmq'
 import { DynamicModule } from '@nestjs/common'
-import { EcoConfigService } from '../eco-configs/eco-config.service'
-import { RedisConnectionUtils } from '../common/redis/redis-connection-utils'
-import { QueueInterface } from '../common/redis/constants'
+import { EcoConfigService } from '@/eco-configs/eco-config.service'
+import { RedisConnectionUtils } from '@/common/redis/redis-connection-utils'
+import { QueueMetadata } from '@/common/redis/constants'
 
 /**
  * Initialize the BullMQ queue with the given token and eco configs
- * @param token the name of the queue
+ * @param {QueueMetadata} queueInterface queue interface
+ * @param {Partial<RegisterQueueOptions>} opts queue options
  * @returns
  */
-export function initBullMQ(queueInterface: QueueInterface): DynamicModule {
+export function initBullMQ(
+  queueInterface: QueueMetadata,
+  opts?: Partial<RegisterQueueOptions>,
+): DynamicModule {
   return BullModule.registerQueueAsync({
     name: queueInterface.queue,
-    useFactory: async (configService: EcoConfigService) => {
-      return RedisConnectionUtils.getQueueOptions(queueInterface, configService.getRedis())
+    useFactory: (configService: EcoConfigService) => {
+      return {
+        ...RedisConnectionUtils.getQueueOptions(queueInterface, configService.getRedis()),
+        ...opts,
+      }
     },
     inject: [EcoConfigService],
   })
 }
+
+/**
+ * Initialize the BullMQ flow with the given name and eco configs
+ * @param {QueueMetadata} queueInterface queue interface
+ * @returns
+ */
+export function initFlowBullMQ(queueInterface: QueueMetadata): DynamicModule {
+  return BullModule.registerFlowProducerAsync({
+    name: queueInterface.queue,
+    useFactory: (configService: EcoConfigService) =>
+      RedisConnectionUtils.getQueueOptions(queueInterface, configService.getRedis()),
+    inject: [EcoConfigService],
+  })
+}

src/common/redis/constants.ts

@@ -30,8 +30,11 @@ export const QUEUES: Record<any, QueueInterface> = {
   },
 }
 
-export interface QueueInterface {
+export interface QueueMetadata {
   queue: string
   prefix: string
+}
+
+export interface QueueInterface extends QueueMetadata {
   jobs: Record<string, string>
 }

src/common/redis/redis-connection-utils.ts

@@ -3,7 +3,7 @@ import { RegisterQueueOptions } from '@nestjs/bullmq'
 import * as Redis from 'ioredis'
 import { EcoError } from '../errors/eco-error'
 import { EcoLogMessage } from '../logging/eco-log-message'
-import { QueueInterface } from './constants'
+import { QueueMetadata } from './constants'
 import { RedisConfig } from '../../eco-configs/eco-config.types'
 import { RedlockRedisClient } from '../../nest-redlock/nest-redlock.service'
 
@@ -23,7 +23,7 @@ export class RedisConnectionUtils {
     return new Redis.Redis(connection as Redis.RedisOptions)
   }
 
-  static getQueueOptions(queue: QueueInterface, redisConfig: RedisConfig): RegisterQueueOptions {
+  static getQueueOptions(queue: QueueMetadata, redisConfig: RedisConfig): RegisterQueueOptions {
     try {
       const connection = redisConfig.connection
       const { queue: name, prefix } = queue

src/eco-configs/eco-config.service.ts

@@ -1,11 +1,13 @@
 import { Injectable, Logger, OnModuleInit } from '@nestjs/common'
+import * as _ from 'lodash'
 import * as config from 'config'
-import { EcoLogMessage } from '../common/logging/eco-log-message'
+import { EcoLogMessage } from '@/common/logging/eco-log-message'
 import { ConfigSource } from './interfaces/config-source.interface'
-import { EcoConfigType, Solver, IntentSource } from './eco-config.types'
+import { EcoConfigType, IntentSource, Solver } from './eco-config.types'
 import { entries } from 'lodash'
 import { getAddress } from 'viem'
-import { addressKeys } from '../common/viem/utils'
+import { addressKeys, getRpcUrl } from '@/common/viem/utils'
+import { ChainsSupported } from '@/common/chains/supported'
 import { getChainConfig } from './utils'
 
 /**
@@ -66,7 +68,10 @@
       })
       const config = getChainConfig(intent.chainID)
       intent.sourceAddress = config.IntentSource
-      intent.provers = [config.Prover, config.HyperProver]
+      intent.provers = [config.HyperProver]
+      if(config.Prover) {
+        intent.provers.push(config.Prover)
+      }
       return intent
     })
     return intents
@@ -128,4 +133,15 @@
   getServer(): EcoConfigType['server'] {
     return this.get('server')
   }
+
+  getChainRPCs() {
+    const { apiKey, networks } = this.getAlchemy()
+    const supportedAlchemyChainIds = _.map(networks, 'id')
+
+    const entries = ChainsSupported.map((chain) => {
+      const rpcApiKey = supportedAlchemyChainIds.includes(chain.id) ? apiKey : undefined
+      return [chain.id, getRpcUrl(chain, rpcApiKey).url]
+    })
+    return Object.fromEntries(entries) as Record<number, string>
+  }
 }

src/eco-configs/eco-config.types.ts

@@ -131,6 +131,7 @@ export interface TargetContract {
   contractType: TargetContractType
   selectors: string[]
   minBalance: number
+  targetBalance: number
 }
 
 /**