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 7ea28797a..a98134835 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -4,24 +4,57 @@ 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"; import { getCachedTokenDecimals } from "../tokenCache"; // 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 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({ + 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. @@ -32,74 +65,117 @@ 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; } +/** + * 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 ──────────── -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 parsed = parseOr400(exportQuerySchema, req.query, res); + if (!parsed) return; + + const effectiveMax = parsed.maxRows ?? DEFAULT_MAX_ROWS; const network = requestNetwork(req); - const where = buildWhere(req.query as Record, network); + const where = buildWhere(parsed, network); + + // 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)) { + for await (const row of streamTransfers(where, effectiveMax)) { csvStream.write({ - id: row.id, - contractId: row.contractId, - eventType: row.eventType, - fromAddress: row.fromAddress ?? "", - toAddress: row.toAddress ?? "", - amount: row.amount, - displayAmount: toDisplayAmount(row.amount, getCachedTokenDecimals(row.contractId, network)), - 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, getCachedTokenDecimals(row.contractId, network)), + ledger: row.ledger, + ledgerClosedAt: row.ledgerClosedAt.toISOString(), + txHash: row.txHash, + eventId: row.eventId, + isSac: row.isSac ?? false, + createdAt: row.createdAt.toISOString(), }); } @@ -109,17 +185,24 @@ 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 parsed = parseOr400(exportQuerySchema, req.query, res); + if (!parsed) return; + + const effectiveMax = parsed.maxRows ?? DEFAULT_MAX_ROWS; const network = requestNetwork(req); - const where = buildWhere(req.query as Record, network); + const where = buildWhere(parsed, network); const schema = new parquet.ParquetSchema({ id: { type: "INT64" }, @@ -139,7 +222,12 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti const writer = await parquet.ParquetWriter.openFile(schema, tmpFile); - for await (const row of streamTransfers(where)) { + // 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, @@ -161,6 +249,10 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti 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); @@ -175,7 +267,7 @@ async function handleParquetExport(req: Request, res: Response, next: NextFuncti } } -// ── Router ─────────────────────────────────────────────────────────────────── +// ── Router ──────────────────────────────────────────────────────────────────── export function createExportsRouter(): Router { const router = Router(); router.get("/transfers.csv", handleCsvExport);