package gateway import ( "context" "encoding/json" "errors" "io" "net/http" "time" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" ) // 本文件归属 D4 已读游标面(Android 协作 B3):推进会话已读游标的 // POST /api/v1/sessions/{id}/read 端点,以及它为清除账号其他浏览器 // 未读徽标(无需重读)而广播的 `session_read` 实时帧。游标本身存于 // chat_sessions.read_at(internal/runtime/history);GET /sessions 由它 // 推导 unreadCount(history.go)。游标是会话级、单所有者的: // 一个账号每个会话一个游标 —— 在任意一端阅读即为所有端标记该 // 会话已读。 // // 同一条游标还喂养**落库那一刻**的实时面(契约 v0.76):单会话的绝对未读数 // 由 unreadCountOf 读回,随 `message_persisted` 帧的 `data.unreadCount` 下发, // 使客户端不必为换角标再拉一次 GET /sessions。两条面共用同一个数字口径 // (`created_at > read_at`),因此不会互相打脸。 // wsSessionRead 是已读游标推进的 WS 事件类型。 const wsSessionRead = "session_read" // sessionReadRequest 是 POST /api/v1/sessions/{id}/read 的请求体。 // Before 是指针,以便把字段缺失(空请求体)与显式取值区分开: // 缺失表示「现在」—— 客户端把它已看到的一切都标记为已读。 type sessionReadRequest struct { Before *int64 `json:"before"` } // handleMarkSessionRead 推进会话的已读游标,并以 Unix 毫秒返回结果 // 游标 {data:{sessionId, readAt}}。与其他历史写入一样走 owner/admin // 门(writeAuth)。 // // 请求体 {before?} 是可选毫秒时间戳,界定标记已读的范围 —— // created_at <= readAt 的消息变为已读。缺失(或空请求体)表示现在。 // 已给出但非正数或非数值的 before 返回 400:游标是真实的毫秒 // 时间戳,「负数游标造成的未读」不是有意义的客户端请求。未知 // 会话 id 返回 404;HistoryStore 缺失返回 501。 // // 成功时新游标作为 `session_read` 帧广播,使该账号其他浏览器也清除 // 徽标(一端阅读 = 处处已读)。失败路径不广播任何东西。 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 把一次已读游标推进作为 `session_read` 帧扇出到 // 每个客户端,携带会话的新游标,使同一账号的其他端无需重读即可 // 清除该会话的未读徽标。它属于 design/11 §7.1 的「状态机信号」 // (只报告游标,绝不报告消息正文),因此帧门让它留在全连接路径上。 // 与其他历史帧一样,它依赖 hub 没有按会话订阅这一点,由客户端按 // sessionId 过滤;游标的唯一事实来源仍是 chat_sessions.read_at, // 因此错过该帧的客户端会在下次 GET /sessions 时重新收敛。 func (h *hub) broadcastSessionRead(sessionID string, readAt int64) { h.broadcast(wsEvent{ Type: wsSessionRead, SessionID: sessionID, Data: map[string]any{"readAt": readAt}, }) } // sessionUnreadCounter 是 HistoryStore 的一个**可选**更窄视图(与 clientMsgIDReader / // turnIDWriter / documentStatusWriter 同一做法):按会话读回由已读游标推导的绝对未读数。 // 未实现该视图的存储替身(测试替身、未接线的构建)**不报错**,只是帧上不带该键 —— // 未读面是增益,绝不能反过来卡住消息广播。 type sessionUnreadCounter interface { UnreadCount(ctx context.Context, sessionID string) (int, error) } const ( // unreadCountTimeout 是**单次**未读数查询的时长上界。它跑在落库之后的广播路径上 // (写完一条就发一帧),必须有一个界:一次库卡死绝不允许把广播挂住。 unreadCountTimeout = 2 * time.Second // unreadCountLogSource 是这条冷路径上日志的来源标签(AGENTS §10 禁无来源裸日志)。 unreadCountLogSource = "gateway/broadcast" ) // unreadCountOf 返回该会话当前的**绝对未读数**(比 chat_sessions.read_at 新的消息数), // 第二个返回值表示「这个数字是真的」—— false 时调用方必须**省略**该键,绝不用 0 冒充 // 「已读完」(缺席 = 老行为,客户端回落到 GET /sessions 收敛)。 // // 三条不产生数字的情形,都不算错误:① 未注入 HistoryStore;② 注入的存储没实现上面那个 // 窄视图;③ 查询失败(记一条带来源的 WARN,因为「读不到」不是「没有未读」)。 func (s *Server) unreadCountOf(sessionID string) (int, bool) { if s.history == nil || sessionID == "" { return 0, false } counter, ok := s.history.(sessionUnreadCounter) if !ok { return 0, false } ctx, cancel := context.WithTimeout(context.Background(), unreadCountTimeout) defer cancel() n, err := counter.UnreadCount(ctx, sessionID) if err != nil { logger.WithSource(s.log, unreadCountLogSource).Warn("unread count read failed", "err", err, "session", sessionID) return 0, false } return n, true }