diff --git a/dist/plugins.ts b/dist/plugins.ts
index 7dd252a..6914a0d 100644
--- a/dist/plugins.ts
+++ b/dist/plugins.ts
@@ -6,3 +6,13 @@ export { ChangeDataCapturePlugin } from '../plugins/cdc'
 export { QueryLogPlugin } from '../plugins/query-log'
 export { ResendPlugin } from '../plugins/resend'
 export { ClerkPlugin } from '../plugins/clerk'
+export {
+    DataSyncPlugin,
+    ReplicationPlugin,
+    DatabaseSyncSource,
+    SyncAdapter,
+    PostgresSyncSource,
+    PostgresSync,
+    MySQLSyncSource,
+    MySQLSync,
+} from '../plugins/data-sync'
diff --git a/plugins/data-sync/README.md b/plugins/data-sync/README.md
new file mode 100644
index 0000000..fa64342
--- /dev/null
+++ b/plugins/data-sync/README.md
@@ -0,0 +1,46 @@
+# Data Sync / Replication Plugin
+
+The Data Sync (`DataSyncPlugin` / `ReplicationPlugin`) plugin enables automatic pull-based replication from external data sources (PostgreSQL, MySQL, etc.) into StarbaseDB's internal Durable Object SQLite database so your instance serves as a fast, close-to-edge replica.
+
+## Features
+
+- **Pluggable Database Adapters**: Built-in `PostgresSyncSource` and `MySQLSyncSource` extending `DatabaseSyncSource` (`SyncAdapter`) to isolate dialect-specific schema introspection, column type mapping, and pagination queries.
+- **Configurable Sync Interval & Tables**: Define `sync_interval` (in seconds) and specific tables (either by string name or per-table `TableSyncConfig` with custom `trackingColumn`, `primaryKey`, `schema`, and `batchSize`).
+- **Incremental Checkpointing**: Automatically tracks the latest value of `trackingColumn` (`updated_at`, `created_at`, primary key, or any custom column) in `tmp_data_sync_metadata` for append/update polling.
+- **Schema Resolution & `beforeQuery` Rewriting**:
+  - Default `public` schema tables are created at the SQLite root (`public.users` -> `users`), and queries to `public.users` on the internal source are automatically rewritten to `users`.
+  - Custom schema tables (`analytics.events`) are mapped to `analytics_events` and transparently rewritten in `beforeQuery`.
+- **Internal `tmp_` Table Convention**: Stores sync checkpoints in `tmp_data_sync_metadata` and error logs in `tmp_data_sync_errors`.
+- **Admin Endpoints**:
+  - `GET /data-sync/status`: Returns per-table sync checkpoints and recent errors.
+  - `POST /data-sync/trigger` (or `POST /data-sync/sync`): Manually triggers replication for all tables or a specific `{ "table": "users" }`.
+
+## Usage
+
+```typescript
+import {
+    DataSyncPlugin,
+    PostgresSyncSource,
+} from '../plugins/data-sync'
+
+const syncStore = new PostgresSyncSource({
+    dialect: 'postgresql',
+})
+
+const plugins = [
+    new DataSyncPlugin(syncStore, {
+        sync_interval: 60,
+        tables: [
+            'users',
+            'person',
+            {
+                table: 'events',
+                schema: 'analytics',
+                trackingColumn: 'event_timestamp',
+                primaryKey: 'event_id',
+                batchSize: 500,
+            },
+        ],
+    }),
+] satisfies StarbasePlugin[]
+```
diff --git a/plugins/data-sync/index.test.ts b/plugins/data-sync/index.test.ts
new file mode 100644
index 0000000..47e760c
--- /dev/null
+++ b/plugins/data-sync/index.test.ts
@@ -0,0 +1,479 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import {
+    DataSyncPlugin,
+    ReplicationPlugin,
+    PostgresSyncSource,
+    MySQLSyncSource,
+    normalizeExternalRows,
+    serializeSqliteValue,
+    quoteSqlIdentifier,
+} from './index'
+import { executeExternalQuery } from '../../src/operation'
+import type { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler'
+import type { DataSource } from '../../src/types'
+
+vi.mock('../../src/operation', () => ({
+    executeExternalQuery: vi.fn(),
+}))
+
+describe('DataSyncPlugin / ReplicationPlugin (#72)', () => {
+    let mockDataSource: DataSource
+    let mockConfig: StarbaseDBConfiguration
+    let executedInternalQueries: Array<{ sql: string; params?: any[] }>
+
+    beforeEach(() => {
+        vi.clearAllMocks()
+        executedInternalQueries = []
+
+        mockDataSource = {
+            source: 'internal',
+            external: {
+                dialect: 'postgresql',
+                host: 'localhost',
+                port: 5432,
+                user: 'postgres',
+                password: 'password',
+                database: 'postgres',
+            },
+            rpc: {
+                executeQuery: vi.fn(
+                    async (opts: { sql: string; params?: any[] }) => {
+                        executedInternalQueries.push(opts)
+                        if (
+                            opts.sql.includes(
+                                'FROM tmp_data_sync_metadata\n        WHERE table_name = ?'
+                            )
+                        ) {
+                            return []
+                        }
+                        if (opts.sql.includes('FROM tmp_data_sync_metadata')) {
+                            return [
+                                {
+                                    table_name: 'users',
+                                    source_schema: 'public',
+                                    source_table: 'users',
+                                    tracking_column: 'updated_at',
+                                    last_sync_timestamp: 1700000000000,
+                                    last_sync_value: '2025-01-01T00:00:00Z',
+                                    rows_synced: 2,
+                                    status: 'synced',
+                                    last_error: null,
+                                },
+                            ]
+                        }
+                        if (opts.sql.includes('FROM tmp_data_sync_errors')) {
+                            return []
+                        }
+                        if (opts.sql.startsWith('PRAGMA table_info')) {
+                            return []
+                        }
+                        return []
+                    }
+                ),
+            },
+        } as unknown as DataSource
+
+        mockConfig = {
+            role: 'admin',
+            outerbaseApiKey: 'mock-key',
+        }
+    })
+
+    describe('PostgresSyncSource & MySQLSyncSource Adapters', () => {
+        it('maps PostgreSQL and MySQL column types to SQLite affinities accurately', () => {
+            const pg = new PostgresSyncSource({ dialect: 'postgresql' })
+            expect(pg.mapColumnType('integer')).toBe('INTEGER')
+            expect(pg.mapColumnType('bigint')).toBe('INTEGER')
+            expect(pg.mapColumnType('boolean')).toBe('INTEGER')
+            expect(pg.mapColumnType('numeric')).toBe('REAL')
+            expect(pg.mapColumnType('double precision')).toBe('REAL')
+            expect(pg.mapColumnType('bytea')).toBe('BLOB')
+            expect(pg.mapColumnType('character varying')).toBe('TEXT')
+            expect(pg.mapColumnType('timestamp with time zone')).toBe('TEXT')
+
+            const mysql = new MySQLSyncSource({ dialect: 'mysql' })
+            expect(mysql.mapColumnType('tinyint')).toBe('INTEGER')
+            expect(mysql.mapColumnType('decimal')).toBe('REAL')
+            expect(mysql.mapColumnType('longblob')).toBe('BLOB')
+            expect(mysql.mapColumnType('varchar')).toBe('TEXT')
+        })
+
+        it('resolves public and custom schema table names without dots', () => {
+            const pg = new PostgresSyncSource()
+            expect(pg.parseTableIdentifier('users')).toEqual({
+                sourceSchema: 'public',
+                sourceTable: 'users',
+                internalTable: 'users',
+            })
+            expect(pg.parseTableIdentifier('public.users')).toEqual({
+                sourceSchema: 'public',
+                sourceTable: 'users',
+                internalTable: 'users',
+            })
+            expect(pg.parseTableIdentifier('analytics.events')).toEqual({
+                sourceSchema: 'analytics',
+                sourceTable: 'events',
+                internalTable: 'analytics_events',
+            })
+        })
+
+        it('builds incremental and full fetch queries for both Postgres and MySQL', () => {
+            const pg = new PostgresSyncSource()
+            expect(
+                pg.getFetchRecordsQuery({
+                    schema: 'public',
+                    table: 'users',
+                    trackingColumn: 'id',
+                    lastValue: 10,
+                    batchSize: 100,
+                    offset: 0,
+                }).sql
+            ).toBe(
+                'SELECT * FROM users WHERE id > 10 ORDER BY id ASC LIMIT 100 OFFSET 0'
+            )
+
+            const mysql = new MySQLSyncSource()
+            expect(
+                mysql.getFetchRecordsQuery({
+                    schema: 'analytics',
+                    table: 'events',
+                    trackingColumn: 'created_at',
+                    lastValue: '2025-01-01',
+                    batchSize: 50,
+                    offset: 0,
+                }).sql
+            ).toBe(
+                "SELECT * FROM analytics.events WHERE created_at > '2025-01-01' ORDER BY created_at ASC LIMIT 50 OFFSET 0"
+            )
+        })
+    })
+
+    describe('DataSyncPlugin Replication Lifecycle', () => {
+        it('replicates multiple tables from Postgres into internal SQLite with tmp_ metadata tables', async () => {
+            const syncStore = new PostgresSyncSource({ dialect: 'postgresql' })
+            const plugin = new DataSyncPlugin(syncStore, {
+                sync_interval: 60,
+                tables: ['users', 'person'],
+            })
+            plugin['dataSource'] = mockDataSource
+            plugin['config'] = mockConfig
+
+            vi.mocked(executeExternalQuery)
+                // 1. users schema
+                .mockResolvedValueOnce([
+                    {
+                        column_name: 'id',
+                        data_type: 'integer',
+                        is_nullable: 'NO',
+                        is_primary_key: 1,
+                    },
+                    {
+                        column_name: 'name',
+                        data_type: 'text',
+                        is_nullable: 'YES',
+                        is_primary_key: 0,
+                    },
+                    {
+                        column_name: 'updated_at',
+                        data_type: 'timestamp',
+                        is_nullable: 'YES',
+                        is_primary_key: 0,
+                    },
+                ])
+                // 2. users rows
+                .mockResolvedValueOnce([
+                    {
+                        id: 1,
+                        name: 'Alice',
+                        updated_at: '2025-01-01T10:00:00Z',
+                    },
+                    {
+                        id: 2,
+                        name: 'Bob',
+                        updated_at: '2025-01-01T11:00:00Z',
+                    },
+                ])
+                // 3. person schema
+                .mockResolvedValueOnce([
+                    {
+                        column_name: 'person_id',
+                        data_type: 'integer',
+                        is_nullable: 'NO',
+                        is_primary_key: 1,
+                    },
+                    {
+                        column_name: 'email',
+                        data_type: 'varchar',
+                        is_nullable: 'NO',
+                        is_primary_key: 0,
+                    },
+                ])
+                // 4. person rows
+                .mockResolvedValueOnce([
+                    { person_id: 101, email: 'alice@example.com' },
+                ])
+
+            const results = await plugin.syncAllTables()
+
+            expect(results).toEqual([
+                { table: 'users', rowsSynced: 2, status: 'synced' },
+                { table: 'person', rowsSynced: 1, status: 'synced' },
+            ])
+
+            const sqlStatements = executedInternalQueries.map((q) => q.sql)
+
+            // Verify tmp_ prefix on internal metadata and error tables
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'CREATE TABLE IF NOT EXISTS tmp_data_sync_metadata'
+                    )
+                )
+            ).toBe(true)
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'CREATE TABLE IF NOT EXISTS tmp_data_sync_errors'
+                    )
+                )
+            ).toBe(true)
+
+            // Verify internal tables are named `users` and `person` (not `users.table`)
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, updated_at TEXT)'
+                    )
+                )
+            ).toBe(true)
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'CREATE TABLE IF NOT EXISTS person (person_id INTEGER PRIMARY KEY, email TEXT NOT NULL)'
+                    )
+                )
+            ).toBe(true)
+
+            // Verify upserts
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'INSERT OR REPLACE INTO users (id, name, updated_at) VALUES (?, ?, ?)'
+                    )
+                )
+            ).toBe(true)
+            expect(
+                sqlStatements.some((s) =>
+                    s.includes(
+                        'INSERT OR REPLACE INTO person (person_id, email) VALUES (?, ?)'
+                    )
+                )
+            ).toBe(true)
+        })
+
+        it('supports custom trackingColumn and schema expansion when new columns appear', async () => {
+            const plugin = new ReplicationPlugin({
+                sync_interval: 120,
+                tables: [
+                    {
+                        table: 'events',
+                        schema: 'analytics',
+                        trackingColumn: 'seq_num',
+                        primaryKey: 'event_uuid',
+                        batchSize: 2,
+                    },
+                ],
+            })
+            plugin['dataSource'] = mockDataSource
+            plugin['config'] = mockConfig
+
+            // Simulate existing table missing the `payload` column in PRAGMA table_info
+            vi.mocked(mockDataSource.rpc.executeQuery).mockImplementation(
+                async (opts: { sql: string; params?: any[] }) => {
+                    executedInternalQueries.push(opts)
+                    if (opts.sql.startsWith('PRAGMA table_info')) {
+                        return [{ name: 'event_uuid' }, { name: 'seq_num' }]
+                    }
+                    return []
+                }
+            )
+
+            vi.mocked(executeExternalQuery)
+                .mockResolvedValueOnce([
+                    {
+                        column_name: 'event_uuid',
+                        data_type: 'uuid',
+                        is_nullable: 'NO',
+                        is_primary_key: 0,
+                    },
+                    {
+                        column_name: 'seq_num',
+                        data_type: 'bigint',
+                        is_nullable: 'NO',
+                        is_primary_key: 0,
+                    },
+                    {
+                        column_name: 'payload',
+                        data_type: 'jsonb',
+                        is_nullable: 'YES',
+                        is_primary_key: 0,
+                    },
+                ])
+                // Batch 1 (full batch of 2)
+                .mockResolvedValueOnce([
+                    {
+                        event_uuid: 'e1',
+                        seq_num: 100,
+                        payload: { action: 'login' },
+                    },
+                    {
+                        event_uuid: 'e2',
+                        seq_num: 101,
+                        payload: { action: 'view' },
+                    },
+                ])
+                // Batch 2 (1 row -> end)
+                .mockResolvedValueOnce([
+                    {
+                        event_uuid: 'e3',
+                        seq_num: 102,
+                        payload: { action: 'logout' },
+                    },
+                ])
+
+            const results = await plugin.syncAllTables()
+            expect(results).toEqual([
+                { table: 'analytics_events', rowsSynced: 3, status: 'synced' },
+            ])
+
+            // Verify ALTER TABLE added the missing `payload` column
+            expect(
+                executedInternalQueries.some(
+                    (q) =>
+                        q.sql ===
+                        'ALTER TABLE analytics_events ADD COLUMN payload TEXT'
+                )
+            ).toBe(true)
+
+            // Verify second batch used checkpoint `seq_num > '101'`
+            expect(executeExternalQuery).toHaveBeenNthCalledWith(
+                3,
+                expect.objectContaining({
+                    sql: "SELECT * FROM analytics.events WHERE seq_num > '101' ORDER BY seq_num ASC LIMIT 2 OFFSET 0",
+                })
+            )
+        })
+
+        it('rewrites public.<table> and <schema>.<table> in beforeQuery for internal queries', async () => {
+            const plugin = new DataSyncPlugin({
+                tables: ['public.users', 'analytics.events'],
+            })
+
+            const q1 = await plugin.beforeQuery({
+                sql: 'SELECT * FROM public.users WHERE id = 1',
+                params: [],
+                dataSource: { source: 'internal' } as DataSource,
+            })
+            expect(q1.sql).toBe('SELECT * FROM users WHERE id = 1')
+
+            const q2 = await plugin.beforeQuery({
+                sql: 'SELECT * FROM "public"."users" JOIN analytics.events ON users.id = analytics.events.user_id',
+                params: [],
+                dataSource: { source: 'internal' } as DataSource,
+            })
+            expect(q2.sql).toBe(
+                'SELECT * FROM "users" JOIN analytics_events ON users.id = analytics_events.user_id'
+            )
+
+            // Does not rewrite when querying external source directly
+            const qExternal = await plugin.beforeQuery({
+                sql: 'SELECT * FROM public.users',
+                params: [],
+                dataSource: { source: 'external' } as DataSource,
+            })
+            expect(qExternal.sql).toBe('SELECT * FROM public.users')
+        })
+
+        it('logs errors to tmp_data_sync_errors when external query fails', async () => {
+            const plugin = new DataSyncPlugin({
+                tables: ['broken_table'],
+            })
+            plugin['dataSource'] = mockDataSource
+            plugin['config'] = mockConfig
+
+            vi.mocked(executeExternalQuery).mockRejectedValueOnce(
+                new Error('Connection refused')
+            )
+
+            const results = await plugin.syncAllTables()
+            expect(results).toEqual([
+                {
+                    table: 'broken_table',
+                    rowsSynced: 0,
+                    status: 'error',
+                    error: 'Connection refused',
+                },
+            ])
+
+            expect(
+                executedInternalQueries.some((q) =>
+                    q.sql.includes('INSERT INTO tmp_data_sync_errors')
+                )
+            ).toBe(true)
+        })
+
+        it('registers routes for /data-sync/status and /data-sync/trigger', async () => {
+            const routes: Record<string, Function> = {}
+            const mockApp = {
+                use: vi.fn(),
+                get: vi.fn((path: string, handler: Function) => {
+                    routes[`GET:${path}`] = handler
+                }),
+                post: vi.fn((path: string, handler: Function) => {
+                    routes[`POST:${path}`] = handler
+                }),
+            } as unknown as StarbaseApp
+
+            const plugin = new DataSyncPlugin({ tables: ['users'] })
+            plugin['dataSource'] = mockDataSource
+            plugin['config'] = mockConfig
+
+            await plugin.register(mockApp)
+
+            const statusRes: Response = await routes['GET:/data-sync/status']()
+            expect(statusRes.status).toBe(200)
+            const statusBody = (await statusRes.json()) as any
+            expect(statusBody.result.tables).toHaveLength(1)
+
+            vi.mocked(executeExternalQuery)
+                .mockResolvedValueOnce([])
+                .mockResolvedValueOnce([])
+
+            const triggerRes: Response = await routes[
+                'POST:/data-sync/trigger'
+            ]({
+                req: {
+                    json: async () => ({ table: 'users' }),
+                },
+            })
+            expect(triggerRes.status).toBe(200)
+        })
+    })
+
+    describe('Helper Utilities', () => {
+        it('normalizes raw and object external rows and serializes values for SQLite', () => {
+            expect(
+                normalizeExternalRows({
+                    columns: ['id', 'name'],
+                    rows: [[1, 'Alice']],
+                })
+            ).toEqual([{ id: 1, name: 'Alice' }])
+
+            expect(serializeSqliteValue(true)).toBe(1)
+            expect(serializeSqliteValue(false)).toBe(0)
+            expect(serializeSqliteValue({ a: 1 })).toBe('{"a":1}')
+            expect(quoteSqlIdentifier('users')).toBe('users')
+            expect(quoteSqlIdentifier('users.table')).toBe('"users.table"')
+        })
+    })
+})
diff --git a/plugins/data-sync/index.ts b/plugins/data-sync/index.ts
new file mode 100644
index 0000000..b96d4f3
--- /dev/null
+++ b/plugins/data-sync/index.ts
@@ -0,0 +1,1085 @@
+import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler'
+import { executeExternalQuery } from '../../src/operation'
+import { StarbasePlugin } from '../../src/plugin'
+import { DataSource } from '../../src/types'
+import { createResponse } from '../../src/utils'
+
+export type SqliteColumnType = 'INTEGER' | 'REAL' | 'TEXT' | 'BLOB'
+
+export interface ColumnDefinition {
+    name: string
+    type: string
+    sqliteType: SqliteColumnType
+    nullable: boolean
+    isPrimaryKey: boolean
+}
+
+export interface TableSyncConfig {
+    table: string
+    schema?: string
+    trackingColumn?: string
+    primaryKey?: string
+    batchSize?: number
+}
+
+export interface NormalizedTableSyncConfig {
+    sourceSchema: string
+    sourceTable: string
+    internalTable: string
+    trackingColumn?: string
+    primaryKey?: string
+    batchSize: number
+}
+
+export interface DataSyncOptions {
+    sync_interval?: number
+    syncInterval?: number
+    batch_size?: number
+    batchSize?: number
+    tables?: Array<string | TableSyncConfig>
+    source?: DatabaseSyncSource
+    autoSyncOnRequest?: boolean
+}
+
+export interface TableSyncMetadata {
+    table_name: string
+    source_schema: string
+    source_table: string
+    tracking_column: string | null
+    last_sync_timestamp: number
+    last_sync_value: string | null
+    rows_synced: number
+    status: string
+    last_error: string | null
+}
+
+const SQL_QUERIES = {
+    CREATE_METADATA_TABLE: `
+        CREATE TABLE IF NOT EXISTS tmp_data_sync_metadata (
+            table_name TEXT PRIMARY KEY,
+            source_schema TEXT NOT NULL,
+            source_table TEXT NOT NULL,
+            tracking_column TEXT,
+            last_sync_timestamp INTEGER NOT NULL DEFAULT 0,
+            last_sync_value TEXT,
+            rows_synced INTEGER NOT NULL DEFAULT 0,
+            status TEXT NOT NULL DEFAULT 'idle',
+            last_error TEXT
+        )
+    `,
+    CREATE_ERRORS_TABLE: `
+        CREATE TABLE IF NOT EXISTS tmp_data_sync_errors (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            table_name TEXT NOT NULL,
+            error_message TEXT NOT NULL,
+            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+        )
+    `,
+    GET_ALL_METADATA: `
+        SELECT
+            table_name,
+            source_schema,
+            source_table,
+            tracking_column,
+            last_sync_timestamp,
+            last_sync_value,
+            rows_synced,
+            status,
+            last_error
+        FROM tmp_data_sync_metadata
+    `,
+    GET_TABLE_METADATA: `
+        SELECT
+            table_name,
+            source_schema,
+            source_table,
+            tracking_column,
+            last_sync_timestamp,
+            last_sync_value,
+            rows_synced,
+            status,
+            last_error
+        FROM tmp_data_sync_metadata
+        WHERE table_name = ?
+    `,
+    UPSERT_METADATA: `
+        INSERT OR REPLACE INTO tmp_data_sync_metadata (
+            table_name,
+            source_schema,
+            source_table,
+            tracking_column,
+            last_sync_timestamp,
+            last_sync_value,
+            rows_synced,
+            status,
+            last_error
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+    `,
+    INSERT_ERROR: `
+        INSERT INTO tmp_data_sync_errors (table_name, error_message)
+        VALUES (?, ?)
+    `,
+    GET_RECENT_ERRORS: `
+        SELECT id, table_name, error_message, created_at
+        FROM tmp_data_sync_errors
+        ORDER BY id DESC
+        LIMIT 20
+    `,
+}
+
+export function quoteSqlIdentifier(identifier: string): string {
+    if (/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(identifier)) {
+        return identifier
+    }
+    return `"${identifier.replace(/"/g, '""')}"`
+}
+
+export function normalizeExternalRows(rawResult: any): Record<string, any>[] {
+    if (!rawResult) {
+        return []
+    }
+    if (Array.isArray(rawResult)) {
+        return rawResult
+    }
+    if (Array.isArray(rawResult.rows) && Array.isArray(rawResult.columns)) {
+        return rawResult.rows.map((row: any) => {
+            if (row && typeof row === 'object' && !Array.isArray(row)) {
+                return row
+            }
+            return rawResult.columns.reduce(
+                (acc: Record<string, any>, col: string, idx: number) => {
+                    acc[col] = row?.[idx]
+                    return acc
+                },
+                {}
+            )
+        })
+    }
+    if (Array.isArray(rawResult.rows)) {
+        return rawResult.rows
+    }
+    if (Array.isArray(rawResult.items)) {
+        return rawResult.items
+    }
+    return []
+}
+
+export function serializeSqliteValue(value: unknown): unknown {
+    if (value === undefined || value === null) {
+        return null
+    }
+    if (typeof value === 'boolean') {
+        return value ? 1 : 0
+    }
+    if (value instanceof Date) {
+        return value.toISOString()
+    }
+    if (
+        typeof value === 'object' &&
+        !(value instanceof Uint8Array) &&
+        !(value instanceof ArrayBuffer)
+    ) {
+        return JSON.stringify(value)
+    }
+    return value
+}
+
+export abstract class DatabaseSyncSource {
+    public readonly dialect: 'postgresql' | 'mysql' | 'sqlite'
+    public readonly defaultSchema: string
+
+    constructor(opts: {
+        dialect: 'postgresql' | 'mysql' | 'sqlite'
+        defaultSchema?: string
+    }) {
+        this.dialect = opts.dialect
+        this.defaultSchema =
+            opts.defaultSchema ||
+            (opts.dialect === 'postgresql'
+                ? 'public'
+                : opts.dialect === 'mysql'
+                  ? 'public'
+                  : 'main')
+    }
+
+    public abstract mapColumnType(externalType: string): SqliteColumnType
+
+    public abstract getTableSchemaQuery(
+        schema: string,
+        table: string
+    ): { sql: string; params: any[] }
+
+    public abstract parseSchemaRows(
+        rows: Record<string, any>[],
+        primaryKeyOverride?: string
+    ): ColumnDefinition[]
+
+    public abstract getFetchRecordsQuery(opts: {
+        schema: string
+        table: string
+        trackingColumn?: string
+        lastValue?: string | number | null
+        batchSize: number
+        offset: number
+    }): { sql: string; params: any[] }
+
+    public parseTableIdentifier(
+        rawTable: string,
+        explicitSchema?: string
+    ): {
+        sourceSchema: string
+        sourceTable: string
+        internalTable: string
+    } {
+        const trimmed = rawTable.trim()
+        let sourceSchema = explicitSchema?.trim() || this.defaultSchema
+        let sourceTable = trimmed
+
+        if (!explicitSchema && trimmed.includes('.')) {
+            const parts = trimmed.split('.')
+            sourceSchema = parts[0].trim() || this.defaultSchema
+            sourceTable = parts.slice(1).join('_').trim()
+        }
+
+        const isDefaultSchema =
+            !sourceSchema ||
+            sourceSchema.toLowerCase() === this.defaultSchema.toLowerCase() ||
+            sourceSchema.toLowerCase() === 'public' ||
+            sourceSchema.toLowerCase() === 'main'
+
+        const internalTable = isDefaultSchema
+            ? sourceTable
+            : `${sourceSchema}_${sourceTable}`
+
+        return {
+            sourceSchema,
+            sourceTable,
+            internalTable,
+        }
+    }
+}
+
+export class PostgresSyncSource extends DatabaseSyncSource {
+    constructor(opts?: { dialect?: 'postgresql'; defaultSchema?: string }) {
+        super({
+            dialect: 'postgresql',
+            defaultSchema: opts?.defaultSchema || 'public',
+        })
+    }
+
+    public override mapColumnType(externalType: string): SqliteColumnType {
+        const normalized = (externalType || '').toLowerCase().trim()
+
+        if (
+            normalized.includes('int') ||
+            normalized === 'serial' ||
+            normalized === 'bigserial' ||
+            normalized === 'smallserial' ||
+            normalized === 'boolean' ||
+            normalized === 'bool'
+        ) {
+            return 'INTEGER'
+        }
+
+        if (
+            normalized.includes('real') ||
+            normalized.includes('float') ||
+            normalized.includes('double') ||
+            normalized.includes('numeric') ||
+            normalized.includes('decimal') ||
+            normalized === 'money'
+        ) {
+            return 'REAL'
+        }
+
+        if (normalized === 'bytea') {
+            return 'BLOB'
+        }
+
+        return 'TEXT'
+    }
+
+    public override getTableSchemaQuery(
+        schema: string,
+        table: string
+    ): { sql: string; params: any[] } {
+        const escapedSchema = schema.replace(/'/g, "''")
+        const escapedTable = table.replace(/'/g, "''")
+        const sql = `
+            SELECT
+                c.column_name AS column_name,
+                c.data_type AS data_type,
+                c.is_nullable AS is_nullable,
+                CASE WHEN pk.column_name IS NOT NULL THEN 1 ELSE 0 END AS is_primary_key
+            FROM information_schema.columns c
+            LEFT JOIN (
+                SELECT kcu.table_schema, kcu.table_name, kcu.column_name
+                FROM information_schema.table_constraints tc
+                JOIN information_schema.key_column_usage kcu
+                    ON tc.constraint_name = kcu.constraint_name
+                    AND tc.table_schema = kcu.table_schema
+                WHERE tc.constraint_type = 'PRIMARY KEY'
+            ) pk
+                ON c.table_schema = pk.table_schema
+                AND c.table_name = pk.table_name
+                AND c.column_name = pk.column_name
+            WHERE c.table_schema = '${escapedSchema}'
+              AND c.table_name = '${escapedTable}'
+            ORDER BY c.ordinal_position ASC
+        `
+        return { sql: sql.trim(), params: [] }
+    }
+
+    public override parseSchemaRows(
+        rows: Record<string, any>[],
+        primaryKeyOverride?: string
+    ): ColumnDefinition[] {
+        return rows.map((row) => {
+            const name = String(
+                row.column_name ?? row.COLUMN_NAME ?? row.name ?? ''
+            )
+            const rawType = String(
+                row.data_type ?? row.DATA_TYPE ?? row.type ?? 'text'
+            )
+            const nullableRaw =
+                row.is_nullable ?? row.IS_NULLABLE ?? row.nullable ?? 'YES'
+            const isNullable =
+                typeof nullableRaw === 'boolean'
+                    ? nullableRaw
+                    : String(nullableRaw).toUpperCase() !== 'NO'
+
+            const isPkFromRow =
+                Boolean(row.is_primary_key) &&
+                row.is_primary_key !== 0 &&
+                row.is_primary_key !== '0' &&
+                row.is_primary_key !== 'false'
+
+            const isPrimaryKey = primaryKeyOverride
+                ? name === primaryKeyOverride
+                : isPkFromRow
+
+            return {
+                name,
+                type: rawType,
+                sqliteType: this.mapColumnType(rawType),
+                nullable: isNullable,
+                isPrimaryKey,
+            }
+        })
+    }
+
+    public override getFetchRecordsQuery(opts: {
+        schema: string
+        table: string
+        trackingColumn?: string
+        lastValue?: string | number | null
+        batchSize: number
+        offset: number
+    }): { sql: string; params: any[] } {
+        const quotedSchema = quoteSqlIdentifier(opts.schema || 'public')
+        const quotedTable = quoteSqlIdentifier(opts.table)
+        const fullTable =
+            !opts.schema || opts.schema === 'public'
+                ? quotedTable
+                : `${quotedSchema}.${quotedTable}`
+
+        if (opts.trackingColumn) {
+            const quotedCol = quoteSqlIdentifier(opts.trackingColumn)
+            if (opts.lastValue !== undefined && opts.lastValue !== null) {
+                const literal =
+                    typeof opts.lastValue === 'number'
+                        ? String(opts.lastValue)
+                        : `'${String(opts.lastValue).replace(/'/g, "''")}'`
+                return {
+                    sql: `SELECT * FROM ${fullTable} WHERE ${quotedCol} > ${literal} ORDER BY ${quotedCol} ASC LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+                    params: [],
+                }
+            }
+            return {
+                sql: `SELECT * FROM ${fullTable} ORDER BY ${quotedCol} ASC LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+                params: [],
+            }
+        }
+
+        return {
+            sql: `SELECT * FROM ${fullTable} LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+            params: [],
+        }
+    }
+}
+
+export class MySQLSyncSource extends DatabaseSyncSource {
+    constructor(opts?: { dialect?: 'mysql'; defaultSchema?: string }) {
+        super({
+            dialect: 'mysql',
+            defaultSchema: opts?.defaultSchema || 'public',
+        })
+    }
+
+    public override mapColumnType(externalType: string): SqliteColumnType {
+        const normalized = (externalType || '').toLowerCase().trim()
+
+        if (
+            normalized.includes('int') ||
+            normalized === 'bit' ||
+            normalized === 'boolean' ||
+            normalized === 'bool'
+        ) {
+            return 'INTEGER'
+        }
+
+        if (
+            normalized.includes('float') ||
+            normalized.includes('double') ||
+            normalized.includes('decimal') ||
+            normalized.includes('numeric')
+        ) {
+            return 'REAL'
+        }
+
+        if (normalized.includes('blob') || normalized.includes('binary')) {
+            return 'BLOB'
+        }
+
+        return 'TEXT'
+    }
+
+    public override getTableSchemaQuery(
+        schema: string,
+        table: string
+    ): { sql: string; params: any[] } {
+        const escapedTable = table.replace(/'/g, "''")
+        const schemaFilter =
+            schema && schema !== 'public'
+                ? `TABLE_SCHEMA = '${schema.replace(/'/g, "''")}'`
+                : 'TABLE_SCHEMA = DATABASE()'
+
+        const sql = `
+            SELECT
+                COLUMN_NAME AS column_name,
+                DATA_TYPE AS data_type,
+                IS_NULLABLE AS is_nullable,
+                CASE WHEN COLUMN_KEY = 'PRI' THEN 1 ELSE 0 END AS is_primary_key
+            FROM information_schema.columns
+            WHERE ${schemaFilter}
+              AND TABLE_NAME = '${escapedTable}'
+            ORDER BY ORDINAL_POSITION ASC
+        `
+        return { sql: sql.trim(), params: [] }
+    }
+
+    public override parseSchemaRows(
+        rows: Record<string, any>[],
+        primaryKeyOverride?: string
+    ): ColumnDefinition[] {
+        return rows.map((row) => {
+            const name = String(
+                row.column_name ?? row.COLUMN_NAME ?? row.name ?? ''
+            )
+            const rawType = String(
+                row.data_type ?? row.DATA_TYPE ?? row.type ?? 'text'
+            )
+            const nullableRaw =
+                row.is_nullable ?? row.IS_NULLABLE ?? row.nullable ?? 'YES'
+            const isNullable =
+                typeof nullableRaw === 'boolean'
+                    ? nullableRaw
+                    : String(nullableRaw).toUpperCase() !== 'NO'
+
+            const isPkFromRow =
+                Boolean(row.is_primary_key) &&
+                row.is_primary_key !== 0 &&
+                row.is_primary_key !== '0' &&
+                row.is_primary_key !== 'false'
+
+            return {
+                name,
+                type: rawType,
+                sqliteType: this.mapColumnType(rawType),
+                nullable: isNullable,
+                isPrimaryKey: primaryKeyOverride
+                    ? name === primaryKeyOverride
+                    : isPkFromRow,
+            }
+        })
+    }
+
+    public override getFetchRecordsQuery(opts: {
+        schema: string
+        table: string
+        trackingColumn?: string
+        lastValue?: string | number | null
+        batchSize: number
+        offset: number
+    }): { sql: string; params: any[] } {
+        const quotedTable = quoteSqlIdentifier(opts.table)
+        const fullTable =
+            !opts.schema || opts.schema === 'public'
+                ? quotedTable
+                : `${quoteSqlIdentifier(opts.schema)}.${quotedTable}`
+
+        if (opts.trackingColumn) {
+            const quotedCol = quoteSqlIdentifier(opts.trackingColumn)
+            if (opts.lastValue !== undefined && opts.lastValue !== null) {
+                const literal =
+                    typeof opts.lastValue === 'number'
+                        ? String(opts.lastValue)
+                        : `'${String(opts.lastValue).replace(/'/g, "''")}'`
+                return {
+                    sql: `SELECT * FROM ${fullTable} WHERE ${quotedCol} > ${literal} ORDER BY ${quotedCol} ASC LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+                    params: [],
+                }
+            }
+            return {
+                sql: `SELECT * FROM ${fullTable} ORDER BY ${quotedCol} ASC LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+                params: [],
+            }
+        }
+
+        return {
+            sql: `SELECT * FROM ${fullTable} LIMIT ${opts.batchSize} OFFSET ${opts.offset}`,
+            params: [],
+        }
+    }
+}
+
+export class DataSyncPlugin extends StarbasePlugin {
+    public override pathPrefix: string = '/data-sync'
+    private syncSource: DatabaseSyncSource
+    private syncIntervalMs: number
+    private defaultBatchSize: number
+    private rawTables: Array<string | TableSyncConfig>
+    private autoSyncOnRequest: boolean
+    private dataSource?: DataSource
+    private config?: StarbaseDBConfiguration
+    private initializedTables: boolean = false
+
+    constructor(
+        sourceOrOpts?: DatabaseSyncSource | DataSyncOptions,
+        maybeOpts?: DataSyncOptions
+    ) {
+        super(
+            'starbasedb:data-sync',
+            {
+                requiresAuth: true,
+            },
+            '/data-sync'
+        )
+
+        let resolvedSource: DatabaseSyncSource | undefined
+        let resolvedOpts: DataSyncOptions = {}
+
+        if (sourceOrOpts instanceof DatabaseSyncSource) {
+            resolvedSource = sourceOrOpts
+            resolvedOpts = maybeOpts || {}
+        } else if (sourceOrOpts && typeof sourceOrOpts === 'object') {
+            resolvedOpts = sourceOrOpts
+            resolvedSource = sourceOrOpts.source
+        }
+
+        this.syncSource = resolvedSource || new PostgresSyncSource()
+        const intervalSeconds =
+            resolvedOpts.sync_interval ?? resolvedOpts.syncInterval ?? 300
+        this.syncIntervalMs = Math.max(1, intervalSeconds) * 1000
+        this.defaultBatchSize =
+            resolvedOpts.batch_size ?? resolvedOpts.batchSize ?? 500
+        this.rawTables = resolvedOpts.tables || []
+        this.autoSyncOnRequest = resolvedOpts.autoSyncOnRequest ?? true
+    }
+
+    public getNormalizedTables(): NormalizedTableSyncConfig[] {
+        return this.rawTables.map((entry) => {
+            if (typeof entry === 'string') {
+                const parsed = this.syncSource.parseTableIdentifier(entry)
+                return {
+                    ...parsed,
+                    batchSize: this.defaultBatchSize,
+                }
+            }
+            const parsed = this.syncSource.parseTableIdentifier(
+                entry.table,
+                entry.schema
+            )
+            return {
+                ...parsed,
+                trackingColumn: entry.trackingColumn,
+                primaryKey: entry.primaryKey,
+                batchSize: entry.batchSize ?? this.defaultBatchSize,
+            }
+        })
+    }
+
+    override async register(app: StarbaseApp): Promise<void> {
+        app.use('*', async (c, next) => {
+            this.dataSource = c?.get('dataSource')
+            this.config = c?.get('config')
+            await this.ensureMetadataTables()
+
+            if (this.autoSyncOnRequest && this.dataSource) {
+                const syncPromise = this.syncDueTables()
+                if (this.dataSource.executionContext?.waitUntil) {
+                    this.dataSource.executionContext.waitUntil(syncPromise)
+                } else {
+                    await syncPromise
+                }
+            }
+
+            await next()
+        })
+
+        app.get(`${this.pathPrefix}/status`, async () => {
+            if (this.config && this.config.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized request', 403)
+            }
+            const status = await this.getSyncStatus()
+            return createResponse(status, undefined, 200)
+        })
+
+        const handleManualSync = async (c: any) => {
+            if (this.config && this.config.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized request', 403)
+            }
+
+            let targetTable: string | undefined
+            try {
+                const body = await c.req.json()
+                if (body && typeof body.table === 'string') {
+                    targetTable = body.table
+                }
+            } catch {
+                // Empty body is valid; syncs all configured tables
+            }
+
+            const results = await this.syncAllTables(targetTable)
+            return createResponse({ results }, undefined, 200)
+        }
+
+        app.post(`${this.pathPrefix}/trigger`, handleManualSync)
+        app.post(`${this.pathPrefix}/sync`, handleManualSync)
+    }
+
+    override async beforeQuery(opts: {
+        sql: string
+        params?: unknown[]
+        dataSource?: DataSource
+        config?: StarbaseDBConfiguration
+    }): Promise<{ sql: string; params?: unknown[] }> {
+        let { sql, params, dataSource } = opts
+        const activeSource = dataSource?.source ?? this.dataSource?.source
+
+        // Only rewrite schema-qualified table names when querying the internal SQLite replica
+        if (activeSource && activeSource !== 'internal') {
+            return { sql, params }
+        }
+
+        // 1. Rewrite configured non-default schema references (e.g. `analytics.events` -> `analytics_events`)
+        for (const tableCfg of this.getNormalizedTables()) {
+            if (
+                tableCfg.sourceSchema &&
+                tableCfg.sourceSchema.toLowerCase() !== 'public' &&
+                tableCfg.sourceSchema.toLowerCase() !== 'main'
+            ) {
+                const pattern = new RegExp(
+                    `(?:"${this.escapeRegExp(tableCfg.sourceSchema)}"|\\b${this.escapeRegExp(tableCfg.sourceSchema)})\\.(?:"${this.escapeRegExp(tableCfg.sourceTable)}"|${this.escapeRegExp(tableCfg.sourceTable)}\\b)`,
+                    'gi'
+                )
+                sql = sql.replace(pattern, tableCfg.internalTable)
+            }
+        }
+
+        // 2. Strip `public.` prefix (e.g. `SELECT * FROM public.users` or `"public"."users"` -> `users`)
+        sql = sql.replace(
+            /(?:"public"|\bpublic)\.(?:"([a-zA-Z_][a-zA-Z0-9_]*)"|([a-zA-Z_][a-zA-Z0-9_]*)\b)/gi,
+            (_match, quotedName, unquotedName) =>
+                quotedName ? `"${quotedName}"` : unquotedName
+        )
+
+        return { sql, params }
+    }
+
+    private escapeRegExp(value: string): string {
+        return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')
+    }
+
+    public async ensureMetadataTables(): Promise<void> {
+        if (!this.dataSource || this.initializedTables) {
+            return
+        }
+
+        await this.dataSource.rpc.executeQuery({
+            sql: SQL_QUERIES.CREATE_METADATA_TABLE,
+            params: [],
+        })
+        await this.dataSource.rpc.executeQuery({
+            sql: SQL_QUERIES.CREATE_ERRORS_TABLE,
+            params: [],
+        })
+        this.initializedTables = true
+    }
+
+    public async getSyncStatus(): Promise<{
+        tables: TableSyncMetadata[]
+        errors: Record<string, any>[]
+    }> {
+        if (!this.dataSource) {
+            return { tables: [], errors: [] }
+        }
+        await this.ensureMetadataTables()
+
+        const tables = ((await this.dataSource.rpc.executeQuery({
+            sql: SQL_QUERIES.GET_ALL_METADATA,
+            params: [],
+        })) || []) as TableSyncMetadata[]
+
+        const errors = ((await this.dataSource.rpc.executeQuery({
+            sql: SQL_QUERIES.GET_RECENT_ERRORS,
+            params: [],
+        })) || []) as Record<string, any>[]
+
+        return { tables, errors }
+    }
+
+    public async syncDueTables(): Promise<void> {
+        if (!this.dataSource || !this.dataSource.external) {
+            return
+        }
+
+        const now = Date.now()
+        for (const tableCfg of this.getNormalizedTables()) {
+            const metaRows = ((await this.dataSource.rpc.executeQuery({
+                sql: SQL_QUERIES.GET_TABLE_METADATA,
+                params: [tableCfg.internalTable],
+            })) || []) as TableSyncMetadata[]
+
+            const lastSync = metaRows[0]?.last_sync_timestamp ?? 0
+            if (now - lastSync >= this.syncIntervalMs) {
+                await this.syncTable(tableCfg)
+            }
+        }
+    }
+
+    public async syncAllTables(
+        onlyTable?: string
+    ): Promise<Array<{ table: string; rowsSynced: number; status: string; error?: string }>> {
+        const results: Array<{
+            table: string
+            rowsSynced: number
+            status: string
+            error?: string
+        }> = []
+
+        const tables = this.getNormalizedTables().filter((t) =>
+            onlyTable
+                ? t.internalTable === onlyTable ||
+                  t.sourceTable === onlyTable ||
+                  `${t.sourceSchema}.${t.sourceTable}` === onlyTable
+                : true
+        )
+
+        for (const tableCfg of tables) {
+            const res = await this.syncTable(tableCfg)
+            results.push(res)
+        }
+
+        return results
+    }
+
+    public async syncTable(
+        tableCfg: NormalizedTableSyncConfig
+    ): Promise<{ table: string; rowsSynced: number; status: string; error?: string }> {
+        if (!this.dataSource) {
+            return {
+                table: tableCfg.internalTable,
+                rowsSynced: 0,
+                status: 'skipped',
+                error: 'No dataSource configured',
+            }
+        }
+
+        await this.ensureMetadataTables()
+
+        const existingMetaRows = ((await this.dataSource.rpc.executeQuery({
+            sql: SQL_QUERIES.GET_TABLE_METADATA,
+            params: [tableCfg.internalTable],
+        })) || []) as TableSyncMetadata[]
+
+        const existingMeta = existingMetaRows[0]
+        let lastSyncValue: string | number | null =
+            existingMeta?.last_sync_value ?? null
+        let totalRowsSynced = Number(existingMeta?.rows_synced ?? 0)
+
+        try {
+            // 1. Introspect external table schema
+            const schemaQuery = this.syncSource.getTableSchemaQuery(
+                tableCfg.sourceSchema,
+                tableCfg.sourceTable
+            )
+            const rawSchemaResult = await executeExternalQuery({
+                sql: schemaQuery.sql,
+                params: schemaQuery.params,
+                dataSource: this.dataSource,
+                config: this.config || { role: 'admin' },
+            })
+            const schemaRows = normalizeExternalRows(rawSchemaResult)
+            let columns = this.syncSource.parseSchemaRows(
+                schemaRows,
+                tableCfg.primaryKey
+            )
+
+            // 2. Determine tracking column for incremental checkpointing
+            let trackingColumn = tableCfg.trackingColumn
+            if (!trackingColumn && columns.length > 0) {
+                const colNames = new Set(columns.map((c) => c.name))
+                if (colNames.has('updated_at')) {
+                    trackingColumn = 'updated_at'
+                } else if (colNames.has('created_at')) {
+                    trackingColumn = 'created_at'
+                } else {
+                    const pkCol = columns.find((c) => c.isPrimaryKey)
+                    if (pkCol) {
+                        trackingColumn = pkCol.name
+                    } else if (colNames.has('id')) {
+                        trackingColumn = 'id'
+                    }
+                }
+            }
+
+            // 3. Create internal table if we have column definitions from schema introspection
+            if (columns.length > 0) {
+                await this.ensureInternalTableSchema(
+                    tableCfg.internalTable,
+                    columns
+                )
+            }
+
+            // 4. Pull records in batches from external database
+            let offset = 0
+            let syncedInRun = 0
+            let tableCreatedFromRow = columns.length > 0
+
+            while (true) {
+                const fetchQuery = this.syncSource.getFetchRecordsQuery({
+                    schema: tableCfg.sourceSchema,
+                    table: tableCfg.sourceTable,
+                    trackingColumn,
+                    lastValue: lastSyncValue,
+                    batchSize: tableCfg.batchSize,
+                    offset,
+                })
+
+                const rawRecords = await executeExternalQuery({
+                    sql: fetchQuery.sql,
+                    params: fetchQuery.params,
+                    dataSource: this.dataSource,
+                    config: this.config || { role: 'admin' },
+                })
+                const rows = normalizeExternalRows(rawRecords)
+
+                if (rows.length === 0) {
+                    break
+                }
+
+                // Fallback schema inference if information_schema returned no rows
+                if (!tableCreatedFromRow) {
+                    columns = this.inferColumnsFromRow(
+                        rows[0],
+                        tableCfg.primaryKey
+                    )
+                    await this.ensureInternalTableSchema(
+                        tableCfg.internalTable,
+                        columns
+                    )
+                    tableCreatedFromRow = true
+                }
+
+                for (const row of rows) {
+                    const rowKeys = Object.keys(row)
+                    if (rowKeys.length === 0) continue
+
+                    const quotedInternalTable = quoteSqlIdentifier(
+                        tableCfg.internalTable
+                    )
+                    const quotedCols = rowKeys
+                        .map((k) => quoteSqlIdentifier(k))
+                        .join(', ')
+                    const placeholders = rowKeys.map(() => '?').join(', ')
+                    const values = rowKeys.map((k) =>
+                        serializeSqliteValue(row[k])
+                    )
+
+                    await this.dataSource.rpc.executeQuery({
+                        sql: `INSERT OR REPLACE INTO ${quotedInternalTable} (${quotedCols}) VALUES (${placeholders})`,
+                        params: values,
+                    })
+
+                    if (
+                        trackingColumn &&
+                        row[trackingColumn] !== undefined &&
+                        row[trackingColumn] !== null
+                    ) {
+                        lastSyncValue = String(row[trackingColumn])
+                    }
+                    syncedInRun++
+                }
+
+                if (rows.length < tableCfg.batchSize) {
+                    break
+                }
+
+                // When filtering by `trackingColumn > lastSyncValue`, `lastSyncValue` advances each batch so offset stays 0.
+                // When doing full-table pagination without a tracking column, advance offset.
+                if (!trackingColumn) {
+                    offset += tableCfg.batchSize
+                }
+            }
+
+            totalRowsSynced += syncedInRun
+
+            await this.dataSource.rpc.executeQuery({
+                sql: SQL_QUERIES.UPSERT_METADATA,
+                params: [
+                    tableCfg.internalTable,
+                    tableCfg.sourceSchema,
+                    tableCfg.sourceTable,
+                    trackingColumn ?? null,
+                    Date.now(),
+                    lastSyncValue !== null ? String(lastSyncValue) : null,
+                    totalRowsSynced,
+                    'synced',
+                    null,
+                ],
+            })
+
+            return {
+                table: tableCfg.internalTable,
+                rowsSynced: syncedInRun,
+                status: 'synced',
+            }
+        } catch (error: any) {
+            const errorMessage =
+                error instanceof Error ? error.message : String(error)
+
+            await this.dataSource.rpc.executeQuery({
+                sql: SQL_QUERIES.UPSERT_METADATA,
+                params: [
+                    tableCfg.internalTable,
+                    tableCfg.sourceSchema,
+                    tableCfg.sourceTable,
+                    tableCfg.trackingColumn ?? null,
+                    Date.now(),
+                    lastSyncValue !== null ? String(lastSyncValue) : null,
+                    totalRowsSynced,
+                    'error',
+                    errorMessage,
+                ],
+            })
+
+            await this.dataSource.rpc.executeQuery({
+                sql: SQL_QUERIES.INSERT_ERROR,
+                params: [tableCfg.internalTable, errorMessage],
+            })
+
+            return {
+                table: tableCfg.internalTable,
+                rowsSynced: 0,
+                status: 'error',
+                error: errorMessage,
+            }
+        }
+    }
+
+    private inferColumnsFromRow(
+        row: Record<string, any>,
+        primaryKeyOverride?: string
+    ): ColumnDefinition[] {
+        const keys = Object.keys(row)
+        return keys.map((key) => {
+            const val = row[key]
+            let sqliteType: SqliteColumnType = 'TEXT'
+            if (typeof val === 'number') {
+                sqliteType = Number.isInteger(val) ? 'INTEGER' : 'REAL'
+            } else if (typeof val === 'boolean') {
+                sqliteType = 'INTEGER'
+            } else if (
+                val instanceof Uint8Array ||
+                val instanceof ArrayBuffer
+            ) {
+                sqliteType = 'BLOB'
+            }
+
+            const isPrimaryKey = primaryKeyOverride
+                ? key === primaryKeyOverride
+                : key === 'id'
+
+            return {
+                name: key,
+                type: sqliteType,
+                sqliteType,
+                nullable: !isPrimaryKey,
+                isPrimaryKey,
+            }
+        })
+    }
+
+    private async ensureInternalTableSchema(
+        internalTable: string,
+        columns: ColumnDefinition[]
+    ): Promise<void> {
+        if (!this.dataSource || columns.length === 0) return
+
+        const quotedTable = quoteSqlIdentifier(internalTable)
+        const pkColumns = columns.filter((c) => c.isPrimaryKey)
+        const hasSinglePk = pkColumns.length === 1
+
+        const colDefs = columns.map((col) => {
+            const parts = [quoteSqlIdentifier(col.name), col.sqliteType]
+            if (hasSinglePk && col.isPrimaryKey) {
+                parts.push('PRIMARY KEY')
+            } else if (!col.nullable) {
+                parts.push('NOT NULL')
+            }
+            return parts.join(' ')
+        })
+
+        if (pkColumns.length > 1) {
+            const compositeKeys = pkColumns
+                .map((c) => quoteSqlIdentifier(c.name))
+                .join(', ')
+            colDefs.push(`PRIMARY KEY (${compositeKeys})`)
+        }
+
+        await this.dataSource.rpc.executeQuery({
+            sql: `CREATE TABLE IF NOT EXISTS ${quotedTable} (${colDefs.join(', ')})`,
+            params: [],
+        })
+
+        // Also check existing columns via PRAGMA table_info and add any missing columns
+        const pragmaRows = ((await this.dataSource.rpc.executeQuery({
+            sql: `PRAGMA table_info(${quotedTable})`,
+            params: [],
+        })) || []) as Record<string, any>[]
+
+        if (pragmaRows.length > 0) {
+            const existingNames = new Set(
+                pragmaRows.map((r) => String(r.name))
+            )
+            for (const col of columns) {
+                if (!existingNames.has(col.name)) {
+                    await this.dataSource.rpc.executeQuery({
+                        sql: `ALTER TABLE ${quotedTable} ADD COLUMN ${quoteSqlIdentifier(col.name)} ${col.sqliteType}`,
+                        params: [],
+                    })
+                }
+            }
+        }
+    }
+}
+
+export {
+    DataSyncPlugin as ReplicationPlugin,
+    DatabaseSyncSource as SyncAdapter,
+    PostgresSyncSource as PostgresSync,
+    PostgresSyncSource as PostgresSyncAdapter,
+    MySQLSyncSource as MySQLSync,
+    MySQLSyncSource as MySQLSyncAdapter,
+}
diff --git a/plugins/data-sync/meta.json b/plugins/data-sync/meta.json
new file mode 100644
index 0000000..fe59f5c
--- /dev/null
+++ b/plugins/data-sync/meta.json
@@ -0,0 +1,16 @@
+{
+    "version": "1.0.0",
+    "resources": {
+        "tables": {
+            "tmp_data_sync_metadata": {},
+            "tmp_data_sync_errors": {}
+        },
+        "secrets": {},
+        "variables": {}
+    },
+    "dependencies": {
+        "tables": {},
+        "secrets": {},
+        "variables": {}
+    }
+}