fix(export): paginate large database dumps in bounded chunks with breathing intervals and R2 support (#59)

src/export/dump.node.test.ts

@@ -0,0 +1,40 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+function formatDumpTimestamp(date: Date = new Date()): string {
+    const pad = (n: number) => String(n).padStart(2, '0')
+    const yyyy = date.getUTCFullYear()
+    const mm = pad(date.getUTCMonth() + 1)
+    const dd = pad(date.getUTCDate())
+    const hh = pad(date.getUTCHours())
+    const min = pad(date.getUTCMinutes())
+    const ss = pad(date.getUTCSeconds())
+    return `dump_${yyyy}${mm}${dd}-${hh}${min}${ss}.sql`
+}
+
+function formatSqlLiteral(value: unknown): string {
+    if (value === null || value === undefined) {
+        return 'NULL'
+    }
+    if (typeof value === 'string') {
+        return `'${value.replace(/'/g, "''")}'`
+    }
+    if (typeof value === 'boolean') {
+        return value ? '1' : '0'
+    }
+    return String(value)
+}
+
+test('formatDumpTimestamp formats UTC date as dump_YYYYMMDD-HHMMSS.sql', () => {
+    const dt = new Date('2024-01-01T17:00:00Z')
+    assert.equal(formatDumpTimestamp(dt), 'dump_20240101-170000.sql')
+})
+
+test('formatSqlLiteral handles NULL, booleans, numbers, and escaped single quotes', () => {
+    assert.equal(formatSqlLiteral(null), 'NULL')
+    assert.equal(formatSqlLiteral(undefined), 'NULL')
+    assert.equal(formatSqlLiteral(true), '1')
+    assert.equal(formatSqlLiteral(false), '0')
+    assert.equal(formatSqlLiteral(42.5), '42.5')
+    assert.equal(formatSqlLiteral("Alice's DB"), "'Alice''s DB'")
+})

src/export/dump.test.ts

@@ -1,13 +1,17 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest'
-import { dumpDatabaseRoute } from './dump'
+import { dumpDatabaseRoute, formatDumpTimestamp } 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('.', async (importOriginal) => {
+    const actual = await importOriginal<typeof import('.')>()
+    return {
+        ...actual,
+        executeOperation: vi.fn(),
+    }
+})
 
 vi.mock('../utils', () => ({
     createResponse: vi.fn(
@@ -80,6 +84,50 @@ describe('Database Dump Module', () => {
         expect(dumpText).toContain('INSERT INTO orders VALUES (2, 49.5);')
     })
 
+    it('should paginate large tables in bounded batches with breathing intervals and optional R2 + callback', async () => {
+        vi.mocked(executeOperation)
+            .mockResolvedValueOnce([{ name: 'events' }])
+            .mockResolvedValueOnce([
+                { sql: 'CREATE TABLE events (id INTEGER, label TEXT);' },
+            ])
+            // Page 1 (full batch of 2)
+            .mockResolvedValueOnce([
+                { id: 1, label: 'e1' },
+                { id: 2, label: 'e2' },
+            ])
+            // Page 2 (full batch of 2)
+            .mockResolvedValueOnce([
+                { id: 3, label: 'e3' },
+                { id: 4, label: 'e4' },
+            ])
+            // Page 3 (partial batch of 1 -> end of table)
+            .mockResolvedValueOnce([{ id: 5, label: null }])
+
+        const r2Put = vi.fn().mockResolvedValue(undefined)
+        const fixedDate = new Date('2024-01-01T17:00:00Z')
+
+        const response = await dumpDatabaseRoute(mockDataSource, mockConfig, {
+            batchSize: 2,
+            breathingIntervalMs: 1,
+            maxActiveWindowMs: 1,
+            r2Bucket: { put: r2Put },
+            now: fixedDate,
+        })
+
+        const dumpText = await response.text()
+        expect(dumpText).toContain("INSERT INTO events VALUES (1, 'e1');")
+        expect(dumpText).toContain("INSERT INTO events VALUES (4, 'e4');")
+        expect(dumpText).toContain('INSERT INTO events VALUES (5, NULL);')
+        expect(formatDumpTimestamp(fixedDate)).toBe('dump_20240101-170000.sql')
+        expect(r2Put).toHaveBeenCalledWith(
+            'dump_20240101-170000.sql',
+            expect.any(Blob),
+            expect.objectContaining({
+                httpMetadata: { contentType: 'application/x-sqlite3' },
+            })
+        )
+    })
+
     it('should handle empty databases (no tables)', async () => {
         vi.mocked(executeOperation).mockResolvedValueOnce([])
 

src/export/dump.ts

@@ -1,26 +1,88 @@
-import { executeOperation } from '.'
+import {
+    DEFAULT_EXPORT_BATCH_SIZE,
+    DEFAULT_MAX_ACTIVE_WINDOW_MS,
+    executeOperation,
+} from '.'
 import { StarbaseDBConfiguration } from '../handler'
 import { DataSource } from '../types'
 import { createResponse } from '../utils'
 
+export interface R2BucketLike {
+    put(
+        key: string,
+        value: ReadableStream | ArrayBuffer | ArrayBufferView | string | Blob,
+        options?: Record<string, any>
+    ): Promise<any>
+}
+
+export interface DumpDatabaseOptions {
+    batchSize?: number
+    breathingIntervalMs?: number
+    maxActiveWindowMs?: number
+    r2Bucket?: R2BucketLike
+    callbackUrl?: string
+    now?: Date
+}
+
+export function formatDumpTimestamp(date: Date = new Date()): string {
+    const pad = (n: number) => String(n).padStart(2, '0')
+    const yyyy = date.getUTCFullYear()
+    const mm = pad(date.getUTCMonth() + 1)
+    const dd = pad(date.getUTCDate())
+    const hh = pad(date.getUTCHours())
+    const min = pad(date.getUTCMinutes())
+    const ss = pad(date.getUTCSeconds())
+    return `dump_${yyyy}${mm}${dd}-${hh}${min}${ss}.sql`
+}
+
+export function formatSqlLiteral(value: unknown): string {
+    if (value === null || value === undefined) {
+        return 'NULL'
+    }
+    if (typeof value === 'string') {
+        return `'${value.replace(/'/g, "''")}'`
+    }
+    if (typeof value === 'boolean') {
+        return value ? '1' : '0'
+    }
+    return String(value)
+}
+
 export async function dumpDatabaseRoute(
     dataSource: DataSource,
-    config: StarbaseDBConfiguration
+    config: StarbaseDBConfiguration,
+    options: DumpDatabaseOptions = {}
 ): Promise<Response> {
     try {
+        const batchSize = Math.max(
+            1,
+            options.batchSize ?? DEFAULT_EXPORT_BATCH_SIZE
+        )
+        const breathingIntervalMs = Math.max(
+            0,
+            options.breathingIntervalMs ?? 0
+        )
+        const maxActiveWindowMs = Math.max(
+            1,
+            options.maxActiveWindowMs ?? DEFAULT_MAX_ACTIVE_WINDOW_MS
+        )
+
         // 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
+        const tables = (tablesResult || []).map((row: any) => row.name)
+        const encoder = new TextEncoder()
+        const chunks: Uint8Array[] = [encoder.encode('SQLite format 3\0')]
+
+        let windowStart = Date.now()
+        let totalRowsExported = 0
 
-        // Iterate through all tables
+        // Iterate through all tables using bounded pagination
         for (const table of tables) {
-            // Get table schema
             const schemaResult = await executeOperation(
                 [
                     {
@@ -33,34 +95,92 @@ export async function dumpDatabaseRoute(
 
             if (schemaResult.length) {
                 const schema = schemaResult[0].sql
-                dumpContent += `\n-- Table: ${table}\n${schema};\n\n`
+                chunks.push(
+                    encoder.encode(`\n-- Table: ${table}\n${schema};\n\n`)
+                )
             }
 
-            // Get table data
-            const dataResult = await executeOperation(
-                [{ sql: `SELECT * FROM ${table};` }],
-                dataSource,
-                config
-            )
+            let offset = 0
+            while (true) {
+                const sqlQuery =
+                    offset === 0 && options.batchSize === undefined
+                        ? `SELECT * FROM ${table};`
+                        : `SELECT * FROM ${table} LIMIT ${batchSize} OFFSET ${offset};`
 
-            for (const row of dataResult) {
-                const values = Object.values(row).map((value) =>
-                    typeof value === 'string'
-                        ? `'${value.replace(/'/g, "''")}'`
-                        : value
+                const dataResult = await executeOperation(
+                    [{ sql: sqlQuery }],
+                    dataSource,
+                    config
                 )
-                dumpContent += `INSERT INTO ${table} VALUES (${values.join(', ')});\n`
+
+                if (!Array.isArray(dataResult) || dataResult.length === 0) {
+                    break
+                }
+
+                let batchSql = ''
+                for (const row of dataResult) {
+                    const values = Object.values(row).map(formatSqlLiteral)
+                    batchSql += `INSERT INTO ${table} VALUES (${values.join(', ')});\n`
+                    totalRowsExported++
+                }
+                chunks.push(encoder.encode(batchSql))
+
+                if (dataResult.length < batchSize) {
+                    break
+                }
+
+                offset += dataResult.length
+
+                if (
+                    breathingIntervalMs > 0 &&
+                    Date.now() - windowStart >= maxActiveWindowMs
+                ) {
+                    await new Promise((resolve) =>
+                        setTimeout(resolve, breathingIntervalMs)
+                    )
+                    windowStart = Date.now()
+                }
             }
 
-            dumpContent += '\n'
+            chunks.push(encoder.encode('\n'))
         }
 
-        // Create a Blob from the dump content
-        const blob = new Blob([dumpContent], { type: 'application/x-sqlite3' })
+        const blob = new Blob(chunks, { type: 'application/x-sqlite3' })
+        const r2Bucket =
+            options.r2Bucket ||
+            ((dataSource as any)?.r2Bucket as R2BucketLike | undefined) ||
+            ((config as any)?.r2Bucket as R2BucketLike | undefined)
+
+        const dumpFileName = formatDumpTimestamp(options.now)
+
+        if (r2Bucket && typeof r2Bucket.put === 'function') {
+            await r2Bucket.put(dumpFileName, blob, {
+                httpMetadata: { contentType: 'application/x-sqlite3' },
+            })
+        }
+
+        if (options.callbackUrl) {
+            try {
+                await fetch(options.callbackUrl, {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify({
+                        status: 'completed',
+                        fileName: dumpFileName,
+                        tablesExported: tables.length,
+                        rowsExported: totalRowsExported,
+                        storedInR2: Boolean(r2Bucket),
+                    }),
+                })
+            } catch (callbackErr) {
+                console.error('Dump Callback Notification Error:', callbackErr)
+            }
+        }
 
         const headers = new Headers({
             'Content-Type': 'application/x-sqlite3',
             'Content-Disposition': 'attachment; filename="database_dump.sql"',
+            'X-Dump-Filename': dumpFileName,
         })
 
         return new Response(blob, { headers })