✨ feat(core): 扩展多源数据库驱动并实现数据同步引擎

- 集成 go-ora, dm, gokb 驱动，封装统一的 Database 接口实现，支持自定义 DSN 连接
- 新增 SyncEngine 同步引擎，支持基于主键的增量数据比对 (Insert/Update)
- 新增 DataSyncModal 组件，实现三步走同步向导逻辑，修复 Transfer 组件空状态显示问题
- 优化 ConnectionModal 交互逻辑，支持驱动参数动态显隐
- 引入 antd/locale/zh_CN，统一应用界面的中文本地化显示

internal/app/methods_sync.go

@@ -0,0 +1,11 @@
+package app
+
+import (
+	"GoNavi-Wails/internal/sync"
+)
+
+// DataSync executes a data synchronization task
+func (a *App) DataSync(config sync.SyncConfig) sync.SyncResult {
+	engine := sync.NewSyncEngine()
+	return engine.RunSync(config)
+}

internal/connection/types.go

@@ -19,14 +19,16 @@ type ConnectionConfig struct {
 	Database string    `json:"database"`
 	UseSSH   bool      `json:"useSSH"`
 	SSH      SSHConfig `json:"ssh"`
+	Driver   string    `json:"driver,omitempty"` // For custom connection
+	DSN      string    `json:"dsn,omitempty"`    // For custom connection
 }
 
 // QueryResult is the standard response format for Wails methods
 type QueryResult struct {
-	Success bool                   `json:"success"`
-	Message string                 `json:"message"`
-	Data    interface{}            `json:"data"`
-	Fields  []string               `json:"fields,omitempty"`
+	Success bool        `json:"success"`
+	Message string      `json:"message"`
+	Data    interface{} `json:"data"`
+	Fields  []string    `json:"fields,omitempty"`
 }
 
 // ColumnDefinition represents a table column
@@ -36,7 +38,7 @@ type ColumnDefinition struct {
 	Nullable string  `json:"nullable"` // YES/NO
 	Key      string  `json:"key"`      // PRI, UNI, MUL
 	Default  *string `json:"default"`
-	Extra    string  `json:"extra"`    // auto_increment
+	Extra    string  `json:"extra"` // auto_increment
 	Comment  string  `json:"comment"`
 }
 

internal/db/custom_impl.go

@@ -0,0 +1,234 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/utils"
+)
+
+type CustomDB struct {
+	conn   *sql.DB
+	driver string
+}
+
+func (c *CustomDB) Connect(config connection.ConnectionConfig) error {
+	if config.Driver == "" || config.DSN == "" {
+		return fmt.Errorf("driver and dsn are required for custom connection")
+	}
+
+	// Verify driver is registered (implicit check by sql.Open)
+	// We might not need explicit check, sql.Open will fail or Ping will fail if driver not found.
+
+	db, err := sql.Open(config.Driver, config.DSN)
+	if err != nil {
+		return err
+	}
+	c.conn = db
+	c.driver = config.Driver
+	return c.Ping()
+}
+
+func (c *CustomDB) Close() error {
+	if c.conn != nil {
+		return c.conn.Close()
+	}
+	return nil
+}
+
+func (c *CustomDB) Ping() error {
+	if c.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+	ctx, cancel := utils.ContextWithTimeout(5 * time.Second)
+	defer cancel()
+	return c.conn.PingContext(ctx)
+}
+
+func (c *CustomDB) Query(query string) ([]map[string]interface{}, []string, error) {
+	if c.conn == nil {
+		return nil, nil, fmt.Errorf("connection not open")
+	}
+
+	rows, err := c.conn.Query(query)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var resultData []map[string]interface{}
+
+	for rows.Next() {
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			continue
+		}
+
+		entry := make(map[string]interface{})
+		for i, col := range columns {
+			var v interface{}
+			val := values[i]
+			b, ok := val.([]byte)
+			if ok {
+				v = string(b)
+			} else {
+				v = val
+			}
+			entry[col] = v
+		}
+		resultData = append(resultData, entry)
+	}
+
+	return resultData, columns, nil
+}
+
+func (c *CustomDB) Exec(query string) (int64, error) {
+	if c.conn == nil {
+		return 0, fmt.Errorf("connection not open")
+	}
+	res, err := c.conn.Exec(query)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+func (c *CustomDB) GetDatabases() ([]string, error) {
+	// Try standard information_schema or some known patterns if we can't guess
+	// For "custom", we can't easily know.
+	// But many DBs support SHOW DATABASES or SELECT datname FROM pg_database
+	// We'll try a generic query or return empty.
+	// Users using custom might know their DB context is single.
+
+	// Best effort:
+	return []string{}, nil
+}
+
+func (c *CustomDB) GetTables(dbName string) ([]string, error) {
+	// ANSI Standard
+	query := "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"
+	// If mysql-like
+	if c.driver == "mysql" {
+		query = "SHOW TABLES"
+		if dbName != "" {
+			query = fmt.Sprintf("SHOW TABLES FROM `%s`", dbName)
+		}
+	} else if c.driver == "postgres" || c.driver == "kingbase" {
+		if dbName != "" && dbName != "public" {
+			query = fmt.Sprintf("SELECT table_name FROM information_schema.tables WHERE table_schema = '%s'", dbName)
+		}
+	} else if c.driver == "sqlite" {
+		query = "SELECT name FROM sqlite_master WHERE type='table'"
+	} else if c.driver == "oracle" || c.driver == "dm" {
+		query = "SELECT table_name FROM user_tables"
+	}
+
+	// Fallback generic execution
+	data, _, err := c.Query(query)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get tables for custom driver %s: %v", c.driver, err)
+	}
+
+	var tables []string
+	for _, row := range data {
+		// iterate keys to find likely column
+		for k, v := range row {
+			if strings.Contains(strings.ToLower(k), "name") || strings.Contains(strings.ToLower(k), "table") {
+				tables = append(tables, fmt.Sprintf("%v", v))
+				break
+			}
+		}
+	}
+	return tables, nil
+}
+
+func (c *CustomDB) GetCreateStatement(dbName, tableName string) (string, error) {
+	return "Not supported for custom connections yet", nil
+}
+
+func (c *CustomDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefinition, error) {
+	// ANSI Standard
+	// SELECT column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_name = '...'
+
+	schema := "public"
+	if dbName != "" {
+		schema = dbName
+	}
+
+	query := fmt.Sprintf(`SELECT column_name, data_type, is_nullable, column_default 
+		FROM information_schema.columns 
+		WHERE table_name = '%s'`, tableName)
+
+	// Adjust for schema if likely supported
+	if c.driver == "postgres" || c.driver == "kingbase" {
+		query += fmt.Sprintf(" AND table_schema = '%s'", schema)
+	} else if c.driver == "mysql" {
+		query = fmt.Sprintf("SHOW FULL COLUMNS FROM `%s`", tableName)
+		if dbName != "" {
+			query = fmt.Sprintf("SHOW FULL COLUMNS FROM `%s`.`%s`", dbName, tableName)
+		}
+	}
+
+	data, _, err := c.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var columns []connection.ColumnDefinition
+	for _, row := range data {
+		col := connection.ColumnDefinition{}
+		// flexible mapping
+		for k, v := range row {
+			kl := strings.ToLower(k)
+			val := fmt.Sprintf("%v", v)
+			if strings.Contains(kl, "field") || strings.Contains(kl, "column_name") {
+				col.Name = val
+			} else if strings.Contains(kl, "type") {
+				col.Type = val
+			} else if strings.Contains(kl, "null") || strings.Contains(kl, "nullable") {
+				col.Nullable = val
+			} else if strings.Contains(kl, "default") {
+				col.Default = &val
+			} else if strings.Contains(kl, "key") {
+				col.Key = val
+			} else if strings.Contains(kl, "comment") {
+				col.Comment = val
+			}
+		}
+		columns = append(columns, col)
+	}
+	return columns, nil
+}
+
+func (c *CustomDB) GetIndexes(dbName, tableName string) ([]connection.IndexDefinition, error) {
+	return nil, fmt.Errorf("not implemented for custom")
+}
+
+func (c *CustomDB) GetForeignKeys(dbName, tableName string) ([]connection.ForeignKeyDefinition, error) {
+	return nil, fmt.Errorf("not implemented for custom")
+}
+
+func (c *CustomDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDefinition, error) {
+	return nil, fmt.Errorf("not implemented for custom")
+}
+
+func (c *CustomDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	return fmt.Errorf("read-only mode for custom")
+}
+
+func (c *CustomDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
+	return nil, fmt.Errorf("not implemented for custom")
+}

