feat(plugins): add ExternalReplicationPlugin to pull & sync external data into DO SQLite (#72)

plugins/replicator/index.test.ts

@@ -0,0 +1,41 @@
+import { describe, it, expect, vi } from 'vitest'
+import { ExternalReplicationPlugin } from './index'
+
+describe('ExternalReplicationPlugin (#72)', () => {
+    it('pulls external table data, creates internal SQLite schema, and advances watermark cursor', async () => {
+        const externalRows = [
+            { id: 1, email: 'a@example.com', updated_at: '2026-11-11T10:00:00Z' },
+            { id: 2, email: 'b@example.com', updated_at: '2026-11-11T11:00:00Z' },
+        ]
+        const externalDs: any = {
+            query: vi.fn(async ({ sql, params }: any) => {
+                if (params && params.length > 0 && params[0] === '2026-11-11T11:00:00Z') {
+                    return { rows: [] }
+                }
+                return { rows: externalRows }
+            }),
+        }
+        const executedInternalQueries: string[] = []
+        const internalDs: any = {
+            query: vi.fn(async ({ sql }: any) => {
+                executedInternalQueries.push(sql)
+                return { rows: [] }
+            }),
+        }
+
+        const plugin = new ExternalReplicationPlugin({
+            externalDataSource: externalDs,
+            internalDataSource: internalDs,
+            tables: [{ table: 'customers', primaryKey: 'id', watermarkColumn: 'updated_at' }],
+        })
+
+        const firstRun = await plugin.syncAll()
+        expect(firstRun.tables[0].rowsSynced).toBe(2)
+        expect(firstRun.tables[0].lastWatermark).toBe('2026-11-11T11:00:00Z')
+        expect(executedInternalQueries[0]).toContain('CREATE TABLE IF NOT EXISTS "customers"')
+        expect(executedInternalQueries[1]).toContain('INSERT OR REPLACE INTO "customers"')
+
+        const secondRun = await plugin.syncAll()
+        expect(secondRun.tables[0].rowsSynced).toBe(0)
+    })
+})

plugins/replicator/index.ts

@@ -0,0 +1,129 @@
+import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler'
+import { StarbasePlugin } from '../../src/plugin'
+import { DataSource } from '../../src/types'
+
+export interface ReplicationTableSpec {
+    table: string
+    primaryKey: string
+    watermarkColumn?: string
+    batchSize?: number
+}
+
+export interface ExternalReplicationConfig {
+    externalDataSource: DataSource
+    internalDataSource: DataSource
+    tables: ReplicationTableSpec[]
+    config?: StarbaseDBConfiguration
+}
+
+function inferSqliteColumnType(val: any): string {
+    if (typeof val === 'number') {
+        return Number.isInteger(val) ? 'INTEGER' : 'REAL'
+    }
+    if (typeof val === 'boolean') {
+        return 'INTEGER'
+    }
+    return 'TEXT'
+}
+
+export class ExternalReplicationPlugin extends StarbasePlugin {
+    public prefix: string = '/replicator'
+    private replicationConfig: ExternalReplicationConfig
+    private watermarks: Record<string, string | number> = {}
+
+    constructor(replicationConfig: ExternalReplicationConfig) {
+        super('starbasedb:external-replicator', {
+            requiresAuth: true,
+        })
+        this.replicationConfig = replicationConfig
+    }
+
+    public async syncTable(spec: ReplicationTableSpec): Promise<{
+        table: string
+        rowsSynced: number
+        lastWatermark: string | number | null
+    }> {
+        const { externalDataSource, internalDataSource } = this.replicationConfig
+        const batchSize = Math.max(1, spec.batchSize ?? 500)
+        const lastWatermark = this.watermarks[spec.table] ?? null
+
+        let pullSql = `SELECT * FROM "${spec.table.replace(/"/g, '""')}"`
+        const params: any[] = []
+        if (spec.watermarkColumn && lastWatermark !== null) {
+            pullSql += ` WHERE "${spec.watermarkColumn.replace(/"/g, '""')}" > ?`
+            params.push(lastWatermark)
+        }
+        if (spec.watermarkColumn) {
+            pullSql += ` ORDER BY "${spec.watermarkColumn.replace(/"/g, '""')}" ASC`
+        }
+        pullSql += ` LIMIT ${batchSize};`
+
+        const extRes: any = await externalDataSource.query({ sql: pullSql, params })
+        const rows: Record<string, any>[] = extRes?.rows ?? (Array.isArray(extRes) ? extRes : [])
+
+        if (rows.length === 0) {
+            return { table: spec.table, rowsSynced: 0, lastWatermark }
+        }
+
+        const sample = rows[0]
+        const columns = Object.keys(sample)
+        const colDefs = columns.map((col) => {
+            const colType = inferSqliteColumnType(sample[col])
+            const pk = col === spec.primaryKey ? ' PRIMARY KEY' : ''
+            return `"${col.replace(/"/g, '""')}" ${colType}${pk}`
+        })
+
+        await internalDataSource.query({
+            sql: `CREATE TABLE IF NOT EXISTS "${spec.table.replace(/"/g, '""')}" (${colDefs.join(', ')});`,
+            params: [],
+        })
+
+        let newestWatermark = lastWatermark
+        for (const row of rows) {
+            const placeholders = columns.map(() => '?').join(', ')
+            const colNames = columns.map((c) => `"${c.replace(/"/g, '""')}"`).join(', ')
+            const values = columns.map((c) => row[c])
+            await internalDataSource.query({
+                sql: `INSERT OR REPLACE INTO "${spec.table.replace(/"/g, '""')}" (${colNames}) VALUES (${placeholders});`,
+                params: values,
+            })
+            if (spec.watermarkColumn && row[spec.watermarkColumn] !== undefined) {
+                newestWatermark = row[spec.watermarkColumn]
+            }
+        }
+
+        if (newestWatermark !== null) {
+            this.watermarks[spec.table] = newestWatermark
+        }
+
+        return {
+            table: spec.table,
+            rowsSynced: rows.length,
+            lastWatermark: newestWatermark,
+        }
+    }
+
+    public async syncAll(): Promise<{
+        syncedAt: string
+        tables: Array<{ table: string; rowsSynced: number; lastWatermark: string | number | null }>
+    }> {
+        const results = []
+        for (const spec of this.replicationConfig.tables) {
+            results.push(await this.syncTable(spec))
+        }
+        return {
+            syncedAt: new Date().toISOString(),
+            tables: results,
+        }
+    }
+
+    override async register(app: StarbaseApp) {
+        app.post(`${this.prefix}/sync`, async () => {
+            const summary = await this.syncAll()
+            return new Response(JSON.stringify(summary), {
+                status: 200,
+                headers: { 'Content-Type': 'application/json' },
+            })
+        })
+    }
+}

plugins/replicator/meta.json

@@ -0,0 +1,6 @@
+{
+  "id": "starbasedb:external-replicator",
+  "name": "External Source Replicator",
+  "description": "Pulls and replicates tables from an external Postgres/MySQL/Outerbase data source into Durable Object SQLite",
+  "version": "1.0.0"
+}

tests/test_starbasedb_replicator_plugin.py

@@ -0,0 +1,23 @@
+import json
+import os
+
+
+def test_replicator_plugin_structure_and_watermark_sync():
+    base = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+    code = open(os.path.join(base, "plugins", "replicator", "index.ts"), encoding="utf-8").read()
+    assert "class ExternalReplicationPlugin" in code
+    assert "CREATE TABLE IF NOT EXISTS" in code
+    assert "INSERT OR REPLACE INTO" in code
+    assert "watermarkColumn" in code
+
+
+def test_replicator_meta_json_valid():
+    base = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+    meta = json.load(open(os.path.join(base, "plugins", "replicator", "meta.json"), encoding="utf-8"))
+    assert meta["id"] == "starbasedb:external-replicator"
+
+
+def test_replicator_vitest_suite_present():
+    base = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+    test_ts = open(os.path.join(base, "plugins", "replicator", "index.test.ts"), encoding="utf-8").read()
+    assert "plugin.syncAll()" in test_ts