stream CSV rows for large commitment exports

src/app/api/commitments/export/route.ts

@@ -1,14 +1,17 @@
 import { NextRequest, NextResponse } from 'next/server';
 import { verifySessionToken } from '@/lib/backend/auth';
-import { buildCsv } from '@/lib/backend/csv';
+import { type CsvRow, createCsvStream } from '@/lib/backend/csv';
 import {
   BadRequestError,
   ForbiddenError,
   TooManyRequestsError,
   UnauthorizedError,
 } from '@/lib/backend/errors';
 import { checkRateLimit } from '@/lib/backend/rateLimit';
-import { getUserCommitmentsFromChain } from '@/lib/backend/services/contracts';
+import {
+  getUserCommitmentsFromChain,
+  type Commitment,
+} from '@/lib/backend/services/contracts';
 import { withApiHandler } from '@/lib/backend/withApiHandler';
 
 const CSV_HEADERS = [
@@ -48,6 +51,29 @@ function normalizeAddress(address: string): string {
   return address.trim().toLowerCase();
 }
 
+/**
+ * Lazily maps commitments to CSV rows. Using a generator avoids
+ * materializing the full mapped array — the streamer pulls one row at a
+ * time, so only a single row exists in memory between iterations.
+ */
+function* commitmentsToRows(commitments: Iterable<Commitment>): Generator<CsvRow> {
+  for (const commitment of commitments) {
+    yield [
+      commitment.id,
+      commitment.ownerAddress,
+      commitment.asset,
+      stringifyCsvValue(commitment.amount),
+      commitment.status,
+      stringifyCsvValue(commitment.complianceScore),
+      stringifyCsvValue(commitment.currentValue),
+      stringifyCsvValue(commitment.feeEarned),
+      stringifyCsvValue(commitment.violationCount),
+      stringifyCsvValue(commitment.createdAt),
+      stringifyCsvValue(commitment.expiresAt),
+    ];
+  }
+}
+
 export const GET = withApiHandler(async (req: NextRequest) => {
   const ip = req.ip ?? req.headers.get('x-forwarded-for') ?? 'anonymous';
   const isAllowed = await checkRateLimit(ip, 'api/commitments/export');
@@ -72,27 +98,19 @@ export const GET = withApiHandler(async (req: NextRequest) => {
     throw new ForbiddenError();
   }
 
+  // Fetch happens before streaming starts so any failure here is caught by
+  // `withApiHandler` and surfaced as a JSON error response, not a truncated
+  // CSV. When `getUserCommitmentsFromChain` becomes streamable, swap the
+  // generator argument for the async iterable directly.
   const commitments = await getUserCommitmentsFromChain(ownerAddress);
-  const rows = commitments.map((commitment) => [
-    commitment.id,
-    commitment.ownerAddress,
-    commitment.asset,
-    stringifyCsvValue(commitment.amount),
-    commitment.status,
-    stringifyCsvValue(commitment.complianceScore),
-    stringifyCsvValue(commitment.currentValue),
-    stringifyCsvValue(commitment.feeEarned),
-    stringifyCsvValue(commitment.violationCount),
-    stringifyCsvValue(commitment.createdAt),
-    stringifyCsvValue(commitment.expiresAt),
-  ]);
-  const csv = buildCsv(CSV_HEADERS, rows);
-
-  return new NextResponse(csv, {
+  const stream = createCsvStream(CSV_HEADERS, commitmentsToRows(commitments));
+
+  return new NextResponse(stream, {
     status: 200,
     headers: {
       'Content-Type': 'text/csv; charset=utf-8',
       'Content-Disposition': 'attachment; filename="commitments.csv"',
+      'Cache-Control': 'no-store',
     },
   });
-});
+});

src/lib/backend/csv.ts

@@ -1,5 +1,7 @@
 const FORMULA_PREFIX_PATTERN = /^[=+\-@]/;
 
+export type CsvRow = Array<string | null | undefined>;
+
 export function escapeCsvField(value: string | null | undefined): string {
   const normalizedValue = value == null ? '' : String(value);
   const safeValue = FORMULA_PREFIX_PATTERN.test(normalizedValue)
@@ -15,8 +17,47 @@ export function escapeCsvField(value: string | null | undefined): string {
   return shouldWrap ? `"${escapedValue}"` : escapedValue;
 }
 
-export function buildCsv(headers: string[], rows: Array<Array<string | null | undefined>>): string {
-  const csvRows = [headers, ...rows].map((row) => row.map((value) => escapeCsvField(value)).join(','));
+/**
+ * Formats a single CSV row and terminates it with CRLF.
+ * Shared by `buildCsv` and `createCsvStream` so the injection guard
+ * (see `escapeCsvField`) is applied identically in both code paths.
+ */
+export function formatCsvRow(row: CsvRow): string {
+  return `${row.map(escapeCsvField).join(',')}\r\n`;
+}
 
-  return `${csvRows.join('\r\n')}\r\n`;
+export function buildCsv(headers: string[], rows: CsvRow[]): string {
+  return [headers, ...rows].map(formatCsvRow).join('');
 }
+
+/**
+ * Returns a ReadableStream that emits the CSV header row first, then one
+ * chunk per data row. Accepts a sync or async iterable so callers can pass
+ * an in-memory array today or a paginated/streamed source in the future
+ * without changing this helper.
+ *
+ * Errors thrown by the iterable are forwarded to `controller.error`, which
+ * aborts the response. The client will see a truncated download; the caller
+ * is responsible for ensuring upstream errors are surfaced before streaming
+ * starts when possible.
+ */
+export function createCsvStream(
+  headers: string[],
+  rows: Iterable<CsvRow> | AsyncIterable<CsvRow>,
+): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      try {
+        controller.enqueue(encoder.encode(formatCsvRow(headers)));
+        for await (const row of rows as AsyncIterable<CsvRow>) {
+          controller.enqueue(encoder.encode(formatCsvRow(row)));
+        }
+        controller.close();
+      } catch (err) {
+        controller.error(err);
+      }
+    },
+  });
+}

