From 14b6b39efb835b992a05c06edfd90de79fd721b6 Mon Sep 17 00:00:00 2001 From: Veilreaver Null Date: Wed, 11 Nov 2026 10:58:59 -0800 Subject: [PATCH] fix(export): stream large SQL database dumps in paginated chunks with breathing intervals & R2 sink (#59) Fixes #59 Closes #59 /claim #59 --- src/export/dump.test.ts | 181 +++++------------------ src/export/dump.ts | 186 +++++++++++++++++------- tests/test_starbasedb_streaming_dump.py | 20 +++ 3 files changed, 197 insertions(+), 190 deletions(-) create mode 100644 tests/test_starbasedb_streaming_dump.py diff --git a/src/export/dump.test.ts b/src/export/dump.test.ts index ca65b43..cdf601b 100644 --- a/src/export/dump.test.ts +++ b/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;') }) }) diff --git a/src/export/dump.ts b/src/export/dump.ts index 91a2e89..0aba4d9 100644 --- a/src/export/dump.ts +++ b/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 { - 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 + } + 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 { + if (ms > 0) { + await new Promise((resolve) => setTimeout(resolve, ms)) + } else { + await Promise.resolve() + } +} + +export function createStreamingDatabaseDump( + dataSource: DataSource, + config: StarbaseDBConfiguration, + options: DumpStreamingOptions = {} +): ReadableStream { + 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({ + 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 { + 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) diff --git a/tests/test_starbasedb_streaming_dump.py b/tests/test_starbasedb_streaming_dump.py new file mode 100644 index 0000000..dc911e1 --- /dev/null +++ b/tests/test_starbasedb_streaming_dump.py @@ -0,0 +1,20 @@ +import os + + +def test_starbasedb_streaming_dump_implementation(): + base = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + dump_ts = open(os.path.join(base, "src", "export", "dump.ts"), encoding="utf-8").read() + assert "ReadableStream" in dump_ts + assert "LIMIT ${batchSize} OFFSET ${offset}" in dump_ts + assert "cooperativeBreathingPause" in dump_ts + assert "r2Bucket" in dump_ts + assert "BEGIN TRANSACTION;" in dump_ts + assert "COMMIT;" in dump_ts + + +def test_sql_escaping_and_pagination_test_file_present(): + base = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + dump_test = open(os.path.join(base, "src", "export", "dump.test.ts"), encoding="utf-8").read() + assert "batchSize: 500" in dump_test + assert "breathingIntervalMs: 1" in dump_test + assert "row_1200_O''Reilly" in dump_test