initial commit

Author: aditya520

apps/price_pusher/package.json

@@ -61,7 +61,7 @@
     "@injectivelabs/networks": "^1.14.6",
     "@injectivelabs/sdk-ts": "1.10.72",
     "@mysten/sui": "^1.3.0",
-    "@pythnetwork/price-service-client": "workspace:*",
+    "@pythnetwork/hermes-client": "^1.3.1",
     "@pythnetwork/price-service-sdk": "workspace:^",
     "@pythnetwork/pyth-fuel-js": "workspace:*",
     "@pythnetwork/pyth-sdk-solidity": "workspace:*",

apps/price_pusher/src/controller.ts

@@ -1,4 +1,4 @@
-import { UnixTimestamp } from "@pythnetwork/price-service-client";
+import { UnixTimestamp } from "@pythnetwork/hermes-client";
 import { DurationInSeconds, sleep } from "./utils";
 import { IPriceListener, IPricePusher } from "./interface";
 import { PriceConfig, shouldUpdate, UpdateCondition } from "./price-config";

apps/price_pusher/src/interface.ts

@@ -1,4 +1,4 @@
-import { HexString, UnixTimestamp } from "@pythnetwork/price-service-client";
+import { HexString, UnixTimestamp } from "@pythnetwork/hermes-client";
 import { DurationInSeconds } from "./utils";
 
 export type PriceItem = {

apps/price_pusher/src/price-config.ts

@@ -1,4 +1,4 @@
-import { HexString } from "@pythnetwork/price-service-client";
+import { HexString } from "@pythnetwork/hermes-client";
 import Joi from "joi";
 import YAML from "yaml";
 import fs from "fs";

apps/price_pusher/src/pyth-price-listener.ts

@@ -1,15 +1,16 @@
 import {
   HexString,
-  PriceFeed,
-  PriceServiceConnection,
-} from "@pythnetwork/price-service-client";
+  HermesClient,
+  PriceUpdate,
+} from "@pythnetwork/hermes-client";
 import { PriceInfo, IPriceListener, PriceItem } from "./interface";
 import { Logger } from "pino";
