Add graph ql subscriptions for live prices

.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.

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.

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<WebSocket> {
+  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<any> {
+  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()
+  })
+})

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: <event> }`.
+  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)
   })
 }