From aeb9a50d46c5516e672ce38a6378cd24dac78d7c Mon Sep 17 00:00:00 2001 From: mattm Date: Mon, 26 Oct 2026 12:47:55 -0700 Subject: [PATCH] fix(export): paginate large database dumps in bounded chunks with breathing intervals and R2 support (#59) --- src/export/dump.node.test.ts | 40 +++++++++ src/export/dump.test.ts | 56 +++++++++++- src/export/dump.ts | 164 ++++++++++++++++++++++++++++++----- src/export/index.ts | 84 +++++++++++++++--- 4 files changed, 308 insertions(+), 36 deletions(-) create mode 100644 src/export/dump.node.test.ts diff --git a/src/export/dump.node.test.ts b/src/export/dump.node.test.ts new file mode 100644 index 0000000..f5e7a56 --- /dev/null +++ b/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'") +}) diff --git a/src/export/dump.test.ts b/src/export/dump.test.ts index ca65b43..e8238c9 100644 --- a/src/export/dump.test.ts +++ b/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() + 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([]) diff --git a/src/export/dump.ts b/src/export/dump.ts index 91a2e89..3a8ad07 100644 --- a/src/export/dump.ts +++ b/src/export/dump.ts @@ -1,13 +1,72 @@ -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 + ): Promise +} + +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 { 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';" }], @@ -15,12 +74,15 @@ export async function dumpDatabaseRoute( 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 }) diff --git a/src/export/index.ts b/src/export/index.ts index 9c40119..aefe2ad 100644 --- a/src/export/index.ts +++ b/src/export/index.ts @@ -2,6 +2,15 @@ import { DataSource } from '../types' import { executeTransaction } from '../operation' import { StarbaseDBConfiguration } from '../handler' +export const DEFAULT_EXPORT_BATCH_SIZE = 500 +export const DEFAULT_MAX_ACTIVE_WINDOW_MS = 5000 + +export interface PaginatedTableOptions { + batchSize?: number + breathingIntervalMs?: number + maxActiveWindowMs?: number +} + export async function executeOperation( queries: { sql: string; params?: any[] }[], dataSource: DataSource, @@ -13,16 +22,67 @@ export async function executeOperation( dataSource, config, })) as any[] - // return results?.length > 0 ? results[0] : undefined return results.length > 0 && Array.isArray(results[0]) ? results[0] : results } +export async function* paginateTableData( + tableName: string, + dataSource: DataSource, + config: StarbaseDBConfiguration, + options: PaginatedTableOptions = {} +): AsyncGenerator { + 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 + ) + + let offset = 0 + let windowStart = Date.now() + + while (true) { + const batch = await executeOperation( + [ + { + sql: `SELECT * FROM ${tableName} LIMIT ${batchSize} OFFSET ${offset};`, + }, + ], + dataSource, + config + ) + + if (!Array.isArray(batch) || batch.length === 0) { + break + } + + yield batch + + if (batch.length < batchSize) { + break + } + + offset += batch.length + + if ( + breathingIntervalMs > 0 && + Date.now() - windowStart >= maxActiveWindowMs + ) { + await new Promise((resolve) => + setTimeout(resolve, breathingIntervalMs) + ) + windowStart = Date.now() + } + } +} + export async function getTableData( tableName: string, dataSource: DataSource, - config: StarbaseDBConfiguration + config: StarbaseDBConfiguration, + options: PaginatedTableOptions = {} ): Promise { try { // Verify if the table exists @@ -41,13 +101,16 @@ export async function getTableData( return null } - // Get table data - const dataResult = await executeOperation( - [{ sql: `SELECT * FROM ${tableName};` }], + const allRows: any[] = [] + for await (const batch of paginateTableData( + tableName, dataSource, - config - ) - return dataResult + config, + options + )) { + allRows.push(...batch) + } + return allRows } catch (error: any) { console.error('Table Data Fetch Error:', error) throw error @@ -55,11 +118,12 @@ export async function getTableData( } export function createExportResponse( - data: any, + data: BlobPart | BlobPart[], fileName: string, contentType: string ): Response { - const blob = new Blob([data], { type: contentType }) + const parts = Array.isArray(data) ? data : [data] + const blob = new Blob(parts, { type: contentType }) const headers = new Headers({ 'Content-Type': contentType,