package gateway

// 本文件钉住 M44：语音面在**实时帧**与**握手回放**上的纯增量覆盖（契约 v0.67 §4.4）。
//
// 五条断言各对应一条取舍：
//  ① `message_persisted` 的 `data.message.voice` 出现，且与 `GET /sessions/{id}/messages`
//     同一行**逐字相等** —— 这是本切片唯一的意义：客户端不必再拉一次 REST 才认出语音气泡；
//  ② 非语音消息该键**整键缺席**（不是 `null`、不是 `{}`）；
//  ③ `channel_persisted`（隔离渠道会话）同样带，且与渠道日志读面同一行逐字相等；
//  ④ 握手 `history` 回放同样带（老客户端因此重连即可拿到，不必等重新拉取）；
//  ⑤ 成本实测：一帧 = **一次**批量读列（`IN (?)`）、回放一页 = **一次**（id 去重）——
//     不是逐条查，也不是「每帧多打几枪」。
//
// 断言一律打在**帧的序列化输出**上（`readFakeFrame` / `ReadMessage` 都是先 marshal 再
// 回读），因此「键缺席」与「零值」的区别是真的可分辨的 —— 解码进结构体做不到这一点。

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice"
)

// countingVoiceStore 在真实存储外面只给 `VoiceColumns` 记账：本切片的核心成本问题是
// 「一条消息多打几枪、一次带几个 id」，所以把调用次数与 id 数**量出来**，而不是凭印象
// 说「一次批量」。
type countingVoiceStore struct {
	*history.Store
	mu    sync.Mutex
	calls int
	ids   []string
}

func (c *countingVoiceStore) VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) {
	c.mu.Lock()
	c.calls++
	c.ids = append(c.ids, ids...)
	c.mu.Unlock()
	return c.Store.VoiceColumns(ctx, ids)
}

// reset 把计数器归零：每个断言段各自量自己的代价。
func (c *countingVoiceStore) reset() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.calls, c.ids = 0, nil
}

// snapshot 返回 (调用次数, 各次查询的 id 并集)。
func (c *countingVoiceStore) snapshot() (int, []string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.calls, append([]string(nil), c.ids...)
}

// voiceFrameEnv 打包实时帧 / 回放测试所需的依赖：真实（内存）历史存储（外套记账层）
// + httptest 网关 + JWT 服务。
type voiceFrameEnv struct {
	srv   *Server
	ts    *httptest.Server
	jwt   *auth.JWTService
	st    *history.Store
	count *countingVoiceStore
}

