RoomService.updateParticipantMetadata, participant permissions

pkg/routing/interfaces.go

@@ -23,7 +23,14 @@ type MessageSource interface {
 	ReadChan() <-chan proto.Message
 }
 
-type NewParticipantCallback func(roomName, identity, metadata string, reconnect bool, requestSource MessageSource, responseSink MessageSink)
+type ParticipantInit struct {
+	Identity   string
+	Metadata   string
+	Reconnect  bool
+	Permission *livekit.ParticipantPermission
+}
+
+type NewParticipantCallback func(roomName string, pi ParticipantInit, requestSource MessageSource, responseSink MessageSink)
 type RTCMessageCallback func(roomName, identity string, msg *livekit.RTCNodeMessage)
 
 // Router allows multiple nodes to coordinate the participant session
@@ -39,7 +46,7 @@ type Router interface {
 	ListNodes() ([]*livekit.Node, error)
 
 	// participant signal connection is ready to start
-	StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error)
+	StartParticipantSignal(roomName string, pi ParticipantInit) (connectionId string, reqSink MessageSink, resSource MessageSource, err error)
 
 	// sends a message to RTC node
 	CreateRTCSink(roomName, identity string) (MessageSink, error)

pkg/routing/localrouter.go

@@ -71,7 +71,7 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) {
 	}, nil
 }
 
-func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
+func (r *LocalRouter) StartParticipantSignal(roomName string, pi ParticipantInit) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
 	// treat it as a new participant connecting
 	if r.onNewParticipant == nil {
 		err = ErrHandlerNotDefined
@@ -79,21 +79,19 @@ func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string
 	}
 
 	// index channels by roomName | identity
-	key := participantKey(roomName, identity)
+	key := participantKey(roomName, pi.Identity)
 	reqChan := r.getOrCreateMessageChannel(r.requestChannels, key)
 	resChan := r.getOrCreateMessageChannel(r.responseChannels, key)
 
 	r.onNewParticipant(
 		roomName,
-		identity,
-		metadata,
-		reconnect,
+		pi,
 		// request source
 		reqChan,
 		// response sink
 		resChan,
 	)
-	return identity, reqChan, resChan, nil
+	return pi.Identity, reqChan, resChan, nil
 }
 
 func (r *LocalRouter) CreateRTCSink(roomName, identity string) (MessageSink, error) {

pkg/routing/redisrouter.go

@@ -129,7 +129,7 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) {
 }
 
 // signal connection sets up paths to the RTC node, and starts to route messages to that message queue
-func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
+func (r *RedisRouter) StartParticipantSignal(roomName string, pi ParticipantInit) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) {
 	// find the node where the room is hosted at
 	rtcNode, err := r.GetNodeForRoom(roomName)
 	if err != nil {
@@ -138,7 +138,7 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string
 
 	// create a new connection id
 	connectionId = utils.NewGuid("CO_")
-	pKey := participantKey(roomName, identity)
+	pKey := participantKey(roomName, pi.Identity)
 
 	// map signal & rtc nodes
 	if err = r.setParticipantSignalNode(connectionId, r.currentNode.Id); err != nil {
@@ -150,11 +150,12 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string
 	// sends a message to start session
 	err = sink.WriteMessage(&livekit.StartSession{
 		RoomName: roomName,
-		Identity: identity,
-		Metadata: metadata,
+		Identity: pi.Identity,
+		Metadata: pi.Metadata,
 		// connection id is to allow the RTC node to identify where to route the message back to
 		ConnectionId: connectionId,
-		Reconnect:    reconnect,
+		Reconnect:    pi.Reconnect,
+		Permission:   pi.Permission,
 	})
 	if err != nil {
 		return
@@ -215,13 +216,18 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK
 		}
 	}
 
+	pi := ParticipantInit{
+		Identity:   ss.Identity,
+		Metadata:   ss.Metadata,
+		Reconnect:  ss.Reconnect,
+		Permission: ss.Permission,
+	}
+
 	reqChan := r.getOrCreateMessageChannel(r.requestChannels, participantKey)
 	resSink := NewSignalNodeSink(r.rc, signalNode, ss.ConnectionId)
 	r.onNewParticipant(
 		ss.RoomName,
-		ss.Identity,
-		ss.Metadata,
-		ss.Reconnect,
+		pi,
 		reqChan,
 		resSink,
 	)

pkg/routing/routingfakes/fake_router.go

@@ -124,13 +124,11 @@ type FakeRouter struct {
 	startReturnsOnCall map[int]struct {
 		result1 error
 	}
-	StartParticipantSignalStub        func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error)
+	StartParticipantSignalStub        func(string, routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error)
 	startParticipantSignalMutex       sync.RWMutex
 	startParticipantSignalArgsForCall []struct {
 		arg1 string
-		arg2 string
-		arg3 string
-		arg4 bool
+		arg2 routing.ParticipantInit
 	}
 	startParticipantSignalReturns struct {
 		result1 string
@@ -757,21 +755,19 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
-func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string, arg4 bool) (string, routing.MessageSink, routing.MessageSource, error) {
+func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error) {
 	fake.startParticipantSignalMutex.Lock()
 	ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)]
 	fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct {
 		arg1 string
-		arg2 string
-		arg3 string
-		arg4 bool
-	}{arg1, arg2, arg3, arg4})
+		arg2 routing.ParticipantInit
+	}{arg1, arg2})
 	stub := fake.StartParticipantSignalStub
 	fakeReturns := fake.startParticipantSignalReturns
