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>
274 lines
7.5 KiB
Go
274 lines
7.5 KiB
Go
package dbsync
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// PushItem 本机 agent → 线上 A 的一条变更（形态 B）。
|
||
type PushItem struct {
|
||
Table string `json:"table"`
|
||
Op string `json:"op"` // insert|update|update_by_id|delete|upsert
|
||
RowPK string `json:"row_pk"`
|
||
Row map[string]any `json:"row"` // 单行；与 Rows 二选一
|
||
Rows []map[string]any `json:"rows"` // 兼容批量 insert
|
||
Version int64 `json:"version"` // outbox 单调 version；0 则用时间戳
|
||
ClientOutboxID string `json:"client_outbox_id"`
|
||
OnlineDBID string `json:"online_db_id"`
|
||
}
|
||
|
||
// PushResult 单条推送结果。
|
||
type PushResult struct {
|
||
OK bool `json:"ok"`
|
||
Applied bool `json:"applied"`
|
||
Skipped bool `json:"skipped"`
|
||
Conflict bool `json:"conflict"`
|
||
AppliedVersion int64 `json:"applied_version"`
|
||
Message string `json:"message"`
|
||
ClientOutboxID string `json:"client_outbox_id,omitempty"`
|
||
}
|
||
|
||
// PushToRemote 仅打开 remote，按 LWW/幂等将变更落到线上 A。不连本机 SQLite。
|
||
func PushToRemote(ctx context.Context, ch *Channel, store *FileStore, item PushItem) (*PushResult, error) {
|
||
if ch == nil {
|
||
return nil, fmt.Errorf("channel is nil")
|
||
}
|
||
table := strings.TrimSpace(item.Table)
|
||
if table == "" {
|
||
return nil, fmt.Errorf("table required")
|
||
}
|
||
// Z4 冻结：通道表白名单不再作为拒收依据；Binding + JWT 才是权限源。
|
||
pkCol := "id"
|
||
if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" {
|
||
pkCol = ch.PKColumns[table]
|
||
}
|
||
|
||
op, payload, rowPK, err := normalizePushPayload(item, pkCol)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
version := item.Version
|
||
if version <= 0 {
|
||
version = time.Now().UnixNano()
|
||
}
|
||
|
||
db, err := AcquireRemote(ch.Remote.Driver, ch.Remote.DSN)
|
||
if err != nil {
|
||
return nil, wrapOpenRemote(err)
|
||
}
|
||
|
||
if err := EnsureMeta(ctx, db, ch.Remote.Driver); err != nil {
|
||
return nil, Retryablef("ensure meta: %v", err)
|
||
}
|
||
|
||
if op != "delete" {
|
||
var row map[string]any
|
||
_ = json.Unmarshal([]byte(payload), &row)
|
||
if err := EnsureTableFromRow(ctx, db, ch.Remote.Driver, table, pkCol, row); err != nil {
|
||
return nil, Retryablef("ensure table: %v", err)
|
||
}
|
||
}
|
||
|
||
res := &PushResult{
|
||
ClientOutboxID: item.ClientOutboxID,
|
||
AppliedVersion: version,
|
||
}
|
||
|
||
tgtVer, has, err := GetMetaVersion(ctx, db, ch.Remote.Driver, table, rowPK)
|
||
if err != nil {
|
||
return nil, Retryablef("meta version: %v", err)
|
||
}
|
||
// 幂等：同 version 已落地 → 跳过
|
||
if has && tgtVer == version {
|
||
res.OK = true
|
||
res.Skipped = true
|
||
res.Message = "already applied (same version)"
|
||
if store != nil {
|
||
_ = store.PatchStats(ch.ID, func(c *Channel) { c.Stats.PushedSkipped++ })
|
||
}
|
||
return res, nil
|
||
}
|
||
if has && tgtVer > version {
|
||
policy := ch.ConflictPolicy
|
||
if policy == "" {
|
||
policy = PolicyLWWSource // B→A 默认偏源
|
||
}
|
||
switch policy {
|
||
case PolicyLWWTarget:
|
||
loser := payload
|
||
winner := SnapshotTargetRow(ctx, db, ch.Remote.Driver, table, pkCol, rowPK)
|
||
RecordLwwOverride(store, LwwOverride{
|
||
TenantID: ch.TenantID,
|
||
ChannelID: ch.ID,
|
||
Table: table,
|
||
RowPK: rowPK,
|
||
Op: op,
|
||
Entry: EntryAgentPush,
|
||
Policy: string(PolicyLWWTarget),
|
||
Outcome: OutcomeKept,
|
||
LoserPayload: loser,
|
||
WinnerPayload: winner,
|
||
TargetVer: tgtVer,
|
||
SourceVer: version,
|
||
})
|
||
res.OK = true
|
||
res.Skipped = true
|
||
res.Message = "target newer; kept (lww_target)"
|
||
if store != nil {
|
||
_ = store.PatchStats(ch.ID, func(c *Channel) { c.Stats.PushedSkipped++ })
|
||
}
|
||
return res, nil
|
||
case PolicyLWWSource:
|
||
loser := SnapshotTargetRow(ctx, db, ch.Remote.Driver, table, pkCol, rowPK)
|
||
if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, op, payload, version); err != nil {
|
||
return nil, Retryablef("apply: %v", err)
|
||
}
|
||
RecordLwwOverride(store, LwwOverride{
|
||
TenantID: ch.TenantID,
|
||
ChannelID: ch.ID,
|
||
Table: table,
|
||
RowPK: rowPK,
|
||
Op: op,
|
||
Entry: EntryAgentPush,
|
||
Policy: string(PolicyLWWSource),
|
||
Outcome: OutcomeApplied,
|
||
LoserPayload: loser,
|
||
WinnerPayload: payload,
|
||
TargetVer: tgtVer,
|
||
SourceVer: version,
|
||
})
|
||
if store != nil {
|
||
_ = store.PatchStats(ch.ID, func(c *Channel) {
|
||
c.Stats.PushedOK++
|
||
c.Stats.PushedApplied++
|
||
})
|
||
}
|
||
res.OK = true
|
||
res.Applied = true
|
||
res.Message = "applied (lww_source override)"
|
||
return res, nil
|
||
default:
|
||
if store != nil {
|
||
_ = store.AddConflict(Conflict{
|
||
TenantID: ch.TenantID,
|
||
ChannelID: ch.ID,
|
||
Table: table,
|
||
RowPK: rowPK,
|
||
Op: op,
|
||
Source: "agent",
|
||
Payload: payload,
|
||
TargetVer: tgtVer,
|
||
SourceVer: version,
|
||
Message: "target version newer than agent push",
|
||
})
|
||
_ = store.PatchStats(ch.ID, func(c *Channel) {
|
||
c.Stats.Conflicts++
|
||
c.Stats.PushedSkipped++
|
||
})
|
||
}
|
||
res.OK = true
|
||
res.Skipped = true
|
||
res.Conflict = true
|
||
res.Message = "queued conflict; target newer"
|
||
return res, nil
|
||
}
|
||
}
|
||
|
||
if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, op, payload, version); err != nil {
|
||
return nil, Retryablef("apply: %v", err)
|
||
}
|
||
if store != nil {
|
||
_ = store.PatchStats(ch.ID, func(c *Channel) {
|
||
c.Stats.PushedOK++
|
||
c.Stats.PushedApplied++
|
||
})
|
||
}
|
||
res.OK = true
|
||
res.Applied = true
|
||
res.Message = "applied"
|
||
return res, nil
|
||
}
|
||
|
||
// PushBatchToRemote 保序批量；遇错即停，已成功条数在返回切片中。
|
||
func PushBatchToRemote(ctx context.Context, ch *Channel, store *FileStore, items []PushItem) ([]PushResult, error) {
|
||
out := make([]PushResult, 0, len(items))
|
||
for i, it := range items {
|
||
r, err := PushToRemote(ctx, ch, store, it)
|
||
if err != nil {
|
||
fail := PushResult{
|
||
OK: false,
|
||
Message: err.Error(),
|
||
ClientOutboxID: it.ClientOutboxID,
|
||
}
|
||
out = append(out, fail)
|
||
return out, fmt.Errorf("item[%d]: %w", i, err)
|
||
}
|
||
out = append(out, *r)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func tableInChannel(ch *Channel, table string) bool {
|
||
// Z4 冻结：通道 tables 仅历史兼容展示，push/pull 不再据此拒收。
|
||
_ = ch
|
||
return strings.TrimSpace(table) != ""
|
||
}
|
||
|
||
func normalizePushPayload(item PushItem, pkCol string) (op string, payload string, rowPK string, err error) {
|
||
rawOp := strings.ToLower(strings.TrimSpace(item.Op))
|
||
if rawOp == "" {
|
||
rawOp = "upsert"
|
||
}
|
||
switch rawOp {
|
||
case "delete":
|
||
op = "delete"
|
||
case "insert", "update", "update_by_id", "upsert":
|
||
op = "upsert"
|
||
default:
|
||
return "", "", "", fmt.Errorf("unsupported op: %s", item.Op)
|
||
}
|
||
|
||
row := item.Row
|
||
if row == nil && len(item.Rows) > 0 {
|
||
row = item.Rows[0]
|
||
}
|
||
rowPK = strings.TrimSpace(item.RowPK)
|
||
if row != nil {
|
||
if rowPK == "" {
|
||
rowPK = strings.TrimSpace(fmt.Sprint(row[pkCol]))
|
||
}
|
||
if op == "upsert" {
|
||
// 保证 payload 含主键
|
||
if _, ok := row[pkCol]; !ok && rowPK != "" {
|
||
row = copyMap(row)
|
||
row[pkCol] = rowPK
|
||
}
|
||
}
|
||
}
|
||
if rowPK == "" || rowPK == "<nil>" {
|
||
return "", "", "", fmt.Errorf("row_pk required")
|
||
}
|
||
if op == "delete" && row == nil {
|
||
row = map[string]any{pkCol: rowPK}
|
||
}
|
||
if row == nil {
|
||
return "", "", "", fmt.Errorf("row required for %s", op)
|
||
}
|
||
b, err := json.Marshal(row)
|
||
if err != nil {
|
||
return "", "", "", err
|
||
}
|
||
return op, string(b), rowPK, nil
|
||
}
|
||
|
||
func copyMap(m map[string]any) map[string]any {
|
||
out := make(map[string]any, len(m)+1)
|
||
for k, v := range m {
|
||
out[k] = v
|
||
}
|
||
return out
|
||
}
|