visionA/local-agent/server/internal/api/ws/hub_broadcast_test.go

package ws

// hub_broadcast_test.go — MAJ-4 補丁：驗證 Hub.BroadcastToRoom 在 system room 的行為
//
// 涵蓋：
//  1. 多個 client 都收到同一則訊息
//  2. 空 room（無 client）不 panic、不 block
//  3. client send channel 滿時不 block hub（會 drop 該 client）

import (
	"encoding/json"
	"testing"
	"time"
)

// makeRegisteredClient 註冊一個 buffered send channel 的 dummy client 到指定 room。
func makeRegisteredClient(h *Hub, room string, bufSize int) *Client {
	c := &Client{Send: make(chan []byte, bufSize)}
	h.RegisterSync(&Subscription{Client: c, Room: room})
	return c
}

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

	c1 := makeRegisteredClient(hub, "system", 4)
	c2 := makeRegisteredClient(hub, "system", 4)
	c3 := makeRegisteredClient(hub, "system", 4)

	payload := map[string]interface{}{
		"type":   "server:shutdown-imminent",
		"reason": "quit",
		"ts":     int64(1234567890),
	}
	hub.BroadcastToRoom("system", payload)

	for i, c := range []*Client{c1, c2, c3} {
		select {
		case msg := <-c.Send:
			var got map[string]interface{}
			if err := json.Unmarshal(msg, &got); err != nil {
				t.Fatalf("client %d bad json: %v", i, err)
			}
			if got["type"] != "server:shutdown-imminent" || got["reason"] != "quit" {
				t.Errorf("client %d wrong payload: %+v", i, got)
			}
		case <-time.After(500 * time.Millisecond):
			t.Fatalf("client %d did not receive broadcast within 500ms", i)
		}
	}
}

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

	// 無 client 註冊在 "system" room → BroadcastToRoom 應該 no-op 且不 panic
	done := make(chan struct{})
	go func() {
		defer close(done)
		hub.BroadcastToRoom("system", map[string]string{"type": "server:shutdown-imminent"})
	}()

	select {
	case <-done:
		// OK
	case <-time.After(500 * time.Millisecond):
		t.Fatalf("BroadcastToRoom with empty room blocked > 500ms")
	}
}

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

	// buffer = 1 的 slow client，先塞滿 → hub 接著 broadcast 時會打到 default case
	// （select 非 blocking send），hub goroutine 必須仍能繼續處理後續訊息。
	//
	// 同時註冊一個 healthy client 觀察：hub 沒被卡住的證據 = healthy client 仍能收到訊息。
	slow := makeRegisteredClient(hub, "system", 1)
	slow.Send <- []byte("pre-existing") // 塞滿 slow 的 buffer

	healthy := makeRegisteredClient(hub, "system", 4)

	// 先量 broadcast 的時間，若 hub 被 slow client 卡住，這行會 block 直到 test timeout
	done := make(chan struct{})
	go func() {
		defer close(done)
		hub.BroadcastToRoom("system", map[string]string{"type": "server:shutdown-imminent"})
	}()
	select {
	case <-done:
		// OK — broadcast 沒 block
	case <-time.After(500 * time.Millisecond):
		t.Fatalf("BroadcastToRoom blocked — slow client 未被 drop")
	}

	// healthy client 必須收到訊息（證明 hub goroutine 沒被 slow client 卡住）
	select {
	case msg := <-healthy.Send:
		var got map[string]interface{}
		if err := json.Unmarshal(msg, &got); err != nil {
			t.Fatalf("healthy client bad json: %v", err)
		}
		if got["type"] != "server:shutdown-imminent" {
			t.Errorf("healthy client wrong payload: %+v", got)
		}
	case <-time.After(500 * time.Millisecond):
		t.Fatalf("healthy client did not receive broadcast — hub 被 slow client 卡住")
	}

	// 再次 broadcast 驗證 hub 持續工作（slow 已被 drop，broadcast 仍該回來）
	done2 := make(chan struct{})
	go func() {
		defer close(done2)
		hub.BroadcastToRoom("system", map[string]string{"type": "server:shutdown-imminent", "n": "2"})
	}()
	select {
	case <-done2:
		// OK
	case <-time.After(500 * time.Millisecond):
		t.Fatalf("second BroadcastToRoom blocked")
	}

	// healthy 應收到第二則
	select {
	case <-healthy.Send:
	case <-time.After(500 * time.Millisecond):
		t.Fatalf("healthy client 未收到第二則")
	}
}