feat: include agent address indexing into xrp indexer

Author: kuco23

package.json

@@ -20,7 +20,6 @@
     "run-native-indexer-debug": "ts-node packages/fasset-indexer-core/src/run/run-indexer.ts",
     "run-native-watchdog": "yarn build-base && node packages/fasset-indexer-core/dist/src/run/run-watchdog.js",
     "run-xrp-indexer": "yarn build-base && node packages/fasset-indexer-xrp/dist/src/run/run-indexer.js",
-    "run-xrp-address-indexer": "ts-node packages/fasset-indexer-xrp/src/run/run-address-indexer.ts",
     "run-btc-indexer": "yarn build-base && node packages/fasset-indexer-btc/dist/src/run/run-indexer.js",
     "run-doge-indexer": "yarn build-base && node packages/fasset-indexer-doge/dist/src/run/run-indexer.js",
     "run-api": "yarn build && node packages/fasset-indexer-api/dist/src/main",

packages/fasset-indexer-xrp/src/indexer/address-transaction-indexer.ts

@@ -1,27 +1,38 @@
 import { FAssetType } from "fasset-indexer-core"
+import * as Entities from "fasset-indexer-core/entities"
 import { findOrCreateEntity, getVar, setVar, type EntityManager } from "fasset-indexer-core/orm"
-import {
-  UnderlyingBlock, UnderlyingAddress, UnderlyingVoutReference, UnderlyingTransaction
-} from "fasset-indexer-core/entities"
 import { PaymentReference } from "fasset-indexer-core/utils"
-import { IXrpMemo, IXrpBlock, IXrpTransaction } from "../client/interface"
-import { BLOCK_HEIGHT_OFFSET } from "../constants"
 import { logger } from "fasset-indexer-core/logger"
+import { IXrpMemo, IXrpBlock, IXrpTransaction } from "../client/interface"
 import type { XrpContext } from "../context"
 
 
