From a172a82cdbd65f14214583312e9650adaf36aa3d Mon Sep 17 00:00:00 2001 From: DevScoopee Date: Sat, 17 Oct 2026 09:26:49 -0400 Subject: [PATCH 1/3] Add a single enumerated error code set with a shared apiError helper, attach code to every { error } response, document the codes in OpenAPI, regenerate the spec, and add a guard test that all emitted codes stay in the set. src/errors.ts defines ERROR_CODES, ErrorCode, and apiError(code, message) preserving existing error text. Routes, API handlers, middleware (network/x402), and the rate limiter now include the appropriate code. openapi.yaml adds ErrorCode enum and Error schema (PaymentRequired/TooManyRequests include code), openapi.json is regenerated. src/__tests__/errorCodes.test.ts guards that the OpenAPI enum matches ERROR_CODES and only enumerated codes are emitted. Closes #261 --- .changeset/api-cursor-pagination.md | 22 +++ openapi.json | 168 +++++++++++++++++ openapi.yaml | 121 +++++++++++++ src/__tests__/benchmark.test.ts | 2 +- src/__tests__/errorCodes.test.ts | 67 +++++++ src/__tests__/pagination.test.ts | 268 ++++++++++++++++++++++++++++ src/__tests__/pairs.test.ts | 25 ++- src/__tests__/pools.test.ts | 46 ++++- src/api/admin.ts | 33 ++-- src/api/auth.ts | 27 +-- src/api/history.ts | 114 +++++++++--- src/api/rest.ts | 107 +++++++---- src/api/schemas.ts | 7 +- src/api/usage.ts | 37 ++-- src/app.ts | 3 +- src/errors.ts | 81 +++++++++ src/middleware/network.ts | 2 +- src/middleware/x402.ts | 13 +- src/pagination.ts | 116 ++++++++++++ src/routes/basket.ts | 11 +- src/routes/benchmark.ts | 5 +- src/routes/book.ts | 5 +- src/routes/candles.ts | 66 +++++-- src/routes/oracle.ts | 5 +- src/routes/pairs.ts | 64 +++++-- src/routes/price.ts | 9 +- src/routes/screener.ts | 53 +++--- src/routes/spreads.ts | 9 +- src/routes/volumes.ts | 7 +- src/routes/webhooks.ts | 13 +- 30 files changed, 1318 insertions(+), 188 deletions(-) create mode 100644 .changeset/api-cursor-pagination.md create mode 100644 src/__tests__/errorCodes.test.ts create mode 100644 src/__tests__/pagination.test.ts create mode 100644 src/errors.ts create mode 100644 src/pagination.ts diff --git a/.changeset/api-cursor-pagination.md b/.changeset/api-cursor-pagination.md new file mode 100644 index 00000000..5e07cfa5 --- /dev/null +++ b/.changeset/api-cursor-pagination.md @@ -0,0 +1,22 @@ +--- +'lens': minor +--- + +Add cursor-and-limit pagination to every list route — `GET /pools`, +`GET /pairs`, `GET /candles/:assetA/:assetB` and `GET /prices/history` — matching +the convention `GET /screener` already used. Each accepts `?cursor=` and +`?limit=` (absent limit → the route default, out-of-range values clamped, a +non-numeric limit is a 400) and now returns `nextCursor` + `hasMore` beside its +own key (`pools`, `pairs`, `candles`, `points`), so an unpaged caller still gets +a sensible first page. `GET /prices/history?format=csv` carries the same +envelope in the `X-Next-Cursor` / `X-Has-More` response headers, keeping the CSV +body the byte-stable `ts,price,volume` contract. + +Pagination is keyset-based, not offset-based: the cursor is an opaque base64 +`[sortValue, id]` and the predicate is `(sort, id) > (value, id)` (or `<` for +descending), so a row inserted while a client is paging can never shift a page +boundary and cause a duplicate or a skipped row. The shared helpers live in +`src/pagination.ts`; the per-route default/maximum page sizes are 50/200 +(`/pools`, `/pairs`), 500/5000 (`/candles`), 1000/10000 (`/prices/history`) and +20/100 (`/screener`), documented in `openapi.yaml` and exercised by a +mid-page-insert regression test. diff --git a/openapi.json b/openapi.json index cb6441b7..653f5087 100644 --- a/openapi.json +++ b/openapi.json @@ -292,6 +292,26 @@ "parameters": [ { "$ref": "#/components/parameters/NetworkQuery" + }, + { + "name": "cursor", + "in": "query", + "required": false, + "description": "Opaque cursor returned as `nextCursor` by the previous page.", + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + } } ], "responses": { @@ -334,6 +354,14 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true, + "description": "Cursor to pass as `?cursor=` for the next page, or null on the last page." + }, + "hasMore": { + "type": "boolean" } } } @@ -409,6 +437,26 @@ "type": "string", "format": "date-time" } + }, + { + "name": "cursor", + "in": "query", + "required": false, + "description": "Opaque cursor returned as `nextCursor` by the previous page.", + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 5000, + "default": 500 + } } ], "responses": { @@ -457,6 +505,14 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true, + "description": "Cursor to pass as `?cursor=` for the next page, or null on the last page." + }, + "hasMore": { + "type": "boolean" } } } @@ -543,6 +599,26 @@ }, { "$ref": "#/components/parameters/NetworkQuery" + }, + { + "name": "cursor", + "in": "query", + "required": false, + "description": "Opaque cursor returned as `nextCursor` by the previous page. When `format=csv`, the cursor is returned in the `X-Next-Cursor` response header instead.", + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 10000, + "default": 1000 + } } ], "responses": { @@ -600,6 +676,14 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true, + "description": "Cursor to pass as `?cursor=` for the next page, or null on the last page." + }, + "hasMore": { + "type": "boolean" } } } @@ -626,6 +710,26 @@ "parameters": [ { "$ref": "#/components/parameters/NetworkQuery" + }, + { + "name": "cursor", + "in": "query", + "required": false, + "description": "Opaque cursor returned as `nextCursor` by the previous page.", + "schema": { + "type": "string" + } + }, + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + } } ], "responses": { @@ -661,6 +765,14 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true, + "description": "Cursor to pass as `?cursor=` for the next page, or null on the last page." + }, + "hasMore": { + "type": "boolean" } } } @@ -1896,6 +2008,16 @@ } }, "responses": { + "Error": { + "description": "Standard error envelope", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, "PaymentRequired": { "description": "Payment Required", "content": { @@ -1929,6 +2051,9 @@ "error": { "type": "string" }, + "code": { + "$ref": "#/components/schemas/ErrorCode" + }, "description": { "type": "string" } @@ -1964,6 +2089,9 @@ "error": { "type": "string" }, + "code": { + "$ref": "#/components/schemas/ErrorCode" + }, "message": { "type": "string" }, @@ -1977,6 +2105,46 @@ } }, "schemas": { + "ErrorCode": { + "type": "string", + "description": "Stable machine-readable error code. Branch on this, never on `error`, which is human-readable.", + "enum": [ + "missing_parameter", + "invalid_parameter", + "bad_network", + "bad_interval", + "bad_window", + "unsupported", + "unknown_pair", + "unwatched_pair", + "pair_already_watched", + "unauthorized", + "not_found", + "upstream_unavailable", + "internal_error", + "rate_limited", + "payment_required", + "payment_invalid", + "payment_verification_failed", + "quota_exceeded" + ] + }, + "Error": { + "type": "object", + "required": [ + "error", + "code" + ], + "properties": { + "error": { + "type": "string", + "description": "Human-readable error message. Do not branch on it." + }, + "code": { + "$ref": "#/components/schemas/ErrorCode" + } + } + }, "Asset": { "type": "object", "properties": { diff --git a/openapi.yaml b/openapi.yaml index 9d511d0b..4788cca1 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -182,6 +182,20 @@ paths: - x402: [] parameters: - $ref: '#/components/parameters/NetworkQuery' + - name: cursor + in: query + required: false + description: Opaque cursor returned as `nextCursor` by the previous page. + schema: + type: string + - name: limit + in: query + required: false + schema: + type: integer + minimum: 1 + maximum: 200 + default: 50 responses: '200': description: List of pools @@ -212,6 +226,12 @@ paths: timestamp: type: string format: date-time + nextCursor: + type: string + nullable: true + description: Cursor to pass as `?cursor=` for the next page, or null on the last page. + hasMore: + type: boolean '402': $ref: '#/components/responses/PaymentRequired' '429': @@ -253,6 +273,20 @@ paths: schema: type: string format: date-time + - name: cursor + in: query + required: false + description: Opaque cursor returned as `nextCursor` by the previous page. + schema: + type: string + - name: limit + in: query + required: false + schema: + type: integer + minimum: 1 + maximum: 5000 + default: 500 responses: '200': description: Candle data @@ -287,6 +321,12 @@ paths: type: number volume: type: number + nextCursor: + type: string + nullable: true + description: Cursor to pass as `?cursor=` for the next page, or null on the last page. + hasMore: + type: boolean '400': description: Invalid parameters '402': @@ -342,6 +382,20 @@ paths: enum: [json, csv] default: json - $ref: '#/components/parameters/NetworkQuery' + - name: cursor + in: query + required: false + description: Opaque cursor returned as `nextCursor` by the previous page. When `format=csv`, the cursor is returned in the `X-Next-Cursor` response header instead. + schema: + type: string + - name: limit + in: query + required: false + schema: + type: integer + minimum: 1 + maximum: 10000 + default: 1000 responses: '200': description: Historical price snapshots @@ -381,6 +435,12 @@ paths: type: number volume: type: number + nextCursor: + type: string + nullable: true + description: Cursor to pass as `?cursor=` for the next page, or null on the last page. + hasMore: + type: boolean '400': description: Invalid parameters '402': @@ -395,6 +455,20 @@ paths: security: [] parameters: - $ref: '#/components/parameters/NetworkQuery' + - name: cursor + in: query + required: false + description: Opaque cursor returned as `nextCursor` by the previous page. + schema: + type: string + - name: limit + in: query + required: false + schema: + type: integer + minimum: 1 + maximum: 200 + default: 50 responses: '200': description: List of pairs @@ -421,6 +495,12 @@ paths: type: string format: date-time nullable: true + nextCursor: + type: string + nullable: true + description: Cursor to pass as `?cursor=` for the next page, or null on the last page. + hasMore: + type: boolean '429': $ref: '#/components/responses/TooManyRequests' post: @@ -1231,6 +1311,12 @@ components: default: testnet responses: + Error: + description: Standard error envelope + content: + application/json: + schema: + $ref: '#/components/schemas/Error' PaymentRequired: description: Payment Required content: @@ -1255,6 +1341,8 @@ components: type: string error: type: string + code: + $ref: '#/components/schemas/ErrorCode' description: type: string TooManyRequests: @@ -1277,12 +1365,45 @@ components: type: integer error: type: string + code: + $ref: '#/components/schemas/ErrorCode' message: type: string retryAfter: type: string schemas: + ErrorCode: + type: string + description: Stable machine-readable error code. Branch on this, never on `error`, which is human-readable. + enum: + - missing_parameter + - invalid_parameter + - bad_network + - bad_interval + - bad_window + - unsupported + - unknown_pair + - unwatched_pair + - pair_already_watched + - unauthorized + - not_found + - upstream_unavailable + - internal_error + - rate_limited + - payment_required + - payment_invalid + - payment_verification_failed + - quota_exceeded + Error: + type: object + required: [error, code] + properties: + error: + type: string + description: Human-readable error message. Do not branch on it. + code: + $ref: '#/components/schemas/ErrorCode' Asset: type: object properties: diff --git a/src/__tests__/benchmark.test.ts b/src/__tests__/benchmark.test.ts index 575099e8..defa1486 100644 --- a/src/__tests__/benchmark.test.ts +++ b/src/__tests__/benchmark.test.ts @@ -37,7 +37,7 @@ describe('GET /benchmark/:asset', () => { }) expect(res.statusCode).toBe(404) - expect(res.json()).toEqual({ error: 'Pair USDC/USD not watched' }) + expect(res.json()).toEqual({ error: 'Pair USDC/USD not watched', code: 'unwatched_pair' }) }) it('calculates benchmark deviation correctly when asset is assetB in the pair (e.g. USD/USDC)', async () => { diff --git a/src/__tests__/errorCodes.test.ts b/src/__tests__/errorCodes.test.ts new file mode 100644 index 00000000..4b4e5db4 --- /dev/null +++ b/src/__tests__/errorCodes.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it } from 'vitest' +import { readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' +import { readdirSync } from 'node:fs' +import YAML from 'js-yaml' +import { ERROR_CODES, ERROR_CODE_VALUES } from '../errors' + +function readSpec(): any { + const specPath = fileURLToPath(new URL('../../openapi.yaml', import.meta.url)) + const spec = YAML.load(readFileSync(specPath, 'utf8')) + return spec +} + +function collectCodesFromFile(filePath: string): string[] { + const content = readFileSync(filePath, 'utf8') + const codes: string[] = [] + const re1 = /apiError\(\s*['"]([a-z0-9_]+)['"]/g + let m: RegExpExecArray | null + while ((m = re1.exec(content))) { + codes.push(m[1]) + } + const re2 = /\bcode:\s*['"]([a-z][a-z0-9_]*)['"]/g + while ((m = re2.exec(content))) { + codes.push(m[1]) + } + return codes +} + +function getModuleFiles(): string[] { + const modules: string[] = [] + const routesDir = fileURLToPath(new URL('../routes', import.meta.url)) + for (const f of readdirSync(routesDir)) { + if (f.endsWith('.ts')) modules.push(fileURLToPath(new URL(`../routes/${f}`, import.meta.url))) + } + const apiFiles = ['rest.ts', 'history.ts', 'admin.ts', 'auth.ts', 'usage.ts'].map(f => + fileURLToPath(new URL(`../api/${f}`, import.meta.url)), + ) + modules.push(...apiFiles) + modules.push(fileURLToPath(new URL('../middleware/network.ts', import.meta.url))) + modules.push(fileURLToPath(new URL('../middleware/x402.ts', import.meta.url))) + modules.push(fileURLToPath(new URL('../app.ts', import.meta.url))) + return modules +} + +describe('error codes', () => { + it('documents the error code enum in openapi and keeps it in sync with ERROR_CODES', () => { + const spec = readSpec() + const enumVals = spec.components?.schemas?.ErrorCode?.enum as string[] | undefined + expect(enumVals).toBeDefined() + const expected = Object.keys(ERROR_CODES).sort() + const actual = [...(enumVals || [])].sort() + expect(actual).toEqual(expected) + }) + + it('no route/error-emitting module emits a code outside the enumerated set', () => { + const allowed = new Set(Object.keys(ERROR_CODES)) + const modules = getModuleFiles() + const all = modules.flatMap(collectCodesFromFile) + const bad = all.filter(c => !allowed.has(c)) + expect(bad, `Codes outside set: ${JSON.stringify(bad)}`).toEqual([]) + }) + + it('error code set is non-empty and matches values', () => { + expect(ERROR_CODE_VALUES.length).toBeGreaterThan(0) + expect(Object.keys(ERROR_CODES).length).toBe(ERROR_CODE_VALUES.length) + }) +}) diff --git a/src/__tests__/pagination.test.ts b/src/__tests__/pagination.test.ts new file mode 100644 index 00000000..34afd5f3 --- /dev/null +++ b/src/__tests__/pagination.test.ts @@ -0,0 +1,268 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +const { mockQuery } = vi.hoisted(() => ({ mockQuery: vi.fn() })) + +vi.mock('../db', () => ({ + prisma: {}, + pgPool: { query: mockQuery }, +})) + +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + SCREENER_DEFAULT_LIMIT, + SCREENER_MAX_LIMIT, + POOLS_DEFAULT_LIMIT, + POOLS_MAX_LIMIT, + PAIRS_DEFAULT_LIMIT, + PAIRS_MAX_LIMIT, + CANDLES_DEFAULT_LIMIT, + CANDLES_MAX_LIMIT, + HISTORY_DEFAULT_LIMIT, +} from '../pagination' +import { registerNetworkSelector } from '../middleware/network' +import { registerHistoryRoutes, MAX_HISTORY_POINTS } from '../api/history' +import { registerCandleRoutes } from '../routes/candles' + +describe('pagination helpers', () => { + describe('parseLimit', () => { + it('falls back to the route default when omitted', () => { + expect(parseLimit(undefined, 50, 200)).toEqual({ limit: 50 }) + }) + + it('clamps out-of-range values into [1, max]', () => { + expect(parseLimit('0', 50, 200)).toEqual({ limit: 1 }) + expect(parseLimit('-3', 50, 200)).toEqual({ limit: 1 }) + expect(parseLimit('9999', 50, 200)).toEqual({ limit: 200 }) + }) + + it('rejects a non-integer limit rather than silently clamping it', () => { + expect(parseLimit('lots', 50, 200)).toEqual({ error: 'limit must be a valid integer' }) + }) + + it('exposes each route default and maximum', () => { + expect([SCREENER_DEFAULT_LIMIT, SCREENER_MAX_LIMIT]).toEqual([20, 100]) + expect([POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT]).toEqual([50, 200]) + expect([PAIRS_DEFAULT_LIMIT, PAIRS_MAX_LIMIT]).toEqual([50, 200]) + expect([CANDLES_DEFAULT_LIMIT, CANDLES_MAX_LIMIT]).toEqual([500, 5000]) + expect(HISTORY_DEFAULT_LIMIT).toBe(1000) + expect(MAX_HISTORY_POINTS).toBe(10_000) + }) + }) + + describe('encodeCursor / decodeCursor', () => { + it('round-trips a [value, id] tuple', () => { + const cursor = encodeCursor(20, 'B/X') + expect(decodeCursor(cursor)).toEqual({ value: 20, id: 'B/X' }) + }) + + it('returns null for a cursor that is not base64 JSON', () => { + expect(decodeCursor('not-base64-json')).toBeNull() + }) + + it("returns 'skip' for a well-formed but degenerate cursor", () => { + expect(decodeCursor(encodeCursor(null, 'B/X'))).toBe('skip') + expect(decodeCursor(encodeCursor(20, ''))).toBe('skip') + }) + }) + + describe('slicePage', () => { + it('trims a limit+1 fetch and reports hasMore', () => { + expect(slicePage([1, 2, 3], 2)).toEqual({ data: [1, 2], hasMore: true }) + }) + + it('keeps the whole page when it did not overflow', () => { + expect(slicePage([1, 2], 2)).toEqual({ data: [1, 2], hasMore: false }) + }) + }) +}) + +describe('keyset paging survives a row inserted mid-page', () => { + // A miniature of a route's keyset query: strictly-after-the-cursor, descending + // by nothing, ascending by id — the same shape every list route uses. + const fetchAfter = (rows: number[], after: number | null) => + rows.filter(id => after === null || id > after) + + it('never duplicates or skips the original rows when one is inserted behind the cursor', () => { + const rows = [1, 2, 3, 4, 5] + const limit = 2 + const seen: number[] = [] + let cursor: string | null = null + let page = 0 + + while (true) { + const decoded = cursor ? decodeCursor(cursor) : null + const after = decoded && decoded !== 'skip' ? Number(decoded.value) : null + const { data, hasMore } = slicePage(fetchAfter(rows, after), limit) + seen.push(...data) + page++ + + // Insert a row between pages, behind where the cursor already points. + // An OFFSET-based page would shift and re-serve row 2 (or skip row 3). + if (page === 1) rows.push(1.5) + + if (hasMore && data.length > 0) { + const last = data[data.length - 1] + cursor = encodeCursor(last, last) + } else { + break + } + } + + expect(seen).toEqual([1, 2, 3, 4, 5]) + }) +}) + +describe('GET /prices/history pagination', () => { + const T0 = new Date('2025-01-01T00:00:00Z') + const minutes = (n: number) => new Date(T0.getTime() + n * 60_000) + + let buckets: Date[] + + beforeEach(() => { + mockQuery.mockReset() + buckets = [minutes(0), minutes(1), minutes(2), minutes(3), minutes(4)] + // Behaves like the aggregate query: honours the keyset predicate ($ afterTs) + // and the trailing LIMIT, exactly as the SQL does. + mockQuery.mockImplementation(async (sql: string, params: unknown[]) => { + const hasCursor = sql.includes('AND ts > $6') + const afterTs = hasCursor ? (params[5] as Date) : null + const limit = hasCursor ? (params[6] as number) : (params[5] as number) + const filtered = buckets.filter(b => !afterTs || b.getTime() > afterTs.getTime()) + return { + rows: filtered.slice(0, limit).map(bucket => ({ bucket, price: '1.25', volume: '10' })), + } + }) + }) + + async function buildApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await registerHistoryRoutes(app) + await app.ready() + return app + } + + it('pages through the whole result set with a row inserted mid-page', async () => { + const app = await buildApp() + + const page1 = ( + await app.inject({ method: 'GET', url: '/prices/history?pair=XLM/USDC&interval=1m&limit=2' }) + ).json() + expect(page1.points.map((p: { ts: string }) => p.ts)).toEqual([minutes(0), minutes(1)].map(d => d.toISOString())) + expect(page1.hasMore).toBe(true) + expect(page1.nextCursor).toBeTruthy() + + // A snapshot lands between the two rows page 1 already served — behind the + // cursor. Keyset paging must ignore it and neither repeat nor skip a row. + buckets.splice(1, 0, new Date(T0.getTime() + 30_000)) + + const page2 = ( + await app.inject({ + method: 'GET', + url: `/prices/history?pair=XLM/USDC&interval=1m&limit=2&cursor=${encodeURIComponent(page1.nextCursor)}`, + }) + ).json() + expect(page2.points.map((p: { ts: string }) => p.ts)).toEqual([minutes(2), minutes(3)].map(d => d.toISOString())) + expect(page2.hasMore).toBe(true) + + const page3 = ( + await app.inject({ + method: 'GET', + url: `/prices/history?pair=XLM/USDC&interval=1m&limit=2&cursor=${encodeURIComponent(page2.nextCursor)}`, + }) + ).json() + expect(page3.points.map((p: { ts: string }) => p.ts)).toEqual([minutes(4)].map(d => d.toISOString())) + expect(page3.hasMore).toBe(false) + expect(page3.nextCursor).toBeNull() + + const seen = [...page1.points, ...page2.points, ...page3.points].map((p: { ts: string }) => p.ts) + expect(seen).toEqual([0, 1, 2, 3, 4].map(n => minutes(n).toISOString())) + + await app.close() + }) + + it('gives an unpaged caller the first page with a default limit', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/prices/history?pair=XLM/USDC&interval=1m' }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.points).toHaveLength(buckets.length) + expect(body.hasMore).toBe(false) + // limit + 1 probes: the default is HISTORY_DEFAULT_LIMIT, so $6 is 1001. + expect(mockQuery.mock.calls[0][1][5]).toBe(HISTORY_DEFAULT_LIMIT + 1) + await app.close() + }) +}) + +describe('GET /candles pagination', () => { + async function buildApp() { + const app = Fastify({ logger: false }) + await registerCandleRoutes(app) + await app.ready() + return app + } + + function candleRow(hour: number) { + return { + time: new Date(`2025-01-01T0${hour}:00:00Z`), + open: '1', + high: '2', + low: '0.5', + close: '1.5', + volume: '10', + } + } + + beforeEach(() => { + mockQuery.mockReset() + }) + + it('returns nextCursor/hasMore and a keyset condition on the next page', async () => { + mockQuery.mockResolvedValue({ rows: [candleRow(0), candleRow(1), candleRow(2)] }) + const app = await buildApp() + + const res = await app.inject({ method: 'GET', url: '/candles/XLM/USDC?interval=1h&limit=2' }) + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.candles).toHaveLength(2) + expect(body.hasMore).toBe(true) + + const expectedEpoch = Math.floor(new Date('2025-01-01T01:00:00Z').getTime() / 1000) + expect(JSON.parse(Buffer.from(body.nextCursor, 'base64').toString())).toEqual([expectedEpoch, 'USDC/XLM']) + expect(mockQuery.mock.calls[0][1]).toEqual([ + 3600, + 'USDC/XLM', + expect.any(Date), + expect.any(Date), + 3, + ]) + + mockQuery.mockClear() + mockQuery.mockResolvedValue({ rows: [candleRow(2)] }) + await app.inject({ + method: 'GET', + url: `/candles/XLM/USDC?interval=1h&limit=2&cursor=${encodeURIComponent(body.nextCursor)}`, + }) + const [sql, params] = mockQuery.mock.calls[0] + expect(sql).toMatch(/floor\(EXTRACT\(EPOCH FROM timestamp\) \/ \$1\) \* \$1 > \$5/) + expect(params[4]).toBe(expectedEpoch) + expect(params[5]).toBe(3) + + await app.close() + }) + + it('rejects a malformed cursor and a non-numeric limit', async () => { + const app = await buildApp() + const badCursor = await app.inject({ method: 'GET', url: '/candles/XLM/USDC?interval=1h&cursor=nope' }) + const badLimit = await app.inject({ method: 'GET', url: '/candles/XLM/USDC?interval=1h&limit=lots' }) + + expect(badCursor.statusCode).toBe(400) + expect(badLimit.statusCode).toBe(400) + await app.close() + }) +}) diff --git a/src/__tests__/pairs.test.ts b/src/__tests__/pairs.test.ts index 4ec2acbc..09bbe0ec 100644 --- a/src/__tests__/pairs.test.ts +++ b/src/__tests__/pairs.test.ts @@ -221,9 +221,30 @@ describe('GET /pairs', () => { const app = await buildApp() const body = (await app.inject({ method: 'GET', url: '/pairs' })).json() + // Sorted ascending by pairKey (the keyset order), so NEW/Y precedes QUIET/X. expect(body.pairs).toHaveLength(2) - expect(body.pairs[0]).toMatchObject({ pairKey: 'QUIET/X', latestPrice: 1.5, lastUpdated: old.toISOString() }) - expect(body.pairs[1]).toMatchObject({ pairKey: 'NEW/Y', latestPrice: null, lastUpdated: null }) + expect(body.pairs[0]).toMatchObject({ pairKey: 'NEW/Y', latestPrice: null, lastUpdated: null }) + expect(body.pairs[1]).toMatchObject({ pairKey: 'QUIET/X', latestPrice: 1.5, lastUpdated: old.toISOString() }) + }) + + it('paginates keyset-style on pairKey, fetching one extra row', async () => { + mockGetActivePairs.mockReturnValue([pair('A/B'), pair('C/D'), pair('E/F')]) + const app = await buildApp() + const body = (await app.inject({ method: 'GET', url: '/pairs?limit=2' })).json() + + expect(body.pairs.map((p: { pairKey: string }) => p.pairKey)).toEqual(['A/B', 'C/D']) + expect(body.hasMore).toBe(true) + expect(JSON.parse(Buffer.from(body.nextCursor, 'base64').toString())).toEqual(['C/D', 'C/D']) + }) + + it('resumes after the cursor without duplicating the boundary row', async () => { + mockGetActivePairs.mockReturnValue([pair('A/B'), pair('C/D'), pair('E/F')]) + const cursor = Buffer.from(JSON.stringify(['C/D', 'C/D'])).toString('base64') + const app = await buildApp() + const body = (await app.inject({ method: 'GET', url: `/pairs?cursor=${cursor}` })).json() + + expect(body.pairs.map((p: { pairKey: string }) => p.pairKey)).toEqual(['E/F']) + expect(body.hasMore).toBe(false) }) }) }) diff --git a/src/__tests__/pools.test.ts b/src/__tests__/pools.test.ts index 1531988f..9bdcafbe 100644 --- a/src/__tests__/pools.test.ts +++ b/src/__tests__/pools.test.ts @@ -46,14 +46,14 @@ describe('GET /pools', () => { expect(sql).not.toMatch(/DISTINCT ON/i) expect(sql).toMatch(/network = \$1/) expect(sql).toMatch(/LIMIT 1/i) - expect(params).toEqual([activeNetwork]) + expect(params).toEqual([activeNetwork, 51]) await app.close() }) it('uses the request network when one is resolved', async () => { const app = await buildApp('mainnet') await app.inject({ method: 'GET', url: '/pools' }) - expect(mockQuery.mock.calls[0][1]).toEqual(['mainnet']) + expect(mockQuery.mock.calls[0][1]).toEqual(['mainnet', 51]) await app.close() }) @@ -72,4 +72,46 @@ describe('GET /pools', () => { ]) await app.close() }) + + it('paginates keyset-style on pool_id, fetching one extra row', async () => { + const old = new Date('2020-01-01T00:00:00Z') + mockQuery.mockResolvedValue({ + rows: [ + { pool_id: 'C1', asset_a: 'XLM', asset_b: 'USDC', timestamp: old }, + { pool_id: 'C2', asset_a: 'XLM', asset_b: 'USDC', timestamp: old }, + { pool_id: 'C3', asset_a: 'XLM', asset_b: 'USDC', timestamp: old }, + ], + }) + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/pools?limit=2' }) + + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.pools.map((p: { pool_id: string }) => p.pool_id)).toEqual(['C1', 'C2']) + expect(body.hasMore).toBe(true) + expect(JSON.parse(Buffer.from(body.nextCursor, 'base64').toString())).toEqual(['C2', 'C2']) + expect(mockQuery.mock.calls[0][1]).toEqual([activeNetwork, 3]) + await app.close() + }) + + it('applies the cursor as a pool_id keyset predicate', async () => { + const cursor = Buffer.from(JSON.stringify(['C2', 'C2'])).toString('base64') + const app = await buildApp() + await app.inject({ method: 'GET', url: `/pools?cursor=${cursor}` }) + + const [sql, params] = mockQuery.mock.calls[0] + expect(sql).toMatch(/pool_id > \$2/) + expect(params).toEqual([activeNetwork, 'C2', 51]) + await app.close() + }) + + it('rejects a malformed cursor and a non-numeric limit', async () => { + const app = await buildApp() + const badCursor = await app.inject({ method: 'GET', url: '/pools?cursor=not-base64-json' }) + const badLimit = await app.inject({ method: 'GET', url: '/pools?limit=lots' }) + + expect(badCursor.statusCode).toBe(400) + expect(badLimit.statusCode).toBe(400) + await app.close() + }) }) diff --git a/src/api/admin.ts b/src/api/admin.ts index b493081e..8b40e744 100644 --- a/src/api/admin.ts +++ b/src/api/admin.ts @@ -2,6 +2,7 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' import { randomBytes, timingSafeEqual } from 'crypto' import { prisma } from '../db' import { hashApiKey } from './auth' +import { apiError } from '../errors' /** Generates a new opaque API key with a recognizable `lens_` prefix. */ export function generateApiKey(): string { @@ -49,16 +50,16 @@ export async function registerAdminRoutes(app: FastifyInstance) { '/admin/keys', { config: { public: true } }, async (req: FastifyRequest<{ Body: CreateKeyBody }>, reply: FastifyReply) => { - if (!isAdminAuthorized(req)) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Provide a valid X-Admin-Token header.', - }) - } + if (!isAdminAuthorized(req)) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Provide a valid X-Admin-Token header.', + }) + } const { label, ratePerMin, ratePerDay, monthlyQuotaCents, dailyQuotaCents, overagePolicy } = req.body ?? {} if (!label || typeof label !== 'string' || label.trim().length === 0) { - return reply.status(400).send({ error: 'label is required' }) + return reply.status(400).send(apiError('missing_parameter', 'label is required')) } // Generate the key, store only its hash, and return the plaintext once. @@ -93,18 +94,18 @@ export async function registerAdminRoutes(app: FastifyInstance) { '/admin/keys/:id', { config: { public: true } }, async (req: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { - if (!isAdminAuthorized(req)) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Provide a valid X-Admin-Token header.', - }) - } + if (!isAdminAuthorized(req)) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Provide a valid X-Admin-Token header.', + }) + } const { id } = req.params const existing = await prisma.apiKey.findUnique({ where: { id } }) - if (!existing) { - return reply.status(404).send({ error: 'API key not found' }) - } + if (!existing) { + return reply.status(404).send(apiError('not_found', 'API key not found')) + } if (existing.revokedAt) { return reply.status(200).send({ id, revoked: true, alreadyRevoked: true }) } diff --git a/src/api/auth.ts b/src/api/auth.ts index 44cf0d12..75121ba5 100644 --- a/src/api/auth.ts +++ b/src/api/auth.ts @@ -2,6 +2,7 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' import { createHash } from 'crypto' import fp from 'fastify-plugin' import { prisma } from '../db' +import { apiError } from '../errors' /** * Metadata for an authenticated API key, attached to each request as @@ -75,20 +76,20 @@ async function apiKeyAuthPlugin(app: FastifyInstance) { if (routeConfig.public) return const token = extractBearer(req.headers['authorization']) - if (!token) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Missing API key. Provide an Authorization: Bearer header.', - }) - } + if (!token) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Missing API key. Provide an Authorization: Bearer header.', + }) + } - const keyContext = await lookupApiKey(token) - if (!keyContext) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Invalid or revoked API key.', - }) - } + const keyContext = await lookupApiKey(token) + if (!keyContext) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Invalid or revoked API key.', + }) + } req.apiKey = keyContext }) diff --git a/src/api/history.ts b/src/api/history.ts index 1f8a7590..f4b25ee0 100644 --- a/src/api/history.ts +++ b/src/api/history.ts @@ -2,6 +2,14 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import '../middleware/network' // declares req.network on the FastifyRequest type +import { apiError } from '../errors' +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + HISTORY_DEFAULT_LIMIT, +} from '../pagination' /** Supported aggregation intervals → bucket width in seconds. */ export const HISTORY_INTERVAL_SECONDS: Record = { @@ -71,14 +79,32 @@ export function historyToCsv(points: HistoryPoint[]): string { * bucket) and the summed volume. Buckets are aligned to the Unix epoch so that * e.g. a 1h bucket always starts on the hour, independent of the query window. * With intervalSecs=60 this returns the raw snapshots unchanged. + * + * `opts.afterTs` applies a keyset predicate — only buckets strictly after that + * timestamp are returned (used by the cursor). `opts.limit` caps the number of + * buckets after grouping and ordering; callers doing limit+1 probing pass + * `limit + 1` and trim the extra row themselves. */ export async function queryHistory( pair: string, from: Date, to: Date, intervalSecs: number, - network: NetworkName + network: NetworkName, + opts?: { afterTs?: Date; limit?: number } ): Promise { + const params: unknown[] = [intervalSecs, pair, from, to, network] + let cursorCondition = '' + if (opts?.afterTs) { + params.push(opts.afterTs) + cursorCondition = ` AND ts > $${params.length}` + } + let limitClause = '' + if (opts?.limit !== undefined) { + params.push(opts.limit) + limitClause = ` LIMIT $${params.length}` + } + const result = await pgPool.query( `SELECT to_timestamp(floor(EXTRACT(EPOCH FROM ts) / $1) * $1) AS bucket, @@ -87,10 +113,10 @@ export async function queryHistory( FROM price_snapshots WHERE pair = $2 AND network = $5 AND ts >= $3 - AND ts <= $4 + AND ts <= $4${cursorCondition} GROUP BY floor(EXTRACT(EPOCH FROM ts) / $1) - ORDER BY bucket ASC`, - [intervalSecs, pair, from, to, network] + ORDER BY bucket ASC${limitClause}`, + params ) return result.rows.map(r => ({ @@ -117,56 +143,102 @@ export async function queryHistory( * side effect. */ export async function registerHistoryRoutes(app: FastifyInstance) { + // GET /prices/history?pair=…&from=…&to=…&interval=1m|5m|1h&cursor=&limit= + // + // Paginated keyset-style on the bucket start time ASC (default limit 1000, + // max MAX_HISTORY_POINTS). The cursor carries the last page's bucket ts as + // an ISO string (the tiebreaker id is the pair, kept so every route emits + // the same cursor shape), and the keyset predicate is applied in the WHERE + // clause before grouping, so a snapshot inserted mid-paging cannot shift a + // page boundary the way an offset would. app.get<{ - Querystring: { pair?: string; from?: string; to?: string; interval?: string; format?: string } + Querystring: { + pair?: string + from?: string + to?: string + interval?: string + format?: string + cursor?: string + limit?: string + } }>('/prices/history', async (req, reply) => { const { pair } = req.query if (!pair) { - return reply.status(400).send({ error: 'pair query parameter is required' }) + return reply.status(400).send(apiError('missing_parameter', 'pair query parameter is required')) } // Validated before the DB round-trip so an unknown format never queries. const format = req.query.format ?? 'json' if (format !== 'json' && format !== 'csv') { - return reply.status(400).send({ error: 'format must be one of: json, csv' }) + return reply.status(400).send(apiError('invalid_parameter', 'format must be one of: json, csv')) } const interval = req.query.interval ?? '1m' const intervalSecs = HISTORY_INTERVAL_SECONDS[interval] if (!intervalSecs) { - return reply.status(400).send({ - error: `interval must be one of: ${Object.keys(HISTORY_INTERVAL_SECONDS).join(', ')}`, - }) + return reply.status(400).send( + apiError('bad_interval', `interval must be one of: ${Object.keys(HISTORY_INTERVAL_SECONDS).join(', ')}`), + ) + } + + const parsedLimit = parseLimit(req.query.limit, HISTORY_DEFAULT_LIMIT, MAX_HISTORY_POINTS) + if ('error' in parsedLimit) { + return reply.status(400).send(apiError('invalid_parameter', parsedLimit.error)) + } + const { limit } = parsedLimit + + let afterTs: Date | undefined + if (req.query.cursor) { + const decoded = decodeCursor(req.query.cursor) + if (decoded === null) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + if (decoded !== 'skip') { + afterTs = new Date(String(decoded.value)) + if (isNaN(afterTs.getTime())) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + } } const from = req.query.from ? new Date(req.query.from) : new Date(Date.now() - 24 * 60 * 60 * 1000) const to = req.query.to ? new Date(req.query.to) : new Date() if (isNaN(from.getTime())) { - return reply.status(400).send({ error: 'from must be a valid ISO 8601 date' }) + return reply.status(400).send(apiError('invalid_parameter', 'from must be a valid ISO 8601 date')) } if (isNaN(to.getTime())) { - return reply.status(400).send({ error: 'to must be a valid ISO 8601 date' }) + return reply.status(400).send(apiError('invalid_parameter', 'to must be a valid ISO 8601 date')) } if (from.getTime() > to.getTime()) { - return reply.status(400).send({ error: 'from must be before to' }) + return reply.status(400).send(apiError('invalid_parameter', 'from must be before to')) } // Guard against unbounded scans: cap the number of buckets the window can produce. const spanSecs = (to.getTime() - from.getTime()) / 1000 if (spanSecs / intervalSecs > MAX_HISTORY_POINTS) { - return reply.status(400).send({ - error: `requested window is too large for interval=${interval} (max ${MAX_HISTORY_POINTS} points); narrow the range or use a coarser interval`, - }) + return reply.status(400).send( + apiError('invalid_parameter', `requested window is too large for interval=${interval} (max ${MAX_HISTORY_POINTS} points); narrow the range or use a coarser interval`), + ) } const network = req.network ?? activeNetwork - const points = await queryHistory(pair, from, to, intervalSecs, network) + const points = await queryHistory(pair, from, to, intervalSecs, network, { + afterTs, + limit: limit + 1, // Fetch one extra bucket to determine hasMore + }) + const { data, hasMore } = slicePage(points, limit) + const last = hasMore && data.length > 0 ? data[data.length - 1] : null + const nextCursor = last ? encodeCursor(last.ts, pair) : null if (format === 'csv') { // Deliberately no response schema here: Fastify's JSON serializer must // not touch this body. charset=utf8 keeps the bytes the golden test pins. - return reply.type('text/csv; charset=utf-8').send(historyToCsv(points)) + // The pagination envelope travels in headers so the CSV body stays the + // pure `ts,price,volume` contract downstream parsers key on. + reply.header('X-Has-More', String(hasMore)) + if (nextCursor) reply.header('X-Next-Cursor', nextCursor) + return reply.type('text/csv; charset=utf-8').send(historyToCsv(data)) } return { @@ -175,8 +247,10 @@ export async function registerHistoryRoutes(app: FastifyInstance) { interval, from: from.toISOString(), to: to.toISOString(), - count: points.length, - points, + count: data.length, + points: data, + nextCursor, + hasMore, } }) } diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..fc1ce0cc 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -16,6 +16,15 @@ import { installResponseValidation, } from './schemas' import { getDepth } from '../pricing/depth' +import { apiError } from '../errors' +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + POOLS_DEFAULT_LIMIT, + POOLS_MAX_LIMIT, +} from '../pagination' function makePairKey(a: string, b: string): string { return [a, b].sort().join('/') @@ -105,7 +114,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { const { assetA, assetB } = req.params const network = req.network ?? activeNetwork const pair = findPair(assetA, assetB, network) - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) + if (!pair) return reply.status(404).send(apiError('unwatched_pair', `Pair ${assetA}/${assetB} not watched on ${network}`)) // Cache key is network-scoped so testnet/mainnet prices for the same // asset codes never collide. @@ -148,8 +157,8 @@ export async function registerRESTRoutes(app: FastifyInstance) { const amount = parseFloat(req.query.amount ?? '1000') const network = req.network ?? activeNetwork const pair = findPair(assetA, assetB, network) - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) - if (isNaN(amount) || amount <= 0) return reply.status(400).send({ error: 'amount must be a positive number' }) + if (!pair) return reply.status(404).send(apiError('unwatched_pair', `Pair ${assetA}/${assetB} not watched on ${network}`)) + if (isNaN(amount) || amount <= 0) return reply.status(400).send(apiError('invalid_parameter', 'amount must be a positive number')) return getBestRoute(pair.assetA, pair.assetB, pair.pairKey, amount, network) } @@ -170,7 +179,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { const network = req.network ?? activeNetwork if (!['1m', '5m', '1h', '24h'].includes(window)) { - return reply.status(400).send({ error: 'window must be one of: 1m, 5m, 1h, 24h' }) + return reply.status(400).send(apiError('bad_window', 'window must be one of: 1m, 5m, 1h, 24h')) } const result = await pgPool.query( @@ -212,32 +221,70 @@ export async function registerRESTRoutes(app: FastifyInstance) { // each step), then one LIMIT 1 index probe per pool fetches its newest row. // Cost is O(pools x log n); the old DISTINCT ON sorted every row of every // 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) + // + // Paginated keyset-style on pool_id ASC (default limit 50, max 200): the + // cursor carries the last page's pool_id, and the skip scan's seed query + // resumes strictly after it, so a pool inserted mid-paging cannot shift a + // page boundary the way an OFFSET would. + app.get<{ Querystring: { cursor?: string; limit?: string } }>( + '/pools', + { schema: { response: { 200: poolsResponseSchema } } }, + async (req, reply) => { + const network = req.network ?? activeNetwork + + const parsedLimit = parseLimit(req.query.limit, POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send(apiError('invalid_parameter', parsedLimit.error)) + } + const { limit } = parsedLimit + + const params: unknown[] = [network] + let seedFilter = '' + if (req.query.cursor) { + const decoded = decodeCursor(req.query.cursor) + if (decoded === null) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + if (decoded !== 'skip') { + params.push(decoded.value) + seedFilter = ` AND pool_id > $${params.length}` + } + } + params.push(limit + 1) // Fetch one extra to determine hasMore + + const result = await pgPool.query( + `WITH RECURSIVE ids AS ( + (SELECT pool_id FROM pool_snapshots WHERE network = $1${seedFilter} 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 - ) - 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 } - }) + ORDER BY l.pool_id ASC + LIMIT $${params.length}`, + params + ) + + const { data, hasMore } = slicePage(result.rows, limit) + const nextCursor = + hasMore && data.length > 0 + ? encodeCursor(data[data.length - 1].pool_id, data[data.length - 1].pool_id) + : null + return { pools: data, nextCursor, hasMore } + } + ) // GET /price/:assetA/:assetB/depth?amount=1000 app.get<{ @@ -252,8 +299,8 @@ export async function registerRESTRoutes(app: FastifyInstance) { const network = req.network ?? activeNetwork const pair = findPair(assetA, assetB, network) - if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) - if (isNaN(amount) || amount <= 0) return reply.status(400).send({ error: 'amount must be a positive number' }) + if (!pair) return reply.status(404).send(apiError('unwatched_pair', `Pair ${assetA}/${assetB} not watched on ${network}`)) + if (isNaN(amount) || amount <= 0) return reply.status(400).send(apiError('invalid_parameter', 'amount must be a positive number')) // NOTE: getDepth reads order-book data with no network column yet — see L048. const depthResult = await getDepth(pair.pairKey, amount) diff --git a/src/api/schemas.ts b/src/api/schemas.ts index c7db8496..94d591b1 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -158,11 +158,12 @@ export const historyResponseSchema = { * * 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. + * envelope, however, is validated — including the cursor-pagination fields + * every list route carries alongside its own key. */ export const poolsResponseSchema = { type: 'object', - required: ['pools'], + required: ['pools', 'nextCursor', 'hasMore'], additionalProperties: false, properties: { pools: { @@ -181,6 +182,8 @@ export const poolsResponseSchema = { }, }, }, + nextCursor: { type: ['string', 'null'] }, + hasMore: { type: 'boolean' }, }, } as const diff --git a/src/api/usage.ts b/src/api/usage.ts index f6cb9413..978403d1 100644 --- a/src/api/usage.ts +++ b/src/api/usage.ts @@ -1,6 +1,7 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' import fp from 'fastify-plugin' import { getUsageSummary, getAllUsageSummaries } from '../x402/metering' +import { apiError } from '../errors' function isAdminAuthorized(req: FastifyRequest): boolean { const adminToken = process.env.ADMIN_TOKEN @@ -20,12 +21,12 @@ export async function registerUsageRoutes(app: FastifyInstance) { '/admin/usage/:keyId', { config: { public: true } }, async (req, reply) => { - if (!isAdminAuthorized(req)) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Provide a valid X-Admin-Token header.', - }) - } + if (!isAdminAuthorized(req)) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Provide a valid X-Admin-Token header.', + }) + } const { keyId } = req.params const summary = await getUsageSummary(keyId) return reply.send(summary) @@ -36,12 +37,12 @@ export async function registerUsageRoutes(app: FastifyInstance) { '/admin/usage', { config: { public: true } }, async (req, reply) => { - if (!isAdminAuthorized(req)) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Provide a valid X-Admin-Token header.', - }) - } + if (!isAdminAuthorized(req)) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Provide a valid X-Admin-Token header.', + }) + } const summaries = await getAllUsageSummaries() return reply.send({ keys: summaries }) } @@ -51,12 +52,12 @@ export async function registerUsageRoutes(app: FastifyInstance) { '/usage/me', async (req, reply) => { const apiKey = (req as any).apiKey as { id: string } | undefined - if (!apiKey) { - return reply.status(401).send({ - error: 'Unauthorized', - message: 'Missing API key. Provide an Authorization: Bearer header.', - }) - } + if (!apiKey) { + return reply.status(401).send({ + ...apiError('unauthorized', 'Unauthorized'), + message: 'Missing API key. Provide an Authorization: Bearer header.', + }) + } const summary = await getUsageSummary(apiKey.id) return reply.send(summary) } diff --git a/src/app.ts b/src/app.ts index c0d2a117..506d1b9d 100644 --- a/src/app.ts +++ b/src/app.ts @@ -27,6 +27,7 @@ import { registerBasketRoutes } from './routes/basket' import { registerDiscoveryRoutes } from './routes/discovery' import { registerSettleRoute } from './routes/facilitator' import { getMetrics } from './metrics' +import { apiError } from './errors' export interface BuildAppOptions { /** @@ -98,7 +99,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise req.url === '/status', errorResponseBuilder: (req, context) => ({ statusCode: 429, - error: 'Too Many Requests', + ...apiError('rate_limited', 'Too Many Requests'), message: `Rate limit exceeded, retry in ${context.after}`, retryAfter: context.after }) diff --git a/src/errors.ts b/src/errors.ts new file mode 100644 index 00000000..b02fb60e --- /dev/null +++ b/src/errors.ts @@ -0,0 +1,81 @@ +/** + * The single, enumerated set of stable machine-readable error codes. + * + * Every JSON error response the API emits carries a `code` beside its existing + * `error` string. `error` stays human-readable and may be reworded at any time; + * `code` is a contract a client branches on. Codes are declared here (and + * mirrored in `openapi.yaml`) rather than invented at each call site, so the set + * can be audited in one place and a typo fails to compile instead of shipping. + * + * Lowercase snake_case is deliberate: it matches the codes the Bazaar and x402 + * surfaces already expose (`rate_limited`, `unexpected_settle_error`, ...). + */ +export const ERROR_CODES = { + /** A required query/body/path parameter was not supplied. */ + missing_parameter: 'missing_parameter', + /** A supplied parameter was present but malformed, out of range, or unknown. */ + invalid_parameter: 'invalid_parameter', + /** `network` was not one of the supported networks (testnet, mainnet). */ + bad_network: 'bad_network', + /** An interval was not one of the supported values for the endpoint. */ + bad_interval: 'bad_interval', + /** A window was not one of the supported values for the endpoint. */ + bad_window: 'bad_window', + /** The request named something the API recognises but deliberately does not serve. */ + unsupported: 'unsupported', + /** No price/data exists for the requested pair or asset. */ + unknown_pair: 'unknown_pair', + /** The requested pair is not in the set this instance watches. */ + unwatched_pair: 'unwatched_pair', + /** The pair is already being watched (conflict). */ + pair_already_watched: 'pair_already_watched', + /** Credentials were missing or invalid. */ + unauthorized: 'unauthorized', + /** A referenced resource (e.g. an API key) does not exist. */ + not_found: 'not_found', + /** A dependency the endpoint relies on (database, venue, oracle) was unreachable. */ + upstream_unavailable: 'upstream_unavailable', + /** An unexpected server-side fault. */ + internal_error: 'internal_error', + /** The request was throttled by a quota or rate limit. */ + rate_limited: 'rate_limited', + /** Payment is required before the resource can be served. */ + payment_required: 'payment_required', + /** The supplied payment was rejected. */ + payment_invalid: 'payment_invalid', + /** The payment could not be verified because verification itself failed. */ + payment_verification_failed: 'payment_verification_failed', + /** The caller's quota is exhausted and further payment is required. */ + quota_exceeded: 'quota_exceeded', +} as const + +/** A member of {@link ERROR_CODES}. */ +export type ErrorCode = keyof typeof ERROR_CODES + +/** Every code as a plain string array, for validation and spec generation. */ +export const ERROR_CODE_VALUES: readonly ErrorCode[] = Object.values(ERROR_CODES) + +const ERROR_CODE_SET: ReadonlySet = new Set(ERROR_CODE_VALUES) + +/** Narrows an unknown value to an {@link ErrorCode}. */ +export function isErrorCode(value: unknown): value is ErrorCode { + return typeof value === 'string' && ERROR_CODE_SET.has(value) +} + +/** The JSON body of every non-2xx error response. */ +export interface ApiError { + /** Human-readable message. Do not branch on this — branch on `code`. */ + error: string + /** Stable code from {@link ERROR_CODES}. */ + code: ErrorCode +} + +/** + * Builds an error body from the enumerated set. + * + * Typing `code` as {@link ErrorCode} is what stops a route inventing its own + * code: an unknown value is a compile error, not a runtime surprise. + */ +export function apiError(code: ErrorCode, message: string): ApiError { + return { error: message, code } +} diff --git a/src/middleware/network.ts b/src/middleware/network.ts index 8519559f..7883881e 100644 --- a/src/middleware/network.ts +++ b/src/middleware/network.ts @@ -60,7 +60,7 @@ async function networkSelectorPlugin(app: FastifyInstance) { app.addHook('onRequest', async (req: FastifyRequest, reply: FastifyReply) => { const resolved = resolveNetworkName(rawNetworkFromRequest(req)) if (!resolved.ok) { - reply.status(400).send({ error: resolved.error }) + reply.status(400).send({ error: resolved.error, code: 'bad_network' }) return } req.network = resolved.network diff --git a/src/middleware/x402.ts b/src/middleware/x402.ts index a2651f0a..19600928 100644 --- a/src/middleware/x402.ts +++ b/src/middleware/x402.ts @@ -4,6 +4,7 @@ import { checkQuota, recordUsage, parseCents, getQuotaConfig } from '../x402/met import { X402_NETWORK_LABEL, paymentAddressFor, isX402Configured, getX402ResourceServer } from '../x402/network' import fp from 'fastify-plugin' import './network' // declares req.network on the FastifyRequest type +import { apiError } from '../errors' export interface GatedRoute { readonly path: string @@ -108,7 +109,7 @@ async function x402Plugin(app: FastifyInstance) { const network = req.network ?? 'testnet' const paymentAddress = paymentAddressFor(network) if (!paymentAddress) { - reply.status(402).send({ error: `x402 payments are not configured for network "${network}"` }) + reply.status(402).send(apiError('payment_required', `x402 payments are not configured for network "${network}"`)) return } @@ -127,7 +128,7 @@ async function x402Plugin(app: FastifyInstance) { reply.status(402).send({ x402Version: 1, accepts: [requirements], - error: 'Payment required', + ...apiError('payment_required', 'Payment required'), description, }) return @@ -145,7 +146,7 @@ async function x402Plugin(app: FastifyInstance) { const resourceServer = await getX402ResourceServer(network, FACILITATOR_URL) const result = await resourceServer.verify(payload, requirements) if (!result.isValid) { - reply.status(402).send({ error: 'Payment invalid', reason: result.invalidReason }) + reply.status(402).send({ ...apiError('payment_invalid', 'Payment invalid'), reason: result.invalidReason }) return } @@ -163,14 +164,14 @@ async function x402Plugin(app: FastifyInstance) { await recordUsage(apiKeyId, parseCents(price)) } else if (config.overagePolicy === 'charge_402') { reply.status(402).send({ - error: 'Quota exceeded — additional payment required', + ...apiError('quota_exceeded', 'Quota exceeded — additional payment required'), policy: 'charge_402', }) return } else { // default: block reply.status(402).send({ - error: 'Quota exceeded', + ...apiError('quota_exceeded', 'Quota exceeded'), policy: 'block', }) return @@ -185,7 +186,7 @@ async function x402Plugin(app: FastifyInstance) { app.log.error({ err }, '[oracle] x402 settle error') }) } catch (err) { - reply.status(402).send({ error: 'Payment verification failed', reason: (err as Error).message }) + reply.status(402).send({ ...apiError('payment_verification_failed', 'Payment verification failed'), reason: (err as Error).message }) } }) } diff --git a/src/pagination.ts b/src/pagination.ts new file mode 100644 index 00000000..4eba620c --- /dev/null +++ b/src/pagination.ts @@ -0,0 +1,116 @@ +/** + * Cursor pagination helpers shared by every list route. + * + * The convention is copied verbatim from GET /screener, which was the first + * paginated route in this codebase and is the reference implementation: + * + * - `limit` is clamped into [1, max] with a per-route default; a non-integer + * limit is a 400, never a silent clamp to some arbitrary page size. + * - The cursor is an opaque base64 encoding of `JSON.stringify([value, id])` + * — the last row's sort value plus its tiebreaker id. Callers must treat it + * as opaque; only the shape matters to them. + * - Pagination is keyset-based, not offset-based: the predicate is + * `(sortCol, idCol) > ($value, $id)` (or `<` for descending order), so a + * row inserted while a client is paging can never shift a page boundary + * and cause a duplicate or a skipped row. + * - Handlers fetch `limit + 1` rows; the extra row is the only hasMore + * signal, so no COUNT(*) over the whole table is needed. + * + * Routes keep their own response keys (`pools`, `pairs`, `candles`, `points`, + * screener's `data`) and add `nextCursor` + `hasMore`, so an unpaged caller + * still gets a sensible first page. + */ + +/** Default page size and hard maximum for GET /screener. */ +export const SCREENER_DEFAULT_LIMIT = 20 +export const SCREENER_MAX_LIMIT = 100 + +/** Default page size and hard maximum for GET /pools. */ +export const POOLS_DEFAULT_LIMIT = 50 +export const POOLS_MAX_LIMIT = 200 + +/** Default page size and hard maximum for GET /pairs. */ +export const PAIRS_DEFAULT_LIMIT = 50 +export const PAIRS_MAX_LIMIT = 200 + +/** Default page size and hard maximum for GET /candles/:assetA/:assetB. */ +export const CANDLES_DEFAULT_LIMIT = 500 +export const CANDLES_MAX_LIMIT = 5_000 + +/** + * Default page size for GET /prices/history. The maximum is + * `MAX_HISTORY_POINTS` (src/api/history.ts) so a page can never exceed the + * point cap a single window may produce. + */ +export const HISTORY_DEFAULT_LIMIT = 1_000 + +/** + * Parse a `limit` query parameter. + * + * Returns the clamped integer on success, or `{ error }` for the route to + * answer 400 with — matching /screener's historical behaviour: absent → + * `def`, out-of-range values are clamped, non-numeric values are rejected. + */ +export function parseLimit( + raw: string | undefined, + def: number, + max: number, +): { limit: number } | { error: string } { + if (raw === undefined) return { limit: def } + const parsed = parseInt(raw, 10) + if (isNaN(parsed)) return { error: 'limit must be a valid integer' } + return { limit: Math.min(Math.max(parsed, 1), max) } +} + +/** + * Encode the last row's sort value and tiebreaker id as an opaque cursor. + * + * The payload is a 2-tuple — `[sortValue, id]` — regardless of how many + * columns the route's keyset actually spans, so every route emits the same + * cursor shape. + */ +export function encodeCursor(value: unknown, id: string | number): string { + return Buffer.from(JSON.stringify([value, id])).toString('base64') +} + +/** Result of {@link decodeCursor}: a payload, a deliberate skip, or malformed. */ +export type DecodedCursor = { value: unknown; id: unknown } | 'skip' | null + +/** + * Decode an opaque cursor back into its `[sortValue, id]` payload. + * + * Three outcomes, matching /screener's historical behaviour exactly: + * - `{ value, id }` — apply the keyset predicate with these values. + * - `'skip'` — the cursor is well-formed JSON but its sort value or id is + * empty; the route serves the first page instead of erroring, because a + * degenerate cursor must not punish a first-page caller. + * - `null` — not base64 JSON, or not destructurable at all; the route must + * answer 400 "Invalid cursor format". + */ +export function decodeCursor(raw: string): DecodedCursor { + let parsed: unknown + try { + parsed = JSON.parse(Buffer.from(raw, 'base64').toString('utf-8')) + } catch { + return null + } + let value: unknown + let id: unknown + try { + ;[value, id] = parsed as [unknown, unknown] + } catch { + return null + } + if (value === null || value === undefined || !id) return 'skip' + return { value, id } +} + +/** + * Trim a `limit + 1` row fetch down to the page and report whether another + * page exists. The rows are expected to already be keyset-filtered and + * ordered by the route's SQL (or, for in-memory lists, by the handler). + */ +export function slicePage(rows: T[], limit: number): { data: T[]; hasMore: boolean } { + const hasMore = rows.length > limit + return { data: hasMore ? rows.slice(0, limit) : rows, hasMore } +} diff --git a/src/routes/basket.ts b/src/routes/basket.ts index 3d2083f2..a9547c6b 100644 --- a/src/routes/basket.ts +++ b/src/routes/basket.ts @@ -1,5 +1,6 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' +import { apiError } from '../errors' async function fetchAssetVWAP(asset: string): Promise { const result = await pgPool.query( @@ -31,18 +32,18 @@ export async function registerBasketRoutes(app: FastifyInstance) { : [] if (assets.length === 0) { - return reply.status(400).send({ error: 'At least one asset is required' }) + return reply.status(400).send(apiError('missing_parameter', 'At least one asset is required')) } if (assets.length < 2) { - return reply.status(400).send({ error: 'Basket requires at least 2 assets' }) + return reply.status(400).send(apiError('invalid_parameter', 'Basket requires at least 2 assets')) } if (assets.length !== weightStrings.length) { - return reply.status(400).send({ error: 'Number of assets and weights must match' }) + return reply.status(400).send(apiError('invalid_parameter', 'Number of assets and weights must match')) } const rawWeightValues = weightStrings.map(w => parseFloat(w)) if (rawWeightValues.some(w => isNaN(w) || w <= 0)) { - return reply.status(400).send({ error: 'All weights must be positive numbers' }) + return reply.status(400).send(apiError('invalid_parameter', 'All weights must be positive numbers')) } const weightSum = rawWeightValues.reduce((a, b) => a + b, 0) @@ -58,7 +59,7 @@ export async function registerBasketRoutes(app: FastifyInstance) { const missingAssets = components.filter(c => c.price === null).map(c => c.asset) if (missingAssets.length > 0) { - return reply.status(404).send({ error: `No price data found for: ${missingAssets.join(', ')}` }) + return reply.status(404).send(apiError('unknown_pair', `No price data found for: ${missingAssets.join(', ')}`)) } const basketPrice = components.reduce( diff --git a/src/routes/benchmark.ts b/src/routes/benchmark.ts index 25d5a87e..d1e82533 100644 --- a/src/routes/benchmark.ts +++ b/src/routes/benchmark.ts @@ -2,6 +2,7 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { getActivePairs } from '../pairsRegistry' import { benchmarkResponseSchema } from '../api/schemas' +import { apiError } from '../errors' function findPair(assetCode: string, targetCode: string) { const normalize = (c: string) => c.toLowerCase() === 'native' ? 'XLM' : c.split(':')[0].toUpperCase() @@ -31,7 +32,7 @@ export async function registerBenchmarkRoutes(app: FastifyInstance) { const pair = findPair(asset, target) if (!pair) { - return reply.status(404).send({ error: `Pair ${asset}/${target} not watched` }) + return reply.status(404).send(apiError('unwatched_pair', `Pair ${asset}/${target} not watched`)) } const isAssetA = pair.assetA.code.toUpperCase() === normAsset @@ -81,7 +82,7 @@ export async function registerBenchmarkRoutes(app: FastifyInstance) { rolling24h, } } catch (err) { - return reply.status(500).send({ error: `Benchmark computation failed: ${(err as Error).message}` }) + return reply.status(500).send(apiError('upstream_unavailable', `Benchmark computation failed: ${(err as Error).message}`)) } } ) diff --git a/src/routes/book.ts b/src/routes/book.ts index 1c989b61..126531d5 100644 --- a/src/routes/book.ts +++ b/src/routes/book.ts @@ -7,6 +7,7 @@ import '../middleware/network' import { getHorizonServer } from '../network/clients' import { aggregateBookLevels, generateAmmBookLevels, type AmmBookPool, type RawBookLevel } from '../pricing/book' import { bookResponseSchema } from '../api/schemas' +import { apiError } from '../errors' const BookQuerySchema = z.object({ depth: z.coerce.number().min(0).max(0.5).default(0.1), @@ -138,7 +139,7 @@ export async function registerBookRoutes(app: FastifyInstance) { }>('/book/:assetA/:assetB', { schema: { response: { 200: bookResponseSchema, 503: bookResponseSchema } } }, async (req, reply) => { const parsedQuery = BookQuerySchema.safeParse(req.query) if (!parsedQuery.success) { - return reply.status(400).send({ error: parsedQuery.error.issues[0]?.message ?? 'Invalid book parameters' }) + return reply.status(400).send(apiError('invalid_parameter', parsedQuery.error.issues[0]?.message ?? 'Invalid book parameters')) } let assetA: ParsedAsset @@ -148,7 +149,7 @@ export async function registerBookRoutes(app: FastifyInstance) { assetB = parseAsset(req.params.assetB) if (assetA.canonical === assetB.canonical) throw new Error('Assets must be different') } catch (err) { - return reply.status(400).send({ error: (err as Error).message }) + return reply.status(400).send(apiError('invalid_parameter', (err as Error).message)) } const network: NetworkName = req.network ?? activeNetwork diff --git a/src/routes/candles.ts b/src/routes/candles.ts index c83c56c5..ac094a74 100644 --- a/src/routes/candles.ts +++ b/src/routes/candles.ts @@ -1,5 +1,14 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' +import { apiError } from '../errors' +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + CANDLES_DEFAULT_LIMIT, + CANDLES_MAX_LIMIT, +} from '../pagination' const INTERVAL_SECONDS: Record = { '1m': 60, @@ -11,20 +20,34 @@ const INTERVAL_SECONDS: Record = { } export async function registerCandleRoutes(app: FastifyInstance) { + // GET /candles/:assetA/:assetB?interval=&from=&to=&cursor=&limit= + // + // Paginated keyset-style on the bucket start time ASC (default limit 500, + // max 5000). The cursor carries the last page's bucket epoch in seconds (the + // tiebreaker id is the pairKey, kept so every route emits the same cursor + // shape), and the keyset predicate is applied to the bucket expression in + // the WHERE clause, so a candle inserted mid-paging cannot shift a page + // boundary the way an offset would. app.get<{ Params: { assetA: string; assetB: string } - Querystring: { interval?: string; from?: string; to?: string } + Querystring: { interval?: string; from?: string; to?: string; cursor?: string; limit?: string } }>('/candles/:assetA/:assetB', async (req, reply) => { const { assetA, assetB } = req.params const interval = req.query.interval ?? '1h' const intervalSecs = INTERVAL_SECONDS[interval] if (!intervalSecs) { - return reply.status(400).send({ - error: `interval must be one of: ${Object.keys(INTERVAL_SECONDS).join(', ')}`, - }) + return reply.status(400).send( + apiError('bad_interval', `interval must be one of: ${Object.keys(INTERVAL_SECONDS).join(', ')}`), + ) } + const parsedLimit = parseLimit(req.query.limit, CANDLES_DEFAULT_LIMIT, CANDLES_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send(apiError('invalid_parameter', parsedLimit.error)) + } + const { limit } = parsedLimit + const pairKey = [assetA, assetB].sort().join('/') const fromRaw = req.query.from @@ -33,12 +56,26 @@ export async function registerCandleRoutes(app: FastifyInstance) { const to = toRaw ? new Date(toRaw) : new Date() if (isNaN(from.getTime())) { - return reply.status(400).send({ error: 'from must be a valid ISO 8601 date' }) + return reply.status(400).send(apiError('invalid_parameter', 'from must be a valid ISO 8601 date')) } if (isNaN(to.getTime())) { - return reply.status(400).send({ error: 'to must be a valid ISO 8601 date' }) + return reply.status(400).send(apiError('invalid_parameter', 'to must be a valid ISO 8601 date')) } + const params: unknown[] = [intervalSecs, pairKey, from, to] + let cursorCondition = '' + if (req.query.cursor) { + const decoded = decodeCursor(req.query.cursor) + if (decoded === null) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + if (decoded !== 'skip') { + params.push(Number(decoded.value)) + cursorCondition = ` AND floor(EXTRACT(EPOCH FROM timestamp) / $1) * $1 > $${params.length}` + } + } + params.push(limit + 1) // Fetch one extra to determine hasMore + const result = await pgPool.query( `SELECT to_timestamp(floor(EXTRACT(EPOCH FROM timestamp) / $1) * $1) AS time, @@ -50,18 +87,25 @@ export async function registerCandleRoutes(app: FastifyInstance) { FROM price_points WHERE pair_key = $2 AND timestamp >= $3 - AND timestamp <= $4 + AND timestamp <= $4${cursorCondition} GROUP BY floor(EXTRACT(EPOCH FROM timestamp) / $1) - ORDER BY time ASC`, - [intervalSecs, pairKey, from, to] + ORDER BY time ASC + LIMIT $${params.length}`, + params ) + const { data, hasMore } = slicePage(result.rows, limit) + const last = hasMore && data.length > 0 ? data[data.length - 1] : null + const nextCursor = last + ? encodeCursor(Math.floor(new Date(last.time).getTime() / 1000), pairKey) + : null + return { pairKey, interval, from: from.toISOString(), to: to.toISOString(), - candles: result.rows.map(r => ({ + candles: data.map(r => ({ time: r.time, open: Number(r.open), high: Number(r.high), @@ -69,6 +113,8 @@ export async function registerCandleRoutes(app: FastifyInstance) { close: Number(r.close), volume: Number(r.volume), })), + nextCursor, + hasMore, } }) } diff --git a/src/routes/oracle.ts b/src/routes/oracle.ts index 474c7003..3eed6be8 100644 --- a/src/routes/oracle.ts +++ b/src/routes/oracle.ts @@ -3,6 +3,7 @@ import { activeNetwork, type NetworkName } from '../config' import type { OracleAdapter } from '../ingest/oracles/adapter' import { reflectorOracleAdapter } from '../ingest/oracles/reflector' import { pgPool } from '../db' +import { apiError } from '../errors' const oracleAdapters: OracleAdapter[] = [reflectorOracleAdapter] @@ -26,7 +27,7 @@ export async function registerOracleRoutes( const assetCode = req.params.asset.toUpperCase() const requestedNetwork = req.query.network if (requestedNetwork !== undefined && requestedNetwork !== 'testnet' && requestedNetwork !== 'mainnet') { - return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) + return reply.status(400).send(apiError('bad_network', 'network must be one of: testnet, mainnet')) } const network: NetworkName = requestedNetwork ?? activeNetwork @@ -60,7 +61,7 @@ export async function registerOracleRoutes( const lensPrice = dbResult.rows[0] ? parseFloat(dbResult.rows[0].vwap) : null const anyOracleConfigured = oracleResults.some(result => result.configured) if (lensPrice === null && !anyOracleConfigured) { - return reply.status(404).send({ error: `No price data found for asset ${assetCode}` }) + return reply.status(404).send(apiError('unknown_pair', `No price data found for asset ${assetCode}`)) } const oracles = oracleResults.map(result => { diff --git a/src/routes/pairs.ts b/src/routes/pairs.ts index f2208e04..79e05381 100644 --- a/src/routes/pairs.ts +++ b/src/routes/pairs.ts @@ -3,6 +3,15 @@ import { getActivePairs, parseAssetStr, makePairKey, registerPair, persistPair, import { pgPool } from '../db' import { activeNetwork } from '../config' import '../middleware/network' // declares req.network on the FastifyRequest type +import { apiError } from '../errors' +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + PAIRS_DEFAULT_LIMIT, + PAIRS_MAX_LIMIT, +} from '../pagination' export async function registerPairsRoutes(app: FastifyInstance) { // GET /pairs — list all active pairs with latest price metadata @@ -16,10 +25,31 @@ export async function registerPairsRoutes(app: FastifyInstance) { // the probe still finds its last print, and the response carries that print's // real `lastUpdated`, so a caller can see it is old. A pair with no rows on // this network at all comes back with null price and timestamp. - app.get('/pairs', async (req) => { + // + // Paginated keyset-style on pairKey ASC (default limit 50, max 200). The rows + // are assembled in memory from the registry, so the keyset filter and the + // limit+1 probe are applied to the merged list; the cursor carries the last + // page's pairKey, so a pair registered mid-paging cannot shift a page + // boundary the way an offset would. + app.get<{ Querystring: { cursor?: string; limit?: string } }>('/pairs', async (req, reply) => { const activePairs = getActivePairs() const network = req.network ?? activeNetwork + const parsedLimit = parseLimit(req.query.limit, PAIRS_DEFAULT_LIMIT, PAIRS_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send(apiError('invalid_parameter', parsedLimit.error)) + } + const { limit } = parsedLimit + + let afterKey: string | null = null + if (req.query.cursor) { + const decoded = decodeCursor(req.query.cursor) + if (decoded === null) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + if (decoded !== 'skip') afterKey = String(decoded.value) + } + const result = await pgPool.query( `SELECT p.pair_key, l.price, l.timestamp FROM unnest($2::text[]) AS p(pair_key) @@ -42,8 +72,8 @@ export async function registerPairsRoutes(app: FastifyInstance) { }) }) - return { - pairs: activePairs.map(p => { + const rows = activePairs + .map(p => { const latest = latestPrices.get(p.pairKey) return { pairKey: p.pairKey, @@ -53,7 +83,15 @@ export async function registerPairsRoutes(app: FastifyInstance) { lastUpdated: latest?.timestamp ?? null, } }) - } + .sort((a, b) => (a.pairKey < b.pairKey ? -1 : a.pairKey > b.pairKey ? 1 : 0)) + + const pageSource = afterKey ? rows.filter(r => r.pairKey > afterKey) : rows + const { data, hasMore } = slicePage(pageSource, limit) + const nextCursor = + hasMore && data.length > 0 + ? encodeCursor(data[data.length - 1].pairKey, data[data.length - 1].pairKey) + : null + return { pairs: data, nextCursor, hasMore } }) // POST /pairs — add a new trading pair at runtime @@ -64,33 +102,33 @@ export async function registerPairsRoutes(app: FastifyInstance) { const ADMIN_API_KEY = process.env.ADMIN_API_KEY const key = req.headers['x-admin-key'] ?? req.headers['authorization']?.replace(/^Bearer /, '') if (!ADMIN_API_KEY || key !== ADMIN_API_KEY) { - return reply.status(401).send({ error: 'Unauthorized — provide a valid X-Admin-Key header' }) + return reply.status(401).send(apiError('unauthorized', 'Unauthorized — provide a valid X-Admin-Key header')) } const { assetA: assetAStr, assetB: assetBStr } = req.body ?? {} if (!assetAStr || !assetBStr) { - return reply.status(400).send({ error: 'assetA and assetB are required' }) + return reply.status(400).send(apiError('missing_parameter', 'assetA and assetB are required')) } const assetA = parseAssetStr(assetAStr) if (!assetA) { - return reply.status(400).send({ - error: `Invalid assetA format "${assetAStr}". Expected CODE or CODE:ISSUER (e.g. XLM:native or USDC:GBBD47...)`, - }) + return reply.status(400).send( + apiError('invalid_parameter', `Invalid assetA format "${assetAStr}". Expected CODE or CODE:ISSUER (e.g. XLM:native or USDC:GBBD47...)`), + ) } const assetB = parseAssetStr(assetBStr) if (!assetB) { - return reply.status(400).send({ - error: `Invalid assetB format "${assetBStr}". Expected CODE or CODE:ISSUER (e.g. XLM:native or USDC:GBBD47...)`, - }) + return reply.status(400).send( + apiError('invalid_parameter', `Invalid assetB format "${assetBStr}". Expected CODE or CODE:ISSUER (e.g. XLM:native or USDC:GBBD47...)`), + ) } const pairKey = makePairKey(assetA, assetB) if (hasPair(pairKey)) { - return reply.status(409).send({ error: `Pair ${pairKey} is already being watched` }) + return reply.status(409).send(apiError('pair_already_watched', `Pair ${pairKey} is already being watched`)) } const pair = { assetA, assetB, pairKey } diff --git a/src/routes/price.ts b/src/routes/price.ts index 241a0609..6f84e0ec 100644 --- a/src/routes/price.ts +++ b/src/routes/price.ts @@ -3,6 +3,7 @@ import { computeTWAP, computeVWAP } from '../pricing/twap' import { z } from 'zod' import '../middleware/network' import { activeNetwork, type NetworkName } from '../config' +import { apiError } from '../errors' const TwapQuerySchema = z.object({ window: z.coerce.number().int().min(1).max(1440).default(60), @@ -39,7 +40,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { const { assetA, assetB } = req.params const parsed = TwapQuerySchema.safeParse(req.query) if (!parsed.success) { - return reply.status(400).send({ error: parsed.error.issues[0].message || 'Invalid parameters' }) + return reply.status(400).send(apiError('invalid_parameter', parsed.error.issues[0].message || 'Invalid parameters')) } const { window: windowMinutes, sampleInterval, method } = parsed.data const network: NetworkName = req.network ?? parsed.data.network ?? activeNetwork @@ -71,7 +72,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { // A driver error message can carry the connection string, and so the // Postgres credentials. Log it; never put it in the response body. console.error('[price] TWAP computation failed:', (err as Error).message) - return reply.status(500).send({ error: 'TWAP computation failed' }) + return reply.status(500).send(apiError('upstream_unavailable', 'TWAP computation failed')) } } ) @@ -90,7 +91,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { const { assetA, assetB } = req.params const parsed = VwapQuerySchema.safeParse(req.query) if (!parsed.success) { - return reply.status(400).send({ error: parsed.error.issues[0].message || 'Invalid parameters' }) + return reply.status(400).send(apiError('invalid_parameter', parsed.error.issues[0].message || 'Invalid parameters')) } const { window: windowMinutes, source, method } = parsed.data @@ -119,7 +120,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { } catch (err) { // Same as TWAP above: the driver message can contain credentials. console.error('[price] VWAP computation failed:', (err as Error).message) - return reply.status(500).send({ error: 'VWAP computation failed' }) + return reply.status(500).send(apiError('upstream_unavailable', 'VWAP computation failed')) } } ) diff --git a/src/routes/screener.ts b/src/routes/screener.ts index 783059d4..b9af5f77 100644 --- a/src/routes/screener.ts +++ b/src/routes/screener.ts @@ -2,6 +2,15 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork } from '../config' import '../middleware/network' // declares req.network on the FastifyRequest type +import { apiError } from '../errors' +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + SCREENER_DEFAULT_LIMIT, + SCREENER_MAX_LIMIT, +} from '../pagination' // Lens has no circulating-supply data, so a market cap cannot be computed. It // used to be reported as a copy of `liquidity`; a caller who sorts or filters on @@ -29,24 +38,25 @@ export async function registerScreenerRoutes(app: FastifyInstance) { const q = req.query // 1. Parse and validate pagination & sorting - const limit = Math.min(Math.max(parseInt(q.limit ?? '20', 10), 1), 100) - if (isNaN(limit)) { - return reply.status(400).send({ error: 'limit must be a valid integer' }) + const parsedLimit = parseLimit(q.limit, SCREENER_DEFAULT_LIMIT, SCREENER_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send(apiError('invalid_parameter', parsedLimit.error)) } + const limit = parsedLimit.limit if (q.market_cap !== undefined || q.sortBy === 'market_cap') { - return reply.status(400).send({ error: MARKET_CAP_UNSUPPORTED }) + return reply.status(400).send(apiError('unsupported', MARKET_CAP_UNSUPPORTED)) } const allowedSortFields = ['volume', 'change_24h', 'price', 'liquidity'] const sortBy = q.sortBy ?? 'volume' if (!allowedSortFields.includes(sortBy)) { - return reply.status(400).send({ error: `sortBy must be one of: ${allowedSortFields.join(', ')}` }) + return reply.status(400).send(apiError('invalid_parameter', `sortBy must be one of: ${allowedSortFields.join(', ')}`)) } const order = (q.order ?? 'desc').toLowerCase() if (order !== 'asc' && order !== 'desc') { - return reply.status(400).send({ error: 'order must be asc or desc' }) + return reply.status(400).send(apiError('invalid_parameter', 'order must be asc or desc')) } // 2. Parse filters @@ -62,7 +72,7 @@ export async function registerScreenerRoutes(app: FastifyInstance) { for (const [key, val] of Object.entries(filters)) { if (val !== undefined && isNaN(val)) { - return reply.status(400).send({ error: `Filter ${key} must be a valid number` }) + return reply.status(400).send(apiError('invalid_parameter', `Filter ${key} must be a valid number`)) } } @@ -100,17 +110,14 @@ export async function registerScreenerRoutes(app: FastifyInstance) { // 4. Cursor decoding and pagination condition if (q.cursor) { - try { - const decoded = Buffer.from(q.cursor, 'base64').toString('utf-8') - const [lastVal, lastId] = JSON.parse(decoded) - - if (lastVal !== null && lastVal !== undefined && lastId) { - const op = order === 'desc' ? '<' : '>' - conditions.push(`(${sortBy}, pair_key) ${op} ($${paramIdx++}, $${paramIdx++})`) - params.push(lastVal, lastId) - } - } catch (e) { - return reply.status(400).send({ error: 'Invalid cursor format' }) + const decoded = decodeCursor(q.cursor) + if (decoded === null) { + return reply.status(400).send(apiError('invalid_parameter', 'Invalid cursor format')) + } + if (decoded !== 'skip') { + const op = order === 'desc' ? '<' : '>' + conditions.push(`(${sortBy}, pair_key) ${op} ($${paramIdx++}, $${paramIdx++})`) + params.push(decoded.value, decoded.id) } } @@ -169,17 +176,13 @@ export async function registerScreenerRoutes(app: FastifyInstance) { // 6. Execute query params.push(limit + 1) // Fetch one extra to determine hasMore const result = await pgPool.query(sql, params) - const rows = result.rows - - const hasMore = rows.length > limit - const data = hasMore ? rows.slice(0, limit) : rows + const { data, hasMore } = slicePage(result.rows, limit) // 7. Generate nextCursor let nextCursor = null if (hasMore && data.length > 0) { const lastRecord = data[data.length - 1] - const cursorPayload = JSON.stringify([lastRecord[sortBy], lastRecord.pair_key]) - nextCursor = Buffer.from(cursorPayload).toString('base64') + nextCursor = encodeCursor(lastRecord[sortBy], lastRecord.pair_key) } return { @@ -190,7 +193,7 @@ export async function registerScreenerRoutes(app: FastifyInstance) { } } catch (error) { console.error('[screener]', error) - return reply.status(500).send({ error: 'Internal server error' }) + return reply.status(500).send(apiError('internal_error', 'Internal server error')) } }) } diff --git a/src/routes/spreads.ts b/src/routes/spreads.ts index bd229e43..4e468cbc 100644 --- a/src/routes/spreads.ts +++ b/src/routes/spreads.ts @@ -1,6 +1,7 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' +import { apiError } from '../errors' // Supported windows → lookback in minutes. A spread is a statement about *now*, // so these are deliberately short: widen the window far enough and you stop @@ -74,13 +75,13 @@ export async function registerSpreadsRoutes(app: FastifyInstance) { const window = req.query.window ?? '5m' if (!asset) { - return reply.status(400).send({ error: 'Asset parameter is required' }) + return reply.status(400).send(apiError('missing_parameter', 'Asset parameter is required')) } if (!(window in WINDOW_MINUTES)) { return reply .status(400) - .send({ error: `window must be one of: ${WINDOWS.join(', ')}` }) + .send(apiError('bad_window', `window must be one of: ${WINDOWS.join(', ')}`)) } // Prices from two chains are not comparable, and a spread computed across @@ -88,7 +89,7 @@ export async function registerSpreadsRoutes(app: FastifyInstance) { // process's own network rather than silently pooling them. const requested = req.query.network if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { - return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) + return reply.status(400).send(apiError('bad_network', 'network must be one of: testnet, mainnet')) } const network: NetworkName = (requested as NetworkName) ?? activeNetwork @@ -159,7 +160,7 @@ export async function registerSpreadsRoutes(app: FastifyInstance) { } catch (err) { return reply .status(500) - .send({ error: `Spread aggregation failed: ${(err as Error).message}` }) + .send(apiError('upstream_unavailable', `Spread aggregation failed: ${(err as Error).message}`)) } }) } diff --git a/src/routes/volumes.ts b/src/routes/volumes.ts index 802ba03d..7d7b67e3 100644 --- a/src/routes/volumes.ts +++ b/src/routes/volumes.ts @@ -1,6 +1,7 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' +import { apiError } from '../errors' // Supported windows → lookback in hours. const WINDOW_HOURS = { @@ -39,12 +40,12 @@ export async function registerVolumeRoutes(app: FastifyInstance) { if (!(window in WINDOW_HOURS)) { return reply .status(400) - .send({ error: `window must be one of: ${WINDOWS.join(', ')}` }) + .send(apiError('bad_window', `window must be one of: ${WINDOWS.join(', ')}`)) } const requested = req.query.network if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { - return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) + return reply.status(400).send(apiError('bad_network', 'network must be one of: testnet, mainnet')) } const network: NetworkName = (requested as NetworkName) ?? activeNetwork @@ -100,7 +101,7 @@ export async function registerVolumeRoutes(app: FastifyInstance) { } catch (err) { return reply .status(500) - .send({ error: `Volume aggregation failed: ${(err as Error).message}` }) + .send(apiError('upstream_unavailable', `Volume aggregation failed: ${(err as Error).message}`)) } }) } diff --git a/src/routes/webhooks.ts b/src/routes/webhooks.ts index 1d9e2426..b60b8665 100644 --- a/src/routes/webhooks.ts +++ b/src/routes/webhooks.ts @@ -3,6 +3,7 @@ import { randomBytes } from 'crypto' import { prisma } from '../db' import { activeNetwork, type NetworkName } from '../config' import { resolveNetworkName } from '../middleware/network' +import { apiError } from '../errors' function isValidHttpsUrl(raw: string): boolean { try { @@ -21,26 +22,26 @@ export async function registerWebhookRoutes(app: FastifyInstance) { const { url, assetA, assetB, threshold, direction, network: bodyNetwork } = req.body ?? {} if (!url || !isValidHttpsUrl(url)) { - return reply.status(400).send({ error: 'url must be a valid HTTPS URL' }) + return reply.status(400).send(apiError('invalid_parameter', 'url must be a valid HTTPS URL')) } if (!assetA || typeof assetA !== 'string') { - return reply.status(400).send({ error: 'assetA is required' }) + return reply.status(400).send(apiError('missing_parameter', 'assetA is required')) } if (!assetB || typeof assetB !== 'string') { - return reply.status(400).send({ error: 'assetB is required' }) + return reply.status(400).send(apiError('missing_parameter', 'assetB is required')) } if (typeof threshold !== 'number' || isNaN(threshold)) { - return reply.status(400).send({ error: 'threshold must be a number' }) + return reply.status(400).send(apiError('invalid_parameter', 'threshold must be a number')) } if (direction !== 'above' && direction !== 'below') { - return reply.status(400).send({ error: 'direction must be "above" or "below"' }) + return reply.status(400).send(apiError('invalid_parameter', 'direction must be "above" or "below"')) } let targetNetwork: NetworkName = req.network ?? activeNetwork if (bodyNetwork !== undefined) { const resolved = resolveNetworkName(bodyNetwork) if (!resolved.ok) { - return reply.status(400).send({ error: resolved.error }) + return reply.status(400).send(apiError('bad_network', resolved.error)) } targetNetwork = resolved.network } From c1be6653c919108d77903907432c5aaaeea5821a Mon Sep 17 00:00:00 2001 From: DevScoopee Date: Wed, 28 Oct 2026 00:07:23 -0400 Subject: [PATCH 2/3] Regenerate TypeScript client for the ErrorCode and Error schemas --- clients/typescript/src/schema.d.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index d167db35..cfdbc158 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -948,6 +948,16 @@ export interface paths { export type webhooks = Record; export interface components { schemas: { + /** + * @description Stable machine-readable error code. Branch on this, never on `error`, which is human-readable. + * @enum {string} + */ + ErrorCode: "missing_parameter" | "invalid_parameter" | "bad_network" | "bad_interval" | "bad_window" | "unsupported" | "unknown_pair" | "unwatched_pair" | "pair_already_watched" | "unauthorized" | "not_found" | "upstream_unavailable" | "internal_error" | "rate_limited" | "payment_required" | "payment_invalid" | "payment_verification_failed" | "quota_exceeded" | "admin_not_configured" | "admin_unauthorized" | "missing" | "malformed" | "unknown" | "revoked" | "expired" | "insufficient_scope" | "discovery_database_unavailable" | "VALIDATION_ERROR" | "UNSUPPORTED_PARAMETER" | "INVALID_NETWORK" | "STALE_DATA" | "INVALID_PRICE" | "INTERNAL_ERROR" | "DB_POOL_EXHAUSTED"; + Error: { + /** @description Human-readable error message. Do not branch on it. */ + error: string; + code: components["schemas"]["ErrorCode"]; + }; LendingRateEntry: { id: string; /** @example blend */ From a2b74a1f1b5497ee1a6cf88d8241259d6bed20c5 Mon Sep 17 00:00:00 2001 From: DevScoopee Date: Wed, 28 Oct 2026 00:14:37 -0400 Subject: [PATCH 3/3] Keep /status 503 body as on main; expect code in benchmark and movers tests statusErrorResponseSchema is additionalProperties: false by design (the route is public and must not grow fields), so adding code made the serializer fail with 500. /status already exposes pagingCondition as its machine-readable field. --- src/__tests__/benchmark.test.ts | 2 +- src/__tests__/movers.test.ts | 2 +- src/api/rest.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/__tests__/benchmark.test.ts b/src/__tests__/benchmark.test.ts index 575099e8..defa1486 100644 --- a/src/__tests__/benchmark.test.ts +++ b/src/__tests__/benchmark.test.ts @@ -37,7 +37,7 @@ describe('GET /benchmark/:asset', () => { }) expect(res.statusCode).toBe(404) - expect(res.json()).toEqual({ error: 'Pair USDC/USD not watched' }) + expect(res.json()).toEqual({ error: 'Pair USDC/USD not watched', code: 'unwatched_pair' }) }) it('calculates benchmark deviation correctly when asset is assetB in the pair (e.g. USD/USDC)', async () => { diff --git a/src/__tests__/movers.test.ts b/src/__tests__/movers.test.ts index 9f17d2c0..402432a5 100644 --- a/src/__tests__/movers.test.ts +++ b/src/__tests__/movers.test.ts @@ -279,7 +279,7 @@ describe('GET /movers', () => { const res = await app.inject({ method: 'GET', url: '/movers' }) expect(res.statusCode).toBe(500) - expect(res.json()).toEqual({ error: 'Internal server error' }) + expect(res.json()).toEqual({ error: 'Internal server error', code: 'internal_error' }) expect(res.body).not.toMatch(/secret/) await app.close() diff --git a/src/api/rest.ts b/src/api/rest.ts index bb75b291..8b70d059 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -131,7 +131,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { return reply.status(503).send({ ok: false, status: 'unhealthy', - ...apiError('upstream_unavailable', 'Database unavailable'), + error: 'Database unavailable', pagingCondition: 'DATABASE_DOWN', }) }