chore: initial commit of ai site platform
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
292
platform/internal/dbsync/apply.go
Normal file
292
platform/internal/dbsync/apply.go
Normal file
@@ -0,0 +1,292 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
func PollOutbox(ctx context.Context, db *sql.DB, driver Driver, limit int) ([]OutboxRow, error) {
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
q := fmt.Sprintf(`SELECT id, table_name, row_pk, op, payload, version, created_at
|
||||
FROM %s WHERE synced_at IS NULL ORDER BY id ASC LIMIT %d`, OutboxTable, limit)
|
||||
rows, err := db.QueryContext(ctx, q)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []OutboxRow
|
||||
for rows.Next() {
|
||||
var r OutboxRow
|
||||
var created any
|
||||
if err := rows.Scan(&r.ID, &r.TableName, &r.RowPK, &r.Op, &r.Payload, &r.Version, &created); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
switch v := created.(type) {
|
||||
case time.Time:
|
||||
r.CreatedAt = v
|
||||
case string:
|
||||
t, _ := time.Parse("2006-01-02 15:04:05", v)
|
||||
r.CreatedAt = t
|
||||
case []byte:
|
||||
t, _ := time.Parse("2006-01-02 15:04:05", string(v))
|
||||
r.CreatedAt = t
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func MarkSynced(ctx context.Context, db *sql.DB, driver Driver, ids []int64) error {
|
||||
if len(ids) == 0 {
|
||||
return nil
|
||||
}
|
||||
nowExpr := "CURRENT_TIMESTAMP"
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
nowExpr = "datetime('now')"
|
||||
case DriverMySQL:
|
||||
nowExpr = "UTC_TIMESTAMP(3)"
|
||||
case DriverPostgres:
|
||||
nowExpr = "now()"
|
||||
}
|
||||
placeholders := make([]string, len(ids))
|
||||
args := make([]any, len(ids))
|
||||
for i, id := range ids {
|
||||
placeholders[i] = "?"
|
||||
if driver == DriverPostgres {
|
||||
placeholders[i] = fmt.Sprintf("$%d", i+1)
|
||||
}
|
||||
args[i] = id
|
||||
}
|
||||
q := fmt.Sprintf(`UPDATE %s SET synced_at = %s WHERE id IN (%s)`, OutboxTable, nowExpr, strings.Join(placeholders, ","))
|
||||
_, err := db.ExecContext(ctx, q, args...)
|
||||
return err
|
||||
}
|
||||
|
||||
func FetchRowJSON(ctx context.Context, db *sql.DB, driver Driver, table, pkCol, pkVal string) (string, map[string]any, error) {
|
||||
q := fmt.Sprintf(`SELECT * FROM %s WHERE %s = ? LIMIT 1`, quoteIdent(driver, table), quoteIdent(driver, pkCol))
|
||||
if driver == DriverPostgres {
|
||||
q = fmt.Sprintf(`SELECT * FROM %s WHERE %s = $1 LIMIT 1`, quoteIdent(driver, table), quoteIdent(driver, pkCol))
|
||||
}
|
||||
rows, err := db.QueryContext(ctx, q, pkVal)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
cols, err := rows.Columns()
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if !rows.Next() {
|
||||
return "", nil, sql.ErrNoRows
|
||||
}
|
||||
raw := make([]any, len(cols))
|
||||
ptrs := make([]any, len(cols))
|
||||
for i := range raw {
|
||||
ptrs[i] = &raw[i]
|
||||
}
|
||||
if err := rows.Scan(ptrs...); err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
m := map[string]any{}
|
||||
for i, c := range cols {
|
||||
m[c] = normalizeValue(raw[i])
|
||||
}
|
||||
b, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
return string(b), m, nil
|
||||
}
|
||||
|
||||
func normalizeValue(v any) any {
|
||||
switch x := v.(type) {
|
||||
case nil:
|
||||
return nil
|
||||
case []byte:
|
||||
return string(x)
|
||||
case time.Time:
|
||||
return x.UTC().Format(time.RFC3339Nano)
|
||||
default:
|
||||
return x
|
||||
}
|
||||
}
|
||||
|
||||
func GetMetaVersion(ctx context.Context, db *sql.DB, driver Driver, table, pk string) (int64, bool, error) {
|
||||
q := fmt.Sprintf(`SELECT version FROM %s WHERE table_name = ? AND row_pk = ?`, MetaTable)
|
||||
if driver == DriverPostgres {
|
||||
q = fmt.Sprintf(`SELECT version FROM %s WHERE table_name = $1 AND row_pk = $2`, MetaTable)
|
||||
}
|
||||
var ver int64
|
||||
err := db.QueryRowContext(ctx, q, table, pk).Scan(&ver)
|
||||
if err == sql.ErrNoRows {
|
||||
return 0, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
return ver, true, nil
|
||||
}
|
||||
|
||||
func UpsertMeta(ctx context.Context, db *sql.DB, driver Driver, table, pk string, version int64) error {
|
||||
now := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
_, err := db.ExecContext(ctx, `
|
||||
INSERT INTO `+MetaTable+`(table_name,row_pk,version,updated_at) VALUES(?,?,?,?)
|
||||
ON CONFLICT(table_name,row_pk) DO UPDATE SET version=excluded.version, updated_at=excluded.updated_at`,
|
||||
table, pk, version, now)
|
||||
return err
|
||||
case DriverMySQL:
|
||||
_, err := db.ExecContext(ctx, `
|
||||
INSERT INTO `+MetaTable+`(table_name,row_pk,version,updated_at) VALUES(?,?,?,?)
|
||||
ON DUPLICATE KEY UPDATE version=VALUES(version), updated_at=VALUES(updated_at)`,
|
||||
table, pk, version, now)
|
||||
return err
|
||||
case DriverPostgres:
|
||||
_, err := db.ExecContext(ctx, `
|
||||
INSERT INTO `+MetaTable+`(table_name,row_pk,version,updated_at) VALUES($1,$2,$3,$4)
|
||||
ON CONFLICT(table_name,row_pk) DO UPDATE SET version=EXCLUDED.version, updated_at=EXCLUDED.updated_at`,
|
||||
table, pk, version, time.Now().UTC())
|
||||
return err
|
||||
default:
|
||||
return fmt.Errorf("unsupported")
|
||||
}
|
||||
}
|
||||
|
||||
func ApplyChange(ctx context.Context, db *sql.DB, driver Driver, table, pkCol, op, payload string, version int64) error {
|
||||
return WithApplying(ctx, db, driver, func() error {
|
||||
return applyChangeInner(ctx, db, driver, table, pkCol, op, payload, version)
|
||||
})
|
||||
}
|
||||
|
||||
func applyChangeInner(ctx context.Context, db *sql.DB, driver Driver, table, pkCol, op, payload string, version int64) error {
|
||||
switch op {
|
||||
case "delete":
|
||||
var row map[string]any
|
||||
_ = json.Unmarshal([]byte(payload), &row)
|
||||
pkVal := ""
|
||||
if row != nil {
|
||||
if v, ok := row[pkCol]; ok {
|
||||
pkVal = fmt.Sprint(v)
|
||||
}
|
||||
}
|
||||
if pkVal == "" {
|
||||
return fmt.Errorf("delete missing pk")
|
||||
}
|
||||
q := fmt.Sprintf(`DELETE FROM %s WHERE %s = ?`, quoteIdent(driver, table), quoteIdent(driver, pkCol))
|
||||
if driver == DriverPostgres {
|
||||
q = fmt.Sprintf(`DELETE FROM %s WHERE %s = $1`, quoteIdent(driver, table), quoteIdent(driver, pkCol))
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, q, pkVal); err != nil {
|
||||
return err
|
||||
}
|
||||
return UpsertMeta(ctx, db, driver, table, pkVal, version)
|
||||
default: // upsert
|
||||
var row map[string]any
|
||||
if err := json.Unmarshal([]byte(payload), &row); err != nil {
|
||||
return err
|
||||
}
|
||||
pkVal := fmt.Sprint(row[pkCol])
|
||||
if pkVal == "" || pkVal == "<nil>" {
|
||||
return fmt.Errorf("upsert missing pk %s", pkCol)
|
||||
}
|
||||
cols := make([]string, 0, len(row))
|
||||
vals := make([]any, 0, len(row))
|
||||
for k, v := range row {
|
||||
cols = append(cols, k)
|
||||
vals = append(vals, v)
|
||||
}
|
||||
return upsertRow(ctx, db, driver, table, pkCol, cols, vals, version)
|
||||
}
|
||||
}
|
||||
|
||||
func upsertRow(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string, cols []string, vals []any, version int64) error {
|
||||
qcols := make([]string, len(cols))
|
||||
ph := make([]string, len(cols))
|
||||
for i, c := range cols {
|
||||
qcols[i] = quoteIdent(driver, c)
|
||||
if driver == DriverPostgres {
|
||||
ph[i] = fmt.Sprintf("$%d", i+1)
|
||||
} else {
|
||||
ph[i] = "?"
|
||||
}
|
||||
}
|
||||
pkVal := ""
|
||||
for i, c := range cols {
|
||||
if c == pkCol {
|
||||
pkVal = fmt.Sprint(vals[i])
|
||||
break
|
||||
}
|
||||
}
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
q := fmt.Sprintf(`INSERT INTO %s (%s) VALUES (%s) ON CONFLICT(%s) DO UPDATE SET %s`,
|
||||
quoteIdent(driver, table),
|
||||
strings.Join(qcols, ","),
|
||||
strings.Join(ph, ","),
|
||||
quoteIdent(driver, pkCol),
|
||||
sqliteSetClause(cols, pkCol, driver),
|
||||
)
|
||||
if _, err := db.ExecContext(ctx, q, vals...); err != nil {
|
||||
return err
|
||||
}
|
||||
case DriverMySQL:
|
||||
sets := make([]string, 0, len(cols))
|
||||
for _, c := range cols {
|
||||
if c == pkCol {
|
||||
continue
|
||||
}
|
||||
qi := quoteIdent(driver, c)
|
||||
sets = append(sets, qi+"=VALUES("+qi+")")
|
||||
}
|
||||
if len(sets) == 0 {
|
||||
sets = append(sets, quoteIdent(driver, pkCol)+"="+quoteIdent(driver, pkCol))
|
||||
}
|
||||
q := fmt.Sprintf(`INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s`,
|
||||
quoteIdent(driver, table), strings.Join(qcols, ","), strings.Join(ph, ","), strings.Join(sets, ","))
|
||||
if _, err := db.ExecContext(ctx, q, vals...); err != nil {
|
||||
return err
|
||||
}
|
||||
case DriverPostgres:
|
||||
sets := make([]string, 0, len(cols))
|
||||
for _, c := range cols {
|
||||
if c == pkCol {
|
||||
continue
|
||||
}
|
||||
qi := quoteIdent(driver, c)
|
||||
sets = append(sets, qi+"=EXCLUDED."+qi)
|
||||
}
|
||||
if len(sets) == 0 {
|
||||
sets = append(sets, quoteIdent(driver, pkCol)+"="+quoteIdent(driver, pkCol))
|
||||
}
|
||||
q := fmt.Sprintf(`INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO UPDATE SET %s`,
|
||||
quoteIdent(driver, table), strings.Join(qcols, ","), strings.Join(ph, ","),
|
||||
quoteIdent(driver, pkCol), strings.Join(sets, ","))
|
||||
if _, err := db.ExecContext(ctx, q, vals...); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return UpsertMeta(ctx, db, driver, table, pkVal, version)
|
||||
}
|
||||
|
||||
func sqliteSetClause(cols []string, pkCol string, driver Driver) string {
|
||||
parts := make([]string, 0, len(cols))
|
||||
for _, c := range cols {
|
||||
if c == pkCol {
|
||||
continue
|
||||
}
|
||||
qi := quoteIdent(driver, c)
|
||||
parts = append(parts, qi+"=excluded."+qi)
|
||||
}
|
||||
if len(parts) == 0 {
|
||||
return quoteIdent(driver, pkCol) + "=" + quoteIdent(driver, pkCol)
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
281
platform/internal/dbsync/dialect.go
Normal file
281
platform/internal/dbsync/dialect.go
Normal file
@@ -0,0 +1,281 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
_ "github.com/lib/pq"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
func Open(driver Driver, dsn string) (*sql.DB, error) {
|
||||
var name string
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
name = "sqlite"
|
||||
if dsn == "" {
|
||||
return nil, fmt.Errorf("sqlite dsn required")
|
||||
}
|
||||
// modernc.org/sqlite 注册名为 sqlite
|
||||
case DriverMySQL:
|
||||
name = "mysql"
|
||||
case DriverPostgres:
|
||||
name = "postgres"
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported driver: %s", driver)
|
||||
}
|
||||
db, err := sql.Open(name, dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
db.SetMaxOpenConns(5)
|
||||
db.SetMaxIdleConns(2)
|
||||
if err := db.Ping(); err != nil {
|
||||
_ = db.Close()
|
||||
return nil, err
|
||||
}
|
||||
return db, nil
|
||||
}
|
||||
|
||||
func quoteIdent(driver Driver, name string) string {
|
||||
name = strings.ReplaceAll(name, "`", "")
|
||||
name = strings.ReplaceAll(name, `"`, "")
|
||||
switch driver {
|
||||
case DriverMySQL:
|
||||
return "`" + name + "`"
|
||||
case DriverPostgres:
|
||||
return `"` + name + `"`
|
||||
default:
|
||||
return `"` + name + `"`
|
||||
}
|
||||
}
|
||||
|
||||
func EnsureOutbox(ctx context.Context, db *sql.DB, driver Driver) error {
|
||||
var ddl string
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + OutboxTable + ` (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
table_name TEXT NOT NULL,
|
||||
row_pk TEXT NOT NULL,
|
||||
op TEXT NOT NULL,
|
||||
payload TEXT NOT NULL DEFAULT '',
|
||||
version INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
synced_at TEXT
|
||||
);`
|
||||
case DriverMySQL:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + OutboxTable + ` (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
table_name VARCHAR(128) NOT NULL,
|
||||
row_pk VARCHAR(255) NOT NULL,
|
||||
op VARCHAR(16) NOT NULL,
|
||||
payload LONGTEXT NOT NULL,
|
||||
version BIGINT NOT NULL,
|
||||
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
synced_at DATETIME(3) NULL,
|
||||
INDEX idx_outbox_unsynced (synced_at, id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;`
|
||||
case DriverPostgres:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + OutboxTable + ` (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
table_name TEXT NOT NULL,
|
||||
row_pk TEXT NOT NULL,
|
||||
op TEXT NOT NULL,
|
||||
payload TEXT NOT NULL DEFAULT '',
|
||||
version BIGINT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
synced_at TIMESTAMPTZ
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_ajz_outbox_unsynced ON ` + OutboxTable + ` (synced_at, id);`
|
||||
default:
|
||||
return fmt.Errorf("unsupported driver")
|
||||
}
|
||||
_, err := db.ExecContext(ctx, ddl)
|
||||
return err
|
||||
}
|
||||
|
||||
func EnsureMeta(ctx context.Context, db *sql.DB, driver Driver) error {
|
||||
var ddl string
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + MetaTable + ` (
|
||||
table_name TEXT NOT NULL,
|
||||
row_pk TEXT NOT NULL,
|
||||
version INTEGER NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (table_name, row_pk)
|
||||
);`
|
||||
case DriverMySQL:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + MetaTable + ` (
|
||||
table_name VARCHAR(128) NOT NULL,
|
||||
row_pk VARCHAR(255) NOT NULL,
|
||||
version BIGINT NOT NULL,
|
||||
updated_at DATETIME(3) NOT NULL,
|
||||
PRIMARY KEY (table_name, row_pk)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;`
|
||||
case DriverPostgres:
|
||||
ddl = `CREATE TABLE IF NOT EXISTS ` + MetaTable + ` (
|
||||
table_name TEXT NOT NULL,
|
||||
row_pk TEXT NOT NULL,
|
||||
version BIGINT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL,
|
||||
PRIMARY KEY (table_name, row_pk)
|
||||
);`
|
||||
}
|
||||
_, err := db.ExecContext(ctx, ddl)
|
||||
return err
|
||||
}
|
||||
|
||||
// InstallTriggers：在业务表上挂触发器，写入 outbox（实时捕获变更）。
|
||||
func InstallTriggers(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string) error {
|
||||
table = strings.TrimSpace(table)
|
||||
pkCol = strings.TrimSpace(pkCol)
|
||||
if table == "" || pkCol == "" {
|
||||
return fmt.Errorf("table and pk required")
|
||||
}
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
return installSQLiteTriggers(ctx, db, table, pkCol)
|
||||
case DriverMySQL:
|
||||
return installMySQLTriggers(ctx, db, table, pkCol)
|
||||
case DriverPostgres:
|
||||
return installPostgresTriggers(ctx, db, table, pkCol)
|
||||
default:
|
||||
return fmt.Errorf("unsupported driver")
|
||||
}
|
||||
}
|
||||
|
||||
func installSQLiteTriggers(ctx context.Context, db *sql.DB, table, pk string) error {
|
||||
if err := EnsureFlagTable(ctx, db, DriverSQLite); err != nil {
|
||||
return err
|
||||
}
|
||||
guard := `(SELECT IFNULL((SELECT v FROM ` + FlagTable + ` WHERE k='applying'),0))=0`
|
||||
stmts := []string{
|
||||
fmt.Sprintf(`DROP TRIGGER IF EXISTS ajz_sync_ai_%s`, table),
|
||||
fmt.Sprintf(`DROP TRIGGER IF EXISTS ajz_sync_au_%s`, table),
|
||||
fmt.Sprintf(`DROP TRIGGER IF EXISTS ajz_sync_ad_%s`, table),
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_ai_%s AFTER INSERT ON "%s" WHEN %s BEGIN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version,created_at)
|
||||
VALUES('%s', CAST(NEW."%s" AS TEXT), 'upsert', '',
|
||||
CAST(strftime('%%s','now') AS INTEGER)*1000, datetime('now'));
|
||||
END;`, table, table, guard, OutboxTable, table, pk),
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_au_%s AFTER UPDATE ON "%s" WHEN %s BEGIN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version,created_at)
|
||||
VALUES('%s', CAST(NEW."%s" AS TEXT), 'upsert', '',
|
||||
CAST(strftime('%%s','now') AS INTEGER)*1000, datetime('now'));
|
||||
END;`, table, table, guard, OutboxTable, table, pk),
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_ad_%s AFTER DELETE ON "%s" WHEN %s BEGIN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version,created_at)
|
||||
VALUES('%s', CAST(OLD."%s" AS TEXT), 'delete', '',
|
||||
CAST(strftime('%%s','now') AS INTEGER)*1000, datetime('now'));
|
||||
END;`, table, table, guard, OutboxTable, table, pk),
|
||||
}
|
||||
for _, s := range stmts {
|
||||
if _, err := db.ExecContext(ctx, s); err != nil {
|
||||
return fmt.Errorf("sqlite trigger %s: %w", table, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func installMySQLTriggers(ctx context.Context, db *sql.DB, table, pk string) error {
|
||||
drops := []string{
|
||||
fmt.Sprintf("DROP TRIGGER IF EXISTS ajz_sync_ai_%s", table),
|
||||
fmt.Sprintf("DROP TRIGGER IF EXISTS ajz_sync_au_%s", table),
|
||||
fmt.Sprintf("DROP TRIGGER IF EXISTS ajz_sync_ad_%s", table),
|
||||
}
|
||||
for _, s := range drops {
|
||||
_, _ = db.ExecContext(ctx, s)
|
||||
}
|
||||
creates := []string{
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_ai_%s AFTER INSERT ON `+"`%s`"+` FOR EACH ROW
|
||||
BEGIN
|
||||
IF IFNULL(@ajz_applying,0)=0 THEN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version)
|
||||
VALUES('%s', CAST(NEW.`+"`%s`"+` AS CHAR), 'upsert', '', UNIX_TIMESTAMP(NOW(3))*1000);
|
||||
END IF;
|
||||
END`, table, table, OutboxTable, table, pk),
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_au_%s AFTER UPDATE ON `+"`%s`"+` FOR EACH ROW
|
||||
BEGIN
|
||||
IF IFNULL(@ajz_applying,0)=0 THEN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version)
|
||||
VALUES('%s', CAST(NEW.`+"`%s`"+` AS CHAR), 'upsert', '', UNIX_TIMESTAMP(NOW(3))*1000);
|
||||
END IF;
|
||||
END`, table, table, OutboxTable, table, pk),
|
||||
fmt.Sprintf(`CREATE TRIGGER ajz_sync_ad_%s AFTER DELETE ON `+"`%s`"+` FOR EACH ROW
|
||||
BEGIN
|
||||
IF IFNULL(@ajz_applying,0)=0 THEN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version)
|
||||
VALUES('%s', CAST(OLD.`+"`%s`"+` AS CHAR), 'delete', '', UNIX_TIMESTAMP(NOW(3))*1000);
|
||||
END IF;
|
||||
END`, table, table, OutboxTable, table, pk),
|
||||
}
|
||||
for _, s := range creates {
|
||||
if _, err := db.ExecContext(ctx, s); err != nil {
|
||||
return fmt.Errorf("mysql trigger %s: %w", table, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func installPostgresTriggers(ctx context.Context, db *sql.DB, table, pk string) error {
|
||||
fn := fmt.Sprintf("ajz_sync_fn_%s", table)
|
||||
_, err := db.ExecContext(ctx, fmt.Sprintf(`
|
||||
CREATE OR REPLACE FUNCTION %s() RETURNS trigger AS $$
|
||||
BEGIN
|
||||
IF COALESCE(current_setting('ajz.applying', true), '0') = '1' THEN
|
||||
IF TG_OP = 'DELETE' THEN RETURN OLD; ELSE RETURN NEW; END IF;
|
||||
END IF;
|
||||
IF TG_OP = 'DELETE' THEN
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version)
|
||||
VALUES('%s', OLD.%s::text, 'delete', '', (EXTRACT(EPOCH FROM clock_timestamp())*1000)::bigint);
|
||||
RETURN OLD;
|
||||
ELSE
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version)
|
||||
VALUES('%s', NEW.%s::text, 'upsert', '', (EXTRACT(EPOCH FROM clock_timestamp())*1000)::bigint);
|
||||
RETURN NEW;
|
||||
END IF;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;`, fn, OutboxTable, table, quoteIdent(DriverPostgres, pk), OutboxTable, table, quoteIdent(DriverPostgres, pk)))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, _ = db.ExecContext(ctx, fmt.Sprintf(`DROP TRIGGER IF EXISTS ajz_sync_trg_%s ON %s`, table, quoteIdent(DriverPostgres, table)))
|
||||
_, err = db.ExecContext(ctx, fmt.Sprintf(`
|
||||
CREATE TRIGGER ajz_sync_trg_%s
|
||||
AFTER INSERT OR UPDATE OR DELETE ON %s
|
||||
FOR EACH ROW EXECUTE PROCEDURE %s();`, table, quoteIdent(DriverPostgres, table), fn))
|
||||
return err
|
||||
}
|
||||
|
||||
func ListTables(ctx context.Context, db *sql.DB, driver Driver) ([]string, error) {
|
||||
var q string
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
q = `SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '_ajz_%' ORDER BY name`
|
||||
case DriverMySQL:
|
||||
q = `SELECT table_name FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name NOT LIKE '\_ajz\_%' ORDER BY table_name`
|
||||
case DriverPostgres:
|
||||
q = `SELECT tablename FROM pg_tables WHERE schemaname='public' AND tablename NOT LIKE '\_ajz\_%' ORDER BY tablename`
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported")
|
||||
}
|
||||
rows, err := db.QueryContext(ctx, q)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []string
|
||||
for rows.Next() {
|
||||
var n string
|
||||
if err := rows.Scan(&n); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
77
platform/internal/dbsync/echo.go
Normal file
77
platform/internal/dbsync/echo.go
Normal file
@@ -0,0 +1,77 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
const FlagTable = "_ajz_sync_flag"
|
||||
|
||||
// EnsureFlagTable：SQLite 用表标记「正在应用远端变更」，避免触发器回声。
|
||||
func EnsureFlagTable(ctx context.Context, db *sql.DB, driver Driver) error {
|
||||
if driver != DriverSQLite {
|
||||
return nil
|
||||
}
|
||||
_, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS `+FlagTable+` (
|
||||
k TEXT PRIMARY KEY,
|
||||
v INTEGER NOT NULL DEFAULT 0
|
||||
);`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, _ = db.ExecContext(ctx, `INSERT OR IGNORE INTO `+FlagTable+`(k,v) VALUES('applying',0)`)
|
||||
return nil
|
||||
}
|
||||
|
||||
// WithApplying 在写入目标库时置位，使触发器不写 outbox（防 A↔B 回声导致「多」）。
|
||||
func WithApplying(ctx context.Context, db *sql.DB, driver Driver, fn func() error) error {
|
||||
switch driver {
|
||||
case DriverSQLite:
|
||||
if err := EnsureFlagTable(ctx, db, driver); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, `UPDATE `+FlagTable+` SET v=1 WHERE k='applying'`); err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_, _ = db.ExecContext(context.Background(), `UPDATE `+FlagTable+` SET v=0 WHERE k='applying'`)
|
||||
}()
|
||||
return fn()
|
||||
case DriverMySQL:
|
||||
if _, err := db.ExecContext(ctx, `SET @ajz_applying = 1`); err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _, _ = db.ExecContext(context.Background(), `SET @ajz_applying = 0`) }()
|
||||
return fn()
|
||||
case DriverPostgres:
|
||||
if _, err := db.ExecContext(ctx, `SELECT set_config('ajz.applying', '1', false)`); err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_, _ = db.ExecContext(context.Background(), `SELECT set_config('ajz.applying', '0', false)`)
|
||||
}()
|
||||
return fn()
|
||||
default:
|
||||
return fn()
|
||||
}
|
||||
}
|
||||
|
||||
func EnqueueOutbox(ctx context.Context, db *sql.DB, driver Driver, table, pk, op, payload string, version int64) error {
|
||||
if version <= 0 {
|
||||
version = time.Now().UnixMilli()
|
||||
}
|
||||
switch driver {
|
||||
case DriverPostgres:
|
||||
_, err := db.ExecContext(ctx, fmt.Sprintf(`
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version) VALUES($1,$2,$3,$4,$5)`, OutboxTable),
|
||||
table, pk, op, payload, version)
|
||||
return err
|
||||
default:
|
||||
_, err := db.ExecContext(ctx, fmt.Sprintf(`
|
||||
INSERT INTO %s(table_name,row_pk,op,payload,version) VALUES(?,?,?,?,?)`, OutboxTable),
|
||||
table, pk, op, payload, version)
|
||||
return err
|
||||
}
|
||||
}
|
||||
306
platform/internal/dbsync/manager.go
Normal file
306
platform/internal/dbsync/manager.go
Normal file
@@ -0,0 +1,306 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Manager struct {
|
||||
store *FileStore
|
||||
mu sync.Mutex
|
||||
runners map[string]context.CancelFunc
|
||||
}
|
||||
|
||||
func NewManager(store *FileStore) *Manager {
|
||||
return &Manager{store: store, runners: map[string]context.CancelFunc{}}
|
||||
}
|
||||
|
||||
func (m *Manager) Store() *FileStore { return m.store }
|
||||
|
||||
func (m *Manager) StartAll(ctx context.Context) {
|
||||
list, err := m.store.ListChannels()
|
||||
if err != nil {
|
||||
log.Printf("dbsync list: %v", err)
|
||||
return
|
||||
}
|
||||
for _, ch := range list {
|
||||
if ch.Enabled {
|
||||
_ = m.StartChannel(ch.ID)
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
m.StopAll()
|
||||
}()
|
||||
}
|
||||
|
||||
func (m *Manager) StartChannel(id string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if _, ok := m.runners[id]; ok {
|
||||
return nil
|
||||
}
|
||||
ch, err := m.store.GetChannel(id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(context.Background())
|
||||
m.runners[id] = cancel
|
||||
go m.loop(runCtx, ch.ID)
|
||||
_ = m.store.PatchStats(id, func(c *Channel) {
|
||||
c.Enabled = true
|
||||
c.LastError = ""
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) StopChannel(id string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if cancel, ok := m.runners[id]; ok {
|
||||
cancel()
|
||||
delete(m.runners, id)
|
||||
}
|
||||
_ = m.store.PatchStats(id, func(c *Channel) { c.Enabled = false })
|
||||
}
|
||||
|
||||
func (m *Manager) StopAll() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for id, cancel := range m.runners {
|
||||
cancel()
|
||||
delete(m.runners, id)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) loop(ctx context.Context, id string) {
|
||||
ticks := 0
|
||||
for {
|
||||
ch, err := m.store.GetChannel(id)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
iv := time.Duration(ch.PollIntervalMS) * time.Millisecond
|
||||
if iv < 100*time.Millisecond {
|
||||
iv = 500 * time.Millisecond
|
||||
}
|
||||
if err := m.tick(ctx, ch); err != nil {
|
||||
_ = m.store.PatchStats(id, func(c *Channel) { c.LastError = err.Error() })
|
||||
log.Printf("dbsync channel %s: %v", id, err)
|
||||
}
|
||||
ticks++
|
||||
// 约每分钟主键对账一次，补漏（漏投递 / 触发器未装时的存量差）
|
||||
if ticks%120 == 0 && ch.Direction == DirBidirectional {
|
||||
if _, rerr := ReconcileChannel(ctx, ch); rerr != nil {
|
||||
log.Printf("dbsync reconcile %s: %v", id, rerr)
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(iv):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) tick(ctx context.Context, ch *Channel) error {
|
||||
local, err := Open(ch.Local.Driver, ch.Local.DSN)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open local: %w", err)
|
||||
}
|
||||
defer local.Close()
|
||||
remote, err := Open(ch.Remote.Driver, ch.Remote.DSN)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open remote: %w", err)
|
||||
}
|
||||
defer remote.Close()
|
||||
|
||||
if err := prepareEndpoint(ctx, local, ch.Local, ch.PKColumns); err != nil {
|
||||
return fmt.Errorf("prepare local: %w", err)
|
||||
}
|
||||
if err := prepareEndpoint(ctx, remote, ch.Remote, ch.PKColumns); err != nil {
|
||||
return fmt.Errorf("prepare remote: %w", err)
|
||||
}
|
||||
|
||||
var n int
|
||||
switch ch.Direction {
|
||||
case DirRemoteToLocal:
|
||||
n, err = m.drain(ctx, ch, "remote", remote, ch.Remote, local, ch.Local)
|
||||
case DirBidirectional:
|
||||
n1, e1 := m.drain(ctx, ch, "local", local, ch.Local, remote, ch.Remote)
|
||||
n2, e2 := m.drain(ctx, ch, "remote", remote, ch.Remote, local, ch.Local)
|
||||
n = n1 + n2
|
||||
if e1 != nil {
|
||||
err = e1
|
||||
} else {
|
||||
err = e2
|
||||
}
|
||||
default: // local_to_remote
|
||||
n, err = m.drain(ctx, ch, "local", local, ch.Local, remote, ch.Remote)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
_ = m.store.PatchStats(ch.ID, func(c *Channel) {
|
||||
c.LastSyncAt = &now
|
||||
c.Stats.LastBatch = n
|
||||
if err != nil {
|
||||
c.LastError = err.Error()
|
||||
} else {
|
||||
c.LastError = ""
|
||||
}
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func prepareEndpoint(ctx context.Context, db *sql.DB, ep Endpoint, pks map[string]string) error {
|
||||
if err := EnsureOutbox(ctx, db, ep.Driver); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := EnsureMeta(ctx, db, ep.Driver); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, t := range ep.Tables {
|
||||
pk := "id"
|
||||
if pks != nil && pks[t] != "" {
|
||||
pk = pks[t]
|
||||
}
|
||||
if err := InstallTriggers(ctx, db, ep.Driver, t, pk); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) drain(ctx context.Context, ch *Channel, sourceName string, src *sql.DB, srcEp Endpoint, dst *sql.DB, dstEp Endpoint) (int, error) {
|
||||
rows, err := PollOutbox(ctx, src, srcEp.Driver, 100)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
var done []int64
|
||||
okCount := 0
|
||||
for _, r := range rows {
|
||||
pkCol := ch.PKColumns[r.TableName]
|
||||
if pkCol == "" {
|
||||
pkCol = "id"
|
||||
}
|
||||
payload := r.Payload
|
||||
if r.Op != "delete" && (payload == "" || payload == "{}") {
|
||||
js, _, ferr := FetchRowJSON(ctx, src, srcEp.Driver, r.TableName, pkCol, r.RowPK)
|
||||
if ferr == sql.ErrNoRows {
|
||||
// 行已删，改 delete
|
||||
r.Op = "delete"
|
||||
payload = fmt.Sprintf(`{%q:%q}`, pkCol, r.RowPK)
|
||||
} else if ferr != nil {
|
||||
_ = m.store.PatchStats(ch.ID, func(c *Channel) { c.Stats.Retries++ })
|
||||
continue
|
||||
} else {
|
||||
payload = js
|
||||
}
|
||||
}
|
||||
if r.Op == "delete" && payload == "" {
|
||||
payload = fmt.Sprintf(`{%q:%q}`, pkCol, r.RowPK)
|
||||
}
|
||||
|
||||
tgtVer, has, _ := GetMetaVersion(ctx, dst, dstEp.Driver, r.TableName, r.RowPK)
|
||||
// 幂等：已同步过相同版本 → 跳过（防「多」）
|
||||
if has && tgtVer == r.Version {
|
||||
done = append(done, r.ID)
|
||||
continue
|
||||
}
|
||||
if has && tgtVer > r.Version {
|
||||
switch ch.ConflictPolicy {
|
||||
case PolicyLWWTarget:
|
||||
done = append(done, r.ID)
|
||||
continue
|
||||
case PolicyLWWSource:
|
||||
// fallthrough apply
|
||||
default:
|
||||
_ = m.store.AddConflict(Conflict{
|
||||
TenantID: ch.TenantID,
|
||||
ChannelID: ch.ID,
|
||||
Table: r.TableName,
|
||||
RowPK: r.RowPK,
|
||||
Op: r.Op,
|
||||
Source: sourceName,
|
||||
Payload: payload,
|
||||
TargetVer: tgtVer,
|
||||
SourceVer: r.Version,
|
||||
Message: "target version newer than source",
|
||||
})
|
||||
_ = m.store.PatchStats(ch.ID, func(c *Channel) { c.Stats.Conflicts++ })
|
||||
done = append(done, r.ID)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
if err := ApplyChange(ctx, dst, dstEp.Driver, r.TableName, pkCol, r.Op, payload, r.Version); err != nil {
|
||||
_ = m.store.PatchStats(ch.ID, func(c *Channel) {
|
||||
c.Stats.Retries++
|
||||
c.LastError = err.Error()
|
||||
})
|
||||
continue
|
||||
}
|
||||
done = append(done, r.ID)
|
||||
okCount++
|
||||
_ = m.store.PatchStats(ch.ID, func(c *Channel) {
|
||||
if sourceName == "local" {
|
||||
c.Stats.PushedOK++
|
||||
} else {
|
||||
c.Stats.PulledOK++
|
||||
}
|
||||
})
|
||||
}
|
||||
if err := MarkSynced(ctx, src, srcEp.Driver, done); err != nil {
|
||||
return okCount, err
|
||||
}
|
||||
return okCount, nil
|
||||
}
|
||||
|
||||
// PrepareChannel 连接两端、建 outbox/触发器，供「测试/启用」调用。
|
||||
func PrepareChannel(ctx context.Context, ch *Channel) error {
|
||||
for _, ep := range []Endpoint{ch.Local, ch.Remote} {
|
||||
db, err := Open(ep.Driver, ep.DSN)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := EnsureOutbox(ctx, db, ep.Driver); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
if err := EnsureMeta(ctx, db, ep.Driver); err != nil {
|
||||
_ = db.Close()
|
||||
return err
|
||||
}
|
||||
for _, t := range ep.Tables {
|
||||
pk := "id"
|
||||
if ch.PKColumns != nil && ch.PKColumns[t] != "" {
|
||||
pk = ch.PKColumns[t]
|
||||
}
|
||||
if err := InstallTriggers(ctx, db, ep.Driver, t, pk); err != nil {
|
||||
_ = db.Close()
|
||||
return fmt.Errorf("%s.%s triggers: %w", ep.Driver, t, err)
|
||||
}
|
||||
}
|
||||
_ = db.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestEndpoint(ctx context.Context, ep Endpoint) TestResult {
|
||||
db, err := Open(ep.Driver, ep.DSN)
|
||||
if err != nil {
|
||||
return TestResult{OK: false, Driver: string(ep.Driver), Message: err.Error()}
|
||||
}
|
||||
defer db.Close()
|
||||
tables, err := ListTables(ctx, db, ep.Driver)
|
||||
if err != nil {
|
||||
return TestResult{OK: false, Driver: string(ep.Driver), Message: err.Error()}
|
||||
}
|
||||
return TestResult{OK: true, Driver: string(ep.Driver), Message: "connected", Tables: tables}
|
||||
}
|
||||
182
platform/internal/dbsync/reconcile.go
Normal file
182
platform/internal/dbsync/reconcile.go
Normal file
@@ -0,0 +1,182 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ReconcileReport struct {
|
||||
Table string `json:"table"`
|
||||
OnlyLocal []string `json:"only_local"` // 在 B 有、A 无 → 需推到线上
|
||||
OnlyRemote []string `json:"only_remote"` // 在 A 有、B 无 → 需拉到本地
|
||||
PatchedPush int `json:"patched_push"`
|
||||
PatchedPull int `json:"patched_pull"`
|
||||
}
|
||||
|
||||
type ReconcileResult struct {
|
||||
ChannelID string `json:"channel_id"`
|
||||
Reports []ReconcileReport `json:"reports"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
// ListPKs 列出表全部主键（用于对账，防漏）。
|
||||
func ListPKs(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string) ([]string, error) {
|
||||
q := fmt.Sprintf(`SELECT %s FROM %s`, quoteIdent(driver, pkCol), quoteIdent(driver, table))
|
||||
rows, err := db.QueryContext(ctx, q)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []string
|
||||
for rows.Next() {
|
||||
var v any
|
||||
if err := rows.Scan(&v); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, fmt.Sprint(v))
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ReconcileChannel 全量主键对账：只在一端的行补投递，保证不漏；已有行靠 outbox/版本，不重复插入。
|
||||
func ReconcileChannel(ctx context.Context, ch *Channel) (*ReconcileResult, error) {
|
||||
local, err := Open(ch.Local.Driver, ch.Local.DSN)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open local: %w", err)
|
||||
}
|
||||
defer local.Close()
|
||||
remote, err := Open(ch.Remote.Driver, ch.Remote.DSN)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open remote: %w", err)
|
||||
}
|
||||
defer remote.Close()
|
||||
|
||||
if err := prepareEndpoint(ctx, local, ch.Local, ch.PKColumns); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := prepareEndpoint(ctx, remote, ch.Remote, ch.PKColumns); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tables := uniqueTables(ch.Local.Tables, ch.Remote.Tables)
|
||||
res := &ReconcileResult{ChannelID: ch.ID, Message: "ok"}
|
||||
for _, table := range tables {
|
||||
pkCol := "id"
|
||||
if ch.PKColumns != nil && ch.PKColumns[table] != "" {
|
||||
pkCol = ch.PKColumns[table]
|
||||
}
|
||||
lpks, err := ListPKs(ctx, local, ch.Local.Driver, table, pkCol)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list local %s: %w", table, err)
|
||||
}
|
||||
rpks, err := ListPKs(ctx, remote, ch.Remote.Driver, table, pkCol)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list remote %s: %w", table, err)
|
||||
}
|
||||
lset := toSet(lpks)
|
||||
rset := toSet(rpks)
|
||||
rep := ReconcileReport{Table: table}
|
||||
for pk := range lset {
|
||||
if !rset[pk] {
|
||||
rep.OnlyLocal = append(rep.OnlyLocal, pk)
|
||||
}
|
||||
}
|
||||
for pk := range rset {
|
||||
if !lset[pk] {
|
||||
rep.OnlyRemote = append(rep.OnlyRemote, pk)
|
||||
}
|
||||
}
|
||||
// 补漏：缺的一端从有的一端取行并 apply（带 WithApplying，不产生回声）
|
||||
for _, pk := range rep.OnlyLocal {
|
||||
js, _, ferr := FetchRowJSON(ctx, local, ch.Local.Driver, table, pkCol, pk)
|
||||
if ferr != nil {
|
||||
continue
|
||||
}
|
||||
ver := time.Now().UnixMilli()
|
||||
if err := ApplyChange(ctx, remote, ch.Remote.Driver, table, pkCol, "upsert", js, ver); err == nil {
|
||||
rep.PatchedPush++
|
||||
_ = UpsertMeta(ctx, local, ch.Local.Driver, table, pk, ver)
|
||||
}
|
||||
}
|
||||
for _, pk := range rep.OnlyRemote {
|
||||
js, _, ferr := FetchRowJSON(ctx, remote, ch.Remote.Driver, table, pkCol, pk)
|
||||
if ferr != nil {
|
||||
continue
|
||||
}
|
||||
ver := time.Now().UnixMilli()
|
||||
if err := ApplyChange(ctx, local, ch.Local.Driver, table, pkCol, "upsert", js, ver); err == nil {
|
||||
rep.PatchedPull++
|
||||
_ = UpsertMeta(ctx, remote, ch.Remote.Driver, table, pk, ver)
|
||||
}
|
||||
}
|
||||
res.Reports = append(res.Reports, rep)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func uniqueTables(a, b []string) []string {
|
||||
m := map[string]struct{}{}
|
||||
var out []string
|
||||
for _, t := range append(append([]string{}, a...), b...) {
|
||||
if t == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := m[t]; ok {
|
||||
continue
|
||||
}
|
||||
m[t] = struct{}{}
|
||||
out = append(out, t)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func toSet(xs []string) map[string]bool {
|
||||
m := make(map[string]bool, len(xs))
|
||||
for _, x := range xs {
|
||||
m[x] = true
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// IngestRows：外部源 C → 写入本地 B（会触发 outbox，再同步到线上 A）。
|
||||
// 按主键 upsert，同一 PK 重复灌入不会「多」出一行。
|
||||
func IngestRows(ctx context.Context, ch *Channel, table string, rows []map[string]any, sourceLabel string) (int, error) {
|
||||
if table == "" {
|
||||
return 0, fmt.Errorf("table required")
|
||||
}
|
||||
pkCol := "id"
|
||||
if ch.PKColumns != nil && ch.PKColumns[table] != "" {
|
||||
pkCol = ch.PKColumns[table]
|
||||
}
|
||||
db, err := Open(ch.Local.Driver, ch.Local.DSN)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer db.Close()
|
||||
if err := prepareEndpoint(ctx, db, ch.Local, ch.PKColumns); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n := 0
|
||||
for _, row := range rows {
|
||||
if row == nil {
|
||||
continue
|
||||
}
|
||||
pkVal := fmt.Sprint(row[pkCol])
|
||||
if pkVal == "" || pkVal == "<nil>" {
|
||||
return n, fmt.Errorf("row missing pk %s", pkCol)
|
||||
}
|
||||
// 不置 applying：让触发器写 outbox，随后 worker 推到 A
|
||||
b, _ := json.Marshal(row)
|
||||
ver := time.Now().UnixMilli()
|
||||
// 正常写入本地：触发器入 outbox → worker 推到线上 A；同 PK upsert 不会多行
|
||||
if err := applyChangeInner(ctx, db, ch.Local.Driver, table, pkCol, "upsert", string(b), ver); err != nil {
|
||||
return n, err
|
||||
}
|
||||
_ = sourceLabel
|
||||
n++
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
327
platform/internal/dbsync/store.go
Normal file
327
platform/internal/dbsync/store.go
Normal file
@@ -0,0 +1,327 @@
|
||||
package dbsync
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// FileStore 持久化通道与冲突队列（JSON），不依赖业务库类型。
|
||||
type FileStore struct {
|
||||
mu sync.Mutex
|
||||
dir string
|
||||
chPath string
|
||||
cfPath string
|
||||
}
|
||||
|
||||
func NewFileStore(dir string) (*FileStore, error) {
|
||||
if dir == "" {
|
||||
dir = "./data/dbsync"
|
||||
}
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &FileStore{
|
||||
dir: dir,
|
||||
chPath: filepath.Join(dir, "channels.json"),
|
||||
cfPath: filepath.Join(dir, "conflicts.json"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) ListChannels() ([]Channel, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.readChannels()
|
||||
}
|
||||
|
||||
func (s *FileStore) ListChannelsByTenant(tenantID int64) ([]Channel, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readChannels()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Channel, 0)
|
||||
for _, c := range list {
|
||||
if c.TenantID == tenantID {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) GetChannel(id string) (*Channel, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readChannels()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for i := range list {
|
||||
if list[i].ID == id {
|
||||
cp := list[i]
|
||||
return &cp, nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("channel not found")
|
||||
}
|
||||
|
||||
// GetChannelForTenant 仅返回属于该租户的通道。
|
||||
func (s *FileStore) GetChannelForTenant(id string, tenantID int64) (*Channel, error) {
|
||||
ch, err := s.GetChannel(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ch.TenantID != tenantID {
|
||||
return nil, fmt.Errorf("channel not found")
|
||||
}
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) SaveChannel(ch Channel) (Channel, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readChannels()
|
||||
if err != nil {
|
||||
return ch, err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if ch.ID == "" {
|
||||
ch.ID = uuid.NewString()
|
||||
ch.CreatedAt = now
|
||||
}
|
||||
ch.UpdatedAt = now
|
||||
if ch.PollIntervalMS <= 0 {
|
||||
ch.PollIntervalMS = 500
|
||||
}
|
||||
if ch.Direction == "" {
|
||||
ch.Direction = DirLocalToRemote
|
||||
}
|
||||
if ch.ConflictPolicy == "" {
|
||||
ch.ConflictPolicy = PolicyQueue
|
||||
}
|
||||
if ch.PKColumns == nil {
|
||||
ch.PKColumns = map[string]string{}
|
||||
}
|
||||
found := false
|
||||
for i := range list {
|
||||
if list[i].ID == ch.ID {
|
||||
// 禁止跨租户覆盖；更新时锁定原 tenant_id
|
||||
if list[i].TenantID != 0 && ch.TenantID != 0 && list[i].TenantID != ch.TenantID {
|
||||
return ch, fmt.Errorf("channel belongs to another tenant")
|
||||
}
|
||||
if ch.TenantID == 0 {
|
||||
ch.TenantID = list[i].TenantID
|
||||
}
|
||||
ch.CreatedAt = list[i].CreatedAt
|
||||
ch.Stats = list[i].Stats
|
||||
list[i] = ch
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
if ch.TenantID <= 0 {
|
||||
return ch, fmt.Errorf("tenant_id required")
|
||||
}
|
||||
list = append(list, ch)
|
||||
}
|
||||
if err := s.writeChannels(list); err != nil {
|
||||
return ch, err
|
||||
}
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) DeleteChannel(id string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readChannels()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
next := list[:0]
|
||||
for _, c := range list {
|
||||
if c.ID != id {
|
||||
next = append(next, c)
|
||||
}
|
||||
}
|
||||
return s.writeChannels(next)
|
||||
}
|
||||
|
||||
func (s *FileStore) DeleteChannelForTenant(id string, tenantID int64) error {
|
||||
ch, err := s.GetChannelForTenant(id, tenantID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_ = ch
|
||||
return s.DeleteChannel(id)
|
||||
}
|
||||
|
||||
func (s *FileStore) PatchStats(id string, fn func(*Channel)) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readChannels()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range list {
|
||||
if list[i].ID == id {
|
||||
fn(&list[i])
|
||||
list[i].UpdatedAt = time.Now().UTC()
|
||||
return s.writeChannels(list)
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("channel not found")
|
||||
}
|
||||
|
||||
func (s *FileStore) AddConflict(c Conflict) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readConflicts()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if c.ID == "" {
|
||||
c.ID = uuid.NewString()
|
||||
}
|
||||
if c.CreatedAt.IsZero() {
|
||||
c.CreatedAt = time.Now().UTC()
|
||||
}
|
||||
list = append(list, c)
|
||||
return s.writeConflicts(list)
|
||||
}
|
||||
|
||||
func (s *FileStore) ListConflicts(unresolvedOnly bool) ([]Conflict, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readConflicts()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !unresolvedOnly {
|
||||
return list, nil
|
||||
}
|
||||
out := make([]Conflict, 0)
|
||||
for _, c := range list {
|
||||
if !c.Resolved {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) ListConflictsByTenant(tenantID int64, unresolvedOnly bool) ([]Conflict, error) {
|
||||
list, err := s.ListConflicts(unresolvedOnly)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Conflict, 0)
|
||||
for _, c := range list {
|
||||
if c.TenantID == tenantID {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) ResolveConflict(id, resolution string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readConflicts()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range list {
|
||||
if list[i].ID == id {
|
||||
list[i].Resolved = true
|
||||
list[i].Resolution = resolution
|
||||
return s.writeConflicts(list)
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("conflict not found")
|
||||
}
|
||||
|
||||
func (s *FileStore) ResolveConflictForTenant(id string, tenantID int64, resolution string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
list, err := s.readConflicts()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range list {
|
||||
if list[i].ID == id {
|
||||
if list[i].TenantID != tenantID {
|
||||
return fmt.Errorf("conflict not found")
|
||||
}
|
||||
list[i].Resolved = true
|
||||
list[i].Resolution = resolution
|
||||
return s.writeConflicts(list)
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("conflict not found")
|
||||
}
|
||||
|
||||
func (s *FileStore) readChannels() ([]Channel, error) {
|
||||
b, err := os.ReadFile(s.chPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return []Channel{}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var list []Channel
|
||||
if len(b) == 0 {
|
||||
return []Channel{}, nil
|
||||
}
|
||||
if err := json.Unmarshal(b, &list); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return list, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) writeChannels(list []Channel) error {
|
||||
b, err := json.MarshalIndent(list, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := s.chPath + ".tmp"
|
||||
if err := os.WriteFile(tmp, b, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, s.chPath)
|
||||
}
|
||||
|
||||
func (s *FileStore) readConflicts() ([]Conflict, error) {
|
||||
b, err := os.ReadFile(s.cfPath)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return []Conflict{}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var list []Conflict
|
||||
if len(b) == 0 {
|
||||
return []Conflict{}, nil
|
||||
}
|
||||
if err := json.Unmarshal(b, &list); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return list, nil
|
||||
}
|
||||
|
||||
func (s *FileStore) writeConflicts(list []Conflict) error {
|
||||
b, err := json.MarshalIndent(list, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := s.cfPath + ".tmp"
|
||||
if err := os.WriteFile(tmp, b, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, s.cfPath)
|
||||
}
|
||||
98
platform/internal/dbsync/types.go
Normal file
98
platform/internal/dbsync/types.go
Normal file
@@ -0,0 +1,98 @@
|
||||
// Package dbsync：跨库实时同步中间件（SQLite / MySQL / Postgres）。
|
||||
// 变更写入各端 _ajz_sync_outbox，由 worker 轮询投递；冲突进入队列。
|
||||
package dbsync
|
||||
|
||||
import "time"
|
||||
|
||||
const OutboxTable = "_ajz_sync_outbox"
|
||||
const MetaTable = "_ajz_sync_meta"
|
||||
|
||||
type Driver string
|
||||
|
||||
const (
|
||||
DriverSQLite Driver = "sqlite"
|
||||
DriverMySQL Driver = "mysql"
|
||||
DriverPostgres Driver = "postgres"
|
||||
)
|
||||
|
||||
type Direction string
|
||||
|
||||
const (
|
||||
DirLocalToRemote Direction = "local_to_remote"
|
||||
DirRemoteToLocal Direction = "remote_to_local"
|
||||
DirBidirectional Direction = "bidirectional"
|
||||
)
|
||||
|
||||
type ConflictPolicy string
|
||||
|
||||
const (
|
||||
PolicyQueue ConflictPolicy = "queue" // 入冲突队列，人工处理
|
||||
PolicyLWWSource ConflictPolicy = "lww_source" // 以源端为准覆盖
|
||||
PolicyLWWTarget ConflictPolicy = "lww_target" // 保留目标端，丢弃源变更
|
||||
)
|
||||
|
||||
type Endpoint struct {
|
||||
Driver Driver `json:"driver"` // sqlite | mysql | postgres
|
||||
DSN string `json:"dsn"` // 连接串；后台可改线上地址
|
||||
Tables []string `json:"tables"` // 要同步的表名
|
||||
}
|
||||
|
||||
type Channel struct {
|
||||
ID string `json:"id"`
|
||||
TenantID int64 `json:"tenant_id"` // 公司/租户隔离；仅该公司顶级权限可配
|
||||
Name string `json:"name"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Direction Direction `json:"direction"`
|
||||
ConflictPolicy ConflictPolicy `json:"conflict_policy"`
|
||||
PollIntervalMS int `json:"poll_interval_ms"`
|
||||
Local Endpoint `json:"local"`
|
||||
Remote Endpoint `json:"remote"`
|
||||
PKColumns map[string]string `json:"pk_columns"` // table -> pk col，默认 id
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
LastSyncAt *time.Time `json:"last_sync_at,omitempty"`
|
||||
Stats ChannelStats `json:"stats"`
|
||||
}
|
||||
|
||||
type ChannelStats struct {
|
||||
PushedOK int64 `json:"pushed_ok"`
|
||||
PulledOK int64 `json:"pulled_ok"`
|
||||
Conflicts int64 `json:"conflicts"`
|
||||
Retries int64 `json:"retries"`
|
||||
LastBatch int `json:"last_batch"`
|
||||
}
|
||||
|
||||
type Conflict struct {
|
||||
ID string `json:"id"`
|
||||
TenantID int64 `json:"tenant_id"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
Table string `json:"table"`
|
||||
RowPK string `json:"row_pk"`
|
||||
Op string `json:"op"`
|
||||
Source string `json:"source"` // local | remote
|
||||
Payload string `json:"payload"`
|
||||
TargetVer int64 `json:"target_ver"`
|
||||
SourceVer int64 `json:"source_ver"`
|
||||
Message string `json:"message"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Resolved bool `json:"resolved"`
|
||||
Resolution string `json:"resolution,omitempty"` // apply_source | keep_target | discard
|
||||
}
|
||||
|
||||
type OutboxRow struct {
|
||||
ID int64
|
||||
TableName string
|
||||
RowPK string
|
||||
Op string // upsert | delete
|
||||
Payload string
|
||||
Version int64
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
type TestResult struct {
|
||||
OK bool `json:"ok"`
|
||||
Driver string `json:"driver"`
|
||||
Message string `json:"message"`
|
||||
Tables []string `json:"tables,omitempty"`
|
||||
}
|
||||
Reference in New Issue
Block a user