diff --git a/platform/internal/dbsync/ensure_table.go b/platform/internal/dbsync/ensure_table.go index ba4a9c4..8641065 100644 --- a/platform/internal/dbsync/ensure_table.go +++ b/platform/internal/dbsync/ensure_table.go @@ -37,8 +37,8 @@ func EnsureTableFromRow(ctx context.Context, db *sql.DB, driver Driver, table, p return EnsureTableFromColumns(ctx, db, driver, table, pkCol, cols) } -// EnsureTableFromColumns 表不存在时按列名建空表(全部 TEXT,指定 PK)。已存在则幂等跳过。 -// 用于空表结构同步:本机有空表 → 线上也建同名空表(无需 outbox 行)。 +// EnsureTableFromColumns 按列名建空表(全部 TEXT,指定 PK)。 +// 已存在时补齐缺失列(ADD COLUMN),解决「先按 _row_id 建表、后 push 注入 id」导致的缺列 503。 func EnsureTableFromColumns(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string, columns []string) error { table = strings.TrimSpace(table) pkCol = strings.TrimSpace(pkCol) @@ -55,27 +55,57 @@ func EnsureTableFromColumns(ctx context.Context, db *sql.DB, driver Driver, tabl if err != nil { return err } - if exists { - return nil - } + want := make([]string, 0, len(columns)+1) 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 { + add := func(c string) { c = strings.TrimSpace(c) if c == "" { - continue + return } if _, ok := seen[c]; ok { - continue + return } seen[c] = struct{}{} - defs = append(defs, fmt.Sprintf("%s TEXT", quoteIdent(driver, c))) + want = append(want, c) } - if len(defs) == 0 { + add(pkCol) + for _, c := range columns { + add(c) + } + if len(want) == 0 { return fmt.Errorf("columns required") } + if exists { + have, err := listColumns(ctx, db, driver, table) + if err != nil { + return err + } + haveSet := map[string]struct{}{} + for _, c := range have { + haveSet[c] = struct{}{} + } + for _, c := range want { + if _, ok := haveSet[c]; ok { + continue + } + ddl := fmt.Sprintf(`ALTER TABLE %s ADD COLUMN %s TEXT`, quoteIdent(driver, table), quoteIdent(driver, c)) + if _, err := db.ExecContext(ctx, ddl); err != nil { + // SQLite 无 IF NOT EXISTS;并发下可能已存在 + if !strings.Contains(strings.ToLower(err.Error()), "duplicate column") { + return err + } + } + } + return nil + } + defs := make([]string, 0, len(want)) + defs = append(defs, fmt.Sprintf("%s TEXT PRIMARY KEY", quoteIdent(driver, pkCol))) + for _, c := range want { + if c == pkCol { + continue + } + defs = append(defs, fmt.Sprintf("%s TEXT", quoteIdent(driver, c))) + } ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s)`, quoteIdent(driver, table), strings.Join(defs, ", ")) _, err = db.ExecContext(ctx, ddl) return err diff --git a/platform/internal/dbsync/fingerprint.go b/platform/internal/dbsync/fingerprint.go new file mode 100644 index 0000000..97dd16d --- /dev/null +++ b/platform/internal/dbsync/fingerprint.go @@ -0,0 +1,199 @@ +package dbsync + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strings" +) + +// FingerprintTable 单表指纹(与宇恒 sync_fingerprint 算法对齐)。 +type FingerprintTable struct { + Name string `json:"name"` + PKColumn string `json:"pk_column"` + RowCount int64 `json:"row_count"` + ContentHash string `json:"content_hash"` + Columns []string `json:"columns,omitempty"` + Error string `json:"error,omitempty"` +} + +// FingerprintRequest agent 拉线上表指纹。 +type FingerprintRequest struct { + OnlineDBID string `json:"online_db_id"` + Tables []string `json:"tables,omitempty"` +} + +// FingerprintResult 批量指纹。 +type FingerprintResult struct { + OK bool `json:"ok"` + Tables []FingerprintTable `json:"tables"` +} + +// FingerprintRemoteTables 计算线上 A 业务表 row_count + content_hash。 +// 哈希:sha256( table+"|"+pk+"|" + Σ json.dumps(row, sort_keys=True)+"\n" ) 取前 32 hex。 +func FingerprintRemoteTables(ctx context.Context, ch *Channel, req FingerprintRequest) (*FingerprintResult, error) { + if ch == nil { + return nil, fmt.Errorf("channel is nil") + } + 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) + } + names, err := ListTables(ctx, db, ch.Remote.Driver) + if err != nil { + return nil, Retryablef("list tables: %v", err) + } + want := map[string]struct{}{} + for _, t := range req.Tables { + t = strings.TrimSpace(t) + if t != "" { + want[t] = struct{}{} + } + } + out := &FingerprintResult{OK: true, Tables: make([]FingerprintTable, 0, len(names))} + for _, name := range names { + if strings.HasPrefix(name, "_ajz_") { + continue + } + if len(want) > 0 { + if _, ok := want[name]; !ok { + continue + } + } + if !tableInChannel(ch, name) { + continue + } + fp, ferr := fingerprintOneTable(ctx, db, ch, name) + if ferr != nil { + fp.Error = ferr.Error() + out.OK = false + } + out.Tables = append(out.Tables, fp) + } + return out, nil +} + +func fingerprintOneTable(ctx context.Context, db *sql.DB, ch *Channel, table string) (FingerprintTable, error) { + cols, err := listColumns(ctx, db, ch.Remote.Driver, table) + if err != nil { + return FingerprintTable{Name: table}, err + } + pkCol := "id" + if ch.PKColumns != nil && strings.TrimSpace(ch.PKColumns[table]) != "" { + pkCol = ch.PKColumns[table] + } + if !containsStr(cols, pkCol) { + for _, c := range []string{"id", "__row_id", "_row_id"} { + if containsStr(cols, c) { + pkCol = c + break + } + } + if !containsStr(cols, pkCol) && len(cols) > 0 { + pkCol = cols[0] + } + } + n, err := countRows(ctx, db, ch.Remote.Driver, table) + if err != nil { + return FingerprintTable{Name: table, PKColumn: pkCol, Columns: cols}, err + } + hash, err := contentHashOrdered(ctx, db, ch.Remote.Driver, table, pkCol) + if err != nil { + return FingerprintTable{Name: table, PKColumn: pkCol, RowCount: n, Columns: cols}, err + } + return FingerprintTable{ + Name: table, + PKColumn: pkCol, + RowCount: n, + ContentHash: hash, + Columns: cols, + }, nil +} + +func containsStr(ss []string, x string) bool { + for _, s := range ss { + if s == x { + return true + } + } + return false +} + +func contentHashOrdered(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string) (string, error) { + q := fmt.Sprintf(`SELECT * FROM %s ORDER BY %s`, quoteIdent(driver, table), quoteIdent(driver, pkCol)) + rows, err := db.QueryContext(ctx, q) + if err != nil { + return "", err + } + defer rows.Close() + colNames, err := rows.Columns() + if err != nil { + return "", err + } + h := sha256.New() + _, _ = h.Write([]byte(table + "|" + pkCol + "|")) + ptrs := make([]any, len(colNames)) + vals := make([]any, len(colNames)) + for i := range vals { + ptrs[i] = &vals[i] + } + for rows.Next() { + for i := range vals { + vals[i] = nil + } + if err := rows.Scan(ptrs...); err != nil { + return "", err + } + m := make(map[string]any, len(colNames)) + for i, c := range colNames { + m[c] = normalizeSQLValue(vals[i]) + } + blob, err := stableJSON(m) + if err != nil { + return "", err + } + _, _ = h.Write(blob) + _, _ = h.Write([]byte("\n")) + } + if err := rows.Err(); err != nil { + return "", err + } + sum := hex.EncodeToString(h.Sum(nil)) + if len(sum) > 32 { + sum = sum[:32] + } + return sum, nil +} + +// stableJSON 按 key 排序,贴近 Python json.dumps(..., sort_keys=True, default=str)。 +func stableJSON(m map[string]any) ([]byte, error) { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + parts := make([]string, 0, len(keys)) + for _, k := range keys { + kb, err := json.Marshal(k) + if err != nil { + return nil, err + } + vb, err := json.Marshal(m[k]) + if err != nil { + // fallback string + vb, err = json.Marshal(fmt.Sprint(m[k])) + if err != nil { + return nil, err + } + } + parts = append(parts, string(kb)+":"+string(vb)) + } + return []byte("{" + strings.Join(parts, ",") + "}"), nil +} diff --git a/platform/internal/handler/agent_sync.go b/platform/internal/handler/agent_sync.go index 17d1950..a9af4f9 100644 --- a/platform/internal/handler/agent_sync.go +++ b/platform/internal/handler/agent_sync.go @@ -403,6 +403,53 @@ func agentSyncSchemaDescribeHandler(svcCtx *svc.ServiceContext) http.HandlerFunc } } +func agentSyncSchemaFingerprintHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + started := time.Now() + reqID := r.Header.Get("X-Request-Id") + if !requireDBSync(svcCtx, w) { + return + } + channelID := pathvar.Vars(r)["id"] + ch, err := svcCtx.DBSync.Store().GetChannelForTenant(channelID, syncTenantID(r)) + if err != nil { + authx.WriteError(w, http.StatusNotFound, err.Error()) + return + } + var req dbsync.FingerprintRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + authx.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + if !authorizeUserOnlineDB(svcCtx, w, r, channelID, req.OnlineDBID, "用户自助 schema/fingerprint 须带 online_db_id,且须为本人 Binding") { + return + } + if !authorizeUserSyncChannel(svcCtx, w, r, channelID) { + return + } + res, err := dbsync.FingerprintRemoteTables(r.Context(), ch, req) + dur := time.Since(started) + if err != nil { + logSyncReq(r, "schema/fingerprint", channelID, "", "", "err", err.Error(), dur, reqID) + if dbsync.IsRetryable(err) { + authx.WriteRetryableError(w, http.StatusServiceUnavailable, err.Error()) + return + } + authx.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + n := 0 + if res != nil { + n = len(res.Tables) + } + logSyncReq(r, "schema/fingerprint", channelID, "", "", "ok", "", dur, reqID) + auditSync(svcCtx, r, "dbsync.schema_fingerprint", map[string]any{ + "channel_id": channelID, "n": n, "ms": dur.Milliseconds(), "req_id": reqID, + }) + httpx.OkJson(w, map[string]any{"success": true, "result": res}) + } +} + func uniqueStringSlice(a, b []string) []string { seen := map[string]struct{}{} var out []string diff --git a/platform/internal/handler/routes.go b/platform/internal/handler/routes.go index a2d59e9..ab2862d 100644 --- a/platform/internal/handler/routes.go +++ b/platform/internal/handler/routes.go @@ -161,6 +161,7 @@ func RegisterHandlers(server *rest.Server, svcCtx *svc.ServiceContext) { {Method: http.MethodPost, Path: "/api/v1/agent/sync/channels/:id/bootstrap", Handler: chain(agentSyncBootstrapHandler(svcCtx), rl, authMW, tenant, syncPush)}, {Method: http.MethodPost, Path: "/api/v1/agent/sync/channels/:id/schema", Handler: chain(agentSyncSchemaDescribeHandler(svcCtx), rl, authMW, tenant, syncPush)}, {Method: http.MethodPost, Path: "/api/v1/agent/sync/channels/:id/schema/ensure", Handler: chain(agentSyncSchemaEnsureHandler(svcCtx), rl, authMW, tenant, syncPush)}, + {Method: http.MethodPost, Path: "/api/v1/agent/sync/channels/:id/schema/fingerprint", Handler: chain(agentSyncSchemaFingerprintHandler(svcCtx), rl, authMW, tenant, syncPush)}, // 存储:POST 创建对象,GET 读取(无 /upload 动词路径;旧路径保留别名防断裂) {Method: http.MethodPost, Path: "/api/v1/storage", Handler: chain(uploadHandler(svcCtx), rl, authMW, tenant, perm(authx.Perm上传文件))}, diff --git a/联调后修改意见-宇恒松离线.md b/联调后修改意见-宇恒松离线.md index f429054..5037199 100644 --- a/联调后修改意见-宇恒松离线.md +++ b/联调后修改意见-宇恒松离线.md @@ -1,7 +1,7 @@ # 联调后修改意见 · 宇恒松离线(形态 B) -> 初稿:2026-08-01 · 持续修订至 **2026-08-05** -> 本次整理:宇恒 Z10c/Z13 **客户端已接**回写;**§0.2** 拆成「已完成 / 仍待」;生产阻塞仅智建运维(部署/配号/Secret) +> 初稿:2026-08-01 · 持续修订至 **2026-08-05 晚** +> 本次整理:生产 platform 已恢复;**§0.3 双方配合**;默认同步通道**默认运行中**;Z10d/Z14 合入待生产 pull;凭票/SMS 引号踩坑回写 > 来源:宇恒客户端 `yuhengyihao_client` ↔ 智建 gateway/platform(本机 `8180`/`8888` 或生产 `aisite.yuxindazhineng.com`) > 依据:`松离线-dbsync方案-最终版.md`、`宇恒-松离线数据同步使用文档.md` @@ -11,80 +11,102 @@ | 级别 | 状态 | 说明 | |------|------|------| -| **硬改(阻塞接口上线)** | **无** | Z1–Z13 API/技能大体齐;**生产联调**仍取决于 §0.2 **B 智建运维** | -| **原建议项 2.1–2.5** | **智建侧基本已落实** | 见 §2 对照表;接口请保持兼容 | -| **用户自助 Z1–Z7** | **已落实** | Z2 宇恒复测通过;Z7 可读名:宇恒 ensure 宜带 `database_name`(见 §0.2 A′) | -| **Z8 模块↔本机库+智能体库** | **已落实** | Z8a–g;见 §5.5 / §5.5.2 | -| **Z9 模块导入默认能力(方案 A)** | **已落实(含存量)** | Z9a–f;见 §5.6 | -| **Z10 空表双侧建齐** | **双方已接** | 智建 `schema/ensure`;宇恒 `sync_schema.ensure_before_drain`(§5.7) | -| **Z11 蓝图↔库列一致** | **智建已落实** | 发布自动 `ADD COLUMN IF NOT EXISTS`;见 §5.8 | -| **Z12 开通落点** | **双方已接** | 智建自动默认同步通道;宇恒换票/唯一通道/过渡 env(§5.9) | -| **Z13 绑定 UX** | **双方代码已接;生产待运维** | 智建 API+凭票+policy;宇恒技能/本机 bind API;见 §5.10、§0.2 B | -| **仍建议关注** | 性能/运维 | SQLite remote 高并发 push 易锁;agent 宜单库串行 drain | +| **硬改(阻塞接口上线)** | **无** | Z1–Z14 代码侧大体齐;剩运维配置 + Z14c ACL | +| **生产栈** | **已恢复(2026-08-05)** | `SMS.Provider: "off"` 须加引号;health `platform/ai=true`;部署脚本 `chmod +x` | +| **用户自助 Z1–Z7** | **已落实** | Z7:宇恒 ensure 宜带 `database_name`(§0.2 A′) | +| **Z8–Z9** | **已落实** | 见 §5.5 / §5.6 | +| **Z10 / Z10d** | **双方已接 / 智建已合入** | ensure + 空表;**Z10d** 已存在表 `ADD COLUMN`(生产须 pull) | +| **Z11** | **智建已落实** | 发布自动补列 | +| **Z12** | **双方已接** | 默认同步通道;**默认 Enabled=运行中**(`cb76824`) | +| **Z13** | **双方代码已接** | 绑定码/手机/凭票/policy;生产配号+Secret 见 §0.2 B | +| **Z14** | **宇恒已接;智建 fingerprint 已合入待 pull** | 见 §5.11;Z14c `_uN` ACL 仍待智建修 | +| **仍建议关注** | 性能 | SQLite remote 串行 drain;高并发易锁 | ### 0.1 绑定产品冻结摘要(Z13 · 试运行) | 约定 | 内容 | |------|------| | 联调号 | 生产「宇信达」用 **`13531041944`**;**禁止**超管号 `13531041945` | -| 换票 | **凭票免登录**:宇恒签 `ticket-exchange`(`YXD_YUHENG_TICKET_SECRET` + `YXD_HOST_KEY` + 手机号);**不要**手写 `YXD_SYNC_LOGIN_PASSWORD` | -| 短信 | **试运行关闭**:`SMS.RequireForBind=false`;`GET /api/v1/auth/bind/policy` → `trial_mode=true`。接入短信后改 `true` | -| 宇恒未绑手机 | **仍可直接输入手机号**绑定(试运行免短信;正式须短信) | -| 同号 | 输入号 = 宇恒已绑号 → 正式可免短信 / 凭票;命中智建成员 → **必须弹「已有账号是否绑定」**,不得静默 | -| 确认 | 无论试运行与否,绑到已有账号都须用户 **确认**(`confirm=true`) | -| 当前阻塞 | **智建生产运维**(§0.2 B):部署含公开路径的 gateway、成员绑号、凭票 Secret 对齐 | +| 换票 | **凭票免登录**:`ticket-exchange`(`YXD_YUHENG_TICKET_SECRET` + `YXD_HOST_KEY` + 手机);**不要**手写登录密码 | +| 短信 | **试运行关闭**:`RequireForBind=false`;`Provider: "off"`(**必须加引号**,否则 platform 起不来) | +| 同号 / 确认 | 命中成员必须弹窗;`confirm=true` 才绑 | +| 通道状态 | 默认同步通道 **默认运行中**;勿依赖用户先点「启动」 | ### 0.2 待改清单(按负责方) -> 明细仍见 §5.7 / §5.9–5.10。智建 Z12/Z13 API 已合入 **`d195aa4`**(凭票、`bind/policy`、试运行关短信);生产须 `./restart.sh --pull` 后才生效。 +> 智建近期提交:`d195aa4` 凭票… → `4222667` SMS 引号 → `cb76824` 默认启动 → (本批)Z10d + fingerprint。生产:`bash ./restart.sh --pull`。 #### A. 宇恒 · 已完成(2026-08-05) | 编号 | 内容 | 落地 | |------|------|------| -| **Z10c** | 空表双侧建齐调用 | `sync_schema.ensure_before_drain`;选「同步」force ensure;drain 前 TTL 节流 | -| **Z13b** | 绑定码兑换 | 表单「绑定码」+ `/database/sync/bind/redeem` | -| **Z13c-1** | 同号弹窗确认 | 技能 `bind_sync` + `/database/sync/bind/phone-confirm`(须 confirm) | -| **Z13c-2** | policy + 凭票 | `GET /database/sync/bind/policy`;试运行免短信;`ticket_exchange`(需 Secret) | -| **Z13d** | 未绑定检测 | 宿主启动 `ensure_sync_bound_interactive` | -| **Z13e** | 选「同步」即用 | `POST /database/sync/mode` → Binding + ensure + agent + full_push | +| **Z10c** | 空表 ensure | `sync_schema.ensure_before_drain` | +| **Z13b–e** | 绑定码 / 弹窗 / policy / 凭票 / 选同步 | 技能 + `/database/sync/bind/*` + `sync/mode` | +| **Z14a** | 表级指纹慢心跳 | `sync_fingerprint.py`;不一致 ensure+full_push | -#### A′. 宇恒 · 仍建议(非阻塞) +#### A′. 宇恒 · 仍建议(非阻塞) / **配合注意** -| 优先级 | 编号 | 待改内容 | 验收要点 | -|--------|------|----------|----------| -| **P1** | **Z12e** | 过渡通道缓存 | 仅 1 条默认通道可写 env;终态以换票 `channel_id` 为准,勿当产品必填 | -| **P1** | **Z7** | Binding 可读名 | ensure / 登记 Binding 时带 `database_name` / `display_name` | -| **P1** | **§2.6** | agent 按库串行 drain | 同一 SQLite remote **勿**多线程齐推;加重试消化锁冲突 | +| 优先级 | 编号 | 内容 | +|--------|------|------| +| **P0 配合** | **数据怎么上云** | 见 **§0.3**:绑定 → 库选「同步」→ agent drain;智建「查看线上表」只反映已 push/ensure | +| **P0 配合** | **online_db_id** | 遇 `_uN` **403** 时:暂用本机 `local_database_id` 作可写 id,或省略 online(有「数据同步」管理权时);等 Z14c | +| **P1** | **Z12e / Z7 / §2.6** | env 过渡;Binding 可读名;按库串行 drain | +| **P1** | **指纹** | 优先调智建 `schema/fingerprint`;未部署则回退 `schema.row_count` | -**推荐调用顺序(未绑定)** +#### B. 智建 · 已完成(生产已验证 / 代码已合入) -```text -GET /auth/bind/policy - → sync_bound=false(换票 / agents/me) - → 有手机:phone-lookup → 弹窗 → phone-confirm 或 ticket-exchange - → 无手机 / 取消:绑定码 redeem - → 成功后库选「同步」→ Binding + drain(Z13e) -``` +| 项 | 说明 | +|----|------| +| gateway 公开 bind/ticket | 生产 health 已通 | +| `SMS.Provider: "off"` | YAML 布尔踩坑已修 | +| 默认同步通道默认启动 | `IsSystemDefault` 创建即 Enabled;重启自动 Start | +| Podman DNS 别名 | compose hostname/aliases | +| 部署脚本可执行位 | `restart.sh` 等 100755 | -#### B. 智建 · 仍待办(运维 / 配置 · **当前生产阻塞**) +#### B′. 智建 · 仍待办 | 优先级 | 项 | 说明 | |--------|----|------| -| **P0** | 生产成员手机 | 「宇信达」成员管理绑定 **`13531041944`**;**勿**绑超管 `13531041945` | -| **P0** | **生产部署含凭票的 gateway** | 须 `./restart.sh --pull` 到含公开路径的版本:`GET /auth/bind/policy`、`POST /auth/yuheng/ticket-exchange`(旧 gateway 会 **401 missing bearer token**) | -| **P0** | 生产通道表白名单 | 形态 B 生产通道 **勿**用「填入测试默认」带 `article` 等演示表白名单;可空(整库 Binding) | -| **P1** | 默认同步 DSN | 生产填 `DBSync.DefaultRemoteDSN`(Postgres);空则仍为 sqlite 联调文件 | -| **P1** | 宇恒凭票 Secret | 生产 `Agent.YuhengTicket.Enabled=true` + Secret;与宇恒 `YXD_YUHENG_TICKET_SECRET` **同一值** | -| **P2** | 正式短信 | 接短信平台后:`SMS.Provider`≠`off` 且 `RequireForBind=true` | +| **P0** | 生产再 pull 本批 | Z10d ADD COLUMN + `schema/fingerprint` + 默认启动;`bash ./restart.sh --pull` | +| **P0** | 成员手机 | 「宇信达」绑 **`13531041944`**;勿超管 `13531041945` | +| **P0** | 通道表白名单 | **勿**「填入测试默认」;形态 B 可空 | +| **P0** | **Z14c** `online_db_id` ACL | ticket/`agents/me` 的 `{channel}_uN` 须与 Binding/JWT 一致,避免 schema/ensure/push **403** | +| **P1** | 凭票 Secret | `YuhengTicket.Secret` ↔ `YXD_YUHENG_TICKET_SECRET` 同值 | +| **P1** | `DefaultRemoteDSN` | 生产 Postgres;空则 sqlite 文件(联调) | +| **P2** | 正式短信 | Provider≠`"off"` 且 `RequireForBind=true` | #### C. 双方已否决 / 不改 | 项 | 说明 | |----|------| -| 终端冲突处理台 | 冲突仅平台超管侧 | -| 智建代改宇恒仓库 | 分工冻结,见 §3 | -| 平台 push 租约 | 正确性靠 version 幂等;并发由宇恒串行 drain | +| 终端冲突台 | 仅平台超管 | +| 智建代改宇恒仓 | 分工冻结 | +| 平台 push 租约 | version 幂等 + 宇恒串行 drain | + +### 0.3 双方怎么配合(开通 → 看到线上表) + +```text +【智建运维】 + 1. bash ./restart.sh --pull(平台 Up,health platform=true) + 2. 成员管理绑 13531041944;通道勿测试表白名单;配凭票 Secret + 3. 启用智能体 → 自动默认同步通道(应显示「运行中」) + 4.(可选)生成绑定码发给终端 + +【宇恒终端】 + 5. sync_bound?否 → policy → lookup/confirm 或 redeem 或 ticket-exchange + 6. 本机库选「同步」→ Binding + ensure + agent drain/full_push(Z13e) + 7. 慢心跳指纹对账(Z14);缺列依赖智建 Z10d + +【智建验收】 + 8. 「查看线上表」刷新 → 应出现业务表(含曾 0 行空表) + 9. 仍空 = 终端未 push,不是「点启动」就能灌表 +``` + +| 角色 | 负责 | 不负责 | +|------|------|--------| +| **智建** | 落点/通道/鉴权/ensure/fingerprint API;控制台查看线上表 | 不替宇恒本机灌历史数据 | +| **宇恒** | 绑定 UX、选同步、outbox、drain、指纹修复、full-push | 不手填 DSN/通道当产品终态 | +| **共同** | 同号 `13531041944`;试运行免短信仍须确认;Secret 对齐 | 不用超管号联调 | 宇恒对照脚本: @@ -111,9 +133,13 @@ GET /auth/bind/policy 3. `POST .../pull`、`POST .../bootstrap` - 下行灌库;响应含 `columns`(空表也返回列,便于本机建表) 4. `POST .../schema`、`POST .../schema/ensure`(**Z10**) - - 拉线上表结构 / 本机空表结构推到线上(`CREATE IF NOT EXISTS`,无需 outbox 行) -5. 鉴权:管理员 JWT(含「数据同步」)、智能体 Token,或**登录用户 JWT**(须本人 Binding + `online_db_id`)均可调 agent API -6. 错误体优先带可读 `message`;忙/上游失败宜带 `retryable: true`(502/503) + - 拉线上表结构 / 本机空表结构推到线上(`CREATE IF NOT EXISTS`;**Z10d** 部署后已存在表亦 `ADD COLUMN`) +5. `POST .../schema/fingerprint`(**Z14** · 建议部署) + - Body:`{ "online_db_id": "..." }` + - 返回各表 `name/row_count/content_hash/pk_column`;与宇恒本机指纹算法对齐(见 §5.11) + - **未部署时**宇恒回退只用 `schema.row_count` 对账(可发现行数差,不能发现「同 count 不同内容」) +6. 鉴权:管理员 JWT(含「数据同步」)、智能体 Token,或**登录用户 JWT**(须本人 Binding + `online_db_id`)均可调 agent API +7. 错误体优先带可读 `message`;忙/上游失败宜带 `retryable: true`(502/503) **联调账号(写入约定,勿再用错租户 / 勿用超管号)** @@ -472,13 +498,14 @@ POST .../schema → result.tables[].name / columns / row_count ``` -列一律按 TEXT + 指定 PK 建空表;已存在幂等跳过。详例见 `宇恒-松离线数据同步使用文档.md`「表结构同步」。 +列一律按 TEXT + 指定 PK 建空表。**Z10d(2026-08-05 晚)**:已存在表不再整表跳过,缺列则 `ALTER TABLE … ADD COLUMN`(`platform/internal/dbsync/ensure_table.go`)。**合入后生产 `./restart.sh --pull` 生效**;未 pull 时对「先按 `_row_id` 建表、后 push 注入 `id`」会持续 `has no column named id`。 **验收** 1. 本机空表 ensure 后,智建「查看线上表」可见同名 **0 行**表。 2. 换机 / 仅线上:schema → 本机建空表 → bootstrap 灌行。 -3. 有数据的表仍走原 push;ensure **不替代** outbox 行同步。 +3. 有数据的表仍走原 push;ensure **不替代** outbox 行同步。 +4. **Z10d**:ensure 带新增列名时,已存在表补列成功,随后 push 含新列不再 503。 ### 5.8 【已落实 Z11】蓝图增字段须迁 Postgres(2026-08-05) @@ -508,12 +535,13 @@ POST .../schema |------|--------|------|------|------| | **Z12a** | **P0** | **换票带回绑定** | `POST /api/v1/auth/token`(及登录)响应含 `channel_id` / `online_db_id` / `database_name` / `sync_bound` | **智建已落实** | | **Z12b** | **P0** | **智能体自查绑定** | `GET /api/v1/agents/me`:智能体 Bearer 可读自己的通道/线上库/状态 | **智建已落实** | -| **Z12c** | **P0** | **启用即自动绑通道** | 创建/启用智能体 → `EnsureSystemDefaultChannel`(`IsSystemDefault`,表白名单可空);写回 `channel_id`+`online_db_id`;DSN 见 `DBSync.DefaultRemoteDSN`(空则 sqlite 联调文件) | **智建已落实** | -| **Z12c-1** | **P0** | **账号级隔离** | Binding / push 校验:用户 JWT 只能写本人 `online_db_id` | **智建已有**(自助 Binding) | +| **Z12c** | **P0** | **启用即自动绑通道** | 创建/启用智能体 → `EnsureAndStartSystemDefaultChannel`(`IsSystemDefault`,表白名单可空,**默认运行中**);写回 `channel_id`+`online_db_id`;DSN 见 `DBSync.DefaultRemoteDSN` | **智建已落实**(含默认启动 `cb76824`) | +| **Z12c-1** | **P0** | **账号级隔离** | Binding / push 校验:用户 JWT 只能写本人 `online_db_id` | **智建已有**(自助 Binding);**Z14c** 仍有 `_uN` 与 JWT 不一致问题 | | **Z12c-2** | **P1** | **公司共享库** | Binding `shared=true`:同租户成员可访问该 `online_db_id`;仅管理员可设 | **智建已落实** | | **Z12d** | **P1** | **SyncPage 展示通道 ID** | 运维可见 + copyable | **智建已落实** | | **Z12e** | **P1** | **唯一默认通道** | 每公司至多 1 条 `is_system_default`;复用已有默认 | **智建已落实** + 宇恒「仅 1 条则写入 env」 | -| **Z12f** | **P1** | **开通 UX** | 启用智能体即可;DSN 用公司级默认 | **智建已落实**(控制台引导文案可再收) | +| **Z12f** | **P1** | **开通 UX** | 启用智能体即可;DSN 用公司级默认 | **智建已落实** | +| **Z12g** | **P1** | **默认同步通道默认启动** | 创建 `Enabled=true`;重启对 `IsSystemDefault` 自动 Start;控制台保存后自动启动 | **智建已落实**(`cb76824`) | **宇恒已做(2026-08-05)** @@ -600,13 +628,56 @@ POST .../schema 6. 未确认 / 他人绑定码,不能绑进别的公司或别人账号。 7. 正式且启凭票:同号走 `ticket-exchange`,不再依赖明文 `attested_same_phone`。 +### 5.11 【新增 Z14 · 2026-08-05 晚】表级指纹对账(慢心跳)+ 联调踩坑 + +> **诉求** +> 仅靠 agent「推送心跳」(pending/pushed)发现不了「本机 96 行、线上 0 行」或「行数相同但内容不同」。 +> 需要**慢周期表级指纹**:`row_count` +(可选)`content_hash`;不一致则 ensure + 对该表 full-push。 + +| 编号 | 优先级 | 诉求 | 现状 | +|------|--------|------|------| +| **Z14a** | **P0** | 宇恒本机指纹 + 对比线上 row_count + 自动修复 | **宇恒已接**:`yxd/app_fastapi/sync_fingerprint.py`;agent 默认 TTL **600s**;`POST /database/sync/fingerprint` | +| **Z14b** | **P1** | 智建返回 `content_hash` | **智建已合入待生产 pull**:`POST /api/v1/agent/sync/channels/{id}/schema/fingerprint` | +| **Z10d** | **P0** | ensure 对已存在表补列 | **智建已合入待生产 pull**;否则 `填土高度` 等缺 `id` 会 503 | +| **Z14c** | **P1** | ticket/`agents/me` 的 `online_db_id` 与 gateway ACL 一致 | **待智建修**;宇恒暂用本机库 id 或省略 online(见 §0.3) | + +**契约摘要(Z14b)** + +```http +POST /api/v1/agent/sync/channels/{id}/schema/fingerprint +{ "online_db_id": "..." } +→ result.tables[]: { name, pk_column, row_count, content_hash } +``` + +哈希约定(双方对齐):`sha256( table+"|"+pk+"|" + Σ json(row, sort_keys=True)+"\n" )` 取前 32 hex。 + +**分工** + +| 方 | 做什么 | +|----|--------| +| 宇恒 | 本机算指纹;优先调 fingerprint API;404 则回退 `schema.row_count`;不一致自动 ensure+`enqueue_full_push`;heartbeat 写入 `fingerprint_*` | +| 智建 | 部署 fingerprint + Z10d;修正 `_uN` ACL(Z14c) | + +**联调踩坑(2026-08-05 · 宇信达生产)** + +1. **数据量对不上**:历史行不会因开通同步自动上云 → 须 full-push;指纹对账可检出差表。 +2. **`填土高度(6标一工区)`**:本机列 `_row_id`,线上曾无 `id`;push 默认 pk=`id` 并注入该列 → `has no column named id`;整队曾因 `mark_error` 未改 status 卡住(宇恒已修)。修复路径:删表重建 / 部署 Z10d 后 ensure 补 `id` 再推。 +3. **`online_db_id=_u3` 403**:ticket/`agents/me` 返回 `{channel}_u3`,但显式带该 id 调 schema/ensure 被拒;省略或用本机库 id 可写。JWT `user_id=1` 与 `_u3` 后缀不一致,属平台 Binding/ACL 问题。 + +**验收** + +1. 本机改行数后 ≤TTL 内指纹检出 mismatch,并排队 full-push。 +2. 未部署 fingerprint 时:仅 count 对账仍可用;`zhijian_fingerprint_api=false`。 +3. 部署后:同 count 不同内容 → `hash_mismatch` → 修复。 +4. Z10d 部署后:缺列 ensure 成功,`填土高度` 可推满。 + --- ## 6. 联系与附件 -- **待改清单(优先看)**:§0.2(宇恒已完成 A / 仍建议 A′ / **智建运维阻塞 B**) +- **待改清单(优先看)**:§0.2;**双方配合**见 **§0.3** - 方案:`松离线-dbsync方案-最终版.md`(含 2026-08-01 联调建议落地记录) - 宇恒使用说明:`宇恒-松离线数据同步使用文档.md`(含 Z10 schema/ensure、Z13 绑定) - 开通说明:`docs/数据同步-开通说明.md` - 绑定策略探测:`GET /api/v1/auth/bind/policy`(`trial_mode` / `require_for_bind`) -- 本意见如与冻结方案冲突,**以冻结方案为准**;§5.1–5.10 为产品增量与复测记录,不推翻 H1–H6 默认无感约束。 +- 本意见如与冻结方案冲突,**以冻结方案为准**;§5.1–5.11 为产品增量与复测记录,不推翻 H1–H6 默认无感约束。