DeleteRoom now removes all participants in the room

pkg/routing/errors.go

@@ -7,6 +7,7 @@ var (
 	ErrHandlerNotDefined    = errors.New("handler not defined")
 	ErrNoAvailableNodes     = errors.New("could not find any available nodes")
 	ErrIncorrectRTCNode     = errors.New("current node isn't the RTC node for the room")
+	ErrNodeNotFound         = errors.New("could not locate the node")
 	errInvalidRouterMessage = errors.New("invalid router message")
 	ErrChannelClosed        = errors.New("channel closed")
 	ErrChannelFull          = errors.New("channel is full")

pkg/routing/redisrouter.go

@@ -273,11 +273,19 @@ func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) erro
 }
 
 func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) {
-	return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result()
+	val, err := r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result()
+	if err == redis.Nil {
+		err = ErrNodeNotFound
+	}
+	return val, err
 }
 
 func (r *RedisRouter) getParticipantSignalNode(connectionId string) (nodeId string, err error) {
-	return r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result()
+	val, err := r.rc.Get(r.ctx, participantSignalKey(connectionId)).Result()
+	if err == redis.Nil {
+		err = ErrNodeNotFound
+	}
+	return val, err
 }
 
 // update node stats and cleanup

pkg/service/redisroomstore.go

@@ -12,13 +12,13 @@ import (
 )
 
 const (
-	// hash of room_name => Room proto
+	// RoomsKey is hash of room_name => Room proto
 	RoomsKey = "rooms"
 
-	// hash of room_id => room name
+	// RoomIdMap is hash of room_id => room name
 	RoomIdMap = "room_id_map"
 
-	// hash of participant_name => ParticipantInfo
+	// RoomParticipantsPrefix is hash of participant_name => ParticipantInfo
 	// a key for each room, with expiration
 	RoomParticipantsPrefix = "room_participants:"
 

pkg/service/roommanager.go

@@ -395,6 +395,11 @@ func (r *RoomManager) handleRTCMessage(roomName, identity string, msg *livekit.R
 		if rm.UpdateParticipant.Permission != nil {
 			participant.SetPermission(rm.UpdateParticipant.Permission)
 		}
+	case *livekit.RTCNodeMessage_DeleteRoom:
+		for _, p := range room.GetParticipants() {
+			_ = p.Close()
+		}
+		room.Close()
 	}
 }
 

pkg/service/roomservice.go

@@ -54,17 +54,38 @@ func (s *RoomService) ListRooms(ctx context.Context, req *livekit.ListRoomsReque
 	return
 }
 
-func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomRequest) (res *livekit.DeleteRoomResponse, err error) {
-	if err = EnsureCreatePermission(ctx); err != nil {
+func (s *RoomService) DeleteRoom(ctx context.Context, req *livekit.DeleteRoomRequest) (*livekit.DeleteRoomResponse, error) {
+	if err := EnsureCreatePermission(ctx); err != nil {
 		return nil, twirpAuthError(err)
 	}
-	err = s.roomManager.DeleteRoom(req.Room)
+	// if the room is currently active, RTC node needs to disconnect clients
+	// here we are using any user's identity, due to how it works with routing
+	participants, err := s.roomManager.roomStore.ListParticipants(req.Room)
 	if err != nil {
-		err = twirp.WrapError(twirp.InternalError("could not delete room"), err)
-		return
+		return nil, err
 	}
-	res = &livekit.DeleteRoomResponse{}
-	return
+
+	if len(participants) > 0 {
+		rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participants[0].Identity)
+		if err != nil {
+			return nil, err
+		}
+
+		defer rtcSink.Close()
+		_ = rtcSink.WriteMessage(&livekit.RTCNodeMessage{
+			Message: &livekit.RTCNodeMessage_DeleteRoom{
+				DeleteRoom: req,
+			},
+		})
+	} else {
+		// if a room hasn't started, delete locally
+		if err = s.roomManager.DeleteRoom(req.Room); err != nil {
+			err = twirp.WrapError(twirp.InternalError("could not delete room"), err)
+			return nil, err
+		}
+	}
+
+	return &livekit.DeleteRoomResponse{}, nil
 }
 
 func (s *RoomService) ListParticipants(ctx context.Context, req *livekit.ListParticipantsRequest) (res *livekit.ListParticipantsResponse, err error) {