Merge pull request #219 from utxostack/develop

Merge develop to mainnet-backend (20240927)

.github/workflows/backend-test.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
 
     services:
-      redis:
+      redis-cache:
         image: redis
         options: >-
           --health-cmd "redis-cli ping"
@@ -23,6 +23,15 @@ jobs:
           --health-retries 5
         ports:
           - 6379:6379
+      redis-queue:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 6380:6379
 
     steps:
       - name: Checkout code
@@ -57,7 +66,8 @@ jobs:
           echo BITCOIN_ELECTRS_API_URL="{{ secrets.BITCOIN_ELECTRS_API_URL }}" >> .env
           echo CKB_EXPLORER_API_URL="${{ secrets.CKB_EXPLORER_API_URL }}" >> .env
           echo CKB_RPC_WEBSOCKET_URL="${{ secrets.CKB_RPC_WEBSOCKET_URL }}" >> .env
-          echo REDIS_URL="redis://localhost:6379" >> .env
+          echo REDIS_CACHE_URL="redis://localhost:6379" >> .env
+          echo REDIS_QUEUE_URL="redis://localhost:6380" >> .env
           echo DATABASE_URL="postgres://postgres:postgres@postgres:5432/explorer?sslmode=disable" >> .env
           cat .env
           pnpm run test

backend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@utxo-stack-explorer/backend",
-  "version": "0.1.1",
+  "version": "0.2.0",
   "description": "",
   "author": "",
   "private": true,
@@ -30,11 +30,13 @@
   },
   "dependencies": {
     "@apollo/server": "^4.11.0",
+    "@apollo/server-plugin-response-cache": "^4.1.3",
     "@as-integrations/fastify": "^2.1.1",
     "@cell-studio/mempool.js": "^2.5.3",
     "@ckb-lumos/bi": "^0.23.0",
     "@ckb-lumos/lumos": "^0.23.0",
     "@ckb-lumos/rpc": "^0.23.0",
+    "@nest-lab/throttler-storage-redis": "^1.0.0",
     "@nestjs/apollo": "^12.2.0",
     "@nestjs/axios": "^3.0.2",
     "@nestjs/bullmq": "^10.2.0",
@@ -64,14 +66,15 @@
     "fastify": "^4.28.1",
     "graphql": "^16.9.0",
     "graphql-query-complexity": "^1.0.0",
+    "ioredis": "^5.4.1",
     "lodash": "^4.17.21",
-    "nestjs-cacheable": "^1.0.0",
     "p-limit": "^3.1.0",
     "prisma": "^5.16.2",
     "redis": "^4.6.7",
     "reflect-metadata": "^0.2.0",
     "rpc-websockets": "^7.11.2",
     "rxjs": "^7.8.1",
+    "serialize-javascript": "^6.0.2",
     "ws": "^8.18.0",
     "zod": "^3.23.8"
   },

backend/redis-queue.conf

