From 97cd662cc2c1464f1f926c9f3fa2ae88b9f0b7ad Mon Sep 17 00:00:00 2001 From: Bathoul Mohammed Date: Sun, 4 Oct 2026 21:03:11 +0100 Subject: [PATCH 1/2] feat(exports): Zod validation, maxRows cap, and truncation signalling --- src/routes/exports.ts | 168 ++++++++++++++++++++++++++++++------------ 1 file changed, 122 insertions(+), 46 deletions(-) diff --git a/src/routes/exports.ts b/src/routes/exports.ts index a09ac930a..bbf5e992b 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -4,23 +4,44 @@ import { prisma, toDisplayAmount } from "../db"; import os from "os"; import path from "path"; import fs from "fs"; +import { z } from "zod"; import { requestNetwork } from "../middleware/network"; +import { parseOr400 } from "../openapi/validation"; import type { Network } from "../network"; // How many rows we fetch per DB round-trip. Keeps memory flat. const BATCH_SIZE = 500; -// ── Shared: parse query params into a Prisma where clause ──────────────────── -function buildWhere(query: Record, network: Network) { - const { - address, - contractId, - fromLedger, - toLedger, - fromDate, - toDate, - eventType, - } = query; +// Hard cap on rows returned per export request. Callers that need more should +// paginate with fromLedger/toLedger or apply a tighter date/address filter. +// The value is deliberately large enough to be useful but small enough to keep +// memory and response time predictable under load. +const DEFAULT_MAX_ROWS = 50_000; +const ABSOLUTE_MAX_ROWS = 500_000; + +// ── Query-param schema ──────────────────────────────────────────────────────── +// Exported so the OpenAPI build can reference it. +export const exportQuerySchema = z.object({ + address: z.string().trim().optional(), + contractId: z.string().trim().optional(), + fromLedger: z.coerce.number().int().min(0).optional(), + toLedger: z.coerce.number().int().min(0).optional(), + fromDate: z.string().datetime({ offset: true, message: "Invalid date — expected ISO 8601 (e.g. 2025-01-01T00:00:00Z)" }) + .transform((v) => new Date(v)).optional(), + toDate: z.string().datetime({ offset: true, message: "Invalid date — expected ISO 8601 (e.g. 2025-01-01T00:00:00Z)" }) + .transform((v) => new Date(v)).optional(), + eventType: z.string().trim().optional(), + maxRows: z.coerce.number().int() + .min(1, "maxRows must be >= 1") + .max(ABSOLUTE_MAX_ROWS, `maxRows must be <= ${ABSOLUTE_MAX_ROWS}`) + .optional(), +}); + +type ExportQuery = z.infer; + +// ── Shared: build a Prisma where clause from validated params ───────────────── +function buildWhere(params: ExportQuery, network: Network) { + const { address, contractId, fromLedger, toLedger, fromDate, toDate, eventType } = params; // Network first: an export must not leak rows from a chain the caller did // not ask for, and every filter below narrows within it. @@ -31,50 +52,66 @@ function buildWhere(query: Record, network: Network) { } if (contractId) where.contractId = contractId; if (eventType) { - const types = String(eventType).split(",").map((s) => s.trim()).filter(Boolean); + const types = eventType.split(",").map((s) => s.trim()).filter(Boolean); if (types.length) where.eventType = { in: types }; } const ledgerRange: Record = {}; - if (fromLedger) ledgerRange.gte = parseInt(String(fromLedger), 10); - if (toLedger) ledgerRange.lte = parseInt(String(toLedger), 10); + if (fromLedger !== undefined) ledgerRange.gte = fromLedger; + if (toLedger !== undefined) ledgerRange.lte = toLedger; if (Object.keys(ledgerRange).length) where.ledger = ledgerRange; const dateRange: Record = {}; - if (fromDate) dateRange.gte = new Date(String(fromDate)); - if (toDate) dateRange.lte = new Date(String(toDate)); + if (fromDate) dateRange.gte = fromDate; + if (toDate) dateRange.lte = toDate; if (Object.keys(dateRange).length) where.ledgerClosedAt = dateRange; return where; } // ── Shared: async generator that yields rows in batches via cursor ──────────── -async function* streamTransfers(where: Record) { +// Stops once `limit` rows have been yielded so callers never pull more than +// they asked for regardless of DB size. +async function* streamTransfers(where: Record, limit: number) { let lastId: number | undefined = undefined; + let yielded = 0; - while (true) { - const rows: Awaited> = await prisma.tokenTransfer.findMany({ - where, - orderBy: { id: "asc" }, - take: BATCH_SIZE, - ...(lastId !== undefined ? { cursor: { id: lastId }, skip: 1 } : {}), - }); + while (yielded < limit) { + const take = Math.min(BATCH_SIZE, limit - yielded); + + const rows: Awaited> = + await prisma.tokenTransfer.findMany({ + where, + orderBy: { id: "asc" }, + take, + ...(lastId !== undefined ? { cursor: { id: lastId }, skip: 1 } : {}), + }); if (rows.length === 0) break; for (const row of rows) { yield row; + yielded++; } - if (rows.length < BATCH_SIZE) break; + if (rows.length < take) break; lastId = rows[rows.length - 1].id; } } -// ── CSV endpoint ───────────────────────────────────────────────────────────── +// ── CSV endpoint ────────────────────────────────────────────────────────────── async function handleCsvExport(req: Request, res: Response, next: NextFunction) { try { - const where = buildWhere(req.query as Record, requestNetwork(req)); + const parsed = parseOr400(exportQuerySchema, req.query, res); + if (!parsed) return; + + const effectiveMax = parsed.maxRows ?? DEFAULT_MAX_ROWS; + const where = buildWhere(parsed, requestNetwork(req)); + + // Fetch one row beyond the cap so we can tell the caller whether the result + // was truncated without a separate COUNT query. + let rowCount = 0; + let truncated = false; res.setHeader("Content-Type", "text/csv"); res.setHeader("Content-Disposition", "attachment; filename=\"transfers.csv\""); @@ -83,22 +120,40 @@ async function handleCsvExport(req: Request, res: Response, next: NextFunction) const csvStream = csvFormat({ headers: true }); csvStream.pipe(res); - for await (const row of streamTransfers(where)) { + for await (const row of streamTransfers(where, effectiveMax + 1)) { + if (rowCount === effectiveMax) { + // We fetched one extra row — result is truncated, don't write this row. + truncated = true; + break; + } + csvStream.write({ - id: row.id, - contractId: row.contractId, - eventType: row.eventType, - fromAddress: row.fromAddress ?? "", - toAddress: row.toAddress ?? "", - amount: row.amount, - displayAmount: toDisplayAmount(row.amount), - ledger: row.ledger, - ledgerClosedAt: row.ledgerClosedAt.toISOString(), - txHash: row.txHash, - eventId: row.eventId, - isSac: row.isSac ?? false, - createdAt: row.createdAt.toISOString(), + id: row.id, + contractId: row.contractId, + eventType: row.eventType, + fromAddress: row.fromAddress ?? "", + toAddress: row.toAddress ?? "", + amount: row.amount, + displayAmount: toDisplayAmount(row.amount), + ledger: row.ledger, + ledgerClosedAt: row.ledgerClosedAt.toISOString(), + txHash: row.txHash, + eventId: row.eventId, + isSac: row.isSac ?? false, + createdAt: row.createdAt.toISOString(), }); + + rowCount++; + } + + // Signal truncation in a trailer header. HTTP/1.1 trailing headers require + // chunked encoding (which we've already set) and the client to opt in; as a + // belt-and-suspenders fallback we also set it as a regular response header + // before the body starts — Express buffers headers until the first write so + // this arrives before any CSV bytes. + if (truncated) { + res.setHeader("X-Truncated", "true"); + res.setHeader("X-Row-Limit", String(effectiveMax)); } csvStream.end(); @@ -107,16 +162,23 @@ async function handleCsvExport(req: Request, res: Response, next: NextFunction) } } -// ── Parquet endpoint ───────────────────────────────────────────────────────── +// ── Parquet endpoint ────────────────────────────────────────────────────────── async function handleParquetExport(req: Request, res: Response, next: NextFunction) { // parquetjs-lite is a CommonJS module — require() avoids ESM interop issues // eslint-disable-next-line @typescript-eslint/no-var-requires const parquet = require("parquetjs-lite"); - const tmpFile = path.join(os.tmpdir(), `transfers-${Date.now()}-${Math.random().toString(36).slice(2)}.parquet`); + const tmpFile = path.join( + os.tmpdir(), + `transfers-${Date.now()}-${Math.random().toString(36).slice(2)}.parquet`, + ); try { - const where = buildWhere(req.query as Record, requestNetwork(req)); + const parsed = parseOr400(exportQuerySchema, req.query, res); + if (!parsed) return; + + const effectiveMax = parsed.maxRows ?? DEFAULT_MAX_ROWS; + const where = buildWhere(parsed, requestNetwork(req)); const schema = new parquet.ParquetSchema({ id: { type: "INT64" }, @@ -136,7 +198,15 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti const writer = await parquet.ParquetWriter.openFile(schema, tmpFile); - for await (const row of streamTransfers(where)) { + let rowCount = 0; + let truncated = false; + + for await (const row of streamTransfers(where, effectiveMax + 1)) { + if (rowCount === effectiveMax) { + truncated = true; + break; + } + await writer.appendRow({ id: row.id, contractId: row.contractId, @@ -152,12 +222,18 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti isSac: row.isSac ?? null, createdAt: row.createdAt.toISOString(), }); + + rowCount++; } await writer.close(); res.setHeader("Content-Type", "application/octet-stream"); res.setHeader("Content-Disposition", "attachment; filename=\"transfers.parquet\""); + if (truncated) { + res.setHeader("X-Truncated", "true"); + res.setHeader("X-Row-Limit", String(effectiveMax)); + } const fileStream = fs.createReadStream(tmpFile); fileStream.pipe(res); @@ -172,7 +248,7 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti } } -// ── Router ─────────────────────────────────────────────────────────────────── +// ── Router ──────────────────────────────────────────────────────────────────── export function createExportsRouter(): Router { const router = Router(); router.get("/transfers.csv", handleCsvExport); From da6911ea0912c803985d23154b7b8ddeafa98feb Mon Sep 17 00:00:00 2001 From: Miracle656 Date: Tue, 6 Oct 2026 11:43:28 +0100 Subject: [PATCH 2/2] fix(exports): decide truncation before the body starts, and test the cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The truncation signal crashed the response on the one path it existed for. `res.setHeader("X-Truncated", ...)` ran after `csvStream.write` had already flushed the response head, so Node threw ERR_HTTP_HEADERS_SENT, which skipped `csvStream.end()` and left the client holding a CSV body that never terminated. Only ever on a truncated export. Reproduced against this repo's own @fast-csv/format, and the new test fails on the previous commit with exactly that error code. Truncation is now decided before any bytes are written, by one indexed `skip: max, take: 1` query that both handlers share — cheaper than a COUNT because it stops at the first row past the cap, and it removes the `rowCount` / `truncated` bookkeeping and the `effectiveMax + 1` fetch from both. Parquet did not have the bug, because it writes its file before setting headers, but it goes through the same helper so the two endpoints cannot answer differently about the same query. The cap is env-configurable as #185 asked: `EXPORT_MAX_ROWS`, clamped to ABSOLUTE_MAX_ROWS so a mistyped value cannot raise it, and falling back rather than becoming NaN, which would make every comparison against it false. Tests, also as #185 asked: `?fromLedger=abc` is a 400 and never reaches the database, an export over a seeded table stops at the cap and says so, and — the case worth having — a result that ends exactly at the cap does NOT claim truncation. The seeded mock honours `skip` and `take`, because one that ignored `skip` would answer "truncated" for every request and the suite would pass for the wrong reason. Documented the third criterion: the endpoints are capped rather than rejected, what the headers mean, and how to narrow or raise the cap. Claude-Session: https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB --- README.md | 9 ++ src/__tests__/routes/exports.test.ts | 149 +++++++++++++++++++++++++++ src/routes/exports.ts | 84 +++++++++------ 3 files changed, 208 insertions(+), 34 deletions(-) create mode 100644 src/__tests__/routes/exports.test.ts diff --git a/README.md b/README.md index e815374db..0959a642a 100644 --- a/README.md +++ b/README.md @@ -411,6 +411,14 @@ Base URL: `http://localhost:3000` Offramp order lookups need a bearer token; see [docs/offramp-orders.md](docs/offramp-orders.md). +`GET /transfers.csv` and `/transfers.parquet` need no filter, and are **capped +rather than rejected**: a request with nothing narrowing it returns the first +`maxRows` rows instead of the whole table. When rows were left behind, the +response says so in `X-Truncated: true` and `X-Row-Limit: `, set before the +body starts — so a client can tell a complete export from a partial one without +counting. Narrow with `address`, `contractId`, `fromLedger`/`toLedger` or +`fromDate`/`toDate`, or raise `maxRows` up to 500 000. + ### Selecting a network Wraith stores testnet and mainnet rows in the same tables, discriminated by a @@ -625,6 +633,7 @@ curl "http://localhost:3000/transfers/tx/abcdef1234567890..." | `RETENTION_DAYS` | `30` | Delete transfers older than N days (keeps DB within free-tier limits) | | `NETWORKS` | *(`STELLAR_NETWORK`)* | Comma-separated networks to index in one process, e.g. `testnet,mainnet`. Also the set the API's `?network=` selector accepts. | | `PORT` | `3000` | REST API port | +| `EXPORT_MAX_ROWS` | `50000` | Default row cap for `/transfers.csv` and `/transfers.parquet`. Clamped to 500 000; a caller may ask for less with `?maxRows=`. | ### RPC URL Resolution diff --git a/src/__tests__/routes/exports.test.ts b/src/__tests__/routes/exports.test.ts new file mode 100644 index 000000000..9a8ee9bbb --- /dev/null +++ b/src/__tests__/routes/exports.test.ts @@ -0,0 +1,149 @@ +import request from "supertest"; + +// ── Module mocks must be declared before any imports that use them ────────── +jest.mock("../../db", () => ({ + queryTransfers: jest.fn(), + queryAllTransfers: jest.fn(), + queryByTxHash: jest.fn(), + querySummary: jest.fn(), + getLastIndexedLedger: jest.fn(), + toDisplayAmount: jest.fn((amount: string) => amount), + prisma: { + $queryRaw: jest.fn(), + tokenMetadata: { findMany: jest.fn() }, + tokenTransfer: { findMany: jest.fn() }, + }, +})); + +jest.mock("../../rpc", () => ({ getLatestLedger: jest.fn() })); + +jest.mock("../../indexer", () => ({ + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), + getIndexerStats: jest + .fn() + .mockReturnValue({ startedAt: "2024-01-01T00:00:00.000Z", uptimeSeconds: 0, totalIndexed: 0 }), +})); + +import { createApp } from "../../api"; +import { prisma } from "../../db"; + +const mockFindMany = prisma.tokenTransfer.findMany as jest.MockedFunction< + typeof prisma.tokenTransfer.findMany +>; + +const CONTRACT = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD2KM"; +const ALICE = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF"; + +function row(id: number) { + return { + id, + contractId: CONTRACT, + eventType: "transfer", + fromAddress: ALICE, + toAddress: ALICE, + amount: "100", + ledger: 1000 + id, + ledgerClosedAt: new Date("2026-01-01T00:00:00Z"), + txHash: `hash${id}`, + eventId: `event${id}`, + isSac: false, + createdAt: new Date("2026-01-01T00:00:00Z"), + }; +} + +/** + * Serve `total` rows, honouring `skip`/`take` the way Prisma does. + * + * The cap probe is a `skip: max, take: 1` query against the same filter, so a + * mock that ignores `skip` would answer "truncated" for every request and the + * tests would pass for the wrong reason. + */ +function seed(total: number) { + mockFindMany.mockImplementation((async (args: { + skip?: number; + take?: number; + cursor?: { id: number }; + }) => { + const all = Array.from({ length: total }, (_, i) => row(i + 1)); + const after = args.cursor ? all.findIndex((r) => r.id === args.cursor!.id) + 1 : 0; + const from = after + (args.skip ?? 0); + return all.slice(from, from + (args.take ?? all.length)); + }) as never); +} + +beforeEach(() => { + jest.clearAllMocks(); +}); + +/** + * Bounding and validating the exports (#185). + * + * The cap is the point of the feature, but the header that announces it is + * where the danger was: setting a response header after the CSV body has begun + * throws ERR_HTTP_HEADERS_SENT, which skips `csvStream.end()` and leaves the + * client holding a response that never terminates — and it only happens on the + * truncation path, the one case the signalling exists for. + */ +describe("GET /transfers.csv", () => { + it("rejects a non-numeric fromLedger with 400, not 500", async () => { + const res = await request(createApp()).get("/transfers.csv?fromLedger=abc"); + + expect(res.status).toBe(400); + expect(res.body.error).toBeTruthy(); + // Nothing was queried: validation happens before the database is touched. + expect(mockFindMany).not.toHaveBeenCalled(); + }); + + it("rejects an unparseable date with 400", async () => { + const res = await request(createApp()).get("/transfers.csv?fromDate=not-a-date"); + expect(res.status).toBe(400); + }); + + it("rejects maxRows above the absolute ceiling", async () => { + const res = await request(createApp()).get("/transfers.csv?maxRows=999999999"); + expect(res.status).toBe(400); + }); + + it("stops at the cap and says so, in headers that arrive before the body", async () => { + seed(10); + + const res = await request(createApp()).get("/transfers.csv?maxRows=4"); + + expect(res.status).toBe(200); + expect(res.headers["x-truncated"]).toBe("true"); + expect(res.headers["x-row-limit"]).toBe("4"); + + // The response terminated — the failure mode here was a body that never ends. + const dataLines = res.text.trim().split("\n").slice(1); + expect(dataLines).toHaveLength(4); + }); + + it("does not claim truncation when the rows end exactly at the cap", async () => { + seed(4); + + const res = await request(createApp()).get("/transfers.csv?maxRows=4"); + + expect(res.status).toBe(200); + expect(res.headers["x-truncated"]).toBeUndefined(); + expect(res.text.trim().split("\n").slice(1)).toHaveLength(4); + }); + + it("returns everything, unflagged, when the result is under the cap", async () => { + seed(3); + + const res = await request(createApp()).get("/transfers.csv?maxRows=10"); + + expect(res.headers["x-truncated"]).toBeUndefined(); + expect(res.text.trim().split("\n").slice(1)).toHaveLength(3); + }); + + it("returns an empty export rather than failing when nothing matches", async () => { + seed(0); + + const res = await request(createApp()).get("/transfers.csv"); + + expect(res.status).toBe(200); + expect(res.headers["x-truncated"]).toBeUndefined(); + }); +}); diff --git a/src/routes/exports.ts b/src/routes/exports.ts index e88fbf861..a98134835 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -17,9 +17,21 @@ const BATCH_SIZE = 500; // paginate with fromLedger/toLedger or apply a tighter date/address filter. // The value is deliberately large enough to be useful but small enough to keep // memory and response time predictable under load. -const DEFAULT_MAX_ROWS = 50_000; const ABSOLUTE_MAX_ROWS = 500_000; +/** + * Default row cap, overridable per deployment with EXPORT_MAX_ROWS. + * + * Clamped to ABSOLUTE_MAX_ROWS rather than trusted: an env var is how an + * operator lowers this without a release, and a mistyped one must not be able + * to raise it past what the process can hold. A non-numeric value falls back + * rather than becoming NaN, which would make every comparison below false. + */ +const DEFAULT_MAX_ROWS = Math.min( + Number(process.env.EXPORT_MAX_ROWS) || 50_000, + ABSOLUTE_MAX_ROWS, +); + // ── Query-param schema ──────────────────────────────────────────────────────── // Exported so the OpenAPI build can reference it. export const exportQuerySchema = z.object({ @@ -70,6 +82,31 @@ function buildWhere(params: ExportQuery, network: Network) { return where; } +/** + * Is there a row beyond the cap? + * + * Asked before any bytes are written, because that is the only time the answer + * can still be put in a header. Detecting truncation mid-stream — by fetching + * one extra row and then calling res.setHeader — throws ERR_HTTP_HEADERS_SENT: + * the first csvStream.write has already flushed the response head. That throw + * skips csvStream.end(), so the client is left holding a CSV body that never + * terminates, and it happens only on the truncation path, which is the one case + * the signalling exists for. + * + * One indexed query against the same filter, and cheaper than a COUNT because + * it stops at the first row past the cap. + */ +async function hasRowsBeyond(where: Record, max: number): Promise { + const beyond = await prisma.tokenTransfer.findMany({ + where, + orderBy: { id: "asc" }, + skip: max, + take: 1, + select: { id: true }, + }); + return beyond.length > 0; +} + // ── Shared: async generator that yields rows in batches via cursor ──────────── // Stops once `limit` rows have been yielded so callers never pull more than // they asked for regardless of DB size. @@ -110,25 +147,21 @@ async function handleCsvExport(req: Request, res: Response, next: NextFunction) const network = requestNetwork(req); const where = buildWhere(parsed, network); - // Fetch one row beyond the cap so we can tell the caller whether the result - // was truncated without a separate COUNT query. - let rowCount = 0; - let truncated = false; + // Before a single byte: once the body starts, headers can no longer be set. + const truncated = await hasRowsBeyond(where, effectiveMax); res.setHeader("Content-Type", "text/csv"); res.setHeader("Content-Disposition", "attachment; filename=\"transfers.csv\""); res.setHeader("Transfer-Encoding", "chunked"); + if (truncated) { + res.setHeader("X-Truncated", "true"); + res.setHeader("X-Row-Limit", String(effectiveMax)); + } const csvStream = csvFormat({ headers: true }); csvStream.pipe(res); - for await (const row of streamTransfers(where, effectiveMax + 1)) { - if (rowCount === effectiveMax) { - // We fetched one extra row — result is truncated, don't write this row. - truncated = true; - break; - } - + for await (const row of streamTransfers(where, effectiveMax)) { csvStream.write({ id: row.id, contractId: row.contractId, @@ -144,18 +177,6 @@ async function handleCsvExport(req: Request, res: Response, next: NextFunction) isSac: row.isSac ?? false, createdAt: row.createdAt.toISOString(), }); - - rowCount++; - } - - // Signal truncation in a trailer header. HTTP/1.1 trailing headers require - // chunked encoding (which we've already set) and the client to opt in; as a - // belt-and-suspenders fallback we also set it as a regular response header - // before the body starts — Express buffers headers until the first write so - // this arrives before any CSV bytes. - if (truncated) { - res.setHeader("X-Truncated", "true"); - res.setHeader("X-Row-Limit", String(effectiveMax)); } csvStream.end(); @@ -201,15 +222,12 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti const writer = await parquet.ParquetWriter.openFile(schema, tmpFile); - let rowCount = 0; - let truncated = false; - - for await (const row of streamTransfers(where, effectiveMax + 1)) { - if (rowCount === effectiveMax) { - truncated = true; - break; - } + // Headers are set after the file is written here, so this could have been + // detected mid-stream — but it goes through the same helper as CSV so the + // two endpoints cannot answer differently about the same query. + const truncated = await hasRowsBeyond(where, effectiveMax); + for await (const row of streamTransfers(where, effectiveMax)) { await writer.appendRow({ id: row.id, contractId: row.contractId, @@ -225,8 +243,6 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti isSac: row.isSac ?? null, createdAt: row.createdAt.toISOString(), }); - - rowCount++; } await writer.close();