Meter and limit GraphQL subscriptions

.env.example

@@ -118,6 +118,8 @@ ADMIN_TOKEN=super-secret-admin-token
 REQUIRE_API_KEY=true
 # Fallback rate limit (requests per minute per IP) for unauthenticated or public requests (default: 100)
 RATE_LIMIT_IP_MAX=100
+# Maximum number of active GraphQL subscriptions allowed per API key (default: 10)
+GRAPHQL_MAX_SUBSCRIPTIONS_PER_KEY=10
 
 # --- x402 Payment Gate ---
 # Stellar public key where API payments should be sent.

README.md

@@ -168,6 +168,7 @@ import WebSocket from "ws"; // browsers already have WebSocket globally
 const client = createClient({
   url: "ws://localhost:3002/graphql",
   webSocketImpl: WebSocket, // omit in the browser
+  connectionParams: { authorization: `Bearer ${process.env.LENS_API_KEY}` },
 });
 
 // `subscribe` returns an unsubscribe function — call it to close the channel.
@@ -189,6 +190,11 @@ const unsubscribe = client.subscribe(
 // unsubscribe();
 ```
 
+Subscriptions require a valid API key in `connectionParams` and are metered at
+the same `$0.10` price as a GraphQL query when each subscription starts. The
+default concurrent subscription limit is 10 per API key; set
+`GRAPHQL_MAX_SUBSCRIPTIONS_PER_KEY` to change it.
+
 > **`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
@@ -202,7 +208,8 @@ const unsubscribe = client.subscribe(
 
 ## Usage Examples
 
-Lens gates `/price`, `/pools`, and `/candles` behind x402 micropayments on Stellar (testnet by default). The `/status` endpoint is free.
+Lens gates `/price`, `/pools`, `/candles`, and GraphQL POST queries behind x402
+micropayments on Stellar (testnet by default). The `/status` endpoint is free.
 
 ### 1. Free health check (no payment)
 
@@ -317,7 +324,8 @@ curl -X POST http://localhost:3002/graphql \
 
 Or interactively at [http://localhost:3002/graphiql](http://localhost:3002/graphiql).
 
-> **Note:** the current `GATED_ROUTES` map in [`src/middleware/x402.ts`](src/middleware/x402.ts) gates `/price`, `/pools`, and `/candles` only — `/graphql` is not gated. If you intend price data from GraphQL to require the same payment as REST, extend `GATED_ROUTES` (or add a per-resolver guard).
+GraphQL POST queries are included in `GATED_ROUTES` at `$0.10`. WebSocket
+subscriptions require an API key and record the same amount once per activation.
 
 ## Documentation
 Detailed system design and data flow diagrams can be found in the [Architecture Overview](docs/architecture.md).
@@ -491,6 +499,7 @@ curl -s http://localhost:3002/price/XLM/USDC | jq '{price, ammPrice, lastUpdated
 | `DATABASE_URL` | PostgreSQL connection string | - | **Yes** |
 | `REDIS_URL` | Redis connection string | - | **Yes** |
 | `PRICE_CACHE_TTL` | Cache duration for price data (seconds) | `10` | No |
+| `GRAPHQL_MAX_SUBSCRIPTIONS_PER_KEY` | Maximum concurrent GraphQL subscriptions per API key | `10` | No |
 | `HORIZON_URL` | Stellar Horizon server URL | - | No |
 | `RPC_URL` | Soroban RPC server URL | - | No |
 | `NETWORK_PASSPHRASE` | Stellar network passphrase | - | No |

src/__tests__/graphqlSubscription.test.ts

@@ -11,9 +11,21 @@ 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: [] } }))
+vi.mock('../api/auth', () => ({
+  extractBearer: (header?: string) => /^Bearer\s+(.+)$/i.exec(header?.trim() ?? '')?.[1]?.trim() ?? null,
+  lookupApiKey: vi.fn(async (token: string) => token === 'valid-test-key'
+    ? { id: 'subscription-test-key', label: 'test', ratePerMin: 60, ratePerDay: 1000 }
+    : null),
+}))
+vi.mock('../x402/metering', () => ({
+  checkQuota: vi.fn(async () => ({ allowed: true })),
+  parseCents: (price: string) => Math.round(Number(price.slice(1)) * 100),
+  recordUsage: vi.fn(async () => undefined),
+}))
 
 import { registerGraphQL } from '../api/graphql'
 import { publishPriceUpdate, priceEmitter } from '../events'
+import { checkQuota, recordUsage } from '../x402/metering'
 
 const SUBPROTOCOL = 'graphql-transport-ws'
 
