From 680d2b8cdeb682eed97057857505e3452836537d Mon Sep 17 00:00:00 2001 From: whm <973418690@qq.com> Date: Thu, 6 Aug 2026 11:49:00 +0800 Subject: [PATCH] feat: add sync checkpoint restore for last successful sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capture rolling online DB snapshots after push/drain/reconcile and expose SyncPage 数据恢复 plus checkpoint/restore APIs; mark Z12h and restore done in coop docs. Co-authored-by: Cursor --- docs/数据同步-开通说明.md | 13 + platform/internal/apidef/catalog.go | 2 + platform/internal/dbsync/checkpoint.go | 453 ++++++++++++++++++++ platform/internal/dbsync/checkpoint_test.go | 89 ++++ platform/internal/dbsync/manager.go | 5 + platform/internal/handler/agent_sync.go | 12 + platform/internal/handler/routes.go | 2 + platform/internal/handler/sync.go | 60 +++ web/src/SyncPage.tsx | 82 ++++ web/src/api.ts | 49 +++ 联调后修改意见-宇恒松离线.md | 109 ++++- 11 files changed, 860 insertions(+), 16 deletions(-) create mode 100644 platform/internal/dbsync/checkpoint.go create mode 100644 platform/internal/dbsync/checkpoint_test.go diff --git a/docs/数据同步-开通说明.md b/docs/数据同步-开通说明.md index e834b28..401fe76 100644 --- a/docs/数据同步-开通说明.md +++ b/docs/数据同步-开通说明.md @@ -99,6 +99,19 @@ Binding：`POST /api/v1/admin/sync/bindings` 登记 `local_database_id → onlin **空表也要两侧建齐**：本机空表 → `POST .../schema/ensure`；线上空表 → `POST .../schema`（或 pull 的 `columns`）在本机 `CREATE IF NOT EXISTS`。仅靠 outbox 行 push **不会**带上空表。 控制台「查看线上表」可 **删表**（`POST .../admin/sync/channels/{id}/drop-table`）：只删当前查看侧，**不同步** `DROP` 到另一侧；本机仍有同名表时下次 push/ensure 可能再建回来。 +## 数据恢复（恢复上次同步 · 已落实） + +成功同步后（agent **自动/手动 push**、通道 drain、控制台「同步修复」），平台对**线上库**打滚动快照（`latest` + `previous`，约 30 秒防抖）。 + +| 操作 | 说明 | +|------|------| +| SyncPage「数据恢复」 | 将线上库恢复为所选快照（写入快照行，并删除快照中不存在的行） | +| 本机 | 平台不直连宇恒 SQLite；恢复线上后，终端下次同步/指纹 pull 可从线上补回 | +| 无快照 | 部署后须先完成至少一次成功同步；按钮会提示「尚无成功同步快照」 | +| 误删后又自动同步成功 | 最新快照可能已含删除后状态 → 确认框选「上一代」 | + +API：`GET .../admin/sync/channels/{id}/checkpoint`、`POST .../admin/sync/channels/{id}/restore`（`confirm=true`，`which=latest|previous`）。详见联调意见 §5.13。 + ## 谁能看什么 | 角色 | 可见 | diff --git a/platform/internal/apidef/catalog.go b/platform/internal/apidef/catalog.go index d03b1e8..a42fb09 100644 --- a/platform/internal/apidef/catalog.go +++ b/platform/internal/apidef/catalog.go @@ -74,6 +74,8 @@ var Catalog = []Entry{ {Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/start", OperationID: "startSyncChannel", Summary: "启动近实时同步", Group: "admin"}, {Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/stop", OperationID: "stopSyncChannel", Summary: "停止同步", Group: "admin"}, {Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/reconcile", OperationID: "reconcileSyncChannel", Summary: "主键对账（同步修复，有限流）", Group: "admin"}, + {Method: "GET", Path: "/api/v1/admin/sync/channels/{id}/checkpoint", OperationID: "getSyncCheckpoint", Summary: "线上库同步快照摘要（latest/previous）", Group: "admin"}, + {Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/restore", OperationID: "restoreSyncChannel", Summary: "按上次同步快照恢复线上库", Group: "admin"}, {Method: "POST", Path: "/api/v1/admin/sync/channels/{id}/ingest", OperationID: "ingestSyncRows", Summary: "外部行写入通道 local", Group: "admin"}, {Method: "GET", Path: "/api/v1/admin/sync/bindings", OperationID: "listSyncBindings", Summary: "列出本机库↔线上库绑定（用户仅本人）", Group: "admin"}, {Method: "POST", Path: "/api/v1/admin/sync/bindings", OperationID: "ensureSyncBinding", Summary: "登记/更新绑定（用户可自助）", Group: "admin"}, diff --git a/platform/internal/dbsync/checkpoint.go b/platform/internal/dbsync/checkpoint.go new file mode 100644 index 0000000..b48afec --- /dev/null +++ b/platform/internal/dbsync/checkpoint.go @@ -0,0 +1,453 @@ +package dbsync + +import ( + "compress/gzip" + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +const ( + checkpointDebounce = 30 * time.Second + checkpointMaxRows = 500_000 // 单次快照行数上限，避免撑爆磁盘 + checkpointFileLatest = "latest.json.gz" + checkpointFilePrev = "previous.json.gz" + checkpointFileMeta = "meta.json" +) + +// CheckpointSlotMeta 一代快照摘要（不读全量）。 +type CheckpointSlotMeta struct { + SyncedAt time.Time `json:"synced_at"` + Source string `json:"source"` + TableCount int `json:"table_count"` + RowCount int `json:"row_count"` +} + +// CheckpointMeta 通道快照目录摘要。 +type CheckpointMeta struct { + ChannelID string `json:"channel_id"` + Latest *CheckpointSlotMeta `json:"latest,omitempty"` + Previous *CheckpointSlotMeta `json:"previous,omitempty"` +} + +// CheckpointTable 单表快照。 +type CheckpointTable struct { + PKColumn string `json:"pk_column"` + Rows map[string]map[string]any `json:"rows"` // pk -> row +} + +// CheckpointPayload 全量快照内容。 +type CheckpointPayload struct { + ChannelID string `json:"channel_id"` + SyncedAt time.Time `json:"synced_at"` + Source string `json:"source"` + Tables map[string]CheckpointTable `json:"tables"` +} + +// RestoreResult 恢复结果。 +type RestoreResult struct { + OK bool `json:"ok"` + Which string `json:"which"` + SyncedAt time.Time `json:"synced_at"` + Source string `json:"source,omitempty"` + Tables int `json:"tables"` + Upserted int `json:"upserted"` + Deleted int `json:"deleted"` + RestoredAt time.Time `json:"restored_at"` +} + +var ( + cpSchedMu sync.Mutex + cpPending = map[string]*time.Timer{} +) + +func (s *FileStore) Dir() string { + if s == nil { + return "./data/dbsync" + } + return s.dir +} + +func checkpointDir(store *FileStore, channelID string) string { + return filepath.Join(store.Dir(), "checkpoints", strings.TrimSpace(channelID)) +} + +// ScheduleCheckpoint 防抖后写入线上库快照；失败仅打日志。 +func ScheduleCheckpoint(store *FileStore, ch *Channel, source string) { + if store == nil || ch == nil || strings.TrimSpace(ch.ID) == "" { + return + } + id := strings.TrimSpace(ch.ID) + chCopy := *ch + src := strings.TrimSpace(source) + if src == "" { + src = "sync" + } + cpSchedMu.Lock() + defer cpSchedMu.Unlock() + if t, ok := cpPending[id]; ok { + t.Stop() + } + cpPending[id] = time.AfterFunc(checkpointDebounce, func() { + cpSchedMu.Lock() + delete(cpPending, id) + cpSchedMu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + if err := SaveCheckpoint(ctx, store, &chCopy, src); err != nil { + log.Printf("dbsync checkpoint channel=%s source=%s: %v", id, src, err) + } + }) +} + +// SaveCheckpoint 导出通道 Remote 业务表，轮转 previous ← latest。 +func SaveCheckpoint(ctx context.Context, store *FileStore, ch *Channel, source string) error { + if store == nil || ch == nil { + return fmt.Errorf("store/channel required") + } + dir := checkpointDir(store, ch.ID) + if err := os.MkdirAll(dir, 0o755); err != nil { + return err + } + payload, err := dumpRemoteCheckpoint(ctx, ch, source) + if err != nil { + return err + } + latestPath := filepath.Join(dir, checkpointFileLatest) + prevPath := filepath.Join(dir, checkpointFilePrev) + // 轮转：现有 latest → previous + if _, err := os.Stat(latestPath); err == nil { + _ = os.Remove(prevPath) + if err := os.Rename(latestPath, prevPath); err != nil { + // Windows 上目标存在时 Rename 可能失败；已删 prev 再试 + _ = os.Remove(prevPath) + if err2 := os.Rename(latestPath, prevPath); err2 != nil { + return fmt.Errorf("rotate checkpoint: %w", err2) + } + } + } + tmp := latestPath + ".tmp" + if err := writeCheckpointGzip(tmp, payload); err != nil { + _ = os.Remove(tmp) + return err + } + if err := os.Rename(tmp, latestPath); err != nil { + _ = os.Remove(latestPath) + if err2 := os.Rename(tmp, latestPath); err2 != nil { + _ = os.Remove(tmp) + return err2 + } + } + meta := CheckpointMeta{ChannelID: ch.ID} + meta.Latest = slotMetaFromPayload(payload) + if prev, err := loadCheckpointPayload(prevPath); err == nil && prev != nil { + meta.Previous = slotMetaFromPayload(prev) + } + return writeCheckpointMeta(dir, meta) +} + +func slotMetaFromPayload(p *CheckpointPayload) *CheckpointSlotMeta { + if p == nil { + return nil + } + rows := 0 + for _, t := range p.Tables { + rows += len(t.Rows) + } + return &CheckpointSlotMeta{ + SyncedAt: p.SyncedAt, + Source: p.Source, + TableCount: len(p.Tables), + RowCount: rows, + } +} + +func dumpRemoteCheckpoint(ctx context.Context, ch *Channel, source string) (*CheckpointPayload, error) { + db, err := AcquireRemote(ch.Remote.Driver, ch.Remote.DSN) + if err != nil { + return nil, wrapOpenRemote(err) + } + names, err := listTablesForInspect(ctx, db, ch.Remote.Driver, false) + if err != nil { + return nil, err + } + out := &CheckpointPayload{ + ChannelID: ch.ID, + SyncedAt: time.Now().UTC(), + Source: source, + Tables: make(map[string]CheckpointTable, len(names)), + } + totalRows := 0 + for _, table := range names { + pkCol := "id" + if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" { + pkCol = strings.TrimSpace(ch.PKColumns[table]) + } + cols, err := listColumns(ctx, db, ch.Remote.Driver, table) + if err != nil { + return nil, fmt.Errorf("columns %s: %w", table, err) + } + rows, err := fetchAllRows(ctx, db, ch.Remote.Driver, table, cols) + if err != nil { + return nil, fmt.Errorf("dump %s: %w", table, err) + } + m := make(map[string]map[string]any, len(rows)) + for _, row := range rows { + pk := fmt.Sprint(row[pkCol]) + if pk == "" || pk == "" { + continue + } + m[pk] = row + } + totalRows += len(m) + if totalRows > checkpointMaxRows { + return nil, fmt.Errorf("checkpoint too large: >%d rows", checkpointMaxRows) + } + out.Tables[table] = CheckpointTable{PKColumn: pkCol, Rows: m} + } + return out, nil +} + +func fetchAllRows(ctx context.Context, db *sql.DB, driver Driver, table string, cols []string) ([]map[string]any, error) { + if len(cols) == 0 { + return nil, nil + } + q := fmt.Sprintf(`SELECT * FROM %s`, quoteIdent(driver, table)) + rows, err := db.QueryContext(ctx, q) + if err != nil { + return nil, err + } + defer rows.Close() + colNames, err := rows.Columns() + if err != nil { + return nil, err + } + var out []map[string]any + for rows.Next() { + raw := make([]any, len(colNames)) + ptrs := make([]any, len(colNames)) + for i := range raw { + ptrs[i] = &raw[i] + } + if err := rows.Scan(ptrs...); err != nil { + return nil, err + } + m := make(map[string]any, len(colNames)) + for i, c := range colNames { + m[c] = normalizeValue(raw[i]) + } + out = append(out, m) + } + return out, rows.Err() +} + +func writeCheckpointGzip(path string, payload *CheckpointPayload) error { + f, err := os.Create(path) + if err != nil { + return err + } + defer f.Close() + zw := gzip.NewWriter(f) + enc := json.NewEncoder(zw) + if err := enc.Encode(payload); err != nil { + _ = zw.Close() + return err + } + if err := zw.Close(); err != nil { + return err + } + return f.Close() +} + +func loadCheckpointPayload(path string) (*CheckpointPayload, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer f.Close() + zr, err := gzip.NewReader(f) + if err != nil { + return nil, err + } + defer zr.Close() + var p CheckpointPayload + if err := json.NewDecoder(zr).Decode(&p); err != nil && err != io.EOF { + return nil, err + } + return &p, nil +} + +func writeCheckpointMeta(dir string, meta CheckpointMeta) error { + b, err := json.MarshalIndent(meta, "", " ") + if err != nil { + return err + } + return os.WriteFile(filepath.Join(dir, checkpointFileMeta), b, 0o644) +} + +// LoadCheckpointMeta 读取摘要；无快照时返回空 meta（不报错）。 +func LoadCheckpointMeta(store *FileStore, channelID string) (*CheckpointMeta, error) { + if store == nil || strings.TrimSpace(channelID) == "" { + return &CheckpointMeta{}, nil + } + dir := checkpointDir(store, channelID) + b, err := os.ReadFile(filepath.Join(dir, checkpointFileMeta)) + if err != nil { + if os.IsNotExist(err) { + // 尝试从文件推断 + meta := &CheckpointMeta{ChannelID: channelID} + if p, e := loadCheckpointPayload(filepath.Join(dir, checkpointFileLatest)); e == nil { + meta.Latest = slotMetaFromPayload(p) + } + if p, e := loadCheckpointPayload(filepath.Join(dir, checkpointFilePrev)); e == nil { + meta.Previous = slotMetaFromPayload(p) + } + return meta, nil + } + return nil, err + } + var meta CheckpointMeta + if err := json.Unmarshal(b, &meta); err != nil { + return nil, err + } + meta.ChannelID = channelID + return &meta, nil +} + +// LoadCheckpoint 加载 latest 或 previous。 +func LoadCheckpoint(store *FileStore, channelID, which string) (*CheckpointPayload, error) { + which = strings.TrimSpace(which) + if which == "" { + which = "latest" + } + if which != "latest" && which != "previous" { + return nil, fmt.Errorf("which 须为 latest 或 previous") + } + name := checkpointFileLatest + if which == "previous" { + name = checkpointFilePrev + } + path := filepath.Join(checkpointDir(store, channelID), name) + p, err := loadCheckpointPayload(path) + if err != nil { + if os.IsNotExist(err) { + return nil, fmt.Errorf("无可用快照（%s）", which) + } + return nil, err + } + return p, nil +} + +// RestoreCheckpoint 将线上库恢复为指定快照（upsert + 删除快照外 PK）。 +func RestoreCheckpoint(ctx context.Context, store *FileStore, ch *Channel, which string) (*RestoreResult, error) { + if store == nil || ch == nil { + return nil, fmt.Errorf("store/channel required") + } + payload, err := LoadCheckpoint(store, ch.ID, which) + if err != nil { + return nil, err + } + 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, err + } + verBase := time.Now().UnixNano() + upserted, deleted := 0, 0 + for table, ct := range payload.Tables { + pkCol := strings.TrimSpace(ct.PKColumn) + if pkCol == "" { + pkCol = "id" + if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" { + pkCol = strings.TrimSpace(ch.PKColumns[table]) + } + } + // 先 upsert 快照行 + i := 0 + for pk, row := range ct.Rows { + if len(row) == 0 { + continue + } + if err := EnsureTableFromRow(ctx, db, ch.Remote.Driver, table, pkCol, row); err != nil { + return nil, fmt.Errorf("ensure %s: %w", table, err) + } + b, err := json.Marshal(row) + if err != nil { + return nil, err + } + ver := verBase + int64(i) + i++ + if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, "upsert", string(b), ver); err != nil { + return nil, fmt.Errorf("upsert %s pk=%s: %w", table, pk, err) + } + upserted++ + } + // 删快照中不存在的行（含空表：清空线上多余行） + livePKs, err := listTablePKs(ctx, db, ch.Remote.Driver, table, pkCol) + if err != nil { + // 表可能尚不存在且快照也空 + if len(ct.Rows) == 0 { + continue + } + return nil, fmt.Errorf("list pk %s: %w", table, err) + } + for _, pk := range livePKs { + if _, ok := ct.Rows[pk]; ok { + continue + } + delPayload, _ := json.Marshal(map[string]any{pkCol: pk}) + ver := verBase + int64(i) + i++ + if err := ApplyChange(ctx, db, ch.Remote.Driver, table, pkCol, "delete", string(delPayload), ver); err != nil { + return nil, fmt.Errorf("delete %s pk=%s: %w", table, pk, err) + } + deleted++ + } + } + // 快照里没有、线上多出来的业务表：不自动 DROP（避免误伤）；仅对快照内表做行级 prune + which = strings.TrimSpace(which) + if which == "" { + which = "latest" + } + return &RestoreResult{ + OK: true, + Which: which, + SyncedAt: payload.SyncedAt, + Source: payload.Source, + Tables: len(payload.Tables), + Upserted: upserted, + Deleted: deleted, + RestoredAt: time.Now().UTC(), + }, nil +} + +func listTablePKs(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 + } + s := fmt.Sprint(normalizeValue(v)) + if s != "" && s != "" { + out = append(out, s) + } + } + return out, rows.Err() +} diff --git a/platform/internal/dbsync/checkpoint_test.go b/platform/internal/dbsync/checkpoint_test.go new file mode 100644 index 0000000..60b1a18 --- /dev/null +++ b/platform/internal/dbsync/checkpoint_test.go @@ -0,0 +1,89 @@ +package dbsync + +import ( + "context" + "path/filepath" + "testing" + "time" +) + +func TestCheckpointSaveRestoreRoundTrip(t *testing.T) { + dir := t.TempDir() + store, err := NewFileStore(dir) + if err != nil { + t.Fatal(err) + } + remoteDSN := "file:" + filepath.ToSlash(filepath.Join(dir, "online.db")) + "?_pragma=busy_timeout(5000)" + ch := &Channel{ + ID: "ch-cp-1", + TenantID: 1, + Name: "test", + Remote: Endpoint{ + Driver: DriverSQLite, + DSN: remoteDSN, + }, + PKColumns: map[string]string{"demo": "id"}, + } + ctx := context.Background() + db, err := Open(DriverSQLite, remoteDSN) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = db.Close() + InvalidateRemote(DriverSQLite, remoteDSN) + }) + if _, err := db.ExecContext(ctx, `CREATE TABLE demo (id TEXT PRIMARY KEY, name TEXT)`); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO demo(id,name) VALUES('a','one'),('b','two')`); err != nil { + t.Fatal(err) + } + if err := EnsureMeta(ctx, db, DriverSQLite); err != nil { + t.Fatal(err) + } + + if err := SaveCheckpoint(ctx, store, ch, "test"); err != nil { + t.Fatal(err) + } + meta, err := LoadCheckpointMeta(store, ch.ID) + if err != nil || meta.Latest == nil || meta.Latest.RowCount != 2 { + t.Fatalf("meta=%+v err=%v", meta, err) + } + + // mutate then restore + if _, err := db.ExecContext(ctx, `DELETE FROM demo WHERE id='a'`); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO demo(id,name) VALUES('c','three')`); err != nil { + t.Fatal(err) + } + res, err := RestoreCheckpoint(ctx, store, ch, "latest") + if err != nil { + t.Fatal(err) + } + if !res.OK || res.Upserted < 2 { + t.Fatalf("restore=%+v", res) + } + var n int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM demo`).Scan(&n); err != nil || n != 2 { + t.Fatalf("count=%d err=%v", n, err) + } + var name string + if err := db.QueryRowContext(ctx, `SELECT name FROM demo WHERE id='a'`).Scan(&name); err != nil || name != "one" { + t.Fatalf("row a name=%q err=%v", name, err) + } + + // second save rotates + time.Sleep(10 * time.Millisecond) + if _, err := db.ExecContext(ctx, `UPDATE demo SET name='one2' WHERE id='a'`); err != nil { + t.Fatal(err) + } + if err := SaveCheckpoint(ctx, store, ch, "test2"); err != nil { + t.Fatal(err) + } + meta2, _ := LoadCheckpointMeta(store, ch.ID) + if meta2.Previous == nil || meta2.Previous.RowCount != 2 { + t.Fatalf("expected previous after rotate: %+v", meta2) + } +} diff --git a/platform/internal/dbsync/manager.go b/platform/internal/dbsync/manager.go index 8e07500..59cecd2 100644 --- a/platform/internal/dbsync/manager.go +++ b/platform/internal/dbsync/manager.go @@ -120,6 +120,7 @@ func (m *Manager) loop(ctx context.Context, id string) { now := time.Now().UTC() c.LastReconcileAt = &now }) + ScheduleCheckpoint(m.store, ch, "reconcile") } } } @@ -176,6 +177,10 @@ func (m *Manager) tick(ctx context.Context, ch *Channel) error { c.LastError = "" } }) + // 成功且本批有变更 → 防抖打线上库快照（数据恢复） + if err == nil && n > 0 { + ScheduleCheckpoint(m.store, ch, "drain") + } return err } diff --git a/platform/internal/handler/agent_sync.go b/platform/internal/handler/agent_sync.go index da27194..164233c 100644 --- a/platform/internal/handler/agent_sync.go +++ b/platform/internal/handler/agent_sync.go @@ -90,6 +90,9 @@ func agentSyncPushHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { if res != nil && res.Skipped { outcome = "skipped" } + if res != nil && res.Applied { + dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "push") + } logSyncReq(r, "push", channelID, item.Table, item.RowPK, outcome, "", dur, reqID) auditSync(svcCtx, r, "dbsync.push", map[string]any{ "channel_id": channelID, "table": item.Table, "row_pk": item.RowPK, @@ -153,6 +156,15 @@ func agentSyncPushBatchHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { httpx.OkJson(w, payload) return } + applied := 0 + for _, r0 := range results { + if r0.Applied { + applied++ + } + } + if applied > 0 { + dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "push_batch") + } logSyncReq(r, "push_batch", channelID, "", "", "ok", "", dur, reqID) auditSync(svcCtx, r, "dbsync.push_batch", map[string]any{ "channel_id": channelID, "n": len(body.Items), "ms": dur.Milliseconds(), "req_id": reqID, diff --git a/platform/internal/handler/routes.go b/platform/internal/handler/routes.go index ab2862d..a2cac17 100644 --- a/platform/internal/handler/routes.go +++ b/platform/internal/handler/routes.go @@ -146,6 +146,8 @@ func RegisterHandlers(server *rest.Server, svcCtx *svc.ServiceContext) { {Method: http.MethodGet, Path: "/api/v1/admin/sync/conflicts", Handler: chain(syncConflictsHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, {Method: http.MethodPost, Path: "/api/v1/admin/sync/conflicts/:id/resolve", Handler: chain(syncResolveConflictHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, {Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/reconcile", Handler: chain(syncReconcileHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, + {Method: http.MethodGet, Path: "/api/v1/admin/sync/channels/:id/checkpoint", Handler: chain(syncCheckpointMetaHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, + {Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/restore", Handler: chain(syncRestoreHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, {Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/ingest", Handler: chain(syncIngestHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, {Method: http.MethodGet, Path: "/api/v1/admin/sync/channels/:id/inspect", Handler: chain(syncInspectHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, {Method: http.MethodPost, Path: "/api/v1/admin/sync/channels/:id/preview", Handler: chain(syncPreviewHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm数据同步))}, diff --git a/platform/internal/handler/sync.go b/platform/internal/handler/sync.go index 5c95b82..a685b24 100644 --- a/platform/internal/handler/sync.go +++ b/platform/internal/handler/sync.go @@ -262,6 +262,7 @@ func syncReconcileHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { _ = svcCtx.DBSync.Store().PatchStats(ch.ID, func(c *dbsync.Channel) { c.LastReconcileAt = &now }) + dbsync.ScheduleCheckpoint(svcCtx.DBSync.Store(), ch, "reconcile") httpx.OkJson(w, res) } } @@ -418,3 +419,62 @@ func syncDropTableHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { httpx.OkJson(w, res) } } + +func syncCheckpointMetaHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if !requireDBSync(svcCtx, w) { + return + } + id := pathvar.Vars(r)["id"] + if _, err := svcCtx.DBSync.Store().GetChannelForTenant(id, syncTenantID(r)); err != nil { + authx.WriteError(w, http.StatusNotFound, err.Error()) + return + } + meta, err := dbsync.LoadCheckpointMeta(svcCtx.DBSync.Store(), id) + if err != nil { + authx.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + httpx.OkJson(w, meta) + } +} + +func syncRestoreHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if !requireDBSync(svcCtx, w) { + return + } + ch, err := svcCtx.DBSync.Store().GetChannelForTenant(pathvar.Vars(r)["id"], syncTenantID(r)) + if err != nil { + authx.WriteError(w, http.StatusNotFound, err.Error()) + return + } + var body struct { + Which string `json:"which"` + Confirm bool `json:"confirm"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + authx.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + if !body.Confirm { + authx.WriteError(w, http.StatusBadRequest, "请确认恢复：confirm=true") + return + } + which := strings.TrimSpace(body.Which) + if which == "" { + which = "latest" + } + res, err := dbsync.RestoreCheckpoint(r.Context(), svcCtx.DBSync.Store(), ch, which) + if err != nil { + authx.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + if svcCtx.Audit != nil { + _ = svcCtx.Audit.Log(r.Context(), syncTenantID(r), authx.UserID(r.Context()), "dbsync.restore", audit.DetailJSON(map[string]any{ + "channel_id": ch.ID, "which": which, "synced_at": res.SyncedAt, "upserted": res.Upserted, "deleted": res.Deleted, + })) + } + httpx.OkJson(w, res) + } +} diff --git a/web/src/SyncPage.tsx b/web/src/SyncPage.tsx index 0c41d4c..4c95030 100644 --- a/web/src/SyncPage.tsx +++ b/web/src/SyncPage.tsx @@ -34,6 +34,7 @@ import { createSyncChannel, deleteSyncChannel, dropSyncTable, + getSyncCheckpointMeta, inspectSyncChannel, listAgents, listApps, @@ -44,6 +45,7 @@ import { revokeBindCode, previewSyncTable, reconcileSyncChannel, + restoreSyncChannel, startSyncChannel, stopSyncChannel, testSyncEndpoints, @@ -308,6 +310,83 @@ export function SyncPage(props: { } } + function formatCheckpointTime(iso?: string) { + if (!iso) return "—"; + const d = new Date(iso); + if (Number.isNaN(d.getTime())) return iso; + return d.toLocaleString(); + } + + async function openRestore(r: SyncChannel) { + setBusy(true); + try { + const meta = await getSyncCheckpointMeta(session, r.id); + if (!meta.latest && !meta.previous) { + message.warning("尚无成功同步快照。请先完成一次同步（含自动推送），约 30 秒后生成。"); + return; + } + let which: "latest" | "previous" = meta.latest ? "latest" : "previous"; + const options: { value: "latest" | "previous"; label: string }[] = []; + if (meta.latest) { + options.push({ + value: "latest", + label: `最近一次 · ${formatCheckpointTime(meta.latest.synced_at)}（${meta.latest.row_count ?? 0} 行 / ${meta.latest.table_count ?? 0} 表）`, + }); + } + if (meta.previous) { + options.push({ + value: "previous", + label: `上一代 · ${formatCheckpointTime(meta.previous.synced_at)}（${meta.previous.row_count ?? 0} 行 / ${meta.previous.table_count ?? 0} 表）`, + }); + } + Modal.confirm({ + title: `数据恢复 · ${r.name || r.id}`, + width: 560, + content: ( +
+

+ 将线上库恢复为所选同步快照（upsert 并删除快照中不存在的行）。本机数据需宇恒下次同步/pull 从线上补回。 +

+

+ 若误删后又发生成功自动同步，最新快照可能已含删除后状态，请改选「上一代」。 +

+
+ 恢复到 +