diff --git a/workspace/data-proxy/src/config/pyth-lazer-module-config.ts b/workspace/data-proxy/src/config/pyth-lazer-module-config.ts index 565f0d9..3b2c390 100644 --- a/workspace/data-proxy/src/config/pyth-lazer-module-config.ts +++ b/workspace/data-proxy/src/config/pyth-lazer-module-config.ts @@ -85,7 +85,8 @@ export interface PythLazerModuleConfig export const PythLazerModuleRouteSchema = v.strictObject({ ...RouteSchema.entries, moduleName: v.string(), - fetchFromModule: v.string(), + // Present: path template. Omitted: POST body is expected instead. + fetchFromModule: v.optional(v.string()), channel: v.optional(PythLazerChannelSchema, PYTH_LAZER_DEFAULT_CHANNEL), type: v.literal("pyth-lazer"), }); diff --git a/workspace/data-proxy/src/modules/pyth-lazer/README.md b/workspace/data-proxy/src/modules/pyth-lazer/README.md index 160dc0b..3a0063d 100644 --- a/workspace/data-proxy/src/modules/pyth-lazer/README.md +++ b/workspace/data-proxy/src/modules/pyth-lazer/README.md @@ -11,7 +11,10 @@ On startup the module: 3. Caches inbound `streamUpdated` messages keyed by `(channel, priceFeedId)`. Ticks for feeds that are no longer in the desired set are ignored. 4. A forked cleanup daemon removes from the desired set feeds that have not been requested within `priceFeedsCleanupTtl`. It deletes their cache entries and updates the desired set. -HTTP requests resolve `fetchFromModule` to one or more comma-separated feed IDs or symbols, add any new feeds to the desired set, send a subscribe of the full set for that channel if it grew, and return the latest cached price for each. If a price is not yet available, the handler waits briefly for an update (shared price-cache timeout: 3 seconds). +This module supports two different types of HTTP requests: + +- **Standard Request** (`fetchFromModule` set): resolve the template to one or more comma-separated feed IDs or symbols, add any new feeds to the desired set for the route channel, send a subscribe of the full set for that channel if it grew, and return the latest cached price for each as a JSON array. If a price is not yet available, the handler waits briefly for an update (shared price-cache timeout: 3 seconds). +- **POST Body-Based Request** (`fetchFromModule` omitted): parse a JSON body (`priceFeedIds` / `priceFeedSymbols`, optional `channel`), subscribe on the selected channel (body overrides route default), and return `{ parsed: { timestampUs, priceFeeds } }`. All-or-fail if any feed misses. Pyth cannot append feeds to an existing subscription id, so a grow sends a **new** subscribe with a new id and the full desired set for that channel. Each channel has at most one **active** (acked) subscription, plus any in-flight ids still waiting for an ack. The module promotes an id only after Pyth acks it (`subscribed` or `subscribedWithInvalidFeedIdsIgnored`) and only if it is higher than the channel’s current active id; older outstanding ids on that channel are then unsubscribed. @@ -37,8 +40,8 @@ Pyth cannot append feeds to an existing subscription id, so a grow sends a **new | `moduleName` | yes | — | Name of a configured Pyth Lazer module. | | `path` | yes | — | Proxy path (supports `{:param}` path params). | | `method` | no | `GET` | HTTP method(s). | -| `fetchFromModule` | yes | — | Template producing one or more comma-separated feed IDs or symbols. | -| `channel` | no | `fixed_rate@200ms` | Channel for this route’s subscriptions and cache lookups (`real_time`, `fixed_rate@50ms`, `fixed_rate@200ms`, `fixed_rate@1000ms`). | +| `fetchFromModule` | no | — | When set: Template producing one or more comma-separated feed IDs or symbols. When omitted: `POST` body is expected instead. | +| `channel` | no | `fixed_rate@200ms` | Default channel for this route’s subscriptions and cache lookups (`real_time`, `fixed_rate@50ms`, `fixed_rate@200ms`, `fixed_rate@1000ms`). On the Pro body surface, a request may override this with a `channel` field. | ### Example Configuration @@ -71,6 +74,13 @@ Pyth cannot append feeds to an existing subscription id, so a grow sends a **new "method": "GET", "fetchFromModule": "{:symbol}", "channel": "real_time" + }, + { + "type": "pyth-lazer", + "moduleName": "pyth", + "path": "/v1/latest_price", + "method": "POST", + "channel": "fixed_rate@200ms" } ] } @@ -151,6 +161,21 @@ On a wait timeout, the entry still appears with `__sedaHasPrice: false` and with Requests with more feeds than `maxFeedsPerRequest`, or a numeric token that is not a u32, return HTTP 400. +### POST Body-Based Request (Latest Price Request) + +When a route omits `fetchFromModule`, the module accepts a JSON body (extra Pyth Pro fields such as `formats` / `properties` are ignored): + +```json +{ + "priceFeedIds": [1, 2], + "channel": "real_time" +} +``` + +Use `priceFeedSymbols` instead of `priceFeedIds` when requesting by symbol. `channel` is optional and falls back to the route’s `channel`. + +The response matches Pyth Pro’s `{ parsed: { timestampUs, priceFeeds } }` envelope. Missing optional feed fields are emitted as `null`. Unlike the path surface, this surface is all-or-fail: if any requested feed never produces a price, the whole request errors. + ## Notes diff --git a/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.test.ts b/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.test.ts index 6024c18..51bcb64 100644 --- a/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.test.ts +++ b/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.test.ts @@ -92,12 +92,16 @@ const makeFakeLazerClient = (options: FakeClientOptions = {}) => { messageListener?.({ type: "json", value } as JsonOrBinaryResponse); }; - const deliverTick = (subscriptionId: number, priceFeedIds: number[]) => { + const deliverTick = ( + subscriptionId: number, + priceFeedIds: number[], + timestampUs = "0", + ) => { emitJson({ type: "streamUpdated", subscriptionId, parsed: { - timestampUs: "0", + timestampUs, priceFeeds: priceFeedIds.map((priceFeedId) => ({ priceFeedId, price: `${priceFeedId * 100}`, @@ -106,6 +110,18 @@ const makeFakeLazerClient = (options: FakeClientOptions = {}) => { }); }; + const deliverParsed = ( + subscriptionId: number, + timestampUs: string, + priceFeeds: Array & { priceFeedId: number }>, + ) => { + emitJson({ + type: "streamUpdated", + subscriptionId, + parsed: { timestampUs, priceFeeds }, + }); + }; + const ack = (subscriptionId: number) => { emitJson({ type: "subscribed", subscriptionId }); }; @@ -161,6 +177,7 @@ const makeFakeLazerClient = (options: FakeClientOptions = {}) => { subscribeCalls, unsubscribeCalls, deliverTick, + deliverParsed, ack, ackWithInvalidIgnored, ackError, @@ -771,3 +788,307 @@ describe("handleRequest symbol resolution", () => { ); }); }); + +const latestPriceRoute = v.parse(PythLazerModuleRouteSchema, { + type: "pyth-lazer", + moduleName: "pyth", + path: "/v1/latest_price", + method: ["POST"], + channel: "fixed_rate@200ms", +}); + +const latestPriceBody = (feeds: { + priceFeedIds?: number[]; + priceFeedSymbols?: string[]; + channel?: string; +}) => + JSON.stringify({ + formats: [], + jsonBinaryEncoding: "base64", + parsed: true, + properties: ["price", "exponent", "feedUpdateTimestamp"], + ...feeds, + }); + +describe("handleRequest latest_price surface (POST body)", () => { + it("returns the parsed Pro shape with null-filled optional fields", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + yield* module.start(); + fake.deliverParsed(fake.subscribeCalls[0].subscriptionId, "100", [ + { + priceFeedId: 1, + price: "111", + exponent: -8, + feedUpdateTimestamp: 1_700_000_000_000_000, + }, + ]); + + const response = yield* module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { method: "POST" }), + latestPriceBody({ priceFeedIds: [1] }), + ); + + expect(response.status).toBe(200); + const body = (yield* Effect.promise(() => response.json())) as { + parsed: { + timestampUs: string; + priceFeeds: Record[]; + }; + }; + expect(body.parsed.timestampUs).toBe("100"); + expect(body.parsed.priceFeeds[0]).toEqual({ + priceFeedId: 1, + exponent: -8, + feedUpdateTimestamp: 1_700_000_000_000_000, + bestAskPrice: null, + bestBidPrice: null, + confidence: null, + emaConfidence: null, + emaPrice: null, + fundingRate: null, + fundingRateInterval: null, + fundingTimestamp: null, + marketSession: null, + price: "111", + publisherCount: null, + }); + expect(JSON.stringify(body)).not.toContain(HAS_PRICE_KEY); + }).pipe( + Effect.provide( + PythLazerModuleService( + makeConfig({ + priceFeedIds: [{ name: "BTC/USD", id: 1 }], + }), + ), + ), + ), + ); + }); + + it("uses the channel from the request body when provided", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + yield* module.start(); + expect(fake.subscribeCalls[0].channel).toBe("fixed_rate@200ms"); + + const fiber = yield* Effect.fork( + module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { + method: "POST", + }), + latestPriceBody({ + priceFeedIds: [1], + channel: "real_time", + }), + ), + ); + yield* TestClock.adjust(Duration.millis(0)); + + const realtimeId = fake.subscribeCalls.find( + (call) => call.channel === "real_time", + )?.subscriptionId; + expect(realtimeId).toBeDefined(); + if (realtimeId === undefined) { + throw new Error("expected real_time subscription id"); + } + fake.deliverTick(realtimeId, [1], "200"); + + const response = yield* Fiber.join(fiber); + expect(response.status).toBe(200); + const body = (yield* Effect.promise(() => response.json())) as { + parsed: { timestampUs: string; priceFeeds: { price: string }[] }; + }; + expect(body.parsed.timestampUs).toBe("200"); + expect(body.parsed.priceFeeds[0].price).toBe("100"); + }).pipe( + Effect.provide( + PythLazerModuleService( + makeConfig({ + priceFeedIds: [{ name: "BTC/USD", id: 1 }], + }), + ), + ), + ), + ); + }); + + it("does not serve a real_time price for a fixed_rate@200ms body request", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + yield* module.start(); + fake.deliverTick(fake.subscribeCalls[0].subscriptionId, [1], "111"); + + const fiber = yield* Effect.fork( + module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { + method: "POST", + }), + latestPriceBody({ + priceFeedIds: [1], + channel: "real_time", + }), + ), + ); + yield* TestClock.adjust(Duration.millis(0)); + + const realtimeId = fake.subscribeCalls.find( + (call) => call.channel === "real_time", + )?.subscriptionId; + expect(realtimeId).toBeDefined(); + if (realtimeId === undefined) { + throw new Error("expected real_time subscription id"); + } + fake.deliverTick(realtimeId, [1], "999"); + + const response = yield* Fiber.join(fiber); + const body = (yield* Effect.promise(() => response.json())) as { + parsed: { timestampUs: string; priceFeeds: { price: string }[] }; + }; + expect(body.parsed.timestampUs).toBe("999"); + expect(body.parsed.priceFeeds[0].price).toBe("100"); + }).pipe( + Effect.provide( + PythLazerModuleService( + makeConfig({ + priceFeedIds: [{ name: "BTC/USD", id: 1 }], + }), + ), + ), + ), + ); + }); + + it("falls back to the route channel when the body omits one", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + yield* module.start(); + fake.deliverTick(fake.subscribeCalls[0].subscriptionId, [1], "50"); + + const response = yield* module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { method: "POST" }), + JSON.stringify({ priceFeedIds: [1] }), + ); + + expect(response.status).toBe(200); + expect( + fake.subscribeCalls.every( + (call) => call.channel === "fixed_rate@200ms", + ), + ).toBe(true); + }).pipe( + Effect.provide( + PythLazerModuleService( + makeConfig({ + priceFeedIds: [{ name: "BTC/USD", id: 1 }], + }), + ), + ), + ), + ); + }); + + it("fails a wait timeout with the missing feed id", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + yield* module.start(); + fake.deliverTick(fake.subscribeCalls[0].subscriptionId, [1], "100"); + + const fiber = yield* Effect.fork( + module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { + method: "POST", + }), + latestPriceBody({ priceFeedIds: [1, 999] }), + ), + ); + yield* TestClock.adjust(Duration.millis(0)); + yield* TestClock.adjust(Duration.seconds(3)); + + const response = yield* Fiber.join(fiber); + expect(response.status).toBe(400); + const body = (yield* Effect.promise(() => response.json())) as { + data_proxy_error: string; + }; + expect(body.data_proxy_error).toContain("Price feed not found: 999"); + }).pipe( + Effect.provide( + PythLazerModuleService( + makeConfig({ + priceFeedIds: [{ name: "BTC/USD", id: 1 }], + }), + ), + ), + ), + ); + }); + + it("rejects a body with neither priceFeedIds nor priceFeedSymbols", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + const response = yield* module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { method: "POST" }), + JSON.stringify({ channel: "real_time" }), + ); + expect(response.status).toBe(400); + }).pipe(Effect.provide(PythLazerModuleService(makeConfig()))), + ); + }); + + it("rejects an invalid channel in the body", async () => { + const fake = makeFakeLazerClient(); + fakeHolder.current = fake; + + await runWithTestClock( + Effect.gen(function* () { + const module = yield* ModuleService; + const response = yield* module.handleRequest( + latestPriceRoute, + {}, + new Request("http://proxy.local/v1/latest_price", { method: "POST" }), + JSON.stringify({ + priceFeedIds: [1], + channel: "fixed_rate@9999ms", + }), + ); + expect(response.status).toBe(400); + }).pipe(Effect.provide(PythLazerModuleService(makeConfig()))), + ); + }); +}); diff --git a/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.ts b/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.ts index 30b4c27..6a9d09e 100644 --- a/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.ts +++ b/workspace/data-proxy/src/modules/pyth-lazer/pyth-lazer.ts @@ -5,6 +5,7 @@ import { type ParsedPayload, PythLazerClient, } from "@pythnetwork/pyth-lazer-sdk"; +import { tryParseSync } from "@seda-protocol/utils"; import { Clock, Data, @@ -16,8 +17,12 @@ import { Option, Runtime, } from "effect"; +import * as v from "valibot"; import type { Route } from "../../config/config-parser"; -import type { PythLazerModuleConfig } from "../../config/pyth-lazer-module-config"; +import { + PythLazerChannelSchema, + type PythLazerModuleConfig, +} from "../../config/pyth-lazer-module-config"; import { HAS_PRICE_KEY } from "../../constants"; import { createErrorResponse } from "../../controllers/create-error-response"; import { forkIdleCleanup } from "../../utils/idle-cleanup"; @@ -58,6 +63,16 @@ export const channelFromSubscriptionKey = ( key: PriceFeedSubscriptionKey, ): Channel => key.slice(0, key.lastIndexOf(":")) as Channel; +interface CachedPriceFeed { + priceFeed: ParsedFeedPayload; + timestampUs: string; +} + +interface PriceFeedWithSymbol extends ParsedFeedPayload { + symbol?: string; + [HAS_PRICE_KEY]: boolean; +} + const PYTH_LAZER_SUBSCRIBE_PROPERTIES = [ "bestAskPrice", "bestBidPrice", @@ -74,10 +89,19 @@ const PYTH_LAZER_SUBSCRIBE_PROPERTIES = [ "publisherCount", ] as const; -interface PriceFeedWithSymbol extends ParsedFeedPayload { - symbol?: string; - [HAS_PRICE_KEY]: boolean; -} +const LatestPriceRequestBodySchema = v.pipe( + v.looseObject({ + priceFeedIds: v.optional(v.array(v.number())), + priceFeedSymbols: v.optional(v.array(v.string())), + channel: v.optional(PythLazerChannelSchema), + }), + v.check( + (body) => + (body.priceFeedIds?.length ?? 0) > 0 || + (body.priceFeedSymbols?.length ?? 0) > 0, + "Request body must include a non-empty priceFeedIds or priceFeedSymbols array", + ), +); /** Pyth Lazer timestamps are Unix microseconds; returns lag in ms, or undefined if missing/invalid. */ export const lagMsFromTimestampUs = ( @@ -116,7 +140,7 @@ export const PythLazerModuleService = (config: PythLazerModuleConfig) => const runtime = yield* Effect.runtime(); const priceCache = yield* createPriceCache< PriceFeedSubscriptionKey, - ParsedFeedPayload + CachedPriceFeed >(); // The timestamp of the last request to the price feed const lastRequestToPriceFeed = MutableHashMap.empty< @@ -414,7 +438,10 @@ export const PythLazerModuleService = (config: PythLazerModuleConfig) => continue; } - priceCache.setPriceSync(key, priceFeed); + priceCache.setPriceSync(key, { + priceFeed, + timestampUs: message.timestampUs, + }); applied += 1; } recordTickHandle( @@ -546,39 +573,22 @@ export const PythLazerModuleService = (config: PythLazerModuleConfig) => }); }).pipe(Effect.annotateLogs("_name", "pyth-lazer")); - const handleRequest = ( - route: Route, - params: Record, - request: Request, - ) => + /** + * Shared pipeline: resolve tokens → grow subscriptions → wait on cache. + * Returns Either per feed so callers choose soft-miss vs all-or-fail. + */ + const resolveAndFetch = (tokens: string[], channel: Channel) => Effect.gen(function* () { - if (route.type !== "pyth-lazer") { + if (tokens.length > config.maxFeedsPerRequest) { return yield* Effect.fail( - new FailedToHandleRequest({ - msg: "Route is not a Pyth Lazer module", + new FailedToHandlePythLazerRequestError({ + error: `Too many price feeds, max is ${config.maxFeedsPerRequest} but got ${tokens.length}`, }), ); } - const priceFeedIdsRaw = replaceParams( - route.fetchFromModule, - params, - ).split(","); - - if (priceFeedIdsRaw.length > config.maxFeedsPerRequest) { - return yield* Effect.succeed( - createErrorResponse( - new FailedToHandlePythLazerRequestError({ - error: `Too many price feed IDs, max is ${config.maxFeedsPerRequest} but got ${priceFeedIdsRaw.length}`, - }), - 400, - ), - ); - } - - // Normalize ids or symbols to price feed ids concurrently. const priceFeedIds = yield* Effect.forEach( - priceFeedIdsRaw, + tokens, (symbolOrId) => Effect.gen(function* () { const parsedId = Number(symbolOrId); @@ -607,53 +617,169 @@ export const PythLazerModuleService = (config: PythLazerModuleConfig) => ); const now = yield* Clock.currentTimeMillis; - let desiredSetGrew = false; for (const priceFeedId of priceFeedIds) { - const key = priceFeedSubscriptionKey(priceFeedId, route.channel); - if (addDesiredFeed(route.channel, priceFeedId)) { + const key = priceFeedSubscriptionKey(priceFeedId, channel); + if (addDesiredFeed(channel, priceFeedId)) { desiredSetGrew = true; } - MutableHashMap.set(lastRequestToPriceFeed, key, now); } if (desiredSetGrew) { - sendSubscriptionForChannel(route.channel); + sendSubscriptionForChannel(channel); } - // Now since the subscriptions are in-flight, we can fetch the prices concurrently. const results = yield* Effect.forEach( priceFeedIds, (priceFeedId) => priceCache.getOrWaitPrice( - priceFeedSubscriptionKey(priceFeedId, route.channel), + priceFeedSubscriptionKey(priceFeedId, channel), ), { concurrency: "unbounded" }, ); - const prices: PriceFeedWithSymbol[] = []; + return { priceFeedIds, results }; + }); + + const handleRequest = ( + route: Route, + params: Record, + _request: Request, + body?: string, + ) => + Effect.gen(function* () { + if (route.type !== "pyth-lazer") { + return yield* Effect.fail( + new FailedToHandleRequest({ + msg: "Route is not a Pyth Lazer module", + }), + ); + } + + // Standard request + if (route.fetchFromModule !== undefined) { + const priceFeedIdsRaw = replaceParams( + route.fetchFromModule, + params, + ).split(","); + + const { priceFeedIds, results } = yield* resolveAndFetch( + priceFeedIdsRaw, + route.channel, + ); + + const prices: PriceFeedWithSymbol[] = []; + for (let i = 0; i < priceFeedIds.length; i++) { + const priceFeedId = priceFeedIds[i]; + const price = results[i]; + + if (price === null) { + prices.push({ + priceFeedId, + symbol: priceFeedIdsRaw.at(i), + [HAS_PRICE_KEY]: false, + }); + } else { + prices.push({ + ...price.priceFeed, + symbol: priceFeedIdsRaw.at(i), + [HAS_PRICE_KEY]: true, + }); + } + } + + return yield* Effect.succeed( + new Response(JSON.stringify(prices), { status: 200 }), + ); + } + + // POST body-based request (latest price request) + const parsedJson = yield* Effect.try({ + try: () => + JSON.parse(body && body.length > 0 ? body : "{}") as unknown, + catch: () => + new FailedToHandlePythLazerRequestError({ + error: "Request body must be valid JSON", + }), + }); + + const parsedRequest = tryParseSync( + LatestPriceRequestBodySchema, + parsedJson, + ); + if (parsedRequest.isErr) { + return yield* Effect.fail( + new FailedToHandlePythLazerRequestError({ + error: parsedRequest.error + .map((issue) => issue.message) + .join(", "), + }), + ); + } + + const requestBody = parsedRequest.value; + const requestedFeeds = + requestBody.priceFeedIds !== undefined && + requestBody.priceFeedIds.length > 0 + ? requestBody.priceFeedIds.map(String) + : (requestBody.priceFeedSymbols ?? []); + const channel = requestBody.channel ?? route.channel; + + const { priceFeedIds, results } = yield* resolveAndFetch( + requestedFeeds, + channel, + ); + + const cachedFeeds: CachedPriceFeed[] = []; for (let i = 0; i < priceFeedIds.length; i++) { - const priceFeedId = priceFeedIds[i]; - const price = results[i]; - - if (price === null) { - prices.push({ - priceFeedId, - symbol: priceFeedIdsRaw.at(i), - [HAS_PRICE_KEY]: false, - }); - } else { - prices.push({ - ...price, - symbol: priceFeedIdsRaw.at(i), - [HAS_PRICE_KEY]: true, - }); + const result = results[i]; + if (result === null) { + return yield* Effect.fail( + new FailedToHandlePythLazerRequestError({ + error: `Price feed not found: ${priceFeedIds[i]}`, + }), + ); } + cachedFeeds.push(result); } + const timestampUs = cachedFeeds + .map((feed) => feed.timestampUs) + .reduce((latest, timestamp) => + BigInt(timestamp) > BigInt(latest) ? timestamp : latest, + ); + + const priceFeeds = cachedFeeds.map( + ({ priceFeed, timestampUs: ts }) => { + const rawFeed = priceFeed as Record; + const out: Record = { + priceFeedId: priceFeed.priceFeedId, + feedUpdateTimestamp: + priceFeed.feedUpdateTimestamp ?? Number.parseInt(ts, 10), + }; + + for (const field of PYTH_LAZER_SUBSCRIBE_PROPERTIES) { + if (field === "feedUpdateTimestamp") { + continue; + } + out[field] = rawFeed[field] ?? null; + } + + return out; + }, + ); + return yield* Effect.succeed( - new Response(JSON.stringify(prices), { status: 200 }), + new Response( + JSON.stringify({ + parsed: { + timestampUs, + priceFeeds, + }, + }), + { status: 200 }, + ), ); }).pipe( Effect.withSpan("handlePythLazerRequest"), diff --git a/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts b/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts index 19ef073..82b7d8c 100644 --- a/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts +++ b/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts @@ -123,7 +123,11 @@ export const createTickerModuleService = < _request: Request, ) => Effect.gen(function* () { - if (route.type !== routeType || !("fetchFromModule" in route)) { + if ( + route.type !== routeType || + !("fetchFromModule" in route) || + typeof route.fetchFromModule !== "string" + ) { return yield* Effect.fail( new FailedToHandleRequest({ msg: `Route is not a ${titleCase(venue)} module`,