rest: streaming CSV / JSON export endpoint (M10.21 / C.21)

Copilot · Copilot · commit 9a1c97f32e99 · 2026-05-18T23:05:53.000+08:00
Add GET /data/:object/export that streams record exports without
buffering the full result set in memory.

Query params:
  format=csv|json     default csv. json emits a JSON array.
  fields=a,b,c        default: object schema fields, falls back to keys
                      of the first row when schema lookup is unavailable.
  filter=&lt;json&gt;       URL-encoded $filter, same shape as list endpoint.
  orderby=field:dir   comma-separated shorthand or a JSON object.
  limit=&lt;n&gt;           default 10 000, hard cap 50 000.
  page=&lt;n&gt;            driver chunk size, default 500, max 5 000.

The handler pages through findData() in chunks of `page` and writes
each chunk straight to the response (Content-Disposition: attachment
with a dated filename). CSV cells are RFC-4180 escaped via the new
formatCsvCell / rowsToCsv helpers (commas, quotes, CR, LF → quoted;
embedded quotes doubled). JSON exports stream as a single JSON array.

Auth: same enforceAuth gate as the existing CRUD endpoints. Project
scoping mirrors the list handler (works under both /api/v1/data and
/api/v1/projects/:projectId/data).

Tests (4 new, 57 total):
  - CSV header + RFC-4180 escaping (commas, embedded quotes, newlines).
  - JSON array emission.
  - filter JSON validation (400 on malformed input).
  - 50 k hard cap is honoured when callers ask for more.

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/packages/rest/src/rest-server.ts b/packages/rest/src/rest-server.ts
@@ -235,6 +235,39 @@ function parseCsvToRows(csv: string, mapping: Record<string, string> = {}): Arra
     return out;
 }
 
