feat: add GraphQL pools query

src/__tests__/graphqlPools.test.ts

@@ -0,0 +1,106 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import Fastify, { type FastifyInstance } from 'fastify'
+
+const { mockQuery } = vi.hoisted(() => ({ mockQuery: vi.fn() }))
+
+vi.mock('../db', () => ({ pgPool: { query: mockQuery }, prisma: {} }))
+vi.mock('../redis', () => ({ getCachedPrice: vi.fn(), setCachedPrice: vi.fn() }))
+vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: vi.fn() }))
+vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: vi.fn() }))
+vi.mock('../pricing/depth', () => ({ getDepth: vi.fn() }))
+vi.mock('../config', () => {
+  const mainnetPair = {
+    pairKey: 'USDC/XLM',
+    assetA: { code: 'XLM', issuer: null },
+    assetB: { code: 'USDC', issuer: null },
+  }
+  return {
+    activeNetwork: 'testnet',
+    config: { pairs: [] },
+    getNetworkConfig: (network: string) => ({ pairs: network === 'mainnet' ? [mainnetPair] : [] }),
+  }
+})
+
+import { registerGraphQL } from '../api/graphql'
+import { registerRESTRoutes } from '../api/rest'
+import { registerNetworkSelector } from '../middleware/network'
+
+const POOL_ID = '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef'
+const SNAPSHOT_TIME = new Date('2026-10-17T08:00:00.000Z')
+const SNAPSHOT = {
+  pool_id: POOL_ID,
+  asset_a: 'XLM',
+  asset_b: 'USDC',
+  reserve_a: 1250.5,
+  reserve_b: 300.25,
+  spot_price: 0.2401,
+  fee_bp: 30,
+  timestamp: SNAPSHOT_TIME,
+}
+
+describe('GraphQL pools query', () => {
+  let app: FastifyInstance
+
+  beforeEach(async () => {
+    vi.clearAllMocks()
+    mockQuery.mockResolvedValue({ rows: [SNAPSHOT] })
+    app = Fastify({ logger: false })
+    await app.register(registerNetworkSelector)
+    await registerRESTRoutes(app)
+    await registerGraphQL(app)
+    await app.ready()
+  })
+
+  afterEach(async () => {
+    await app.close()
+  })
+
+  it('returns the same complete network-specific pools as REST', async () => {
+    const graphResponse = await app.inject({
+      method: 'POST',
+      url: '/graphql',
+      payload: {
+        query: `query {
+          pools(network: "mainnet") {
+            pool_id asset_a asset_b reserve_a reserve_b spot_price fee_bp timestamp
+          }
+        }`,
+      },
+    })
+    const restResponse = await app.inject({ method: 'GET', url: '/pools?network=mainnet' })
+
+    expect(graphResponse.statusCode).toBe(200)
+    expect(restResponse.statusCode).toBe(200)
+    expect(graphResponse.json().errors).toBeUndefined()
+    expect(graphResponse.json().data.pools).toEqual(restResponse.json().pools)
+    expect(graphResponse.json().data.pools[0].pool_id).toBe(POOL_ID)
+    expect(mockQuery.mock.calls.map(([, params]) => params)).toEqual([['mainnet'], ['mainnet']])
+  })
+
+  it('filters to a watched pair and rejects pairs not watched on that network', async () => {
+    const selected = await app.inject({
+      method: 'POST',
+      url: '/graphql',
+      payload: {
+        query: `query {
+          pools(pair: "XLM/USDC", network: "mainnet") { pool_id }
+        }`,
+      },
+    })
+    expect(selected.json().data.pools).toEqual([{ pool_id: POOL_ID }])
+    expect(mockQuery.mock.calls[0][1]).toEqual(['mainnet', 'XLM', 'USDC'])
+
+    const notWatched = await app.inject({
+      method: 'POST',
+      url: '/graphql',
+      payload: {
+        query: `query {
+          pools(pair: "XLM/USDC", network: "testnet") { pool_id }
+        }`,
+      },
+    })
+    expect(notWatched.json().data.pools).toBeNull()
+    expect(notWatched.json().errors[0].extensions.code).toBe('PAIR_NOT_FOUND')
+    expect(mockQuery).toHaveBeenCalledTimes(1)
+  })
+})

