Cyrene_v2/internal/services/gateway/broadcast_unread_test.go

package gateway

import (
	"context"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)

// 本文件钉住契约 v0.76 的**未读实时面**：`message_persisted` 帧的 data 携带该会话
// 此刻的**绝对未读数**（与 GET /api/v1/sessions 的 unreadCount 同一口径、同一条
// 读游标）。三条硬约束各有断言：值是对的；只在**确实算得出来且非零**时出现
// （缺席 = 老行为）；算不出来时绝不冒出一个 0 去冒充「已读完」。

// unreadOf 从帧 data 里读未读数并返回 (值, 是否在场)。它刻意先断言「在场性」：
// 本面的关键取舍就是「缺席」与「0」是两回事。
func unreadOf(t *testing.T, ev wsEvent) (int, bool) {
	t.Helper()
	data, ok := ev.Data.(map[string]any)
	if !ok {
		t.Fatalf("frame data = %+v, want a JSON object", ev.Data)
	}
	raw, present := data["unreadCount"]
	if !present {
		return 0, false
	}
	n, ok := raw.(float64)
	if !ok {
		t.Fatalf("frame data.unreadCount = %v (%T), want a number", raw, raw)
	}
	return int(n), true
}

// TestBroadcastPersistedCarriesUnreadCount 覆盖顺路径与两条边界：帧上的数字是
// **绝对**未读数（不是「自上次以来」），游标推进之后同一条广播路径不再带该键
// （0 不顶位），游标之后的新消息重新点亮并给出刷新后的绝对值。末段顺带钉住
// `channel_persisted` 与 `message_persisted` 走**同一个**载荷构造函数，因此隔离
// 渠道会话（没有 chat_sessions 行）同样不带该键 —— 它连读游标都没有。
func TestBroadcastPersistedCarriesUnreadCount(t *testing.T) {
	t.Parallel()
	srv, _, _, st := newHistoryEnv(t)
	c := newOwnerConn(srv.hub, 32)
	srv.hub.register(c)
	ctx := context.Background()

	appendAt := func(at int64, content string) history.Message {
		t.Helper()
		full, err := st.AppendFull(ctx, history.Message{
			SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: content, CreatedAt: at,
		})
		if err != nil {
			t.Fatalf("append %q: %v", content, err)
		}
		return full
	}

	// ① 历史一条（1000）+ 刚落库的一条（2000），游标仍是 0 ⇒ 两条都未读。
	if err := st.Append(ctx, history.Message{
		SessionID: history.MainSessionID, Role: history.RoleUser, Content: "old", CreatedAt: 1000,
	}); err != nil {
		t.Fatalf("seed old: %v", err)
	}
	srv.BroadcastPersisted(appendAt(2000, "new"))
	ev := readFakeFrame(t, c)
	if ev.Type != "message_persisted" {
		t.Fatalf("frame type = %q, want message_persisted", ev.Type)
	}
	if got, present := unreadOf(t, ev); !present || got != 2 {
		t.Fatalf("frame data.unreadCount = %d (present=%v), want 2 (绝对未读，不是本次 +1)", got, present)
	}

	// ② 游标推进到最新消息 ⇒ 该会话已读：同一条广播路径**不再**带该键。
	if _, err := st.MarkSessionRead(ctx, history.MainSessionID, time.UnixMilli(2000)); err != nil {
		t.Fatalf("mark session read: %v", err)
	}
	srv.BroadcastPersisted(appendAt(2000, "exactly at the cursor"))
	if got, present := unreadOf(t, readFakeFrame(t, c)); present {
		t.Fatalf("frame data.unreadCount = %d，want 该键缺席：未读为 0 时不顶位（缺席 = 老行为，客户端回落 GET /sessions 收敛）", got)
	}

	// ③ 游标之后的新消息 ⇒ 值是刷新后的绝对值。
	srv.BroadcastPersisted(appendAt(3000, "after the cursor"))
	if got, present := unreadOf(t, readFakeFrame(t, c)); !present || got != 1 {
		t.Fatalf("frame data.unreadCount = %d (present=%v), want 1", got, present)
	}

	// ④ 隔离渠道会话走同一个构造函数 ⇒ 口径不分叉；它没有游标，因此不带该键。
	if err := st.Append(ctx, history.Message{
		SessionID: "sess-g", Role: history.RoleUser, Content: "isolated", CreatedAt: 5000,
	}); err != nil {
		t.Fatalf("seed isolated: %v", err)
	}
	srv.BroadcastPersisted(history.Message{
		ID: "m-iso", SessionID: "sess-g", Role: history.RoleUser, Content: "isolated", CreatedAt: 5000,
	})
	iso := readFakeFrame(t, c)
	if iso.Type != "channel_persisted" {
		t.Fatalf("frame type = %q, want channel_persisted", iso.Type)
	}
	if got, present := unreadOf(t, iso); present {
		t.Fatalf("隔离会话的帧带上了 unreadCount = %d，want 该键缺席（没有会话行就没有读游标）", got)
	}
}

// TestBroadcastPersistedUnreadCountAbsentWhenUnknown 钉住三条**诚实缺席**路径：
// 未注入 HistoryStore、注入的存储没实现单会话未读视图（`sessionUnreadCounter`）、
// 以及隔离渠道会话本身。三条都不许冒出 `unreadCount:0` —— 把「算不出来」写成
// 「没有未读」就是编数据。
func TestBroadcastPersistedUnreadCountAbsentWhenUnknown(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, 32)
		srv.hub.register(c)

		srv.BroadcastPersisted(pending)
		if got, present := unreadOf(t, readFakeFrame(t, c)); present {
			t.Fatalf("未接线历史存储的帧带上了 unreadCount = %d，want 缺席", got)
		}
	})

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

		srv.BroadcastPersisted(pending)
		if got, present := unreadOf(t, readFakeFrame(t, c)); present {
			t.Fatalf("未实现 sessionUnreadCounter 的存储替身带上了 unreadCount = %d，want 缺席", got)
		}
	})

	t.Run("isolated session has no cursor", func(t *testing.T) {
		srv, _, _, st := newHistoryEnv(t)
		c := newOwnerConn(srv.hub, 32)
		srv.hub.register(c)
		if err := st.Append(context.Background(), history.Message{
			SessionID: "sess-g", Role: history.RoleUser, Content: "iso", CreatedAt: 1000,
		}); err != nil {
			t.Fatalf("seed isolated: %v", err)
		}

		srv.BroadcastPersisted(history.Message{
			ID: "m-iso", SessionID: "sess-g", Role: history.RoleUser, Content: "iso", CreatedAt: 1000,
		})
		ev := readFakeFrame(t, c)
		if ev.Type != "channel_persisted" {
			t.Fatalf("frame type = %q, want channel_persisted", ev.Type)
		}
		if got, present := unreadOf(t, ev); present {
			t.Fatalf("隔离会话的帧带上了 unreadCount = %d，want 缺席", got)
		}
	})
}