@@ -0,0 +1,80 @@
+# Redis configuration
+#
+# Example: https://raw.githubusercontent.com/redis/redis/7.4/redis.conf
+
+################################## NETWORK #####################################
+bind 0.0.0.0
+
+################################ SNAPSHOTTING  ################################
+
+# Save the DB to disk.
+#
+# save <seconds> <changes> [<seconds> <changes> ...]
+#
+# Redis will save the DB if the given number of seconds elapsed and it
+# surpassed the given number of write operations against the DB.
+#
+# Snapshotting can be completely disabled with a single empty string argument
+# as in following example:
+#
+# save ""
+#
+# Unless specified otherwise, by default Redis will save the DB:
+#   * After 3600 seconds (an hour) if at least 1 change was performed
+#   * After 300 seconds (5 minutes) if at least 100 changes were performed
+#   * After 60 seconds if at least 10000 changes were performed
+#
+# You can set these explicitly by uncommenting the following line.
+#
+save 3600 1 300 100 60 10000
+
+############################## APPEND ONLY MODE ###############################
+
+# By default Redis asynchronously dumps the dataset on disk. This mode is
+# good enough in many applications, but an issue with the Redis process or
+# a power outage may result into a few minutes of writes lost (depending on
+# the configured save points).
+#
+# The Append Only File is an alternative persistence mode that provides
+# much better durability. For instance using the default data fsync policy
+# (see later in the config file) Redis can lose just one second of writes in a
+# dramatic event like a server power outage, or a single write if something
+# wrong with the Redis process itself happens, but the operating system is
+# still running correctly.
+#
+# AOF and RDB persistence can be enabled at the same time without problems.
+# If the AOF is enabled on startup Redis will load the AOF, that is the file
+# with the better durability guarantees.
+#
+# Please check https://redis.io/topics/persistence for more information.
+appendonly yes
+
+# Redis can create append-only base files in either RDB or AOF formats. Using
+# the RDB format is always faster and more efficient, and disabling it is only
+# supported for backward compatibility purposes.
+aof-use-rdb-preamble yes
+
+# Set a memory usage limit to the specified amount of bytes.
+# When the memory limit is reached Redis will try to remove keys
+# according to the eviction policy selected (see maxmemory-policy).
+#
+# If Redis can't remove keys according to the policy, or if the policy is
+# set to 'noeviction', Redis will start to reply with errors to commands
+# that would use more memory, like SET, LPUSH, and so on, and will continue
+# to reply to read-only commands like GET.
+#
+# This option is usually useful when using Redis as an LRU or LFU cache, or to
+# set a hard memory limit for an instance (using the 'noeviction' policy).
+#
+# WARNING: If you have replicas attached to an instance with maxmemory on,
+# the size of the output buffers needed to feed the replicas are subtracted
+# from the used memory count, so that network problems / resyncs will
+# not trigger a loop where keys are evicted, and in turn the output
+# buffer of replicas is full with DELs of keys evicted triggering the deletion
+# of more keys, and so forth until the database is completely emptied.
+#
+# In short... if you have replicas attached it is suggested that you set a lower
+# limit for maxmemory so that there is some free RAM on the system for replica
+# output buffers (but this is not needed if the policy is 'noeviction').
+maxmemory 2gb
+maxmemory-policy noeviction

backend/src/app.module.ts

@@ -7,7 +7,6 @@ import { redisStore } from 'cache-manager-redis-yet';
 import { Env } from './env';
 import { CoreModule } from './core/core.module';
 import { ApiModule } from './modules/api.module';
-import { CacheableModule } from 'nestjs-cacheable';
 import { ScheduleModule } from '@nestjs/schedule';
 import { BullModule } from '@nestjs/bullmq';
 import configModule from './config';
