数据同步

1. 页面概览

1.1 是什么

数据同步页面（SyncPage.vue）是 LightFoundry「数据入平台」的关键一环，负责把外部数据源 / 本体对象按计划抽取为平台内的数据集。它补全了"数据入平台"的闭环：数据源接入（数据源页）→ 数据抽取入库（本页）→ 数据集管理（数据集页）→ 质量画像（数据质量页）→ 血缘追踪（数据血缘页）。

页面的核心对象是「同步目标」（Sync Target）：一个同步目标把"某个来源"（平台数据源查询 platform_ds，或本体对象 object）以「全量重写（full）」或「增量水位追加（incremental）」两种模式写入「目标数据集」。用户可以给目标配置 cron 调度表达式实现定时抽取，也可以手动点击「立即运行」触发一次抽取。

同步引擎是 V5 Stage 1（B1-3）的产物，与平台统一调度器（scheduler）和异步任务系统（task manager）深度集成：手动运行走"任务化"模式——后端先落一条 running 运行记录，再提交一个 type=sync_run 的异步任务；前端拿到 task_id 后按 1.5 秒间隔轮询 GET /system/tasks/:id，实时展示进度条，直到任务进入 success / failed / cancelled 终态。这一"立即运行 → 任务化轮询 → 终态"语义是本页区别于普通表单页的核心机制。

1.2 核心价值

1.3 一句话总结

数据同步页面把"外部数据抽取为平台数据集"变成可计划、可观察、可追溯的一等公民操作。

2. 访问入口

2.1 路由与菜单

路由注册见 action/web/src/router/index.js：

{ path: 'sync', name: 'FoundrySync', component: SyncPage, meta: { title: 'Foundry 数据同步', requiresAuth: true } },

页面挂在 FoundryLayout 布局下（子路由 children），侧边栏入口见 action/web/src/views/FoundryLayout.vue 的 menuItems 数组 { path: '/foundry/sync', label: '数据同步' }。接线时路由注册与侧边栏入口由接线 agent 统一处理。

2.2 认证与权限

• 路由 requiresAuth: true：未登录访问会被前端路由守卫拦截并跳转登录页。
• 后端 API 全部挂在 protected 组：authMiddleware 校验 Authorization: Bearer <token>，token 缺失或非法返回 401 {"code":"AUTH_ERROR","error":"..."}。
• Token 类型：aip_token（localStorage 键名 aip_token）。请求拦截器自动读取并附带 Bearer ${token}；401 响应由拦截器统一处理：清理 aip_token / aip_username，若当前不在登录页则 window.location.href = '/login' 跳转登录。
• 角色限制：本页 API 未做细粒度角色门禁（各 handler 直接调用服务），任何登录用户均可操作；但写操作（创建 / 删除 / 启停 / 运行）会经 currentUserID 记录操作者。
• 404 排错提示：若访问 /foundry/sync 显示空白或路由未匹配，检查 Foundry 后端（18081）是否启动、Vite 代理 /api → 18081 是否配置、以及路由是否在 /foundry children 下注册。

2.3 端口与 API 前缀

• Foundry 后端端口：18081。
• API 前缀：/api/v1（syncApi.js baseURL 为 /api/v1，由 Vite 开发服务器代理 /api → Foundry 18081）。
• 任务轮询接口同样走 /api/v1/system/tasks/:id（Foundry 侧任务路由挂 /system 前缀，见 server.go 中 task.RegisterRoutes(protected.Group("/system"), ...)）。
• 数据集下拉辅助接口走 /api/v1/datasets（datasetApi.js，60 秒超时）。

3. 界面布局

+--------------------------------------------------------------+
| 数据同步引擎                                                    |
| 外部数据源/对象按计划抽取为平台内数据集：全量重写（full）与增量水位追加（incremental）。 |
+--------------------------------------------------------------+
| [alert 操作结果提示条（可关闭）]                                    |
+--------------------------------------------------------------+
| [同步任务进度条卡片（job.progress !== null 时显示）]                 |
|  同步任务进度（task: xxx）  [████████████░░░░] 45% · 执行中...      |
+--------------------------------------------------------------+
| [工具栏]  [新建同步目标]        [数据集下拉过滤]  [刷新]              |
+--------------------------------------------------------------+
| [新建向导卡片（showCreateForm 时显示）]                            |
|  第一步：选择目标数据集与来源 / 第二步：写入与调度策略                  |
+--------------------------------------------------------------+
| [目标列表卡片] 同步目标（N）                                       |
|  名称 | 目标数据集 | 来源 | 模式 | 调度 | 水位 | 最近运行 | 状态 | 操作 |
+--------------------------------------------------------------+
| [运行历史抽屉（runsTarget 时右侧滑出）]                             |
|  运行历史 · {目标名}  状态/模式/同步行数/触发/开始/结束/任务            |
+--------------------------------------------------------------+

