diff --git a/.changeset/graphql-price-subscriptions.md b/.changeset/graphql-price-subscriptions.md new file mode 100644 index 0000000..09ddf06 --- /dev/null +++ b/.changeset/graphql-price-subscriptions.md @@ -0,0 +1,5 @@ +--- +"lens": minor +--- + +Add a `priceUpdated(pair: String!)` GraphQL subscription that streams live prices over the existing `/graphql` endpoint (graphql-transport-ws protocol). Every ingester (SDEX, Horizon AMM, Soroswap) now publishes `{ pair, price, ts }` on each new price; subscribers receive only the pair they request. diff --git a/README.md b/README.md index 614bfcb..0585d75 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,8 @@ curl "https://api.example.com/price/XLM/USDC?network=mainnet" ``` ### GraphQL -Available at `/graphql` with GraphiQL IDE at `/graphiql`. +Available at `/graphql` with GraphiQL IDE at `/graphiql`. Real-time price +streaming is available via the `priceUpdated` [subscription](#graphql-subscriptions-live-prices). ```graphql query { @@ -91,6 +92,67 @@ histogram_quantile(0.95, See [`docs/http-metrics.md`](docs/http-metrics.md) for the full label reference, the bucket rationale and suggested alerting rules. +### GraphQL Subscriptions (live prices) + +Lens exposes a `priceUpdated(pair)` subscription that pushes a message every time +an ingester (SDEX, Horizon AMM, or Soroswap) records a new price for the pair. +It runs over the same `/graphql` endpoint using the `graphql-transport-ws` +protocol, so any [`graphql-ws`](https://github.com/enisdenjo/graphql-ws) client works. + +```graphql +subscription { + priceUpdated(pair: "XLM/USDC", network: "mainnet") { + pair + price + ts + network + } +} +``` + +```bash +npm install graphql-ws ws +``` + +```typescript +import { createClient } from "graphql-ws"; +import WebSocket from "ws"; // browsers already have WebSocket globally + +const client = createClient({ + url: "ws://localhost:3002/graphql", + webSocketImpl: WebSocket, // omit in the browser +}); + +// `subscribe` returns an unsubscribe function — call it to close the channel. +const unsubscribe = client.subscribe( + { + query: `subscription ($pair: String!, $network: String) { + priceUpdated(pair: $pair, network: $network) { pair price ts network } + }`, + variables: { pair: "XLM/USDC", network: "mainnet" }, + }, + { + next: ({ data }) => console.log("price:", data.priceUpdated), + error: (err) => console.error("subscription error:", err), + complete: () => console.log("subscription closed"), + }, +); + +// Later — stop receiving updates and close the socket cleanly: +// unsubscribe(); +``` + +> **`network` is optional but you almost always want it.** Since #117 every +> enabled network runs its own ingester loop and they all publish to the same +> stream, so omitting it interleaves testnet and mainnet prices for the same +> pair. Every message carries its own `network` field, so an omitted argument +> is safe *if* you read that field — and misleading if you do not. + +> **Note:** the `pair` argument is the canonical `pairKey` (alphabetically +> sorted, e.g. `XLM:native/USDC:GA5...`). Use the `listPairs` query to discover +> the exact keys being indexed. Only the pair you subscribe to is delivered; +> updates for other pairs are filtered out server-side. + ## Usage Examples Lens gates `/price`, `/pools`, and `/candles` behind x402 micropayments on Stellar (testnet by default). The `/status` endpoint is free. diff --git a/src/__tests__/graphqlSubscription.test.ts b/src/__tests__/graphqlSubscription.test.ts new file mode 100644 index 0000000..27f3e4a --- /dev/null +++ b/src/__tests__/graphqlSubscription.test.ts @@ -0,0 +1,183 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' +import WebSocket from 'ws' +import type { AddressInfo } from 'net' + +// graphql.ts pulls in redis / db / aggregator / config at import time — stub the +// ones the Query resolvers touch so importing the module is side-effect free. +// The subscription path under test does not use any of them. +vi.mock('../redis', () => ({ getCachedPrice: vi.fn() })) +vi.mock('../db', () => ({ pgPool: { query: vi.fn() } })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: vi.fn() })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: vi.fn() })) +vi.mock('../config', () => ({ config: { pairs: [] } })) + +import { registerGraphQL } from '../api/graphql' +import { publishPriceUpdate, priceEmitter } from '../events' + +const SUBPROTOCOL = 'graphql-transport-ws' + +async function buildServer(): Promise<{ app: FastifyInstance; url: string }> { + const app = Fastify({ logger: false }) + await registerGraphQL(app) + await app.listen({ port: 0, host: '127.0.0.1' }) + const { port } = app.server.address() as AddressInfo + return { app, url: `ws://127.0.0.1:${port}/graphql` } +} + +/** Open a graphql-transport-ws connection and complete the connection_init handshake. */ +function connect(url: string): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url, SUBPROTOCOL) + ws.on('error', reject) + ws.on('open', () => ws.send(JSON.stringify({ type: 'connection_init' }))) + ws.on('message', function onAck(raw) { + const msg = JSON.parse(raw.toString()) + if (msg.type === 'connection_ack') { + ws.off('message', onAck) + resolve(ws) + } + }) + }) +} + +/** Wait for the next message of a given type, with a timeout. */ +function waitFor(ws: WebSocket, type: string, timeoutMs = 2000): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + ws.off('message', onMsg) + reject(new Error(`timed out waiting for "${type}"`)) + }, timeoutMs) + function onMsg(raw: WebSocket.RawData) { + const msg = JSON.parse(raw.toString()) + if (msg.type === type) { + clearTimeout(timer) + ws.off('message', onMsg) + resolve(msg) + } + } + ws.on('message', onMsg) + }) +} + +const SUBSCRIBE = (pair: string, network?: string) => ({ + id: '1', + type: 'subscribe', + payload: { + query: `subscription($pair: String!, $network: String) { + priceUpdated(pair: $pair, network: $network) { pair price ts network } + }`, + variables: { pair, network: network ?? null }, + }, +}) + +describe('GraphQL priceUpdated subscription', () => { + let app: FastifyInstance + let url: string + + beforeEach(async () => { + ;({ app, url } = await buildServer()) + }) + + afterEach(async () => { + await app.close() + // app.close() fires the onClose hook that detaches the bridge listener. + expect(priceEmitter.listenerCount('price:published')).toBe(0) + }) + + it('streams updates for the subscribed pair', async () => { + const ws = await connect(url) + ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC'))) + + // Give the server a tick to register the subscription before publishing. + await new Promise(r => setTimeout(r, 100)) + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.1234, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' }) + + const next = await waitFor(ws, 'next') + expect(next.id).toBe('1') + expect(next.payload.data.priceUpdated).toEqual({ + pair: 'XLM/USDC', + price: 0.1234, + ts: '2026-06-02T00:00:00.000Z', + network: 'testnet', + }) + + ws.close() + }) + + it('does not deliver updates for other pairs', async () => { + const ws = await connect(url) + ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC'))) + await new Promise(r => setTimeout(r, 100)) + + // A different pair must be filtered out… + publishPriceUpdate({ pair: 'BTC/USDC', price: 99, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' }) + // …while the subscribed pair still comes through. + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.5, ts: '2026-06-02T00:00:01.000Z', network: 'testnet' }) + + const next = await waitFor(ws, 'next') + expect(next.payload.data.priceUpdated.pair).toBe('XLM/USDC') + expect(next.payload.data.priceUpdated.price).toBe(0.5) + + ws.close() + }) + + it('does not deliver another network\'s price for the same pair', async () => { + // Since #117 every enabled network runs its own ingester loop and all of + // them publish to this one emitter. Without the network filter a mainnet + // print and a testnet print for XLM/USDC arrive on the same stream, + // indistinguishable — a chart that looks noisy rather than wrong, which is + // the harder kind of bug to notice. + const ws = await connect(url) + ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC', 'mainnet'))) + await new Promise(r => setTimeout(r, 100)) + + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.11, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' }) + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.22, ts: '2026-06-02T00:00:01.000Z', network: 'mainnet' }) + + const next = await waitFor(ws, 'next') + expect(next.payload.data.priceUpdated.network).toBe('mainnet') + expect(next.payload.data.priceUpdated.price).toBe(0.22) + + ws.close() + }) + + it('delivers every network when none is requested, each tagged with its own', async () => { + // Omitting the argument is allowed, but only because the message carries + // its own network — a subscriber can still tell the two chains apart. + const ws = await connect(url) + ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC'))) + await new Promise(r => setTimeout(r, 100)) + + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.11, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' }) + const first = await waitFor(ws, 'next') + expect(first.payload.data.priceUpdated.network).toBe('testnet') + + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.22, ts: '2026-06-02T00:00:01.000Z', network: 'mainnet' }) + const second = await waitFor(ws, 'next') + expect(second.payload.data.priceUpdated.network).toBe('mainnet') + + ws.close() + }) + + it('closes the channel cleanly on complete', async () => { + const ws = await connect(url) + ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC'))) + await new Promise(r => setTimeout(r, 100)) + + // Client-initiated unsubscribe. + ws.send(JSON.stringify({ id: '1', type: 'complete' })) + await new Promise(r => setTimeout(r, 100)) + + // After completing, further publishes for that pair must not arrive. + let received = false + ws.on('message', raw => { + if (JSON.parse(raw.toString()).type === 'next') received = true + }) + publishPriceUpdate({ pair: 'XLM/USDC', price: 1, ts: '2026-06-02T00:00:02.000Z', network: 'testnet' }) + await new Promise(r => setTimeout(r, 200)) + + expect(received).toBe(false) + ws.close() + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index 24ba53e..b9c7ebb 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,11 +1,17 @@ import type { FastifyInstance } from 'fastify' import { price_requests_total } from '../metrics' -import mercurius from 'mercurius' +import mercurius, { withFilter, type MercuriusContext } from 'mercurius' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' import { config } from '../config' +import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' + +// Mercurius pubsub topic that carries every new price. A single app-level +// listener bridges the ingesters' in-process `priceEmitter` onto this topic; +// each subscriber then filters it down to the pair they asked for. +const PRICE_TOPIC = 'PRICE_UPDATED' const schema = ` type AggregatedPrice { @@ -53,12 +59,33 @@ const schema = ` low: Float } + type PriceUpdate { + pair: String! + price: Float! + ts: String! + """Which chain the price came from — testnet or mainnet.""" + network: String! + } + type Query { getPrice(assetA: String!, assetB: String!): AggregatedPrice getBestRoute(assetA: String!, assetB: String!, amount: Float!): RouteInfo getPriceHistory(assetA: String!, assetB: String!, window: String!, limit: Int): [PriceBucket] listPairs: [String]! } + + type Subscription { + """ + Streams a PriceUpdate every time an ingester records a new price for the + given pair. + + The network argument narrows the stream to one chain. It is optional, and + omitting it delivers every enabled network — the right default only if you + read the network field on each message, since a dual-network deployment + otherwise interleaves two chains prices on one stream. + """ + priceUpdated(pair: String!, network: String): PriceUpdate! + } ` function makePairKey(a: string, b: string): string { @@ -142,6 +169,29 @@ const resolvers = { return config.pairs.map(p => p.pairKey) }, }, + + Subscription: { + priceUpdated: { + subscribe: withFilter< + { priceUpdated: PricePublishedEvent }, + unknown, + MercuriusContext, + { pair: string; network?: string | null } + >( + (_root, _args, { pubsub }) => pubsub.subscribe(PRICE_TOPIC), + // Both loops publish to one topic, so the network filter has to happen + // here. Omitting `network` keeps every chain — the message carries its + // own `network` field, so the subscriber can still tell them apart. + (payload, { pair, network }) => + payload.priceUpdated.pair === pair && + // == null, not === undefined: a client that passes the variable + // explicitly sends null rather than omitting it, and both mean + // "every network". Comparing against undefined alone would filter + // out every message for those clients. + (network == null || payload.priceUpdated.network === network) + ), + }, + }, } export async function registerGraphQL(app: FastifyInstance) { @@ -150,5 +200,28 @@ export async function registerGraphQL(app: FastifyInstance) { resolvers, graphiql: true, path: '/graphql', + subscription: { + // Speak the `graphql-transport-ws` subprotocol (the modern `graphql-ws` + // library) rather than the legacy `subscriptions-transport-ws`. + fullWsTransport: true, + wsDefaultSubprotocol: 'graphql-transport-ws', + }, + }) + + // Bridge: forward every price the ingesters emit onto the GraphQL pubsub + // topic. Mercurius wraps the payload under the subscription field name so + // `withFilter` and the resolver receive `{ priceUpdated: }`. + const onPricePublished = (event: PricePublishedEvent) => { + app.graphql.pubsub.publish({ + topic: PRICE_TOPIC, + payload: { priceUpdated: event }, + }) + } + priceEmitter.on(PRICE_PUBLISHED, onPricePublished) + + // Detach the listener when the server shuts down so repeated + // register/close cycles (e.g. in tests) don't leak listeners. + app.addHook('onClose', async () => { + priceEmitter.off(PRICE_PUBLISHED, onPricePublished) }) } diff --git a/src/events.ts b/src/events.ts index 5e46e1e..7b1c525 100644 --- a/src/events.ts +++ b/src/events.ts @@ -1,4 +1,5 @@ import { EventEmitter } from 'events' +import type { NetworkName } from './config' export const priceEmitter = new EventEmitter() @@ -11,3 +12,38 @@ export interface PriceUpdateEvent { currentPrice: number timestamp: Date } + +/** + * Emitted by every ingester after it records a new price for a pair. This is + * the feed that backs the GraphQL `priceUpdated` subscription. It is kept + * deliberately minimal — `{ pair, price, ts }` — so it is cheap to publish on + * the ingester hot path and the GraphQL layer can filter purely on `pair`. + */ +export const PRICE_PUBLISHED = 'price:published' + +export interface PricePublishedEvent { + /** pairKey, e.g. "XLM:native/USDC:GA5..." */ + pair: string + price: number + /** ISO-8601 timestamp of when the price was recorded */ + ts: string + /** + * Which chain this price came from. + * + * Load-bearing since #117: every enabled network runs its own ingester loop + * and they all publish to this one emitter. Without a discriminator a + * subscriber to XLM/USDC receives testnet and mainnet prices interleaved, + * indistinguishable, on the same stream — a chart that looks noisy rather + * than wrong, which is the harder kind to notice. + */ + network: NetworkName +} + +/** + * Publish a new price to all live subscribers (GraphQL `priceUpdated`). + * Fire-and-forget: emitting is synchronous and never throws, so ingesters can + * call it without a try/catch on their hot path. + */ +export function publishPriceUpdate(event: PricePublishedEvent): void { + priceEmitter.emit(PRICE_PUBLISHED, event) +} diff --git a/src/ingesters/amm.ts b/src/ingesters/amm.ts index a91841e..8a62304 100644 --- a/src/ingesters/amm.ts +++ b/src/ingesters/amm.ts @@ -3,6 +3,7 @@ import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../co import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints, getIndexerCursor, setIndexerCursor, prisma } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' +import { publishPriceUpdate } from '../events' import type { WatchedPair } from '../types' const lastPrice = new Map() @@ -34,7 +35,11 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ } } -export async function snapshotPool(pool: any, pair: WatchedPair): Promise { +export async function snapshotPool( + pool: any, + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { try { const r0 = pool.reserves[0] const r1 = pool.reserves[1] @@ -83,6 +88,14 @@ export async function snapshotPool(pool: any, pair: WatchedPair): Promise const previousPrice = lastPrice.get(pair.pairKey) ?? spotPrice lastPrice.set(pair.pairKey, spotPrice) + + publishPriceUpdate({ + pair: pair.pairKey, + price: spotPrice, + ts: new Date().toISOString(), + network, + }) + dispatchPriceUpdate({ assetA: pair.assetA.code, assetB: pair.assetB.code, @@ -148,6 +161,13 @@ export async function ingestPoolTrades( await setIndexerCursor(stateId, lastCursor) console.log(`[amm] Pool ${pool.id.slice(0, 8)}: ingested ${points.length} trades`) + publishPriceUpdate({ + pair: pair.pairKey, + price: currentPrice, + ts: points[points.length - 1].timestamp.toISOString(), + network, + }) + dispatchPriceUpdate({ assetA: pair.assetA.code, assetB: pair.assetB.code, @@ -172,7 +192,7 @@ export async function startAMMIngester(network: NetworkName = activeNetwork): Pr console.log(`[amm] ${pair.pairKey}: found ${pools.length} AMM pools`) await Promise.all(pools.map(async pool => { - await snapshotPool(pool, pair) + await snapshotPool(pool, pair, network) await ingestPoolTrades(pool, pair, network) })) } diff --git a/src/ingesters/sdex.ts b/src/ingesters/sdex.ts index e5022cc..0e6bba3 100644 --- a/src/ingesters/sdex.ts +++ b/src/ingesters/sdex.ts @@ -5,6 +5,7 @@ import { getHorizonServer } from '../network/clients' import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints, getIndexerCursor, setIndexerCursor } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' +import { publishPriceUpdate } from '../events' import type { WatchedPair } from '../types' // Last seen price per pairKey — used for threshold crossing detection @@ -70,6 +71,13 @@ export async function ingestPair(pair: WatchedPair, network: NetworkName = activ await setIndexerCursor(stateId, lastCursor) console.log(`[sdex] ${pair.pairKey}: ingested ${points.length} trades`) + publishPriceUpdate({ + pair: pair.pairKey, + price: currentPrice, + ts: points[points.length - 1].timestamp.toISOString(), + network, + }) + dispatchPriceUpdate({ assetA: pair.assetA.code, assetB: pair.assetB.code, diff --git a/src/ingesters/soroswap.ts b/src/ingesters/soroswap.ts index 183e69f..7709817 100644 --- a/src/ingesters/soroswap.ts +++ b/src/ingesters/soroswap.ts @@ -24,6 +24,7 @@ import { getRpcServer } from '../network/clients' import { getActivePairs } from '../pairsRegistry' import { upsertPricePoints } from '../db' import { dispatchPriceUpdate } from '../webhookDispatcher' +import { publishPriceUpdate } from '../events' import type { WatchedPair } from '../types' // ── Constants ───────────────────────────────────────────────────────────────── @@ -240,6 +241,13 @@ export async function ingestPool( const previousPrice = lastPrice.get(pair.pairKey) ?? spotPrice lastPrice.set(pair.pairKey, spotPrice) + publishPriceUpdate({ + pair: pair.pairKey, + price: spotPrice, + ts: new Date().toISOString(), + network, + }) + dispatchPriceUpdate({ assetA: pair.assetA.code, assetB: pair.assetB.code,