internal/db/dameng_impl.go

@@ -0,0 +1,340 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/ssh"
+	"GoNavi-Wails/internal/utils"
+
+	_ "gitee.com/chunanyong/dm"
+)
+
+type DamengDB struct {
+	conn *sql.DB
+}
+
+func (d *DamengDB) getDSN(config connection.ConnectionConfig) string {
+	// dm://user:password@host:port?schema=...
+	// or dm://user:password@host:port
+
+	address := fmt.Sprintf("%s:%d", config.Host, config.Port)
+	if config.UseSSH {
+		// SSH logic similar to others, assumes port forwarding
+		_, err := ssh.RegisterSSHNetwork(config.SSH)
+		if err == nil {
+			// DM driver likely uses standard net.Dial, so we might need a local listener
+			// or assume port forwarding is handled externally or implicitly via "tcp" override if driver allows.
+			// Similar to Oracle, we skip complex custom dialer injection for now.
+		}
+	}
+
+	dsn := fmt.Sprintf("dm://%s:%s@%s", config.User, config.Password, address)
+	if config.Database != "" {
+		dsn += fmt.Sprintf("?schema=%s", config.Database)
+	}
+	return dsn
+}
+
+func (d *DamengDB) Connect(config connection.ConnectionConfig) error {
+	dsn := d.getDSN(config)
+	db, err := sql.Open("dm", dsn)
+	if err != nil {
+		return err
+	}
+	d.conn = db
+	return d.Ping()
+}
+
+func (d *DamengDB) Close() error {
+	if d.conn != nil {
+		return d.conn.Close()
+	}
+	return nil
+}
+
+func (d *DamengDB) Ping() error {
+	if d.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+	ctx, cancel := utils.ContextWithTimeout(5 * time.Second)
+	defer cancel()
+	return d.conn.PingContext(ctx)
+}
+
+func (d *DamengDB) Query(query string) ([]map[string]interface{}, []string, error) {
+	if d.conn == nil {
+		return nil, nil, fmt.Errorf("connection not open")
+	}
+
+	rows, err := d.conn.Query(query)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var resultData []map[string]interface{}
+
+	for rows.Next() {
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			continue
+		}
+
+		entry := make(map[string]interface{})
+		for i, col := range columns {
+			var v interface{}
+			val := values[i]
+			b, ok := val.([]byte)
+			if ok {
+				v = string(b)
+			} else {
+				v = val
+			}
+			entry[col] = v
+		}
+		resultData = append(resultData, entry)
+	}
+
+	return resultData, columns, nil
+}
+
+func (d *DamengDB) Exec(query string) (int64, error) {
+	if d.conn == nil {
+		return 0, fmt.Errorf("connection not open")
+	}
+	res, err := d.conn.Exec(query)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+func (d *DamengDB) GetDatabases() ([]string, error) {
+	// DM: List Users/Schemas
+	data, _, err := d.Query("SELECT username FROM dba_users")
+	if err != nil {
+		// Fallback if dba_users not accessible
+		data, _, err = d.Query("SELECT username FROM all_users")
+		if err != nil {
+			return nil, err
+		}
+	}
+	var dbs []string
+	for _, row := range data {
+		if val, ok := row["USERNAME"]; ok {
+			dbs = append(dbs, fmt.Sprintf("%v", val))
+		}
+	}
+	return dbs, nil
+}
+
+func (d *DamengDB) GetTables(dbName string) ([]string, error) {
+	query := fmt.Sprintf("SELECT table_name FROM all_tables WHERE owner = '%s'", strings.ToUpper(dbName))
+	if dbName == "" {
+		query = "SELECT table_name FROM user_tables"
+	}
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var tables []string
+	for _, row := range data {
+		if val, ok := row["TABLE_NAME"]; ok {
+			tables = append(tables, fmt.Sprintf("%v", val))
+		}
+	}
+	return tables, nil
+}
+
+func (d *DamengDB) GetCreateStatement(dbName, tableName string) (string, error) {
+	// DM: SP_TABLEDEF usually returns definition
+	// Or standard Oracle way if supported.
+	// We'll try a common DM approach.
+	// SELECT DBMS_METADATA.GET_DDL('TABLE', 'TABLE_NAME', 'OWNER') FROM DUAL;
+
+	query := fmt.Sprintf("SELECT DBMS_METADATA.GET_DDL('TABLE', '%s', '%s') as ddl FROM DUAL",
+		strings.ToUpper(tableName), strings.ToUpper(dbName))
+
+	if dbName == "" {
+		query = fmt.Sprintf("SELECT DBMS_METADATA.GET_DDL('TABLE', '%s') as ddl FROM DUAL", strings.ToUpper(tableName))
+	}
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return "", err
+	}
+
+	if len(data) > 0 {
+		if val, ok := data[0]["DDL"]; ok {
+			return fmt.Sprintf("%v", val), nil
+		}
+	}
+	return "", fmt.Errorf("create statement not found")
+}
+
+func (d *DamengDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefinition, error) {
+	query := fmt.Sprintf(`SELECT column_name, data_type, nullable, data_default 
+		FROM all_tab_columns 
+		WHERE owner = '%s' AND table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	if dbName == "" {
+		query = fmt.Sprintf(`SELECT column_name, data_type, nullable, data_default 
+			FROM user_tab_columns 
+			WHERE table_name = '%s'`, strings.ToUpper(tableName))
+	}
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var columns []connection.ColumnDefinition
+	for _, row := range data {
+		col := connection.ColumnDefinition{
+			Name:     fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			Type:     fmt.Sprintf("%v", row["DATA_TYPE"]),
+			Nullable: fmt.Sprintf("%v", row["NULLABLE"]),
+		}
+
+		if row["DATA_DEFAULT"] != nil {
+			def := fmt.Sprintf("%v", row["DATA_DEFAULT"])
+			col.Default = &def
+		}
+
+		columns = append(columns, col)
+	}
+	return columns, nil
+}
+
+func (d *DamengDB) GetIndexes(dbName, tableName string) ([]connection.IndexDefinition, error) {
+	query := fmt.Sprintf(`SELECT index_name, column_name, uniqueness 
+		FROM all_ind_columns 
+		JOIN all_indexes USING (index_name, owner) 
+		WHERE table_owner = '%s' AND table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	if dbName == "" {
+		query = fmt.Sprintf(`SELECT index_name, column_name, uniqueness 
+			FROM user_ind_columns 
+			JOIN user_indexes USING (index_name) 
+			WHERE table_name = '%s'`, strings.ToUpper(tableName))
+	}
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var indexes []connection.IndexDefinition
+	for _, row := range data {
+		unique := 1
+		if val, ok := row["UNIQUENESS"]; ok && val == "UNIQUE" {
+			unique = 0
+		}
+
+		idx := connection.IndexDefinition{
+			Name:       fmt.Sprintf("%v", row["INDEX_NAME"]),
+			ColumnName: fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			NonUnique:  unique,
+			IndexType:  "BTREE",
+		}
+		indexes = append(indexes, idx)
+	}
+	return indexes, nil
+}
+
+func (d *DamengDB) GetForeignKeys(dbName, tableName string) ([]connection.ForeignKeyDefinition, error) {
+	// Reusing Oracle style query as DM is highly compatible
+	query := fmt.Sprintf(`SELECT a.constraint_name, a.column_name, c_pk.table_name r_table_name, b.column_name r_column_name
+		FROM all_cons_columns a
+		JOIN all_constraints c ON a.owner = c.owner AND a.constraint_name = c.constraint_name
+		JOIN all_constraints c_pk ON c.r_owner = c_pk.owner AND c.r_constraint_name = c_pk.constraint_name
+		JOIN all_cons_columns b ON c_pk.owner = b.owner AND c_pk.constraint_name = b.constraint_name AND a.position = b.position
+		WHERE c.constraint_type = 'R' AND a.owner = '%s' AND a.table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var fks []connection.ForeignKeyDefinition
+	for _, row := range data {
+		fk := connection.ForeignKeyDefinition{
+			Name:           fmt.Sprintf("%v", row["CONSTRAINT_NAME"]),
+			ColumnName:     fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			RefTableName:   fmt.Sprintf("%v", row["R_TABLE_NAME"]),
+			RefColumnName:  fmt.Sprintf("%v", row["R_COLUMN_NAME"]),
+			ConstraintName: fmt.Sprintf("%v", row["CONSTRAINT_NAME"]),
+		}
+		fks = append(fks, fk)
+	}
+	return fks, nil
+}
+
+func (d *DamengDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDefinition, error) {
+	query := fmt.Sprintf(`SELECT trigger_name, trigger_type, triggering_event 
+		FROM all_triggers 
+		WHERE table_owner = '%s' AND table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var triggers []connection.TriggerDefinition
+	for _, row := range data {
+		trig := connection.TriggerDefinition{
+			Name:      fmt.Sprintf("%v", row["TRIGGER_NAME"]),
+			Timing:    fmt.Sprintf("%v", row["TRIGGER_TYPE"]),
+			Event:     fmt.Sprintf("%v", row["TRIGGERING_EVENT"]),
+			Statement: "SOURCE HIDDEN",
+		}
+		triggers = append(triggers, trig)
+	}
+	return triggers, nil
+}
+
+func (d *DamengDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	return fmt.Errorf("read-only mode implemented for Dameng so far")
+}
+
+func (d *DamengDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
+	query := fmt.Sprintf(`SELECT table_name, column_name, data_type 
+		FROM all_tab_columns 
+		WHERE owner = '%s'`, strings.ToUpper(dbName))
+
+	data, _, err := d.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var cols []connection.ColumnDefinitionWithTable
+	for _, row := range data {
+		col := connection.ColumnDefinitionWithTable{
+			TableName: fmt.Sprintf("%v", row["TABLE_NAME"]),
+			Name:      fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			Type:      fmt.Sprintf("%v", row["DATA_TYPE"]),
+		}
+		cols = append(cols, col)
+	}
+	return cols, nil
+}

internal/db/database.go

@@ -1,8 +1,8 @@
 package db
 
 import (
-	"fmt"
 	"GoNavi-Wails/internal/connection"
+	"fmt"
 )
 
 type Database interface {
@@ -34,6 +34,14 @@ func NewDatabase(dbType string) (Database, error) {
 		return &PostgresDB{}, nil
 	case "sqlite":
 		return &SQLiteDB{}, nil
+	case "oracle":
+		return &OracleDB{}, nil
+	case "dameng":
+		return &DamengDB{}, nil
+	case "kingbase":
+		return &KingbaseDB{}, nil
+	case "custom":
+		return &CustomDB{}, nil
 	default:
 		// Default to MySQL for backward compatibility if empty
 		if dbType == "" {

internal/db/kingbase_impl.go

@@ -0,0 +1,363 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/ssh"
+	"GoNavi-Wails/internal/utils"
+
+	_ "gitea.com/kingbase/gokb" // Registers "kingbase" driver
+)
+
+type KingbaseDB struct {
+	conn *sql.DB
+}
+
+func (k *KingbaseDB) getDSN(config connection.ConnectionConfig) string {
+	// Kingbase DSN usually similar to Postgres:
+	// host=localhost port=54321 user=system password=... dbname=TEST sslmode=disable
+
+	address := config.Host
+	port := config.Port
+
+	if config.UseSSH {
+		netName, err := ssh.RegisterSSHNetwork(config.SSH)
+		if err == nil {
+			// Kingbase/Postgres lib/pq allows custom dialer via "host" if using unix socket,
+			// but for custom network it's harder.
+			// Ideally we use a local forwarder.
+			// For now, we assume standard TCP or handle SSH externally.
+			// If we implement the net.Dial override for "kingbase" driver (which might use lib/pq internally),
+			// we might need to check if it supports "cloudsql" style or similar custom dialers.
+			// Similar to others, skipping SSH deep integration here for now.
+			_ = netName
+		}
+	}
+
+	// Construct DSN
+	dsn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
+		address, port, config.User, config.Password, config.Database)
+
+	return dsn
+}
+
+func (k *KingbaseDB) Connect(config connection.ConnectionConfig) error {
+	dsn := k.getDSN(config)
+	// Open using "kingbase" driver
+	db, err := sql.Open("kingbase", dsn)
+	if err != nil {
+		return err
+	}
+	k.conn = db
+	return k.Ping()
+}
+
+func (k *KingbaseDB) Close() error {
+	if k.conn != nil {
+		return k.conn.Close()
+	}
+	return nil
+}
+
+func (k *KingbaseDB) Ping() error {
+	if k.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+	ctx, cancel := utils.ContextWithTimeout(5 * time.Second)
+	defer cancel()
+	return k.conn.PingContext(ctx)
+}
+
+func (k *KingbaseDB) Query(query string) ([]map[string]interface{}, []string, error) {
+	if k.conn == nil {
+		return nil, nil, fmt.Errorf("connection not open")
+	}
+
+	rows, err := k.conn.Query(query)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var resultData []map[string]interface{}
+
+	for rows.Next() {
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			continue
+		}
+
+		entry := make(map[string]interface{})
+		for i, col := range columns {
+			var v interface{}
+			val := values[i]
+			b, ok := val.([]byte)
+			if ok {
+				v = string(b)
+			} else {
+				v = val
+			}
+			entry[col] = v
+		}
+		resultData = append(resultData, entry)
+	}
+
+	return resultData, columns, nil
+}
+
+func (k *KingbaseDB) Exec(query string) (int64, error) {
+	if k.conn == nil {
+		return 0, fmt.Errorf("connection not open")
+	}
+	res, err := k.conn.Exec(query)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+func (k *KingbaseDB) GetDatabases() ([]string, error) {
+	// Postgres/Kingbase style
+	data, _, err := k.Query("SELECT datname FROM pg_database WHERE datistemplate = false")
+	if err != nil {
+		return nil, err
+	}
+	var dbs []string
+	for _, row := range data {
+		if val, ok := row["datname"]; ok {
+			dbs = append(dbs, fmt.Sprintf("%v", val))
+		}
+	}
+	return dbs, nil
+}
+
+func (k *KingbaseDB) GetTables(dbName string) ([]string, error) {
+	// Usually restricted to current database connection in PG/Kingbase
+	// dbName param is often Schema in PG context, or ignored if we are connected to a specific DB.
+	// But in PG, cross-database queries are not standard without dblink.
+	// We assume dbName here might mean Schema (public, etc.)
+
+	query := "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"
+	if dbName != "" && dbName != "public" {
+		query = fmt.Sprintf("SELECT table_name FROM information_schema.tables WHERE table_schema = '%s'", dbName)
+	}
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var tables []string
+	for _, row := range data {
+		if val, ok := row["table_name"]; ok {
+			tables = append(tables, fmt.Sprintf("%v", val))
+		}
+	}
+	return tables, nil
+}
+
+func (k *KingbaseDB) GetCreateStatement(dbName, tableName string) (string, error) {
+	// Kingbase doesn't have "SHOW CREATE TABLE".
+	// We can try pg_dump logic or use a query to reconstruction.
+	// A simple approach is just returning basic info or "Not Supported".
+	// Or we can query information_schema to build it.
+	return "SHOW CREATE TABLE not directly supported in Kingbase/Postgres via SQL", nil
+}
+
+func (k *KingbaseDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefinition, error) {
+	schema := "public"
+	if dbName != "" {
+		schema = dbName
+	}
+
+	query := fmt.Sprintf(`SELECT column_name, data_type, is_nullable, column_default 
+		FROM information_schema.columns 
+		WHERE table_schema = '%s' AND table_name = '%s' 
+		ORDER BY ordinal_position`, schema, tableName)
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var columns []connection.ColumnDefinition
+	for _, row := range data {
+		col := connection.ColumnDefinition{
+			Name:     fmt.Sprintf("%v", row["column_name"]),
+			Type:     fmt.Sprintf("%v", row["data_type"]),
+			Nullable: fmt.Sprintf("%v", row["is_nullable"]),
+		}
+
+		if row["column_default"] != nil {
+			def := fmt.Sprintf("%v", row["column_default"])
+			col.Default = &def
+		}
+
+		columns = append(columns, col)
+	}
+	return columns, nil
+}
+
+func (k *KingbaseDB) GetIndexes(dbName, tableName string) ([]connection.IndexDefinition, error) {
+	// Postgres/Kingbase index query
+	query := fmt.Sprintf(`
+		SELECT
+			i.relname as index_name,
+			a.attname as column_name,
+			ix.indisunique as is_unique
+		FROM
+			pg_class t,
+			pg_class i,
+			pg_index ix,
+			pg_attribute a,
+			pg_namespace n
+		WHERE
+			t.oid = ix.indrelid
+			AND i.oid = ix.indexrelid
+			AND a.attrelid = t.oid
+			AND a.attnum = ANY(ix.indkey)
+			AND t.relkind = 'r'
+			AND t.relname = '%s'
+			AND n.oid = t.relnamespace
+			AND n.nspname = '%s'
+	`, tableName, "public") // Default to public if dbName (schema) not clear.
+
+	if dbName != "" {
+		// Update query to use dbName as schema
+		query = strings.Replace(query, "'public'", fmt.Sprintf("'%s'", dbName), 1)
+	}
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var indexes []connection.IndexDefinition
+	for _, row := range data {
+		nonUnique := 1
+		if val, ok := row["is_unique"]; ok {
+			if b, ok := val.(bool); ok && b {
+				nonUnique = 0
+			}
+		}
+
+		idx := connection.IndexDefinition{
+			Name:       fmt.Sprintf("%v", row["index_name"]),
+			ColumnName: fmt.Sprintf("%v", row["column_name"]),
+			NonUnique:  nonUnique,
+			IndexType:  "BTREE", // Default
+		}
+		indexes = append(indexes, idx)
+	}
+	return indexes, nil
+}
+
+func (k *KingbaseDB) GetForeignKeys(dbName, tableName string) ([]connection.ForeignKeyDefinition, error) {
+	schema := "public"
+	if dbName != "" {
+		schema = dbName
+	}
+
+	query := fmt.Sprintf(`
+		SELECT
+			tc.constraint_name, 
+			kcu.column_name, 
+			ccu.table_name AS foreign_table_name,
+			ccu.column_name AS foreign_column_name 
+		FROM 
+			information_schema.table_constraints AS tc 
+			JOIN information_schema.key_column_usage AS kcu
+			  ON tc.constraint_name = kcu.constraint_name
+			  AND tc.table_schema = kcu.table_schema
+			JOIN information_schema.constraint_column_usage AS ccu
+			  ON ccu.constraint_name = tc.constraint_name
+			  AND ccu.table_schema = tc.table_schema
+		WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='%s' AND tc.table_schema='%s'`,
+		tableName, schema)
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var fks []connection.ForeignKeyDefinition
+	for _, row := range data {
+		fk := connection.ForeignKeyDefinition{
+			Name:           fmt.Sprintf("%v", row["constraint_name"]),
+			ColumnName:     fmt.Sprintf("%v", row["column_name"]),
+			RefTableName:   fmt.Sprintf("%v", row["foreign_table_name"]),
+			RefColumnName:  fmt.Sprintf("%v", row["foreign_column_name"]),
+			ConstraintName: fmt.Sprintf("%v", row["constraint_name"]),
+		}
+		fks = append(fks, fk)
+	}
+	return fks, nil
+}
+
+func (k *KingbaseDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDefinition, error) {
+	query := fmt.Sprintf(`SELECT trigger_name, action_timing, event_manipulation 
+		FROM information_schema.triggers 
+		WHERE event_object_table = '%s'`, tableName)
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var triggers []connection.TriggerDefinition
+	for _, row := range data {
+		trig := connection.TriggerDefinition{
+			Name:      fmt.Sprintf("%v", row["trigger_name"]),
+			Timing:    fmt.Sprintf("%v", row["action_timing"]),
+			Event:     fmt.Sprintf("%v", row["event_manipulation"]),
+			Statement: "SOURCE HIDDEN",
+		}
+		triggers = append(triggers, trig)
+	}
+	return triggers, nil
+}
+
+func (k *KingbaseDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	return fmt.Errorf("read-only mode implemented for Kingbase so far")
+}
+
+func (k *KingbaseDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
+	schema := "public"
+	if dbName != "" {
+		schema = dbName
+	}
+
+	query := fmt.Sprintf(`SELECT table_name, column_name, data_type 
+		FROM information_schema.columns 
+		WHERE table_schema = '%s'`, schema)
+
+	data, _, err := k.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var cols []connection.ColumnDefinitionWithTable
+	for _, row := range data {
+		col := connection.ColumnDefinitionWithTable{
+			TableName: fmt.Sprintf("%v", row["table_name"]),
+			Name:      fmt.Sprintf("%v", row["column_name"]),
+			Type:      fmt.Sprintf("%v", row["data_type"]),
+		}
+		cols = append(cols, col)
+	}
+	return cols, nil
+}

internal/db/oracle_impl.go

@@ -0,0 +1,350 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/ssh"
+	"GoNavi-Wails/internal/utils"
+
+	_ "github.com/sijms/go-ora/v2"
+)
+
+type OracleDB struct {
+	conn *sql.DB
+}
+
+func (o *OracleDB) getDSN(config connection.ConnectionConfig) string {
+	// oracle://user:pass@host:port/service_name
+	database := config.Database
+	if database == "" {
+		database = config.User // Default to user service/schema if empty?
+	}
+
+	address := fmt.Sprintf("%s:%d", config.Host, config.Port)
+	if config.UseSSH {
+		_, err := ssh.RegisterSSHNetwork(config.SSH)
+		if err == nil {
+			// Oracle driver might not support custom dialer via DSN easily without extra config
+			// But go-ora v2 supports some advanced options.
+			// For simplicity, we assume standard TCP or we might need a workaround for SSH.
+			// go-ora v2 is pure Go, so we can potentially use a custom dialer if we manually open.
+			// But for now, let's just use the address.
+			// SSH tunneling via net.Dialer override is complex in sql.Open("oracle", ...).
+			// We might need to forward a local port if using SSH.
+			// Since ssh.RegisterSSHNetwork creates a custom network "ssh-via-...",
+			// we need to see if go-ora supports custom networks.
+			// Checking go-ora docs (simulated): It supports "unix" and "tcp".
+			// We might need to map the custom network to a local proxy.
+			// For now, we will assume direct connection or handle SSH separately later.
+			// We'll leave the protocol implementation as is in MySQL for now, hoping go-ora uses standard net.Dial.
+			// Note: go-ora connection string: oracle://user:pass@host:port/service
+			// It parses host/port. It doesn't easily take a custom "network" parameter in URL.
+			// We will proceed with standard TCP string.
+		}
+	}
+
+	// go-ora url structure: oracle://user:password@address:port/service_name
+	return fmt.Sprintf("oracle://%s:%s@%s/%s",
+		config.User, config.Password, address, database)
+}
+
+func (o *OracleDB) Connect(config connection.ConnectionConfig) error {
+	dsn := o.getDSN(config)
+	db, err := sql.Open("oracle", dsn)
+	if err != nil {
+		return err
+	}
+	o.conn = db
+	return o.Ping()
+}
+
+func (o *OracleDB) Close() error {
+	if o.conn != nil {
+		return o.conn.Close()
+	}
+	return nil
+}
+
+func (o *OracleDB) Ping() error {
+	if o.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+	ctx, cancel := utils.ContextWithTimeout(5 * time.Second)
+	defer cancel()
+	return o.conn.PingContext(ctx)
+}
+
+func (o *OracleDB) Query(query string) ([]map[string]interface{}, []string, error) {
+	if o.conn == nil {
+		return nil, nil, fmt.Errorf("connection not open")
+	}
+
+	rows, err := o.conn.Query(query)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var resultData []map[string]interface{}
+
+	for rows.Next() {
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			continue
+		}
+
+		entry := make(map[string]interface{})
+		for i, col := range columns {
+			var v interface{}
+			val := values[i]
+			b, ok := val.([]byte)
+			if ok {
+				v = string(b)
+			} else {
+				v = val
+			}
+			entry[col] = v
+		}
+		resultData = append(resultData, entry)
+	}
+
+	return resultData, columns, nil
+}
+
+func (o *OracleDB) Exec(query string) (int64, error) {
+	if o.conn == nil {
+		return 0, fmt.Errorf("connection not open")
+	}
+	res, err := o.conn.Exec(query)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+func (o *OracleDB) GetDatabases() ([]string, error) {
+	// Oracle treats Users/Schemas as "Databases" in this context
+	data, _, err := o.Query("SELECT username FROM all_users ORDER BY username")
+	if err != nil {
+		return nil, err
+	}
+	var dbs []string
+	for _, row := range data {
+		if val, ok := row["USERNAME"]; ok {
+			dbs = append(dbs, fmt.Sprintf("%v", val))
+		}
+	}
+	return dbs, nil
+}
+
+func (o *OracleDB) GetTables(dbName string) ([]string, error) {
+	// dbName is Schema/Owner
+	query := "SELECT table_name FROM user_tables"
+	if dbName != "" {
+		query = fmt.Sprintf("SELECT table_name FROM all_tables WHERE owner = '%s'", strings.ToUpper(dbName))
+	}
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var tables []string
+	for _, row := range data {
+		if val, ok := row["TABLE_NAME"]; ok {
+			tables = append(tables, fmt.Sprintf("%v", val))
+		}
+	}
+	return tables, nil
+}
+
+func (o *OracleDB) GetCreateStatement(dbName, tableName string) (string, error) {
+	// Oracle provides DBMS_METADATA.GET_DDL
+	// Note: LONG type might be tricky, but basic string scan should work for smaller DDLs
+	query := fmt.Sprintf("SELECT DBMS_METADATA.GET_DDL('TABLE', '%s', '%s') as ddl FROM DUAL",
+		strings.ToUpper(tableName), strings.ToUpper(dbName))
+
+	if dbName == "" {
+		query = fmt.Sprintf("SELECT DBMS_METADATA.GET_DDL('TABLE', '%s') as ddl FROM DUAL", strings.ToUpper(tableName))
+	}
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return "", err
+	}
+
+	if len(data) > 0 {
+		if val, ok := data[0]["DDL"]; ok {
+			return fmt.Sprintf("%v", val), nil
+		}
+	}
+	return "", fmt.Errorf("create statement not found")
+}
+
+func (o *OracleDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefinition, error) {
+	query := fmt.Sprintf(`SELECT column_name, data_type, nullable, data_default 
+		FROM all_tab_columns 
+		WHERE owner = '%s' AND table_name = '%s' 
+		ORDER BY column_id`, strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	if dbName == "" {
+		query = fmt.Sprintf(`SELECT column_name, data_type, nullable, data_default 
+			FROM user_tab_columns 
+			WHERE table_name = '%s' 
+			ORDER BY column_id`, strings.ToUpper(tableName))
+	}
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var columns []connection.ColumnDefinition
+	for _, row := range data {
+		col := connection.ColumnDefinition{
+			Name:     fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			Type:     fmt.Sprintf("%v", row["DATA_TYPE"]),
+			Nullable: fmt.Sprintf("%v", row["NULLABLE"]),
+		}
+
+		if row["DATA_DEFAULT"] != nil {
+			d := fmt.Sprintf("%v", row["DATA_DEFAULT"])
+			col.Default = &d
+		}
+
+		columns = append(columns, col)
+	}
+	return columns, nil
+}
+
+func (o *OracleDB) GetIndexes(dbName, tableName string) ([]connection.IndexDefinition, error) {
+	query := fmt.Sprintf(`SELECT index_name, column_name, uniqueness 
+		FROM all_ind_columns 
+		JOIN all_indexes USING (index_name, owner) 
+		WHERE table_owner = '%s' AND table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	if dbName == "" {
+		query = fmt.Sprintf(`SELECT index_name, column_name, uniqueness 
+			FROM user_ind_columns 
+			JOIN user_indexes USING (index_name) 
+			WHERE table_name = '%s'`, strings.ToUpper(tableName))
+	}
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var indexes []connection.IndexDefinition
+	for _, row := range data {
+		unique := 1
+		if val, ok := row["UNIQUENESS"]; ok && val == "UNIQUE" {
+			unique = 0
+		}
+
+		idx := connection.IndexDefinition{
+			Name:       fmt.Sprintf("%v", row["INDEX_NAME"]),
+			ColumnName: fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			NonUnique:  unique,
+			// SeqInIndex is harder to get in simple join, omitting or estimating
+			IndexType: "BTREE", // Default assumption
+		}
+		indexes = append(indexes, idx)
+	}
+	return indexes, nil
+}
+
+func (o *OracleDB) GetForeignKeys(dbName, tableName string) ([]connection.ForeignKeyDefinition, error) {
+	// Simplified query for FKs
+	query := fmt.Sprintf(`SELECT a.constraint_name, a.column_name, c_pk.table_name r_table_name, b.column_name r_column_name
+		FROM all_cons_columns a
+		JOIN all_constraints c ON a.owner = c.owner AND a.constraint_name = c.constraint_name
+		JOIN all_constraints c_pk ON c.r_owner = c_pk.owner AND c.r_constraint_name = c_pk.constraint_name
+		JOIN all_cons_columns b ON c_pk.owner = b.owner AND c_pk.constraint_name = b.constraint_name AND a.position = b.position
+		WHERE c.constraint_type = 'R' AND a.owner = '%s' AND a.table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var fks []connection.ForeignKeyDefinition
+	for _, row := range data {
+		fk := connection.ForeignKeyDefinition{
+			Name:           fmt.Sprintf("%v", row["CONSTRAINT_NAME"]),
+			ColumnName:     fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			RefTableName:   fmt.Sprintf("%v", row["R_TABLE_NAME"]),
+			RefColumnName:  fmt.Sprintf("%v", row["R_COLUMN_NAME"]),
+			ConstraintName: fmt.Sprintf("%v", row["CONSTRAINT_NAME"]),
+		}
+		fks = append(fks, fk)
+	}
+	return fks, nil
+}
+
+func (o *OracleDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDefinition, error) {
+	query := fmt.Sprintf(`SELECT trigger_name, trigger_type, triggering_event 
+		FROM all_triggers 
+		WHERE table_owner = '%s' AND table_name = '%s'`,
+		strings.ToUpper(dbName), strings.ToUpper(tableName))
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var triggers []connection.TriggerDefinition
+	for _, row := range data {
+		trig := connection.TriggerDefinition{
+			Name:      fmt.Sprintf("%v", row["TRIGGER_NAME"]),
+			Timing:    fmt.Sprintf("%v", row["TRIGGER_TYPE"]),
+			Event:     fmt.Sprintf("%v", row["TRIGGERING_EVENT"]),
+			Statement: "SOURCE HIDDEN", // Requires more complex query to get body
+		}
+		triggers = append(triggers, trig)
+	}
+	return triggers, nil
+}
+
+func (o *OracleDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	// TODO: Implement batch application for Oracle using correct syntax
+	return fmt.Errorf("read-only mode implemented for Oracle so far")
+}
+
+func (o *OracleDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
+	query := fmt.Sprintf(`SELECT table_name, column_name, data_type 
+		FROM all_tab_columns 
+		WHERE owner = '%s'`, strings.ToUpper(dbName))
+
+	data, _, err := o.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var cols []connection.ColumnDefinitionWithTable
+	for _, row := range data {
+		col := connection.ColumnDefinitionWithTable{
+			TableName: fmt.Sprintf("%v", row["TABLE_NAME"]),
+			Name:      fmt.Sprintf("%v", row["COLUMN_NAME"]),
+			Type:      fmt.Sprintf("%v", row["DATA_TYPE"]),
+		}
+		cols = append(cols, col)
+	}
+	return cols, nil
+}

internal/sync/sync_engine.go

@@ -0,0 +1,179 @@
+package sync
+
+import (
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/db"
+	"fmt"
+)
+
+// SyncConfig defines the parameters for a synchronization task
+type SyncConfig struct {
+	SourceConfig connection.ConnectionConfig `json:"sourceConfig"`
+	TargetConfig connection.ConnectionConfig `json:"targetConfig"`
+	Tables       []string                    `json:"tables"` // Tables to sync
+	Mode         string                      `json:"mode"`   // "insert_update", "full_overwrite"
+}
+
+// SyncResult holds the result of the sync operation
+type SyncResult struct {
+	Success      bool     `json:"success"`
+	Message      string   `json:"message"`
+	Logs         []string `json:"logs"`
+	TablesSynced int      `json:"tablesSynced"`
+	RowsInserted int      `json:"rowsInserted"`
+	RowsUpdated  int      `json:"rowsUpdated"`
+	RowsDeleted  int      `json:"rowsDeleted"`
+}
+
+type SyncEngine struct {
+}
+
+func NewSyncEngine() *SyncEngine {
+	return &SyncEngine{}
+}
+
+// CompareAndSync performs the synchronization
+func (s *SyncEngine) RunSync(config SyncConfig) SyncResult {
+	result := SyncResult{Success: true, Logs: []string{}}
+
+	sourceDB, err := db.NewDatabase(config.SourceConfig.Type)
+	if err != nil {
+		return s.fail(result, "初始化源数据库驱动失败: "+err.Error())
+	}
+	if config.SourceConfig.Type == "custom" {
+		// Custom DB setup would go here if needed
+	}
+
+	targetDB, err := db.NewDatabase(config.TargetConfig.Type)
+	if err != nil {
+		return s.fail(result, "初始化目标数据库驱动失败: "+err.Error())
+	}
+
+	// Connect Source
+	result.Logs = append(result.Logs, fmt.Sprintf("正在连接源数据库: %s...", config.SourceConfig.Host))
+	if err := sourceDB.Connect(config.SourceConfig); err != nil {
+		return s.fail(result, "源数据库连接失败: "+err.Error())
+	}
+	defer sourceDB.Close()
+
+	// Connect Target
+	result.Logs = append(result.Logs, fmt.Sprintf("正在连接目标数据库: %s...", config.TargetConfig.Host))
+	if err := targetDB.Connect(config.TargetConfig); err != nil {
+		return s.fail(result, "目标数据库连接失败: "+err.Error())
+	}
+	defer targetDB.Close()
+
+	// Iterate Tables
+	for _, tableName := range config.Tables {
+		result.Logs = append(result.Logs, fmt.Sprintf("正在同步表: %s", tableName))
+
+		// 1. Get Columns & PKs (Naive approach: assume same schema)
+		cols, err := sourceDB.GetColumns(config.SourceConfig.Database, tableName)
+		if err != nil {
+			result.Logs = append(result.Logs, fmt.Sprintf("获取表 %s 的列信息失败: %v", tableName, err))
+			continue
+		}
+
+		pkCol := ""
+		for _, col := range cols {
+			if col.Key == "PRI" || col.Key == "PK" {
+				pkCol = col.Name
+				break
+			}
+		}
+
+		if pkCol == "" {
+			result.Logs = append(result.Logs, fmt.Sprintf("跳过表 %s: 未找到主键 (同步需要主键)", tableName))
+			continue
+		}
+
+		// 2. Fetch Data (MEMORY INTENSIVE - PROTOTYPE ONLY)
+		// TODO: Implement paging/streaming
+		sourceRows, _, err := sourceDB.Query(fmt.Sprintf("SELECT * FROM %s", tableName))
+		if err != nil {
+			result.Logs = append(result.Logs, fmt.Sprintf("读取源表 %s 失败: %v", tableName, err))
+			continue
+		}
+
+		targetRows, _, err := targetDB.Query(fmt.Sprintf("SELECT * FROM %s", tableName))
+		if err != nil {
+			// Table might not exist in target?
+			// Check if error is "table not found" -> Try to Create?
+			// For now, assume table exists.
+			result.Logs = append(result.Logs, fmt.Sprintf("读取目标表 %s 失败: %v", tableName, err))
+			continue
+		}
+
+		// 3. Compare (In-Memory Hash Map)
+		targetMap := make(map[string]map[string]interface{})
+		for _, row := range targetRows {
+			pkVal := fmt.Sprintf("%v", row[pkCol])
+			targetMap[pkVal] = row
+		}
+
+		var inserts []map[string]interface{}
+		var updates []connection.UpdateRow
+		// var deletes []map[string]interface{} // Not implemented in "insert_update" mode usually
+
+		for _, sRow := range sourceRows {
+			pkVal := fmt.Sprintf("%v", sRow[pkCol])
+
+			if tRow, exists := targetMap[pkVal]; exists {
+				// Update? Compare values
+				// Simplified: Compare string representations or iterate keys
+				// For prototype: assume update if exists
+				// Optimization: Check diff
+				changes := make(map[string]interface{})
+				for k, v := range sRow {
+					if fmt.Sprintf("%v", v) != fmt.Sprintf("%v", tRow[k]) {
+						changes[k] = v
+					}
+				}
+				if len(changes) > 0 {
+					updates = append(updates, connection.UpdateRow{
+						Keys:   map[string]interface{}{pkCol: pkVal},
+						Values: changes,
+					})
+				}
+			} else {
+				// Insert
+				inserts = append(inserts, sRow)
+			}
+		}
+
+		// 4. Apply Changes
+		changeSet := connection.ChangeSet{
+			Inserts: inserts,
+			Updates: updates,
+		}
+
+		if len(inserts) > 0 || len(updates) > 0 {
+			result.Logs = append(result.Logs, fmt.Sprintf("  -> 需插入: %d 行, 需更新: %d 行", len(inserts), len(updates)))
+
+			// We need a BatchApplier interface or assume Database implements ApplyChanges
+			if applier, ok := targetDB.(db.BatchApplier); ok {
+				if err := applier.ApplyChanges(tableName, changeSet); err != nil {
+					result.Logs = append(result.Logs, fmt.Sprintf("  -> 应用变更失败: %v", err))
+				} else {
+					result.RowsInserted += len(inserts)
+					result.RowsUpdated += len(updates)
+				}
+			} else {
+				result.Logs = append(result.Logs, "  -> 目标驱动不支持应用数据变更 (ApplyChanges).")
+			}
+		} else {
+			result.Logs = append(result.Logs, "  -> 数据一致，无需变更.")
+		}
+
+		result.TablesSynced++
+	}
+
+	return result
+}
+
+func (s *SyncEngine) fail(res SyncResult, msg string) SyncResult {
+	res.Success = false
+	res.Message = msg
+	res.Logs = append(res.Logs, "致命错误: "+msg)
+	return res
+}