[pice-pusher] add injective pusher (#633)

* add injective pusher

* remove cwPriceServiceConnection

Author: Dev Kalra

price_pusher/package-lock.json

@@ -44,7 +44,7 @@
   },
   "dependencies": {
     "@injectivelabs/sdk-ts": "^1.0.457",
-    "@pythnetwork/pyth-common-js": "^1.2.0",
+    "@pythnetwork/pyth-common-js": "^1.4.0",
     "@pythnetwork/pyth-evm-js": "^1.1.0",
     "@pythnetwork/pyth-sdk-solidity": "^2.2.0",
     "@truffle/hdwallet-provider": "^2.1.3",

price_pusher/package.json

@@ -103,11 +103,18 @@ async function injectiveRun() {
     { pollingFrequency: argv.pollingFrequency }
   );
 
+  const injectivePricePusher = new InjectivePricePusher(
+    connection,
+    argv.pythContract,
+    argv.endpoint,
+    fs.readFileSync(argv.mnemonicFile, "utf-8").trim()
+  );
+
   const handler = new Controller(
     priceConfigs,
     pythPriceListener,
     injectivePriceListener,
-    new InjectivePricePusher(),
+    injectivePricePusher,
     {
       cooldownDuration: argv.cooldownDuration,
     }

price_pusher/src/index.ts

@@ -1,8 +1,18 @@
-import { HexString } from "@pythnetwork/pyth-common-js";
+import { HexString, PriceServiceConnection } from "@pythnetwork/pyth-common-js";
 import { ChainPricePusher, PriceInfo, PriceListener } from "./interface";
 import { DurationInSeconds } from "./utils";
 import { PriceConfig } from "./price-config";
-import { ChainGrpcWasmApi } from "@injectivelabs/sdk-ts";
+import {
+  ChainGrpcAuthApi,
+  ChainGrpcWasmApi,
+  DEFAULT_STD_FEE,
+  MsgExecuteContract,
+  Msgs,
+  PrivateKey,
+  TxGrpcClient,
+  TxResponse,
+  createTransactionFromMsg,
+} from "@injectivelabs/sdk-ts";
 
 type PriceQueryResponse = {
   price_feed: {
@@ -16,6 +26,11 @@ type PriceQueryResponse = {
   };
 };
 
+type UpdateFeeResponse = {
+  denom: string;
+  amount: string;
+};
+
 // this use price without leading 0x
 // FIXME: implement common methods in the parent class
 export class InjectivePriceListener implements PriceListener {
@@ -103,10 +118,118 @@ export class InjectivePriceListener implements PriceListener {
 }
 
 export class InjectivePricePusher implements ChainPricePusher {
+  private wallet: PrivateKey;
+  constructor(
+    private priceServiceConnection: PriceServiceConnection,
+    private pythContract: string,
+    private grpcEndpoint: string,
+    mnemonic: string
+  ) {
+    this.wallet = PrivateKey.fromMnemonic(mnemonic);
+  }
+
+  private injectiveAddress(): string {
+    return this.wallet.toBech32();
+  }
+
+  private async signAndBroadcastMsg(
+    msg: Msgs,
+    fee = DEFAULT_STD_FEE
+  ): Promise<TxResponse> {
+    const chainGrpcAuthApi = new ChainGrpcAuthApi(this.grpcEndpoint);
+    const account = await chainGrpcAuthApi.fetchAccount(
+      this.injectiveAddress()
+    );
+    const { signBytes, txRaw } = createTransactionFromMsg({
+      sequence: account.baseAccount.sequence,
+      accountNumber: account.baseAccount.accountNumber,
+      message: msg,
+      chainId: "injective-888",
+      fee,
+      pubKey: this.wallet.toPublicKey().toBase64(),
+    });
+
+    const sig = await this.wallet.sign(Buffer.from(signBytes));
+
+    /** Append Signatures */
+    txRaw.setSignaturesList([sig]);
+
+    const txService = new TxGrpcClient(this.grpcEndpoint);
+    const txResponse = await txService.broadcast(txRaw);
+
+    return txResponse;
+  }
+
+  async getPriceFeedUpdateObject(priceIds: string[]): Promise<any> {
+    const vaas = await this.priceServiceConnection.getLatestVaas(priceIds);
+
+    return {
+      update_price_feeds: {
+        data: vaas,
+      },
+    };
+  }
+
   async updatePriceFeed(
     priceIds: string[],
     pubTimesToPush: number[]
   ): Promise<void> {
-    console.log("dummy pushed");
+    if (priceIds.length === 0) {
+      return;
+    }
+
+    if (priceIds.length !== pubTimesToPush.length)
+      throw new Error("Invalid arguments");
+
+    let priceFeedUpdateObject;
+    try {
+      // get the latest VAAs for updatePriceFeed and then push them
+      priceFeedUpdateObject = await this.getPriceFeedUpdateObject(priceIds);
+    } catch (e) {
+      console.error("Error fetching the latest vaas to push");
+      console.error(e);
+      return;
+    }
+
+    let updateFeeQueryResponse: UpdateFeeResponse;
+    try {
+      const api = new ChainGrpcWasmApi(this.grpcEndpoint);
+      const { data } = await api.fetchSmartContractState(
+        this.pythContract,
+        Buffer.from(
+          JSON.stringify({
+            get_update_fee: {
+              vaas: priceFeedUpdateObject.update_price_feeds.data,
+            },
+          })
+        ).toString("base64")
+      );
+
+      const json = Buffer.from(data as string, "base64").toString();
+      updateFeeQueryResponse = JSON.parse(json);
+    } catch (e) {
+      console.error("Error fetching update fee");
+      console.error(e);
+      return;
+    }
+
+    // TODO: add specific error messages
+    try {
+      const executeMsg = MsgExecuteContract.fromJSON({
+        sender: this.injectiveAddress(),
+        contractAddress: this.pythContract,
+        msg: priceFeedUpdateObject,
+        funds: [updateFeeQueryResponse],
+      });
+
+      const rs = await this.signAndBroadcastMsg(executeMsg);
+
+      if (rs.code !== 0) throw new Error("Error: transaction failed");
+
+      console.log("Succesfully broadcasted txHash:", rs.txHash);
+    } catch (e) {
+      console.error("Error executing messages");
+      console.log(e);
+    }
   }
 }