@@ -26,11 +38,14 @@ async function buildServer(): Promise<{ app: FastifyInstance; url: string }> {
 }
 
 /** Open a graphql-transport-ws connection and complete the connection_init handshake. */
-function connect(url: string): Promise<WebSocket> {
+function connect(url: string, authorization: string | undefined = 'Bearer valid-test-key'): 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('open', () => ws.send(JSON.stringify({
+      type: 'connection_init',
+      payload: authorization ? { authorization } : undefined,
+    })))
     ws.on('message', function onAck(raw) {
       const msg = JSON.parse(raw.toString())
       if (msg.type === 'connection_ack') {
@@ -60,8 +75,8 @@ function waitFor(ws: WebSocket, type: string, timeoutMs = 2000): Promise<any> {
   })
 }
 
-const SUBSCRIBE = (pair: string, network?: string) => ({
-  id: '1',
+const SUBSCRIBE = (pair: string, network?: string, id = '1') => ({
+  id,
   type: 'subscribe',
   payload: {
     query: `subscription($pair: String!, $network: String) {
@@ -76,11 +91,13 @@ describe('GraphQL priceUpdated subscription', () => {
   let url: string
 
   beforeEach(async () => {
+    vi.clearAllMocks()
     ;({ app, url } = await buildServer())
   })
 
   afterEach(async () => {
     await app.close()
+    vi.unstubAllEnvs()
     // app.close() fires the onClose hook that detaches the bridge listener.
     expect(priceEmitter.listenerCount('price:published')).toBe(0)
   })
@@ -105,6 +122,68 @@ describe('GraphQL priceUpdated subscription', () => {
     ws.close()
   })
 
+  it('refuses an unauthenticated connection during connection_init', async () => {
+    const ws = new WebSocket(url, SUBPROTOCOL)
+    const rejected = waitFor(ws, 'connection_error')
+    ws.on('error', () => undefined)
+    ws.on('open', () => ws.send(JSON.stringify({ type: 'connection_init' })))
+
+    const error = await rejected
+    expect(error.payload.message).toBe('Forbidden')
+    ws.close()
+  })
+
+  it('refuses an invalid API key during connection_init', async () => {
+    const ws = new WebSocket(url, SUBPROTOCOL)
+    const rejected = waitFor(ws, 'connection_error')
+    ws.on('error', () => undefined)
+    ws.on('open', () => ws.send(JSON.stringify({
+      type: 'connection_init',
+      payload: { authorization: 'Bearer invalid-test-key' },
+    })))
+
+    const error = await rejected
+    expect(error.payload.message).toBe('Forbidden')
+    ws.close()
+  })
+
+  it('meters an active subscription at the GraphQL request price', async () => {
+    const ws = await connect(url)
+    ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC')))
+    await new Promise(r => setTimeout(r, 100))
+
+    expect(checkQuota).toHaveBeenCalledWith('subscription-test-key')
+    expect(recordUsage).toHaveBeenCalledWith('subscription-test-key', 10)
+
+    ws.close()
+  })
+
+  it('caps active subscriptions per key and releases a slot on complete', async () => {
+    await app.close()
+    vi.stubEnv('GRAPHQL_MAX_SUBSCRIPTIONS_PER_KEY', '1')
+    ;({ app, url } = await buildServer())
+
+    const ws = await connect(url)
+    ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC', undefined, '1')))
+    await new Promise(r => setTimeout(r, 100))
+
+    ws.send(JSON.stringify(SUBSCRIBE('BTC/USDC', undefined, '2')))
+    const rejected = await waitFor(ws, 'error')
+    expect(rejected.id).toBe('2')
+    expect(rejected.payload[0].message).toContain('Concurrent subscription limit exceeded')
+
+    ws.send(JSON.stringify({ id: '1', type: 'complete' }))
+    await new Promise(r => setTimeout(r, 100))
+    ws.send(JSON.stringify(SUBSCRIBE('BTC/USDC', undefined, '3')))
+    await new Promise(r => setTimeout(r, 100))
+    publishPriceUpdate({ pair: 'BTC/USDC', price: 99, ts: '2026-06-02T00:00:00.000Z', network: 'testnet' })
+
+    const next = await waitFor(ws, 'next')
+    expect(next.id).toBe('3')
+    expect(recordUsage).toHaveBeenCalledTimes(2)
+    ws.close()
+  })
+
   it('does not deliver updates for other pairs', async () => {
     const ws = await connect(url)
     ws.send(JSON.stringify(SUBSCRIBE('XLM/USDC')))

src/api/graphql.ts

@@ -1,6 +1,9 @@
 import type { FastifyInstance } from 'fastify'
 import { price_requests_total } from '../metrics'
 import mercurius, { withFilter, type MercuriusContext } from 'mercurius'
+import { extractBearer, lookupApiKey, type ApiKeyContext } from './auth'
+import { checkQuota, parseCents, recordUsage } from '../x402/metering'
+import { GRAPHQL_REQUEST_PRICE } from '../x402/pricing'
 import { getCachedPrice } from '../redis'
 import { getAggregatedPrice } from '../aggregator/vwap'
 import { getBestRoute } from '../aggregator/bestRoute'
@@ -12,6 +15,16 @@ import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../even
 // 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 activeSubscriptionsByKey = new Map<string, number>()
+
+interface GraphQLSubscriptionContext extends MercuriusContext {
+  apiKey?: ApiKeyContext
+}
+
+function maxSubscriptionsPerKey(): number {
+  const configuredLimit = Number.parseInt(process.env.GRAPHQL_MAX_SUBSCRIPTIONS_PER_KEY ?? '10', 10)
+  return Number.isSafeInteger(configuredLimit) && configuredLimit > 0 ? configuredLimit : 10
+}
 
 const schema = `
   type AggregatedPrice {
@@ -182,10 +195,44 @@ const resolvers = {
       subscribe: withFilter<
         { priceUpdated: PricePublishedEvent },
         unknown,
-        MercuriusContext,
+        GraphQLSubscriptionContext,
         { pair: string; network?: string | null }
       >(
-        (_root, _args, { pubsub }) => pubsub.subscribe(PRICE_TOPIC),
+        async (_root, _args, { pubsub, apiKey }) => {
+          if (!apiKey) throw new Error('A valid API key is required for subscriptions')
+          const activeCount = activeSubscriptionsByKey.get(apiKey.id) ?? 0
+          if (activeCount >= maxSubscriptionsPerKey()) {
+            throw new Error('Concurrent subscription limit exceeded')
+          }
+          activeSubscriptionsByKey.set(apiKey.id, activeCount + 1)
+
+          let released = false
+          const release = () => {
+            if (released) return
+            released = true
+            const count = activeSubscriptionsByKey.get(apiKey.id) ?? 0
+            if (count <= 1) activeSubscriptionsByKey.delete(apiKey.id)
+            else activeSubscriptionsByKey.set(apiKey.id, count - 1)
+          }
+
+          try {
+            const quota = await checkQuota(apiKey.id)
+            if (!quota.allowed) throw new Error('API key usage quota exceeded')
+            await recordUsage(apiKey.id, parseCents(GRAPHQL_REQUEST_PRICE))
+
+            const source = await pubsub.subscribe(PRICE_TOPIC)
+            return (async function* () {
+              try {
+                for await (const payload of source) yield payload
+              } finally {
+                release()
+              }
+            })()
+          } catch (error) {
+            release()
+            throw error
+          }
+        },
         // 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.
@@ -212,6 +259,19 @@ export async function registerGraphQL(app: FastifyInstance) {
       // library) rather than the legacy `subscriptions-transport-ws`.
       fullWsTransport: true,
       wsDefaultSubprotocol: 'graphql-transport-ws',
+      onConnect: async ({ payload }) => {
+        const initPayload = payload && typeof payload === 'object'
+          ? payload as Record<string, unknown>
+          : {}
+        const authorization = initPayload.authorization ?? initPayload.Authorization
+        const token = typeof initPayload.apiKey === 'string'
+          ? initPayload.apiKey
+          : extractBearer(typeof authorization === 'string' ? authorization : undefined)
+        if (!token) return false
+
+        const apiKey = await lookupApiKey(token)
+        return apiKey ? { apiKey } : false
+      },
     },
   })
 

src/middleware/x402.ts

@@ -2,6 +2,7 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify'
 import { x402_payments_received_total } from '../metrics'
 import { checkQuota, recordUsage, parseCents, getQuotaConfig } from '../x402/metering'
 import { X402_NETWORK_LABEL, paymentAddressFor, isX402Configured, getX402ResourceServer } from '../x402/network'
+import { GRAPHQL_REQUEST_PRICE } from '../x402/pricing'
 import fp from 'fastify-plugin'
 import './network' // declares req.network on the FastifyRequest type
 
@@ -35,7 +36,7 @@ export const GATED_ROUTES: readonly GatedRoute[] = [
   {
     path: '/graphql',
     method: 'POST',
-    price: '$0.10',
+    price: GRAPHQL_REQUEST_PRICE,
     description: 'GraphQL queries for price data and market information',
   },
 ] as const

src/x402/pricing.ts

@@ -0,0 +1 @@
+export const GRAPHQL_REQUEST_PRICE = '$0.10'