增加arbitrm事件的解析

.vscode/launch.json

@@ -17,5 +17,18 @@
       ],
       "type": "pwa-node"
     },
+    {
+      "name": "Debug Monitor",
+      "request": "launch",
+      "runtimeArgs": [
+          "run-script",
+          "dev:monitor"
+      ],
+      "runtimeExecutable": "npm",
+      "skipFiles": [
+          "<node_internals>/**"
+      ],
+      "type": "pwa-node"
+    },
   ]
 }

package.json

@@ -11,7 +11,8 @@
     "dev:api": "npx ts-node src/api.ts",
     "lint": "eslint --ext .ts src/**",
     "format": "eslint --ext .ts src/** --fix",
-    "dev:monitor": "NODE_ENV=development ts-node -r tsconfig-paths/register src/monitor.ts",
+    "dev:monitor": "npx ts-node src/monitor.ts",
+    "dev:monitor2": "NODE_ENV=development ts-node -r tsconfig-paths/register src/monitor.ts",
     "prod:monitor": "NODE_PATH=./dist node dist/monitor.js"
   },
   "author": "z",

src/chain/allchain.ts

@@ -210,7 +210,7 @@ export const AllChains = [
   {
     name: 'Arbitrum One',
     type: 'Mainnet',
-    rpc: 'https://endpoints.omniatech.io/v1/arbitrum/one/public|https://rpc.ankr.com/arbitrum',
+    rpc: 'https://arb-mainnet.g.alchemy.com/v2/tFPlLg-MT8uGkM5wKHyoBqEHWJJ0o3Nt|https://endpoints.omniatech.io/v1/arbitrum/one/public|https://rpc.ankr.com/arbitrum',
     id: 42161,
     network: 'ARBITRUM',
     symbol: 'ETH',

src/config/events.json

@@ -1,9 +1,10 @@
 [
   {
-    "address": "0x8135D4F16A7AAA269cbf61CE9659D3A272BF541f",
+    "address": "0x3F13F83E6363D97d0353cAAfACA08B05D9BF3637",
-    "event": "Confirmation",
+    "chain": 42161,
-    "abi": "BEMultiSigWallet",
+    "event": "Transfer",
-    "fromBlock": 34353697,
+    "abi": "ERC721",
-    "eventProcesser": "ScheduleConfirmEvent"
+    "fromBlock": 105016690,
+    "eventProcesser": "NftTransferEvent"
   }
 ]

src/models/FtTransferEvent.ts

@@ -3,12 +3,14 @@ import { dbconn } from 'decorators/dbconn'
 import { BaseModule } from './Base'
 
 @dbconn()
-@index({ address: 1 }, { unique: false })
+@index({ address: 1, chain: 1 }, { unique: false })
-@index({ transactionHash: 1, from: 1, to: 1 }, { unique: true })
+@index({ chain: 1, transactionHash: 1, address: 1, from: 1, to: 1 }, { unique: true })
 @modelOptions({
   schemaOptions: { collection: 'ft_transfer_event', timestamps: true },
 })
 export class FtTransferEventClass extends BaseModule {
+  @prop()
+  public chain: number
   @prop({ required: true })
   public address!: string
   @prop()
@@ -16,6 +18,10 @@ export class FtTransferEventClass extends BaseModule {
   @prop({ required: true })
   public transactionHash: string
   @prop()
+  public transactionIndex: number
+  @prop()
+  public logIndex: number
+  @prop()
   public blockNumber: number
   @prop()
   public blockHash: string
@@ -28,6 +34,8 @@ export class FtTransferEventClass extends BaseModule {
   @prop()
   public amount: string
   @prop()
+  public eventId: string
+  @prop()
   public blockTime: number
 
   public static async saveEvent(event: any) {
@@ -35,21 +43,36 @@ export class FtTransferEventClass extends BaseModule {
     if (amount == undefined) {
       return
     }
-    const from = event.returnValues?.from
+    const from = (event.source || event.returnValues?.from).toLowerCase()
-    const to = event.returnValues?.to
+    const to = (event.target || event.returnValues?.to).toLowerCase()
+    const address = (event.address || event.tokenAddress).toLowerCase()
     const data = {
-      address: event.address,
+      chain: event.chain,
-      blockNumber: event.blockNumber,
+      address,
+      blockNumber: event.blockHeight || event.blockNumber,
       blockHash: event.blockHash,
       removed: event.removed,
       event: event.event,
       from,
       to,
-      transactionHash: event.transactionHash,
+      transactionHash: event.transactionHash || event.hash,
+      transactionIndex: event.transactionIndex,
       amount,
-      blockTime: event.timestamp * 1000
+      eventId: event.id,
+      logIndex: event.logIndex,
+      blockTime: event.timestamp,
+      $inc: { version: 1 },
     }
-    return FtTransferEvent.insertOrUpdate({ transactionHash: event.transactionHash, amount, from, to }, data)
+    return FtTransferEvent.insertOrUpdate(
+      {
+        chain: event.chain,
+        address,
+        amount,
+        from,
+        to,
+      },
+      data,
+    )
   }
 }
 

src/models/NftTransferEvent.ts

@@ -3,12 +3,14 @@ import { dbconn } from 'decorators/dbconn'
 import { BaseModule } from './Base'
 
 @dbconn()
-@index({ tokenId: 1 }, { unique: false })
+@index({ address: 1, chain: 1 }, { unique: false })
-@index({ transactionHash: 1, tokenId: 1, from: 1, to: 1 }, { unique: true })
+@index({ chain: 1, transactionHash: 1, address: 1, tokenId: 1, from: 1, to: 1 }, { unique: true })
 @modelOptions({
   schemaOptions: { collection: 'nft_transfer_event', timestamps: true },
 })
 export class NftTransferEventClass extends BaseModule {
+  @prop({ required: true })
+  public chain: number
   @prop({ required: true })
   public address!: string
   @prop()
@@ -16,6 +18,10 @@ export class NftTransferEventClass extends BaseModule {
   @prop({ required: true })
   public transactionHash: string
   @prop()
+  public transactionIndex: number
+  @prop()
+  public logIndex: number
+  @prop()
   public blockNumber: number
   @prop()
   public blockHash: string
@@ -28,33 +34,48 @@ export class NftTransferEventClass extends BaseModule {
   @prop()
   public tokenId: string
   @prop()
+  public eventId: string
+  @prop()
   public blockTime: number
   @prop({ default: 0 })
   public version: number
 
   public static async saveEvent(event: any) {
-    if (!event.success) {
+    const tokenId = event.tokenId || event.returnValues?.tokenId
-      return
-    }
-    const tokenId = event.tokenId
     if (!tokenId) {
       return
     }
-    const from = event.source
+    const from = (event.source || event.returnValues?.from).toLowerCase()
-    const to = event.target
+    const to = (event.target || event.returnValues?.to).toLowerCase()
+    const address = (event.address || event.tokenAddress).toLowerCase()
     const data = {
-      address: event.tokenAddress,
+      chain: event.chain,
-      blockNumber: event.blockHeight,
+      address,
+      blockNumber: event.blockHeight || event.blockNumber,
+      blockHash: event.blockHash,
       removed: event.removed,
       from,
       to,
-      transactionHash: event.hash,
+      transactionHash: event.transactionHash || event.hash,
+      transactionIndex: event.transactionIndex,
       tokenId,
-      blockTime: new Date(event.time).getTime(),
+      eventId: event.id,
+      logIndex: event.logIndex,
+      blockTime: event.timestamp,
       $inc: { version: 1 },
     }
 
-    return NftTransferEvent.insertOrUpdate({ transactionHash: event.hash, tokenId, from, to }, data)
+    return NftTransferEvent.insertOrUpdate(
+      {
+        chain: event.chain,
+        transactionHash: event.transactionHash || event.hash,
+        address,
+        tokenId,
+        from,
+        to,
+      },
+      data,
+    )
   }
 }
 

src/models/ScheduleConfirmEvent.ts

@@ -1,7 +1,6 @@
 import { getModelForClass, index, modelOptions, prop } from '@typegoose/typegoose'
 import { dbconn } from 'decorators/dbconn'
 import logger from 'logger/logger'
-import { TaskSvr } from 'service/task.service'
 import { BaseModule } from './Base'
 
 @dbconn()
@@ -52,7 +51,7 @@ export class ScheduleConfirmEventClass extends BaseModule {
     if (record.version === 1) {
       logger.log('receive events: ' + JSON.stringify(record.scheduleIds))
       for (let id of record.scheduleIds) {
-        await new TaskSvr().parseOneSchedule(id)
+        // await new TaskSvr().parseOneSchedule(id)
       }
     }
     return record

src/monitor.ts

@@ -6,8 +6,6 @@ dotenv.config({ path: envFile })
 import { EventSyncSvr } from 'service/event.sync.service'
 import { NftTransferEvent } from 'models/NftTransferEvent'
 import { FtTransferEvent } from 'models/FtTransferEvent'
-import { ScheduleConfirmEvent } from 'models/ScheduleConfirmEvent'
-import { ScheduleExecutedEvent } from 'models/ScheduleExecutedEvent'
 
 import 'common/Extend'
 
@@ -17,8 +15,6 @@ let lock = false
 let eventProcessers = {
   NftTransferEvent: NftTransferEvent,
   FtTransferEvent: FtTransferEvent,
-  ScheduleConfirmEvent: ScheduleConfirmEvent,
-  ScheduleExecutedEvent: ScheduleExecutedEvent,
 }
 
 const events = require('config/events.json')
@@ -27,7 +23,8 @@ async function initEventSvrs() {
   // let nfts = [{ address: '0x37c30a2945799a53c5358636a721b442458fa691' }]
   for (let event of events) {
     let eventSvr = new EventSyncSvr({
-      address: event.address,
+      chain: event.chain,
+      address: event.address.toLowerCase(),
       event: event.event,
       abi: require('abis/' + event.abi + '.json').abi,
       fromBlock: event.fromBlock,

src/service/event.sync.service.ts

@@ -2,7 +2,6 @@ import assert from 'assert'
 import { AllChains } from 'chain/allchain'
 import { HttpRetryProvider } from 'chain/HttpRetryProvider'
 import logger from 'logger/logger'
-import { NftTransferEvent } from 'models/NftTransferEvent'
 import { RedisClient } from 'redis/RedisClient'
 
 import { clearTimeCache, getPastEventsIter, processEvents } from 'utils/contract.util'
@@ -17,24 +16,27 @@ export class EventSyncSvr {
   event: string
   blockKey = ''
   address: string
+  chain: number
   eventProcesser: any
 
   constructor({
+    chain,
     address,
     event,
     abi,
     fromBlock,
     eventProcesser,
   }: {
+    chain: number
     address: string
     event: string
     abi: any
     fromBlock: number
     eventProcesser: any
   }) {
-    const defaultChain = parseInt(process.env.CHAIN_DEFAULT)
+    this.chain = chain
-    const chainData = AllChains.find(o => o.id === defaultChain)
+    const chainData = AllChains.find(o => o.id === chain)
-    assert(chainData, `chain data with ${defaultChain} not found`)
+    assert(chainData, `chain data with ${this.chain} not found`)
     this.provider = new HttpRetryProvider(chainData.rpc.split('|'))
     // @ts-ignore
     this.web3 = new Web3(this.provider)
@@ -42,7 +44,7 @@ export class EventSyncSvr {
     this.address = this.contract.options.address
     this.event = event
     this.fromBlock = fromBlock
-    this.blockKey = `${address.toLowerCase()}_${event}`
+    this.blockKey = `${chain}_${address.toLowerCase()}_${event}`
     this.eventProcesser = eventProcesser
   }
 
@@ -58,13 +60,14 @@ export class EventSyncSvr {
     }
     logger.log(`query events:: ${this.event} address: ${this.address} from: ${this.fromBlock} to: ${this.toBlock}`)
     let events = getPastEventsIter({
+      chain: this.chain,
       contract: this.contract,
       event: this.event,
       fromBlock: this.fromBlock,
       toBlock: this.toBlock,
     })
     // this.fromBlock = this.toBlock
-    await processEvents(this.web3, events, this.eventProcesser.saveEvent)
+    await processEvents(this.web3, events, this.chain, this.eventProcesser.saveEvent)
     // 处理完一种nft后, 清楚block的timestamp缓存
     clearTimeCache()
   }

src/utils/contract.util.ts

@@ -170,12 +170,14 @@ export async function getPastEvents({
 }
 
 export function* getPastEventsIter({
+  chain,
   contract,
   event,
   fromBlock,
   toBlock,
   options,
 }: {
+  chain: number
   contract: any
   event: string
   fromBlock: number
@@ -183,7 +185,7 @@ export function* getPastEventsIter({
   options?: any
 }) {
   const address = contract.options.address
-  const redisKey = `${address.toLowerCase()}_${event}`
+  const redisKey = `${chain}_${address.toLowerCase()}_${event}`
   logger.debug(`*getPastEventsIter: ${event} from: ${fromBlock} to: ${toBlock}`)
   let from = toBN(fromBlock)
   let to = toBN(fromBlock).add(queryRange)
@@ -198,7 +200,7 @@ export function* getPastEventsIter({
   yield new RedisClient().set(redisKey, toBlockBN.add(ONE) + '')
 }
 
-export async function processEvents(web3, iterator, processedEvent) {
+export async function processEvents(web3, iterator, chain: number, processedEvent) {
   for (const getPastEventPromise of iterator) {
     const events = await getPastEventPromise
     for (const event of events) {
@@ -211,6 +213,7 @@ export async function processEvents(web3, iterator, processedEvent) {
           blockTimeMap.set(event.blockNumber, blockData.timestamp)
         }
       }
+      event.chain = chain
       await processedEvent(event)
     }
   }