From b802532af36780dc8bfa062837aea0489029eb83 Mon Sep 17 00:00:00 2001 From: jevb Date: Thu, 19 Mar 2026 16:33:47 +0100 Subject: [PATCH] feat: extend SFU to forward video tracks and enforce MaxVideo limit - setupOnTrack now handles both audio and video tracks via kind branch instead of returning early on non-audio tracks - RTP forwarding works identically for both kinds; speaker detection only runs for audio tracks - handleVoiceCamera enforces MaxVideo limit before DB update, rejecting with VIDEO_LIMIT error when the cap is reached - Added TestHandleVoiceCamera_MaxVideoEnforced test --- Server/ws/voice_handlers.go | 55 +++++++++++++++-- Server/ws/voice_handlers_test.go | 100 +++++++++++++++++++++++++++++++ 2 files changed, 149 insertions(+), 6 deletions(-) diff --git a/Server/ws/voice_handlers.go b/Server/ws/voice_handlers.go index 39339722..f458b039 100644 --- a/Server/ws/voice_handlers.go +++ b/Server/ws/voice_handlers.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "strings" "time" "github.com/pion/webrtc/v4" @@ -445,6 +446,28 @@ func (h *Hub) handleVoiceCamera(c *Client, payload json.RawMessage) { return } + // Enforce MaxVideo limit when enabling camera. + if p.Enabled { + room := h.GetVoiceRoom(voiceChID) + if room != nil { + cfg := room.Config() + if cfg.MaxVideo > 0 { + allTracks := room.GetTracks() + videoCount := 0 + for _, vt := range allTracks { + if vt.Local != nil && strings.HasPrefix(vt.Local.ID(), "video-") { + videoCount++ + } + } + if videoCount >= cfg.MaxVideo { + c.sendMsg(buildErrorMsg("VIDEO_LIMIT", + fmt.Sprintf("maximum %d video streams reached", cfg.MaxVideo))) + return + } + } + } + } + if err := h.db.UpdateVoiceCamera(c.userID, p.Enabled); err != nil { slog.Error("ws handleVoiceCamera UpdateVoiceCamera", "err", err, "user_id", c.userID) c.sendMsg(buildErrorMsg("INTERNAL", "failed to update camera state")) @@ -683,25 +706,33 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { } pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { - if track.Kind() != webrtc.RTPCodecTypeAudio { + // Determine kind from the remote track. + var kind string + switch track.Kind() { + case webrtc.RTPCodecTypeAudio: + kind = "audio" + case webrtc.RTPCodecTypeVideo: + kind = "video" + default: return } slog.Info("SFU OnTrack", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "codec", track.Codec().MimeType, ) // Create local track for fan-out using the remote track's codec. local, err := webrtc.NewTrackLocalStaticRTP( track.Codec().RTPCodecCapability, - fmt.Sprintf("audio-%d", c.userID), + fmt.Sprintf("%s-%d", kind, c.userID), fmt.Sprintf("user-%d", c.userID), ) if err != nil { slog.Error("setupOnTrack NewTrackLocalStaticRTP", - "err", err, "user_id", c.userID) + "err", err, "user_id", c.userID, "kind", kind) return } @@ -711,8 +742,8 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { } // Store track on room. - room.SetTrack(c.userID, "audio", track, local) - vt := room.GetTrack(c.userID, "audio") + room.SetTrack(c.userID, kind, track, local) + vt := room.GetTrack(c.userID, kind) // Collect other participant IDs (lock ordering: VoiceRoom.mu released before voiceMu). participantIDs := room.ParticipantIDs() @@ -749,6 +780,7 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { slog.Info("SFU track fan-out", "from_user", c.userID, "channel_id", channelID, + "kind", kind, "participants", len(participantIDs), "tracks_added", addedCount) @@ -790,7 +822,8 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { noPacketTimer := time.AfterFunc(5*time.Second, func() { slog.Warn("RTP: no packets received after 5s", "user_id", c.userID, - "channel_id", channelID) + "channel_id", channelID, + "kind", kind) }) defer noPacketTimer.Stop() @@ -800,6 +833,7 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { case <-done: slog.Info("RTP goroutine exiting via done signal", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "packets_forwarded", pktCount) return default: @@ -810,6 +844,7 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { slog.Info("RTP read ended", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "packets_forwarded", pktCount, "err", readErr.Error()) return @@ -820,6 +855,7 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { slog.Info("RTP write ended", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "packets_forwarded", pktCount, "err", writeErr.Error()) return @@ -830,14 +866,21 @@ func (h *Hub) setupOnTrack(c *Client, channelID int64) { slog.Info("RTP first packet received", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "bytes", n) } else if pktCount%1000 == 0 { slog.Info("RTP forwarding", "user_id", c.userID, "channel_id", channelID, + "kind", kind, "packets", pktCount) } + // Speaker detection only applies to audio tracks. + if kind != "audio" { + continue + } + // Extract audio level directly from raw RTP bytes (avoids full Unmarshal). level, ok := extractAudioLevel(buf, n) if !ok { diff --git a/Server/ws/voice_handlers_test.go b/Server/ws/voice_handlers_test.go index 1e99b2e5..074a2b2c 100644 --- a/Server/ws/voice_handlers_test.go +++ b/Server/ws/voice_handlers_test.go @@ -2,10 +2,13 @@ package ws_test import ( "encoding/json" + "fmt" "testing" "testing/fstest" "time" + "github.com/pion/webrtc/v4" + "github.com/owncord/server/auth" "github.com/owncord/server/db" "github.com/owncord/server/ws" @@ -1627,3 +1630,100 @@ func TestVoice_Join_SameChannel_IsIdempotent(t *testing.T) { } } +// ─── MaxVideo enforcement ───────────────────────────────────────────────────── + +// makeVideoTrack creates a TrackLocalStaticRTP with ID "video-{userID}" for testing. +func makeVideoTrack(t *testing.T, userID int64) *webrtc.TrackLocalStaticRTP { + t.Helper() + local, err := webrtc.NewTrackLocalStaticRTP( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, + fmt.Sprintf("video-%d", userID), + fmt.Sprintf("user-%d", userID), + ) + if err != nil { + t.Fatalf("NewTrackLocalStaticRTP: %v", err) + } + return local +} + +// TestHandleVoiceCamera_MaxVideoEnforced verifies that when the MaxVideo limit +// is reached, a voice_camera enable request is rejected with VIDEO_LIMIT error. +func TestHandleVoiceCamera_MaxVideoEnforced(t *testing.T) { + hub, database := newVoiceHub(t) + chanID := seedVoiceChan(t, database, "vc-maxvideo") + + // Pre-create the room with MaxVideo=2 so handleVoiceJoin reuses it. + hub.GetOrCreateVoiceRoom(chanID, ws.VoiceRoomConfig{ + ChannelID: chanID, + MaxUsers: 10, + Quality: "medium", + MixingThreshold: 10, + TopSpeakers: 3, + MaxVideo: 2, + }) + + // Create 3 users: user1 and user2 will have video tracks, user3 will be rejected. + user1 := seedVoiceOwner(t, database, "maxvid-user1") + user2 := seedVoiceOwner(t, database, "maxvid-user2") + user3 := seedVoiceOwner(t, database, "maxvid-user3") + + send1 := make(chan []byte, 32) + c1 := ws.NewTestClientWithUser(hub, user1, chanID, send1) + hub.Register(c1) + + send2 := make(chan []byte, 32) + c2 := ws.NewTestClientWithUser(hub, user2, chanID, send2) + hub.Register(c2) + + send3 := make(chan []byte, 64) + c3 := ws.NewTestClientWithUser(hub, user3, chanID, send3) + hub.Register(c3) + time.Sleep(20 * time.Millisecond) + + // All three join the voice channel. + hub.HandleMessageForTest(c1, voiceJoinMsg(chanID)) + time.Sleep(30 * time.Millisecond) + hub.HandleMessageForTest(c2, voiceJoinMsg(chanID)) + time.Sleep(30 * time.Millisecond) + hub.HandleMessageForTest(c3, voiceJoinMsg(chanID)) + time.Sleep(30 * time.Millisecond) + + // Simulate user1 and user2 having video tracks by setting them on the room. + room := hub.GetVoiceRoom(chanID) + if room == nil { + t.Fatal("VoiceRoom should exist after joins") + } + room.SetTrack(user1.ID, "video", nil, makeVideoTrack(t, user1.ID)) + room.SetTrack(user2.ID, "video", nil, makeVideoTrack(t, user2.ID)) + + // Drain all messages from prior operations. + drainChan(send1) + drainChan(send2) + drainChan(send3) + + // User3 tries to enable camera — should be rejected with VIDEO_LIMIT. + hub.HandleMessageForTest(c3, voiceCameraMsg(true)) + time.Sleep(50 * time.Millisecond) + + msgs := drainChan(send3) + foundVideoLimit := false + for _, m := range msgs { + if extractCode(t, m) == "VIDEO_LIMIT" { + foundVideoLimit = true + break + } + } + if !foundVideoLimit { + t.Error("expected VIDEO_LIMIT error when MaxVideo limit is reached") + } + + // Verify DB state was NOT updated (camera should still be false). + state, err := database.GetVoiceState(user3.ID) + if err != nil { + t.Fatalf("GetVoiceState: %v", err) + } + if state != nil && state.Camera { + t.Error("camera should not be enabled after VIDEO_LIMIT rejection") + } +} +