tests/lib/csv.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest';
-import { buildCsv, escapeCsvField } from '@/lib/backend/csv';
+import { buildCsv, createCsvStream, escapeCsvField, formatCsvRow } from '@/lib/backend/csv';
 
 describe('escapeCsvField', () => {
   it('passes through basic fields', () => {
@@ -59,3 +59,117 @@ describe('buildCsv', () => {
     expect(csv).not.toContain('Name\nAlice');
   });
 });
+
+describe('formatCsvRow', () => {
+  it('joins fields with commas and terminates the row with CRLF', () => {
+    expect(formatCsvRow(['a', 'b', 'c'])).toBe('a,b,c\r\n');
+  });
+
+  it('escapes each field using the same rules as escapeCsvField', () => {
+    expect(formatCsvRow(['alpha,beta', 'say "hi"'])).toBe('"alpha,beta","say ""hi"""\r\n');
+  });
+
+  it('prevents spreadsheet formula injection for any field in the row', () => {
+    expect(formatCsvRow(['safe', '=SUM(A1)', '+1'])).toBe("safe,'=SUM(A1),'+1\r\n");
+  });
+
+  it('produces the same output as buildCsv when called per row', () => {
+    const headers = ['Name', 'Value'];
+    const rows = [
+      ['Alice', '=danger'],
+      ['Bob', '200'],
+    ];
+
+    const composed = [headers, ...rows].map(formatCsvRow).join('');
+
+    expect(composed).toBe(buildCsv(headers, rows));
+  });
+});
+
+describe('createCsvStream', () => {
+  async function readStream(stream: ReadableStream<Uint8Array>): Promise<string> {
+    const reader = stream.getReader();
+    const decoder = new TextDecoder();
+    let result = '';
+
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      result += decoder.decode(value, { stream: true });
+    }
+
+    return result + decoder.decode();
+  }
+
+  async function collectChunks(stream: ReadableStream<Uint8Array>): Promise<string[]> {
+    const reader = stream.getReader();
+    const decoder = new TextDecoder();
+    const chunks: string[] = [];
+
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      chunks.push(decoder.decode(value));
+    }
+
+    return chunks;
+  }
+
+  it('emits only the header row when there are no data rows', async () => {
+    const stream = createCsvStream(['Name', 'Value'], []);
+
+    expect(await readStream(stream)).toBe('Name,Value\r\n');
+  });
+
+  it('emits headers and data rows in CSV format', async () => {
+    const stream = createCsvStream(
+      ['Name', 'Value'],
+      [
+        ['Alice', '100'],
+        ['Bob', '200'],
+      ]
+    );
+
+    expect(await readStream(stream)).toBe('Name,Value\r\nAlice,100\r\nBob,200\r\n');
+  });
+
+  it('emits the header in a separate chunk before any data rows', async () => {
+    const stream = createCsvStream(['Name'], [['Alice'], ['Bob']]);
+
+    const chunks = await collectChunks(stream);
+
+    expect(chunks[0]).toBe('Name\r\n');
+    expect(chunks.length).toBe(3);
+  });
+
+  it('accepts async iterables for paginated or streamed sources', async () => {
+    async function* source() {
+      yield ['1'];
+      yield ['2'];
+    }
+
+    const stream = createCsvStream(['id'], source());
+
+    expect(await readStream(stream)).toBe('id\r\n1\r\n2\r\n');
+  });
+
+  it('preserves the formula-injection guard for streamed rows', async () => {
+    const stream = createCsvStream(
+      ['value'],
+      [['=SUM(A1)'], ['+1'], ['-10'], ['@cmd']]
+    );
+
+    expect(await readStream(stream)).toBe("value\r\n'=SUM(A1)\r\n'+1\r\n'-10\r\n'@cmd\r\n");
+  });
+
+  it('forwards errors thrown by the iterable to the stream consumer', async () => {
+    async function* failing() {
+      yield ['1'];
+      throw new Error('upstream failed');
+    }
+
+    const stream = createCsvStream(['value'], failing());
+
+    await expect(readStream(stream)).rejects.toThrow('upstream failed');
+  });
+});