refactor: PbxEvents out of DB Watcher (#32372)

apps/meteor/app/lib/server/lib/notifyListener.ts

@@ -1,6 +1,6 @@
 import { api, dbWatchersDisabled } from '@rocket.chat/core-services';
-import type { IRocketChatRecord, IRoom } from '@rocket.chat/core-typings';
-import { Rooms } from '@rocket.chat/models';
+import type { IPbxEvent, IRocketChatRecord, IRoom } from '@rocket.chat/core-typings';
+import { PbxEvents, Rooms } from '@rocket.chat/models';
 
 type ClientAction = 'inserted' | 'updated' | 'removed';
 
@@ -65,3 +65,18 @@ export async function notifyOnRoomChangedByUserDM<T extends IRoom>(
 		void api.broadcast('watch.rooms', { clientAction, room: item });
 	}
 }
+
+export async function notifyOnPbxEventChangedById<T extends IPbxEvent>(
+	id: T['_id'],
+	clientAction: ClientAction = 'updated',
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const item = await PbxEvents.findOneById(id);
+
+	if (item) {
+		void api.broadcast('watch.pbxevents', { clientAction, id, data: item });
+	}
+}

apps/meteor/server/database/watchCollections.ts

@@ -39,7 +39,6 @@ export function getWatchCollections(): string[] {
 		IntegrationHistory.getCollectionName(),
 		Integrations.getCollectionName(),
 		EmailInbox.getCollectionName(),
-		PbxEvents.getCollectionName(),
 		Settings.getCollectionName(),
 		LivechatPriority.getCollectionName(),
 		Subscriptions.getCollectionName(),
@@ -50,6 +49,7 @@ export function getWatchCollections(): string[] {
 		collections.push(Messages.getCollectionName());
 		collections.push(Roles.getCollectionName());
 		collections.push(Rooms.getCollectionName());
+		collections.push(PbxEvents.getCollectionName());
 	}
 
 	if (onlyCollections.length > 0) {

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

@@ -47,11 +47,11 @@ import { Logger } from '@rocket.chat/logger';
 import { Users, PbxEvents } from '@rocket.chat/models';
 import type { Db } from 'mongodb';
 
+import { notifyOnPbxEventChangedById } from '../../../../../../app/lib/server/lib/notifyListener';
 import { Command, CommandType } from '../Command';
 import { Commands } from '../Commands';
 import { ACDQueue } from './ACDQueue';
 import { CallbackContext } from './CallbackContext';
-// import { sendMessage } from '../../../../../../app/lib/server/functions/sendMessage';
 
 export class ContinuousMonitor extends Command {
 	private logger: Logger;
@@ -140,13 +140,15 @@ export class ContinuousMonitor extends Command {
 				// This event represents when an agent drops a call because of disconnection
 				// May happen for any reason outside of our control, like closing the browswer
 				// Or network/power issues
-				await PbxEvents.insertOne({
+				const { insertedId } = await PbxEvents.insertOne({
 					event: eventName,
 					uniqueId: `${eventName}-${event.contactstatus}-${now.getTime()}`,
 					ts: now,
 					agentExtension: event.aor,
 				});
 
+				void notifyOnPbxEventChangedById(insertedId, 'inserted');
+
 				return;
 			}
 
@@ -159,7 +161,7 @@ export class ContinuousMonitor extends Command {
 			// NOTE: using the uniqueId prop of event is not the recommented approach, since it's an opaque ID
 			// However, since we're not using it for anything special, it's a "fair use"
 			// uniqueId => {server}/{epoch}.{id of channel associated with this call}
-			await PbxEvents.insertOne({
+			const { insertedId } = await PbxEvents.insertOne({
 				uniqueId,
 				event: eventName,
 				ts: now,
@@ -170,6 +172,8 @@ export class ContinuousMonitor extends Command {
 				callUniqueIdFallback: event.linkedid,
 				agentExtension: event?.connectedlinenum,
 			});
+
+			void notifyOnPbxEventChangedById(insertedId, 'inserted');
 		} catch (e) {
 			this.logger.debug('Event was handled by other instance');
 		}
@@ -282,7 +286,7 @@ export class ContinuousMonitor extends Command {
 		 * and event.calleridnum is the extension that is initiating a call.
 		 */
 		try {
-			await PbxEvents.insertOne({
+			const { insertedId } = await PbxEvents.insertOne({
 				uniqueId: `${event.event}-${event.calleridnum}-${event.channel}-${event.destchannel}-${event.uniqueid}`,
 				event: event.event,
 				ts: new Date(),
@@ -291,6 +295,8 @@ export class ContinuousMonitor extends Command {
 				callUniqueIdFallback: event.linkedid,
 				agentExtension: event.calleridnum,
 			});
+
+			void notifyOnPbxEventChangedById(insertedId, 'inserted');
 		} catch (e) {
 			// This could mean we received a duplicate event
 			// This is quite common since DialEnd event happens "multiple times" at the end of the call