+import { PriceFeed } from "@pythnetwork/price-service-sdk";
 
 type TimestampInMs = number & { readonly _: unique symbol };
 
 export class PythPriceListener implements IPriceListener {
-  private connection: PriceServiceConnection;
+  private connection: HermesClient;
   private priceIds: HexString[];
   private priceIdToAlias: Map<HexString, string>;
   private latestPriceInfo: Map<HexString, PriceInfo>;
@@ -18,7 +19,7 @@ export class PythPriceListener implements IPriceListener {
   private healthCheckInterval?: NodeJS.Timeout;
 
   constructor(
-    connection: PriceServiceConnection,
+    connection: HermesClient,
     priceItems: PriceItem[],
     logger: Logger
   ) {
@@ -34,107 +35,29 @@ export class PythPriceListener implements IPriceListener {
   // This method should be awaited on and once it finishes it has the latest value
   // for the given price feeds (if they exist).
   async start() {
-    // Set custom error handler for websocket errors
-    this.connection.onWsError = (error: Error) => {
-      if (error.message.includes("not found")) {
-        // Extract invalid feed IDs from error message
-        const match = error.message.match(/\[(.*?)\]/);
-        if (match) {
-          const invalidFeedIds = match[1].split(",").map((id) => {
-            // Remove '0x' prefix if present to match our stored IDs
-            return id.trim().replace(/^0x/, "");
-          });
-
-          // Log invalid feeds with their aliases
-          invalidFeedIds.forEach((id) => {
-            this.logger.error(
-              `Price feed ${id} (${this.priceIdToAlias.get(
-                id
-              )}) not found for subscribePriceFeedUpdates`
-            );
-          });
-
-          // Filter out invalid feeds and resubscribe with valid ones
-          const validFeeds = this.priceIds.filter(
-            (id) => !invalidFeedIds.includes(id)
-          );
-
-          this.priceIds = validFeeds;
-
-          if (validFeeds.length > 0) {
-            this.logger.info("Resubscribing with valid feeds only");
-            this.connection.subscribePriceFeedUpdates(
-              validFeeds,
-              this.onNewPriceFeed.bind(this)
-            );
-          }
-        }
-      } else {
-        this.logger.error("Websocket error occurred:", error);
-      }
-    };
-
-    this.connection.subscribePriceFeedUpdates(
+    this.connection.getPriceUpdatesStream(
       this.priceIds,
-      this.onNewPriceFeed.bind(this)
     );
 
     try {
-      const priceFeeds = await this.connection.getLatestPriceFeeds(
-        this.priceIds
+      const priceUpdates = await this.connection.getLatestPriceUpdates(
+        this.priceIds,
+        {
+          encoding: "hex",
+          parsed: true,
+          ignoreInvalidPriceIds: true,
+        }
       );
-      priceFeeds?.forEach((priceFeed) => {
-        const latestAvailablePrice = priceFeed.getPriceUnchecked();
-        this.latestPriceInfo.set(priceFeed.id, {
-          price: latestAvailablePrice.price,
-          conf: latestAvailablePrice.conf,
-          publishTime: latestAvailablePrice.publishTime,
+      priceUpdates.parsed?.forEach((priceUpdate) => {
+        this.latestPriceInfo.set(priceUpdate.id, {
+          price: priceUpdate.price.price,
+          conf: priceUpdate.price.conf,
+          publishTime: priceUpdate.price.publish_time,
         });
       });
     } catch (error: any) {
       // Always log the HTTP error first
       this.logger.error("Failed to get latest price feeds:", error);
-
-      if (error.response.data.includes("Price ids not found:")) {
-        // Extract invalid feed IDs from error message
-        const invalidFeedIds = error.response.data
-          .split("Price ids not found:")[1]
-          .split(",")
-          .map((id: string) => id.trim().replace(/^0x/, ""));
-
-        // Log invalid feeds with their aliases
-        invalidFeedIds.forEach((id: string) => {
-          this.logger.error(
-            `Price feed ${id} (${this.priceIdToAlias.get(
-              id
-            )}) not found for getLatestPriceFeeds`
-          );
-        });
-
-        // Filter out invalid feeds and retry
-        const validFeeds = this.priceIds.filter(
-          (id) => !invalidFeedIds.includes(id)
-        );
-
-        this.priceIds = validFeeds;
-
-        if (validFeeds.length > 0) {
-          this.logger.info(
-            "Retrying getLatestPriceFeeds with valid feeds only"
-          );
-          const validPriceFeeds = await this.connection.getLatestPriceFeeds(
-            validFeeds
-          );
-          validPriceFeeds?.forEach((priceFeed) => {
-            const latestAvailablePrice = priceFeed.getPriceUnchecked();
-            this.latestPriceInfo.set(priceFeed.id, {
-              price: latestAvailablePrice.price,
-              conf: latestAvailablePrice.conf,
-              publishTime: latestAvailablePrice.publishTime,
-            });
-          });
-        }
-      }
     }
 
     // Store health check interval reference
@@ -148,29 +71,6 @@ export class PythPriceListener implements IPriceListener {
     }, 5000);
   }
 
-  private onNewPriceFeed(priceFeed: PriceFeed) {
-    this.logger.debug(
-      `Received new price feed update from Pyth price service: ${this.priceIdToAlias.get(
-        priceFeed.id
-      )} ${priceFeed.id}`
-    );
-
-    // Consider price to be currently available if it is not older than 60s
-    const currentPrice = priceFeed.getPriceNoOlderThan(60);
-    if (currentPrice === undefined) {
-      return;
-    }
-
-    const priceInfo: PriceInfo = {
-      conf: currentPrice.conf,
-      price: currentPrice.price,
-      publishTime: currentPrice.publishTime,
-    };
-
-    this.latestPriceInfo.set(priceFeed.id, priceInfo);
-    this.lastUpdated = Date.now() as TimestampInMs;
-  }
-
   getLatestPriceInfo(priceId: string): PriceInfo | undefined {
     return this.latestPriceInfo.get(priceId);
   }