From 7b0d851bbb516c30591bb4b135a43fa96bd921f2 Mon Sep 17 00:00:00 2001
From: ahonn <yuexunjiang@gmail.com>
Date: Thu, 26 Sep 2024 17:30:41 +1000
Subject: [PATCH] refactor(indexer): improve asset indexing and health check

---
 backend/src/constants.ts                      |  1 +
 backend/src/core/indexer/flow/assets.flow.ts  | 30 +++++++++++++++++--
 backend/src/core/indexer/indexer.health.ts    | 11 ++++++-
 backend/src/core/indexer/indexer.queue.ts     |  8 +++--
 .../processor/block-assets.processor.ts       |  2 +-
 backend/src/filters/all-exceptions.filter.ts  | 15 ++++++----
 6 files changed, 54 insertions(+), 13 deletions(-)

diff --git a/backend/src/constants.ts b/backend/src/constants.ts
index c3263cab..1b0572f5 100644
--- a/backend/src/constants.ts
+++ b/backend/src/constants.ts
@@ -19,6 +19,7 @@ export const BtcTestnetTypeMap: Record<NetworkType.testnet | NetworkType.signet,
 };
 
 export const CKB_MIN_SAFE_CONFIRMATIONS = 24;
+export const CKB_ONE_DAY_BLOCKS = 8640;
 export const CKB_CHAIN_ID = 1;
 
 export const XUDT_TYPESCRIPTS = {
diff --git a/backend/src/core/indexer/flow/assets.flow.ts b/backend/src/core/indexer/flow/assets.flow.ts
index 1327d561..e3523e95 100644
--- a/backend/src/core/indexer/flow/assets.flow.ts
+++ b/backend/src/core/indexer/flow/assets.flow.ts
@@ -1,7 +1,7 @@
 import { Logger } from '@nestjs/common';
 import { AssetType, Chain } from '@prisma/client';
 import { EventEmitter } from 'node:events';
-import { CKB_MIN_SAFE_CONFIRMATIONS } from 'src/constants';
+import { CKB_MIN_SAFE_CONFIRMATIONS, CKB_ONE_DAY_BLOCKS } from 'src/constants';
 import { BlockchainService } from 'src/core/blockchain/blockchain.service';
 import { PrismaService } from 'src/core/database/prisma/prisma.service';
 import { IndexerQueueService } from '../indexer.queue';
@@ -24,6 +24,21 @@ export class IndexerAssetsFlow extends EventEmitter {
   }
 
   public async start() {
+    const latestAsset = await this.getLatestAsset();
+    if (latestAsset) {
+      this.logger.log(`Latest asset block number: ${latestAsset.blockNumber}`);
+      const tipBlockNumber = await this.blockchainService.getTipBlockNumber();
+      if (
+        tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS - latestAsset.blockNumber <
+        CKB_ONE_DAY_BLOCKS
+      ) {
+        this.logger.log(`Latest asset is near tip block number, skip indexing assets...`);
+        this.startBlockAssetsIndexing();
+        this.setupBlockAssetsIndexedListener();
+        return;
+      }
+    }
+
     const assetTypeScripts = await this.prismaService.assetType.findMany({
       where: { chainId: this.chain.id },
     });
@@ -32,6 +47,15 @@ export class IndexerAssetsFlow extends EventEmitter {
     this.setupAssetIndexedListener(assetTypeScripts.length);
   }
 
+  private async getLatestAsset() {
+    const latestAsset = await this.prismaService.asset.findFirst({
+      select: { blockNumber: true },
+      where: { chainId: this.chain.id },
+      orderBy: { blockNumber: 'desc' },
+    });
+    return latestAsset;
+  }
+
   private async indexAssets(assetType: AssetType) {
     const cursor = await this.indexerQueueService.getLatestAssetJobCursor(assetType);
     if (cursor === '0x') {
@@ -62,7 +86,7 @@ export class IndexerAssetsFlow extends EventEmitter {
   private async startBlockAssetsIndexing() {
     const tipBlockNumber = await this.blockchainService.getTipBlockNumber();
 
-    let latestIndexedBlockNumber = await this.indexerQueueService.getLatestIndexedBlock(
+    let latestIndexedBlockNumber = await this.indexerQueueService.getLatestIndexedAssetsBlock(
       this.chain.id,
     );
     if (!latestIndexedBlockNumber) {
@@ -75,7 +99,7 @@ export class IndexerAssetsFlow extends EventEmitter {
     }
     const targetBlockNumber = tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS;
     if (targetBlockNumber <= latestIndexedBlockNumber) {
-      this.emit(IndexerAssetsEvent.BlockAssetsIndexed);
+      this.emit(IndexerAssetsEvent.BlockAssetsIndexed, latestIndexedBlockNumber);
       return;
     }
 
diff --git a/backend/src/core/indexer/indexer.health.ts b/backend/src/core/indexer/indexer.health.ts
index 3afac5a5..ac27f98a 100644
--- a/backend/src/core/indexer/indexer.health.ts
+++ b/backend/src/core/indexer/indexer.health.ts
@@ -60,7 +60,16 @@ export class IndexerHealthIndicator extends HealthIndicator {
     const blockchainService = this.blockchainServiceFactory.getService(CKB_CHAIN_ID);
     const tipBlockNumber = await blockchainService.getTipBlockNumber();
     const targetBlockNumber = tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS;
-    const currentBlockNumber = await this.indexerQueueService.getLatestIndexedBlock(CKB_CHAIN_ID);
+
+    let currentBlockNumber = await this.indexerQueueService.getLatestIndexedAssetsBlock(CKB_CHAIN_ID);
+    if (!currentBlockNumber) {
+      const latestAsset = await this.prismaService.asset.findFirst({
+        select: { blockNumber: true },
+        where: { chainId: CKB_CHAIN_ID },
+        orderBy: { blockNumber: 'desc' },
+      });
+      currentBlockNumber = latestAsset?.blockNumber;
+    }
 
     const isHealthy =
       !!currentBlockNumber && currentBlockNumber >= targetBlockNumber - INDEEXR_HEALTH_THRESHOLD;
diff --git a/backend/src/core/indexer/indexer.queue.ts b/backend/src/core/indexer/indexer.queue.ts
index d0f7beb1..55049c96 100644
--- a/backend/src/core/indexer/indexer.queue.ts
+++ b/backend/src/core/indexer/indexer.queue.ts
@@ -76,13 +76,17 @@ export class IndexerQueueService {
     await this.cacheManager.set(`${INDEXER_ASSETS_QUEUE}:${typeHash}`, cursor || '');
   }
 
-  public async getLatestIndexedBlock(chainId: number) {
+  public async getLatestIndexedAssetsBlock(chainId: number) {
     const blockNumber = await this.cacheManager.get<number>(
       `${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`,
     );
     return blockNumber;
   }
 
+  public async setLatestIndexedAssetsBlock(chainId: number, blockNumber: number) {
+    await this.cacheManager.set(`${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`, blockNumber);
+  }
+
   public async addBlockAssetsJob(data: IndexerBlockAssetsJobData) {
     const { chainId, blockNumber } = data;
     const params = new URLSearchParams();
@@ -95,7 +99,7 @@ export class IndexerQueueService {
       `Added block assets job ${jobId} for chain ${chainId} with block number ${blockNumber}`,
     );
     await this.blockAssetsQueue.add(jobId, data, { jobId });
-    await this.cacheManager.set(`${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`, blockNumber);
+    await this.setLatestIndexedAssetsBlock(chainId, blockNumber);
   }
 
   public async addLockJob(data: IndexerLockJobData) {
diff --git a/backend/src/core/indexer/processor/block-assets.processor.ts b/backend/src/core/indexer/processor/block-assets.processor.ts
index b2b00818..f84daf86 100644
--- a/backend/src/core/indexer/processor/block-assets.processor.ts
+++ b/backend/src/core/indexer/processor/block-assets.processor.ts
@@ -107,7 +107,7 @@ export class IndexerBlockAssetsProcessor extends WorkerHost {
     } else {
       const indexerServiceFactory = this.moduleRef.get(IndexerServiceFactory);
       const indexerService = await indexerServiceFactory.getService(chainId);
-      indexerService.assetsFlow.emit(IndexerAssetsEvent.BlockAssetsIndexed, block);
+      indexerService.assetsFlow.emit(IndexerAssetsEvent.BlockAssetsIndexed, blockNumber);
       return;
     }
   }
diff --git a/backend/src/filters/all-exceptions.filter.ts b/backend/src/filters/all-exceptions.filter.ts
index 5f831d39..309a7b08 100644
--- a/backend/src/filters/all-exceptions.filter.ts
+++ b/backend/src/filters/all-exceptions.filter.ts
@@ -11,12 +11,15 @@ export class AllExceptionsFilter extends SentryGlobalGraphQLFilter {
   }
 
   catch(exception: unknown, host: ArgumentsHost) {
-    const ctx = host.switchToHttp();
-    const request = ctx.getRequest();
-    if (SKIP_REQUEST_URLS.includes(request.url)) {
-      const response = (exception as HttpException).getResponse();
-      this.httpAdapterHost.httpAdapter.reply(ctx.getResponse(), response, 200);
-      return;
+    const type = host.getType();
+    if (type === 'http') {
+      const ctx = host.switchToHttp();
+      const request = ctx.getRequest();
+      if (SKIP_REQUEST_URLS.includes(request.url)) {
+        const response = (exception as HttpException).getResponse();
+        this.httpAdapterHost.httpAdapter.reply(ctx.getResponse(), response, 200);
+        return;
+      }
     }
 
     super.catch(exception, host);