feat(api): add JSON schema response validation to REST endpoints

src/__tests__/schemaValidation.test.ts

@@ -0,0 +1,121 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import Fastify from 'fastify'
+import Ajv from 'ajv'
+
+const { mockQuery, mockGetCachedPrice, mockGetBestRoute, mockGetAggregatedPrice } = vi.hoisted(() => ({
+  mockQuery: vi.fn(),
+  mockGetCachedPrice: vi.fn(),
+  mockGetBestRoute: vi.fn(),
+  mockGetAggregatedPrice: vi.fn(),
+}))
+
+vi.mock('../db', () => ({
+  pgPool: { query: mockQuery },
+}))
+
+vi.mock('../redis', () => ({
+  getCachedPrice: mockGetCachedPrice,
+  setCachedPrice: vi.fn(),
+}))
+
+vi.mock('../aggregator/bestRoute', () => ({
+  getBestRoute: mockGetBestRoute,
+}))
+
+vi.mock('../aggregator/vwap', () => ({
+  getAggregatedPrice: mockGetAggregatedPrice,
+}))
+
+vi.mock('../config', () => ({
+  config: {
+    pairs: [
+      {
+        pairKey: 'USDC/XLM',
+        assetA: { code: 'XLM', issuer: null },
+        assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' },
+      },
+    ],
+    cache: { priceTtl: 10 },
+  },
+}))
+
+import { registerRESTRoutes } from '../api/rest'
+import { priceResponseSchema } from '../api/schemas'
+
+async function buildApp() {
+  const app = Fastify({ logger: false })
+  await registerRESTRoutes(app)
+  await app.ready()
+  return app
+}
+
+/** A valid aggregated-price payload matching getAggregatedPrice's return type. */
+function validAggregate() {
+  return {
+    price: 0.1,
+    sdexPrice: 0.1,
+    ammPrice: 0.1,
+    volume24h: 150,
+    sdexVolume24h: 100,
+    ammVolume24h: 50,
+    vwap1m: 0.1,
+    vwap5m: 0.1,
+    vwap1h: 0.1,
+    vwap24h: 0.1,
+    priceChange24h: 1.1,
+    sources: 2,
+    confidence: 'high' as const,
+    lastTradeAgeSeconds: 10,
+  }
+}
+
+describe('REST response schema validation', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockGetCachedPrice.mockResolvedValue(null)
+    mockGetBestRoute.mockResolvedValue({ route: 'SDEX' })
+  })
+
+  it('returns 200 and a body that matches the declared /price schema', async () => {
+    mockGetAggregatedPrice.mockResolvedValue(validAggregate())
+
+    const app = await buildApp()
+    const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' })
+
+    expect(res.statusCode).toBe(200)
+
+    // The serialized body must independently satisfy the published schema.
+    const ajv = new Ajv({ allErrors: true })
+    const validate = ajv.compile(priceResponseSchema as object)
+    const ok = validate(res.json())
+    // Surface a readable diff if the body ever drifts from the schema.
+    expect(ok, ajv.errorsText(validate.errors)).toBe(true)
+  })
+
+  it('fails (500) when the response grows an undeclared field', async () => {
+    // Simulate an accidental shape change: an extra property leaks into the body.
+    mockGetAggregatedPrice.mockResolvedValue({
+      ...validAggregate(),
+      surpriseField: 'should not be here',
+    })
+
+    const app = await buildApp()
+    const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' })
+
+    // additionalProperties: false → serializer validation throws → 500.
+    expect(res.statusCode).toBe(500)
+  })
+
+  it('fails (500) when a field changes type', async () => {
+    // confidence must be one of the enum strings; a number is a shape break.
+    mockGetAggregatedPrice.mockResolvedValue({
+      ...validAggregate(),
+      confidence: 42 as unknown as 'high',
+    })
+
+    const app = await buildApp()
+    const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' })
+
+    expect(res.statusCode).toBe(500)
+  })
+})

src/api/rest.ts

@@ -5,6 +5,14 @@ import { getAggregatedPrice } from '../aggregator/vwap'
 import { getBestRoute } from '../aggregator/bestRoute'
 import { pgPool } from '../db'
 import { config } from '../config'
