From 84582ea4ab802fee46bbd07bc6a15611ddbef691 Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Wed, 13 May 2026 14:42:06 +0530
Subject: [PATCH 1/8] test: cover template page hydration

---
 .../interface/pages/template/index.test.tsx   | 47 +++++++++++++++++++
 1 file changed, 47 insertions(+)
 create mode 100644 plugins/interface/pages/template/index.test.tsx

diff --git a/plugins/interface/pages/template/index.test.tsx b/plugins/interface/pages/template/index.test.tsx
new file mode 100644
index 0000000..6e3f13c
--- /dev/null
+++ b/plugins/interface/pages/template/index.test.tsx
@@ -0,0 +1,47 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const hydrateRoot = vi.hoisted(() => vi.fn())
+
+vi.mock('hono/jsx/dom/client', () => ({
+    hydrateRoot,
+}))
+
+vi.mock('../../public/global.css', () => ({}))
+
+describe('template page entrypoint', () => {
+    beforeEach(() => {
+        vi.resetModules()
+        hydrateRoot.mockClear()
+    })
+
+    afterEach(() => {
+        vi.unstubAllGlobals()
+    })
+
+    it('does not hydrate when the template root is missing', async () => {
+        const querySelector = vi.fn(() => null)
+        vi.stubGlobal('document', { querySelector })
+
+        await import('./index')
+
+        expect(querySelector).toHaveBeenCalledWith(
+            '#root[data-client="template"]'
+        )
+        expect(hydrateRoot).not.toHaveBeenCalled()
+    })
+
+    it('hydrates the template page when the server root is present', async () => {
+        const root = {
+            dataset: {
+                serverProps: '{}',
+            },
+        }
+        const querySelector = vi.fn(() => root)
+        vi.stubGlobal('document', { querySelector })
+
+        await import('./index')
+
+        expect(hydrateRoot).toHaveBeenCalledTimes(1)
+        expect(hydrateRoot).toHaveBeenCalledWith(root, expect.any(Object))
+    })
+})

From ede63420ff0e32a98ee9aa89764d71fcae8c949b Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Wed, 13 May 2026 21:06:56 +0530
Subject: [PATCH 2/8] test: cover interface primitives

---
 .../interface/components/primitives.test.tsx  | 98 +++++++++++++++++++
 1 file changed, 98 insertions(+)
 create mode 100644 plugins/interface/components/primitives.test.tsx

diff --git a/plugins/interface/components/primitives.test.tsx b/plugins/interface/components/primitives.test.tsx
new file mode 100644
index 0000000..d701376
--- /dev/null
+++ b/plugins/interface/components/primitives.test.tsx
@@ -0,0 +1,98 @@
+import { renderToString } from 'hono/jsx/dom/server'
+import { describe, expect, it } from 'vitest'
+
+import { Avatar } from './avatar'
+import { Card } from './card'
+import { Label } from './label/Label'
+import { Loader } from './loader/Loader'
+import { Toggle } from './toggle'
+
+describe('interface primitive components', () => {
+    it('renders avatar links with fallback initials and custom classes', () => {
+        const html = renderToString(
+            <Avatar as="a" href="/profile" username="outerbase" class="extra" />
+        )
+
+        expect(html).toContain('<a')
+        expect(html).toContain('href="/profile"')
+        expect(html).toContain('extra')
+        expect(html).toContain('>O</p>')
+    })
+
+    it('renders avatar images with accessible alt text and selected state', () => {
+        const html = renderToString(
+            <Avatar
+                image="/avatar.png"
+                toggled
+                username="Ada"
+                data-testid="avatar"
+            />
+        )
+
+        expect(html).toContain('<button')
+        expect(html).toContain('after:opacity-100')
+        expect(html).toContain('src="/avatar.png"')
+        expect(html).toContain('alt="Ada"')
+        expect(html).toContain('data-testid="avatar"')
+    })
+
+    it('renders cards as links or divs with variant classes', () => {
+        const link = renderToString(
+            <Card as="a" href="/docs" variant="primary">
+                Docs
+            </Card>
+        )
+        const panel = renderToString(
+            <Card variant="secondary" data-testid="card">
+                Panel
+            </Card>
+        )
+
+        expect(link).toContain('<a')
+        expect(link).toContain('href="/docs"')
+        expect(link).toContain('btn-primary')
+        expect(link).toContain('Docs')
+
+        expect(panel).toContain('<div')
+        expect(panel).toContain('btn-secondary')
+        expect(panel).toContain('data-testid="card"')
+    })
+
+    it('renders labels with validation messaging only when invalid', () => {
+        const invalid = renderToString(
+            <Label
+                title="Database"
+                required
+                requiredDescription="Required"
+                isValid={false}
+            >
+                <input />
+            </Label>
+        )
+        const valid = renderToString(
+            <Label
+                title="Database"
+                required
+                requiredDescription="Required"
+                isValid
+            />
+        )
+
+        expect(invalid).toContain('Database')
+        expect(invalid).toContain('*')
+        expect(invalid).toContain('Required')
+        expect(valid).not.toContain('Required')
+    })
+
+    it('renders loader and toggle sizing/state classes', () => {
+        const loader = renderToString(<Loader size={18} class="spin" />)
+        const toggle = renderToString(
+            <Toggle onClick={() => undefined} size="lg" toggled />
+        )
+
+        expect(loader).toContain('class="spin"')
+        expect(loader).toContain('style="height: 18px; width: 18px"')
+        expect(toggle).toContain('h-7.5 w-12.5')
+        expect(toggle).toContain('translate-x-full')
+    })
+})

From a85df987d4bf675e7c9bff5476791b8817d8386e Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Wed, 13 May 2026 23:34:01 +0530
Subject: [PATCH 3/8] test: cover input primitive states

---
 .../interface/components/primitives.test.tsx  | 35 +++++++++++++++++++
 1 file changed, 35 insertions(+)

diff --git a/plugins/interface/components/primitives.test.tsx b/plugins/interface/components/primitives.test.tsx
index d701376..12ec5fc 100644
--- a/plugins/interface/components/primitives.test.tsx
+++ b/plugins/interface/components/primitives.test.tsx
@@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest'
 
 import { Avatar } from './avatar'
 import { Card } from './card'
+import { Input } from './input/Input'
 import { Label } from './label/Label'
 import { Loader } from './loader/Loader'
 import { Toggle } from './toggle'
@@ -95,4 +96,38 @@ describe('interface primitive components', () => {
         expect(toggle).toContain('h-7.5 w-12.5')
         expect(toggle).toContain('translate-x-full')
     })
+
+    it('renders input wrappers with prefix, suffix, and invalid state', () => {
+        const wrapped = renderToString(
+            <Input
+                initialValue="abc"
+                isValid={false}
+                onValueChange={() => undefined}
+                placeholder="Filter"
+                preText="$"
+                postText="USD"
+                size="sm"
+            />
+        )
+        const plain = renderToString(
+            <Input
+                className="extra-input"
+                initialValue="plain"
+                onValueChange={() => undefined}
+                size="lg"
+            />
+        )
+
+        expect(wrapped).toContain('<div')
+        expect(wrapped).toContain('ob-size-sm')
+        expect(wrapped).toContain('>$</span>')
+        expect(wrapped).toContain('>USD</span>')
+        expect(wrapped).toContain('placeholder="Filter"')
+        expect(wrapped).toContain('text-ob-destructive')
+
+        expect(plain).toContain('<input')
+        expect(plain).toContain('extra-input')
+        expect(plain).toContain('ob-size-lg')
+        expect(plain).toContain('value="plain"')
+    })
 })

From 8496c2f55e0597466c4257321d84f7c8bbc96906 Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Thu, 14 May 2026 00:07:25 +0530
Subject: [PATCH 4/8] test: cover public package entrypoints

---
 src/public-entrypoints.test.ts | 54 ++++++++++++++++++++++++++++++++++
 1 file changed, 54 insertions(+)
 create mode 100644 src/public-entrypoints.test.ts

