ai_site/docs/数据同步-中间件.md

# 跨库数据同步中间件

支持方言 **Postgres（生产线上库 A 推荐）** / MySQL / SQLite。  
**形态 B（本机 agent）**：终端 agent 经平台 **push** 写线上 A；平台**不直连用户本机 SQLite**。

冲突策略默认 **自动 LWW（源端覆盖）**；落败写入**平台超级管理员**覆盖日志。公司管理员**无冲突台**。

## 权限与隔离

| 项 | 说明 |
|----|------|
| 谁可配通道 | 公司**顶级权限（管理员）**，「数据同步」 |
| Agent push / 拉白名单 | JWT 含「数据同步」（人类管理员或智能体凭证） |
| LWW 覆盖审计 | **仅平台超级管理员**；公司 top → 403 |
| 数据隔离 | 通道带 `tenant_id`；公司 A 看不到公司 B |

## 推荐场景（松离线 B→A，A=Postgres）

| 端 | 角色 |
|----|------|
| **B** | 本机正式库（UUID 主键）；开通且白名单表本地可见 |
| **A** | 线上 **Postgres**（通道 `remote`）；agent 经平台 push 幂等写入 |
| **Agent** | 读本机 outbox → `POST /api/v1/agent/sync/channels/:id/push` |

推荐配置：

1. 通道：`remote.driver=postgres` + DSN；方向 **`local_to_remote`**；策略 **`lww_source`**。  
2. 表白名单须 UUID TEXT/UUID PK + FK 闭包（保存时校验）。  
3. 客户端显式 `local_dbsync`（见开通说明）；未开通用户零感。  
4. 怀疑漏数时点 **同步修复（对账）**（有最小间隔限流）。

### 单服务器 / 多服务器（同一套 API）

| 部署 | 配置 |
|------|------|
| 单服务器 | 一条通道指向该 Postgres；一个 `channel_id` |
| 多服务器 | **每库一条通道**（不同 remote DSN）；客户端按库选择 `channel_id` / Binding |

协议相同，均为「经智建 push」，无需第二套接口。

如何保证**不漏、不多**：

| 手段 | 防什么 |
|------|--------|
| 触发器 / 本机 outbox → agent push | 漏 |
| `WithApplying` / 远端应用不回写 outbox | 多（回声） |
| 目标 meta 同 version 跳过 | 多（重复投递） |
| LWW + 超管覆盖日志 | 并发同 PK |
| 主键对账 reconcile | 漏（存量差） |

## 能力一览

| 项 | 说明 |
|----|------|
| 方言 | 生产 A：`postgres`；兼容 `mysql`；`sqlite` 仅联调临时 A（不宜多 writer） |
| 方向 | 本地→线上（推荐）/ 线上→本地 / 双向 |
| 冲突 | 默认 `lww_source`；`lww_target`；`queue` 仅调试（租户不可见） |
| Agent | `GET .../agent/sync/.../whitelist`；`POST .../push`（管理员/智能体「数据同步」，或登录用户 + 本人 Binding） |
| Binding | `GET/POST /api/v1/admin/sync/bindings`（同上鉴权；普通用户仅本人） |
| 对账 | `POST .../reconcile`（默认最少间隔 300s；仍需「数据同步」） |
| 审计 | `GET /api/v1/platform/dbsync/lww-overrides`（超管） |

持久化：`data/dbsync/channels.json`、`bindings.json`、`lww_overrides.json`（及遗留 `conflicts.json`）。

## 用户自助同步（库级三态）

终端用户可在客户端对名下库选择仅本地 / 仅线上 / 同步；选 **同步** 后用**登录 JWT**（无需单独「数据同步」管理员权）调 agent API。

| 角色 | 范围 |
|------|------|
| 「数据同步」管理员 / 智能体 | 租户内通道；push 不强制 Binding |
| 普通登录用户 | 仅本人 Binding：`channel_id` + `online_db_id`；push **须带** `online_db_id` |

### 表白名单策略（Z4，已冻结）

| 项 | 行为 |
|----|------|
| **权限源** | 登录 JWT + 本人 Binding 的 `online_db_id`（不是通道 `tables`） |
| **push** | 接受 Binding 库内**任意表**；表不存在时**自动建表**（TEXT 列 + PK） |
| **通道 tables** | 仅历史兼容 / whitelist 接口展示；**不拒收** |

管理员不必再为业务表维护白名单。宇恒默认 `YXD_SYNC_ENFORCE_WHITELIST=0`。

### 下行 / 仅线上（Z5）

形态 B：平台**不直连本机 B**，只读线上 A，把行返回给本机 agent 写入 B（建议 `WithApplying` 防回声）。

| mode | 用途 |
|------|------|
| `bootstrap` | 全量分页灌库（「仅线上」首启 / 本地空库） |
| `pks` | 只列 A 上主键，客户端 diff 后再 `rows` |
| `rows` | 按 `row_pks` 取行补齐 |

```http
POST /api/v1/agent/sync/channels/{id}/pull
POST /api/v1/agent/sync/channels/{id}/bootstrap   # 强制 mode=bootstrap
```