各板块职责：

• 页头：标题「数据同步引擎」+ 一句功能描述（全量重写与增量水位追加）。
• 操作结果提示条：所有操作（创建 / 运行 / 启停 / 删除）的结果以 alert 呈现，绿色 = 成功（alert-success）、红色 = 失败（alert-error）、蓝色 = 信息（alert-info），右侧「关闭」按钮清空 message。
• 同步任务进度条：任务化运行期间显示任务 ID 与 0~100% 进度条，进度百分比与 message 实时更新；任务终态后 stopPolling 并把 progress 置回 null，卡片消失。
• 工具栏：左侧「新建同步目标」按钮切换新建向导显隐；右侧数据集下拉（按目标数据集过滤列表，切换即触发 fetchTargets）+「刷新」按钮（fetchAll = 数据集 + 目标列表都刷新）。
• 新建向导：两步式表单，第一步选目标数据集与来源（platform_ds / object），第二步配置写入与调度策略。
• 目标列表：表格展示全部同步目标及其来源 / 模式 / 调度 / 水位 / 最近运行 / 启停状态，行内提供「立即运行」「运行历史」「停用/启用」「删除」四个操作。
• 运行历史抽屉：右侧滑出面板（遮罩点击自身可关闭），展示某目标最近的 50 条运行记录，含状态徽标、同步行数、触发来源、起止时间与关联任务 ID。

4. 交互元素详解

4.1 工具栏与列表操作

4.2 新建向导表单

4.3 运行历史抽屉

5. 后端关联

5.1 API 客户端

syncApi.js 为独立 axios 实例：

• baseURL：/api/v1
• timeout：120000（2 分钟；任务化接口 15s 等待窗口内完成会直接返回，超时返回 202/task_id 供轮询）
• 请求拦截器：从 localStorage 读 aip_token，附加 Authorization: Bearer ${token}
• 响应拦截器：401 时移除 aip_token/aip_username 并跳转 /login

导出函数：

datasetApi.js 提供 listDatasets(status='', limit=0, offset=0) → GET /datasets，用于加载数据集下拉（本页调用 listDatasets('') 取全部活跃数据集，取 data.data.datasets）。

5.2 端点表

5.3 响应结构

列表响应（foundry 统一包装 {code:0, data:{...}}）：

{
  "code": 0,
  "data": {
    "targets": [
      {
        "id": "0f8f3b12-...",
        "name": "orders_hourly_sync",
        "dataset_id": "ds-3f1a",
        "source_type": "platform_ds",
        "source_ref": "{\"ds_id\":\"1\",\"query\":\"SELECT order_id, amount, updated_at FROM orders\",\"watermark_col\":\"updated_at\"}",
        "mode": "incremental",
        "schedule": "0 */2 * * *",
        "field_map": "{\"order_id\":\"order_id\"}",
        "enabled": true,
        "last_run_at": "2026-08-30T10:00:00+08:00",
        "watermark_value": "2026-08-30 09:58:32",
        "created_at": "2026-08-30T09:00:00+08:00",
        "updated_at": "2026-08-30T10:00:00+08:00"
      }
    ],
    "total": 1
  }
}

运行接口响应（任务化，POST /sync/targets/:id/run）：

{
  "code": 0,
  "data": {
    "run": { "id": "run-uuid", "target_id": "...", "status": "running", "mode": "incremental",
             "rows_synced": 0, "task_id": "task-uuid", "started_at": "...", "triggered_by": "manual" },
    "task_id": "task-uuid"
  }
}

降级同步执行时 run 已终态（status 为 success / failed），前端据此走"已完成"提示分支。

任务轮询响应（GET /system/tasks/:id）：

