[IMPROVE] Unify voip streams into single stream

.gitignore

@@ -40,3 +40,5 @@ yarn-error.log*
 !.yarn/releases
 !.yarn/sdks
 !.yarn/versions
+
+.nvmrc

apps/meteor/client/lib/voip/QueueAggregator.ts

@@ -39,7 +39,7 @@ export class QueueAggregator {
 	}
 
 	private updateQueueInfo(queueName: string, queuedCalls: number): void {
-		if (!this.currentQueueMembershipStatus[queueName]) {
+		if (!this.currentQueueMembershipStatus?.[queueName]) {
 			// something is wrong. Queue is not found in the membership details.
 			return;
 		}
@@ -48,14 +48,14 @@ export class QueueAggregator {
 
 	setMembership(subscription: IQueueMembershipSubscription): void {
 		this.extension = subscription.extension;
-		for (let i = 0; i < subscription.queues.length; i++) {
-			const queue = subscription.queues[i];
+
+		subscription.queues.forEach((queue) => {
 			const queueInfo: IQueueInfo = {
 				queueName: queue.name,
 				callsInQueue: 0,
 			};
 			this.currentQueueMembershipStatus[queue.name] = queueInfo;
-		}
+		});
 	}
 
 	queueJoined(joiningDetails: { queuename: string; callerid: { id: string }; queuedcalls: string }): void {
@@ -77,7 +77,7 @@ export class QueueAggregator {
 
 	memberRemoved(queue: { queuename: string; queuedcalls: string }): void {
 		// current user is removed from the queue which has queue count |queuedcalls|
-		if (!this.currentQueueMembershipStatus[queue.queuename]) {
+		if (!this.currentQueueMembershipStatus?.[queue.queuename]) {
 			// something is wrong. Queue is not found in the membership details.
 			return;
 		}
@@ -89,23 +89,19 @@ export class QueueAggregator {
 	}
 
 	getCallWaitingCount(): number {
-		let totalCallWaitingCount = 0;
-		Object.entries(this.currentQueueMembershipStatus).forEach(([, value]) => {
-			totalCallWaitingCount += value.callsInQueue;
-		});
-		return totalCallWaitingCount;
+		return Object.entries(this.currentQueueMembershipStatus).reduce((acc, [_, value]) => acc + value.callsInQueue, 0);
 	}
 
 	getCurrentQueueName(): string {
-		if (this.currentlyServing.queueInfo) {
+		if (this.currentlyServing?.queueInfo) {
 			return this.currentlyServing.queueInfo.queueName;
 		}
 
 		return '';
 	}
 
 	callRinging(queueInfo: { queuename: string; callerid: { id: string; name: string } }): void {
-		if (!this.currentQueueMembershipStatus[queueInfo.queuename]) {
+		if (!this.currentQueueMembershipStatus?.[queueInfo.queuename]) {
 			return;
 		}
 

apps/meteor/client/providers/CallProvider/CallProvider.tsx

@@ -1,5 +1,13 @@
-import type { IVoipRoom, IUser } from '@rocket.chat/core-typings';
-import { ICallerInfo } from '@rocket.chat/core-typings';
+import type { IVoipRoom, IUser, VoipEventDataSignature } from '@rocket.chat/core-typings';
+import {
+	ICallerInfo,
+	isVoipEventAgentCalled,
+	isVoipEventAgentConnected,
+	isVoipEventCallerJoined,
+	isVoipEventQueueMemberAdded,
+	isVoipEventQueueMemberRemoved,
+	isVoipEventCallAbandoned,
+} from '@rocket.chat/core-typings';
 import { Random } from 'meteor/random';
 import React, { useMemo, FC, useRef, useCallback, useEffect, useState } from 'react';
 import { createPortal } from 'react-dom';
@@ -66,90 +74,54 @@ export const CallProvider: FC = ({ children }) => {
 			return;
 		}
 
-		const handleAgentCalled = async (queue: {
-			queuename: string;
-			callerId: { id: string; name: string };
-			queuedcalls: string;
-		}): Promise<void> => {
-			queueAggregator.callRinging({ queuename: queue.queuename, callerid: queue.callerId });
-			setQueueName(queueAggregator.getCurrentQueueName());
-		};
-
-		return subscribeToNotifyUser(`${user._id}/agentcalled`, handleAgentCalled);
-	}, [subscribeToNotifyUser, user, voipEnabled, queueAggregator]);
-
-	useEffect(() => {
-		if (!voipEnabled || !user || !queueAggregator) {
-			return;
-		}
-
-		const handleQueueJoined = async (joiningDetails: {
-			queuename: string;
-			callerid: { id: string };
-			queuedcalls: string;
-		}): Promise<void> => {
-			queueAggregator.queueJoined(joiningDetails);
-			setQueueCounter(queueAggregator.getCallWaitingCount());
-		};
-
-		return subscribeToNotifyUser(`${user._id}/callerjoined`, handleQueueJoined);
-	}, [subscribeToNotifyUser, user, voipEnabled, queueAggregator]);
-
-	useEffect(() => {
-		if (!voipEnabled || !user || !queueAggregator) {
-			return;
-		}
-
-		const handleAgentConnected = (queue: { queuename: string; queuedcalls: string; waittimeinqueue: string }): void => {
-			queueAggregator.callPickedup(queue);
-			setQueueName(queueAggregator.getCurrentQueueName());
-			setQueueCounter(queueAggregator.getCallWaitingCount());
-		};
-
-		return subscribeToNotifyUser(`${user._id}/agentconnected`, handleAgentConnected);
-	}, [queueAggregator, subscribeToNotifyUser, user, voipEnabled]);
-
-	useEffect(() => {
-		if (!voipEnabled || !user || !queueAggregator) {
-			return;
-		}
-
-		const handleMemberAdded = (queue: { queuename: string; queuedcalls: string }): void => {
-			queueAggregator.memberAdded(queue);
-			setQueueName(queueAggregator.getCurrentQueueName());
-			setQueueCounter(queueAggregator.getCallWaitingCount());
-		};
-
-		return subscribeToNotifyUser(`${user._id}/queuememberadded`, handleMemberAdded);
-	}, [queueAggregator, subscribeToNotifyUser, user, voipEnabled]);
-
-	useEffect(() => {
-		if (!voipEnabled || !user || !queueAggregator) {
-			return;
-		}
-
-		const handleMemberRemoved = (queue: { queuename: string; queuedcalls: string }): void => {
-			queueAggregator.memberRemoved(queue);
-			setQueueCounter(queueAggregator.getCallWaitingCount());
-		};
-
-		return subscribeToNotifyUser(`${user._id}/queuememberremoved`, handleMemberRemoved);
-	}, [queueAggregator, subscribeToNotifyUser, user, voipEnabled]);
-
-	useEffect(() => {
-		if (!voipEnabled || !user || !queueAggregator) {
-			return;
-		}
-
-		const handleCallAbandon = (queue: { queuename: string; queuedcallafterabandon: string }): void => {
-			queueAggregator.queueAbandoned(queue);
-			setQueueName(queueAggregator.getCurrentQueueName());
-			setQueueCounter(queueAggregator.getCallWaitingCount());
+		const handleEventReceived = async (event: VoipEventDataSignature): Promise<void> => {
+			if (isVoipEventAgentCalled(event)) {
+				const { data } = event;
+				queueAggregator.callRinging({ queuename: data.queue, callerid: data.callerId });
+				setQueueName(queueAggregator.getCurrentQueueName());
+				return;
+			}
+			if (isVoipEventAgentConnected(event)) {
+				const { data } = event;
+				queueAggregator.callPickedup({ queuename: data.queue, queuedcalls: data.queuedCalls, waittimeinqueue: data.waitTimeInQueue });
+				setQueueName(queueAggregator.getCurrentQueueName());
+				setQueueCounter(queueAggregator.getCallWaitingCount());
+				return;
+			}
+			if (isVoipEventCallerJoined(event)) {
+				const { data } = event;
+				queueAggregator.queueJoined({ queuename: data.queue, callerid: data.callerId, queuedcalls: data.queuedCalls });
+				setQueueCounter(queueAggregator.getCallWaitingCount());
+				return;
+			}
+			if (isVoipEventQueueMemberAdded(event)) {
+				const { data } = event;
+				queueAggregator.memberAdded({ queuename: data.queue, queuedcalls: data.queuedCalls });
+				setQueueName(queueAggregator.getCurrentQueueName());
+				setQueueCounter(queueAggregator.getCallWaitingCount());
+				return;
+			}
+			if (isVoipEventQueueMemberRemoved(event)) {
+				const { data } = event;
+				queueAggregator.memberRemoved({ queuename: data.queue, queuedcalls: data.queuedCalls });
+				setQueueCounter(queueAggregator.getCallWaitingCount());
+				return;
+			}
+			if (isVoipEventCallAbandoned(event)) {
+				const { data } = event;
+				queueAggregator.queueAbandoned({ queuename: data.queue, queuedcallafterabandon: data.queuedCallAfterAbandon });
+				setQueueName(queueAggregator.getCurrentQueueName());
+				setQueueCounter(queueAggregator.getCallWaitingCount());
+				return;
+			}
+
+			console.warn('Unknown event received');
 		};
 
-		return subscribeToNotifyUser(`${user._id}/callabandoned`, handleCallAbandon);
-	}, [queueAggregator, subscribeToNotifyUser, user, voipEnabled]);
+		return subscribeToNotifyUser(`${user._id}/voip.events`, handleEventReceived);
+	}, [subscribeToNotifyUser, user, queueAggregator, voipEnabled]);
 
+	// This was causing event duplication before, so we'll leave this here for now
 	useEffect(() => {
 		if (!voipEnabled || !user || !queueAggregator) {
 			return;
@@ -160,7 +132,7 @@ export const CallProvider: FC = ({ children }) => {
 			openWrapUpModal();
 		};
 
-		return subscribeToNotifyUser(`${user._id}/call.callerhangup`, handleCallHangup);
+		return subscribeToNotifyUser(`${user._id}/call.hangup`, handleCallHangup);
 	}, [openWrapUpModal, queueAggregator, subscribeToNotifyUser, user, voipEnabled]);
 
 	useEffect(() => {

apps/meteor/server/modules/listeners/listeners.module.ts

@@ -271,23 +271,13 @@ export class ListenersModule {
 		service.onEvent('banner.enabled', (bannerId): void => {
 			notifications.notifyLoggedInThisInstance('banner-changed', { bannerId });
 		});
-		service.onEvent('queue.agentcalled', (userId, queuename, callerId): void => {
-			notifications.notifyUserInThisInstance(userId, 'agentcalled', { queuename, callerId });
-		});
-		service.onEvent('queue.agentconnected', (userId, queuename: string, queuedcalls: string, waittimeinqueue: string): void => {
-			notifications.notifyUserInThisInstance(userId, 'agentconnected', { queuename, queuedcalls, waittimeinqueue });
-		});
-		service.onEvent('queue.callerjoined', (userId, queuename, callerid, queuedcalls): void => {
-			notifications.notifyUserInThisInstance(userId, 'callerjoined', { queuename, callerid, queuedcalls });
-		});
-		service.onEvent('queue.queuememberadded', (userId, queuename: string, queuedcalls: string): void => {
-			notifications.notifyUserInThisInstance(userId, 'queuememberadded', { queuename, queuedcalls });
-		});
-		service.onEvent('queue.queuememberremoved', (userId, queuename: string, queuedcalls: string): void => {
-			notifications.notifyUserInThisInstance(userId, 'queuememberremoved', { queuename, queuedcalls });
+
+		service.onEvent('voip.events', (userId, data): void => {
+			notifications.notifyUserInThisInstance(userId, 'voip.events', data);
 		});
-		service.onEvent('queue.callabandoned', (userId, queuename: string, queuedcallafterabandon: string): void => {
-			notifications.notifyUserInThisInstance(userId, 'callabandoned', { queuename, queuedcallafterabandon });
+
+		service.onEvent('call.callerhangup', (userId, data): void => {
+			notifications.notifyUserInThisInstance(userId, 'call.hangup', data);
 		});
 
 		service.onEvent('notify.desktop', (uid, notification): void => {

apps/meteor/server/modules/watchers/watchers.module.ts

@@ -344,6 +344,12 @@ export function initWatchers(models: IModelsParam, broadcast: BroadcastCallback,
 	// TODO: Prevent flood from database on username change, what causes changes on all past messages from that user
 	// and most of those messages are not loaded by the clients.
 	watch<IUser>(Users, ({ clientAction, id, data, diff, unset }) => {
+		// LivechatCount is updated each time an agent is routed to a chat. This prop is not used on the UI so we don't need
+		// to broadcast events originated by it when it's the only update on the user
+		if (diff && Object.keys(diff).length === 1 && 'livechatCount' in diff) {
+			return;
+		}
+
 		broadcast('watch.users', { clientAction, data, diff, unset, id });
 	});
 

apps/meteor/server/sdk/lib/Events.ts

@@ -22,6 +22,7 @@ import type {
 	IInvite,
 	IWebdavAccount,
 	ICustomSound,
+	VoipEventDataSignature,
 } from '@rocket.chat/core-typings';
 
 import { AutoUpdateRecord } from '../types/IMeteor';
@@ -123,11 +124,9 @@ export type EventSignatures = {
 		diff?: undefined | Record<string, any>;
 		id: string;
 	}): void;
-	'queue.agentcalled'(userid: string, queuename: string, callerid: Record<string, string>): void;
-	'queue.agentconnected'(userid: string, queuename: string, queuedcalls: string, waittimeinqueue: string): void;
-	'queue.callerjoined'(userid: string, queuename: string, callerid: Record<string, string>, queuedcalls: string): void;
-	'queue.queuememberadded'(userid: string, queuename: string, queuedcalls: string): void;
-	'queue.queuememberremoved'(userid: string, queuename: string, queuedcalls: string): void;
-	'queue.callabandoned'(userid: string, queuename: string, queuedcallafterabandon: string): void;
+
+	// Send all events from here
+	'voip.events'(userId: string, data: VoipEventDataSignature): void;
+	'call.callerhangup'(userId: string, data: { roomId: string }): void;
 	'watch.pbxevents'(data: { clientAction: ClientAction; data: Partial<IPbxEvent>; id: string }): void;
 };

apps/meteor/server/services/omnichannel-voip/service.ts

@@ -28,7 +28,7 @@ import { VoipRoomsRaw } from '../../../app/models/server/raw/VoipRooms';
 import { PbxEventsRaw } from '../../../app/models/server/raw/PbxEvents';
 import { sendMessage } from '../../../app/lib/server/functions/sendMessage';
 import { FindVoipRoomsParams } from './internalTypes';
-import { Notifications } from '../../../app/notifications/server';
+import { api } from '../../sdk/api';
 
 export class OmnichannelVoipService extends ServiceClassInternal implements IOmnichannelVoipService {
 	protected name = 'omnichannel-voip';
@@ -80,8 +80,7 @@ export class OmnichannelVoipService extends ServiceClassInternal implements IOmn
 			return;
 		}
 		this.logger.debug(`Notifying agent ${agent._id} of hangup on room ${currentRoom._id}`);
-		// TODO evalute why this is 'notifyUserInThisInstance'
-		Notifications.notifyUserInThisInstance(agent._id, 'call.callerhangup', { roomId: currentRoom._id });
+		api.broadcast('call.callerhangup', agent._id, { roomId: currentRoom._id });
 	}
 
 	private async processAgentDisconnect(extension: string): Promise<void> {

apps/meteor/server/services/voip/connector/asterisk/ami/ContinuousMonitor.ts

@@ -89,8 +89,8 @@ export class ContinuousMonitor extends Command {
 
 	async processQueueMembershipChange(event: IQueueMemberAdded | IQueueMemberRemoved): Promise<void> {
 		const extension = event.interface.toLowerCase().replace('pjsip/', '');
-		const queueName = event.queue;
-		const queueDetails = await this.getQueueDetails(queueName);
+		const { queue } = event;
+		const queueDetails = await this.getQueueDetails(queue);
 		const { calls } = queueDetails;
 		const user = await this.users.findOneByExtension(extension, {
 			projection: {
@@ -101,9 +101,9 @@ export class ContinuousMonitor extends Command {
 		});
 		if (user) {
 			if (isIQueueMemberAddedEvent(event)) {
-				api.broadcast(`queue.queuememberadded`, user._id, queueName, calls);
+				api.broadcast(`voip.events`, user._id, { data: { queue, queuedCalls: calls }, event: 'queue-member-added' });
 			} else if (isIQueueMemberRemovedEvent(event)) {
-				api.broadcast(`queue.queuememberremoved`, user._id, queueName, calls);
+				api.broadcast(`voip.events`, user._id, { event: 'queue-member-removed', data: { queue, queuedCalls: calls } });
 			}
 		}
 	}
@@ -130,7 +130,8 @@ export class ContinuousMonitor extends Command {
 			name: event.calleridname,
 		};
 
-		api.broadcast('queue.agentcalled', user._id, event.queue, callerId);
+		api.broadcast('voip.events', user._id, { event: 'agent-called', data: { callerId, queue: event.queue } });
+		// api.broadcast('queue.agentcalled', user._id, event.queue, callerId);
 	}
 
 	async storePbxEvent(event: IQueueEvent | IContactStatus, eventName: string): Promise<void> {
@@ -185,7 +186,7 @@ export class ContinuousMonitor extends Command {
 				await this.storePbxEvent(event, 'QueueCallerJoin');
 				this.logger.debug(`Broadcasting event queue.callerjoined to ${members.length} agents on queue ${event.queue}`);
 				members.forEach((m) => {
-					api.broadcast('queue.callerjoined', m, event.queue, callerId, event.count);
+					api.broadcast('voip.events', m, { event: 'caller-joined', data: { callerId, queue: event.queue, queuedCalls: event.count } });
 				});
 				break;
 			}
@@ -194,7 +195,7 @@ export class ContinuousMonitor extends Command {
 				await this.storePbxEvent(event, 'QueueCallerAbandon');
 				this.logger.debug(`Broadcasting event queue.callabandoned to ${members.length} agents on queue ${event.queue}`);
 				members.forEach((m) => {
-					api.broadcast('queue.callabandoned', m, event.queue, calls);
+					api.broadcast('voip.events', m, { event: 'call-abandoned', data: { queue: event.queue, queuedCallAfterAbandon: calls } });
 				});
 				break;
 			}
@@ -205,7 +206,10 @@ export class ContinuousMonitor extends Command {
 				this.logger.debug(`Broadcasting event queue.agentconnected to ${members.length} agents on queue ${event.queue}`);
 				members.forEach((m) => {
 					// event.holdtime signifies wait time in the queue.
-					api.broadcast('queue.agentconnected', m, event.queue, calls, event.holdtime);
+					api.broadcast('voip.events', m, {
+						event: 'agent-connected',
+						data: { queue: event.queue, queuedCalls: calls, waitTimeInQueue: event.holdtime },
+					});
 				});
 				break;
 			}

packages/core-typings/src/index.ts

@@ -108,6 +108,7 @@ export * from "./voip/VoipEvents";
 export * from "./voip/VoIpCallerInfo";
 export * from "./voip/VoipClientEvents";
 export * from "./voip/WorkflowTypes";
+export * from "./voip/IVoipClientEvents";
 export * from "./ACDQueues";
 export * from "./IVoipConnectorResult";
 export * from "./IVoipServerConfig";