src/api/graphql.ts

@@ -8,9 +8,12 @@ import { getCachedPrice } from '../redis'
 import { getAggregatedPrice } from '../aggregator/vwap'
 import { getBestRoute } from '../aggregator/bestRoute'
 import { pgPool } from '../db'
-import { config, activeNetwork } from '../config'
+import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../config'
 import { parseAssetStr } from '../pairsRegistry'
 import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events'
+import { getPools } from './pools'
+import { resolveNetworkName } from '../middleware/network'
+import type { AssetId } from '../types'
 
 // Mercurius pubsub topic that carries every new price. A single app-level
 // listener bridges the ingesters' in-process `priceEmitter` onto this topic;
@@ -108,6 +111,17 @@ const schema = `
     network: String!
   }
 
+  type Pool {
+    pool_id: String!
+    asset_a: String!
+    asset_b: String!
+    reserve_a: Float
+    reserve_b: Float
+    spot_price: Float
+    fee_bp: Int
+    timestamp: String!
+  }
+
   type Query {
     """
     Aggregated price for a pair. Three outcomes are kept distinct:
@@ -126,6 +140,12 @@ const schema = `
     getBestRoute(assetA: String!, assetB: String!, amount: Float!): RouteInfo
     getPriceHistory(assetA: String!, assetB: String!, window: String!, limit: Int): [PriceBucket]
     listPairs: [String]!
