Provide the InputVideo/AudioState to Ingress in WHIPRTCConnectionNotify (#3982)

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397
-	github.com/livekit/protocol v1.42.1-0.20250930033119-68992df259bc
+	github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724
 	github.com/livekit/psrpc v0.7.0
 	github.com/mackerelio/go-osstat v0.2.6
 	github.com/magefile/mage v1.15.0

go.sum

@@ -171,8 +171,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397 h1:Z7j2mY+bvG05UC80MpnJkitlJju8sSDWsr0Bb4dPceo=
 github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A=
-github.com/livekit/protocol v1.42.1-0.20250930033119-68992df259bc h1:ZotDBmtFumH723+RVlnMee3bL399LbJ874wj2V9sj2I=
-github.com/livekit/protocol v1.42.1-0.20250930033119-68992df259bc/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A=
+github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724 h1:oz2RQahwZ8LPaGX+2RzAR08s+zqJJSep+OCvWuD90I4=
+github.com/livekit/protocol v1.42.1-0.20251006173230-a7bc52ad9724/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A=
 github.com/livekit/psrpc v0.7.0 h1:rtfqfjYN06WJYloE/S0nmkJ/Y04x4pxLQLe8kQ4FVHU=
 github.com/livekit/psrpc v0.7.0/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk=
 github.com/mackerelio/go-osstat v0.2.6 h1:gs4U8BZeS1tjrL08tt5VUliVvSWP26Ai2Ob8Lr7f2i0=

pkg/service/roommanager_service.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -18,6 +19,10 @@ import (
 	"github.com/livekit/psrpc"
 )
 
+const (
+	whipSessionNotifyInterval = 10 * time.Second
+)
+
 type whipService struct {
 	*RoomManager
 
@@ -114,18 +119,32 @@ func (s whipService) Create(ctx context.Context, req *rpc.WHIPCreateRequest) (*r
 	}
 
 	if req.FromIngress {
+		aliveCtx, cancel := context.WithCancel(context.Background())
+
 		lp.AddOnClose(types.ParticipantCloseKeyWHIP, func(lp types.LocalParticipant) {
+			cancel()
+
 			go func() {
 				lp.GetLogger().Debugw("whip service: notify participant closed")
+
+				video, audio := getMediaStateForParticipant(lp)
+
 				_, err := s.ingressRpcCli.WHIPRTCConnectionNotify(context.Background(), string(lp.ID()), &rpc.WHIPRTCConnectionNotifyRequest{
 					ParticipantId: string(lp.ID()),
 					Closed:        true,
+					Audio:         audio,
+					Video:         video,
 				}, psrpc.WithRequestTimeout(rpc.DefaultPSRPCConfig.Timeout))
 				if err != nil {
 					lp.GetLogger().Warnw("whip service: could not notify ingress of participant closed", err)
 				}
 			}()
 		})
+		go func() {
+			if err := s.notifySession(aliveCtx, lp); err != nil {
+				cancel()
+			}
+		}()
 	}
 
 	var iceServers []*livekit.ICEServer
@@ -145,6 +164,92 @@ func (s whipService) Create(ctx context.Context, req *rpc.WHIPCreateRequest) (*r
 	}, nil
 }
 
+func (s whipService) notifySession(ctx context.Context, participant types.Participant) error {
+	ticker := time.NewTicker(whipSessionNotifyInterval)
+	defer ticker.Stop()
+
+	err := s.sendConnectionNotify(ctx, participant)
+	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			return nil
+		}
+	}
+
+	for {
+		select {
+		case <-ticker.C:
+			err := s.sendConnectionNotify(ctx, participant)
+			if err != nil {
+				if errors.Is(err, context.Canceled) {
+					return nil
+				}
+			}
+
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (s whipService) sendConnectionNotify(ctx context.Context, participant types.Participant) error {
+	video, audio := getMediaStateForParticipant(participant)
+
+	_, err := s.ingressRpcCli.WHIPRTCConnectionNotify(ctx, string(participant.ID()), &rpc.WHIPRTCConnectionNotifyRequest{
+		ParticipantId: string(participant.ID()),
+		Video:         video,
+		Audio:         audio,
+	}, psrpc.WithRequestTimeout(rpc.DefaultPSRPCConfig.Timeout))
+
+	return err
+}
+
+func getMediaStateForParticipant(participant types.Participant) (*livekit.InputVideoState, *livekit.InputAudioState) {
+	pParticipant := participant.ToProto()
+
+	var video *livekit.InputVideoState
+	var audio *livekit.InputAudioState
+
+	for _, v := range pParticipant.Tracks {
+		if v == nil {
+			continue
+		}
+
+		if v.Type != livekit.TrackType_VIDEO {
+			continue
+		}
+
+		video = &livekit.InputVideoState{}
+
+		video.MimeType = v.MimeType
+		video.Height = v.Height
+		video.Width = v.Width
+
+		break
+	}
+
+	for _, a := range pParticipant.Tracks {
+		if a == nil {
+			continue
+		}
+
+		if a.Type != livekit.TrackType_AUDIO {
+			continue
+		}
+
+		audio = &livekit.InputAudioState{}
+
+		audio.MimeType = a.MimeType
+		audio.Channels = 1
+		if a.Stereo {
+			audio.Channels = 2
+		}
+
+		break
+	}
+
+	return video, audio
+}
+
 // -------------------------------------------
 
 type whipParticipantService struct {