kefu_cloud/server/internal/ws/bus_test.go

package ws

import (
	"errors"
	"sync"
	"testing"
	"time"
)

type fakeBus struct {
	mu        sync.Mutex
	published []*busEnvelope
	fail      bool
}

func (f *fakeBus) Publish(env *busEnvelope) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.fail {
		return errors.New("redis down")
	}
	f.published = append(f.published, env)
	return nil
}

func (f *fakeBus) Subscribe(dispatch func(env *busEnvelope)) {}

func (f *fakeBus) Close() error { return nil }

func (f *fakeBus) count() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.published)
}

func TestBroadcastPublishesToBusInsteadOfLocalDelivery(t *testing.T) {
	hub := NewHub()
	go hub.Run()
	bus := &fakeBus{}
	hub.mu.Lock()
	hub.bus = bus
	hub.mu.Unlock()

	sessionID := uint(12)
	agentID := uint(7)
	assigned := &Client{TenantID: 1, UserID: agentID, Role: "agent", Kind: "agent", Send: make(chan []byte, 1)}
	visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &sessionID, Send: make(chan []byte, 1)}
	for _, client := range []*Client{assigned, visitor} {
		hub.register <- client
	}

	hub.BroadcastToSession(1, sessionID, &agentID, []byte(`{"type":"message"}`))
	if bus.count() != 1 {
		t.Fatalf("应向总线发布 1 条信封，got=%d", bus.count())
	}
	// 发布模式下本进程不直接投递（由订阅回调统一投递，避免双发）
	select {
	case <-assigned.Send:
		t.Fatal("发布模式不应直接投递本地客户端")
	case <-time.After(100 * time.Millisecond):
	}

	// 订阅回调 dispatch 后应在本进程按范围投递
	hub.dispatch(&busEnvelope{Scope: busScopeSession, TenantID: 1, SessionID: sessionID, AgentID: &agentID, Message: []byte(`{"type":"message"}`)})
	for _, client := range []*Client{assigned, visitor} {
		select {
		case <-client.Send:
		case <-time.After(time.Second):
			t.Fatalf("dispatch 后客户端未收到消息：%+v", client)
		}
	}
}

func TestDispatchRoutesAllScopes(t *testing.T) {
	hub := NewHub()
	go hub.Run()

	sessionID := uint(41)
	agentID := uint(3)
	assigned := &Client{TenantID: 1, UserID: agentID, Role: "agent", Kind: "agent", Send: make(chan []byte, 1)}
	supervisor := &Client{TenantID: 1, UserID: 4, Role: "supervisor", Kind: "agent", Send: make(chan []byte, 1)}
	visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &sessionID, Send: make(chan []byte, 1)}
	other := &Client{TenantID: 2, Kind: "agent", UserID: 9, Role: "agent", Send: make(chan []byte, 1)}
	for _, client := range []*Client{assigned, supervisor, visitor, other} {
		hub.register <- client
	}

	// tenant_staff
	hub.dispatch(&busEnvelope{Scope: busScopeTenantStaff, TenantID: 1, Message: []byte(`{"type":"session_updated","session_id":41}`)})
	for _, client := range []*Client{assigned, supervisor} {
		select {
		case <-client.Send:
		case <-time.After(time.Second):
			t.Fatalf("租户工作人员应收到 session_updated：%+v", client)
		}
	}
	select {
	case <-visitor.Send:
		t.Fatal("访客不应收到 tenant_staff 广播")
	case <-other.Send:
		t.Fatal("跨租户客户端不应收到广播")
	case <-time.After(100 * time.Millisecond):
	}

	// visitor
	hub.dispatch(&busEnvelope{Scope: busScopeVisitor, TenantID: 1, SessionID: sessionID, Message: []byte(`{"type":"typing"}`)})
	select {
	case <-visitor.Send:
	case <-time.After(time.Second):
		t.Fatal("会话访客应收到消息")
	}

	// disconnect：跨节点踢下线（踢 supervisor，避免 assigned/visitor 被断开）
	hub.dispatch(&busEnvelope{Scope: busScopeDisconnect, UserID: 4, Text: "账号已停用"})
	deadline := time.After(time.Second)
	for {
		select {
		case <-supervisor.Send:
			// 先收到 kicked 通知，继续等连接关闭
		case <-deadline:
			t.Fatal("被踢客户端未收到任何消息")
		default:
		}
		hub.mu.RLock()
		_, stillOnline := hub.clients[supervisor]
		hub.mu.RUnlock()
		if !stillOnline {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func TestPublishFailureFallsBackToLocalDelivery(t *testing.T) {
	hub := NewHub()
	go hub.Run()
	bus := &fakeBus{fail: true}
	hub.mu.Lock()
	hub.bus = bus
	hub.mu.Unlock()

	visitorSession := uint(51)
	visitor := &Client{TenantID: 1, Kind: "visitor", SessionID: &visitorSession, Send: make(chan []byte, 1)}
	hub.register <- visitor

	hub.BroadcastToVisitor(1, visitorSession, []byte(`{"type":"typing"}`))
	select {
	case <-visitor.Send:
	case <-time.After(time.Second):
		t.Fatal("总线不可用时应兜底本进程投递")
	}
}