84 lines
4.2 KiB
Go
84 lines
4.2 KiB
Go
package history
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"strings"
|
||
)
|
||
|
||
// 本文件归属聊天消息的**回合标识**列（design/15 §2.5.9）：一条消息属于哪一路回合。
|
||
//
|
||
// 它与会话编排器分摊同一件事：编排器是那个**生成**权威 turnId 的地方（每个在飞的
|
||
// 路一份，挂进 loop.WithTurnID 并盖在该路全部流帧上），存储只是把它落进行里，使
|
||
// 「回复 ↔ 提问」的配对在**重连 / 回放**时依然成立 —— 流帧是易失的，历史行不是。
|
||
//
|
||
// 写入是**两段式**，与 document 列完全同一手法：
|
||
// - 消息落库时它所属的路已经存在（编排器新起的一路、以及该路的助手回复）⇒ 随
|
||
// INSERT 一起写入（Message.TurnID），因此 `message_persisted` 那一帧也带着它；
|
||
// - 排队（忙时入队）或**被转向（steer）吃进**的消息在落库那一刻**还没有自己的
|
||
// 路** —— 用户消息在入队时就落库，而它由哪一路回答要等那一轮才定。这种行的
|
||
// turn_id 只能在它**实际被处理的那一路**起跑 / 消费时补写，即本文件的
|
||
// SetMessagesTurnID。
|
||
//
|
||
// 空串（列的默认值）意味着「这一行不属于任何已知的回合」：加入该列之前的历史行、
|
||
// 以及**本就不属于任何回合**的写入路径（被静默登记的非主人消息、无入站回合的主动
|
||
// 推送）都是这种情形，读起来与以往逐字相同。IM 车道不在其列 —— 它没有编排器，但
|
||
// 渠道桥自己就是那条车道的回合权威，在落库前生成并写入这一列
|
||
// （internal/services/channelbridge/turnid.go）。
|
||
|
||
// SetMessagesTurnID 把 ids 所指的消息行的 turn_id 落定为 turnID —— 两段式写入的
|
||
// 第二段（起跑 / 被 steer 消费时补写）。
|
||
//
|
||
// 三条纪律：
|
||
// - **先到先得**：只写给仍然是空串的行（`AND turn_id = ”`）。一行只属于一路，
|
||
// 后来者绝不改写已经带标识的行 —— 否则「被哪一路回答」会随结算顺序漂移；
|
||
// - **幂等**：重复补写同一条（重启、重试、同一批消息被两处补写）是空操作；
|
||
// - **参数化**：id 列表按占位符展开，绝不把值拼进 SQL。
|
||
//
|
||
// 空 ids 或空 turnID 是空操作：调用方在「这一路没有落库任何用户消息」或「这一路
|
||
// 没有回合标识」（未接编排器的旧路径）时不必先判断。未知 id 被忽略而不是报错 ——
|
||
// 第二段的常态是「刚落的行」，一行找不到只可能是它已被删除，那不是本次写入的失败。
|
||
func (s *Store) SetMessagesTurnID(ctx context.Context, ids []string, turnID string) error {
|
||
unique := make([]string, 0, len(ids))
|
||
seen := make(map[string]bool, len(ids))
|
||
for _, id := range ids {
|
||
if id == "" || seen[id] {
|
||
continue
|
||
}
|
||
seen[id] = true
|
||
unique = append(unique, id)
|
||
}
|
||
if len(unique) == 0 || turnID == "" {
|
||
return nil
|
||
}
|
||
placeholders := strings.TrimSuffix(strings.Repeat("?, ", len(unique)), ", ")
|
||
args := make([]any, 0, len(unique)+1)
|
||
args = append(args, turnID)
|
||
for _, id := range unique {
|
||
args = append(args, id)
|
||
}
|
||
query := `UPDATE chat_messages SET turn_id = ? WHERE turn_id = '' AND id IN (` + placeholders + `)`
|
||
if _, err := s.db.ExecContext(ctx, query, args...); err != nil {
|
||
return fmt.Errorf("history: set turn id for %d messages: %w", len(unique), err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ensureTurnIDColumn 在 chat_messages 早于本列（0018 迁移）建表时把它前滚：缺列则
|
||
// 补加，已有则忽略重复列错误。它是幂等的，因此 NewStore 可在全新 schema 或既有
|
||
// schema 上调用（与 ensureClientMsgIDColumn / ensureDocumentColumn 同一手法）。
|
||
func ensureTurnIDColumn(ctx context.Context, db *sql.DB) error {
|
||
_, err := db.ExecContext(ctx, `ALTER TABLE chat_messages
|
||
ADD COLUMN turn_id TEXT NOT NULL DEFAULT ''`)
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
// 列已存在时（由 createMessagesTable 建出，或上一次运行补过）SQLite 报
|
||
// "duplicate column name: turn_id"。
|
||
if strings.Contains(err.Error(), "duplicate column name") {
|
||
return nil
|
||
}
|
||
return fmt.Errorf("history: roll forward turn_id column: %w", err)
|
||
}
|