From e8eccd2bb285c081ab77d9c40fbe7c93c99aa84d Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 16:47:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(history,gateway):=20D4=20=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E6=9C=AA=E8=AF=BB/=E5=B7=B2=E8=AF=BB=E6=B8=B8?= =?UTF-8?q?=E6=A0=87=20=E2=80=94=20unreadCount=20+=20POST=20/sessions/{id}?= =?UTF-8?q?/read=20+=20session=5Fread=20=E5=B9=BF=E6=92=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - chat_sessions 增 read_at（毫秒，与 chat_messages.created_at 同单位）；迁移 0010 并把既有会话回填为各自最新消息时间（升级当天不误报） - ListSessions 单查询派生 unreadCount（无 N+1，复用 idx_chat_msgs_session_created 覆盖索引） - MarkSessionRead 只前进 MAX(read_at, ?)；未知会话 ErrSessionNotFound；游标方法挂在网关消费侧窄接口（沿用 GetMessage/messageLookup 先例，避免污染 runtime 接口与其 fake） - POST /api/v1/sessions/{id}/read：writeAuth + 400/404/501，成功广播 session_read、失败路径不广播 - Docs/api-client v0.37（§1 unreadCount / 新行 8a / §3.9 session_read / §4.2 用法） - 含迁移回填测试、store/handler 测试与变异验证；顺带把 migrate_pinned_test 的「回退一步=0009」硬编码改为对迁移数量免疫 --- Docs/api-client.md | 15 +- internal/db/migrate_pinned_test.go | 9 +- .../0010_chat_session_read_at.down.sql | 9 + .../0010_chat_session_read_at.up.sql | 34 ++ internal/runtime/history/session_preview.go | 44 ++- internal/runtime/history/session_read.go | 52 +++ internal/runtime/history/session_read_test.go | 351 ++++++++++++++++++ internal/runtime/history/store.go | 13 +- internal/services/gateway/contracts.go | 5 + internal/services/gateway/history.go | 7 + internal/services/gateway/routes.go | 4 + .../gateway/session_orchestrator_test.go | 7 + internal/services/gateway/session_read.go | 97 +++++ .../services/gateway/session_read_test.go | 233 ++++++++++++ 14 files changed, 855 insertions(+), 25 deletions(-) create mode 100644 internal/db/migrations/0010_chat_session_read_at.down.sql create mode 100644 internal/db/migrations/0010_chat_session_read_at.up.sql create mode 100644 internal/runtime/history/session_read.go create mode 100644 internal/runtime/history/session_read_test.go create mode 100644 internal/services/gateway/session_read.go create mode 100644 internal/services/gateway/session_read_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 6a5b251..08d827b 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -26,13 +26,14 @@ | 1b | `/api/v1/auth/logout` | POST | refreshToken 或 JWT | ✅ | **登出**（B2）：body `{refreshToken}` 撤**单个** grant（幂等）；带 `Authorization: Bearer ` 且无 body → 撤**该账号全部** grant（全设备登出，返回 `{revokedCount}`）。refresh 相关操作成功才审计 | | 2 | `/api/v1/health` | GET | 否 | ✅ | 健康检查 `{data:{status:"ok"}}` | | 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,deviceId?,mode?,stream?}` → `{data:{reply}}`；`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL，入视觉路由（design/06）——**真视觉需配 `LLM_VISION_MODEL`/凭据，无凭据时降级为主模型直接看图或视觉·OCR 工具文本描述**；`deviceId` 为**客户端生成的具体端标识**（design/16 §4.5，如 `web:chrome:a1b2c3`，可选），落库到消息、便于主人知道从哪个端发来；`mode` 为**行为模式**（design/01 §3 / design/16 §5.1，如 `chat`/`code`/`home`/`docs`/…，可选），随该轮次落库——IM 恒 `chat`、缺省回落 `chat`，供前端按模式渲染气泡（仅 `chat` 才解析 `

` 拆气泡）。**会话编排（design/15）**：非流式且该会话「正忙」时**不阻塞**，返回 `202` `{data:{queued:true,queueDepth}}`（稍后按上下文续接）；待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | -| 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化（design/16）**：注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt,pinned,lastMessage?}]}`（**排序（D5）：`pinned=true` 优先，其次主会话 `id="web"` 恒在前，再按 `createdAt` 升序**；次级 `web:`；`pinned` 恒返回 bool；`lastMessage` 为**最后一条消息预览** `{content,role(user|assistant),createdAt}`，content 折叠空白截前 60 rune，空会话省略）；未注入时退化回渠道会话列表 | +| 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化（design/16）**：注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt,pinned,unreadCount,lastMessage?}]}`（**排序（D5）：`pinned=true` 优先，其次主会话 `id="web"` 恒在前，再按 `createdAt` 升序**；次级 `web:`；`pinned` 恒返回 bool；**`unreadCount`（D4）恒返回整数**——该会话中 `createdAt >` 已读游标的消息条数（0 = 无未读），与 `lastMessage` **同一条 SQL 派生**、无 N+1；`lastMessage` 为**最后一条消息预览** `{content,role(user|assistant),createdAt}`，content 折叠空白截前 60 rune，空会话省略）；未注入时退化回渠道会话列表 | | 5 | `/api/v1/sessions` | POST | JWT(owner/admin) | ✅ | **新建次级会话** → `201 {data:{sessionId}}`（`sessionId="web:"`）；未注入 HistoryStore 回 `501 not_implemented` | | 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**：`?limit=&before=&platform=&after=`（A5）默认 `limit=50`（上限 200）。**`before`**=上一页最旧 `createdAt`（毫秒），按时间**倒序**；**`after`**=<`createdAt` 毫秒>，按时间**升序**返回 `created_at > after` 的消息（**正向增量**，客户端把上次看到的 `createdAt` 当游标补断线期间漏的，解决重连只回放 50 条的缺口）；**`after` 与 `before` 互斥**，同传回 `400`；负数/非数字回 `400`。`platform` 为**可选**来源平台过滤（design/16 §4.4）——Web 端不传（=全量共享主会话），IM 适配器传**本平台**（如 `telegram`）只拉本渠道消息；`deviceId` 为**客户端具体端标识**；`senderName` 为**第三方发送者昵称**；**`q`**（D2）=**会话内内容搜索**（大小写不敏感 ASCII LIKE，`%`/`_`/`\` 已转义为字面量），仍按时间**倒序**、`before`/`limit` 同分页语义；`q` 与 `after` **互斥**（同传 `400`），空 `q` 视为未传（走普通分页），搜索**严格限定在 `{id}` 会话内**（绝不跨会话，群聊/陌生人隔离会话不会外泄）；非法 `limit`/`before`/`after` 回 `400`；未注入 HistoryStore 回 `501` | | 6a | `/api/v1/messages/{id}` | PATCH | JWT(owner/admin) | ✅ | **编辑单条消息**（D3）：body `{content}`（非空、≤4000 rune，空/纯空白/超长回 `400`）→ **`{data:<消息DTO>}`**（`data` **直接就是**更新后的完整消息 DTO，含 `id/sessionId/role/content/.../createdAt`，**无 `message` 包一层**；**只改 `content`**，role/时间戳/平台保持不变）；消息不存在回 `404 not_found`；未注入 HistoryStore 回 `501`。**成功后广播下行 `message_edited`**（§3.8，账号其它端免刷新同步）。**重发不单独设端点**——客户端用「编辑后的内容再 `POST /chat`」即可（避免重复轮次语义） | | 6b | `/api/v1/messages/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除单条消息**（D3）→ `{data:{deleted:}}`；不存在回 `404 not_found`；未注入 HistoryStore 回 `501`。**成功后广播下行 `message_deleted`**（§3.8，账号其它端按 `id` 移除） | | 7 | `/api/v1/sessions/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除次级会话及其消息** → `{data:{deleted:}}`；主会话不可删回 `403 forbidden`；未注入 HistoryStore 回 `501` | | 8 | `/api/v1/sessions/{id}` | PATCH | JWT(owner/admin) | ✅ | **会话元数据更新**（A4 改名 + D5 置顶）：body `{title?,pinned?}`（**至少一个字段**，都缺回 `400`）。`title` trim、≤80 rune，空/超长回 `400`；`pinned` 为 bool（置顶/取消，主会话与次级会话均可）。响应只回显**本次变更**的字段：`{data:{id,title?}}` / `{data:{id,pinned?}}` / 两者都改则都有（**title-only 保持 A4 原 `{id,title}` 形状**）；不存在回 `404`；主会话(`id="web"`)可改名/置顶但不可删；未注入 HistoryStore 回 `501`。另：会话**首条用户消息到达时自动命名** title（取前 20 rune 折叠空白；纯空白回落 `主会话`/`次级会话`），列表不再空名 | +| 8a | `/api/v1/sessions/{id}/read` | POST | JWT(owner/admin) | ✅ | **推进已读游标**（D4 / Android 协同 B3）：body `{before?}`——`before` 为**可选**的毫秒 Unix 时间戳，**缺省（或空 body）= 现在**，即「我看到的都算已读」。→ `200 {data:{sessionId,readAt}}`（`readAt` 为推进后的游标毫秒）。**判定**：`createdAt <= readAt` 的消息算已读（列表 `unreadCount` 只数 `createdAt > readAt` 的）；**语义**：**会话级单一游标、单 owner 账号**——读一处 = **该账号所有端已读**（不做 per-device 游标，与「同一账号所有浏览器都属于同一 owner」一致）；**只前进不后退**（SQL `read_at = MAX(read_at, before)`），所以乱序/并发/重试请求幂等安全、绝不会把已读变回未读。`before` 为 `0`/负数/非数字、或 body 不是 JSON 对象 → `400`；未知会话（含隔离 `sess-xxx`，不在 `chat_sessions`）→ `404`；未注入 HistoryStore → `501 not_implemented`。**成功后广播下行 `session_read`**（§3.9）让同账号其它端清红点；**失败路径不广播**。**升级回填**：迁移 `0010` 把既有会话游标回填为「该会话现有最新消息时间」，升级当天不误报未读 | | 9 | `/api/v1/personas` | GET | JWT | ✅ | `{data:[persona...]}` | | 10 | `/api/v1/audio/transcribe` | POST | JWT(owner) | ✅ | **语音→文本**：body `{audio:"",format?,deviceId?}` → `{data:{text, provider?}}`（走后端 STT，百炼 `qwen-audio-3.0-asr-flash`；**当结果为空文本/silent 降级时响应体带 `provider`**（如 `"silent"`，非空文本省略），客户端据此提示「没识别到声音」） | | 11 | `/api/v1/audio/speech` | POST | JWT(owner) | ✅ | **文本→音频**：body `{text,voice?,format?,instruction?}` → `200` 音频字节（`Content-Type: audio/`，非 JSON envelope，走后端 TTS，百炼 `cosyvoice-v3-flash`，复刻音色可带 `instruction` 情绪指令；**静默降级空音频时仍 `500 tts_empty`，并附 `X-Voice-Provider: silent` 响应头**，客户端据此提示「没生成声音」） | @@ -144,7 +145,7 @@ ## 3. WebSocket `/ws/chat`（实时） - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。 -- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤；**v0.36 起结构化流帧 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 也带 `sessionId` + `turnId`**——多会话并发流式时客户端据此把 delta 路由到激活会话，见 §3.1）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood`(情绪变化，见 §3.7) / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**（会话持久化增量，见 §3.3）/ **`message_edited`**、**`message_deleted`**（消息编辑/删除增量，见 §3.8）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 +- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤；**v0.36 起结构化流帧 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 也带 `sessionId` + `turnId`**——多会话并发流式时客户端据此把 delta 路由到激活会话，见 §3.1）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood`(情绪变化，见 §3.7) / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**（会话持久化增量，见 §3.3）/ **`message_edited`**、**`message_deleted`**（消息编辑/删除增量，见 §3.8）/ **`session_read`**（已读游标推进，见 §3.9）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 > **`history` 回放事件（design/16，新增）**：WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,createdAt}]}}`——回放主会话最新 50 条（`thinking` 已剥离，不泄给非 owner；默认主会话，次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取）。客户端据此在重连/新开连接后补全上下文。消息体含 `senderName?`（第三方昵称，design/16 §4.6，主人私聊为空）。 @@ -218,6 +219,12 @@ - **客户端处理**：与 `message_persisted` **同一过滤规则**——仅当 `sessionId === activeSessionId` 时处理（`message_deleted` 的 `sessionId` 缺失时按 `id` 兜底：消息 `id` 全局唯一，只会命中它自己那条）。`message_edited` → **按 `id` 原位更新该条的 `content`**（不新增行：找不到该 `id` 即忽略，避免把别的会话的消息注入当前列表）；`message_deleted` → **按 `id` 从列表移除**。两个操作都**不触发整页重拉**——这正是它相对「重拉绕行」的收益。 - **失败不广播**：`404`（消息不存在）/ `400`（内容非法）/ `501`（未注入 HistoryStore）**不**产生任何事件，客户端不会因一次失败请求而丢掉本地消息。 +### 3.9 已读游标事件（`session_read`，D4） +- **触发**：`POST /api/v1/sessions/{id}/read` **成功推进游标后**由网关广播（与 `message_persisted` 同一条 hub 广播路径：**全连接广播 + 客户端按 `sessionId` 过滤**，hub 无按会话订阅）。用途是**同账号多端清红点**：一端读了，其它端不用重拉列表就把该会话的未读清零。 +- **载荷**：`{type:"session_read", sessionId, data:{readAt}}`——`readAt` 为该会话推进后的**已读游标**（毫秒 Unix 时间戳，与该会话消息的 `createdAt` 同单位）。 +- **客户端处理**：仅当 `sessionId === activeSessionId`（或该会话在自己维护的会话列表里）时处理 → 把该会话 `unreadCount` 置 0（更精确的做法是按 `readAt` 重算本地未读；本地没有更完整的数据时直接清零也不会错，因为 `readAt` 只前进）。**它只驱动渲染**：唯一真源仍是后端 `chat_sessions.read_at`，漏收该帧的端在下一次 `GET /api/v1/sessions` 自然收敛（不需要补发、不需要重连）。 +- **失败不广播**：`400`（`before` 非法）/ `404`（会话不存在）/ `501`（未注入 HistoryStore）**不**产生事件，其它端不会误清红点。 + ## 4. 客户端接入指引（Web / 移动 / 原生 / IM bridge 通用） ### 4.1 一次性接入流程（三步走） @@ -226,8 +233,9 @@ 3. **带端标识**（可选但推荐）：每条消息/请求可带 `deviceId`（见 §4.3），让她知道你从哪个端来。 ### 4.2 会话与消息用法（聊天） -- **列出会话**：`GET /api/v1/sessions` → `[{id,type:"main"|"sub",title?,createdAt,pinned}]`。**排序：置顶会话（`pinned=true`）最前，其次主会话 `id="web"`**（她与你的主线，最大化继承上下文），再按创建时间升序；次级 `web:` 由 `POST /api/v1/sessions` 新建（临时小事/专业问题，独立上下文回溯）。 +- **列出会话**：`GET /api/v1/sessions` → `[{id,type:"main"|"sub",title?,createdAt,pinned,unreadCount,lastMessage?}]`。**排序：置顶会话（`pinned=true`）最前，其次主会话 `id="web"`**（她与你的主线，最大化继承上下文），再按创建时间升序；次级 `web:` 由 `POST /api/v1/sessions` 新建（临时小事/专业问题，独立上下文回溯）。 - **改名 / 置顶**：`PATCH /api/v1/sessions/{id}` body `{title?}`（≤80 rune）或 `{pinned?:bool}`（至少一个字段）→ `{data:{id,title?,pinned?}}`（只回显变更字段）。 +- **未读 / 已读游标（D4）**：列表项的 **`unreadCount`** 直接用于会话列表红点（恒返回整数，0 = 无未读）。客户端进入会话、或滚到底看到最新消息后调用 **`POST /api/v1/sessions/{id}/read`**：body `{"before":<毫秒>}`（如「本次已渲染的最新一条的 `createdAt`」），或**省略 body** = 标记到**现在**。后端**只前进**地推进该会话**唯一**游标（单 owner 账号：读一处 = 所有端已读），并广播 `session_read`（§3.9）——同账号其它端**免刷新**清红点。列表 → 读 → 再看列表即可自洽（漏收广播的端下次拉列表自然收敛）。 - **拉取激活会话历史**：`GET /api/v1/sessions/{id}/messages?limit=50&before=&platform=&q=`。 - **热分页**：`limit` 默认 50（上限 200），`before` = 当前最旧一条的 `createdAt`（毫秒），向上翻页。`platform` 可选（IM 适配器传本平台只拉本渠道；Web/通用客户端不传 = 全量共享主会话）。 - **搜索**：`q` 非空 = **会话内**内容搜索（倒序，仍支持 `before`/`limit` 分页）；空 `q` = 普通分页；`q` 与 `after` 互斥。**不跨会话**。 @@ -253,6 +261,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `v0.37`（**D4 未读状态 / 已读游标**，对客户端**新增一个语义字段、一个 REST 端点、一个下行事件**）：① `chat_sessions` 新增 **`read_at INTEGER NOT NULL DEFAULT 0`**（迁移 **0010**，down 可逆；**毫秒 Unix 时间戳**，与 `chat_messages.created_at` **同单位**）。② **语义**：**会话级单一已读游标 + 单 owner 账号**——同一账号所有端共用一个游标，**读一处 = 全端已读**（不做 per-device/per-user 游标，与「同一账号所有浏览器都属于同一 owner」一致）；`createdAt <= readAt` 的消息算已读，未读数 = `createdAt > readAt` 的条数。③ **`GET /api/v1/sessions` 每项新增恒返回的 `unreadCount`**（整数，0 = 无未读）：与 `lastMessage` **同一条 SQL 派生**（关联子查询 `COUNT(*) WHERE session_id = s.id AND created_at > s.read_at`，走已有 `idx_chat_msgs_session_created`，**无 N+1、不新增索引**）。④ **新增 `POST /api/v1/sessions/{id}/read`**（owner/admin，与 `PATCH /sessions/{id}` 同鉴权）：body `{before?}`（可选毫秒；缺省= 现在）→ `200 {data:{sessionId,readAt}}`；`before` 为 0/负数/非数字/body 非对象 → `400`，未知会话（含隔离 `sess-xxx`）→ `404`，未注入 HistoryStore → `501`。**游标只前进不后退**（`read_at = MAX(read_at, before)`），乱序/并发/重试请求幂等安全。⑤ **新增下行事件 `session_read`**：`{type:"session_read", sessionId, data:{readAt}}`，全连接广播 + 客户端按 `sessionId` 过滤，用于同账号其它端清红点；失败路径不广播（§3.9）。⑥ **升级回填**：0010 把既有会话游标回填为「该会话现有最新消息时间」，**升级当天不误报未读**（既有历史全部视为已读）。⑦ `HistoryStore`（网关消费侧契约）新增 `MarkSessionRead(ctx, sessionID, before)`；`SessionMeta` 新增 `UnreadCount`。**向后兼容**：仅新增字段/端点/事件，既有契约形状不变；不消费 `unreadCount`/`session_read` 的客户端行为不变。详见 §1 行 4/8a 与 §3.9。 - `v0.36`（**A1 流式帧附加会话 / 回合标识**，对客户端**新增两个可选字段、无端点变更**）：结构化流帧 `message.start/delta/end`、`thinking.start/delta/end`、`tool_call.start/delta/end`、`tool_execution.update` 现在带 `sessionId` + `turnId`（**向后兼容**：只新增帧字段，`data` 结构与既有事件形状不变）。`sessionId` 由 loop 在**回合边界**从运行上下文注入（模型适配器在 loop 之下、天然不感知会话，保持零值；loop 外无会话上下文时该字段为空）；`turnId` 由网关**每回合**新生成（`newTurnID`）并经 `loop.WithTurnID` 挂进运行上下文，`runTurn` 是编排层与 legacy 直连路径**唯一**的回合收口点，两路都覆盖——同一会话同时只有一个 in-flight 回合，故同轮内所有帧的 `turnId` 一致。此前这几帧属于 design/16 §4.2 标注的「若知道就带上」缺口（v0.10 起记录）：**多会话并发流式时前端会把别的会话的 delta 接进当前会话气泡**，现在客户端按 `sessionId === activeSessionId` 过滤即可路由到激活会话。`stream_chunk` / `stream_end` 的 `sessionId`/`turnId` 行为不变。详见 §3.1。 - `v0.35`（**可观测性收尾：后端日志 tail 端点**，对客户端**新增一个 admin 只读端点**）：`GET /api/v1/admin/observability/logs?level=&limit=`（owner/admin）→ `200 {data:[{time,level,source,msg}]}`，**最新在前**。数据源为 logger 新增的**有界环形缓冲**（`internal/logger`，容量 2000，满则丢最旧、并发安全、不阻塞、不改动既有控制台/文件输出），组合根在构建 logger 时挂上、在网关就绪后经 `SetLogBuffer` 注入；缓冲未注入回 `501 not_implemented`（不伪造空 tail）。`time` 为 **RFC3339 UTC（秒精度）**，与 `trace.startedAt`/`audit.when` 一致；`level ∈ debug|info|warn|error`（大小写不敏感）为**最低等级过滤**，空 = 全部，未知值 `400 invalid_level`；`limit` 缺省 `200`、上限 `1000`。`source` 取 `logger.WithSource` 的 `source=`（`Scoped` 记录回落 `module/sub`），无来源为空串。详见 §2「可观测性契约」。 - `v0.34`（**D3 收尾：消息编辑 / 删除的 WS 实时广播**，对客户端**新增两个下行事件、无 REST 契约变更**）：`PATCH /api/v1/messages/{id}` 成功后广播 **`message_edited`**（`{type, sessionId, data:<更新后的完整消息DTO>}`，`thinking` 已剥离）、`DELETE /api/v1/messages/{id}` 成功后广播 **`message_deleted`**（`{type, sessionId, data:{id}}`）；两者复用 `message_persisted` 的同一 hub 广播路径（**全连接广播 + 客户端按 `sessionId` 过滤**，hub 无按会话订阅），客户端**按 `id` 原位改 `content` / 按 `id` 移除**，不再需要「改/删后重拉」的绕行。`message_deleted` 的 `sessionId` 由后端在删除前读取消息所属会话填入（`HistoryStore` 新增 `GetMessage`；读不到时客户端按全局唯一的 `id` 兜底）。`404`/`400`/`501` 等失败路径**不广播**。Web 端已消费（`lib/ws.ts` + `store/useAppStore` 新增 `updateMessageById`/`removeMessageById`）。详见 §3.8。 diff --git a/internal/db/migrate_pinned_test.go b/internal/db/migrate_pinned_test.go index 977c0d3..2a36ac2 100644 --- a/internal/db/migrate_pinned_test.go +++ b/internal/db/migrate_pinned_test.go @@ -40,9 +40,12 @@ func TestMigrate0009PinnedColumnIdempotent(t *testing.T) { t.Fatal("chat_sessions.pinned should still exist after idempotent re-run") } - // Down one step rolls back only 0009, dropping the column. - if err := MigrateDown(ctx, conn, 1, testLogger()); err != nil { - t.Fatalf("MigrateDown(1): %v", err) + // Roll back every migration newer than 0008 (0009 itself plus any migration added + // after it, currently 0010), so this test keeps validating the 0009 down no matter + // how many later migrations the repo has grown. + steps := len(upMigrationIDs(t)) - 8 + if err := MigrateDown(ctx, conn, steps, testLogger()); err != nil { + t.Fatalf("MigrateDown(%d): %v", steps, err) } if columnExists(t, ctx, conn, "chat_sessions", "pinned") { t.Fatal("chat_sessions.pinned should be dropped by the 0009 down") diff --git a/internal/db/migrations/0010_chat_session_read_at.down.sql b/internal/db/migrations/0010_chat_session_read_at.down.sql new file mode 100644 index 0000000..42baaa1 --- /dev/null +++ b/internal/db/migrations/0010_chat_session_read_at.down.sql @@ -0,0 +1,9 @@ +-- 0010_chat_session_read_at down: drop the read cursor added by the up migration. +-- SQLite DROP COLUMN (3.35+) is already relied on by 0006/0007/0009, and read_at is +-- not referenced by any index, view or constraint, so a plain drop is safe. The +-- rolled-back state is the pre-D4 one: no unread badge data at all (the cursor is +-- derived display state, not data a session needs to stay valid). The backfilled +-- values cannot be restored by a re-up beyond the newest-message rule, which is +-- exactly the upgrade default the up migration applies. + +ALTER TABLE chat_sessions DROP COLUMN read_at; diff --git a/internal/db/migrations/0010_chat_session_read_at.up.sql b/internal/db/migrations/0010_chat_session_read_at.up.sql new file mode 100644 index 0000000..9690213 --- /dev/null +++ b/internal/db/migrations/0010_chat_session_read_at.up.sql @@ -0,0 +1,34 @@ +-- 0010_chat_session_read_at: per-session read cursor backing the session-list +-- unread badge (DevDocs/design/16 §2.2/§4.2, Android 协同批次 D4/B3). There is a +-- single owner account and no per-device identity at this layer, so the cursor is +-- session-scoped: reading on any one end marks the session read for every end of +-- the account (one cursor, no per-user/per-device cursor table). +-- +-- read_at is a Unix MILLISECOND timestamp in the same unit as +-- chat_messages.created_at (both are INTEGER ms columns; the store assigns +-- time.Now().UnixMilli()), so the unread count is a plain +-- `created_at > read_at` comparison. 0 means "nothing read yet". +-- +-- Backfill on upgrade: every existing session's cursor is set to its newest +-- message timestamp, so an upgrade does not suddenly light up the whole session +-- list as unread (a session with no messages — and a session whose messages do not +-- exist — stays at 0, which is also 0 unread). +-- +-- The 0003 chat_sessions table already exists at this point, so ALTER ADD COLUMN is +-- the correct, non-destructive step. The canonical schema (including this column) is +-- also re-issued idempotently by internal/runtime/history.NewStore so a fresh +-- in-memory store used by tests stays self-contained; the migration is the +-- authoritative path for an existing database. +-- +-- No new index is needed: the unread count filters +-- `session_id = ? AND created_at > ?`, which the existing +-- idx_chat_msgs_session_created (session_id, created_at DESC) already covers as an +-- equality + range scan. + +ALTER TABLE chat_sessions ADD COLUMN read_at INTEGER NOT NULL DEFAULT 0; + +UPDATE chat_sessions +SET read_at = COALESCE( + (SELECT MAX(m.created_at) FROM chat_messages m WHERE m.session_id = chat_sessions.id), + 0 +); diff --git a/internal/runtime/history/session_preview.go b/internal/runtime/history/session_preview.go index e4bca18..2637e55 100644 --- a/internal/runtime/history/session_preview.go +++ b/internal/runtime/history/session_preview.go @@ -7,11 +7,12 @@ import ( "strings" ) -// This file owns the last-message preview carried in the session list -// (GET /api/v1/sessions): a display-only summary of each session's newest message so a -// client can render a conversation row without one request per session. The preview is -// truncated for display and never feeds identity or permission decisions — identity -// comes from channel actor metadata / JWT (project red line). +// This file owns the two derived per-session display fields carried in the session +// list (GET /api/v1/sessions): the last-message preview and the unread count. Both are +// read-only projections so a client can render a conversation row (with its unread +// badge) without one request per session. The preview is truncated for display, and +// neither field ever feeds identity or permission decisions — identity comes from +// channel actor metadata / JWT (project red line). // lastMessagePreviewRunes bounds the preview content so a long message cannot bloat the // session-list response. A longer message is whitespace-collapsed and truncated to this @@ -25,14 +26,19 @@ type MessageSummary struct { CreatedAt int64 // Unix milliseconds } -// listSessionsQuery reads every session plus its newest message in one statement. The -// LEFT JOIN's correlated subquery resolves each session's newest row via the -// (session_id, created_at DESC) index (rowid DESC breaks a same-millisecond tie by -// insertion order), so a session with no messages still appears with NULL preview -// columns. Ordering floats pinned sessions first, then keeps the main session before -// unpinned subs, then subs oldest-first. +// listSessionsQuery reads every session plus its newest message and its unread count +// in one statement. The LEFT JOIN's correlated subquery resolves each session's newest +// row via the (session_id, created_at DESC) index (rowid DESC breaks a same-millisecond +// tie by insertion order), so a session with no messages still appears with NULL preview +// columns. The unread count is a correlated COUNT(*) over the same index +// (session_id = s.id AND created_at > s.read_at is an equality + range scan), so no +// per-session round trip (N+1) is needed and no extra index is required. Ordering floats +// pinned sessions first, then keeps the main session before unpinned subs, then subs +// oldest-first. const listSessionsQuery = `SELECT s.id, s.type, s.title, s.pinned, s.created_at, - m.role, m.content, m.created_at + m.role, m.content, m.created_at, + (SELECT COUNT(*) FROM chat_messages c + WHERE c.session_id = s.id AND c.created_at > s.read_at) FROM chat_sessions s LEFT JOIN chat_messages m ON m.rowid = ( SELECT rowid FROM chat_messages @@ -44,8 +50,9 @@ const listSessionsQuery = `SELECT s.id, s.type, s.title, s.pinned, s.created_at, // ListSessions returns pinned sessions first, then the main session, then every sub // session, oldest first, each carrying a LastMessage preview of its newest message (nil -// when the session has none). Sessions and previews are read in a single query, so the -// query count does not grow with the number of sessions. The returned slice is non-nil. +// when the session has none) and the UnreadCount derived from the session's read cursor. +// Sessions, previews and unread counts are read in a single query, so the query count +// does not grow with the number of sessions. The returned slice is non-nil. func (s *Store) ListSessions(ctx context.Context) ([]SessionMeta, error) { rows, err := s.db.QueryContext(ctx, listSessionsQuery, SessionMain) if err != nil { @@ -67,9 +74,10 @@ func (s *Store) ListSessions(ctx context.Context) ([]SessionMeta, error) { return sessions, nil } -// scanSessionMeta reads one joined chat_sessions + last-message row. The three message -// columns are NULL for a session that has never received a message, so they scan into -// nullable holders and only produce a LastMessage when a row is present. +// scanSessionMeta reads one joined chat_sessions + last-message + unread-count row. The +// three message columns are NULL for a session that has never received a message, so they +// scan into nullable holders and only produce a LastMessage when a row is present. The +// unread count is a non-NULL COUNT(*), so it scans into a plain int. func scanSessionMeta(sc rowScanner) (SessionMeta, error) { var ( meta SessionMeta @@ -79,7 +87,7 @@ func scanSessionMeta(sc rowScanner) (SessionMeta, error) { content sql.NullString lastAt sql.NullInt64 ) - if err := sc.Scan(&meta.ID, &typeStr, &meta.Title, &pinned, &meta.CreatedAt, &role, &content, &lastAt); err != nil { + if err := sc.Scan(&meta.ID, &typeStr, &meta.Title, &pinned, &meta.CreatedAt, &role, &content, &lastAt, &meta.UnreadCount); err != nil { return SessionMeta{}, err } meta.Type = typeStr diff --git a/internal/runtime/history/session_read.go b/internal/runtime/history/session_read.go new file mode 100644 index 0000000..715c317 --- /dev/null +++ b/internal/runtime/history/session_read.go @@ -0,0 +1,52 @@ +package history + +import ( + "context" + "fmt" + "time" +) + +// This file owns the per-session read cursor (design/16 §4.2, Android 协同批次 +// D4/B3). There is a single owner account and no per-device identity at this +// layer, so the cursor is session-scoped: reading on any one end marks the session +// read for every end of the account. The cursor lives on chat_sessions.read_at in +// the same millisecond unit as chat_messages.created_at, so the unread badge is a +// plain `created_at > read_at` count (see listSessionsQuery). + +// MarkSessionRead advances the session's read cursor to before and returns the +// resulting cursor as a Unix millisecond timestamp. A zero before means "now" — the +// caller's "mark everything I have seen" case, which does not have to read the +// clock itself. +// +// The cursor only ever moves forward: the UPDATE applies SQLite's two-argument +// scalar MAX, so a stale or out-of-order request (two ends racing, a retried +// request) can never un-read messages. A negative before therefore leaves the stored +// value untouched rather than moving it backwards. +// +// An unknown session id returns ErrSessionNotFound, matching RenameSession/SetPinned +// — isolated (group/stranger) sessions have no chat_sessions row and are not +// addressable here. +func (s *Store) MarkSessionRead(ctx context.Context, sessionID string, before time.Time) (int64, error) { + ms := before.UnixMilli() + if before.IsZero() { + ms = time.Now().UnixMilli() + } + res, err := s.db.ExecContext(ctx, + `UPDATE chat_sessions SET read_at = MAX(read_at, ?) WHERE id = ?`, ms, sessionID) + if err != nil { + return 0, fmt.Errorf("history: mark session read %s: %w", sessionID, err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("history: mark session read %s rows: %w", sessionID, err) + } + if n == 0 { + return 0, fmt.Errorf("history: mark session read %s: %w", sessionID, ErrSessionNotFound) + } + var readAt int64 + if err := s.db.QueryRowContext(ctx, + `SELECT read_at FROM chat_sessions WHERE id = ?`, sessionID).Scan(&readAt); err != nil { + return 0, fmt.Errorf("history: read session cursor %s: %w", sessionID, err) + } + return readAt, nil +} diff --git a/internal/runtime/history/session_read_test.go b/internal/runtime/history/session_read_test.go new file mode 100644 index 0000000..66668f7 --- /dev/null +++ b/internal/runtime/history/session_read_test.go @@ -0,0 +1,351 @@ +package history + +import ( + "context" + "database/sql" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// These tests cover the D4 per-session read cursor (design/16 §4.2, Android +// collaboration B3): the unread count derived in ListSessions, the forward-only +// cursor semantics of MarkSessionRead, and the 0010 migration's upgrade backfill. + +// TestListSessionsUnreadCountFromCursor pins the unread derivation: a message counts as +// unread only when it is strictly newer than the session's cursor (created_at > read_at), +// so the cursor is inclusive of the timestamp it names (a client that has rendered a +// message sent at T marks T read). A fresh session starts at cursor 0 with everything +// unread, and a cursor at the newest message clears the badge. +func TestListSessionsUnreadCountFromCursor(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + for _, ts := range []int64{1000, 2000, 3000} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "m", CreatedAt: ts}); err != nil { + t.Fatalf("seed %d: %v", ts, err) + } + } + // Never read: all three messages are unread (a brand-new account/session). + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 3 { + t.Fatalf("initial unreadCount = %d, want 3", got) + } + // Cursor on a message's own timestamp counts that message as read. + readAt, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(2000)) + if err != nil { + t.Fatalf("mark read: %v", err) + } + if readAt != 2000 { + t.Fatalf("readAt = %d, want 2000", readAt) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 1 { + t.Fatalf("unreadCount after cursor 2000 = %d, want 1 (only 3000 newer)", got) + } + // A cursor between two messages reads the earlier one only. + if _, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(2500)); err != nil { + t.Fatalf("mark read 2500: %v", err) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 1 { + t.Fatalf("unreadCount after cursor 2500 = %d, want 1", got) + } + // A cursor at the newest message clears the badge. + if _, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(3000)); err != nil { + t.Fatalf("mark read 3000: %v", err) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 0 { + t.Fatalf("unreadCount after cursor 3000 = %d, want 0", got) + } +} + +// TestListSessionsUnreadCountSessionIsolation pins the per-session scope: each session +// carries its own cursor, so reading the main session must not clear (or inflate) a sub +// session's badge, and vice versa. +func TestListSessionsUnreadCountSessionIsolation(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + sub, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + for _, m := range []Message{ + {SessionID: MainSessionID, Role: RoleUser, Content: "main-1", CreatedAt: 1000}, + {SessionID: MainSessionID, Role: RoleAssistant, Content: "main-2", CreatedAt: 2000}, + {SessionID: sub, Role: RoleUser, Content: "sub-1", CreatedAt: 1500}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 2 { + t.Fatalf("main unreadCount = %d, want 2", got) + } + if got := sessionOf(t, st, sub).UnreadCount; got != 1 { + t.Fatalf("sub unreadCount = %d, want 1", got) + } + + // Reading the main session must leave the sub untouched. + if _, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(2000)); err != nil { + t.Fatalf("mark main read: %v", err) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 0 { + t.Fatalf("main unreadCount after read = %d, want 0", got) + } + if got := sessionOf(t, st, sub).UnreadCount; got != 1 { + t.Fatalf("sub unreadCount after main read = %d, want 1 (cursors are per-session)", got) + } + + // A message arriving after the read re-opens the main badge. + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "main-3", CreatedAt: 3000}); err != nil { + t.Fatalf("append main-3: %v", err) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 1 { + t.Fatalf("main unreadCount after new message = %d, want 1", got) + } +} + +// TestMarkSessionReadForwardOnly covers the racing/malformed-request side of the cursor +// contract: the cursor never moves backwards, so a stale or repeated request is +// idempotent and can never resurrect already-read messages. +func TestMarkSessionReadForwardOnly(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + for _, ts := range []int64{1000, 2000, 3000} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "m", CreatedAt: ts}); err != nil { + t.Fatalf("seed %d: %v", ts, err) + } + } + // Advance to the newest message. + readAt, err := st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(3000)) + if err != nil || readAt != 3000 { + t.Fatalf("mark 3000 = (%d, %v), want (3000, nil)", readAt, err) + } + // A stale request (an end that was behind, or a retry of an older read) must not + // move the cursor back and re-open the badge. + readAt, err = st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(1000)) + if err != nil { + t.Fatalf("stale mark: %v", err) + } + if readAt != 3000 { + t.Fatalf("stale mark readAt = %d, want the cursor to stay at 3000", readAt) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 0 { + t.Fatalf("unreadCount after stale mark = %d, want 0 (never un-read)", got) + } + // Repeating the same cursor is a no-op, and a negative cursor is clamped by the + // same MAX (it cannot move the stored value at all). + if readAt, err = st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(3000)); err != nil || readAt != 3000 { + t.Fatalf("repeat mark = (%d, %v), want (3000, nil)", readAt, err) + } + if readAt, err = st.MarkSessionRead(ctx, MainSessionID, time.UnixMilli(-5000)); err != nil || readAt != 3000 { + t.Fatalf("negative mark = (%d, %v), want (3000, nil)", readAt, err) + } +} + +// TestMarkSessionReadZeroTimeMeansNow covers the "mark everything I have seen" call the +// HTTP layer makes when the body omits before: the zero time resolves to the wall clock, +// which reads every message that already exists. +func TestMarkSessionReadZeroTimeMeansNow(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + seedAt := time.Now().UnixMilli() + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "m", CreatedAt: seedAt}); err != nil { + t.Fatalf("seed: %v", err) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 1 { + t.Fatalf("unreadCount = %d, want 1", got) + } + readAt, err := st.MarkSessionRead(ctx, MainSessionID, time.Time{}) + if err != nil { + t.Fatalf("mark zero time: %v", err) + } + if readAt < seedAt { + t.Fatalf("readAt = %d, want >= the seeded message time %d", readAt, seedAt) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 0 { + t.Fatalf("unreadCount after mark-now = %d, want 0", got) + } +} + +// TestMarkSessionReadUnknownSession pins the error contract: an id with no chat_sessions +// row (including an isolated group/stranger session) is ErrSessionNotFound, so the HTTP +// layer can answer 404 instead of silently reporting success for a session that does not +// exist. +func TestMarkSessionReadUnknownSession(t *testing.T) { + t.Parallel() + st := newStore(t) + if _, err := st.MarkSessionRead(context.Background(), "web:missing", time.UnixMilli(1000)); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("unknown session err = %v, want ErrSessionNotFound", err) + } +} + +// migrationReadAtSQL reads one of the 0010 migration files from the source tree, so the +// test applies the same SQL text the embedded migration runs rather than a copy. +func migrationReadAtSQL(t *testing.T, name string) string { + t.Helper() + b, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", name)) + if err != nil { + t.Fatalf("read migration %s: %v", name, err) + } + return string(b) +} + +// columnPresent reports whether table has a column named column. +func columnPresent(t *testing.T, conn *sql.DB, table, column string) bool { + t.Helper() + var n int + if err := conn.QueryRowContext(context.Background(), + "SELECT count(*) FROM pragma_table_info(?) WHERE name = ?", table, column).Scan(&n); err != nil { + t.Fatalf("pragma_table_info(%s): %v", table, err) + } + return n == 1 +} + +// openLegacyDB opens an in-memory DB carrying the pre-0010 schema (chat_sessions with no +// read_at column) plus two legacy sessions with prior history — the real fixed main +// session "web" and a sub session — so the 0010 migration can be applied to a database +// that really predates this feature. +func openLegacyDB(t *testing.T) *sql.DB { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("open memory db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + ddls := []string{ + `CREATE TABLE chat_sessions ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + pinned INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL)`, + `CREATE TABLE chat_messages ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + mode TEXT, + mood TEXT, + image TEXT, + thinking TEXT, + platform TEXT NOT NULL DEFAULT 'web', + device_id TEXT NOT NULL DEFAULT '', + sender_name TEXT NOT NULL DEFAULT '', + adapter_id TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL)`, + `CREATE INDEX idx_chat_msgs_session_created ON chat_messages (session_id, created_at DESC)`, + } + for _, ddl := range ddls { + if _, err := conn.ExecContext(ctx, ddl); err != nil { + t.Fatalf("legacy ddl: %v", err) + } + } + legacy := []struct { + sessionID string + sessionType string + messageAt []int64 + }{ + {sessionID: MainSessionID, sessionType: SessionMain, messageAt: []int64{1000, 4000}}, + {sessionID: "web:legacy", sessionType: SessionSub, messageAt: []int64{1000, 2000}}, + } + for _, l := range legacy { + if _, err := conn.ExecContext(ctx, + `INSERT INTO chat_sessions (id, type, title, pinned, created_at) VALUES (?, ?, '', 0, ?)`, + l.sessionID, l.sessionType, int64(100)); err != nil { + t.Fatalf("insert legacy session %s: %v", l.sessionID, err) + } + for i, ts := range l.messageAt { + if _, err := conn.ExecContext(ctx, + `INSERT INTO chat_messages (id, session_id, role, content, created_at) VALUES (?, ?, ?, ?, ?)`, + fmt.Sprintf("%s-%d", l.sessionID, i), l.sessionID, RoleUser, "legacy message", ts); err != nil { + t.Fatalf("insert legacy message: %v", err) + } + } + } + return conn +} + +// TestMigration0010ReadAtBackfill exercises the real 0010 migration against a legacy +// database: up adds the column and backfills each session's cursor to its newest message, +// so upgrade day reports 0 unread for existing history — including the fixed main session, +// whose existing history is the whole point of the upgrade — and the store (which +// re-issues the canonical DDL idempotently) then works on top of it; a message that arrives +// after the migration is unread until it is marked read; down drops the column again. +func TestMigration0010ReadAtBackfill(t *testing.T) { + ctx := context.Background() + conn := openLegacyDB(t) + if columnPresent(t, conn, "chat_sessions", "read_at") { + t.Fatal("legacy schema should not have read_at before the migration") + } + + if _, err := conn.ExecContext(ctx, migrationReadAtSQL(t, "0010_chat_session_read_at.up.sql")); err != nil { + t.Fatalf("apply 0010 up: %v", err) + } + if !columnPresent(t, conn, "chat_sessions", "read_at") { + t.Fatal("read_at should exist after the 0010 up migration") + } + // The backfill is each session's newest message timestamp, not a blanket 0 (which + // would mark the whole existing history unread) and not the first message's. + for _, tc := range []struct { + sessionID string + want int64 + }{ + {sessionID: MainSessionID, want: 4000}, + {sessionID: "web:legacy", want: 2000}, + } { + var backfilled int64 + if err := conn.QueryRowContext(ctx, + `SELECT read_at FROM chat_sessions WHERE id = ?`, tc.sessionID).Scan(&backfilled); err != nil { + t.Fatalf("read backfilled cursor %s: %v", tc.sessionID, err) + } + if backfilled != tc.want { + t.Fatalf("backfilled read_at for %s = %d, want its newest legacy message %d", tc.sessionID, backfilled, tc.want) + } + } + + // The store comes up on the migrated schema and reports no unread for either + // pre-existing session (upgrade day must not light up every row, main session + // included); the store's INSERT OR IGNORE for the main session must not overwrite + // the backfilled cursor. + st, err := NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("new store on migrated db: %v", err) + } + for _, sessionID := range []string{MainSessionID, "web:legacy"} { + if got := sessionOf(t, st, sessionID).UnreadCount; got != 0 { + t.Fatalf("legacy session %s unreadCount = %d, want 0 after backfill", sessionID, got) + } + } + // A message written after the migration is unread, and marking it read clears it. + if err := st.Append(ctx, Message{SessionID: "web:legacy", Role: RoleAssistant, Content: "new", CreatedAt: 9000}); err != nil { + t.Fatalf("append after migration: %v", err) + } + if got := sessionOf(t, st, "web:legacy").UnreadCount; got != 1 { + t.Fatalf("unreadCount after post-migration message = %d, want 1", got) + } + if got := sessionOf(t, st, MainSessionID).UnreadCount; got != 0 { + t.Fatalf("main unreadCount = %d, want 0 (untouched by the sub session's new message)", got) + } + readAt, err := st.MarkSessionRead(ctx, "web:legacy", time.UnixMilli(9000)) + if err != nil || readAt != 9000 { + t.Fatalf("mark read = (%d, %v), want (9000, nil)", readAt, err) + } + if got := sessionOf(t, st, "web:legacy").UnreadCount; got != 0 { + t.Fatalf("unreadCount after mark = %d, want 0", got) + } + + if _, err := conn.ExecContext(ctx, migrationReadAtSQL(t, "0010_chat_session_read_at.down.sql")); err != nil { + t.Fatalf("apply 0010 down: %v", err) + } + if columnPresent(t, conn, "chat_sessions", "read_at") { + t.Fatal("read_at should be dropped by the 0010 down migration") + } +} diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 0770691..4290721 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -121,10 +121,20 @@ type SessionMeta struct { // A client uses it to render the conversation row without a per-session request; // identity and permissions never derive from it. LastMessage *MessageSummary + // UnreadCount is the number of messages newer than the session's read cursor + // (design/16 §4.2, D4), computed by ListSessions in the same query. The cursor + // is session-scoped and single-owner: reading on any one end marks the session + // read for every end of the account. Like LastMessage it is display-only — + // identity and permissions never derive from it. + UnreadCount int } // HistoryStore abstracts the chat-history store so callers (the gateway) and tests can -// hold an interface rather than a concrete *Store. +// hold an interface rather than a concrete *Store. It is deliberately kept to the calls +// the runtime-level consumers (channelbridge, backthink) actually make: the D4 read +// cursor (MarkSessionRead) is declared on the consumer-side interface in +// internal/services/gateway (contracts.go) instead, so a narrower store fake elsewhere +// does not have to implement a cursor it never advances. *Store satisfies both. type HistoryStore interface { // Append persists msg, assigning an id and timestamp when the caller omitted them. Append(ctx context.Context, msg Message) error @@ -579,6 +589,7 @@ CREATE TABLE IF NOT EXISTS chat_sessions ( type TEXT NOT NULL, title TEXT NOT NULL DEFAULT '', pinned INTEGER NOT NULL DEFAULT 0, + read_at INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL )` diff --git a/internal/services/gateway/contracts.go b/internal/services/gateway/contracts.go index 6f01cb2..7a95095 100644 --- a/internal/services/gateway/contracts.go +++ b/internal/services/gateway/contracts.go @@ -116,6 +116,11 @@ type HistoryStore interface { // history.ErrSessionNotFound. SetPinned(ctx context.Context, sessionID string, pinned bool) error NewSubSession(ctx context.Context, ownerID string) (string, error) + // MarkSessionRead advances a session's read cursor (D4) to before and returns + // the resulting cursor as a Unix millisecond timestamp; a zero before means + // "now". The cursor only moves forward. An unknown id returns + // history.ErrSessionNotFound. + MarkSessionRead(ctx context.Context, sessionID string, before time.Time) (int64, error) ListSessions(ctx context.Context) ([]history.SessionMeta, error) } diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 5ade642..73581f6 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -26,6 +26,12 @@ type historySessionSummary struct { // per-session request. It is omitted for a session with no messages. Content is // already truncated by the store. LastMessage *historyLastMessage `json:"lastMessage,omitempty"` + // UnreadCount is the number of messages newer than the session's read cursor + // (D4), derived in the same list query. It is always returned (0 when nothing + // is unread) so a client can render/clear the badge without a second call. The + // cursor is session-scoped and single-owner: reading on any one end clears the + // badge for every end of the account. + UnreadCount int `json:"unreadCount"` } // historyLastMessage is the wire shape of the last-message preview in a session summary. @@ -79,6 +85,7 @@ func (s *Server) handleHistorySessions(w http.ResponseWriter, r *http.Request) { CreatedAt: m.CreatedAt, Pinned: m.Pinned, LastMessage: toHistoryLastMessage(m.LastMessage), + UnreadCount: m.UnreadCount, }) } writeJSON(w, http.StatusOK, out) diff --git a/internal/services/gateway/routes.go b/internal/services/gateway/routes.go index 590f039..69edd54 100644 --- a/internal/services/gateway/routes.go +++ b/internal/services/gateway/routes.go @@ -83,6 +83,10 @@ func (s *Server) registerHistoryRoutes(mux *http.ServeMux, writeAuth func(http.H mux.Handle("POST /api/v1/sessions", writeAuth(http.HandlerFunc(s.handleCreateSubSession))) mux.Handle("DELETE /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handleDeleteSession))) mux.Handle("PATCH /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handlePatchSession))) + // Read cursor (D4 / Android collaboration B3): advancing it clears the session's + // unread badge on every end of the account, so it is an owner/admin history write + // like the other session mutations. It broadcasts a `session_read` frame. + mux.Handle("POST /api/v1/sessions/{id}/read", writeAuth(http.HandlerFunc(s.handleMarkSessionRead))) // Single-message edit/delete (Android collaboration batch ③a, D3). A message is // addressed by its own opaque id; both are owner/admin-gated history writes. mux.Handle("PATCH /api/v1/messages/{id}", writeAuth(http.HandlerFunc(s.handlePatchMessage))) diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index ba7a116..ca1746f 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -231,6 +231,13 @@ func (f *fakeHistory) ListSessions(context.Context) ([]history.SessionMeta, erro return nil, nil } +// MarkSessionRead satisfies the HistoryStore contract for the read-cursor path (D4). +// The orchestrator never advances a cursor, so the fake only reports the requested +// cursor back without storing it. +func (f *fakeHistory) MarkSessionRead(_ context.Context, _ string, before time.Time) (int64, error) { + return before.UnixMilli(), nil +} + func (f *fakeHistory) records() []history.Message { f.mu.Lock() defer f.mu.Unlock() diff --git a/internal/services/gateway/session_read.go b/internal/services/gateway/session_read.go new file mode 100644 index 0000000..8e7ede3 --- /dev/null +++ b/internal/services/gateway/session_read.go @@ -0,0 +1,97 @@ +package gateway + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "time" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// This file owns the D4 read-cursor surface (Android collaboration B3): the +// POST /api/v1/sessions/{id}/read endpoint that advances a session's read cursor, +// plus the `session_read` realtime frame it broadcasts so the account's other +// browsers clear the unread badge without a re-read. The cursor itself lives in +// chat_sessions.read_at (internal/runtime/history); GET /sessions derives the +// unreadCount from it (history.go). The cursor is session-scoped and single-owner: +// one account, one cursor per session — reading on any one end marks the session +// read for every end. + +// wsSessionRead is the WS event type for a read-cursor advance. +const wsSessionRead = "session_read" + +// sessionReadRequest is the POST /api/v1/sessions/{id}/read body. Before is a +// pointer so an absent field (an empty body) is distinguishable from an explicit +// value: absent means "now" — the client marks everything it has seen read. +type sessionReadRequest struct { + Before *int64 `json:"before"` +} + +// handleMarkSessionRead advances a session's read cursor and answers +// {data:{sessionId, readAt}} with the resulting cursor in Unix milliseconds. +// It is owner/admin-gated like the other history writes (writeAuth). +// +// body {before?} is an optional millisecond timestamp bounding what is marked +// read — messages with created_at <= readAt become read. Absent (or an empty body) +// means now. A present non-positive or non-numeric before answers 400: the cursor +// is a real millisecond timestamp, and "unread by a negative cursor" is not a +// meaningful client request. An unknown session id answers 404; a missing +// HistoryStore answers 501. +// +// On success the new cursor is broadcast as a `session_read` frame so the account's +// other browsers clear the badge too (reading on one end = read everywhere). A +// failure path broadcasts nothing. +func (s *Server) handleMarkSessionRead(w http.ResponseWriter, r *http.Request) { + if s.history == nil { + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) + return + } + id := r.PathValue("id") + if id == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "session id is required")) + return + } + var req sessionReadRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil && !errors.Is(err, io.EOF) { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) + return + } + var before time.Time + if req.Before != nil { + if *req.Before <= 0 { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "before must be a positive millisecond timestamp")) + return + } + before = time.UnixMilli(*req.Before) + } + readAt, err := s.history.MarkSessionRead(r.Context(), id, before) + if err != nil { + if errors.Is(err, history.ErrSessionNotFound) { + writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "session %q not found", id)) + return + } + s.log.Error("history mark session read", "err", err, "session", id) + writeError(w, apperr.Wrap(apperr.ErrInternal, "session_failed", "failed to mark session read")) + return + } + s.hub.broadcastSessionRead(id, readAt) + writeJSON(w, http.StatusOK, map[string]any{"sessionId": id, "readAt": readAt}) +} + +// broadcastSessionRead fans a read-cursor advance out to every client as a +// `session_read` frame carrying the session's new cursor, so the same account's +// other ends clear the session's unread badge without a re-read. Like the other +// history frames it is an all-connections broadcast (the hub has no per-session +// subscription), and the client filters it by sessionId; the cursor's single source +// of truth stays chat_sessions.read_at, so a client that missed the frame converges +// on its next GET /sessions. +func (h *hub) broadcastSessionRead(sessionID string, readAt int64) { + h.broadcast(wsEvent{ + Type: wsSessionRead, + SessionID: sessionID, + Data: map[string]any{"readAt": readAt}, + }) +} diff --git a/internal/services/gateway/session_read_test.go b/internal/services/gateway/session_read_test.go new file mode 100644 index 0000000..2600d98 --- /dev/null +++ b/internal/services/gateway/session_read_test.go @@ -0,0 +1,233 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// These tests cover the D4 read-cursor contract (Android collaboration B3): the +// unreadCount carried by GET /api/v1/sessions, the advance endpoint +// POST /api/v1/sessions/{id}/read, and the `session_read` frame that clears the badge +// on the account's other browsers. + +// unreadBySession decodes GET /api/v1/sessions into a map of session id → unreadCount via +// the wire key, so the test asserts the JSON contract rather than the in-package struct. +func unreadBySession(t *testing.T, ts *httptest.Server, tok string) map[string]int { + t.Helper() + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list sessions status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var sessions []struct { + ID string `json:"id"` + UnreadCount int `json:"unreadCount"` + } + decodeDataSlice(t, body, &sessions) + out := make(map[string]int, len(sessions)) + for _, s := range sessions { + out[s.ID] = s.UnreadCount + } + return out +} + +// TestSessionsUnreadCountAndMarkRead covers the happy path end to end: a session with +// unread messages reports the count, POST /read moves the cursor and answers +// {sessionId, readAt}, the list drops to the messages sent after the cursor, and a +// body-less POST (mark everything seen) clears the badge. +func TestSessionsUnreadCountAndMarkRead(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + sub, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + for _, m := range []history.Message{ + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "a", CreatedAt: 1000}, + {SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "b", CreatedAt: 2000}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "c", CreatedAt: 3000}, + {SessionID: sub, Role: history.RoleUser, Content: "sub", CreatedAt: 1500}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/read" + + // Nothing read yet: the main session is fully unread, the sub session independently so. + if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 3 || got[sub] != 1 { + t.Fatalf("initial unread = %+v, want main 3 / sub 1", got) + } + + // Advance the main cursor to 2000: messages at or before it are read. + resp, body := doReq(t, http.MethodPost, base, map[string]any{"before": 2000}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("mark read status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + env := decodeBody(t, body) + if len(env) != 1 { + t.Fatalf("envelope = %v, want only a data field", env) + } + var data struct { + SessionID string `json:"sessionId"` + ReadAt int64 `json:"readAt"` + } + if err := json.Unmarshal(env["data"], &data); err != nil { + t.Fatalf("decode read data: %v (body=%s)", err, body) + } + if data.SessionID != history.MainSessionID || data.ReadAt != 2000 { + t.Fatalf("read data = %+v, want sessionId %s readAt 2000", data, history.MainSessionID) + } + + // Only the later message stays unread, and the sub session is untouched. + if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 1 || got[sub] != 1 { + t.Fatalf("unread after cursor 2000 = %+v, want main 1 / sub 1", got) + } + + // No body = "mark everything I have seen read" (resolved to now server-side). + resp, body = doReq(t, http.MethodPost, base, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("body-less mark status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &data); err != nil { + t.Fatalf("decode body-less read data: %v (body=%s)", err, body) + } + if data.ReadAt < 3000 { + t.Fatalf("body-less readAt = %d, want >= the newest message 3000", data.ReadAt) + } + if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 0 || got[sub] != 1 { + t.Fatalf("unread after mark-now = %+v, want main 0 / sub 1", got) + } + markedNow := data.ReadAt + + // The cursor only moves forward: re-posting a stale cursor must not re-open the badge. + resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("stale mark status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &data); err != nil { + t.Fatalf("decode stale read data: %v", err) + } + if data.ReadAt != markedNow { + t.Fatalf("stale readAt = %d, want the cursor to stay at %d", data.ReadAt, markedNow) + } + if got := unreadBySession(t, ts, tok); got[history.MainSessionID] != 0 { + t.Fatalf("unread after stale mark = %d, want 0", got[history.MainSessionID]) + } +} + +// TestMarkSessionReadValidation covers the failure contract: a present but unusable +// `before` answers 400, an unknown session 404, a viewer 403 and a missing HistoryStore +// 501 — none of which may be mistaken for a successful read. +func TestMarkSessionReadValidation(t *testing.T) { + t.Parallel() + _, ts, jwt, _ := newHistoryEnv(t) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + base := ts.URL + "/api/v1/sessions/web/read" + + for _, tc := range []struct { + name string + body any + }{ + {name: "zero", body: map[string]any{"before": 0}}, + {name: "negative", body: map[string]any{"before": -1}}, + {name: "non-numeric", body: map[string]any{"before": "abc"}}, + {name: "not an object", body: "not-an-object"}, + } { + t.Run(tc.name, func(t *testing.T) { + resp, body := doReq(t, http.MethodPost, base, tc.body, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (body=%s)", resp.StatusCode, body) + } + wantErrorShape(t, body) + }) + } + + // An empty object means "now", not "no-op". + resp, body := doReq(t, http.MethodPost, base, map[string]any{}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("empty object status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + + // Unknown session → 404; the session id is not auto-created. + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web:missing/read", map[string]any{"before": 1000}, tok) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("unknown session status = %d, want 404 (body=%s)", resp.StatusCode, body) + } + + // A viewer cannot write the cursor (owner/admin history write, like PATCH /sessions). + viewer := signAny(t, jwt, "acct", auth.RoleViewer) + resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, viewer) + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("viewer status = %d, want 403 (body=%s)", resp.StatusCode, body) + } + + // And no token at all is 401, not a silent success. + resp, body = doReq(t, http.MethodPost, base, map[string]any{"before": 1000}, "") + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("anonymous status = %d, want 401 (body=%s)", resp.StatusCode, body) + } +} + +// TestMarkSessionReadNotWired pins the honest degradation: with no HistoryStore injected +// the endpoint answers 501 not_implemented rather than pretending the cursor moved. +func TestMarkSessionReadNotWired(t *testing.T) { + t.Parallel() + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web/read", map[string]any{"before": 1000}, tok) + if resp.StatusCode != http.StatusNotImplemented { + t.Fatalf("status = %d, want 501 (body=%s)", resp.StatusCode, body) + } + wantErrorShape(t, body) +} + +// TestMarkSessionReadBroadcasts asserts the cross-end sync frame: a successful advance +// pushes exactly one `session_read` frame scoped to that session and carrying the new +// cursor, while a failed one (404) pushes nothing. +func TestMarkSessionReadBroadcasts(t *testing.T) { + t.Parallel() + srv, ts, jwt, _ := newHistoryEnv(t) + c := newFakeConn(srv.hub, 32) + srv.hub.register(c) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/read", map[string]any{"before": 2000}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + ev := readFakeFrame(t, c) + if ev.Type != "session_read" { + t.Fatalf("frame type = %q, want session_read (frame=%+v)", ev.Type, ev) + } + if ev.SessionID != history.MainSessionID { + t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, history.MainSessionID) + } + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame data = %+v, want object", ev.Data) + } + if data["readAt"] != float64(2000) { + t.Fatalf("frame data.readAt = %v, want 2000", data["readAt"]) + } + if err := connHasNoFrame(c); err != nil { + t.Fatal(err) + } + + // A failed advance must not tell other ends the badge was cleared. + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions/web:missing/read", map[string]any{"before": 2000}, tok) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("unknown status = %d, want 404 (body=%s)", resp.StatusCode, body) + } + if err := connHasNoFrame(c); err != nil { + t.Fatal(err) + } +}