feat(export): stream /export/dump to R2 with DO alarm resumption (#59)

The legacy /export/dump route buffers the entire dump in memory and runs
synchronously, so it falls over on databases that exceed the 30s Worker
timeout or the Durable Object memory ceiling (currently 1GB, soon 10GB).

This change adds a streaming path that lives inside the Durable Object:

- POST /export/dump kicks off a job, opens an R2 multipart upload, and
  returns 202 with a jobId. Supports format=sql|csv|json plus optional
  callbackUrl/table/chunkSize.
- GET /export/dump/status/:jobId returns progress (tables, rows, bytes,
  parts uploaded) and a downloadUrl once status is 'completed'.
- GET /export/dump/download/:jobId streams the finished object back from
  R2 to the client.
- DELETE /export/dump/:jobId aborts an in-flight upload.

The engine paginates 1000 rows at a time, buffers up to the R2 multipart
5 MiB minimum, flushes parts as they fill, and budgets each tick at 20s.
When a tick yields, the leftover bytes are persisted to a temp R2 object
(DO storage values are capped at 128 KiB and cannot hold the buffer
directly). The DO alarm() handler dispatches dump work first, then falls
through to the existing cron logic, so the two co-exist on the same
alarm channel.

A new [[r2_buckets]] binding named DATABASE_DUMPS gates the streaming
path. The legacy GET /export/dump remains untouched for small databases
and existing clients.

Tests: 17 new unit tests covering the engine (mid-tick yield/resume,
multipart flushing at the 5 MiB threshold, error abort, BLOB literals,
empty databases, CSV/JSON formats) and the HTTP routes.

Author: ViperDroid

README.md

@@ -243,6 +243,54 @@ curl --location 'https://starbasedb.YOUR-ID-HERE.workers.dev/export/dump' \
 </code>
 </pre>
 
+<h4>Streaming SQL dumps for large databases</h4>
+The synchronous endpoint above buffers the whole dump in memory and is bounded by the 30-second Worker timeout. For databases that exceed that budget, switch to the streaming variant — it paginates rows, writes them straight into R2 over (potentially many) Durable Object alarm ticks, and lets you download the finished artifact when ready.
+
+Add an R2 binding called `DATABASE_DUMPS` to your `wrangler.toml`:
+
+<pre>
+<code>
+[[r2_buckets]]
+binding = "DATABASE_DUMPS"
+bucket_name = "starbasedb-dumps"
+</code>
+</pre>
+
+Kick off the job (returns 202 with a `jobId`):
+
+<pre>
+<code>
+curl -X POST 'https://starbasedb.YOUR-ID-HERE.workers.dev/export/dump' \
+--header 'Authorization: Bearer ABC123' \
+--header 'Content-Type: application/json' \
+--data '{ "format": "sql", "callbackUrl": "https://hooks.example.com/dump-done" }'
+</code>
+</pre>
+
+`format` may be `sql`, `csv`, or `json`. Optional fields: `callbackUrl` (POSTed the status view on completion), `table` (export a single table only), `chunkSize` (rows per SELECT batch; default 1000).
+
+Poll the status, then download once it reads `completed`:
+
+<pre>
+<code>
+curl 'https://starbasedb.YOUR-ID-HERE.workers.dev/export/dump/status/JOB_ID' \
+--header 'Authorization: Bearer ABC123'
+
+curl 'https://starbasedb.YOUR-ID-HERE.workers.dev/export/dump/download/JOB_ID' \
+--header 'Authorization: Bearer ABC123' \
+--output database_dump.sql
+</code>
+</pre>
+
+To cancel an in-flight job:
+
+<pre>
+<code>
+curl -X DELETE 'https://starbasedb.YOUR-ID-HERE.workers.dev/export/dump/JOB_ID' \
+--header 'Authorization: Bearer ABC123'
+</code>
+</pre>
+
 <h3>JSON Data Export</h3>
 <pre>
 <code>

src/do.ts

@@ -1,4 +1,17 @@
 import { DurableObject } from 'cloudflare:workers'
+import { runTick } from './export/dump-engine'
+import { createDumpHost, jobStateKey } from './export/do-dump-host'
+import {
+    DumpFormat,
+    DumpJobOptions,
+    DumpJobState,
+    DumpJobStatusView,
+    newJobState,
+    toStatusView,
+} from './export/streaming-dump'
+
+const ACTIVE_DUMP_KEY = 'dump:active'
+const DUMP_RESUME_DELAY_MS = 1_000
 
 export class StarbaseDBDurableObject extends DurableObject {
     // Durable storage for the SQL database
@@ -9,6 +22,8 @@ export class StarbaseDBDurableObject extends DurableObject {
     public connections = new Map<string, WebSocket>()
     // Store the client auth token for requests back to our Worker
     private clientAuthToken: string
+    // R2 bucket binding for streaming dump output (optional).
+    private dumpBucket: R2Bucket | undefined
 
     /**
      * The constructor is invoked once upon creation of the Durable Object, i.e. the first call to
@@ -22,6 +37,7 @@ export class StarbaseDBDurableObject extends DurableObject {
         this.clientAuthToken = env.CLIENT_AUTHORIZATION_TOKEN
         this.sql = ctx.storage.sql
         this.storage = ctx.storage
+        this.dumpBucket = env.DATABASE_DUMPS
 
         // Install default necessary `tmp_` tables for various features here.
         const cacheStatement = `
@@ -72,6 +88,10 @@ export class StarbaseDBDurableObject extends DurableObject {
             deleteAlarm: this.deleteAlarm.bind(this),
             getStatistics: this.getStatistics.bind(this),
             executeQuery: this.executeQuery.bind(this),
+            startDumpJob: this.startDumpJob.bind(this),
+            getDumpJob: this.getDumpJob.bind(this),
+            getDumpDownloadBody: this.getDumpDownloadBody.bind(this),
+            cancelDumpJob: this.cancelDumpJob.bind(this),
         }
     }
 
@@ -105,12 +125,28 @@ export class StarbaseDBDurableObject extends DurableObject {
     }
 
     async alarm() {
+        // Dispatch streaming dump work first so a long-running export does not
+        // get starved by other alarm-driven features. The dump engine sets its
+        // own continuation alarm if it needs another tick.
+        try {
+            await this.continueActiveDumpJob()
+        } catch (err) {
+            console.error('Failed to continue dump job:', err)
+        }
+
         try {
             // Fetch all the tasks that are marked to emit an event for this cycle.
-            const task = (await this.executeQuery({
-                sql: 'SELECT * FROM tmp_cron_tasks WHERE is_active = 1;',
-                isRaw: false,
-            })) as Record<string, SqlStorageValue>[]
+            // The cron table is created lazily by the CronPlugin; if it does not
+            // exist yet (fresh DO with no cron plugin installed) we silently skip.
+            let task: Record<string, SqlStorageValue>[] = []
+            try {
+                task = (await this.executeQuery({
+                    sql: 'SELECT * FROM tmp_cron_tasks WHERE is_active = 1;',
+                    isRaw: false,
+                })) as Record<string, SqlStorageValue>[]
+            } catch {
+                return
+            }
 
             if (!task.length) {
                 return
@@ -148,6 +184,219 @@ export class StarbaseDBDurableObject extends DurableObject {
         }
     }
 
+    // -- Streaming dump job machinery -------------------------------------
+
+    /**
+     * Begin a new streaming dump job. Creates the R2 multipart upload,
+     * persists initial state, and schedules an immediate alarm to kick off
+     * the first tick. Returns a status view the worker can hand back to the
+     * client right away.
+     */
+    public async startDumpJob(
+        options: DumpJobOptions
+    ): Promise<DumpJobStatusView> {
+        if (!this.dumpBucket) {
+            throw new Error(
+                'Streaming dump requires the DATABASE_DUMPS R2 binding. ' +
+                    'Add an [[r2_buckets]] entry to wrangler.toml.'
+            )
+        }
+
+        const tables = await this.listUserTables(options.table)
+        const jobId = crypto.randomUUID()
+        const state = newJobState(jobId, options, tables)
+
+        // Open the R2 multipart upload up-front so part numbers and uploadId
+        // are stable across alarm-driven continuations.
+        const upload = await this.dumpBucket.createMultipartUpload(
+            state.objectKey,
+            {
+                httpMetadata: {
+                    contentType: this.contentTypeFor(options.format),
+                    contentDisposition: `attachment; filename="${state.objectKey.split('/').pop()}"`,
+                },
+            }
+        )
+        state.uploadId = upload.uploadId
+
+        await this.storage.put(jobStateKey(jobId), state)
+        await this.storage.put(ACTIVE_DUMP_KEY, jobId)
+        await this.setAlarm(Date.now() + 1000)
+        return toStatusView(state)
+    }
+
+    /** Read the current state of a dump job for the status endpoint. */
+    public async getDumpJob(jobId: string): Promise<DumpJobStatusView | null> {
+        const state = await this.storage.get<DumpJobState>(jobStateKey(jobId))
+        if (!state) return null
+        return toStatusView(state)
+    }
+
+    /**
+     * Stream the dump body from R2 back to the worker so it can be relayed to
+     * the client. Returns null when the job is not finished yet or missing.
+     */
+    public async getDumpDownloadBody(jobId: string): Promise<{
+        body: ReadableStream
+        size: number
+        contentType: string
+        filename: string
+    } | null> {
+        if (!this.dumpBucket) return null
+        const state = await this.storage.get<DumpJobState>(jobStateKey(jobId))
+        if (!state || state.status !== 'completed') return null
+        const obj = await this.dumpBucket.get(state.objectKey)
+        if (!obj) return null
+        return {
+            body: obj.body,
+            size: obj.size,
+            contentType:
+                obj.httpMetadata?.contentType ??
+                this.contentTypeFor(state.format),
+            filename: state.objectKey.split('/').pop() ?? state.objectKey,
+        }
+    }
+
+    /** Cancel an in-flight dump job. Aborts the R2 multipart and marks failed. */
+    public async cancelDumpJob(
+        jobId: string
+    ): Promise<DumpJobStatusView | null> {
+        const state = await this.storage.get<DumpJobState>(jobStateKey(jobId))
+        if (!state) return null
+        if (state.status === 'completed' || state.status === 'failed')
+            return toStatusView(state)
+
+        if (this.dumpBucket && state.uploadId) {
+            try {
+                const upload = this.dumpBucket.resumeMultipartUpload(
+                    state.objectKey,
+                    state.uploadId
+                )
+                await upload.abort()
+            } catch (err) {
+                console.error('Failed to abort R2 multipart upload:', err)
+            }
+        }
+        if (this.dumpBucket && state.pendingTempKey) {
+            await this.dumpBucket.delete(state.pendingTempKey).catch(() => {})
+        }
+        state.status = 'cancelled'
+        state.error = 'Cancelled by user'
+        state.progress.updatedAt = Date.now()
+        await this.storage.put(jobStateKey(jobId), state)
+        const active = await this.storage.get<string>(ACTIVE_DUMP_KEY)
+        if (active === jobId) await this.storage.delete(ACTIVE_DUMP_KEY)
+        return toStatusView(state)
+    }
+
+    /**
+     * Run one tick of the active dump job (if any). Called from alarm().
+     * Persists state after every tick and re-arms the alarm if more work
+     * remains.
+     */
+    private async continueActiveDumpJob(): Promise<void> {
+        const activeId = await this.storage.get<string>(ACTIVE_DUMP_KEY)
+        if (!activeId) return
+        if (!this.dumpBucket) {
+            console.error(
+                'Active dump job exists but DATABASE_DUMPS R2 binding is missing.'
+            )
+            await this.storage.delete(ACTIVE_DUMP_KEY)
+            return
+        }
+        const state = await this.storage.get<DumpJobState>(
+            jobStateKey(activeId)
+        )
+        if (!state) {
+            await this.storage.delete(ACTIVE_DUMP_KEY)
+            return
+        }
+        if (
+            state.status === 'completed' ||
+            state.status === 'failed' ||
+            state.status === 'cancelled'
+        ) {
+            await this.storage.delete(ACTIVE_DUMP_KEY)
+            return
+        }
+
+        const host = createDumpHost({
+            sql: this.sql,
+            storage: this.storage,
+            bucket: this.dumpBucket,
+        })
+
+        try {
+            const { done } = await runTick(state, host)
+            await this.storage.put(jobStateKey(state.jobId), state)
+            if (done) {
+                await this.storage.delete(ACTIVE_DUMP_KEY)
+                if (state.callbackUrl) {
+                    this.fireDumpCallback(state).catch((err) =>
+                        console.error('Dump callback failed:', err)
+                    )
+                }
+            } else {
+                await this.setAlarm(Date.now() + DUMP_RESUME_DELAY_MS)
+            }
+        } catch (err) {
+            console.error('Dump engine error:', err)
+            // The engine sets state.status='failed' before throwing. Persist
+            // the failure so callers see it via the status endpoint.
+            state.status = 'failed'
+            state.error =
+                err instanceof Error ? err.message : String(err ?? 'unknown')
+            state.progress.updatedAt = Date.now()
+            await this.storage.put(jobStateKey(state.jobId), state)
+            await this.storage.delete(ACTIVE_DUMP_KEY)
+            if (state.callbackUrl) {
+                this.fireDumpCallback(state).catch((cbErr) =>
+                    console.error('Dump failure callback failed:', cbErr)
+                )
+            }
+        }
+    }
+
+    private async fireDumpCallback(state: DumpJobState): Promise<void> {
+        if (!state.callbackUrl) return
+        try {
+            await fetch(state.callbackUrl, {
+                method: 'POST',
+                headers: { 'Content-Type': 'application/json' },
+                body: JSON.stringify(toStatusView(state)),
+            })
+        } catch (err) {
+            console.error('Failed to fire dump completion callback:', err)
+        }
+    }
+
+    /**
+     * List the user-facing tables this DO holds. Excludes SQLite internal
+     * tables and the tmp_* feature tables which would otherwise leak into
+     * the export.
+     */
+    private async listUserTables(only?: string): Promise<string[]> {
+        if (only) {
+            const exists = (await this.executeQuery({
+                sql: 'SELECT name FROM sqlite_master WHERE type = ? AND name = ?;',
+                params: ['table', only],
+                isRaw: false,
+            })) as Record<string, SqlStorageValue>[]
+            return exists.length ? [only] : []
+        }
+        const rows = (await this.executeQuery({
+            sql: "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'tmp_%' AND name NOT LIKE '_cf_%' ORDER BY name;",
+            isRaw: false,
+        })) as Record<string, SqlStorageValue>[]
+        return rows.map((r) => String(r.name))
+    }
+
+    private contentTypeFor(format: DumpFormat): string {
+        if (format === 'csv') return 'text/csv'
+        if (format === 'json') return 'application/json'
+        return 'application/sql'
+    }
+
     public async getStatistics(): Promise<{
         databaseSize: number
         activeConnections: number