Merge pull request #11 from helix-bridge/xiaoch05-reindexer

reindexer

apollo/.env

@@ -1,2 +1,22 @@
 SUBQL=https://api.subquery.network/sq/helix-bridge/
-THEGRAPH=https://crab-thegraph.darwinia.network/subgraphs/name/wormhole/Sub2SubMappingTokenFactory
+THEGRAPH=https://crab-thegraph.darwinia.network/subgraphs/name/wormhole/Sub2SubMappingTokenFactory
+FETCH_DARWINIA_TO_CRAB_LOCK_DATA_INTERVAL=10000
+UPDATE_DARWINIA_TO_CRAB_LOCK_DATA_INTERVAL=5000
+FETCH_DARWINIA_TO_CRAB_LOCK_DATA_FIRST=10
+UPDATE_DARWINIA_TO_CRAB_LOCK_DATA_FIRST=10
+FETCH_DARWINIA_TO_CRAB_BURN_DATA_INTERVAL=10000
+UPDATE_DARWINIA_TO_CRAB_BURN_DATA_INTERVAL=15000
+FETCH_DARWINIA_TO_CRAB_BURN_DATA_FIRST=10
+UPDATE_DARWINIA_TO_CRAB_BURN_DATA_FIRST=10
+
+FETCH_CRAB_TO_CRAB_DVM_DATA_INTERVAL=10000
+FETCH_CRAB_TO_CRAB_DVM_DATA_FIRST=10
+
+# This was inserted by `prisma init`:
+# Environment variables declared in this file are automatically made available to Prisma.
+# See the documentation for more detail: https://pris.ly/d/prisma-schema#accessing-environment-variables-from-the-schema
+
+# Prisma supports the native connection string format for PostgreSQL, MySQL, SQLite, SQL Server, MongoDB and CockroachDB (Preview).
+# See the documentation for all the connection string options: https://pris.ly/d/connection-strings
+
+DATABASE_URL="postgresql://admin:admin@localhost:5432/apollo_graph?schema=public"

apollo/package.json

@@ -28,12 +28,15 @@
     "@nestjs/core": "^8.0.0",
     "@nestjs/graphql": "^10.0.9",
     "@nestjs/platform-express": "^8.0.0",
+    "@nestjs/schedule": "^1.1.0",
+    "@prisma/client": "^3.14.0",
     "apollo-server-express": "^3.6.7",
     "apollo-type-bigint": "^0.1.3",
     "axios": "^0.26.1",
     "date-fns": "^2.28.0",
     "graphql": "^16.3.0",
     "lodash": "^4.17.21",
+    "pg": "^8.7.3",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
     "rxjs": "^7.2.0"
@@ -50,6 +53,7 @@
     "@vercel/ncc": "^0.33.4",
     "copyfiles": "^2.4.1",
     "jest": "^27.2.5",
+    "prisma": "^3.14.0",
     "source-map-support": "^0.5.20",
     "supertest": "^6.1.3",
     "ts-jest": "^27.0.3",

apollo/prisma/migrations/20220512130618_init/migration.sql

@@ -0,0 +1,21 @@
+-- CreateTable
+CREATE TABLE "HistoryRecord" (
+    "id" TEXT NOT NULL,
+    "fromChain" TEXT NOT NULL,
+    "toChain" TEXT NOT NULL,
+    "bridge" TEXT NOT NULL,
+    "laneId" TEXT NOT NULL,
+    "nonce" TEXT NOT NULL,
+    "requestTxHash" TEXT NOT NULL,
+    "responseTxHash" TEXT,
+    "sender" TEXT NOT NULL,
+    "recipient" TEXT NOT NULL,
+    "token" TEXT NOT NULL,
+    "amount" TEXT NOT NULL,
+    "startTime" INTEGER NOT NULL,
+    "endTime" INTEGER,
+    "result" INTEGER NOT NULL,
+    "fee" TEXT NOT NULL,
+
+    CONSTRAINT "HistoryRecord_pkey" PRIMARY KEY ("id")
+);

apollo/prisma/migrations/migration_lock.toml

@@ -0,0 +1,3 @@
+# Please do not edit this file manually
+# It should be added in your version-control system (i.e. Git)
+provider = "postgresql"

apollo/prisma/schema.prisma

@@ -0,0 +1,30 @@
+// This is your Prisma schema file,
+// learn more about it in the docs: https://pris.ly/d/prisma-schema
+
+generator client {
+  provider = "prisma-client-js"
+}
+
+datasource db {
+  provider = "postgresql"
+  url      = env("DATABASE_URL")
+}
+
+model HistoryRecord {
+  id String @id
+  fromChain String
+  toChain String
+  bridge String
+  laneId String
+  nonce String
+  requestTxHash String
+  responseTxHash String?
+  sender String
+  recipient String
+  token String
+  amount String
+  startTime Int
+  endTime Int?
+  result Int
+  fee String
+}

apollo/src/aggregation/aggregation.history.graphql

