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

package ws

// hub_video_race_test.go — video-inference-stuck 修法 A2/B 的 Hub 行為測試
//
// 涵蓋：
//  A2  WaitForRoomClient：room 有 client 時立即返回、無 client 時阻塞到 register、ctx 取消返回 false
//  B   late-join replay：inference room 緩存最近 N 筆、join 時補送、ClearRoomReplay 清除
//  隔離 非 inference room（如 flash:）不緩存 replay

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

// drainN 從 client.Send 收 n 筆訊息，逾時 fail。
func drainN(t *testing.T, c *Client, n int, timeout time.Duration) [][]byte {
	t.Helper()
	out := make([][]byte, 0, n)
	deadline := time.After(timeout)
	for len(out) < n {
		select {
		case msg := <-c.Send:
			out = append(out, msg)
		case <-deadline:
			t.Fatalf("只收到 %d/%d 筆訊息就逾時", len(out), n)
		}
	}
	return out
}

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

	makeRegisteredClient(hub, "inference:DEV1", 4)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if !hub.WaitForRoomClient(ctx, "inference:DEV1") {
		t.Fatal("room 已有 client，WaitForRoomClient 應立即回 true")
	}
}

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

	got := make(chan bool, 1)
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		got <- hub.WaitForRoomClient(ctx, "inference:DEV2")
	}()

	// 確保 waiter 已掛上（尚未有 client）
	select {
	case <-got:
		t.Fatal("尚無 client 時 WaitForRoomClient 不該返回")
	case <-time.After(100 * time.Millisecond):
	}

	// 現在 register 一個 client → 應喚醒 waiter
	makeRegisteredClient(hub, "inference:DEV2", 4)

	select {
	case ok := <-got:
		if !ok {
			t.Fatal("client join 後 WaitForRoomClient 應回 true")
		}
	case <-time.After(time.Second):
		t.Fatal("client join 後 WaitForRoomClient 未在時限內返回")
	}
}

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

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	if hub.WaitForRoomClient(ctx, "inference:NEVER") {
		t.Fatal("無 client 且 ctx 逾時，應回 false")
	}
}

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

	room := "inference:DEV3"
	// 在無 client 時廣播 3 筆（模擬「WS 連上前的早期結果」）
	for i := 0; i < 3; i++ {
		hub.BroadcastToRoom(room, map[string]int{"frame": i})
	}
	// 讓 broadcast 都被 Run() 處理完（進 replay ring）
	time.Sleep(50 * time.Millisecond)

	// 現在 client late-join → 應補送到那 3 筆
	c := makeRegisteredClient(hub, room, 16)
	msgs := drainN(t, c, 3, time.Second)

	for i, m := range msgs {
		var got map[string]int
		if err := json.Unmarshal(m, &got); err != nil {
			t.Fatalf("replay 第 %d 筆 bad json: %v", i, err)
		}
		if got["frame"] != i {
			t.Errorf("replay 順序錯：第 %d 筆 frame=%d，預期 %d", i, got["frame"], i)
		}
	}
}

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

	room := "inference:DEV4"
	total := replayBufferSize + 10
	for i := 0; i < total; i++ {
		hub.BroadcastToRoom(room, map[string]int{"frame": i})
	}
	time.Sleep(80 * time.Millisecond)

	c := makeRegisteredClient(hub, room, replayBufferSize+8)
	msgs := drainN(t, c, replayBufferSize, time.Second)

	// 應只保留最後 replayBufferSize 筆，第一筆 frame 應為 total-replayBufferSize
	var first map[string]int
	if err := json.Unmarshal(msgs[0], &first); err != nil {
		t.Fatalf("bad json: %v", err)
	}
	if first["frame"] != total-replayBufferSize {
		t.Errorf("ring 未正確截斷：首筆 frame=%d，預期 %d", first["frame"], total-replayBufferSize)
	}
	// 不應再有第 replayBufferSize+1 筆
	select {
	case extra := <-c.Send:
		t.Errorf("replay 超出 buffer 上限，仍收到多餘訊息: %s", extra)
	case <-time.After(150 * time.Millisecond):
	}
}

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

	room := "inference:DEV5"
	hub.BroadcastToRoom(room, map[string]int{"frame": 0})
	time.Sleep(50 * time.Millisecond)

	hub.ClearRoomReplay(room)
	time.Sleep(50 * time.Millisecond)

	// clear 之後 late-join 不該收到任何 replay
	c := makeRegisteredClient(hub, room, 4)
	select {
	case msg := <-c.Send:
		t.Errorf("ClearRoomReplay 後仍補送 replay: %s", msg)
	case <-time.After(150 * time.Millisecond):
	}
}

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

	room := "flash:DEV6" // 非 inference 前綴 → 不緩存
	hub.BroadcastToRoom(room, map[string]string{"type": "progress"})
	time.Sleep(50 * time.Millisecond)

	c := makeRegisteredClient(hub, room, 4)
	select {
	case msg := <-c.Send:
		t.Errorf("非 inference room 不應緩存 replay，卻補送: %s", msg)
	case <-time.After(150 * time.Millisecond):
	}
}

// TestHub_ReplayAndLiveBroadcast_Ordering：late-join client 先收 replay、再收後續 live 訊息。
func TestHub_ReplayAndLiveBroadcast_Ordering(t *testing.T) {
	hub := NewHub()
	go hub.Run()

	room := "inference:DEV7"
	hub.BroadcastToRoom(room, map[string]int{"frame": 0}) // 早期（進 replay）
	time.Sleep(50 * time.Millisecond)

	c := makeRegisteredClient(hub, room, 8)
	// join 後再來一筆 live
	hub.BroadcastToRoom(room, map[string]int{"frame": 1})

	msgs := drainN(t, c, 2, time.Second)
	for i, m := range msgs {
		var got map[string]int
		_ = json.Unmarshal(m, &got)
		if got["frame"] != i {
			t.Errorf("順序錯：第 %d 筆 frame=%d，預期 %d（replay 應在 live 之前）", i, got["frame"], i)
		}
	}
}