diff --git a/src/public-entrypoints.test.ts b/src/public-entrypoints.test.ts
new file mode 100644
index 0000000..cdd5b46
--- /dev/null
+++ b/src/public-entrypoints.test.ts
@@ -0,0 +1,54 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import { ChangeDataCapturePlugin } from '../plugins/cdc'
+import { ClerkPlugin } from '../plugins/clerk'
+import { QueryLogPlugin } from '../plugins/query-log'
+import { ResendPlugin } from '../plugins/resend'
+import { SqlMacrosPlugin } from '../plugins/sql-macros'
+import { StripeSubscriptionPlugin } from '../plugins/stripe'
+import { StudioPlugin } from '../plugins/studio'
+import { WebSocketPlugin } from '../plugins/websocket'
+import * as publicApi from '../dist'
+import * as pluginApi from '../dist/plugins'
+import { StarbaseDBDurableObject } from './do'
+import { StarbaseDB } from './handler'
+
+vi.mock('cloudflare:workers', () => {
+    return {
+        DurableObject: class MockDurableObject {},
+    }
+})
+
+describe('public package entrypoints', () => {
+    it('exposes runtime APIs from the root package export', () => {
+        expect(publicApi.StarbaseDB).toBe(StarbaseDB)
+        expect(publicApi.StarbaseDBDurableObject).toBe(StarbaseDBDurableObject)
+        expect(Object.keys(publicApi).sort()).toEqual([
+            'StarbaseDB',
+            'StarbaseDBDurableObject',
+        ])
+    })
+
+    it('exposes documented plugin constructors from the plugin export', () => {
+        expect(pluginApi.StudioPlugin).toBe(StudioPlugin)
+        expect(pluginApi.WebSocketPlugin).toBe(WebSocketPlugin)
+        expect(pluginApi.SqlMacrosPlugin).toBe(SqlMacrosPlugin)
+        expect(pluginApi.StripeSubscriptionPlugin).toBe(
+            StripeSubscriptionPlugin
+        )
+        expect(pluginApi.ChangeDataCapturePlugin).toBe(ChangeDataCapturePlugin)
+        expect(pluginApi.QueryLogPlugin).toBe(QueryLogPlugin)
+        expect(pluginApi.ResendPlugin).toBe(ResendPlugin)
+        expect(pluginApi.ClerkPlugin).toBe(ClerkPlugin)
+        expect(Object.keys(pluginApi).sort()).toEqual([
+            'ChangeDataCapturePlugin',
+            'ClerkPlugin',
+            'QueryLogPlugin',
+            'ResendPlugin',
+            'SqlMacrosPlugin',
+            'StripeSubscriptionPlugin',
+            'StudioPlugin',
+            'WebSocketPlugin',
+        ])
+    })
+})

From 52c842401e4ca4af10214e5de2d75a9a02dce1fa Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Thu, 14 May 2026 14:54:42 +0530
Subject: [PATCH 5/8] test: cover JSON import validation paths

---
 src/import/json.test.ts | 139 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 139 insertions(+)

diff --git a/src/import/json.test.ts b/src/import/json.test.ts
index 04b4ed1..053dcf6 100644
--- a/src/import/json.test.ts
+++ b/src/import/json.test.ts
@@ -83,6 +83,38 @@ describe('JSON Import Module', () => {
         expect(jsonResponse.error).toContain('Invalid JSON format')
     })
 
