对接

internal/integration/live/server.go

@@ -322,6 +322,11 @@ func (m *Manager) handleChat(session *melody.Session, payload chatSendPayload) {
 		m.send(session, "Error", map[string]any{"code": "CHAT_UNAVAILABLE", "message": "chat is only available in live rooms"})
 		return
 	}
+	available, err := m.roomCatalog.IsLive(context.Background(), roomID)
+	if err != nil || !available {
+		m.send(session, "Error", map[string]any{"code": "ROOM_UNAVAILABLE", "message": "live room is unavailable"})
+		return
+	}
 	item, err := m.chat.Send(context.Background(), roomID, p.userID, payload.Content)
 	if err != nil {
 		m.send(session, "Error", map[string]any{"code": "INVALID_CHAT", "message": err.Error()})
@@ -379,13 +384,10 @@ func (m *Manager) authorizeRoomPublisher(streamID string, userID uint) error {
 	if err != nil || id == 0 {
 		return errors.New("invalid live room stream id")
 	}
-	item, err := m.roomCatalog.Get(context.Background(), uint(id))
+	item, err := m.roomCatalog.GetPublic(context.Background(), uint(id))
 	if err != nil {
 		return errors.New("live room is unavailable")
 	}
-	if !item.Enabled {
-		return errors.New("live room is disabled")
-	}
 	if item.AnchorUserID != userID {
 		return errors.New("current user is not the room anchor")
 	}

internal/integration/live/server_test.go

@@ -2,6 +2,7 @@ package live
 
 import (
 	"context"
+	"encoding/json"
 	"sync"
 	"testing"
 
@@ -9,6 +10,7 @@ import (
 	"github.com/pion/webrtc/v4"
 
 	bizlive "kra/internal/biz/live"
+	bizchat "kra/internal/biz/livechat"
 	platformws "kra/pkg/websocket"
 )
 
@@ -39,28 +41,41 @@ func (h *liveHubStub) Send(session *platformws.Session, message []byte) error {
 
 type liveCatalogStub struct {
 	bizlive.CatalogRepo
-	room *bizlive.Room
-	err  error
+	room        *bizlive.Room
+	category    *bizlive.Category
+	err         error
+	categoryErr error
 }
 
-func (s *liveCatalogStub) GetRoom(context.Context, uint) (*bizlive.Room, 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
+	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) {
@@ -86,6 +101,38 @@ func (s *liveStreamStub) List(context.Context) ([]*bizlive.Stream, error) {
 	}
 	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()
@@ -97,15 +144,49 @@ func (s *liveStreamStub) RegisterSubscriber(_ context.Context, streamID string,
 		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{}),
@@ -132,6 +213,11 @@ func TestAnonymousSubscribersReceiveDistinctRuntimeIdentities(t *testing.T) {
 	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) {
@@ -147,7 +233,7 @@ func TestSubscriberCannotCreateUnknownCatalogRoom(t *testing.T) {
 
 func TestSubscriberCannotCreateOfflineRuntimeState(t *testing.T) {
 	streams := &liveStreamStub{}
-	manager := newAnonymousTestManager(&liveCatalogStub{room: &bizlive.Room{ID: 7, CategoryID: 1, Enabled: true}}, streams)
+	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"})
@@ -155,6 +241,9 @@ func TestSubscriberCannotCreateOfflineRuntimeState(t *testing.T) {
 	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) {
@@ -168,6 +257,198 @@ func TestAnonymousPublisherIsRejected(t *testing.T) {
 	}
 }
 
+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 {