From 3073fae396aa6491f73fb4f44e5c0747b25435c1 Mon Sep 17 00:00:00 2001 From: collins-uzu Date: Fri, 16 Oct 2026 22:27:51 +0000 Subject: [PATCH] Meter and limit GraphQL subscriptions --- .env.example | 2 + README.md | 13 +++- src/__tests__/graphqlSubscription.test.ts | 87 +++++++++++++++++++++-- src/api/graphql.ts | 64 ++++++++++++++++- src/middleware/x402.ts | 3 +- src/x402/pricing.ts | 1 + 6 files changed, 161 insertions(+), 9 deletions(-) create mode 100644 src/x402/pricing.ts diff --git a/.env.example b/.env.example index d37f04a6..654cc430 100644 --- a/.env.example +++ b/.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. diff --git a/README.md b/README.md index 92d0ada6..4cecef0b 100644 --- a/README.md +++ b/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 | diff --git a/src/__tests__/graphqlSubscription.test.ts b/src/__tests__/graphqlSubscription.test.ts index 27f3e4ac..d6ddcdd9 100644 --- a/src/__tests__/graphqlSubscription.test.ts +++ b/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 { +function connect(url: string, authorization: string | undefined = 'Bearer valid-test-key'): 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('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 { }) } -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'))) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index b03f3210..b96eec16 100644 --- a/src/api/graphql.ts +++ b/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() + +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 + : {} + 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 + }, }, }) diff --git a/src/middleware/x402.ts b/src/middleware/x402.ts index a2651f0a..3ee63dc7 100644 --- a/src/middleware/x402.ts +++ b/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 diff --git a/src/x402/pricing.ts b/src/x402/pricing.ts new file mode 100644 index 00000000..aca28fbe --- /dev/null +++ b/src/x402/pricing.ts @@ -0,0 +1 @@ +export const GRAPHQL_REQUEST_PRICE = '$0.10' \ No newline at end of file