feat(gateway): D3 审查读侧/revert 端点 + 语音用户可见面(L45)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

(a) 记忆入库前审查的两条 admin 端点(主人判「有必要就做」):
  GET /api/v1/admin/memories/reviews?limit= (留痕读侧,不含记忆正文)
  POST /api/v1/admin/memories/{id}/revert (回滚软合并,落审计 memory.review.revert)
  两条 writeAuth(裁定 DECISIONS #10:留痕属私有面,不给 viewer)
(b) 语音用户可见面:DTO voice(含服务端派生 url)+ GET /api/v1/media/{name}
  (200/206/304/416/HEAD、Accept-Ranges、非法名/穿越统一 404 media_not_found)
  + documents.Store.SaveAudio(内容寻址、$DATA_DIR 0600)+ history voice 列读写与引用计数差动

缺口(如实):message_persisted/握手回放仍不带 voice(文件面外,需给 history.Message 加字段);
AMR 需一行 VoiceTranscoder 注入(DECISIONS #11 已排)。4 个变异自检(含一条自认的等价变异)。
This commit is contained in:
AskaEth committed 2026-10-06 14:12:27 +08:00
1 parent 4a937bd269
commit 1420e8aaa6
15 files changed
+2202 -5

No files matched your search

+203
View File
@@ -0,0 +1,203 @@
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)
}
@@ -0,0 +1,213 @@
package history
// 本文件锁定 chat_messages.voice 列的**读写视图**(design/24 §10.4 / §10.9-4):
// ① 写入后能读回同样的 JSON 单对象,且列不存在时先幂等前滚(迁移 0020 之前建出的表);
// ② 坏形状**拒绝写入**(绝不落下读不回来的列);
// ③ 未知消息 id → ErrMessageNotFound(不假装写成功);
// ④ 音频副本的名字并入**同一套**引用计数(按新旧名字集合的差增减,同一事务);
// ⑤ 批量读是**按 id 集合一次查**,取不到的 id 不出现在映射里(与「取到空串」可区分)。
import (
"context"
"errors"
"testing"
voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice"
)
// seedVoiceMessage 经真实存储追加一条消息并返回已存行(含生成的 id)。
func seedVoiceMessage(t *testing.T, st *Store, role, content string) Message {
t.Helper()
got, err := st.AppendFull(context.Background(), Message{SessionID: MainSessionID, Role: role, Content: content})
if err != nil {
t.Fatalf("append %q: %v", content, err)
}
return got
}
// voiceRawWithCopy 编码一条「有副本」的语音事实(ASR 转写成功 + 副本三字段齐备)。
func voiceRawWithCopy(t *testing.T, name string) string {
t.Helper()
raw, err := voiceref.Encode(voiceref.Ref{
Source: voiceref.SourceASR,
Status: voiceref.StatusTranscribed,
Name: name,
MIME: "audio/ogg",
Size: 1234,
DurationMs: 3200,
})
if err != nil {
t.Fatalf("编码语音事实: %v", err)
}
return raw
}
// voiceRawWithoutCopy 编码一条「没有副本」的语音事实(微信按既有裁决不取字节)。
func voiceRawWithoutCopy(t *testing.T) string {
t.Helper()
raw, err := voiceref.Encode(voiceref.Ref{
Source: voiceref.SourcePlatform,
Status: voiceref.StatusTranscribed,
})
if err != nil {
t.Fatalf("编码语音事实: %v", err)
}
return raw
}
func TestUpdateMessageVoiceRoundTrip(t *testing.T) {
t.Parallel()
refs := &recordingRefs{}
st := newRefStore(t, refs)
ctx := context.Background()
msg := seedVoiceMessage(t, st, RoleUser, "(语音)")
name := "0123456789abcdef0123456789abcdef.ogg"
raw := voiceRawWithCopy(t, name)
// 全新库由本包 DDL 建表(**不含** voice 列)⇒ 这一次写入顺带把列幂等前滚。
if err := st.UpdateMessageVoice(ctx, msg.ID, raw); err != nil {
t.Fatalf("UpdateMessageVoice: %v", err)
}
cols, err := st.VoiceColumns(ctx, []string{msg.ID, "does-not-exist"})
if err != nil {
t.Fatalf("VoiceColumns: %v", err)
}
if got := cols[msg.ID]; got != raw {
t.Fatalf("读回 = %q, want %q", got, raw)
}
if _, ok := cols["does-not-exist"]; ok {
t.Fatal("取不到的 id 不该出现在映射里(与「取到空串」必须可区分)")
}
if len(refs.retained) != 1 || refs.retained[0] != name {
t.Fatalf("引用计数 retain = %v, want [%s]", refs.retained, name)
}
if len(refs.released) != 0 {
t.Fatalf("首次写入不该 release:%v", refs.released)
}
// 清空(= 撤销语音事实)是合法值,且副本的名字被释放一次。
if err := st.UpdateMessageVoice(ctx, msg.ID, ""); err != nil {
t.Fatalf("清空 voice: %v", err)
}
if len(refs.released) != 1 || refs.released[0] != name {
t.Fatalf("清空后 release = %v, want [%s]", refs.released, name)
}
cols, err = st.VoiceColumns(ctx, []string{msg.ID})
if err != nil {
t.Fatalf("VoiceColumns(清空后): %v", err)
}
if got, ok := cols[msg.ID]; !ok || got != "" {
t.Fatalf("清空后列 = (%q, %v), want (\"\", true)", got, ok)
}
}
func TestUpdateMessageVoiceRejectsMalformedAndUnknown(t *testing.T) {
t.Parallel()
refs := &recordingRefs{}
st := newRefStore(t, refs)
ctx := context.Background()
msg := seedVoiceMessage(t, st, RoleUser, "(语音)")
good := voiceRawWithCopy(t, "0123456789abcdef0123456789abcdef.wav")
if err := st.UpdateMessageVoice(ctx, msg.ID, good); err != nil {
t.Fatalf("先写一条合法值: %v", err)
}
// 坏形状全部拒绝,且**列保持原值**(绝不落下读不回来的内容)。
bad := []string{
"{not json",
`{"source":"asr"}`, // 状态缺席(闭集外)
`{"source":"telepathy","status":"transcribed"}`, // 来源不在闭集
`{"source":"asr","status":"transcribed","name":"x.ogg"}`, // 有 name 没有 mime/size(同生共死被破)
`{"source":"asr","status":"transcribed","mime":"audio/x"}`, // 有 mime 没有 name
`{"source":"asr","status":"transcribed","durationMs":-1}`, // 负时长
}
for _, raw := range bad {
err := st.UpdateMessageVoice(ctx, msg.ID, raw)
if !errors.Is(err, voiceref.ErrBadRef) {
t.Fatalf("UpdateMessageVoice(%q) err = %v, want wrap of ErrBadRef", raw, err)
}
}
cols, err := st.VoiceColumns(ctx, []string{msg.ID})
if err != nil {
t.Fatalf("VoiceColumns: %v", err)
}
if cols[msg.ID] != good {
t.Fatalf("坏写入改动了列:%q, want %q", cols[msg.ID], good)
}
// 未知 id ⇒ ErrMessageNotFound(绝不假装写成功)。
if err := st.UpdateMessageVoice(ctx, "no-such-message", good); !errors.Is(err, ErrMessageNotFound) {
t.Fatalf("未知 id err = %v, want ErrMessageNotFound", err)
}
}
func TestUpdateMessageVoiceSyncsRefsByDifference(t *testing.T) {
t.Parallel()
refs := &recordingRefs{}
st := newRefStore(t, refs)
ctx := context.Background()
msg := seedVoiceMessage(t, st, RoleUser, "(语音)")
first := voiceRawWithCopy(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.ogg")
second := voiceRawWithCopy(t, "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3")
// ① 没有 → 有副本:retain 一次。
if err := st.UpdateMessageVoice(ctx, msg.ID, first); err != nil {
t.Fatalf("第一次写入: %v", err)
}
// ② 内容逐字不变:一次都不增减(重复计数会让音频永远回收不掉)。
if err := st.UpdateMessageVoice(ctx, msg.ID, first); err != nil {
t.Fatalf("同值重写: %v", err)
}
if len(refs.retained) != 1 || len(refs.released) != 0 {
t.Fatalf("同值重写后 retain/release = %v/%v, want 1/0", refs.retained, refs.released)
}
// ③ 换成另一份副本:旧的释放、新的保留(各一次)。
if err := st.UpdateMessageVoice(ctx, msg.ID, second); err != nil {
t.Fatalf("换副本: %v", err)
}
if len(refs.retained) != 2 || refs.retained[1] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3" {
t.Fatalf("换副本后 retain = %v", refs.retained)
}
if len(refs.released) != 1 || refs.released[0] != "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.ogg" {
t.Fatalf("换副本后 release = %v", refs.released)
}
// ④ 有副本 → 无副本(微信那条路):release 一次,不新增 retain。
if err := st.UpdateMessageVoice(ctx, msg.ID, voiceRawWithoutCopy(t)); err != nil {
t.Fatalf("写成无副本: %v", err)
}
if len(refs.retained) != 2 {
t.Fatalf("无副本写法不该 retain:%v", refs.retained)
}
if len(refs.released) != 2 || refs.released[1] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3" {
t.Fatalf("无副本写法 release = %v", refs.released)
}
}
func TestVoiceColumnsBatchesAndSkipsBlanks(t *testing.T) {
t.Parallel()
st := newRefStore(t, &recordingRefs{})
ctx := context.Background()
withVoice := seedVoiceMessage(t, st, RoleUser, "语音")
without := seedVoiceMessage(t, st, RoleAssistant, "文本")
raw := voiceRawWithCopy(t, "cccccccccccccccccccccccccccccccc.webm")
if err := st.UpdateMessageVoice(ctx, withVoice.ID, raw); err != nil {
t.Fatalf("UpdateMessageVoice: %v", err)
}
cols, err := st.VoiceColumns(ctx, []string{withVoice.ID, without.ID, "", withVoice.ID})
if err != nil {
t.Fatalf("VoiceColumns: %v", err)
}
if len(cols) != 2 {
t.Fatalf("映射大小 = %d, want 2", len(cols))
}
if cols[withVoice.ID] != raw {
t.Fatalf("语音行 = %q, want %q", cols[withVoice.ID], raw)
}
if got, ok := cols[without.ID]; !ok || got != "" {
t.Fatalf("文本行 = (%q, %v), want (\"\", true)", got, ok)
}
if empty, err := st.VoiceColumns(ctx, nil); err != nil || len(empty) != 0 {
t.Fatalf("空 id 列表 ⇒ 空映射且无错误,得到 (%v, %v)", empty, err)
}
}