From f56a1000d0219ba1912a6647e39708cfdc302cf3 Mon Sep 17 00:00:00 2001 From: Alex Maxy Date: Wed, 11 Nov 2026 01:42:09 +0400 Subject: [PATCH] feat(plugins): add replication plugin for external database sync (#72) - Implement ReplicationPlugin extending StarbasePlugin - Support pull-based synchronization from external data sources (Postgres, Supabase, MySQL) to internal Durable Object SQLite - Cursor tracking via configurable cursorColumn (id, updated_at, sequence) for incremental sync - Idempotent upserts with INSERT OR REPLACE INTO keyed by primaryKey - Schema-aware auto-provisioning of destination tables with type inference and identifier escaping - Concurrency lock guard per table to prevent overlapping sync tasks - REST API under /replication (GET /status, POST /sync, POST /configure, POST /reset) - Cron/Alarm integration support via handleCron() and syncAll() - Comprehensive Vitest unit & integration test coverage (17/17 tests passing) - Complete technical documentation in README.md and meta.json --- plugins/replication/README.md | 195 ++++++++++ plugins/replication/index.test.ts | 455 +++++++++++++++++++++++ plugins/replication/index.ts | 583 ++++++++++++++++++++++++++++++ plugins/replication/meta.json | 19 + 4 files changed, 1252 insertions(+) create mode 100644 plugins/replication/README.md create mode 100644 plugins/replication/index.test.ts create mode 100644 plugins/replication/index.ts create mode 100644 plugins/replication/meta.json diff --git a/plugins/replication/README.md b/plugins/replication/README.md new file mode 100644 index 0000000..3c4e986 --- /dev/null +++ b/plugins/replication/README.md @@ -0,0 +1,195 @@ +# StarbaseDB Replication Plugin (`starbasedb:replication`) + +A pull-based replication plugin for **StarbaseDB** that continuously or on-demand replicates data from external databases (PostgreSQL/Supabase, MySQL, etc.) into the internal Durable Object SQLite instance, serving as an ultra-fast, close-to-the-edge read replica. + +--- + +## Features + +- **Edge Replica Architecture:** Pull data from your central database (e.g. Supabase Postgres) directly into Cloudflare Durable Object SQLite to provide sub-millisecond read access at the edge. +- **Incremental Cursor Sync:** Support for cursor-based incremental sync (`cursorColumn` like `id`, `updated_at`, or sequential timestamps), querying only newly modified or added rows (`WHERE cursorColumn > previousCursor`). +- **Idempotent Upserts:** Employs `INSERT OR REPLACE INTO` keyed by `primaryKey` to ensure idempotent replication and prevent duplicate records. +- **Schema-Aware Auto-Creation:** Automatically provisions target tables in SQLite on first sync with type inference (`INTEGER`, `REAL`, `TEXT`) and primary key assignment. Handles schema-qualified table names (e.g. `public.users` -> `users`). +- **Resilient State Persistence:** Tracks sync status, last cursor, total synced rows, and error diagnostics in `_starbase_replication_state`. +- **Concurrency Guard:** Locks in-flight sync jobs per table to prevent race conditions from concurrent triggers. +- **Full REST API:** Endpoints under `/replication/*` to check health, trigger manual syncs, add dynamic configurations, or reset cursors. +- **Cron / Alarm Ready:** Exposes `syncAll()` and `handleCron()` for scheduled synchronization in Cloudflare Workers. + +--- + +## Configuration + +### Table Replication Config (`TableReplicationConfig`) + +| Property | Type | Default | Description | +| :-------------------- | :--------- | :-------------------------- | :----------------------------------------------------------------------------- | +| `name` | `string` | _(Required)_ | Source table name (e.g. `"public.users"` or `"orders"`). | +| `targetTable` | `string` | `name` (stripped of schema) | Destination table in local SQLite. | +| `cursorColumn` | `string` | `undefined` | Column to track for incremental sync (`id`, `updated_at`, etc.). | +| `primaryKey` | `string` | `"id"` | Local table primary key used for upserting. | +| `batchSize` | `number` | `500` | Max rows pulled per sync batch (clamped 1 - 5000). | +| `syncIntervalSeconds` | `number` | `60` | Background sync interval. | +| `whereClause` | `string` | `undefined` | Optional SQL predicate applied to external query (e.g. `"status = 'active'"`). | +| `columns` | `string[]` | `["*"]` | Specific columns to replicate. | + +--- + +## Installation & Setup + +Import and register `ReplicationPlugin` in your `src/index.ts`: + +```typescript +import { ReplicationPlugin } from '../plugins/replication' +import type { StarbasePlugin } from './plugin' + +const replicationPlugin = new ReplicationPlugin({ + tables: [ + { + name: 'public.users', + targetTable: 'users', + cursorColumn: 'id', + primaryKey: 'id', + batchSize: 1000, + }, + { + name: 'public.orders', + cursorColumn: 'updated_at', + primaryKey: 'id', + whereClause: "created_at >= '2024-01-01'", + }, + ], + defaultIntervalSeconds: 60, + autoCreateTable: true, +}) + +const plugins = [ + // ...other plugins + replicationPlugin, +] satisfies StarbasePlugin[] +``` + +--- + +## REST API Endpoints + +### 1. `GET /replication/status` + +Returns the status of all replicated tables, number of configured tables, and any currently active sync operations. + +**Response (200 OK):** + +```json +{ + "result": { + "tables": [ + { + "table_name": "public.users", + "cursor_value": "15420", + "last_sync_at": 1716300000000, + "total_rows_synced": 15420, + "status": "synced", + "last_error": null + } + ], + "configuredCount": 2, + "syncingTables": [] + }, + "error": null +} +``` + +### 2. `POST /replication/sync` + +Triggers an immediate replication cycle. + +**Body (optional):** + +```json +{ + "table": "public.users", + "forceFull": false +} +``` + +- If `table` is omitted, all configured tables are synchronized sequentially. +- If `forceFull: true`, ignores existing cursor and performs a full pull from the start. + +**Response (200 OK):** + +```json +{ + "result": { + "table": "public.users", + "rowsSynced": 250, + "newCursor": "15670", + "durationMs": 42 + }, + "error": null +} +``` + +### 3. `POST /replication/configure` + +Dynamically register or update a table configuration at runtime. + +**Body:** + +```json +{ + "name": "public.products", + "cursorColumn": "id", + "batchSize": 200 +} +``` + +### 4. `POST /replication/reset` + +Resets replication state and cursor, allowing the next sync to re-pull from the start. + +**Body (optional):** + +```json +{ + "table": "public.users" +} +``` + +_(Omit `table` to reset state across all tables)._ + +--- + +## Scheduled Background Sync (Cron / Durable Object Alarms) + +To sync automatically on a schedule, call `handleCron()` from your Worker `scheduled` event: + +```typescript +export default { + async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) { + ctx.waitUntil(replicationPlugin.handleCron()) + }, +} +``` + +--- + +## State Table Schema + +The plugin maintains its state in an internal SQLite table: + +```sql +CREATE TABLE IF NOT EXISTS _starbase_replication_state ( + table_name TEXT PRIMARY KEY, + cursor_value TEXT, + last_sync_at INTEGER, + total_rows_synced INTEGER DEFAULT 0, + status TEXT DEFAULT 'idle', -- 'idle' | 'syncing' | 'synced' | 'error' + last_error TEXT +); +``` + +--- + +## Notes & Limitations + +- **Updates & Deletes:** When `cursorColumn` is set to `updated_at` or a timestamp, updated source rows are pulled and upserted via `INSERT OR REPLACE INTO`. For hard deletes in source databases without soft-delete flags, trigger a full sync with `{ forceFull: true }` or use CDC webhooks. +- **Identifier Safety:** Table and column names are escaped to support schema qualification and SQL reserved keywords safely. diff --git a/plugins/replication/index.test.ts b/plugins/replication/index.test.ts new file mode 100644 index 0000000..ac340a5 --- /dev/null +++ b/plugins/replication/index.test.ts @@ -0,0 +1,455 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { Hono } from 'hono' +import { ReplicationPlugin, TableReplicationConfig } from './index' +import type { DataSource } from '../../src/types' +import type { StarbaseApp } from '../../src/handler' + +describe('ReplicationPlugin', () => { + let mockDataSource: DataSource + let executedRpcQueries: { sql: string; params?: unknown[] }[] + let executedExternalQueries: { sql: string; params?: unknown[] }[] + let externalMockData: Record[] + let stateMockTable: Record[] + + beforeEach(() => { + vi.clearAllMocks() + executedRpcQueries = [] + executedExternalQueries = [] + externalMockData = [] + stateMockTable = [] + + mockDataSource = { + source: 'external', + external: { + dialect: 'postgres', + executeQuery: vi.fn( + async ({ + sql, + params, + }: { + sql: string + params?: unknown[] + }) => { + executedExternalQueries.push({ sql, params }) + return externalMockData + } + ), + } as any, + rpc: { + executeQuery: vi.fn( + async ({ + sql, + params, + }: { + sql: string + params?: unknown[] + }) => { + executedRpcQueries.push({ sql, params }) + + if ( + sql.includes('SELECT') && + sql.includes('_starbase_replication_state') + ) { + if (sql.includes('WHERE table_name = ?')) { + const tableName = params?.[0] + const found = stateMockTable.filter( + (r) => r.table_name === tableName + ) + return found + } + return stateMockTable + } + + if ( + sql.includes( + 'INSERT OR REPLACE INTO _starbase_replication_state' + ) + ) { + const [ + table_name, + cursor_value, + last_sync_at, + total_rows_synced, + status, + last_error, + ] = (params ?? []) as any[] + const idx = stateMockTable.findIndex( + (r) => r.table_name === table_name + ) + const record = { + table_name, + cursor_value, + last_sync_at, + total_rows_synced, + status, + last_error, + } + if (idx >= 0) { + stateMockTable[idx] = record + } else { + stateMockTable.push(record) + } + return [] + } + + if ( + sql.includes('UPDATE _starbase_replication_state') + ) { + if (sql.includes('WHERE table_name = ?')) { + const tableName = params?.[0] + const row = stateMockTable.find( + (r) => r.table_name === tableName + ) + if (row) { + row.cursor_value = null + row.last_sync_at = null + row.total_rows_synced = 0 + row.status = 'idle' + row.last_error = null + } + } else { + stateMockTable.forEach((row) => { + row.cursor_value = null + row.last_sync_at = null + row.total_rows_synced = 0 + row.status = 'idle' + row.last_error = null + }) + } + return [] + } + + return [] + } + ), + }, + } as any + }) + + describe('Initialization & Configuration', () => { + it('should initialize with default options', () => { + const plugin = new ReplicationPlugin() + expect(plugin.name).toBe('starbasedb:replication') + expect(plugin.prefix).toBe('/replication') + expect(plugin.pathPrefix).toBe('/replication') + expect(plugin.getTableConfigs()).toEqual([]) + }) + + it('should initialize with pre-configured tables', () => { + const plugin = new ReplicationPlugin({ + tables: [ + { name: 'users', cursorColumn: 'id' }, + { name: 'orders', primaryKey: 'order_id', batchSize: 250 }, + ], + defaultIntervalSeconds: 120, + }) + + const configs = plugin.getTableConfigs() + expect(configs).toHaveLength(2) + expect(configs[0].name).toBe('users') + expect(configs[0].targetTable).toBe('users') + expect(configs[0].batchSize).toBe(500) + expect(configs[0].syncIntervalSeconds).toBe(120) + + expect(configs[1].name).toBe('orders') + expect(configs[1].primaryKey).toBe('order_id') + expect(configs[1].batchSize).toBe(250) + }) + + it('should strip schema prefix for default targetTable', () => { + const plugin = new ReplicationPlugin() + plugin.configureTable({ name: 'public.customers' }) + + const config = plugin.getTableConfig('public.customers') + expect(config).toBeDefined() + expect(config?.targetTable).toBe('customers') + }) + + it('should throw an error when configuring a table without a valid name', () => { + const plugin = new ReplicationPlugin() + expect(() => plugin.configureTable({ name: '' })).toThrow( + 'Table name is required for replication configuration' + ) + expect(() => plugin.configureTable({ name: ' ' })).toThrow( + 'Table name is required for replication configuration' + ) + }) + + it('should initialize schema state table in init()', async () => { + const plugin = new ReplicationPlugin() + plugin.setDataSource(mockDataSource) + await plugin.init() + + expect(executedRpcQueries).toHaveLength(1) + expect(executedRpcQueries[0].sql).toContain( + 'CREATE TABLE IF NOT EXISTS _starbase_replication_state' + ) + }) + }) + + describe('HTTP Routes & Hono Middleware', () => { + let app: StarbaseApp + let plugin: ReplicationPlugin + + beforeEach(async () => { + app = new Hono() as unknown as StarbaseApp + app.use('*', async (c, next) => { + c.set('dataSource', mockDataSource) + await next() + }) + + plugin = new ReplicationPlugin({ + tables: [{ name: 'products', cursorColumn: 'id' }], + }) + await plugin.register(app) + }) + + it('GET /replication/status returns replication status', async () => { + const res = await app.request('/replication/status', { + method: 'GET', + }) + expect(res.status).toBe(200) + + const body = await res.json() + expect(body.error).toBeUndefined() + expect(body.result.configuredCount).toBe(1) + expect(body.result.syncingTables).toEqual([]) + expect(Array.isArray(body.result.tables)).toBe(true) + }) + + it('POST /replication/configure dynamically adds table config', async () => { + const res = await app.request('/replication/configure', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + name: 'invoices', + cursorColumn: 'created_at', + batchSize: 100, + }), + }) + expect(res.status).toBe(200) + const body = await res.json() + expect(body.result.success).toBe(true) + expect(body.result.table).toBe('invoices') + expect(plugin.getTableConfigs()).toHaveLength(2) + }) + + it('POST /replication/configure returns 400 when name is invalid', async () => { + const res = await app.request('/replication/configure', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ name: '' }), + }) + expect(res.status).toBe(400) + const body = await res.json() + expect(body.error).toBe( + 'Table name is required for replication configuration' + ) + }) + + it('POST /replication/reset resets single table or all tables', async () => { + stateMockTable.push({ + table_name: 'products', + cursor_value: '100', + last_sync_at: Date.now(), + total_rows_synced: 50, + status: 'synced', + last_error: null, + }) + + const res = await app.request('/replication/reset', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ table: 'products' }), + }) + expect(res.status).toBe(200) + const row = stateMockTable.find((r) => r.table_name === 'products') + expect(row?.cursor_value).toBeNull() + expect(row?.total_rows_synced).toBe(0) + expect(row?.status).toBe('idle') + }) + + it('POST /replication/sync triggers table sync', async () => { + externalMockData = [ + { id: 1, name: 'Widget A', price: 9.99 }, + { id: 2, name: 'Widget B', price: 19.99 }, + ] + + const res = await app.request('/replication/sync', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ table: 'products' }), + }) + expect(res.status).toBe(200) + const body = await res.json() + expect(body.result.table).toBe('products') + expect(body.result.rowsSynced).toBe(2) + expect(body.result.newCursor).toBe('2') + }) + }) + + describe('Replication Sync Execution', () => { + let plugin: ReplicationPlugin + + beforeEach(() => { + plugin = new ReplicationPlugin({ + tables: [ + { + name: 'public.users', + targetTable: 'users', + cursorColumn: 'id', + primaryKey: 'id', + whereClause: 'active = 1', + batchSize: 10, + }, + ], + }) + plugin.setDataSource(mockDataSource) + }) + + it('should handle zero rows returned from external source gracefully', async () => { + externalMockData = [] + const result = await plugin.syncTable('public.users') + + expect(result.table).toBe('public.users') + expect(result.rowsSynced).toBe(0) + expect(result.newCursor).toBeNull() + + const state = stateMockTable.find( + (r) => r.table_name === 'public.users' + ) + expect(state?.status).toBe('synced') + expect(state?.total_rows_synced).toBe(0) + }) + + it('should auto-create local SQLite table with correct types and primary key', async () => { + externalMockData = [ + { id: 1, username: 'alice', is_admin: 1, balance: 125.5 }, + ] + + await plugin.syncTable('public.users') + + const createTableQuery = executedRpcQueries.find((q) => + q.sql.includes('CREATE TABLE IF NOT EXISTS "users"') + ) + expect(createTableQuery).toBeDefined() + expect(createTableQuery?.sql).toContain('"id" INTEGER PRIMARY KEY') + expect(createTableQuery?.sql).toContain('"username" TEXT') + expect(createTableQuery?.sql).toContain('"balance" REAL') + }) + + it('should perform incremental sync using cursorColumn', async () => { + // First batch: rows 1 and 2 + externalMockData = [ + { id: 1, name: 'Alice' }, + { id: 2, name: 'Bob' }, + ] + + const firstResult = await plugin.syncTable('public.users') + expect(firstResult.rowsSynced).toBe(2) + expect(firstResult.newCursor).toBe('2') + + const stateAfterFirst = stateMockTable.find( + (r) => r.table_name === 'public.users' + ) + expect(stateAfterFirst?.cursor_value).toBe('2') + expect(stateAfterFirst?.total_rows_synced).toBe(2) + + // Second batch: rows 3 and 4 + externalMockData = [ + { id: 3, name: 'Charlie' }, + { id: 4, name: 'Diana' }, + ] + + const secondResult = await plugin.syncTable('public.users') + expect(secondResult.rowsSynced).toBe(2) + expect(secondResult.newCursor).toBe('4') + + // Verify query in second run included cursor condition + const lastExtQuery = + executedExternalQueries[executedExternalQueries.length - 1] + expect(lastExtQuery.sql).toContain('"id" > ?') + expect(lastExtQuery.params).toContain('2') + + const stateAfterSecond = stateMockTable.find( + (r) => r.table_name === 'public.users' + ) + expect(stateAfterSecond?.cursor_value).toBe('4') + expect(stateAfterSecond?.total_rows_synced).toBe(4) + }) + + it('should support forceFull = true to bypass existing cursor', async () => { + stateMockTable.push({ + table_name: 'public.users', + cursor_value: '50', + last_sync_at: Date.now(), + total_rows_synced: 50, + status: 'synced', + last_error: null, + }) + + externalMockData = [{ id: 1, name: 'Root' }] + + const result = await plugin.syncTable('public.users', true) + expect(result.rowsSynced).toBe(1) + expect(result.newCursor).toBe('1') + + const lastExtQuery = + executedExternalQueries[executedExternalQueries.length - 1] + expect(lastExtQuery.sql).not.toContain('"id" > ?') + }) + + it('should prevent concurrent sync on the same table with lock guard', async () => { + let resolveExternalQuery: (value: any) => void + const delayPromise = new Promise((resolve) => { + resolveExternalQuery = resolve + }) + + mockDataSource.external!.executeQuery = vi + .fn() + .mockImplementation(async () => { + await delayPromise + return [{ id: 10, name: 'Pending' }] + }) + + const firstSyncPromise = plugin.syncTable('public.users') + const secondSyncResult = await plugin.syncTable('public.users') + + expect(secondSyncResult.error).toContain('is already in progress') + expect(secondSyncResult.rowsSynced).toBe(0) + + resolveExternalQuery!([{ id: 10, name: 'Pending' }]) + const firstSyncResult = await firstSyncPromise + expect(firstSyncResult.rowsSynced).toBe(1) + }) + + it('should record error state in SQLite when external source fails', async () => { + mockDataSource.external!.executeQuery = vi + .fn() + .mockRejectedValue(new Error('Connection timeout to Supabase')) + + const result = await plugin.syncTable('public.users') + expect(result.rowsSynced).toBe(0) + expect(result.error).toBe('Connection timeout to Supabase') + + const state = stateMockTable.find( + (r) => r.table_name === 'public.users' + ) + expect(state?.status).toBe('error') + expect(state?.last_error).toContain( + 'Connection timeout to Supabase' + ) + }) + + it('should sync all configured tables with syncAll() and handleCron()', async () => { + plugin.configureTable({ name: 'orders', cursorColumn: 'id' }) + externalMockData = [{ id: 100 }] + + const results = await plugin.syncAll() + expect(results).toHaveLength(2) + + const cronResults = await plugin.handleCron() + expect(cronResults).toHaveLength(2) + }) + }) +}) diff --git a/plugins/replication/index.ts b/plugins/replication/index.ts new file mode 100644 index 0000000..6bb9b4a --- /dev/null +++ b/plugins/replication/index.ts @@ -0,0 +1,583 @@ +import { StarbaseApp } from '../../src/handler' +import { StarbasePlugin } from '../../src/plugin' +import { DataSource, QueryResult } from '../../src/types' +import { createResponse } from '../../src/utils' + +export interface TableReplicationConfig { + /** Name of the table in the external database (can be schema-qualified, e.g. "public.users") */ + name: string + /** Target table name in the internal SQLite database (defaults to the table name without schema) */ + targetTable?: string + /** Column used for incremental sync cursor tracking (e.g. "id", "updated_at", "created_at") */ + cursorColumn?: string + /** Primary key for local upsert conflict resolution (defaults to "id") */ + primaryKey?: string + /** Batch size of rows to pull per iteration (defaults to 500, clamped between 1 and 5000) */ + batchSize?: number + /** Sync interval in seconds for background replication (defaults to plugin default) */ + syncIntervalSeconds?: number + /** Optional SQL filter condition applied to external query (e.g. "is_active = true") */ + whereClause?: string + /** Specific columns to pull (defaults to all columns: "*") */ + columns?: string[] +} + +export interface ReplicationPluginOptions { + /** Pre-configured table replication settings */ + tables?: TableReplicationConfig[] + /** Default sync interval in seconds */ + defaultIntervalSeconds?: number + /** Automatically create the destination table in SQLite if it does not exist */ + autoCreateTable?: boolean + /** Whether endpoints require authentication (defaults to true) */ + requiresAuth?: boolean +} + +export interface TableSyncState { + table_name: string + cursor_value: string | null + last_sync_at: number | null + total_rows_synced: number + status: 'idle' | 'syncing' | 'synced' | 'error' + last_error: string | null +} + +export interface TableSyncResult { + table: string + rowsSynced: number + newCursor: string | null + durationMs: number + error?: string +} + +function escapeIdentifier(id: string): string { + return `"${id.replace(/"/g, '""')}"` +} + +function escapeTableOrSchemaName(name: string): string { + if (name.includes('.')) { + const parts = name.split('.') + return parts.map(escapeIdentifier).join('.') + } + return escapeIdentifier(name) +} + +function stripSchema(name: string): string { + if (name.includes('.')) { + const parts = name.split('.') + return parts[parts.length - 1] + } + return name +} + +const SQL_QUERIES = { + CREATE_STATE_TABLE: ` + CREATE TABLE IF NOT EXISTS _starbase_replication_state ( + table_name TEXT PRIMARY KEY, + cursor_value TEXT, + last_sync_at INTEGER, + total_rows_synced INTEGER DEFAULT 0, + status TEXT DEFAULT 'idle', + last_error TEXT + ) + `, + GET_TABLE_STATE: ` + SELECT table_name, cursor_value, last_sync_at, total_rows_synced, status, last_error + FROM _starbase_replication_state + WHERE table_name = ? + `, + GET_ALL_STATES: ` + SELECT table_name, cursor_value, last_sync_at, total_rows_synced, status, last_error + FROM _starbase_replication_state + `, + UPSERT_TABLE_STATE: ` + INSERT OR REPLACE INTO _starbase_replication_state ( + table_name, cursor_value, last_sync_at, total_rows_synced, status, last_error + ) VALUES (?, ?, ?, ?, ?, ?) + `, + RESET_TABLE_STATE: ` + UPDATE _starbase_replication_state + SET cursor_value = NULL, last_sync_at = NULL, total_rows_synced = 0, status = 'idle', last_error = NULL + WHERE table_name = ? + `, + RESET_ALL_STATES: ` + UPDATE _starbase_replication_state + SET cursor_value = NULL, last_sync_at = NULL, total_rows_synced = 0, status = 'idle', last_error = NULL + `, +} + +export class ReplicationPlugin extends StarbasePlugin { + public prefix: string = '/replication' + public override pathPrefix: string = '/replication' + private dataSource?: DataSource + private tableConfigs: Map = new Map() + private defaultIntervalSeconds: number + private autoCreateTable: boolean + private syncingTables: Set = new Set() + private isInitialized: boolean = false + + constructor(options: ReplicationPluginOptions = {}) { + super('starbasedb:replication', { + requiresAuth: options.requiresAuth ?? true, + }) + this.defaultIntervalSeconds = options.defaultIntervalSeconds ?? 60 + this.autoCreateTable = options.autoCreateTable ?? true + + if (options.tables) { + options.tables.forEach((t) => this.configureTable(t)) + } + } + + public setDataSource(dataSource: DataSource): void { + this.dataSource = dataSource + } + + public configureTable(config: TableReplicationConfig): void { + if ( + !config || + !config.name || + typeof config.name !== 'string' || + !config.name.trim() + ) { + throw new Error( + 'Table name is required for replication configuration' + ) + } + const cleanName = config.name.trim() + const effectiveConfig: TableReplicationConfig = { + batchSize: 500, + primaryKey: 'id', + syncIntervalSeconds: this.defaultIntervalSeconds, + targetTable: config.targetTable ?? stripSchema(cleanName), + ...config, + name: cleanName, + } + this.tableConfigs.set(cleanName, effectiveConfig) + } + + public getTableConfigs(): TableReplicationConfig[] { + return Array.from(this.tableConfigs.values()) + } + + public getTableConfig(name: string): TableReplicationConfig | undefined { + return this.tableConfigs.get(name) + } + + override async register(app: StarbaseApp): Promise { + // Middleware to attach dataSource and ensure schema state table exists + app.use(async (c, next) => { + const ds = c?.get?.('dataSource') + if (ds) { + this.dataSource = ds + if (!this.isInitialized) { + await this.init() + } + } + await next() + }) + + // GET /replication/status + app.get(`${this.pathPrefix}/status`, async () => { + try { + const status = await this.getStatus() + return createResponse(status, undefined, 200) + } catch (err: any) { + return createResponse( + null, + err?.message ?? 'Failed to get status', + 500 + ) + } + }) + + // POST /replication/sync + app.post(`${this.pathPrefix}/sync`, async (c) => { + try { + let body: { table?: string; forceFull?: boolean } = {} + try { + body = (await c.req.json()) ?? {} + } catch { + // Empty or missing json body is acceptable + } + + if (body.table) { + const result = await this.syncTable( + body.table, + body.forceFull ?? false + ) + return createResponse(result, undefined, 200) + } + + const results = await this.syncAll(body.forceFull ?? false) + return createResponse(results, undefined, 200) + } catch (err: any) { + return createResponse( + null, + err?.message ?? 'Replication sync failed', + 500 + ) + } + }) + + // POST /replication/configure + app.post(`${this.pathPrefix}/configure`, async (c) => { + try { + const body = (await c.req.json()) as TableReplicationConfig + if (!body || !body.name) { + return createResponse( + null, + 'Table name is required for replication configuration', + 400 + ) + } + this.configureTable(body) + return createResponse( + { success: true, table: body.name }, + undefined, + 200 + ) + } catch (err: any) { + return createResponse( + null, + err?.message ?? 'Configuration failed', + 400 + ) + } + }) + + // POST /replication/reset + app.post(`${this.pathPrefix}/reset`, async (c) => { + try { + let body: { table?: string } = {} + try { + body = (await c.req.json()) ?? {} + } catch { + // Empty json body is fine + } + await this.resetState(body.table) + return createResponse( + { success: true, table: body.table ?? 'all' }, + undefined, + 200 + ) + } catch (err: any) { + return createResponse(null, err?.message ?? 'Reset failed', 500) + } + }) + } + + public async init(): Promise { + if (!this.dataSource) return + try { + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.CREATE_STATE_TABLE, + params: [], + }) + this.isInitialized = true + } catch (error) { + console.error( + 'Error initializing ReplicationPlugin state table:', + error + ) + } + } + + public async getStatus(): Promise<{ + tables: TableSyncState[] + configuredCount: number + syncingTables: string[] + }> { + if (!this.dataSource) { + throw new Error( + 'ReplicationPlugin not initialized: DataSource is missing' + ) + } + + const rows = (await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.GET_ALL_STATES, + params: [], + })) as unknown as TableSyncState[] + + return { + tables: Array.isArray(rows) ? rows : [], + configuredCount: this.tableConfigs.size, + syncingTables: Array.from(this.syncingTables), + } + } + + public async resetState(tableName?: string): Promise { + if (!this.dataSource) { + throw new Error( + 'ReplicationPlugin not initialized: DataSource is missing' + ) + } + + if (tableName) { + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.RESET_TABLE_STATE, + params: [tableName], + }) + } else { + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.RESET_ALL_STATES, + params: [], + }) + } + } + + public async syncAll( + forceFull: boolean = false + ): Promise { + const results: TableSyncResult[] = [] + for (const tableName of this.tableConfigs.keys()) { + const res = await this.syncTable(tableName, forceFull) + results.push(res) + } + return results + } + + public async handleCron(): Promise { + return await this.syncAll(false) + } + + public async syncTable( + tableName: string, + forceFull: boolean = false + ): Promise { + const startTime = Date.now() + const config = this.tableConfigs.get(tableName) + if (!config) { + throw new Error( + `Table '${tableName}' is not configured for replication` + ) + } + + if (!this.dataSource) { + throw new Error( + 'ReplicationPlugin not initialized: DataSource is missing' + ) + } + + if (this.syncingTables.has(tableName)) { + return { + table: tableName, + rowsSynced: 0, + newCursor: null, + durationMs: 0, + error: `Replication for table '${tableName}' is already in progress`, + } + } + + this.syncingTables.add(tableName) + + try { + // Retrieve current replication cursor state + const stateResult = (await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.GET_TABLE_STATE, + params: [tableName], + })) as unknown as TableSyncState[] + + const currentState = + Array.isArray(stateResult) && stateResult.length > 0 + ? stateResult[0] + : null + const previousCursor = forceFull + ? null + : (currentState?.cursor_value ?? null) + const previousTotal = forceFull + ? 0 + : (currentState?.total_rows_synced ?? 0) + + // Construct pull query for external database + const targetTable = config.targetTable ?? stripSchema(config.name) + const batchSize = Math.max( + 1, + Math.min(5000, config.batchSize ?? 500) + ) + const columnsStr = + config.columns && config.columns.length > 0 + ? config.columns.map(escapeIdentifier).join(', ') + : '*' + + const sourceTableEscaped = escapeTableOrSchemaName(config.name) + let pullSql = `SELECT ${columnsStr} FROM ${sourceTableEscaped}` + const whereClauses: string[] = [] + const queryParams: unknown[] = [] + + if (config.whereClause) { + whereClauses.push(`(${config.whereClause})`) + } + + if (config.cursorColumn && previousCursor !== null) { + whereClauses.push( + `${escapeIdentifier(config.cursorColumn)} > ?` + ) + queryParams.push(previousCursor) + } + + if (whereClauses.length > 0) { + pullSql += ` WHERE ${whereClauses.join(' AND ')}` + } + + if (config.cursorColumn) { + pullSql += ` ORDER BY ${escapeIdentifier(config.cursorColumn)} ASC` + } + + pullSql += ` LIMIT ${batchSize}` + + // Fetch data from external database source + let externalRows: Record[] = [] + if (this.dataSource.external) { + const extResult = await this.dataSource.external.executeQuery({ + sql: pullSql, + params: queryParams, + }) + externalRows = Array.isArray(extResult) + ? (extResult as Record[]) + : [] + } else { + // Fallback: internal RPC query for testing environment + const extResult = (await this.dataSource.rpc.executeQuery({ + sql: pullSql, + params: queryParams, + })) as unknown as Record[] + externalRows = Array.isArray(extResult) ? extResult : [] + } + + if (!externalRows || externalRows.length === 0) { + // No new records to sync + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.UPSERT_TABLE_STATE, + params: [ + tableName, + previousCursor, + Date.now(), + previousTotal, + 'synced', + null, + ], + }) + return { + table: tableName, + rowsSynced: 0, + newCursor: previousCursor, + durationMs: Date.now() - startTime, + } + } + + // Auto-create destination table in SQLite if enabled + if (this.autoCreateTable && externalRows.length > 0) { + await this.ensureLocalTable( + targetTable, + externalRows[0], + config.primaryKey ?? 'id' + ) + } + + // Batch insert or replace into local SQLite database + let newCursor: string | null = previousCursor + const rowKeys = Object.keys(externalRows[0]) + const quotedCols = rowKeys.map(escapeIdentifier).join(', ') + const placeholders = rowKeys.map(() => '?').join(', ') + const insertSql = `INSERT OR REPLACE INTO ${escapeIdentifier(targetTable)} (${quotedCols}) VALUES (${placeholders})` + + for (const row of externalRows) { + const values = rowKeys.map((k) => + row[k] === undefined ? null : row[k] + ) + await this.dataSource.rpc.executeQuery({ + sql: insertSql, + params: values, + }) + + if ( + config.cursorColumn && + row[config.cursorColumn] !== undefined && + row[config.cursorColumn] !== null + ) { + newCursor = String(row[config.cursorColumn]) + } + } + + const totalSynced = previousTotal + externalRows.length + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.UPSERT_TABLE_STATE, + params: [ + tableName, + newCursor, + Date.now(), + totalSynced, + 'synced', + null, + ], + }) + + return { + table: tableName, + rowsSynced: externalRows.length, + newCursor, + durationMs: Date.now() - startTime, + } + } catch (error: any) { + const errorMsg = error?.message ?? String(error) + console.error(`Replication error on table '${tableName}':`, error) + + if (this.dataSource) { + try { + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.UPSERT_TABLE_STATE, + params: [ + tableName, + null, + Date.now(), + 0, + 'error', + errorMsg, + ], + }) + } catch { + // Ignore secondary error while reporting state + } + } + + return { + table: tableName, + rowsSynced: 0, + newCursor: null, + durationMs: Date.now() - startTime, + error: errorMsg, + } + } finally { + this.syncingTables.delete(tableName) + } + } + + private async ensureLocalTable( + targetTable: string, + sampleRow: Record, + primaryKey: string + ): Promise { + if (!this.dataSource) return + + const columnDefs = Object.entries(sampleRow).map(([col, val]) => { + let colType = 'TEXT' + if (typeof val === 'number') { + colType = Number.isInteger(val) ? 'INTEGER' : 'REAL' + } else if (typeof val === 'boolean') { + colType = 'INTEGER' + } + const escapedCol = escapeIdentifier(col) + if (col === primaryKey) { + return `${escapedCol} ${colType} PRIMARY KEY` + } + return `${escapedCol} ${colType}` + }) + + const escapedTable = escapeIdentifier(targetTable) + const createSql = `CREATE TABLE IF NOT EXISTS ${escapedTable} (${columnDefs.join(', ')})` + await this.dataSource.rpc.executeQuery({ + sql: createSql, + params: [], + }) + } +} diff --git a/plugins/replication/meta.json b/plugins/replication/meta.json new file mode 100644 index 0000000..3e127ab --- /dev/null +++ b/plugins/replication/meta.json @@ -0,0 +1,19 @@ +{ + "version": "1.0.0", + "name": "starbasedb:replication", + "description": "Replicates data from external sources (Postgres, Supabase, MySQL) into internal Durable Object SQLite as an edge replica.", + "resources": { + "tables": { + "_starbase_replication_state": "Tracks replication cursors, sync timestamps, and status per table" + }, + "secrets": {}, + "variables": { + "intervalSeconds": "Default background sync interval in seconds" + } + }, + "dependencies": { + "tables": {}, + "secrets": {}, + "variables": {} + } +}