feat: add ripple address balance monitor

Author: kuco23

packages/fasset-indexer-core/src/orm/entities/index.ts

@@ -40,7 +40,7 @@ export { TokenBalance } from "./state/balance"
 export { AssetManagerSettings } from './state/settings'
 // underlying data
 export { UnderlyingBlock } from "./underlying/block"
-export { UnderlyingAddress } from "./underlying/address"
+export { UnderlyingAddress, UnderlyingBalance } from "./underlying/address"
 export { UnderlyingReference as UnderlyingVoutReference } from "./underlying/reference"
 export { UnderlyingTransaction } from "./underlying/transaction"
 // building event bound stuff

packages/fasset-indexer-core/src/orm/entities/underlying/address.ts

@@ -1,5 +1,6 @@
-import { Entity, Enum, PrimaryKey, Property } from "@mikro-orm/core"
+import { Entity, Enum, ManyToMany, ManyToOne, PrimaryKey, Property } from "@mikro-orm/core"
 import { AddressType } from "../../interface"
+import { uint256 } from "../../custom/uint"
 
 
 @Entity()
@@ -18,4 +19,19 @@ export class UnderlyingAddress {
     this.text = text
     this.type = type
   }
+}
+
+@Entity()
+export class UnderlyingBalance {
+
+  @ManyToOne({ entity: () => UnderlyingAddress, primary: true })
+  address: UnderlyingAddress
+
+  @Property({ type: new uint256() })
+  balance: bigint
+
+  constructor(address: UnderlyingAddress, balance: bigint) {
+    this.address = address
+    this.balance = balance
+  }
 }

packages/fasset-indexer-core/src/orm/index.ts

@@ -2,5 +2,5 @@ export { createOrm } from "./mikro-orm.config"
 export { raw } from "@mikro-orm/core"
 export { EntityManager, SelectQueryBuilder } from "@mikro-orm/knex"
 export { ZeroAddress } from 'ethers' // for database lookup
-export { getVar, setVar, findOrCreateUnderlyingAddress, findOrCreateUnderlyingTransaction } from "./utils"
+export { getVar, setVar, findOrCreateUnderlyingAddress, findOrCreateUnderlyingTransaction, updateUnderlyingBalance } from "./utils"
 export { OrmOptions, ORM, AddressType } from "./interface"

packages/fasset-indexer-core/src/orm/mikro-orm.config.ts

@@ -4,7 +4,11 @@ import { EvmAddress } from "./entities/evm/address"
 import { EvmBlock } from "./entities/evm/block"
 import { EvmTransaction } from "./entities/evm/transaction"
 import { EvmLog } from "./entities/evm/log"
-import { AgentVaultCreated, AgentSettingChanged, SelfClose, VaultCollateralWithdrawalAnnounced, PoolTokenRedemptionAnnounced, UnderlyingWithdrawalAnnounced, UnderlyingWithdrawalConfirmed } from "./entities/events/agent"
+import {
+  AgentVaultCreated, AgentSettingChanged, SelfClose,
+  VaultCollateralWithdrawalAnnounced, PoolTokenRedemptionAnnounced,
+  UnderlyingWithdrawalAnnounced, UnderlyingWithdrawalConfirmed
+} from "./entities/events/agent"
 import {
   CollateralReserved, MintingExecuted,
   MintingPaymentDefault, CollateralReservationDeleted,
@@ -50,7 +54,7 @@ import { AssetManagerSettings, CoreVaultManagerSettings } from "./entities/state
 import { FtsoPrice } from "./entities/state/price"
 import { TokenBalance } from "./entities/state/balance"
 import { UnderlyingBlock } from "./entities/underlying/block"
-import { UnderlyingAddress } from "./entities/underlying/address"
+import { UnderlyingAddress, UnderlyingBalance } from "./entities/underlying/address"
 import { UnderlyingReference } from "./entities/underlying/reference"
 import { updateSchema } from "./utils"
 import { MIN_DATABASE_POOL_CONNECTIONS, MAX_DATABASE_POOL_CONNECTIONS } from "../config/constants"
@@ -86,7 +90,7 @@ export const ORM_OPTIONS: Options<AbstractSqlDriver> = defineConfig({
     EscrowFinished, CoreVaultManagerCustodianAddressUpdated,
     AssetManagerSettings, CoreVaultManagerSettings,
     // underlying
-    UnderlyingBlock, UnderlyingReference, UnderlyingAddress
+    UnderlyingBlock, UnderlyingReference, UnderlyingAddress, UnderlyingBalance
   ],
   pool: {
     min: MIN_DATABASE_POOL_CONNECTIONS,

packages/fasset-indexer-core/src/orm/utils.ts

@@ -1,7 +1,7 @@
 import { Var } from "../orm/entities/state/var"
-import { UnderlyingAddress } from "../orm/entities/underlying/address"
+import { UnderlyingAddress, UnderlyingBalance } from "../orm/entities/underlying/address"
 import type { EntityManager } from "@mikro-orm/core"
-import type { ORM, SchemaUpdate, AddressType } from "./interface"
+import { ORM, SchemaUpdate, AddressType } from "./interface"
 import { UnderlyingBlock, UnderlyingTransaction } from "./entities";
 
 
@@ -59,3 +59,15 @@ export async function findOrCreateUnderlyingTransaction(
   }
   return underlyingTransaction
 }
+
+export async function updateUnderlyingBalance(em: EntityManager, address: string, amount: bigint): Promise<UnderlyingBalance> {
+  let balance = await em.findOne(UnderlyingBalance, { address: { text: address } })
+  if (balance == null) {
+    const _address = await findOrCreateUnderlyingAddress(em, address, AddressType.USER)
+    balance = new UnderlyingBalance(_address, amount)
+  } else {
+    balance.balance = amount
+  }
+  em.persist(balance)
+  return balance
+}

packages/fasset-indexer-xrp/src/client/interface.ts

@@ -46,4 +46,12 @@ export interface IXrpServerInfoResponse {
     }
     status: string
   }
+}
+
+export interface IXrpAccountInfoResponse {
+  result: {
+    account_data: {
+      Balance: string
+    }
+  }
 }

