diff --git a/dist/plugins.ts b/dist/plugins.ts
index 7dd252a..03babc3 100644
--- a/dist/plugins.ts
+++ b/dist/plugins.ts
@@ -6,3 +6,9 @@ export { ChangeDataCapturePlugin } from '../plugins/cdc'
 export { QueryLogPlugin } from '../plugins/query-log'
 export { ResendPlugin } from '../plugins/resend'
 export { ClerkPlugin } from '../plugins/clerk'
+export { DataReplicationPlugin } from '../plugins/replication'
+export { SyncAdapter } from '../plugins/replication/adapter'
+export { PostgresSyncAdapter } from '../plugins/replication/adapters/postgres'
+export { MySQLSyncAdapter } from '../plugins/replication/adapters/mysql'
+export { GenericSyncAdapter } from '../plugins/replication/adapters/generic'
+export * from '../plugins/replication/types'
diff --git a/plugins/replication/adapter.test.ts b/plugins/replication/adapter.test.ts
new file mode 100644
index 0000000..c78439e
--- /dev/null
+++ b/plugins/replication/adapter.test.ts
@@ -0,0 +1,398 @@
+import { describe, it, expect } from 'vitest'
+import { PostgresSyncAdapter } from './adapters/postgres'
+import { MySQLSyncAdapter } from './adapters/mysql'
+import { GenericSyncAdapter } from './adapters/generic'
+
+describe('PostgresSyncAdapter', () => {
+    const adapter = new PostgresSyncAdapter()
+
+    describe('buildIntrospectQuery', () => {
+        it('should target information_schema.columns with schema and table params', () => {
+            const query = adapter.buildIntrospectQuery('users', 'public')
+            expect(query.sql).toContain('information_schema.columns')
+            expect(query.sql).toContain('table_schema = ?')
+            expect(query.sql).toContain('table_name = ?')
+            expect(query.sql).toContain('ORDER BY ordinal_position')
+            expect(query.params).toEqual(['public', 'users'])
+        })
+    })
+
+    describe('parseIntrospectResult & mapType', () => {
+        it('should map Postgres integer and boolean types to SQLite INTEGER', () => {
+            const raw = [
+                { column_name: 'id', data_type: 'bigint', is_nullable: 'NO' },
+                {
+                    column_name: 'age',
+                    data_type: 'integer',
+                    is_nullable: 'YES',
+                },
+                {
+                    column_name: 'active',
+                    data_type: 'boolean',
+                    is_nullable: 'NO',
+                },
+                { column_name: 'seq', data_type: 'serial', is_nullable: 'NO' },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0]).toEqual({
+                name: 'id',
+                sourceType: 'bigint',
+                sqliteType: 'INTEGER',
+                nullable: false,
+            })
+            expect(parsed[1].sqliteType).toBe('INTEGER')
+            expect(parsed[1].nullable).toBe(true)
+            expect(parsed[2].sqliteType).toBe('INTEGER')
+            expect(parsed[3].sqliteType).toBe('INTEGER')
+        })
+
+        it('should map Postgres floating-point types to SQLite REAL', () => {
+            const raw = [
+                {
+                    column_name: 'price',
+                    data_type: 'numeric',
+                    is_nullable: 'YES',
+                },
+                {
+                    column_name: 'score',
+                    data_type: 'double precision',
+                    is_nullable: 'YES',
+                },
+                { column_name: 'ratio', data_type: 'real', is_nullable: 'YES' },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            parsed.forEach((col) => expect(col.sqliteType).toBe('REAL'))
+        })
+
+        it('should map bytea to SQLite BLOB', () => {
+            const raw = [
+                { column_name: 'data', data_type: 'bytea', is_nullable: 'YES' },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0].sqliteType).toBe('BLOB')
+        })
+
+        it('should map json and jsonb to SQLite TEXT for compatibility with json() functions', () => {
+            const raw = [
+                {
+                    column_name: 'metadata',
+                    data_type: 'json',
+                    is_nullable: 'YES',
+                },
+                {
+                    column_name: 'settings',
+                    data_type: 'jsonb',
+                    is_nullable: 'YES',
+                },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0].sqliteType).toBe('TEXT')
+            expect(parsed[1].sqliteType).toBe('TEXT')
+        })
+
+        it('should map strings and timestamps to SQLite TEXT', () => {
+            const raw = [
+                {
+                    column_name: 'name',
+                    data_type: 'character varying',
+                    is_nullable: 'YES',
+                },
+                { column_name: 'bio', data_type: 'text', is_nullable: 'YES' },
+                {
+                    column_name: 'created_at',
+                    data_type: 'timestamp with time zone',
+                    is_nullable: 'NO',
+                },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            parsed.forEach((col) => expect(col.sqliteType).toBe('TEXT'))
+            expect(parsed[2].nullable).toBe(false)
+        })
+    })
+
+    describe('buildFetchQuery', () => {
+        it('should generate an initial fetch query without WHERE when cursorValue is null', () => {
+            const query = adapter.buildFetchQuery({
+                table: 'users',
+                schema: 'public',
+                cursorColumn: 'id',
+                cursorValue: null,
+                limit: 100,
+            })
+            expect(query.sql).toContain('"public"."users"')
+            expect(query.sql).toContain('ORDER BY "id" ASC LIMIT ?')
+            expect(query.sql).not.toContain('WHERE')
+            expect(query.params).toEqual([100])
+        })
+
+        it('should generate an incremental query with WHERE when cursorValue is present', () => {
+            const query = adapter.buildFetchQuery({
+                table: 'users',
+                schema: 'public',
+                cursorColumn: 'id',
+                cursorValue: '1050',
+                limit: 50,
+            })
+            expect(query.sql).toContain('WHERE "id" > ?')
+            expect(query.sql).toContain('ORDER BY "id" ASC LIMIT ?')
+            expect(query.params).toEqual(['1050', 50])
+        })
+
+        it('should select specific columns when provided', () => {
+            const query = adapter.buildFetchQuery({
+                table: 'users',
+                schema: 'public',
+                cursorColumn: 'created_at',
+                cursorValue: '2026-01-01T00:00:00Z',
+                columns: ['id', 'email', 'created_at'],
+                limit: 10,
+            })
+            expect(query.sql).toContain('SELECT "id", "email", "created_at"')
+            expect(query.sql).toContain('WHERE "created_at" > ?')
+            expect(query.params).toEqual(['2026-01-01T00:00:00Z', 10])
+        })
+
+        it('should omit schema from table name when schema is empty', () => {
+            const query = adapter.buildFetchQuery({
+                table: 'items',
+                schema: '',
+                cursorColumn: 'id',
+                cursorValue: null,
+                limit: 25,
+            })
+            expect(query.sql).toContain('FROM "items"')
+            expect(query.sql).not.toContain('"."')
+        })
+    })
+})
+
+describe('MySQLSyncAdapter', () => {
+    const adapter = new MySQLSyncAdapter()
+
+    describe('buildIntrospectQuery', () => {
+        it('should target INFORMATION_SCHEMA.COLUMNS', () => {
+            const query = adapter.buildIntrospectQuery('orders', 'store')
+            expect(query.sql).toContain('INFORMATION_SCHEMA.COLUMNS')
+            expect(query.sql).toContain('TABLE_SCHEMA = ?')
+            expect(query.sql).toContain('TABLE_NAME = ?')
+            expect(query.params).toEqual(['store', 'orders'])
+        })
+    })
+
+    describe('parseIntrospectResult & mapType', () => {
+        it('should map MySQL integer types to SQLite INTEGER', () => {
+            const raw = [
+                {
+                    column_name: 'id',
+                    data_type: 'bigint(20)',
+                    is_nullable: 'NO',
+                },
+                {
+                    column_name: 'active',
+                    data_type: 'tinyint(1)',
+                    is_nullable: 'NO',
+                },
+                { column_name: 'count', data_type: 'int', is_nullable: 'YES' },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0].sqliteType).toBe('INTEGER')
+            expect(parsed[1].sqliteType).toBe('INTEGER')
+            expect(parsed[2].sqliteType).toBe('INTEGER')
+        })
+
+        it('should map MySQL float and decimal types to SQLite REAL', () => {
+            const raw = [
+                {
+                    column_name: 'price',
+                    data_type: 'decimal(10,2)',
+                    is_nullable: 'YES',
+                },
+                {
+                    column_name: 'weight',
+                    data_type: 'float',
+                    is_nullable: 'YES',
+                },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0].sqliteType).toBe('REAL')
+            expect(parsed[1].sqliteType).toBe('REAL')
+        })
+
+        it('should map MySQL blob types to SQLite BLOB', () => {
+            const raw = [
+                {
+                    column_name: 'payload',
+                    data_type: 'blob',
+                    is_nullable: 'YES',
+                },
+                {
+                    column_name: 'binary_hash',
+                    data_type: 'varbinary(64)',
+                    is_nullable: 'YES',
+                },
+            ]
+            const parsed = adapter.parseIntrospectResult(raw)
+            expect(parsed[0].sqliteType).toBe('BLOB')
+            expect(parsed[1].sqliteType).toBe('BLOB')
+        })
+    })
+
+    describe('buildFetchQuery', () => {
+        it('should use backticks for identifiers in MySQL queries', () => {
+            const query = adapter.buildFetchQuery({
+                table: 'orders',
+                schema: 'shop',
+                cursorColumn: 'id',
+                cursorValue: '200',
+                columns: ['id', 'amount'],
+                limit: 100,
+            })
+            expect(query.sql).toContain(
+                'SELECT `id`, `amount` FROM `shop`.`orders`'
+            )
+            expect(query.sql).toContain('WHERE `id` > ?')
+            expect(query.sql).toContain('ORDER BY `id` ASC LIMIT ?')
+            expect(query.params).toEqual(['200', 100])
+        })
+    })
+})
+
+describe('GenericSyncAdapter', () => {
+    const adapter = new GenericSyncAdapter()
+
+    it('should build PRAGMA table_info query', () => {
+        const query = adapter.buildIntrospectQuery('products', '')
+        expect(query.sql).toBe('PRAGMA table_info("products")')
+    })
+
+    it('should parse PRAGMA table_info result correctly', () => {
+        const raw = [
+            { name: 'id', type: 'INTEGER', notnull: 1 },
+            { name: 'title', type: 'TEXT', notnull: 0 },
+            { name: 'rating', type: 'REAL', notnull: 0 },
+        ]
+        const parsed = adapter.parseIntrospectResult(raw)
+        expect(parsed[0]).toEqual({
+            name: 'id',
+            sourceType: 'INTEGER',
+            sqliteType: 'INTEGER',
+            nullable: false,
+        })
+        expect(parsed[1]).toEqual({
+            name: 'title',
+            sourceType: 'TEXT',
+            sqliteType: 'TEXT',
+            nullable: true,
+        })
+        expect(parsed[2]).toEqual({
+            name: 'rating',
+            sourceType: 'REAL',
+            sqliteType: 'REAL',
+            nullable: true,
+        })
+    })
+})
+
+describe('SyncAdapter Base Class Methods', () => {
+    const adapter = new PostgresSyncAdapter()
+
+    describe('resolveTargetTableName', () => {
+        it('should prefix schema name with an underscore when schema is provided', () => {
+            expect(adapter.resolveTargetTableName('users', 'public')).toBe(
+                'public_users'
+            )
+            expect(adapter.resolveTargetTableName('accounts', 'auth')).toBe(
+                'auth_accounts'
+            )
+        })
+
+        it('should return sourceTable unchanged when schema is empty or omitted', () => {
+            expect(adapter.resolveTargetTableName('users', '')).toBe('users')
+            expect(adapter.resolveTargetTableName('users', undefined)).toBe(
+                'users'
+            )
+        })
+
+        it('should allow customTargetTable to override schema prefixing', () => {
+            expect(
+                adapter.resolveTargetTableName(
+                    'users',
+                    'public',
+                    'edge_users_replica'
+                )
+            ).toBe('edge_users_replica')
+        })
+    })
+
+    describe('buildCreateTableSQL', () => {
+        it('should generate valid SQLite CREATE TABLE statement with types and primary key', () => {
+            const columns = [
+                {
+                    name: 'id',
+                    sourceType: 'integer',
+                    sqliteType: 'INTEGER' as const,
+                    nullable: false,
+                },
+                {
+                    name: 'email',
+                    sourceType: 'varchar',
+                    sqliteType: 'TEXT' as const,
+                    nullable: false,
+                },
+                {
+                    name: 'metadata',
+                    sourceType: 'jsonb',
+                    sqliteType: 'TEXT' as const,
+                    nullable: true,
+                },
+            ]
+            const sql = adapter.buildCreateTableSQL(
+                'public_users',
+                columns,
+                'id'
+            )
+            expect(sql).toBe(
+                'CREATE TABLE IF NOT EXISTS "public_users" ("id" INTEGER NOT NULL, "email" TEXT NOT NULL, "metadata" TEXT, PRIMARY KEY ("id"))'
+            )
+        })
+
+        it('should omit PRIMARY KEY clause when primaryKey argument is omitted', () => {
+            const columns = [
+                {
+                    name: 'log_line',
+                    sourceType: 'text',
+                    sqliteType: 'TEXT' as const,
+                    nullable: true,
+                },
+            ]
+            const sql = adapter.buildCreateTableSQL('logs', columns)
+            expect(sql).toBe(
+                'CREATE TABLE IF NOT EXISTS "logs" ("log_line" TEXT)'
+            )
+        })
+    })
+
+    describe('buildUpsertSQL', () => {
+        it('should generate INSERT OR REPLACE statement for replace strategy', () => {
+            const sql = adapter.buildUpsertSQL(
+                'public_users',
+                ['id', 'email', 'name'],
+                'replace'
+            )
+            expect(sql).toBe(
+                'INSERT OR REPLACE INTO "public_users" ("id", "email", "name") VALUES (?, ?, ?)'
+            )
+        })
+
+        it('should generate INSERT OR IGNORE statement for ignore strategy', () => {
+            const sql = adapter.buildUpsertSQL(
+                'public_users',
+                ['id', 'email'],
+                'ignore'
+            )
+            expect(sql).toBe(
+                'INSERT OR IGNORE INTO "public_users" ("id", "email") VALUES (?, ?)'
+            )
+        })
+    })
+})
diff --git a/plugins/replication/adapter.ts b/plugins/replication/adapter.ts
new file mode 100644
index 0000000..604c381
--- /dev/null
+++ b/plugins/replication/adapter.ts
@@ -0,0 +1,74 @@
+import type { ConflictStrategy, CursorType } from './types'
+
+export interface ColumnDefinition {
+    name: string
+    sourceType: string
+    sqliteType: 'TEXT' | 'INTEGER' | 'REAL' | 'BLOB'
+    nullable: boolean
+}
+
+export interface FetchRowsOpts {
+    table: string
+    schema: string
+    cursorColumn: string
+    cursorValue: string | number | null
+    cursorType?: CursorType
+    columns?: string[]
+    limit: number
+}
+
+export abstract class SyncAdapter {
+    abstract buildIntrospectQuery(
+        table: string,
+        schema: string
+    ): { sql: string; params: unknown[] }
+
+    abstract parseIntrospectResult(
+        rows: Record<string, unknown>[]
+    ): ColumnDefinition[]
+
+    abstract buildFetchQuery(opts: FetchRowsOpts): {
+        sql: string
+        params: unknown[]
+    }
+
+    resolveTargetTableName(
+        sourceTable: string,
+        schema?: string,
+        customTargetTable?: string
+    ): string {
+        if (customTargetTable && customTargetTable.trim().length > 0) {
+            return customTargetTable
+        }
+        if (schema && schema.trim().length > 0) {
+            return `${schema}_${sourceTable}`
+        }
+        return sourceTable
+    }
+
+    buildCreateTableSQL(
+        targetTable: string,
+        columns: ColumnDefinition[],
+        primaryKey?: string
+    ): string {
+        const cols = columns
+            .map(
+                (c) =>
+                    `"${c.name}" ${c.sqliteType}${c.nullable ? '' : ' NOT NULL'}`
+            )
+            .join(', ')
+        const pkClause = primaryKey ? `, PRIMARY KEY ("${primaryKey}")` : ''
+        return `CREATE TABLE IF NOT EXISTS "${targetTable}" (${cols}${pkClause})`
+    }
+
+    buildUpsertSQL(
+        targetTable: string,
+        columns: string[],
+        strategy: ConflictStrategy = 'replace'
+    ): string {
+        const or = strategy === 'replace' ? 'OR REPLACE' : 'OR IGNORE'
+        const colList = columns.map((c) => `"${c}"`).join(', ')
+        const placeholders = columns.map(() => '?').join(', ')
+        return `INSERT ${or} INTO "${targetTable}" (${colList}) VALUES (${placeholders})`
+    }
+}
diff --git a/plugins/replication/adapters/generic.ts b/plugins/replication/adapters/generic.ts
new file mode 100644
index 0000000..15cc58c
--- /dev/null
+++ b/plugins/replication/adapters/generic.ts
@@ -0,0 +1,67 @@
+import {
+    SyncAdapter,
+    type ColumnDefinition,
+    type FetchRowsOpts,
+} from '../adapter'
+
+export class GenericSyncAdapter extends SyncAdapter {
+    buildIntrospectQuery(table: string, _schema: string) {
+        return {
+            sql: `PRAGMA table_info("${table}")`,
+            params: [] as unknown[],
+        }
+    }
+
+    parseIntrospectResult(rows: Record<string, unknown>[]): ColumnDefinition[] {
+        return rows.map((r) => ({
+            name: String(r.name),
+            sourceType: String(r.type || 'TEXT'),
+            sqliteType: this.mapType(String(r.type || 'TEXT').toLowerCase()),
+            nullable: Number(r.notnull) === 0,
+        }))
+    }
+
+    buildFetchQuery(opts: FetchRowsOpts) {
+        const { table, cursorColumn, cursorValue, columns, limit } = opts
+        const cols = columns?.map((c) => `"${c}"`).join(', ') ?? '*'
+
+        if (cursorValue === null || cursorValue === undefined) {
+            return {
+                sql: `SELECT ${cols} FROM "${table}" ORDER BY "${cursorColumn}" ASC LIMIT ?`,
+                params: [limit] as unknown[],
+            }
+        }
+
+        return {
+            sql: `SELECT ${cols} FROM "${table}" WHERE "${cursorColumn}" > ? ORDER BY "${cursorColumn}" ASC LIMIT ?`,
+            params: [cursorValue, limit] as unknown[],
+        }
+    }
+
+    private mapType(typeStr: string): 'TEXT' | 'INTEGER' | 'REAL' | 'BLOB' {
+        const t = typeStr.toLowerCase()
+
+        if (
+            [
+                'int',
+                'integer',
+                'tinyint',
+                'smallint',
+                'mediumint',
+                'bigint',
+            ].some((k) => t.includes(k))
+        ) {
+            return 'INTEGER'
+        }
+
+        if (['real', 'floa', 'doub'].some((k) => t.includes(k))) {
+            return 'REAL'
+        }
+
+        if (t.includes('blob')) {
+            return 'BLOB'
+        }
+
+        return 'TEXT'
+    }
+}
diff --git a/plugins/replication/adapters/mysql.ts b/plugins/replication/adapters/mysql.ts
new file mode 100644
index 0000000..e65b937
--- /dev/null
+++ b/plugins/replication/adapters/mysql.ts
@@ -0,0 +1,87 @@
+import {
+    SyncAdapter,
+    type ColumnDefinition,
+    type FetchRowsOpts,
+} from '../adapter'
+
+export class MySQLSyncAdapter extends SyncAdapter {
+    buildIntrospectQuery(table: string, schema: string) {
+        return {
+            sql: `SELECT COLUMN_NAME AS column_name, DATA_TYPE AS data_type, IS_NULLABLE AS is_nullable
+                  FROM INFORMATION_SCHEMA.COLUMNS
+                  WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?
+                  ORDER BY ORDINAL_POSITION`,
+            params: [schema, table] as unknown[],
+        }
+    }
+
+    parseIntrospectResult(rows: Record<string, unknown>[]): ColumnDefinition[] {
+        return rows.map((r) => ({
+            name: String(r.column_name),
+            sourceType: String(r.data_type),
+            sqliteType: this.mapType(String(r.data_type).toLowerCase()),
+            nullable: String(r.is_nullable).toUpperCase() === 'YES',
+        }))
+    }
+
+    buildFetchQuery(opts: FetchRowsOpts) {
+        const { table, schema, cursorColumn, cursorValue, columns, limit } =
+            opts
+        const cols = columns?.map((c) => `\`${c}\``).join(', ') ?? '*'
+        const tbl = schema ? `\`${schema}\`.\`${table}\`` : `\`${table}\``
+
+        if (cursorValue === null || cursorValue === undefined) {
+            return {
+                sql: `SELECT ${cols} FROM ${tbl} ORDER BY \`${cursorColumn}\` ASC LIMIT ?`,
+                params: [limit] as unknown[],
+            }
+        }
+
+        return {
+            sql: `SELECT ${cols} FROM ${tbl} WHERE \`${cursorColumn}\` > ? ORDER BY \`${cursorColumn}\` ASC LIMIT ?`,
+            params: [cursorValue, limit] as unknown[],
+        }
+    }
+
+    private mapType(mysqlType: string): 'TEXT' | 'INTEGER' | 'REAL' | 'BLOB' {
+        const t = mysqlType.toLowerCase()
+
+        if (
+            [
+                'int',
+                'integer',
+                'tinyint',
+                'smallint',
+                'mediumint',
+                'bigint',
+                'boolean',
+                'bool',
+            ].some((k) => t === k || t.startsWith(k + '('))
+        ) {
+            return 'INTEGER'
+        }
+
+        if (
+            ['float', 'double', 'decimal', 'numeric', 'real'].some(
+                (k) => t === k || t.startsWith(k + '(')
+            )
+        ) {
+            return 'REAL'
+        }
+
+        if (
+            [
+                'blob',
+                'tinyblob',
+                'mediumblob',
+                'longblob',
+                'binary',
+                'varbinary',
+            ].some((k) => t === k || t.startsWith(k))
+        ) {
+            return 'BLOB'
+        }
+
+        return 'TEXT'
+    }
+}
diff --git a/plugins/replication/adapters/postgres.ts b/plugins/replication/adapters/postgres.ts
new file mode 100644
index 0000000..22429b4
--- /dev/null
+++ b/plugins/replication/adapters/postgres.ts
@@ -0,0 +1,89 @@
+import {
+    SyncAdapter,
+    type ColumnDefinition,
+    type FetchRowsOpts,
+} from '../adapter'
+
+export class PostgresSyncAdapter extends SyncAdapter {
+    buildIntrospectQuery(table: string, schema: string) {
+        return {
+            sql: `SELECT column_name, data_type, is_nullable
+                  FROM information_schema.columns
+                  WHERE table_schema = ? AND table_name = ?
+                  ORDER BY ordinal_position`,
+            params: [schema, table] as unknown[],
+        }
+    }
+
+    parseIntrospectResult(rows: Record<string, unknown>[]): ColumnDefinition[] {
+        return rows.map((r) => ({
+            name: String(r.column_name),
+            sourceType: String(r.data_type),
+            sqliteType: this.mapType(String(r.data_type).toLowerCase()),
+            nullable: String(r.is_nullable).toUpperCase() === 'YES',
+        }))
+    }
+
+    buildFetchQuery(opts: FetchRowsOpts) {
+        const { table, schema, cursorColumn, cursorValue, columns, limit } =
+            opts
+        const cols = columns?.map((c) => `"${c}"`).join(', ') ?? '*'
+        const tbl = schema ? `"${schema}"."${table}"` : `"${table}"`
+
+        if (cursorValue === null || cursorValue === undefined) {
+            return {
+                sql: `SELECT ${cols} FROM ${tbl} ORDER BY "${cursorColumn}" ASC LIMIT ?`,
+                params: [limit] as unknown[],
+            }
+        }
+
+        return {
+            sql: `SELECT ${cols} FROM ${tbl} WHERE "${cursorColumn}" > ? ORDER BY "${cursorColumn}" ASC LIMIT ?`,
+            params: [cursorValue, limit] as unknown[],
+        }
+    }
+
+    private mapType(pgType: string): 'TEXT' | 'INTEGER' | 'REAL' | 'BLOB' {
+        if (
+            [
+                'integer',
+                'bigint',
+                'smallint',
+                'serial',
+                'bigserial',
+                'boolean',
+                'int',
+                'int2',
+                'int4',
+                'int8',
+            ].includes(pgType)
+        ) {
+            return 'INTEGER'
+        }
+
+        if (
+            [
+                'real',
+                'double precision',
+                'numeric',
+                'decimal',
+                'float',
+                'float4',
+                'float8',
+            ].some((t) => pgType.startsWith(t))
+        ) {
+            return 'REAL'
+        }
+
+        if (pgType === 'bytea') {
+            return 'BLOB'
+        }
+
+        // json and jsonb must map to TEXT in SQLite so built-in json functions work seamlessly
+        if (['json', 'jsonb'].includes(pgType)) {
+            return 'TEXT'
+        }
+
+        return 'TEXT'
+    }
+}
diff --git a/plugins/replication/index.test.ts b/plugins/replication/index.test.ts
new file mode 100644
index 0000000..cc746a4
--- /dev/null
+++ b/plugins/replication/index.test.ts
@@ -0,0 +1,575 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { DataReplicationPlugin } from './index'
+import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler'
+import { DataSource } from '../../src/types'
+import * as operationModule from '../../src/operation'
+
+let mockDataSource: DataSource
+let mockAdminConfig: StarbaseDBConfiguration
+let mockClientConfig: StarbaseDBConfiguration
+
+beforeEach(() => {
+    vi.restoreAllMocks()
+
+    mockDataSource = {
+        rpc: {
+            executeQuery: vi.fn().mockResolvedValue([]),
+        },
+        source: 'internal',
+        external: {
+            dialect: 'postgresql',
+            host: 'localhost',
+            port: 5432,
+            user: 'test',
+            password: 'pwd',
+            database: 'testdb',
+            defaultSchema: 'public',
+        },
+        executionContext: {
+            waitUntil: vi.fn(),
+            passThroughOnException: vi.fn(),
+        },
+    } as unknown as DataSource
+
+    mockAdminConfig = {
+        role: 'admin',
+        features: { allowlist: false, rls: false, rest: true },
+    }
+
+    mockClientConfig = {
+        role: 'client',
+        features: { allowlist: false, rls: false, rest: true },
+    }
+})
+
+describe('DataReplicationPlugin - Route Registration and Initialization', () => {
+    it('should register middleware and all admin routes on app', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: { tables: [] },
+        })
+
+        const registeredRoutes: { method: string; path: string }[] = []
+        const mockApp = {
+            use: vi.fn(),
+            get: vi.fn((path: string) =>
+                registeredRoutes.push({ method: 'GET', path })
+            ),
+            post: vi.fn((path: string) =>
+                registeredRoutes.push({ method: 'POST', path })
+            ),
+            delete: vi.fn((path: string) =>
+                registeredRoutes.push({ method: 'DELETE', path })
+            ),
+        } as unknown as StarbaseApp
+
+        await plugin.register(mockApp)
+
+        expect(mockApp.use).toHaveBeenCalled()
+        expect(registeredRoutes).toContainEqual({
+            method: 'GET',
+            path: '/replication/status',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'GET',
+            path: '/replication/status/:table',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'POST',
+            path: '/replication/sync',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'POST',
+            path: '/replication/sync/:table',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'DELETE',
+            path: '/replication/state/:table',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'GET',
+            path: '/replication/history',
+        })
+        expect(registeredRoutes).toContainEqual({
+            method: 'GET',
+            path: '/replication/history/:table',
+        })
+    })
+
+    it('should create tmp_replication_state and tmp_replication_history tables on init', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: { tables: [] },
+        })
+
+        let middlewareFn: (
+            context: unknown,
+            next: () => Promise<void>
+        ) => Promise<void>
+        const mockApp = {
+            use: vi.fn((fn) => {
+                middlewareFn = fn
+            }),
+            get: vi.fn(),
+            post: vi.fn(),
+            delete: vi.fn(),
+        } as unknown as StarbaseApp
+
+        await plugin.register(mockApp)
+
+        const mockContext = {
+            get: vi.fn((key: string) => {
+                if (key === 'dataSource') return mockDataSource
+                if (key === 'config') return mockAdminConfig
+                return undefined
+            }),
+        }
+
+        await middlewareFn(mockContext, vi.fn())
+
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining(
+                    'CREATE TABLE IF NOT EXISTS tmp_replication_state'
+                ),
+            })
+        )
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining(
+                    'CREATE TABLE IF NOT EXISTS tmp_replication_history'
+                ),
+            })
+        )
+    })
+})
+
+describe('DataReplicationPlugin - Admin Authorization', () => {
+    async function setupPluginWithRole(role: 'admin' | 'client') {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'users',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                        cursorType: 'integer',
+                    },
+                ],
+            },
+        })
+
+        const handlers: Record<
+            string,
+            (context: unknown) => Promise<Response>
+        > = {}
+        let middlewareFn: (
+            context: unknown,
+            next: () => Promise<void>
+        ) => Promise<void>
+        const mockApp = {
+            use: vi.fn((fn) => {
+                middlewareFn = fn
+            }),
+            get: vi.fn((path, handler) => {
+                handlers[`GET ${path}`] = handler
+            }),
+            post: vi.fn((path, handler) => {
+                handlers[`POST ${path}`] = handler
+            }),
+            delete: vi.fn((path, handler) => {
+                handlers[`DELETE ${path}`] = handler
+            }),
+        } as unknown as StarbaseApp
+
+        await plugin.register(mockApp)
+
+        const mockContext = {
+            get: vi.fn((key: string) => {
+                if (key === 'dataSource') return mockDataSource
+                if (key === 'config')
+                    return role === 'admin' ? mockAdminConfig : mockClientConfig
+                return undefined
+            }),
+            req: {
+                param: vi.fn((paramName: string) =>
+                    paramName === 'table' ? 'users' : ''
+                ),
+            },
+        }
+
+        await middlewareFn(mockContext, vi.fn())
+        return { plugin, handlers, mockContext }
+    }
+
+    it('should reject non-admin users with 401 on all endpoints', async () => {
+        const { handlers, mockContext } = await setupPluginWithRole('client')
+
+        const routes = [
+            'GET /replication/status',
+            'GET /replication/status/:table',
+            'POST /replication/sync',
+            'POST /replication/sync/:table',
+            'DELETE /replication/state/:table',
+            'GET /replication/history',
+            'GET /replication/history/:table',
+        ]
+
+        for (const route of routes) {
+            const handler = handlers[route]
+            expect(handler).toBeDefined()
+            const res = await handler(mockContext)
+            expect(res.status).toBe(401)
+        }
+    })
+
+    it('should allow admin users on status, history and state endpoints', async () => {
+        const { handlers, mockContext } = await setupPluginWithRole('admin')
+
+        vi.mocked(mockDataSource.rpc.executeQuery).mockResolvedValue([
+            {
+                source_table: 'users',
+                target_table: 'public_users',
+                status: 'idle',
+            } as unknown as Record<string, unknown>,
+        ])
+
+        const statusRes = await handlers['GET /replication/status'](mockContext)
+        expect(statusRes.status).toBe(200)
+
+        const tableStatusRes =
+            await handlers['GET /replication/status/:table'](mockContext)
+        expect(tableStatusRes.status).toBe(200)
+
+        const deleteStateRes =
+            await handlers['DELETE /replication/state/:table'](mockContext)
+        expect(deleteStateRes.status).toBe(200)
+
+        const historyRes =
+            await handlers['GET /replication/history'](mockContext)
+        expect(historyRes.status).toBe(200)
+    })
+})
+
+describe('DataReplicationPlugin - beforeQuery Table Resolution', () => {
+    it('should rewrite public.users to public_users for SQLite compatibility', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'users',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                    },
+                    {
+                        sourceTable: 'orders',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                        targetTable: 'my_edge_orders',
+                    },
+                ],
+            },
+        })
+
+        const query1 = await plugin.beforeQuery({
+            sql: 'SELECT * FROM public.users WHERE id = 1',
+            params: [],
+            isRaw: false,
+        })
+        expect(query1.sql).toBe('SELECT * FROM "public_users" WHERE id = 1')
+
+        const query2 = await plugin.beforeQuery({
+            sql: 'SELECT * FROM "public"."users" JOIN public.orders ON public.users.id = public.orders.user_id',
+            params: [],
+            isRaw: false,
+        })
+        expect(query2.sql).toContain('"public_users"')
+        expect(query2.sql).toContain('"my_edge_orders"')
+
+        const query3 = await plugin.beforeQuery({
+            sql: 'SELECT * FROM `public`.`users`',
+            params: [],
+            isRaw: false,
+        })
+        expect(query3.sql).toBe('SELECT * FROM "public_users"')
+    })
+
+    it('should leave unrelated queries unchanged', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'users',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                    },
+                ],
+            },
+        })
+
+        const query = await plugin.beforeQuery({
+            sql: 'SELECT * FROM local_cache WHERE key = ?',
+            params: ['token'],
+            isRaw: false,
+        })
+        expect(query.sql).toBe('SELECT * FROM local_cache WHERE key = ?')
+        expect(query.params).toEqual(['token'])
+    })
+})
+
+describe('DataReplicationPlugin - Replication Execution Logic', () => {
+    it('should introspect external schema and create SQLite table on initial sync', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'customers',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                        cursorType: 'integer',
+                    },
+                ],
+            },
+        })
+
+        // Mock introspect query returning columns
+        vi.spyOn(operationModule, 'executeExternalQuery')
+            .mockResolvedValueOnce([
+                { column_name: 'id', data_type: 'integer', is_nullable: 'NO' },
+                {
+                    column_name: 'email',
+                    data_type: 'varchar',
+                    is_nullable: 'NO',
+                },
+                {
+                    column_name: 'balance',
+                    data_type: 'numeric',
+                    is_nullable: 'YES',
+                },
+            ])
+            // Fetch query returning 2 rows
+            .mockResolvedValueOnce([
+                { id: 1, email: 'alice@example.com', balance: 100.5 },
+                { id: 2, email: 'bob@example.com', balance: 250.0 },
+            ])
+
+        const results = await plugin.syncAll(mockDataSource, mockAdminConfig)
+
+        expect(results).toHaveLength(1)
+        expect(results[0].success).toBe(true)
+        expect(results[0].rowsSynced).toBe(2)
+        expect(results[0].targetTable).toBe('public_customers')
+
+        // Verify CREATE TABLE IF NOT EXISTS was executed on SQLite
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringMatching(
+                    /CREATE TABLE IF NOT EXISTS "public_customers"/
+                ),
+            })
+        )
+
+        // Verify rows were inserted via INSERT OR REPLACE
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringMatching(
+                    /INSERT OR REPLACE INTO "public_customers"/
+                ),
+                params: [1, 'alice@example.com', 100.5],
+            })
+        )
+
+        // Verify tmp_replication_history audit record was written
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining(
+                    'INSERT INTO tmp_replication_history'
+                ),
+                params: expect.arrayContaining([
+                    'customers',
+                    'public_customers',
+                    2,
+                    expect.any(Number),
+                    'success',
+                ]),
+            })
+        )
+    })
+
+    it('should perform incremental sync using cursor value when state already exists', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'events',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                        cursorType: 'integer',
+                    },
+                ],
+            },
+        })
+
+        // Mock getState returning existing checkpoint cursor at '100'
+        vi.mocked(mockDataSource.rpc.executeQuery).mockImplementation(
+            async (opts: { sql: string; params?: unknown[] }) => {
+                if (
+                    opts.sql.includes(
+                        'SELECT source_table, target_table, last_cursor_value'
+                    )
+                ) {
+                    return [
+                        {
+                            source_table: 'events',
+                            target_table: 'public_events',
+                            last_cursor_value: '100',
+                            rows_synced: 100,
+                            total_rows_synced: 100,
+                            status: 'success',
+                            error_message: null,
+                        },
+                    ] as unknown as Record<string, unknown>[]
+                }
+                return [] as unknown as Record<string, unknown>[]
+            }
+        )
+
+        const executeExternalSpy = vi
+            .spyOn(operationModule, 'executeExternalQuery')
+            .mockResolvedValueOnce([
+                { id: 101, event_type: 'login' },
+                { id: 102, event_type: 'click' },
+            ])
+
+        const results = await plugin.syncAll(mockDataSource, mockAdminConfig)
+
+        expect(results[0].success).toBe(true)
+        expect(results[0].rowsSynced).toBe(2)
+
+        // Verify fetch query passed cursorValue '100'
+        expect(executeExternalSpy).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining('WHERE "id" > ?'),
+                params: ['100', 1000],
+            })
+        )
+
+        // Verify cursor advanced to '102'
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining(
+                    'INSERT INTO tmp_replication_state'
+                ),
+                params: expect.arrayContaining([
+                    'events',
+                    'public_events',
+                    '102',
+                ]),
+            })
+        )
+    })
+
+    it('should isolate errors so one failing table does not block others', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'failing_table',
+                        cursorColumn: 'id',
+                    },
+                    {
+                        sourceTable: 'healthy_table',
+                        cursorColumn: 'id',
+                    },
+                ],
+            },
+        })
+
+        vi.spyOn(operationModule, 'executeExternalQuery').mockImplementation(
+            async (opts: { sql: string }) => {
+                if (opts.sql.includes('failing_table')) {
+                    throw new Error('Connection timeout to external shard')
+                }
+                if (opts.sql.includes('information_schema')) {
+                    return [
+                        {
+                            column_name: 'id',
+                            data_type: 'integer',
+                            is_nullable: 'NO',
+                        },
+                    ]
+                }
+                return [{ id: 1 }]
+            }
+        )
+
+        const results = await plugin.syncAll(mockDataSource, mockAdminConfig)
+
+        expect(results).toHaveLength(2)
+        expect(results[0].table).toBe('failing_table')
+        expect(results[0].success).toBe(false)
+        expect(results[0].error).toContain(
+            'Connection timeout to external shard'
+        )
+
+        expect(results[1].table).toBe('healthy_table')
+        expect(results[1].success).toBe(true)
+        expect(results[1].rowsSynced).toBe(1)
+
+        // Verify failure was logged to tmp_replication_history
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringContaining(
+                    'INSERT INTO tmp_replication_history'
+                ),
+                params: expect.arrayContaining([
+                    'failing_table',
+                    'public_failing_table',
+                    0,
+                    expect.any(Number),
+                    'error',
+                    'Connection timeout to external shard',
+                ]),
+            })
+        )
+    })
+
+    it('should correctly serialize JSON and nested objects into SQLite TEXT', async () => {
+        const plugin = new DataReplicationPlugin({
+            config: {
+                tables: [
+                    {
+                        sourceTable: 'configs',
+                        schema: 'public',
+                        cursorColumn: 'id',
+                    },
+                ],
+            },
+        })
+
+        vi.spyOn(operationModule, 'executeExternalQuery')
+            .mockResolvedValueOnce([
+                { column_name: 'id', data_type: 'integer', is_nullable: 'NO' },
+                {
+                    column_name: 'payload',
+                    data_type: 'jsonb',
+                    is_nullable: 'YES',
+                },
+            ])
+            .mockResolvedValueOnce([
+                { id: 1, payload: { theme: 'dark', notifications: true } },
+            ])
+
+        await plugin.syncAll(mockDataSource, mockAdminConfig)
+
+        expect(mockDataSource.rpc.executeQuery).toHaveBeenCalledWith(
+            expect.objectContaining({
+                sql: expect.stringMatching(
+                    /INSERT OR REPLACE INTO "public_configs"/
+                ),
+                params: [
+                    1,
+                    JSON.stringify({ theme: 'dark', notifications: true }),
+                ],
+            })
+        )
+    })
+})
diff --git a/plugins/replication/index.ts b/plugins/replication/index.ts
new file mode 100644
index 0000000..de96df8
--- /dev/null
+++ b/plugins/replication/index.ts
@@ -0,0 +1,701 @@
+import { StarbaseApp, StarbaseDBConfiguration } from '../../src/handler'
+import { StarbasePlugin } from '../../src/plugin'
+import { DataSource } from '../../src/types'
+import { executeExternalQuery } from '../../src/operation'
+import { createResponse } from '../../src/utils'
+import { CronPlugin } from '../cron'
+import { PostgresSyncAdapter } from './adapters/postgres'
+import { MySQLSyncAdapter } from './adapters/mysql'
+import { GenericSyncAdapter } from './adapters/generic'
+import type { SyncAdapter } from './adapter'
+import type {
+    ReplicationConfig,
+    ReplicationState,
+    SyncResult,
+    TableReplicationConfig,
+} from './types'
+
+export function createSyncAdapter(dialect: string): SyncAdapter {
+    switch (dialect.toLowerCase()) {
+        case 'postgresql':
+        case 'postgres':
+            return new PostgresSyncAdapter()
+        case 'mysql':
+            return new MySQLSyncAdapter()
+        default:
+            return new GenericSyncAdapter()
+    }
+}
+
+const DEFAULT_BATCH_SIZE = 1000
+const DEFAULT_SYNC_INTERVAL_MS = 60 * 1000
+const CRON_TASK_NAME = 'starbasedb:replication:sync'
+
+const SQL = {
+    CREATE_STATE_TABLE: `
+        CREATE TABLE IF NOT EXISTS tmp_replication_state (
+            source_table      TEXT PRIMARY KEY,
+            target_table      TEXT NOT NULL,
+            last_cursor_value TEXT,
+            last_sync_at      TEXT,
+            rows_synced       INTEGER DEFAULT 0,
+            total_rows_synced INTEGER DEFAULT 0,
+            status            TEXT DEFAULT 'idle',
+            error_message     TEXT,
+            updated_at        TEXT DEFAULT (datetime('now'))
+        )
+    `,
+    CREATE_HISTORY_TABLE: `
+        CREATE TABLE IF NOT EXISTS tmp_replication_history (
+            id            INTEGER PRIMARY KEY AUTOINCREMENT,
+            source_table  TEXT NOT NULL,
+            target_table  TEXT NOT NULL,
+            rows_synced   INTEGER NOT NULL,
+            duration_ms   INTEGER NOT NULL,
+            status        TEXT NOT NULL,
+            error_message TEXT,
+            started_at    TEXT NOT NULL,
+            completed_at  TEXT DEFAULT (datetime('now'))
+        )
+    `,
+    GET_ALL_STATE: `
+        SELECT source_table, target_table, last_cursor_value, last_sync_at,
+               rows_synced, total_rows_synced, status, error_message, updated_at
+        FROM tmp_replication_state
+        ORDER BY source_table ASC
+    `,
+    GET_STATE: `
+        SELECT source_table, target_table, last_cursor_value, last_sync_at,
+               rows_synced, total_rows_synced, status, error_message, updated_at
+        FROM tmp_replication_state
+        WHERE source_table = ?
+    `,
+    UPSERT_STATE: `
+        INSERT INTO tmp_replication_state (
+            source_table, target_table, last_cursor_value, last_sync_at,
+            rows_synced, total_rows_synced, status, error_message, updated_at
+        )
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
+        ON CONFLICT(source_table) DO UPDATE SET
+            target_table      = excluded.target_table,
+            last_cursor_value = excluded.last_cursor_value,
+            last_sync_at      = excluded.last_sync_at,
+            rows_synced       = excluded.rows_synced,
+            total_rows_synced = tmp_replication_state.total_rows_synced + excluded.rows_synced,
+            status            = excluded.status,
+            error_message     = excluded.error_message,
+            updated_at        = datetime('now')
+    `,
+    UPDATE_STATUS: `
+        UPDATE tmp_replication_state
+        SET status = ?, error_message = ?, updated_at = datetime('now')
+        WHERE source_table = ?
+    `,
+    RESET_CHECKPOINT: `
+        UPDATE tmp_replication_state
+        SET last_cursor_value = NULL, rows_synced = 0, total_rows_synced = 0,
+            status = 'idle', error_message = NULL, updated_at = datetime('now')
+        WHERE source_table = ?
+    `,
+    INSERT_HISTORY: `
+        INSERT INTO tmp_replication_history (
+            source_table, target_table, rows_synced, duration_ms, status, error_message, started_at, completed_at
+        )
+        VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now'))
+    `,
+    GET_HISTORY: `
+        SELECT id, source_table, target_table, rows_synced, duration_ms, status, error_message, started_at, completed_at
+        FROM tmp_replication_history
+        ORDER BY id DESC LIMIT 50
+    `,
+    GET_HISTORY_FOR_TABLE: `
+        SELECT id, source_table, target_table, rows_synced, duration_ms, status, error_message, started_at, completed_at
+        FROM tmp_replication_history
+        WHERE source_table = ?
+        ORDER BY id DESC LIMIT 50
+    `,
+}
+
+export class DataReplicationPlugin extends StarbasePlugin {
+    public pathPrefix: string = '/replication'
+    private config: ReplicationConfig
+    private cronPlugin?: CronPlugin
+    private dataSource?: DataSource
+    private pluginConfig?: StarbaseDBConfiguration
+    private cronRegistered = false
+
+    constructor(opts: {
+        config: ReplicationConfig
+        cronPlugin?: CronPlugin
+        pathPrefix?: string
+    }) {
+        super(
+            'starbasedb:replication',
+            { requiresAuth: true },
+            opts.pathPrefix || '/replication'
+        )
+        this.config = opts.config
+        this.cronPlugin = opts.cronPlugin
+        if (opts.pathPrefix) {
+            this.pathPrefix = opts.pathPrefix
+        }
+    }
+
+    override async register(app: StarbaseApp) {
+        app.use(async (c, next) => {
+            this.dataSource = c.get('dataSource')
+            this.pluginConfig = c.get('config')
+            await this.init()
+            await this.maybeScheduleLazySync()
+            await next()
+        })
+
+        app.get(`${this.pathPrefix}/status`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            const rows = (await this.dataSource?.rpc.executeQuery({
+                sql: SQL.GET_ALL_STATE,
+                params: [],
+            })) as unknown as ReplicationState[]
+
+            return createResponse(rows ?? [], undefined, 200)
+        })
+
+        app.get(`${this.pathPrefix}/status/:table`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            const table = c.req.param('table')
+            const rows = (await this.dataSource?.rpc.executeQuery({
+                sql: SQL.GET_STATE,
+                params: [table],
+            })) as unknown as ReplicationState[]
+
+            if (!rows || rows.length === 0) {
+                return createResponse(undefined, 'Table state not found', 404)
+            }
+
+            return createResponse(rows[0], undefined, 200)
+        })
+
+        app.post(`${this.pathPrefix}/sync`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            if (!this.dataSource) {
+                return createResponse(undefined, 'Data source unavailable', 503)
+            }
+
+            const ds = this.dataSource
+            const cfg = this.pluginConfig
+            const syncPromise = this.syncAll(ds, cfg)
+            ds.executionContext?.waitUntil(syncPromise) ?? syncPromise
+
+            return createResponse(
+                { message: 'Sync triggered for all tables' },
+                undefined,
+                202
+            )
+        })
+
+        app.post(`${this.pathPrefix}/sync/:table`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            if (!this.dataSource) {
+                return createResponse(undefined, 'Data source unavailable', 503)
+            }
+
+            const tableName = c.req.param('table')
+            const tableConfig = this.config.tables.find(
+                (t) => t.sourceTable === tableName
+            )
+
+            if (!tableConfig) {
+                return createResponse(
+                    undefined,
+                    `Table '${tableName}' not found in replication config`,
+                    404
+                )
+            }
+
+            const ds = this.dataSource
+            const cfg = this.pluginConfig
+            const syncPromise = this.syncTable(tableConfig, ds, cfg)
+            ds.executionContext?.waitUntil(syncPromise) ?? syncPromise
+
+            return createResponse(
+                { message: `Sync triggered for table '${tableName}'` },
+                undefined,
+                202
+            )
+        })
+
+        app.delete(`${this.pathPrefix}/state/:table`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            const table = c.req.param('table')
+            await this.dataSource?.rpc.executeQuery({
+                sql: SQL.RESET_CHECKPOINT,
+                params: [table],
+            })
+
+            return createResponse(
+                { message: `Checkpoint reset for '${table}'` },
+                undefined,
+                200
+            )
+        })
+
+        app.get(`${this.pathPrefix}/history`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            const rows = await this.dataSource?.rpc.executeQuery({
+                sql: SQL.GET_HISTORY,
+                params: [],
+            })
+
+            return createResponse(rows ?? [], undefined, 200)
+        })
+
+        app.get(`${this.pathPrefix}/history/:table`, async (c) => {
+            if (this.pluginConfig?.role !== 'admin') {
+                return createResponse(undefined, 'Unauthorized', 401)
+            }
+
+            const table = c.req.param('table')
+            const rows = await this.dataSource?.rpc.executeQuery({
+                sql: SQL.GET_HISTORY_FOR_TABLE,
+                params: [table],
+            })
+
+            return createResponse(rows ?? [], undefined, 200)
+        })
+
+        if (this.cronPlugin && this.config.cronSchedule) {
+            this.cronPlugin.onEvent(async ({ name }) => {
+                if (name !== CRON_TASK_NAME || !this.dataSource) return
+                await this.syncAll(this.dataSource, this.pluginConfig)
+            })
+        }
+    }
+
+    override async beforeQuery(opts: {
+        sql: string
+        params?: unknown[]
+        isRaw: boolean
+        dataSource?: DataSource
+        config?: StarbaseDBConfiguration
+    }): Promise<{ sql: string; params?: unknown[] }> {
+        let sql = opts.sql
+        if (!sql) return { sql, params: opts.params }
+
+        for (const tableConfig of this.config.tables) {
+            const schema = tableConfig.schema ?? 'public'
+            const source = tableConfig.sourceTable
+            const target =
+                tableConfig.targetTable ??
+                (schema ? `${schema}_${source}` : source)
+
+            // Matches schema.sourceTable with optional double quotes or backticks,
+            // e.g. public.users, "public"."users", `public`.`users`, "public".users, public."users"
+            const schemaQualifiedRegex = new RegExp(
+                `["\`]?${schema}["\`]?\\s*\\.\\s*["\`]?${source}["\`]?`,
+                'gi'
+            )
+            sql = sql.replace(schemaQualifiedRegex, `"${target}"`)
+        }
+
+        return {
+            sql,
+            params: opts.params,
+        }
+    }
+
+    private async init() {
+        if (!this.dataSource) return
+
+        await this.dataSource.rpc.executeQuery({
+            sql: SQL.CREATE_STATE_TABLE,
+            params: [],
+        })
+
+        await this.dataSource.rpc.executeQuery({
+            sql: SQL.CREATE_HISTORY_TABLE,
+            params: [],
+        })
+
+        if (
+            !this.cronRegistered &&
+            this.cronPlugin &&
+            this.config.cronSchedule &&
+            this.config.callbackHost
+        ) {
+            try {
+                await this.cronPlugin.addEvent(
+                    this.config.cronSchedule,
+                    CRON_TASK_NAME,
+                    {},
+                    this.config.callbackHost
+                )
+                this.cronRegistered = true
+            } catch {
+                // Task may already exist; safe to ignore.
+            }
+        }
+    }
+
+    private async maybeScheduleLazySync() {
+        if (this.cronPlugin || !this.config.syncIntervalMs) return
+        if (!this.dataSource) return
+
+        const intervalMs =
+            this.config.syncIntervalMs ?? DEFAULT_SYNC_INTERVAL_MS
+
+        const rows = (await this.dataSource.rpc.executeQuery({
+            sql: `SELECT MIN(last_sync_at) AS earliest FROM tmp_replication_state`,
+            params: [],
+        })) as unknown as { earliest: string | null }[]
+
+        const earliest = rows?.[0]?.earliest
+        if (!earliest) {
+            return
+        }
+
+        const lastSyncMs = new Date(earliest).getTime()
+        if (Date.now() - lastSyncMs > intervalMs) {
+            const ds = this.dataSource
+            const cfg = this.pluginConfig
+            const syncPromise = this.syncAll(ds, cfg)
+            ds.executionContext?.waitUntil(syncPromise) ?? syncPromise
+        }
+    }
+
+    async syncAll(
+        dataSource: DataSource,
+        config?: StarbaseDBConfiguration
+    ): Promise<SyncResult[]> {
+        const results: SyncResult[] = []
+
+        for (const tableConfig of this.config.tables) {
+            try {
+                const result = await this.syncTable(
+                    tableConfig,
+                    dataSource,
+                    config
+                )
+                results.push(result)
+            } catch (err) {
+                // Isolated error handling: a failure in one table must not crash syncAll
+                const errorMessage =
+                    err instanceof Error ? err.message : String(err)
+                results.push({
+                    table: tableConfig.sourceTable,
+                    targetTable:
+                        tableConfig.targetTable ?? tableConfig.sourceTable,
+                    rowsSynced: 0,
+                    durationMs: 0,
+                    success: false,
+                    error: errorMessage,
+                })
+            }
+        }
+
+        return results
+    }
+
+    async syncTable(
+        tableConfig: TableReplicationConfig,
+        dataSource: DataSource,
+        config?: StarbaseDBConfiguration
+    ): Promise<SyncResult> {
+        const schema =
+            tableConfig.schema ??
+            (dataSource?.external && 'defaultSchema' in dataSource.external
+                ? (dataSource.external.defaultSchema ?? 'public')
+                : 'public')
+
+        const dialect =
+            dataSource?.external && 'dialect' in dataSource.external
+                ? dataSource.external.dialect
+                : 'sqlite'
+
+        const adapter = createSyncAdapter(dialect)
+        const targetTable = adapter.resolveTargetTableName(
+            tableConfig.sourceTable,
+            schema,
+            tableConfig.targetTable
+        )
+
+        if (!dataSource?.external) {
+            return {
+                table: tableConfig.sourceTable,
+                targetTable,
+                rowsSynced: 0,
+                durationMs: 0,
+                success: false,
+                error: 'No external data source configured',
+            }
+        }
+
+        const conflictStrategy = tableConfig.conflictStrategy ?? 'replace'
+        const batchSize = this.config.batchSize ?? DEFAULT_BATCH_SIZE
+        const startedAt = new Date().toISOString()
+        const startMs = Date.now()
+        let totalSynced = 0
+
+        await this.setStatus(
+            dataSource,
+            tableConfig.sourceTable,
+            targetTable,
+            'running'
+        )
+
+        try {
+            const existingState = await this.getState(
+                dataSource,
+                tableConfig.sourceTable,
+                targetTable
+            )
+
+            // Auto schema creation on first sync
+            if (!existingState || existingState.last_cursor_value === null) {
+                const introspect = adapter.buildIntrospectQuery(
+                    tableConfig.sourceTable,
+                    schema
+                )
+
+                const introspectRows = (await executeExternalQuery({
+                    sql: introspect.sql,
+                    params: introspect.params,
+                    dataSource,
+                    config: config ?? { role: 'admin' },
+                })) as Record<string, unknown>[]
+
+                if (introspectRows && introspectRows.length > 0) {
+                    const columnDefs =
+                        adapter.parseIntrospectResult(introspectRows)
+                    const createSQL = adapter.buildCreateTableSQL(
+                        targetTable,
+                        columnDefs,
+                        tableConfig.cursorColumn
+                    )
+                    await dataSource.rpc.executeQuery({
+                        sql: createSQL,
+                        params: [],
+                    })
+                }
+            }
+
+            let cursorValue = existingState?.last_cursor_value ?? null
+
+            while (true) {
+                const fetchQuery = adapter.buildFetchQuery({
+                    table: tableConfig.sourceTable,
+                    schema,
+                    cursorColumn: tableConfig.cursorColumn,
+                    cursorValue,
+                    cursorType: tableConfig.cursorType,
+                    columns: tableConfig.columns,
+                    limit: batchSize,
+                })
+
+                const rows = (await executeExternalQuery({
+                    sql: fetchQuery.sql,
+                    params: fetchQuery.params,
+                    dataSource,
+                    config: config ?? { role: 'admin' },
+                })) as Record<string, unknown>[]
+
+                if (!rows || rows.length === 0) {
+                    break
+                }
+
+                const columnNames = Object.keys(rows[0])
+                const upsertSQL = adapter.buildUpsertSQL(
+                    targetTable,
+                    columnNames,
+                    conflictStrategy
+                )
+
+                for (let i = 0; i < rows.length; i++) {
+                    const values = columnNames.map((col) => {
+                        const val = rows[i][col]
+                        if (val !== null && typeof val === 'object') {
+                            return JSON.stringify(val)
+                        }
+                        return val ?? null
+                    })
+
+                    await dataSource.rpc.executeQuery({
+                        sql: upsertSQL,
+                        params: values,
+                    })
+
+                    // Yield event loop every 100 rows to prevent DO starvation
+                    if (i > 0 && i % 100 === 0) {
+                        await new Promise<void>((resolve) =>
+                            setTimeout(resolve, 0)
+                        )
+                    }
+                }
+
+                const lastRow = rows[rows.length - 1]
+                cursorValue = String(lastRow[tableConfig.cursorColumn])
+                totalSynced += rows.length
+
+                await dataSource.rpc.executeQuery({
+                    sql: SQL.UPSERT_STATE,
+                    params: [
+                        tableConfig.sourceTable,
+                        targetTable,
+                        cursorValue,
+                        new Date().toISOString(),
+                        rows.length,
+                        'running',
+                        null,
+                    ],
+                })
+
+                if (rows.length < batchSize) {
+                    break
+                }
+            }
+
+            const durationMs = Date.now() - startMs
+
+            await dataSource.rpc.executeQuery({
+                sql: SQL.UPSERT_STATE,
+                params: [
+                    tableConfig.sourceTable,
+                    targetTable,
+                    cursorValue,
+                    new Date().toISOString(),
+                    0,
+                    'success',
+                    null,
+                ],
+            })
+
+            await dataSource.rpc.executeQuery({
+                sql: SQL.INSERT_HISTORY,
+                params: [
+                    tableConfig.sourceTable,
+                    targetTable,
+                    totalSynced,
+                    durationMs,
+                    'success',
+                    null,
+                    startedAt,
+                ],
+            })
+
+            return {
+                table: tableConfig.sourceTable,
+                targetTable,
+                rowsSynced: totalSynced,
+                durationMs,
+                success: true,
+            }
+        } catch (err) {
+            const errorMessage =
+                err instanceof Error ? err.message : String(err)
+            const durationMs = Date.now() - startMs
+
+            await this.setStatus(
+                dataSource,
+                tableConfig.sourceTable,
+                targetTable,
+                'error',
+                errorMessage
+            )
+
+            await dataSource.rpc.executeQuery({
+                sql: SQL.INSERT_HISTORY,
+                params: [
+                    tableConfig.sourceTable,
+                    targetTable,
+                    0,
+                    durationMs,
+                    'error',
+                    errorMessage,
+                    startedAt,
+                ],
+            })
+
+            console.error(
+                `[replication] sync failed for '${tableConfig.sourceTable}':`,
+                errorMessage
+            )
+
+            return {
+                table: tableConfig.sourceTable,
+                targetTable,
+                rowsSynced: 0,
+                durationMs,
+                success: false,
+                error: errorMessage,
+            }
+        }
+    }
+
+    private async getState(
+        dataSource: DataSource,
+        sourceTable: string,
+        targetTable: string
+    ): Promise<ReplicationState | null> {
+        const rows = (await dataSource.rpc.executeQuery({
+            sql: SQL.GET_STATE,
+            params: [sourceTable],
+        })) as unknown as ReplicationState[]
+
+        if (rows && rows.length > 0) return rows[0]
+
+        await dataSource.rpc.executeQuery({
+            sql: `INSERT OR IGNORE INTO tmp_replication_state (source_table, target_table) VALUES (?, ?)`,
+            params: [sourceTable, targetTable],
+        })
+
+        return null
+    }
+
+    private async setStatus(
+        dataSource: DataSource,
+        sourceTable: string,
+        targetTable: string,
+        status: string,
+        errorMessage?: string
+    ) {
+        const existing = (await dataSource.rpc.executeQuery({
+            sql: `SELECT source_table FROM tmp_replication_state WHERE source_table = ?`,
+            params: [sourceTable],
+        })) as unknown as { source_table: string }[]
+
+        if (!existing || existing.length === 0) {
+            await dataSource.rpc.executeQuery({
+                sql: `INSERT INTO tmp_replication_state (source_table, target_table, status, error_message)
+                      VALUES (?, ?, ?, ?)`,
+                params: [
+                    sourceTable,
+                    targetTable,
+                    status,
+                    errorMessage ?? null,
+                ],
+            })
+        } else {
+            await dataSource.rpc.executeQuery({
+                sql: SQL.UPDATE_STATUS,
+                params: [status, errorMessage ?? null, sourceTable],
+            })
+        }
+    }
+}
diff --git a/plugins/replication/meta.json b/plugins/replication/meta.json
new file mode 100644
index 0000000..1f68610
--- /dev/null
+++ b/plugins/replication/meta.json
@@ -0,0 +1,6 @@
+{
+    "name": "Data Replication Plugin",
+    "description": "Replicates data from an external data source (Postgres, MySQL, SQLite) into the internal SQLite database.",
+    "version": "0.1.0",
+    "author": "Outerbase"
+}
diff --git a/plugins/replication/types.ts b/plugins/replication/types.ts
new file mode 100644
index 0000000..a7a714a
--- /dev/null
+++ b/plugins/replication/types.ts
@@ -0,0 +1,54 @@
+export type CursorType = 'integer' | 'timestamp' | 'string'
+export type ConflictStrategy = 'replace' | 'ignore'
+export type SyncStatus = 'idle' | 'running' | 'success' | 'error'
+
+export interface TableReplicationConfig {
+    sourceTable: string
+    targetTable?: string
+    schema?: string
+    cursorColumn: string
+    cursorType?: CursorType
+    columns?: string[]
+    conflictStrategy?: ConflictStrategy
+}
+
+export interface ReplicationConfig {
+    tables: TableReplicationConfig[]
+    batchSize?: number
+    syncIntervalMs?: number
+    cronSchedule?: string
+    callbackHost?: string
+}
+
+export interface ReplicationState {
+    source_table: string
+    target_table: string
+    last_cursor_value: string | null
+    last_sync_at: string | null
+    rows_synced: number
+    total_rows_synced: number
+    status: SyncStatus
+    error_message: string | null
+    updated_at: string
+}
+
+export interface SyncHistoryRecord {
+    id?: number
+    source_table: string
+    target_table: string
+    rows_synced: number
+    duration_ms: number
+    status: SyncStatus
+    error_message: string | null
+    started_at: string
+    completed_at?: string
+}
+
+export interface SyncResult {
+    table: string
+    targetTable: string
+    rowsSynced: number
+    durationMs: number
+    success: boolean
+    error?: string
+}
diff --git a/src/index.ts b/src/index.ts
index 4d08932..aed6ee0 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -12,6 +12,7 @@ import { QueryLogPlugin } from '../plugins/query-log'
 import { StatsPlugin } from '../plugins/stats'
 import { CronPlugin } from '../plugins/cron'
 import { InterfacePlugin } from '../plugins/interface'
