流事件规则（CDC 事件化与规则投递）

1. 页面概览

1.1 是什么

「流事件化规则」页面（页面内标题为 流事件化规则）是 LightFoundry 的 CDC 事件 → 规则条件求值 → 执行动作 的自动化闭环管理界面，V5 Stage 3（B3-4）交付。页面描述把整条链路浓缩为：「CDC 事件 → 规则条件求值（{{change_type}}/table/pk_value/payload）→ 执行动作（webhook/通知/同步/画像/写回）。」

在数据集成场景里，数据库的表变化（插入、更新、删除）会被 CDC（Change Data Capture）检测器捕获并发布到平台事件总线（platform/cdc），每条事件带有 change_type（insert/update/delete/snapshot）、table_name、pk_value、payload 等信息。业务上常常需要「当订单表插入一条新记录时立刻通知主管」「当某张表的金额超过阈值时触发一次数据同步」「当订单状态变为已发货时把状态写回另一套系统」这类实时响应。本页面就是配置这些响应规则的地方。

一条规则 = 绑定一个 CDC watcher（数据源）+ 一个条件表达式 + 一个动作。规则启用后即订阅该 watcher 的全部事件；事件到达时后端对条件求值，命中则执行动作并落一条投递记录，未命中则不产生任何记录。页面同时提供「测试触发」——构造一条模拟事件直接走完整处理流水线（不经过事件总线订阅），以及「投递历史」抽屉——查看每次动作执行的 event_id、状态与错误。

1.2 核心价值

1.3 一句话总结

2. 访问入口

2.1 路由与菜单

• 路由：/foundry/stream
• 路由名称：FoundryStream
• 路由 meta：title: 流事件规则，requiresAuth: true，挂在父路由 /foundry（FoundryLayout）下
• 菜单位置：Foundry 左侧边栏「流事件规则」（FoundryLayout 菜单项）
• 前端源码：action/web/src/views/StreamPage.vue
• API 客户端：action/web/src/api/streamApi.js

2.2 认证与权限

