package history // 本文件归属聊天消息的**语音事实**列(design/24 §10.4 / §10.9-4;迁移 0020): // 一条消息来自一条语音吗、那次转写成不成、盘上有没有一份可回放的副本。 // // 列的内容是**单个 JSON 对象**(形状见 `internal/runtime/voice.Ref`;空串 = 这条消息 // 不是语音消息)。它与 document 列的演进史不同:自诞生起就是单对象,因此没有需要 // 兼容的旧形状。 // // 写入是**两段式**(与 document 列同一手法):入站语音的消息行先落库,副本落盘拿到 // 内容寻址名之后再由渠道桥回填同一列 —— 因为「可播放副本」只有在转码/落盘之后才 // 存在,而那时消息行早已写好。回填入口 = 本文件的 UpdateMessageVoice(渠道桥按 // 窄接口 `voiceColumnWriter` 发现它,见 channelbridge/inbound_voice_copy.go)。 // // 音频副本与文档共用 `documents` 表与同一套引用计数(design/24 §10.5):副本一旦被 // 引用就要 +1,否则 GC 会把一份仍被播放的语音回收掉。因此本文件的每一次列变更都按 // **新旧名字集合的差**在同一条事务里增减计数(与 UpdateMessageDocument 逐字同族)。 import ( "context" "database/sql" "errors" "fmt" "strings" voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" ) // voiceColumn 是语音事实的列名(迁移 0020 加的列)。 const voiceColumn = "voice" // UpdateMessageVoice 只替换 id 所指消息的 voice 列,其余任何列保持已存值。它是 // 「入站语音落库后回填来源事实与副本引用」的写入端(design/24 §10.2 第 5 步)。 // // 四条纪律: // - **未知 id 返回 ErrMessageNotFound**(与 UpdateMessage / UpdateMessageDocument // 同一口径),调用方能区分「那一行不见了」与「写失败」,并据此如实记 WARN; // - **坏形状拒绝写**:非空值必须解码成自洽的语音事实(来源/状态在闭集内、副本三 // 字段同生共死)。写一份读不回来的 JSON 等于让客户端永远看不到这条语音,所以这里 // fail loud,绝不静默落一份半截数据。空串是**合法**值(这条消息不是语音); // - **引用计数与消息列同一个事务**:按新旧名字集合的差增减 `documents.ref_count` // (切片 S7 的既有通道),任何一边失败就整体回滚 —— 绝不留下「消息认了这份音频、 // 计数没加」这种让 GC 误删仍被引用文件的窗口; // - **列不存在时前滚**:迁移 0020 之前建出的表(含本包 DDL 造的测试库)在这里幂等 // 补列,与 ensureDocumentColumn / ensureTurnIDColumn 同一手法。 func (s *Store) UpdateMessageVoice(ctx context.Context, id, raw string) error { if _, err := voiceref.Decode(raw); err != nil { return fmt.Errorf("history: update message voice %s: %w", id, err) } if err := ensureVoiceColumn(ctx, s.db); err != nil { return err } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("history: update message voice %s: %w", id, err) } defer func() { _ = tx.Rollback() }() var previous string err = tx.QueryRowContext(ctx, `SELECT `+voiceColumn+` FROM chat_messages WHERE id = ?`, id).Scan(&previous) if err != nil { if errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("history: update message voice %s: %w", id, ErrMessageNotFound) } return fmt.Errorf("history: read message voice %s: %w", id, err) } res, err := tx.ExecContext(ctx, `UPDATE chat_messages SET `+voiceColumn+` = ? WHERE id = ?`, raw, id) if err != nil { return fmt.Errorf("history: update message voice %s: %w", id, err) } n, err := res.RowsAffected() if err != nil { return fmt.Errorf("history: update message voice %s rows: %w", id, err) } if n == 0 { return fmt.Errorf("history: update message voice %s: %w", id, ErrMessageNotFound) } if err := s.syncVoiceRefs(ctx, tx, previous, raw); err != nil { return err } if err := tx.Commit(); err != nil { return fmt.Errorf("history: update message voice %s: %w", id, err) } return nil } // VoiceColumns 按 id 批量取回各行的 voice 列原文(**一次** `IN (...)`,禁 N+1)。 // 返回「消息 id → 列内容」;查询不到的 id(已删除、id 拼错)**不出现在映射里**, // 与「取到空串」(这条消息不是语音)因此可以区分。 // // 它只读、不迁移、不建连接池(用本 Store 的活连接)。列不存在(迁移 0020 之前建出的 // 表且从未写过语音)时返回错误,由调用方决定怎么如实报告 —— 这里**不**静默回空映射, // 「没有这一面」与「这些消息都不是语音」是两件事。 func (s *Store) VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) { unique := uniqueMessageIDs(ids) if len(unique) == 0 { return map[string]string{}, nil } placeholders := strings.TrimSuffix(strings.Repeat("?, ", len(unique)), ", ") args := make([]any, 0, len(unique)) for _, id := range unique { args = append(args, id) } query := `SELECT id, ` + voiceColumn + ` FROM chat_messages WHERE id IN (` + placeholders + `)` rows, err := s.db.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("history: read voice columns for %d messages: %w", len(unique), err) } defer func() { _ = rows.Close() }() out := make(map[string]string, len(unique)) for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { return nil, fmt.Errorf("history: scan voice column: %w", err) } out[id] = raw } if err := rows.Err(); err != nil { return nil, fmt.Errorf("history: iterate voice columns: %w", err) } return out, nil } // uniqueMessageIDs 去重并丢掉空 id(保持首次出现顺序)。 func uniqueMessageIDs(ids []string) []string { 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) } return unique } // syncVoiceRefs 把一行消息的 voice 列从 oldRaw 换成 newRaw 时按**差**增减引用计数。 // 两次内容相同时(含都是空)直接返回。 // // 与文档侧同一取舍:列被外部改坏(解不出名字)时**不让写入失败**(消息行才是用户的 // 事实),只记 WARN 并跳过本次增减 —— 保守方向是「计数偏低 ⇒ 只会晚回收,绝不早删」。 func (s *Store) syncVoiceRefs(ctx context.Context, ex SQLExecer, oldRaw, newRaw string) error { if s.refs == nil || oldRaw == newRaw { return nil } oldNames, err := voiceNames(oldRaw) if err != nil { s.log.Warn("voice refs sync skipped: malformed old column", "err", err) return nil } newNames, err := voiceNames(newRaw) if err != nil { s.log.Warn("voice refs sync skipped: malformed new column", "err", err) return nil } add, remove := diffDocumentNames(oldNames, newNames) if len(add) > 0 { if err := s.refs.Retain(ctx, ex, add); err != nil { return fmt.Errorf("history: retain voice refs: %w", err) } } if len(remove) > 0 { if err := s.refs.Release(ctx, ex, remove); err != nil { return fmt.Errorf("history: release voice refs: %w", err) } } return nil } // voiceNames 解出 voice 列涉及的音频副本名(有副本时恰好一个,没有副本/不是语音时为空)。 // 与本包 documentNames 同形,因此复用同一套差集计算。 func voiceNames(raw string) ([]string, error) { ref, err := voiceref.Decode(raw) if err != nil { return nil, err } if !ref.HasCopy() { return nil, nil } return []string{ref.Name}, nil } // ensureVoiceColumn 在 chat_messages 早于本列(0020 迁移)建表时把它前滚:缺列则 // 补加,已有则忽略重复列错误。它是幂等的(与 ensureDocumentColumn / ensureTurnIDColumn // 同一手法),因此本包的写入路径可以在全新 schema 或既有 schema 上直接调用。 // // 只在**写入路径**调用:读路径不迁移(只读纪律),而一条消息要带语音事实必然先经过 // 写入路径,因此列在读到它之前一定已经补上。 func ensureVoiceColumn(ctx context.Context, db *sql.DB) error { _, err := db.ExecContext(ctx, `ALTER TABLE chat_messages ADD COLUMN voice TEXT NOT NULL DEFAULT ''`) if err == nil { return nil } // 列已存在时(由 0020 迁移补过,或上一次写入补过)SQLite 报 // "duplicate column name: voice"。 if strings.Contains(err.Error(), "duplicate column name") { return nil } return fmt.Errorf("history: roll forward voice column: %w", err) }