feat: adds support for price SSE stream

Author: stalniy

README.md

@@ -175,6 +175,7 @@ akash query wasm contract-state smart $CONTRACT_ADDRESS '{"get_config":{}}'
 | `WALLET_SECRET` | Yes | - | Either `privateKey:<private key in hex format>` or `mnemonic:<12/24 words>` |
 | `HERMES_ENDPOINT` | No | `https://hermes.pyth.network` | Pyth Hermes API |
 | `PRICE_DEVIATION_TOLERANCE` | No | 0 | absolute or percentage value for price deviations which should be ignored (e.g., `100` or `10%`) |
+| `PRICE_FETCHING_METHOD` | No | polling | `polling` or `sse` |
 | `UPDATE_INTERVAL_MS` | No | `5000` | Update interval (default 5 sec) |
 | `GAS_PRICE` | No | `0.025uakt` | Gas price |
 | `DENOM` | No | `uakt` | Token denomination |

package-lock.json

@@ -44,6 +44,7 @@
         "@opentelemetry/resources": "^2.5.0",
         "@opentelemetry/sdk-node": "^0.211.0",
         "commander": "^12.0.0",
+        "fetch-event-stream": "^0.1.6",
         "zod": "^4.3.6"
     },
     "devDependencies": {

package.json

@@ -2,7 +2,8 @@ import { z } from "zod";
 import { HermesClient, type HermesConfig } from "../hermes-client.ts";
 import { validateContractAddress, validateWalletSecret } from "../validation.ts";
 import type { PriceProducerFactoryOptions } from "../types.ts";
-import { pollPriceStream } from "../price-stream/polling-price-stream.ts";
+import { pollPriceStream } from "../price-stream/polling-price-stream/polling-price-stream.ts";
+import { priceSSEStream } from "../price-stream/price-sse-stream/price-sse-stream.ts";
 
 export interface CommandConfig extends HermesConfig {
     createHermesClient: (config: HermesConfig) => Promise<HermesClient>;
@@ -41,6 +42,7 @@ const configSchema = z.object({
             });
         }
     }).optional(),
+    PRICE_FETCHING_METHOD: z.enum(["polling", "sse"]).default("polling"),
     UPDATE_INTERVAL_MS: z.coerce.number().int().nonnegative().default(5 * 1000), // Default to 5 seconds
     HEALTHCHECK_PORT: z.coerce.number().int().min(1).max(65535).default(3000),
     GAS_PRICE: z.string().regex(/^(\d+)(\.\d+)?uakt$/, { message: 'GAS_PRICE must be a valid number with unit (e.g., "0.025uakt")' }).default("0.025uakt"),
