From 8733f47d9ccd708f4584f71f1ab747f5ffcc7424 Mon Sep 17 00:00:00 2001 From: sparkysolutions10-pixelmy Date: Sat, 17 Oct 2026 17:03:13 +0530 Subject: [PATCH] feat(graphql): schema artifact, metrics, persisted queries, and test suite (#372, #374, #375, #378) - Export canonical GraphQL SDL artifact at schema.graphql with zero-drift CI guard (closes #372) - Add bounded Prometheus metrics for operations, resolver durations, and error categories (closes #374) - Add comprehensive GraphQL query test suite with REST parity assertions and issuer preservation (closes #375) - Support persisted queries with configurable arbitrary document toggle and registry tooling (closes #378) --- ...raphql-schema-metrics-persisted-queries.md | 5 + .env.example | 6 + README.md | 6 + docs/graphql.md | 97 ++++ package.json | 3 + queries/manifest.json | 27 ++ schema.graphql | 86 ++++ scripts/export-schema.ts | 34 ++ scripts/register-query.ts | 75 ++++ src/__tests__/graphqlMetrics.test.ts | 224 ++++++++++ src/__tests__/graphqlPersistedQueries.test.ts | 209 +++++++++ src/__tests__/graphqlQueries.test.ts | 294 ++++++++++++ src/__tests__/graphqlSchema.test.ts | 58 +++ src/api/graphql.ts | 421 ++++++++++-------- src/api/persistedQueries.ts | 129 ++++++ src/api/schema.ts | 97 ++++ src/metrics.ts | 29 ++ 17 files changed, 1624 insertions(+), 176 deletions(-) create mode 100644 .changeset/graphql-schema-metrics-persisted-queries.md create mode 100644 queries/manifest.json create mode 100644 schema.graphql create mode 100644 scripts/export-schema.ts create mode 100644 scripts/register-query.ts create mode 100644 src/__tests__/graphqlMetrics.test.ts create mode 100644 src/__tests__/graphqlPersistedQueries.test.ts create mode 100644 src/__tests__/graphqlQueries.test.ts create mode 100644 src/__tests__/graphqlSchema.test.ts create mode 100644 src/api/persistedQueries.ts create mode 100644 src/api/schema.ts diff --git a/.changeset/graphql-schema-metrics-persisted-queries.md b/.changeset/graphql-schema-metrics-persisted-queries.md new file mode 100644 index 00000000..2805f78a --- /dev/null +++ b/.changeset/graphql-schema-metrics-persisted-queries.md @@ -0,0 +1,5 @@ +--- +'lens': minor +--- + +Export GraphQL SDL schema artifact, enforce zero drift check in CI, add bounded Prometheus metrics, implement persisted query support with configurable arbitrary query toggle, and add comprehensive query execution test suite (#372, #374, #375, #378). diff --git a/.env.example b/.env.example index 41e4465f..4a02a5e9 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,12 @@ PRICE_CACHE_TTL=10 # Default timeout for outbound HTTP requests in milliseconds (default: 10000) HTTP_TIMEOUT_MS=10000 +# --- GraphQL --- +# Whether arbitrary (unpersisted) GraphQL documents are accepted. +# Defaults to true in development/test, and false in production (enforcing persisted queries only). +GRAPHQL_ALLOW_ARBITRARY_QUERIES=true + + # --- Active Network --- # Which network the node operates on: "testnet" or "mainnet" (default: testnet) # This drives all config.horizon / config.rpc / config.soroswap shortcut paths. diff --git a/README.md b/README.md index 8351617e..3a85a115 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,12 @@ curl "https://api.example.com/price/XLM/USDC?network=mainnet" Available at `/graphql` with GraphiQL IDE at `/graphiql`. Real-time price streaming is available via the `priceUpdated` [subscription](#graphql-subscriptions-live-prices). +- **Schema Artifact**: Committed SDL artifact at [`schema.graphql`](schema.graphql). + - Refresh schema: `npm run schema:export` + - CI drift check: `npm run schema:check` +- **Persisted Queries**: Apollo APQ and hash execution supported with configurable arbitrary-document toggle (`GRAPHQL_ALLOW_ARBITRARY_QUERIES`, default `false` in production). Register queries via `npm run query:register`. +- **Metrics**: GraphQL operations (`graphql_operations_total`), database resolver durations (`graphql_resolver_duration_seconds`), and categorized errors (`graphql_errors_total`) with bounded cardinality on `/metrics`. + For complete GraphQL schema documentation, all queries with runnable `curl` examples, WebSocket `graphql-transport-ws` handshake specifications, and an explicit comparison against the REST surface, see [`docs/graphql.md`](docs/graphql.md). ```graphql diff --git a/docs/graphql.md b/docs/graphql.md index b4385131..07b8418d 100644 --- a/docs/graphql.md +++ b/docs/graphql.md @@ -19,6 +19,9 @@ Lens exposes a high-performance GraphQL API alongside its REST interface, powere - [Runnable Client Example (Raw WebSocket & Handshake)](#runnable-client-example-raw-websocket--handshake) - [Runnable Client Example (`graphql-ws` Library)](#runnable-client-example-graphql-ws-library) 5. [Capabilities & Gaps Against REST](#capabilities--gaps-against-rest) +6. [Schema Artifact & CI Drift Guard](#schema-artifact--ci-drift-guard) +7. [Persisted Queries & Security](#persisted-queries--security) +8. [Prometheus Metrics & Observability](#prometheus-metrics--observability) --- @@ -548,3 +551,97 @@ The following matrix documents the differences explicitly: | **Indexer Synchronization & Health** | `GET /status` | *No equivalent* | **REST Only**: Checking ledger heights, node lag seconds, and indexer operational status is REST-only. | | **x402 Micropayment Facilitator** | `/verify`, `/settle`, `/supported`, `/discovery/resources`, `/usage/me` | *No equivalent* | **REST Only**: All x402 payment validation, settlement, and bazaar resource discovery routes are REST-only. | | **Webhook Management** | `POST /webhooks`, `DELETE /webhooks/:id` | *No equivalent* | **REST Only**: Subscribing to price-threshold alert webhooks is REST-only. | + +--- + +## Schema Artifact & CI Drift Guard + +Lens provides a committed GraphQL Schema Definition Language (SDL) artifact at [`schema.graphql`](../schema.graphql). This allows client code generators (Apollo, GraphQL Code Generator, Relay) and IDE tooling to inspect the full schema statically without booting a live server. + +### Commands + +- **Export/Regenerate Schema:** + ```bash + npm run schema:export + ``` + Exports the canonical, formatted SDL from `src/api/schema.ts` directly into `schema.graphql`. + +- **Verify Zero Schema Drift (CI Check):** + ```bash + npm run schema:check + ``` + Fails with exit code `1` if `schema.graphql` does not match the schema defined in TypeScript. + +--- + +## Persisted Queries & Security + +To prevent Denial of Service (DoS) via expensive, unbounded adversarial GraphQL documents, Lens supports **persisted queries**. Clients send a cryptographic SHA-256 hash instead of a full query string. + +### Arbitrary Query Toggle (`GRAPHQL_ALLOW_ARBITRARY_QUERIES`) + +- **Development / Test (`true`)**: Arbitrary raw GraphQL query strings are accepted alongside registered hashes. +- **Production (`false` by default)**: Only registered persisted queries are accepted. Any arbitrary query string is rejected with `400 Bad Request`: + ```json + { + "data": null, + "errors": [ + { + "message": "Arbitrary GraphQL queries are disabled. Only registered persisted queries are allowed." + } + ] + } + ``` + +### Executing a Persisted Query + +Clients can execute persisted queries using the standard **Apollo Automatic Persisted Queries (APQ)** format: + +```bash +curl -X POST http://localhost:3002/graphql \ + -H "Content-Type: application/json" \ + -d '{ + "extensions": { + "persistedQuery": { + "version": 1, + "sha256Hash": "99d1aac6bd493749bd5db715d4ccb398a3fd85796ee1fbaccfc39740a9548a67" + } + }, + "variables": { + "assetA": "XLM", + "assetB": "USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN" + } + }' +``` + +Or via the Mercurius persisted format: +```bash +curl -X POST http://localhost:3002/graphql \ + -H "Content-Type: application/json" \ + -d '{ + "persisted": true, + "query": "99d1aac6bd493749bd5db715d4ccb398a3fd85796ee1fbaccfc39740a9548a67", + "variables": { "assetA": "XLM", "assetB": "USDC" } + }' +``` + +### Registering Queries + +The query manifest lives in [`queries/manifest.json`](../queries/manifest.json). To register a new query file or string: +```bash +npm run query:register [OperationName] +``` +The script validates the document against the schema, computes its SHA-256 hash, and appends it to the manifest. + +--- + +## Prometheus Metrics & Observability + +Lens exports dedicated GraphQL metrics on `GET /metrics` with strictly bounded label cardinality: + +| Metric Name | Type | Labels | Description | +|---|---|---|---| +| `graphql_operations_total` | Counter | `operation_name`, `operation_type` | Counts GraphQL operations by operation name (`getPrice`, `getBestRoute`, `getPriceHistory`, `listPairs`, `priceUpdated`, `other`) and operation type (`query`, `subscription`). Unknown client-provided names are folded into `"other"` to prevent time-series cardinality explosion. | +| `graphql_resolver_duration_seconds` | Histogram | `field` | Execution duration in seconds for expensive database-backed resolvers (`getPrice`, `getBestRoute`, `getPriceHistory`). | +| `graphql_errors_total` | Counter | `category` | GraphQL errors classified into bounded categories: `"validation"`, `"not_found"`, `"unauthorized"`, `"persisted_query"`, `"internal"`, or `"other"`. | + diff --git a/package.json b/package.json index c25cfd9e..545ebb17 100644 --- a/package.json +++ b/package.json @@ -9,6 +9,9 @@ "start": "node dist/index.js", "test": "vitest run", "openapi:gen": "tsx scripts/generate-openapi.ts", + "schema:export": "tsx scripts/export-schema.ts", + "schema:check": "tsx scripts/export-schema.ts --check", + "query:register": "tsx scripts/register-query.ts", "changeset": "changeset", "version-packages": "changeset version", "release": "changeset tag", diff --git a/queries/manifest.json b/queries/manifest.json new file mode 100644 index 00000000..05159aef --- /dev/null +++ b/queries/manifest.json @@ -0,0 +1,27 @@ +{ + "99d1aac6bd493749bd5db715d4ccb398a3fd85796ee1fbaccfc39740a9548a67": { + "name": "GetPrice", + "hash": "99d1aac6bd493749bd5db715d4ccb398a3fd85796ee1fbaccfc39740a9548a67", + "query": "query GetPrice($assetA: String!, $assetB: String!) {\n getPrice(assetA: $assetA, assetB: $assetB) {\n assetA\n assetB\n pairKey\n price\n sdexPrice\n ammPrice\n bestRoute\n volume24h\n sdexVolume24h\n ammVolume24h\n vwap1m\n vwap5m\n vwap1h\n vwap24h\n priceChange24h\n lastUpdated\n sources\n confidence\n lastTradeAgeSeconds\n }\n}" + }, + "741b1ab3f437d3f016304fe4102d5df6d00858b8f4cb65c87c4251c24d47f566": { + "name": "GetBestRoute", + "hash": "741b1ab3f437d3f016304fe4102d5df6d00858b8f4cb65c87c4251c24d47f566", + "query": "query GetBestRoute($assetA: String!, $assetB: String!, $amount: Float!) {\n getBestRoute(assetA: $assetA, assetB: $assetB, amount: $amount) {\n route\n sdexPrice\n ammPrice\n estimatedOutput\n slippagePct\n recommendation\n }\n}" + }, + "47c189a4cea77c48994a7d9c8b0ce810733f5e0312a71a00743e8331ba64d1f1": { + "name": "GetPriceHistory", + "hash": "47c189a4cea77c48994a7d9c8b0ce810733f5e0312a71a00743e8331ba64d1f1", + "query": "query GetPriceHistory($assetA: String!, $assetB: String!, $window: String!, $limit: Int) {\n getPriceHistory(assetA: $assetA, assetB: $assetB, window: $window, limit: $limit) {\n bucket\n window\n vwap\n sdexVwap\n ammVwap\n volume\n tradeCount\n open\n close\n high\n low\n }\n}" + }, + "c655423a31d157afc1f5b492f50189bff13ad7c781a2147bd82721a3eeee0bab": { + "name": "ListPairs", + "hash": "c655423a31d157afc1f5b492f50189bff13ad7c781a2147bd82721a3eeee0bab", + "query": "query ListPairs {\n listPairs\n}" + }, + "c991c45a10395c6d0a0c5d9b93f0ef5c4ef23033638d9b9716c371ab7de5f7bd": { + "name": "PriceUpdated", + "hash": "c991c45a10395c6d0a0c5d9b93f0ef5c4ef23033638d9b9716c371ab7de5f7bd", + "query": "subscription PriceUpdated($pair: String!, $network: String) {\n priceUpdated(pair: $pair, network: $network) {\n pair\n price\n ts\n network\n }\n}" + } +} diff --git a/schema.graphql b/schema.graphql new file mode 100644 index 00000000..bdf720ec --- /dev/null +++ b/schema.graphql @@ -0,0 +1,86 @@ +type AggregatedPrice { + assetA: String! + assetB: String! + pairKey: String! + price: Float! + sdexPrice: Float! + ammPrice: Float! + bestRoute: String! + volume24h: Float! + sdexVolume24h: Float! + ammVolume24h: Float! + vwap1m: Float! + vwap5m: Float! + vwap1h: Float! + vwap24h: Float! + priceChange24h: Float! + lastUpdated: String! + sources: Int! + confidence: String! + lastTradeAgeSeconds: Int +} + +type RouteInfo { + route: String! + sdexPrice: Float! + ammPrice: Float! + estimatedOutput: Float! + slippagePct: Float! + recommendation: String! +} + +type PriceBucket { + bucket: String! + window: String! + vwap: Float! + sdexVwap: Float + ammVwap: Float + volume: Float! + tradeCount: Int! + open: Float + close: Float + high: Float + low: Float +} + +type PriceUpdate { + pair: String! + price: Float! + ts: String! + + """Which chain the price came from — testnet or mainnet.""" + network: String! +} + +type Query { + """ + Aggregated price for a pair. Three outcomes are kept distinct: + + - Malformed asset input (an asset that is not CODE or CODE:ISSUER) fails + with a GraphQL error whose extensions.code is INVALID_INPUT. + - A well-formed pair that is not tracked on the active network fails with + a GraphQL error whose extensions.code is PAIR_NOT_FOUND. + - A tracked pair with no trade data yet is not an error: it resolves to an + AggregatedPrice with zeroed figures rather than null or an error. + + The result is only null if the resolver itself cannot produce a payload; + clients should branch on extensions.code, not on null. + """ + 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! +} diff --git a/scripts/export-schema.ts b/scripts/export-schema.ts new file mode 100644 index 00000000..1257b1d0 --- /dev/null +++ b/scripts/export-schema.ts @@ -0,0 +1,34 @@ +#!/usr/bin/env tsx +import { readFileSync, writeFileSync, existsSync } from 'node:fs' +import { resolve } from 'node:path' +import { getCanonicalSchemaSDL } from '../src/api/schema' + +const rootDir = resolve(__dirname, '..') +const targetFile = resolve(rootDir, 'schema.graphql') + +const isCheck = process.argv.includes('--check') + +const generatedSDL = getCanonicalSchemaSDL() + +if (isCheck) { + if (!existsSync(targetFile)) { + console.error(`❌ Error: schema.graphql not found at ${targetFile}`) + console.error(`Run 'npm run schema:export' to generate it.`) + process.exit(1) + } + + const existingSDL = readFileSync(targetFile, 'utf8').replace(/\r\n/g, '\n') + const normalizedGenerated = generatedSDL.replace(/\r\n/g, '\n') + + if (existingSDL.trim() !== normalizedGenerated.trim()) { + console.error(`❌ Error: schema.graphql is out of sync with src/api/schema.ts`) + console.error(`Run 'npm run schema:export' to regenerate the GraphQL SDL artifact.`) + process.exit(1) + } + + console.log(`✅ schema.graphql is in sync with src/api/schema.ts`) + process.exit(0) +} else { + writeFileSync(targetFile, generatedSDL, 'utf8') + console.log(`✅ Exported GraphQL schema SDL to ${targetFile}`) +} diff --git a/scripts/register-query.ts b/scripts/register-query.ts new file mode 100644 index 00000000..6851d234 --- /dev/null +++ b/scripts/register-query.ts @@ -0,0 +1,75 @@ +#!/usr/bin/env tsx +import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'node:fs' +import { resolve } from 'node:path' +import { parse, validate, buildSchema } from 'graphql' +import { schema } from '../src/api/schema' +import { hashQuery, registerPersistedQuery } from '../src/api/persistedQueries' + +const rootDir = resolve(__dirname, '..') +const manifestDir = resolve(rootDir, 'queries') +const manifestFile = resolve(manifestDir, 'manifest.json') + +function loadManifest(): Record { + if (existsSync(manifestFile)) { + try { + return JSON.parse(readFileSync(manifestFile, 'utf8')) + } catch { + return {} + } + } + return {} +} + +function saveManifest(data: Record) { + if (!existsSync(manifestDir)) { + mkdirSync(manifestDir, { recursive: true }) + } + writeFileSync(manifestFile, JSON.stringify(data, null, 2) + '\n', 'utf8') +} + +export function registerQueryFromFileOrString(input: string, name?: string): { hash: string; query: string } { + let queryText = input + if (existsSync(input)) { + queryText = readFileSync(input, 'utf8') + } + + // Validate the query against the schema + const compiledSchema = buildSchema(schema) + const document = parse(queryText) + const errors = validate(compiledSchema, document) + if (errors.length > 0) { + const messages = errors.map(e => e.message).join('; ') + throw new Error(`Invalid GraphQL query: ${messages}`) + } + + const hash = registerPersistedQuery(queryText) + const manifest = loadManifest() + manifest[hash] = { + name: name || undefined, + hash, + query: queryText.trim(), + } + saveManifest(manifest) + return { hash, query: queryText.trim() } +} + +if (require.main === module || process.argv[1]?.endsWith('register-query.ts')) { + const args = process.argv.slice(2) + if (args.length === 0) { + console.log('Usage: npx tsx scripts/register-query.ts [operationName]') + process.exit(0) + } + + try { + const input = args[0] + const opName = args[1] + const result = registerQueryFromFileOrString(input, opName) + console.log(`✅ Successfully registered persisted query:`) + console.log(` Hash: ${result.hash}`) + if (opName) console.log(` Name: ${opName}`) + console.log(` Manifest updated at: ${manifestFile}`) + } catch (err: any) { + console.error(`❌ Failed to register query: ${err.message}`) + process.exit(1) + } +} diff --git a/src/__tests__/graphqlMetrics.test.ts b/src/__tests__/graphqlMetrics.test.ts new file mode 100644 index 00000000..96a183d6 --- /dev/null +++ b/src/__tests__/graphqlMetrics.test.ts @@ -0,0 +1,224 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' + +const mocks = vi.hoisted(() => ({ + getCachedPrice: vi.fn(), + getAggregatedPrice: vi.fn(), + getBestRoute: vi.fn(), + pgPoolQuery: vi.fn(), +})) + +vi.mock('../redis', () => ({ getCachedPrice: mocks.getCachedPrice })) +vi.mock('../db', () => ({ pgPool: { query: mocks.pgPoolQuery }, prisma: {} })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: mocks.getAggregatedPrice })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mocks.getBestRoute })) +vi.mock('../config', () => ({ + activeNetwork: 'testnet', + config: { + cache: { priceTtl: 10 }, + pairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: null }, + }, + ], + }, +})) + +import { registerGraphQL } from '../api/graphql' +import { + graphql_operations_total, + graphql_resolver_duration_seconds, + graphql_errors_total, + getMetrics, +} from '../metrics' + +describe('GraphQL Prometheus Metrics & Label Cardinality Bounding', () => { + let app: FastifyInstance + + beforeEach(async () => { + vi.clearAllMocks() + mocks.getCachedPrice.mockResolvedValue(null) + mocks.getAggregatedPrice.mockResolvedValue({ + price: 0.125, + sdexPrice: 0.124, + ammPrice: 0.126, + volume24h: 1000, + sdexVolume24h: 500, + ammVolume24h: 500, + vwap1m: 0.125, + vwap5m: 0.125, + vwap1h: 0.125, + vwap24h: 0.125, + priceChange24h: 1.5, + sources: 2, + confidence: 'high', + lastTradeAgeSeconds: 10, + }) + mocks.getBestRoute.mockResolvedValue({ + route: 'SDEX', + sdexPrice: 0.124, + ammPrice: 0.126, + estimatedOutput: 125, + slippagePct: 0.05, + recommendation: 'Use SDEX', + }) + mocks.pgPoolQuery.mockResolvedValue({ + rows: [ + { + bucket: new Date('2026-06-01T00:00:00Z'), + window: '1h', + vwap: 0.125, + sdex_vwap: 0.124, + amm_vwap: 0.126, + volume: 5000, + trade_count: 50, + open_price: 0.12, + close_price: 0.125, + high_price: 0.13, + low_price: 0.119, + }, + ], + }) + + app = Fastify({ logger: false }) + await registerGraphQL(app, { allowArbitraryQueries: true }) + await app.ready() + }) + + afterEach(async () => { + await app.close() + }) + + it('counts operations by operation name and type for named queries', async () => { + const beforeVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'GetPrice' && v.labels.operation_type === 'query' + )?.value ?? 0 + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `query GetPrice { getPrice(assetA: "XLM", assetB: "USDC") { price } }`, + }, + }) + expect(res.statusCode).toBe(200) + + const afterVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'GetPrice' && v.labels.operation_type === 'query' + )?.value ?? 0 + + expect(afterVal).toBe(beforeVal + 1) + }) + + it('infers root field for anonymous queries without exploding labels', async () => { + const beforeVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'getBestRoute' && v.labels.operation_type === 'query' + )?.value ?? 0 + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getBestRoute(assetA: "XLM", assetB: "USDC", amount: 100) { route } }`, + }, + }) + expect(res.statusCode).toBe(200) + + const afterVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'getBestRoute' && v.labels.operation_type === 'query' + )?.value ?? 0 + + expect(afterVal).toBe(beforeVal + 1) + }) + + it('folds arbitrary/client-generated operation names into "other" to bound cardinality', async () => { + const beforeVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'other' && v.labels.operation_type === 'query' + )?.value ?? 0 + + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `query MaliciousClientRandomUUID_987123 { listPairs }`, + }, + }) + expect(res.statusCode).toBe(200) + + const afterVal = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'other' && v.labels.operation_type === 'query' + )?.value ?? 0 + + expect(afterVal).toBe(beforeVal + 1) + + // Verify the arbitrary string is NOT created as a label + const customSeries = (await graphql_operations_total.get()).values.find( + v => v.labels.operation_name === 'MaliciousClientRandomUUID_987123' + ) + expect(customSeries).toBeUndefined() + }) + + it('records resolver duration for database-backed fields', async () => { + await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getPrice(assetA: "XLM", assetB: "USDC") { price } }`, + }, + }) + + const durationValues = (await graphql_resolver_duration_seconds.get()).values + const priceResolverSamples = durationValues.filter(v => v.labels.field === 'getPrice') + expect(priceResolverSamples.length).toBeGreaterThan(0) + }) + + it('counts errors classified into bounded categories', async () => { + const beforeValidation = (await graphql_errors_total.get()).values.find( + v => v.labels.category === 'validation' + )?.value ?? 0 + + // Trigger validation error + const valRes = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getPrice(assetA: "INVALID:::STRING", assetB: "USDC") { price } }`, + }, + }) + expect(valRes.statusCode).toBe(200) // GraphQL returns errors array in 200 payload + expect(valRes.json().errors).toBeDefined() + + const afterValidation = (await graphql_errors_total.get()).values.find( + v => v.labels.category === 'validation' + )?.value ?? 0 + expect(afterValidation).toBe(beforeValidation + 1) + + // Trigger not_found error + const beforeNotFound = (await graphql_errors_total.get()).values.find( + v => v.labels.category === 'not_found' + )?.value ?? 0 + + const notFoundRes = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getPrice(assetA: "XLM", assetB: "NONEXISTENT") { price } }`, + }, + }) + expect(notFoundRes.json().errors[0].extensions.code).toBe('PAIR_NOT_FOUND') + + const afterNotFound = (await graphql_errors_total.get()).values.find( + v => v.labels.category === 'not_found' + )?.value ?? 0 + expect(afterNotFound).toBe(beforeNotFound + 1) + }) + + it('renders all GraphQL metrics in the /metrics endpoint registry', async () => { + const metricsOutput = await getMetrics() + expect(metricsOutput).toContain('graphql_operations_total') + expect(metricsOutput).toContain('graphql_resolver_duration_seconds') + expect(metricsOutput).toContain('graphql_errors_total') + }) +}) diff --git a/src/__tests__/graphqlPersistedQueries.test.ts b/src/__tests__/graphqlPersistedQueries.test.ts new file mode 100644 index 00000000..1ea9c4a5 --- /dev/null +++ b/src/__tests__/graphqlPersistedQueries.test.ts @@ -0,0 +1,209 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' + +const mocks = vi.hoisted(() => ({ + getCachedPrice: vi.fn(), + getAggregatedPrice: vi.fn(), + getBestRoute: vi.fn(), + pgPoolQuery: vi.fn(), +})) + +vi.mock('../redis', () => ({ getCachedPrice: mocks.getCachedPrice })) +vi.mock('../db', () => ({ pgPool: { query: mocks.pgPoolQuery }, prisma: {} })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: mocks.getAggregatedPrice })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mocks.getBestRoute })) +vi.mock('../config', () => ({ + activeNetwork: 'testnet', + config: { + cache: { priceTtl: 10 }, + pairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: null }, + }, + ], + }, +})) + +import { registerGraphQL } from '../api/graphql' +import { + hashQuery, + registerPersistedQuery, + CANONICAL_QUERIES, +} from '../api/persistedQueries' + +describe('GraphQL Persisted Queries & Arbitrary Document Toggle', () => { + beforeEach(() => { + vi.clearAllMocks() + mocks.getCachedPrice.mockResolvedValue(null) + mocks.getAggregatedPrice.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: 1, + confidence: 'high', + lastTradeAgeSeconds: 5, + }) + }) + + describe('when arbitrary queries are permitted (default in dev/test)', () => { + let app: FastifyInstance + + beforeEach(async () => { + app = Fastify({ logger: false }) + await registerGraphQL(app, { allowArbitraryQueries: true }) + await app.ready() + }) + + afterEach(async () => { + await app.close() + }) + + it('executes an arbitrary raw GraphQL query string', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ listPairs }`, + }, + }) + expect(res.statusCode).toBe(200) + expect(res.json().data.listPairs).toEqual(['USDC/XLM']) + }) + + it('executes a registered query by SHA-256 hash using Apollo APQ extensions', async () => { + const hash = hashQuery(CANONICAL_QUERIES.ListPairs) + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + extensions: { + persistedQuery: { + version: 1, + sha256Hash: hash, + }, + }, + }, + }) + expect(res.statusCode).toBe(200) + expect(res.json().data.listPairs).toEqual(['USDC/XLM']) + }) + + it('executes a registered query by hash using mercurius persisted format', async () => { + const hash = hashQuery(CANONICAL_QUERIES.ListPairs) + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + persisted: true, + query: hash, + }, + }) + expect(res.statusCode).toBe(200) + expect(res.json().data.listPairs).toEqual(['USDC/XLM']) + }) + + it('returns a clear error when an unknown hash is supplied', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + extensions: { + persistedQuery: { + version: 1, + sha256Hash: 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855', + }, + }, + }, + }) + expect(res.statusCode).toBe(400) + const body = res.json() + expect(body.errors).toBeDefined() + expect(body.errors[0].message).toContain('Persisted query not found') + }) + }) + + describe('when arbitrary queries are disabled (safe production mode)', () => { + let app: FastifyInstance + + beforeEach(async () => { + app = Fastify({ logger: false }) + await registerGraphQL(app, { allowArbitraryQueries: false }) + await app.ready() + }) + + afterEach(async () => { + await app.close() + }) + + it('rejects arbitrary unpersisted queries with clear security error', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ listPairs }`, + }, + }) + expect(res.statusCode).toBe(400) + const body = res.json() + expect(body.errors).toBeDefined() + expect(body.errors[0].message).toContain('Arbitrary GraphQL queries are disabled') + }) + + it('still executes registered queries by hash when arbitrary queries are disabled', async () => { + const hash = hashQuery(CANONICAL_QUERIES.ListPairs) + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + extensions: { + persistedQuery: { + version: 1, + sha256Hash: hash, + }, + }, + }, + }) + expect(res.statusCode).toBe(200) + expect(res.json().data.listPairs).toEqual(['USDC/XLM']) + }) + + it('allows dynamically registering a new query and executing it by hash', async () => { + const customQuery = `query CustomPairQuery { listPairs }` + const customHash = registerPersistedQuery(customQuery) + + // Re-register or inject into instance + const dynamicApp = Fastify({ logger: false }) + await registerGraphQL(dynamicApp, { + allowArbitraryQueries: false, + persistedQueries: { [customHash]: customQuery }, + }) + await dynamicApp.ready() + + const res = await dynamicApp.inject({ + method: 'POST', + url: '/graphql', + payload: { + extensions: { + persistedQuery: { + version: 1, + sha256Hash: customHash, + }, + }, + }, + }) + expect(res.statusCode).toBe(200) + expect(res.json().data.listPairs).toEqual(['USDC/XLM']) + await dynamicApp.close() + }) + }) +}) diff --git a/src/__tests__/graphqlQueries.test.ts b/src/__tests__/graphqlQueries.test.ts new file mode 100644 index 00000000..54d16b32 --- /dev/null +++ b/src/__tests__/graphqlQueries.test.ts @@ -0,0 +1,294 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' + +const mocks = vi.hoisted(() => ({ + ISSUER_CIRCLE: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN', + ISSUER_OTHER: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + + getCachedPrice: vi.fn(), + getAggregatedPrice: vi.fn(), + getBestRoute: vi.fn(), + pgPoolQuery: vi.fn(), +})) + +vi.mock('../redis', () => ({ getCachedPrice: mocks.getCachedPrice })) +vi.mock('../db', () => ({ pgPool: { query: mocks.pgPoolQuery }, prisma: {} })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: mocks.getAggregatedPrice })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mocks.getBestRoute })) +vi.mock('../config', () => ({ + activeNetwork: 'testnet', + config: { + cache: { priceTtl: 10 }, + pairs: [ + { + pairKey: `USDC:${mocks.ISSUER_CIRCLE}/XLM`, + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: mocks.ISSUER_CIRCLE }, + }, + ], + }, +})) + + +import { registerGraphQL } from '../api/graphql' + +describe('GraphQL Query Execution & REST Parity Suite', () => { + let app: FastifyInstance + + const MOCK_AGGREGATE = { + price: 0.1825, + sdexPrice: 0.182, + ammPrice: 0.183, + bestRoute: 'SDEX', + volume24h: 1250000, + sdexVolume24h: 750000, + ammVolume24h: 500000, + vwap1m: 0.1824, + vwap5m: 0.1822, + vwap1h: 0.1819, + vwap24h: 0.181, + priceChange24h: 2.14, + sources: 2, + confidence: 'high', + lastTradeAgeSeconds: 12, + } + + const MOCK_ROUTE = { + route: 'SDEX', + sdexPrice: 0.182, + ammPrice: 0.183, + estimatedOutput: 182.4, + slippagePct: 0.08, + recommendation: 'SDEX provides best execution for 1000 XLM', + } + + const MOCK_DB_BUCKETS = [ + { + bucket: new Date('2026-06-01T12:00:00.000Z'), + window: '1h', + vwap: 0.182, + sdex_vwap: 0.1818, + amm_vwap: 0.1822, + volume: 45000, + trade_count: 120, + open_price: 0.18, + close_price: 0.1825, + high_price: 0.183, + low_price: 0.179, + }, + ] + + beforeEach(async () => { + vi.clearAllMocks() + mocks.getCachedPrice.mockResolvedValue(null) + mocks.getAggregatedPrice.mockResolvedValue(MOCK_AGGREGATE) + mocks.getBestRoute.mockResolvedValue(MOCK_ROUTE) + mocks.pgPoolQuery.mockResolvedValue({ rows: MOCK_DB_BUCKETS }) + + app = Fastify({ logger: false }) + await registerGraphQL(app, { allowArbitraryQueries: true }) + await app.ready() + }) + + afterEach(async () => { + await app.close() + }) + + describe('1. getPrice query & REST parity', () => { + const GET_PRICE_GQL = `query GetPrice($a: String!, $b: String!) { + getPrice(assetA: $a, assetB: $b) { + assetA assetB pairKey price sdexPrice ammPrice bestRoute + volume24h sdexVolume24h ammVolume24h vwap1m vwap5m vwap1h vwap24h + priceChange24h sources confidence lastTradeAgeSeconds + } + }` + + it('executes getPrice through GraphQL and matches aggregate data', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_PRICE_GQL, + variables: { a: 'XLM', b: 'USDC' }, + }, + }) + + expect(res.statusCode).toBe(200) + const data = res.json().data.getPrice + + // Field-by-field parity with aggregator / REST representation + expect(data.pairKey).toBe(`USDC:${mocks.ISSUER_CIRCLE}/XLM`) + expect(data.price).toBe(MOCK_AGGREGATE.price) + expect(data.sdexPrice).toBe(MOCK_AGGREGATE.sdexPrice) + expect(data.ammPrice).toBe(MOCK_AGGREGATE.ammPrice) + expect(data.bestRoute).toBe(MOCK_ROUTE.route) + expect(data.volume24h).toBe(MOCK_AGGREGATE.volume24h) + expect(data.vwap1m).toBe(MOCK_AGGREGATE.vwap1m) + expect(data.confidence).toBe(MOCK_AGGREGATE.confidence) + expect(data.sources).toBe(MOCK_AGGREGATE.sources) + }) + + it('enforces issuer matching in GraphQL (retains issuer instead of discarding)', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_PRICE_GQL, + variables: { a: 'XLM', b: `USDC:${mocks.ISSUER_CIRCLE}` }, + }, + }) + + expect(res.statusCode).toBe(200) + const data = res.json().data.getPrice + expect(data.pairKey).toBe(`USDC:${mocks.ISSUER_CIRCLE}/XLM`) + expect(data.price).toBe(MOCK_AGGREGATE.price) + }) + + it('returns PAIR_NOT_FOUND when issuer does not match tracked pair', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_PRICE_GQL, + variables: { a: 'XLM', b: `USDC:${mocks.ISSUER_OTHER}` }, + }, + }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.data.getPrice).toBeNull() + expect(body.errors).toBeDefined() + expect(body.errors[0].extensions.code).toBe('PAIR_NOT_FOUND') + }) + }) + + describe('2. getBestRoute query & REST parity', () => { + const GET_ROUTE_GQL = `query GetRoute($a: String!, $b: String!, $amount: Float!) { + getBestRoute(assetA: $a, assetB: $b, amount: $amount) { + route sdexPrice ammPrice estimatedOutput slippagePct recommendation + } + }` + + it('executes getBestRoute and returns exact route calculation', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_ROUTE_GQL, + variables: { a: 'XLM', b: 'USDC', amount: 1000 }, + }, + }) + + expect(res.statusCode).toBe(200) + const data = res.json().data.getBestRoute + + expect(data).toEqual({ + route: MOCK_ROUTE.route, + sdexPrice: MOCK_ROUTE.sdexPrice, + ammPrice: MOCK_ROUTE.ammPrice, + estimatedOutput: MOCK_ROUTE.estimatedOutput, + slippagePct: MOCK_ROUTE.slippagePct, + recommendation: MOCK_ROUTE.recommendation, + }) + }) + + it('returns null when pair is unknown for getBestRoute', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_ROUTE_GQL, + variables: { a: 'XLM', b: 'UNTRACKED', amount: 500 }, + }, + }) + + expect(res.statusCode).toBe(200) + expect(res.json().data.getBestRoute).toBeNull() + }) + }) + + describe('3. getPriceHistory query & REST parity', () => { + const GET_HISTORY_GQL = `query GetHistory($a: String!, $b: String!, $window: String!, $limit: Int) { + getPriceHistory(assetA: $a, assetB: $b, window: $window, limit: $limit) { + bucket window vwap sdexVwap ammVwap volume tradeCount open close high low + } + }` + + it('executes getPriceHistory and maps price aggregate buckets matching database schema', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: GET_HISTORY_GQL, + variables: { a: 'XLM', b: 'USDC', window: '1h', limit: 10 }, + }, + }) + + expect(res.statusCode).toBe(200) + const history = res.json().data.getPriceHistory + expect(history).toHaveLength(1) + expect(history[0]).toEqual({ + bucket: '2026-06-01T12:00:00.000Z', + window: '1h', + vwap: 0.182, + sdexVwap: 0.1818, + ammVwap: 0.1822, + volume: 45000, + tradeCount: 120, + open: 0.18, + close: 0.1825, + high: 0.183, + low: 0.179, + }) + }) + }) + + describe('4. listPairs query & parity', () => { + it('executes listPairs and returns watched pair keys', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ listPairs }`, + }, + }) + + expect(res.statusCode).toBe(200) + const pairs = res.json().data.listPairs + expect(pairs).toEqual([`USDC:${mocks.ISSUER_CIRCLE}/XLM`]) + }) + }) + + describe('5. Error paths coverage', () => { + it('rejects malformed asset format with INVALID_INPUT', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getPrice(assetA: "THISISWAYTOOLONGASSETNAME", assetB: "USDC") { price } }`, + }, + }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.data.getPrice).toBeNull() + expect(body.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(body.errors[0].extensions.field).toBe('assetA') + }) + + it('rejects completely unknown pair with PAIR_NOT_FOUND', async () => { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { + query: `{ getPrice(assetA: "XLM", assetB: "EURT") { price } }`, + }, + }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.data.getPrice).toBeNull() + expect(body.errors[0].extensions.code).toBe('PAIR_NOT_FOUND') + }) + }) +}) diff --git a/src/__tests__/graphqlSchema.test.ts b/src/__tests__/graphqlSchema.test.ts new file mode 100644 index 00000000..9dceb367 --- /dev/null +++ b/src/__tests__/graphqlSchema.test.ts @@ -0,0 +1,58 @@ +import { describe, it, expect } from 'vitest' +import { readFileSync, existsSync } from 'node:fs' +import { resolve } from 'node:path' +import { buildSchema, printSchema } from 'graphql' +import { schema, getCanonicalSchemaSDL } from '../api/schema' + +const rootDir = resolve(__dirname, '../..') +const schemaArtifactPath = resolve(rootDir, 'schema.graphql') + +describe('GraphQL Schema SDL Artifact & Drift Guard', () => { + it('committed schema.graphql artifact exists at repository root', () => { + expect(existsSync(schemaArtifactPath)).toBe(true) + }) + + it('committed schema.graphql matches the in-code schema with zero drift', () => { + const committedSDL = readFileSync(schemaArtifactPath, 'utf8').replace(/\r\n/g, '\n').trim() + const canonicalSDL = getCanonicalSchemaSDL().replace(/\r\n/g, '\n').trim() + expect(committedSDL).toBe(canonicalSDL) + }) + + it('committed schema.graphql is valid SDL parseable by standard tooling', () => { + const committedSDL = readFileSync(schemaArtifactPath, 'utf8') + const parsed = buildSchema(committedSDL) + expect(parsed).toBeDefined() + + const typeMap = parsed.getTypeMap() + expect(typeMap.AggregatedPrice).toBeDefined() + expect(typeMap.RouteInfo).toBeDefined() + expect(typeMap.PriceBucket).toBeDefined() + expect(typeMap.PriceUpdate).toBeDefined() + expect(typeMap.Query).toBeDefined() + expect(typeMap.Subscription).toBeDefined() + }) + + it('query root type contains all 4 queries with documentation', () => { + const parsed = buildSchema(getCanonicalSchemaSDL()) + const queryType = parsed.getQueryType() + expect(queryType).toBeDefined() + + const fields = queryType!.getFields() + expect(fields.getPrice).toBeDefined() + expect(fields.getBestRoute).toBeDefined() + expect(fields.getPriceHistory).toBeDefined() + expect(fields.listPairs).toBeDefined() + + expect(fields.getPrice.description).toContain('Aggregated price for a pair') + }) + + it('subscription root type contains priceUpdated stream', () => { + const parsed = buildSchema(getCanonicalSchemaSDL()) + const subType = parsed.getSubscriptionType() + expect(subType).toBeDefined() + + const fields = subType!.getFields() + expect(fields.priceUpdated).toBeDefined() + expect(fields.priceUpdated.description).toContain('Streams a PriceUpdate') + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index c8e24e29..f0aa1d31 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,8 +1,13 @@ import type { FastifyInstance } from 'fastify' import { ErrorWithProps } from 'mercurius' -import { price_requests_total } from '../metrics' +import { + price_requests_total, + graphql_operations_total, + graphql_resolver_duration_seconds, + graphql_errors_total, +} from '../metrics' import mercurius, { withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' -import { GraphQLError, type ExecutionResult } from 'graphql' +import { GraphQLError, type ExecutionResult, type OperationDefinitionNode, type FieldNode } from 'graphql' import { randomUUID } from 'node:crypto' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' @@ -11,6 +16,10 @@ import { pgPool } from '../db' import { config, activeNetwork } from '../config' import { parseAssetStr } from '../pairsRegistry' import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' +import { schema } from './schema' +import { DEFAULT_PERSISTED_QUERIES, hashQuery } from './persistedQueries' + +export { schema } // Mercurius pubsub topic that carries every new price. A single app-level // listener bridges the ingesters' in-process `priceEmitter` onto this topic; @@ -20,14 +29,21 @@ 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', + 'PERSISTED_QUERY_NOT_FOUND', + 'PERSISTED_QUERY_NOT_SUPPORTED', + 'FORBIDDEN', +]) const INTERNAL_ERROR_MESSAGE = 'Internal server error' 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. const original = err.originalError as - | { extensions?: { code?: unknown }; errors?: unknown; statusCode?: unknown } + | { extensions?: { code?: unknown }; errors?: unknown; statusCode?: unknown; message?: unknown } | undefined if (!original) return true // Mercurius wraps document validation failures (and other request-level @@ -35,7 +51,10 @@ function isUserFacing(err: GraphQLError): boolean { if (Array.isArray(original.errors)) return true if (typeof original.statusCode === 'number' && original.statusCode < 500) return true const code = original.extensions?.code - return typeof code === 'string' && USER_FACING_CODES.has(code) + if (typeof code === 'string' && USER_FACING_CODES.has(code)) return true + const msg = String(original.message ?? err.message ?? '') + if (msg.includes('Persisted query') || msg.includes('Arbitrary GraphQL queries are disabled')) return true + return false } export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusContext) { @@ -43,6 +62,31 @@ export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusCo const log = ctx?.reply?.log ?? ctx?.app?.log const errors = (execution.errors ?? []).map(err => { + // Record GraphQL error metric with strictly bounded category labels + const original = err.originalError as any + const code = original?.extensions?.code ?? (err.extensions as any)?.code + const msg = String(original?.message ?? err.message ?? '') + + let category = 'other' + if (code === 'INVALID_INPUT' || msg.includes('Syntax Error') || msg.includes('Cannot query field') || msg.includes('Variable')) { + category = 'validation' + } else if (code === 'PAIR_NOT_FOUND') { + category = 'not_found' + } else if (code === 'UNAUTHORIZED') { + category = 'unauthorized' + } else if ( + code === 'PERSISTED_QUERY_NOT_FOUND' || + code === 'PERSISTED_QUERY_NOT_SUPPORTED' || + msg.includes('Persisted query') || + msg.includes('Arbitrary GraphQL') + ) { + category = 'persisted_query' + } else if (code === 'INTERNAL_SERVER_ERROR' || !isUserFacing(err)) { + category = 'internal' + } + + graphql_errors_total.inc({ category }) + if (isUserFacing(err)) return err log?.error({ err: err.originalError ?? err, requestId, path: err.path }, 'unexpected graphql error') return new GraphQLError(INTERNAL_ERROR_MESSAGE, { @@ -54,149 +98,82 @@ export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusCo return defaultErrorFormatter({ ...execution, errors }, ctx) } -const schema = ` - type AggregatedPrice { - assetA: String! - assetB: String! - pairKey: String! - price: Float! - sdexPrice: Float! - ammPrice: Float! - bestRoute: String! - volume24h: Float! - sdexVolume24h: Float! - ammVolume24h: Float! - vwap1m: Float! - vwap5m: Float! - vwap1h: Float! - vwap24h: Float! - priceChange24h: Float! - lastUpdated: String! - sources: Int! - confidence: String! - lastTradeAgeSeconds: Int - } - - type RouteInfo { - route: String! - sdexPrice: Float! - ammPrice: Float! - estimatedOutput: Float! - slippagePct: Float! - recommendation: String! - } - - type PriceBucket { - bucket: String! - window: String! - vwap: Float! - sdexVwap: Float - ammVwap: Float - volume: Float! - tradeCount: Int! - open: Float - close: Float - high: Float - low: Float - } - - type PriceUpdate { - pair: String! - price: Float! - ts: String! - """Which chain the price came from — testnet or mainnet.""" - network: String! - } - - type Query { - """ - Aggregated price for a pair. Three outcomes are kept distinct: - - - Malformed asset input (an asset that is not CODE or CODE:ISSUER) fails - with a GraphQL error whose extensions.code is INVALID_INPUT. - - A well-formed pair that is not tracked on the active network fails with - a GraphQL error whose extensions.code is PAIR_NOT_FOUND. - - A tracked pair with no trade data yet is not an error: it resolves to an - AggregatedPrice with zeroed figures rather than null or an error. - - The result is only null if the resolver itself cannot produce a payload; - clients should branch on extensions.code, not on null. - """ - 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 { return [a, b].sort().join('/') } +/** + * Resolve a requested pair against the ones this network watches. + * Enforces issuer parity with REST routes (prevents asset confusion across issuers). + */ function findPair(assetA: string, assetB: string) { - const normalize = (a: string) => a.toLowerCase() === 'native' ? 'XLM' : a.split(':')[0].toUpperCase() - const cA = normalize(assetA) - const cB = normalize(assetB) - return config.pairs.find(p => { - const pA = p.assetA.code.toUpperCase() - const pB = p.assetB.code.toUpperCase() - return (cA === pA && cB === pB) || (cA === pB && cB === pA) - }) + const parse = (a: string) => { + if (a.toLowerCase() === 'native') return { code: 'XLM', issuer: null as string | null } + const [code, issuer] = a.split(':') + return { code: (code ?? '').toUpperCase(), issuer: issuer ?? null } + } + const qA = parse(assetA) + const qB = parse(assetB) + + const matches = ( + q: { code: string; issuer: string | null }, + side: { code: string; issuer?: string | null }, + ) => { + if (q.code !== side.code.toUpperCase()) return false + // Only enforce the issuer when both sides name one; XLM has none, and a + // caller passing a bare code is not asserting which issuer they meant. + if (!q.issuer || !side.issuer) return true + return q.issuer === side.issuer + } + + return config.pairs.find( + p => + (matches(qA, p.assetA) && matches(qB, p.assetB)) || + (matches(qA, p.assetB) && matches(qB, p.assetA)) + ) } const resolvers = { Query: { async getPrice(_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) { + const timer = graphql_resolver_duration_seconds.startTimer({ field: 'getPrice' }) price_requests_total.inc() - for (const [field, value] of [['assetA', assetA], ['assetB', assetB]] as const) { - if (!parseAssetStr(value)) { + try { + for (const [field, value] of [['assetA', assetA], ['assetB', assetB]] as const) { + if (!parseAssetStr(value)) { + throw new ErrorWithProps( + `Invalid ${field}: expected CODE or CODE:ISSUER, got "${value}"`, + { code: 'INVALID_INPUT', field }, + ) + } + } + const pair = findPair(assetA, assetB) + if (!pair) { throw new ErrorWithProps( - `Invalid ${field}: expected CODE or CODE:ISSUER, got "${value}"`, - { code: 'INVALID_INPUT', field }, + `Pair ${assetA}/${assetB} is not tracked on ${activeNetwork}`, + { code: 'PAIR_NOT_FOUND' }, ) } - } - const pair = findPair(assetA, assetB) - if (!pair) { - throw new ErrorWithProps( - `Pair ${assetA}/${assetB} is not tracked on ${activeNetwork}`, - { code: 'PAIR_NOT_FOUND' }, - ) - } - const pairKey = pair.pairKey - - // These queries carry no network argument, so they resolve against the - // active network's pair list (findPair above) and read that same - // network's price data — scoped, not pooled across both chains. - // Per-request network selection on the GraphQL surface is separate work. - // The cache key is network-prefixed to match the REST route and the - // refresh worker, so no deployment can serve another network's cached - // payload under a network-less key. - const cacheKey = `${activeNetwork}:${pairKey}` - const cached = await getCachedPrice(cacheKey) - if (cached) { - try { return JSON.parse(cached) } catch { /* fall through */ } - } + const pairKey = pair.pairKey + + // These queries carry no network argument, so they resolve against the + // active network's pair list (findPair above) and read that same + // network's price data — scoped, not pooled across both chains. + const cacheKey = `${activeNetwork}:${pairKey}` + const cached = await getCachedPrice(cacheKey) + if (cached) { + try { return JSON.parse(cached) } catch { /* fall through */ } + } - const agg = await getAggregatedPrice(pairKey, activeNetwork) - const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, activeNetwork) - return { - assetA, assetB, pairKey, ...agg, - bestRoute: route.route, - lastUpdated: new Date().toISOString(), + const agg = await getAggregatedPrice(pairKey, activeNetwork) + const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, activeNetwork) + return { + assetA, assetB, pairKey, ...agg, + bestRoute: route.route, + lastUpdated: new Date().toISOString(), + } + } finally { + timer() } }, @@ -204,39 +181,49 @@ const resolvers = { _: unknown, { assetA, assetB, amount }: { assetA: string; assetB: string; amount: number } ) { - const pair = findPair(assetA, assetB) - if (!pair) return null - return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount) + const timer = graphql_resolver_duration_seconds.startTimer({ field: 'getBestRoute' }) + try { + const pair = findPair(assetA, assetB) + if (!pair) return null + return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount) + } finally { + timer() + } }, async getPriceHistory( _: unknown, { assetA, assetB, window, limit = 100 }: { assetA: string; assetB: string; window: string; limit?: number } ) { - const pairKey = makePairKey(assetA, assetB) - const result = await pgPool.query( - `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, - volume::float, trade_count, open_price::float, close_price::float, - high_price::float, low_price::float - FROM price_aggregates - WHERE pair_key = $1 AND window = $2 - ORDER BY bucket DESC - LIMIT $3`, - [pairKey, window, Math.min(limit, 1000)] - ) - return result.rows.map(r => ({ - bucket: r.bucket.toISOString(), - window: r.window, - vwap: r.vwap, - sdexVwap: r.sdex_vwap, - ammVwap: r.amm_vwap, - volume: r.volume, - tradeCount: r.trade_count, - open: r.open_price, - close: r.close_price, - high: r.high_price, - low: r.low_price, - })) + const timer = graphql_resolver_duration_seconds.startTimer({ field: 'getPriceHistory' }) + try { + const pairKey = makePairKey(assetA, assetB) + const result = await pgPool.query( + `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, + volume::float, trade_count, open_price::float, close_price::float, + high_price::float, low_price::float + FROM price_aggregates + WHERE pair_key = $1 AND window = $2 + ORDER BY bucket DESC + LIMIT $3`, + [pairKey, window, Math.min(limit, 1000)] + ) + return result.rows.map(r => ({ + bucket: r.bucket.toISOString(), + window: r.window, + vwap: r.vwap, + sdexVwap: r.sdex_vwap, + ammVwap: r.amm_vwap, + volume: r.volume, + tradeCount: r.trade_count, + open: r.open_price, + close: r.close_price, + high: r.high_price, + low: r.low_price, + })) + } finally { + timer() + } }, listPairs() { @@ -253,39 +240,122 @@ const resolvers = { { 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) { +const KNOWN_OPERATIONS = new Set([ + 'getPrice', + 'getBestRoute', + 'getPriceHistory', + 'listPairs', + 'priceUpdated', + 'GetPrice', + 'GetBestRoute', + 'GetPriceHistory', + 'ListPairs', + 'PriceUpdated', + 'IntrospectionQuery', +]) + +export interface RegisterGraphQLOptions { + /** + * Whether unpersisted arbitrary documents are accepted. + * Defaults to false in production, true in development/test, or drives from + * GRAPHQL_ALLOW_ARBITRARY_QUERIES env variable. + */ + allowArbitraryQueries?: boolean + /** Additional persisted queries to register on this instance */ + persistedQueries?: Record +} + +export async function registerGraphQL(app: FastifyInstance, options?: RegisterGraphQLOptions) { + const allowArbitrary = options?.allowArbitraryQueries ?? ( + process.env.GRAPHQL_ALLOW_ARBITRARY_QUERIES !== undefined + ? (process.env.GRAPHQL_ALLOW_ARBITRARY_QUERIES === 'true' || process.env.GRAPHQL_ALLOW_ARBITRARY_QUERIES === '1') + : process.env.NODE_ENV !== 'production' + ) + + const queryStore: Record = { + ...DEFAULT_PERSISTED_QUERIES, + ...(options?.persistedQueries ?? {}), + } + + const persistedQueryProvider = { + isPersistedQuery: (req: any) => { + if (!allowArbitrary) return true + return Boolean( + req.persisted || + req.extensions?.persistedQuery || + (req.query && queryStore[req.query.trim()]) + ) + }, + isPersistedQueryRetry: (req: any) => { + return Boolean(req.query && req.extensions?.persistedQuery) + }, + getHash: (req: any) => { + if (req.extensions?.persistedQuery?.sha256Hash) { + return req.extensions.persistedQuery.sha256Hash + } + if (req.persisted && req.query) { + return req.query.trim() + } + if (req.query && queryStore[req.query.trim()]) { + return req.query.trim() + } + return false + }, + getQueryFromHash: async (hash: string) => { + return queryStore[hash] + }, + getHashForQuery: (query: string) => { + return hashQuery(query) + }, + saveQuery: async (hash: string, query: string) => { + queryStore[hash] = query.trim() + }, + notFoundError: 'Persisted query not found for given hash', + notSupportedError: 'Arbitrary GraphQL queries are disabled. Only registered persisted queries are allowed.', + } + await app.register(mercurius, { schema, resolvers, - graphiql: true, + graphiql: allowArbitrary, errorFormatter: formatGraphQLErrors, + persistedQueryProvider, 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: }`. + // Pre-execution hook: track GraphQL operations with strictly bounded label cardinality + app.graphql.addHook('preExecution', async (_schema, document) => { + const opDef = document.definitions.find(d => d.kind === 'OperationDefinition') as OperationDefinitionNode | undefined + const opType = opDef?.operation ?? 'unknown' + const rawName = opDef?.name?.value + let opName = 'anonymous' + + if (rawName) { + opName = KNOWN_OPERATIONS.has(rawName) ? rawName : 'other' + } else if (opDef?.selectionSet?.selections?.[0]?.kind === 'Field') { + const rootField = (opDef.selectionSet.selections[0] as FieldNode).name.value + opName = KNOWN_OPERATIONS.has(rootField) ? rootField : 'other' + } + + graphql_operations_total.inc({ + operation_name: opName, + operation_type: opType, + }) + }) + + // Bridge: forward every price the ingesters emit onto the GraphQL pubsub topic. const onPricePublished = (event: PricePublishedEvent) => { app.graphql.pubsub.publish({ topic: PRICE_TOPIC, @@ -294,8 +364,7 @@ export async function registerGraphQL(app: FastifyInstance) { } 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. + // Detach listener when server closes app.addHook('onClose', async () => { priceEmitter.off(PRICE_PUBLISHED, onPricePublished) }) diff --git a/src/api/persistedQueries.ts b/src/api/persistedQueries.ts new file mode 100644 index 00000000..a73002c2 --- /dev/null +++ b/src/api/persistedQueries.ts @@ -0,0 +1,129 @@ +import { createHash } from 'node:crypto' + +/** + * Standard query documents for canonical client operations. + */ +export const CANONICAL_QUERIES = { + GetPrice: `query GetPrice($assetA: String!, $assetB: String!) { + getPrice(assetA: $assetA, assetB: $assetB) { + assetA + assetB + pairKey + price + sdexPrice + ammPrice + bestRoute + volume24h + sdexVolume24h + ammVolume24h + vwap1m + vwap5m + vwap1h + vwap24h + priceChange24h + lastUpdated + sources + confidence + lastTradeAgeSeconds + } +}`, + + GetBestRoute: `query GetBestRoute($assetA: String!, $assetB: String!, $amount: Float!) { + getBestRoute(assetA: $assetA, assetB: $assetB, amount: $amount) { + route + sdexPrice + ammPrice + estimatedOutput + slippagePct + recommendation + } +}`, + + GetPriceHistory: `query GetPriceHistory($assetA: String!, $assetB: String!, $window: String!, $limit: Int) { + getPriceHistory(assetA: $assetA, assetB: $assetB, window: $window, limit: $limit) { + bucket + window + vwap + sdexVwap + ammVwap + volume + tradeCount + open + close + high + low + } +}`, + + ListPairs: `query ListPairs { + listPairs +}`, + + PriceUpdated: `subscription PriceUpdated($pair: String!, $network: String) { + priceUpdated(pair: $pair, network: $network) { + pair + price + ts + network + } +}`, +} as const + +/** + * Normalizes and calculates the SHA-256 hash of a GraphQL query document. + */ +export function hashQuery(query: string): string { + // Normalize whitespace to ensure consistent hashing across line endings / indentations + const normalized = query + .replace(/\r\n/g, '\n') + .trim() + return createHash('sha256').update(normalized, 'utf8').digest('hex') +} + +/** + * In-memory registry of persisted queries initialized with canonical operations. + */ +const _registry = new Map() + +// Populate registry with canonical queries +for (const [name, doc] of Object.entries(CANONICAL_QUERIES)) { + const hash = hashQuery(doc) + _registry.set(hash, doc) +} + +/** + * Returns all currently registered persisted queries as a hash -> document map. + */ +export function getDefaultPersistedQueries(): Record { + const result: Record = {} + for (const [hash, doc] of _registry.entries()) { + result[hash] = doc + } + return result +} + +export const DEFAULT_PERSISTED_QUERIES: Record = getDefaultPersistedQueries() + + +/** + * Registers a new query document in the persisted query store and returns its SHA-256 hash. + */ +export function registerPersistedQuery(query: string): string { + const hash = hashQuery(query) + _registry.set(hash, query.trim()) + return hash +} + +/** + * Looks up a query document by its SHA-256 hash. + */ +export function getPersistedQuery(hash: string): string | undefined { + return _registry.get(hash) +} + +/** + * Exports a manifest of registered queries for tooling and clients. + */ +export function getPersistedQueryManifest(): Array<{ hash: string; query: string }> { + return Array.from(_registry.entries()).map(([hash, query]) => ({ hash, query })) +} diff --git a/src/api/schema.ts b/src/api/schema.ts new file mode 100644 index 00000000..c7dcbc74 --- /dev/null +++ b/src/api/schema.ts @@ -0,0 +1,97 @@ +import { buildSchema, printSchema } from 'graphql' + +export const schema = ` + type AggregatedPrice { + assetA: String! + assetB: String! + pairKey: String! + price: Float! + sdexPrice: Float! + ammPrice: Float! + bestRoute: String! + volume24h: Float! + sdexVolume24h: Float! + ammVolume24h: Float! + vwap1m: Float! + vwap5m: Float! + vwap1h: Float! + vwap24h: Float! + priceChange24h: Float! + lastUpdated: String! + sources: Int! + confidence: String! + lastTradeAgeSeconds: Int + } + + type RouteInfo { + route: String! + sdexPrice: Float! + ammPrice: Float! + estimatedOutput: Float! + slippagePct: Float! + recommendation: String! + } + + type PriceBucket { + bucket: String! + window: String! + vwap: Float! + sdexVwap: Float + ammVwap: Float + volume: Float! + tradeCount: Int! + open: Float + close: Float + high: Float + low: Float + } + + type PriceUpdate { + pair: String! + price: Float! + ts: String! + """Which chain the price came from — testnet or mainnet.""" + network: String! + } + + type Query { + """ + Aggregated price for a pair. Three outcomes are kept distinct: + + - Malformed asset input (an asset that is not CODE or CODE:ISSUER) fails + with a GraphQL error whose extensions.code is INVALID_INPUT. + - A well-formed pair that is not tracked on the active network fails with + a GraphQL error whose extensions.code is PAIR_NOT_FOUND. + - A tracked pair with no trade data yet is not an error: it resolves to an + AggregatedPrice with zeroed figures rather than null or an error. + + The result is only null if the resolver itself cannot produce a payload; + clients should branch on extensions.code, not on null. + """ + 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! + } +` + +/** + * Returns canonical, formatted GraphQL SDL generated from the in-code schema. + */ +export function getCanonicalSchemaSDL(): string { + const parsed = buildSchema(schema) + return printSchema(parsed).trim() + '\n' +} diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..4bec3131 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -98,6 +98,35 @@ export const db_query_duration_seconds = new Histogram({ registers: [register] }) +// ── GraphQL Metrics (Bounded Cardinality) ──────────────────────────────────── + +export const graphql_operations_total = new Counter({ + name: 'graphql_operations_total', + help: 'Total number of GraphQL operations processed by operation name and type', + // labelNames: ['operation_name', 'operation_type'] + // Cardinality is strictly bounded: unknown/client-generated operation names + // are folded into 'other' or 'unknown' to avoid time-series explosion. + labelNames: ['operation_name', 'operation_type'], + registers: [register], +}) + +export const graphql_resolver_duration_seconds = new Histogram({ + name: 'graphql_resolver_duration_seconds', + help: 'Duration of database-backed GraphQL resolvers in seconds', + labelNames: ['field'], + buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5], + registers: [register], +}) + +export const graphql_errors_total = new Counter({ + name: 'graphql_errors_total', + help: 'Total number of GraphQL errors classified by bounded category', + // Bounded categories: 'validation' | 'not_found' | 'unauthorized' | 'persisted_query' | 'internal' | 'other' + labelNames: ['category'], + registers: [register], +}) + export async function getMetrics() { return await register.metrics() } +