{
  "code": 0,
  "data": {
    "id": "task-uuid", "type": "sync_run", "status": "running",
    "progress": 0.45, "message": "写入目标数据集",
    "result": null, "error": "",
    "created_by": "user-id", "created_at": "...", "started_at": "...", "finished_at": null
  }
}

5.4 关联模块表

5.5 关键机制

任务化与轮询（核心）：手动运行 POST /sync/targets/:id/run 时，后端 RunSync 先落一条 fs_sync_runs（status=running），再向 task.Manager 提交 type=sync_run 的任务；15 秒等待窗口（taskWaitTimeout）内完成则直接返回结果 + task_id（快路径，前端看到的是"已完成"提示而非进度条），超时则前端经 GET /system/tasks/:id 轮询。前端 startPolling 用 setInterval(..., 1500) 每 1.5 秒调 getTask，读取 progress（0~1）与 message 更新进度条，直到 status ∈ {success, failed, cancelled} 判定终态，stopPolling 并刷新列表。单次轮询请求失败（网络抖动）静默忽略、下一轮继续，不打断任务。

full 模式：单事务内 DELETE 目标行表全部旧行后逐行参数化 INSERT 重写，保证目标与来源一致。

incremental 模式：源查询被包装为 SELECT * FROM ( <query> ) WHERE watermark_col > ?，水位值参数化绑定（可解析为数值时按数值绑定，规避 SQLite 数值/文本比较失真）；首次运行无水位时全量拉取并建立水位；成功后推进 watermark_value。配置了 field_map 时按幂等键（目标列，业务主键语义）冲突跳过，未配置时以水位过滤本身作为防重机制。

调度注册：enabled+schedule 的目标注册 sync:{targetID} 到统一调度器，cron 触发走内部同步执行（不经任务系统，保持调度器防重入语义），运行记录 triggeredBy=cron。创建目标时注册失败（如非法 cron）会回滚删除目标行。

列名安全白名单：schema 列 / watermark 列 / field_map 列一律过 ^[a-z][a-z0-9_]*$ 白名单校验，非法即拒绝（validateColumnName），不做静默改写；全部行值参数化绑定，watermark 值不内联拼 SQL。

6. 核心流程详解

6.1 创建同步目标

1. 点击「新建同步目标」展开向导（showCreateForm = true）。
2. 填写目标名称、选择目标数据集（下拉来自 GET /datasets，展示 名称（来源类型，v当前版本））。
3. 选择来源类型：

• platform_ds：填写数据源 ID（ds_id）、来源查询 SQL（增量模式再填水位列 watermark_col，提示文案提示增量模式自动包装 SELECT * FROM (查询) WHERE 水位列 > 上次值，建议带 LIMIT）；
• object：填写对象 ID（提示 object 来源需平台注入对象读取器，且仅支持 full 模式）。

4. 第二步选择同步模式 full / incremental、填写 cron 调度（可空）、可选 field_map JSON、勾选「创建后立即启用」。
5. 点击「创建目标」：前端先本地 parseFieldMap 解析 field_map（非法 JSON 会提示「field_map 解析失败：...」且不提交），组装 payload 后 POST /sync/targets。
6. 成功后提示「同步目标已创建」，收起表单、清空名称/调度/field_map 字段，刷新目标列表。
7. 后端 CreateTarget 校验：name 非空、dataset_id 非空且目标数据集存在未归档、source_type/mode/source_ref/field_map 白名单校验；enabled+schedule 时注册调度条目，注册失败回滚删除刚插入的目标行并返回错误。

6.2 立即运行（任务化主流程）

1. 点击行内「立即运行」（busy 期间按钮禁用）。
2. 前端先 stopPolling() 清掉上一次轮询残留，再 POST /sync/targets/:id/run。
3. 响应解析（handleRun）：

• 若 run.status !== 'running'（降级同步执行，已终态）：提示「同步已完成（success/failed，N 行）」，刷新列表，流程结束；
• 若返回 task_id：设置 job.taskId、job.progress = 0、job.message = '任务排队中...'，提示「同步任务已提交，进度轮询中（task: xxx）」，启动 startPolling(t.id)；
• 若既无终态 run 也无 task_id：提示「同步已触发」，刷新列表。

4. 轮询循环（startPolling，每 1.5s）：

