Cyrene_v2/internal/services/gateway/proactive_test.go

package gateway

import (
	"context"
	"testing"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)

// TestHub_SendProactive_OwnerAndAdmin 验证主动推送恰好抵达
// owner/admin 连接并返回投递计数；viewer 或未认证
// （nil claims）连接绝不收到它。
func TestHub_SendProactive_OwnerAndAdmin(t *testing.T) {
	t.Parallel()
	hb := newTestHub(nil, nil)
	owner := newFakeConn(hb, 8)
	owner.claims = &auth.Claims{Role: auth.RoleOwner}
	admin := newFakeConn(hb, 8)
	admin.claims = &auth.Claims{Role: auth.RoleAdmin}
	viewer := newFakeConn(hb, 8)
	viewer.claims = &auth.Claims{Role: auth.RoleViewer}
	bare := newFakeConn(hb, 8) // claims 为 nil：未认证
	for _, c := range []*wsConn{owner, admin, viewer, bare} {
		hb.register(c)
	}

	msg := adapter.UnifiedMessage{
		Platform:    "cyrene-backthink",
		ChannelType: adapter.ChannelDirect,
		Content:     "记得喝水哦",
		ContentType: adapter.ContentText,
		Sender:      adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
		ChannelID:   "kook:123",
		SenderID:    "uid-999",
		MessageID:   "m-42",
	}
	n, err := hb.SendProactive(context.Background(), msg)
	if err != nil {
		t.Fatalf("SendProactive err = %v", err)
	}
	if n.Connections != 2 {
		t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n.Connections)
	}

	for _, c := range []*wsConn{owner, admin} {
		ev := readFakeFrame(t, c)
		if ev.Type != "proactive_message" {
			t.Fatalf("frame type = %q, want proactive_message (frame=%+v)", ev.Type, ev)
		}
		data, ok := ev.Data.(map[string]any)
		if !ok {
			t.Fatalf("frame data type = %T, want map[string]any", ev.Data)
		}
		// 正文经共享规整器（turnreply.NormalizeParagraphs）：纯散文也会被包成
		// 一个 `<p>` —— 这条路由与 Web 回合 / IM 出站 / IM 侧主动推送**同源**，
		// 未包裹的裸文本同样会包上（M25 修的就是本路由曾绕过它）。
		if data["content"] != "<p>记得喝水哦</p>" {
			t.Fatalf("frame content = %q, want the normalized proactive text", data["content"])
		}
		if data["sender"] != "Cyrene" {
			t.Fatalf("frame sender = %q, want Cyrene", data["sender"])
		}
		if data["platform"] != "cyrene-backthink" {
			t.Fatalf("frame platform = %q, want cyrene-backthink", data["platform"])
		}
	}
	for _, c := range []*wsConn{viewer, bare} {
		if err := connHasNoFrame(c); err != nil {
			t.Fatalf("unexpected frame for a non-owner/admin conn: %v", err)
		}
	}
}

// TestHub_SendProactive_NoOnlineOwner 验证没有存活 owner/admin 连接时
// 计数为零（无错误），使调用方回退到 IM bridge。
func TestHub_SendProactive_NoOnlineOwner(t *testing.T) {
	t.Parallel()
	hb := newTestHub(nil, nil)
	viewer := newFakeConn(hb, 8)
	viewer.claims = &auth.Claims{Role: auth.RoleViewer}
	hb.register(viewer)

	n, err := hb.SendProactive(context.Background(), adapter.UnifiedMessage{Content: "hi"})
	if err != nil {
		t.Fatalf("SendProactive err = %v", err)
	}
	if n.Connections != 0 {
		t.Fatalf("count = %d, want 0", n.Connections)
	}
}

// TestHub_SendProactive_CancelledContext 验证被取消的 context 拒绝该
// 推送（fail closed）并返回被 wrap 的错误。
func TestHub_SendProactive_CancelledContext(t *testing.T) {
	t.Parallel()
	hb := newTestHub(nil, nil)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	if _, err := hb.SendProactive(ctx, adapter.UnifiedMessage{Content: "hi"}); err == nil {
		t.Fatalf("expected a context error, got nil")
	}
}

// TestServer_SendProactive_DelegatesToHub 验证 Server.SendProactive 镜像
// hub，使组合根能把网关自身交给汇聚点设置器。
func TestServer_SendProactive_DelegatesToHub(t *testing.T) {
	t.Parallel()
	env := newEnv(t, nil, false)
	c := newFakeConn(env.srv.hub, 8)
	c.claims = &auth.Claims{Role: auth.RoleOwner}
	env.srv.hub.register(c)

	n, err := env.srv.SendProactive(context.Background(), adapter.UnifiedMessage{Content: "hi"})
	if err != nil {
		t.Fatalf("SendProactive err = %v", err)
	}
	if n.Connections != 1 {
		t.Fatalf("count = %d, want 1", n.Connections)
	}
	ev := readFakeFrame(t, c)
	if ev.Type != "proactive_message" {
		t.Fatalf("frame type = %q, want proactive_message", ev.Type)
	}
}