@@ -71,6 +73,13 @@ export function parseConfig(config: Record<string, string | undefined>): ParseCo
         priceDeviationTolerance: result.data.PRICE_DEVIATION_TOLERANCE,
         smartContractConfigCacheTTLMs: result.data.SMART_CONTRACT_CONFIG_CACHE_TTL_MS,
         priceProducerFactory(options: PriceProducerFactoryOptions) {
+            if (result.data.PRICE_FETCHING_METHOD === "sse") {
+                return priceSSEStream({
+                    ...options,
+                    unsafeAllowInsecureEndpoints,
+                    baseUrl: result.data.HERMES_ENDPOINT,
+                });
+            }
             return pollPriceStream({
                 ...options,
                 unsafeAllowInsecureEndpoints,

src/cli-commands/command-config.ts

@@ -528,6 +528,17 @@ export class HermesClient {
             const consumePrices = async () => {
                 for await (const priceUpdate of priceStream) {
                     priceUpdates.set(priceUpdate);
+
+                    const price = priceUpdate.priceData.price;
+                    this.#logger?.log(
+                        `Received price from Hermes: ${price.price} (expo: ${price.expo})`,
+                    );
+                    this.#logger?.log(
+                        `  Confidence: ${price.conf}, Publish time: ${price.publish_time}`,
+                    );
+                    this.#logger?.log(
+                        `  VAA size: ${priceUpdate.vaa.length} bytes (base64)`,
+                    );
                 }
                 controller.abort();
             };

src/hermes-client.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it, vi } from "vitest";
-import type { HermesResponse } from "../types.ts";
+import type { HermesResponse } from "../../types.ts";
 import { pollPriceStream, type PollPriceStreamOptions } from "./polling-price-stream.ts";
 
 describe("pollPriceStream", () => {
@@ -106,22 +106,6 @@ describe("pollPriceStream", () => {
         expect(result.value).toEqual({ priceData: goodData.parsed[0], vaa: goodData.binary.data[0] });
     });
 
-    it("logs price details on successful fetch", async () => {
-        const logger = { log: vi.fn(), error: vi.fn(), warn: vi.fn() };
-        const data = createHermesResponse();
-        const options = createOptions({
-            fetch: vi.fn().mockResolvedValueOnce(mockFetchResponse(data)),
-            logger,
-        });
-
-        const gen = pollPriceStream(options);
-        await gen.next();
-
-        expect(logger.log).toHaveBeenCalledWith(expect.stringContaining("Fetched price from Hermes: 1000"));
-        expect(logger.log).toHaveBeenCalledWith(expect.stringContaining("Confidence: 10"));
-        expect(logger.log).toHaveBeenCalledWith(expect.stringContaining("VAA size:"));
-    });
-
     it("polls repeatedly yielding updates", async () => {
         const data1 = createHermesResponse();
         const data2 = createHermesResponse({

…rice-stream/polling-price-stream.test.ts …rice-stream/polling-price-stream.test.tssrc/price-stream/polling-price-stream.test.ts renamed to src/price-stream/polling-price-stream/polling-price-stream.test.ts src/price-stream/polling-price-stream.test.ts renamed to src/price-stream/polling-price-stream/polling-price-stream.test.ts

@@ -3,9 +3,10 @@ import https from "node:https";
 import { performance } from "node:perf_hooks";
 import { Readable } from "node:stream";
 import { setTimeout as delay } from "node:timers/promises";
-import { hermesFetchDuration } from "../metrics.ts";
-import type { HermesResponse, PriceProducerFactoryOptions, PriceUpdate, PythPriceData } from "../types.ts";
-import { validateEndpointUrl } from "../validation.ts";
+import { hermesFetchDuration } from "../../metrics.ts";
+import type { HermesResponse, PriceProducerFactoryOptions, PriceUpdate } from "../../types.ts";
+import { validateEndpointUrl } from "../../validation.ts";
+import { parsePriceUpdate } from "../utils.ts";
 
 export async function *pollPriceStream(options: PollPriceStreamOptions): AsyncGenerator<PriceUpdate> {
     if (!options.priceFeedId) {
@@ -37,7 +38,6 @@ export async function *pollPriceStream(options: PollPriceStreamOptions): AsyncGe
             continue;
         } finally {
             hermesFetchDuration.record(performance.now() - fetchStart, { status });
-            console.log(`Fetch from Hermes completed with status ${status} in ${performance.now() - fetchStart} ms`);
         }
 
         if (!response.ok) {
@@ -48,32 +48,22 @@ export async function *pollPriceStream(options: PollPriceStreamOptions): AsyncGe
             continue;
         }
 
-        const data = await response.json() as HermesResponse;
-
-        if (!data.parsed || data.parsed.length === 0) {
-            options.logger?.error("No price data returned from Hermes");
+        let parsedData: HermesResponse;
+        try {
+            parsedData = await response.json() as HermesResponse;
+        } catch (error) {
+            options.logger?.error(`Error parsing JSON from Hermes: ${(error as Error).message}`);
             continue;
         }
 
-        if (!data.binary?.data || data.binary.data.length === 0) {
-            options.logger?.error("No VAA binary data returned from Hermes");
+        const priceUpdateResult = parsePriceUpdate(parsedData);
+
+        if (!priceUpdateResult.ok) {
+            options.logger?.error(priceUpdateResult.message);
             continue;
         }
 
-        const priceData: PythPriceData = data.parsed[0];
-        const vaa: string = data.binary.data[0];
-
-        options.logger?.log(
-            `Fetched price from Hermes: ${priceData.price.price} (expo: ${priceData.price.expo})`,
-        );
-        options.logger?.log(
-            `  Confidence: ${priceData.price.conf}, Publish time: ${priceData.price.publish_time}`,
-        );
-        options.logger?.log(
-            `  VAA size: ${vaa.length} bytes (base64)`,
-        );
-
-        yield { priceData, vaa };
+        yield priceUpdateResult.value;
         if (options.pollingIntervalMs > 0) {
             await delay(options.pollingIntervalMs, undefined, { signal: options.signal })
                 .catch((error) => options.logger?.warn(`Polling delay interrupted: ${(error as Error).message}`));