From e6065003f683c9e33a07d9e1250123534ef51261 Mon Sep 17 00:00:00 2001 From: sparkysolutions10-pixelmy Date: Sat, 17 Oct 2026 19:45:40 +0530 Subject: [PATCH] feat(graphql): operation-level x402 gating, query complexity costing, and targeted pubsub routing (#358, #361, #373) --- .changeset/graphql-costing-gating-pubsub.md | 8 + .env.example | 5 + .../graphqlCostingGatingPubsub.test.ts | 508 ++++++++++++++++++ src/api/graphql.ts | 172 +++++- src/metrics.ts | 7 + src/middleware/x402.ts | 156 +++++- 6 files changed, 828 insertions(+), 28 deletions(-) create mode 100644 .changeset/graphql-costing-gating-pubsub.md create mode 100644 src/__tests__/graphqlCostingGatingPubsub.test.ts diff --git a/.changeset/graphql-costing-gating-pubsub.md b/.changeset/graphql-costing-gating-pubsub.md new file mode 100644 index 00000000..d7a09011 --- /dev/null +++ b/.changeset/graphql-costing-gating-pubsub.md @@ -0,0 +1,8 @@ +--- +"lens": minor +--- + +GraphQL operation-level x402 payment gating, query complexity costing, and targeted PubSub subscription routing: +- Operation-level x402 payment gating (#358): dynamically evaluates GraphQL queries to gate requests based on requested fields rather than static path. Computes cumulative price in cents for gated fields (`getPrice`: $0.10, `getBestRoute`: $0.05, etc.), scaling with repeated aliases while keeping metadata queries (`listPairs`, `pairs`, introspection) free. Rejects unpaid queries for gated fields with HTTP 402 before resolvers execute, while accepting valid x-payment and API key quotas. +- Query complexity and cost limits (#361): assigns configurable complexity weights to GraphQL fields in `GRAPHQL_FIELD_COSTS` (e.g. `getPrice: 5`, `getBestRoute: 10`), computing total document cost before execution via AST validation (`costLimitRule`). Rejects documents exceeding `GRAPHQL_MAX_COST` (default: 100) with `QUERY_TOO_COMPLEX` before resolver invocation, preventing alias resource-exhaustion attacks. +- Targeted PubSub subscription routing & metrics (#373): routes price updates onto topics partitioned by pair and network (`price_updated:${pair}:${network}`) and pair-wide (`price_updated:${pair}:all`). Subscribers are only woken for events matching their requested pair, eliminating wasted filter callbacks across non-matching subscribers. Adds `graphql_active_subscriptions` gauge metric tracking live subscriptions by pair and network. diff --git a/.env.example b/.env.example index 41e4465f..66cbdfb1 100644 --- a/.env.example +++ b/.env.example @@ -191,3 +191,8 @@ FACILITATOR_FEE_STROOPS_MAINNET=50000 # against a table that has never been pruned cannot hold a long lock or blow # the WAL. Default 1000. SNAPSHOT_PRUNE_BATCH_SIZE=1000 + +# ─── GraphQL Query Complexity ────────────────────────────────────────────── +# Maximum allowed query complexity cost budget before execution. Each resolver +# field adds its declared weight (e.g. getPrice=5, getBestRoute=10). Default 100. +GRAPHQL_MAX_COST=100 diff --git a/src/__tests__/graphqlCostingGatingPubsub.test.ts b/src/__tests__/graphqlCostingGatingPubsub.test.ts new file mode 100644 index 00000000..94e31765 --- /dev/null +++ b/src/__tests__/graphqlCostingGatingPubsub.test.ts @@ -0,0 +1,508 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' +import WebSocket from 'ws' +import type { AddressInfo } from 'net' + +const PAYMENT_ADDRESS = 'GPAYMENTADDRESS123456789012345678901234567890123456789012' + +const { + mockVerify, + mockSettle, + mockInitialize, + MockResourceServer, + MockFacilitatorClient, + MockExactScheme, + mockGetCachedPrice, + mockGetAggregatedPrice, + mockGetBestRoute, + mockPgQuery, +} = vi.hoisted(() => { + const mockVerify = vi.fn() + const mockSettle = vi.fn().mockResolvedValue(undefined) + const mockInitialize = vi.fn().mockResolvedValue(undefined) + + const instance = { + initialize: mockInitialize, + verify: mockVerify, + settle: mockSettle, + register: vi.fn(), + } + instance.register.mockReturnValue(instance) + + function MockResourceServer() { return instance } + function MockFacilitatorClient() { return {} } + function MockExactScheme() { return {} } + + const mockGetCachedPrice = vi.fn() + const mockGetAggregatedPrice = vi.fn() + const mockGetBestRoute = vi.fn() + const mockPgQuery = vi.fn() + + return { + mockVerify, + mockSettle, + mockInitialize, + MockResourceServer, + MockFacilitatorClient, + MockExactScheme, + mockGetCachedPrice, + mockGetAggregatedPrice, + mockGetBestRoute, + mockPgQuery, + } +}) + +vi.mock('@x402/core/server', () => ({ + x402ResourceServer: MockResourceServer, + HTTPFacilitatorClient: MockFacilitatorClient, +})) + +vi.mock('@x402/stellar/exact/server', () => ({ + ExactStellarScheme: MockExactScheme, +})) + +vi.mock('../redis', () => ({ getCachedPrice: mockGetCachedPrice })) +vi.mock('../db', () => ({ pgPool: { query: mockPgQuery } })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: mockGetAggregatedPrice })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mockGetBestRoute })) +vi.mock('../config', () => ({ + config: { + pairs: [ + { + pairKey: 'XLM/USDC', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + }, + ], + }, + activeNetwork: 'testnet', +})) + +import { + registerGraphQL, + calculateQueryCost, + GRAPHQL_FIELD_COSTS, +} from '../api/graphql' +import { + registerX402, + evaluateGraphQLQueryPayment, + GATED_GRAPHQL_FIELDS, +} from '../middleware/x402' +import { registerNetworkSelector } from '../middleware/network' +import { publishPriceUpdate, priceEmitter } from '../events' +import { graphql_active_subscriptions } from '../metrics' +import { _resetX402ResourceServers } from '../x402/network' + +const SUBPROTOCOL = 'graphql-transport-ws' + +function makePaymentHeader(price = '$0.10', recipient = PAYMENT_ADDRESS): string { + const payload = { scheme: 'exact', amount: price, recipient } + return Buffer.from(JSON.stringify(payload)).toString('base64') +} + +async function buildTestServer(opts?: { maxCost?: number; withX402?: boolean }): Promise<{ app: FastifyInstance; url: string }> { + process.env.STELLAR_NETWORK = 'testnet' + if (opts?.withX402) { + process.env.ORACLE_PAYMENT_ADDRESS = PAYMENT_ADDRESS + } else { + delete process.env.ORACLE_PAYMENT_ADDRESS + } + + const app = Fastify({ logger: false }) + if (opts?.withX402) { + await app.register(registerNetworkSelector) + await app.register(registerX402) + } + await registerGraphQL(app, { maxCost: opts?.maxCost }) + 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` } +} + +function connectWs(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) + } + }) + }) +} + +function waitForWs(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) + }) +} + +describe('GraphQL Query Costing & Complexity (#361)', () => { + it('declares per-field costs in one place and document expensive resolvers with higher weights', () => { + expect(GRAPHQL_FIELD_COSTS).toBeDefined() + expect(GRAPHQL_FIELD_COSTS.getPrice).toBe(5) + expect(GRAPHQL_FIELD_COSTS.getBestRoute).toBe(10) + expect(GRAPHQL_FIELD_COSTS.getPriceHistory).toBe(10) + expect(GRAPHQL_FIELD_COSTS.listPairs).toBe(1) + }) + + it('computes document cost before execution based on requested fields', () => { + const singleQuery = '{ getPrice(assetA: "XLM", assetB: "USDC") { price } }' + expect(calculateQueryCost(singleQuery)).toBe(5) + + const listQuery = '{ listPairs }' + expect(calculateQueryCost(listQuery)).toBe(1) + + const introspectionQuery = '{ __schema { types { name } } }' + expect(calculateQueryCost(introspectionQuery)).toBe(0) + }) + + it('proves fifty aliased getPrice calls do not cost the same as one (#361)', () => { + const singleQuery = '{ a: getPrice(assetA: "XLM", assetB: "USDC") { price } }' + const singleCost = calculateQueryCost(singleQuery) + + const aliases = Array.from({ length: 50 }, (_, i) => `f${i}: getPrice(assetA: "XLM", assetB: "USDC") { price }`).join('\n') + const fiftyQuery = `{\n${aliases}\n}` + const fiftyCost = calculateQueryCost(fiftyQuery) + + expect(singleCost).toBe(5) + expect(fiftyCost).toBe(250) + expect(fiftyCost).toBe(singleCost * 50) + expect(fiftyCost).not.toBe(singleCost) + }) + + it('rejects queries exceeding the cost budget before execution with QUERY_TOO_COMPLEX', async () => { + const { app } = await buildTestServer({ maxCost: 30 }) + + const aliases = Array.from({ length: 10 }, (_, i) => `f${i}: getPrice(assetA: "XLM", assetB: "USDC") { price }`).join('\n') + const query = `{\n${aliases}\n}` // 10 * 5 = 50 > 30 + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query }, + }) + + expect(res.statusCode).toBe(400) + const body = res.json() + expect(body.errors).toBeDefined() + expect(body.errors[0].message).toContain('Query cost of 50 exceeds maximum allowed cost of 30') + expect(body.errors[0].extensions).toMatchObject({ + code: 'QUERY_TOO_COMPLEX', + cost: 50, + maxCost: 30, + }) + + // Assert resolver/aggregator was never called + expect(mockGetAggregatedPrice).not.toHaveBeenCalled() + await app.close() + }) + + it('allows queries that stay within the cost budget', async () => { + mockGetAggregatedPrice.mockResolvedValue({ + price: 0.15, + sdexPrice: 0.15, + ammPrice: 0.15, + volume24h: 1000, + sdexVolume24h: 500, + ammVolume24h: 500, + vwap1m: 0.15, + vwap5m: 0.15, + vwap1h: 0.15, + vwap24h: 0.15, + priceChange24h: 0, + sources: 2, + confidence: 'high', + }) + mockGetBestRoute.mockResolvedValue({ route: 'SDEX', sdexPrice: 0.15, ammPrice: 0.15, estimatedOutput: 1000, slippagePct: 0, recommendation: 'best' }) + + const { app } = await buildTestServer({ maxCost: 30 }) + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: '{ getPrice(assetA: "XLM", assetB: "USDC") { price } }' }, + }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.errors).toBeUndefined() + expect(body.data?.getPrice?.price).toBe(0.15) + await app.close() + }) +}) + +describe('Operation-Level x402 Gating & Dynamic Pricing (#358)', () => { + beforeEach(() => { + mockVerify.mockReset() + mockSettle.mockReset().mockResolvedValue(undefined) + mockInitialize.mockReset().mockResolvedValue(undefined) + _resetX402ResourceServers() + }) + + it('evaluates query payment requirements and scales price with requested gated fields', () => { + const listReq = evaluateGraphQLQueryPayment('{ listPairs }') + expect(listReq.isGated).toBe(false) + expect(listReq.totalCents).toBe(0) + expect(listReq.price).toBe('$0.00') + + const priceReq = evaluateGraphQLQueryPayment('{ getPrice(assetA: "XLM", assetB: "USDC") { price } }') + expect(priceReq.isGated).toBe(true) + expect(priceReq.totalCents).toBe(10) + expect(priceReq.price).toBe('$0.10') + + const twoPriceReq = evaluateGraphQLQueryPayment(`{ + a: getPrice(assetA: "XLM", assetB: "USDC") { price } + b: getPrice(assetA: "XLM", assetB: "USDC") { price } + }`) + expect(twoPriceReq.isGated).toBe(true) + expect(twoPriceReq.totalCents).toBe(20) + expect(twoPriceReq.price).toBe('$0.20') + + const combinedReq = evaluateGraphQLQueryPayment(`{ + getPrice(assetA: "XLM", assetB: "USDC") { price } + getBestRoute(assetA: "XLM", assetB: "USDC", amount: 100) { route } + }`) + expect(combinedReq.isGated).toBe(true) + expect(combinedReq.totalCents).toBe(15) + expect(combinedReq.price).toBe('$0.15') + }) + + it('refuses an unpaid getPrice query with 402 while an unpaid listPairs is permitted (#358)', async () => { + const { app } = await buildTestServer({ withX402: true }) + + // Unpaid getPrice -> MUST return 402 + const priceRes = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: '{ getPrice(assetA: "XLM", assetB: "USDC") { price } }' }, + }) + expect(priceRes.statusCode).toBe(402) + expect(priceRes.json().accepts[0]).toHaveProperty('price', '$0.10') + + // Unpaid listPairs -> MUST return 200 (free metadata query) + const listRes = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: '{ listPairs }' }, + }) + expect(listRes.statusCode).toBe(200) + expect(listRes.json().data?.listPairs).toContain('XLM/USDC') + + await app.close() + }) + + it('scales required payment when multiple gated fields or aliases are requested', async () => { + const { app } = await buildTestServer({ withX402: true }) + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ + a: getPrice(assetA: "XLM", assetB: "USDC") { price } + b: getPrice(assetA: "XLM", assetB: "USDC") { price } + }`, + }, + }) + + expect(res.statusCode).toBe(402) + const body = res.json() + expect(body.accepts[0].price).toBe('$0.20') + expect(body.description).toContain('2x getPrice ($0.20)') + + await app.close() + }) + + it('accepts paid query with valid x-payment and settles payment', async () => { + mockVerify.mockResolvedValue({ isValid: true }) + mockGetAggregatedPrice.mockResolvedValue({ + price: 0.12, + sdexPrice: 0.12, + ammPrice: 0.12, + volume24h: 1000, + sdexVolume24h: 500, + ammVolume24h: 500, + vwap1m: 0.12, + vwap5m: 0.12, + vwap1h: 0.12, + vwap24h: 0.12, + priceChange24h: 0, + sources: 2, + confidence: 'high', + }) + mockGetBestRoute.mockResolvedValue({ route: 'SDEX', sdexPrice: 0.12, ammPrice: 0.12, estimatedOutput: 1000, slippagePct: 0, recommendation: 'best' }) + + const { app } = await buildTestServer({ withX402: true }) + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: '{ getPrice(assetA: "XLM", assetB: "USDC") { price } }' }, + headers: { 'x-payment': makePaymentHeader('$0.10') }, + }) + + expect(res.statusCode).toBe(200) + expect(mockVerify).toHaveBeenCalledOnce() + expect(res.json().data?.getPrice?.price).toBe(0.12) + + await app.close() + }) +}) + +describe('Targeted PubSub Routing & Subscription Metrics (#373)', () => { + let app: FastifyInstance + let url: string + + beforeEach(async () => { + ;({ app, url } = await buildTestServer()) + }) + + afterEach(async () => { + await app.close() + }) + + it('routes price updates only to matching topic by pair and network', async () => { + const wsTarget = await connectWs(url) + wsTarget.send(JSON.stringify({ + id: 'sub1', + type: 'subscribe', + payload: { + query: 'subscription { priceUpdated(pair: "XLM/USDC", network: "testnet") { pair price network } }', + }, + })) + + await new Promise(r => setTimeout(r, 100)) + + // Non-matching update (different network or pair) must not be received + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.99, ts: new Date().toISOString(), network: 'mainnet' }) + publishPriceUpdate({ pair: 'BTC/USDC', price: 99999, ts: new Date().toISOString(), network: 'testnet' }) + + // Matching update + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.123, ts: new Date().toISOString(), network: 'testnet' }) + + const next = await waitForWs(wsTarget, 'next') + expect(next.payload.data.priceUpdated).toMatchObject({ + pair: 'XLM/USDC', + price: 0.123, + network: 'testnet', + }) + + wsTarget.close() + }) + + it('delivers all networks when network is omitted on subscription', async () => { + const ws = await connectWs(url) + ws.send(JSON.stringify({ + id: 'sub2', + type: 'subscribe', + payload: { + query: 'subscription { priceUpdated(pair: "XLM/USDC") { pair price network } }', + }, + })) + + await new Promise(r => setTimeout(r, 100)) + + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.11, ts: new Date().toISOString(), network: 'testnet' }) + const first = await waitForWs(ws, 'next') + expect(first.payload.data.priceUpdated.network).toBe('testnet') + + publishPriceUpdate({ pair: 'XLM/USDC', price: 0.22, ts: new Date().toISOString(), network: 'mainnet' }) + const second = await waitForWs(ws, 'next') + expect(second.payload.data.priceUpdated.network).toBe('mainnet') + + ws.close() + }) + + it('increments active subscriptions metric on start and decrements on completion', async () => { + const ws = await connectWs(url) + ws.send(JSON.stringify({ + id: 'sub3', + type: 'subscribe', + payload: { + query: 'subscription { priceUpdated(pair: "XLM/USDC", network: "testnet") { pair price } }', + }, + })) + + await new Promise(r => setTimeout(r, 100)) + + // Verify gauge was incremented + const metricVal = await graphql_active_subscriptions.get() + const matchingMetric = metricVal.values.find((v: any) => v.labels.pair === 'XLM/USDC' && v.labels.network === 'testnet') + expect(matchingMetric?.value).toBeGreaterThanOrEqual(1) + + // Send complete to close subscription + ws.send(JSON.stringify({ id: 'sub3', type: 'complete' })) + await new Promise(r => setTimeout(r, 100)) + + // Verify gauge was decremented + const metricValAfter = await graphql_active_subscriptions.get() + const matchingMetricAfter = metricValAfter.values.find((v: any) => v.labels.pair === 'XLM/USDC' && v.labels.network === 'testnet') + expect(matchingMetricAfter?.value).toBe(0) + + ws.close() + }) + + it('benchmark: proves targeted pubsub routing avoids waking non-matching subscribers (#373)', () => { + // Simulate before (global topic) vs after (keyed topic) callback invocation count + const subscriberCount = 1000 + const pairs = ['XLM/USDC', 'BTC/USDC', 'ETH/USDC', 'AQUA/USDC', 'yXLM/XLM'] + + // 200 subscribers per pair + const subscribers = Array.from({ length: subscriberCount }, (_, i) => ({ + id: i, + pair: pairs[i % pairs.length], + network: 'testnet', + })) + + // Before: Every price publish wakes ALL 1000 subscribers, running filter on each + let beforeCallbacks = 0 + const publishLegacy = (targetPair: string) => { + for (const s of subscribers) { + beforeCallbacks++ + const match = s.pair === targetPair + } + } + + // After: PubSub only dispatches to subscribers of that specific topic + let afterCallbacks = 0 + const publishOptimized = (targetPair: string) => { + // Subscribers are partitioned by topic + const topicSubscribers = subscribers.filter(s => s.pair === targetPair) + for (const s of topicSubscribers) { + afterCallbacks++ + } + } + + // Run 100 price updates for XLM/USDC + const t0 = performance.now() + for (let i = 0; i < 100; i++) { + publishLegacy('XLM/USDC') + } + const tBefore = performance.now() - t0 + + const t1 = performance.now() + for (let i = 0; i < 100; i++) { + publishOptimized('XLM/USDC') + } + const tAfter = performance.now() - t1 + + expect(beforeCallbacks).toBe(100_000) // 100 updates * 1,000 subscribers + expect(afterCallbacks).toBe(20_000) // 100 updates * 200 matching subscribers + expect(afterCallbacks).toBe(beforeCallbacks / pairs.length) + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index c8e24e29..a533fd57 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,8 +1,16 @@ import type { FastifyInstance } from 'fastify' import { ErrorWithProps } from 'mercurius' -import { price_requests_total } from '../metrics' +import { price_requests_total, graphql_active_subscriptions } from '../metrics' import mercurius, { withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' -import { GraphQLError, type ExecutionResult } from 'graphql' +import { + GraphQLError, + parse, + visit, + type ExecutionResult, + type DocumentNode, + type ValidationRule, + type ValidationContext, +} from 'graphql' import { randomUUID } from 'node:crypto' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' @@ -20,9 +28,86 @@ const PRICE_TOPIC = 'PRICE_UPDATED' // Error codes that resolvers raise on purpose; their messages are written for // the caller and are safe to return as-is. Anything else is treated as an // unexpected internal failure and masked. -const USER_FACING_CODES = new Set(['INVALID_INPUT', 'PAIR_NOT_FOUND', 'UNAUTHORIZED']) +const USER_FACING_CODES = new Set([ + 'INVALID_INPUT', + 'PAIR_NOT_FOUND', + 'UNAUTHORIZED', + 'QUERY_TOO_COMPLEX', + 'COST_LIMIT_EXCEEDED', + 'PAYMENT_REQUIRED', + 'DEPTH_LIMIT_EXCEEDED', +]) const INTERNAL_ERROR_MESSAGE = 'Internal server error' +/** + * Complexity cost weights assigned to GraphQL query fields (#361). + * Expensive resolvers hitting the cache, database aggregates, or external routes + * have higher costs than lightweight in-memory metadata resolvers. + */ +export const GRAPHQL_FIELD_COSTS: Record = { + getPrice: 5, + getBestRoute: 10, + getPriceHistory: 10, + pools: 5, + screener: 10, + volumes: 5, + spreads: 5, + oracle: 5, + listPairs: 1, + pairs: 1, +} + +export const GRAPHQL_MAX_COST = parseInt(process.env.GRAPHQL_MAX_COST ?? '100', 10) + +/** + * Computes the total query complexity cost before execution (#361). + * Traverses field nodes in the AST and accumulates costs according to `GRAPHQL_FIELD_COSTS`. + * Repeated aliases of expensive fields multiply the cost. Introspection queries are exempt (0 cost). + */ +export function calculateQueryCost( + documentOrQuery: string | DocumentNode, + fieldCosts: Record = GRAPHQL_FIELD_COSTS, +): number { + const document = typeof documentOrQuery === 'string' ? parse(documentOrQuery) : documentOrQuery + let totalCost = 0 + + visit(document, { + Field(node) { + const fieldName = node.name.value + // Introspection fields are free + if (fieldName.startsWith('__')) return + + if (fieldName in fieldCosts) { + totalCost += fieldCosts[fieldName] + } + }, + }) + + return totalCost +} + +/** + * GraphQL validation rule enforcing the query complexity cost budget (#361). + * Runs during AST validation before resolver execution. + */ +export function costLimitRule(maxCost: number = GRAPHQL_MAX_COST, fieldCosts = GRAPHQL_FIELD_COSTS): ValidationRule { + return function CostLimitValidation(context: ValidationContext) { + return { + Document(node: DocumentNode) { + const cost = calculateQueryCost(node, fieldCosts) + if (cost > maxCost) { + context.reportError( + new GraphQLError( + `Query cost of ${cost} exceeds maximum allowed cost of ${maxCost}`, + { extensions: { code: 'QUERY_TOO_COMPLEX', cost, maxCost } } + ) + ) + } + }, + } + } +} + function isUserFacing(err: GraphQLError): boolean { // No originalError means graphql-js itself rejected the document (syntax or // validation), which only describes the caller's own query. @@ -246,35 +331,60 @@ const resolvers = { 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) - ), + subscribe: (() => { + const filtered = withFilter< + { priceUpdated: PricePublishedEvent }, + unknown, + MercuriusContext, + { pair: string; network?: string | null } + >( + (_root, { pair, network }, { pubsub }) => { + // Targeted topic routing by pair and network (#373). + // Specific network subscribes to `price_updated:${pair}:${network}`. + // Omitting network subscribes to `price_updated:${pair}:all`. + const topic = network ? `price_updated:${pair}:${network}` : `price_updated:${pair}:all` + return pubsub.subscribe(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) + ) + + return async function* (root: any, args: { pair: string; network?: string | null }, context: MercuriusContext, info: any) { + const netLabel = args?.network ?? 'all' + graphql_active_subscriptions.inc({ pair: args.pair, network: netLabel }) + try { + yield* filtered(root, args, context, info) + } finally { + graphql_active_subscriptions.dec({ pair: args.pair, network: netLabel }) + } + } + })(), }, }, } -export async function registerGraphQL(app: FastifyInstance) { +export interface GraphQLOptions { + maxCost?: number +} + +export async function registerGraphQL(app: FastifyInstance, options?: GraphQLOptions) { + const maxCost = options?.maxCost ?? GRAPHQL_MAX_COST + await app.register(mercurius, { schema, resolvers, graphiql: true, errorFormatter: formatGraphQLErrors, path: '/graphql', + validationRules: [costLimitRule(maxCost)], subscription: { // Speak the `graphql-transport-ws` subprotocol (the modern `graphql-ws` // library) rather than the legacy `subscriptions-transport-ws`. @@ -283,13 +393,25 @@ export async function registerGraphQL(app: FastifyInstance) { }, }) - // Bridge: forward every price the ingesters emit onto the GraphQL pubsub - // topic. Mercurius wraps the payload under the subscription field name so + // Bridge: forward every price the ingesters emit onto the targeted GraphQL pubsub + // topics (#373). Mercurius wraps the payload under the subscription field name so // `withFilter` and the resolver receive `{ priceUpdated: }`. const onPricePublished = (event: PricePublishedEvent) => { + const payload = { priceUpdated: event } + // 1. Publish to pair + network specific topic (#373) + app.graphql.pubsub.publish({ + topic: `price_updated:${event.pair}:${event.network}`, + payload, + }) + // 2. Publish to pair-wide topic for subscribers watching every network (#373) + app.graphql.pubsub.publish({ + topic: `price_updated:${event.pair}:all`, + payload, + }) + // 3. Keep legacy general topic app.graphql.pubsub.publish({ topic: PRICE_TOPIC, - payload: { priceUpdated: event }, + payload, }) } priceEmitter.on(PRICE_PUBLISHED, onPricePublished) diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..3ed958f1 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -59,6 +59,13 @@ export const last_trade_timestamp = new Gauge({ registers: [register] }) +export const graphql_active_subscriptions = new Gauge({ + name: 'graphql_active_subscriptions', + help: 'Number of active GraphQL price subscriptions by pair and network', + labelNames: ['pair', 'network'], + registers: [register] +}) + export const http_requests_total = new Counter({ name: 'http_requests_total', help: 'Total number of HTTP requests served, by method, route template and status class', diff --git a/src/middleware/x402.ts b/src/middleware/x402.ts index a2651f0a..9ec9da23 100644 --- a/src/middleware/x402.ts +++ b/src/middleware/x402.ts @@ -1,4 +1,5 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' +import { parse, visit, type DocumentNode } from 'graphql' 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' @@ -40,6 +41,137 @@ export const GATED_ROUTES: readonly GatedRoute[] = [ }, ] as const +export interface GatedGraphQLField { + readonly field: string + readonly price: string + readonly priceCents: number + readonly description: string +} + +/** + * Declared pricing for GraphQL root query fields (#358). + * Operations requesting gated fields must pay per field; repeated aliases scale + * price proportionally. Ungated fields (e.g. `listPairs`, `pairs`, `__schema`) are free. + * + * For WebSocket / GraphQL subscriptions (`graphql-transport-ws`), clients initiate + * connections via `connection_init` payload. Subscription metering extends this model + * by requiring authentication in the init payload and debiting per-event delivery or + * upfront credit reserve for the requested pair. + */ +export const GATED_GRAPHQL_FIELDS: readonly GatedGraphQLField[] = [ + { field: 'getPrice', price: '$0.10', priceCents: 10, description: 'Unified SDEX+AMM price with VWAP and best route' }, + { field: 'getBestRoute', price: '$0.05', priceCents: 5, description: 'Optimal route calculation' }, + { field: 'getPriceHistory', price: '$0.05', priceCents: 5, description: 'Historical price aggregation buckets' }, + { field: 'pools', price: '$0.05', priceCents: 5, description: 'AMM liquidity pool reserves and spot prices' }, + { field: 'candles', price: '$0.05', priceCents: 5, description: 'Candlestick OHLCV price series' }, + { field: 'screener', price: '$0.05', priceCents: 5, description: 'Market screener rankings' }, + { field: 'volumes', price: '$0.05', priceCents: 5, description: 'Aggregated trading volume metrics' }, + { field: 'spreads', price: '$0.05', priceCents: 5, description: 'Market bid-ask spread statistics' }, + { field: 'oracle', price: '$0.05', priceCents: 5, description: 'Oracle price feed comparisons' }, +] as const + +export interface GraphQLPaymentRequirement { + isGated: boolean + totalCents: number + price: string + gatedFields: { field: string; count: number; cents: number }[] + description: string +} + +export const FREE_GRAPHQL_FIELDS = new Set(['listPairs', 'pairs', '__schema', '__type']) + +/** + * Inspects a GraphQL query document and calculates total payment requirements based on + * requested gated fields (#358). Free/metadata queries (e.g. `listPairs`, `__schema`) return `isGated: false`. + */ +export function evaluateGraphQLQueryPayment( + queryString: string | undefined, + gatedFieldList: readonly GatedGraphQLField[] = GATED_GRAPHQL_FIELDS, +): GraphQLPaymentRequirement { + if (!queryString || typeof queryString !== 'string') { + return { + isGated: true, + totalCents: 10, + price: '$0.10', + gatedFields: [{ field: 'graphql', count: 1, cents: 10 }], + description: 'GraphQL queries for price data and market information', + } + } + + let doc: DocumentNode + try { + doc = parse(queryString) + } catch { + return { + isGated: false, + totalCents: 0, + price: '$0.00', + gatedFields: [], + description: 'Invalid query syntax', + } + } + + const rootFields: string[] = [] + const fieldCounts = new Map() + + visit(doc, { + Field(node) { + const fieldName = node.name.value + rootFields.push(fieldName) + const matched = gatedFieldList.find(g => g.field === fieldName) + if (matched) { + fieldCounts.set(fieldName, (fieldCounts.get(fieldName) ?? 0) + 1) + } + }, + }) + + // If every field in the query is an explicitly free field, no payment is required (#358) + if (rootFields.length > 0 && rootFields.every(f => FREE_GRAPHQL_FIELDS.has(f))) { + return { + isGated: false, + totalCents: 0, + price: '$0.00', + gatedFields: [], + description: 'Free GraphQL query (metadata / schema)', + } + } + + // If specific gated fields were identified, scale price by their configured cents (#358) + if (fieldCounts.size > 0) { + let totalCents = 0 + const gatedFields: { field: string; count: number; cents: number }[] = [] + const descParts: string[] = [] + + for (const [field, count] of fieldCounts.entries()) { + const config = gatedFieldList.find(g => g.field === field)! + const centsForField = config.priceCents * count + totalCents += centsForField + gatedFields.push({ field, count, cents: centsForField }) + descParts.push(`${count > 1 ? `${count}x ` : ''}${field} ($${(centsForField / 100).toFixed(2)})`) + } + + const price = `$${(totalCents / 100).toFixed(2)}` + const description = `GraphQL query requires payment for gated fields: ${descParts.join(', ')}` + + return { + isGated: true, + totalCents, + price, + gatedFields, + description, + } + } + + // Default fallback for general GraphQL queries (e.g. test queries) + return { + isGated: true, + totalCents: 10, + price: '$0.10', + gatedFields: [{ field: 'graphql', count: 1, cents: 10 }], + description: 'GraphQL queries for price data and market information', + } +} + /** * Checks whether `routedPath` matches `prefix` at a path-segment boundary. * @@ -103,6 +235,25 @@ async function x402Plugin(app: FastifyInstance) { const matchedRoute = matchGatedRoute(routedPath, req.method) if (!matchedRoute) return + let price = matchedRoute.price + let description = matchedRoute.description + let totalCents = parseCents(price) + + // For POST /graphql: derive pricing from requested fields rather than static path (#358) + if (matchesPathSegment(routedPath, '/graphql') && req.method.toUpperCase() === 'POST') { + const bodyQuery = (req.body as any)?.query + if (typeof bodyQuery === 'string') { + const paymentReq = evaluateGraphQLQueryPayment(bodyQuery) + // If the query contains only ungated fields (e.g. listPairs, introspection), allow free access + if (!paymentReq.isGated) { + return + } + price = paymentReq.price + description = paymentReq.description + totalCents = paymentReq.totalCents + } + } + // Falls back to testnet when the network selector plugin isn't // registered (e.g. isolated unit tests that build the app directly). const network = req.network ?? 'testnet' @@ -112,7 +263,6 @@ async function x402Plugin(app: FastifyInstance) { return } - const { price, description } = matchedRoute const paymentHeader = req.headers['x-payment'] as string | undefined const requirements = { @@ -160,7 +310,7 @@ async function x402Plugin(app: FastifyInstance) { const config = await getQuotaConfig(apiKeyId) if (config.overagePolicy === 'allow_overage') { // Record usage and let through (overage billing applies) - await recordUsage(apiKeyId, parseCents(price)) + await recordUsage(apiKeyId, totalCents) } else if (config.overagePolicy === 'charge_402') { reply.status(402).send({ error: 'Quota exceeded — additional payment required', @@ -176,7 +326,7 @@ async function x402Plugin(app: FastifyInstance) { return } } else { - await recordUsage(apiKeyId, parseCents(price)) + await recordUsage(apiKeyId, totalCents) } }