func newVoiceFrameEnv(t *testing.T) *voiceFrameEnv {
	t.Helper()
	ctx := context.Background()
	conn, err := db.Open(ctx, ":memory:", testLogger())
	if err != nil {
		t.Fatalf("打开内存库: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })
	st, err := history.NewStore(ctx, conn, testLogger())
	if err != nil {
		t.Fatalf("构造历史存储: %v", err)
	}
	count := &countingVoiceStore{Store: st}
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{History: count})
	return &voiceFrameEnv{srv: srv, ts: ts, jwt: jwt, st: st, count: count}
}

// seed 追加一条消息，并把（可选的）语音事实写进它的 voice 列 —— 这与生产里渠道桥
// 「落库后回填同一列」的两段式同形。
func (e *voiceFrameEnv) seed(t *testing.T, m history.Message, raw string) history.Message {
	t.Helper()
	got, err := e.st.AppendFull(context.Background(), m)
	if err != nil {
		t.Fatalf("落库消息: %v", err)
	}
	if raw == "" {
		return got
	}
	if err := e.st.UpdateMessageVoice(context.Background(), got.ID, raw); err != nil {
		t.Fatalf("写 voice 列: %v", err)
	}
	return got
}

// encodeVoiceFact 编码一条语音事实；失败即测试前置条件坏掉。
func encodeVoiceFact(t *testing.T, ref voiceref.Ref) string {
	t.Helper()
	raw, err := voiceref.Encode(ref)
	if err != nil {
		t.Fatalf("编码语音事实: %v", err)
	}
	return raw
}

// canonicalJSON 把一段 JSON 归一成「键序由 json.Marshal 的 map 排序决定」的文本。
// JSON 对象的键**无序**，因此「逐字相等」在对象层面就该按归一后的文本来判：两侧都走
// 同一套归一，任何一处多一个键 / 少一个键 / 值不同都会让文本不同。
func canonicalJSON(t *testing.T, raw json.RawMessage) string {
	t.Helper()
	var v any
	if err := json.Unmarshal(raw, &v); err != nil {
		t.Fatalf("解析 JSON（raw=%s）: %v", raw, err)
	}
	b, err := json.Marshal(v)
	if err != nil {
		t.Fatalf("归一 JSON（raw=%s）: %v", raw, err)
	}
	return string(b)
}

// frameMessageVoice 从**已序列化并回读**的帧里取出 `data.message.voice`，返回归一后的
// JSON 文本与「该键是否在场」。本面的关键取舍就是「缺席」，因此先断言在场性再看内容。
func frameMessageVoice(t *testing.T, ev wsEvent) (string, bool) {
	t.Helper()
	data, ok := ev.Data.(map[string]any)
	if !ok {
		t.Fatalf("帧 data = %+v, want 一个 JSON 对象", ev.Data)
	}
	msg, ok := data["message"].(map[string]any)
	if !ok {
		t.Fatalf("帧 data.message = %+v, want 一个 JSON 对象", data["message"])
	}
	raw, present := msg["voice"]
	if !present {
		return "", false
	}
	b, err := json.Marshal(raw)
	if err != nil {
		t.Fatalf("归一帧里的 voice（%v）: %v", raw, err)
	}
	return string(b), true
}

// replayMessageVoice 与 frameMessageVoice 同形，只是取的是握手 `history` 帧里
// `data.messages[]` 的某一条（按 id 定位）。
func replayMessageVoice(t *testing.T, ev wsEvent, id string) (string, bool) {
	t.Helper()
	data, ok := ev.Data.(map[string]any)
	if !ok {
		t.Fatalf("回放帧 data = %+v, want 一个 JSON 对象", ev.Data)
	}
	rows, ok := data["messages"].([]any)
	if !ok {
		t.Fatalf("回放帧 data.messages = %+v, want 一个数组", data["messages"])
	}
	for _, row := range rows {
		msg, ok := row.(map[string]any)
		if !ok {
			t.Fatalf("回放消息 = %+v, want 一个 JSON 对象", row)
		}
		if msg["id"] != id {
			continue
		}
		raw, present := msg["voice"]
		if !present {
			return "", false
		}
		b, err := json.Marshal(raw)
		if err != nil {
			t.Fatalf("归一回放里的 voice（%v）: %v", raw, err)
		}
		return string(b), true
	}
	t.Fatalf("回放里没有 id=%q 的消息", id)
	return "", false
}

// restVoiceOf 取 REST 读面里**同一行**消息的 `voice`，归一成与帧侧可比的文本。
// 键缺席时返回 ("", false)。
func restVoiceOf(t *testing.T, rows []map[string]json.RawMessage, id string) (string, bool) {
	t.Helper()
	for _, row := range rows {
		var gotID string
		if err := json.Unmarshal(row["id"], &gotID); err != nil {
			t.Fatalf("解析 id: %v", err)
		}
		if gotID != id {
			continue
		}
		raw, present := row["voice"]
		if !present {
			return "", false
		}
		return canonicalJSON(t, raw), true
	}
	t.Fatalf("REST 读面里没有 id=%q 的消息", id)
	return "", false
}

// TestBroadcastPersistedCarriesVoiceMatchingRestRead 断言 ①②⑤：语音消息的
// `message_persisted` 帧带 `voice`、与 REST 读面逐字相等；非语音消息整键缺席；
// 一帧的代价是一次批量读列（一个 id）。
func TestBroadcastPersistedCarriesVoiceMatchingRestRead(t *testing.T) {
	t.Parallel()
	env := newVoiceFrameEnv(t)
	tok := signAny(t, env.jwt, "acct", auth.RoleOwner)
	name := "0123456789abcdef0123456789abcdef.ogg"
	voiceMsg := env.seed(t, history.Message{
		SessionID: history.MainSessionID, Role: history.RoleUser, Content: "转写文本",
	}, encodeVoiceFact(t, voiceref.Ref{
		Source:     voiceref.SourceASR,
		Status:     voiceref.StatusTranscribed,
		Name:       name,
		MIME:       "audio/ogg",
		Size:       4242,
		DurationMs: 3200,
	}))
	plainMsg := env.seed(t, history.Message{
		SessionID: history.MainSessionID, Role: history.RoleUser, Content: "普通文本",
	}, "")

	c := newOwnerConn(env.srv.hub, 8)
	env.srv.hub.register(c)

	env.count.reset()
	env.srv.BroadcastPersisted(voiceMsg)
	ev := readFakeFrame(t, c)
	if ev.Type != "message_persisted" {
		t.Fatalf("帧类型 = %q, want message_persisted", ev.Type)
	}
	frameVoice, present := frameMessageVoice(t, ev)
	if !present {
		t.Fatalf("语音消息的 message_persisted 帧缺 data.message.voice（帧=%+v）", ev)
	}
	// ⑤ 成本：一帧 = 一次批量读列，恰好一个 id（没有逐条多打一枪）。**先**取快照，
	// 免得下面那次 REST 拉取的记账混进来。
	calls, ids := env.count.snapshot()

	// 与 REST 读面同一行**逐字相等**：这是本切片唯一的意义。
	resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("REST status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	restVoice, restOK := restVoiceOf(t, messagePage(t, body), voiceMsg.ID)
	if !restOK {
		t.Fatalf("REST 读面同一行缺 voice 键（body=%s）", body)
	}
	if frameVoice != restVoice {
		t.Fatalf("帧里的 voice = %s, REST 读面同一行 = %s（两处各算一份就失去意义）", frameVoice, restVoice)
	}

	// 字段值逐项落地（给失败信息可读性；形状正确性由上面的逐字比较兜底）。
	var fields historyVoice
	if err := json.Unmarshal([]byte(frameVoice), &fields); err != nil {
		t.Fatalf("解析帧里的 voice: %v（raw=%s）", err, frameVoice)
	}
	if fields.Source != "asr" || fields.Status != "transcribed" ||
		fields.Name != name || fields.MIME != "audio/ogg" || fields.Size != 4242 ||
		fields.DurationMs != 3200 || fields.URL != voiceref.MediaPathPrefix+name {
		t.Fatalf("帧里的 voice 字段 = %+v", fields)
	}

	// ⑤ 成本：一帧 = 一次批量读列，恰好一个 id（没有逐条多打一枪）。
	if calls != 1 || len(ids) != 1 || ids[0] != voiceMsg.ID {
		t.Fatalf("一帧的读列代价 = %d 次调用 / ids=%v，want 1 次 / [%s]", calls, ids, voiceMsg.ID)
	}

	// ② 非语音消息：该键**整键缺席**（不是 `null`、不是 `{}`）。
	env.srv.BroadcastPersisted(plainMsg)
	plainFrame := readFakeFrame(t, c)
	if got, present := frameMessageVoice(t, plainFrame); present {
		t.Fatalf("非语音消息的帧带上了 voice = %s，want 整键缺席", got)
	}
}

// TestChannelPersistedCarriesVoiceMatchingChannelLogRead 断言 ③⑤：隔离渠道会话
// （群聊 / 陌生人）走的是同一个构造函数，因此 `channel_persisted` 同样带 `voice`，
// 且与渠道日志读面同一行逐字相等 —— 「Web 帧带、IM 帧不带」的分叉不存在。
func TestChannelPersistedCarriesVoiceMatchingChannelLogRead(t *testing.T) {
	t.Parallel()
	env := newVoiceFrameEnv(t)
	tok := signAny(t, env.jwt, "acct", auth.RoleOwner)
	adapter := "qq-obv11:10001"
	name := "ffffffffffffffffffffffffffffffff.mp3"
	msg := env.seed(t, history.Message{
		SessionID: "sess-group-1", Role: history.RoleUser, Content: "群里的语音",
		Platform: "qq-obv11", AdapterID: adapter, SenderName: "某人",
	}, encodeVoiceFact(t, voiceref.Ref{
		Source: voiceref.SourceASR, Status: voiceref.StatusTranscribed,
		Name: name, MIME: "audio/mpeg", Size: 99,
	}))

	c := newOwnerConn(env.srv.hub, 8)
	env.srv.hub.register(c)

	env.count.reset()
	env.srv.BroadcastPersisted(msg)
	ev := readFakeFrame(t, c)
	if ev.Type != "channel_persisted" {
		t.Fatalf("帧类型 = %q, want channel_persisted（隔离会话不是 Web 主会话）", ev.Type)
	}
	frameVoice, present := frameMessageVoice(t, ev)
	if !present {
		t.Fatalf("隔离会话的 channel_persisted 帧缺 data.message.voice（帧=%+v）", ev)
	}
	// ⑤ 成本快照**先**取，免得下面那次渠道日志拉取的记账混进来。
	calls, ids := env.count.snapshot()

	resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/channels/messages?adapter="+adapter, nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("渠道日志 status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	restVoice, restOK := restVoiceOf(t, messagePage(t, body), msg.ID)
	if !restOK {
		t.Fatalf("渠道日志同一行缺 voice 键（body=%s）", body)
	}
	if frameVoice != restVoice {
		t.Fatalf("帧里的 voice = %s, 渠道日志同一行 = %s", frameVoice, restVoice)
	}

	if calls != 1 || len(ids) != 1 {
		t.Fatalf("一帧的读列代价 = %d 次调用 / ids=%v，want 1 次 / 1 个 id", calls, ids)
	}
}

// TestWSHandshakeReplayCarriesVoice 断言 ④⑤：握手 `history` 回放同样带 `voice`
// （与 REST 读面同一行逐字相等），非语音行之键缺席，且整页只查一次列。
func TestWSHandshakeReplayCarriesVoice(t *testing.T) {
	t.Parallel()
	env := newVoiceFrameEnv(t)
	tok := signAny(t, env.jwt, "acct", auth.RoleOwner)
	name := "abcdefabcdefabcdefabcdefabcdefab.ogg"
	voiceMsg := env.seed(t, history.Message{
		SessionID: history.MainSessionID, Role: history.RoleUser, Content: "回放里的语音", CreatedAt: 1000,
	}, encodeVoiceFact(t, voiceref.Ref{
		Source: voiceref.SourceASR, Status: voiceref.StatusTranscribed,
		Name: name, MIME: "audio/ogg", Size: 777, DurationMs: 1200,
	}))
	plainMsg := env.seed(t, history.Message{
		SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "回放里的纯文本", CreatedAt: 1500,
	}, "")

	env.count.reset()
	conn := dialWS(t, env.ts.URL, tok)
	defer conn.Close()
	_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
	_, raw, err := conn.ReadMessage()
	if err != nil {
		t.Fatalf("读取握手帧: %v", err)
	}
	var ev wsEvent
	if err := json.Unmarshal(raw, &ev); err != nil {
		t.Fatalf("解析握手帧 %s: %v", raw, err)
	}
	if ev.Type != "history" {
		t.Fatalf("首帧类型 = %q, want history（%s）", ev.Type, raw)
	}
	replayVoice, present := replayMessageVoice(t, ev, voiceMsg.ID)
	if !present {
		t.Fatalf("握手回放里语音行缺 voice（帧=%s）", raw)
	}
	// ⑤ 整页一次批量：恰好 1 次调用、把两行的 id 都带上。快照**先**取，免得下面那次
	// REST 拉取的记账混进来。
	calls, ids := env.count.snapshot()

	// 与 REST 读面同一行逐字相等。
	resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("REST status = %d, want 200 (body=%s)", resp.StatusCode, body)
	}
	restVoice, restOK := restVoiceOf(t, messagePage(t, body), voiceMsg.ID)
	if !restOK {
		t.Fatalf("REST 读面同一行缺 voice 键（body=%s）", body)
	}
	if replayVoice != restVoice {
		t.Fatalf("回放里的 voice = %s, REST 读面同一行 = %s", replayVoice, restVoice)
	}
	if got, present := replayMessageVoice(t, ev, plainMsg.ID); present {
		t.Fatalf("回放里非语音行带上了 voice = %s，want 整键缺席", got)
	}

	// ⑤ 整页一次批量：恰好 1 次调用，且把两行的 id 都带上了。
	if calls != 1 || len(ids) != 2 {
		t.Fatalf("回放一页的读列代价 = %d 次调用 / ids=%v，want 1 次 / 2 个 id", calls, ids)
	}
}

// TestBroadcastPersistedVoiceAbsentWhenStoreCannotAnswer 钉住两条**诚实缺席**路径：
// 未接线历史存储、以及存储没提供语音列读面（测试替身 / 只做基本读写的实现）。
// 两者都不许 panic、不许把帧吞掉 —— 那条路径上「问不到」的结果就是字段缺席。
func TestBroadcastPersistedVoiceAbsentWhenStoreCannotAnswer(t *testing.T) {
	t.Parallel()
	pending := history.Message{
		ID: "row-1", SessionID: history.MainSessionID, Role: history.RoleUser, Content: "hi", CreatedAt: 1000,
	}

	t.Run("history not wired", func(t *testing.T) {
		srv, _, _ := buildEnv(t, cfgDefault(), Deps{})
		c := newOwnerConn(srv.hub, 8)
		srv.hub.register(c)

		srv.BroadcastPersisted(pending)
		ev := readFakeFrame(t, c)
		if ev.Type != "message_persisted" {
			t.Fatalf("帧类型 = %q, want message_persisted（问不到语音列不该吞掉帧）", ev.Type)
		}
		if got, present := frameMessageVoice(t, ev); present {
			t.Fatalf("未接线历史存储的帧带上了 voice = %s，want 缺席", got)
		}
	})

	t.Run("store without the voice view", func(t *testing.T) {
		srv, _, _ := buildEnv(t, cfgDefault(), Deps{History: &fakeHistory{}})
		c := newOwnerConn(srv.hub, 8)
		srv.hub.register(c)

		srv.BroadcastPersisted(pending)
		ev := readFakeFrame(t, c)
		if got, present := frameMessageVoice(t, ev); present {
			t.Fatalf("未实现语音列读面的存储替身带上了 voice = %s，want 缺席", got)
		}
	})
}