-	fake.recordInvocation("StartParticipantSignal", []interface{}{arg1, arg2, arg3, arg4})
+	fake.recordInvocation("StartParticipantSignal", []interface{}{arg1, arg2})
 	fake.startParticipantSignalMutex.Unlock()
 	if stub != nil {
-		return stub(arg1, arg2, arg3, arg4)
+		return stub(arg1, arg2)
 	}
 	if specificReturn {
 		return ret.result1, ret.result2, ret.result3, ret.result4
@@ -785,17 +781,17 @@ func (fake *FakeRouter) StartParticipantSignalCallCount() int {
 	return len(fake.startParticipantSignalArgsForCall)
 }
 
-func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error)) {
+func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, routing.ParticipantInit) (string, routing.MessageSink, routing.MessageSource, error)) {
 	fake.startParticipantSignalMutex.Lock()
 	defer fake.startParticipantSignalMutex.Unlock()
 	fake.StartParticipantSignalStub = stub
 }
 
-func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (string, string, string, bool) {
+func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (string, routing.ParticipantInit) {
 	fake.startParticipantSignalMutex.RLock()
 	defer fake.startParticipantSignalMutex.RUnlock()
 	argsForCall := fake.startParticipantSignalArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
+	return argsForCall.arg1, argsForCall.arg2
 }
 
 func (fake *FakeRouter) StartParticipantSignalReturns(result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) {

pkg/rtc/participant.go

@@ -38,9 +38,10 @@ type ParticipantImpl struct {
 	conf         *WebRTCConfig
 	identity     string
 	// JSON encoded metadata to pass to clients
-	metadata string
-	state    atomic.Value // livekit.ParticipantInfo_State
-	rtcpCh   chan []rtcp.Packet
+	metadata   string
+	permission *livekit.ParticipantPermission
+	state      atomic.Value // livekit.ParticipantInfo_State
+	rtcpCh     chan []rtcp.Packet
 
 	// hold reference for MediaTrack
 	twcc *twcc.Responder
@@ -59,6 +60,7 @@ type ParticipantImpl struct {
 	onTrackPublished func(types.Participant, types.PublishedTrack)
 	onTrackUpdated   func(types.Participant, types.PublishedTrack)
 	onStateChange    func(p types.Participant, oldState livekit.ParticipantInfo_State)
+	onMetadataUpdate func(types.Participant)
 	onClose          func(types.Participant)
 }
 
@@ -140,18 +142,24 @@ func (p *ParticipantImpl) IsReady() bool {
 func (p *ParticipantImpl) SetMetadata(metadata map[string]interface{}) error {
 	if metadata == nil {
 		p.metadata = ""
-		return nil
-	}
-
-	if data, err := json.Marshal(metadata); err != nil {
-		return err
 	} else {
-		p.metadata = string(data)
+		if data, err := json.Marshal(metadata); err != nil {
+			return err
+		} else {
+			p.metadata = string(data)
+		}
 	}
 
+	if p.onMetadataUpdate != nil {
+		p.onMetadataUpdate(p)
+	}
 	return nil
 }
 
+func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermission) {
+	p.permission = permission
+}
+
 func (p *ParticipantImpl) RTCPChan() chan []rtcp.Packet {
 	return p.rtcpCh
 }
@@ -197,6 +205,10 @@ func (p *ParticipantImpl) OnTrackUpdated(callback func(types.Participant, types.
 	p.onTrackUpdated = callback
 }
 
+func (p *ParticipantImpl) OnMetadataUpdate(callback func(types.Participant)) {
+	p.onMetadataUpdate = callback
+}
+
 func (p *ParticipantImpl) OnClose(callback func(types.Participant)) {
 	p.onClose = callback
 }
@@ -459,6 +471,14 @@ func (p *ParticipantImpl) GetAudioLevel() (level uint8, noisy bool) {
 	return
 }
 
+func (p *ParticipantImpl) CanPublish() bool {
+	return p.permission == nil || p.permission.CanPublish
+}
+
+func (p *ParticipantImpl) CanSubscribe() bool {
+	return p.permission == nil || p.permission.CanSubscribe
+}
+
 func (p *ParticipantImpl) SubscriberPC() *webrtc.PeerConnection {
 	return p.subscriber.pc
 }
@@ -580,6 +600,12 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		"remoteTrack", track.ID(),
 		"rid", track.RID())
 
+	if !p.CanPublish() {
+		logger.Warnw("no permission to publish mediaTrack",
+			"participant", p.Identity())
+		return
+	}
+
 	// delete pending track if it's not simulcasting
 	ti := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), track.RID() == "")
 	if ti == nil {
@@ -620,6 +646,12 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
 	}
 	logger.Debugw("dataChannel added", "participant", p.Identity(), "label", dc.Label())
 
+	if !p.CanPublish() {
+		logger.Warnw("no permission to publish dataTrack",
+			"participant", p.Identity())
+		return
+	}
+
 	// data channels have numeric ids, so we use its label to identify
 	ti := p.getPendingTrack(dc.Label(), livekit.TrackType_DATA, true)
 	if ti == nil {

pkg/rtc/room.go

@@ -137,7 +137,7 @@ func (r *Room) Join(participant types.Participant) error {
 		if r.onParticipantChanged != nil {
 			r.onParticipantChanged(participant)
 		}
-		r.broadcastParticipantState(p)
+		r.broadcastParticipantState(p, true)
 
 		if p.State() == livekit.ParticipantInfo_ACTIVE {
 			// subscribe participant to existing publishedTracks
@@ -161,7 +161,7 @@ func (r *Room) Join(participant types.Participant) error {
 		}
 	})
 	participant.OnTrackUpdated(r.onTrackUpdated)
-
+	participant.OnMetadataUpdate(r.onParticipantMetadataUpdate)
 	logger.Infow("new participant joined",
 		"id", participant.ID(),
 		"identity", participant.Identity(),
@@ -213,7 +213,7 @@ func (r *Room) RemoveParticipant(identity string) {
 		if r.onParticipantChanged != nil {
 			r.onParticipantChanged(p)
 		}
-		r.broadcastParticipantState(p)
+		r.broadcastParticipantState(p, true)
 	}
 }
 
@@ -288,7 +288,7 @@ func (r *Room) OnParticipantChanged(f func(participant types.Participant)) {
 // a ParticipantImpl in the room added a new remoteTrack, subscribe other participants to it
 func (r *Room) onTrackAdded(participant types.Participant, track types.PublishedTrack) {
 	// publish participant update, since track state is changed
-	r.broadcastParticipantState(participant)
+	r.broadcastParticipantState(participant, true)
 
 	r.lock.RLock()
 	defer r.lock.RUnlock()
@@ -322,19 +322,27 @@ func (r *Room) onTrackAdded(participant types.Participant, track types.Published
 }
 
 func (r *Room) onTrackUpdated(p types.Participant, track types.PublishedTrack) {
-	r.broadcastParticipantState(p)
+	// send track updates to everyone, especially if track was updated by admin
+	r.broadcastParticipantState(p, false)
+	if r.onParticipantChanged != nil {
+		r.onParticipantChanged(p)
+	}
+}
+
+func (r *Room) onParticipantMetadataUpdate(p types.Participant) {
+	r.broadcastParticipantState(p, false)
 	if r.onParticipantChanged != nil {
 		r.onParticipantChanged(p)
 	}
 }
 
 // broadcast an update about participant p
-func (r *Room) broadcastParticipantState(p types.Participant) {
+func (r *Room) broadcastParticipantState(p types.Participant, skipSource bool) {
 	updates := ToProtoParticipants([]types.Participant{p})
 	participants := r.GetParticipants()
 	for _, op := range participants {
 		// skip itself && closed participants
-		if p.ID() == op.ID() || op.State() == livekit.ParticipantInfo_DISCONNECTED {
+		if (skipSource && p.ID() == op.ID()) || op.State() == livekit.ParticipantInfo_DISCONNECTED {
 			continue
 		}
 

pkg/rtc/types/interfaces.go

@@ -29,6 +29,7 @@ type Participant interface {
 	ToProto() *livekit.ParticipantInfo
 	RTCPChan() chan []rtcp.Packet
 	SetMetadata(metadata map[string]interface{}) error
+	SetPermission(permission *livekit.ParticipantPermission)
 	GetResponseSink() routing.MessageSink
 	SetResponseSink(sink routing.MessageSink)
 	SubscriberMediaEngine() *webrtc.MediaEngine
@@ -47,6 +48,10 @@ type Participant interface {
 	SetTrackMuted(trackId string, muted bool)
 	GetAudioLevel() (level uint8, noisy bool)
 
+	// permissions
+	CanPublish() bool
+	CanSubscribe() bool
+
 	Start()
 	Close() error
 
@@ -56,6 +61,7 @@ type Participant interface {
 	OnTrackPublished(func(Participant, PublishedTrack))
 	// OnTrackUpdated - one of its publishedTracks changed in status
 	OnTrackUpdated(callback func(Participant, PublishedTrack))
+	OnMetadataUpdate(callback func(Participant))
 	OnClose(func(Participant))
 
 	// package methods

pkg/rtc/types/typesfakes/fake_participant.go

@@ -48,6 +48,26 @@ type FakeParticipant struct {
 		arg2 string
 		arg3 livekit.TrackType
 	}
+	CanPublishStub        func() bool
+	canPublishMutex       sync.RWMutex
+	canPublishArgsForCall []struct {
+	}
+	canPublishReturns struct {
+		result1 bool
+	}
+	canPublishReturnsOnCall map[int]struct {
+		result1 bool
+	}
+	CanSubscribeStub        func() bool
+	canSubscribeMutex       sync.RWMutex
+	canSubscribeArgsForCall []struct {
+	}
+	canSubscribeReturns struct {
+		result1 bool
+	}
+	canSubscribeReturnsOnCall map[int]struct {
+		result1 bool
+	}
 	CloseStub        func() error
 	closeMutex       sync.RWMutex
 	closeArgsForCall []struct {
@@ -159,6 +179,11 @@ type FakeParticipant struct {
 	onCloseArgsForCall []struct {
 		arg1 func(types.Participant)
 	}
+	OnMetadataUpdateStub        func(func(types.Participant))
+	onMetadataUpdateMutex       sync.RWMutex
+	onMetadataUpdateArgsForCall []struct {
+		arg1 func(types.Participant)
+	}
 	OnStateChangeStub        func(func(p types.Participant, oldState livekit.ParticipantInfo_State))
 	onStateChangeMutex       sync.RWMutex
 	onStateChangeArgsForCall []struct {
@@ -241,6 +266,11 @@ type FakeParticipant struct {
 	setMetadataReturnsOnCall map[int]struct {
 		result1 error
 	}
+	SetPermissionStub        func(*livekit.ParticipantPermission)
+	setPermissionMutex       sync.RWMutex
+	setPermissionArgsForCall []struct {
+		arg1 *livekit.ParticipantPermission
+	}
 	SetResponseSinkStub        func(routing.MessageSink)
 	setResponseSinkMutex       sync.RWMutex
 	setResponseSinkArgsForCall []struct {
@@ -490,6 +520,112 @@ func (fake *FakeParticipant) AddTrackArgsForCall(i int) (string, string, livekit
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
+func (fake *FakeParticipant) CanPublish() bool {
+	fake.canPublishMutex.Lock()
+	ret, specificReturn := fake.canPublishReturnsOnCall[len(fake.canPublishArgsForCall)]
+	fake.canPublishArgsForCall = append(fake.canPublishArgsForCall, struct {
+	}{})
+	stub := fake.CanPublishStub
+	fakeReturns := fake.canPublishReturns
+	fake.recordInvocation("CanPublish", []interface{}{})
+	fake.canPublishMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) CanPublishCallCount() int {
+	fake.canPublishMutex.RLock()
+	defer fake.canPublishMutex.RUnlock()
+	return len(fake.canPublishArgsForCall)
+}
+
+func (fake *FakeParticipant) CanPublishCalls(stub func() bool) {
+	fake.canPublishMutex.Lock()
+	defer fake.canPublishMutex.Unlock()
+	fake.CanPublishStub = stub
+}
+
+func (fake *FakeParticipant) CanPublishReturns(result1 bool) {
+	fake.canPublishMutex.Lock()
+	defer fake.canPublishMutex.Unlock()
+	fake.CanPublishStub = nil
+	fake.canPublishReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeParticipant) CanPublishReturnsOnCall(i int, result1 bool) {
+	fake.canPublishMutex.Lock()
+	defer fake.canPublishMutex.Unlock()
+	fake.CanPublishStub = nil
+	if fake.canPublishReturnsOnCall == nil {
+		fake.canPublishReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.canPublishReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeParticipant) CanSubscribe() bool {
+	fake.canSubscribeMutex.Lock()
+	ret, specificReturn := fake.canSubscribeReturnsOnCall[len(fake.canSubscribeArgsForCall)]
+	fake.canSubscribeArgsForCall = append(fake.canSubscribeArgsForCall, struct {
+	}{})
+	stub := fake.CanSubscribeStub
+	fakeReturns := fake.canSubscribeReturns
+	fake.recordInvocation("CanSubscribe", []interface{}{})
+	fake.canSubscribeMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) CanSubscribeCallCount() int {
+	fake.canSubscribeMutex.RLock()
+	defer fake.canSubscribeMutex.RUnlock()
+	return len(fake.canSubscribeArgsForCall)
+}
+
+func (fake *FakeParticipant) CanSubscribeCalls(stub func() bool) {
+	fake.canSubscribeMutex.Lock()
+	defer fake.canSubscribeMutex.Unlock()
+	fake.CanSubscribeStub = stub
+}
+
+func (fake *FakeParticipant) CanSubscribeReturns(result1 bool) {
+	fake.canSubscribeMutex.Lock()
+	defer fake.canSubscribeMutex.Unlock()
+	fake.CanSubscribeStub = nil
+	fake.canSubscribeReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeParticipant) CanSubscribeReturnsOnCall(i int, result1 bool) {
+	fake.canSubscribeMutex.Lock()
+	defer fake.canSubscribeMutex.Unlock()
+	fake.CanSubscribeStub = nil
+	if fake.canSubscribeReturnsOnCall == nil {
+		fake.canSubscribeReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.canSubscribeReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
+}
+
 func (fake *FakeParticipant) Close() error {
 	fake.closeMutex.Lock()
 	ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)]
@@ -1074,6 +1210,38 @@ func (fake *FakeParticipant) OnCloseArgsForCall(i int) func(types.Participant) {
 	return argsForCall.arg1
 }
 
+func (fake *FakeParticipant) OnMetadataUpdate(arg1 func(types.Participant)) {
+	fake.onMetadataUpdateMutex.Lock()
+	fake.onMetadataUpdateArgsForCall = append(fake.onMetadataUpdateArgsForCall, struct {
+		arg1 func(types.Participant)
+	}{arg1})
+	stub := fake.OnMetadataUpdateStub
+	fake.recordInvocation("OnMetadataUpdate", []interface{}{arg1})
+	fake.onMetadataUpdateMutex.Unlock()
+	if stub != nil {
+		fake.OnMetadataUpdateStub(arg1)
+	}
+}
+
+func (fake *FakeParticipant) OnMetadataUpdateCallCount() int {
+	fake.onMetadataUpdateMutex.RLock()
+	defer fake.onMetadataUpdateMutex.RUnlock()
+	return len(fake.onMetadataUpdateArgsForCall)
+}
+
+func (fake *FakeParticipant) OnMetadataUpdateCalls(stub func(func(types.Participant))) {
+	fake.onMetadataUpdateMutex.Lock()
+	defer fake.onMetadataUpdateMutex.Unlock()
+	fake.OnMetadataUpdateStub = stub
+}
+
+func (fake *FakeParticipant) OnMetadataUpdateArgsForCall(i int) func(types.Participant) {
+	fake.onMetadataUpdateMutex.RLock()
+	defer fake.onMetadataUpdateMutex.RUnlock()
+	argsForCall := fake.onMetadataUpdateArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeParticipant) OnStateChange(arg1 func(p types.Participant, oldState livekit.ParticipantInfo_State)) {
 	fake.onStateChangeMutex.Lock()
 	fake.onStateChangeArgsForCall = append(fake.onStateChangeArgsForCall, struct {
@@ -1554,6 +1722,38 @@ func (fake *FakeParticipant) SetMetadataReturnsOnCall(i int, result1 error) {
 	}{result1}
 }
 
+func (fake *FakeParticipant) SetPermission(arg1 *livekit.ParticipantPermission) {
+	fake.setPermissionMutex.Lock()
+	fake.setPermissionArgsForCall = append(fake.setPermissionArgsForCall, struct {
+		arg1 *livekit.ParticipantPermission
+	}{arg1})
+	stub := fake.SetPermissionStub
+	fake.recordInvocation("SetPermission", []interface{}{arg1})
+	fake.setPermissionMutex.Unlock()
+	if stub != nil {
+		fake.SetPermissionStub(arg1)
+	}
+}
+
+func (fake *FakeParticipant) SetPermissionCallCount() int {
+	fake.setPermissionMutex.RLock()
+	defer fake.setPermissionMutex.RUnlock()
+	return len(fake.setPermissionArgsForCall)
+}
+
+func (fake *FakeParticipant) SetPermissionCalls(stub func(*livekit.ParticipantPermission)) {
+	fake.setPermissionMutex.Lock()
+	defer fake.setPermissionMutex.Unlock()
+	fake.SetPermissionStub = stub
+}
+
+func (fake *FakeParticipant) SetPermissionArgsForCall(i int) *livekit.ParticipantPermission {
+	fake.setPermissionMutex.RLock()
+	defer fake.setPermissionMutex.RUnlock()
+	argsForCall := fake.setPermissionArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeParticipant) SetResponseSink(arg1 routing.MessageSink) {
 	fake.setResponseSinkMutex.Lock()
 	fake.setResponseSinkArgsForCall = append(fake.setResponseSinkArgsForCall, struct {
@@ -1866,6 +2066,10 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.addSubscriberMutex.RUnlock()
 	fake.addTrackMutex.RLock()
 	defer fake.addTrackMutex.RUnlock()
+	fake.canPublishMutex.RLock()
+	defer fake.canPublishMutex.RUnlock()
+	fake.canSubscribeMutex.RLock()
+	defer fake.canSubscribeMutex.RUnlock()
 	fake.closeMutex.RLock()
 	defer fake.closeMutex.RUnlock()
 	fake.getAudioLevelMutex.RLock()
@@ -1888,6 +2092,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.isReadyMutex.RUnlock()
 	fake.onCloseMutex.RLock()
 	defer fake.onCloseMutex.RUnlock()
+	fake.onMetadataUpdateMutex.RLock()
+	defer fake.onMetadataUpdateMutex.RUnlock()
 	fake.onStateChangeMutex.RLock()
 	defer fake.onStateChangeMutex.RUnlock()
 	fake.onTrackPublishedMutex.RLock()
@@ -1908,6 +2114,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.sendParticipantUpdateMutex.RUnlock()
 	fake.setMetadataMutex.RLock()
 	defer fake.setMetadataMutex.RUnlock()
+	fake.setPermissionMutex.RLock()
+	defer fake.setPermissionMutex.RUnlock()
 	fake.setResponseSinkMutex.RLock()
 	defer fake.setResponseSinkMutex.RUnlock()
 	fake.setTrackMutedMutex.RLock()

pkg/service/roommanager.go

@@ -172,22 +172,22 @@ func (r *RoomManager) CloseIdleRooms() {
 }
 
 // starts WebRTC session when a new participant is connected, takes place on RTC node
-func (r *RoomManager) StartSession(roomName, identity, metadata string, reconnect bool, requestSource routing.MessageSource, responseSink routing.MessageSink) {
+func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit, requestSource routing.MessageSource, responseSink routing.MessageSink) {
 	room, err := r.getOrCreateRoom(roomName)
 	if err != nil {
 		logger.Errorw("could not create room", "error", err)
 		return
 	}
 
-	participant := room.GetParticipant(identity)
+	participant := room.GetParticipant(pi.Identity)
 	if participant != nil {
 		// When reconnecting, it means WS has interrupted by underlying peer connection is still ok
 		// in this mode, we'll keep the participant SID, and just swap the sink for the underlying connection
-		if reconnect {
+		if pi.Reconnect {
 			logger.Debugw("resuming RTC session",
 				"room", roomName,
 				"node", r.currentNode.Id,
-				"participant", identity,
+				"participant", pi.Identity,
 			)
 			// close previous sink, and link to new one
 			prevSink := participant.GetResponseSink()
@@ -206,22 +206,26 @@ func (r *RoomManager) StartSession(roomName, identity, metadata string, reconnec
 	logger.Debugw("starting RTC session",
 		"room", roomName,
 		"node", r.currentNode.Id,
-		"participant", identity,
+		"participant", pi.Identity,
 		"num_participants", len(room.GetParticipants()),
 	)
 
-	participant, err = rtc.NewParticipant(identity, r.rtcConfig, responseSink, r.config.Audio)
+	participant, err = rtc.NewParticipant(pi.Identity, r.rtcConfig, responseSink, r.config.Audio)
 	if err != nil {
 		logger.Errorw("could not create participant", "error", err)
 		return
 	}
-	if metadata != "" {
+	if pi.Metadata != "" {
 		var md map[string]interface{}
-		if err := json.Unmarshal([]byte(metadata), &md); err == nil {
+		if err := json.Unmarshal([]byte(pi.Metadata), &md); err == nil {
 			participant.SetMetadata(md)
 		}
 	}
 
+	if pi.Permission != nil {
+		participant.SetPermission(pi.Permission)
+	}
+
 	// join room
 	if err := room.Join(participant); err != nil {
 		logger.Errorw("could not join room", "error", err)
@@ -334,7 +338,13 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 			case *livekit.SignalRequest_Mute:
 				participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted)
 			case *livekit.SignalRequest_Subscription:
-				room.UpdateSubscriptions(participant, msg.Subscription)
+				if participant.CanSubscribe() {
+					room.UpdateSubscriptions(participant, msg.Subscription)
+				} else {
+					logger.Warnw("rejected participant subscription",
+						"participant", participant.Identity(),
+						"tracks", msg.Subscription.TrackSids)
+				}
 			case *livekit.SignalRequest_TrackSetting:
 				for _, subTrack := range participant.GetSubscribedTracks() {
 					for _, sid := range msg.TrackSetting.TrackSids {
@@ -373,6 +383,16 @@ func (r *RoomManager) handleRTCMessage(roomName, identity string, msg *livekit.R
 		logger.Debugw("setting track muted", "room", roomName, "participant", identity,
 			"track", rm.MuteTrack.TrackSid, "muted", rm.MuteTrack.Muted)
 		participant.SetTrackMuted(rm.MuteTrack.TrackSid, rm.MuteTrack.Muted)
+	case *livekit.RTCNodeMessage_UpdateMetadata:
+		logger.Debugw("updating metadata", "room", roomName, "participant", identity)
+		var md map[string]interface{}
+		if rm.UpdateMetadata.Metadata != "" {
+			if err := json.Unmarshal([]byte(rm.UpdateMetadata.Metadata), &md); err != nil {
+				logger.Errorw("could not update metadata", "error", err)
+				return
+			}
+		}
+		participant.SetMetadata(md)
 	}
 }
 

pkg/service/roomservice.go

@@ -7,6 +7,7 @@ import (
 	"github.com/thoas/go-funk"
 	"github.com/twitchtv/twirp"
 
+	"github.com/livekit/livekit-server/pkg/routing"
 	livekit "github.com/livekit/livekit-server/proto"
 )
 
@@ -97,16 +98,8 @@ func (s *RoomService) GetParticipant(ctx context.Context, req *livekit.RoomParti
 }
 
 func (s *RoomService) RemoveParticipant(ctx context.Context, req *livekit.RoomParticipantIdentity) (res *livekit.RemoveParticipantResponse, err error) {
-	if err = EnsureAdminPermission(ctx, req.Room); err != nil {
-		return nil, twirpAuthError(err)
-	}
+	rtcSink, err := s.createRTCSink(ctx, req.Room, req.Identity)
 
-	participant, err := s.roomManager.roomStore.GetParticipant(req.Room, req.Identity)
-	if err != nil {
-		return
-	}
-
-	rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity)
 	if err != nil {
 		return
 	}
@@ -122,30 +115,16 @@ func (s *RoomService) RemoveParticipant(ctx context.Context, req *livekit.RoomPa
 }
 
 func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteRoomTrackRequest) (res *livekit.MuteRoomTrackResponse, err error) {
-	if err = EnsureAdminPermission(ctx, req.Room); err != nil {
-		return nil, twirpAuthError(err)
-	}
-
-	participant, err := s.roomManager.roomStore.GetParticipant(req.Room, req.Identity)
-	if err != nil {
-		return
-	}
-
-	rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity)
+	rtcSink, err := s.createRTCSink(ctx, req.Room, req.Identity)
 	if err != nil {
 		return
 	}
 	defer rtcSink.Close()
-	err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
-		Message: &livekit.RTCNodeMessage_MuteTrack{
-			MuteTrack: req,
-		},
-	})
 
+	participant, err := s.roomManager.roomStore.GetParticipant(req.Room, req.Identity)
 	if err != nil {
-		return
+		return nil, err
 	}
-
 	// find the track
 	track := funk.Find(participant.Tracks, func(t *livekit.TrackInfo) bool {
 		return t.Sid == req.TrackSid
@@ -153,8 +132,55 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR
 	if track == nil {
 		return nil, twirp.NotFoundError(ErrTrackNotFound.Error())
 	}
+
+	err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
+		Message: &livekit.RTCNodeMessage_MuteTrack{
+			MuteTrack: req,
+		},
+	})
+	if err != nil {
+		return
+	}
+
 	res = &livekit.MuteRoomTrackResponse{
 		Track: track.(*livekit.TrackInfo),
 	}
+	// mute might not have happened, reflect desired state
+	res.Track.Muted = req.Muted
 	return
 }
+
+func (s *RoomService) UpdateParticipantMetadata(ctx context.Context, req *livekit.UpdateParticipantMetadataRequest) (*livekit.ParticipantInfo, error) {
+	rtcSink, err := s.createRTCSink(ctx, req.Target.Room, req.Target.Identity)
+	if err != nil {
+		return nil, err
+	}
+	defer rtcSink.Close()
+
+	participant, err := s.roomManager.roomStore.GetParticipant(req.Target.Room, req.Target.Identity)
+	if err != nil {
+		return nil, err
+	}
+
+	err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
+		Message: &livekit.RTCNodeMessage_UpdateMetadata{
+			UpdateMetadata: req,
+		},
+	})
+
+	participant.Metadata = req.Metadata
+	return participant, nil
+}
+
+func (s *RoomService) createRTCSink(ctx context.Context, room, identity string) (routing.MessageSink, error) {
+	if err := EnsureAdminPermission(ctx, room); err != nil {
+		return nil, twirpAuthError(err)
+	}
+
+	_, err := s.roomManager.roomStore.GetParticipant(room, identity)
+	if err != nil {
+		return nil, err
+	}
+
+	return s.roomManager.router.CreateRTCSink(room, identity)
+}

pkg/service/rtcservice.go

@@ -56,14 +56,24 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	roomName := r.FormValue("room")
 	reconnectParam := r.FormValue("reconnect")
-	isReconnect := reconnectParam == "1" || reconnectParam == "true"
+
 	claims := GetGrants(r.Context())
 	// require a claim
 	if claims == nil || claims.Video == nil {
 		handleError(w, http.StatusUnauthorized, rtc.ErrPermissionDenied.Error())
 		return
 	}
-	identity := claims.Identity
+	pi := routing.ParticipantInit{
+		Reconnect: reconnectParam == "1" || reconnectParam == "true",
+		Identity:  claims.Identity,
+	}
+	// only use permissions if any of them are set, default permissive
+	if claims.Video.CanPublish || claims.Video.CanSubscribe {
+		pi.Permission = &livekit.ParticipantPermission{
+			CanSubscribe: claims.Video.CanSubscribe,
+			CanPublish:   claims.Video.CanPublish,
+		}
+	}
 
 	onlyName, err := EnsureJoinPermission(r.Context())
 	if err != nil {
@@ -81,17 +91,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var metadata string
 	if claims.Metadata != nil {
 		if data, err := json.Marshal(claims.Metadata); err != nil {
 			logger.Warnw("unable to encode metadata", "error", err)
 		} else {
-			metadata = string(data)
+			pi.Metadata = string(data)
 		}
 	}
 
 	// this needs to be started first *before* using router functions on this node
-	connId, reqSink, resSource, err := s.router.StartParticipantSignal(roomName, identity, metadata, isReconnect)
+	connId, reqSink, resSource, err := s.router.StartParticipantSignal(roomName, pi)
 	if err != nil {
 		handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error())
 		return
@@ -100,7 +109,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	done := make(chan bool, 1)
 	// function exits when websocket terminates, it'll close the event reading off of response sink as well
 	defer func() {
-		logger.Infow("WS connection closed", "participant", identity, "connectionId", connId)
+		logger.Infow("WS connection closed", "participant", pi.Identity, "connectionId", connId)
 		reqSink.Close()
 		close(done)
 	}()
@@ -120,7 +129,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		"connectionId", connId,
 		"room", rm.Sid,
 		"roomName", rm.Name,
-		"name", identity,
+		"name", pi.Identity,
 	)
 
 	// handle responses
@@ -137,7 +146,8 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			case msg := <-resSource.ReadChan():
 				if msg == nil {
-					logger.Infow("source closed connection", "participant", identity,
+					logger.Infow("source closed connection",
+						"participant", pi.Identity,
 						"connectionId", connId)
 					return
 				}
@@ -145,7 +155,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				if !ok {
 					logger.Errorw("unexpected message type",
 						"type", fmt.Sprintf("%T", msg),
-						"participant", identity,
+						"participant", pi.Identity,
 						"connectionId", connId)
 					continue
 				}
@@ -172,8 +182,10 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 		if err := reqSink.WriteMessage(req); err != nil {
-			logger.Warnw("error writing to request sink", "error", err,
-				"participant", identity, "connectionId", connId)
+			logger.Warnw("error writing to request sink",
+				"error", err,
+				"participant", pi.Identity,
+				"connectionId", connId)
 		}
 	}
 }