From 90e33e834e461bc2cab10fd5c24d9b31086154a6 Mon Sep 17 00:00:00 2001 From: DevScoopee Date: Sat, 17 Oct 2026 08:24:53 -0400 Subject: [PATCH] =?UTF-8?q?Every=20list=20route=20in=20the=20API=20?= =?UTF-8?q?=E2=80=94=20/pools,=20/pairs,=20/candles=20and=20/prices/histor?= =?UTF-8?q?y=20=E2=80=94=20now=20pages=20with=20the=20same=20cursor=20and?= =?UTF-8?q?=20limit=20parameters=20and=20the=20same=20opaque=20cursor=20sh?= =?UTF-8?q?ape=20that=20/screener=20already=20used,=20backed=20by=20keyset?= =?UTF-8?q?=20pagination=20so=20rows=20inserted=20while=20a=20client=20pag?= =?UTF-8?q?es=20can=20never=20shift=20a=20page=20boundary,=20with=20all=20?= =?UTF-8?q?defaults=20and=20maxima=20documented=20per=20route=20in=20opena?= =?UTF-8?q?pi.yaml=20and=20the=20behaviour=20locked=20down=20by=20tests=20?= =?UTF-8?q?including=20one=20that=20inserts=20a=20row=20mid-page.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem. /pools, /pairs, /candles and /prices/history returned everything they had in one response. That is tolerable today, but each venue added — Phoenix, Blend and DeFindex are queued — grows those responses without bound, and the first casualty will be a mobile client timing out on /pools. Offset pagination would not have been a fix either, because a row inserted while a client pages shifts every subsequent page and produces duplicates and skipped rows. What changed. The cursor-and-limit convention already proven on /screener has been extracted into a shared helper (src/pagination.ts, providing parseLimit, encodeCursor/decodeCursor and slicePage) and applied to every list route rather than reinvented per route. Every route now accepts the same two parameters: an opaque cursor, which is a base64 JSON [sortValue, id] tuple passed through unchanged from the previous page's nextCursor, and a limit, which is clamped into [1, max] and rejected with a 400 if it is not a valid integer. The cursor shape is identical everywhere, so a client can write a single paging loop for the whole API. Pagination is keyset-based, not offset-based: each route filters strictly past the cursor's sort value, so a row inserted mid-page cannot shift a page boundary. Handlers fetch limit + 1 rows to compute hasMore, so no COUNT(*) over the whole table is needed. Specifically, /pools keysets on pool_id with a default limit of 50 and a maximum of 200, /pairs on pairKey with the same 50/200, /candles on the bucket start time with 500/5000, and /prices/history on the snapshot timestamp with 1000/10000, reusing MAX_HISTORY_POINTS as the ceiling. When /prices/history is called with format=csv, the byte-exact CSV body stays untouched and the pagination envelope travels in the X-Next-Cursor and X-Has-More response headers instead. Closes #263 --- openapi.json | 176 +++++++++++++++++++-- openapi.yaml | 125 ++++++++++++++- src/__tests__/candles.test.ts | 50 ++++++ src/__tests__/history.test.ts | 101 ++++++++++++ src/__tests__/pagination.test.ts | 253 +++++++++++++++++++++++++++++++ src/__tests__/pairs.test.ts | 7 +- src/__tests__/pools.test.ts | 7 +- src/api/history.ts | 83 ++++++++-- src/api/rest.ts | 95 +++++++++--- src/api/schemas.ts | 7 +- src/pagination.ts | 116 ++++++++++++++ src/routes/candles.ts | 55 ++++++- src/routes/pairs.ts | 45 +++++- src/routes/screener.ts | 40 ++--- 14 files changed, 1071 insertions(+), 89 deletions(-) create mode 100644 src/__tests__/pagination.test.ts create mode 100644 src/pagination.ts diff --git a/openapi.json b/openapi.json index be8c1005..739f927c 100644 --- a/openapi.json +++ b/openapi.json @@ -284,6 +284,7 @@ "get": { "operationId": "getPools", "summary": "Get AMM liquidity pool reserves and spot prices", + "description": "Latest snapshot per pool for one network. Paginated with the shared keyset convention: ordered by pool_id ascending, default limit 50, maximum 200.", "security": [ { "x402": [] @@ -292,6 +293,21 @@ "parameters": [ { "$ref": "#/components/parameters/NetworkQuery" + }, + { + "$ref": "#/components/parameters/CursorQuery" + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Page size. Out-of-range values are clamped into [1, 200].", + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + } } ], "responses": { @@ -301,6 +317,11 @@ "application/json": { "schema": { "type": "object", + "required": [ + "pools", + "nextCursor", + "hasMore" + ], "properties": { "pools": { "type": "array", @@ -334,12 +355,22 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true + }, + "hasMore": { + "type": "boolean" } } } } } }, + "400": { + "description": "Invalid limit or cursor" + }, "402": { "$ref": "#/components/responses/PaymentRequired" }, @@ -353,6 +384,7 @@ "get": { "operationId": "getCandles", "summary": "Get OHLCV candle data", + "description": "OHLCV candles for a pair. Paginated with the shared keyset convention: ordered by bucket start time ascending, default limit 500, maximum 5000.", "security": [ { "x402": [] @@ -409,6 +441,21 @@ "type": "string", "format": "date-time" } + }, + { + "$ref": "#/components/parameters/CursorQuery" + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Page size. Out-of-range values are clamped into [1, 5000].", + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 5000, + "default": 500 + } } ], "responses": { @@ -418,6 +465,15 @@ "application/json": { "schema": { "type": "object", + "required": [ + "pairKey", + "interval", + "from", + "to", + "candles", + "nextCursor", + "hasMore" + ], "properties": { "pairKey": { "type": "string" @@ -426,10 +482,12 @@ "type": "string" }, "from": { - "type": "string" + "type": "string", + "format": "date-time" }, "to": { - "type": "string" + "type": "string", + "format": "date-time" }, "candles": { "type": "array", @@ -457,6 +515,13 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true + }, + "hasMore": { + "type": "boolean" } } } @@ -479,7 +544,7 @@ "get": { "operationId": "getPriceSnapshots", "summary": "Get historical 1-minute price snapshots", - "description": "Returns minute-resolution price snapshots for a pair over [from, to], optionally aggregated into 5m or 1h buckets. Snapshots are retained for ~30 days. Gated by x402 (matches the /price prefix).", + "description": "Returns minute-resolution price snapshots for a pair over [from, to], optionally aggregated into 5m or 1h buckets. Snapshots are retained for ~30 days. Paginated with the shared keyset convention: ordered by bucket start time ascending, default limit 1000, maximum 10000 points. With format=csv the pagination envelope travels in the X-Next-Cursor and X-Has-More response headers; with format=json it is in the body. Gated by x402 (matches the /price prefix).", "security": [ { "x402": [] @@ -541,13 +606,46 @@ "default": "json" } }, + { + "$ref": "#/components/parameters/CursorQuery" + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Page size (number of buckets). Out-of-range values are clamped into [1, 10000].", + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 10000, + "default": 1000 + } + }, { "$ref": "#/components/parameters/NetworkQuery" } ], "responses": { "200": { - "description": "Historical price snapshots", + "description": "Historical price snapshots. When more buckets exist beyond this page, the response carries X-Next-Cursor (pass it back as ?cursor=) and X-Has-More: true.", + "headers": { + "X-Next-Cursor": { + "description": "Opaque cursor for the next page; absent on the last page.", + "schema": { + "type": "string" + } + }, + "X-Has-More": { + "description": "'true' when another page exists, 'false' otherwise.", + "schema": { + "type": "string", + "enum": [ + "true", + "false" + ] + } + } + }, "content": { "text/csv": { "schema": { @@ -558,6 +656,17 @@ "application/json": { "schema": { "type": "object", + "required": [ + "pair", + "network", + "interval", + "from", + "to", + "count", + "points", + "nextCursor", + "hasMore" + ], "properties": { "pair": { "type": "string" @@ -581,7 +690,8 @@ "format": "date-time" }, "count": { - "type": "integer" + "type": "integer", + "description": "Number of points in this page (not the total across pages)." }, "points": { "type": "array", @@ -600,6 +710,13 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true + }, + "hasMore": { + "type": "boolean" } } } @@ -622,10 +739,26 @@ "get": { "operationId": "listPairs", "summary": "List all active trading pairs", + "description": "Active trading pairs with their latest price metadata. Paginated with the shared keyset convention: ordered by pairKey ascending, default limit 50, maximum 200.", "security": [], "parameters": [ { "$ref": "#/components/parameters/NetworkQuery" + }, + { + "$ref": "#/components/parameters/CursorQuery" + }, + { + "name": "limit", + "in": "query", + "required": false, + "description": "Page size. Out-of-range values are clamped into [1, 200].", + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 200, + "default": 50 + } } ], "responses": { @@ -635,6 +768,11 @@ "application/json": { "schema": { "type": "object", + "required": [ + "pairs", + "nextCursor", + "hasMore" + ], "properties": { "pairs": { "type": "array", @@ -661,12 +799,22 @@ } } } + }, + "nextCursor": { + "type": "string", + "nullable": true + }, + "hasMore": { + "type": "boolean" } } } } } }, + "400": { + "description": "Invalid limit or cursor" + }, "429": { "$ref": "#/components/responses/TooManyRequests" } @@ -1208,18 +1356,13 @@ } }, { - "name": "cursor", - "in": "query", - "required": false, - "description": "Opaque cursor returned as `nextCursor` by the previous page.", - "schema": { - "type": "string" - } + "$ref": "#/components/parameters/CursorQuery" }, { "name": "limit", "in": "query", "required": false, + "description": "Page size. Out-of-range values are clamped into [1, 100].", "schema": { "type": "integer", "minimum": 1, @@ -1809,6 +1952,15 @@ ], "default": "testnet" } + }, + "CursorQuery": { + "name": "cursor", + "in": "query", + "required": false, + "description": "Opaque cursor returned as `nextCursor` by the previous page. Every list route uses the same keyset-cursor convention: pass it unchanged to fetch the next page; rows inserted while paging never shift a page boundary. An invalid cursor is rejected with 400.", + "schema": { + "type": "string" + } } }, "responses": { diff --git a/openapi.yaml b/openapi.yaml index 685c29ff..b771e99c 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -178,10 +178,24 @@ paths: get: operationId: getPools summary: Get AMM liquidity pool reserves and spot prices + description: >- + Latest snapshot per pool for one network. Paginated with the shared + keyset convention: ordered by pool_id ascending, default limit 50, + maximum 200. security: - x402: [] parameters: - $ref: '#/components/parameters/NetworkQuery' + - $ref: '#/components/parameters/CursorQuery' + - name: limit + in: query + required: false + description: Page size. Out-of-range values are clamped into [1, 200]. + schema: + type: integer + minimum: 1 + maximum: 200 + default: 50 responses: '200': description: List of pools @@ -189,6 +203,7 @@ paths: application/json: schema: type: object + required: [pools, nextCursor, hasMore] properties: pools: type: array @@ -212,6 +227,13 @@ paths: timestamp: type: string format: date-time + nextCursor: + type: string + nullable: true + hasMore: + type: boolean + '400': + description: Invalid limit or cursor '402': $ref: '#/components/responses/PaymentRequired' '429': @@ -221,6 +243,10 @@ paths: get: operationId: getCandles summary: Get OHLCV candle data + description: >- + OHLCV candles for a pair. Paginated with the shared keyset convention: + ordered by bucket start time ascending, default limit 500, + maximum 5000. security: - x402: [] parameters: @@ -253,6 +279,16 @@ paths: schema: type: string format: date-time + - $ref: '#/components/parameters/CursorQuery' + - name: limit + in: query + required: false + description: Page size. Out-of-range values are clamped into [1, 5000]. + schema: + type: integer + minimum: 1 + maximum: 5000 + default: 500 responses: '200': description: Candle data @@ -260,6 +296,7 @@ paths: application/json: schema: type: object + required: [pairKey, interval, from, to, candles, nextCursor, hasMore] properties: pairKey: type: string @@ -267,8 +304,10 @@ paths: type: string from: type: string + format: date-time to: type: string + format: date-time candles: type: array items: @@ -287,6 +326,11 @@ paths: type: number volume: type: number + nextCursor: + type: string + nullable: true + hasMore: + type: boolean '400': description: Invalid parameters '402': @@ -301,7 +345,11 @@ paths: description: >- Returns minute-resolution price snapshots for a pair over [from, to], optionally aggregated into 5m or 1h buckets. Snapshots are retained for - ~30 days. Gated by x402 (matches the /price prefix). + ~30 days. Paginated with the shared keyset convention: ordered by + bucket start time ascending, default limit 1000, maximum 10000 points. + With format=csv the pagination envelope travels in the X-Next-Cursor + and X-Has-More response headers; with format=json it is in the body. + Gated by x402 (matches the /price prefix). security: - x402: [] parameters: @@ -341,10 +389,35 @@ paths: type: string enum: [json, csv] default: json + - $ref: '#/components/parameters/CursorQuery' + - name: limit + in: query + required: false + description: >- + Page size (number of buckets). Out-of-range values are clamped + into [1, 10000]. + schema: + type: integer + minimum: 1 + maximum: 10000 + default: 1000 - $ref: '#/components/parameters/NetworkQuery' responses: '200': - description: Historical price snapshots + description: >- + Historical price snapshots. When more buckets exist beyond this + page, the response carries X-Next-Cursor (pass it back as + ?cursor=) and X-Has-More: true. + headers: + X-Next-Cursor: + description: Opaque cursor for the next page; absent on the last page. + schema: + type: string + X-Has-More: + description: "'true' when another page exists, 'false' otherwise." + schema: + type: string + enum: ['true', 'false'] content: text/csv: schema: @@ -353,6 +426,7 @@ paths: application/json: schema: type: object + required: [pair, network, interval, from, to, count, points, nextCursor, hasMore] properties: pair: type: string @@ -369,6 +443,7 @@ paths: format: date-time count: type: integer + description: Number of points in this page (not the total across pages). points: type: array items: @@ -381,6 +456,11 @@ paths: type: number volume: type: number + nextCursor: + type: string + nullable: true + hasMore: + type: boolean '400': description: Invalid parameters '402': @@ -392,9 +472,23 @@ paths: get: operationId: listPairs summary: List all active trading pairs + description: >- + Active trading pairs with their latest price metadata. Paginated with + the shared keyset convention: ordered by pairKey ascending, default + limit 50, maximum 200. security: [] parameters: - $ref: '#/components/parameters/NetworkQuery' + - $ref: '#/components/parameters/CursorQuery' + - name: limit + in: query + required: false + description: Page size. Out-of-range values are clamped into [1, 200]. + schema: + type: integer + minimum: 1 + maximum: 200 + default: 50 responses: '200': description: List of pairs @@ -402,6 +496,7 @@ paths: application/json: schema: type: object + required: [pairs, nextCursor, hasMore] properties: pairs: type: array @@ -421,6 +516,13 @@ paths: type: string format: date-time nullable: true + nextCursor: + type: string + nullable: true + hasMore: + type: boolean + '400': + description: Invalid limit or cursor '429': $ref: '#/components/responses/TooManyRequests' post: @@ -764,15 +866,11 @@ paths: type: string enum: [asc, desc] default: desc - - name: cursor - in: query - required: false - description: Opaque cursor returned as `nextCursor` by the previous page. - schema: - type: string + - $ref: '#/components/parameters/CursorQuery' - name: limit in: query required: false + description: Page size. Out-of-range values are clamped into [1, 100]. schema: type: integer minimum: 1 @@ -1166,6 +1264,17 @@ components: type: string enum: [testnet, mainnet] default: testnet + CursorQuery: + name: cursor + in: query + required: false + description: >- + Opaque cursor returned as `nextCursor` by the previous page. Every list + route uses the same keyset-cursor convention: pass it unchanged to + fetch the next page; rows inserted while paging never shift a page + boundary. An invalid cursor is rejected with 400. + schema: + type: string responses: PaymentRequired: diff --git a/src/__tests__/candles.test.ts b/src/__tests__/candles.test.ts index 343044fa..baf1e4c4 100644 --- a/src/__tests__/candles.test.ts +++ b/src/__tests__/candles.test.ts @@ -135,4 +135,54 @@ describe('GET /candles/:assetA/:assetB', () => { expect(new Date(params[2]).toISOString()).toBe(from) expect(new Date(params[3]).toISOString()).toBe(to) }) + + describe('cursor pagination', () => { + it('fetches limit+1 buckets and reports hasMore with a cursor', async () => { + const app = await buildApp() + mockQuery.mockResolvedValue({ + rows: [ + makeRow({ time: new Date('2025-01-01T00:00:00Z') }), + makeRow({ time: new Date('2025-01-01T01:00:00Z') }), + makeRow({ time: new Date('2025-01-01T02:00:00Z') }), + ], + }) + + 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) + // The cursor carries the last kept bucket's epoch seconds (tiebroken by pairKey). + expect(JSON.parse(Buffer.from(body.nextCursor, 'base64').toString())).toEqual([ + Math.floor(Date.parse('2025-01-01T01:00:00Z') / 1000), + 'USDC/XLM', + ]) + // The probe binds limit+1. + expect(mockQuery.mock.calls[0][1].at(-1)).toBe(3) + await app.close() + }) + + it('applies the cursor as a keyset condition on the bucket expression', async () => { + const app = await buildApp() + mockQuery.mockResolvedValue({ rows: [] }) + const cursor = Buffer.from(JSON.stringify([Math.floor(Date.parse('2025-01-01T01:00:00Z') / 1000), 'USDC/XLM'])).toString('base64') + + const res = await app.inject({ method: 'GET', url: `/candles/XLM/USDC?interval=1h&cursor=${cursor}` }) + + expect(res.statusCode).toBe(200) + const [sql, params] = mockQuery.mock.calls[0] + expect(sql).toMatch(/floor\(EXTRACT\(EPOCH FROM timestamp\) \/ \$1\) \* \$1 > \$5/) + expect(params[4]).toBe(Math.floor(Date.parse('2025-01-01T01:00:00Z') / 1000)) + await app.close() + }) + + it('rejects a malformed cursor with 400', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/candles/XLM/USDC?interval=1h&cursor=@@@' }) + expect(res.statusCode).toBe(400) + expect(res.json().error).toMatch(/cursor/) + await app.close() + }) + }) }) diff --git a/src/__tests__/history.test.ts b/src/__tests__/history.test.ts index 70992553..29e0706a 100644 --- a/src/__tests__/history.test.ts +++ b/src/__tests__/history.test.ts @@ -166,6 +166,107 @@ describe('GET /prices/history', () => { }) }) +describe('GET /prices/history cursor pagination', () => { + beforeEach(() => { + mockQuery.mockReset() + }) + + // Three 1m buckets; the mock honours the route's SQL contract — params are + // [interval, pair, from, to, network, (afterTs)?, limit + 1] — so a bucket + // inserted mid-paging lands exactly where a real insert would. + const makeTable = () => [ + { bucket: new Date('2025-01-01T00:00:00Z'), price: '1.0', volume: '1' }, + { bucket: new Date('2025-01-01T00:01:00Z'), price: '1.1', volume: '1' }, + { bucket: new Date('2025-01-01T00:02:00Z'), price: '1.2', volume: '1' }, + ] + + function useKeysetTable(table: ReturnType) { + mockQuery.mockImplementation(async (_sql: string, params: unknown[]) => { + const limitPlus1 = params[params.length - 1] as number + const afterTs = params.length === 7 ? (params[5] as Date) : null + let rows = [...table] + if (afterTs) rows = rows.filter(r => r.bucket > afterTs) + rows.sort((a, b) => a.bucket.getTime() - b.bucket.getTime()) + return { rows: rows.slice(0, limitPlus1) } + }) + } + + it('pages without duplicates or skips when a snapshot is inserted mid-page', async () => { + const app = await buildApp() + const table = makeTable() + useKeysetTable(table) + + const page1 = await app.inject({ method: 'GET', url: '/prices/history?pair=XLM/USDC&interval=1m&limit=2' }) + expect(page1.statusCode).toBe(200) + const body1 = page1.json() + expect(body1.points.map((p: { ts: string }) => p.ts)).toEqual([ + '2025-01-01T00:00:00.000Z', + '2025-01-01T00:01:00.000Z', + ]) + expect(body1.count).toBe(2) + expect(body1.hasMore).toBe(true) + expect(body1.nextCursor).toBeTruthy() + + // A snapshot lands inside page 1's range mid-paging. An offset page 2 + // would return 00:01 again and skip 00:02; the keyset cursor must not. + table.splice(1, 0, { bucket: new Date('2025-01-01T00:00:30Z'), price: '1.05', volume: '1' }) + + const page2 = await app.inject({ + method: 'GET', + url: `/prices/history?pair=XLM/USDC&interval=1m&limit=2&cursor=${encodeURIComponent(body1.nextCursor)}`, + }) + expect(page2.statusCode).toBe(200) + const body2 = page2.json() + expect(body2.points.map((p: { ts: string }) => p.ts)).toEqual(['2025-01-01T00:02:00.000Z']) + expect(body2.hasMore).toBe(false) + expect(body2.nextCursor).toBeNull() + }) + + it('applies the cursor as a keyset condition and binds limit+1', async () => { + const app = await buildApp() + mockQuery.mockResolvedValue({ rows: [] }) + const cursorTs = '2025-01-01T00:01:00.000Z' + const cursor = Buffer.from(JSON.stringify([cursorTs, 'XLM/USDC'])).toString('base64') + + const res = await app.inject({ method: 'GET', url: `/prices/history?pair=XLM/USDC&cursor=${cursor}&limit=50` }) + + expect(res.statusCode).toBe(200) + const [sql, params] = mockQuery.mock.calls[0] + expect(sql).toMatch(/AND ts > \$6/) + expect(sql).toMatch(/LIMIT \$7/) + expect(params[5]).toEqual(new Date(cursorTs)) + expect(params[6]).toBe(51) + }) + + it('rejects a malformed cursor with 400', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/prices/history?pair=XLM/USDC&cursor=@@@' }) + expect(res.statusCode).toBe(400) + expect(res.json().error).toMatch(/cursor/) + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('carries the pagination envelope in headers for format=csv', async () => { + const app = await buildApp() + useKeysetTable(makeTable()) + + const page1 = await app.inject({ method: 'GET', url: '/prices/history?pair=XLM/USDC&interval=1m&limit=2&format=csv' }) + expect(page1.statusCode).toBe(200) + expect(page1.headers['x-has-more']).toBe('true') + expect(page1.headers['x-next-cursor']).toBeTruthy() + // The CSV body stays the pure ts,price,volume contract. + expect(page1.body).toBe('ts,price,volume\n2025-01-01T00:00:00.000Z,1,1\n2025-01-01T00:01:00.000Z,1.1,1\n') + + const page2 = await app.inject({ + method: 'GET', + url: `/prices/history?pair=XLM/USDC&interval=1m&limit=2&format=csv&cursor=${encodeURIComponent(page1.headers['x-next-cursor'] as string)}`, + }) + expect(page2.statusCode).toBe(200) + expect(page2.headers['x-has-more']).toBe('false') + expect(page2.headers['x-next-cursor']).toBeUndefined() + }) +}) + describe('GET /prices/history network selection', () => { beforeEach(() => { mockQuery.mockReset() diff --git a/src/__tests__/pagination.test.ts b/src/__tests__/pagination.test.ts new file mode 100644 index 00000000..101c5918 --- /dev/null +++ b/src/__tests__/pagination.test.ts @@ -0,0 +1,253 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify from 'fastify' + +const { mockQuery, mockGetActivePairs } = vi.hoisted(() => ({ + mockQuery: vi.fn(), + mockGetActivePairs: vi.fn(), +})) + +vi.mock('../db', () => ({ + prisma: {}, + pgPool: { query: mockQuery }, +})) + +vi.mock('../redis', () => ({ + getCachedPrice: vi.fn(), + setCachedPrice: vi.fn(), +})) + +vi.mock('../aggregator/bestRoute', () => ({ + getBestRoute: vi.fn(), +})) + +vi.mock('../pairsRegistry', () => ({ + getActivePairs: mockGetActivePairs, + hasPair: vi.fn(), + registerPair: vi.fn(), + persistPair: vi.fn(), + parseAssetStr: vi.fn(), + makePairKey: vi.fn(), +})) + +import { + parseLimit, + encodeCursor, + decodeCursor, + slicePage, + SCREENER_DEFAULT_LIMIT, + SCREENER_MAX_LIMIT, + POOLS_DEFAULT_LIMIT, + POOLS_MAX_LIMIT, +} from '../pagination' +import { registerRESTRoutes } from '../api/rest' +import { registerPairsRoutes } from '../routes/pairs' + +describe('parseLimit', () => { + it('returns the default when the parameter is absent', () => { + expect(parseLimit(undefined, 20, 100)).toEqual({ limit: 20 }) + }) + + it('clamps out-of-range values into [1, max]', () => { + expect(parseLimit('0', 20, 100)).toEqual({ limit: 1 }) + expect(parseLimit('-5', 20, 100)).toEqual({ limit: 1 }) + expect(parseLimit('1000', 20, 100)).toEqual({ limit: 100 }) + }) + + it('accepts a limit inside the range', () => { + expect(parseLimit('42', 20, 100)).toEqual({ limit: 42 }) + expect(parseLimit('1', 50, 200)).toEqual({ limit: 1 }) + expect(parseLimit('200', 50, 200)).toEqual({ limit: 200 }) + }) + + it('rejects a non-integer limit with the screener error message', () => { + expect(parseLimit('abc', 20, 100)).toEqual({ error: 'limit must be a valid integer' }) + expect(parseLimit('', 20, 100)).toEqual({ error: 'limit must be a valid integer' }) + }) +}) + +describe('encodeCursor / decodeCursor', () => { + it('round-trips a [value, id] payload through base64 JSON', () => { + const cursor = encodeCursor(20, 'B/X') + expect(decodeCursor(cursor)).toEqual({ value: 20, id: 'B/X' }) + }) + + it('is byte-identical to screener\'s historical cursor encoding', () => { + expect(encodeCursor(20, 'B/X')).toBe(Buffer.from(JSON.stringify([20, 'B/X'])).toString('base64')) + }) + + it('returns null for malformed cursors so the route answers 400', () => { + expect(decodeCursor('not-base64-json')).toBeNull() + expect(decodeCursor(Buffer.from('null').toString('base64'))).toBeNull() + expect(decodeCursor(Buffer.from('{"a":1}').toString('base64'))).toBeNull() + expect(decodeCursor(Buffer.from('123').toString('base64'))).toBeNull() + }) + + it("returns 'skip' for a well-formed but degenerate cursor", () => { + expect(decodeCursor(Buffer.from('[null, "B/X"]').toString('base64'))).toBe('skip') + expect(decodeCursor(Buffer.from('[20, ""]').toString('base64'))).toBe('skip') + expect(decodeCursor(Buffer.from('[1]').toString('base64'))).toBe('skip') + }) +}) + +describe('slicePage', () => { + it('reports hasMore and trims the probe row', () => { + expect(slicePage([1, 2, 3], 2)).toEqual({ data: [1, 2], hasMore: true }) + }) + + it('keeps every row when the probe found no extra row', () => { + expect(slicePage([1, 2], 2)).toEqual({ data: [1, 2], hasMore: false }) + expect(slicePage([], 2)).toEqual({ data: [], hasMore: false }) + }) +}) + +/** + * The mid-page-insert acceptance tests below run against the real route + * handlers with a mutable fake table standing in for Postgres. The mock + * honours the same contract the SQL does — keyset filter, ascending order, + * LIMIT — so a row inserted between page fetches lands exactly where a real + * insert would, and an offset implementation would demonstrably return a + * duplicate. + */ + +async function buildPoolsApp() { + const app = Fastify({ logger: false }) + await registerRESTRoutes(app) + await app.ready() + return app +} + +describe('GET /pools cursor pagination', () => { + let table: { pool_id: string }[] + + beforeEach(() => { + table = [{ pool_id: 'A' }, { pool_id: 'B' }, { pool_id: 'C' }] + // Contract of the route's SQL: params are [network, (cursor pool_id)?, + // limit + 1]; rows come back ordered by pool_id with the keyset applied. + mockQuery.mockImplementation(async (_sql: string, params: unknown[]) => { + const limitPlus1 = params[params.length - 1] as number + const after = params.length === 3 ? (params[1] as string) : null + let rows = table.map(r => ({ ...r, asset_a: 'X', asset_b: 'Y', reserve_a: 1, reserve_b: 2, spot_price: 2, fee_bp: 30, timestamp: new Date('2025-01-01T00:00:00Z') })) + if (after) rows = rows.filter(r => r.pool_id > after) + rows.sort((a, b) => (a.pool_id < b.pool_id ? -1 : 1)) + return { rows: rows.slice(0, limitPlus1) } + }) + }) + + afterEach(() => { + vi.clearAllMocks() + }) + + it('pages without duplicates or skips when a pool is inserted mid-page', async () => { + const app = await buildPoolsApp() + + // Page 1: default limit 50 would swallow everything, so ask for 2. + const page1 = await app.inject({ method: 'GET', url: '/pools?limit=2' }) + expect(page1.statusCode).toBe(200) + const body1 = page1.json() + expect(body1.pools.map((p: { pool_id: string }) => p.pool_id)).toEqual(['A', 'B']) + expect(body1.hasMore).toBe(true) + expect(body1.nextCursor).toBeTruthy() + + // A pool sorts into the range page 1 already covered. An OFFSET-based + // page 2 would now return B again (and skip C); the keyset cursor must not. + table.push({ pool_id: 'A2' }) + + const page2 = await app.inject({ + method: 'GET', + url: `/pools?limit=2&cursor=${encodeURIComponent(body1.nextCursor)}`, + }) + expect(page2.statusCode).toBe(200) + const body2 = page2.json() + expect(body2.pools.map((p: { pool_id: string }) => p.pool_id)).toEqual(['C']) + expect(body2.hasMore).toBe(false) + expect(body2.nextCursor).toBeNull() + await app.close() + }) + + it('returns the first page and hasMore=false when everything fits', async () => { + const app = await buildPoolsApp() + const res = await app.inject({ method: 'GET', url: '/pools' }) + const body = res.json() + expect(body.pools).toHaveLength(3) + expect(body.hasMore).toBe(false) + expect(body.nextCursor).toBeNull() + await app.close() + }) + + it('rejects a malformed cursor and an invalid limit with 400', async () => { + const app = await buildPoolsApp() + const badCursor = await app.inject({ method: 'GET', url: '/pools?cursor=@@@' }) + expect(badCursor.statusCode).toBe(400) + const badLimit = await app.inject({ method: 'GET', url: '/pools?limit=abc' }) + expect(badLimit.statusCode).toBe(400) + await app.close() + }) + + it('uses the documented default and maximum limits', () => { + expect(SCREENER_DEFAULT_LIMIT).toBe(20) + expect(SCREENER_MAX_LIMIT).toBe(100) + expect(POOLS_DEFAULT_LIMIT).toBe(50) + expect(POOLS_MAX_LIMIT).toBe(200) + }) +}) + +describe('GET /pairs cursor pagination', () => { + const pair = (pairKey: string) => ({ + pairKey, + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GTEST' }, + }) + + let watched: string[] + + beforeEach(() => { + watched = ['B/C', 'A/B', 'C/D'] + mockGetActivePairs.mockImplementation(() => watched.map(pair)) + mockQuery.mockResolvedValue({ rows: [] }) + }) + + afterEach(() => { + vi.clearAllMocks() + }) + + async function buildPairsApp() { + const app = Fastify({ logger: false }) + await registerPairsRoutes(app) + await app.ready() + return app + } + + it('pages without duplicates or skips when a pair is registered mid-page', async () => { + const app = await buildPairsApp() + + const page1 = await app.inject({ method: 'GET', url: '/pairs?limit=2' }) + expect(page1.statusCode).toBe(200) + const body1 = page1.json() + // Sorted by pairKey regardless of registry order. + expect(body1.pairs.map((p: { pairKey: string }) => p.pairKey)).toEqual(['A/B', 'B/C']) + expect(body1.hasMore).toBe(true) + + // Insert a pair that sorts into page 1's range mid-paging. + watched.push('A2/X') + + const page2 = await app.inject({ + method: 'GET', + url: `/pairs?limit=2&cursor=${encodeURIComponent(body1.nextCursor)}`, + }) + expect(page2.statusCode).toBe(200) + const body2 = page2.json() + // Keyset: everything strictly after B/C — no duplicate B/C, no skipped C/D. + // (A2/X sorts before B/C, so it belongs to page 1 and is not repeated.) + expect(body2.pairs.map((p: { pairKey: string }) => p.pairKey)).toEqual(['C/D']) + expect(body2.hasMore).toBe(false) + expect(body2.nextCursor).toBeNull() + await app.close() + }) + + it('rejects a malformed cursor with 400', async () => { + const app = await buildPairsApp() + const res = await app.inject({ method: 'GET', url: '/pairs?cursor=@@@' }) + expect(res.statusCode).toBe(400) + await app.close() + }) +}) diff --git a/src/__tests__/pairs.test.ts b/src/__tests__/pairs.test.ts index 4ec2acbc..c5633d61 100644 --- a/src/__tests__/pairs.test.ts +++ b/src/__tests__/pairs.test.ts @@ -222,8 +222,11 @@ describe('GET /pairs', () => { const body = (await app.inject({ method: 'GET', url: '/pairs' })).json() 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 }) + // Sorted by pairKey ascending, so NEW/Y precedes QUIET/X. + 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() }) + expect(body.nextCursor).toBeNull() + expect(body.hasMore).toBe(false) }) }) }) diff --git a/src/__tests__/pools.test.ts b/src/__tests__/pools.test.ts index 1531988f..0cab37e0 100644 --- a/src/__tests__/pools.test.ts +++ b/src/__tests__/pools.test.ts @@ -46,14 +46,15 @@ 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]) + // Last bind is the limit+1 probe (default 50 + 1). + 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() }) @@ -70,6 +71,8 @@ describe('GET /pools', () => { expect(res.json().pools).toEqual([ { pool_id: 'C1', asset_a: 'XLM', asset_b: 'USDC', reserve_a: 10, reserve_b: 20, spot_price: 2, fee_bp: 30, timestamp: old.toISOString() }, ]) + expect(res.json().nextCursor).toBeNull() + expect(res.json().hasMore).toBe(false) await app.close() }) }) diff --git a/src/api/history.ts b/src/api/history.ts index 1f8a7590..2210c212 100644 --- a/src/api/history.ts +++ b/src/api/history.ts @@ -1,6 +1,13 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' +import { + HISTORY_DEFAULT_LIMIT, + decodeCursor, + encodeCursor, + parseLimit, + slicePage, +} from '../pagination' import '../middleware/network' // declares req.network on the FastifyRequest type /** Supported aggregation intervals → bucket width in seconds. */ @@ -71,14 +78,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 +112,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,8 +142,16 @@ 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) { @@ -139,6 +172,26 @@ export async function registerHistoryRoutes(app: FastifyInstance) { }) } + const parsedLimit = parseLimit(req.query.limit, HISTORY_DEFAULT_LIMIT, MAX_HISTORY_POINTS) + if ('error' in parsedLimit) { + return reply.status(400).send({ error: 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({ error: 'Invalid cursor format' }) + } + if (decoded !== 'skip') { + afterTs = new Date(String(decoded.value)) + if (isNaN(afterTs.getTime())) { + return reply.status(400).send({ error: '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() @@ -161,12 +214,22 @@ export async function registerHistoryRoutes(app: FastifyInstance) { } 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 +238,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..84517c28 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -16,6 +16,14 @@ import { installResponseValidation, } from './schemas' import { getDepth } from '../pricing/depth' +import { + POOLS_DEFAULT_LIMIT, + POOLS_MAX_LIMIT, + decodeCursor, + encodeCursor, + parseLimit, + slicePage, +} from '../pagination' function makePairKey(a: string, b: string): string { return [a, b].sort().join('/') @@ -204,7 +212,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { } ) - // GET /pools + // GET /pools?cursor=&limit= // // Latest snapshot per pool for one network, without reading history. The pool // ids are discovered with a recursive "skip scan" over the @@ -212,32 +220,69 @@ 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({ error: 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({ error: '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<{ diff --git a/src/api/schemas.ts b/src/api/schemas.ts index 141e7aee..f18e1b40 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/pagination.ts b/src/pagination.ts new file mode 100644 index 00000000..648b2940 --- /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 Iterable + } 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/candles.ts b/src/routes/candles.ts index c83c56c5..442ac2b4 100644 --- a/src/routes/candles.ts +++ b/src/routes/candles.ts @@ -1,5 +1,13 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' +import { + CANDLES_DEFAULT_LIMIT, + CANDLES_MAX_LIMIT, + decodeCursor, + encodeCursor, + parseLimit, + slicePage, +} from '../pagination' const INTERVAL_SECONDS: Record = { '1m': 60, @@ -11,9 +19,17 @@ 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' @@ -25,6 +41,12 @@ export async function registerCandleRoutes(app: FastifyInstance) { }) } + const parsedLimit = parseLimit(req.query.limit, CANDLES_DEFAULT_LIMIT, CANDLES_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send({ error: parsedLimit.error }) + } + const { limit } = parsedLimit + const pairKey = [assetA, assetB].sort().join('/') const fromRaw = req.query.from @@ -39,6 +61,20 @@ export async function registerCandleRoutes(app: FastifyInstance) { return reply.status(400).send({ error: '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({ error: '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 +86,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 +112,8 @@ export async function registerCandleRoutes(app: FastifyInstance) { close: Number(r.close), volume: Number(r.volume), })), + nextCursor, + hasMore, } }) } diff --git a/src/routes/pairs.ts b/src/routes/pairs.ts index f2208e04..caae13dd 100644 --- a/src/routes/pairs.ts +++ b/src/routes/pairs.ts @@ -2,10 +2,18 @@ import type { FastifyInstance } from 'fastify' import { getActivePairs, parseAssetStr, makePairKey, registerPair, persistPair, hasPair } from '../pairsRegistry' import { pgPool } from '../db' import { activeNetwork } from '../config' +import { + PAIRS_DEFAULT_LIMIT, + PAIRS_MAX_LIMIT, + decodeCursor, + encodeCursor, + parseLimit, + slicePage, +} from '../pagination' import '../middleware/network' // declares req.network on the FastifyRequest type export async function registerPairsRoutes(app: FastifyInstance) { - // GET /pairs — list all active pairs with latest price metadata + // GET /pairs?cursor=&limit= — list active pairs with latest price metadata // // One index probe per watched pair on (network, pair_key, timestamp DESC): // `ORDER BY timestamp DESC LIMIT 1` reads a single index entry, so the cost is @@ -16,10 +24,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({ error: 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({ error: '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 +71,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 +82,13 @@ 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 diff --git a/src/routes/screener.ts b/src/routes/screener.ts index 783059d4..14414a94 100644 --- a/src/routes/screener.ts +++ b/src/routes/screener.ts @@ -1,6 +1,14 @@ import type { FastifyInstance } from 'fastify' import { pgPool } from '../db' import { activeNetwork } from '../config' +import { + SCREENER_DEFAULT_LIMIT, + SCREENER_MAX_LIMIT, + decodeCursor, + encodeCursor, + parseLimit, + slicePage, +} from '../pagination' import '../middleware/network' // declares req.network on the FastifyRequest type // Lens has no circulating-supply data, so a market cap cannot be computed. It @@ -29,10 +37,11 @@ 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({ error: parsedLimit.error }) } + const limit = parsedLimit.limit if (q.market_cap !== undefined || q.sortBy === 'market_cap') { return reply.status(400).send({ error: MARKET_CAP_UNSUPPORTED }) @@ -100,18 +109,15 @@ 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) { + const decoded = decodeCursor(q.cursor) + if (decoded === null) { return reply.status(400).send({ error: 'Invalid cursor format' }) } + if (decoded !== 'skip') { + const op = order === 'desc' ? '<' : '>' + conditions.push(`(${sortBy}, pair_key) ${op} ($${paramIdx++}, $${paramIdx++})`) + params.push(decoded.value, decoded.id) + } } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '' @@ -169,17 +175,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 {