feat: bill of sudt

offchain-modules/packages/app-monitor/src/balanceProvider.ts

@@ -1,11 +1,11 @@
-import CKB from '@nervosnetwork/ckb-sdk-core';
-import { CkbIndexer, Order, ScriptType, SearchKey } from '@force-bridge/x/dist/ckb/tx-helper/indexer';
 import { utils } from '@ckb-lumos/base';
+import { EthAsset } from '@force-bridge/x/dist/ckb/model/asset';
+import { CkbIndexer, Order, ScriptType, SearchKey } from '@force-bridge/x/dist/ckb/tx-helper/indexer';
+import { getOwnerTypeHash } from '@force-bridge/x/dist/ckb/tx-helper/multisig/multisig_helper';
+import { logger } from '@force-bridge/x/dist/utils/logger';
+import CKB from '@nervosnetwork/ckb-sdk-core';
 import Web3 from 'web3';
 import { AbiItem } from 'web3-utils';
-import { logger } from '@force-bridge/x/dist/utils/logger';
-import { getOwnerTypeHash } from '@force-bridge/x/dist/ckb/tx-helper/multisig/multisig_helper';
-import { EthAsset } from '@force-bridge/x/dist/ckb/model/asset';
 
 const minERC20ABI = [
   // balanceOf

offchain-modules/packages/app-monitor/src/duration.ts

@@ -25,6 +25,8 @@ export interface EthConfig {
 
 export interface CkbConfig {
   lastHandledBlock: number;
+  sudtBillReconcBlock: number;
+  sudtBillLastReconcBlock: number;
   matchCount: {
     burn: number;
     mint: number;
@@ -55,6 +57,8 @@ export function NewDurationCfg(): Duration {
     },
     ckb: {
       lastHandledBlock: ForceBridgeCore.config.ckb.startBlockHeight,
+      sudtBillLastReconcBlock: 0,
+      sudtBillReconcBlock: 0,
       matchCount: {
         mint: 0,
         burn: 0,

offchain-modules/packages/app-monitor/src/monitor.ts

@@ -1,23 +1,25 @@
 import { parseAddress } from '@ckb-lumos/helpers';
-import { CkbBurnRecord, CkbMintRecord, EthLockRecord, EthUnlockRecord } from '@force-bridge/reconc/dist';
+import { CkbBurnRecord, CkbMintRecord, EthLockRecord, EthUnlockRecord, SudtRecord } from '@force-bridge/reconc/dist';
 import { nonNullable } from '@force-bridge/x';
 import { IndexerCollector } from '@force-bridge/x/dist/ckb/tx-helper/collector';
 import { getOwnerTypeHash } from '@force-bridge/x/dist/ckb/tx-helper/multisig/multisig_helper';
 import { verifierEndpoint, feeAccounts } from '@force-bridge/x/dist/config';
 import { bootstrap, ForceBridgeCore } from '@force-bridge/x/dist/core';
+import { KVDb } from '@force-bridge/x/dist/db/kv';
+import { SudtDb } from '@force-bridge/x/dist/db/sudt';
 import { CKBRecordObservable } from '@force-bridge/x/dist/reconc';
-import { asyncSleep, foreverPromise } from '@force-bridge/x/dist/utils';
+import { asyncSleep, foreverPromise, getDBConnection } from '@force-bridge/x/dist/utils';
 import { logger } from '@force-bridge/x/dist/utils/logger';
 import {
   createCKBRecordObservable,
   createETHRecordObservable,
   EthRecordObservable,
 } from '@force-bridge/xchain-eth/dist/reconc';
 import axios from 'axios';
+import dayjs from 'dayjs';
 import { ethers } from 'ethers';
 import { assetListPriceChange } from './assetPrice';
 import { BalanceProvider } from './balanceProvider';
-import dayjs from 'dayjs';
 import { WebHook } from './discord';
 import { Duration, EventItem, monitorEvent, NewDurationCfg, readMonitorConfig, writeMonitorConfig } from './duration';
 
@@ -105,6 +107,8 @@ export class Monitor {
   private durationConfig: Duration;
   private gasPriceRecorder: GasPriceRecorder;
   private balanceProvider: BalanceProvider;
+  private sudtDb: SudtDb;
+  private kvDb: KVDb;
   webHookInfoUrl: string;
   webHookErrorUrl: string;
 
@@ -143,6 +147,22 @@ export class Monitor {
     this.durationConfig.ckb.pending.mints.set(mint.fromTxId!, newEvent(mint));
   }
 
+  async onSudtRecord(record: SudtRecord): Promise<void> {
+    logger.info(`Receive sudt:${JSON.stringify(record)}`);
+    try {
+      await this.sudtDb.createSudtTransferRecord(
+        record.txId,
+        record.direction,
+        record.lock,
+        record.token,
+        record.amount,
+        record.index,
+      );
+    } catch (e) {
+      logger.error(e.stack);
+    }
+  }
+
   async onCkbBurnRecord(burn: CkbBurnRecord): Promise<void> {
     logger.info(`Receive ckbBurn:${JSON.stringify(burn)}`);
     this.durationConfig.ckb.pending.burns.set(burn.txId, newEvent(burn));
@@ -212,6 +232,7 @@ export class Monitor {
     this.observeCkbEvent().catch((err) => {
       logger.error(`Monitor observeCkbEvent error:${err.stack}`);
     });
+    void this.reconcSudtBill();
     // this.observeAssetPrice().catch((err) => {
     //   logger.error(`Monitor observeAssetPrice error:${err.stack}`);
     // });
@@ -230,7 +251,38 @@ export class Monitor {
     }
   }
 
+  async reconcSudtBill(): Promise<void> {
+    let fromBlock = this.durationConfig.ckb.sudtBillLastReconcBlock;
+    let toBlock = fromBlock + 100;
+    while (fromBlock < this.durationConfig.ckb.sudtBillReconcBlock) {
+      toBlock > this.durationConfig.ckb.sudtBillReconcBlock ? this.durationConfig.ckb.sudtBillReconcBlock : toBlock;
+      try {
+        logger.info(`Monitor reconcSudtBill fromBlock: ${fromBlock} toBlock: ${toBlock}`);
+        await this.ckbRecordObservable
+          .observeSudtRecord({ fromBlock: `0x${fromBlock.toString(16)}`, toBlock: `0x${toBlock.toString(16)}` })
+          .subscribe((records) => {
+            records.forEach((record) => {
+              void this.onSudtRecord(record);
+            });
+          });
+      } catch (e) {
+        logger.error(e.stack);
+      }
+      try {
+        await this.kvDb.set('sudt_bill_last_handle_block', toBlock.toString());
+      } catch (e) {
+        logger.error(e.stack);
+      }
+      fromBlock = toBlock + 1;
+      toBlock = fromBlock + 100;
+      await asyncSleep(10);
+    }
+  }
+
   async init(): Promise<void> {
+    const conn = await getDBConnection();
+    this.sudtDb = new SudtDb(conn);
+    this.kvDb = new KVDb(conn);
     let durationConfig = readMonitorConfig();
     if (!durationConfig) {
       durationConfig = NewDurationCfg();
@@ -246,6 +298,19 @@ export class Monitor {
     if (ForceBridgeCore.config.monitor!.expiredCheckInterval > 0) {
       expiredCheckInterval = ForceBridgeCore.config.monitor!.expiredCheckInterval;
     }
+
+    this.durationConfig.ckb.sudtBillReconcBlock = parseInt(
+      (await this.kvDb.get('sudt_bill_reconc_handle_block')) ?? '0',
+    );
+    if (this.durationConfig.ckb.sudtBillReconcBlock == 0) {
+      await this.kvDb.set('sudt_bill_reconc_handle_block', this.durationConfig.ckb.lastHandledBlock.toString());
+      this.durationConfig.ckb.sudtBillReconcBlock = this.durationConfig.ckb.lastHandledBlock;
+    }
+
+    this.durationConfig.ckb.sudtBillLastReconcBlock = parseInt(
+      (await this.kvDb.get('sudt_bill_last_handle_block')) ?? '0',
+    );
+    console.log(this.durationConfig);
   }
 
   checkExpiredEvent(): void {
@@ -514,6 +579,12 @@ export class Monitor {
           })
           .subscribe((record) => this.onCkbBurnRecord(record));
 
+        await this.ckbRecordObservable.observeSudtRecord({ fromBlock, toBlock }).subscribe((records) => {
+          records.forEach((record) => {
+            void this.onSudtRecord(record);
+          });
+        });
+
         this.durationConfig.ckb.lastHandledBlock = toBlockNum;
       },
       {

offchain-modules/packages/reconc/src/index.ts

@@ -58,6 +58,13 @@ export type CkbMintRecord = ToRecord & {
   blockHash?: string;
 };
 
+export type SudtRecord = FromRecord & {
+  token: string;
+  lock: string;
+  direction: 'in' | 'out';
+  index: number;
+};
+
 export class Reconciliation {
   constructor(public from: FromRecord[], public to: ToRecord[]) {}
 

offchain-modules/packages/scripts/src/testnetDocker.ts

@@ -1,16 +1,16 @@
 import fs from 'fs';
 import path from 'path';
 import { KeyStore } from '@force-bridge/keystore/dist';
+import { nonNullable } from '@force-bridge/x';
 import { OwnerCellConfig } from '@force-bridge/x/dist/ckb/tx-helper/deploy';
 import { Config, WhiteListEthAsset, CkbDeps } from '@force-bridge/x/dist/config';
 import { getFromEnv, privateKeyToCkbPubkeyHash, writeJsonToFile } from '@force-bridge/x/dist/utils';
 import { logger, initLog } from '@force-bridge/x/dist/utils/logger';
 import * as dotenv from 'dotenv';
 import * as lodash from 'lodash';
 import * as Mustache from 'mustache';
-import { execShellCmd, PATH_PROJECT_ROOT, pathFromProjectRoot } from './utils';
+import { execShellCmd, pathFromProjectRoot } from './utils';
 import { deployDev } from './utils/deploy';
-import { nonNullable } from '@force-bridge/x';
 
 dotenv.config({ path: process.env.DOTENV_PATH || '.env' });
 
@@ -99,12 +99,12 @@ async function generateConfig(
   writeJsonToFile({ forceBridge: watcherConfig }, path.join(configPath, 'watcher/force_bridge.json'));
   //monitor
   const monitorConfig: Config = lodash.cloneDeep(baseConfig);
-  monitorConfig.common.orm = undefined;
   monitorConfig.common.port = undefined;
   monitorConfig.common.openMetric = false;
   monitorConfig.common.role = 'watcher';
   monitorConfig.common.log.identity = 'monitor';
   monitorConfig.common.log.logFile = path.join(configPath, 'monitor/force_bridge.log');
+  monitorConfig.common.orm!.host = 'monitor_db';
   monitorConfig.monitor = {
     discordWebHook: monitorDiscordWebHook,
     expiredTime: 1800000, //30 minutes
@@ -236,6 +236,13 @@ services:
     depends_on:
       - {{name}}_db
 {{/verifiers}}
+  monitor_db:
+    image: mysql:5.7
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: forcebridge
+    ports:
+      - 3064:3306
   monitor:
     image: node:14.18.1-bullseye
     restart: on-failure
@@ -250,6 +257,8 @@ services:
       cd /app/offchain-modules;
       npx ts-node ./packages/app-cli/src/index.ts monitor -cfg /data/force_bridge.json
       '
+    depends_on:
+      - monitor_db
 volumes:
   force-bridge-node-modules:
     external: true

offchain-modules/packages/x/src/db/entity/sudt.ts

@@ -0,0 +1,32 @@
+import { Column, CreateDateColumn, Entity, Index, PrimaryGeneratedColumn, UpdateDateColumn } from 'typeorm';
+
+@Entity()
+@Index(['txHash', 'index', 'direction'], { unique: true })
+export class Sudt {
+  @PrimaryGeneratedColumn('increment')
+  id: number;
+
+  @Column()
+  txHash: string;
+
+  @Column()
+  direction: number;
+
+  @Column({ length: 10240 })
+  address: string;
+
+  @Column()
+  sudtArgs: string;
+
+  @Column()
+  index: number;
+
+  @Column()
+  amount: string;
+
+  @CreateDateColumn()
+  createdAt: string;
+
+  @UpdateDateColumn()
+  updatedAt: string;
+}

offchain-modules/packages/x/src/db/index.ts

@@ -5,3 +5,4 @@ export * from './tron';
 export * from './btc';
 export * from './eos';
 export * from './kv';
+export * from './sudt';

offchain-modules/packages/x/src/db/sudt.ts

@@ -0,0 +1,29 @@
+import { Connection, Repository } from 'typeorm';
+import { Sudt } from './entity/sudt';
+
+export class SudtDb {
+  private sudtRepository: Repository<Sudt>;
+
+  constructor(private connection: Connection) {
+    this.sudtRepository = connection.getRepository(Sudt);
+  }
+
+  async createSudtTransferRecord(
+    hash: string,
+    direction: 'in' | 'out',
+    address: string,
+    sudtArgs: string,
+    amount: string,
+    index: number,
+  ): Promise<void> {
+    const record = this.sudtRepository.create({
+      txHash: hash,
+      direction: direction == 'in' ? 1 : -1,
+      address,
+      sudtArgs,
+      amount,
+      index,
+    });
+    await this.sudtRepository.save(record);
+  }
+}