Cyrene_v2/internal/runtime/history/voice_column.go

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)
}