merge from develop

server/lib/Lobby.js

@@ -47,7 +47,8 @@ class Lobby extends EventEmitter
 		return Object.values(this._peers).map((peer) =>
 			({
 				peerId      : peer.id,
-				displayName : peer.displayName 
+				displayName : peer.displayName,
+				picture     : peer.picture
 			}));
 	}
 
@@ -62,8 +63,8 @@ class Lobby extends EventEmitter
 
 		for (const peer in this._peers)
 		{
-			if (peer.socket)
-				this.promotePeer(peer.id);
+			if (!this._peers[peer].closed)
+				this.promotePeer(peer);
 		}
 	}
 

server/lib/Peer.js

@@ -23,9 +23,15 @@ class Peer extends EventEmitter
 
 		this._joined = false;
 
+		this._joinedTimestamp = null;
+
 		this._inLobby = false;
 
-		this._roles = [ userRoles.ALL ];
+		this._authenticated = false;
+
+		this._authenticatedTimestamp = null;
+
+		this._roles = [ userRoles.NORMAL ];
 
 		this._displayName = false;
 
@@ -33,10 +39,14 @@ class Peer extends EventEmitter
 
 		this._email = null;
 
+		this._routerId = null;
+
 		this._rtpCapabilities = null;
 
 		this._raisedHand = false;
 
+		this._raisedHandTimestamp = null;
+
 		this._transports = new Map();
 
 		this._producers = new Map();
@@ -133,9 +143,18 @@ class Peer extends EventEmitter
 
 	set joined(joined)
 	{
+		joined ?
+			this._joinedTimestamp = Date.now() :
+			this._joinedTimestamp = null;
+
 		this._joined = joined;
 	}
 
+	get joinedTimestamp()
+	{
+		return this._joinedTimestamp;
+	}
+
 	get inLobby()
 	{
 		return this._inLobby;
@@ -146,6 +165,32 @@ class Peer extends EventEmitter
 		this._inLobby = inLobby;
 	}
 
+	get authenticated()
+	{
+		return this._authenticated;
+	}
+
+	set authenticated(authenticated)
+	{
+		if (authenticated !== this._authenticated)
+		{
+			authenticated ?
+				this._authenticatedTimestamp = Date.now() :
+				this._authenticatedTimestamp = null;
+
+			const oldAuthenticated = this._authenticated;
+
+			this._authenticated = authenticated;
+
+			this.emit('authenticationChanged', { oldAuthenticated });
+		}
+	}
+
+	get authenticatedTimestamp()
+	{
+		return this._authenticatedTimestamp;
+	}
+
 	get roles()
 	{
 		return this._roles;
@@ -195,6 +240,16 @@ class Peer extends EventEmitter
 		this._email = email;
 	}
 
