feat: streaming SQL/CSV/JSON dumps for large DBs (#59)

src/export/csv.test.ts

@@ -1,13 +1,11 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest'
 import { exportTableToCsvRoute } from './csv'
-import { getTableData, createExportResponse } from './index'
-import { createResponse } from '../utils'
+import { executeOperation } from './index'
 import type { DataSource } from '../types'
 import type { StarbaseDBConfiguration } from '../handler'
 
 vi.mock('./index', () => ({
-    getTableData: vi.fn(),
-    createExportResponse: vi.fn(),
+    executeOperation: vi.fn(),
 }))
 
 vi.mock('../utils', () => ({
@@ -24,14 +22,12 @@ let mockDataSource: DataSource
 let mockConfig: StarbaseDBConfiguration
 
 beforeEach(() => {
-    vi.clearAllMocks()
+    vi.mocked(executeOperation).mockReset()
 
     mockDataSource = {
         source: 'external',
         external: { dialect: 'sqlite' },
-        rpc: {
-            executeQuery: vi.fn(),
-        },
+        rpc: { executeQuery: vi.fn() },
     } as any
 
     mockConfig = {
@@ -41,118 +37,91 @@ beforeEach(() => {
     }
 })
 
-describe('CSV Export Module', () => {
-    it('should return a CSV file when table data exists', async () => {
-        vi.mocked(getTableData).mockResolvedValue([
+/** Wire up an existence check followed by a single page of rows. */
+function mockTable(rows: any[]) {
+    vi.mocked(executeOperation)
+        .mockResolvedValueOnce([{ name: 'tbl' }]) // existence check
+        .mockResolvedValueOnce(rows) // first page
+        .mockResolvedValueOnce([]) // empty page → terminate
+}
+
+describe('CSV Export Module (streaming)', () => {
+    it('streams a CSV body when the table has data', async () => {
+        mockTable([
             { id: 1, name: 'Alice', age: 30 },
             { id: 2, name: 'Bob', age: 25 },
         ])
 
-        vi.mocked(createExportResponse).mockReturnValue(
-            new Response('mocked-csv-content', {
-                headers: { 'Content-Type': 'text/csv' },
-            })
-        )
-
         const response = await exportTableToCsvRoute(
             'users',
             mockDataSource,
             mockConfig
         )
 
-        expect(getTableData).toHaveBeenCalledWith(
-            'users',
-            mockDataSource,
-            mockConfig
-        )
-        expect(createExportResponse).toHaveBeenCalledWith(
-            'id,name,age\n1,Alice,30\n2,Bob,25\n',
-            'users_export.csv',
-            'text/csv'
-        )
         expect(response.headers.get('Content-Type')).toBe('text/csv')
+        expect(response.headers.get('Content-Disposition')).toBe(
+            'attachment; filename="users_export.csv"'
+        )
+        expect(response.body).toBeInstanceOf(ReadableStream)
+
+        const csv = await response.text()
+        expect(csv).toBe('id,name,age\n1,Alice,30\n2,Bob,25\n')
     })
 
-    it('should return 404 if table does not exist', async () => {
-        vi.mocked(getTableData).mockResolvedValue(null)
+    it('returns 404 if the table does not exist', async () => {
+        vi.mocked(executeOperation).mockResolvedValueOnce([])
 
         const response = await exportTableToCsvRoute(
             'non_existent_table',
             mockDataSource,
             mockConfig
         )
 
-        expect(getTableData).toHaveBeenCalledWith(
-            'non_existent_table',
-            mockDataSource,
-            mockConfig
-        )
         expect(response.status).toBe(404)
-
-        const jsonResponse: { error: string } = await response.json()
-        expect(jsonResponse.error).toBe(
-            "Table 'non_existent_table' does not exist."
-        )
+        const body = (await response.json()) as { error: string }
+        expect(body.error).toBe("Table 'non_existent_table' does not exist.")
     })
 
-    it('should handle empty table (return only headers)', async () => {
-        vi.mocked(getTableData).mockResolvedValue([])
-
-        vi.mocked(createExportResponse).mockReturnValue(
-            new Response('mocked-csv-content', {
-                headers: { 'Content-Type': 'text/csv' },
-            })
-        )
+    it('emits an empty body for an empty table (header row needs at least one row to know columns)', async () => {
+        mockTable([])
+        // mockTable above queues two empty pages after the existence check; we
+        // only need one. Re-prime to keep this scenario explicit.
+        vi.mocked(executeOperation).mockReset()
+        vi.mocked(executeOperation)
+            .mockResolvedValueOnce([{ name: 'empty_table' }])
+            .mockResolvedValueOnce([])
 
         const response = await exportTableToCsvRoute(
             'empty_table',
             mockDataSource,
             mockConfig
         )
 
-        expect(getTableData).toHaveBeenCalledWith(
-            'empty_table',
-            mockDataSource,
-            mockConfig
-        )
-        expect(createExportResponse).toHaveBeenCalledWith(
-            '',
-            'empty_table_export.csv',
-            'text/csv'
-        )
         expect(response.headers.get('Content-Type')).toBe('text/csv')
+        const csv = await response.text()
+        expect(csv).toBe('')
     })
 
-    it('should escape commas and quotes in CSV values', async () => {
-        vi.mocked(getTableData).mockResolvedValue([
-            { id: 1, name: 'Sahithi, is', bio: 'my forever "penguin"' },
-        ])
-
-        vi.mocked(createExportResponse).mockReturnValue(
-            new Response('mocked-csv-content', {
-                headers: { 'Content-Type': 'text/csv' },
-            })
-        )
+    it('quotes fields containing commas, quotes, or newlines', async () => {
+        mockTable([{ id: 1, name: 'Sahithi, is', bio: 'my forever "penguin"' }])
 
         const response = await exportTableToCsvRoute(
             'special_chars',
             mockDataSource,
             mockConfig
         )
 
-        expect(createExportResponse).toHaveBeenCalledWith(
-            'id,name,bio\n1,"Sahithi, is","my forever ""penguin"""\n',
-            'special_chars_export.csv',
-            'text/csv'
+        const csv = await response.text()
+        expect(csv).toBe(
+            'id,name,bio\n1,"Sahithi, is","my forever ""penguin"""\n'
         )
-        expect(response.headers.get('Content-Type')).toBe('text/csv')
     })
 
-    it('should return 500 on an unexpected error', async () => {
-        const consoleErrorMock = vi
-            .spyOn(console, 'error')
-            .mockImplementation(() => {})
-        vi.mocked(getTableData).mockRejectedValue(new Error('Database Error'))
+    it('returns 500 when the existence check throws', async () => {
+        vi.spyOn(console, 'error').mockImplementation(() => {})
+        vi.mocked(executeOperation).mockRejectedValueOnce(
+            new Error('Database Error')
+        )
 
         const response = await exportTableToCsvRoute(
             'users',
@@ -161,7 +130,26 @@ describe('CSV Export Module', () => {
         )
 
         expect(response.status).toBe(500)
-        const jsonResponse: { error: string } = await response.json()
-        expect(jsonResponse.error).toBe('Failed to export table to CSV')
+        const body = (await response.json()) as { error: string }
+        expect(body.error).toBe('Failed to export table to CSV')
+    })
+
+    it('reads pages with parameterised LIMIT/OFFSET', async () => {
+        mockTable([{ id: 1, name: 'Alice' }])
+
+        const response = await exportTableToCsvRoute(
+            'users',
+            mockDataSource,
+            mockConfig
+        )
+        await response.text()
+
+        const dataCall = vi
+            .mocked(executeOperation)
+            .mock.calls.find(([qs]) =>
+                qs[0].sql.startsWith('SELECT * FROM users')
+            )
+        expect(dataCall).toBeDefined()
+        expect(dataCall![0][0].sql).toContain('LIMIT ? OFFSET ?')
     })
 })

src/export/csv.ts

@@ -1,54 +1,81 @@
-import { getTableData, createExportResponse } from './index'
+import { executeOperation } from './index'
 import { createResponse } from '../utils'
 import { DataSource } from '../types'
 import { StarbaseDBConfiguration } from '../handler'
+import {
+    DEFAULT_PAGE_SIZE,
+    chunksToStream,
+    iterateTableRows,
+    streamingResponse,
+} from './streaming'
+
+/**
+ * RFC-4180-ish quoting: only quote the field if it contains a delimiter,
+ * quote, or newline; double internal quotes. Matches the previous buffered
+ * implementation byte-for-byte.
+ */
+function csvField(value: unknown): string {
+    if (value === null || value === undefined) return ''
+    if (
+        typeof value === 'string' &&
+        (value.includes(',') || value.includes('"') || value.includes('\n'))
+    ) {
+        return `"${value.replace(/"/g, '""')}"`
+    }
+    return String(value)
+}
+
+async function* csvChunks(
+    tableName: string,
+    dataSource: DataSource,
+    config: StarbaseDBConfiguration,
+    pageSize: number = DEFAULT_PAGE_SIZE
+): AsyncGenerator<string, void, void> {
+    let headersEmitted = false
+    for await (const row of iterateTableRows(
+        tableName,
+        dataSource,
+        config,
+        pageSize
+    )) {
+        if (!headersEmitted) {
+            yield Object.keys(row).join(',') + '\n'
+            headersEmitted = true
+        }
+        yield Object.values(row).map(csvField).join(',') + '\n'
+    }
+    // For an empty table we deliberately emit nothing — same observable
+    // output as the buffered version, which used `csvContent = ''`.
+}
 
 export async function exportTableToCsvRoute(
     tableName: string,
     dataSource: DataSource,
     config: StarbaseDBConfiguration
 ): Promise<Response> {
     try {
-        const data = await getTableData(tableName, dataSource, config)
-
-        if (data === null) {
+        // Confirm table existence up front so 404 still returns synchronously
+        // with a JSON body rather than a half-streamed file.
+        const exists = await executeOperation(
+            [
+                {
+                    sql: `SELECT name FROM sqlite_master WHERE type='table' AND name=?;`,
+                    params: [tableName],
+                },
+            ],
+            dataSource,
+            config
+        )
+        if (!exists || exists.length === 0) {
             return createResponse(
                 undefined,
                 `Table '${tableName}' does not exist.`,
                 404
             )
         }
 
-        // Convert the result to CSV
-        let csvContent = ''
-        if (data.length > 0) {
-            // Add headers
-            csvContent += Object.keys(data[0]).join(',') + '\n'
-
-            // Add data rows
-            data.forEach((row: any) => {
-                csvContent +=
-                    Object.values(row)
-                        .map((value) => {
-                            if (
-                                typeof value === 'string' &&
-                                (value.includes(',') ||
-                                    value.includes('"') ||
-                                    value.includes('\n'))
-                            ) {
-                                return `"${value.replace(/"/g, '""')}"`
-                            }
-                            return value
-                        })
-                        .join(',') + '\n'
-            })
-        }
-
-        return createExportResponse(
-            csvContent,
-            `${tableName}_export.csv`,
-            'text/csv'
-        )
+        const stream = chunksToStream(csvChunks(tableName, dataSource, config))
+        return streamingResponse(stream, `${tableName}_export.csv`, 'text/csv')
     } catch (error: any) {
         console.error('CSV Export Error:', error)
         return createResponse(undefined, 'Failed to export table to CSV', 500)