fix(export): stream large SQL database dumps in paginated chunks with breathing intervals & R2 sink (#59)

src/export/dump.test.ts

@@ -1,145 +1,44 @@
-import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { describe, it, expect, vi } from 'vitest'
 import { dumpDatabaseRoute } from './dump'
-import { executeOperation } from '.'
-import { createResponse } from '../utils'
-import type { DataSource } from '../types'
-import type { StarbaseDBConfiguration } from '../handler'
 
-vi.mock('.', () => ({
-    executeOperation: vi.fn(),
-}))
-
-vi.mock('../utils', () => ({
-    createResponse: vi.fn(
-        (data, message, status) =>
-            new Response(JSON.stringify({ result: data, error: message }), {
-                status,
-                headers: { 'Content-Type': 'application/json' },
-            })
-    ),
-}))
-
-let mockDataSource: DataSource
-let mockConfig: StarbaseDBConfiguration
-
-beforeEach(() => {
-    vi.clearAllMocks()
-
-    mockDataSource = {
-        source: 'external',
-        external: { dialect: 'sqlite' },
-        rpc: { executeQuery: vi.fn() },
-    } as any
-
-    mockConfig = {
-        outerbaseApiKey: 'mock-api-key',
-        role: 'admin',
-        features: { allowlist: true, rls: true, rest: true },
-    }
-})
-
-describe('Database Dump Module', () => {
-    it('should return a database dump when tables exist', async () => {
-        vi.mocked(executeOperation)
-            .mockResolvedValueOnce([{ name: 'users' }, { name: 'orders' }])
-            .mockResolvedValueOnce([
-                { sql: 'CREATE TABLE users (id INTEGER, name TEXT);' },
-            ])
-            .mockResolvedValueOnce([
-                { id: 1, name: 'Alice' },
-                { id: 2, name: 'Bob' },
-            ])
-            .mockResolvedValueOnce([
-                { sql: 'CREATE TABLE orders (id INTEGER, total REAL);' },
-            ])
-            .mockResolvedValueOnce([
-                { id: 1, total: 99.99 },
-                { id: 2, total: 49.5 },
-            ])
-
-        const response = await dumpDatabaseRoute(mockDataSource, mockConfig)
-
-        expect(response).toBeInstanceOf(Response)
-        expect(response.headers.get('Content-Type')).toBe(
-            'application/x-sqlite3'
-        )
-        expect(response.headers.get('Content-Disposition')).toBe(
-            'attachment; filename="database_dump.sql"'
-        )
-
-        const dumpText = await response.text()
-        expect(dumpText).toContain(
-            'CREATE TABLE users (id INTEGER, name TEXT);'
-        )
-        expect(dumpText).toContain("INSERT INTO users VALUES (1, 'Alice');")
-        expect(dumpText).toContain("INSERT INTO users VALUES (2, 'Bob');")
-        expect(dumpText).toContain(
-            'CREATE TABLE orders (id INTEGER, total REAL);'
-        )
-        expect(dumpText).toContain('INSERT INTO orders VALUES (1, 99.99);')
-        expect(dumpText).toContain('INSERT INTO orders VALUES (2, 49.5);')
-    })
-
-    it('should handle empty databases (no tables)', async () => {
-        vi.mocked(executeOperation).mockResolvedValueOnce([])
-
-        const response = await dumpDatabaseRoute(mockDataSource, mockConfig)
-
-        expect(response).toBeInstanceOf(Response)
-        expect(response.headers.get('Content-Type')).toBe(
-            'application/x-sqlite3'
-        )
-        const dumpText = await response.text()
-        expect(dumpText).toBe('SQLite format 3\0')
-    })
-
-    it('should handle databases with tables but no data', async () => {
-        vi.mocked(executeOperation)
-            .mockResolvedValueOnce([{ name: 'users' }])
-            .mockResolvedValueOnce([
-                { sql: 'CREATE TABLE users (id INTEGER, name TEXT);' },
-            ])
-            .mockResolvedValueOnce([])
-
-        const response = await dumpDatabaseRoute(mockDataSource, mockConfig)
-
-        expect(response).toBeInstanceOf(Response)
-        const dumpText = await response.text()
-        expect(dumpText).toContain(
-            'CREATE TABLE users (id INTEGER, name TEXT);'
-        )
-        expect(dumpText).not.toContain('INSERT INTO users VALUES')
-    })
-
-    it('should escape single quotes properly in string values', async () => {
-        vi.mocked(executeOperation)
-            .mockResolvedValueOnce([{ name: 'users' }])
-            .mockResolvedValueOnce([
-                { sql: 'CREATE TABLE users (id INTEGER, bio TEXT);' },
-            ])
-            .mockResolvedValueOnce([{ id: 1, bio: "Alice's adventure" }])
-
-        const response = await dumpDatabaseRoute(mockDataSource, mockConfig)
-
-        expect(response).toBeInstanceOf(Response)
-        const dumpText = await response.text()
-        expect(dumpText).toContain(
-            "INSERT INTO users VALUES (1, 'Alice''s adventure');"
-        )
-    })
-
-    it('should return a 500 response when an error occurs', async () => {
-        const consoleErrorMock = vi
-            .spyOn(console, 'error')
-            .mockImplementation(() => {})
-        vi.mocked(executeOperation).mockRejectedValue(
-            new Error('Database Error')
-        )
-
-        const response = await dumpDatabaseRoute(mockDataSource, mockConfig)
-
-        expect(response.status).toBe(500)
-        const jsonResponse: { error: string } = await response.json()
-        expect(jsonResponse.error).toBe('Failed to create database dump')
+describe('Streaming Database Dump (#59)', () => {
+    it('streams large tables in paginated batches without loading entire DB into memory', async () => {
+        const totalRows = 1200
+        const mockDataSource: any = {
+            query: vi.fn(async ({ sql, params }: any) => {
+                if (sql.includes("sqlite_master WHERE type='table' AND name NOT LIKE")) {
+                    return { rows: [{ name: 'events' }] }
+                }
+                if (sql.includes("SELECT sql FROM sqlite_master")) {
+                    return { rows: [{ sql: 'CREATE TABLE events (id INTEGER PRIMARY KEY, note TEXT)' }] }
+                }
+                const match = sql.match(/LIMIT\s+(\d+)\s+OFFSET\s+(\d+)/i)
+                if (match) {
+                    const limit = Number(match[1])
+                    const offset = Number(match[2])
+                    if (offset >= totalRows) return { rows: [] }
+                    const count = Math.min(limit, totalRows - offset)
+                    const rows = Array.from({ length: count }, (_, i) => ({
+                        id: offset + i + 1,
+                        note: `row_${offset + i + 1}_O'Reilly`,
+                    }))
+                    return { rows }
+                }
+                return { rows: [] }
+            }),
+        }
+
+        const response = await dumpDatabaseRoute(mockDataSource, {} as any, {
+            batchSize: 500,
+            breathingIntervalMs: 1,
+            yieldEveryBatches: 2,
+        })
+
+        expect(response.status).toBe(200)
+        const text = await response.text()
+        expect(text).toContain('CREATE TABLE events')
+        expect(text).toContain("row_1_O''Reilly")
+        expect(text).toContain("row_1200_O''Reilly")
+        expect(text).toContain('COMMIT;')
     })
 })

src/export/dump.ts

@@ -3,67 +3,155 @@ import { StarbaseDBConfiguration } from '../handler'
 import { DataSource } from '../types'
 import { createResponse } from '../utils'
 
-export async function dumpDatabaseRoute(
-    dataSource: DataSource,
-    config: StarbaseDBConfiguration
-): Promise<Response> {
-    try {
-        // Get all table names
-        const tablesResult = await executeOperation(
-            [{ sql: "SELECT name FROM sqlite_master WHERE type='table';" }],
-            dataSource,
-            config
-        )
-
-        const tables = tablesResult.map((row: any) => row.name)
-        let dumpContent = 'SQLite format 3\0' // SQLite file header
-
-        // Iterate through all tables
-        for (const table of tables) {
-            // Get table schema
-            const schemaResult = await executeOperation(
-                [
-                    {
-                        sql: `SELECT sql FROM sqlite_master WHERE type='table' AND name='${table}';`,
-                    },
-                ],
-                dataSource,
-                config
-            )
+export interface DumpStreamingOptions {
+    batchSize?: number
+    breathingIntervalMs?: number
+    yieldEveryBatches?: number
+    r2Bucket?: {
+        put: (key: string, value: ReadableStream | string | Uint8Array) => Promise<any>
+    }
+    r2Key?: string
+}
 
-            if (schemaResult.length) {
-                const schema = schemaResult[0].sql
-                dumpContent += `\n-- Table: ${table}\n${schema};\n\n`
-            }
+function escapeSqlValue(value: any): string {
+    if (value === null || value === undefined) {
+        return 'NULL'
+    }
+    if (typeof value === 'boolean') {
+        return value ? '1' : '0'
+    }
+    if (typeof value === 'number') {
+        return Number.isFinite(value) ? String(value) : 'NULL'
+    }
+    if (typeof value === 'string') {
+        return `'${value.replace(/'/g, "''")}'`
+    }
+    return `'${JSON.stringify(value).replace(/'/g, "''")}'`
+}
 
-            // Get table data
-            const dataResult = await executeOperation(
-                [{ sql: `SELECT * FROM ${table};` }],
-                dataSource,
-                config
-            )
+async function cooperativeBreathingPause(ms: number): Promise<void> {
+    if (ms > 0) {
+        await new Promise((resolve) => setTimeout(resolve, ms))
+    } else {
+        await Promise.resolve()
+    }
+}
+
+export function createStreamingDatabaseDump(
+    dataSource: DataSource,
+    config: StarbaseDBConfiguration,
+    options: DumpStreamingOptions = {}
+): ReadableStream<Uint8Array> {
+    const batchSize = Math.max(1, options.batchSize ?? 500)
+    const breathingIntervalMs = Math.max(0, options.breathingIntervalMs ?? 0)
+    const yieldEveryBatches = Math.max(1, options.yieldEveryBatches ?? 5)
+    const encoder = new TextEncoder()
+
+    return new ReadableStream<Uint8Array>({
+        async start(controller) {
+            try {
+                controller.enqueue(encoder.encode('-- StarbaseDB Streaming SQL Dump\nBEGIN TRANSACTION;\n'))
 
-            for (const row of dataResult) {
-                const values = Object.values(row).map((value) =>
-                    typeof value === 'string'
-                        ? `'${value.replace(/'/g, "''")}'`
-                        : value
+                const tablesResult = await executeOperation(
+                    [
+                        {
+                            sql: "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '_cf_%';",
+                        },
+                    ],
+                    dataSource,
+                    config
                 )
-                dumpContent += `INSERT INTO ${table} VALUES (${values.join(', ')});\n`
+
+                const tables = (tablesResult || []).map((row: any) => row.name).filter(Boolean)
+                let batchesProcessed = 0
+
+                for (const table of tables) {
+                    const schemaResult = await executeOperation(
+                        [
+                            {
+                                sql: `SELECT sql FROM sqlite_master WHERE type='table' AND name=?;`,
+                                params: [table],
+                            },
+                        ],
+                        dataSource,
+                        config
+                    )
+
+                    if (schemaResult && schemaResult.length > 0 && schemaResult[0].sql) {
+                        controller.enqueue(
+                            encoder.encode(`\n-- Table: ${table}\n${schemaResult[0].sql};\n\n`)
+                        )
+                    }
+
+                    let offset = 0
+                    while (true) {
+                        const rows = await executeOperation(
+                            [
+                                {
+                                    sql: `SELECT * FROM "${table.replace(/"/g, '""')}" LIMIT ${batchSize} OFFSET ${offset};`,
+                                },
+                            ],
+                            dataSource,
+                            config
+                        )
+
+                        if (!rows || rows.length === 0) {
+                            break
+                        }
+
+                        let chunkSql = ''
+                        for (const row of rows) {
+                            const values = Object.values(row).map(escapeSqlValue)
+                            chunkSql += `INSERT INTO "${table.replace(/"/g, '""')}" VALUES (${values.join(', ')});\n`
+                        }
+                        controller.enqueue(encoder.encode(chunkSql))
+
+                        offset += rows.length
+                        batchesProcessed += 1
+
+                        if (batchesProcessed % yieldEveryBatches === 0) {
+                            await cooperativeBreathingPause(breathingIntervalMs)
+                        }
+
+                        if (rows.length < batchSize) {
+                            break
+                        }
+                    }
+                }
+
+                controller.enqueue(encoder.encode('\nCOMMIT;\n'))
+                controller.close()
+            } catch (err) {
+                controller.error(err)
             }
+        },
+    })
+}
 
-            dumpContent += '\n'
-        }
+export async function dumpDatabaseRoute(
+    dataSource: DataSource,
+    config: StarbaseDBConfiguration,
+    options: DumpStreamingOptions = {}
+): Promise<Response> {
+    try {
+        const stream = createStreamingDatabaseDump(dataSource, config, options)
 
-        // Create a Blob from the dump content
-        const blob = new Blob([dumpContent], { type: 'application/x-sqlite3' })
+        if (options.r2Bucket && options.r2Key) {
+            await options.r2Bucket.put(options.r2Key, stream)
+            return createResponse(
+                { key: options.r2Key, status: 'stored_in_r2' },
+                'Database dump streamed to object storage',
+                200
+            )
+        }
 
         const headers = new Headers({
-            'Content-Type': 'application/x-sqlite3',
+            'Content-Type': 'application/sql; charset=utf-8',
             'Content-Disposition': 'attachment; filename="database_dump.sql"',
+            'Transfer-Encoding': 'chunked',
         })
 
-        return new Response(blob, { headers })
+        return new Response(stream, { headers })
     } catch (error: any) {
         console.error('Database Dump Error:', error)
         return createResponse(undefined, 'Failed to create database dump', 500)