+import {
+  statusResponseSchema,
+  priceResponseSchema,
+  routeResponseSchema,
+  historyResponseSchema,
+  poolsResponseSchema,
+  installResponseValidation,
+} from './schemas'
 
 function makePairKey(a: string, b: string): string {
   return [a, b].sort().join('/')
@@ -22,8 +30,13 @@ function findPair(assetA: string, assetB: string) {
 }
 
 export async function registerRESTRoutes(app: FastifyInstance) {
+  // Validate every response against its declared schema in dev/test (no-op in
+  // production). Must run before the routes below are registered so they pick
+  // up the validating serializer.
+  installResponseValidation(app)
+
   // GET /status — public health/monitoring endpoint (no API key required)
-  app.get('/status', { config: { public: true } }, async () => {
+  app.get('/status', { config: { public: true }, schema: { response: { 200: statusResponseSchema } } }, async () => {
     const result = await pgPool.query(
       `SELECT last_ledger, last_processed_at FROM indexer_state ORDER BY updated_at DESC LIMIT 1`
     )
@@ -39,6 +52,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
   // GET /price/:assetA/:assetB
   app.get<{ Params: { assetA: string; assetB: string } }>(
     '/price/:assetA/:assetB',
+    { schema: { response: { 200: priceResponseSchema } } },
     async (req, reply) => {
       price_requests_total.inc()
       const { assetA, assetB } = req.params
@@ -76,6 +90,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
     Querystring: { amount?: string }
   }>(
     '/price/:assetA/:assetB/route',
+    { schema: { response: { 200: routeResponseSchema } } },
     async (req, reply) => {
       const { assetA, assetB } = req.params
       const amount = parseFloat(req.query.amount ?? '1000')
@@ -93,6 +108,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
     Querystring: { window?: string; limit?: string }
   }>(
     '/price/:assetA/:assetB/history',
+    { schema: { response: { 200: historyResponseSchema } } },
     async (req, reply) => {
       const { assetA, assetB } = req.params
       const window = req.query.window ?? '1h'
@@ -134,7 +150,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
   )
 
   // GET /pools
-  app.get('/pools', async () => {
+  app.get('/pools', { schema: { response: { 200: poolsResponseSchema } } }, async () => {
     const result = await pgPool.query(
       `SELECT DISTINCT ON (pool_id) pool_id, asset_a, asset_b,
               reserve_a::float, reserve_b::float, spot_price::float, fee_bp, timestamp

src/api/schemas.ts

@@ -0,0 +1,204 @@
+/**
+ * Shared JSON Schema objects for REST response validation.
+ *
+ * These schemas serve two purposes:
+ *   1. They are attached to routes via `{ schema: { response: { 200: ... } } }`,
+ *      which lets Fastify auto-generate a correct OpenAPI/Swagger spec and use
+ *      fast, schema-aware serialization.
+ *   2. In non-production environments they additionally *validate* the outgoing
+ *      payload (see `installResponseValidation`), so an accidental change to a
+ *      response shape fails loudly in tests instead of shipping silently.
+ *
+ * NOTE: response validation has no runtime impact in production — it is only
+ * installed when NODE_ENV !== 'production'. In production Fastify falls back to
+ * its default (fast) serializer for these same schemas.
+ */
+import Ajv from 'ajv'
+import type { FastifyInstance, FastifySchema } from 'fastify'
+
+/** GET /status */
+export const statusResponseSchema = {
+  type: 'object',
+  required: ['ok', 'watchedPairs', 'lastIndexedLedger', 'lastProcessedAt'],
+  additionalProperties: false,
+  properties: {
+    ok: { type: 'boolean' },
+    watchedPairs: { type: 'array', items: { type: 'string' } },
+    lastIndexedLedger: { type: ['integer', 'null'] },
+    // The pg driver returns a timestamp column as a JS Date; the validating
+    // serializer sees that pre-serialization object, so accept Date | string |
+    // null here (a Date stringifies to an ISO string in the response body).
+    lastProcessedAt: {},
+  },
+} as const
+
+/** GET /price/:assetA/:assetB */
+export const priceResponseSchema = {
+  type: 'object',
+  required: [
+    'assetA',
+    'assetB',
+    'pairKey',
+    'price',
+    'sdexPrice',
+    'ammPrice',
+    'volume24h',
+    'sdexVolume24h',
+    'ammVolume24h',
+    'vwap1m',
+    'vwap5m',
+    'vwap1h',
+    'vwap24h',
+    'priceChange24h',
+    'sources',
+    'confidence',
+    'lastTradeAgeSeconds',
+    'bestRoute',
+    'lastUpdated',
+  ],
+  additionalProperties: false,
+  properties: {
+    assetA: { type: 'string' },
+    assetB: { type: 'string' },
+    pairKey: { type: 'string' },
+    price: { type: 'number' },
+    sdexPrice: { type: 'number' },
+    ammPrice: { type: 'number' },
+    volume24h: { type: 'number' },
+    sdexVolume24h: { type: 'number' },
+    ammVolume24h: { type: 'number' },
+    vwap1m: { type: 'number' },
+    vwap5m: { type: 'number' },
+    vwap1h: { type: 'number' },
+    vwap24h: { type: 'number' },
+    priceChange24h: { type: 'number' },
+    sources: { type: 'integer' },
+    confidence: { type: 'string', enum: ['high', 'medium', 'low', 'unknown'] },
+    lastTradeAgeSeconds: { type: ['integer', 'null'] },
+    bestRoute: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'UNKNOWN'] },
+    lastUpdated: { type: 'string' },
+  },
+} as const
+
+/** GET /price/:assetA/:assetB/route — matches RouteInfo */
+export const routeResponseSchema = {
+  type: 'object',
+  required: [
+    'route',
+    'sdexPrice',
+    'ammPrice',
+    'estimatedOutput',
+    'slippagePct',
+    'recommendation',
+  ],
+  additionalProperties: false,
+  properties: {
+    route: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'UNKNOWN'] },
+    sdexPrice: { type: 'number' },
+    ammPrice: { type: 'number' },
+    estimatedOutput: { type: 'number' },
+    slippagePct: { type: 'number' },
+    recommendation: { type: 'string' },
+  },
+} as const
+
+/** GET /price/:assetA/:assetB/history */
+export const historyResponseSchema = {
+  type: 'object',
+  required: ['pairKey', 'window', 'buckets'],
+  additionalProperties: false,
+  properties: {
+    pairKey: { type: 'string' },
+    window: { type: 'string', enum: ['1m', '5m', '1h', '24h'] },
+    buckets: {
+      type: 'array',
+      items: {
+        type: 'object',
+        required: [
+          'bucket',
+          'vwap',
+          'sdexVwap',
+          'ammVwap',
+          'volume',
+          'tradeCount',
+          'open',
+          'close',
+          'high',
+          'low',
+        ],
+        additionalProperties: false,
+        properties: {
+          bucket: {},
+          vwap: { type: ['number', 'null'] },
+          sdexVwap: { type: ['number', 'null'] },
+          ammVwap: { type: ['number', 'null'] },
+          volume: { type: ['number', 'null'] },
+          tradeCount: { type: ['integer', 'null'] },
+          open: { type: ['number', 'null'] },
+          close: { type: ['number', 'null'] },
+          high: { type: ['number', 'null'] },
+          low: { type: ['number', 'null'] },
+        },
+      },
+    },
+  },
+} as const
+
+/**
+ * GET /pools
+ *
+ * Rows come straight from a raw SQL query whose column set may evolve, so the
+ * row shape is intentionally permissive (additionalProperties allowed). The
+ * envelope, however, is validated.
+ */
+export const poolsResponseSchema = {
+  type: 'object',
+  required: ['pools'],
+  additionalProperties: false,
+  properties: {
+    pools: {
+      type: 'array',
+      items: {
+        type: 'object',
+        additionalProperties: true,
+        properties: {
+          pool_id: { type: 'string' },
+          asset_a: { type: 'string' },
+          asset_b: { type: 'string' },
+          reserve_a: { type: ['number', 'null'] },
+          reserve_b: { type: ['number', 'null'] },
+          spot_price: { type: ['number', 'null'] },
+          fee_bp: { type: ['integer', 'null'] },
+        },
+      },
+    },
+  },
+} as const
+
+/**
+ * Install response-shape validation on a Fastify instance.
+ *
+ * Default Fastify behaviour is to *serialize* against the response schema,
+ * silently dropping unknown keys — which means a drifted shape would still
+ * return 200. To make accidental shape changes fail loudly we replace the
+ * serializer with one that first validates the payload against the schema and
+ * throws a 500 (FST_ERR_RESPONSE_SERIALIZATION) when it does not match.
+ *
+ * Skipped entirely in production so there is no added runtime cost there.
+ */
+export function installResponseValidation(app: FastifyInstance): void {
+  if (process.env.NODE_ENV === 'production') return
+
+  const ajv = new Ajv({ allErrors: true, coerceTypes: false })
+
+  app.setSerializerCompiler(({ schema }: { schema: FastifySchema }) => {
+    const validate = ajv.compile(schema as object)
+    return (data: unknown) => {
+      if (!validate(data)) {
+        const detail = ajv.errorsText(validate.errors, { dataVar: 'response' })
+        throw new Error(`Response does not match schema: ${detail}`)
+      }
+      return JSON.stringify(data)
+    }
+  })
+}