From de70394e92127ae6b16f132929f5dbd44d54ce6e Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 18:37:55 -0500 Subject: [PATCH 1/3] feat(plugins): add ReplicationPlugin for external-to-internal data replication (#72) --- plugins/replication/index.ts | 424 +++++++++++++++++++++++++++++++++++ 1 file changed, 424 insertions(+) create mode 100644 plugins/replication/index.ts diff --git a/plugins/replication/index.ts b/plugins/replication/index.ts new file mode 100644 index 0000000..38e6124 --- /dev/null +++ b/plugins/replication/index.ts @@ -0,0 +1,424 @@ +import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler' +import { executeExternalQuery } from '../../src/operation' +import { StarbasePlugin } from '../../src/plugin' +import { DataSource, QueryResult } from '../../src/types' +import { createResponse } from '../../src/utils' + +export interface ReplicationTableConfig { + table: string + schema?: string + cursorColumn?: string + primaryKey?: string + intervalSeconds?: number + batchSize?: number + columns?: string[] +} + +export interface ReplicationPluginOptions { + tables: (string | ReplicationTableConfig)[] + intervalSeconds?: number + batchSize?: number + autoSyncOnQuery?: boolean + ctx?: ExecutionContext +} + +export interface ReplicationStatus { + table_name: string + cursor_column: string | null + last_cursor_value: string | null + last_synced_at: number + rows_synced: number +} + +const SQL_QUERIES = { + CREATE_STATE_TABLE: ` + CREATE TABLE IF NOT EXISTS tmp_replication_state ( + table_name TEXT NOT NULL UNIQUE PRIMARY KEY, + cursor_column TEXT, + last_cursor_value TEXT, + last_synced_at INTEGER NOT NULL, + rows_synced INTEGER DEFAULT 0 + ) + `, + GET_STATE: ` + SELECT table_name, cursor_column, last_cursor_value, last_synced_at, rows_synced + FROM tmp_replication_state + WHERE table_name = ? + `, + GET_ALL_STATES: ` + SELECT table_name, cursor_column, last_cursor_value, last_synced_at, rows_synced + FROM tmp_replication_state + `, + UPSERT_STATE: ` + INSERT OR REPLACE INTO tmp_replication_state ( + table_name, + cursor_column, + last_cursor_value, + last_synced_at, + rows_synced + ) VALUES (?, ?, ?, ?, ?) + `, +} + +function sanitizeIdentifier(identifier: string): string { + if (!/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(identifier)) { + throw new Error(`Invalid SQL identifier: ${identifier}`) + } + return identifier +} + +function inferSqliteColumnType(value: unknown): string { + if (typeof value === 'number') { + return Number.isInteger(value) ? 'INTEGER' : 'REAL' + } + if (typeof value === 'boolean') { + return 'INTEGER' + } + return 'TEXT' +} + +export class ReplicationPlugin extends StarbasePlugin { + public override pathPrefix: string = '/replication' + public tables: ReplicationTableConfig[] + public intervalSeconds: number + public batchSize: number + public autoSyncOnQuery: boolean + + private dataSource?: DataSource + private config?: StarbaseDBConfiguration + private executionContext?: ExecutionContext + private initialized: boolean = false + + constructor(opts: ReplicationPluginOptions) { + super('starbasedb:replication', { requiresAuth: true }, '/replication') + this.intervalSeconds = opts.intervalSeconds ?? 300 + this.batchSize = opts.batchSize ?? 500 + this.autoSyncOnQuery = opts.autoSyncOnQuery ?? false + this.executionContext = opts.ctx + this.tables = (opts.tables || []).map((t) => + typeof t === 'string' + ? { + table: t, + intervalSeconds: this.intervalSeconds, + batchSize: this.batchSize, + } + : { + ...t, + intervalSeconds: + t.intervalSeconds ?? this.intervalSeconds, + batchSize: t.batchSize ?? this.batchSize, + } + ) + } + + override async register(app: StarbaseApp): Promise { + app.use('*', async (c, next) => { + this.dataSource = c?.get('dataSource') + this.config = c?.get('config') + await this.init() + await next() + }) + + app.get(`${this.pathPrefix}/status`, async () => { + try { + const status = await this.getStatus() + return createResponse({ tables: status }, undefined, 200) + } catch (error: any) { + return createResponse( + undefined, + error?.message || 'Failed to fetch replication status', + 500 + ) + } + }) + + app.post(`${this.pathPrefix}/sync`, async (c) => { + if (this.config && this.config.role !== 'admin') { + return createResponse(undefined, 'Unauthorized request', 401) + } + + try { + let targetTable: string | undefined + try { + const body = await c.req.json() + targetTable = body?.table + } catch { + targetTable = undefined + } + + if (targetTable) { + const tableCfg = this.tables.find( + (t) => t.table === targetTable + ) + if (!tableCfg) { + return createResponse( + undefined, + `Table ${targetTable} is not configured for replication`, + 400 + ) + } + const result = await this.syncTable(tableCfg, true) + return createResponse(result, undefined, 200) + } + + const results = await this.syncAll(true) + return createResponse({ results }, undefined, 200) + } catch (error: any) { + return createResponse( + undefined, + error?.message || 'Replication sync failed', + 500 + ) + } + }) + } + + private async init(): Promise { + if (this.initialized || !this.dataSource) return + + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.CREATE_STATE_TABLE, + params: [], + }) + this.initialized = true + } + + override async beforeQuery(opts: { + sql: string + params?: unknown[] + dataSource?: DataSource + config?: StarbaseDBConfiguration + }): Promise<{ sql: string; params?: unknown[] }> { + if (opts.dataSource) { + this.dataSource = opts.dataSource + } + if (opts.config) { + this.config = opts.config + } + + if (this.autoSyncOnQuery && this.dataSource) { + await this.init() + const syncPromise = this.syncAll(false) + const ctx = + this.executionContext ?? this.dataSource.executionContext + if (ctx) { + ctx.waitUntil(syncPromise) + } else { + await syncPromise + } + } + + return { + sql: opts.sql, + params: opts.params, + } + } + + public async getStatus(): Promise { + if (!this.dataSource) return [] + await this.init() + + const rows = (await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.GET_ALL_STATES, + params: [], + })) as QueryResult[] + + return (rows || []).map((row) => ({ + table_name: String(row.table_name), + cursor_column: + row.cursor_column !== null && row.cursor_column !== undefined + ? String(row.cursor_column) + : null, + last_cursor_value: + row.last_cursor_value !== null && + row.last_cursor_value !== undefined + ? String(row.last_cursor_value) + : null, + last_synced_at: Number(row.last_synced_at ?? 0), + rows_synced: Number(row.rows_synced ?? 0), + })) + } + + public async syncAll(force: boolean = false): Promise< + { + table: string + syncedRows: number + lastCursorValue: string | null + skipped: boolean + }[] + > { + const results = [] + for (const tableCfg of this.tables) { + const res = await this.syncTable(tableCfg, force) + results.push(res) + } + return results + } + + public async syncTable( + tableCfg: ReplicationTableConfig, + force: boolean = false + ): Promise<{ + table: string + syncedRows: number + lastCursorValue: string | null + skipped: boolean + }> { + if (!this.dataSource || !this.config) { + throw new Error('ReplicationPlugin is not initialized with a dataSource') + } + + await this.init() + + const tableName = sanitizeIdentifier(tableCfg.table) + const schemaName = tableCfg.schema + ? sanitizeIdentifier(tableCfg.schema) + : undefined + const cursorCol = tableCfg.cursorColumn + ? sanitizeIdentifier(tableCfg.cursorColumn) + : undefined + const pkCol = tableCfg.primaryKey + ? sanitizeIdentifier(tableCfg.primaryKey) + : cursorCol + + const stateRows = (await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.GET_STATE, + params: [tableName], + })) as QueryResult[] + + const currentState = stateRows?.[0] + const now = Date.now() + const intervalMs = + (tableCfg.intervalSeconds ?? this.intervalSeconds) * 1000 + + if ( + !force && + currentState && + now - Number(currentState.last_synced_at ?? 0) < intervalMs + ) { + return { + table: tableName, + syncedRows: 0, + lastCursorValue: + currentState.last_cursor_value !== null && + currentState.last_cursor_value !== undefined + ? String(currentState.last_cursor_value) + : null, + skipped: true, + } + } + + const lastCursorValue = + currentState?.last_cursor_value !== null && + currentState?.last_cursor_value !== undefined + ? String(currentState.last_cursor_value) + : null + + const selectCols = + tableCfg.columns && tableCfg.columns.length > 0 + ? tableCfg.columns.map(sanitizeIdentifier).join(', ') + : '*' + const qualifiedTable = schemaName + ? `${schemaName}.${tableName}` + : tableName + const limit = tableCfg.batchSize ?? this.batchSize + + let externalSql = `SELECT ${selectCols} FROM ${qualifiedTable}` + const externalParams: unknown[] = [] + + if (cursorCol && lastCursorValue !== null) { + externalSql += ` WHERE ${cursorCol} > ?` + externalParams.push(lastCursorValue) + } + + if (cursorCol) { + externalSql += ` ORDER BY ${cursorCol} ASC` + } + + externalSql += ` LIMIT ${Number(limit)}` + + const externalRows = (await executeExternalQuery({ + sql: externalSql, + params: externalParams, + dataSource: this.dataSource, + config: this.config, + })) as Record[] + + if (!Array.isArray(externalRows) || externalRows.length === 0) { + const prevSynced = Number(currentState?.rows_synced ?? 0) + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.UPSERT_STATE, + params: [ + tableName, + cursorCol ?? null, + lastCursorValue, + now, + prevSynced, + ], + }) + return { + table: tableName, + syncedRows: 0, + lastCursorValue, + skipped: false, + } + } + + const firstRow = externalRows[0] + const columns = Object.keys(firstRow).map(sanitizeIdentifier) + + const colDefs = columns.map((col) => { + const colType = inferSqliteColumnType(firstRow[col]) + const pkSuffix = pkCol && col === pkCol ? ' PRIMARY KEY' : '' + return `"${col}" ${colType}${pkSuffix}` + }) + + await this.dataSource.rpc.executeQuery({ + sql: `CREATE TABLE IF NOT EXISTS "${tableName}" (${colDefs.join(', ')})`, + params: [], + }) + + const placeholders = columns.map(() => '?').join(', ') + const quotedCols = columns.map((c) => `"${c}"`).join(', ') + const insertSql = `INSERT OR REPLACE INTO "${tableName}" (${quotedCols}) VALUES (${placeholders})` + + let newCursorValue = lastCursorValue + for (const row of externalRows) { + const values = columns.map((col) => { + const v = row[col] + if (typeof v === 'boolean') return v ? 1 : 0 + if (v !== null && typeof v === 'object') return JSON.stringify(v) + return v ?? null + }) + await this.dataSource.rpc.executeQuery({ + sql: insertSql, + params: values, + }) + + if (cursorCol && row[cursorCol] !== undefined && row[cursorCol] !== null) { + newCursorValue = String(row[cursorCol]) + } + } + + const totalSynced = + Number(currentState?.rows_synced ?? 0) + externalRows.length + + await this.dataSource.rpc.executeQuery({ + sql: SQL_QUERIES.UPSERT_STATE, + params: [ + tableName, + cursorCol ?? null, + newCursorValue, + now, + totalSynced, + ], + }) + + return { + table: tableName, + syncedRows: externalRows.length, + lastCursorValue: newCursorValue, + skipped: false, + } + } +} From 81e682aade9d7abfb2326441357e79f4a41c1278 Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 18:39:17 -0500 Subject: [PATCH 2/3] test(plugins): add unit tests for ReplicationPlugin (#72) --- plugins/replication/index.test.ts | 174 ++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 plugins/replication/index.test.ts diff --git a/plugins/replication/index.test.ts b/plugins/replication/index.test.ts new file mode 100644 index 0000000..79ad3ae --- /dev/null +++ b/plugins/replication/index.test.ts @@ -0,0 +1,174 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { ReplicationPlugin } from './index' +import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler' +import { DataSource } from '../../src/types' +import * as operation from '../../src/operation' + +vi.mock('../../src/operation', () => ({ + executeExternalQuery: vi.fn(), +})) + +describe('ReplicationPlugin', () => { + let plugin: ReplicationPlugin + let mockDataSource: DataSource + let mockConfig: StarbaseDBConfiguration + + beforeEach(() => { + vi.clearAllMocks() + + mockDataSource = { + rpc: { + executeQuery: vi.fn().mockResolvedValue([]), + }, + source: 'internal', + } as unknown as DataSource + + mockConfig = { + role: 'admin', + outerbaseApiKey: 'test-key', + } + + plugin = new ReplicationPlugin({ + tables: [ + { + table: 'users', + cursorColumn: 'id', + primaryKey: 'id', + intervalSeconds: 60, + batchSize: 100, + }, + 'orders', + ], + intervalSeconds: 300, + batchSize: 500, + }) + }) + + it('should initialize with normalized table configs', () => { + expect(plugin.name).toBe('starbasedb:replication') + expect(plugin.pathPrefix).toBe('/replication') + expect(plugin.tables).toHaveLength(2) + expect(plugin.tables[0].table).toBe('users') + expect(plugin.tables[0].cursorColumn).toBe('id') + expect(plugin.tables[0].intervalSeconds).toBe(60) + expect(plugin.tables[1].table).toBe('orders') + expect(plugin.tables[1].intervalSeconds).toBe(300) + }) + + it('should create replication state table on register middleware', async () => { + const mockApp = { + use: vi.fn((_path, middleware) => + middleware( + { + get: vi.fn((key: string) => + key === 'dataSource' ? mockDataSource : mockConfig + ), + }, + vi.fn() + ) + ), + get: vi.fn(), + post: vi.fn(), + } as unknown as StarbaseApp + + await plugin.register(mockApp) + + expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith({ + sql: expect.stringContaining( + 'CREATE TABLE IF NOT EXISTS tmp_replication_state' + ), + params: [], + }) + }) + + it('should pull external rows and insert into internal SQLite with cursor tracking', async () => { + plugin['dataSource'] = mockDataSource + plugin['config'] = mockConfig + + vi.mocked(operation.executeExternalQuery).mockResolvedValueOnce([ + { id: 10, name: 'Alice', active: true }, + { id: 11, name: 'Bob', active: false }, + ]) + + const res = await plugin.syncTable(plugin.tables[0], true) + + expect(operation.executeExternalQuery).toHaveBeenCalledWith({ + sql: 'SELECT * FROM users ORDER BY id ASC LIMIT 100', + params: [], + dataSource: mockDataSource, + config: mockConfig, + }) + expect(res).toEqual({ + table: 'users', + syncedRows: 2, + lastCursorValue: '11', + skipped: false, + }) + expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith({ + sql: expect.stringContaining('CREATE TABLE IF NOT EXISTS "users"'), + params: [], + }) + expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith({ + sql: 'INSERT OR REPLACE INTO "users" ("id", "name", "active") VALUES (?, ?, ?)', + params: [10, 'Alice', 1], + }) + expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith({ + sql: 'INSERT OR REPLACE INTO "users" ("id", "name", "active") VALUES (?, ?, ?)', + params: [11, 'Bob', 0], + }) + }) + + it('should use incremental WHERE cursorColumn > ? when last_cursor_value exists', async () => { + plugin['dataSource'] = mockDataSource + plugin['config'] = mockConfig + + vi.mocked(mockDataSource.rpc.executeQuery) + .mockResolvedValueOnce([]) // CREATE_STATE_TABLE + .mockResolvedValueOnce([ + { + table_name: 'users', + cursor_column: 'id', + last_cursor_value: '11', + last_synced_at: 1000, + rows_synced: 2, + }, + ]) + + vi.mocked(operation.executeExternalQuery).mockResolvedValueOnce([ + { id: 12, name: 'Charlie', active: true }, + ]) + + const res = await plugin.syncTable(plugin.tables[0], false) + + expect(operation.executeExternalQuery).toHaveBeenCalledWith({ + sql: 'SELECT * FROM users WHERE id > ? ORDER BY id ASC LIMIT 100', + params: ['11'], + dataSource: mockDataSource, + config: mockConfig, + }) + expect(res.syncedRows).toBe(1) + expect(res.lastCursorValue).toBe('12') + }) + + it('should skip sync when within configured intervalSeconds and force is false', async () => { + plugin['dataSource'] = mockDataSource + plugin['config'] = mockConfig + + vi.mocked(mockDataSource.rpc.executeQuery) + .mockResolvedValueOnce([]) // CREATE_STATE_TABLE + .mockResolvedValueOnce([ + { + table_name: 'users', + cursor_column: 'id', + last_cursor_value: '12', + last_synced_at: Date.now(), + rows_synced: 3, + }, + ]) + + const res = await plugin.syncTable(plugin.tables[0], false) + + expect(res.skipped).toBe(true) + expect(operation.executeExternalQuery).not.toHaveBeenCalled() + }) +}) From e6fce4afa8263505c2006eabefc87b9c83d786d9 Mon Sep 17 00:00:00 2001 From: jeffrey-graham Date: Fri, 6 Nov 2026 18:40:37 -0500 Subject: [PATCH 3/3] feat(plugins): export ReplicationPlugin from dist/plugins.ts (#72) --- dist/plugins.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/dist/plugins.ts b/dist/plugins.ts index 7dd252a..0a5cac5 100644 --- a/dist/plugins.ts +++ b/dist/plugins.ts @@ -6,3 +6,4 @@ export { ChangeDataCapturePlugin } from '../plugins/cdc' export { QueryLogPlugin } from '../plugins/query-log' export { ResendPlugin } from '../plugins/resend' export { ClerkPlugin } from '../plugins/clerk' +export { ReplicationPlugin } from '../plugins/replication'