Files
kra-new/internal/integration/live/server_test.go
T
2026-09-23 09:03:17 +08:00

510 lines
18 KiB
Go

package live
import (
"context"
"encoding/json"
"strings"
"sync"
"testing"
"github.com/olahol/melody"
"github.com/pion/webrtc/v4"
bizlive "kra/internal/biz/live"
bizchat "kra/internal/biz/livechat"
platformws "kra/pkg/websocket"
)
type liveHubStub struct {
mu sync.Mutex
messages map[*melody.Session][][]byte
}
func (h *liveHubStub) Enabled() bool { return true }
func (h *liveHubStub) Broadcast([]byte) error { return nil }
func (h *liveHubStub) BroadcastBinary([]byte) error { return nil }
func (h *liveHubStub) Sessions() ([]*platformws.Session, error) { return nil, nil }
func (h *liveHubStub) Len() int { return 0 }
func (h *liveHubStub) SendBinary(*platformws.Session, []byte) error { return nil }
func (h *liveHubStub) OnMessage(func(*platformws.Session, []byte)) {}
func (h *liveHubStub) OnBinaryMessage(func(*platformws.Session, []byte)) {}
func (h *liveHubStub) OnConnect(func(*platformws.Session)) {}
func (h *liveHubStub) OnDisconnect(func(*platformws.Session)) {}
func (h *liveHubStub) Send(session *platformws.Session, message []byte) error {
h.mu.Lock()
defer h.mu.Unlock()
if h.messages == nil {
h.messages = make(map[*melody.Session][][]byte)
}
h.messages[session] = append(h.messages[session], append([]byte(nil), message...))
return nil
}
type liveCatalogStub struct {
bizlive.CatalogRepo
room *bizlive.Room
category *bizlive.Category
err error
categoryErr error
}
func (s *liveCatalogStub) GetRoom(_ context.Context, id uint) (*bizlive.Room, error) {
if s.err != nil {
return nil, s.err
}
if s.room == nil || s.room.ID != id {
return nil, bizlive.ErrRoomNotFound
}
copy := *s.room
return &copy, nil
}
func (s *liveCatalogStub) GetCategory(context.Context, uint) (*bizlive.Category, error) {
if s.categoryErr != nil {
return nil, s.categoryErr
}
if s.category != nil {
copy := *s.category
return &copy, nil
}
return &bizlive.Category{ID: s.room.CategoryID, Enabled: true}, nil
}
type liveStreamStub struct {
bizlive.StreamRepo
mu sync.Mutex
streams map[string]*bizlive.Stream
users map[uint]struct{}
registerCalls int
publisherRegisterCalls int
}
func (s *liveStreamStub) Get(_ context.Context, streamID string) (*bizlive.Stream, error) {
s.mu.Lock()
defer s.mu.Unlock()
stream := s.streams[streamID]
if stream == nil {
return &bizlive.Stream{StreamID: streamID}, nil
}
copy := *stream
return &copy, nil
}
func (s *liveStreamStub) List(context.Context) ([]*bizlive.Stream, error) {
s.mu.Lock()
defer s.mu.Unlock()
items := make([]*bizlive.Stream, 0, len(s.streams))
for _, stream := range s.streams {
if stream.PublisherOnline {
copy := *stream
items = append(items, &copy)
}
}
return items, nil
}
func (s *liveStreamStub) RegisterPublisher(_ context.Context, streamID string, userID uint) (*bizlive.Stream, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.streams == nil {
s.streams = make(map[string]*bizlive.Stream)
}
stream := s.streams[streamID]
if stream == nil {
stream = &bizlive.Stream{StreamID: streamID}
s.streams[streamID] = stream
}
if stream.PublisherOnline {
return nil, bizlive.ErrPublisherExists
}
s.publisherRegisterCalls++
stream.PublisherOnline, stream.PublisherID = true, userID
copy := *stream
return &copy, nil
}
func (s *liveStreamStub) UnregisterPublisher(_ context.Context, streamID string, userID uint) error {
s.mu.Lock()
defer s.mu.Unlock()
stream := s.streams[streamID]
if stream == nil || !stream.PublisherOnline || stream.PublisherID != userID {
return bizlive.ErrPublisherNotFound
}
stream.PublisherOnline, stream.PublisherID = false, 0
return nil
}
func (s *liveStreamStub) RegisterSubscriber(_ context.Context, streamID string, userID uint) (*bizlive.Stream, error) {
s.mu.Lock()
defer s.mu.Unlock()
s.registerCalls++
if s.users == nil {
s.users = make(map[uint]struct{})
}
if _, exists := s.users[userID]; exists {
return nil, bizlive.ErrSubscriberExists
}
s.users[userID] = struct{}{}
if stream := s.streams[streamID]; stream != nil {
stream.SubscriberCount = len(s.users)
}
return &bizlive.Stream{StreamID: streamID, SubscriberCount: len(s.users)}, nil
}
func (s *liveStreamStub) UnregisterSubscriber(_ context.Context, streamID string, userID uint) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, exists := s.users[userID]; !exists {
return bizlive.ErrSubscriberNotFound
}
delete(s.users, userID)
if stream := s.streams[streamID]; stream != nil {
stream.SubscriberCount = len(s.users)
}
return nil
}
type liveChatStub struct {
bizchat.Repo
cleared []uint
messages []*bizchat.Message
}
func (s *liveChatStub) Clear(_ context.Context, roomID uint) error {
s.cleared = append(s.cleared, roomID)
return nil
}
func (s *liveChatStub) CreateChat(_ context.Context, roomID, userID uint, content string) (*bizchat.Message, error) {
item := &bizchat.Message{ID: uint(len(s.messages) + 1), RoomID: roomID, UserID: userID, Content: content, Type: bizchat.TypeChat}
s.messages = append(s.messages, item)
return item, nil
}
func newAnonymousTestManager(catalog bizlive.CatalogRepo, streams bizlive.StreamRepo) *Manager {
streamUC := bizlive.NewStreamUsecase(streams)
return &Manager{
hub: &liveHubStub{},
streams: streamUC,
roomCatalog: bizlive.NewRoomUsecase(catalog, streamUC),
chat: bizchat.NewUsecase(&liveChatStub{}),
rooms: make(map[string]*room),
peers: make(map[*melody.Session]*peer),
observers: make(map[*melody.Session]struct{}),
stop: make(chan struct{}),
}
}
func TestAnonymousSubscribersReceiveDistinctRuntimeIdentities(t *testing.T) {
streams := &liveStreamStub{streams: map[string]*bizlive.Stream{"room-7": {StreamID: "room-7", PublisherOnline: true}}}
manager := newAnonymousTestManager(&liveCatalogStub{room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: true}}, streams)
first := &melody.Session{Keys: map[string]any{}}
second := &melody.Session{Keys: map[string]any{}}
manager.join(first, joinPayload{Role: "subscriber", StreamID: "room-7"})
manager.join(second, joinPayload{Role: "subscriber", StreamID: "room-7"})
p1, p2 := manager.peer(first), manager.peer(second)
if p1 == nil || p2 == nil {
t.Fatalf("anonymous subscribers did not join: first=%+v second=%+v", p1, p2)
}
if p1.authenticated || p2.authenticated || p1.userID == 0 || p2.userID == 0 || p1.userID == p2.userID {
t.Fatalf("anonymous identities are not unique: first=%d second=%d", p1.userID, p2.userID)
}
if len(streams.users) != 2 {
t.Fatalf("subscriber count = %d, want 2", len(streams.users))
}
for _, session := range []*melody.Session{first, second} {
if payload := lastMessagePayload(t, manager, session, "Joined"); payload["userId"] != float64(0) {
t.Fatalf("anonymous runtime identity leaked in response: %+v", payload)
}
}
}
func TestSubscriberCannotCreateUnknownCatalogRoom(t *testing.T) {
manager := newAnonymousTestManager(&liveCatalogStub{err: bizlive.ErrRoomNotFound}, &liveStreamStub{})
session := &melody.Session{Keys: map[string]any{}}
manager.join(session, joinPayload{Role: "subscriber", StreamID: "room-999999"})
if manager.peer(session) != nil || len(manager.rooms) != 0 {
t.Fatalf("unknown room created runtime state: peers=%d rooms=%d", len(manager.peers), len(manager.rooms))
}
}
func TestSubscriberCannotCreateOfflineRuntimeState(t *testing.T) {
streams := &liveStreamStub{}
manager := newAnonymousTestManager(&liveCatalogStub{room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: true, ReplayURL: "https://example.test/replay.mp4"}}, streams)
session := &melody.Session{Keys: map[string]any{}}
manager.join(session, joinPayload{Role: "subscriber", StreamID: "room-7"})
if manager.peer(session) != nil || len(manager.rooms) != 0 || len(streams.users) != 0 || streams.registerCalls != 0 {
t.Fatalf("offline room created runtime state: peers=%d rooms=%d users=%d registrations=%d", len(manager.peers), len(manager.rooms), len(streams.users), streams.registerCalls)
}
if payload := lastMessagePayload(t, manager, session, "JoinRejected"); payload["code"] != "ROOM_UNAVAILABLE" {
t.Fatalf("offline join rejection=%+v", payload)
}
}
func TestAnonymousPublisherIsRejected(t *testing.T) {
manager := newAnonymousTestManager(&liveCatalogStub{room: &bizlive.Room{ID: 7, Enabled: true}}, &liveStreamStub{})
session := &melody.Session{Keys: map[string]any{}}
manager.join(session, joinPayload{Role: "publisher", StreamID: "room-7"})
if manager.peer(session) != nil {
t.Fatal("anonymous publisher joined")
}
}
func TestBoundAnchorCanPublishOfflineRoomAndDisconnectRestoresReplay(t *testing.T) {
catalog := &liveCatalogStub{room: &bizlive.Room{ID: 7, CategoryID: 1, AnchorUserID: 42, Enabled: true, ReplayURL: "https://example.test/replay.mp4"}}
streams := &liveStreamStub{}
manager := newAnonymousTestManager(catalog, streams)
chat := &liveChatStub{}
manager.chat = bizchat.NewUsecase(chat)
publisher := &melody.Session{Keys: map[string]any{"user_id": uint(42)}}
viewer := &melody.Session{Keys: map[string]any{"user_id": uint(9)}}
before, err := manager.roomCatalog.GetPublic(context.Background(), 7)
if err != nil || before.Status != bizlive.RoomStatusOffline {
t.Fatalf("offline public room=%+v error=%v", before, err)
}
manager.join(publisher, joinPayload{Role: "publisher", StreamID: "room-7"})
peer := manager.peer(publisher)
if peer == nil || peer.role != "publisher" || !peer.authenticated || peer.userID != 42 {
t.Fatalf("bound anchor did not join offline room: peer=%+v", peer)
}
if payload := lastMessagePayload(t, manager, publisher, "Joined"); payload["role"] != "publisher" || payload["streamId"] != "room-7" {
t.Fatalf("publisher join response=%+v", payload)
}
live, err := manager.roomCatalog.GetPublic(context.Background(), 7)
if err != nil || live.Status != bizlive.RoomStatusLive || streams.publisherRegisterCalls != 1 {
t.Fatalf("published room=%+v registrations=%d error=%v", live, streams.publisherRegisterCalls, err)
}
manager.join(viewer, joinPayload{Role: "subscriber", StreamID: "room-7"})
if manager.peer(viewer) == nil {
t.Fatal("viewer could not join published room")
}
if payload := lastMessagePayload(t, manager, viewer, "Joined"); payload["userId"] != float64(9) {
t.Fatalf("authenticated viewer response=%+v", payload)
}
manager.handleChat(viewer, chatSendPayload{Content: "Hello"})
if len(chat.messages) != 1 || chat.messages[0].RoomID != 7 || chat.messages[0].UserID != 9 {
t.Fatalf("live chat messages=%+v", chat.messages)
}
manager.onDisconnect(publisher)
after, err := manager.roomCatalog.GetPublic(context.Background(), 7)
if err != nil || after.Status != bizlive.RoomStatusOffline || after.ReplayURL != before.ReplayURL {
t.Fatalf("room after disconnect=%+v error=%v", after, err)
}
if payload := lastMessagePayload(t, manager, viewer, "PublisherStatus"); payload["ready"] != false || payload["reason"] != "disconnected" {
t.Fatalf("offline notification=%+v", payload)
}
if available, err := manager.roomCatalog.IsLive(context.Background(), 7); err != nil || available {
t.Fatalf("room remained live after disconnect: live=%v error=%v", available, err)
}
manager.handleChat(viewer, chatSendPayload{Content: "Not live"})
if len(chat.messages) != 1 {
t.Fatalf("offline message was persisted: %+v", chat.messages)
}
if payload := lastMessagePayload(t, manager, viewer, "Error"); payload["code"] != "ROOM_UNAVAILABLE" {
t.Fatalf("offline chat rejection=%+v", payload)
}
manager.onDisconnect(viewer)
if len(manager.peers) != 0 || len(manager.rooms) != 0 || len(streams.users) != 0 {
t.Fatalf("disconnected peers remain: peers=%d rooms=%d users=%d", len(manager.peers), len(manager.rooms), len(streams.users))
}
if len(chat.cleared) != 2 || chat.cleared[0] != 7 || chat.cleared[1] != 7 {
t.Fatalf("chat reset rooms=%v", chat.cleared)
}
}
func TestPublisherJoinRequiresEnabledRoomCategoryAndBoundAnchor(t *testing.T) {
for _, test := range []struct {
name string
userID uint
enabled bool
category *bizlive.Category
roomErr error
categoryErr error
code string
}{
{name: "anonymous", enabled: true, code: "AUTH_REQUIRED"},
{name: "wrong anchor", userID: 9, enabled: true, code: "PUBLISH_FORBIDDEN"},
{name: "disabled room", userID: 42, code: "PUBLISH_FORBIDDEN"},
{name: "disabled category", userID: 42, enabled: true, category: &bizlive.Category{ID: 1, Enabled: false}, code: "PUBLISH_FORBIDDEN"},
{name: "missing room", userID: 42, enabled: true, roomErr: bizlive.ErrRoomNotFound, code: "PUBLISH_FORBIDDEN"},
{name: "missing category", userID: 42, enabled: true, categoryErr: bizlive.ErrCategoryNotFound, code: "PUBLISH_FORBIDDEN"},
} {
t.Run(test.name, func(t *testing.T) {
streams := &liveStreamStub{}
manager := newAnonymousTestManager(&liveCatalogStub{
room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: test.enabled, AnchorUserID: 42},
category: test.category,
err: test.roomErr,
categoryErr: test.categoryErr,
}, streams)
session := &melody.Session{Keys: map[string]any{"user_id": test.userID}}
manager.join(session, joinPayload{Role: "publisher", StreamID: "room-7"})
if manager.peer(session) != nil || len(manager.rooms) != 0 || streams.publisherRegisterCalls != 0 {
t.Fatalf("forbidden publisher created state: peers=%d rooms=%d registrations=%d", len(manager.peers), len(manager.rooms), streams.publisherRegisterCalls)
}
if payload := lastMessagePayload(t, manager, session, "JoinRejected"); payload["code"] != test.code {
t.Fatalf("publisher rejection=%+v, want code=%s", payload, test.code)
}
})
}
}
func TestSubscriberJoinCannotBypassDisabledRoomOrCategory(t *testing.T) {
for _, test := range []struct {
name string
enabled bool
category *bizlive.Category
}{
{name: "disabled room"},
{name: "disabled category", enabled: true, category: &bizlive.Category{ID: 1, Enabled: false}},
} {
t.Run(test.name, func(t *testing.T) {
streams := &liveStreamStub{streams: map[string]*bizlive.Stream{"room-7": {StreamID: "room-7", PublisherOnline: true}}}
manager := newAnonymousTestManager(&liveCatalogStub{
room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: test.enabled},
category: test.category,
}, streams)
session := &melody.Session{Keys: map[string]any{"user_id": uint(9)}}
manager.join(session, joinPayload{Role: "subscriber", StreamID: "room-7"})
if manager.peer(session) != nil || len(manager.rooms) != 0 || streams.registerCalls != 0 {
t.Fatalf("unavailable room created subscriber state: peers=%d rooms=%d registrations=%d", len(manager.peers), len(manager.rooms), streams.registerCalls)
}
if payload := lastMessagePayload(t, manager, session, "JoinRejected"); payload["code"] != "ROOM_UNAVAILABLE" {
t.Fatalf("subscriber rejection=%+v", payload)
}
})
}
}
func TestChatRechecksRoomAndCategoryVisibilityAfterJoin(t *testing.T) {
for _, disableCategory := range []bool{false, true} {
name := "disabled room"
if disableCategory {
name = "disabled category"
}
t.Run(name, func(t *testing.T) {
catalog := &liveCatalogStub{
room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: true},
category: &bizlive.Category{ID: 1, Enabled: true},
}
streams := &liveStreamStub{streams: map[string]*bizlive.Stream{"room-7": {StreamID: "room-7", PublisherOnline: true}}}
manager := newAnonymousTestManager(catalog, streams)
chat := &liveChatStub{}
manager.chat = bizchat.NewUsecase(chat)
session := &melody.Session{Keys: map[string]any{"user_id": uint(9)}}
manager.join(session, joinPayload{Role: "subscriber", StreamID: "room-7"})
if manager.peer(session) == nil {
t.Fatal("viewer did not join enabled room")
}
if disableCategory {
catalog.category.Enabled = false
} else {
catalog.room.Enabled = false
}
manager.handleChat(session, chatSendPayload{Content: "Hello"})
if len(chat.messages) != 0 {
t.Fatalf("unavailable chat was persisted: %+v", chat.messages)
}
if payload := lastMessagePayload(t, manager, session, "Error"); payload["code"] != "ROOM_UNAVAILABLE" {
t.Fatalf("unavailable chat rejection=%+v", payload)
}
})
}
}
func lastMessagePayload(t *testing.T, manager *Manager, session *melody.Session, msgType string) map[string]any {
t.Helper()
hub := manager.hub.(*liveHubStub)
hub.mu.Lock()
defer hub.mu.Unlock()
messages := hub.messages[session]
for index := len(messages) - 1; index >= 0; index-- {
var msg struct {
Type string `json:"type"`
Payload map[string]any `json:"payload"`
}
if err := json.Unmarshal(messages[index], &msg); err != nil {
t.Fatal(err)
}
if msg.Type == msgType {
return msg.Payload
}
}
t.Fatalf("missing %s message in %q", msgType, messages)
return nil
}
func TestClosedSupersededPeerConnectionDoesNotRemovePeer(t *testing.T) {
oldPC, err := webrtc.NewPeerConnection(webrtc.Configuration{})
if err != nil {
t.Fatal(err)
}
defer oldPC.Close()
newPC, err := webrtc.NewPeerConnection(webrtc.Configuration{})
if err != nil {
t.Fatal(err)
}
defer newPC.Close()
p := &peer{pc: oldPC}
replaced, installed := p.replacePeerConnection(newPC)
if !installed || replaced != oldPC {
t.Fatalf("replaced peer connection = %p, want %p", replaced, oldPC)
}
manager := &Manager{}
manager.handleICEConnectionStateChange(p, oldPC, webrtc.ICEConnectionStateClosed)
cleanupAvailable := false
p.cleanup.Do(func() { cleanupAvailable = true })
if !cleanupAvailable {
t.Fatal("superseded peer connection consumed peer cleanup")
}
if !p.hasPeerConnection(newPC) {
t.Fatal("replacement peer connection is no longer active")
}
}
var _ platformws.Hub = (*liveHubStub)(nil)
func TestNewPionAPIRejectsConflictingUDPConfiguration(t *testing.T) {
t.Setenv("KRA_LIVE_UDP_PORT", "40000")
t.Setenv("KRA_LIVE_UDP_MIN_PORT", "40000")
t.Setenv("KRA_LIVE_UDP_MAX_PORT", "40010")
_, closer, err := newPionAPI()
if closer != nil {
_ = closer.Close()
}
if err == nil || !strings.Contains(err.Error(), "cannot be combined") {
t.Fatalf("newPionAPI error = %v, want conflicting UDP configuration", err)
}
}
func TestNewPionAPIRejectsInvalidPublicIP(t *testing.T) {
t.Setenv("KRA_LIVE_PUBLIC_IP", "live.example.com")
_, closer, err := newPionAPI()
if closer != nil {
_ = closer.Close()
}
if err == nil || !strings.Contains(err.Error(), "invalid KRA_LIVE_PUBLIC_IP") {
t.Fatalf("newPionAPI error = %v, want invalid public IP", err)
}
}