+    """
+    Latest AMM pool snapshots, optionally narrowed to a watched pair.
+    This query is covered by the paid /graphql route in GATED_ROUTES; keep it
+    behind payment if GraphQL route pricing or gating is changed.
+    """
+    pools(pair: String, network: String): [Pool!]!
   }
 
   type Subscription {
@@ -159,6 +179,62 @@ function findPair(assetA: string, assetB: string) {
 
 const resolvers = {
   Query: {
+    async pools(
+      _: unknown,
+      { pair, network }: { pair?: string | null; network?: string | null },
+      context: MercuriusContext
+    ) {
+      const requestNetwork = (context.reply.request as typeof context.reply.request & { network?: string }).network
+      const resolved = resolveNetworkName(network ?? requestNetwork)
+      if (!resolved.ok) {
+        throw new ErrorWithProps(resolved.error, { code: 'INVALID_INPUT', field: 'network' })
+      }
+
+      const selectedNetwork: NetworkName = resolved.network
+      let pairFilter: { assetA: string; assetB: string } | undefined
+      if (pair != null) {
+        const assets = pair.split('/')
+        if (assets.length !== 2) {
+          throw new ErrorWithProps(
+            'Invalid pair: expected ASSET_A/ASSET_B, with each asset in CODE or CODE:ISSUER format',
+            { code: 'INVALID_INPUT', field: 'pair' },
+          )
+        }
+        const requestedA = parseAssetStr(assets[0])
+        const requestedB = parseAssetStr(assets[1])
+        if (!requestedA || !requestedB) {
+          throw new ErrorWithProps(
+            'Invalid pair: expected ASSET_A/ASSET_B, with each asset in CODE or CODE:ISSUER format',
+            { code: 'INVALID_INPUT', field: 'pair' },
+          )
+        }
+
+        const matches = getNetworkConfig(selectedNetwork).pairs.filter(configured => {
+          const matchesAsset = (
+            requested: AssetId,
+            actual: typeof configured.assetA,
+          ) => requested.code === actual.code.toUpperCase() &&
+            (requested.issuer == null || requested.issuer === actual.issuer)
+          return (
+            (matchesAsset(requestedA, configured.assetA) && matchesAsset(requestedB, configured.assetB)) ||
+            (matchesAsset(requestedA, configured.assetB) && matchesAsset(requestedB, configured.assetA))
+          )
+        })
+        if (matches.length !== 1) {
+          throw new ErrorWithProps(
+            `Pair ${pair} is not uniquely watched on ${selectedNetwork}`,
+            { code: 'PAIR_NOT_FOUND' },
+          )
+        }
+        pairFilter = {
+          assetA: matches[0].assetA.code,
+          assetB: matches[0].assetB.code,
+        }
+      }
+
+      return getPools(selectedNetwork, pairFilter)
+    },
+
     async getPrice(_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) {
       price_requests_total.inc()
       for (const [field, value] of [['assetA', assetA], ['assetB', assetB]] as const) {

src/api/pools.ts

@@ -0,0 +1,45 @@
+import { pgPool } from '../db'
+import type { NetworkName } from '../config'
+
+export interface PoolPairFilter {
+  assetA: string
+  assetB: string
+}
+
+/**
+ * Read the latest snapshot for each pool on one network. The pair filter is
+ * optional; when supplied it narrows results to the pair's stored asset codes.
+ */
+export async function getPools(network: NetworkName, pair?: PoolPairFilter) {
+  const pairFilter = pair
+    ? `AND ((s.asset_a = $2 AND s.asset_b = $3) OR (s.asset_a = $3 AND s.asset_b = $2))`
+    : ''
+  const params = pair ? [network, pair.assetA, pair.assetB] : [network]
+  const result = await pgPool.query(
+    `WITH RECURSIVE ids AS (
+       (SELECT pool_id FROM pool_snapshots WHERE network = $1 ORDER BY pool_id LIMIT 1)
+       UNION ALL
+       SELECT (SELECT s.pool_id FROM pool_snapshots s
+                WHERE s.network = $1 AND s.pool_id > ids.pool_id
+                ORDER BY s.pool_id LIMIT 1)
+         FROM ids
+        WHERE ids.pool_id IS NOT NULL
+     )
+     SELECT l.pool_id, l.asset_a, l.asset_b,
+            l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp
+       FROM ids
+      CROSS JOIN LATERAL (
+            SELECT * FROM pool_snapshots s
+             WHERE s.network = $1 AND s.pool_id = ids.pool_id ${pairFilter}
+             ORDER BY s.timestamp DESC
+             LIMIT 1
+           ) l
+      WHERE ids.pool_id IS NOT NULL`,
+    params
+  )
+
+  return result.rows.map(row => ({
+    ...row,
+    timestamp: row.timestamp instanceof Date ? row.timestamp.toISOString() : row.timestamp,
+  }))
+}

src/api/rest.ts

@@ -16,6 +16,7 @@ import {
   installResponseValidation,
 } from './schemas'
 import { getDepth } from '../pricing/depth'
+import { getPools } from './pools'
 
 function makePairKey(a: string, b: string): string {
   return [a, b].sort().join('/')
@@ -214,29 +215,7 @@ export async function registerRESTRoutes(app: FastifyInstance) {
   // network. A quiet pool is kept with its real `timestamp`, never dropped.
   app.get('/pools', { schema: { response: { 200: poolsResponseSchema } } }, async (req) => {
     const network = req.network ?? activeNetwork
-    const result = await pgPool.query(
-      `WITH RECURSIVE ids AS (
-         (SELECT pool_id FROM pool_snapshots WHERE network = $1 ORDER BY pool_id LIMIT 1)
-         UNION ALL
-         SELECT (SELECT s.pool_id FROM pool_snapshots s
-                  WHERE s.network = $1 AND s.pool_id > ids.pool_id
-                  ORDER BY s.pool_id LIMIT 1)
-           FROM ids
-          WHERE ids.pool_id IS NOT NULL
-       )
-       SELECT l.pool_id, l.asset_a, l.asset_b,
-              l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp
-         FROM ids
-        CROSS JOIN LATERAL (
-              SELECT * FROM pool_snapshots s
-               WHERE s.network = $1 AND s.pool_id = ids.pool_id
-               ORDER BY s.timestamp DESC
-               LIMIT 1
-             ) l
-        WHERE ids.pool_id IS NOT NULL`,
-      [network]
-    )
-    return { pools: result.rows }
+    return { pools: await getPools(network) }
   })
 
   // GET /price/:assetA/:assetB/depth?amount=1000