@@ -0,0 +1,25 @@
+
+type HistoryRecord {
+  id: String!
+  fromChain: String!
+  toChain: String!
+  bridge: String!
+  laneId: String!
+  nonce: String!
+  requestTxHash: String!
+  responseTxHash: String
+  sender: String!
+  recipient: String!
+  token: String!
+  amount: String!
+  startTime: Int!
+  endTime: Int
+  result: Int!
+  fee: String!
+}
+
+type Query {
+  historyRecordById(id: String): HistoryRecord
+  historyRecords(sender: String, recipient: String, row: Int, page: Int): [HistoryRecord]
+}
+

apollo/src/aggregation/aggregation.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { AggregationService } from './aggregation.service';
+import { AggregationResolver } from './aggregation.resolver';
+
+@Module({
+  providers: [AggregationService, AggregationResolver],
+  exports: [AggregationService]
+})
+export class AggregationModule {}

apollo/src/aggregation/aggregation.resolver.ts

@@ -0,0 +1,42 @@
+import { Args, Query, Resolver } from '@nestjs/graphql';
+import { AggregationService } from './aggregation.service';
+import { HistoryRecord, Prisma, PrismaClient } from '@prisma/client';
+
+@Resolver()
+export class AggregationResolver {
+  constructor(private aggregationService: AggregationService) {}
+
+  @Query()
+  async historyRecordById(
+    @Args('id') id: string
+  ) {
+    return this.aggregationService.queryHistoryRecordById({
+      id: id,
+    });
+  }
+
+  @Query()
+  async historyRecords(
+    @Args('sender') sender: string,
+    @Args('recipient') recipient: string,
+    @Args('row') row: number,
+    @Args('page') page: number
+  ) {
+    let skip = row * page || undefined;
+    let take = row || undefined;
+    let filter = new Array();
+    if (sender) {
+      filter.push({ sender: sender });
+    }
+    if (recipient) {
+      filter.push({ recipient: recipient });
+    }
+
+    let where = (sender || recipient) ? { OR: filter } : undefined;
+    return this.aggregationService.queryHistoryRecords({
+      skip,
+      take,
+      where,
+    });
+  }
+}

apollo/src/aggregation/aggregation.service.ts

@@ -0,0 +1,64 @@
+import { INestApplication, Injectable, OnModuleInit } from '@nestjs/common';
+import { HistoryRecord, Prisma, PrismaClient } from '@prisma/client';
+
+@Injectable()
+export class AggregationService extends PrismaClient implements OnModuleInit {
+  async onModuleInit() {
+    await this.$connect();
+  }
+
+  async enableShutdownHooks(app: INestApplication) {
+    this.$on('beforeExit', async () => {
+      await app.close();
+    });
+  }
+
+  async createHistoryRecord(data: Prisma.HistoryRecordCreateInput): Promise<HistoryRecord> {
+      return this.historyRecord.create({
+          data,
+      });
+  }
+
+  async updateHistoryRecord(params: {
+    where: Prisma.HistoryRecordWhereUniqueInput,
+    data: Prisma.HistoryRecordUpdateInput,
+  }): Promise<HistoryRecord> {
+    const { where, data } = params;
+    return this.historyRecord.update({
+      data,
+      where,
+    });
+  }
+
+  async queryHistoryRecordById(
+      historyRecordWhereUniqueInput: Prisma.HistoryRecordWhereUniqueInput,
+  ): Promise<HistoryRecord | null> {
+    return this.historyRecord.findUnique({
+        where: historyRecordWhereUniqueInput,
+    });
+  }
+
+  async queryHistoryRecordFirst(
+    historyRecordWhereInput: Prisma.HistoryRecordWhereInput,
+  ): Promise<HistoryRecord | null> {
+    return this.historyRecord.findFirst({
+      where: historyRecordWhereInput,
+      orderBy: { startTime: 'desc' }
+    });
+  }
+
+  // const count = this.historyRecord.count();
+  async queryHistoryRecords(params: {
+    skip?: number;
+    take?: number;
+    where?: Prisma.HistoryRecordWhereInput;
+  }): Promise<HistoryRecord[]> {
+    const { skip, take, where } = params;
+    return this.historyRecord.findMany({
+      skip,
+      take,
+      where,
+      orderBy: {startTime: 'desc'}
+    });
+  }
+}

apollo/src/app.module.ts

@@ -8,6 +8,11 @@ import { AccountModule } from './account/account.module';
 import { AppController } from './app.controller';
 import { AppService } from './app.service';
 import { Substrate2substrateModule } from './substrate2substrate/substrate2substrate.module';
+import { TasksModule } from './tasks/tasks.module';
+import { ScheduleModule } from '@nestjs/schedule';
+import { AggregationModule } from './aggregation/aggregation.module';
+import { Darwinia2crabModule } from './darwinia2crab/darwinia2crab.module';
+import { Crab2smartModule } from './crab2smart/crab2smart.module';
 
 @Scalar('BigInt')
 export class BigIntScalar extends BigInt {}
