diff --git a/workspace/data-proxy/package.json b/workspace/data-proxy/package.json index 31cb8ed6..08b0101e 100644 --- a/workspace/data-proxy/package.json +++ b/workspace/data-proxy/package.json @@ -1,7 +1,7 @@ { "name": "@seda-protocol/data-proxy", "main": "./src/index.ts", - "version": "6.0.2", + "version": "6.1.0", "devDependencies": { "@types/big.js": "^6.2.2", "@types/ws": "^8.18.1", diff --git a/workspace/data-proxy/src/config/lighter-module-config.ts b/workspace/data-proxy/src/config/lighter-module-config.ts index 7ad915fb..388b0059 100644 --- a/workspace/data-proxy/src/config/lighter-module-config.ts +++ b/workspace/data-proxy/src/config/lighter-module-config.ts @@ -1,85 +1,36 @@ -import { Duration, Effect, Option } from "effect"; import * as v from "valibot"; -import { RouteSchema } from "./route-config"; +import { + keepaliveIntervalField, + tickerModuleBaseFields, + tickerModuleRouteSchema, + validateTickerModuleRoute, +} from "./ticker-module-config"; + +export const LIGHTER_STREAM_TYPES = ["ticker", "order_book", "trade"] as const; + +export type LighterStreamType = (typeof LIGHTER_STREAM_TYPES)[number]; export const LighterModuleConfigSchema = v.strictObject({ - name: v.string(), type: v.literal("lighter"), - wsUrl: v.optional(v.string(), "wss://mainnet.zklighter.elliot.ai/stream"), - // Lighter numeric market ids (e.g. 1 for BTC) to subscribe at startup. - subscriptionMarketIds: v.optional(v.array(v.number()), []), - maxMarketsPerRequest: v.optional(v.number(), 100), - // Lighter allows 200 client WS messages per minute. - maxMessagesPerMinute: v.optional( - v.pipe( - v.number(), - v.minValue(1, "maxMessagesPerMinute must be at least 1"), - ), - 180, - ), + ...tickerModuleBaseFields({ + // ?readonly=true avoids geo-restriction on the public stream. + wsUrl: "wss://mainnet.zklighter.elliot.ai/stream?readonly=true", + // Lighter allows 200 client WS messages per minute. + maxMessages: 180, + maxMessagesWindow: "1 minute", + }), // Lighter closes a connection with no client frames for 2 minutes; // ping on a shorter cadence. - keepaliveInterval: v.pipe( - v.optional(v.union([v.number(), v.string()]), "60 seconds"), - v.transform((value) => - Option.getOrThrowWith( - Duration.decodeUnknown(value), - () => new Error("Invalid keepaliveInterval duration"), - ), - ), - ), - reconnectMaxBackoff: v.pipe( - v.optional(v.union([v.number(), v.string()]), "30 seconds"), - v.transform((value) => - Option.getOrThrowWith( - Duration.decodeUnknown(value), - () => new Error("Invalid reconnectMaxBackoff duration"), - ), - ), - ), - reconnectStableThreshold: v.pipe( - v.optional(v.union([v.number(), v.string()]), "30 seconds"), - v.transform((value) => - Option.getOrThrowWith( - Duration.decodeUnknown(value), - () => new Error("Invalid reconnectStableThreshold duration"), - ), - ), - ), - marketsCleanupTtl: v.pipe( - v.optional(v.union([v.number(), v.string()]), "1 hour"), - v.transform((ttl) => - Option.getOrThrowWith( - Duration.decodeUnknown(ttl), - () => new Error("Invalid markets cleanup TTL"), - ), - ), - ), - marketsCleanupInterval: v.pipe( - v.optional(v.union([v.number(), v.string()]), "30 seconds"), - v.transform((interval) => - Option.getOrThrowWith( - Duration.decodeUnknown(interval), - () => new Error("Invalid markets cleanup interval"), - ), - ), - ), + keepaliveInterval: keepaliveIntervalField("60 seconds"), + streamType: v.optional(v.picklist(LIGHTER_STREAM_TYPES), "ticker"), }); export type LighterModuleConfig = v.InferOutput< typeof LighterModuleConfigSchema >; -export const LighterModuleRouteSchema = v.strictObject({ - ...RouteSchema.entries, - moduleName: v.string(), - fetchFromModule: v.string(), - type: v.literal("lighter"), -}); +export const LighterModuleRouteSchema = tickerModuleRouteSchema("lighter"); export type LighterModuleRoute = v.InferOutput; -export const validateLighterModuleRoute = (_route: LighterModuleRoute) => - Effect.gen(function* () { - return yield* Effect.void; - }); +export const validateLighterModuleRoute = validateTickerModuleRoute; diff --git a/workspace/data-proxy/src/config/ticker-module-config.ts b/workspace/data-proxy/src/config/ticker-module-config.ts index 084c1512..2f0febdd 100644 --- a/workspace/data-proxy/src/config/ticker-module-config.ts +++ b/workspace/data-proxy/src/config/ticker-module-config.ts @@ -13,7 +13,7 @@ const durationField = (defaultValue: string, invalidMessage: string) => ), ); -/** Shared fields for string-keyed public ticker modules. */ +/** Shared fields for public ticker modules. */ export const tickerModuleBaseFields = (defaults: { wsUrl: string; maxMessages: number; diff --git a/workspace/data-proxy/src/modules/lighter/errors.ts b/workspace/data-proxy/src/modules/lighter/errors.ts deleted file mode 100644 index ac45caca..00000000 --- a/workspace/data-proxy/src/modules/lighter/errors.ts +++ /dev/null @@ -1,7 +0,0 @@ -import { Data } from "effect"; - -export class FailedToHandleLighterRequestError extends Data.TaggedError( - "FailedToHandleLighterRequestError", -)<{ error: string; status: number }> { - message = `Lighter error: ${this.error}`; -} diff --git a/workspace/data-proxy/src/modules/lighter/lighter.ts b/workspace/data-proxy/src/modules/lighter/lighter.ts deleted file mode 100644 index 3c7d401c..00000000 --- a/workspace/data-proxy/src/modules/lighter/lighter.ts +++ /dev/null @@ -1,161 +0,0 @@ -import { Clock, Effect, Layer, MutableHashMap } from "effect"; -import type { Route } from "../../config/config-parser"; -import type { LighterModuleConfig } from "../../config/lighter-module-config"; -import { HAS_PRICE_KEY } from "../../constants"; -import { createErrorResponse } from "../../controllers/create-error-response"; -import { forkIdleCleanup } from "../../utils/idle-cleanup"; -import { replaceParams } from "../../utils/replace-params"; -import { FailedToHandleRequest, ModuleService } from "../module"; -import { createPriceCache } from "../shared/price-cache"; -import { FailedToHandleLighterRequestError } from "./errors"; -import { type LighterPriceFrame, createLighterWS } from "./ws-client"; - -type LighterPriceEntry = - | ({ marketId: string } & LighterPriceFrame & { [HAS_PRICE_KEY]: true }) - | { marketId: string; [HAS_PRICE_KEY]: false }; - -// Request tokens are Lighter numeric market ids (e.g. "1" for BTC). A token that -// is not a non-negative integer can never name a market, so it resolves to a miss. -const parseMarketId = (token: string): number | null => { - const id = Number(token); - return Number.isInteger(id) && id >= 0 ? id : null; -}; - -export const LighterModuleService = (config: LighterModuleConfig) => - Layer.effect( - ModuleService, - Effect.gen(function* () { - yield* Effect.logInfo("Initializing Lighter module", { - name: config.name, - wsUrl: config.wsUrl, - }); - - const cache = yield* createPriceCache(); - const ws = yield* createLighterWS(config, cache); - // Market id -> timestamp of the last request, drives the idle cleanup pass. - const lastRequestToMarketId = MutableHashMap.empty(); - - const start = () => - Effect.gen(function* () { - yield* Effect.logInfo("Starting Lighter module", { - name: config.name, - }); - - yield* ws.start(); - - if (config.subscriptionMarketIds.length > 0) { - const now = yield* Clock.currentTimeMillis; - for (const marketId of config.subscriptionMarketIds) { - MutableHashMap.set(lastRequestToMarketId, marketId, now); - } - yield* ws.subscribe(config.subscriptionMarketIds); - } - - yield* forkIdleCleanup({ - lastRequest: lastRequestToMarketId, - ttl: config.marketsCleanupTtl, - interval: config.marketsCleanupInterval, - onExpire: (marketId) => - Effect.gen(function* () { - yield* Effect.logInfo("Cleaning up idle market", { marketId }); - yield* cache.deletePrice(marketId); - yield* ws.unsubscribe([marketId]); - }), - }); - }).pipe(Effect.annotateLogs("_name", "lighter")); - - const handleRequest = ( - route: Route, - params: Record, - _request: Request, - ) => - Effect.gen(function* () { - if (route.type !== "lighter") { - return yield* Effect.fail( - new FailedToHandleRequest({ - msg: "Route is not a Lighter module", - }), - ); - } - - const requestedTokens = replaceParams(route.fetchFromModule, params) - .split(",") - .map((token) => token.trim()) - .filter((token) => token.length > 0); - - if (requestedTokens.length > config.maxMarketsPerRequest) { - return yield* Effect.fail( - new FailedToHandleLighterRequestError({ - error: `Too many markets, max is ${config.maxMarketsPerRequest} but got ${requestedTokens.length}`, - status: 400, - }), - ); - } - - const requested = requestedTokens.map((token) => ({ - token, - marketId: parseMarketId(token), - })); - - const now = yield* Clock.currentTimeMillis; - const socketHealthy = !(yield* ws.hasError()); - const newMarketIds: number[] = []; - for (const { marketId } of requested) { - if (marketId === null) continue; - if (!MutableHashMap.has(lastRequestToMarketId, marketId)) { - newMarketIds.push(marketId); - } - MutableHashMap.set(lastRequestToMarketId, marketId, now); - } - - if (newMarketIds.length > 0) { - yield* ws.subscribe(newMarketIds); - } - - // Subscriptions are in-flight; resolve every requested market concurrently. - // A token that is not a market id can never land in the cache, so it - // short-circuits to a miss instead of blocking on the wait. - const results = yield* Effect.forEach( - requested, - ({ marketId }) => { - if (marketId === null) { - return Effect.succeed(null); - } - return cache.getOrWaitPrice(marketId); - }, - { concurrency: "unbounded" }, - ); - - const prices: LighterPriceEntry[] = []; - for (let i = 0; i < requested.length; i++) { - const { token } = requested[i]; - const result = results[i]; - - if (result === null || !socketHealthy) { - prices.push({ marketId: token, [HAS_PRICE_KEY]: false }); - } else { - prices.push({ - marketId: token, - ...result, - [HAS_PRICE_KEY]: true, - }); - } - } - - return new Response(JSON.stringify(prices), { - status: 200, - headers: { "Content-Type": "application/json" }, - }); - }).pipe( - Effect.withSpan("handleLighterRequest"), - Effect.catchAll((error) => - Effect.succeed(createErrorResponse(error, error.status)), - ), - ); - - return { - start, - handleRequest, - }; - }), - ); diff --git a/workspace/data-proxy/src/modules/lighter/ws-client.ts b/workspace/data-proxy/src/modules/lighter/ws-client.ts deleted file mode 100644 index 36d8dd1b..00000000 --- a/workspace/data-proxy/src/modules/lighter/ws-client.ts +++ /dev/null @@ -1,346 +0,0 @@ -import { - Deferred, - Duration, - Effect, - type Fiber, - Metric, - MutableHashMap, - Option, - Queue, - Runtime, - Schedule, - Stream, -} from "effect"; -import type { LighterModuleConfig } from "../../config/lighter-module-config"; -import { forkInboundControl } from "../shared/inbound-control"; -import { recordTickHandle } from "../shared/tick-metrics"; - -/** A Lighter `ticker` payload. Always carries the symbol in `s`; the best - * bid/ask sit in `b`/`a` and are relayed verbatim. */ -export interface LighterPriceFrame { - s: string; - [key: string]: unknown; -} - -type OutboundMessageType = "subscribe" | "unsubscribe" | "ping" | "pong"; - -type OutboundMessage = { - frame: string; - type: OutboundMessageType; -}; - -/** Metrics for the Lighter WebSocket client. */ -const activeSubscriptions = Metric.gauge("lighter_active_subscriptions", { - description: "Number of active Lighter WebSocket market subscriptions", -}); - -const messagesSent = Metric.counter("lighter_messages_sent", { - description: - "Outbound Lighter WebSocket messages sent (subscribe/unsubscribe/ping/pong)", - incremental: true, -}); - -const connectionAttempts = Metric.counter("lighter_connection_attempts", { - description: "Lighter WebSocket connection attempts", - incremental: true, -}); - -/** The subset of the shared price cache the WS daemon writes to, keyed by market id. */ -interface PriceSink { - setPriceSync: (key: number, price: LighterPriceFrame) => void; -} - -const PING_FRAME = JSON.stringify({ type: "ping" }); -const PONG_FRAME = JSON.stringify({ type: "pong" }); - -export const buildSubscribeFrame = (marketId: number): string => - JSON.stringify({ type: "subscribe", channel: `ticker/${marketId}` }); - -export const buildUnsubscribeFrame = (marketId: number): string => - JSON.stringify({ type: "unsubscribe", channel: `ticker/${marketId}` }); - -const isRecord = (value: unknown): value is Record => - typeof value === "object" && value !== null; - -/** The inbound `channel` is `ticker:{id}` (colon) even though subscribe sends - * `ticker/{id}` (slash); accept either separator. */ -const parseMarketId = (channel: unknown): number | null => { - if (typeof channel !== "string") return null; - const last = channel.split(/[:/]/).pop(); - if (last === undefined) return null; - const id = Number(last); - return Number.isInteger(id) ? id : null; -}; - -export type ParsedInbound = - | { kind: "ping" } - | { kind: "ticker"; marketId: number | null; frame: LighterPriceFrame } - | { kind: "error"; code: number | null; message: string | null }; - -/** Classifies an inbound message: a keepalive ping, a ticker payload (snapshot - * `subscribed/ticker` or `update/ticker`, both carry `.ticker`), a venue error - * frame, or null for control frames like `{type:connected}` and malformed input. */ -export const parseInboundFrame = (raw: string): ParsedInbound | null => { - let json: unknown; - try { - json = JSON.parse(raw); - } catch { - return null; - } - if (!isRecord(json)) return null; - if (json.type === "ping") return { kind: "ping" }; - - const err = json.error; - if (isRecord(err)) { - return { - kind: "error", - code: typeof err.code === "number" ? err.code : null, - message: typeof err.message === "string" ? err.message : null, - }; - } - - const ticker = json.ticker; - if (isRecord(ticker) && typeof ticker.s === "string") { - return { - kind: "ticker", - marketId: parseMarketId(json.channel), - frame: ticker as LighterPriceFrame, - }; - } - return null; -}; - -export const defaultReconnectSchedule = (config: LighterModuleConfig) => - Schedule.exponential(Duration.seconds(1)).pipe( - Schedule.either(Schedule.spaced(config.reconnectMaxBackoff)), - Schedule.resetAfter(config.reconnectStableThreshold), - ); - -export interface LighterWS { - /** Forks the WS daemon (reconnect with backoff, resubscribe on open), the - * paced outbound sender, and the keepalive daemon. */ - start(): Effect.Effect, never, never>; - /** Adds the market ids to the desired set and enqueues a subscribe frame for - * each new one. Idempotent. Frames are paced by the outbound sender. */ - subscribe(marketIds: number[]): Effect.Effect; - /** Removes the market ids from the desired set and enqueues an unsubscribe - * frame for each removed one. Idempotent. Frames are paced by the outbound sender. */ - unsubscribe(marketIds: number[]): Effect.Effect; - /** True while there is no open connection, so cached prices may be stale. */ - hasError(): Effect.Effect; -} - -export interface CreateLighterWSOptions { - reconnectSchedule?: Schedule.Schedule; -} - -export const createLighterWS = ( - config: LighterModuleConfig, - cache: PriceSink, - options?: CreateLighterWSOptions, -): Effect.Effect => - Effect.gen(function* () { - const runtime = yield* Effect.runtime(); - const desiredMarkets = MutableHashMap.empty(); - let currentWS: WebSocket | null = null; - const schedule = - options?.reconnectSchedule ?? defaultReconnectSchedule(config); - // Outbound queue to enfore maxMessagesPerMinute rate limit. - const outbound = yield* Queue.unbounded(); - const inboundControl = yield* forkInboundControl( - (event: Extract) => - Effect.logWarning("Lighter WS error frame", { - code: event.code, - message: event.message, - }), - ); - - const withModuleName = ( - metric: Metric.Metric, - ) => Metric.tagged(metric, "module", config.name); - - const setActiveSubscriptions = () => - Metric.set( - withModuleName(activeSubscriptions), - MutableHashMap.size(desiredMarkets), - ); - - const enqueue = (frame: string, type: OutboundMessageType) => - Queue.offer(outbound, { frame, type }).pipe(Effect.asVoid); - - const clearOutbound = () => Queue.takeAll(outbound).pipe(Effect.asVoid); - - const sendOutbound = ({ frame, type }: OutboundMessage) => - Effect.gen(function* () { - const ws = currentWS; - if (ws === null || ws.readyState !== WebSocket.OPEN) { - // Drop; handleOpen re-enqueues current desiredMarkets on reconnect. - return; - } - - try { - ws.send(frame); - yield* Metric.increment( - Metric.tagged(withModuleName(messagesSent), "type", type), - ); - yield* Effect.sleep(Duration.decode(Duration.minutes(1))); - } catch (err) { - yield* Effect.logWarning("Lighter WS send failed", { - error: String(err), - }); - try { - ws.close(); - } catch { - // best-effort; the close listener will trigger the reconnect loop. - } - } - }); - - const sendLoop = Stream.fromQueue(outbound).pipe( - Stream.mapEffect(sendOutbound, { - concurrency: config.maxMessagesPerMinute, - }), - Stream.runDrain, - ); - - const subscribe = (marketIds: number[]) => - Effect.gen(function* () { - let added = false; - for (const marketId of marketIds) { - if (Option.isSome(MutableHashMap.get(desiredMarkets, marketId))) - continue; - MutableHashMap.set(desiredMarkets, marketId, true); - added = true; - yield* enqueue(buildSubscribeFrame(marketId), "subscribe"); - } - if (added) { - yield* setActiveSubscriptions(); - } - }); - - const unsubscribe = (marketIds: number[]) => - Effect.gen(function* () { - let removed = false; - for (const marketId of marketIds) { - if (Option.isNone(MutableHashMap.get(desiredMarkets, marketId))) - continue; - MutableHashMap.remove(desiredMarkets, marketId); - removed = true; - yield* enqueue(buildUnsubscribeFrame(marketId), "unsubscribe"); - } - if (removed) { - yield* setActiveSubscriptions(); - } - }); - - const handleOpen = (ws: WebSocket) => - Effect.gen(function* () { - yield* Effect.logInfo("Lighter WS open", { name: config.name }); - currentWS = ws; - for (const [marketId] of desiredMarkets) { - yield* enqueue(buildSubscribeFrame(marketId), "subscribe"); - } - }); - - const handleInboundMessage = (raw: string): void => { - const started = performance.now(); - const parsed = parseInboundFrame(raw); - if (!parsed) return; - if (parsed.kind === "ping") { - outbound.unsafeOffer({ frame: PONG_FRAME, type: "pong" }); - return; - } - if (parsed.kind === "error") { - inboundControl.offer(parsed); - return; - } - // Drop frames with no parseable id, or for a market we have since - // unsubscribed (post-unsubscribe race). - if ( - parsed.marketId === null || - Option.isNone(MutableHashMap.get(desiredMarkets, parsed.marketId)) - ) { - return; - } - cache.setPriceSync(parsed.marketId, parsed.frame); - recordTickHandle("lighter", config.name, performance.now() - started); - }; - - const handleDisconnect = (closed: Deferred.Deferred) => - Effect.gen(function* () { - currentWS = null; - // Drop queued frames so a reconnect isn't preceded by a stale backlog. - yield* clearOutbound(); - yield* Deferred.fail(closed, undefined); - }); - - const connectOnce = Effect.gen(function* () { - const deferred = yield* Deferred.make(); - - yield* Effect.logInfo("Lighter WS connecting", { name: config.name }); - yield* Metric.increment(withModuleName(connectionAttempts)); - - const ws = yield* Effect.acquireRelease( - Effect.sync(() => new WebSocket(config.wsUrl)), - (socket) => - Effect.sync(() => { - if (socket.readyState !== WebSocket.CLOSED) { - socket.close(); - } - }), - ); - - ws.addEventListener("open", () => { - Runtime.runSync(runtime, handleOpen(ws)); - }); - ws.addEventListener("message", (event) => { - if (typeof event.data !== "string") return; - handleInboundMessage(event.data); - }); - ws.addEventListener("error", () => { - Runtime.runSync( - runtime, - Effect.logWarning("Lighter WS error event", { name: config.name }), - ); - }); - ws.addEventListener("close", (event) => { - Runtime.runSync( - runtime, - Effect.gen(function* () { - yield* Effect.logWarning("Lighter WS disconnected", { - code: event.code, - closeReason: event.reason, - wasClean: event.wasClean, - }); - yield* handleDisconnect(deferred); - }), - ); - }); - - yield* Deferred.await(deferred); - }).pipe(Effect.scoped); - - const loop = connectOnce.pipe( - Effect.tapError(() => Effect.logWarning("Lighter WS connect failed")), - Effect.retry(schedule), - ); - - const keepaliveLoop = enqueue(PING_FRAME, "ping").pipe( - Effect.schedule(Schedule.spaced(config.keepaliveInterval)), - ); - - const cachedStart = yield* Effect.cached( - Effect.gen(function* () { - yield* Effect.forkDaemon(sendLoop); - yield* Effect.forkDaemon(keepaliveLoop); - return yield* Effect.forkDaemon(loop); - }), - ); - const start = () => cachedStart; - - // The socket is healthy only while a connection is open; currentWS is - // nulled on every disconnect and reset on open. - const hasError = () => Effect.sync(() => currentWS === null); - - return { start, subscribe, unsubscribe, hasError } satisfies LighterWS; - }); diff --git a/workspace/data-proxy/src/modules/shared/json.ts b/workspace/data-proxy/src/modules/shared/json.ts new file mode 100644 index 00000000..4937f1c1 --- /dev/null +++ b/workspace/data-proxy/src/modules/shared/json.ts @@ -0,0 +1,30 @@ +export const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +/** `null` for invalid JSON, non-objects, and arrays. */ +export const parseJsonRecord = ( + raw: string, +): Record | null => { + try { + const json: unknown = JSON.parse(raw); + return isRecord(json) ? json : null; + } catch { + return null; + } +}; + +/** Walks a payload that is either one object or an array of objects, skipping + * items that are not objects or that `map` rejects. */ +export const mapDataItems = ( + data: unknown, + map: (item: Record) => T | null, +): T[] => { + const items = Array.isArray(data) ? data : isRecord(data) ? [data] : []; + const out: T[] = []; + for (const item of items) { + if (!isRecord(item)) continue; + const mapped = map(item); + if (mapped !== null) out.push(mapped); + } + return out; +}; diff --git a/workspace/data-proxy/src/modules/shared/venue-ws.test.ts b/workspace/data-proxy/src/modules/shared/venue-ws.test.ts index 5b5c32d7..b5d82137 100644 --- a/workspace/data-proxy/src/modules/shared/venue-ws.test.ts +++ b/workspace/data-proxy/src/modules/shared/venue-ws.test.ts @@ -49,7 +49,7 @@ const startClient = (options?: { ? undefined : { interval: options.keepaliveInterval, - frame: "ping", + pingFrame: "ping", }, buildSubscribeFrame: (keys) => JSON.stringify({ op: "subscribe", keys }), buildUnsubscribeFrame: (keys) => @@ -80,7 +80,7 @@ const startClient = (options?: { describe("createVenueWS", () => { it("opens the WS at the configured url and batches the subscribe on open", async () => { const { fiber, ws: service } = await Effect.runPromise( - startClient({ preSubscribed: ["btc", "eth"] }), + startClient({ preSubscribed: ["BTC", "ETH"] }), ); await flush(); @@ -191,6 +191,105 @@ describe("createVenueWS", () => { await Effect.runPromise(Fiber.interrupt(fiber)); }); + it("enqueues one frame per key when the builder returns an array", async () => { + const { ws: service, fiber } = await Effect.runPromise( + Effect.gen(function* () { + const cache = yield* createPriceCache< + string, + Record + >(); + const ws = yield* createVenueWS({ + venue: "venue", + config: { + name: "venue-test", + wsUrl: "wss://example.test/ws", + maxMessages: 5, + maxMessagesWindow: Duration.seconds(1), + reconnectMaxBackoff: Duration.seconds(30), + reconnectStableThreshold: Duration.seconds(30), + }, + cache, + reconnectSchedule: Schedule.spaced(Duration.minutes(10)), + buildSubscribeFrame: (keys) => + keys.map((key) => JSON.stringify({ op: "subscribe", keys: [key] })), + buildUnsubscribeFrame: (keys) => + keys.map((key) => + JSON.stringify({ op: "unsubscribe", keys: [key] }), + ), + parseInboundFrame: () => null, + }); + const fiber = yield* ws.start(); + return { ws, fiber }; + }).pipe(Logger.withMinimumLogLevel(LogLevel.None)), + ); + await flush(); + const ws = FakeWebSocket.instances[0]; + ws.triggerOpen(); + await flush(); + + await Effect.runPromise(service.subscribe(["BTC", "ETH"])); + await flush(); + + expect(ws.sent.map(parseControl)).toEqual([ + { op: "subscribe", keys: ["BTC"] }, + { op: "subscribe", keys: ["ETH"] }, + ]); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); + + it("replies to an inbound ping immediately when pongFrame is configured", async () => { + const { fiber } = await Effect.runPromise( + Effect.gen(function* () { + const cache = yield* createPriceCache< + string, + Record + >(); + const ws = yield* createVenueWS({ + venue: "venue", + config: { + name: "venue-test", + wsUrl: "wss://example.test/ws", + maxMessages: 5, + maxMessagesWindow: Duration.seconds(1), + reconnectMaxBackoff: Duration.seconds(30), + reconnectStableThreshold: Duration.seconds(30), + }, + cache, + reconnectSchedule: Schedule.spaced(Duration.minutes(10)), + keepalive: { + interval: Duration.minutes(10), + pingFrame: "ping", + pongFrame: "pong", + }, + buildSubscribeFrame: (keys) => + JSON.stringify({ op: "subscribe", keys }), + buildUnsubscribeFrame: (keys) => + JSON.stringify({ op: "unsubscribe", keys }), + parseInboundFrame: (raw) => { + if (raw === "ping") return { kind: "ping" }; + return null; + }, + }); + yield* ws.subscribe(["BTC"]); + const fiber = yield* ws.start(); + return { fiber }; + }).pipe(Logger.withMinimumLogLevel(LogLevel.None)), + ); + await flush(); + const ws = FakeWebSocket.instances[0]; + ws.triggerOpen(); + await flush(); + ws.sent.length = 0; + + ws.triggerMessage("ping"); + await flush(); + + expect(ws.sent).toEqual(["pong"]); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); + it("subscribe batches multiple new keys into one frame", async () => { const { ws: service, fiber } = await Effect.runPromise( startClient({ preSubscribed: [] }), @@ -269,7 +368,7 @@ describe("createVenueWS", () => { it("resubscribes desired keys after reconnect", async () => { const { fiber } = await Effect.runPromise( startClient({ - preSubscribed: ["btc", "eth"], + preSubscribed: ["BTC", "ETH"], reconnectSchedule: Schedule.spaced(Duration.millis(10)), }), ); diff --git a/workspace/data-proxy/src/modules/shared/venue-ws.ts b/workspace/data-proxy/src/modules/shared/venue-ws.ts index 03f4e0ad..864241b7 100644 --- a/workspace/data-proxy/src/modules/shared/venue-ws.ts +++ b/workspace/data-proxy/src/modules/shared/venue-ws.ts @@ -33,23 +33,24 @@ export const defaultReconnectSchedule = (config: ReconnectBackoffConfig) => Schedule.resetAfter(config.reconnectStableThreshold), ); -export interface VenueWS { +export interface VenueWS { /** Forks the WS daemon. The daemon owns reconnect with backoff and resubscribes on each open. */ start(): Effect.Effect, never, never>; - /** Adds the keys to the desired set and enqueues one subscribe frame for the new ones. Idempotent. */ - subscribe(keys: string[]): Effect.Effect; - /** Removes the keys from the desired set and enqueues one unsubscribe frame for the removed ones. Idempotent. */ - unsubscribe(keys: string[]): Effect.Effect; + /** Adds the keys to the desired set and enqueues subscribe frame(s) for the new ones. Idempotent. */ + subscribe(keys: TKey[]): Effect.Effect; + /** Removes the keys from the desired set and enqueues unsubscribe frame(s) for the removed ones. Idempotent. */ + unsubscribe(keys: TKey[]): Effect.Effect; /** True while the socket is disconnected, errored, or has a pending send failure. */ hasError(): Effect.Effect; } -export type VenueParsedInbound = +export type VenueParsedInbound = + | { kind: "ping" } | { kind: "pong" } | { kind: "error"; code: string | number | null; message: string | null } | { kind: "tickers"; - frames: Array<{ key: string; frame: TFrame }>; + frames: Array<{ key: TKey; frame: TFrame }>; }; export interface VenueWSConfig extends ReconnectBackoffConfig { @@ -59,23 +60,25 @@ export interface VenueWSConfig extends ReconnectBackoffConfig { maxMessagesWindow: Duration.Duration; } -export interface CreateVenueWSParams { +export interface CreateVenueWSParams { venue: string; config: VenueWSConfig; - cache: PriceCache; + cache: PriceCache; reconnectSchedule?: Schedule.Schedule; keepalive?: { interval: VenueWSConfig["reconnectMaxBackoff"]; - frame: string; + pingFrame: string; + /** When set, an inbound ping is answered immediately with this frame. */ + pongFrame?: string; }; - buildSubscribeFrame: (keys: string[]) => string; - buildUnsubscribeFrame: (keys: string[]) => string; - parseInboundFrame: (raw: string) => VenueParsedInbound | null; + buildSubscribeFrame: (keys: TKey[]) => string | string[]; + buildUnsubscribeFrame: (keys: TKey[]) => string | string[]; + parseInboundFrame: (raw: string) => VenueParsedInbound | null; } -export const createVenueWS = ( - params: CreateVenueWSParams, -): Effect.Effect => +export const createVenueWS = ( + params: CreateVenueWSParams, +): Effect.Effect, never, never> => Effect.gen(function* () { const { venue, @@ -91,7 +94,7 @@ export const createVenueWS = ( params.reconnectSchedule ?? defaultReconnectSchedule(config); const inboundControl = yield* forkInboundControl( - (event: Extract, { kind: "error" }>) => + (event: Extract, { kind: "error" }>) => Effect.logWarning(`${venue} WS error frame`, { code: event.code, message: event.message, @@ -99,7 +102,7 @@ export const createVenueWS = ( ); const runtime = yield* Effect.runtime(); - const desiredKeys = MutableHashMap.empty(); + const desiredKeys = MutableHashMap.empty(); let currentWS: WebSocket | null = null; let unhealthy = false; const outbound = yield* Queue.unbounded(); @@ -133,12 +136,26 @@ export const createVenueWS = ( const incrementConnectionAttempts = () => Metric.increment(withModuleName(connectionAttempts)); - const isDesired = (key: string): boolean => - Option.isSome(MutableHashMap.get(desiredKeys, key.toUpperCase())); + const isDesired = (key: TKey): boolean => + Option.isSome(MutableHashMap.get(desiredKeys, key)); const enqueue = (frame: string, type: OutboundMessageType) => Queue.offer(outbound, { frame, type }).pipe(Effect.asVoid); + const enqueueFrames = ( + frames: string | string[], + type: OutboundMessageType, + ) => + Effect.gen(function* () { + if (Array.isArray(frames)) { + for (const frame of frames) { + yield* enqueue(frame, type); + } + return; + } + yield* enqueue(frames, type); + }); + const clearOutbound = () => Queue.takeAll(outbound).pipe(Effect.asVoid); const closeOnSendFailure = (ws: WebSocket, err: unknown, action: string) => @@ -154,6 +171,18 @@ export const createVenueWS = ( } }); + const sendImmediate = (frame: string, type: string) => + Effect.gen(function* () { + const ws = currentWS; + if (ws === null || ws.readyState !== WebSocket.OPEN) return; + try { + ws.send(frame); + yield* incrementMessagesSent(type); + } catch (err) { + yield* closeOnSendFailure(ws, err, type); + } + }); + const sendOutbound = ({ frame, type }: OutboundMessage) => Effect.gen(function* () { const ws = currentWS; @@ -179,32 +208,30 @@ export const createVenueWS = ( Stream.runDrain, ); - const subscribe = (keys: string[]) => + const subscribe = (keys: TKey[]) => Effect.gen(function* () { - const fresh: string[] = []; - for (const raw of keys) { - const key = raw.toUpperCase(); + const fresh: TKey[] = []; + for (const key of keys) { if (Option.isSome(MutableHashMap.get(desiredKeys, key))) continue; MutableHashMap.set(desiredKeys, key, true); fresh.push(key); } if (fresh.length === 0) return; yield* setActiveSubscriptions(); - yield* enqueue(buildSubscribeFrame(fresh), "subscribe"); + yield* enqueueFrames(buildSubscribeFrame(fresh), "subscribe"); }); - const unsubscribe = (keys: string[]) => + const unsubscribe = (keys: TKey[]) => Effect.gen(function* () { - const removed: string[] = []; - for (const raw of keys) { - const key = raw.toUpperCase(); + const removed: TKey[] = []; + for (const key of keys) { if (Option.isNone(MutableHashMap.get(desiredKeys, key))) continue; MutableHashMap.remove(desiredKeys, key); removed.push(key); } if (removed.length === 0) return; yield* setActiveSubscriptions(); - yield* enqueue(buildUnsubscribeFrame(removed), "unsubscribe"); + yield* enqueueFrames(buildUnsubscribeFrame(removed), "unsubscribe"); }); const hasError = () => Effect.sync(() => unhealthy); @@ -214,10 +241,10 @@ export const createVenueWS = ( yield* Effect.logInfo(`${venue} WS open`, { name }); unhealthy = false; currentWS = ws; - const keys: string[] = []; + const keys: TKey[] = []; for (const [key] of desiredKeys) keys.push(key); if (keys.length > 0) { - yield* enqueue(buildSubscribeFrame(keys), "subscribe"); + yield* enqueueFrames(buildSubscribeFrame(keys), "subscribe"); } }); @@ -235,6 +262,12 @@ export const createVenueWS = ( const parsed = parseInboundFrame(raw); if (!parsed) return; if (parsed.kind === "pong") return; + if (parsed.kind === "ping") { + if (keepalive?.pongFrame !== undefined) { + Runtime.runSync(runtime, sendImmediate(keepalive.pongFrame, "pong")); + } + return; + } if (parsed.kind === "error") { inboundControl.offer(parsed); return; @@ -308,14 +341,7 @@ export const createVenueWS = ( const sendPing = () => Effect.gen(function* () { if (keepalive === undefined) return; - const ws = currentWS; - if (ws === null || ws.readyState !== WebSocket.OPEN) return; - try { - ws.send(keepalive.frame); - yield* incrementMessagesSent("ping"); - } catch (err) { - yield* closeOnSendFailure(ws, err, "ping"); - } + yield* sendImmediate(keepalive.pingFrame, "ping"); }); const cachedStart = yield* Effect.cached( @@ -333,5 +359,5 @@ export const createVenueWS = ( ); const start = () => cachedStart; - return { start, subscribe, unsubscribe, hasError } satisfies VenueWS; + return { start, subscribe, unsubscribe, hasError } satisfies VenueWS; }); diff --git a/workspace/data-proxy/src/modules/ticker-module/README.md b/workspace/data-proxy/src/modules/ticker-module/README.md index 1e32cfa9..1a732edf 100644 --- a/workspace/data-proxy/src/modules/ticker-module/README.md +++ b/workspace/data-proxy/src/modules/ticker-module/README.md @@ -1,22 +1,23 @@ # Ticker module -Shared HTTP + cache + idle-cleanup loop for string-keyed public ticker venues. Each module supplies only its WebSocket protocol; request handling, price caching, and subscription lifecycle are the same. +Shared HTTP + cache + idle-cleanup loop for public ticker venues. Each module supplies only its WebSocket protocol and a `parseKey` that turns request tokens into cache keys; request handling, price caching, and subscription lifecycle are the same. ## Overview On startup each venue module: 1. Connects to `wsUrl`. -2. Subscribes to every symbol in `subscriptionSymbols`. -3. Caches inbound ticker frames by uppercased symbol. -4. Sets up an idle cleanup loop to unsubscribe symbols that have not been requested within `symbolsCleanupTtl`. +2. Subscribes to every parsed key in `subscriptionSymbols`. +3. Caches inbound ticker frames by key. +4. Sets up an idle cleanup loop to unsubscribe keys that have not been requested within `symbolsCleanupTtl`. For HTTP requests the handler: -1. Resolves `fetchFromModule` to one or more comma-separated symbols. +1. Resolves `fetchFromModule` to one or more comma-separated tokens. 2. Rejects the request with HTTP 400 if the count exceeds `maxSymbolsPerRequest`. -3. Subscribes to any new symbols over WebSocket. -4. Returns the latest cached frame for each symbol. If a price is not yet available, the handler waits briefly (shared price-cache timeout: 3 seconds). +3. Parses tokens into keys (`parseKey`); invalid tokens miss immediately. +4. Subscribes to any new keys over WebSocket. +5. Returns the latest cached frame for each token. If a price is not yet available, the handler waits briefly (shared price-cache timeout: 3 seconds). While the socket is disconnected or errored, every item is returned with `__sedaHasPrice: false` even if a stale frame is still in cache. @@ -28,10 +29,10 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). | Field | Required | Default | Description | | --- | --- | --- | --- | -| `type` | yes | — | `"binance"`, `"bybit"`, or `"okx"` | +| `type` | yes | — | `"binance"`, `"bybit"`, `"okx"`, or `"lighter"` | | `name` | yes | — | Module name referenced by routes as `moduleName`. | | `wsUrl` | no | venue default (below) | Public WebSocket URL. | -| `subscriptionSymbols` | no | `[]` | Symbols to subscribe to on start. | +| `subscriptionSymbols` | no | `[]` | Tokens to subscribe to on start (strings; Lighter uses market ids such as `"1"`). | | `maxSymbolsPerRequest` | no | `100` | Max symbols allowed in a single request. | | `maxMessages` | no | venue default (below) | Max outbound WS control frames per `maxMessagesWindow`. | | `maxMessagesWindow` | no | venue default (below) | Rolling window for `maxMessages`. | @@ -47,12 +48,13 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). | Binance | `wss://stream.binance.com:9443/stream` | `5` / `"1 second"` | `streamType` (default `"bookTicker"`): `bookTicker`, `aggTrade`, `trade`, `ticker`, `miniTicker`. No keepalive. | | OKX | `wss://ws.okx.com:8443/ws/v5/public` | `480` / `"1 hour"` | `keepaliveInterval` (default `"20 seconds"`). | | Bybit | `wss://stream.bybit.com/v5/public/spot` | `5` / `"1 second"` | `keepaliveInterval` (default `"15 seconds"`). | +| Lighter | `wss://mainnet.zklighter.elliot.ai/stream?readonly=true` | `180` / `"1 minute"` | `keepaliveInterval` (default `"60 seconds"`). `?readonly=true` avoids geo-restriction. Keys are numeric market ids. | ### Route | Field | Required | Description | | --- | --- | --- | -| `type` | yes | Must match the module (`"binance"`, `"bybit"`, or `"okx"`). | +| `type` | yes | Must match the module (`"binance"`, `"bybit"`, `"okx"`, or `"lighter"`). | | `moduleName` | yes | Name of a configured ticker module. | | `path` | yes | Proxy path (supports `{:param}` path params). | | `method` | no | HTTP method(s); typically `GET`. | @@ -65,7 +67,8 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). "modules": [ { "type": "binance", "name": "binance" }, { "type": "okx", "name": "okx" }, - { "type": "bybit", "name": "bybit" } + { "type": "bybit", "name": "bybit" }, + { "type": "lighter", "name": "lighter" } ], "routes": [ { @@ -88,6 +91,13 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). "path": "/bybit/:symbols", "method": ["GET"], "fetchFromModule": "{:symbols}" + }, + { + "type": "lighter", + "moduleName": "lighter", + "path": "/lighter/:markets", + "method": ["GET"], + "fetchFromModule": "{:markets}" } ] } @@ -97,6 +107,7 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). curl -s "http://127.0.0.1:5384/proxy/binance/BTCUSDT,ETHUSDT" | jq . curl -s "http://127.0.0.1:5384/proxy/okx/BTC-USDT,ETH-USDT" | jq . curl -s "http://127.0.0.1:5384/proxy/bybit/BTCUSDT,ETHUSDT" | jq . +curl -s "http://127.0.0.1:5384/proxy/lighter/1,0" | jq . ``` ## Response shape @@ -152,9 +163,25 @@ Bybit (identity field `symbol`): ] ``` +Lighter (identity field `marketId`): + +```jsonc +[ + { + "s": "BTC", + "marketId": "1", + "__sedaHasPrice": true + }, + { + "marketId": "NOPE", + "__sedaHasPrice": false + } +] +``` + | Field | Present when | Description | | --- | --- | --- | -| Identity (`symbol` or `instId`) | always | The raw request token from `fetchFromModule` (original casing). | +| Identity (`symbol`, `instId`, or `marketId`) | always | The raw request token from `fetchFromModule` (original casing). | | Venue ticker fields | `__sedaHasPrice: true` | Relayed verbatim from the stream. | | `__sedaHasPrice` | always | `true` when a cached price was returned; `false` on wait timeout, miss, or unhealthy socket. | @@ -162,17 +189,16 @@ Requests with more symbols than `maxSymbolsPerRequest` return HTTP 400. ## Adding a venue -For another string-keyed public ticker feed: +For another public ticker feed: 1. Add venue config with `tickerModuleBaseFields` (and keepalive / extra fields if needed). -2. Implement `parseInboundFrame`, subscribe/unsubscribe builders, and `createWS` via `createVenueWS`. -3. Wrap with `createTickerModuleService` (`venue`, `routeType`, `identityField`, `createWS`). +2. Implement `parseInboundFrame`, subscribe/unsubscribe builders (`string | string[]` frames), and `createWS` via `createVenueWS`. +3. Wrap with `createTickerModuleService` (`venue`, `routeType`, `identityField`, `parseKey`, `createWS`). 4. Register the module in `module-config.ts` and `proxy-server.ts`. ## Notes -- Symbols are uppercased for subscribe/cache keys; the identity field on the response keeps the request token as written. -- The first request for a new symbol may wait up to 3 seconds for the first tick; a miss still returns 200 with `__sedaHasPrice: false`. - Binance docs: https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams - OKX tickers channel: https://www.okx.com/docs-v5/en/#order-book-trading-market-data-ws-tickers-channel - Bybit ticker stream: https://bybit-exchange.github.io/docs/v5/websocket/public/ticker +- Lighter: https://apidocs.lighter.xyz/docs/websocket-reference diff --git a/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts index 4383b569..d198382f 100644 --- a/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts @@ -66,23 +66,24 @@ describe("parseInboundFrame", () => { it("unwraps a combined-stream envelope", () => { const raw = JSON.stringify({ stream: "btcusdt@bookTicker", data: btcBook }); expect(parseInboundFrame(raw)).toEqual({ - kind: "ticker", - symbol: "BTCUSDT", - frame: btcBook, + kind: "tickers", + frames: [{ key: "BTCUSDT", frame: btcBook }], }); }); it("accepts a bare (raw-stream) payload", () => { expect(parseInboundFrame(JSON.stringify(ethBook))).toEqual({ - kind: "ticker", - symbol: "ETHUSDT", - frame: ethBook, + kind: "tickers", + frames: [{ key: "ETHUSDT", frame: ethBook }], }); }); it("uppercases the symbol so cache keys stay consistent", () => { const result = parseInboundFrame(JSON.stringify({ s: "btcusdt", b: "1" })); - expect(result).toMatchObject({ kind: "ticker", symbol: "BTCUSDT" }); + expect(result).toMatchObject({ + kind: "tickers", + frames: [{ key: "BTCUSDT" }], + }); }); it("returns null for a control ack with no symbol", () => { diff --git a/workspace/data-proxy/src/modules/ticker-module/binance.ts b/workspace/data-proxy/src/modules/ticker-module/binance.ts index 57a169e5..1c85385f 100644 --- a/workspace/data-proxy/src/modules/ticker-module/binance.ts +++ b/workspace/data-proxy/src/modules/ticker-module/binance.ts @@ -1,8 +1,16 @@ import type { Effect, Schedule } from "effect"; import type { BinanceModuleConfig } from "../../config/binance-module-config"; +import { isRecord, parseJsonRecord } from "../shared/json"; import type { PriceCache } from "../shared/price-cache"; -import { type VenueWS, createVenueWS } from "../shared/venue-ws"; -import { createTickerModuleService } from "./ticker-module"; +import { + type VenueParsedInbound, + type VenueWS, + createVenueWS, +} from "../shared/venue-ws"; +import { + createTickerModuleService, + parseUppercaseSymbol, +} from "./ticker-module"; export const BinanceModuleService = (config: BinanceModuleConfig) => createTickerModuleService({ @@ -10,12 +18,11 @@ export const BinanceModuleService = (config: BinanceModuleConfig) => routeType: "binance", identityField: "symbol", config, + parseKey: parseUppercaseSymbol, createWS: createBinanceWS, extraInitLog: { streamType: config.streamType }, }); -/** A raw Binance market-data payload. Always carries the symbol in `s`; the rest - * of the fields depend on the configured stream type and are relayed verbatim. */ export interface BinancePriceFrame { s: string; [key: string]: unknown; @@ -34,23 +41,11 @@ export const buildUnsubscribeFrame = ( id: number, ): string => JSON.stringify({ method: "UNSUBSCRIBE", params: streamNames, id }); -const isRecord = (value: unknown): value is Record => - typeof value === "object" && value !== null; - -export type ParsedInbound = - | { kind: "ticker"; symbol: string; frame: BinancePriceFrame } - | { kind: "error"; code: number | null; message: string | null }; - -/** Classifies an inbound message: a market-data payload, a venue error, or null - * for other messages. */ -export const parseInboundFrame = (raw: string): ParsedInbound | null => { - let json: unknown; - try { - json = JSON.parse(raw); - } catch { - return null; - } - if (!isRecord(json)) return null; +export const parseInboundFrame = ( + raw: string, +): VenueParsedInbound | null => { + const json = parseJsonRecord(raw); + if (!json) return null; const err = json.error; if (isRecord(err)) { @@ -72,9 +67,13 @@ export const parseInboundFrame = (raw: string): ParsedInbound | null => { } return { - kind: "ticker", - symbol: payload.s.toUpperCase(), - frame: payload as BinancePriceFrame, + kind: "tickers", + frames: [ + { + key: payload.s.toUpperCase(), + frame: payload as BinancePriceFrame, + }, + ], }; }; @@ -97,14 +96,6 @@ export const createBinanceWS = ( buildSubscribeFrame(streamNamesFor(keys), nextControlId()), buildUnsubscribeFrame: (keys) => buildUnsubscribeFrame(streamNamesFor(keys), nextControlId()), - parseInboundFrame: (raw) => { - const parsed = parseInboundFrame(raw); - if (!parsed) return null; - if (parsed.kind === "error") return parsed; - return { - kind: "tickers", - frames: [{ key: parsed.symbol, frame: parsed.frame }], - }; - }, + parseInboundFrame, }); }; diff --git a/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts index 24032dcc..3e20a786 100644 --- a/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts @@ -71,7 +71,7 @@ describe("parseInboundFrame", () => { it("extracts tickers from a snapshot payload", () => { expect(parseInboundFrame(tickerMessage(btcTicker))).toEqual({ kind: "tickers", - frames: [{ symbol: "BTCUSDT", frame: btcTicker }], + frames: [{ key: "BTCUSDT", frame: btcTicker }], }); }); @@ -81,7 +81,7 @@ describe("parseInboundFrame", () => { ); expect(result).toMatchObject({ kind: "tickers", - frames: [{ symbol: "BTCUSDT" }], + frames: [{ key: "BTCUSDT" }], }); }); @@ -98,8 +98,8 @@ describe("parseInboundFrame", () => { ).toEqual({ kind: "tickers", frames: [ - { symbol: "BTCUSDT", frame: btcTicker }, - { symbol: "ETHUSDT", frame: ethTicker }, + { key: "BTCUSDT", frame: btcTicker }, + { key: "ETHUSDT", frame: ethTicker }, ], }); }); @@ -117,7 +117,7 @@ describe("parseInboundFrame", () => { ), ).toEqual({ kind: "tickers", - frames: [{ symbol: "BTCUSDT", frame: { ...rest, symbol: "BTCUSDT" } }], + frames: [{ key: "BTCUSDT", frame: { ...rest, symbol: "BTCUSDT" } }], }); }); @@ -133,7 +133,7 @@ describe("parseInboundFrame", () => { kind: "tickers", frames: [ { - symbol: "BTCUSDT", + key: "BTCUSDT", frame: { symbol: "BTCUSDT", lastPrice: "77000" }, }, ], diff --git a/workspace/data-proxy/src/modules/ticker-module/bybit.ts b/workspace/data-proxy/src/modules/ticker-module/bybit.ts index 58991eb7..4f7d5891 100644 --- a/workspace/data-proxy/src/modules/ticker-module/bybit.ts +++ b/workspace/data-proxy/src/modules/ticker-module/bybit.ts @@ -1,8 +1,16 @@ import type { Effect, Schedule } from "effect"; import type { BybitModuleConfig } from "../../config/bybit-module-config"; +import { mapDataItems, parseJsonRecord } from "../shared/json"; import type { PriceCache } from "../shared/price-cache"; -import { type VenueWS, createVenueWS } from "../shared/venue-ws"; -import { createTickerModuleService } from "./ticker-module"; +import { + type VenueParsedInbound, + type VenueWS, + createVenueWS, +} from "../shared/venue-ws"; +import { + createTickerModuleService, + parseUppercaseSymbol, +} from "./ticker-module"; export const TICKERS_TOPIC_PREFIX = "tickers."; export const PING_FRAME = JSON.stringify({ op: "ping" }); @@ -27,12 +35,11 @@ export const BybitModuleService = (config: BybitModuleConfig) => routeType: "bybit", identityField: "symbol", config, + parseKey: parseUppercaseSymbol, createWS: createBybitWS, cacheApply: applyBybitFrame, }); -/** A raw Bybit tickers payload. Always carries `symbol`; the rest of the - * fields are relayed verbatim. */ export interface BybitPriceFrame { symbol: string; [key: string]: unknown; @@ -50,33 +57,17 @@ export const buildUnsubscribeFrame = (symbols: string[]): string => args: symbols.map((symbol) => `${TICKERS_TOPIC_PREFIX}${symbol}`), }); -const isRecord = (value: unknown): value is Record => - typeof value === "object" && value !== null; - -export type ParsedInbound = - | { kind: "pong" } - | { - kind: "tickers"; - frames: Array<{ symbol: string; frame: BybitPriceFrame }>; - } - | { kind: "error"; code: string | null; message: string | null }; - const codeFromJson = (json: Record): string | null => { if (typeof json.ret_code === "string") return json.ret_code; if (typeof json.ret_code === "number") return String(json.ret_code); return null; }; -/** Classifies an inbound message: a tickers payload, a venue error, a - * keepalive pong, or null for other control frames. */ -export const parseInboundFrame = (raw: string): ParsedInbound | null => { - let json: unknown; - try { - json = JSON.parse(raw); - } catch { - return null; - } - if (!isRecord(json)) return null; +export const parseInboundFrame = ( + raw: string, +): VenueParsedInbound | null => { + const json = parseJsonRecord(raw); + if (!json) return null; if (json.op === "pong" || json.ret_msg === "pong") { return { kind: "pong" }; @@ -96,18 +87,10 @@ export const parseInboundFrame = (raw: string): ParsedInbound | null => { } const topicSymbol = topic.slice(TICKERS_TOPIC_PREFIX.length).toUpperCase(); - const items = Array.isArray(json.data) - ? json.data - : isRecord(json.data) - ? [json.data] - : []; - - const frames: Array<{ symbol: string; frame: BybitPriceFrame }> = []; - for (const item of items) { - if (!isRecord(item)) continue; + const frames = mapDataItems(json.data, (item) => { const symbol = typeof item.symbol === "string" ? item.symbol.toUpperCase() : topicSymbol; - if (!symbol) continue; + if (!symbol) return null; const frame = { ...item, symbol } as BybitPriceFrame; Object.defineProperty(frame, frameType, { @@ -115,8 +98,8 @@ export const parseInboundFrame = (raw: string): ParsedInbound | null => { // Non-enumerable so object spread leaves frameType out of the cached ticker. enumerable: false, }); - frames.push({ symbol, frame }); - } + return { key: symbol, frame }; + }); if (frames.length === 0) return null; return { kind: "tickers", frames }; @@ -134,20 +117,9 @@ export const createBybitWS = ( reconnectSchedule, keepalive: { interval: config.keepaliveInterval, - frame: PING_FRAME, + pingFrame: PING_FRAME, }, buildSubscribeFrame, buildUnsubscribeFrame, - parseInboundFrame: (raw) => { - const parsed = parseInboundFrame(raw); - if (!parsed) return null; - if (parsed.kind === "pong" || parsed.kind === "error") return parsed; - return { - kind: "tickers", - frames: parsed.frames.map(({ symbol, frame }) => ({ - key: symbol, - frame, - })), - }; - }, + parseInboundFrame, }); diff --git a/workspace/data-proxy/src/modules/lighter/ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/lighter-ws-client.test.ts similarity index 69% rename from workspace/data-proxy/src/modules/lighter/ws-client.test.ts rename to workspace/data-proxy/src/modules/ticker-module/lighter-ws-client.test.ts index 9dd94aae..728c4143 100644 --- a/workspace/data-proxy/src/modules/lighter/ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/lighter-ws-client.test.ts @@ -4,14 +4,13 @@ import type { LighterModuleConfig } from "../../config/lighter-module-config"; import { createPriceCache } from "../shared/price-cache"; import { type LighterPriceFrame, + PING_FRAME, + PONG_FRAME, buildSubscribeFrame, buildUnsubscribeFrame, createLighterWS, parseInboundFrame, -} from "./ws-client"; - -const PING = JSON.stringify({ type: "ping" }); -const PONG = JSON.stringify({ type: "pong" }); +} from "./lighter"; const innerTicker = (symbol: string) => ({ s: symbol, @@ -36,46 +35,61 @@ const tickerMessage = ( describe("buildSubscribeFrame / buildUnsubscribeFrame", () => { it("sends the ticker channel with a slash separator", () => { - expect(JSON.parse(buildSubscribeFrame(1))).toEqual({ + expect(JSON.parse(buildSubscribeFrame(1, "ticker"))).toEqual({ type: "subscribe", channel: "ticker/1", }); - expect(JSON.parse(buildUnsubscribeFrame(42))).toEqual({ + expect(JSON.parse(buildUnsubscribeFrame(42, "ticker"))).toEqual({ type: "unsubscribe", channel: "ticker/42", }); }); + + it("uses the configured stream type in the channel", () => { + expect(JSON.parse(buildSubscribeFrame(0, "order_book"))).toEqual({ + type: "subscribe", + channel: "order_book/0", + }); + expect(JSON.parse(buildUnsubscribeFrame(0, "trade"))).toEqual({ + type: "unsubscribe", + channel: "trade/0", + }); + }); }); describe("parseInboundFrame", () => { it("extracts market id and verbatim frame from an update/ticker", () => { - expect(parseInboundFrame(tickerMessage(1, "BTC"))).toEqual({ - kind: "ticker", - marketId: 1, - frame: innerTicker("BTC"), + expect(parseInboundFrame(tickerMessage(1, "BTC"), "ticker")).toEqual({ + kind: "tickers", + frames: [{ key: 1, frame: innerTicker("BTC") }], }); }); it("treats the subscribed/ticker snapshot the same as an update", () => { const parsed = parseInboundFrame( tickerMessage(2, "ETH", "subscribed/ticker"), + "ticker", ); expect(parsed).toEqual({ - kind: "ticker", - marketId: 2, - frame: innerTicker("ETH"), + kind: "tickers", + frames: [{ key: 2, frame: innerTicker("ETH") }], }); }); it("classifies a keepalive ping", () => { - expect(parseInboundFrame(JSON.stringify({ type: "ping" }))).toEqual({ + expect( + parseInboundFrame(JSON.stringify({ type: "ping" }), "ticker"), + ).toEqual({ kind: "ping", }); }); it("returns null for the connected control frame", () => { expect( - parseInboundFrame(JSON.stringify({ session_id: "x", type: "connected" })), + parseInboundFrame( + JSON.stringify({ session_id: "x", type: "connected" }), + "ticker", + ), ).toBeNull(); }); @@ -83,6 +97,7 @@ describe("parseInboundFrame", () => { expect( parseInboundFrame( JSON.stringify({ error: { code: 30005, message: "Invalid Channel" } }), + "ticker", ), ).toEqual({ kind: "error", @@ -97,6 +112,7 @@ describe("parseInboundFrame", () => { JSON.stringify({ error: { code: 30010, message: "Too Many Inflight Messages!" }, }), + "ticker", ), ).toEqual({ kind: "error", @@ -106,7 +122,50 @@ describe("parseInboundFrame", () => { }); it("returns null for malformed JSON", () => { - expect(parseInboundFrame("not json")).toBeNull(); + expect(parseInboundFrame("not json", "ticker")).toBeNull(); + }); + + it("extracts an order_book payload when that stream is configured", () => { + const orderBook = { + code: 0, + asks: [{ price: "2064.54", size: "0.3285" }], + bids: [{ price: "2064.53", size: "1.0" }], + }; + expect( + parseInboundFrame( + JSON.stringify({ + channel: "order_book:0", + order_book: orderBook, + type: "update/order_book", + }), + "order_book", + ), + ).toEqual({ + kind: "tickers", + frames: [{ key: 0, frame: orderBook }], + }); + }); + + it("extracts trade arrays when that stream is configured", () => { + const trades = [{ trade_id: 1, price: "2181.83", size: "0.1336" }]; + expect( + parseInboundFrame( + JSON.stringify({ + channel: "trade:0", + trades, + liquidation_trades: [], + type: "update/trade", + }), + "trade", + ), + ).toEqual({ + kind: "tickers", + frames: [{ key: 0, frame: { trades, liquidation_trades: [] } }], + }); + }); + + it("returns null when the channel prefix does not match the stream type", () => { + expect(parseInboundFrame(tickerMessage(1, "BTC"), "order_book")).toBeNull(); }); }); @@ -168,14 +227,16 @@ const baseConfig: LighterModuleConfig = { name: "lighter", type: "lighter", wsUrl: "wss://lighter.test/stream", - subscriptionMarketIds: [], - maxMarketsPerRequest: 100, - maxMessagesPerMinute: 180, + subscriptionSymbols: [], + maxSymbolsPerRequest: 100, + maxMessages: 180, + maxMessagesWindow: Duration.minutes(1), keepaliveInterval: Duration.seconds(60), reconnectMaxBackoff: Duration.seconds(30), reconnectStableThreshold: Duration.seconds(30), - marketsCleanupTtl: Duration.hours(1), - marketsCleanupInterval: Duration.seconds(30), + symbolsCleanupTtl: Duration.hours(1), + symbolsCleanupInterval: Duration.seconds(30), + streamType: "ticker", }; const originalWebSocket = globalThis.WebSocket; @@ -193,14 +254,14 @@ afterEach(() => { const startService = ( config: LighterModuleConfig, preSubscribed: number[] = [], - options?: Parameters[2], + reconnectSchedule?: Schedule.Schedule, ) => Effect.gen(function* () { const cache = yield* createPriceCache(); const ws = yield* createLighterWS( config, cache, - options ?? { reconnectSchedule: Schedule.spaced(Duration.minutes(10)) }, + reconnectSchedule ?? Schedule.spaced(Duration.minutes(10)), ); if (preSubscribed.length > 0) { yield* ws.subscribe(preSubscribed); @@ -221,7 +282,10 @@ describe("createLighterWS", () => { ws.triggerOpen(); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1), buildSubscribeFrame(2)]); + expect(ws.sent).toEqual([ + buildSubscribeFrame(1, "ticker"), + buildSubscribeFrame(2, "ticker"), + ]); await Effect.runPromise(Fiber.interrupt(fiber)); }); @@ -273,7 +337,7 @@ describe("createLighterWS", () => { ws.triggerMessage(JSON.stringify({ type: "ping" })); await flush(); - expect(ws.sent).toEqual([PONG]); + expect(ws.sent).toEqual([PONG_FRAME]); await Effect.runPromise(Fiber.interrupt(fiber)); }); @@ -293,7 +357,9 @@ describe("createLighterWS", () => { await new Promise((r) => setTimeout(r, 50)); - expect(ws.sent.filter((frame) => frame === PING).length).toBeGreaterThan(0); + expect( + ws.sent.filter((frame) => frame === PING_FRAME).length, + ).toBeGreaterThan(0); await Effect.runPromise(Fiber.interrupt(fiber)); }); @@ -306,13 +372,13 @@ describe("createLighterWS", () => { const ws = FakeWebSocket.instances[0]; ws.triggerOpen(); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1)]); + expect(ws.sent).toEqual([buildSubscribeFrame(1, "ticker")]); await Effect.runPromise(service.subscribe([1])); await Effect.runPromise(service.subscribe([1])); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1)]); + expect(ws.sent).toEqual([buildSubscribeFrame(1, "ticker")]); await Effect.runPromise(Fiber.interrupt(fiber)); }); @@ -326,34 +392,39 @@ describe("createLighterWS", () => { ws.triggerOpen(); await flush(); - // Unknown market id: no frame. await Effect.runPromise(service.unsubscribe([2])); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1)]); + expect(ws.sent).toEqual([buildSubscribeFrame(1, "ticker")]); await Effect.runPromise(service.unsubscribe([1])); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1), buildUnsubscribeFrame(1)]); + expect(ws.sent).toEqual([ + buildSubscribeFrame(1, "ticker"), + buildUnsubscribeFrame(1, "ticker"), + ]); - // Repeat: already removed, no frame. await Effect.runPromise(service.unsubscribe([1])); await flush(); - expect(ws.sent).toEqual([buildSubscribeFrame(1), buildUnsubscribeFrame(1)]); + expect(ws.sent).toEqual([ + buildSubscribeFrame(1, "ticker"), + buildUnsubscribeFrame(1, "ticker"), + ]); await Effect.runPromise(Fiber.interrupt(fiber)); }); it("reconnects after a close and re-subscribes every desired market", async () => { const { fiber } = await Effect.runPromise( - startService(baseConfig, [1, 2], { - reconnectSchedule: Schedule.spaced(Duration.millis(10)), - }), + startService(baseConfig, [1, 2], Schedule.spaced(Duration.millis(10))), ); await flush(); const ws1 = FakeWebSocket.instances[0]; ws1.triggerOpen(); await flush(); - expect(ws1.sent).toEqual([buildSubscribeFrame(1), buildSubscribeFrame(2)]); + expect(ws1.sent).toEqual([ + buildSubscribeFrame(1, "ticker"), + buildSubscribeFrame(2, "ticker"), + ]); ws1.triggerClose(); await new Promise((r) => setTimeout(r, 40)); @@ -364,7 +435,10 @@ describe("createLighterWS", () => { ws2.triggerOpen(); await flush(); - expect(ws2.sent).toEqual([buildSubscribeFrame(1), buildSubscribeFrame(2)]); + expect(ws2.sent).toEqual([ + buildSubscribeFrame(1, "ticker"), + buildSubscribeFrame(2, "ticker"), + ]); await Effect.runPromise(Fiber.interrupt(fiber)); }); @@ -378,9 +452,7 @@ describe("createLighterWS", () => { }; const { fiber } = await Effect.runPromise( - startService(baseConfig, [1], { - reconnectSchedule: Schedule.spaced(Duration.millis(10)), - }), + startService(baseConfig, [1], Schedule.spaced(Duration.millis(10))), ); await flush(); const ws1 = FakeWebSocket.instances[0]; @@ -393,15 +465,15 @@ describe("createLighterWS", () => { ws2.triggerOpen(); await flush(); - expect(ws2.sent).toEqual([buildSubscribeFrame(1)]); + expect(ws2.sent).toEqual([buildSubscribeFrame(1, "ticker")]); await Effect.runPromise(Fiber.interrupt(fiber)); }); - it("paces outbound frames to stay under maxMessagesPerMinute", async () => { + it("paces outbound frames to stay under maxMessages per maxMessagesWindow", async () => { const { fiber } = await Effect.runPromise( startService( - { ...baseConfig, maxMessagesPerMinute: 6 }, + { ...baseConfig, maxMessages: 6 }, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], ), ); @@ -411,21 +483,21 @@ describe("createLighterWS", () => { await flush(); expect(ws.sent).toEqual([ - buildSubscribeFrame(1), - buildSubscribeFrame(2), - buildSubscribeFrame(3), - buildSubscribeFrame(4), - buildSubscribeFrame(5), - buildSubscribeFrame(6), + buildSubscribeFrame(1, "ticker"), + buildSubscribeFrame(2, "ticker"), + buildSubscribeFrame(3, "ticker"), + buildSubscribeFrame(4, "ticker"), + buildSubscribeFrame(5, "ticker"), + buildSubscribeFrame(6, "ticker"), ]); await new Promise((r) => setTimeout(r, 50)); expect(ws.sent).toEqual([ - buildSubscribeFrame(1), - buildSubscribeFrame(2), - buildSubscribeFrame(3), - buildSubscribeFrame(4), - buildSubscribeFrame(5), - buildSubscribeFrame(6), + buildSubscribeFrame(1, "ticker"), + buildSubscribeFrame(2, "ticker"), + buildSubscribeFrame(3, "ticker"), + buildSubscribeFrame(4, "ticker"), + buildSubscribeFrame(5, "ticker"), + buildSubscribeFrame(6, "ticker"), ]); await Effect.runPromise(Fiber.interrupt(fiber)); diff --git a/workspace/data-proxy/src/modules/lighter/lighter.test.ts b/workspace/data-proxy/src/modules/ticker-module/lighter.test.ts similarity index 82% rename from workspace/data-proxy/src/modules/lighter/lighter.test.ts rename to workspace/data-proxy/src/modules/ticker-module/lighter.test.ts index 116ac9ba..2965704d 100644 --- a/workspace/data-proxy/src/modules/lighter/lighter.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/lighter.test.ts @@ -4,8 +4,7 @@ import type { Route } from "../../config/config-parser"; import type { LighterModuleConfig } from "../../config/lighter-module-config"; import { HAS_PRICE_KEY } from "../../constants"; import { ModuleService } from "../module"; -import { LighterModuleService } from "./lighter"; -import { buildSubscribeFrame } from "./ws-client"; +import { LighterModuleService, buildSubscribeFrame } from "./lighter"; const innerTicker = (symbol: string) => ({ s: symbol, @@ -72,14 +71,16 @@ const baseConfig: LighterModuleConfig = { name: "lighter", type: "lighter", wsUrl: "wss://lighter.test/stream", - subscriptionMarketIds: [], - maxMarketsPerRequest: 100, - maxMessagesPerMinute: 180, + subscriptionSymbols: [], + maxSymbolsPerRequest: 100, + maxMessages: 180, + maxMessagesWindow: Duration.minutes(1), keepaliveInterval: Duration.seconds(60), reconnectMaxBackoff: Duration.seconds(30), reconnectStableThreshold: Duration.seconds(30), - marketsCleanupTtl: Duration.hours(1), - marketsCleanupInterval: Duration.seconds(30), + symbolsCleanupTtl: Duration.hours(1), + symbolsCleanupInterval: Duration.seconds(30), + streamType: "ticker", }; const originalWebSocket = globalThis.WebSocket; @@ -105,7 +106,7 @@ describe("LighterModuleService", () => { it("subscribes seeded markets, caches a delivered ticker, and serves it in request order", async () => { const service = await buildService({ ...baseConfig, - subscriptionMarketIds: [1], + subscriptionSymbols: ["1"], }); await Effect.runPromise(quiet(service.start())); await flush(); @@ -114,8 +115,7 @@ describe("LighterModuleService", () => { ws.triggerOpen(); await flush(); - // The seeded market's subscribe frame actually went out. - expect(ws.sent).toContain(buildSubscribeFrame(1)); + expect(ws.sent).toContain(buildSubscribeFrame(1, "ticker")); ws.triggerMessage(tickerMessage(1, "BTC")); await flush(); @@ -124,7 +124,6 @@ describe("LighterModuleService", () => { service.handleRequest(routeFor("1,NOPE"), {}, new Request("http://x")), ); expect(response.status).toBe(200); - // "1" resolves to a price; "NOPE" is not a market id, so it short-circuits to a miss. expect(await response.json()).toEqual([ { marketId: "1", ...innerTicker("BTC"), [HAS_PRICE_KEY]: true }, { marketId: "NOPE", [HAS_PRICE_KEY]: false }, @@ -139,13 +138,11 @@ describe("LighterModuleService", () => { ws.triggerOpen(); await flush(); - // Market 2 was not seeded; the request itself drives the subscription, then - // waits on the cache. Deliver the ticker while that wait is in flight. const responsePromise = Effect.runPromise( service.handleRequest(routeFor("2"), {}, new Request("http://x")), ); await flush(); - expect(ws.sent).toContain(buildSubscribeFrame(2)); + expect(ws.sent).toContain(buildSubscribeFrame(2, "ticker")); ws.triggerMessage(tickerMessage(2, "ETH")); const response = await responsePromise; @@ -158,7 +155,7 @@ describe("LighterModuleService", () => { it("stops vouching for cached prices once the socket has errored", async () => { const service = await buildService({ ...baseConfig, - subscriptionMarketIds: [1], + subscriptionSymbols: ["1"], }); await Effect.runPromise(quiet(service.start())); await flush(); @@ -173,24 +170,21 @@ describe("LighterModuleService", () => { ); expect((await fresh.json())[0][HAS_PRICE_KEY]).toBe(true); - // Socket drops; the ws-client reports hasError until it reconnects. ws.close(); await flush(); const afterError = await Effect.runPromise( service.handleRequest(routeFor("1"), {}, new Request("http://x")), ); - // Market 1 is still cached, but the unhealthy socket means it is no longer - // presented as a live price. expect(await afterError.json()).toEqual([ { marketId: "1", [HAS_PRICE_KEY]: false }, ]); }); - it("rejects a request over maxMarketsPerRequest with 400", async () => { + it("rejects a request over maxSymbolsPerRequest with 400", async () => { const service = await buildService({ ...baseConfig, - maxMarketsPerRequest: 1, + maxSymbolsPerRequest: 1, }); const response = await Effect.runPromise( diff --git a/workspace/data-proxy/src/modules/ticker-module/lighter.ts b/workspace/data-proxy/src/modules/ticker-module/lighter.ts new file mode 100644 index 00000000..7d1aa032 --- /dev/null +++ b/workspace/data-proxy/src/modules/ticker-module/lighter.ts @@ -0,0 +1,153 @@ +import { type Effect, Match, type Schedule } from "effect"; +import type { + LighterModuleConfig, + LighterStreamType, +} from "../../config/lighter-module-config"; +import { isRecord, parseJsonRecord } from "../shared/json"; +import type { PriceCache } from "../shared/price-cache"; +import { + type VenueParsedInbound, + type VenueWS, + createVenueWS, +} from "../shared/venue-ws"; +import { createTickerModuleService } from "./ticker-module"; + +export const PING_FRAME = JSON.stringify({ type: "ping" }); +export const PONG_FRAME = JSON.stringify({ type: "pong" }); + +export const LighterModuleService = (config: LighterModuleConfig) => + createTickerModuleService({ + venue: "lighter", + routeType: "lighter", + identityField: "marketId", + config, + parseKey: parseMarketId, + createWS: createLighterWS, + extraInitLog: { streamType: config.streamType }, + }); + +export interface LighterPriceFrame { + [key: string]: unknown; +} + +export const buildSubscribeFrame = ( + marketId: number, + streamType: LighterStreamType, +): string => + JSON.stringify({ type: "subscribe", channel: `${streamType}/${marketId}` }); + +export const buildUnsubscribeFrame = ( + marketId: number, + streamType: LighterStreamType, +): string => + JSON.stringify({ + type: "unsubscribe", + channel: `${streamType}/${marketId}`, + }); + +export const parseMarketId = (token: string): number | null => { + const id = Number(token); + return Number.isInteger(id) && id >= 0 ? id : null; +}; + +/** The inbound `channel` is `{streamType}:{id}` even though subscribe sends + * `{streamType}/{id}`. Accept either separator. */ +const parseMarketIdFromChannel = ( + channel: unknown, + streamType: LighterStreamType, +): number | null => { + if (typeof channel !== "string") return null; + const [prefix, idPart, ...rest] = channel.split(/[:/]/); + if (rest.length > 0 || prefix !== streamType || idPart === undefined) { + return null; + } + const id = Number(idPart); + return Number.isInteger(id) ? id : null; +}; + +type ReadFrame = (json: Record) => LighterPriceFrame | null; + +const readTradeFrame: ReadFrame = (json) => { + if (!Array.isArray(json.trades)) return null; + const frame: LighterPriceFrame = { trades: json.trades }; + if (Array.isArray(json.liquidation_trades)) { + frame.liquidation_trades = json.liquidation_trades; + } + return frame; +}; + +const readRecordFrame = + (kind: "ticker" | "order_book"): ReadFrame => + (json) => { + const payload = json[kind]; + if (!isRecord(payload)) return null; + return payload; + }; + +/** Stream type is fixed for a connection, so the frame reader is chosen once. */ +const streamHandlers = (streamType: LighterStreamType) => { + const readFrame = Match.value(streamType).pipe( + Match.when("trade", () => readTradeFrame), + Match.whenOr("ticker", "order_book", (kind) => readRecordFrame(kind)), + Match.exhaustive, + ); + + const parseFrame = ( + raw: string, + ): VenueParsedInbound | null => { + const json = parseJsonRecord(raw); + if (!json) return null; + if (json.type === "ping") return { kind: "ping" }; + + const err = json.error; + if (isRecord(err)) { + return { + kind: "error", + code: typeof err.code === "number" ? err.code : null, + message: typeof err.message === "string" ? err.message : null, + }; + } + + const marketId = parseMarketIdFromChannel(json.channel, streamType); + if (marketId === null) return null; + const frame = readFrame(json); + if (frame === null) return null; + return { kind: "tickers", frames: [{ key: marketId, frame }] }; + }; + + return { + buildSubscribeFrame: (keys: number[]) => + keys.map((marketId) => buildSubscribeFrame(marketId, streamType)), + buildUnsubscribeFrame: (keys: number[]) => + keys.map((marketId) => buildUnsubscribeFrame(marketId, streamType)), + parseInboundFrame: parseFrame, + }; +}; + +export const parseInboundFrame = ( + raw: string, + streamType: LighterStreamType, +): VenueParsedInbound | null => + streamHandlers(streamType).parseInboundFrame(raw); + +export const createLighterWS = ( + config: LighterModuleConfig, + cache: PriceCache, + reconnectSchedule?: Schedule.Schedule, +): Effect.Effect, never, never> => { + const handlers = streamHandlers(config.streamType); + return createVenueWS({ + venue: "lighter", + config, + cache, + reconnectSchedule, + keepalive: { + interval: config.keepaliveInterval, + pingFrame: PING_FRAME, + pongFrame: PONG_FRAME, + }, + buildSubscribeFrame: handlers.buildSubscribeFrame, + buildUnsubscribeFrame: handlers.buildUnsubscribeFrame, + parseInboundFrame: handlers.parseInboundFrame, + }); +}; diff --git a/workspace/data-proxy/src/modules/ticker-module/okx-ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/okx-ws-client.test.ts index f2d003b1..0bb392ee 100644 --- a/workspace/data-proxy/src/modules/ticker-module/okx-ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/okx-ws-client.test.ts @@ -82,7 +82,7 @@ describe("parseInboundFrame", () => { it("extracts tickers from a push payload", () => { expect(parseInboundFrame(tickerMessage(btcTicker))).toEqual({ kind: "tickers", - frames: [{ instId: "BTC-USDT", frame: btcTicker }], + frames: [{ key: "BTC-USDT", frame: btcTicker }], }); }); @@ -92,7 +92,7 @@ describe("parseInboundFrame", () => { ); expect(result).toMatchObject({ kind: "tickers", - frames: [{ instId: "BTC-USDT" }], + frames: [{ key: "BTC-USDT" }], }); }); @@ -107,8 +107,8 @@ describe("parseInboundFrame", () => { ).toEqual({ kind: "tickers", frames: [ - { instId: "BTC-USDT", frame: btcTicker }, - { instId: "ETH-USDT", frame: ethTicker }, + { key: "BTC-USDT", frame: btcTicker }, + { key: "ETH-USDT", frame: ethTicker }, ], }); }); diff --git a/workspace/data-proxy/src/modules/ticker-module/okx.ts b/workspace/data-proxy/src/modules/ticker-module/okx.ts index 663de374..1cbe6717 100644 --- a/workspace/data-proxy/src/modules/ticker-module/okx.ts +++ b/workspace/data-proxy/src/modules/ticker-module/okx.ts @@ -1,9 +1,13 @@ import type { Effect, Schedule } from "effect"; import type { OkxModuleConfig } from "../../config/okx-module-config"; +import { isRecord, mapDataItems, parseJsonRecord } from "../shared/json"; import type { PriceCache } from "../shared/price-cache"; import { createVenueWS } from "../shared/venue-ws"; -import type { VenueWS } from "../shared/venue-ws"; -import { createTickerModuleService } from "./ticker-module"; +import type { VenueParsedInbound, VenueWS } from "../shared/venue-ws"; +import { + createTickerModuleService, + parseUppercaseSymbol, +} from "./ticker-module"; export const TICKERS_CHANNEL = "tickers"; export const PING_FRAME = "ping"; @@ -15,11 +19,10 @@ export const OkxModuleService = (config: OkxModuleConfig) => routeType: "okx", identityField: "instId", config, + parseKey: parseUppercaseSymbol, createWS: createOkxWS, }); -/** A raw OKX tickers-channel payload. Always carries `instId`; the rest of the - * fields are relayed verbatim. */ export interface OkxPriceFrame { instId: string; [key: string]: unknown; @@ -45,24 +48,13 @@ export const buildUnsubscribeFrame = (instIds: string[], id: string): string => })), }); -const isRecord = (value: unknown): value is Record => - typeof value === "object" && value !== null; - -export type ParsedInbound = - | { kind: "pong" } - | { kind: "tickers"; frames: Array<{ instId: string; frame: OkxPriceFrame }> } - | { kind: "error"; code: string | null; message: string | null }; - -export const parseInboundFrame = (raw: string): ParsedInbound | null => { +export const parseInboundFrame = ( + raw: string, +): VenueParsedInbound | null => { if (raw === PONG_FRAME) return { kind: "pong" }; - let json: unknown; - try { - json = JSON.parse(raw); - } catch { - return null; - } - if (!isRecord(json)) return null; + const json = parseJsonRecord(raw); + if (!json) return null; if (json.event === "error" || json.event === "notice") { return { @@ -73,23 +65,21 @@ export const parseInboundFrame = (raw: string): ParsedInbound | null => { } const arg = json.arg; - const data = json.data; if ( !isRecord(arg) || arg.channel !== TICKERS_CHANNEL || - !Array.isArray(data) + !Array.isArray(json.data) ) { return null; } - const frames: Array<{ instId: string; frame: OkxPriceFrame }> = []; - for (const item of data) { - if (!isRecord(item) || typeof item.instId !== "string") continue; - frames.push({ - instId: item.instId.toUpperCase(), + const frames = mapDataItems(json.data, (item) => { + if (typeof item.instId !== "string") return null; + return { + key: item.instId.toUpperCase(), frame: item as OkxPriceFrame, - }); - } + }; + }); if (frames.length === 0) return null; return { kind: "tickers", frames }; @@ -110,22 +100,11 @@ export const createOkxWS = ( reconnectSchedule, keepalive: { interval: config.keepaliveInterval, - frame: PING_FRAME, + pingFrame: PING_FRAME, }, buildSubscribeFrame: (keys) => buildSubscribeFrame(keys, nextControlId()), buildUnsubscribeFrame: (keys) => buildUnsubscribeFrame(keys, nextControlId()), - parseInboundFrame: (raw) => { - const parsed = parseInboundFrame(raw); - if (!parsed) return null; - if (parsed.kind === "pong" || parsed.kind === "error") return parsed; - return { - kind: "tickers", - frames: parsed.frames.map(({ instId, frame }) => ({ - key: instId, - frame, - })), - }; - }, + parseInboundFrame, }); }; 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 f32498e1..19ef073b 100644 --- a/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts +++ b/workspace/data-proxy/src/modules/ticker-module/ticker-module.ts @@ -18,7 +18,11 @@ export interface TickerModuleServiceConfig { symbolsCleanupInterval: Duration.Duration; } +export const parseUppercaseSymbol = (token: string): string => + token.toUpperCase(); + export interface CreateTickerModuleServiceParams< + TKey, TFrame, TConfig extends TickerModuleServiceConfig, > { @@ -26,10 +30,11 @@ export interface CreateTickerModuleServiceParams< routeType: Route["type"]; identityField: string; config: TConfig; + parseKey: (token: string) => TKey | null; createWS: ( config: TConfig, - cache: PriceCache, - ) => Effect.Effect; + cache: PriceCache, + ) => Effect.Effect, never, never>; cacheApply?: (prev: TFrame | undefined, next: TFrame) => TFrame; extraInitLog?: Record; } @@ -38,14 +43,16 @@ const titleCase = (venue: string) => `${venue.charAt(0).toUpperCase()}${venue.slice(1)}`; /** - * HTTP + cache + idle-cleanup loop shared by string-keyed public ticker - * modules. Venue protocol stays in each createWS. + * HTTP + cache + idle-cleanup loop shared by public ticker modules. + * Venue protocol stays in each createWS; parseKey turns request tokens into + * cache/subscribe keys. */ export const createTickerModuleService = < + TKey, TFrame, TConfig extends TickerModuleServiceConfig, >( - params: CreateTickerModuleServiceParams, + params: CreateTickerModuleServiceParams, ) => Layer.effect( ModuleService, @@ -55,6 +62,7 @@ export const createTickerModuleService = < routeType, identityField, config, + parseKey, createWS, cacheApply, extraInitLog, @@ -66,11 +74,11 @@ export const createTickerModuleService = < ...extraInitLog, }); - const cache = yield* createPriceCache({ + const cache = yield* createPriceCache({ apply: cacheApply, }); const ws = yield* createWS(config, cache); - const lastRequestToSymbol = MutableHashMap.empty(); + const lastRequestToKey = MutableHashMap.empty(); const start = () => Effect.gen(function* () { @@ -82,24 +90,29 @@ export const createTickerModuleService = < if (config.subscriptionSymbols.length > 0) { const now = yield* Clock.currentTimeMillis; - const seeded = config.subscriptionSymbols.map((symbol) => - symbol.toUpperCase(), - ); - for (const symbol of seeded) { - MutableHashMap.set(lastRequestToSymbol, symbol, now); + const seeded: TKey[] = []; + for (const token of config.subscriptionSymbols) { + const key = parseKey(token); + if (key === null) continue; + if (!MutableHashMap.has(lastRequestToKey, key)) { + seeded.push(key); + } + MutableHashMap.set(lastRequestToKey, key, now); + } + if (seeded.length > 0) { + yield* ws.subscribe(seeded); } - yield* ws.subscribe(seeded); } yield* forkIdleCleanup({ - lastRequest: lastRequestToSymbol, + lastRequest: lastRequestToKey, ttl: config.symbolsCleanupTtl, interval: config.symbolsCleanupInterval, - onExpire: (symbol) => + onExpire: (key) => Effect.gen(function* () { - yield* Effect.logInfo("Cleaning up idle symbol", { symbol }); - yield* cache.deletePrice(symbol); - yield* ws.unsubscribe([symbol]); + yield* Effect.logInfo("Cleaning up idle key", { key }); + yield* cache.deletePrice(key); + yield* ws.unsubscribe([key]); }), }); }).pipe(Effect.annotateLogs("_name", venue)); @@ -118,59 +131,65 @@ export const createTickerModuleService = < ); } - const requestedSymbols = replaceParams( + const requestedTokens = replaceParams( route.fetchFromModule, routeParams, ) .split(",") - .map((symbol) => symbol.trim()) - .filter((symbol) => symbol.length > 0); + .map((token) => token.trim()) + .filter((token) => token.length > 0); - if (requestedSymbols.length > config.maxSymbolsPerRequest) { + if (requestedTokens.length > config.maxSymbolsPerRequest) { return yield* Effect.fail( new FailedToHandleTickerRequestError({ - error: `Too many symbols, max is ${config.maxSymbolsPerRequest} but got ${requestedSymbols.length}`, + error: `Too many symbols, max is ${config.maxSymbolsPerRequest} but got ${requestedTokens.length}`, status: 400, moduleName: config.name, }), ); } + const requested = requestedTokens.map((token) => ({ + token, + key: parseKey(token), + })); + const now = yield* Clock.currentTimeMillis; const socketHealthy = !(yield* ws.hasError()); - const newSymbols: string[] = []; - for (const requested of requestedSymbols) { - const symbol = requested.toUpperCase(); - if (!MutableHashMap.has(lastRequestToSymbol, symbol)) { - newSymbols.push(symbol); + const newKeys: TKey[] = []; + for (const { key } of requested) { + if (key === null) continue; + if (!MutableHashMap.has(lastRequestToKey, key)) { + newKeys.push(key); } - MutableHashMap.set(lastRequestToSymbol, symbol, now); + MutableHashMap.set(lastRequestToKey, key, now); } - if (newSymbols.length > 0) { - yield* ws.subscribe(newSymbols); + if (newKeys.length > 0) { + yield* ws.subscribe(newKeys); } const results = yield* Effect.forEach( - requestedSymbols, - (requested) => cache.getOrWaitPrice(requested.toUpperCase()), + requested, + ({ key }) => + key === null ? Effect.succeed(null) : cache.getOrWaitPrice(key), { concurrency: "unbounded" }, ); const prices: Array> = []; - for (let i = 0; i < requestedSymbols.length; i++) { - const requested = requestedSymbols[i]; + for (let i = 0; i < requested.length; i++) { + const { token } = requested[i]; const result = results[i]; if (result === null || !socketHealthy) { prices.push({ - [identityField]: requested, + [identityField]: token, [HAS_PRICE_KEY]: false, }); } else { prices.push({ ...result, - [identityField]: requested, + [identityField]: token, [HAS_PRICE_KEY]: true, }); } diff --git a/workspace/data-proxy/src/proxy-server.ts b/workspace/data-proxy/src/proxy-server.ts index 38e4ff0a..18fe2c9d 100644 --- a/workspace/data-proxy/src/proxy-server.ts +++ b/workspace/data-proxy/src/proxy-server.ts @@ -12,13 +12,13 @@ import { handleProxyRequest } from "./controllers/proxy/handle-proxy-request"; import { ChainlinkStreamsModuleService } from "./modules/chainlink-streams/chainlink-streams"; import { DxFeedModuleService } from "./modules/dxfeed/dxfeed"; import { HydromancerModuleService } from "./modules/hydromancer/hydromancer"; -import { LighterModuleService } from "./modules/lighter/lighter"; import { LoTechModuleService } from "./modules/lo-tech/lo-tech"; import { type ModuleHandlers, ModuleService } from "./modules/module"; import { PmInsightsModuleService } from "./modules/pm-insights/pm-insights"; import { PythLazerModuleService } from "./modules/pyth-lazer/pyth-lazer"; import { BinanceModuleService } from "./modules/ticker-module/binance"; import { BybitModuleService } from "./modules/ticker-module/bybit"; +import { LighterModuleService } from "./modules/ticker-module/lighter"; import { OkxModuleService } from "./modules/ticker-module/okx"; import { VolmexModuleService } from "./modules/volmex/volmex"; import type { HttpClientService } from "./services/http-client"; diff --git a/workspace/data-proxy/src/scripts/performance.bench.ts b/workspace/data-proxy/src/scripts/performance.bench.ts index 43f4a860..a9e7ccd4 100644 --- a/workspace/data-proxy/src/scripts/performance.bench.ts +++ b/workspace/data-proxy/src/scripts/performance.bench.ts @@ -299,8 +299,8 @@ const buildConfigJson = (mock: MockVenueServer): unknown => { name: "lig", type: "lighter", wsUrl: wssUrl("lighter"), - subscriptionMarketIds: lighterMarketIds, - maxMarketsPerRequest: 200, + subscriptionSymbols: lighterMarketIds.map(String), + maxSymbolsPerRequest: 200, }); routes.push({ type: "lighter",