• 页面路由挂 requiresAuth: true，未登录访问重定向到 /login。
• 数据请求走 streamApi 实例：baseURL /api/v1，请求拦截器自动附带 Authorization: Bearer <aip_token>；响应拦截器 401 时清理 aip_token / aip_username 并跳转 /login。
• streamApi 用 unwrap() 解析统一响应体 {code, data}：code !== 0 时抛出 body.error || body.message，页面在 alert 中展示。
• 后端 /stream/* 与 /cdc/watchers 均挂 protected 语义，未带有效令牌返回 401。
• Watcher 列表（GET /cdc/watchers）不可用时（如 CDC 未接线）前端不阻断规则管理，仅 watchers 下拉为空。

2.3 端口与 API 前缀

• Foundry 后端端口：18081。
• API 前缀：/api/v1（streamApi 的 baseURL）。
• 完整请求示例：GET /api/v1/stream/rules、POST /api/v1/stream/rules/:id/test。

3. 界面布局

页面为单栏布局，从上到下依次为工具栏、规则编辑表单（条件显示）、规则列表、测试触发面板（条件显示）、投递历史抽屉（条件显示）：

┌────────────────────────────────────────────────────────────────┐
│ 流事件化规则（页头 + 描述）                                       │
│ [alert 操作结果提示条（可关闭）]                                   │
│ 工具栏：[新建规则]                        [刷新]                  │
├────────────────────────────────────────────────────────────────┤
│ [新建/编辑规则表单，v-if form.show]                               │
│  规则名称 * | Watcher（CDC 检测源）*                               │
│  条件表达式（可空=恒真；{{var}} 语法）                              │
│  动作类型 * | 启用（保存后立即订阅该 watcher 事件）                  │
│  动作配置 action_config（JSON）*   [保存] [取消]                   │
├────────────────────────────────────────────────────────────────┤
│ 流规则（N）  表格：名称 / Watcher / 条件 / 动作 / 状态 / 触发次数 / │
│  最近触发 / 操作（测试|投递|编辑|停用|启用|删除）                    │
├────────────────────────────────────────────────────────────────┤
│ [测试触发：规则名，v-if testRule]                                  │
│  change_type / table_name / pk_value / payload（JSON 摘要文本）    │
│  [触发测试]  命中= 状态= （skipped/error）                         │
├────────────────────────────────────────────────────────────────┤
│ [投递历史：规则名（N），v-if deliveryRule]                         │
│  表格：event_id / 状态 / 错误 / 时间        [关闭]                 │
└────────────────────────────────────────────────────────────────┘

各板块职责：

• 工具栏：「新建规则」打开编辑表单；「刷新」同时重载规则列表与 watcher 列表。
• 新建/编辑规则表单：录入规则名称、绑定 watcher、条件表达式、动作类型与动作配置 JSON、启用开关；「保存」提交到后端并自动同步订阅。
• 流规则列表：全部规则的汇总表，行内可直接「测试」「投递」「编辑」「停用/启用」「删除」。
• 测试触发面板：构造模拟事件（change_type/table_name/pk_value/payload），走完整处理流水线验证规则；展示命中、状态、跳过原因或错误。
• 投递历史抽屉：某条规则的全部投递记录，含 event_id、状态（success/failed）、错误、时间。

4. 交互元素详解

4.1 工具栏

4.2 新建/编辑规则表单

4.3 规则列表

4.4 测试触发面板

4.5 投递历史抽屉

5. 后端关联

5.1 API 客户端

• 文件：action/web/src/api/streamApi.js
• baseURL：/api/v1；超时 30000ms；Content-Type: application/json
• 拦截器：请求自动附带 Authorization: Bearer <aip_token>；响应 401 清除令牌并跳登录
• unwrap()：解析 {code, data}，code !== 0 抛 body.error || body.message
• 导出函数：listStreamRules / getStreamRule / createStreamRule / updateStreamRule / deleteStreamRule / testStreamRule / listStreamDeliveries / listWatchers，以及常量 ACTION_TYPE_LABELS

5.2 端点表

5.3 响应结构

规则对象（rules 数组元素，{code:0, data:{rules:[...]}}）：

{ "id": "...", "name": "订单插入→通知主管", "watcher_id": "w_orders",
  "condition": "{{change_type}} == 'insert' && {{table}} == 'orders'",
  "action_type": "webhook", "action_config": "{\"url\":\"http://host:8080/hook\"}",
  "enabled": true, "last_fired_at": "2026-08-30T...", "fire_count": 3,
  "created_at": "...", "updated_at": "..." }

投递记录（deliveries 数组元素）：

{ "id": "...", "rule_id": "...", "event_id": "evt_xxx", "status": "success",
  "error": "", "created_at": "2026-08-30T..." }

测试触发结果（data 直接为 TestResult）：

{ "rule_id": "...", "event_id": "evt_auto_xxx", "matched": true,
  "status": "success", "error": "", "skipped": "" }

5.4 关联模块表

5.5 关键机制

• 订阅生命周期：Start(ctx) 对每条 enabled 规则执行 bus.Subscribe(watcherID, handler)；handler 在事件到达时重查规则（防陈旧配置/已删除/已停用）再处理；CRUD 内部自动同步订阅；SyncSubscription 支持外部直改库后手动重载。
• 条件求值：eventVars 把 payload 字段顶层展开，并保留 payload / change_type / table / watcher_id / watcher_name / datasource_id / pk_value / event_id / occurred_at；EvalCondition 求值失败记日志不报错，未命中不产生投递记录。
• 幂等防重入：先落一条 status=failed、error="pending" 的投递记录占位，(rule_id, event_id) 复合唯一索引冲突时跳过（幂等重放）；动作执行后回写终态 success/failed。
• 动作语义：webhook 为 HTTP POST JSON（超时 10s、失败重试 1 次、请求前 SSRF 校验 fail-closed）；notification 为审计落点 + 通知渠道（未注入降级为审计+投递记录不报错）；sync_run / quality_run 依赖注入接口（未注入记录不可用）；writeback 经 writepath Executor VALIDATE_AND_EXECUTE 走全五步流水线，幂等键 stream:<rule>:<event>。
• 触发计数：命中并执行后规则表 fire_count + 1、last_fired_at 更新（best-effort，失败不阻断投递结果）。

6. 核心流程详解

6.1 主流程：建规则 → 订阅 → 事件到达 → 求值 → 执行 → 落投递

1. 新建规则：点「新建规则」，填写规则名称、选择 watcher（需先在 CDC watcher 数据源存在，watchers 由 cdc REST 创建）、写条件表达式、选动作类型并填 action_config JSON（可先切换动作类型自动填充模板）、勾选启用，点「保存」。
2. 订阅生效：保存后前端提示「规则已创建（已订阅事件源）」；后端对 enabled 规则 bus.Subscribe(watcherID, handler)，此后该 watcher 的全部事件都会进入规则。
3. 事件处理：事件到达 → handler 重查规则（确保仍启用）→ EvalCondition 求值 → 命中则幂等 claim 占位 → 执行动作 → 回写投递终态 → 更新规则触发计数。
4. 验证与追溯：用「测试」构造模拟事件直接走处理流水线（不落订阅）验证条件与动作；用「投递」抽屉查看每次执行的 event_id、状态与错误。

6.2 条件表达式要点

• 变量引用：{{change_type}}、{{table}}、{{pk_value}}、{{payload}}，payload 内字段顶层展开可直接 {{count}}。
• 运算符：== != > < && || !（注意 HTML 中需写 > < &&）。
• 示例：{{change_type}} == 'insert' && {{table}} == 'orders'；条件留空表示恒真（每条事件都触发）。
• 求值失败：后端记 warn 日志并跳过该事件（不产生投递记录），不会误触发动作。

6.3 分支流程：测试触发

1. 点规则行的「测试」打开面板（预填 change_type=insert）。
2. 修改 change_type、table_name、pk_value、payload。
3. 点「触发测试」：前端组装 {watcher_id, table_name: table_name || watcher_id, change_type, pk_value, payload} 提交。
4. 结果展示：命中=true 状态=success（绿）/ 命中=false 状态=（条件未命中，未触发动作） / 命中=true 状态=failed 错误=xxx（红）/ 幂等重放时 skipped 提示。
5. 测试会真实执行动作并落投递记录（webhook 会真的发起 HTTP 请求），测试后规则 fire_count 会增加。

6.4 投递记录语义

• 只有条件命中才产生投递记录；未命中无记录（抽屉空态文案明确说明）。
• status 只有 success / failed 两种终态；执行中的占位记录 error="pending" 理论上只在崩溃残留时可见。
• 同规则同事件重复触发（幂等重放）被跳过，不产生第二条投递记录，也不重复执行动作。

7. 权限与安全

• 认证：全部端点经 authMiddleware（JWT，aip_token），未授权 401。
• webhook SSRF 防护：所有 webhook 触发路径（订阅触发与 TestRule 模拟触发）均经 validateWebhookTarget 校验目标 URL 与 method，拒绝内网/私网/云元数据等目标，fail-closed。
• 动作依赖注入：sync_run / quality_run / writeback 依赖后端注入的接口，未注入时记录不可用（failed + 明确错误），不会静默失败。
• 写操作防护：删除规则用 window.confirm 二次确认且提示「投递历史保留」；停用/启用直接提交。
• 幂等安全：(rule_id, event_id) 唯一索引防止重复执行动作，重放/重复投递安全。

8. 常见问题与排错

9. 已知缺陷与边界