whm
b04b180d30
Enable Binding-scoped agent push/pull, empty-table schema ensure, SyncPage inspect/drop-table, default module import, and agent-bound publish docs from the 宇恒联调意见. Co-authored-by: Cursor <cursoragent@cursor.com>
106 lines
2.9 KiB
Go
106 lines
2.9 KiB
Go
package dbsync
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"strings"
|
||
)
|
||
|
||
// EnsureTableFromRow 表不存在时按行字段自动建表（TEXT 列 + PK），满足 Z4「按 Binding 接受任意表」。
|
||
func EnsureTableFromRow(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string, row map[string]any) error {
|
||
table = strings.TrimSpace(table)
|
||
pkCol = strings.TrimSpace(pkCol)
|
||
if table == "" || pkCol == "" {
|
||
return fmt.Errorf("table and pk required")
|
||
}
|
||
if row == nil {
|
||
row = map[string]any{pkCol: ""}
|
||
}
|
||
cols := make([]string, 0, len(row))
|
||
seen := map[string]struct{}{}
|
||
if _, ok := row[pkCol]; !ok {
|
||
cols = append(cols, pkCol)
|
||
seen[pkCol] = struct{}{}
|
||
}
|
||
for k := range row {
|
||
k = strings.TrimSpace(k)
|
||
if k == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[k]; ok {
|
||
continue
|
||
}
|
||
seen[k] = struct{}{}
|
||
cols = append(cols, k)
|
||
}
|
||
return EnsureTableFromColumns(ctx, db, driver, table, pkCol, cols)
|
||
}
|
||
|
||
// EnsureTableFromColumns 表不存在时按列名建空表（全部 TEXT，指定 PK）。已存在则幂等跳过。
|
||
// 用于空表结构同步：本机有空表 → 线上也建同名空表（无需 outbox 行）。
|
||
func EnsureTableFromColumns(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string, columns []string) error {
|
||
table = strings.TrimSpace(table)
|
||
pkCol = strings.TrimSpace(pkCol)
|
||
if table == "" {
|
||
return fmt.Errorf("table required")
|
||
}
|
||
if strings.HasPrefix(table, "_ajz_") {
|
||
return fmt.Errorf("sync system table not allowed: %s", table)
|
||
}
|
||
if pkCol == "" {
|
||
pkCol = "id"
|
||
}
|
||
exists, err := tableExists(ctx, db, driver, table)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if exists {
|
||
return nil
|
||
}
|
||
seen := map[string]struct{}{}
|
||
defs := make([]string, 0, len(columns)+1)
|
||
defs = append(defs, fmt.Sprintf("%s TEXT PRIMARY KEY", quoteIdent(driver, pkCol)))
|
||
seen[pkCol] = struct{}{}
|
||
for _, c := range columns {
|
||
c = strings.TrimSpace(c)
|
||
if c == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[c]; ok {
|
||
continue
|
||
}
|
||
seen[c] = struct{}{}
|
||
defs = append(defs, fmt.Sprintf("%s TEXT", quoteIdent(driver, c)))
|
||
}
|
||
if len(defs) == 0 {
|
||
return fmt.Errorf("columns required")
|
||
}
|
||
ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s)`, quoteIdent(driver, table), strings.Join(defs, ", "))
|
||
_, err = db.ExecContext(ctx, ddl)
|
||
return err
|
||
}
|
||
|
||
func tableExists(ctx context.Context, db *sql.DB, driver Driver, table string) (bool, error) {
|
||
var q string
|
||
switch driver {
|
||
case DriverSQLite:
|
||
q = `SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1`
|
||
case DriverMySQL:
|
||
q = `SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=? LIMIT 1`
|
||
case DriverPostgres:
|
||
q = `SELECT 1 FROM information_schema.tables WHERE table_schema=current_schema() AND table_name=$1 LIMIT 1`
|
||
default:
|
||
return false, fmt.Errorf("unsupported driver")
|
||
}
|
||
var n int
|
||
err := db.QueryRowContext(ctx, q, table).Scan(&n)
|
||
if err == sql.ErrNoRows {
|
||
return false, nil
|
||
}
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return true, nil
|
||
}
|