+/**
+ * Escape a single value into an RFC-4180 CSV cell. Values containing
+ * commas, quotes, CR, or LF are wrapped in double-quotes with embedded
+ * quotes doubled. `null` / `undefined` become an empty cell. Objects and
+ * arrays are serialised as compact JSON so nested data round-trips
+ * without flattening surprises.
+ */
+function formatCsvCell(value: any): string {
+    if (value === null || value === undefined) return '';
+    let s: string;
+    if (typeof value === 'string') s = value;
+    else if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') s = String(value);
+    else if (value instanceof Date) s = value.toISOString();
+    else { try { s = JSON.stringify(value); } catch { s = String(value); } }
+    if (/[",\r\n]/.test(s)) {
+        return `"${s.replace(/"/g, '""')}"`;
+    }
+    return s;
+}
+
+/**
+ * Serialise a list of rows to RFC-4180 CSV text. Caller supplies the
+ * ordered list of field names; unknown fields produce empty cells.
+ */
+function rowsToCsv(fields: string[], rows: Array<Record<string, any>>, includeHeader: boolean): string {
+    const lines: string[] = [];
+    if (includeHeader) lines.push(fields.map(formatCsvCell).join(','));
+    for (const row of rows) {
+        lines.push(fields.map(f => formatCsvCell(row?.[f])).join(','));
+    }
+    return lines.join('\r\n') + (lines.length > 0 ? '\r\n' : '');
+}
+
 /**
  * Structural subset of `KernelManager` that RestServer needs in order to
  * resolve a per-project protocol at request time. Typed locally to avoid
@@ -1628,6 +1661,156 @@ export class RestServer {
                 tags: ['data', 'import'],
             },
         });
+
+        // GET /data/:object/export  — streaming export (M10.21 / C.21)
+        //
+        // Query params:
+        //   format=csv|json     (default: csv. json emits a JSON array.)
+        //   fields=a,b,c        (default: derive from object schema; falls back to keys of the first row)
+        //   filter=<json>       ($filter as URL-encoded JSON, same shape as list endpoint)
+        //   orderby=field:desc  (optional ordering, mirrors $orderby semantics)
+        //   limit=<n>           (default 10000, hard cap 50000)
+        //   page=<n>            (driver chunk size, default 500, max 5000)
+        //
+        // Streams the response so 50k-row exports do not buffer in memory.
+        // Filename suggests `${object}-${YYYY-MM-DD}.${ext}` for browsers.
+        this.routeManager.register({
+            method: 'GET',
+            path: `${dataPath}/:object/export`,
+            handler: async (req: any, res: any) => {
+                try {
+                    const projectId = isScoped ? req.params?.projectId : undefined;
+                    const p = await this.resolveProtocol(projectId, req);
+                    const context = await this.resolveExecCtx(projectId, req);
+                    if (this.enforceAuth(req, res, context)) return;
+                    const objectName = String(req.params.object || '');
+                    if (!objectName) {
+                        res.status(400).json({ code: 'INVALID_REQUEST', error: 'object is required' });
+                        return;
+                    }
+                    const q = req.query ?? {};
+                    const format = (String(q.format ?? 'csv')).toLowerCase() === 'json' ? 'json' : 'csv';
+                    const HARD_CAP = 50_000;
+                    const MAX_CHUNK = 5_000;
+                    const requestedLimit = q.limit != null ? Math.max(1, Number(q.limit) || 0) : 10_000;
+                    const limit = Math.min(requestedLimit, HARD_CAP);
+                    const chunkSize = Math.min(MAX_CHUNK, Math.max(50, q.page != null ? Number(q.page) || 500 : 500));
+
+                    let filter: any = undefined;
+                    if (typeof q.filter === 'string' && q.filter.length > 0) {
+                        try { filter = JSON.parse(q.filter); }
+                        catch {
+                            res.status(400).json({ code: 'INVALID_REQUEST', error: 'filter must be JSON' });
+                            return;
+                        }
+                    } else if (q.filter && typeof q.filter === 'object') {
+                        filter = q.filter;
+                    }
+
+                    let orderby: any = undefined;
+                    if (typeof q.orderby === 'string' && q.orderby.length > 0) {
+                        // Accept "field:dir,field2:dir" shorthand or a JSON object.
+                        if (q.orderby.startsWith('{') || q.orderby.startsWith('[')) {
+                            try { orderby = JSON.parse(q.orderby); } catch { /* leave undefined */ }
+                        } else {
+                            const obj: Record<string, 'asc' | 'desc'> = {};
+                            for (const part of q.orderby.split(',')) {
+                                const [field, dir] = part.split(':').map((s: string) => s.trim());
+                                if (field) obj[field] = dir?.toLowerCase() === 'desc' ? 'desc' : 'asc';
+                            }
+                            if (Object.keys(obj).length > 0) orderby = obj;
+                        }
+                    }
+
+                    // Resolve fields: explicit param > schema fields > derived from first row.
+                    let fields: string[] | undefined;
+                    if (typeof q.fields === 'string' && q.fields.length > 0) {
+                        fields = q.fields.split(',').map((s: string) => s.trim()).filter(Boolean);
+                    } else if (Array.isArray(q.fields)) {
+                        fields = q.fields.filter((s: any) => typeof s === 'string' && s.length > 0);
+                    }
+                    if (!fields || fields.length === 0) {
+                        try {
+                            const schema = await (p as any).getObjectSchema?.(objectName, projectId);
+                            const schemaFields = schema?.fields;
+                            if (Array.isArray(schemaFields)) {
+                                fields = schemaFields.map((f: any) => f.name).filter((n: any) => typeof n === 'string');
+                            }
+                        } catch { /* fall back to first-row derivation */ }
+                    }
+
+                    // Prepare streaming response. Set headers BEFORE first write.
+                    const stamp = new Date().toISOString().slice(0, 10);
+                    const safeObj = objectName.replace(/[^A-Za-z0-9_.-]/g, '_');
+                    if (format === 'csv') {
+                        res.header('Content-Type', 'text/csv; charset=utf-8');
+                        res.header('Content-Disposition', `attachment; filename="${safeObj}-${stamp}.csv"`);
+                    } else {
+                        res.header('Content-Type', 'application/json; charset=utf-8');
+                        res.header('Content-Disposition', `attachment; filename="${safeObj}-${stamp}.json"`);
+                    }
+                    res.header('X-Export-Format', format);
+                    res.header('X-Export-Limit', String(limit));
+                    res.header('Cache-Control', 'no-store');
+
+                    let exported = 0;
+                    let firstChunk = true;
+                    let skip = 0;
+                    if (format === 'json') res.write('[');
+
+                    while (exported < limit) {
+                        const take = Math.min(chunkSize, limit - exported);
+                        const findArgs: any = {
+                            object: objectName,
+                            query: {
+                                ...(filter ? { $filter: filter } : {}),
+                                ...(orderby ? { $orderby: orderby } : {}),
+                                $top: take,
+                                $skip: skip,
+                            },
+                            ...(projectId ? { projectId } : {}),
+                            ...(context ? { context } : {}),
+                        };
+                        const result: any = await (p as any).findData(findArgs);
+                        const rows: any[] = Array.isArray(result?.data) ? result.data
+                            : Array.isArray(result?.rows) ? result.rows
+                                : Array.isArray(result) ? result : [];
+
+                        if (rows.length === 0) break;
+
+                        if (format === 'csv') {
+                            // Derive fields from the first row if schema lookup failed.
+                            if ((!fields || fields.length === 0) && firstChunk) {
+                                fields = Object.keys(rows[0] ?? {});
+                            }
+                            const text = rowsToCsv(fields ?? [], rows, firstChunk);
+                            res.write(text);
+                        } else {
+                            for (let i = 0; i < rows.length; i++) {
+                                const prefix = (firstChunk && i === 0) ? '' : ',';
+                                res.write(prefix + JSON.stringify(rows[i]));
+                            }
+                        }
+                        firstChunk = false;
+                        exported += rows.length;
+                        skip += rows.length;
+                        if (rows.length < take) break;
+                    }
+                    if (format === 'json') res.write(']');
+                    res.end();
+                } catch (error: any) {
+                    logError('[REST] Unhandled error:', error);
+                    // Best-effort error envelope; if headers already sent the
+                    // client receives a truncated stream which signals failure.
+                    try { sendError(res, error, String(req.params?.object || '')); }
+                    catch { try { res.end(); } catch { /* swallow */ } }
+                }
+            },
+            metadata: {
+                summary: 'Streaming export of object rows (CSV or JSON)',
+                tags: ['data', 'export'],
+            },
+        });
     }
 
     /**
diff --git a/packages/rest/src/rest.test.ts b/packages/rest/src/rest.test.ts
@@ -566,6 +566,120 @@ describe('RestServer', () => {
       });
     });
   });
+
+  // -----------------------------------------------------------------------
+  // GET /data/:object/export — streaming CSV / JSON export (M10.21 / C.21)
+  // -----------------------------------------------------------------------
+  describe('export handler', () => {
+    function getExportRoute(rest: any) {
+      const routes = rest.getRoutes();
+      return routes.find(
+        (r: any) => r.method === 'GET' && r.path === '/api/v1/data/:object/export',
+      );
+    }
+
+    function makeRes() {
+      const chunks: string[] = [];
+      const headers: Record<string, string> = {};
+      let status = 200;
+      const res: any = {
+        write: (s: string) => { chunks.push(s); },
+        end: vi.fn(),
+        header: (n: string, v: string) => { headers[n] = v; return res; },
+        status: (code: number) => { status = code; return res; },
+        json: vi.fn(),
+      };
+      return { res, chunks, headers, getStatus: () => status };
+    }
+
+    it('streams CSV with header row and quotes risky cells', async () => {
+      const rest = new RestServer(server as any, protocol as any);
+      rest.registerRoutes();
+      const route = getExportRoute(rest);
+      expect(route).toBeDefined();
+
+      protocol.findData.mockResolvedValueOnce({
+        data: [
+          { id: '1', name: 'Acme, Inc.', note: 'line1\nline2' },
+          { id: '2', name: 'Beta "Co"', note: null },
+        ],
+      });
+      // Second call returns empty -> ends the stream
+      protocol.findData.mockResolvedValueOnce({ data: [] });
+
+      const { res, chunks, headers } = makeRes();
+      const req = {
+        params: { object: 'account' },
+        query: { format: 'csv', fields: 'id,name,note', limit: '500' },
+      };
+
+      await route!.handler(req as any, res);
+
+      expect(headers['Content-Type']).toBe('text/csv; charset=utf-8');
+      expect(headers['Content-Disposition']).toMatch(/attachment; filename="account-\d{4}-\d{2}-\d{2}\.csv"/);
+      const text = chunks.join('');
+      expect(text.startsWith('id,name,note\r\n')).toBe(true);
+      expect(text).toContain('1,"Acme, Inc.","line1\nline2"');
+      expect(text).toContain('2,"Beta ""Co""",');
+      expect(res.end).toHaveBeenCalled();
+    });
+
+    it('streams JSON array when format=json', async () => {
+      const rest = new RestServer(server as any, protocol as any);
+      rest.registerRoutes();
+      const route = getExportRoute(rest);
+
+      protocol.findData.mockResolvedValueOnce({
+        data: [{ id: 'a' }, { id: 'b' }],
+      });
+      protocol.findData.mockResolvedValueOnce({ data: [] });
+
+      const { res, chunks, headers } = makeRes();
+      await route!.handler({
+        params: { object: 'lead' },
+        query: { format: 'json' },
+      } as any, res);
+
+      expect(headers['Content-Type']).toBe('application/json; charset=utf-8');
+      const text = chunks.join('');
+      expect(text).toBe('[{"id":"a"},{"id":"b"}]');
+    });
+
+    it('rejects invalid JSON in filter query', async () => {
+      const rest = new RestServer(server as any, protocol as any);
+      rest.registerRoutes();
+      const route = getExportRoute(rest);
+
+      const { res } = makeRes();
+      await route!.handler({
+        params: { object: 'account' },
+        query: { filter: '{not json' },
+      } as any, res);
+      expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ code: 'INVALID_REQUEST' }));
+    });
+
+    it('honours the hard 50k row cap', async () => {
+      const rest = new RestServer(server as any, protocol as any);
+      rest.registerRoutes();
+      const route = getExportRoute(rest);
+
+      // Always return full chunks so the loop is bounded only by `limit`.
+      protocol.findData.mockImplementation(async ({ query }: any) => {
+        const take = query?.$top ?? 0;
+        return { data: Array.from({ length: take }, (_v, i) => ({ id: String((query?.$skip ?? 0) + i) })) };
+      });
+
+      const { res, chunks } = makeRes();
+      await route!.handler({
+        params: { object: 'big' },
+        query: { format: 'csv', limit: '999999', fields: 'id' },
+      } as any, res);
+
+      const lines = chunks.join('').split('\r\n').filter(Boolean);
+      // 1 header + 50 000 rows.
+      expect(lines.length).toBe(50_001);
+    });
+  });
 });
 
 // ---------------------------------------------------------------------------