+import { DataReplicationPlugin } from '../plugins/replication'
 
 export { StarbaseDBDurableObject } from './do'
 
@@ -210,6 +211,13 @@ export default {
             }, ctx)
 
             const interfacePlugin = new InterfacePlugin()
+            const replicationPlugin = new DataReplicationPlugin({
+                cronPlugin,
+                config: {
+                    tables: [],
+                    batchSize: 1000,
+                },
+            })
 
             const plugins = [
                 webSocketPlugin,
@@ -226,6 +234,7 @@ export default {
                 cronPlugin,
                 new StatsPlugin(),
                 interfacePlugin,
+                replicationPlugin,
             ] satisfies StarbasePlugin[]
 
             const starbase = new StarbaseDB({
diff --git a/src/operation.ts b/src/operation.ts
index 4abc0dd..e98e708 100644
--- a/src/operation.ts
+++ b/src/operation.ts
@@ -15,6 +15,7 @@ import {
 import {
     CloudflareD1Source,
     DataSource,
+    PostgresSource,
     RemoteSource,
     StarbaseDBSource,
     TursoDBSource,
@@ -447,7 +448,9 @@ export async function executeSDKQuery(opts: {
     let connection: SqlConnection
 
     if (external.dialect === 'postgresql') {
-        const { database } = await createSDKPostgresConnection(external)
+        const { database } = await createSDKPostgresConnection(
+            external as PostgresSource
+        )
         connection = database
     } else if (external.dialect === 'mysql') {
         const { database } = await createSDKMySQLConnection(external)