@@ -16,7 +21,7 @@ export class BigIntScalar extends BigInt {}
   imports: [
     GraphQLModule.forRoot<ApolloDriverConfig>({
       driver: ApolloDriver,
-      typePaths: ['./**/*.graphql'],
+      typePaths: ['./src/**/*.graphql'],
       definitions: {
         path: join(process.cwd(), 'src/graphql.ts'),
         outputAs: 'class',
@@ -25,6 +30,11 @@ export class BigIntScalar extends BigInt {}
     Substrate2substrateModule,
     AccountModule,
     ConfigModule.forRoot(),
+    ScheduleModule.forRoot(),
+    TasksModule,
+    AggregationModule,
+    Darwinia2crabModule,
+    Crab2smartModule,
   ],
   controllers: [AppController],
   providers: [AppService, BigIntScalar],

apollo/src/crab2smart/crab2smart.module.ts

@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
+import { Crab2smartService } from './crab2smart.service';
+import { AggregationModule } from '../aggregation/aggregation.module';
+import { TasksModule } from '../tasks/tasks.module';
+
+@Module({
+  imports: [ ConfigModule, AggregationModule, TasksModule ],
+  providers: [Crab2smartService]
+})
+export class Crab2smartModule {}

apollo/src/crab2smart/crab2smart.service.ts

@@ -0,0 +1,73 @@
+import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { AggregationService } from '../aggregation/aggregation.service';
+import { TasksService } from '../tasks/tasks.service';
+import axios from 'axios';
+import { getUnixTime } from 'date-fns';
+
+@Injectable()
+export class Crab2smartService implements OnModuleInit {
+  private readonly logger = new Logger(TasksService.name);
+
+  private readonly crabUrl = this.configService.get<string>('SUBQL') + 'crab';
+  private readonly fetchDataInterval = this.configService.get<number>('FETCH_CRAB_TO_CRAB_DVM_DATA_INTERVAL');
+  private readonly fetchDataFirst = this.configService.get<number>('FETCH_CRAB_TO_CRAB_DVM_DATA_FIRST');
+
+  constructor(
+      private configService: ConfigService,
+      private aggregationService: AggregationService,
+      private taskService: TasksService
+  ) {}
+
+  async onModuleInit() {
+    const self = this;
+    this.taskService.addInterval('crab2crabdvm-fetchdata', this.fetchDataInterval, function() {
+        self.fetchRecords()
+    });
+  }
+
+  async fetchRecords() {
+    const first = this.fetchDataFirst;
+    try {
+      let firstRecord = await this.aggregationService.queryHistoryRecordFirst({
+        OR: [
+          { fromChain: 'crab', toChain: 'crab-dvm' },
+          { fromChain: 'crab-dvm', toChain: 'crab' }
+        ],
+        bridge: 'helix',
+      });
+      var latestNonce: number = firstRecord ? Number(firstRecord.nonce) : 0;
+      const res = await axios.post(this.crabUrl, {
+        query: `query { transfers (first: ${first}, orderBy: TIMESTAMP_ASC, offset: ${latestNonce}) { totalCount nodes{id, senderId, recipientId, fromChain, toChain, amount, timestamp }}}`,
+        variables: null,
+      });
+      const nodes = res.data?.data?.transfers?.nodes;
+      if (nodes) {
+        for (let node of nodes) {
+          latestNonce = latestNonce + 1;
+          await this.aggregationService.createHistoryRecord({
+            id: 'crab2crabdvm-' + node.id,
+            fromChain: node.fromChain,
+            toChain: node.toChain,
+            bridge: 'helix',
+            laneId: "0",
+            nonce: latestNonce.toString(),
+            requestTxHash: node.id,
+            responseTxHash: node.id,
+            sender: node.senderId,
+            recipient: node.recipientId,
+            token: "Crab",
+            amount: node.amount,
+            startTime: getUnixTime(new Date(node.timestamp)),
+            endTime: getUnixTime(new Date(node.timestamp)),
+            result: 1,
+            fee: "0",
+          });
+        }
+      }
+      this.logger.log(`save new Darwinia to Crab lock records success, latestNonce: ${latestNonce}, added: ${nodes.length}`);
+    } catch(e) {
+      this.logger.warn(`update Crab to Crab DVM records failed ${e}`);
+    }
+  }
+}

apollo/src/darwinia2crab/darwinia2crab.graphql

@@ -0,0 +1,26 @@
+scalar BigInt
+scalar ID
+scalar JSON
+
+type S2sEvent {
+  id: ID!
+  laneId: String!
+  nonce: String!
+  requestTxHash: String! # TokenLocked tx hash
+  responseTxHash: String # TokenLockedConfirmed tx hash
+  senderId: String!
+  result: Int! # 0 TokenLocked 1 TokenLockedConfirmed success 2 TokenLockedConfirmed fail
+  recipient: String!
+  token: String! # token address
+  amount: String!
+  startTimestamp: String!
+  endTimestamp: String
+  fee: String!
+}
+
+type DailyStatistic {
+  id: ID!
+  dailyVolume: BigInt
+  dailyCount: Int
+}
+