AskaEth
7512d593c2
- G2：`stream:true` 原经**有损总线**且**先回 200** ⇒ 总线饱和时客户端拿到 200 却什么都没发生。
改为**直达编排器**（与 ws chat 帧既有先例同一原则）：新文件 chat_stream_dispatch.go；200 变成有意义的
承诺（同步穿过「编排器接收 + 用户消息落库」才回）；顺带对齐 `202 {queued,queueDepth}` / `409 chat_busy`；
兜底路径（未接编排器才走总线）现在**返回错误**、ctx 与请求脱钩（WithoutCancel + 30min 上界）→ `503 chat_dispatch_failed`；
未接 loop 由 200 改为 `500 chat_unavailable`（与同步路径同口径，已同步改既有用例并新增钉住用例）
- G4：`chat`（WS 帧 + REST 体）新增**可选** `clientMsgId` ⇒ 幂等。
原子性两层，均非「先查再插」：① 同进程在 `o.mu` 同一临界区内完成「查认领表→查存储→预分配 id→认领」；
② 跨重启靠 `(session_id, client_msg_id)` **部分**唯一索引 + `INSERT … ON CONFLICT DO NOTHING`（命中则读回既有行）。
重复语义：REST `200 {duplicate:true, messageId}`；WS 新帧 `chat_duplicate`（owner/admin 档）；重复**不落库、不起回合**。
`clientMsgId` 回填进 DTO ⇒ `message_persisted` 与读接口都可见（客户端可据此对齐本地气泡，不再靠文本相等）
- 迁移 `0013_chat_client_msg_id`（up：加列 + 部分唯一索引 `idx_chat_messages_client_msg_id`；down：先删索引再删列），
已在 `data/cyrene.db` **副本**上验 up → 判重 → down → 再 up
- 拆分：store.go 647→498 行（`log_reads.go` 抽走四个读路径）；七个 SELECT 清单抽成 `messageColumns` 一处
- 复核：**全树 `go test ./... -count=1` 0 FAIL** + build/vet 绿 + linux/darwin 交叉编译绿；
四条真断言（总线饱和仍落库出帧且 `ws.chat` 零发布 / 同键两次仅一回合＋并发两条也仅一回合 / 省略键逐字不变 /
迁移副本 up-down-up）；**3 条变异反证**均已还原
- 越界说明（均已报备）：`history.go` DTO 字段（否则「读接口可见」不可能）、`policy_store_test.go`
（0013 使「回滚最后一步 == 0012」前提失效，改为按 schema_migrations 现算）、`extra_test.go`（补 runner）
- 顺带补交 P0-A 遗漏的新测试文件 `loop/memory_writeback_test.go`
72 lines
3.5 KiB
Go
72 lines
3.5 KiB
Go
package history
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
// 本文件归属聊天消息的**客户端幂等键**（Android 协同 §G4）。
|
|
//
|
|
// 它是与会话编排器（internal/services/gateway/session_orchestrator.go）分摊的
|
|
// 一件事的两半：编排器决定「这一轮还要不要跑」，存储决定「这一行还要不要写」。
|
|
// 存储这半边的核心是 (session_id, client_msg_id) 上的部分唯一索引（见 0013
|
|
// 迁移与本文件的 createClientMsgIDIndex）—— 判重由数据库原子裁决，因此并发的
|
|
// 两次重发只有一个赢家，不存在「先查再插」的竞态窗口。
|
|
//
|
|
// client_msg_id 为空串表示「这条消息没有幂等键」（老客户端、IM 镜像、历史行），
|
|
// 部分索引刻意把它排除在外：否则同一会话的第二条空键消息就会撞上唯一约束。
|
|
|
|
// NewMessageID 返回一个新的消息 id（128 位密码学随机、十六进制编码），与
|
|
// insert 在调用方省略 ID 时分配的 id 同源。它被会话编排器用来**预分配**一条
|
|
// 用户消息的 id：这样同一条 clientMsgId 的并发重发即使赶在第一条真正落库之前
|
|
// 到达，也能立刻拿到一个确定且终将存在的 messageId（而不是空值）。
|
|
func NewMessageID() (string, error) {
|
|
return newID()
|
|
}
|
|
|
|
// FindByClientMsgID 按 (sessionID, clientMsgID) 读回既有消息，供幂等判重。
|
|
// 没有该行时返回包装了 ErrMessageNotFound 的错误；clientMsgID 为空时同理
|
|
// （空键不是一条可判重的消息）。查询走部分唯一索引，至多命中一行。
|
|
func (s *Store) FindByClientMsgID(ctx context.Context, sessionID, clientMsgID string) (Message, error) {
|
|
if clientMsgID == "" {
|
|
return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, ErrMessageNotFound)
|
|
}
|
|
row := s.db.QueryRowContext(ctx, `SELECT `+messageColumns+`
|
|
FROM chat_messages WHERE session_id = ? AND client_msg_id = ?`, sessionID, clientMsgID)
|
|
m, err := scanMessage(row)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, ErrMessageNotFound)
|
|
}
|
|
return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, err)
|
|
}
|
|
return m, nil
|
|
}
|
|
|
|
// createClientMsgIDIndex 建 (session_id, client_msg_id) 的**部分**唯一索引
|
|
// （只覆盖非空键）。它同时是幂等写入的冲突裁决者：insert 带非空
|
|
// ClientMsgID 时用 ON CONFLICT DO NOTHING，由该索引决定谁写下这一行。
|
|
const createClientMsgIDIndex = `
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_messages_client_msg_id
|
|
ON chat_messages (session_id, client_msg_id) WHERE client_msg_id != ''`
|
|
|
|
// ensureClientMsgIDColumn 在 chat_messages 早于本列（0013 迁移）建表时把它
|
|
// 前滚：缺列则补加，已有则忽略重复列错误。它是幂等的，因此 NewStore 可在全新
|
|
// schema 或既有 schema 上调用（与 memory 包的 ensureProjectIDColumn 同一手法）。
|
|
func ensureClientMsgIDColumn(ctx context.Context, db *sql.DB) error {
|
|
_, err := db.ExecContext(ctx, `ALTER TABLE chat_messages
|
|
ADD COLUMN client_msg_id TEXT NOT NULL DEFAULT ''`)
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
// 列已存在时（由 createMessagesTable 建出，或上一次运行补过）SQLite 报
|
|
// "duplicate column name: client_msg_id"。
|
|
if strings.Contains(err.Error(), "duplicate column name") {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("history: roll forward client_msg_id column: %w", err)
|
|
}
|