• getTask(job.taskId) → t = data.data → 更新 progress 与 message；
• 判断终态 ['success', 'failed', 'cancelled'].includes(t.status)：

• success：提示「同步任务成功」，stopPolling、刷新列表、若运行历史抽屉正打开则 fetchRuns 刷新；
• failed/cancelled：提示「同步任务结束（failed/cancelled）：错误详情」，stopPolling、刷新列表；

• 轮询请求异常被 catch 静默（// 轮询失败不打断（网络抖动容错），连续失败由用户手动刷新）。

5. 页面卸载（onBeforeUnmount）调用 stopPolling，防止内存泄漏。

6.3 状态机与任务终态语义

任务状态机（platform/task）：

queued ──→ running ──→ success
   │           │          │
   │           └──→ failed
   └──(cancel)─→ cancelled

• queued：任务已入队，等待 worker 取走（前端 message 显示「任务排队中...」）；
• running：worker 正在执行，progress 0~1 递增，message 为阶段描述（如「加载数据集」「拉取源数据」「写入目标数据集」「更新数据集元数据」「血缘与质量画像」）；
• success：执行成功，result 回写，fs_sync_runs 终态 success + rows_synced 落库；
• failed：执行失败，error 携带失败详情，fs_sync_runs 终态 failed；
• cancelled：任务被取消（queued 直接取消，running 走 context 取消传播）。

前端把 ['success', 'failed', 'cancelled'] 视为终态集合，命中即 stopPolling。运行记录（fs_sync_runs）状态机与任务大体平行：running → success / failed。

6.4 启停 / 删除 / 运行历史

• 停用 / 启用：PUT /sync/targets/:id 传 {enabled: !enabled}，成功提示「目标已启用/停用」，刷新列表；停用会反注册调度条目，启用则按当前 schedule 重新注册（syncSchedule 幂等）。
• 删除：window.confirm('确认删除同步目标「xxx」？运行历史将保留。') 二次确认后 DELETE /sync/targets/:id，成功提示「同步目标已删除」；后端硬删目标、保留 fs_sync_runs 运行历史、反注册调度条目。
• 运行历史：点击「运行历史」打开抽屉并 fetchRuns（limit=50），表格按 started_at 倒序展示；抽屉底部若 runsError 非空显示「最近失败：...」。点击「关闭」或遮罩（@click.self）关闭并清空。

7. 权限与安全

7.1 认证

• JWT Bearer 认证（aip_token）；401 统一跳转登录页（前端拦截器）。
• 后端 authMiddleware 双路径：lfk_ 前缀走 API Key 校验（SHA256 查库，校验 active/未过期），否则走 securityService.ParseToken 解析 JWT；成功注入 user_id/username/roles 到 gin 上下文。

7.2 数据级安全

本页面向"行表写入"场景，同步写路径的安全口径是列名白名单 + 参数化绑定：

• 列名（schema 列 / watermark 列 / field_map 列）一律过 ^[a-z][a-z0-9_]*$ 白名单校验（validateColumnName），非法即拒绝，不做静默改写（防注入语义可测）；
• 全部行值经参数化绑定（? 占位），watermark 值不内联拼 SQL（经 Connect 句柄 Raw 参数绑定执行）；
• 目标行表名经 dataset.RowTableName 纯标识符构造，物理类型映射白名单（TEXT/REAL/INTEGER）；
• 目标数据集已归档（archived）不可作为同步目标、不可同步，返回明确错误「目标数据集已归档...」。

7.3 写操作防护

• 创建/更新/删除/运行均记录操作者（currentUserID，缺省 anonymous）到日志；
• 运行历史记录 triggered_by（manual / cron），便于审计"谁在何时触发"；
• 删除有前端 window.confirm 二次确认，且保留运行历史（可回溯）；
• 目标 enabled 字段显式落库（无 GORM default 标签），避免建行即启用的隐式行为。

8. 常见问题与排错

8.1 任务卡在「任务排队中...」进度条不动

• 现象：点击「立即运行」后，进度条一直停留在 0%，message 长期为「任务排队中...」。
• 原因：task.Manager worker 池繁忙或任务队列积压；或 taskMgr 未装配（此时应看到"同步已完成"的降级提示而非进度条）。
• 排查步骤：

