kra-new/internal/integration/live/server_test.go

package live

import (
	"context"
	"sync"
	"testing"

	"github.com/olahol/melody"
	"github.com/pion/webrtc/v4"

	bizlive "kra/internal/biz/live"
	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
	err  error
}

func (s *liveCatalogStub) GetRoom(context.Context, uint) (*bizlive.Room, error) {
	if s.err != nil {
		return nil, s.err
	}
	copy := *s.room
	return &copy, nil
}

func (s *liveCatalogStub) GetCategory(context.Context, uint) (*bizlive.Category, error) {
	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
}

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) 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{}{}
	return &bizlive.Stream{StreamID: streamID, SubscriberCount: len(s.users)}, 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),
		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))
	}
}

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}}, 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)
	}
}

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 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)