-export class XrpReferenceIndexer {
+export class XrpIndexer {
+  private tracked = new Set<string>()
 
   constructor(
     public readonly context: XrpContext,
     public readonly firstUnhandledBlockDbKey: string,
-    public readonly minBlockDbKey: string
-  ) { }
+    public readonly minBlockDbKey: string,
+    public readonly initialTracked: string[]
+  ) {
+    for (const address of initialTracked) {
+      this.tracked.add(address)
+    }
+  }
 
   async run(startBlock?: number): Promise<void> {
     return this.runHistoric(startBlock)
   }
 
+  async updateTrackedAddresses(): Promise<void> {
+    const agents = await this.context.orm.em.fork().findAll(
+      Entities.AgentVault, { populate: [ 'underlyingAddress' ] })
+    for (const agent of agents) {
+      this.tracked.add(agent.underlyingAddress.text)
+    }
+  }
+
   async runHistoric(startBlock?: number, endBlock?: number): Promise<void> {
     const firstUnhandledBlock = await this.firstUnhandledBlock(startBlock)
     if (startBlock === undefined || firstUnhandledBlock > startBlock) {
@@ -31,6 +42,7 @@ export class XrpReferenceIndexer {
     if (endBlock === undefined || endBlock > lastBlockToHandle) {
       endBlock = lastBlockToHandle
     }
+    await this.updateTrackedAddresses()
     for (let i = startBlock; i <= endBlock; i += 1) {
       const block = await this.context.provider.block(i)
       await this.processBlock(block)
@@ -44,8 +56,14 @@ export class XrpReferenceIndexer {
     return (block !== null ? Number(block.value!) : startBlock) ?? await this.minBlock()
   }
 
+  // note: this is so new agent vault creation events are indexed before
+  //       the xrp indexer skips the associated agent vault underlying addresses
+  // opt: return await this.context.provider.blockHeight() - BLOCK_HEIGHT_OFFSET
   async lastBlockToHandle(): Promise<number> {
-    return await this.context.provider.blockHeight() - BLOCK_HEIGHT_OFFSET
+    const blocks = await this.context.orm.em.fork().find(
+      Entities.CurrentUnderlyingBlockUpdated, {},
+      { limit: 1, orderBy: { underlyingBlockNumber: 'desc' } })
+    return blocks[0]?.underlyingBlockNumber ?? 0
   }
 
   async minBlock(): Promise<number> {
@@ -61,50 +79,74 @@ export class XrpReferenceIndexer {
 
   protected async processBlock(block: IXrpBlock): Promise<void> {
     await this.context.orm.em.transactional(async em => {
-    let blockEnt = await em.findOne(UnderlyingBlock, { hash: block.ledger_hash })
-    if (blockEnt != null) return
+    let blockEnt = await em.findOne(Entities.UnderlyingBlock, { hash: block.ledger_hash })
+    if (blockEnt == null) {
       blockEnt = await this.storeXrpBlock(em, block)
       for (const tx of block.transactions) {
-        await this.processTx(em, tx, blockEnt)
+        await this.processTransaction(em, tx, blockEnt)
       }
-    })
+    }})
   }
 
-  protected async processTx(
+  protected async processTransaction(
     em: EntityManager,
     transaction: IXrpTransaction,
-    block: UnderlyingBlock
+    block: Entities.UnderlyingBlock
   ): Promise<void> {
+    let utransaction = null
+    const { Account, Destination } = transaction
+    if (this.tracked.has(Account) || this.tracked.has(Destination!)) {
+      utransaction = await this.storeTransaction(em, transaction, block)
+    }
     if (transaction.Memos == null) return
     for (const memo of transaction.Memos) {
       const reference = this.extractReference(memo)
       if (reference == null) continue
-      if (transaction.Account == null) {
-        logger.error(`reference but no account present at index ${block.height}`)
+      if (utransaction == null) {
+        utransaction = await this.storeTransaction(em, transaction, block)
+      }
+      if (utransaction.source == null) {
+        logger.warn(`transaction ${transaction.hash} has valid FAsset reference ${reference} but no source`)
         continue
       }
-      const source = await findOrCreateEntity(em, UnderlyingAddress, { text: transaction.Account })
-      const target = transaction.Destination == null ? undefined :
-        await findOrCreateEntity(em, UnderlyingAddress, { text: transaction.Destination })
-      const utransaction = await findOrCreateEntity(em, UnderlyingTransaction, {
-         hash: transaction.hash, block, value: BigInt(transaction.Amount!), source, target})
-      await this.storeReference(em, reference, utransaction, source, block)
+      await this.storeReference(em, reference, utransaction, utransaction.source, block)
       break
     }
   }
 
-  private async storeXrpBlock(em: EntityManager, xrpBlock: IXrpBlock): Promise<UnderlyingBlock> {
-    return em.create(UnderlyingBlock, {
+  protected async storeTransaction(
+    em: EntityManager,
+    transaction: IXrpTransaction,
+    block: Entities.UnderlyingBlock
+  ): Promise<Entities.UnderlyingTransaction> {
+    let source = undefined
+    let target = undefined
+    const { Account, Destination, Amount } = transaction
+    if (Account != null) {
+      source = await findOrCreateEntity(em, Entities.UnderlyingAddress, { text: Account })
+    }
+    if(Destination != null) {
+      target = await findOrCreateEntity(em, Entities.UnderlyingAddress, { text: Destination })
+    }
+    return findOrCreateEntity(em, Entities.UnderlyingTransaction, {
+      hash: transaction.hash, block, value: BigInt(Amount ?? 0), source, target
+    })
+  }
+
+  private async storeXrpBlock(em: EntityManager, xrpBlock: IXrpBlock): Promise<Entities.UnderlyingBlock> {
+    return em.create(Entities.UnderlyingBlock, {
       hash: xrpBlock.ledger_hash, height: xrpBlock.ledger_index, timestamp: xrpBlock.close_time
     })
   }
 
   private async storeReference(
-    em: EntityManager, reference: string,
-    transaction: UnderlyingTransaction,
-    address: UnderlyingAddress, block: UnderlyingBlock
-  ): Promise<UnderlyingVoutReference> {
-    return em.create(UnderlyingVoutReference, {
+    em: EntityManager,
+    reference: string,
+    transaction: Entities.UnderlyingTransaction,
+    address: Entities.UnderlyingAddress,
+    block: Entities.UnderlyingBlock
+  ): Promise<Entities.UnderlyingVoutReference> {
+    return em.create(Entities.UnderlyingVoutReference, {
       fasset: FAssetType.FXRP, reference, transaction, address, block
     })
   }

…xer-xrp/src/indexer/reference-indexer.ts …t-indexer-xrp/src/indexer/xrp-indexer.tspackages/fasset-indexer-xrp/src/indexer/reference-indexer.ts renamed to packages/fasset-indexer-xrp/src/indexer/xrp-indexer.ts packages/fasset-indexer-xrp/src/indexer/reference-indexer.ts renamed to packages/fasset-indexer-xrp/src/indexer/xrp-indexer.ts

@@ -1,5 +1,5 @@
 import { IndexerRunner } from "fasset-indexer-core"
-import { XrpReferenceIndexer } from "../indexer/reference-indexer"
+import { XrpIndexer } from "../indexer/xrp-indexer"
 import { XrpConfigLoader } from "../config/config"
 import { XrpContext } from "../context"
 import { monitor } from "./run-monitor"
@@ -12,7 +12,12 @@ import {
 async function runIndexer() {
   const config = new XrpConfigLoader()
   const context = await XrpContext.create(config)
-  const indexer = new XrpReferenceIndexer(context, FIRST_UNHANDLED_XRP_BLOCK_DB_KEY, MIN_XRP_BLOCK_NUMBER_DB_KEY)
+  const indexer = new XrpIndexer(
+    context,
+    FIRST_UNHANDLED_XRP_BLOCK_DB_KEY,
+    MIN_XRP_BLOCK_NUMBER_DB_KEY,
+    config.xrpMonitoredAddresses
+  )
   const runner = new IndexerRunner(indexer, 'xrp')
   await Promise.all([monitor(config, context), runner.run(config.xrpMinBlockNumber)])
 }