@@ -18,13 +17,12 @@ import { BootstrapService } from './bootstrap.service';
   imports: [
     configModule,
     SentryModule.forRoot(),
-    CacheableModule.register(),
     CacheModule.registerAsync<RedisClientOptions>({
       isGlobal: true,
       imports: [ConfigModule],
       useFactory: async (configService: ConfigService<Env>) => {
         const store = (await redisStore({
-          url: configService.get('REDIS_URL'),
+          url: configService.get('REDIS_CACHE_URL'),
           isCacheable: (value) => value !== undefined,
         })) as unknown as CacheStore;
         return {
@@ -36,7 +34,7 @@ import { BootstrapService } from './bootstrap.service';
     BullModule.forRootAsync({
       imports: [ConfigModule],
       useFactory: async (configService: ConfigService<Env>) => {
-        const url = new URL(configService.get('REDIS_URL')!);
+        const url = new URL(configService.get('REDIS_QUEUE_URL')!);
         return {
           connection: {
             host: url.hostname,

backend/src/bootstrap.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { PrismaService } from './core/database/prisma/prisma.service';
 import { IndexerServiceFactory } from './core/indexer/indexer.factory';
+import cluster from 'node:cluster';
 
 @Injectable()
 export class BootstrapService {
@@ -9,7 +10,13 @@ export class BootstrapService {
   constructor(
     private prismaService: PrismaService,
     private IndexerServiceFactory: IndexerServiceFactory,
-  ) {}
+  ) { }
+
+  public async bootstrap() {
+    if (cluster.isPrimary) {
+      await this.bootstrapAssetsIndex();
+    }
+  }
 
   public async bootstrapAssetsIndex() {
     const chains = await this.prismaService.chain.findMany();

backend/src/cluster.service.ts

@@ -0,0 +1,30 @@
+import { Injectable, Logger } from '@nestjs/common';
+import cluster from 'node:cluster';
+import * as process from 'node:process';
+import * as os from 'node:os';
+import { envSchema } from './env';
+
+const numCPUs = os.cpus().length;
+const env = envSchema.parse(process.env);
+
+@Injectable()
+export class ClusterService {
+  private static logger = new Logger(ClusterService.name);
+
+  public static clusterize(callback: Function): void {
+    if (cluster.isPrimary) {
+      this.logger.log(`PRIMIRY PROCESS (${process.pid}) IS RUNNING `);
+      const workersNum = Math.min(env.CLUSTER_WORKERS_NUM, numCPUs);
+      for (let i = 0; i < workersNum; i++) {
+        cluster.fork();
+      }
+      cluster.on('exit', (worker) => {
+        this.logger.log(`WORKER ${worker.process.pid} DIED, FORKING NEW ONE`);
+        cluster.fork();
+      });
+    } else {
+      this.logger.log(`WORKER PROCESS (${process.pid}) IS RUNNING`);
+      callback();
+    }
+  }
+}

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 = {

backend/src/core/bitcoin-api/bitcoin-api.service.ts

@@ -9,8 +9,8 @@ import { MempoolService } from './provider/mempool.service';
 import { ChainInfo, Transaction } from './bitcoin-api.schema';
 import { ONE_HOUR_MS, ONE_MONTH_MS, TEN_MINUTES_MS } from 'src/common/date';
 import { Cacheable } from 'src/decorators/cacheable.decorator';
-import { PLimit } from 'src/decorators/plimit.decorator';
 import * as Sentry from '@sentry/nestjs';
+import { PLimit } from 'src/decorators/plimit.decorator';
 
 type MethodParameters<T, K extends keyof T> = T[K] extends (...args: infer P) => any ? P : never;
 type MethodReturnType<T, K extends keyof T> = T[K] extends (...args: any[]) => infer R ? R : never;
@@ -88,6 +88,7 @@ export class BitcoinApiService {
     }
   }
 
+  @PLimit({ concurrency: 200 })
   private async call<K extends keyof IBitcoinDataProvider>(
     method: K,
     ...args: MethodParameters<IBitcoinDataProvider, K>

backend/src/core/blockchain/blockchain.service.ts

@@ -14,6 +14,7 @@ import { ONE_MONTH_MS } from 'src/common/date';
 import { CKB_MIN_SAFE_CONFIRMATIONS } from 'src/constants';
 import * as Sentry from '@sentry/nestjs';
 import { Chain } from '@prisma/client';
+import { PLimit } from 'src/decorators/plimit.decorator';
 
 class WebsocketError extends Error {
   constructor(message: string) {
@@ -70,6 +71,12 @@ export class BlockchainService {
     });
   }
 
+  @PLimit({ concurrency: 200 })
+  private async call(method: string, params: any[]): Promise<any> {
+    await this.websocketReady;
+    return this.websocket.call(method, params);
+  }
+
   private handleReconnection() {
     if (this.reconnectAttempts < this.maxReconnectAttempts) {
       this.reconnectAttempts++;
@@ -122,7 +129,7 @@ export class BlockchainService {
   ): Promise<TransactionWithStatusResponse> {
     await this.websocketReady;
     this.logger.debug(`get_transaction - txHash: ${txHash}`);
-    const response = await this.websocket.call('get_transaction', [txHash]);
+    const response = await this.call('get_transaction', [txHash]);
     const tx = response as TransactionWithStatusResponse;
     // XXX: we don't need these fields by default, remove them to save cache/memory space
     if (!withData) {
@@ -162,7 +169,7 @@ export class BlockchainService {
   ): Promise<Block> {
     await this.websocketReady;
     this.logger.debug(`get_block - blockHash: ${blockHash}`);
-    const response = await this.websocket.call('get_block', [blockHash]);
+    const response = await this.call('get_block', [blockHash]);
     const block = response as Block;
     if (!withTxData) {
       block.transactions = block.transactions.map((tx) => {
@@ -204,7 +211,7 @@ export class BlockchainService {
   ): Promise<Block> {
     await this.websocketReady;
     this.logger.debug(`get_block_by_number - blockNumber: ${blockNumber}`);
-    const response = await this.websocket.call('get_block_by_number', [
+    const response = await this.call('get_block_by_number', [
       BI.from(blockNumber).toHexString(),
     ]);
     const block = response as Block;
@@ -231,7 +238,7 @@ export class BlockchainService {
   public async getBlockEconomicState(blockHash: string): Promise<BlockEconomicState> {
     await this.websocketReady;
     this.logger.debug(`get_block_economic_state - blockHash: ${blockHash}`);
-    const blockEconomicState = await this.websocket.call('get_block_economic_state', [blockHash]);
+    const blockEconomicState = await this.call('get_block_economic_state', [blockHash]);
     return blockEconomicState as BlockEconomicState;
   }
 
@@ -244,10 +251,16 @@ export class BlockchainService {
   public async getTipBlockNumber(): Promise<number> {
     await this.websocketReady;
     this.logger.debug('get_tip_block_number');
-    const tipBlockNumber = await this.websocket.call('get_tip_block_number', []);
+    const tipBlockNumber = await this.call('get_tip_block_number', []);
     return BI.from(tipBlockNumber).toNumber();
   }
 
+  @Cacheable({
+    namespace: 'BlockchainService',
+    key: (searchKey: SearchKey, order: 'asc' | 'desc', limit: string, after?: string) =>
+      `getTransactions:${JSON.stringify(searchKey)}:${order}:${limit}:${after}`,
+    ttl: 10_000,
+  })
   public async getTransactions(
     searchKey: SearchKey,
     order: 'asc' | 'desc',
@@ -258,11 +271,17 @@ export class BlockchainService {
     this.logger.debug(
       `get_transactions - searchKey: ${JSON.stringify(searchKey)}, order: ${order}, limit: ${limit}, after: ${after}`,
     );
-    const result = await this.websocket.call('get_transactions', [searchKey, order, limit, after]);
+    const result = await this.call('get_transactions', [searchKey, order, limit, after]);
     const transactions = result as GetTransactionsResult;
     return transactions;
   }
 
+  @Cacheable({
+    namespace: 'BlockchainService',
+    key: (searchKey: SearchKey, order: 'asc' | 'desc', limit: string, after?: string) =>
+      `getCells:${JSON.stringify(searchKey)}:${order}:${limit}:${after}`,
+    ttl: 10_000,
+  })
   public async getCells(
     searchKey: SearchKey,
     order: 'asc' | 'desc',
@@ -274,7 +293,7 @@ export class BlockchainService {
     this.logger.debug(
       `get_cells - searchKey: ${JSON.stringify(searchKey)}, order: ${order}, limit: ${limit}, after: ${after}`,
     );
-    const result = await this.websocket.call('get_cells', [searchKey, order, limit, after]);
+    const result = await this.call('get_cells', [searchKey, order, limit, after]);
     const cells = result as GetCellsResult;
     cells.objects = cells.objects.map((cell) => {
       if (!withData) {

backend/src/core/core.service.ts

@@ -14,6 +14,8 @@ import { Env } from 'src/env';
 import { Transaction } from './blockchain/blockchain.interface';
 import { BlockchainServiceFactory } from './blockchain/blockchain.factory';
 import { LeapDirection } from '@prisma/client';
+import { ONE_MONTH_MS } from 'src/common/date';
+import { Cacheable } from 'src/decorators/cacheable.decorator';
 
 export const CELLBASE_TX_HASH =
   '0x0000000000000000000000000000000000000000000000000000000000000000';
@@ -72,6 +74,13 @@ export class CoreService {
     );
   }
 
+  @Cacheable({
+    namespace: 'CoreService',
+    key: (chainId: number, ckbTx: Transaction) => {
+      return `getLeapDirectionByCkbTx:${chainId}:${ckbTx.hash}`;
+    },
+    ttl: ONE_MONTH_MS,
+  })
   public async getLeapDirectionByCkbTx(chainId: number, ckbTx: Transaction) {
     const blockchainService = this.blockchainServiceFactory.getService(chainId);
     const inputCells = await Promise.all(