mirror of
https://github.com/sinanyuntu/trade-message-center.git
synced 2026-09-17 13:22:11 +08:00
Merge pull request #44 from sinanyuntu/09-11-websocket-routing-single-source-of-truth
统一 WebSocket 路由单一事实来源
This commit is contained in:
YBF
@@ -100,7 +100,7 @@ type OneTalkServiceWorkerFrameRouter = {
|
||||
### 4.3 Contracts
|
||||
|
||||
- Bright transport 只负责 socket/reconnect、decode、scope guard、`ws.hello`/heartbeat、`ws.accepted`/generic `ws.error` 状态转换、诊断和 guarded `send(frame)`；不得公开 sync/profile/buyer/send-confirmation 专用 `send*` facade，也不决定业务 frame consumer。
|
||||
- router 以静态表单播到 owner：anchor 与四类 ACK 到 sync、`contact.profile.ack` 和精确 `profile_observed_at_future` 到 profile、`buyer.facts.ack` 到 buyer、`send.command` 到 send Flow。connection-only frame 不进入 router；没有 event bus、多播、自动注册或 fallback owner。
|
||||
- router 以一张 typed executable table 单播到 owner：该表自身同时定义本地业务 type、runtime narrowing 和 owner delegation，不能另写平行 literal union、key list 或 `switch` selector。anchor 与四类 ACK 到 sync、`contact.profile.ack` 和精确 `profile_observed_at_future` 到 profile、`buyer.facts.ack` 到 buyer、`send.command` 到 send Flow。connection-only frame 不进入 router；没有 event bus、多播、自动注册或 fallback owner。
|
||||
- router 必须在调用 sync 前将 `OneTalkFrame` 收窄到 `OneTalkSyncServerFrame`。sync engine 保留已有 runtime guards/ACK 次序，但不能重新实现 general-frame routing。
|
||||
- outbound Flow 只能用 `@trade-message-center/onetalk-contract` 的 canonical creator/type 构造业务 frame，再调用 transport `send(frame)`；不得复制 decoder/schema、绕开 guarded send、创建 outbox 或对未确认发送自动重试。
|
||||
- send Flow 将 `send.command` 交给现有 `routePageCommand`，保持 `channelAccountId + conversationId` 精确路由，page `requestId` 与 `sendRequestId` 分离，并只回写 `confirmed_sent`、`rejected_before_send` 或 `delivery_unknown` 的 canonical confirmation。页面路由异常只能收敛为 `delivery_unknown/send_state_lost`。
|
||||
|
||||
@@ -71,6 +71,62 @@ app.get("/health", async () => ({ status: "ok" }));
|
||||
|
||||
The health boundary is stable and secret-free; database probing belongs in a later operational contract。
|
||||
|
||||
## Scenario: OneTalk authenticated WebSocket route metadata
|
||||
|
||||
### 1. Scope / Trigger
|
||||
|
||||
- Trigger：新增、迁移或重构已认证 OneTalk client frame 的 endpoint ownership、authorization 或生产分发。
|
||||
- Scope：`websocket/authenticated-router.ts` 拥有 non-hello route metadata；`endpoint-routes.ts` 用它校验固定 endpoint 的 per-socket handler table 并分发。`handler.ts`/`protocol-kernel.ts` 继续拥有 decode、固定 endpoint/scope gate 和 socket FIFO；业务 Flow、registry 与 pending-send coordinator 继续拥有状态及最终副作用。
|
||||
- Excluded：不改 shared wire/decoder、`ws.hello` bootstrap、connection registry 或业务 Flow 的授权/commit/ACK/publish 次序。
|
||||
|
||||
### 2. Signatures
|
||||
|
||||
```ts
|
||||
createOneTalkAuthenticatedRouter() -> OneTalkAuthenticatedRouter
|
||||
createOneTalkEndpointAuthenticatedRouter(endpoint, router, handlers) -> { dispatch(frame): Promise<void> }
|
||||
defineOneTalkEndpointRouteHandler(type, handler) -> OneTalkEndpointRouteHandler
|
||||
```
|
||||
|
||||
### 3. Contracts
|
||||
|
||||
- `ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS` 是每个 non-hello client frame 的 endpoint ownership、`session | pending_send` authorization 与 operation 的唯一声明；必须从 `ONETALK_CLIENT_FRAME_TYPES` 完整校验 missing、duplicate、unknown 与 `ws.hello`。
|
||||
- `installWebsocket` 只构造一次 shared router 并注入 Plugin/Mind endpoint。每个已建立 socket 都从同一 metadata 建立 fixed-endpoint view；view 的 handler table 必须与该 endpoint 的可拥有 type 完全相等，不能有 no-op/fallback handler 或另一份 endpoint type list。
|
||||
- `dispatch` 在调用业务 collaborator 前检查 frame connection type 与 endpoint ownership。正常生产顺序保持 `decode -> admission -> fixed endpoint/scope -> FIFO -> endpoint router -> existing handler`；router 不得重新解析 raw data、创建第二个 session/FIFO 或把 `pending_send` 路径改为通用 session authorization。
|
||||
- `ws.hello` 永不注册为 authenticated route；它继续由 endpoint bootstrap 处理。
|
||||
|
||||
### 4. Validation & Error Matrix
|
||||
|
||||
| 条件 | 必须行为 |
|
||||
| --- | --- |
|
||||
| metadata 缺少、重复、未知或包含 `ws.hello` | router construction 失败 |
|
||||
| endpoint handler 缺少、重复或不归属该 endpoint | endpoint view construction 失败 |
|
||||
| 合法 client frame 发往另一个 fixed endpoint | kernel 在业务 collaborator 前发 `ws.error(scope_mismatch)` 并 close `1008` |
|
||||
| `send.request` / `send.confirmation` | 仍走原 pending-send coordinator；不得改为 session authorization |
|
||||
| 同 endpoint 的非 client/未认证/重复 hello | 保持 kernel 既有 error/close 优先级，不进入 router |
|
||||
|
||||
### 5. Good / Base / Bad Cases
|
||||
|
||||
- Good：Plugin 的 handler table 从 shared metadata 验证后只把收窄 frame 委派给既有 sync/profile/buyer/confirmation Flow；Mind 的 `send.request` 仍委派现有 request Flow。
|
||||
- Base：`heartbeat` 是唯一双 endpoint route，但两个 endpoint 各有自己的 authorization adapter 与 handler closure。
|
||||
- Bad：在 Plugin/Mind endpoint 保留 `if (frame.type === ...)` 作为第二份 owner selector，或只在启动期检查独立 `ONETALK_*_ROUTE_TYPES`。
|
||||
|
||||
### 6. Tests Required
|
||||
|
||||
- router tests 必须从 canonical metadata 断言全量 non-hello coverage、metadata failure、exact endpoint handler table 与 wrong-ownership collaborator 零调用。
|
||||
- `injectWS` 必须发送合法 Mind frame 到 `/ws/plugin` 和合法 Plugin frame 到 `/ws/mind`，断言 `scope_mismatch`、close `1008`，并观察目标业务 collaborator 为零调用。
|
||||
- 保留 full client-frame wire matrix 与 pending confirmation、profile/buyer post-write fence、observation ACK/publish、sync completion flush 的 focused tests；重构不能以私有 router 调用替代这些 wire 断言。
|
||||
|
||||
### 7. Wrong vs Correct
|
||||
|
||||
```ts
|
||||
// Wrong: endpoint type list and production selector can drift independently.
|
||||
if (frame.type === "send.request") return sendRequestFlow.handleRequest(socket, frame, epoch);
|
||||
|
||||
// Correct: the fixed endpoint dispatches an exact handler table validated by shared metadata.
|
||||
const router = createOneTalkEndpointAuthenticatedRouter("mind_page", metadata, handlers);
|
||||
return router.dispatch(frame);
|
||||
```
|
||||
|
||||
## Scenario: Bright v3 authorization and commit fences
|
||||
|
||||
### 1. Scope / Trigger
|
||||
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
{"file":".trellis/spec/project/architecture.md","reason":"检查唯一事实来源、入口组合和依赖方向。"}
|
||||
{"file":".trellis/spec/project/module-organization.md","reason":"检查分发、endpoint 与业务 Flow 没有职责回流。"}
|
||||
{"file":".trellis/spec/server/backend/quality-guidelines.md","reason":"执行服务端 strict TypeScript、构建、测试和格式化门禁。"}
|
||||
{"file":".trellis/spec/server/backend/error-handling.md","reason":"检查稳定 WebSocket 错误、close 语义与 wire matrix。"}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/quality-guidelines.md","reason":"执行扩展严格类型、测试、构建和格式化门禁。"}
|
||||
{"file":".trellis/tasks/09-11-websocket-routing-single-source-of-truth/research/server-routing.md","reason":"核对服务端 route ownership、error/close及 Flow fence。"}
|
||||
{"file":".trellis/tasks/09-11-websocket-routing-single-source-of-truth/research/client-routing.md","reason":"核对客户端单播、特殊错误与 optional owner 行为。"}
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
# 设计：可执行的端点路由声明
|
||||
|
||||
## 目标边界
|
||||
|
||||
本设计只替换“已认证、已通过固定 endpoint 和 session/scope gate 的 client frame 如何选择 endpoint 业务 handler”这一层。`protocol-kernel.ts` 与 `handler.ts` 继续拥有 raw WebSocket decode、cutover、endpoint/session gate、FIFO、诊断和 close；Plugin/Mind endpoint 继续拥有 hello、授权适配以及业务 Flow 的构造；各 Flow、registry 和 pending coordinator 继续拥有其可变状态与最终动作。
|
||||
|
||||
```text
|
||||
decoded raw frame
|
||||
→ ProtocolKernel (decode, admission, endpoint, session/scope, FIFO)
|
||||
→ endpoint callback
|
||||
→ endpoint-bound authenticated Router (metadata ownership + typed handler)
|
||||
→ existing Plugin/Mind Flow
|
||||
```
|
||||
|
||||
## 服务端单一来源
|
||||
|
||||
在 `authenticated-router.ts` 保留并强化一份静态 authenticated route metadata：每个非 `ws.hello` client type 恰一次，包含 allowed endpoint(s)、`authorization` 和 `operation`。`heartbeat` 是唯一双 endpoint route；Plugin 业务 frame 与 Mind `send.request` 各只有一个 owner；`send.request` 与 `send.confirmation` 保留 `pending_send`。
|
||||
|
||||
Router construction 负责由 contract client type 集合校验 metadata 的 missing/duplicate/unknown/hello 边界。`endpoint-routes.ts` 基于这同一份 metadata 生成固定 endpoint view，并接收该 endpoint 在当前 socket 下已绑定的 typed handler table；它必须校验 handler table 与 endpoint 可拥有的 route 集合完全一致。`dispatch` 先校验 endpoint ownership 与 frame connection type，再调用唯一 handler，不创建 no-op/fallback handler。
|
||||
|
||||
`websocket/index.ts` 在安装端点前构造并校验 shared metadata Router，并把它显式注入 Plugin/Mind handler options。由 per-socket callback 绑定 Flow closure 后构造 endpoint view，`onAuthenticatedFrame` 只调用 view 的 `dispatch`。因此启动期验证、endpoint ownership 和运行时选择都观察同一对象/声明；旧 `ONETALK_PLUGIN_ROUTE_TYPES` 与 `ONETALK_MIND_ROUTE_TYPES` 删除，不能保留为第二清单。
|
||||
|
||||
## endpoint handler 适配
|
||||
|
||||
Plugin/Mind route handler table 的函数只将当前已有分支提取为已收窄 frame 的委托，保留原有公共授权段与 Flow 调用顺序：
|
||||
|
||||
- Plugin `heartbeat` 仍执行 heartbeat authorization/liveness/ack；普通 session routes 仍走原 binding/scope/version/permission/policy/canonical guard；各 discovery、observation、completion、profile、buyer handler 继续委托原 Flow/service。
|
||||
- Plugin `send.confirmation` 与 Mind `send.request` 直接保持既有 pending-send Flow/coordinator 路径，不以 metadata 把它们改为普通 session authorization。
|
||||
- Mind `heartbeat` 继续用 Cookie session capability 或 authorization reader 及现有 error-only/close 语义。
|
||||
|
||||
`ws.hello`、decoder-invalid、wrong fixed endpoint、unauthenticated、scope mismatch、same-scope non-client unknown frame 都在 Router 前结束，故本重构不得调整其输出或优先级。route metadata 是 selection/ownership 的 canonical owner，不迁移业务 authorization、commit guard 或 timer/state。
|
||||
|
||||
## 客户端收敛
|
||||
|
||||
`frame-router.ts` 仅维护一份 typed executable business route definition。该定义同时提供业务 frame type、narrowed frame handler与其单播 owner，删除独立 `RoutedFrame` literal union及重复 `switch`。如 TypeScript 对 correlated union 需要 helper，该 helper 必须在 discriminant 检查后调用窄 handler，不能将宽 `OneTalkFrame` 断言为具体类型。
|
||||
|
||||
`ws.error` 继续作为表外的精确 `profile_observed_at_future` 分支；transport lifecycle、other server/Mind-page frame 与 default 保持 no-op。Service Worker composition roots、Bright transport 与各 owner 的 public types不变。
|
||||
|
||||
## 风险与回滚
|
||||
|
||||
最高风险是将 selection 重构误变成业务行为重写：尤其 pending-send、Plugin commit/authorization fence 和 Mind heartbeat Cookie capability。实现必须先以当前 injectWS wire matrix characterization，再以真实 output/close/collaborator calls 验证新路径；不得只测试私有 route function。
|
||||
|
||||
若出现 wire output、close、ACK/publish 次序或 authorization 回归，回滚只恢复 route wiring 与局部 tests，不触碰 contract、Flow、registry 或持久化。此结构变更不要求迁移、feature flag 或部署切换。
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
{"file":".trellis/spec/project/architecture.md","reason":"约束路由、共享类型和入口的唯一所有权。"}
|
||||
{"file":".trellis/spec/project/module-organization.md","reason":"约束协议内核、Router、endpoint 与业务 Flow 的职责边界。"}
|
||||
{"file":".trellis/spec/project/async-state-boundaries.md","reason":"保护认证、commit guard、ACK 与 publish 的 await fence。"}
|
||||
{"file":".trellis/spec/server/backend/index.md","reason":"服务端包边界、质量基线与 OneTalk 领域规范索引。"}
|
||||
{"file":".trellis/spec/server/backend/error-handling.md","reason":"保护 WebSocket 稳定错误、关闭和真实 wire characterization。"}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/type-safety.md","reason":"约束 Service Worker 的严格 frame narrowing 和类型所有权。"}
|
||||
{"file":".trellis/tasks/09-11-websocket-routing-single-source-of-truth/research/server-routing.md","reason":"当前服务端生产调用图、route matrix 和回归风险。"}
|
||||
{"file":".trellis/tasks/09-11-websocket-routing-single-source-of-truth/research/client-routing.md","reason":"当前客户端路由重复点、owner 边界和测试矩阵。"}
|
||||
+44
@@ -0,0 +1,44 @@
|
||||
# 实施计划：收敛 WebSocket 路由分发
|
||||
|
||||
## 1. 确立服务端 canonical metadata 与 endpoint view
|
||||
|
||||
- 修改 `apps/server/src/websocket/authenticated-router.ts` 与 `endpoint-routes.ts`，让同一份静态 metadata 完整覆盖 contract 的 non-hello client frames，并能绑定每个 fixed endpoint 的 typed handler table。
|
||||
- 删除/替换独立 endpoint route type lists；在 `apps/server/src/websocket/index.ts` 的 composition root 构造/验证 shared Router 后显式注入两端。
|
||||
- 保持 factory 构造失败条件、endpoint ownership guard 与 `isOneTalkAuthenticatedClientFrame` production predicate；不改 `protocol-kernel.ts`、contract、registry 或 pending coordinator。
|
||||
|
||||
## 2. 让生产 endpoint 通过 Router 分发
|
||||
|
||||
- 在 `apps/server/src/websocket/plugin/index.ts` 将认证帧的 type-to-owner branches 变为 endpoint-bound route handlers；保留已有 common authorization、Flow/service 调用、await/fence、ACK/publish和错误处理。
|
||||
- 在 `apps/server/src/websocket/mind/index.ts` 对 heartbeat/send request 做同样的 endpoint-bound handler wiring，保留 Cookie session capability 与 pending-send path。
|
||||
- 审计生产 endpoint `onAuthenticatedFrame` 不再维护平行 route type list/selector；Flow 内按 payload、业务结果或已收窄 frame 的分支可保留。
|
||||
|
||||
## 3. 收敛客户端局部 dispatcher
|
||||
|
||||
- 修改 `apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts`，由一张 typed executable table 完成 frame narrowing和 owner delegation；保留特殊 future-skew `ws.error` 分支、optional owner no-op及其它 no-op。
|
||||
- 不改 Bright transport、composition subscriptions、sync/profile/buyer/send owner 的入参或共享 contract。
|
||||
|
||||
## 4. 测试与结构审计
|
||||
|
||||
- 更新 `apps/server/test/authenticated-router.test.ts`，使用生产 metadata/endpoint route builder 验证全量、ownership、metadata mismatch及 collaborator 不调用。
|
||||
- 更新 server WebSocket regression，真实 `injectWS` 覆盖 `/ws/plugin` 收到合法 Mind `send.request` 与 `/ws/mind` 收到合法 Plugin `conversation.discovered` 时的 `scope_mismatch` + close `1008` + 零业务调用；保留全量 client frame wire matrix和各 Flow focused test。
|
||||
- 更新 `apps/chrome-extension/test/onetalk-frame-router.test.js`，锁住八类 frame 的单播、future-skew精确处理、generic/transport no-op与 optional owner no-op；不以另一份产品 literal list测试实现细节。
|
||||
- 静态审计 production source 只由 canonical metadata派生 endpoint view/dispatch，并确认 client router 不再重复 union/switch route list。
|
||||
|
||||
## Validation
|
||||
|
||||
按风险由小到大执行：
|
||||
|
||||
1. `pnpm --filter @trade-message-center/server exec node --experimental-strip-types --test test/authenticated-router.test.ts test/websocket.test.ts test/onetalk-websocket.test.ts`
|
||||
2. `pnpm --filter @trade-message-center/chrome-extension exec node --experimental-strip-types --test test/onetalk-frame-router.test.js`
|
||||
3. `pnpm --filter @trade-message-center/server typecheck` 与 `pnpm --filter @trade-message-center/chrome-extension typecheck`
|
||||
4. `pnpm --filter @trade-message-center/server build` 与 `pnpm --filter @trade-message-center/chrome-extension build`
|
||||
5. `pnpm --filter @trade-message-center/server test` 与 `pnpm --filter @trade-message-center/chrome-extension test`；服务端每次测试运行必须以 60 秒为上限。
|
||||
6. `pnpm format:check`、`git diff --check`，并以 GitNexus `detect_changes` 确认只影响预期 WebSocket/extension routing flow。
|
||||
|
||||
真实 Chromium、数据库、Mind HTTP、TLS/NGINX 不在本次重构的自动化验收范围；若执行，需使用隔离 binding/account，并单独记录为 runtime evidence。
|
||||
|
||||
## Review Gates
|
||||
|
||||
- 编码前：对每个将修改的 Router/endpoint/client symbol 重跑 GitNexus upstream impact；出现 HIGH/CRITICAL 先报告再继续。
|
||||
- 编码后：检查没有 fallback/no-op 处理不支持的 server endpoint route，没有宽类型断言，没有 duplicate route metadata，没有改变 Flow 的 state owner或 final action次序。
|
||||
- 提交前：Trellis independent check、GitNexus detect_changes、完整 validation与 diff review。
|
||||
@@ -0,0 +1,58 @@
|
||||
# 收敛 WebSocket 路由分发单一事实来源
|
||||
|
||||
## Goal
|
||||
|
||||
让 OneTalk 服务端的已认证 WebSocket 路由声明同时成为端点归属校验和生产帧分发的唯一事实来源；同时收敛 Chrome Extension Service Worker 入站业务帧的重复类型清单。这样新增或迁移 frame 不会只更新启动期列表或某个手写分支而导致二者漂移。
|
||||
|
||||
## Background
|
||||
|
||||
当前不是“两套 Router 同时运行”：`authenticated-router.ts` 的 factory 与 `endpoint-routes.ts` 的 endpoint view 只由 `authenticated-router.test.ts` 调用；生产仅在 `websocket/index.ts` 启动时用 `ONETALK_PLUGIN_ROUTE_TYPES` 与 `ONETALK_MIND_ROUTE_TYPES` 做集合校验，随后由 Plugin/Mind endpoint 中的 `onAuthenticatedFrame` 手写条件分发业务 frame。`isOneTalkAuthenticatedClientFrame` 仍是 `handler.ts` 的生产 type predicate，不能误删或误称为未使用。
|
||||
|
||||
客户端的 `frame-router.ts` 只有一张实际执行的 `routes` 表，但 `RoutedFrame`、表的 key 和 `switch` 三次维护同一组八个业务 type；`ws.error/profile_observed_at_future` 是独立异常路由，必须保留精确匹配。
|
||||
|
||||
## Requirements
|
||||
|
||||
### R1 — 服务端路由的唯一事实来源
|
||||
|
||||
- 已认证 client frame 的 metadata 必须从 `ONETALK_CLIENT_FRAME_TYPES`（排除 `ws.hello`）完整覆盖，并明确每个 frame 的 endpoint ownership、`session`/`pending_send` 授权类别和 operation。
|
||||
- 生产 Plugin/Mind endpoint 必须通过该声明式 Router 选择并调用本端业务 handler；不能只把 Router 包在现有 `if` 外面，同时继续以 `if`/`switch` 维护 type-to-owner 映射。
|
||||
- 端点类型集合校验必须从同一份可执行声明派生，或被其直接替代。不得继续独立维护 `ONETALK_PLUGIN_ROUTE_TYPES`、`ONETALK_MIND_ROUTE_TYPES` 与生产分发规则。
|
||||
- `ws.hello` 继续由 endpoint handshake/bootstrap 处理，永不注册为 authenticated route。
|
||||
|
||||
### R2 — 行为与所有权保持
|
||||
|
||||
- 不改变 wire protocol version、共享 contract/decoder、公开 frame、错误码、close code/reason、诊断边界或 `/ws/plugin`、`/ws/mind` URL。
|
||||
- 保持 protocol kernel 的 decode → admission → fixed endpoint → session/scope → authenticated-frame 分类顺序及单 socket FIFO；不得由 Router 重新解析原始数据、创建第二份 session/FIFO 或绕过共享 handler。
|
||||
- 保持 Plugin/Mind 的现有授权细节、pending-send coordinator、registry/canonical connection、commit guard、Flow 的 await fence 与 commit → ACK → publish 顺序。特别是 `send.request`/`send.confirmation` 的 `pending_send` 路径不得被通用 session 授权替代。
|
||||
- 收到另一个 fixed endpoint 的合法 frame 时，必须在业务 collaborator 前维持现有 `scope_mismatch` + close `1008` 行为；未知、未认证及重复 hello 的现有优先级不变。
|
||||
|
||||
### R3 — 客户端本地路由收敛
|
||||
|
||||
- `frame-router.ts` 以一份 typed、可执行的本地业务路由定义完成 type narrowing 和单播；不把协议 direction list 当成本地 owner 列表，也不扩大 `OneTalkSyncEngine` 等 owner 的窄输入类型。
|
||||
- 保持 optional profile/buyer/send owner 缺失时的 no-op、transport/lifecycle frame 的 no-op，以及只有 `profile_observed_at_future` 交给 profile coordinator 的 `ws.error` 特例。
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- 变更 `apps/onetalk-contract` 的 frame 类型、payload、方向、版本或 decoder。
|
||||
- 修改 OneTalk registry、pending-send、同步/profile/buyer/observation Flow 的状态所有权、持久化、业务时序或授权策略。
|
||||
- 增加动态路由注册、fallback route、事件总线或新的网络/数据库/浏览器行为。
|
||||
- 以启动服务、连接存在或浏览器加载替代真实的 WebSocket 行为断言。
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] 服务端生产链路 `createApp → installWebsocket → fixed endpoint → protocol kernel → authenticated Router → endpoint handler` 可由真实 WebSocket 测试证明；声明式 route factory 不再仅在合成单元测试中执行。
|
||||
- [ ] 任一非 hello client frame 的 endpoint ownership、授权 metadata 与生产 dispatch 均来自同一份声明；缺失、重复、未知或 `ws.hello` entry 在构造/注册时失败。
|
||||
- [ ] 现有全量 client-frame wire matrix 继续覆盖 contract 的全部 client type，且公开输出、错误与 close 语义不变；新增 wrong-path 真实 WebSocket 回归断言 `scope_mismatch`、close `1008` 与零业务 collaborator 调用。
|
||||
- [ ] Plugin 的 pending confirmation、profile/buyer post-write fence、observation ACK/publish、sync completion flush，及 Mind send request 行为仍由原有 focused tests 锁住。
|
||||
- [ ] 客户端八类业务帧仍各只委派给一个 owner；generic `ws.error`、transport-only 和 Mind-page frame 不调用 owner；optional owner 缺失不抛错。
|
||||
- [ ] 严格 TypeScript、相关 source/compiled tests、format check、build 与 `git diff --check` 按实施计划通过，或记录与本变更无关的明确阻碍。
|
||||
|
||||
## Confirmed Facts
|
||||
|
||||
- `createOneTalkAuthenticatedRouter`（`apps/server/src/websocket/authenticated-router.ts:102`）与 `createOneTalkEndpointAuthenticatedRouter`（`apps/server/src/websocket/endpoint-routes.ts:42`）当前只有 router tests 调用；生产 `websocket/index.ts:66` 仅调用集合 validator。
|
||||
- 真实 server dispatch 在 `plugin/index.ts:273-390` 与 `mind/index.ts:125-156`；共享 `protocol-kernel.ts:96-165` 在此前完成端点、会话、scope 与 FIFO 门禁。
|
||||
- 客户端重复点在 `apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts:10-80`；两个生产 composition root 只订阅同一 router 的 `handle`。
|
||||
|
||||
## Open Questions
|
||||
|
||||
无阻塞产品或兼容性问题。技术形状以 `design.md` 的单一 metadata + endpoint-bound handler table 为准；若实现发现它无法在严格 TypeScript 下维持 discriminated-union narrowing，必须在不扩大 owner 入参、不用无依据断言的前提下调整 helper 形状。
|
||||
+92
@@ -0,0 +1,92 @@
|
||||
# 研究：Chrome Extension Service Worker frame-router 重复路由类型
|
||||
|
||||
- 查询：核查 `frame-router.ts` 的 `RoutedFrame`、`routes`、`switch` 及所有生产调用者，确认 OneTalk wire contract、Bright transport、sync/profile/buyer/send owner 和相关测试；提出不改变协议行为的最小收敛方案。
|
||||
- 范围：internal
|
||||
- 日期：2026-09-11
|
||||
- 协调类别：package-local
|
||||
- 阻塞：Chrome Extension Service Worker frame-router 清理及其 focused tests；不阻塞服务端 Router 研究，但服务端生产分发与客户端收敛应由主会话分别综合。
|
||||
- 共享边界：`apps/onetalk-contract/src/wire.ts` 的 `OneTalkFrame`/协议类型是只读共享边界；本研究不建议修改 contract。客户端内部共享边界是 `frame-router.ts` 暴露的 `OneTalkServiceWorkerFrameRouter.handle(OneTalkFrame)`，以及各 owner 的窄 `handle*` 接口。
|
||||
- 证据基线：checkout `/Users/ybf/code/trade-message-center-worktree`，分支 `main`，HEAD `f81d5bcb9d4567f09863f652f9b4338f1f243a72`；工作树只有本任务下未跟踪的 `prd.md`、`task.json`、`implement.jsonl`、`check.jsonl`，目标源码与测试无 dirty path。已刷新 source anchors：`frame-router.ts`、`sync-runtime.ts`、`configured-sync-session.ts`、`sync-engine.ts`、`bright-client.ts`、`onetalk-contract/src/wire.ts` 及相关测试。
|
||||
- 复用证据与缺口：没有发现本任务已有 client-routing 研究文件；复用了 Chrome OneTalk runtime 记忆作为仓库定位提示，并以当前 checkout/source/test/GitNexus 结果重新核验。GitNexus impact（`createOneTalkServiceWorkerFrameRouter`，upstream，含 tests）为 LOW：直接调用者 2 个、受影响 Service-worker module 1 个、受影响 execution processes 2 个。未执行真实 Chromium、Bright WebSocket、数据库或生产 ZIP 运行时探针。
|
||||
|
||||
## 发现
|
||||
|
||||
### 1. `frame-router.ts` 当前有三处生产路由类型维护
|
||||
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts:10-23` 的 `RoutedFrame` 手写了 8 个业务帧类型：`anchor.snapshot`、4 类 sync ACK、`contact.profile.ack`、`buyer.facts.ack`、`send.command`。
|
||||
- 同文件 `:36-47` 的 `routes` mapped object 再以 8 个 object keys 维护同一集合，并把每个 key 绑定到一个唯一 owner：前 5 个给 `options.sync.handleServerFrame`，profile/buyer ACK 分别给对应 coordinator，`send.command` 给 send Flow。optional owner 缺失时当前行为是 optional chaining 后静默不处理，不能在收敛时改变成 fallback 或广播。
|
||||
- 同文件 `:48-73` 的 `switch (frame.type)` 第三次重复 8 个 literal，并只负责调用同名 `routes` entry 后 return；它没有第二套业务逻辑，只是手动做从 `OneTalkFrame` 到各具体 frame type 的 narrowing。
|
||||
- `:74-81` 的 `ws.error` 是独立特殊分支：仅当 code 为 `ONETALK_ERROR_CODES.profileObservedAtFuture` 时转给 profile；其它错误和 default 均忽略。它不属于上述 8 个业务路由，不能因删除 switch 而丢失，也不应把 generic `ws.error` 交给 profile。
|
||||
|
||||
### 2. 生产调用者和 owner 边界
|
||||
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/sync-runtime.ts:91-96` 创建 router（sync、可选 profile、send），并在该 composition root 对 Bright `subscribe` 只挂一个 `router.handle` listener。
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/configured-sync-session.ts:274-282` 在配置会话中创建 router（sync、可选 profile、buyer、send），Bright listener 另有 `currentRevision === this.revision` fence；`:284-301` 的 stale revision cleanup 会取消该 listener。
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/sync-engine.ts:78-88` 定义 `OneTalkSyncServerFrame`，只包含 anchor 和四类 ACK；`:179-181` 的公开 `handleServerFrame` 接收该窄类型，`:238-274` 的 `handleBrightFrame` 继续拥有 anchor/ACK 状态机、checkpoint、ACK 和 bootstrap side effects。router 重构必须继续在调用 sync 前完成相同 narrowing，不能把 `OneTalkFrame` 直接暴露给 sync。
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/flows/send-command-flow.ts:10-22` 的 send owner 只接受 `send.command` frame；`:36-95` 负责页面精确路由、page-result 收窄和 confirmation，不应被 router 重写或合并。
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/contact-profile-coordinator.ts:46-54` 与 `buyer-fact-coordinator.ts:207-256` 分别拥有 profile/buyer ledger、ACK 和状态；router 只做单播委派，不拥有它们的可变状态。
|
||||
|
||||
### 3. Bright transport 已拥有生命周期处理，router 不应扩大范围
|
||||
|
||||
- `apps/chrome-extension/src/onetalk/service-worker/transport/bright-client.ts:487-520` 完成 inbound decode、诊断、plugin scope 校验并通知 subscribers；`:521-545` 自己处理 `ws.accepted`、generic `ws.error`、重连/关闭策略。
|
||||
- 因此 `ws.accepted`、`heartbeat.ack`、`plugin.status`、`sync.status`、`conversation.updated`、`message.created`、`send.result` 等 transport-only 或 mind-page frames 不应进入业务 router。当前 dedicated test `apps/chrome-extension/test/onetalk-frame-router.test.js:46-69` 已验证这些帧不调用任何 owner。
|
||||
- `apps/onetalk-contract/src/wire.ts:43-71` 是完整 `ONETALK_FRAME_TYPES`，`:74-105` 是 client/server direction lists，`:137-164` 是 `OneTalkFrame` union。它们是协议事实源；客户端 router 的 8 帧只是本地消费投影，不能复制或改写 contract lists。
|
||||
|
||||
### 4. 现有测试覆盖与缺口
|
||||
|
||||
- `apps/chrome-extension/test/onetalk-frame-router.test.js:10-44` 覆盖 8 个业务帧各调用一次且只调用一个 owner，并覆盖 `profile_observed_at_future` 的特殊错误路由；`:46-69` 覆盖 connection-only/mind-page frames 被忽略。
|
||||
- `apps/chrome-extension/test/onetalk-sync-runtime.test.js:424-443` 覆盖 standalone runtime 创建后只有一个 frame listener，dispose 后 listener/status listener 清空并 disconnect Bright；这保护 composition root 的订阅生命周期，但不校验每个 owner 的 production composition 路由。
|
||||
- `apps/chrome-extension/test/onetalk-configured-sync-session.test.js:240-305` 通过真实 `OneTalkConfiguredSyncSession` 测试 `send.command` 页面路由和 confirmation；`:307-337` 测试 profile coordinator 被配置并在 authenticated 后发送，但没有直接断言 profile ACK/future error 经 session router 到达 profile。
|
||||
- `apps/chrome-extension/test/onetalk-sync-engine.test.js:233-240` 的测试 helper 为 engine 单独创建 router 并订阅 FakeBright；这是测试组合，不是第二个产品路由实现。同步状态机本身从 `:410` 起使用 `anchor.snapshot`、`:490-519` 使用 ACK，现有行为测试依赖 router 驱动。
|
||||
- 当前没有静态测试断言“业务 route 类型只定义一次”，也没有针对 optional profile/buyer/send owner 缺失时路由不抛异常的 focused case。收敛后应补/调整 focused test，使测试矩阵保护行为而不是复制生产的三份 literal list。
|
||||
|
||||
## 候选 Scope 与依赖
|
||||
|
||||
### 推荐 package-local scope：收敛 Service Worker inbound router
|
||||
|
||||
- 语义职责：让 `frame-router.ts` 只保留一份可执行声明式业务路由定义，并由该定义驱动分发；保留 `OneTalkServiceWorkerFrameRouter.handle(OneTalkFrame)` public shape、owner 单播、optional owner no-op、`ws.error` future-skew 特殊分支和现有调用顺序。
|
||||
- 拥有路径：`apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts`；focused behavior tests `apps/chrome-extension/test/onetalk-frame-router.test.js`。必要时只更新测试 fixture/断言，不修改协议 contract、transport、sync/profile/buyer/send Flow。
|
||||
- 排除路径：`apps/onetalk-contract/src/*`、`apps/server/src/websocket/*`、`bright-client.ts`、`sync-engine.ts` 及 owner 内部状态机；它们属于其它功能闭环或已验证边界。
|
||||
- 所需接口：`OneTalkFrame` 继续作为 router ingress；sync handler 继续只接 `OneTalkSyncServerFrame`，profile/buyer/send 继续使用各自窄 frame handler。若采用 typed helper，必须在 router 内完成 discriminant narrowing，不得让 owner 接受宽泛 `OneTalkFrame`。
|
||||
- 实际并行因素：该 scope 与服务端 Router 收敛在源码上 write-disjoint、runtime-isolated、无共享写入，可并行研究/实现；但共同验收需要确认不会改变跨端协议 type 集合。客户端 scope 不需要等待服务端实现，服务端不应因客户端重构而改 contract。
|
||||
|
||||
### 最小实现方向（供 implementer 选择具体 TypeScript 形状）
|
||||
|
||||
1. 以一个 typed declarative route definition 作为业务路由唯一事实来源；其 key/type parameter 同时提供 route key 与具体 frame narrowing。移除独立 `RoutedFrame` literal union 和逐项 `switch`，避免继续维护三份集合。
|
||||
2. 通过 keyed lookup 或 typed dispatch helper 调用同一张表；如果 TypeScript 的 correlated-union 索引需要辅助函数，辅助函数应接受经过运行时 discriminant check 的具体 frame，不能用无校验的 `as never` 把宽类型强行传给 owner。
|
||||
3. 将 `ws.error` future-skew 保留为表外 transport-error 特殊分支，或者显式作为与业务路由不同类别的单一特殊定义；必须保持 generic error 不触 profile 的行为。connection-only/default 继续 no-op。
|
||||
4. 不要把 `ONETALK_SERVER_FRAME_TYPES` 直接当作 router routes：该协议列表包含 transport/mind-page 帧，且 router 的 owner mapping 不是 wire direction 的同义词。router 只能维护本地“哪些 inbound business frames 由哪个 owner 消费”的投影。
|
||||
|
||||
## Invariant 与验收探针
|
||||
|
||||
### 必须成立的 invariant / owner
|
||||
|
||||
- `OneTalkFrame` 是 wire contract 唯一事实源，由 `apps/onetalk-contract/src/wire.ts:137-164` 拥有；本 task 不新增第二份跨包协议列表。
|
||||
- `frame-router.ts` 是 inbound business-frame dispatcher 唯一 owner；每个业务 type 最多一个 owner，且每次 `handle` 最多一次委派。路由表不拥有 sync/profile/buyer/send 的状态、重试、ACK 或页面 side effect。
|
||||
- `OneTalkSyncEngine.handleServerFrame` 的窄 `OneTalkSyncServerFrame` 是 sync ingress invariant（`sync-engine/model.ts:78-88`）；router 负责 narrowing，sync engine 负责既有状态机。
|
||||
- Bright transport 是 lifecycle/error owner（`bright-client.ts:487-545`）；router 仅消费 subscriber 帧，不能接管 accepted/error/reconnect/heartbeat 状态。
|
||||
- 两个 composition root 各只拥有一个 router subscription，并在 dispose/revision replacement 后取消旧 listener（`sync-runtime.ts:91-107`、`configured-sync-session.ts:274-301`）。
|
||||
|
||||
### Static / unit probes（不启动服务，不产生运行时外部状态）
|
||||
|
||||
- `node --experimental-strip-types --test apps/chrome-extension/test/onetalk-frame-router.test.js`：当前基线已通过（2 tests）；重构后继续断言 8 个业务帧分别单播，`profile_observed_at_future` 仅到 profile，generic `ws.error`、transport-only、mind-page frame 全部不调用 owner。
|
||||
- 新增/调整 focused matrix：对 profile、buyer、send owner 缺失分别发送 `contact.profile.ack`、`buyer.facts.ack`、`send.command`，断言无 throw、sync owner 仍不受影响；对重复同 type 断言每次仅一次调用。测试可保留行为期望列表，但不要再让产品源码存在第二份路由 literal 列表。
|
||||
- `pnpm --filter @trade-message-center/chrome-extension typecheck`：验证 typed table/dispatch 仍将 5 个 sync frame narrowing 到 `OneTalkSyncServerFrame`，且不会通过宽类型或无依据 assertion 绕过严格检查。
|
||||
- `pnpm --filter @trade-message-center/chrome-extension test`：覆盖 contract build 后的扩展 focused suite；重点回归 `onetalk-sync-engine.test.js` 的 anchor/ACK 状态机、`onetalk-configured-sync-session.test.js` 的 send/profile flow 和 `onetalk-sync-runtime.test.js` 的 subscription disposal。
|
||||
- `pnpm format:check` 与 `git diff --check`：验证 Oxfmt/空白；再用 `rg` 审计 `frame-router.ts` 不再出现独立 `switch` route literal 或第二份业务 type union（特殊 `ws.error` literal 除外）。
|
||||
- 结构静态审计：确认只有 `sync-runtime.ts:96` 和 `configured-sync-session.ts:280-282` 两个生产 composition subscription，未新增 `bright.subscribe((frame) => sync.handleServerFrame(...))`、event bus、多播或 fallback owner。
|
||||
|
||||
### Runtime probes（当前未执行，需隔离）
|
||||
|
||||
- Real Chromium/extension probe：加载与当前构建匹配的 unpacked extension，建立真实 Bright authenticated plugin connection，观察 anchor/ACK、profile future-skew、buyer ACK 和 send command 的 owner side effect；要求独立测试账号、页面和可回收测试数据，不能用启动 server 代替证据。
|
||||
- Bright WebSocket probe：使用隔离的 test binding/account，逐类发送合法 decoded plugin frames，确认 transport lifecycle 状态与 router business delivery 顺序；不应触碰生产 binding/DB。
|
||||
- Runtime evidence 只能证明当前构建产物的行为；源码测试、typecheck 和 build 不能代替 Chromium/网络验证。当前报告没有声称这些探针通过。
|
||||
|
||||
## 注意事项 / 未找到内容
|
||||
|
||||
- GitNexus index 对 router 的符号有 Function/Const 双候选；使用明确 Function UID 后得到 LOW upstream impact，直接 caller 为 `createOneTalkServiceWorkerSyncRuntime` 和 `OneTalkConfiguredSyncSession.configure`。不要把索引列出的跨 server execution flow 当成客户端 caller。
|
||||
- 当前 `routes` object 本身已经是实际唯一执行表；问题是 type alias 与 switch 对它的重复描述，不是客户端存在两套独立业务路由。重构验收应聚焦“单一声明 + 行为保持”，避免夸大为两个同时运行的 Router。
|
||||
- `ws.error` future-skew 是 profile ledger 的显式异常通道；删除 switch 时最容易误删它，或错误地把所有 `ws.error` 转给 profile。必须保留 code 精确匹配和 generic error no-op。
|
||||
- 不建议把 route keys 提升到 `src/lib` 或 `onetalk-contract`：它们包含 Service Worker owner 语义，且扩展规范要求渠道/上下文领域概念留在 `src/onetalk`（`.trellis/spec/chrome-extension/frontend/architecture.md`）。
|
||||
- 不建议把当前 optional owner 的 no-op 改成抛错、默认 owner 或自动注册；standalone runtime 当前只组合 sync/profile/send，而 configured session 才可组合 buyer，这是已有产品行为边界。
|
||||
- 真实 browser/WebSocket/DB/production-bundle 证据尚未获得；本次仅完成 source、static test、GitNexus impact 和规范核查。
|
||||
+163
@@ -0,0 +1,163 @@
|
||||
# 研究：服务端 WebSocket 路由生产调用图
|
||||
|
||||
- 查询：核对 `authenticated-router.ts`、`endpoint-routes.ts`、`websocket/index.ts`、`plugin/index.ts`、`mind/index.ts`、`handler.ts`、`protocol-kernel.ts`、OneTalk contract decoder 与服务端 WebSocket/router 测试；追踪两个 Fastify WebSocket endpoint 的真实生产调用与每个 client frame 的公开输出。
|
||||
- 范围：internal
|
||||
- 日期：2026-09-11
|
||||
- 协调类别：cross-cutting
|
||||
- 阻塞：服务端路由收敛实现；实现前需确定“路由元数据/所有权”与 per-socket Flow handler 的接口，以及保留 `ws.hello`、session authorization、pending-send coordinator 的边界。Plugin/Mind endpoint writer 必须等待这个共享 route contract；业务 Flow 的时序和 state owner 不需要重设计。
|
||||
- 共享边界：`apps/server/src/websocket/index.ts`（唯一 Fastify 安装/组合点）、`handler.ts`/`protocol-kernel.ts`（共享传输、scope/endpoint gate、FIFO）、`authenticated-router.ts`/`endpoint-routes.ts`（当前声明式路由与校验）、`apps/onetalk-contract/src/wire.ts`/`decoder.ts`（frame 类型和方向事实源）、`registry.ts`/`pending-send-coordinator.ts`（connection/pending-send 唯一 owner）。
|
||||
- 证据基线：checkout `/Users/ybf/code/trade-message-center-worktree`，branch `main`，HEAD `f81d5bcb9d4567f09863f652f9b4338f1f243a72`（2026-09-11）；相关 dirty path 只有当前任务目录 `.trellis/tasks/09-11-websocket-routing-single-source-of-truth/`，无产品源码 dirty path。本次刷新了上述 server source、contract source、`authenticated-router.test.ts`、`websocket.test.ts`、`onetalk-websocket.test.ts` 和 app wiring。
|
||||
- 复用证据与缺口：`.trellis/tasks/archive/2026-09/09-10-server-ws-role-code-organization/research/endpoint-code-organization.md` 与归档 role-split design 已覆盖 endpoint/shared owner；本报告刷新了其旧 checkout/旧行号。当前缺口正是该报告指出的 runtime route wiring：生产只调用 `assertOneTalkEndpointFrameSets`，没有调用声明式 router factory；还没有实际 endpoint wrong-path 的 injectWS 防回归矩阵。
|
||||
|
||||
## 发现
|
||||
|
||||
### 1. 生产调用图：声明式 Router 未接入
|
||||
|
||||
```text
|
||||
createApp(app.ts:71-133)
|
||||
-> installWebsocket(websocket/index.ts:139-182)
|
||||
-> registerWebsocketRoutes(...)(app,...)
|
||||
-> assertOneTalkEndpointFrameSets(index.ts:66-69) [启动期校验，不分发]
|
||||
-> /ws/plugin -> createOneTalkPluginWebSocketHandler(index.ts:125-127)
|
||||
-> /ws/mind -> createOneTalkMindWebSocketHandler(index.ts:128-130)
|
||||
-> createOneTalkWebSocketHandler(handler.ts:198-306)
|
||||
-> createOneTalkProtocolKernel(protocol-kernel.ts:263-306)
|
||||
-> decode + endpoint gate + auth/scope gate
|
||||
-> endpoint.onAuthenticatedFrame(frame)
|
||||
-> plugin/index.ts:273-390 手写 if 分发
|
||||
-> mind/index.ts:125-156 手写 if 分发
|
||||
```
|
||||
|
||||
- `createOneTalkAuthenticatedRouter` (`authenticated-router.ts:102-128`) 的工厂在生产源码中没有调用；唯一 factory 调用者是 `authenticated-router.test.ts:39,51,55,60,71,83`。
|
||||
- `createOneTalkEndpointAuthenticatedRouter` (`endpoint-routes.ts:42-75`) 的唯一调用者是 `authenticated-router.test.ts:120`；生产 `index.ts:66-69` 只调用 `assertOneTalkEndpointFrameSets`。
|
||||
- `authenticated-router.ts:55-59` 的 `isOneTalkAuthenticatedClientFrame` 是例外：它被 `handler.ts:23-25,263-273` 生产使用，作为 protocol kernel 的 client-frame type predicate；这不等于生产使用了 Router 的 route table/dispatch。
|
||||
- `ONETALK_PLUGIN_ROUTE_TYPES` (`plugin/index.ts:34-44`) 与 `ONETALK_MIND_ROUTE_TYPES` (`mind/index.ts:20`) 只进入启动期集合校验；真正业务分派规则是 `plugin/index.ts:273-390`、`mind/index.ts:125-156` 的条件分支。由此形成“端点类型列表 + 手写分支 + 未接入声明式 Router”三处维护面，当前实际运行的仍只有手写分支。
|
||||
|
||||
### 2. 共享 transport 的顺序和错误边界
|
||||
|
||||
- `protocol-kernel.ts:96-111` 先执行 cutover admission，再检查 fixed `expectedConnectionType`；错误调用 `onAdmissionRejected` 或 `onEndpointRejected`，后者由 `handler.ts:276-279` 发 `ws.error(scope_mismatch)` 并关闭 `1008`。因此 decoder 合法但发往错误 URL 的另一端 frame 在业务 handler 前被拒绝。
|
||||
- `protocol-kernel.ts:113-144` 记录 inbound frame 后，未认证只允许 `ws.hello`（`handler.ts:288-291` 发 `auth_required` 并关 `1008`）；已认证重复 `ws.hello` 为 `unknown_request`（`handler.ts:292-300`），同 scope 的非 client server frame 也为 `unknown_request`；scope/connection type mismatch 先于 client classification，发 `scope_mismatch` 并关 `1008`。
|
||||
- `protocol-kernel.ts:146-165` 以每 socket Promise chain 保证 FIFO；`handler.ts:263-306` 是共享 kernel 装配点。route integration 不应让 endpoint Router重新 parse raw data、建立第二个 FIFO/session state，或改变这些错误顺序。
|
||||
- `decoder.ts:74-112` 是方向事实源：plugin inbound frames 包括 discovery/sync/profile/buyer/observation/`send.confirmation`；Mind inbound frames 包括 `send.request`；`heartbeat` 双端合法。`wire.ts:74-86` 的 `ONETALK_CLIENT_FRAME_TYPES` 当前 11 项，且包含 `ws.hello`，authenticated route 需要排除它。
|
||||
|
||||
### 3. 每个 endpoint 的 frame、当前 handler 与语义
|
||||
|
||||
| endpoint / frame | 当前生产 handler 与成功输出 | 授权/错误/未知语义（当前行为） | 风险 |
|
||||
| --- | --- | --- | --- |
|
||||
| Plugin `heartbeat` | `plugin/index.ts:278-312`；operation `heartbeat`，记录 liveness 后 `heartbeat.ack` | `sessionAuthorizationFailure(..., null)` 检查 allowed、binding、Mind scope、version，不要求 read；明确拒绝或非 authorizationRejected 错误发 `ws.error` + close `1008`，缺 permission 情形为 error-only；无 canonical connection 时无 ACK | 中：Router metadata 必须表达双端 ownership 和 heartbeat 的特殊 permission |
|
||||
| Plugin `conversation.discovered` | `plugin/index.ts:369-375` -> `syncFlows.handleDiscovery`；成功 `conversation.ack`，并发布 sync status | operation `sync` + `read`；授权拒绝/撤权/version/scope 等按 `plugin/index.ts:285-307`，通常 error + `1008`，缺 read 为 `authorization_rejected` error-only；DB/guard 错误由 Flow 转稳定错误并关闭；无效 discovery batch 为 `invalid_message` + `1003` | 高：手写 branch 与 route metadata 都决定 Plugin ownership |
|
||||
| Plugin `conversations.discovered` | 同上 `syncFlows.handleDiscovery`；fragment batch 最终 `conversations.ack` | 同上；Flow 自有 fragment sequence/duplicate/size/timer 校验（`sync-flows.ts:217-315`） | 高：批次 timer/state 不能被 Router 复制 |
|
||||
| Plugin `sync.complete` | `plugin/index.ts:377-381` -> `syncFlows.handleCompletion`；flush observation 后发布 `sync.status`、`conversation.updated`，不返回 client ACK | 同上 `sync` + `read`；rejected complete 发 `invalid_message` error-only；必须先等 observation flush（`sync-flows.ts:398-450`） | 高：Router 只能委托，不能改变 await/发布顺序 |
|
||||
| Plugin `contact.profile.observed` | `plugin/index.ts:359-367` -> `profileFlow.handle`；guarded write/post-write update 后 `contact.profile.ack` | `sync` + `read`；future skew 发 `profile_observed_at_future` error-only；写入/guard/reauth/canonical fence 由 `profile-flow.ts:66-126` 拥有；reauth 失败发错误并关 `1008` | 高：跨 await 最终动作不能移到通用 Router |
|
||||
| Plugin `buyer.facts.observed` | `plugin/index.ts:322-357`；`buyerFactService.ingestFacts` 后再次授权，成功 `buyer.facts.ack` | `sync` + `read`；缺 service 发 `database_unavailable` error-only；存储异常 `database_unavailable` + `1011`；post-write reauth 失败发错误 + `1008`；commit guard 失效丢弃 | 高：当前仍是 Plugin handler 内一段独立业务流程，不能仅以 `type` switch 搬动 |
|
||||
| Plugin `message.observed` | `plugin/index.ts:383-388` -> `syncFlows.handleObservation`；batch flush 后 `message.ack`，仅 live accepted 发布 `message.created`、`conversation.updated` | `sync` + `read`；observation failure/DB failure 由 Flow 发稳定错误并关闭；history/incremental 或 duplicate 不发布 live event（`sync-flows.ts:318-396`） | 高：ObservationBatcher 是 per-connection state owner |
|
||||
| Plugin `messages.observed` | 同上；批量 `messages.ack`，发布规则同上 | 同上，且 batch 输入结果数必须严格匹配 | 高：同上 |
|
||||
| Plugin `send.confirmation` | `plugin/index.ts:279-280` -> `sendConfirmationFlow.handleConfirmation`；coordinator 完成 claim/ingest 后向 Mind 发布并 resolve `send.result` | 跳过 generic session auth；`pending-send-coordinator.ts:287-363` 用 pending snapshot、plugin socket/scope、send authorization、generation/epoch/guard；无 pending/迟到/错 scope静默 no-op；non-confirmed status -> `delivery_unknown` | 高：`pending_send` 是特殊授权策略，必须保留 coordinator 唯一 owner，不可套普通 session route |
|
||||
| Plugin `send.request`（反端输入） | `plugin/index.ts:273-276` 明确回 `authorization_rejected`，不进入 coordinator；正常合法方向为 Mind endpoint | 正常 decoder 要求 `send.request` 为 `mind_page`；发到 `/ws/plugin` 时先被 kernel expected endpoint gate 拒绝 `scope_mismatch` + close `1008`，手写 fallback 仅覆盖极端已到达 callback 的情况 | 中：应由 fixed endpoint router/contract gate 锁住，不要让 fallback 成为第二分发规则 |
|
||||
| Mind `heartbeat` | `mind/index.ts:129-156`；使用 Cookie session capability 或 authorization reader，记录 liveness 后 `heartbeat.ack` | operation `heartbeat` + `read`；拒绝或 binding/scope/version 错误通常 error + `1008`，允许但无 read 为 `authorization_rejected` error-only；无合法 Mind scope 回 authorizationRejected error-only | 高：Plugin/Mind 共享 heartbeat 但授权策略在 endpoint context 内不同 |
|
||||
| Mind `send.request` | `mind/index.ts:125-128` -> `sendRequestFlow.handleRequest`；`registry.requestSend` 完成后 Mind 收 `send.result` | 跳过 generic session auth；coordinator 在 `pending-send-coordinator.ts:123-284` 负责 duplicate/request/conversation route、双方连接和 send 权限、授权窗口、dispatch fence、timeout、终态；失败为 `rejected_before_send`/`delivery_unknown` 稳定 payload | 高：必须维持 `pending_send` metadata 和唯一 coordinator |
|
||||
|
||||
### 4. `ws.hello`、未知帧和 route set
|
||||
|
||||
- `ws.hello` 不属于 authenticated router（`authenticated-router.ts:10-13,48-59,71-74`）；Plugin bootstrap 在 `plugin/index.ts:192-271` 保留 `accepted -> register -> anchor.snapshot`，Mind bootstrap 在 `mind/index.ts:48-124` 保留 `accepted -> register` 并由 registry 发布 Plugin status。
|
||||
- `ONETALK_CLIENT_FRAME_TYPES` 的非 hello 集合应恰覆盖上述 10 个 authenticated client route；`endpoint-routes.ts:82-112` 当前只对两个类型列表做 unknown、duplicate、heartbeat 双归属和非 heartbeat 单 owner 校验。它没有看到实际 handler，也不能发现 `if` 分支与列表不一致。
|
||||
- `createOneTalkAuthenticatedRouter` 的 factory 会对全量非 hello client route 做 missing/duplicate/unknown/hello runtime 校验，并在 `authenticated-router.ts:118-126` 检查 route connection type 后调用 handler；但现在只在合成测试中执行。
|
||||
|
||||
### 5. 已有测试位置与应扩展处
|
||||
|
||||
- `apps/server/test/authenticated-router.test.ts:38-100` 覆盖全量非 hello route、missing/duplicate/unknown/hello、connection mismatch 不调用 collaborator；`:102-148` 只用合成 route table 测 endpoint view；`:150-189` 覆盖 heartbeat 双端及其它 route 单 owner。它证明 factory 本身，不证明生产 handlers 通过 factory。
|
||||
- `apps/server/test/onetalk-websocket.test.ts:413-778` 通过真实 `createApp`/`injectWS` 对 `ONETALK_CLIENT_FRAME_TYPES` 逐项断言 wire output；`CLIENT_FRAME_COVERAGE_BASELINES` 在 `:42-54` 以 `satisfies Record<...>` 锁住 11 项。该矩阵是业务行为基线，应继续保留并在 Router 接入后不改公开 frame 顺序。
|
||||
- `apps/server/test/websocket.test.ts:209-257` 覆盖 Plugin handshake/heartbeat，`:259-292` 覆盖未认证 gate，`:1148-1273` 覆盖 protocol upgrade、scope mismatch、同 scope non-client unknown 优先级；`:294-372` 和 `:805-854` 覆盖 send/pending 与 permission 行为。应新增 endpoint wrong-path 的真实 `injectWS` 测试，断言 route collaborator/coordinator 不调用。
|
||||
- `apps/server/test/app.test.ts:387-397` 只验证 WebSocket server 和 `/ws`、`/ws/plugin`、`/ws/mind` 路由注册，不验证 route table 在生产 handler 中被构造或 dispatch。
|
||||
- 推荐扩展 `authenticated-router.test.ts`：用实际 route-definition builder 或生产暴露的 route metadata 做完整性断言；断言 Plugin/Mind endpoint router 真实 dispatch 各自 handler，错误 endpoint 不调用 collaborator；增加 route metadata `authorization`/`operation` mismatch 的静态或运行时拒绝。不要只增加字符串列表测试。
|
||||
- 推荐扩展 `websocket.test.ts`：完成握手后在 `/ws/plugin` 发送合法 `mind_page/send.request`，在 `/ws/mind` 发送合法 `plugin/conversation.discovered`；预期 `ws.error(scope_mismatch)` + close `1008` 且 service/registry collaborator 调用数为零。该测试与 decoder 反方向 invalid-message 测试分开。
|
||||
- 生产 frame 每项的 ACK/publish/terminal 行为继续由 `onetalk-websocket.test.ts`、`onetalk-profile-websocket.test.ts`、`onetalk-buyer-websocket.test.ts` 锁定；新增 Router 不应把这些 domain/fence 测试替换成私有函数调用。
|
||||
|
||||
## 候选 Scope 与依赖
|
||||
|
||||
1. **Cross-cutting route contract and production wiring**：由 `authenticated-router.ts`/`endpoint-routes.ts` 统一拥有 route metadata、client-frame completeness、endpoint ownership、`session`/`pending_send` 和 operation；由 `plugin/index.ts`/`mind/index.ts` 在 per-socket Flow 创建后装配对应 handler 并调用 Router dispatch。排除 contract wire/decoder 变更、registry/pending state 重写、hello bootstrap 和业务 Flow 时序变化。该 scope 必须先定好 route handler context，Plugin/Mind 实现均等待。
|
||||
2. **Plugin endpoint route adapters**：把当前 Plugin `onAuthenticatedFrame` 的分支变为 route handler 委托，保留 sync/profile/buyer/confirmation Flow 和现有错误/guard/ACK/publish 顺序；拥有路径为 `plugin/index.ts` 及必要的 route adapter，排除 shared registry/contract/handler transport。依赖 scope 1 的 route contract。
|
||||
3. **Mind endpoint route adapters**：把 Mind heartbeat/send request 分支变为 route handler 委托，保留 Cookie session authorization 和 pending-send coordinator；拥有路径为 `mind/index.ts` 及必要 adapter，排除 shared publisher/coordinator/state。依赖 scope 1，且与 Plugin 若共享 route builder/context 文件则不能并行写。
|
||||
4. **Production-path regression tests**：扩展 router unit 和 `injectWS` endpoint wrong-path/完整 wire matrix；测试只能观察 public outbound frame、close code 与 collaborator call count。依赖 scope 1-3 的接口确定；若测试与各 endpoint 文件分离且只读公共 API，可以在实现稳定后并行，但当前不满足 dependency-ready。
|
||||
|
||||
最小可兼容方案建议：保留 `protocol-kernel`/`handler` 的 shared gate 与现有 Flow；把当前 route metadata/handler wiring 变成真正传给 `createOneTalkEndpointAuthenticatedRouter(...).dispatch` 的生产路径。route metadata 应从单一显式定义派生 Plugin/Mind route type views，或由同一个 builder 同时生成 route definitions 与 endpoint type sets；不要继续让 `ONETALK_*_ROUTE_TYPES`、`authenticated-router.test.ts` 合成 routes 和生产 `if/switch` 各自维护相同映射。`ws.hello` 继续留在 endpoint bootstrap，不塞进 authenticated Router。`send.request`/`send.confirmation` 标为 `pending_send`，dispatch 只调用已有 registry coordinator；普通 session route 仍由 endpoint context 执行现有 authorization/fence。若为适配不同 endpoint context 改 factory API，应保留 construction-time missing/duplicate/unknown/hello 失败和 endpoint ownership guard，禁止 unsupported endpoint 的 no-op/fallback handler。
|
||||
|
||||
## Invariant 与验收探针
|
||||
|
||||
### 结构性 owner / source of truth
|
||||
|
||||
| 不变量 | 唯一 owner | 需保护的 await/副作用边界 |
|
||||
| --- | --- | --- |
|
||||
| client frame 类型和 direction | `apps/onetalk-contract/src/wire.ts:74-86` + `decoder.ts:74-112` | route builder 不复制 wire union；decoder 失败在业务前结束 |
|
||||
| endpoint ownership/metadata | 生产使用的声明式 endpoint Router（由单一 route definition builder 生成） | Plugin/Mind handler 不再有第二份 type-to-owner mapping；`ws.hello` 永不进入 authenticated Router |
|
||||
| decode、endpoint gate、scope gate、FIFO | `protocol-kernel.ts:96-165` + `handler.ts:263-306` | 先 endpoint/scope 再 unknown/client classification；每 socket Promise chain 不变 |
|
||||
| connection/generation/commit guard | `connection-store.ts`/`registry.ts` | 任何 DB/授权/publish await 后继续沿用现有 canonical/guard fence |
|
||||
| pending send | `pending-send-coordinator.ts:79-83,123-284,287-363` | Router 不授权复制 pending map；request/confirmation 的 dispatch、timeout、claim、terminal 顺序不变 |
|
||||
| Plugin sync/profile observation Flow | `plugin/flows/*` 与 domain services | commit -> ACK -> publish、flush-before-complete、profile post-write fences 不变 |
|
||||
|
||||
### Static/unit probes（不启动服务）
|
||||
|
||||
- `rg -n "createOneTalkAuthenticatedRouter|createOneTalkEndpointAuthenticatedRouter|ONETALK_.*ROUTE_TYPES|frame\.type ===|frame\.type !==" apps/server/src/websocket apps/server/test`：验收目标是生产 endpoint handler 通过 Router dispatch，手写分发不再作为第二份 route ownership 规则；允许 Flow 内部按 payload/结果分支，不允许按 client frame type 重新拥有 route map。
|
||||
- `pnpm --filter @trade-message-center/server exec node --experimental-strip-types --test test/authenticated-router.test.ts test/websocket.test.ts test/onetalk-websocket.test.ts`：覆盖 factory、握手、error/close 和 11 项 public wire matrix；后续运行 compiled `dist/test` 对应测试。
|
||||
- `pnpm --filter @trade-message-center/server typecheck`、`pnpm --filter @trade-message-center/server build`、`pnpm format:check`、`git diff --check`；backend 测试 hard timeout 60 秒。生产路由变更前后均应保持 strict TypeScript，不以 `as never` 伪造 frame/context。
|
||||
|
||||
### Runtime probes（使用 injectWS/应用状态；本研究未执行）
|
||||
|
||||
- 用 `injectWS` 建立 `/ws/plugin` 与 `/ws/mind`，验证 route table 在真实 `createApp -> installWebsocket -> fixed handler -> protocol kernel -> router` 链路中生效；重点记录 outbound frame/close code，不把服务启动或 socket 存在视为证据。
|
||||
- wrong-path matrix：合法 decoder frame `send.request`（connectionType `mind_page`）发往 `/ws/plugin`；合法 `conversation.discovered`（connectionType `plugin`）发往 `/ws/mind`。期待 `ws.error { code: "scope_mismatch" }`、close `1008`、无 service/registry business collaborator 调用。
|
||||
- 既有真实 wire matrix 应继续证明 `ws.hello -> accepted/anchor`、heartbeat ACK、Plugin discovery/profile/buyer/observation/sync outputs、Mind send command/result、Plugin confirmation publish；用 deferred/latch 的已有 fence 测试确认 Router 接入没有越过 await 或重排 commit/ACK/publish。
|
||||
- 本研究未启动服务、未连接 PostgreSQL/Mind HTTP、未使用真实浏览器/插件账号，也未进行 TLS/proxy/生产授权验证；这些均为 runtime-unverified。
|
||||
|
||||
## 已验证事实、假设与外部边界
|
||||
|
||||
- 已验证事实：当前 HEAD 上 focused router/WebSocket source tests 共 29 个子测试全部通过（`authenticated-router.test.ts` 5 + `websocket.test.ts` 24）；测试通过不改变生产 factory 未被调用的静态事实。
|
||||
- 已验证事实：`createOneTalkAuthenticatedRouter` 和 `createOneTalkEndpointAuthenticatedRouter` 的生产调用搜索仅命中定义/测试；生产 handler 只通过 `isOneTalkAuthenticatedClientFrame` 做 client-frame predicate。
|
||||
- 已验证事实：contract decoder 对 frame direction 做严格检查；因此错误 URL 的“合法另一端 frame”可以在 kernel endpoint gate 被拒绝，伪造错误 connectionType 的同名 frame 则可能在 decoder 阶段得到 `invalid_message`，两者测试必须分开。
|
||||
- 假设：实现会保持当前 wire version 6、frame/payload/error code、close code/reason、Flow 时序和 registry API；若产品要改变任一项，应拆成独立行为变更，不归本路由结构重构。
|
||||
- 外部/未验证边界：Fastify upgrade 的真实网络行为、PostgreSQL transaction、Mind authorization HTTP、TLS/NGINX、真实 Chrome plugin runtime 未在本研究验证。
|
||||
|
||||
## 找到的文件与代码模式
|
||||
|
||||
- `apps/server/src/app.ts:71-133`：创建 Fastify app，并在 `:110` 调用唯一 `installWebsocket`。
|
||||
- `apps/server/src/websocket/index.ts:53-136,139-182`：注册前 route-set 校验、Origin/cutover guard、固定 `/ws/plugin`/`/ws/mind` handler 组合；不执行业务 frame dispatch。
|
||||
- `apps/server/src/websocket/handler.ts:198-306`：共享 socket context、error mapping、protocol kernel wiring；只把 typed authenticated frame 交给 endpoint callback。
|
||||
- `apps/server/src/websocket/protocol-kernel.ts:96-165`：admission、expected endpoint、auth hello gate、scope gate、unknown classification、per-socket FIFO。
|
||||
- `apps/server/src/websocket/authenticated-router.ts:10-128`：authenticated frame predicate、完整静态 route factory、connection type guard、handler dispatch；factory 当前未接生产。
|
||||
- `apps/server/src/websocket/endpoint-routes.ts:15-112`：端点 route set 校验和 endpoint view factory；生产只用同文件导出的 set assertion（间接由 index 导入）。
|
||||
- `apps/server/src/websocket/plugin/index.ts:34-44,108-115,192-390`：Plugin route type 列表、handler 建立、hello、手写 authenticated frame branches。
|
||||
- `apps/server/src/websocket/mind/index.ts:20,31-47,48-156`：Mind route type 列表、handler 建立、Cookie hello、手写 heartbeat/send request branches。
|
||||
- `apps/server/src/websocket/plugin/flows/sync-flows.ts:152-458`：Plugin discovery/observation/completion Flow 与独占 batch/timer。
|
||||
- `apps/server/src/websocket/plugin/flows/profile-flow.ts:62-129`：profile guarded write、reauthorization、conversation update、ACK。
|
||||
- `apps/server/src/websocket/plugin/flows/send-confirmation-flow.ts:32-81`：Plugin confirmation 对 pending coordinator 的委托。
|
||||
- `apps/server/src/websocket/mind/send-request-flow.ts:29-58`：Mind request 对 pending coordinator 的委托和 terminal projection。
|
||||
- `apps/server/src/websocket/pending-send-coordinator.ts:71-370`：pending/terminal、授权窗口、dispatch/confirmation claim、timeout 的唯一 owner。
|
||||
- `apps/onetalk-contract/src/wire.ts:41-105`：protocol/frame/client/server 类型全集；`apps/onetalk-contract/src/decoder.ts:74-112,177-194`：direction 与 strict frame decode。
|
||||
- `apps/server/test/authenticated-router.test.ts:38-189`：只证明声明式 Router/validator 本身。
|
||||
- `apps/server/test/onetalk-websocket.test.ts:42-54,413-778`：11 项 client frame 的 typed coverage 和真实 wire matrix。
|
||||
- `apps/server/test/websocket.test.ts:209-292,1148-1273`：handshake/heartbeat/unauthenticated/scope/unknown/close baseline。
|
||||
- `apps/server/test/app.test.ts:387-397`：仅 route registration smoke。
|
||||
|
||||
## 外部参考（文档、版本）
|
||||
|
||||
- 无外部网络参考；本研究依据当前仓库源码、测试和 Trellis 规范。
|
||||
- 运行时基线：Node `>=22.22.2 <23`、Fastify `^5.12.1`、`@fastify/websocket` `^11.3.0`（`apps/server/package.json`；`.trellis/spec/server/backend/quality-guidelines.md`）。
|
||||
|
||||
## 相关 spec
|
||||
|
||||
- `.trellis/spec/project/architecture.md`
|
||||
- `.trellis/spec/project/module-organization.md`
|
||||
- `.trellis/spec/project/module-ownership.md`
|
||||
- `.trellis/spec/project/async-state-boundaries.md`
|
||||
- `.trellis/spec/server/backend/error-handling.md`
|
||||
- `.trellis/spec/server/backend/quality-guidelines.md`
|
||||
- `.trellis/spec/server/backend/service-foundation.md:300-381`
|
||||
- `.trellis/tasks/archive/2026-09/09-10-server-websocket-role-split/design.md`
|
||||
- `.trellis/tasks/archive/2026-09/09-10-server-ws-role-code-organization/research/endpoint-code-organization.md`
|
||||
|
||||
## 注意事项 / 未找到内容
|
||||
|
||||
- 没有发现生产 code 调用 route factory；不能把当前 route unit tests 或 startup `assertOneTalkEndpointFrameSets` 报告成生产 Router 已接入。
|
||||
- `handler.ts` 仍然生产使用 `isOneTalkAuthenticatedClientFrame`，报告“Router 未接入”时应精确限定为 route table/factory/dispatch 未接入。
|
||||
- `endpoint-routes.ts` 当前 `allRoutes` 合并逻辑（`:49-57`）用于测试视图，若改为生产使用必须避免 unsupported endpoint handler 的 no-op/fallback，并保留 ownership 检查。
|
||||
- `ONETALK_*_ROUTE_TYPES` 当前是额外事实来源；最小可兼容改动也必须明确它们是由生产 route definition 派生，或删除并以单一 builder 输出，不能仅把 `router.dispatch` 套在现有 if 后面造成“声明式外壳 + 手写第二分支”。
|
||||
- 本研究只写入当前任务的 `research/server-routing.md`，未修改产品源码、测试、spec、任务根文件或 Git index。
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"id": "websocket-routing-single-source-of-truth",
|
||||
"name": "websocket-routing-single-source-of-truth",
|
||||
"title": "收敛 WebSocket 路由分发单一事实来源",
|
||||
"description": "让服务端认证 WebSocket 的端点归属校验与生产帧分发共用一份声明式路由定义，并消除客户端路由类型列表的重复维护。",
|
||||
"status": "completed",
|
||||
"dev_type": null,
|
||||
"scope": null,
|
||||
"package": null,
|
||||
"priority": "P2",
|
||||
"creator": "ybf",
|
||||
"assignee": "ybf",
|
||||
"createdAt": "2026-09-11",
|
||||
"completedAt": "2026-09-12",
|
||||
"branch": "09-11-websocket-routing-single-source-of-truth",
|
||||
"base_branch": "main",
|
||||
"worktree_path": null,
|
||||
"commit": null,
|
||||
"pr_url": null,
|
||||
"subtasks": [],
|
||||
"children": [],
|
||||
"parent": null,
|
||||
"relatedFiles": [],
|
||||
"notes": "",
|
||||
"meta": {}
|
||||
}
|
||||
@@ -8,8 +8,8 @@
|
||||
|
||||
<!-- @@@auto:current-status -->
|
||||
- **Active File**: `journal-1.md`
|
||||
- **Total Sessions**: 59
|
||||
- **Last Active**: 2026-09-11
|
||||
- **Total Sessions**: 61
|
||||
- **Last Active**: 2026-09-12
|
||||
<!-- @@@/auto:current-status -->
|
||||
|
||||
---
|
||||
@@ -19,7 +19,7 @@
|
||||
<!-- @@@auto:active-documents -->
|
||||
| File | Lines | Status |
|
||||
|------|-------|--------|
|
||||
| `journal-1.md` | ~1288 | Active |
|
||||
| `journal-1.md` | ~1321 | Active |
|
||||
<!-- @@@/auto:active-documents -->
|
||||
|
||||
---
|
||||
@@ -29,6 +29,7 @@
|
||||
<!-- @@@auto:session-history -->
|
||||
| # | Date | Title | Commits | Branch |
|
||||
|---|------|-------|---------|--------|
|
||||
| 61 | 2026-09-12 | 收敛 OneTalk WebSocket 路由分发 | `bd9cc6a` | `09-11-websocket-routing-single-source-of-truth` |
|
||||
| 60 | 2026-09-11 | 隔离发送观测 pending | `1a899a9` | `send-observation-pending` |
|
||||
| 59 | 2026-09-11 | 移除 pending-send 终态历史 | `63d9e5b` | `fix/remove-terminal-send-history` |
|
||||
| 58 | 2026-09-11 | 移除 OneTalk 图片宽高链路 | `282194e` | `09-11-remove-image-dimensions` |
|
||||
|
||||
@@ -1297,3 +1297,25 @@ Moved Mind-only and Plugin-only WebSocket implementations to canonical endpoint
|
||||
### Status
|
||||
|
||||
[OK] **Completed**
|
||||
|
||||
|
||||
## Session 61: 收敛 OneTalk WebSocket 路由分发
|
||||
<!-- trellis-session: v=2 fp=df00ad1bc5438aed -->
|
||||
|
||||
**Date**: 2026-09-12
|
||||
**Task**: 收敛 OneTalk WebSocket 路由分发
|
||||
**Branch**: `09-11-websocket-routing-single-source-of-truth`
|
||||
|
||||
### Summary
|
||||
|
||||
将认证 WebSocket 路由元数据收敛为唯一事实来源，并让 Plugin/Mind 生产分发与扩展 Service Worker 路由复用声明式表；聚焦 WebSocket 与路由测试通过。完整类型、构建和全测仍受既有 business-card 夹具、缺失 migration 和构建哈希测试阻碍。
|
||||
|
||||
### Git Commits
|
||||
|
||||
| Hash | Message |
|
||||
|------|---------|
|
||||
| `bd9cc6a` | refactor: centralize WebSocket route dispatch |
|
||||
|
||||
### Status
|
||||
|
||||
[OK] **Completed**
|
||||
|
||||
@@ -7,20 +7,24 @@ import type { OneTalkContactProfileCoordinator } from "../contact-profile-coordi
|
||||
import type { OneTalkSendCommandFlow } from "../flows/send-command-flow.ts";
|
||||
import type { OneTalkSyncEngine } from "../sync-engine.ts";
|
||||
|
||||
type RoutedFrame = Extract<
|
||||
OneTalkFrame,
|
||||
{
|
||||
type:
|
||||
| "anchor.snapshot"
|
||||
| "message.ack"
|
||||
| "messages.ack"
|
||||
| "conversation.ack"
|
||||
| "conversations.ack"
|
||||
| "contact.profile.ack"
|
||||
| "buyer.facts.ack"
|
||||
| "send.command";
|
||||
}
|
||||
>;
|
||||
type OneTalkBusinessRoute = {
|
||||
dispatch: (frame: OneTalkFrame) => boolean;
|
||||
};
|
||||
|
||||
const defineOneTalkBusinessRoute = <TType extends OneTalkFrame["type"]>(
|
||||
type: TType,
|
||||
handle: (frame: Extract<OneTalkFrame, { type: TType }>) => void,
|
||||
): OneTalkBusinessRoute => {
|
||||
const matches = (frame: OneTalkFrame): frame is Extract<OneTalkFrame, { type: TType }> =>
|
||||
frame.type === type;
|
||||
return {
|
||||
dispatch: (frame) => {
|
||||
if (!matches(frame)) return false;
|
||||
handle(frame);
|
||||
return true;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
export type OneTalkServiceWorkerFrameRouter = {
|
||||
handle: (frame: OneTalkFrame) => void;
|
||||
@@ -33,51 +37,35 @@ export const createOneTalkServiceWorkerFrameRouter = (options: {
|
||||
buyer?: Pick<OneTalkBuyerFactCoordinator, "handleFrame">;
|
||||
send?: Pick<OneTalkSendCommandFlow, "handle">;
|
||||
}): OneTalkServiceWorkerFrameRouter => {
|
||||
const routes: {
|
||||
[K in RoutedFrame["type"]]: (frame: Extract<RoutedFrame, { type: K }>) => void;
|
||||
} = {
|
||||
"anchor.snapshot": (frame) => options.sync.handleServerFrame(frame),
|
||||
"message.ack": (frame) => options.sync.handleServerFrame(frame),
|
||||
"messages.ack": (frame) => options.sync.handleServerFrame(frame),
|
||||
"conversation.ack": (frame) => options.sync.handleServerFrame(frame),
|
||||
"conversations.ack": (frame) => options.sync.handleServerFrame(frame),
|
||||
"contact.profile.ack": (frame) => options.profile?.handleFrame(frame),
|
||||
"buyer.facts.ack": (frame) => options.buyer?.handleFrame(frame),
|
||||
"send.command": (frame) => options.send?.handle(frame),
|
||||
};
|
||||
const routes = [
|
||||
defineOneTalkBusinessRoute("anchor.snapshot", (frame) =>
|
||||
options.sync.handleServerFrame(frame),
|
||||
),
|
||||
defineOneTalkBusinessRoute("message.ack", (frame) => options.sync.handleServerFrame(frame)),
|
||||
defineOneTalkBusinessRoute("messages.ack", (frame) =>
|
||||
options.sync.handleServerFrame(frame),
|
||||
),
|
||||
defineOneTalkBusinessRoute("conversation.ack", (frame) =>
|
||||
options.sync.handleServerFrame(frame),
|
||||
),
|
||||
defineOneTalkBusinessRoute("conversations.ack", (frame) =>
|
||||
options.sync.handleServerFrame(frame),
|
||||
),
|
||||
defineOneTalkBusinessRoute("contact.profile.ack", (frame) =>
|
||||
options.profile?.handleFrame(frame),
|
||||
),
|
||||
defineOneTalkBusinessRoute("buyer.facts.ack", (frame) => options.buyer?.handleFrame(frame)),
|
||||
defineOneTalkBusinessRoute("send.command", (frame) => options.send?.handle(frame)),
|
||||
];
|
||||
const handle = (frame: OneTalkFrame): void => {
|
||||
switch (frame.type) {
|
||||
case "anchor.snapshot":
|
||||
routes["anchor.snapshot"](frame);
|
||||
return;
|
||||
case "message.ack":
|
||||
routes["message.ack"](frame);
|
||||
return;
|
||||
case "messages.ack":
|
||||
routes["messages.ack"](frame);
|
||||
return;
|
||||
case "conversation.ack":
|
||||
routes["conversation.ack"](frame);
|
||||
return;
|
||||
case "conversations.ack":
|
||||
routes["conversations.ack"](frame);
|
||||
return;
|
||||
case "contact.profile.ack":
|
||||
routes["contact.profile.ack"](frame);
|
||||
return;
|
||||
case "buyer.facts.ack":
|
||||
routes["buyer.facts.ack"](frame);
|
||||
return;
|
||||
case "send.command":
|
||||
routes["send.command"](frame);
|
||||
return;
|
||||
case "ws.error":
|
||||
if (frame.payload.code === ONETALK_ERROR_CODES.profileObservedAtFuture) {
|
||||
options.profile?.handleFrame(frame);
|
||||
}
|
||||
return;
|
||||
default:
|
||||
return;
|
||||
for (const route of routes) {
|
||||
if (route.dispatch(frame)) return;
|
||||
}
|
||||
if (
|
||||
frame.type === "ws.error" &&
|
||||
frame.payload.code === ONETALK_ERROR_CODES.profileObservedAtFuture
|
||||
) {
|
||||
options.profile?.handleFrame(frame);
|
||||
}
|
||||
};
|
||||
return { handle };
|
||||
|
||||
@@ -67,3 +67,17 @@ test("does not route transport-only or mind-page frames", () => {
|
||||
|
||||
assert.deepEqual(calls, []);
|
||||
});
|
||||
|
||||
test("optional business owners are explicit no-ops", () => {
|
||||
const calls = [];
|
||||
const router = createOneTalkServiceWorkerFrameRouter({
|
||||
sync: { handleServerFrame: (value) => calls.push(["sync", value.type]) },
|
||||
});
|
||||
|
||||
assert.doesNotThrow(() => {
|
||||
router.handle(frame("contact.profile.ack"));
|
||||
router.handle(frame("buyer.facts.ack"));
|
||||
router.handle(frame("send.command"));
|
||||
});
|
||||
assert.deepEqual(calls, []);
|
||||
});
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// 定义 OneTalk 已认证客户端帧的静态路由。
|
||||
// 定义 OneTalk 已认证客户端帧的唯一静态路由元数据。
|
||||
|
||||
import {
|
||||
ONETALK_CLIENT_FRAME_TYPES,
|
||||
@@ -14,35 +14,27 @@ export type OneTalkAuthenticatedClientFrame = Exclude<
|
||||
|
||||
export type OneTalkAuthenticatedRouteAuthorization = "session" | "pending_send";
|
||||
|
||||
export type OneTalkAuthenticatedRouteDefinition<
|
||||
TContext,
|
||||
TFrame extends OneTalkAuthenticatedClientFrame = OneTalkAuthenticatedClientFrame,
|
||||
> = {
|
||||
type: TFrame["type"];
|
||||
connectionType: TFrame["connectionType"] | readonly TFrame["connectionType"][];
|
||||
export type OneTalkAuthenticatedRouteDefinition = {
|
||||
type: OneTalkAuthenticatedClientFrame["type"];
|
||||
connectionType:
|
||||
| OneTalkAuthenticatedClientFrame["connectionType"]
|
||||
| readonly OneTalkAuthenticatedClientFrame["connectionType"][];
|
||||
authorization: OneTalkAuthenticatedRouteAuthorization;
|
||||
operation: OneTalkAuthorizationOperation;
|
||||
handler: (
|
||||
context: TContext,
|
||||
frame: TFrame,
|
||||
route: OneTalkAuthenticatedRouteDefinition<TContext, TFrame>,
|
||||
) => Promise<void>;
|
||||
};
|
||||
|
||||
type OneTalkAuthenticatedRouteTable<TContext> = ReadonlyMap<
|
||||
type OneTalkAuthenticatedRouteTable = ReadonlyMap<
|
||||
OneTalkAuthenticatedClientFrame["type"],
|
||||
OneTalkAuthenticatedRouteDefinition<TContext>
|
||||
OneTalkAuthenticatedRouteDefinition
|
||||
>;
|
||||
|
||||
export type OneTalkAuthenticatedRouter<TContext> = {
|
||||
get: (
|
||||
type: OneTalkAuthenticatedClientFrame["type"],
|
||||
) => OneTalkAuthenticatedRouteDefinition<TContext>;
|
||||
export type OneTalkAuthenticatedRouter = {
|
||||
all: () => readonly OneTalkAuthenticatedRouteDefinition[];
|
||||
get: (type: OneTalkAuthenticatedClientFrame["type"]) => OneTalkAuthenticatedRouteDefinition;
|
||||
allowsConnectionType: (
|
||||
route: OneTalkAuthenticatedRouteDefinition<TContext>,
|
||||
route: OneTalkAuthenticatedRouteDefinition,
|
||||
connectionType: OneTalkConnectionType,
|
||||
) => boolean;
|
||||
dispatch: (context: TContext, frame: OneTalkAuthenticatedClientFrame) => Promise<void>;
|
||||
};
|
||||
|
||||
const authenticatedClientFrameTypes = (): readonly OneTalkAuthenticatedClientFrame["type"][] => {
|
||||
@@ -51,6 +43,76 @@ const authenticatedClientFrameTypes = (): readonly OneTalkAuthenticatedClientFra
|
||||
);
|
||||
};
|
||||
|
||||
const connectionTypesFor = (
|
||||
route: OneTalkAuthenticatedRouteDefinition,
|
||||
): readonly OneTalkConnectionType[] => {
|
||||
return Array.isArray(route.connectionType) ? route.connectionType : [route.connectionType];
|
||||
};
|
||||
|
||||
/** 已认证帧的端点、授权类别和 operation 的唯一声明。 */
|
||||
export const ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS = [
|
||||
{
|
||||
type: "heartbeat",
|
||||
connectionType: ["plugin", "mind_page"],
|
||||
authorization: "session",
|
||||
operation: "heartbeat",
|
||||
},
|
||||
{
|
||||
type: "conversation.discovered",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "conversations.discovered",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "sync.complete",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "contact.profile.observed",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "buyer.facts.observed",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "message.observed",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "messages.observed",
|
||||
connectionType: "plugin",
|
||||
authorization: "session",
|
||||
operation: "sync",
|
||||
},
|
||||
{
|
||||
type: "send.request",
|
||||
connectionType: "mind_page",
|
||||
authorization: "pending_send",
|
||||
operation: "send",
|
||||
},
|
||||
{
|
||||
type: "send.confirmation",
|
||||
connectionType: "plugin",
|
||||
authorization: "pending_send",
|
||||
operation: "send",
|
||||
},
|
||||
] as const satisfies readonly OneTalkAuthenticatedRouteDefinition[];
|
||||
|
||||
/** 判断已解码帧是否可进入已认证路由。 */
|
||||
export const isOneTalkAuthenticatedClientFrame = (
|
||||
frame: OneTalkFrame,
|
||||
@@ -58,13 +120,13 @@ export const isOneTalkAuthenticatedClientFrame = (
|
||||
return authenticatedClientFrameTypes().some((type) => type === frame.type);
|
||||
};
|
||||
|
||||
const assertValidAuthenticatedRoutes = <TContext>(
|
||||
routes: readonly OneTalkAuthenticatedRouteDefinition<TContext>[],
|
||||
): OneTalkAuthenticatedRouteTable<TContext> => {
|
||||
const assertValidAuthenticatedRoutes = (
|
||||
routes: readonly OneTalkAuthenticatedRouteDefinition[],
|
||||
): OneTalkAuthenticatedRouteTable => {
|
||||
const expected = new Set(authenticatedClientFrameTypes());
|
||||
const table = new Map<
|
||||
OneTalkAuthenticatedClientFrame["type"],
|
||||
OneTalkAuthenticatedRouteDefinition<TContext>
|
||||
OneTalkAuthenticatedRouteDefinition
|
||||
>();
|
||||
|
||||
for (const route of routes) {
|
||||
@@ -78,6 +140,20 @@ const assertValidAuthenticatedRoutes = <TContext>(
|
||||
if (table.has(route.type)) {
|
||||
throw new Error(`Duplicate OneTalk authenticated route: ${route.type}`);
|
||||
}
|
||||
const connectionTypes = connectionTypesFor(route);
|
||||
if (
|
||||
route.type === "heartbeat" &&
|
||||
(connectionTypes.length !== 2 ||
|
||||
!connectionTypes.includes("plugin") ||
|
||||
!connectionTypes.includes("mind_page"))
|
||||
) {
|
||||
throw new Error("OneTalk heartbeat route must be owned by plugin and mind_page");
|
||||
}
|
||||
if (route.type !== "heartbeat" && connectionTypes.length !== 1) {
|
||||
throw new Error(
|
||||
`OneTalk authenticated route ${route.type} must have exactly one endpoint owner`,
|
||||
);
|
||||
}
|
||||
table.set(route.type, route);
|
||||
}
|
||||
|
||||
@@ -88,41 +164,30 @@ const assertValidAuthenticatedRoutes = <TContext>(
|
||||
return table;
|
||||
};
|
||||
|
||||
const routeAllowsConnectionType = <TContext>(
|
||||
route: OneTalkAuthenticatedRouteDefinition<TContext>,
|
||||
const routeAllowsConnectionType = (
|
||||
route: OneTalkAuthenticatedRouteDefinition,
|
||||
connectionType: OneTalkConnectionType,
|
||||
): boolean => {
|
||||
const connectionTypes = Array.isArray(route.connectionType)
|
||||
? route.connectionType
|
||||
: [route.connectionType];
|
||||
return connectionTypes.includes(connectionType);
|
||||
return connectionTypesFor(route).includes(connectionType);
|
||||
};
|
||||
|
||||
/** 创建并校验完整、静态的已认证客户端帧路由表。 */
|
||||
export const createOneTalkAuthenticatedRouter = <TContext>(
|
||||
routes: readonly OneTalkAuthenticatedRouteDefinition<TContext>[],
|
||||
): OneTalkAuthenticatedRouter<TContext> => {
|
||||
export const createOneTalkAuthenticatedRouter = (
|
||||
routes: readonly OneTalkAuthenticatedRouteDefinition[] = ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS,
|
||||
): OneTalkAuthenticatedRouter => {
|
||||
const table = assertValidAuthenticatedRoutes(routes);
|
||||
|
||||
const get = (
|
||||
type: OneTalkAuthenticatedClientFrame["type"],
|
||||
): OneTalkAuthenticatedRouteDefinition<TContext> => {
|
||||
): OneTalkAuthenticatedRouteDefinition => {
|
||||
const route = table.get(type);
|
||||
if (!route) throw new Error(`Unknown OneTalk authenticated route: ${type}`);
|
||||
return route;
|
||||
};
|
||||
|
||||
return {
|
||||
all: () => [...table.values()],
|
||||
get,
|
||||
allowsConnectionType: routeAllowsConnectionType,
|
||||
dispatch: async (context, frame) => {
|
||||
const route = get(frame.type);
|
||||
if (!routeAllowsConnectionType(route, frame.connectionType)) {
|
||||
throw new Error(
|
||||
`OneTalk route ${frame.type} rejects ${frame.connectionType} connections`,
|
||||
);
|
||||
}
|
||||
await route.handler(context, frame, route);
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
@@ -1,65 +1,77 @@
|
||||
// 定义 WebSocket 端点专属的已认证帧路由。
|
||||
// 将共享路由元数据绑定到固定 WebSocket 端点的业务 handler。
|
||||
|
||||
import type { OneTalkConnectionType } from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import {
|
||||
ONETALK_CLIENT_FRAME_TYPES,
|
||||
type OneTalkConnectionType,
|
||||
} from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import {
|
||||
createOneTalkAuthenticatedRouter,
|
||||
type OneTalkAuthenticatedClientFrame,
|
||||
type OneTalkAuthenticatedRouteDefinition,
|
||||
type OneTalkAuthenticatedRouter,
|
||||
} from "./authenticated-router.ts";
|
||||
|
||||
type OneTalkEndpointRoutes<TContext> = {
|
||||
plugin: readonly OneTalkAuthenticatedRouteDefinition<TContext>[];
|
||||
mind_page: readonly OneTalkAuthenticatedRouteDefinition<TContext>[];
|
||||
export type OneTalkEndpointRouteHandler = {
|
||||
type: OneTalkAuthenticatedClientFrame["type"];
|
||||
matches: (frame: OneTalkAuthenticatedClientFrame) => boolean;
|
||||
handle: (frame: OneTalkAuthenticatedClientFrame) => Promise<void>;
|
||||
};
|
||||
|
||||
const routesForEndpoint = <TContext>(
|
||||
endpoint: OneTalkConnectionType,
|
||||
routes: OneTalkEndpointRoutes<TContext>,
|
||||
): readonly OneTalkAuthenticatedRouteDefinition<TContext>[] => {
|
||||
return endpoint === "plugin" ? routes.plugin : routes.mind_page;
|
||||
export type OneTalkEndpointAuthenticatedRouter = {
|
||||
dispatch: (frame: OneTalkAuthenticatedClientFrame) => Promise<void>;
|
||||
};
|
||||
|
||||
const assertEndpointRoutes = <TContext>(
|
||||
/** 在 runtime discriminant 收窄后调用端点的具体业务 handler。 */
|
||||
export const defineOneTalkEndpointRouteHandler = <
|
||||
TType extends OneTalkAuthenticatedClientFrame["type"],
|
||||
>(
|
||||
type: TType,
|
||||
handler: (frame: Extract<OneTalkAuthenticatedClientFrame, { type: TType }>) => Promise<void>,
|
||||
): OneTalkEndpointRouteHandler => {
|
||||
const matches = (
|
||||
frame: OneTalkAuthenticatedClientFrame,
|
||||
): frame is Extract<OneTalkAuthenticatedClientFrame, { type: TType }> => frame.type === type;
|
||||
return {
|
||||
type,
|
||||
matches,
|
||||
handle: async (frame) => {
|
||||
if (!matches(frame)) throw new Error(`OneTalk route handler rejects ${frame.type}`);
|
||||
await handler(frame);
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const assertEndpointHandlers = (
|
||||
endpoint: OneTalkConnectionType,
|
||||
routes: readonly OneTalkAuthenticatedRouteDefinition<TContext>[],
|
||||
router: OneTalkAuthenticatedRouter,
|
||||
handlers: readonly OneTalkEndpointRouteHandler[],
|
||||
): void => {
|
||||
for (const route of routes) {
|
||||
const connectionTypes = Array.isArray(route.connectionType)
|
||||
? route.connectionType
|
||||
: [route.connectionType];
|
||||
if (!connectionTypes.includes(endpoint)) {
|
||||
throw new Error(`OneTalk ${endpoint} route ${route.type} has no endpoint ownership`);
|
||||
}
|
||||
const expected = new Set(
|
||||
router
|
||||
.all()
|
||||
.filter((route) => router.allowsConnectionType(route, endpoint))
|
||||
.map((route) => route.type),
|
||||
);
|
||||
const actual = new Set(handlers.map((handler) => handler.type));
|
||||
if (actual.size !== handlers.length)
|
||||
throw new Error(`Duplicate OneTalk ${endpoint} route handler`);
|
||||
const unknown = handlers.filter((handler) => !expected.has(handler.type));
|
||||
if (unknown.length > 0) {
|
||||
throw new Error(
|
||||
`OneTalk ${endpoint} route handler has no endpoint ownership: ${unknown[0].type}`,
|
||||
);
|
||||
}
|
||||
const missing = [...expected].filter((type) => !actual.has(type));
|
||||
if (missing.length > 0) {
|
||||
throw new Error(`Missing OneTalk ${endpoint} route handlers: ${missing.join(", ")}`);
|
||||
}
|
||||
};
|
||||
|
||||
/** 校验并创建固定端点的完整路由视图。 */
|
||||
export const createOneTalkEndpointAuthenticatedRouter = <TContext>(
|
||||
/** 以共享 metadata 校验端点 ownership，并分发给唯一绑定的业务 handler。 */
|
||||
export const createOneTalkEndpointAuthenticatedRouter = (
|
||||
endpoint: OneTalkConnectionType,
|
||||
routes: OneTalkEndpointRoutes<TContext>,
|
||||
): OneTalkAuthenticatedRouter<TContext> => {
|
||||
const endpointRoutes = routesForEndpoint(endpoint, routes);
|
||||
assertEndpointRoutes(endpoint, endpointRoutes);
|
||||
|
||||
// The shared router remains the single exhaustive contract validator. Each endpoint
|
||||
// receives only its own view while heartbeat deliberately belongs to both views.
|
||||
const allRoutes = [
|
||||
...routes.plugin,
|
||||
...routes.mind_page.filter(
|
||||
(candidate) => !routes.plugin.some((route) => route.type === candidate.type),
|
||||
),
|
||||
];
|
||||
const router = createOneTalkAuthenticatedRouter(allRoutes);
|
||||
router: OneTalkAuthenticatedRouter,
|
||||
handlers: readonly OneTalkEndpointRouteHandler[],
|
||||
): OneTalkEndpointAuthenticatedRouter => {
|
||||
assertEndpointHandlers(endpoint, router, handlers);
|
||||
return {
|
||||
get: router.get,
|
||||
allowsConnectionType: (route, connectionType) =>
|
||||
connectionType === endpoint && router.allowsConnectionType(route, connectionType),
|
||||
dispatch: async (context, frame) => {
|
||||
dispatch: async (frame) => {
|
||||
const route = router.get(frame.type);
|
||||
if (
|
||||
frame.connectionType !== endpoint ||
|
||||
@@ -69,45 +81,10 @@ export const createOneTalkEndpointAuthenticatedRouter = <TContext>(
|
||||
`OneTalk endpoint ${endpoint} rejects ${frame.connectionType} ${frame.type}`,
|
||||
);
|
||||
}
|
||||
await router.dispatch(context, frame);
|
||||
const handler = handlers.find((candidate) => candidate.type === frame.type);
|
||||
if (!handler)
|
||||
throw new Error(`Missing OneTalk ${endpoint} route handler: ${frame.type}`);
|
||||
await handler.handle(frame);
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
/** 校验两端静态帧归属覆盖完整协议，heartbeat 是唯一共享帧。 */
|
||||
export const assertOneTalkEndpointFrameSets = (sets: {
|
||||
plugin: readonly OneTalkAuthenticatedClientFrame["type"][];
|
||||
mind_page: readonly OneTalkAuthenticatedClientFrame["type"][];
|
||||
}): void => {
|
||||
const expected = ONETALK_CLIENT_FRAME_TYPES.filter(
|
||||
(type): type is OneTalkAuthenticatedClientFrame["type"] => type !== "ws.hello",
|
||||
);
|
||||
const plugin = new Set(sets.plugin);
|
||||
const mind = new Set(sets.mind_page);
|
||||
const unknown = [...plugin, ...mind].filter((type) => !expected.includes(type));
|
||||
if (unknown.length > 0)
|
||||
throw new Error(`Unknown OneTalk endpoint routes: ${unknown.join(", ")}`);
|
||||
if (
|
||||
sets.plugin.filter((type) => type === "heartbeat").length !== 1 ||
|
||||
sets.mind_page.filter((type) => type === "heartbeat").length !== 1
|
||||
) {
|
||||
throw new Error("OneTalk endpoint routes require heartbeat exactly once per endpoint");
|
||||
}
|
||||
if (plugin.size !== sets.plugin.length || mind.size !== sets.mind_page.length) {
|
||||
throw new Error("Duplicate OneTalk endpoint route");
|
||||
}
|
||||
const covered = new Set([...plugin, ...mind]);
|
||||
const missing = expected.filter((type) => !covered.has(type));
|
||||
if (missing.length > 0)
|
||||
throw new Error(`Missing OneTalk endpoint routes: ${missing.join(", ")}`);
|
||||
for (const type of expected) {
|
||||
if (type !== "heartbeat" && plugin.has(type) === mind.has(type)) {
|
||||
throw new Error(`OneTalk endpoint route must have one owner: ${type}`);
|
||||
}
|
||||
}
|
||||
for (const type of plugin) {
|
||||
if (mind.has(type) && type !== "heartbeat") {
|
||||
throw new Error(`Duplicate OneTalk endpoint route: ${type}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@@ -9,9 +9,8 @@ import {
|
||||
} from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import { createOneTalkMindWebSocketHandler } from "./mind/index.ts";
|
||||
import { ONETALK_MIND_ROUTE_TYPES } from "./mind/index.ts";
|
||||
import { ONETALK_PLUGIN_ROUTE_TYPES, createOneTalkPluginWebSocketHandler } from "./plugin/index.ts";
|
||||
import { assertOneTalkEndpointFrameSets } from "./endpoint-routes.ts";
|
||||
import { createOneTalkPluginWebSocketHandler } from "./plugin/index.ts";
|
||||
import { createOneTalkAuthenticatedRouter } from "./authenticated-router.ts";
|
||||
import { createOneTalkConnectionRegistry, type OneTalkConnectionRegistry } from "./registry.ts";
|
||||
import type { OneTalkPublishFailureSink } from "./mind/publisher.ts";
|
||||
import type {
|
||||
@@ -63,10 +62,7 @@ const registerWebsocketRoutes = (
|
||||
cutoverPolicy: OneTalkCutoverPolicy | undefined,
|
||||
): FastifyPluginCallback => {
|
||||
return (app, _options, done) => {
|
||||
assertOneTalkEndpointFrameSets({
|
||||
plugin: ONETALK_PLUGIN_ROUTE_TYPES,
|
||||
mind_page: ONETALK_MIND_ROUTE_TYPES,
|
||||
});
|
||||
const authenticatedRouter = createOneTalkAuthenticatedRouter();
|
||||
app.addHook("onRequest", async (request, reply) => {
|
||||
const origin = request.headers.origin;
|
||||
const routeType = request.url.startsWith("/ws/mind")
|
||||
@@ -121,6 +117,7 @@ const registerWebsocketRoutes = (
|
||||
...(pluginOrigins === undefined ? {} : { pluginOrigins }),
|
||||
},
|
||||
cutoverPolicy,
|
||||
authenticatedRouter,
|
||||
});
|
||||
app.get("/ws/plugin", { websocket: true }, (socket, request) =>
|
||||
createOneTalkPluginWebSocketHandler(endpointOptions(request))(socket),
|
||||
|
||||
@@ -14,11 +14,14 @@ import {
|
||||
sessionAuthorizationFailure,
|
||||
type OneTalkWebSocketHandlerOptions,
|
||||
} from "../handler.ts";
|
||||
import {
|
||||
createOneTalkEndpointAuthenticatedRouter,
|
||||
defineOneTalkEndpointRouteHandler,
|
||||
} from "../endpoint-routes.ts";
|
||||
import type { OneTalkAuthenticatedRouter } from "../authenticated-router.ts";
|
||||
import { createOneTalkSendRequestFlow } from "./send-request-flow.ts";
|
||||
import { OneTalkMindSessionAuthorization } from "./session-authorization.ts";
|
||||
|
||||
export const ONETALK_MIND_ROUTE_TYPES = ["heartbeat", "send.request"] as const;
|
||||
|
||||
const CLOSE_POLICY_VIOLATION = 1008;
|
||||
const CLOSE_INTERNAL_ERROR = 1011;
|
||||
export type OneTalkMindWebSocketHandlerOptions = Omit<
|
||||
@@ -26,6 +29,7 @@ export type OneTalkMindWebSocketHandlerOptions = Omit<
|
||||
"expectedConnectionType"
|
||||
> & {
|
||||
// Mind only owns the request half of the send bridge.
|
||||
authenticatedRouter: OneTalkAuthenticatedRouter;
|
||||
};
|
||||
|
||||
/** 创建只接纳 Mind Cookie/session 会话的发送端点处理器。 */
|
||||
@@ -44,6 +48,47 @@ export const createOneTalkMindWebSocketHandler = (
|
||||
type === "mind_page" && context.isPolicyCurrent(epoch),
|
||||
closeForPause: context.closeForPause,
|
||||
});
|
||||
const authenticatedRouter = createOneTalkEndpointAuthenticatedRouter(
|
||||
"mind_page",
|
||||
options.authenticatedRouter,
|
||||
[
|
||||
defineOneTalkEndpointRouteHandler("send.request", async (frame) => {
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
await sendRequestFlow.handleRequest(socket, frame, epoch);
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler("heartbeat", async (frame) => {
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
const route = options.authenticatedRouter.get(frame.type);
|
||||
if (!isOneTalkMindScope(frame.scope)) {
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.authorizationRejected);
|
||||
return;
|
||||
}
|
||||
const decision = await (state.sessionAuthorization
|
||||
? state.sessionAuthorization.authorize(frame.scope, "heartbeat")
|
||||
: context.authorize({
|
||||
connectionType: "mind_page",
|
||||
operation: route.operation,
|
||||
scope: frame.scope,
|
||||
}));
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
const code = sessionAuthorizationFailure(state, decision, "read");
|
||||
if (code !== null) {
|
||||
const authorizationDenied = !decision.allowed;
|
||||
if (code !== ONETALK_ERROR_CODES.authorizationRejected)
|
||||
state.unregister?.();
|
||||
context.sendError(frame, code);
|
||||
if (
|
||||
authorizationDenied ||
|
||||
code !== ONETALK_ERROR_CODES.authorizationRejected
|
||||
)
|
||||
context.close(CLOSE_POLICY_VIOLATION, code);
|
||||
return;
|
||||
}
|
||||
if (context.options.registry.recordHeartbeat(socket))
|
||||
context.sendFrame(createOneTalkHeartbeatAckFrame(frame));
|
||||
}),
|
||||
],
|
||||
);
|
||||
return {
|
||||
onHello: async (frame) => {
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
@@ -122,38 +167,7 @@ export const createOneTalkMindWebSocketHandler = (
|
||||
requestId: frame.requestId,
|
||||
});
|
||||
},
|
||||
onAuthenticatedFrame: async (frame) => {
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
if (frame.type === "send.request")
|
||||
return sendRequestFlow.handleRequest(socket, frame, epoch);
|
||||
if (frame.type !== "heartbeat" || !isOneTalkMindScope(frame.scope)) {
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.authorizationRejected);
|
||||
return;
|
||||
}
|
||||
const decision = await (state.sessionAuthorization
|
||||
? state.sessionAuthorization.authorize(frame.scope, "heartbeat")
|
||||
: context.authorize({
|
||||
connectionType: "mind_page",
|
||||
operation: "heartbeat",
|
||||
scope: frame.scope,
|
||||
}));
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
const code = sessionAuthorizationFailure(state, decision, "read");
|
||||
if (code !== null) {
|
||||
const authorizationDenied = !decision.allowed;
|
||||
if (code !== ONETALK_ERROR_CODES.authorizationRejected)
|
||||
state.unregister?.();
|
||||
context.sendError(frame, code);
|
||||
if (
|
||||
authorizationDenied ||
|
||||
code !== ONETALK_ERROR_CODES.authorizationRejected
|
||||
)
|
||||
context.close(CLOSE_POLICY_VIOLATION, code);
|
||||
return;
|
||||
}
|
||||
if (context.options.registry.recordHeartbeat(socket))
|
||||
context.sendFrame(createOneTalkHeartbeatAckFrame(frame));
|
||||
},
|
||||
onAuthenticatedFrame: authenticatedRouter.dispatch,
|
||||
};
|
||||
},
|
||||
},
|
||||
|
||||
@@ -26,23 +26,19 @@ import {
|
||||
type OneTalkWebSocketEndpointContext,
|
||||
type OneTalkWebSocketHandlerOptions,
|
||||
} from "../handler.ts";
|
||||
import {
|
||||
createOneTalkEndpointAuthenticatedRouter,
|
||||
defineOneTalkEndpointRouteHandler,
|
||||
} from "../endpoint-routes.ts";
|
||||
import type {
|
||||
OneTalkAuthenticatedClientFrame,
|
||||
OneTalkAuthenticatedRouter,
|
||||
} from "../authenticated-router.ts";
|
||||
import { createOneTalkProfileFlow } from "./flows/profile-flow.ts";
|
||||
import { createOneTalkSendConfirmationFlow } from "./flows/send-confirmation-flow.ts";
|
||||
import { createOneTalkSyncFlows } from "./flows/sync-flows.ts";
|
||||
import type { OneTalkRegisteredConnection } from "../registry.ts";
|
||||
|
||||
export const ONETALK_PLUGIN_ROUTE_TYPES = [
|
||||
"heartbeat",
|
||||
"conversation.discovered",
|
||||
"conversations.discovered",
|
||||
"sync.complete",
|
||||
"contact.profile.observed",
|
||||
"buyer.facts.observed",
|
||||
"message.observed",
|
||||
"messages.observed",
|
||||
"send.confirmation",
|
||||
] as const;
|
||||
|
||||
const CLOSE_POLICY_VIOLATION = 1008;
|
||||
const CLOSE_UNSUPPORTED_DATA = 1003;
|
||||
const CLOSE_INTERNAL_ERROR = 1011;
|
||||
@@ -54,6 +50,7 @@ export type OneTalkPluginWebSocketHandlerOptions = Omit<
|
||||
profileService: OneTalkProfileService;
|
||||
buyerFactService?: OneTalkBuyerFactService;
|
||||
readService?: OneTalkReadService;
|
||||
authenticatedRouter: OneTalkAuthenticatedRouter;
|
||||
};
|
||||
const isCommitGuardFailure = (error: unknown): boolean =>
|
||||
error instanceof Error && error.message === "connection_commit_invalid";
|
||||
@@ -188,6 +185,231 @@ export const createOneTalkPluginWebSocketHandler = (
|
||||
input.policyEpoch,
|
||||
),
|
||||
});
|
||||
const authorizeSessionRoute = async (
|
||||
frame: OneTalkAuthenticatedClientFrame,
|
||||
): Promise<number | null> => {
|
||||
if (!isOneTalkPluginScope(frame.scope)) {
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.authorizationRejected);
|
||||
return null;
|
||||
}
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
const route = options.authenticatedRouter.get(frame.type);
|
||||
const decision = await context.authorize({
|
||||
connectionType: "plugin",
|
||||
operation: route.operation,
|
||||
scope: frame.scope,
|
||||
binding: state.binding ?? "",
|
||||
});
|
||||
if (!context.isPolicyCurrent(epoch)) {
|
||||
context.closeForPause();
|
||||
return null;
|
||||
}
|
||||
const code = sessionAuthorizationFailure(
|
||||
state,
|
||||
decision,
|
||||
route.operation === "heartbeat" ? null : "read",
|
||||
);
|
||||
if (code === null) return epoch;
|
||||
const authorizationDenied = !decision.allowed;
|
||||
if (code !== ONETALK_ERROR_CODES.authorizationRejected) state.unregister?.();
|
||||
context.sendError(frame, code);
|
||||
if (authorizationDenied || code !== ONETALK_ERROR_CODES.authorizationRejected)
|
||||
context.close(CLOSE_POLICY_VIOLATION, code);
|
||||
return null;
|
||||
};
|
||||
const authenticatedRouter = createOneTalkEndpointAuthenticatedRouter(
|
||||
"plugin",
|
||||
options.authenticatedRouter,
|
||||
[
|
||||
defineOneTalkEndpointRouteHandler("heartbeat", async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch !== null && context.options.registry.recordHeartbeat(socket))
|
||||
context.sendFrame(createOneTalkHeartbeatAckFrame(frame));
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler("send.confirmation", async (frame) =>
|
||||
sendConfirmationFlow.handleConfirmation(socket, frame),
|
||||
),
|
||||
defineOneTalkEndpointRouteHandler(
|
||||
"conversation.discovered",
|
||||
async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical
|
||||
? sourceContextForConnection(canonical)
|
||||
: null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await syncFlows.handleDiscovery(frame, {
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(
|
||||
canonical,
|
||||
epoch,
|
||||
),
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
},
|
||||
),
|
||||
defineOneTalkEndpointRouteHandler(
|
||||
"conversations.discovered",
|
||||
async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical
|
||||
? sourceContextForConnection(canonical)
|
||||
: null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await syncFlows.handleDiscovery(frame, {
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(
|
||||
canonical,
|
||||
epoch,
|
||||
),
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
},
|
||||
),
|
||||
defineOneTalkEndpointRouteHandler("sync.complete", async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical ? sourceContextForConnection(canonical) : null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await syncFlows.handleCompletion(frame, {
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(canonical, epoch),
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler(
|
||||
"contact.profile.observed",
|
||||
async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical
|
||||
? sourceContextForConnection(canonical)
|
||||
: null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await profileFlow.handle({
|
||||
socket,
|
||||
frame,
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(
|
||||
canonical,
|
||||
epoch,
|
||||
),
|
||||
policyEpoch: epoch,
|
||||
canonical,
|
||||
});
|
||||
},
|
||||
),
|
||||
defineOneTalkEndpointRouteHandler("buyer.facts.observed", async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical ? sourceContextForConnection(canonical) : null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
const guard = context.options.registry.createCommitGuard(
|
||||
canonical,
|
||||
epoch,
|
||||
);
|
||||
if (!options.buyerFactService)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.databaseUnavailable,
|
||||
);
|
||||
try {
|
||||
const result = await options.buyerFactService.ingestFacts({
|
||||
channelAccountId: source.channelAccountId,
|
||||
facts: frame.payload.facts,
|
||||
commitGuard: guard,
|
||||
});
|
||||
guard.assertValid();
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
if (
|
||||
context.options.registry.getCanonicalConnection(socket) !==
|
||||
canonical
|
||||
)
|
||||
return;
|
||||
const current = await reauthorize(context, frame);
|
||||
guard.assertValid();
|
||||
if (!current.ok) {
|
||||
state.unregister?.();
|
||||
context.sendError(frame, current.code);
|
||||
context.close(CLOSE_POLICY_VIOLATION, current.code);
|
||||
return;
|
||||
}
|
||||
context.sendFrame(
|
||||
createOneTalkBuyerFactsAckFrame(frame, result.inputFactCount),
|
||||
);
|
||||
} catch (error: unknown) {
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
if (isCommitGuardFailure(error)) return;
|
||||
context.closeForDatabaseFailure(frame);
|
||||
}
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler("message.observed", async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical ? sourceContextForConnection(canonical) : null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await syncFlows.handleObservation(frame, {
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(canonical, epoch),
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler("messages.observed", async (frame) => {
|
||||
const epoch = await authorizeSessionRoute(frame);
|
||||
if (epoch === null) return;
|
||||
const canonical =
|
||||
context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical ? sourceContextForConnection(canonical) : null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
await syncFlows.handleObservation(frame, {
|
||||
context: source,
|
||||
guard: context.options.registry.createCommitGuard(canonical, epoch),
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
}),
|
||||
],
|
||||
);
|
||||
return {
|
||||
onHello: async (frame) => {
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
@@ -270,124 +492,7 @@ export const createOneTalkPluginWebSocketHandler = (
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
onAuthenticatedFrame: async (frame) => {
|
||||
if (frame.type === "send.request") {
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.authorizationRejected);
|
||||
return;
|
||||
}
|
||||
const epoch = context.capturePolicyEpoch();
|
||||
if (frame.type === "send.confirmation")
|
||||
return sendConfirmationFlow.handleConfirmation(socket, frame);
|
||||
if (!isOneTalkPluginScope(frame.scope)) {
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.authorizationRejected);
|
||||
return;
|
||||
}
|
||||
const decision = await context.authorize({
|
||||
connectionType: "plugin",
|
||||
operation: frame.type === "heartbeat" ? "heartbeat" : "sync",
|
||||
scope: frame.scope,
|
||||
binding: state.binding ?? "",
|
||||
});
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
const code = sessionAuthorizationFailure(
|
||||
state,
|
||||
decision,
|
||||
frame.type === "heartbeat" ? null : "read",
|
||||
);
|
||||
if (code !== null) {
|
||||
const authorizationDenied = !decision.allowed;
|
||||
if (code !== ONETALK_ERROR_CODES.authorizationRejected)
|
||||
state.unregister?.();
|
||||
context.sendError(frame, code);
|
||||
if (
|
||||
authorizationDenied ||
|
||||
code !== ONETALK_ERROR_CODES.authorizationRejected
|
||||
)
|
||||
context.close(CLOSE_POLICY_VIOLATION, code);
|
||||
return;
|
||||
}
|
||||
if (frame.type === "heartbeat") {
|
||||
if (context.options.registry.recordHeartbeat(socket))
|
||||
context.sendFrame(createOneTalkHeartbeatAckFrame(frame));
|
||||
return;
|
||||
}
|
||||
const canonical = context.options.registry.getCanonicalConnection(socket);
|
||||
const source = canonical ? sourceContextForConnection(canonical) : null;
|
||||
if (!canonical || !source)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.authorizationRejected,
|
||||
);
|
||||
const guard = context.options.registry.createCommitGuard(canonical, epoch);
|
||||
if (frame.type === "buyer.facts.observed") {
|
||||
if (!options.buyerFactService)
|
||||
return context.sendError(
|
||||
frame,
|
||||
ONETALK_ERROR_CODES.databaseUnavailable,
|
||||
);
|
||||
try {
|
||||
const result = await options.buyerFactService.ingestFacts({
|
||||
channelAccountId: source.channelAccountId,
|
||||
facts: frame.payload.facts,
|
||||
commitGuard: guard,
|
||||
});
|
||||
guard.assertValid();
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
if (
|
||||
context.options.registry.getCanonicalConnection(socket) !==
|
||||
canonical
|
||||
)
|
||||
return;
|
||||
const current = await reauthorize(context, frame);
|
||||
guard.assertValid();
|
||||
if (!current.ok) {
|
||||
state.unregister?.();
|
||||
context.sendError(frame, current.code);
|
||||
context.close(CLOSE_POLICY_VIOLATION, current.code);
|
||||
return;
|
||||
}
|
||||
context.sendFrame(
|
||||
createOneTalkBuyerFactsAckFrame(frame, result.inputFactCount),
|
||||
);
|
||||
} catch (error: unknown) {
|
||||
if (!context.isPolicyCurrent(epoch)) return context.closeForPause();
|
||||
if (isCommitGuardFailure(error)) return;
|
||||
context.closeForDatabaseFailure(frame);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (frame.type === "contact.profile.observed")
|
||||
return profileFlow.handle({
|
||||
socket,
|
||||
frame,
|
||||
context: source,
|
||||
guard,
|
||||
policyEpoch: epoch,
|
||||
canonical,
|
||||
});
|
||||
if (
|
||||
frame.type === "conversation.discovered" ||
|
||||
frame.type === "conversations.discovered"
|
||||
)
|
||||
return syncFlows.handleDiscovery(frame, {
|
||||
context: source,
|
||||
guard,
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
if (frame.type === "sync.complete")
|
||||
return syncFlows.handleCompletion(frame, {
|
||||
context: source,
|
||||
guard,
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
if (frame.type === "message.observed" || frame.type === "messages.observed")
|
||||
return syncFlows.handleObservation(frame, {
|
||||
context: source,
|
||||
guard,
|
||||
policyEpoch: epoch,
|
||||
});
|
||||
context.sendError(frame, ONETALK_ERROR_CODES.unknownRequest);
|
||||
},
|
||||
onAuthenticatedFrame: authenticatedRouter.dispatch,
|
||||
dispose: syncFlows.dispose,
|
||||
};
|
||||
},
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// 验证 OneTalk 已认证帧路由的完整性与拒绝边界。
|
||||
// 验证 OneTalk canonical authenticated route metadata 与端点 handler 绑定。
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
@@ -6,185 +6,116 @@ import {
|
||||
ONETALK_CLIENT_FRAME_TYPES,
|
||||
ONETALK_PROTOCOL_VERSION,
|
||||
} from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import {
|
||||
ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS,
|
||||
type OneTalkAuthenticatedClientFrame,
|
||||
type OneTalkAuthenticatedRouteDefinition,
|
||||
createOneTalkAuthenticatedRouter,
|
||||
} from "../src/websocket/authenticated-router.ts";
|
||||
import {
|
||||
assertOneTalkEndpointFrameSets,
|
||||
createOneTalkEndpointAuthenticatedRouter,
|
||||
defineOneTalkEndpointRouteHandler,
|
||||
} from "../src/websocket/endpoint-routes.ts";
|
||||
|
||||
type RouteContext = { invocations: number };
|
||||
|
||||
const authenticatedFrameTypes = ONETALK_CLIENT_FRAME_TYPES.filter(
|
||||
(type): type is OneTalkAuthenticatedClientFrame["type"] => type !== "ws.hello",
|
||||
);
|
||||
|
||||
const createCompleteRoutes = (): OneTalkAuthenticatedRouteDefinition<RouteContext>[] => {
|
||||
return authenticatedFrameTypes.map((type) => ({
|
||||
type,
|
||||
connectionType: type === "send.request" ? "mind_page" : "plugin",
|
||||
authorization:
|
||||
type === "send.request" || type === "send.confirmation" ? "pending_send" : "session",
|
||||
operation: type === "heartbeat" ? "heartbeat" : type.startsWith("send.") ? "send" : "sync",
|
||||
handler: async (context: RouteContext): Promise<void> => {
|
||||
context.invocations += 1;
|
||||
},
|
||||
}));
|
||||
};
|
||||
|
||||
test("authenticated router requires every non-hello contract client frame exactly once", () => {
|
||||
const router = createOneTalkAuthenticatedRouter(createCompleteRoutes());
|
||||
|
||||
test("canonical metadata covers every non-hello client frame exactly once", () => {
|
||||
const router = createOneTalkAuthenticatedRouter();
|
||||
assert.deepEqual(
|
||||
authenticatedFrameTypes.map((type) => router.get(type).type).sort(),
|
||||
[...authenticatedFrameTypes].sort(),
|
||||
);
|
||||
assert.equal(router.get("send.request").authorization, "pending_send");
|
||||
assert.equal(router.get("send.confirmation").authorization, "pending_send");
|
||||
});
|
||||
|
||||
test("authenticated router rejects missing, duplicate, unknown, and hello routes", () => {
|
||||
const routes = createCompleteRoutes();
|
||||
|
||||
test("canonical metadata rejects missing, duplicate, unknown, hello, and invalid endpoint ownership", () => {
|
||||
assert.throws(
|
||||
() => createOneTalkAuthenticatedRouter(routes.slice(1)),
|
||||
() => createOneTalkAuthenticatedRouter(ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS.slice(1)),
|
||||
/Missing OneTalk authenticated routes: heartbeat/,
|
||||
);
|
||||
assert.throws(
|
||||
() => createOneTalkAuthenticatedRouter([...routes, routes[0]]),
|
||||
() =>
|
||||
createOneTalkAuthenticatedRouter([
|
||||
...ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS,
|
||||
ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS[0],
|
||||
]),
|
||||
/Duplicate OneTalk authenticated route: heartbeat/,
|
||||
);
|
||||
assert.throws(
|
||||
() =>
|
||||
createOneTalkAuthenticatedRouter([
|
||||
...routes,
|
||||
{
|
||||
...routes[0],
|
||||
type: "ws.hello" as never,
|
||||
},
|
||||
...ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS,
|
||||
{ ...ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS[0], type: "ws.hello" as never },
|
||||
]),
|
||||
/ws\.hello must not be an authenticated route/,
|
||||
);
|
||||
assert.throws(
|
||||
() =>
|
||||
createOneTalkAuthenticatedRouter([
|
||||
...routes,
|
||||
{
|
||||
...routes[0],
|
||||
type: "unknown.route" as never,
|
||||
},
|
||||
...ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS,
|
||||
{ ...ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS[0], type: "unknown.route" as never },
|
||||
]),
|
||||
/Unknown OneTalk authenticated route: unknown\.route/,
|
||||
);
|
||||
assert.throws(
|
||||
() =>
|
||||
createOneTalkAuthenticatedRouter(
|
||||
ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS.map((route) =>
|
||||
route.type === "conversation.discovered"
|
||||
? { ...route, connectionType: ["plugin", "mind_page"] as const }
|
||||
: route,
|
||||
),
|
||||
),
|
||||
/conversation\.discovered must have exactly one endpoint owner/,
|
||||
);
|
||||
});
|
||||
|
||||
test("connection-type mismatch cannot invoke a route collaborator", async () => {
|
||||
const router = createOneTalkAuthenticatedRouter(createCompleteRoutes());
|
||||
const context = { invocations: 0 };
|
||||
test("endpoint view rejects ownership mismatch before invoking business handlers", async () => {
|
||||
const router = createOneTalkAuthenticatedRouter();
|
||||
let invocations = 0;
|
||||
const endpoint = createOneTalkEndpointAuthenticatedRouter("mind_page", router, [
|
||||
defineOneTalkEndpointRouteHandler("heartbeat", async () => {
|
||||
invocations += 1;
|
||||
}),
|
||||
defineOneTalkEndpointRouteHandler("send.request", async () => {
|
||||
invocations += 1;
|
||||
}),
|
||||
]);
|
||||
const frame = {
|
||||
protocolVersion: ONETALK_PROTOCOL_VERSION,
|
||||
connectionType: "mind_page",
|
||||
connectionType: "plugin",
|
||||
type: "heartbeat",
|
||||
requestId: "wrong-connection-type",
|
||||
scope: {
|
||||
channelAccountId: "account-1",
|
||||
mindUserId: "mind-user-1",
|
||||
workspaceId: "workspace-1",
|
||||
},
|
||||
payload: {},
|
||||
} as OneTalkAuthenticatedClientFrame;
|
||||
requestId: "wrong-endpoint",
|
||||
scope: { channelAccountId: "account-1", deviceId: "device-1" },
|
||||
payload: { sentAtMs: 1_700_000_000_000 },
|
||||
} satisfies OneTalkAuthenticatedClientFrame;
|
||||
|
||||
await assert.rejects(() => router.dispatch(context, frame), /rejects mind_page connections/);
|
||||
assert.equal(context.invocations, 0);
|
||||
});
|
||||
|
||||
test("endpoint router accepts the shared heartbeat but keeps business frames endpoint-owned", async () => {
|
||||
const routes = createCompleteRoutes();
|
||||
const plugin = routes
|
||||
.filter((route) => route.type !== "send.request")
|
||||
.map((route) => ({
|
||||
...route,
|
||||
connectionType:
|
||||
route.type === "heartbeat"
|
||||
? (["plugin", "mind_page"] as const)
|
||||
: route.connectionType,
|
||||
}));
|
||||
const mindPage: OneTalkAuthenticatedRouteDefinition<RouteContext>[] = routes
|
||||
.filter((route) => route.type === "heartbeat" || route.type === "send.request")
|
||||
.map((route) => ({
|
||||
...route,
|
||||
connectionType:
|
||||
route.type === "heartbeat" ? (["plugin", "mind_page"] as const) : "mind_page",
|
||||
}));
|
||||
const router = createOneTalkEndpointAuthenticatedRouter("mind_page", {
|
||||
plugin,
|
||||
mind_page: mindPage,
|
||||
});
|
||||
|
||||
assert.equal(router.allowsConnectionType(router.get("heartbeat"), "mind_page"), true);
|
||||
assert.equal(router.allowsConnectionType(router.get("send.request"), "mind_page"), true);
|
||||
assert.equal(
|
||||
router.allowsConnectionType(router.get("conversation.discovered"), "mind_page"),
|
||||
false,
|
||||
);
|
||||
await assert.rejects(
|
||||
() =>
|
||||
router.dispatch(
|
||||
{ invocations: 0 },
|
||||
{
|
||||
...({
|
||||
protocolVersion: ONETALK_PROTOCOL_VERSION,
|
||||
connectionType: "plugin",
|
||||
type: "heartbeat",
|
||||
requestId: "wrong-endpoint",
|
||||
scope: { channelAccountId: "account-1", deviceId: "device-1" },
|
||||
payload: { sentAtMs: 1_700_000_000_000 },
|
||||
} satisfies OneTalkAuthenticatedClientFrame),
|
||||
},
|
||||
),
|
||||
() => endpoint.dispatch(frame),
|
||||
/endpoint mind_page rejects plugin heartbeat/,
|
||||
);
|
||||
assert.equal(invocations, 0);
|
||||
});
|
||||
|
||||
test("endpoint route validation requires heartbeat on both sides and one owner per other frame", () => {
|
||||
const routes = createCompleteRoutes();
|
||||
const plugin = routes
|
||||
.filter((route) => route.type !== "send.request")
|
||||
.map((route) => ({
|
||||
...route,
|
||||
connectionType:
|
||||
route.type === "heartbeat"
|
||||
? (["plugin", "mind_page"] as const)
|
||||
: route.connectionType,
|
||||
}));
|
||||
const mindPage = routes
|
||||
.filter((route) => route.type === "heartbeat" || route.type === "send.request")
|
||||
.map((route) => ({
|
||||
...route,
|
||||
connectionType:
|
||||
route.type === "heartbeat" ? (["plugin", "mind_page"] as const) : "mind_page",
|
||||
}));
|
||||
const pluginTypes = plugin.map((route) => route.type);
|
||||
const mindTypes = mindPage.map((route) => route.type);
|
||||
|
||||
assert.doesNotThrow(() =>
|
||||
assertOneTalkEndpointFrameSets({ plugin: pluginTypes, mind_page: mindTypes }),
|
||||
test("endpoint view requires an exact handler table derived from canonical metadata", () => {
|
||||
const router = createOneTalkAuthenticatedRouter();
|
||||
assert.throws(
|
||||
() =>
|
||||
createOneTalkEndpointAuthenticatedRouter("mind_page", router, [
|
||||
defineOneTalkEndpointRouteHandler("heartbeat", async () => {}),
|
||||
]),
|
||||
/Missing OneTalk mind_page route handlers: send.request/,
|
||||
);
|
||||
assert.throws(
|
||||
() =>
|
||||
assertOneTalkEndpointFrameSets({
|
||||
plugin: pluginTypes.filter((type) => type !== "heartbeat"),
|
||||
mind_page: mindTypes,
|
||||
}),
|
||||
/heartbeat exactly once per endpoint/,
|
||||
);
|
||||
assert.throws(
|
||||
() =>
|
||||
assertOneTalkEndpointFrameSets({
|
||||
plugin: pluginTypes,
|
||||
mind_page: [...mindTypes, "conversation.discovered"],
|
||||
}),
|
||||
/one owner|Duplicate OneTalk endpoint route/,
|
||||
createOneTalkEndpointAuthenticatedRouter("mind_page", router, [
|
||||
defineOneTalkEndpointRouteHandler("heartbeat", async () => {}),
|
||||
defineOneTalkEndpointRouteHandler("send.request", async () => {}),
|
||||
defineOneTalkEndpointRouteHandler("conversation.discovered", async () => {}),
|
||||
]),
|
||||
/has no endpoint ownership: conversation.discovered/,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -256,6 +256,102 @@ test("accepts a mock-authorized plugin handshake and heartbeat", async () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects a legal Mind send frame on the plugin endpoint before business dispatch", async () => {
|
||||
const authorization = createMockAuthorizationReader([authorizationRecord]);
|
||||
let sendRequests = 0;
|
||||
const delegate = createOneTalkConnectionRegistry({
|
||||
authorization,
|
||||
onPublishFailure: () => {},
|
||||
});
|
||||
const registry: OneTalkConnectionRegistry = {
|
||||
...delegate,
|
||||
requestSend: async (input) => {
|
||||
sendRequests += 1;
|
||||
return delegate.requestSend(input);
|
||||
},
|
||||
};
|
||||
const app = createApp(testConfig, {
|
||||
database: createDatabaseStub(),
|
||||
authorization,
|
||||
oneTalkService: createServiceStub(),
|
||||
oneTalkRegistry: registry,
|
||||
});
|
||||
const socket = await openSocket(app);
|
||||
|
||||
try {
|
||||
const handshake = nextMessages(socket, 2);
|
||||
socket.send(JSON.stringify(helloFrame()));
|
||||
await handshake;
|
||||
|
||||
const errorMessage = nextMessage(socket);
|
||||
const closeCode = nextCloseCode(socket);
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
protocolVersion: ONETALK_PROTOCOL_VERSION,
|
||||
connectionType: "mind_page",
|
||||
type: "send.request",
|
||||
requestId: "mind-frame-on-plugin-endpoint",
|
||||
sendRequestId: "request-1",
|
||||
scope: mindScope,
|
||||
payload: {
|
||||
conversationId: "conversation-1",
|
||||
content: { kind: "text", text: "hello" },
|
||||
},
|
||||
}),
|
||||
);
|
||||
assert.deepEqual((await errorMessage).payload, {
|
||||
code: ONETALK_ERROR_CODES.scopeMismatch,
|
||||
});
|
||||
assert.equal(await closeCode, 1008);
|
||||
assert.equal(sendRequests, 0);
|
||||
} finally {
|
||||
await closeApp(app, socket);
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects a legal Plugin discovery frame on the Mind endpoint before business dispatch", async () => {
|
||||
let discoveries = 0;
|
||||
const service = createServiceStub();
|
||||
service.discoverConversation = async (...args) => {
|
||||
discoveries += 1;
|
||||
return createServiceStub().discoverConversation(...args);
|
||||
};
|
||||
const authorization = createMockAuthorizationReader([authorizationRecord]);
|
||||
const app = createApp(testConfig, {
|
||||
database: createDatabaseStub(),
|
||||
authorization,
|
||||
oneTalkService: service,
|
||||
});
|
||||
const socket = await openSocket(app, "/ws/mind");
|
||||
|
||||
try {
|
||||
const errorMessage = nextMessage(socket);
|
||||
const closeCode = nextCloseCode(socket);
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
protocolVersion: ONETALK_PROTOCOL_VERSION,
|
||||
connectionType: "plugin",
|
||||
type: "conversation.discovered",
|
||||
requestId: "plugin-frame-on-mind-endpoint",
|
||||
scope: bindingScope,
|
||||
payload: {
|
||||
conversationId: "conversation-1",
|
||||
conversationType: "direct",
|
||||
lastContactTimeLong: null,
|
||||
messagePreview: null,
|
||||
},
|
||||
}),
|
||||
);
|
||||
assert.deepEqual((await errorMessage).payload, {
|
||||
code: ONETALK_ERROR_CODES.scopeMismatch,
|
||||
});
|
||||
assert.equal(await closeCode, 1008);
|
||||
assert.equal(discoveries, 0);
|
||||
} finally {
|
||||
await closeApp(app, socket);
|
||||
}
|
||||
});
|
||||
|
||||
test("rejects an unauthenticated business frame before authorization", async () => {
|
||||
const app = createApp(testConfig, {
|
||||
database: createDatabaseStub(),
|
||||
|
||||
Reference in New Issue
Block a user