+	get routerId()
+	{
+		return this._routerId;
+	}
+
+	set routerId(routerId)
+	{
+		this._routerId = routerId;
+	}
+
 	get rtpCapabilities()
 	{
 		return this._rtpCapabilities;
@@ -212,9 +267,18 @@ class Peer extends EventEmitter
 
 	set raisedHand(raisedHand)
 	{
+		raisedHand ?
+			this._raisedHandTimestamp = Date.now() :
+			this._raisedHandTimestamp = null;
+
 		this._raisedHand = raisedHand;
 	}
 
+	get raisedHandTimestamp()
+	{
+		return this._raisedHandTimestamp;
+	}
+
 	get transports()
 	{
 		return this._transports;
@@ -314,10 +378,12 @@ class Peer extends EventEmitter
 	{
 		const peerInfo =
 		{
-			id          : this.id,
-			displayName : this.displayName,
-			picture     : this.picture,
-			roles       : this.roles
+			id                  : this.id,
+			displayName         : this.displayName,
+			picture             : this.picture,
+			roles               : this.roles,
+			raisedHand          : this.raisedHand,
+			raisedHandTimestamp : this.raisedHandTimestamp
 		};
 
 		return peerInfo;

server/lib/Room.js

@@ -2,11 +2,37 @@ const EventEmitter = require('events').EventEmitter;
 const axios = require('axios');
 const Logger = require('./Logger');
 const Lobby = require('./Lobby');
+const { v4: uuidv4 } = require('uuid');
+const jwt = require('jsonwebtoken');
 const userRoles = require('../userRoles');
 const config = require('../config/config');
 
 const logger = new Logger('Room');
 
+// In case they are not configured properly
+const accessFromRoles =
+{
+	BYPASS_ROOM_LOCK : [ userRoles.ADMIN ],
+	BYPASS_LOBBY     : [ userRoles.NORMAL ],
+	...config.accessFromRoles
+};
+
+const permissionsFromRoles =
+{
+	CHANGE_ROOM_LOCK : [ userRoles.NORMAL ],
+	PROMOTE_PEER     : [ userRoles.NORMAL ],
+	SEND_CHAT        : [ userRoles.NORMAL ],
+	MODERATE_CHAT    : [ userRoles.MODERATOR ],
+	SHARE_SCREEN     : [ userRoles.NORMAL ],
+	EXTRA_VIDEO      : [ userRoles.NORMAL ],
+	SHARE_FILE       : [ userRoles.NORMAL ],
+	MODERATE_FILES   : [ userRoles.MODERATOR ],
+	MODERATE_ROOM    : [ userRoles.MODERATOR ],
+	...config.permissionsFromRoles
+};
+
+const ROUTER_SCALE_SIZE = config.routerScaleSize || 40;
+
 class Room extends EventEmitter
 {
 	/**
@@ -14,38 +40,57 @@ class Room extends EventEmitter
 	 *
 	 * @async
 	 *
-	 * @param {mediasoup.Worker} mediasoupWorker - The mediasoup Worker in which a new
+	 * @param {mediasoup.Worker} mediasoupWorkers - The mediasoup Worker in which a new
 	 *   mediasoup Router must be created.
 	 * @param {String} roomId - Id of the Room instance.
 	 */
-	static async create({ mediasoupWorker, roomId })
+	static async create({ mediasoupWorkers, roomId })
 	{
 		logger.info('create() [roomId:"%s"]', roomId);
 
+		// Shuffle workers to get random cores
+		let shuffledWorkers = mediasoupWorkers.sort(() => Math.random() - 0.5);
+
 		// Router media codecs.
 		const mediaCodecs = config.mediasoup.router.mediaCodecs;
 
-		// Create a mediasoup Router.
-		const mediasoupRouter = await mediasoupWorker.createRouter({ mediaCodecs });
+		const mediasoupRouters = new Map();
 
-		// Create a mediasoup AudioLevelObserver.
-		const audioLevelObserver = await mediasoupRouter.createAudioLevelObserver(
+		let firstRouter = null;
+
+		for (const worker of shuffledWorkers)
+		{
+			const router = await worker.createRouter({ mediaCodecs });
+
+			if (!firstRouter)
+				firstRouter = router;
+
+			mediasoupRouters.set(router.id, router);
+		}
+
+		// Create a mediasoup AudioLevelObserver on first router
+		const audioLevelObserver = await firstRouter.createAudioLevelObserver(
 			{
 				maxEntries : 1,
 				threshold  : -80,
 				interval   : 800
 			});
 
-		return new Room({ roomId, mediasoupRouter, audioLevelObserver });
+		firstRouter = null;
+		shuffledWorkers = null;
+
+		return new Room({ roomId, mediasoupRouters, audioLevelObserver });
 	}
 
-	constructor({ roomId, mediasoupRouter, audioLevelObserver })
+	constructor({ roomId, mediasoupRouters, audioLevelObserver })
 	{
 		logger.info('constructor() [roomId:"%s"]', roomId);
 
 		super();
 		this.setMaxListeners(Infinity);
 
+		this._uuid = uuidv4();
+
 		// Room ID.
 		this._roomId = roomId;
 
@@ -55,12 +100,6 @@ class Room extends EventEmitter
 		// Locked flag.
 		this._locked = false;
 
-		// Required roles to access
-		this._requiredRoles = [ userRoles.ALL ];
-
-		if ('requiredRolesForAccess' in config)
-			this._requiredRoles = config.requiredRolesForAccess;
-
 		// if true: accessCode is a possibility to open the room
 		this._joinByAccesCode = true;
 
@@ -78,8 +117,13 @@ class Room extends EventEmitter
 
 		this._peers = {};
 
-		// mediasoup Router instance.
-		this._mediasoupRouter = mediasoupRouter;
+		// Array of mediasoup Router instances.
+		this._mediasoupRouters = mediasoupRouters;
+
+		// The router we are currently putting peers in
+		this._routerIterator = this._mediasoupRouters.values();
+
+		this._currentRouter = this._routerIterator.next().value;
 
 		// mediasoup AudioLevelObserver.
 		this._audioLevelObserver = audioLevelObserver;
@@ -102,8 +146,14 @@ class Room extends EventEmitter
 
 		this._closed = true;
 
+		this._chatHistory = null;
+
+		this._fileHistory = null;
+
 		this._lobby.close();
 
+		this._lobby = null;
+
 		// Close the peers.
 		for (const peer in this._peers)
 		{
@@ -113,41 +163,86 @@ class Room extends EventEmitter
 
 		this._peers = null;
 
-		// Close the mediasoup Router.
-		this._mediasoupRouter.close();
+		// Close the mediasoup Routers.
+		for (const router of this._mediasoupRouters.values())
+		{
+			router.close();
+		}
+
+		this._routerIterator = null;
+
+		this._currentRouter = null;
+
+		this._mediasoupRouters.clear();
+
+		this._audioLevelObserver = null;
 
 		// Emit 'close' event.
 		this.emit('close');
 	}
 
-	handlePeer(peer)
+	verifyPeer({ id, token })
 	{
-		logger.info('handlePeer() [peer:"%s", roles:"%s"]', peer.id, peer.roles);
+		try
+		{
+			const decoded = jwt.verify(token, this._uuid);
 
-		// Allow reconnections, remove old peer
+			logger.info('verifyPeer() [decoded:"%o"]', decoded);
+
+			return decoded.id === id;
+		}
+		catch (err)
+		{
+			logger.warn('verifyPeer() | invalid token');
+		}
+
+		return false;
+	}
+
+	handlePeer({ peer, returning })
+	{
+		logger.info('handlePeer() [peer:"%s", roles:"%s", returning:"%s"]', peer.id, peer.roles, returning);
+
+		// Should not happen
 		if (this._peers[peer.id])
 		{
 			logger.warn(
 				'handleConnection() | there is already a peer with same peerId [peer:"%s"]',
 				peer.id);
-
-			this._peers[peer.id].close();
 		}
 
-		// Always let ADMIN in, even if locked
-		if (peer.roles.includes(userRoles.ADMIN))
+		// Returning user
+		if (returning)
+			this._peerJoining(peer, true);
+		else if ( // Has a role that is allowed to bypass room lock
+			peer.roles.some((role) => accessFromRoles.BYPASS_ROOM_LOCK.includes(role))
+		)
 			this._peerJoining(peer);
+		else if (
+			'maxUsersPerRoom' in config &&
+			(
+				Object.keys(this._peers).length +
+				this._lobby.peerList().length
+			) >= config.maxUsersPerRoom)
+		{
+			this._handleOverRoomLimit(peer);
+		}
 		else if (this._locked)
 			this._parkPeer(peer);
 		else
 		{
-			// If the user has a role in config.requiredRolesForAccess, let them in
-			peer.roles.some((role) => this._requiredRoles.includes(role)) ?
+			// Has a role that is allowed to bypass lobby
+			peer.roles.some((role) => accessFromRoles.BYPASS_LOBBY.includes(role)) ?
 				this._peerJoining(peer) :
 				this._handleGuest(peer);
 		}
 	}
 
+	_handleOverRoomLimit(peer)
+	{
+		this._notification(peer.socket, 'overRoomLimit');
+	}
+
 	_handleGuest(peer)
 	{
 		if (config.activateOnHostJoin && !this.checkEmpty())
@@ -169,7 +264,11 @@ class Room extends EventEmitter
 
 			this._peerJoining(promotedPeer);
 
-			for (const peer of this._getJoinedPeers())
+			for (
+				const peer of this._getPeersWithPermission({
+					permission : permissionsFromRoles.PROMOTE_PEER
+				})
+			)
 			{
 				this._notification(peer.socket, 'lobby:promotedPeer', { peerId: id });
 			}
@@ -177,18 +276,18 @@ class Room extends EventEmitter
 
 		this._lobby.on('peerRolesChanged', (peer) =>
 		{
-			// Always let admin in, even if locked
-			if (peer.roles.includes(userRoles.ADMIN))
+			if ( // Has a role that is allowed to bypass room lock
+				peer.roles.some((role) => accessFromRoles.BYPASS_ROOM_LOCK.includes(role))
+			)
 			{
 				this._lobby.promotePeer(peer.id);
 
 				return;
 			}
 
-			// If the user has a role in config.requiredRolesForAccess, let them in
-			if (
+			if ( // Has a role that is allowed to bypass lobby
 				!this._locked &&
-				peer.roles.some((role) => this._requiredRoles.includes(role))
+				peer.roles.some((role) => accessFromRoles.BYPASS_LOBBY.includes(role))
 			)
 			{
 				this._lobby.promotePeer(peer.id);
@@ -201,7 +300,11 @@ class Room extends EventEmitter
 		{
 			const { id, displayName } = changedPeer;
 
-			for (const peer of this._getJoinedPeers())
+			for (
+				const peer of this._getPeersWithPermission({
+					permission : permissionsFromRoles.PROMOTE_PEER
+				})
+			)
 			{
 				this._notification(peer.socket, 'lobby:changeDisplayName', { peerId: id, displayName });
 			}
@@ -211,7 +314,11 @@ class Room extends EventEmitter
 		{
 			const { id, picture } = changedPeer;
 
-			for (const peer of this._getJoinedPeers())
+			for (
+				const peer of this._getPeersWithPermission({
+					permission : permissionsFromRoles.PROMOTE_PEER
+				})
+			)
 			{
 				this._notification(peer.socket, 'lobby:changePicture', { peerId: id, picture });
 			}
@@ -223,7 +330,11 @@ class Room extends EventEmitter
 
 			const { id } = closedPeer;
 
-			for (const peer of this._getJoinedPeers())
+			for (
+				const peer of this._getPeersWithPermission({
+					permission : permissionsFromRoles.PROMOTE_PEER
+				})
+			)
 			{
 				this._notification(peer.socket, 'lobby:peerClosed', { peerId: id });
 			}
@@ -326,13 +437,17 @@ class Room extends EventEmitter
 	{
 		this._lobby.parkPeer(parkPeer);
 
-		for (const peer of this._getJoinedPeers())
+		for (
+			const peer of this._getPeersWithPermission({
+				permission : permissionsFromRoles.PROMOTE_PEER
+			})
+		)
 		{
 			this._notification(peer.socket, 'parkedPeer', { peerId: parkPeer.id });
 		}
 	}
 
-	async _peerJoining(peer)
+	async _peerJoining(peer, returning = false)
 	{
 		peer.socket.join(this._roomId);
 
@@ -341,47 +456,60 @@ class Room extends EventEmitter
 
 		this._peers[peer.id] = peer;
 
+		// Assign routerId
+		peer.routerId = await this._getRouterId();
+
 		this._handlePeer(peer);
 
-		let turnServers;
-
-		if ('turnAPIURI' in config)
+		if (returning)
 		{
-			try
-			{
-				const { data } = await axios.get(
-					config.turnAPIURI,
-					{
-						params : {
-							'uri_schema'  : 'turn',
-							'transport'   : 'tcp',
-							'ip_ver'      : 'ipv4',
-							'servercount' : '2',
-							'api_key'     : config.turnAPIKey,
-							'ip'          : peer.socket.request.connection.remoteAddress
-						}
-					});
-	
-				turnServers = [ {
-					urls       : data.uris,
-					username   : data.username,
-					credential : data.password
-				} ];
-			}
-			catch (error)
-			{
-				if ('backupTurnServers' in config)
-					turnServers = config.backupTurnServers;
-	
-				logger.error('_peerJoining() | error on REST turn [error:"%o"]', error);
-			}
+			this._notification(peer.socket, 'roomBack');
 		}
-		else if ('backupTurnServers' in config)
+		else
 		{
-			turnServers = config.backupTurnServers;
-		}
+			const token = jwt.sign({ id: peer.id }, this._uuid, { noTimestamp: true });
 
-		this._notification(peer.socket, 'roomReady', { turnServers });
+			peer.socket.handshake.session.token = token;
+
+			peer.socket.handshake.session.save();
+
+			let turnServers;
+	
+			if ('turnAPIURI' in config)
+			{
+				try
+				{
+					const { data } = await axios.get(
+						config.turnAPIURI,
+						{
+							params : {
+								...config.turnAPIparams,
+								'api_key' : config.turnAPIKey,
+								'ip'      : peer.socket.request.connection.remoteAddress
+							}
+						});
+		
+					turnServers = [ {
+						urls       : data.uris,
+						username   : data.username,
+						credential : data.password
+					} ];
+				}
+				catch (error)
+				{
+					if ('backupTurnServers' in config)
+						turnServers = config.backupTurnServers;
+		
+					logger.error('_peerJoining() | error on REST turn [error:"%o"]', error);
+				}
+			}
+			else if ('backupTurnServers' in config)
+			{
+				turnServers = config.backupTurnServers;
+			}
+	
+			this._notification(peer.socket, 'roomReady', { turnServers });
+		}
 	}
 
 	_handlePeer(peer)
@@ -461,6 +589,17 @@ class Room extends EventEmitter
 				peerId : peer.id,
 				role   : newRole
 			}, true, true);
+
+			// Got permission to promote peers, notify peer of
+			// peers in lobby
+			if (permissionsFromRoles.PROMOTE_PEER.includes(newRole))
+			{
+				const lobbyPeers = this._lobby.peerList();
+
+				lobbyPeers.length > 0 && this._notification(peer.socket, 'parkedPeers', {
+					lobbyPeers
+				});
+			}
 		});
 
 		peer.on('lostRole', ({ oldRole }) =>
@@ -479,11 +618,14 @@ class Room extends EventEmitter
 
 	async _handleSocketRequest(peer, request, cb)
 	{
+		const router =
+			this._mediasoupRouters.get(peer.routerId);
+
 		switch (request.method)
 		{
 			case 'getRouterRtpCapabilities':
 			{
-				cb(null, this._mediasoupRouter.rtpCapabilities);
+				cb(null, router.rtpCapabilities);
 
 				break;
 			}
@@ -517,9 +659,21 @@ class Room extends EventEmitter
 					.filter((joinedPeer) => joinedPeer.id !== peer.id)
 					.map((joinedPeer) => (joinedPeer.peerInfo));
 
+				const lobbyPeers = this._lobby.peerList();
+
 				cb(null, {
-					roles : peer.roles,
-					peers : peerInfos
+					roles                : peer.roles,
+					peers                : peerInfos,
+					tracker              : config.fileTracker,
+					authenticated        : peer.authenticated,
+					permissionsFromRoles : permissionsFromRoles,
+					userRoles            : userRoles,
+					chatHistory          : this._chatHistory,
+					fileHistory          : this._fileHistory,
+					lastNHistory         : this._lastN,
+					locked               : this._locked,
+					lobbyPeers           : lobbyPeers,
+					accessCode           : this._accessCode
 				});
 
 				// Mark the new Peer as joined.
@@ -580,7 +734,7 @@ class Room extends EventEmitter
 					webRtcTransportOptions.enableTcp = true;
 				}
 
-				const transport = await this._mediasoupRouter.createWebRtcTransport(
+				const transport = await router.createWebRtcTransport(
 					webRtcTransportOptions
 				);
 
@@ -646,12 +800,27 @@ class Room extends EventEmitter
 
 			case 'produce':
 			{
+				let { appData } = request.data;
+
+				if (
+					appData.source === 'screen' &&
+					!peer.roles.some(
+						(role) => permissionsFromRoles.SHARE_SCREEN.includes(role))
+				)
+					throw new Error('peer not authorized');
+
+				if (
+					appData.source === 'extravideo' &&
+					!peer.roles.some(
+						(role) => permissionsFromRoles.EXTRA_VIDEO.includes(role))
+				)
+					throw new Error('peer not authorized');
+
 				// Ensure the Peer is joined.
 				if (!peer.joined)
 					throw new Error('Peer not yet joined');
 
 				const { transportId, kind, rtpParameters } = request.data;
-				let { appData } = request.data;
 				const transport = peer.getTransport(transportId);
 
 				if (!transport)
@@ -664,6 +833,19 @@ class Room extends EventEmitter
 				const producer =
 					await transport.produce({ kind, rtpParameters, appData });
 
+				const pipeRouters = this._getRoutersToPipeTo(peer.routerId);
+
+				for (const [ routerId, destinationRouter ] of this._mediasoupRouters)
+				{
+					if (pipeRouters.includes(routerId))
+					{
+						await router.pipeToRouter({
+							producerId : producer.id,
+							router     : destinationRouter
+						});
+					}
+				}
+
 				// Store the Producer into the Peer data Object.
 				peer.addProducer(producer.id, producer);
 
@@ -910,16 +1092,10 @@ class Room extends EventEmitter
 					throw new Error('Peer not yet joined');
 
 				const { displayName } = request.data;
-				const oldDisplayName = peer.displayName;
 
 				peer.displayName = displayName;
 
-				// Spread to others
-				this._notification(peer.socket, 'changeDisplayName', {
-					peerId         : peer.id,
-					displayName    : displayName,
-					oldDisplayName : oldDisplayName
-				}, true);
+				// This will be spread through events from the peer object
 
 				// Return no error
 				cb();
@@ -927,7 +1103,7 @@ class Room extends EventEmitter
 				break;
 			}
 
-			case 'changePicture':
+			/* case 'changePicture':
 			{
 				// Ensure the Peer is joined.
 				if (!peer.joined)
@@ -947,10 +1123,15 @@ class Room extends EventEmitter
 				cb();
 
 				break;
-			}
+			} */
 
 			case 'chatMessage':
 			{
+				if (
+					!peer.roles.some((role) => permissionsFromRoles.SEND_CHAT.includes(role))
+				)
+					throw new Error('peer not authorized');
+
 				const { chatMessage } = request.data;
 	
 				this._chatHistory.push(chatMessage);
@@ -967,28 +1148,35 @@ class Room extends EventEmitter
 				break;
 			}
 
-			case 'serverHistory':
+			case 'moderator:clearChat':
 			{
-				// Return to sender
-				const lobbyPeers = this._lobby.peerList();
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_CHAT.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+	
+				this._chatHistory = [];
 
-				cb(
-					null,
-					{
-						chatHistory  : this._chatHistory,
-						fileHistory  : this._fileHistory,
-						lastNHistory : this._lastN,
-						locked       : this._locked,
-						lobbyPeers   : lobbyPeers,
-						accessCode   : this._accessCode
-					}
-				);
+				// Spread to others
+				this._notification(peer.socket, 'moderator:clearChat', null, true);
+
+				// Return no error
+				cb();
 
 				break;
 			}
 
 			case 'lockRoom':
 			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.CHANGE_ROOM_LOCK.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+
 				this._locked = true;
 
 				// Spread to others
@@ -1004,6 +1192,13 @@ class Room extends EventEmitter
 
 			case 'unlockRoom':
 			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.CHANGE_ROOM_LOCK.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+
 				this._locked = false;
 
 				// Spread to others
@@ -1059,6 +1254,13 @@ class Room extends EventEmitter
 
 			case 'promotePeer':
 			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.PROMOTE_PEER.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+
 				const { peerId } = request.data;
 
 				this._lobby.promotePeer(peerId);
@@ -1071,6 +1273,13 @@ class Room extends EventEmitter
 
 			case 'promoteAllPeers':
 			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.PROMOTE_PEER.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+
 				this._lobby.promoteAllPeers();
 
 				// Return no error
@@ -1081,6 +1290,13 @@ class Room extends EventEmitter
 
 			case 'sendFile':
 			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.SHARE_FILE.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+
 				const { magnetUri } = request.data;
 	
 				this._fileHistory.push({ peerId: peer.id, magnetUri: magnetUri });
@@ -1097,16 +1313,37 @@ class Room extends EventEmitter
 				break;
 			}
 
-			case 'raiseHand':
+			case 'moderator:clearFileSharing':
+			{
+				if (
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_FILES.includes(role)
+					)
+				)
+					throw new Error('peer not authorized');
+	
+				this._fileHistory = [];
+
+				// Spread to others
+				this._notification(peer.socket, 'moderator:clearFileSharing', null, true);
+
+				// Return no error
+				cb();
+
+				break;
+			}
+
+			case 'raisedHand':
 			{
 				const { raisedHand } = request.data;
 
 				peer.raisedHand = raisedHand;
 
 				// Spread to others
-				this._notification(peer.socket, 'raiseHand', {
-					peerId     : peer.id,
-					raisedHand : raisedHand
+				this._notification(peer.socket, 'raisedHand', {
+					peerId              : peer.id,
+					raisedHand          : raisedHand,
+					raisedHandTimestamp : peer.raisedHandTimestamp
 				}, true);
 
 				// Return no error
@@ -1118,15 +1355,14 @@ class Room extends EventEmitter
 			case 'moderator:muteAll':
 			{
 				if (
-					!peer.hasRole(userRoles.MODERATOR) &&
-					!peer.hasRole(userRoles.ADMIN)
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_ROOM.includes(role)
+					)
 				)
-					throw new Error('peer does not have moderator priveleges');
+					throw new Error('peer not authorized');
 
 				// Spread to others
-				this._notification(peer.socket, 'moderator:mute', {
-					peerId : peer.id
-				}, true);
+				this._notification(peer.socket, 'moderator:mute', null, true);
 
 				cb();
 
@@ -1136,15 +1372,14 @@ class Room extends EventEmitter
 			case 'moderator:stopAllVideo':
 			{
 				if (
-					!peer.hasRole(userRoles.MODERATOR) &&
-					!peer.hasRole(userRoles.ADMIN)
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_ROOM.includes(role)
+					)
 				)
-					throw new Error('peer does not have moderator priveleges');
+					throw new Error('peer not authorized');
 
 				// Spread to others
-				this._notification(peer.socket, 'moderator:stopVideo', {
-					peerId : peer.id
-				}, true);
+				this._notification(peer.socket, 'moderator:stopVideo', null, true);
 
 				cb();
 
@@ -1154,17 +1389,13 @@ class Room extends EventEmitter
 			case 'moderator:closeMeeting':
 			{
 				if (
-					!peer.hasRole(userRoles.MODERATOR) &&
-					!peer.hasRole(userRoles.ADMIN)
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_ROOM.includes(role)
+					)
 				)
-					throw new Error('peer does not have moderator priveleges');
+					throw new Error('peer not authorized');
 
-				this._notification(
-					peer.socket,
-					'moderator:kick',
-					null,
-					true
-				);
+				this._notification(peer.socket, 'moderator:kick', null,	true);
 
 				cb();
 
@@ -1177,10 +1408,11 @@ class Room extends EventEmitter
 			case 'moderator:kickPeer':
 			{
 				if (
-					!peer.hasRole(userRoles.MODERATOR) &&
-					!peer.hasRole(userRoles.ADMIN)
+					!peer.roles.some(
+						(role) => permissionsFromRoles.MODERATE_ROOM.includes(role)
+					)
 				)
-					throw new Error('peer does not have moderator priveleges');
+					throw new Error('peer not authorized');
 
 				const { peerId } = request.data;
 
@@ -1189,10 +1421,7 @@ class Room extends EventEmitter
 				if (!kickPeer)
 					throw new Error(`peer with id "${peerId}" not found`);
 
-				this._notification(
-					kickPeer.socket,
-					'moderator:kick'
-				);
+				this._notification(kickPeer.socket, 'moderator:kick');
 
 				kickPeer.close();
 
@@ -1224,6 +1453,8 @@ class Room extends EventEmitter
 			producer.id
 		);
 
+		const router = this._mediasoupRouters.get(producerPeer.routerId);
+
 		// Optimization:
 		// - Create the server-side Consumer. If video, do it paused.
 		// - Tell its Peer about it and wait for its response.
@@ -1234,7 +1465,7 @@ class Room extends EventEmitter
 		// NOTE: Don't create the Consumer if the remote Peer cannot consume it.
 		if (
 			!consumerPeer.rtpCapabilities ||
-			!this._mediasoupRouter.canConsume(
+			!router.canConsume(
 				{
 					producerId      : producer.id,
 					rtpCapabilities : consumerPeer.rtpCapabilities
@@ -1369,6 +1600,19 @@ class Room extends EventEmitter
 			.filter((peer) => peer.joined && peer !== excludePeer);
 	}
 
+	_getPeersWithPermission({ permission = null, excludePeer = undefined, joined = true })
+	{
+		return Object.values(this._peers)
+			.filter(
+				(peer) =>
+					peer.joined === joined &&
+					peer !== excludePeer &&
+					peer.roles.some(
+						(role) => permission.includes(role)
+					)
+			);
+	}
+
 	_timeoutCallback(callback)
 	{
 		let called = false;
@@ -1433,6 +1677,84 @@ class Room extends EventEmitter
 			socket.emit('notification', { method, data });
 		}
 	}
+
+	async _pipeProducersToNewRouter()
+	{
+		const peersToPipe =
+			Object.values(this._peers)
+				.filter((peer) => peer.routerId !== this._currentRouter.id);
+
+		for (const peer of peersToPipe)
+		{
+			const srcRouter = this._mediasoupRouters.get(peer.routerId);
+
+			for (const producerId of peer.producers.keys())
+			{
+				await srcRouter.pipeToRouter({
+					producerId,
+					router : this._currentRouter
+				});
+			}
+		}
+	}
+
+	async _getRouterId()
+	{
+		if (this._currentRouter)
+		{
+			const routerLoad =
+				Object.values(this._peers)
+					.filter((peer) => peer.routerId === this._currentRouter.id).length;
+
+			if (routerLoad >= ROUTER_SCALE_SIZE)
+			{
+				this._currentRouter = this._routerIterator.next().value;
+
+				if (this._currentRouter)
+				{
+					await this._pipeProducersToNewRouter();
+
+					return this._currentRouter.id;
+				}
+			}
+			else
+			{
+				return this._currentRouter.id;
+			}
+		}
+
+		return this._getLeastLoadedRouter();
+	}
+
+	// Returns an array of router ids we need to pipe to
+	_getRoutersToPipeTo(originRouterId)
+	{
+		return Object.values(this._peers)
+			.map((peer) => peer.routerId)
+			.filter((routerId, index, self) =>
+				routerId !== originRouterId && self.indexOf(routerId) === index
+			);
+	}
+
+	_getLeastLoadedRouter()
+	{
+		let load = Infinity;
+		let id;
+
+		for (const routerId of this._mediasoupRouters.keys())
+		{
+			const routerLoad = 
+				Object.values(this._peers).filter((peer) => peer.routerId === routerId).length;
+
+			if (routerLoad < load)
+			{
+				id = routerId;
+				load = routerLoad;
+			}
+		}
+
+		return id;
+	}
 }
 
 module.exports = Room;

server/lib/promExporter.js

@@ -0,0 +1,284 @@
+const { Resolver } = require('dns').promises;
+const express = require('express');
+const mediasoup = require('mediasoup');
+const prom = require('prom-client');
+
+const Logger = require('./Logger');
+
+const logger = new Logger('prom');
+const resolver = new Resolver();
+const workers = new Map();
+
+const labelNames = [
+	'pid', 'room_id', 'peer_id', 'display_name', 'user_agent', 'transport_id',
+	'proto', 'local_addr', 'remote_addr', 'id', 'kind', 'codec', 'type'
+];
+
+const metadata = {
+	'byteCount' : { metricType: prom.Counter, unit: 'bytes' },
+	'score'     : { metricType: prom.Gauge }
+};
+
+module.exports = async function(rooms, peers, config)
+{
+	const collect = async function(registry)
+	{
+		const newMetrics = function(subsystem)
+		{
+			const namespace = 'mediasoup';
+			const metrics = new Map();
+
+			for (const key in metadata)
+			{
+				if (Object.prototype.hasOwnProperty.call(metadata, key))
+				{
+					const value = metadata[key];
+					const name = key.split(/(?=[A-Z])/).join('_')
+						.toLowerCase();
+					const unit = value.unit;
+					const metricType = value.metricType;
+					let s = `${namespace}_${subsystem}_${name}`;
+
+					if (unit)
+					{
+						s += `_${unit}`;
+					}
+					const m = new metricType({
+						name : s, help : `${subsystem}.${key}`, labelNames : labelNames, registers : [ registry ] });
+
+					metrics.set(key, m);
+				}
+			}
+
+			return metrics;
+		};
+
+		const commonLabels = function(both, fn)
+		{
+			for (const roomId of rooms.keys())
+			{
+				for (const [ peerId, peer ] of peers)
+				{
+					if (fn(peer))
+					{
+						const displayName = peer._displayName;
+						const userAgent = peer._socket.client.request.headers['user-agent'];
+						const kind = both.kind;
+						const codec = both.rtpParameters.codecs[0].mimeType.split('/')[1];
+
+						return { roomId, peerId, displayName, userAgent, kind, codec };
+					}
+				}
+			}
+			throw new Error('cannot find common labels');
+		};
+
+		const addr = async function(ip, port)
+		{
+			if (config.deidentify)
+			{
+				const a = ip.split('.');
+
+				for (let i = 0; i < a.length - 2; i++)
+				{
+					a[i] = 'xx';
+				}
+
+				return `${a.join('.')}:${port}`;
+			}
+			else if (config.numeric)
+			{
+				return `${ip}:${port}`;
+			}
+			else
+			{
+				try
+				{
+					const a = await resolver.reverse(ip);
+
+					ip = a[0];
+				}
+				catch (err)
+				{
+					logger.error(`reverse DNS query failed: ${ip} ${err.code}`);
+				}
+
+				return `${ip}:${port}`;
+			}
+		};
+
+		const quiet = function(s)
+		{
+			return config.quiet ? '' : s;
+		};
+
+		const setValue = function(key, m, labels, v)
+		{
+			logger.debug(`setValue key=${key} v=${v}`);
+			switch (metadata[key].metricType)
+			{
+				case prom.Counter:
+					m.inc(labels, v);
+					break;
+				case prom.Gauge:
+					m.set(labels, v);
+					break;
+				default:
+					throw new Error(`unexpected metric: ${m}`);
+			}
+		};
+
+		logger.debug('collect');
+		const mRooms = new prom.Gauge({ name: 'edumeet_rooms', help: '#rooms', registers: [ registry ] });
+
+		mRooms.set(rooms.size);
+		const mPeers = new prom.Gauge({ name: 'edumeet_peers', help: '#peers', labelNames: [ 'room_id' ], registers: [ registry ] });
+
+		for (const [ roomId, room ] of rooms)
+		{
+			mPeers.labels(roomId).set(Object.keys(room._peers).length);
+		}
+
+		const mConsumer = newMetrics('consumer');
+		const mProducer = newMetrics('producer');
+
+		for (const [ pid, worker ] of workers)
+		{
+			logger.debug(`visiting worker ${pid}`);
+			for (const router of worker._routers)
+			{
+				logger.debug(`visiting router ${router.id}`);
+				for (const [ transportId, transport ] of router._transports)
+				{
+					logger.debug(`visiting transport ${transportId}`);
+					const transportJson = await transport.dump();
+
+					if (transportJson.iceState != 'completed')
+					{
+						logger.debug(`skipping transport ${transportId}}: ${transportJson.iceState}`);
+						continue;
+					}
+					const iceSelectedTuple = transportJson.iceSelectedTuple;
+					const proto = iceSelectedTuple.protocol;
+					const localAddr = await addr(iceSelectedTuple.localIp,
+						iceSelectedTuple.localPort);
+					const remoteAddr = await addr(iceSelectedTuple.remoteIp,
+						iceSelectedTuple.remotePort);
+
+					for (const [ producerId, producer ] of transport._producers)
+					{
+						logger.debug(`visiting producer ${producerId}`);
+						const { roomId, peerId, displayName, userAgent, kind, codec } = 
+							commonLabels(producer, (peer) => peer._producers.has(producerId));
+						const a = await producer.getStats();
+
+						for (const x of a)
+						{
+							const type = x.type;
+							const labels = {
+								'pid'          : pid,
+								'room_id'      : roomId,
+								'peer_id'      : peerId,
+								'display_name' : displayName,
+								'user_agent'   : userAgent,
+								'transport_id' : quiet(transportId),
+								'proto'        : proto,
+								'local_addr'   : localAddr,
+								'remote_addr'  : remoteAddr,
+								'id'           : quiet(producerId),
+								'kind'         : kind,
+								'codec'        : codec,
+								'type'         : type
+							};
+
+							for (const [ key, m ] of mProducer)
+							{
+								setValue(key, m, labels, x[key]);
+							}
+						}
+					}
+					for (const [ consumerId, consumer ] of transport._consumers)
+					{
+						logger.debug(`visiting consumer ${consumerId}`);
+						const { roomId, peerId, displayName, userAgent, kind, codec } = 
+							commonLabels(consumer, (peer) => peer._consumers.has(consumerId));
+						const a = await consumer.getStats();
+
+						for (const x of a)
+						{
+							if (x.type == 'inbound-rtp')
+							{
+								continue;
+							}
+							const type = x.type;
+							const labels =
+							{
+								'pid'          : pid,
+								'room_id'      : roomId,
+								'peer_id'      : peerId,
+								'display_name' : displayName,
+								'user_agent'   : userAgent,
+								'transport_id' : quiet(transportId),
+								'proto'        : proto,
+								'local_addr'   : localAddr,
+								'remote_addr'  : remoteAddr,
+								'id'           : quiet(consumerId),
+								'kind'         : kind,
+								'codec'        : codec,
+								'type'         : type
+							};
+
+							for (const [ key, m ] of mConsumer)
+							{
+								setValue(key, m, labels, x[key]);
+							}
+						}
+					}
+				}
+			}
+		}
+	};
+
+	try
+	{
+		logger.debug(`config.deidentify=${config.deidentify}`);
+		logger.debug(`config.numeric=${config.numeric}`);
+		logger.debug(`config.port=${config.port}`);
+		logger.debug(`config.quiet=${config.quiet}`);
+
+		mediasoup.observer.on('newworker', (worker) =>
+		{
+			logger.debug(`observing newworker ${worker.pid} #${workers.size}`);
+			workers.set(worker.pid, worker);
+			worker.observer.on('close', () =>
+			{
+				logger.debug(`observing close worker ${worker.pid} #${workers.size - 1}`);
+				workers.delete(worker.pid);
+			});
+		});
+
+		const app = express();
+
+		app.get('/', async (req, res) =>
+		{
+			logger.debug(`GET ${req.originalUrl}`);
+			const registry = new prom.Registry();
+
+			await collect(registry);
+			res.set('Content-Type', registry.contentType);
+			const data = registry.metrics();
+
+			res.end(data);
+		});
+		const server = app.listen(config.port || 8889, () =>
+		{
+			const address = server.address();
+
+			logger.info(`listening ${address.address}:${address.port}`);
+		});
+	}
+	catch (err)
+	{
+		logger.error(err);
+	}
+};