move zero into processor and add redis to the existing docker image

Dockerfile

@@ -11,6 +11,9 @@ COPY package*.json ./
 RUN npm install
 RUN npm run build
 
+RUN apk --update add redis
+RUN npm install -g concurrently
+
 EXPOSE 3000
 
-CMD [ "npm", "run", "start" ]
+CMD concurrently "/usr/bin/redis-server --bind '0.0.0.0'" "sleep 5s; npm run start"

src/config.ts

@@ -21,9 +21,9 @@ export interface NotificationConfig {
 
 export interface XdcZeroConfig {
   isEnabled: boolean;
-  subnetZeroContractAddress: string;
-  parentChainZeroContractAddress: string;
-  walletPk: string;
+  subnetZeroContractAddress?: string;
+  parentChainZeroContractAddress?: string;
+  walletPk?: string;
 }
 
 export interface Config {
@@ -32,13 +32,14 @@ export interface Config {
   cronJob: {
     liteJobExpression: string;
     jobExpression: string;
+    zeroJobExpression: string;
   };
   subnet: SubnetConfig;
   mainnet: MainnetConfig;
   reBootstrapWaitingTime: number;
   notification: NotificationConfig;
   chunkSize: number;
-  xdcZero: XdcZeroConfig | undefined;
+  xdcZero: XdcZeroConfig;
   relayerCsc: {
     isEnabled: boolean;
   }
@@ -53,8 +54,8 @@ const getZeroConfig = (): XdcZeroConfig => {
     isEnabled,
     subnetZeroContractAddress: process.env.SUBNET_ZERO_CONTRACT,
     parentChainZeroContractAddress: process.env.PARENTNET_ZERO_CONTRACT,
-    walletPk: process.env.PARENTNET_ZERO_WALLET_PK
-  }: undefined;
+    walletPk: process.env.PARENTNET_ZERO_WALLET_PK.startsWith("0x") ? process.env.PARENTNET_ZERO_WALLET_PK : `0x${process.env.PARENTNET_ZERO_WALLET_PK}`
+  }: { isEnabled: false };
 };
 
 const config: Config = {
@@ -63,6 +64,7 @@ const config: Config = {
   cronJob: {
     liteJobExpression: "0 */2 * * * *", // every 2min
     jobExpression: "*/20 * * * * *", // every 20s
+    zeroJobExpression: "*/10 * * * * *", // every 10s
   },
   subnet: {
     url: process.env.SUBNET_URL || "https://devnetstats.apothem.network/subnet",

src/processors/README.md

@@ -0,0 +1,7 @@
+# How to add new processors?
+
+1. Read our `lite.ts`(simple version) or the `standard.ts`(more complex version) as examples
+2. Assume you plan to add a new processor called `XXX`. First create the file `XXX.ts` under current directory.
+3. Add `export class XXX implements ProcessorInterface` where all our processors has some common methods such as `init` and `reset`. Implement those methods.
+4. Go to `index.ts` in this directory, register your processors with `enum Mode`, `private processors` (class property), `reset` method and add your custom start up condition in `getRunningModes` method
+5. Done

src/processors/index.ts

@@ -1,3 +1,4 @@
+import { Zero } from "./zero";
 import { config } from "./../config";
 import bunyan from "bunyan";
 import * as _ from "lodash";
@@ -17,14 +18,16 @@ export class Processors implements ProcessorInterface {
   private processors: {
     lite: Lite;
     standard: Standard;
+    zero: Zero;
   }
   private mainnetService: MainnetService;
 
   constructor(logger: bunyan) {
     this.logger = logger;
     this.processors = {
       lite: new Lite(logger),
-      standard: new Standard(logger)
+      standard: new Standard(logger),
+      zero: new Zero(logger)
       // Register more processors here
     };
     this.mainnetService = new MainnetService(config.mainnet, logger);
@@ -50,6 +53,9 @@ export class Processors implements ProcessorInterface {
         case Mode.STANDARD:
           await this.processors.standard.reset();
           break;
+        case Mode.ZERO:
+          await this.processors.zero.reset();
+          break;
         default:
           throw new Error("No avaiable modes to choose from");
       }
@@ -72,7 +78,10 @@ export class Processors implements ProcessorInterface {
       }
     }
 
-    // TODO: Now check xdc-zero
+    if (config.xdcZero.isEnabled) {
+      modes.push(Mode.ZERO);
+    }
+
     this.logger.info("Running modes: ", modes);
     return modes;
   }

src/processors/lite.ts

@@ -22,6 +22,7 @@ export class Lite implements ProcessorInterface {
   }
 
   init() {
+    this.logger.info("Initialising XDC Lite relayer");
     this.queue.process(async (_, done) => {
       this.logger.info("⏰ Executing lite flow periodically");
       try {

src/processors/standard.ts

@@ -29,6 +29,7 @@ export class Standard implements ProcessorInterface {
   }
 
   init() {
+    this.logger.info("Initialising XDC relayer");
     this.queue.process(async (_, done) => {
       this.logger.info("⏰ Executing normal flow periodically");
       try {
@@ -50,20 +51,18 @@ export class Standard implements ProcessorInterface {
     try {
       // Stop and remove repeatable jobs
       await this.queue.removeRepeatable(NAME, REPEAT_JOB_OPT.repeat);
-
       // Clean timer
       this.cache.cleanCache();
       // Pull latest confirmed tx from mainnet
       const smartContractData = await this.mainnetService.getLastAudittedBlock();
       // Pull latest confirm block from subnet
       const lastestSubnetCommittedBlock =
         await this.subnetService.getLastCommittedBlockInfo();
-
       const { shouldProcess, from } = await this.shouldProcessSync(
         smartContractData,
         lastestSubnetCommittedBlock
       );
-
+      
       if (shouldProcess) {
         await this.submitTxs(
           from,

src/processors/zero.ts

@@ -0,0 +1,80 @@
+import Bull from "bull";
+import bunyan from "bunyan";
+import { ProcessorInterface } from "./type";
+import { ZeroService } from "../service/zero";
+import { config } from "../config";
+
+const NAME = "ZERO";
+
+export class Zero implements ProcessorInterface {
+  private queue: Bull.Queue;
+  private logger: bunyan;
+  private zeroService: ZeroService;
+
+  constructor(logger: bunyan) {
+    this.logger = logger;
+    this.queue = new Bull(NAME);
+    this.zeroService = new ZeroService(logger);
+  }
+  init() {
+    this.logger.info("Initialising XDC-Zero");
+    this.queue.process(async (_, done) => {
+      this.logger.info("⏰ Executing xdc-zero periodically");
+      try {
+        done(null, await this.processEvent());
+      } catch (error) {
+        this.logger.error("Fail to process xdc-zero job", {
+          message: error.message,
+        });
+        // Report the error
+        done(error);
+        await this.reset();
+      }
+    });
+    return this;
+  }
+
+  async reset(): Promise<void> {
+    await this.queue.add({}, { jobId: NAME, repeat: { cron: config.cronJob.zeroJobExpression}});
+  }
+
+  async processEvent() {
+    const payloads = await this.zeroService.getPayloads();
+    if (payloads.length) {
+      this.logger.info("Nothing to process in xdc-zero, wait for the next event log");
+      return;
+    }
+    const lastPayload = payloads[payloads.length - 1];
+    const lastIndexFromSubnet = lastPayload[0];
+
+    const lastIndexfromParentnet = await this.zeroService.getIndexFromParentnet();
+
+    const lastBlockNumber = lastPayload[7];
+    const cscBlockNumber = await this.zeroService.getLatestBlockNumberFromCsc();
+    if (cscBlockNumber < lastBlockNumber) {
+      this.logger.info(
+        "wait for csc block lastBlockNumber:" +
+          lastBlockNumber +
+          " cscBlockNumber:" +
+          cscBlockNumber
+      );
+      return;
+    }
+
+    if (lastIndexFromSubnet > lastIndexfromParentnet) {
+      for (let i = lastIndexfromParentnet; i < lastIndexFromSubnet; i++) {
+        if (payloads?.[i]?.[6]) {
+          const proof = await this.zeroService.getProof(payloads[i][6]);
+          await this.zeroService.validateTransactionProof(
+            proof.key,
+            proof.receiptProofValues,
+            proof.txProofValues,
+            proof.blockHash
+          );
+          this.logger.info("sync zero index " + i + " success");
+        }
+      }
+    }
+    this.logger.info("Completed the xdc-zero sync, wait for the next cycle");
+  }
+}

src/service/mainnet/extensions.ts

@@ -0,0 +1,21 @@
+import Web3 from "web3";
+import { NetworkInformation } from "../types";
+
+const MAINNET_EXTENSION_NAME = "xdcMainnet";
+
+export interface Web3WithExtension extends Web3 {
+  xdcMainnet: {
+    getNetworkInformation: () => Promise<NetworkInformation>
+  }
+}
+
+export const mainnetExtensions = {
+  property: MAINNET_EXTENSION_NAME,
+  methods: [
+    {
+      name: "networkInformation",
+      call: "XDPoS_networkInformation"
+    }
+  ]
+};
+

src/service/mainnet/index.ts

@@ -8,6 +8,8 @@ import { MainnetConfig } from "../../config";
 import { sleep } from "../../utils/index";
 import FullABI from "./ABI/FullABI.json";
 import LiteABI from "./ABI/LiteABI.json";
+import { Web3WithExtension, mainnetExtensions } from "./extensions";
+import { NetworkInformation } from "../types";
 
 export interface SmartContractData {
   smartContractHash: string;
@@ -19,7 +21,7 @@ export interface SmartContractData {
 const TRANSACTION_GAS_NUMBER = 12500000000;
 
 export class MainnetService {
-  private web3: Web3;
+  private web3: Web3WithExtension;
   private smartContractInstance: Contract;
   private mainnetAccount: Account;
   private mainnetConfig: MainnetConfig;
@@ -32,7 +34,7 @@ export class MainnetService {
       keepAlive: true,
       agent: { https: keepaliveAgent },
     });
-    this.web3 = new Web3(provider);
+    this.web3 = new Web3(provider).extend(mainnetExtensions);
     this.smartContractInstance = new this.web3.eth.Contract(
       FullABI as AbiItem[],
       config.smartContractAddress
@@ -42,6 +44,10 @@ export class MainnetService {
     );
     this.mainnetConfig = config;
   }
+
+  async getNetworkInformation(): Promise<NetworkInformation> {
+    return this.web3.xdcMainnet.getNetworkInformation();
+  }
 
   /*
     A method to fetch the last subnet block that has been stored/audited in mainnet XDC

src/service/subnet/extensions.ts

@@ -1,4 +1,5 @@
 import Web3 from "web3";
+import { NetworkInformation } from "../types";
 
 const SUBNET_EXTENSION_NAME = "xdcSubnet";
 
@@ -18,12 +19,25 @@ export interface FetchedV2BlockInfo {
   Error: string;
 }
 
+export interface TxReceiptProof {
+  blockHash: string;
+  key: string;
+  receiptProofKeys: string[];
+  receiptProofValues: string[];
+  receiptRoot: string;
+  txProofKeys: string;
+  txProofValues: string[];
+  txRoot: string;
+}
+
 export interface Web3WithExtension extends Web3 {
   xdcSubnet: {
     getLatestCommittedBlockInfo:  () => Promise<CommittedBlockInfo>
     getV2Block: (number: string) => Promise<FetchedV2BlockInfo>
     getV2BlockByNumber: (bluckNum: string) => Promise<FetchedV2BlockInfo>
     getV2BlockByHash: (blockHash: string) => Promise<FetchedV2BlockInfo>
+    getNetworkInformation: () => Promise<NetworkInformation>
+    getTransactionAndReceiptProof: (txHash: string) => Promise<TxReceiptProof>
   }
 }
 
@@ -48,6 +62,15 @@ export const subnetExtensions = {
       name: "getV2BlockByHash",
       params: 1,
       call: "XDPoS_getV2BlockByHash"
+    }, {
+      name: "networkInformation",
+      params: 0,
+      call: "XDPoS_networkInformation"
+    },
+    {
+      name: "getTransactionAndReceiptProof",
+      params: 1,
+      call: "eth_getTransactionAndReceiptProof"
     }
   ]
 };

src/service/subnet/index.ts

@@ -4,6 +4,7 @@ import bunyan from "bunyan";
 import { SubnetConfig } from "../../config";
 import { sleep } from "../../utils/index";
 import { subnetExtensions, Web3WithExtension } from "./extensions";
+import { NetworkInformation } from "../types";
 
 export interface SubnetBlockInfo {
   subnetBlockHash: string;
@@ -14,8 +15,8 @@ export interface SubnetBlockInfo {
 }
 
 export class SubnetService {
-  private web3: Web3WithExtension;
-  private subnetConfig: SubnetConfig;
+  protected web3: Web3WithExtension;
+  protected subnetConfig: SubnetConfig;
   logger: bunyan;
 
   constructor(config: SubnetConfig, logger: bunyan) {
@@ -29,7 +30,11 @@ export class SubnetService {
     this.subnetConfig = config;
     this.web3 = new Web3(provider).extend(subnetExtensions);
   }
-
+
+  async getNetworkInformation(): Promise<NetworkInformation> {
+    return this.web3.xdcSubnet.getNetworkInformation();
+  }
+
   async getLastCommittedBlockInfo(): Promise<SubnetBlockInfo> {
     try {
       const { Hash, Number, Round, HexRLP, ParentHash } =
@@ -122,6 +127,15 @@ export class SubnetService {
       throw error;
     }
   }
+
+  async getTransactionAndReceiptProof(txHash: string) {
+    try {
+      return this.web3.xdcSubnet.getTransactionAndReceiptProof(txHash);  
+    } catch (error) {
+      this.logger.error("Error while trying to fetch the transaction receipt proof", error);
+      throw error;
+    }
+  }
 
   async bulkGetRlpHeaders(
     startingBlockNumber: number,

src/service/types.ts

@@ -0,0 +1,5 @@
+export interface NetworkInformation {
+  Denom: string;
+  NetworkId: number;
+  NetworkName: string;
+}