From 283815a5401784bdb86c9f4ce53249a2b0f61347 Mon Sep 17 00:00:00 2001 From: Veilreaver Null Date: Wed, 11 Nov 2026 10:59:34 -0800 Subject: [PATCH] feat(plugins): add ExternalReplicationPlugin to pull & sync external data into DO SQLite (#72) Fixes #72 Closes #72 /claim #72 --- plugins/replicator/index.test.ts | 41 +++++++ plugins/replicator/index.ts | 129 +++++++++++++++++++++ plugins/replicator/meta.json | 6 + tests/test_starbasedb_replicator_plugin.py | 23 ++++ 4 files changed, 199 insertions(+) create mode 100644 plugins/replicator/index.test.ts create mode 100644 plugins/replicator/index.ts create mode 100644 plugins/replicator/meta.json create mode 100644 tests/test_starbasedb_replicator_plugin.py diff --git a/plugins/replicator/index.test.ts b/plugins/replicator/index.test.ts new file mode 100644 index 0000000..99e2719 --- /dev/null +++ b/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) + }) +}) diff --git a/plugins/replicator/index.ts b/plugins/replicator/index.ts new file mode 100644 index 0000000..2b3f8a7 --- /dev/null +++ b/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 = {} + + 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[] = 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' }, + }) + }) + } +} diff --git a/plugins/replicator/meta.json b/plugins/replicator/meta.json new file mode 100644 index 0000000..6afc5dc --- /dev/null +++ b/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" +} diff --git a/tests/test_starbasedb_replicator_plugin.py b/tests/test_starbasedb_replicator_plugin.py new file mode 100644 index 0000000..20762ab --- /dev/null +++ b/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