packages/fasset-indexer-xrp/src/client/xrp-client.ts

@@ -1,5 +1,5 @@
 import { XrpConfigLoader } from "../config/config"
-import { IXrpBlock, IXrpBlockQueryResponse, IXrpLedgerCurrentResponse, IXrpServerInfoResponse } from "./interface"
+import { IXrpAccountInfoResponse, IXrpBlock, IXrpBlockQueryResponse, IXrpLedgerCurrentResponse, IXrpServerInfoResponse } from "./interface"
 
 export class XrpClient {
 
@@ -25,6 +25,15 @@ export class XrpClient {
     return current_block
   }
 
+  async balanceOf(account: string): Promise<bigint> {
+    const resp = await this.getAccountInfo(account)
+    return BigInt(resp.result.account_data.Balance)
+  }
+
+  private async getAccountInfo(account: string): Promise<IXrpAccountInfoResponse> {
+    return this.request('account_info', [{ account, ledger_index: 'validated' }])
+  }
+
   private async getLedgerCurrent(): Promise<IXrpLedgerCurrentResponse> {
     return this.request('ledger_current', [])
   }

packages/fasset-indexer-xrp/src/config/config.ts

@@ -17,4 +17,9 @@ export class XrpConfigLoader extends ConfigLoader {
   get xrpNodeIsAmendmentBlocked(): boolean {
     return process.env.XRP_RPC_AMENDMENT_BLOCKED === "true"
   }
+
+  get xrpMonitoredAddresses(): string[] {
+    const addresses = process.env.XRP_MONITORED_ADDRESSES
+    return (addresses != null) ? addresses.split(',') : []
+  }
 }

packages/fasset-indexer-xrp/src/run/run-indexer.ts

@@ -2,14 +2,15 @@ import { IndexerRunner } from "fasset-indexer-core"
 import { XrpIndexer } from "../indexer/indexer"
 import { XrpConfigLoader } from "../config/config"
 import { XrpContext } from "../context"
+import { monitor } from "./run-monitor"
 
 
 async function runIndexer() {
   const config = new XrpConfigLoader()
   const context = await XrpContext.create(config)
   const indexer = new XrpIndexer(context)
   const runner = new IndexerRunner(indexer, 'xrp')
-  await runner.run(config.xrpMinBlockNumber)
+  await Promise.all([monitor(config, context), runner.run(config.xrpMinBlockNumber)])
 }
 
 runIndexer()

packages/fasset-indexer-xrp/src/run/run-monitor.ts

@@ -0,0 +1,34 @@
+import { sleep } from "fasset-indexer-core/utils"
+import { logger } from "fasset-indexer-core/logger"
+import { updateUnderlyingBalance } from "fasset-indexer-core/orm"
+import { XrpConfigLoader } from "../config/config"
+import { XrpContext } from "../context"
+
+const SLEEP_CYCLE_MS = 60_000
+
+export async function monitor(config: XrpConfigLoader, context: XrpContext) {
+  while (true) {
+    await updateBalances(config.xrpMonitoredAddresses, context)
+    await sleep(SLEEP_CYCLE_MS)
+  }
+}
+
+async function updateBalances(addresses: string[], context: XrpContext) {
+  for (const address of addresses) {
+    const balance = await fetchBalanceOf(address, context)
+    if (balance == null) continue
+    await context.orm.em.transactional(async em => {
+      await updateUnderlyingBalance(em, address, balance)
+    })
+    logger.info(`successfully updated ripple balance for ${address}`)
+  }
+}
+
+async function fetchBalanceOf(address: string, context: XrpContext): Promise<bigint | null> {
+  try {
+    return await context.provider.balanceOf(address)
+  } catch (err: any) {
+    logger.error(`could not fetch ripple balance for ${address} due to: ${err}`)
+    return null
+  }
+}