Room Metadata Implementation (#126)

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/wire v0.5.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b
-	github.com/livekit/protocol v0.9.3
+	github.com/livekit/protocol v0.9.4
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -246,8 +246,8 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
 github.com/livekit/ion-sfu v1.20.9 h1:Ih+tE5CXqdL1o6iXfZn+uBoddBXWceaOKVIun6aVuZM=
 github.com/livekit/ion-sfu v1.20.9/go.mod h1:g8hwobZI5fvX1RXvayf4ZXkgP7spV5YGE4yTSsumpB4=
-github.com/livekit/protocol v0.9.3 h1:rzJwes4AddFF/h47G6kU1/GQF8/meqZ3uihFIKPPxz0=
-github.com/livekit/protocol v0.9.3/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo=
+github.com/livekit/protocol v0.9.4 h1:1vA49MwawrriOS3VvPqYfLiFI94XWvYGaWfkHN6ah6Y=
+github.com/livekit/protocol v0.9.4/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo=
 github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ=
 github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE=
 github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=

pkg/rtc/participant.go

@@ -578,6 +578,16 @@ func (p *ParticipantImpl) SendDataPacket(dp *livekit.DataPacket) error {
 	return dc.Send(data)
 }
 
+func (p *ParticipantImpl) SendRoomUpdate(room *livekit.Room) error {
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_RoomUpdate{
+			RoomUpdate: &livekit.RoomUpdate{
+				Room: room,
+			},
+		},
+	})
+}
+
 func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool, fromAdmin bool) {
 	isPending := false
 	p.lock.RLock()

pkg/rtc/room.go

@@ -47,6 +47,7 @@ type Room struct {
 	statsReporter *stats.RoomStatsReporter
 
 	onParticipantChanged func(p types.Participant)
+	onMetadataUpdate     func(metadata string)
 	onClose              func()
 }
 
@@ -73,6 +74,7 @@ func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICES
 	}
 	r.statsReporter.RoomStarted()
 	go r.audioUpdateWorker()
+
 	return r
 }
 
@@ -374,6 +376,30 @@ func (r *Room) SendDataPacket(up *livekit.UserPacket, kind livekit.DataPacket_Ki
 	r.onDataPacket(nil, dp)
 }
 
+func (r *Room) SetMetadata(metadata string) {
+	r.Room.Metadata = metadata
+
+	// Send update to participants
+	for _, p := range r.GetParticipants() {
+		if !p.IsReady() {
+			continue
+		}
+
+		err := p.SendRoomUpdate(r.Room)
+		if err != nil {
+			logger.Warnw("failed to send room update", err, "room", r.Room.Name, "participant", p.Identity())
+		}
+	}
+
+	if r.onMetadataUpdate != nil {
+		r.onMetadataUpdate(metadata)
+	}
+}
+
+func (r *Room) OnMetadataUpdate(f func(metadata string)) {
+	r.onMetadataUpdate = f
+}
+
 // checks if participant should be autosubscribed to new tracks, assumes lock is already acquired
 func (r *Room) autoSubscribe(participant types.Participant) bool {
 	if !participant.CanSubscribe() {

pkg/rtc/room_test.go

@@ -520,6 +520,20 @@ func TestHiddenParticipants(t *testing.T) {
 	})
 }
 