1. 打开浏览器 DevTools Network，确认 GET /system/tasks/:id 返回 200 且 status 仍为 queued；
2. 若 queued 长期不变，说明 worker 未取走任务——检查 Foundry 进程是否卡顿、worker 并发是否被其他长任务（如管道 run）占满；
3. 观察后端日志是否有「同步任务化失败，降级同步执行」告警（此告警说明走了降级路径）；
4. 若确认任务系统死锁，可重启 Foundry 进程（taskMgr.Restore 会把重启前僵尸 running 任务标记 failed、queued 任务按类型注册表重派）。

8.2 增量同步运行成功但一行都不写入

• 现象：incremental 模式运行成功（success）但 rows_synced 为 0。
• 原因：水位推进逻辑——上次运行已把 watermark_value 推到当前最新值，本次查询 WHERE watermark_col > 上次值 无新数据；或 field_map 幂等键冲突把本批行全部跳过。
• 排查步骤：

1. 打开运行历史抽屉，查看该目标最近几次运行的「同步行数」与列表「水位」列；
2. 确认源表新增行的 watermark 列值是否大于目标记录的 watermark_value；
3. 若配置了 field_map 且键值与已存在行重复，属预期冲突跳过——检查是否误把业务主键配置成幂等键；
4. 若源表确实新增但未同步，检查来源查询 SQL 是否包含 watermark 列、列名与水位列一致（列名不一致会在运行失败分支报「源查询结果缺少 watermark 列」）。

8.3 运行失败，错误信息含 SQL 执行错误

• 现象：点击「立即运行」后任务终态 failed，错误详情含 SQL 执行错误 / 连接失败。
• 原因：来源查询 SQL 语法错误、数据源连接失败（如连接器白名单未放行）、或目标数据集 schema 与源结果列无映射。
• 排查步骤：

1. 查看错误详情（任务终态提示或运行历史「最近失败」）区分是连接问题还是 SQL 问题；
2. 先在数据源页对该数据源点「测试连接」，确认连接可达（注意 connector 工厂默认仅放行 MYSQL/SQLSERVER/POSTGRESQL，其余类型需 AllowAdvancedTypes 显式授权）；
3. 将来源 SQL 拿到数据库客户端手工执行，确认可运行且结果含所需列；
4. 确认目标数据集 schema 列名与源结果列名可映射（列名差异可用 field_map 显式指定 {"源列":"目标列"}），否则报「源结果列 ... 与目标 schema 列 ... 无可映射列」。

8.4 删除目标后旧任务仍在轮询

• 现象：删除同步目标后，之前提交的任务还在页面进度条上轮询。
• 原因：删除只反注册调度条目并删除 fs_sync_targets 行，已提交的异步任务不会随目标删除而撤销。
• 排查步骤：确认任务最终会自然终态（success/failed）；若需中断，可调用 POST /system/tasks/:id/cancel 取消（queued 直接取消，running 走 context 取消传播）；前端重进页面或刷新后轮询自然停止。

9. 已知缺陷与边界

10.2 后端文件

• action/products/foundry/sync/rest.go（路由与 handler）
• action/products/foundry/sync/models.go（fs_sync_targets / fs_sync_runs 表结构）
• action/products/foundry/sync/service.go（CRUD / 调度 / 任务化 / doSync / 列名白名单）
• action/platform/task/models.go、rest.go、manager.go（异步任务系统与 /system/tasks 端点）
• action/products/foundry/server/server.go（路由组装与接线）

10.3 项目文档

• action/wiki/upgrade-v5/dev-story/stage-1.md（B1-3 同步引擎设计）
• action/wiki/changelog/2026-08-29-2-v5-stage1.md（Stage 1 变更记录）
• action/wiki/upgrade-v5/UPGRADE-PLAN.md（§Batch 1 计划）
• action/wiki/frontend-intro-v5/markdown/foundry/index.md（页面索引）

10.4 相邻页面

• 数据集（datasets） — 同步落数的目标数据集管理
• 数据源（data-sources） — 同步来源 platform_ds 的数据源接入
• 数据血缘（lineage） — 同步成功自动打点 source→dataset 血缘边
• 数据质量（quality） — 同步成功链触发的质量画像
• SQL 工作台（sql-workbench） — 数据查询与物理翻译