+    it.each([
+        ['missing data', {}],
+        ['null data', { data: null }],
+        ['object data', { data: { id: 1, name: 'Alice' } }],
+    ])(
+        'should return 400 without inserts for application/json with %s',
+        async (_caseName, payload) => {
+            const request = new Request('http://localhost', {
+                method: 'POST',
+                headers: { 'Content-Type': 'application/json' },
+                body: JSON.stringify(payload),
+            })
+
+            const response = await importTableFromJsonRoute(
+                'users',
+                request,
+                mockDataSource,
+                mockConfig
+            )
+
+            expect(response.status).toBe(400)
+            expect(executeOperation).not.toHaveBeenCalled()
+            const jsonResponse = (await response.json()) as {
+                error?: string
+                result?: any
+            }
+            expect(jsonResponse.error).toBe(
+                'Invalid JSON format. Expected an object with "data" array and optional "columnMapping".'
+            )
+        }
+    )
+
     it('should return 400 if no file is uploaded in multipart form-data', async () => {
         const formData = new FormData()
 
@@ -106,6 +138,36 @@ describe('JSON Import Module', () => {
         expect(jsonResponse.error).toBe('No file uploaded')
     })
 
+    it('should return 400 if uploaded JSON file is invalid', async () => {
+        const formData = new FormData()
+        formData.set(
+            'file',
+            new File(['not json'], 'users.json', {
+                type: 'application/json',
+            })
+        )
+
+        const request = new Request('http://localhost', {
+            method: 'POST',
+            body: formData,
+        })
+
+        const response = await importTableFromJsonRoute(
+            'users',
+            request,
+            mockDataSource,
+            mockConfig
+        )
+
+        expect(response.status).toBe(400)
+        expect(executeOperation).not.toHaveBeenCalled()
+        const jsonResponse = (await response.json()) as {
+            error?: string
+            result?: any
+        }
+        expect(jsonResponse.error).toBe('Invalid file upload')
+    })
+
     it('should successfully insert valid JSON data into the table', async () => {
         vi.mocked(executeOperation).mockResolvedValue([])
 
@@ -136,6 +198,83 @@ describe('JSON Import Module', () => {
         )
     })
 
+    it('should apply column mapping when inserting JSON records', async () => {
+        vi.mocked(executeOperation).mockResolvedValue([])
+
+        const request = new Request('http://localhost', {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({
+                data: [{ fullName: 'Alice', emailAddress: 'alice@test.dev' }],
+                columnMapping: {
+                    fullName: 'name',
+                    emailAddress: 'email',
+                },
+            }),
+        })
+
+        const response = await importTableFromJsonRoute(
+            'users',
+            request,
+            mockDataSource,
+            mockConfig
+        )
+
+        expect(response.status).toBe(200)
+        expect(executeOperation).toHaveBeenCalledWith(
+            [
+                {
+                    sql: 'INSERT INTO users (name, email) VALUES (?, ?)',
+                    params: ['Alice', 'alice@test.dev'],
+                },
+            ],
+            mockDataSource,
+            mockConfig
+        )
+    })
+
+    it('should insert valid JSON data from multipart file upload', async () => {
+        vi.mocked(executeOperation).mockResolvedValue([])
+
+        const formData = new FormData()
+        formData.set(
+            'file',
+            new File(
+                [
+                    JSON.stringify({
+                        data: [{ id: 1, name: 'Alice' }],
+                    }),
+                ],
+                'users.json',
+                { type: 'application/json' }
+            )
+        )
+
+        const request = new Request('http://localhost', {
+            method: 'POST',
+            body: formData,
+        })
+
+        const response = await importTableFromJsonRoute(
+            'users',
+            request,
+            mockDataSource,
+            mockConfig
+        )
+
+        expect(response.status).toBe(200)
+        expect(executeOperation).toHaveBeenCalledWith(
+            [
+                {
+                    sql: 'INSERT INTO users (id, name) VALUES (?, ?)',
+                    params: [1, 'Alice'],
+                },
+            ],
+            mockDataSource,
+            mockConfig
+        )
+    })
+
     it('should return partial success if some inserts fail', async () => {
         vi.mocked(executeOperation)
             .mockResolvedValueOnce([])

From 49ac2f384113827481765b352f5e1ec4e3543f06 Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Tue, 8 Sep 2026 20:06:47 +0530
Subject: [PATCH 6/8] fix: reject malformed JSON import batches before database
 writes

---
 src/import/json.test.ts | 50 +++++++++++++++++++++++++++++++++++++++++
 src/import/json.ts      | 19 +++++++++++++++-
 2 files changed, 68 insertions(+), 1 deletion(-)

diff --git a/src/import/json.test.ts b/src/import/json.test.ts
index 053dcf6..041c830 100644
--- a/src/import/json.test.ts
+++ b/src/import/json.test.ts
@@ -332,4 +332,54 @@ describe('JSON Import Module', () => {
         }
         expect(jsonResponse.error).toBe('Failed to import JSON data')
     })
+
+    describe.each(['application/json', 'multipart/form-data'])(
+        '%s validation',
+        (contentType) => {
+            it.each([
+                ['null document', null],
+                ['null row', { data: [{ id: 1 }, null] }],
+                ['string row', { data: [{ id: 1 }, 'invalid'] }],
+                ['number row', { data: [{ id: 1 }, 42] }],
+                ['array row', { data: [{ id: 1 }, ['invalid']] }],
+            ])(
+                'rejects %s before any database write',
+                async (_label, payload) => {
+                    vi.mocked(executeOperation).mockResolvedValue([])
+                    const body = JSON.stringify(payload)
+                    let request: Request
+                    if (contentType === 'application/json') {
+                        request = new Request('http://localhost', {
+                            method: 'POST',
+                            headers: { 'Content-Type': contentType },
+                            body,
+                        })
+                    } else {
+                        const formData = new FormData()
+                        formData.set(
+                            'file',
+                            new File([body], 'rows.json', {
+                                type: 'application/json',
+                            })
+                        )
+                        request = new Request('http://localhost', {
+                            method: 'POST',
+                            body: formData,
+                        })
+                    }
+                    const response = await importTableFromJsonRoute(
+                        'users',
+                        request,
+                        mockDataSource,
+                        mockConfig
+                    )
+                    expect(response.status).toBe(400)
+                    expect(executeOperation).not.toHaveBeenCalled()
+                    expect(
+                        ((await response.json()) as { error: string }).error
+                    ).toContain('Invalid JSON format')
+                }
+            )
+        }
+    )
 })
diff --git a/src/import/json.ts b/src/import/json.ts
index 04039aa..9052dae 100644
--- a/src/import/json.ts
+++ b/src/import/json.ts
@@ -50,7 +50,7 @@ export async function importTableFromJsonRoute(
             return createResponse(undefined, 'Unsupported Content-Type', 400)
         }
 
-        if (!Array.isArray(jsonData.data)) {
+        if (!jsonData || !Array.isArray(jsonData.data)) {
             return createResponse(
                 undefined,
                 'Invalid JSON format. Expected an object with "data" array and optional "columnMapping".',
@@ -58,6 +58,23 @@ export async function importTableFromJsonRoute(
             )
         }
 
+        // Validate the complete batch before executing any insert. A malformed
+        // later record must not leave an earlier record partially imported.
+        if (
+            jsonData.data.some(
+                (record) =>
+                    record === null ||
+                    typeof record !== 'object' ||
+                    Array.isArray(record)
+            )
+        ) {
+            return createResponse(
+                undefined,
+                'Invalid JSON format. Each record in "data" must be an object.',
+                400
+            )
+        }
+
         const { data, columnMapping = {} } = jsonData
 
         const failedStatements: { statement: string; error: string }[] = []

From b63b1d0133f02bd0982d0402cc3f12398f05aff4 Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Wed, 9 Sep 2026 17:36:34 +0530
Subject: [PATCH 7/8] Handle cron callback failures and preserve async delivery
 lifetime

---
 plugins/cron/index.test.ts | 96 ++++++++++++++++++++++++++++++++++++++
 plugins/cron/index.ts      | 28 +++++------
 2 files changed, 110 insertions(+), 14 deletions(-)
 create mode 100644 plugins/cron/index.test.ts

diff --git a/plugins/cron/index.test.ts b/plugins/cron/index.test.ts
new file mode 100644
index 0000000..d4cb54e
--- /dev/null
+++ b/plugins/cron/index.test.ts
@@ -0,0 +1,96 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { CronPlugin } from './index'
+const events = [
+    { name: 'first', cron_tab: '* * * * *', payload: {} },
+    { name: 'second', cron_tab: '* * * * *', payload: {} },
+]
+async function route(plugin: CronPlugin) {
+    const app = { use: vi.fn(), post: vi.fn() }
+    await plugin.register(app as any)
+    return () =>
+        app.post.mock.calls[0][1]({ req: { json: async () => events } })
+}
+afterEach(() => vi.restoreAllMocks())
+describe('cron callback delivery', () => {
+    it.each(['sync', 'async'])(
+        'contains %s failures for every event and still delivers to other listeners',
+        async (mode) => {
+            const error = vi
+                .spyOn(console, 'error')
+                .mockImplementation(() => {})
+            const plugin = new CronPlugin()
+            plugin.onEvent(() => {
+                if (mode === 'sync') throw new Error('listener failed')
+                return Promise.reject(new Error('listener failed'))
+            })
+            const healthy = vi.fn()
+            plugin.onEvent(healthy)
+            const response = await (await route(plugin))()
+            expect(response.status).toBe(200)
+            expect(await response.json()).toEqual({ result: { success: true } })
+            expect(healthy.mock.calls.map(([event]) => event.name)).toEqual([
+                'first',
+                'second',
+            ])
+            expect(error).toHaveBeenCalledTimes(2)
+        }
+    )
+    it('waits for asynchronous delivery without an execution context', async () => {
+        const plugin = new CronPlugin()
+        let finish!: () => void
+        const pending = new Promise<void>((resolve) => {
+            finish = resolve
+        })
+        const done = vi.fn()
+        plugin.onEvent(async () => {
+            await pending
+            done()
+        })
+        const handler = await route(plugin)
+        let replied = false
+        const response = handler().then(() => {
+            replied = true
+        })
+        await Promise.resolve()
+        await Promise.resolve()
+        expect(replied).toBe(false)
+        finish()
+        await response
+        expect(done).toHaveBeenCalledTimes(2)
+    })
+    it('defers pending delivery through waitUntil without blocking the response', async () => {
+        const plugin = new CronPlugin()
+        let finish!: () => void
+        const pending = new Promise<void>((resolve) => {
+            finish = resolve
+        })
+        const waitUntil = vi.fn()
+        plugin.onEvent(() => pending, { waitUntil } as any)
+        const response = await (await route(plugin))()
+        expect(response.status).toBe(200)
+        expect(waitUntil).toHaveBeenCalledTimes(2)
+        finish()
+        await Promise.all(waitUntil.mock.calls.map(([promise]) => promise))
+    })
+    it('handles rejected deferred callbacks before passing them to waitUntil', async () => {
+        const error = vi.spyOn(console, 'error').mockImplementation(() => {})
+        const waitUntil = vi.fn()
+        const plugin = new CronPlugin()
+        plugin.onEvent(
+            async () => {
+                throw new Error('late failure')
+            },
+            { waitUntil } as any
+        )
+        await (
+            await route(plugin)
+        )()
+        await expect(
+            Promise.all(waitUntil.mock.calls.map(([promise]) => promise))
+        ).resolves.toEqual([undefined, undefined])
+        expect(error).toHaveBeenCalledTimes(2)
+    })
+    it('accepts a batch with no subscribers', async () => {
+        expect((await (await route(new CronPlugin()))()).status).toBe(200)
+    })
+})
diff --git a/plugins/cron/index.ts b/plugins/cron/index.ts
index 313ebbb..6ab3b51 100644
--- a/plugins/cron/index.ts
+++ b/plugins/cron/index.ts
@@ -54,7 +54,8 @@ export interface CronEventPayload {
 export class CronPlugin extends StarbasePlugin {
     public pathPrefix: string = '/cron'
     private dataSource?: DataSource
-    private eventCallbacks: ((payload: CronEventPayload) => void)[] = []
+    private eventCallbacks: ((payload: CronEventPayload) => Promise<void>)[] =
+        []
 
     constructor() {
         super('starbasedb:cron', {
@@ -73,15 +74,11 @@ export class CronPlugin extends StarbasePlugin {
         app.post(`${this.pathPrefix}/callback`, async (c) => {
             const payload = (await c.req.json()) as CronEventPayload[]
 
-            this.eventCallbacks.forEach((callback) => {
-                try {
-                    payload.forEach((element) => {
-                        callback(element)
-                    })
-                } catch (error) {
-                    console.error('Error in Cron event callback:', error)
-                }
-            })
+            await Promise.all(
+                this.eventCallbacks.flatMap((callback) =>
+                    payload.map((element) => callback(element))
+                )
+            )
 
             return createResponse({ success: true }, undefined, 200)
         })
@@ -192,10 +189,13 @@ export class CronPlugin extends StarbasePlugin {
         ctx?: ExecutionContext
     ) {
         const wrappedCallback = async (payload: CronEventPayload) => {
-            const result = callback(payload)
-            if (result instanceof Promise && ctx) {
-                ctx.waitUntil(result)
-            }
+            const delivery = Promise.resolve()
+                .then(() => callback(payload))
+                .catch((error) => {
+                    console.error('Error in Cron event callback:', error)
+                })
+            if (ctx) ctx.waitUntil(delivery)
+            else await delivery
         }
 
         this.eventCallbacks.push(wrappedCallback)

From 89c78c6dfcb1b870253a32378104954b4852e75a Mon Sep 17 00:00:00 2001
From: taherd <183945978+taherdhanera@users.noreply.github.com>
Date: Tue, 8 Sep 2026 21:28:51 +0530
Subject: [PATCH 8/8] Cover worker authorization, dispatch, imports and durable
 object behavior

---
 src/allowlist/index.test.ts    |  76 +++++++++
 src/do.behavior.test.ts        | 227 +++++++++++++++++++++++++
 src/import/csv.test.ts         |  94 +++++++++++
 src/index.test.ts              | 263 +++++++++++++++++++++++++++++
 src/operation.dispatch.test.ts | 296 +++++++++++++++++++++++++++++++++
 5 files changed, 956 insertions(+)
 create mode 100644 src/allowlist/index.test.ts
 create mode 100644 src/do.behavior.test.ts
 create mode 100644 src/import/csv.test.ts
 create mode 100644 src/index.test.ts
 create mode 100644 src/operation.dispatch.test.ts

diff --git a/src/allowlist/index.test.ts b/src/allowlist/index.test.ts
new file mode 100644
index 0000000..6784631
--- /dev/null
+++ b/src/allowlist/index.test.ts
@@ -0,0 +1,76 @@
+import { beforeEach, expect, it, vi } from 'vitest'
+import { isQueryAllowed } from './index'
+let source: any
+const config = { role: 'client' } as any
+beforeEach(() => {
+    source = {
+        source: 'internal',
+        rpc: { executeQuery: vi.fn().mockResolvedValue([]) },
+    }
+})
+const check = (sql: string, isEnabled = true, configuration = config) =>
+    isQueryAllowed({
+        sql,
+        isEnabled,
+        dataSource: source,
+        config: configuration,
+    })
+it('bypasses disabled allowlist without storage access', async () => {
+    expect(await check('SELECT 1', false)).toBe(true)
+    expect(source.rpc.executeQuery).not.toHaveBeenCalled()
+})
+it('bypasses explicit administrator', async () => {
+    expect(await check('SELECT 1', true, { role: 'admin' })).toBe(true)
+    expect(source.rpc.executeQuery).not.toHaveBeenCalled()
+})
+it('matches whitespace and trailing semicolon after source filtering', async () => {
+    source.rpc.executeQuery.mockResolvedValue([
+        { source: 'external', sql_statement: 'SELECT 9' },
+        {
+            source: 'internal',
+            sql_statement: 'SELECT id FROM users WHERE id = 1',
+        },
+    ])
+    expect(await check(' SELECT id FROM users WHERE id = 1; ')).toBe(true)
+})
+it.each([
+    'SELECT id FROM users WHERE id = 2',
+    'SELECT id, name FROM users WHERE id = 1',
+    'SELECT id FROM users',
+    'DELETE FROM users',
+    'SELECT id FROM users WHERE id IN (1,2)',
+])('rejects changed query structure or values: %s', async (sql) => {
+    source.rpc.executeQuery.mockResolvedValueOnce([
+        {
+            source: 'internal',
+            sql_statement: 'SELECT id FROM users WHERE id = 1',
+        },
+    ])
+    await expect(check(sql)).rejects.toThrow('Query not allowed')
+    expect(source.rpc.executeQuery).toHaveBeenLastCalledWith({
+        sql: 'INSERT INTO tmp_allowlist_rejections (sql_statement, source) VALUES (?, ?)',
+        params: [sql, 'internal'],
+    })
+})
+it('returns the empty-query error contract', async () => {
+    expect(await check('')).toBeInstanceOf(Error)
+})
+it('rejects malformed SQL', async () => {
+    await expect(check('not valid SQL @@')).rejects.toThrow()
+})
+it('denies when policy storage fails', async () => {
+    source.rpc.executeQuery.mockRejectedValue(new Error('offline'))
+    await expect(check('SELECT 1')).rejects.toThrow('Query not allowed')
+})
+it('keeps rejection even if audit insertion fails', async () => {
+    source.rpc.executeQuery
+        .mockResolvedValueOnce([])
+        .mockRejectedValueOnce(new Error('audit failed'))
+    await expect(check('SELECT 1')).rejects.toThrow('Query not allowed')
+})
+it('records an audit result returned by storage', async () => {
+    source.rpc.executeQuery
+        .mockResolvedValueOnce([])
+        .mockResolvedValueOnce([{ sql_statement: 'SELECT 1' }])
+    await expect(check('SELECT 1')).rejects.toThrow('Query not allowed')
+})
diff --git a/src/do.behavior.test.ts b/src/do.behavior.test.ts
new file mode 100644
index 0000000..3c01f28
--- /dev/null
+++ b/src/do.behavior.test.ts
@@ -0,0 +1,227 @@
+import { afterEach, beforeEach, expect, it, vi } from 'vitest'
+import { StarbaseDBDurableObject } from './do'
+vi.mock('cloudflare:workers', () => ({
+    DurableObject: class {
+        constructor(public ctx: any) {}
+    },
+}))
+let instance: StarbaseDBDurableObject
+let storage: any
+let ctx: any
+beforeEach(() => {
+    storage = {
+        getAlarm: vi.fn().mockResolvedValue(123),
+        setAlarm: vi.fn().mockResolvedValue(undefined),
+        deleteAlarm: vi.fn().mockResolvedValue(undefined),
+        sql: {
+            databaseSize: 4096,
+            exec: vi.fn().mockReturnValue({
+                columnNames: ['id'],
+                rowsRead: 1,
+                rowsWritten: 0,
+                raw: () => [[7]],
+                toArray: () => [{ id: 7 }],
+            }),
+        },
+    }
+    ctx = { storage, getTags: vi.fn().mockReturnValue([]) }
+    instance = new StarbaseDBDurableObject(ctx, {
+        CLIENT_AUTHORIZATION_TOKEN: 'client',
+    } as any)
+    vi.spyOn(console, 'error').mockImplementation(() => {})
+})
+afterEach(() => {
+    vi.restoreAllMocks()
+    vi.unstubAllGlobals()
+})
+it('exposes bound RPC methods', async () => {
+    const rpc = instance.init()
+    expect(await rpc.getAlarm()).toBe(123)
+    await rpc.deleteAlarm()
+    expect(storage.deleteAlarm).toHaveBeenCalledOnce()
+})
+it.each([0, new Date(0), 5000])(
+    'clamps alarm scheduling at least one second ahead: %s',
+    async (time) => {
+        vi.spyOn(Date, 'now').mockReturnValue(1000)
+        await instance.setAlarm(time)
+        expect(storage.setAlarm).toHaveBeenCalledWith(
+            Math.max(Number(time), 2000),
+            undefined
+        )
+    }
+)
+it('propagates scheduling failure', async () => {
+    storage.setAlarm.mockRejectedValue(new Error('storage'))
+    await expect(instance.setAlarm(5000)).rejects.toThrow('storage')
+})
+it('does not fetch a callback for an empty task queue', async () => {
+    vi.spyOn(instance, 'executeQuery').mockResolvedValue([])
+    const fetcher = vi.fn()
+    vi.stubGlobal('fetch', fetcher)
+    await instance.alarm()
+    expect(fetcher).not.toHaveBeenCalled()
+})
+it('sends active tasks to the callback with client authorization', async () => {
+    const tasks = [{ callback_host: 'https://callback.test', name: 'daily' }]
+    vi.spyOn(instance, 'executeQuery').mockResolvedValue(tasks)
+    const fetcher = vi.fn().mockResolvedValue(new Response())
+    vi.stubGlobal('fetch', fetcher)
+    await instance.alarm()
+    expect(fetcher).toHaveBeenCalledWith(
+        'https://callback.test/cron/callback',
+        expect.objectContaining({
+            method: 'POST',
+            body: JSON.stringify(tasks),
+            headers: expect.objectContaining({
+                Authorization: 'Bearer client',
+            }),
+        })
+    )
+})
+it.each(['callback', 'query'])(
+    'reschedules after %s failure',
+    async (stage) => {
+        vi.spyOn(Date, 'now').mockReturnValue(1000)
+        const query = vi.spyOn(instance, 'executeQuery')
+        query.mockResolvedValue([{ callback_host: 'https://callback.test' }])
+        if (stage === 'query') query.mockRejectedValue(new Error('query'))
+        vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('callback')))
+        await instance.alarm()
+        expect(storage.setAlarm).toHaveBeenCalledWith(61000, undefined)
+    }
+)
+it.each(['callback', 'query'])(
+    'handles a failed recovery alarm after %s failure',
+    async (stage) => {
+        const query = vi.spyOn(instance, 'executeQuery')
+        query.mockResolvedValue([{ callback_host: 'https://callback.test' }])
+        if (stage === 'query') query.mockRejectedValue(new Error('query'))
+        storage.setAlarm.mockRejectedValue(new Error('retry'))
+        vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('callback')))
+        await expect(instance.alarm()).resolves.toBeUndefined()
+        expect(console.error).toHaveBeenCalledWith(
+            'Failed to set recovery alarm:',
+            expect.any(Error)
+        )
+    }
+)
+it.each([[{ count: '3' }], []].map((rows) => ({ rows })))(
+    'reports statistics with available or empty query history %j',
+    async ({ rows }) => {
+        vi.spyOn(instance, 'executeQuery').mockResolvedValue(rows)
+        instance.connections.set('one', {} as any)
+        expect(await instance.getStatistics()).toEqual({
+            databaseSize: 4096,
+            activeConnections: 1,
+            recentQueries: rows.length ? 3 : 0,
+        })
+    }
+)
+it('rejects a non-websocket socket request', async () => {
+    expect(
+        (await instance.fetch(new Request('https://test/socket'))).status
+    ).toBe(400)
+})
+it.each(['?sessionId=s1', ''])(
+    'passes optional session ID on upgrades %s',
+    async (suffix) => {
+        const connect = vi
+            .spyOn(instance, 'clientConnected')
+            .mockResolvedValue(new Response('upgrade'))
+        await instance.fetch(
+            new Request('https://test/socket' + suffix, {
+                headers: { upgrade: 'websocket' },
+            })
+        )
+        expect(connect).toHaveBeenCalledWith(suffix ? 's1' : undefined)
+    }
+)
+it('targets one session and removes dead broadcast connections', async () => {
+    const selected = { send: vi.fn() },
+        other = { send: vi.fn() },
+        dead = {
+            send: vi.fn().mockImplementation(() => {
+                throw new Error('closed')
+            }),
+        }
+    instance.connections.set('selected', selected as any)
+    instance.connections.set('other', other as any)
+    await instance.fetch(
+        new Request('https://test/socket/broadcast?sessionId=selected', {
+            method: 'POST',
+            body: '{"event":1}',
+        })
+    )
+    expect(selected.send).toHaveBeenCalledWith('{"event":1}')
+    expect(other.send).not.toHaveBeenCalled()
+    instance.connections.set('dead', dead as any)
+    await instance.fetch(
+        new Request('https://test/socket/broadcast', {
+            method: 'POST',
+            body: '{}',
+        })
+    )
+    expect(other.send).toHaveBeenCalledWith('{}')
+    expect(instance.connections.has('dead')).toBe(false)
+})
+it('executes websocket query messages and ignores other actions', async () => {
+    const execute = vi
+        .spyOn(instance, 'executeTransaction')
+        .mockResolvedValue([{ id: 1 }])
+    const ws = { send: vi.fn() } as any
+    await instance.webSocketMessage(
+        ws,
+        JSON.stringify({ action: 'query', sql: 'SELECT ?', params: [1] })
+    )
+    expect(execute).toHaveBeenCalledWith(
+        [{ sql: 'SELECT ?', params: [1] }],
+        false
+    )
+    expect(ws.send).toHaveBeenCalledWith('[{"id":1}]')
+    await instance.webSocketMessage(ws, '{"action":"ping"}')
+    expect(execute).toHaveBeenCalledOnce()
+})
+it.each([[], ['session']].map((tags) => ({ tags })))(
+    'cleans tagged sockets on close %j',
+    async ({ tags }) => {
+        ctx.getTags.mockReturnValue(tags)
+        instance.connections.set('session', {} as any)
+        const ws = { close: vi.fn() } as any
+        await instance.webSocketClose(ws, 1000, 'done', true)
+        expect(ws.close).toHaveBeenCalledWith(
+            1000,
+            'StarbaseDB is closing WebSocket connection'
+        )
+        expect(instance.connections.has('session')).toBe(tags.length === 0)
+    }
+)
+it.each([undefined, [], [7]].map((params) => ({ params })))(
+    'returns raw rows and forwards optional parameters %j',
+    async ({ params }) => {
+        expect(
+            await instance.executeQuery({
+                sql: 'SELECT ?',
+                params,
+                isRaw: true,
+            })
+        ).toEqual({
+            columns: ['id'],
+            rows: [[7]],
+            meta: { rows_read: 1, rows_written: 0 },
+        })
+        expect(storage.sql.exec).toHaveBeenLastCalledWith(
+            'SELECT ?',
+            ...(params || [])
+        )
+    }
+)
+it('stops transaction processing on query failure', async () => {
+    const execute = vi
+        .spyOn(instance, 'executeQuery')
+        .mockRejectedValue(new Error('bad query'))
+    await expect(
+        instance.executeTransaction([{ sql: 'bad' }, { sql: 'later' }], false)
+    ).rejects.toThrow('bad query')
+    expect(execute).toHaveBeenCalledOnce()
+})
diff --git a/src/import/csv.test.ts b/src/import/csv.test.ts
new file mode 100644
index 0000000..53d3c87
--- /dev/null
+++ b/src/import/csv.test.ts
@@ -0,0 +1,94 @@
+import { beforeEach, expect, it, vi } from 'vitest'
+import { importTableFromCsvRoute } from './csv'
+import { executeOperation } from '../export'
+vi.mock('../export', () => ({ executeOperation: vi.fn() }))
+const run = (request: Request) =>
+    importTableFromCsvRoute('users', request, {} as any, {} as any)
+const req = (body: string, type = 'text/csv') =>
+    new Request('https://test/import', {
+        method: 'POST',
+        headers: { 'Content-Type': type },
+        body,
+    })
+beforeEach(() => {
+    vi.mocked(executeOperation).mockReset().mockResolvedValue([])
+})
+it('rejects a missing body', async () => {
+    expect((await run(new Request('https://test'))).status).toBe(400)
+    expect(executeOperation).not.toHaveBeenCalled()
+})
+it.each(['text/plain', ''])(
+    'rejects unsupported content type %s',
+    async (type) => {
+        expect((await run(req('id\n1', type))).status).toBe(400)
+        expect(executeOperation).not.toHaveBeenCalled()
+    }
+)
+it('maps JSON-wrapped CSV columns and parameterizes values', async () => {
+    const response = await run(
+        req(
+            JSON.stringify({
+                data: ' name , age\n Alice , 20',
+                columnMapping: { name: 'full_name' },
+            }),
+            'application/json'
+        )
+    )
+    expect(response.status).toBe(200)
+    expect(executeOperation).toHaveBeenCalledWith(
+        [
+            {
+                sql: 'INSERT INTO users (full_name, age) VALUES (?, ?)',
+                params: ['Alice', '20'],
+            },
+        ],
+        {},
+        {}
+    )
+})
+it('imports raw CSV and ignores rows with mismatched field counts', async () => {
+    expect((await run(req('id,name\n1,Alice\ninvalid'))).status).toBe(200)
+    expect(executeOperation).toHaveBeenCalledTimes(1)
+})
+it('rejects headers without records', async () => {
+    expect((await run(req('id,name'))).status).toBe(400)
+    expect(executeOperation).not.toHaveBeenCalled()
+})
+it('requires a multipart file', async () => {
+    const form = new FormData()
+    form.set('other', 'x')
+    expect(
+        (await run(new Request('https://test', { method: 'POST', body: form })))
+            .status
+    ).toBe(400)
+})
+it('imports an uploaded file', async () => {
+    const form = new FormData()
+    form.set(
+        'file',
+        new Blob(['id,name\n1,Alice'], { type: 'text/csv' }),
+        'users.csv'
+    )
+    expect(
+        (await run(new Request('https://test', { method: 'POST', body: form })))
+            .status
+    ).toBe(200)
+    expect(executeOperation).toHaveBeenCalledOnce()
+})
+it.each([new Error('write failed'), {}])(
+    'reports partial failures: %j',
+    async (error) => {
+        vi.mocked(executeOperation)
+            .mockRejectedValueOnce(error)
+            .mockResolvedValueOnce([])
+        const response = await run(req('id\n1\n2'))
+        const body = (await response.json()) as any
+        expect(response.status).toBe(200)
+        expect(JSON.stringify(body)).toContain('Imported 1 out of 2')
+        expect(JSON.stringify(body)).toContain('1 records failed')
+    }
+)
+it('reports malformed JSON', async () => {
+    expect((await run(req('{', 'application/json'))).status).toBe(500)
+    expect(executeOperation).not.toHaveBeenCalled()
+})
diff --git a/src/index.test.ts b/src/index.test.ts
new file mode 100644
index 0000000..67956cf
--- /dev/null
+++ b/src/index.test.ts
@@ -0,0 +1,263 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import worker from './index'
+import { StarbaseDB } from './handler'
+import { jwtVerify } from 'jose'
+const state = {
+    handle: vi.fn(),
+    pre: vi.fn(),
+    route: vi.fn(),
+    options: null as any,
+}
+vi.mock('./do', () => ({ StarbaseDBDurableObject: class {} }))
+vi.mock('./handler', () => ({
+    StarbaseDB: vi.fn().mockImplementation((options) => {
+        state.options = options
+        return { handle: state.handle, handlePreAuth: state.pre }
+    }),
+}))
+vi.mock('jose', () => ({ createRemoteJWKSet: vi.fn(), jwtVerify: vi.fn() }))
+vi.mock('../plugins/websocket', () => ({ WebSocketPlugin: class {} }))
+vi.mock('../plugins/studio', () => ({ StudioPlugin: class {} }))
+vi.mock('../plugins/sql-macros', () => ({ SqlMacrosPlugin: class {} }))
+vi.mock('../plugins/query-log', () => ({ QueryLogPlugin: class {} }))
+vi.mock('../plugins/stats', () => ({ StatsPlugin: class {} }))
+vi.mock('../plugins/cdc', () => ({
+    ChangeDataCapturePlugin: class {
+        onEvent = vi.fn()
+    },
+}))
+vi.mock('../plugins/cron', () => ({
+    CronPlugin: class {
+        onEvent = vi.fn()
+    },
+}))
+vi.mock('../plugins/interface', () => ({
+    InterfacePlugin: class {
+        matchesRoute = state.route
+    },
+}))
+let env: any
+let stub: any
+const ctx = { waitUntil: vi.fn() } as any
+const request = (headers = {}, suffix = '') =>
+    new Request('https://worker.test/query' + suffix, { headers })
+beforeEach(() => {
+    vi.clearAllMocks()
+    state.pre.mockResolvedValue(undefined)
+    state.route.mockReturnValue(false)
+    state.handle.mockResolvedValue(new Response('handled'))
+    stub = { init: vi.fn().mockResolvedValue({ executeQuery: vi.fn() }) }
+    env = {
+        ADMIN_AUTHORIZATION_TOKEN: 'admin',
+        CLIENT_AUTHORIZATION_TOKEN: 'client',
+        DATABASE_DURABLE_OBJECT: {
+            idFromName: vi.fn().mockReturnValue('id'),
+            get: vi.fn().mockReturnValue(stub),
+        },
+    }
+})
+describe('Worker authentication and routing', () => {
+    it('answers preflight before opening a durable object', async () => {
+        expect(
+            (
+                await worker.fetch(
+                    new Request('https://worker.test', { method: 'OPTIONS' }),
+                    env,
+                    ctx
+                )
+            ).status
+        ).toBe(204)
+        expect(stub.init).not.toHaveBeenCalled()
+    })
+    it('rejects missing credentials before handling a query', async () => {
+        expect((await worker.fetch(request(), env, ctx)).status).toBe(401)
+        expect(state.handle).not.toHaveBeenCalled()
+    })
+    it.each([
+        ['admin', 'admin'],
+        ['client', 'client'],
+    ])('assigns the %s role', async (token, role) => {
+        expect(
+            (
+                await worker.fetch(
+                    request({ Authorization: 'Bearer ' + token }),
+                    env,
+                    ctx
+                )
+            ).status
+        ).toBe(200)
+        expect(state.options.config.role).toBe(role)
+    })
+    it('rejects an unknown token without a JWT provider', async () => {
+        expect(
+            (
+                await worker.fetch(
+                    request({ Authorization: 'Bearer wrong' }),
+                    env,
+                    ctx
+                )
+            ).status
+        ).toBe(400)
+        expect(state.handle).not.toHaveBeenCalled()
+    })
+    it.each([undefined, 'RS256'])(
+        'validates JWTs with optional configured algorithm %s',
+        async (algorithm) => {
+            env.AUTH_JWKS_ENDPOINT = 'https://issuer.test/jwks'
+            env.AUTH_ALGORITHM = algorithm
+            vi.mocked(jwtVerify).mockResolvedValue({
+                payload: { sub: 'user' },
+            } as any)
+            expect(
+                (
+                    await worker.fetch(
+                        request({ Authorization: 'Bearer jwt' }),
+                        env,
+                        ctx
+                    )
+                ).status
+            ).toBe(200)
+            expect(jwtVerify).toHaveBeenCalledWith('jwt', undefined, {
+                algorithms: algorithm ? [algorithm] : undefined,
+            })
+        }
+    )
+    it('rejects a JWT without a subject', async () => {
+        env.AUTH_JWKS_ENDPOINT = 'https://issuer.test/jwks'
+        vi.mocked(jwtVerify).mockResolvedValue({ payload: {} } as any)
+        expect(
+            (
+                await worker.fetch(
+                    request({ Authorization: 'Bearer jwt' }),
+                    env,
+                    ctx
+                )
+            ).status
+        ).toBe(400)
+        expect(state.handle).not.toHaveBeenCalled()
+    })
+    it('handles verifier rejection without an error message', async () => {
+        env.AUTH_JWKS_ENDPOINT = 'https://issuer.test/jwks'
+        vi.mocked(jwtVerify).mockRejectedValue(null)
+        const response = await worker.fetch(
+            request({ Authorization: 'Bearer jwt' }),
+            env,
+            ctx
+        )
+        expect(response.status).toBe(400)
+        expect(await response.text()).toContain('Unable to process request')
+    })
+    it('accepts websocket query credentials', async () => {
+        expect(
+            (
+                await worker.fetch(
+                    request({ Upgrade: 'websocket' }, '?token=client'),
+                    env,
+                    ctx
+                )
+            ).status
+        ).toBe(200)
+        expect(state.handle).toHaveBeenCalledOnce()
+    })
+    it('rejects a websocket without a token', async () => {
+        expect(
+            (await worker.fetch(request({ Upgrade: 'websocket' }), env, ctx))
+                .status
+        ).toBe(401)
+    })
+    it('honors a plugin pre-auth response', async () => {
+        state.pre.mockResolvedValue(new Response('plugin', { status: 202 }))
+        expect((await worker.fetch(request(), env, ctx)).status).toBe(202)
+        expect(state.handle).not.toHaveBeenCalled()
+    })
+    it('delegates a matching public interface route', async () => {
+        state.route.mockReturnValue(true)
+        expect((await worker.fetch(request(), env, ctx)).status).toBe(200)
+        expect(state.handle).toHaveBeenCalledOnce()
+    })
+    it.each([new Error('offline'), 'offline'])(
+        'reports initialization failures %s',
+        async (error) => {
+            stub.init.mockRejectedValue(error)
+            expect((await worker.fetch(request(), env, ctx)).status).toBe(400)
+            expect(state.handle).not.toHaveBeenCalled()
+        }
+    )
+})
+describe('Worker data source configuration', () => {
+    it.each([
+        ['external', 'external'],
+        [' HYPERDRIVE ', 'hyperdrive'],
+        ['unknown', 'internal'],
+    ])('normalizes source %s', async (source, expected) => {
+        await worker.fetch(
+            request({
+                'X-Starbase-Source': source,
+                Authorization: 'Bearer client',
+            }),
+            env,
+            ctx
+        )
+        expect(state.options.dataSource.source).toBe(expected)
+    })
+    it('uses query source, cache flag and region hint', async () => {
+        env.REGION = 'weur'
+        await worker.fetch(
+            request(
+                { 'X-Starbase-Cache': 'true', Authorization: 'Bearer client' },
+                '?source=external'
+            ),
+            env,
+            ctx
+        )
+        expect(state.options.dataSource.source).toBe('external')
+        expect(state.options.dataSource.cache).toBe(true)
+        expect(env.DATABASE_DURABLE_OBJECT.get).toHaveBeenCalledWith('id', {
+            locationHint: 'weur',
+        })
+    })
+    it.each(['postgresql', 'mysql'])(
+        'passes %s connection configuration',
+        async (dialect) => {
+            Object.assign(env, {
+                EXTERNAL_DB_TYPE: dialect,
+                EXTERNAL_DB_HOST: 'db.test',
+                EXTERNAL_DB_DEFAULT_SCHEMA: 'tenant',
+            })
+            await worker.fetch(
+                request({ Authorization: 'Bearer client' }),
+                env,
+                ctx
+            )
+            expect(state.options.dataSource.external).toMatchObject({
+                dialect,
+                host: 'db.test',
+                defaultSchema: 'tenant',
+            })
+        }
+    )
+    it.each([
+        [{ EXTERNAL_DB_CLOUDFLARE_API_KEY: 'key' }, 'cloudflare-d1'],
+        [{ EXTERNAL_DB_STARBASEDB_URI: 'https://db.test' }, 'starbase'],
+        [{ EXTERNAL_DB_TURSO_URI: 'libsql://db.test' }, 'turso'],
+    ])('selects the SQLite provider %j', async (extra, provider) => {
+        Object.assign(env, { EXTERNAL_DB_TYPE: 'sqlite' }, extra)
+        await worker.fetch(
+            request({ Authorization: 'Bearer client' }),
+            env,
+            ctx
+        )
+        expect(state.options.dataSource.external.provider).toBe(provider)
+    })
+    it('uses the Hyperdrive binding when present', async () => {
+        env.HYPERDRIVE = { connectionString: 'postgres://db.test' }
+        await worker.fetch(
+            request({ Authorization: 'Bearer client' }),
+            env,
+            ctx
+        )
+        expect(state.options.dataSource.external.connectionString).toBe(
+            'postgres://db.test'
+        )
+    })
+})
diff --git a/src/operation.dispatch.test.ts b/src/operation.dispatch.test.ts
new file mode 100644
index 0000000..e37d5a7
--- /dev/null
+++ b/src/operation.dispatch.test.ts
@@ -0,0 +1,296 @@
+import { afterEach, beforeEach, expect, it, vi } from 'vitest'
+import {
+    executeQuery,
+    executeExternalQuery,
+    executeSDKQuery,
+    executeTransaction,
+} from './operation'
+import { applyRLS } from './rls'
+import { beforeQueryCache, afterQueryCache } from './cache'
+import { Client as PgClient } from 'pg'
+import { createConnection } from 'mysql2'
+import { createClient } from '@libsql/client/web'
+import postgres from 'postgres'
+import {
+    PostgreSQLConnection,
+    MySQLConnection,
+    TursoConnection,
+    CloudflareD1Connection,
+    StarbaseConnection,
+} from '@outerbase/sdk'
+vi.mock('pg', () => ({ Client: vi.fn() }))
+vi.mock('mysql2', () => ({ createConnection: vi.fn() }))
+vi.mock('@libsql/client/web', () => ({ createClient: vi.fn() }))
+vi.mock('postgres', () => ({ default: vi.fn() }))
+vi.mock('@outerbase/sdk', () => ({
+    PostgreSQLConnection: vi.fn(),
+    MySQLConnection: vi.fn(),
+    TursoConnection: vi.fn(),
+    CloudflareD1Connection: vi.fn(),
+    StarbaseConnection: vi.fn(),
+}))
+const drivers = {
+    connect: vi.fn(),
+    raw: vi.fn(),
+    unsafe: vi.fn(),
+    end: vi.fn(),
+    pg: vi.mocked(PgClient),
+    mysql: vi.mocked(createConnection),
+    turso: vi.mocked(createClient),
+    hyper: vi.mocked(postgres),
+}
+vi.mock('./allowlist', () => ({
+    isQueryAllowed: vi.fn().mockResolvedValue(true),
+}))
+vi.mock('./rls', () => ({ applyRLS: vi.fn() }))
+vi.mock('./cache', () => ({
+    beforeQueryCache: vi.fn(),
+    afterQueryCache: vi.fn(),
+}))
+let source: any
+let config: any
+const run = (extra = {}) =>
+    executeQuery({
+        sql: 'SELECT ?',
+        params: [1],
+        isRaw: false,
+        dataSource: source,
+        config,
+        ...extra,
+    })
+beforeEach(() => {
+    vi.clearAllMocks()
+    for (const Connection of [
+        PostgreSQLConnection,
+        MySQLConnection,
+        TursoConnection,
+        CloudflareD1Connection,
+        StarbaseConnection,
+    ]) {
+        vi.mocked(Connection).mockImplementation(
+            () => ({ connect: drivers.connect, raw: drivers.raw }) as any
+        )
+    }
+    source = {
+        source: 'internal',
+        rpc: { executeQuery: vi.fn().mockResolvedValue([{ id: 1 }]) },
+    }
+    config = { role: 'client', features: { allowlist: false, rls: false } }
+    vi.mocked(applyRLS).mockImplementation(async ({ sql }) => sql)
+    vi.mocked(beforeQueryCache).mockResolvedValue(null)
+    vi.mocked(afterQueryCache).mockResolvedValue(undefined)
+    drivers.connect.mockResolvedValue(undefined)
+    drivers.raw.mockResolvedValue({ data: [{ id: 2 }] })
+    drivers.unsafe.mockResolvedValue([{ id: 3 }])
+    drivers.end.mockResolvedValue(undefined)
+    drivers.hyper.mockReturnValue({ unsafe: drivers.unsafe, end: drivers.end })
+    vi.spyOn(console, 'error').mockImplementation(() => {})
+})
+afterEach(() => {
+    vi.restoreAllMocks()
+    vi.unstubAllGlobals()
+})
+it('returns empty results when source is absent', async () => {
+    expect(await run({ dataSource: undefined })).toEqual([])
+    expect(
+        await executeTransaction({
+            queries: [],
+            isRaw: false,
+            dataSource: undefined as any,
+            config,
+        })
+    ).toEqual([])
+})
+it('uses feature defaults when configuration is absent', async () => {
+    await run({ config: undefined })
+    expect(applyRLS).toHaveBeenCalledWith(
+        expect.objectContaining({ isEnabled: true })
+    )
+})
+it('returns a cache hit without querying the database', async () => {
+    vi.mocked(beforeQueryCache).mockResolvedValue([{ cached: true }])
+    expect(await run()).toEqual([{ cached: true }])
+    expect(source.rpc.executeQuery).not.toHaveBeenCalled()
+})
+it('passes query hook rewrites to storage and result hooks to the caller', async () => {
+    source.registry = {
+        beforeQuery: vi
+            .fn()
+            .mockResolvedValue({ sql: 'SELECT 2', params: [2] }),
+        afterQuery: vi.fn().mockResolvedValue([{ hooked: true }]),
+    }
+    expect(await run()).toEqual([{ hooked: true }])
+    expect(source.rpc.executeQuery).toHaveBeenCalledWith({
+        sql: 'SELECT 2',
+        params: [2],
+        isRaw: false,
+    })
+    expect(afterQueryCache).toHaveBeenCalledOnce()
+})
+it('preserves results if an after-query hook fails', async () => {
+    source.registry = {
+        beforeQuery: vi.fn().mockResolvedValue({ sql: 'SELECT 1' }),
+        afterQuery: vi.fn().mockRejectedValue(new Error('hook')),
+    }
+    expect(await run()).toEqual([{ id: 1 }])
+})
+it('returns empty results for a null internal result', async () => {
+    source.rpc.executeQuery.mockResolvedValue(null)
+    expect(await run()).toEqual([])
+})
+it.each(
+    [[{ id: 1 }], { columns: ['id'], rows: [[1]] }, []].map((result) => ({
+        result,
+    }))
+)('normalizes raw results without cache access %j', async ({ result }) => {
+    source.rpc.executeQuery.mockResolvedValue(result)
+    const actual = (await run({ isRaw: true })) as any
+    expect(actual.columns).toEqual(
+        Array.isArray(result) && result.length === 0 ? [] : ['id']
+    )
+    expect(beforeQueryCache).not.toHaveBeenCalled()
+})
+it('returns an empty raw result for missing rows', async () => {
+    source.rpc.executeQuery.mockResolvedValue({ columns: [] })
+    expect(await run({ isRaw: true })).toEqual({
+        columns: [],
+        rows: [],
+        meta: { rows_read: 0, rows_written: 0 },
+    })
+})
+it.each([undefined, {}])(
+    'requires a Hyperdrive connection string %j',
+    async (external) => {
+        source.source = 'hyperdrive'
+        source.external = external
+        await expect(run()).rejects.toThrow(
+            'Hyperdrive connection string not found'
+        )
+    }
+)
+it.each([false, true])(
+    'closes Hyperdrive connections with execution context %s',
+    async (withContext) => {
+        source.source = 'hyperdrive'
+        source.external = { connectionString: 'postgres://test' }
+        if (withContext) source.executionContext = { waitUntil: vi.fn() }
+        expect(await run()).toEqual([{ id: 3 }])
+        expect(drivers.unsafe).toHaveBeenCalledWith('SELECT ?', [1])
+        expect(drivers.end).toHaveBeenCalledOnce()
+    }
+)
+it('propagates Hyperdrive query errors', async () => {
+    source.source = 'hyperdrive'
+    source.external = { connectionString: 'postgres://test' }
+    drivers.unsafe.mockRejectedValue(new Error('query failed'))
+    await expect(run()).rejects.toThrow('query failed')
+})
+it('dispatches an external request through the SDK', async () => {
+    source.source = 'external'
+    source.external = { dialect: 'postgresql' }
+    expect(await run()).toEqual([{ id: 2 }])
+})
+it('executes transaction queries in order', async () => {
+    await executeTransaction({
+        queries: [{ sql: 'SELECT 1' }, { sql: 'SELECT 2', params: [2] }],
+        isRaw: false,
+        dataSource: source,
+        config,
+    })
+    expect(
+        source.rpc.executeQuery.mock.calls.map((c: any) => c[0].sql)
+    ).toEqual(['SELECT 1', 'SELECT 2'])
+})
+it('requires external connection information', async () => {
+    await expect(
+        executeExternalQuery({
+            sql: 'SELECT 1',
+            params: [],
+            dataSource: source,
+            config,
+        })
+    ).rejects.toThrow('External connection not found')
+})
+it.each([[1, 2], { named: 1 }].map((params) => ({ params })))(
+    'formats Outerbase API requests with parameters %j',
+    async ({ params }) => {
+        source.external = { dialect: 'mysql' }
+        config.outerbaseApiKey = 'test-token'
+        const fetcher = vi.fn().mockResolvedValue({
+            json: async () => ({
+                response: { results: { items: [{ id: 8 }] } },
+            }),
+        })
+        vi.stubGlobal('fetch', fetcher)
+        expect(
+            await executeExternalQuery({
+                sql: 'SELECT ?\n, ?',
+                params,
+                dataSource: source,
+                config,
+            })
+        ).toEqual([{ id: 8 }])
+        const body = JSON.parse(fetcher.mock.calls[0][1].body)
+        expect(body.query).not.toContain('\n')
+        expect(body.params).toEqual(
+            Array.isArray(params) ? { param0: 1, param1: 2 } : params
+        )
+    }
+)
+it.each([null, {}, { response: {} }, { response: { results: {} } }])(
+    'handles malformed API response %j',
+    async (result) => {
+        source.external = { dialect: 'mysql' }
+        config.outerbaseApiKey = 'test-token'
+        vi.stubGlobal(
+            'fetch',
+            vi.fn().mockResolvedValue({ json: async () => result })
+        )
+        expect(
+            await executeExternalQuery({
+                sql: 'SELECT 1',
+                params: undefined,
+                dataSource: source,
+                config,
+            })
+        ).toEqual([])
+    }
+)
+it('returns no SDK results without a connection', async () => {
+    expect(
+        await executeSDKQuery({ sql: 'SELECT 1', dataSource: source, config })
+    ).toEqual([])
+})
+it.each([
+    { dialect: 'postgresql' },
+    { dialect: 'postgresql', defaultSchema: 'custom' },
+    { dialect: 'mysql' },
+    { dialect: 'mysql', defaultSchema: 'custom' },
+    { dialect: 'sqlite', provider: 'turso', uri: 'libsql://test' },
+    { dialect: 'sqlite', provider: 'turso', defaultSchema: 'custom' },
+    { dialect: 'sqlite', provider: 'cloudflare-d1' },
+    { dialect: 'sqlite', provider: 'cloudflare-d1', defaultSchema: 'custom' },
+    { dialect: 'sqlite', provider: 'starbase' },
+    { dialect: 'sqlite', provider: 'starbase', defaultSchema: 'custom' },
+])(
+    'connects and runs through the supported SDK provider %j',
+    async (external) => {
+        source.external = external
+        expect(
+            await executeSDKQuery({
+                sql: 'SELECT ?',
+                params: [7],
+                dataSource: source,
+                config,
+            })
+        ).toEqual([{ id: 2 }])
+        expect(drivers.connect).toHaveBeenCalledOnce()
+        expect(drivers.raw).toHaveBeenCalledWith('SELECT ?', [7])
+    }
+)
+it('rejects unsupported SDK providers', async () => {
+    source.external = { dialect: 'sqlite', provider: 'other' }
+    await expect(
+        executeSDKQuery({ sql: 'SELECT 1', dataSource: source, config })
+    ).rejects.toThrow('Unsupported external database type')
+})