```json
{
  "mode": "bootstrap",
  "table": "orders",
  "after_pk": "",
  "limit": 200,
  "online_db_id": "用户自助必填"
}
```

循环直到 `result.has_more=false`。单页默认 200、上限 500。

## Agent push 幂等与重复投递

- **正确性**：同 `(table, row_pk, version)` 已落地 → `result.skipped=true` 且 HTTP 2xx；线上 A **不双行**。  
- **允许**客户端重复 push（例如多 agent / 双 drain）；去重是**客户端优化**（进程内 drain 锁），不是平台必做。  
- 通道 stats：`pushed_applied` / `pushed_skipped` 可观察重推噪声。  
- remote 暂不可达：HTTP **503** + `retryable: true`（agent 可稍后重试，勿换 UUID）。

- **Postgres A** 适合正常并发；若联调仍用 SQLite 作 A，请客户端**按库串行 drain**。
- 平台对 remote 连接做**池化复用 + 通道启动预热**，避免每次 push 冷开 SQLite 触发超时。

## 请求日志怎么查

| 来源 | 位置 / 接口 | 内容 |
|------|-------------|------|
| Platform HTTP | `.runtime/logs/platform.err.log`（go-zero） | 路径、耗时、状态码；超时曾见 `slowcall(3000ms)` |
| Platform dbsync | 同文件，关键字 `dbsync push` / `dbsync pull` | channel、table、pk、user、outcome、dur、`req=`(X-Request-Id) |
| Gateway | `.runtime/logs/gateway.out.log` | sync 路径与 ≥500：`gateway POST /api/... status=... req=` |
| 审计 API | `GET /api/v1/audit/logs`（需「查看审计」） | `action` 含 `dbsync.push` / `dbsync.pull` / `*.error`，detail JSON 带 `req_id` |

联调时用响应头 / 请求头 `X-Request-Id` 可在三处日志对齐同一笔请求。

## 线上 Postgres DSN 示例

```text
postgres://sync_user:pass@127.0.0.1:5432/app_online?sslmode=disable
```

联调临时 SQLite A（不推荐生产）：

```text
file:E:/path/to/remote.db?_pragma=foreign_keys(1)
```

勿用反斜杠；平台会自动规范化并补 `busy_timeout`。

## 控制台用法

1. **数据同步** → 新建通道（remote 选 **postgres**）→ 测试 → 保存 → 启动  
2. 默认策略选 **源端覆盖**；方向首期用 **本地 → 线上**  
3. 点 **同步修复** 做主键对账（勿连续狂点，有限流）  
4. LWW 明细在 **平台工作台**（超管），不在公司同步页  
5. 多服务器：为每台库再新建一条通道，把对应 `channel_id` 发给客户端  

开通与迁移：见 [数据同步-开通说明.md](./数据同步-开通说明.md)、[数据同步-迁移手册.md](./数据同步-迁移手册.md)。

## API

### 公司管理员（「数据同步」）

| 方法 | 路径 |
|------|------|
| GET/POST | `/api/v1/admin/sync/channels` |
| GET/PUT/DELETE | `/api/v1/admin/sync/channels/{id}` |
| POST | `/api/v1/admin/sync/test` |
| POST | `/api/v1/admin/sync/channels/{id}/prepare\|start\|stop` |
| POST | `/api/v1/admin/sync/channels/{id}/reconcile` |
| POST | `/api/v1/admin/sync/channels/{id}/ingest`（写通道 local，偏形态 A） |
| GET/POST | `/api/v1/admin/sync/bindings` |
| GET/POST | `/api/v1/admin/sync/conflicts*` → **403**（已迁超管审计） |

### 本机 Agent / 用户自助

鉴权：Bearer = 「数据同步」**或** 已登录人类 JWT（后者须先 `POST .../bindings` 且 push 带本人 `online_db_id`）。

| 方法 | 路径 |
|------|------|
| GET | `/api/v1/agent/sync/channels/{id}/whitelist` |
| POST | `/api/v1/agent/sync/channels/{id}/push` |
| POST | `/api/v1/agent/sync/channels/{id}/push/batch` |
| POST | `/api/v1/agent/sync/channels/{id}/pull` |
| POST | `/api/v1/agent/sync/channels/{id}/bootstrap` |
| GET/POST | `/api/v1/admin/sync/bindings`（用户自助亦可） |

### 平台超级管理员

| 方法 | 路径 |
|------|------|
| GET | `/api/v1/platform/dbsync/lww-overrides` |
| POST | `/api/v1/platform/dbsync/lww-overrides/{id}/rollback` |

### push 示例

```json
POST /api/v1/agent/sync/channels/{id}/push
{
  "table": "orders",
  "op": "insert",
  "row_pk": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
  "row": { "id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", "title": "x" },
  "version": 1710000000000000000,
  "client_outbox_id": "..."
}
```

## 相关

- [同步表约定.md](./同步表约定.md)  
- [数据同步-开通说明.md](./数据同步-开通说明.md)  
- [数据同步-迁移手册.md](./数据同步-迁移手册.md)