wip

Author: aditya520

apps/price_pusher/gg.txt

@@ -12,3 +12,8 @@
     time_difference: 30
     price_deviation: 0.5
     confidence_ratio: 0.1
+- alias: PYTH/USD
+  id: 2f95862b045670cd22bee3114c39763a4a08beeb663b145d283c31d7d1101c23
+  time_difference: 60
+  price_deviation: 0.5
+  confidence_ratio: 1

apps/price_pusher/price-config.stable.sample.yaml

@@ -164,7 +164,7 @@ export class EvmPricePusher implements IPricePusher {
 
     try {
       updateFee = await this.pythContract.read.getUpdateFee([
-        priceFeedUpdateData,
+        priceFeedUpdateData.map((data) => addLeading0x(data)),
       ]);
       updateFee = BigInt(
         Math.round(Number(updateFee) * (this.updateFeeMultiplier || 1))

apps/price_pusher/src/evm/evm.ts

@@ -1,11 +1,15 @@
-import { HexString, HermesClient } from "@pythnetwork/hermes-client";
+import {
+  HexString,
+  HermesClient,
+  PriceUpdate,
+} from "@pythnetwork/hermes-client";
 import { PriceInfo, IPriceListener, PriceItem } from "./interface";
 import { Logger } from "pino";
 
 type TimestampInMs = number & { readonly _: unique symbol };
 
 export class PythPriceListener implements IPriceListener {
-  private connection: HermesClient;
+  private hermesClient: HermesClient;
   private priceIds: HexString[];
   private priceIdToAlias: Map<HexString, string>;
   private latestPriceInfo: Map<HexString, PriceInfo>;
@@ -14,11 +18,11 @@ export class PythPriceListener implements IPriceListener {
   private healthCheckInterval?: NodeJS.Timeout;
 
   constructor(
-    connection: HermesClient,
+    hermesClient: HermesClient,
     priceItems: PriceItem[],
     logger: Logger
   ) {
-    this.connection = connection;
+    this.hermesClient = hermesClient;
     this.priceIds = priceItems.map((priceItem) => priceItem.id);
     this.priceIdToAlias = new Map(
       priceItems.map((priceItem) => [priceItem.id, priceItem.alias])
@@ -30,28 +34,101 @@ 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() {
-    this.connection.getPriceUpdatesStream(this.priceIds);
-
-    try {
-      const priceUpdates = await this.connection.getLatestPriceUpdates(
-        this.priceIds,
-        {
-          encoding: "hex",
-          parsed: true,
-          ignoreInvalidPriceIds: true,
+    const priceMetadata = await this.hermesClient.getPriceFeeds();
+    const allPriceIds = priceMetadata.map((priceMetadata) => priceMetadata.id);
+
+    // Filter out invalid price ids
+    const { existingPriceIds, invalidPriceIds } = this.priceIds.reduce<{
+      existingPriceIds: string[];
+      invalidPriceIds: string[];
+    }>(
+      (acc, id) => {
+        if (allPriceIds.includes(id)) {
+          acc.existingPriceIds.push(id);
+        } else {
+          acc.invalidPriceIds.push(id);
         }
-      );
+        return acc;
+      },
+      { existingPriceIds: [], invalidPriceIds: [] }
+    );
+
+    const invalidPriceIdsWithAlias = invalidPriceIds.map((id) =>
+      this.priceIdToAlias.get(id)
+    );
+    this.logger.error(
+      `Invalid price id submitted for: ${invalidPriceIdsWithAlias.join(", ")}`
+    );
+
+    this.priceIds = existingPriceIds;
+    // TODO: We can just remove the invalid price ids from the map
+    this.priceIdToAlias = new Map(
+      existingPriceIds.map(
+        (id) => [id, this.priceIdToAlias.get(id)] as [HexString, string]
+      )
+    );
+
+    const eventSource = await this.hermesClient.getPriceUpdatesStream(
+      this.priceIds,
+      {
+        parsed: true,
+        ignoreInvalidPriceIds: true,
+      }
+    );
+    eventSource.onmessage = (event: MessageEvent<string>) => {
+      const priceUpdates = JSON.parse(event.data) as PriceUpdate;
       priceUpdates.parsed?.forEach((priceUpdate) => {
-        this.latestPriceInfo.set(priceUpdate.id, {
-          price: priceUpdate.price.price,
-          conf: priceUpdate.price.conf,
-          publishTime: priceUpdate.price.publish_time,
-        });
+        this.logger.debug(
+          `Received new price feed update from Pyth price service: ${this.priceIdToAlias.get(
+            priceUpdate.id
+          )} ${priceUpdate.id}`
+        );
+
+        // Consider price to be currently available if it is not older than 60s
+        const currentPrice =
+          Date.now() / 1000 - priceUpdate.price.publish_time > 60
+            ? undefined
+            : priceUpdate.price;
+        if (currentPrice === undefined) {
+          return;
+        }
+
+        const priceInfo: PriceInfo = {
+          conf: currentPrice.conf,
+          price: currentPrice.price,
+          publishTime: currentPrice.publish_time,
+        };
+
+        this.latestPriceInfo.set(priceUpdate.id, priceInfo);
+        this.lastUpdated = Date.now() as TimestampInMs;
       });
-    } catch (error: any) {
-      // Always log the HTTP error first
-      this.logger.error("Failed to get latest price feeds:", error);
-    }
+    };
+
+    eventSource.onerror = (error: Event) => {
+      console.error("Error receiving updates from Hermes:", error);
+      eventSource.close();
+    };
+
+    // try {
+    //   const priceUpdates = await this.hermesClient.getLatestPriceUpdates(
+    //     this.priceIds,
+    //     {
+    //       encoding: "hex",
+    //       parsed: true,
+    //       ignoreInvalidPriceIds: true,
+    //     }
+    //   );
+    //   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);
+    // }
 
     // Store health check interval reference
     this.healthCheckInterval = setInterval(() => {