+func TestRoomUpdate(t *testing.T) {
+	t.Run("participants should receive metadata update", func(t *testing.T) {
+		rm := newRoomWithParticipants(t, testRoomOpts{num: 2})
+		defer rm.Close()
+
+		rm.SetMetadata("test metadata...")
+
+		for _, op := range rm.GetParticipants() {
+			fp := op.(*typesfakes.FakeParticipant)
+			require.Equal(t, 1, fp.SendRoomUpdateCallCount())
+		}
+	})
+}
+
 type testRoomOpts struct {
 	num                  int
 	numHidden            int
@@ -548,6 +562,7 @@ func newRoomWithParticipants(t *testing.T, opts testRoomOpts) *rtc.Room {
 		participant := newMockParticipant(identity, opts.protocol, i >= opts.num)
 		err := rm.Join(participant, &rtc.ParticipantOptions{AutoSubscribe: true})
 		participant.StateReturns(livekit.ParticipantInfo_ACTIVE)
+		participant.IsReadyReturns(true)
 		require.NoError(t, err)
 	}
 	return rm

pkg/rtc/types/interfaces.go

@@ -50,6 +50,7 @@ type Participant interface {
 	SendParticipantUpdate(participants []*livekit.ParticipantInfo) error
 	SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error
 	SendDataPacket(packet *livekit.DataPacket) error
+	SendRoomUpdate(room *livekit.Room) error
 	SetTrackMuted(trackId string, muted bool, fromAdmin bool)
 	GetAudioLevel() (level uint8, active bool)
 

pkg/rtc/types/typesfakes/fake_participant.go

@@ -325,6 +325,17 @@ type FakeParticipant struct {
 	sendParticipantUpdateReturnsOnCall map[int]struct {
 		result1 error
 	}
+	SendRoomUpdateStub        func(*livekit.Room) error
+	sendRoomUpdateMutex       sync.RWMutex
+	sendRoomUpdateArgsForCall []struct {
+		arg1 *livekit.Room
+	}
+	sendRoomUpdateReturns struct {
+		result1 error
+	}
+	sendRoomUpdateReturnsOnCall map[int]struct {
+		result1 error
+	}
 	SendSpeakerUpdateStub        func([]*livekit.SpeakerInfo) error
 	sendSpeakerUpdateMutex       sync.RWMutex
 	sendSpeakerUpdateArgsForCall []struct {
@@ -2117,6 +2128,67 @@ func (fake *FakeParticipant) SendParticipantUpdateReturnsOnCall(i int, result1 e
 	}{result1}
 }
 
+func (fake *FakeParticipant) SendRoomUpdate(arg1 *livekit.Room) error {
+	fake.sendRoomUpdateMutex.Lock()
+	ret, specificReturn := fake.sendRoomUpdateReturnsOnCall[len(fake.sendRoomUpdateArgsForCall)]
+	fake.sendRoomUpdateArgsForCall = append(fake.sendRoomUpdateArgsForCall, struct {
+		arg1 *livekit.Room
+	}{arg1})
+	stub := fake.SendRoomUpdateStub
+	fakeReturns := fake.sendRoomUpdateReturns
+	fake.recordInvocation("SendRoomUpdate", []interface{}{arg1})
+	fake.sendRoomUpdateMutex.Unlock()
+	if stub != nil {
+		return stub(arg1)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) SendRoomUpdateCallCount() int {
+	fake.sendRoomUpdateMutex.RLock()
+	defer fake.sendRoomUpdateMutex.RUnlock()
+	return len(fake.sendRoomUpdateArgsForCall)
+}
+
+func (fake *FakeParticipant) SendRoomUpdateCalls(stub func(*livekit.Room) error) {
+	fake.sendRoomUpdateMutex.Lock()
+	defer fake.sendRoomUpdateMutex.Unlock()
+	fake.SendRoomUpdateStub = stub
+}
+
+func (fake *FakeParticipant) SendRoomUpdateArgsForCall(i int) *livekit.Room {
+	fake.sendRoomUpdateMutex.RLock()
+	defer fake.sendRoomUpdateMutex.RUnlock()
+	argsForCall := fake.sendRoomUpdateArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *FakeParticipant) SendRoomUpdateReturns(result1 error) {
+	fake.sendRoomUpdateMutex.Lock()
+	defer fake.sendRoomUpdateMutex.Unlock()
+	fake.SendRoomUpdateStub = nil
+	fake.sendRoomUpdateReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *FakeParticipant) SendRoomUpdateReturnsOnCall(i int, result1 error) {
+	fake.sendRoomUpdateMutex.Lock()
+	defer fake.sendRoomUpdateMutex.Unlock()
+	fake.SendRoomUpdateStub = nil
+	if fake.sendRoomUpdateReturnsOnCall == nil {
+		fake.sendRoomUpdateReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.sendRoomUpdateReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *FakeParticipant) SendSpeakerUpdate(arg1 []*livekit.SpeakerInfo) error {
 	var arg1Copy []*livekit.SpeakerInfo
 	if arg1 != nil {
@@ -2675,6 +2747,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.sendJoinResponseMutex.RUnlock()
 	fake.sendParticipantUpdateMutex.RLock()
 	defer fake.sendParticipantUpdateMutex.RUnlock()
+	fake.sendRoomUpdateMutex.RLock()
+	defer fake.sendRoomUpdateMutex.RUnlock()
 	fake.sendSpeakerUpdateMutex.RLock()
 	defer fake.sendSpeakerUpdateMutex.RUnlock()
 	fake.setMetadataMutex.RLock()

pkg/service/roommanager.go

@@ -304,6 +304,12 @@ func (r *LocalRoomManager) getOrCreateRoom(ctx context.Context, roomName string)
 			"outgoingStats", room.GetOutgoingStats().Copy(),
 		)
 	})
+	room.OnMetadataUpdate(func(metadata string) {
+		err := r.StoreRoom(ctx, room.Room)
+		if err != nil {
+			logger.Errorw("could not handle metadata update", err)
+		}
+	})
 	room.OnParticipantChanged(func(p types.Participant) {
 		var err error
 		if p.State() == livekit.ParticipantInfo_DISCONNECTED {
@@ -482,6 +488,9 @@ func (r *LocalRoomManager) handleRTCMessage(ctx context.Context, roomName, ident
 			DestinationSids: rm.SendData.DestinationSids,
 		}
 		room.SendDataPacket(up, rm.SendData.Kind)
+	case *livekit.RTCNodeMessage_UpdateRoomMetadata:
+		logger.Debugw("updating room", "room", roomName)
+		room.SetMetadata(rm.UpdateRoomMetadata.Metadata)
 	}
 }
 

pkg/service/roomservice.go

@@ -215,6 +215,37 @@ func (s *RoomService) SendData(ctx context.Context, req *livekit.SendDataRequest
 	return &livekit.SendDataResponse{}, nil
 }
 
+func (s *RoomService) UpdateRoomMetadata(ctx context.Context, req *livekit.UpdateRoomMetadataRequest) (*livekit.Room, error) {
+	if err := EnsureAdminPermission(ctx, req.Room); err != nil {
+		return nil, twirpAuthError(err)
+	}
+
+	room, err := s.roomStore.LoadRoom(ctx, req.Room)
+	if err != nil {
+		return nil, err
+	}
+
+	room.Metadata = req.Metadata
+
+	participants, err := s.roomStore.ListParticipants(ctx, req.Room)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(participants) > 0 {
+		err := s.writeMessage(ctx, req.Room, participants[0].Identity, &livekit.RTCNodeMessage{
+			Message: &livekit.RTCNodeMessage_UpdateRoomMetadata{
+				UpdateRoomMetadata: req,
+			},
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return room, nil
+}
+
 func (s *RoomService) writeMessage(ctx context.Context, room, identity string, msg *livekit.RTCNodeMessage) error {
 	if err := EnsureAdminPermission(ctx, room); err != nil {
 		return twirpAuthError(err)