refactor: LivechatInquiry out of DB Watcher

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

@@ -9,6 +9,7 @@ import type {
 	IIntegration,
 	IPbxEvent,
 	LoginServiceConfiguration as LoginServiceConfigurationData,
+	ILivechatInquiryRecord,
 	ILivechatPriority,
 	IEmailInbox,
 	IIntegrationHistory,
@@ -23,6 +24,7 @@ import {
 	Integrations,
 	LoginServiceConfiguration,
 	IntegrationHistory,
+	LivechatInquiry,
 } from '@rocket.chat/models';
 
 type ClientAction = 'inserted' | 'updated' | 'removed';
@@ -278,6 +280,76 @@ export async function notifyOnEmailInboxChanged<T extends IEmailInbox>(
 	void api.broadcast('watch.emailInbox', { clientAction, id: data._id, data });
 }
 
+export async function notifyOnLivechatInquiryChanged(
+	data: ILivechatInquiryRecord | ILivechatInquiryRecord[],
+	clientAction: ClientAction = 'updated',
+	diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const items = Array.isArray(data) ? data : [data];
+
+	for (const item of items) {
+		void api.broadcast('watch.inquiries', { clientAction, inquiry: item, diff });
+	}
+}
+
+export async function notifyOnLivechatInquiryChangedById(
+	id: ILivechatInquiryRecord['_id'],
+	clientAction: ClientAction = 'updated',
+	diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const inquiry = clientAction === 'removed' ? await LivechatInquiry.trashFindOneById(id) : await LivechatInquiry.findOneById(id);
+
+	if (!inquiry) {
+		return;
+	}
+
+	void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+}
+
+export async function notifyOnLivechatInquiryChangedByRoom(
+	rid: ILivechatInquiryRecord['rid'],
+	clientAction: ClientAction = 'updated',
+	diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const inquiry = await LivechatInquiry.findOneByRoomId(rid, {});
+
+	if (!inquiry) {
+		return;
+	}
+
+	void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+}
+
+export async function notifyOnLivechatInquiryChangedByToken(
+	token: ILivechatInquiryRecord['v']['token'],
+	clientAction: ClientAction = 'updated',
+	diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const inquiry = await LivechatInquiry.findOneByToken(token);
+
+	if (!inquiry) {
+		return;
+	}
+
+	void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+}
+
 export async function notifyOnIntegrationHistoryChanged<T extends IIntegrationHistory>(
 	data: AtLeast<T, '_id'>,
 	clientAction: ClientAction = 'updated',

apps/meteor/app/livechat/server/hooks/markRoomResponded.ts

@@ -4,6 +4,7 @@ import { LivechatRooms, LivechatVisitors, LivechatInquiry } from '@rocket.chat/m
 import moment from 'moment';
 
 import { callbacks } from '../../../../lib/callbacks';
+import { notifyOnLivechatInquiryChanged } from '../../../lib/server/lib/notifyListener';
 
 callbacks.add(
 	'afterSaveMessage',
@@ -37,10 +38,11 @@ callbacks.add(
 		}
 
 		if (!room.v?.activity?.includes(monthYear)) {
-			await Promise.all([
-				LivechatRooms.markVisitorActiveForPeriod(room._id, monthYear),
-				LivechatInquiry.markInquiryActiveForPeriod(room._id, monthYear),
-			]);
+			void LivechatRooms.markVisitorActiveForPeriod(room._id, monthYear);
+			const livechatInquiry = await LivechatInquiry.markInquiryActiveForPeriod(room._id, monthYear);
+			if (livechatInquiry) {
+				void notifyOnLivechatInquiryChanged(livechatInquiry, 'updated', { v: livechatInquiry.v });
+			}
 		}
 
 		if (room.responseBy) {

apps/meteor/app/livechat/server/hooks/saveLastMessageToInquiry.ts

@@ -2,6 +2,7 @@ import { isOmnichannelRoom, isEditedMessage } from '@rocket.chat/core-typings';
 import { LivechatInquiry } from '@rocket.chat/models';
 
 import { callbacks } from '../../../../lib/callbacks';
+import { notifyOnLivechatInquiryChanged } from '../../../lib/server/lib/notifyListener';
 import { settings } from '../../../settings/server';
 import { RoutingManager } from '../lib/RoutingManager';
 
@@ -21,7 +22,10 @@ callbacks.add(
 			return message;
 		}
 
-		await LivechatInquiry.setLastMessageByRoomId(room._id, message);
+		const livechatInquiry = await LivechatInquiry.setLastMessageByRoomId(room._id, message);
+		if (livechatInquiry) {
+			void notifyOnLivechatInquiryChanged(livechatInquiry, 'updated', { lastMessage: message });
+		}
 
 		return message;
 	},

apps/meteor/app/livechat/server/lib/Contacts.ts

@@ -6,7 +6,7 @@ import type { MatchKeysAndValues, OnlyFieldsOfType } from 'mongodb';
 
 import { callbacks } from '../../../../lib/callbacks';
 import { trim } from '../../../../lib/utils/stringUtils';
-import { notifyOnRoomChangedById } from '../../../lib/server/lib/notifyListener';
+import { notifyOnRoomChangedById, notifyOnLivechatInquiryChangedByRoom } from '../../../lib/server/lib/notifyListener';
 import { i18n } from '../../../utils/lib/i18n';
 
 type RegisterContactProps = {
@@ -144,6 +144,7 @@ export const Contacts = {
 					Subscriptions.updateDisplayNameByRoomId(rid, name),
 				]);
 
+				void notifyOnLivechatInquiryChangedByRoom(rid, 'updated', { name });
 				void notifyOnRoomChangedById(rid);
 			}
 		}

apps/meteor/app/livechat/server/lib/LivechatTyped.ts

@@ -57,7 +57,12 @@ import { FileUpload } from '../../../file-upload/server';
 import { deleteMessage } from '../../../lib/server/functions/deleteMessage';
 import { sendMessage } from '../../../lib/server/functions/sendMessage';
 import { updateMessage } from '../../../lib/server/functions/updateMessage';
-import { notifyOnRoomChangedById } from '../../../lib/server/lib/notifyListener';
+import {
+	notifyOnLivechatInquiryChanged,
+	notifyOnLivechatInquiryChangedByRoom,
+	notifyOnRoomChangedById,
+	notifyOnLivechatInquiryChangedByToken,
+} from '../../../lib/server/lib/notifyListener';
 import * as Mailer from '../../../mailer/server/api';
 import { metrics } from '../../../metrics/server';
 import { settings } from '../../../settings/server';
@@ -297,6 +302,7 @@ class LivechatClass {
 		if (removedInquiry && removedInquiry.deletedCount !== 1) {
 			throw new Error('Error removing inquiry');
 		}
+		void notifyOnLivechatInquiryChangedByRoom(rid, 'removed');
 
 		const updatedRoom = await LivechatRooms.closeRoomById(rid, closeData);
 		if (!updatedRoom || updatedRoom.modifiedCount !== 1) {
@@ -508,6 +514,8 @@ class LivechatClass {
 			LivechatRooms.removeById(rid),
 		]);
 
+		void notifyOnLivechatInquiryChangedByRoom(rid, 'removed');
+
 		for (const r of result) {
 			if (r.status === 'rejected') {
 				this.logger.error(`Error removing room ${rid}: ${r.reason}`);
@@ -1265,11 +1273,11 @@ class LivechatClass {
 			]);
 		}
 
-		await Promise.all([
-			Subscriptions.removeByVisitorToken(token),
-			LivechatRooms.removeByVisitorToken(token),
-			LivechatInquiry.removeByVisitorToken(token),
-		]);
+		await Promise.all([Subscriptions.removeByVisitorToken(token), LivechatRooms.removeByVisitorToken(token)]);
+
+		const livechatInquiries = await LivechatInquiry.findIdsByVisitorToken(token).toArray();
+		await LivechatInquiry.removeByIds(livechatInquiries.map(({ _id }) => _id));
+		void notifyOnLivechatInquiryChanged(livechatInquiries, 'removed');
 	}
 
 	async deleteMessage({ guest, message }: { guest: ILivechatVisitor; message: IMessage }) {
@@ -1658,8 +1666,13 @@ class LivechatClass {
 	}
 
 	async notifyGuestStatusChanged(token: string, status: UserStatus) {
-		await LivechatInquiry.updateVisitorStatus(token, status);
 		await LivechatRooms.updateVisitorStatus(token, status);
+
+		const inquiryVisitorStatus = await LivechatInquiry.updateVisitorStatus(token, status);
+
+		if (inquiryVisitorStatus.modifiedCount) {
+			void notifyOnLivechatInquiryChangedByToken(token, 'updated', { v: { status } });
+		}
 	}
 
 	async setUserStatusLivechat(userId: string, status: ILivechatAgentStatus) {
@@ -1804,14 +1817,19 @@ class LivechatClass {
 		if (guestData?.name?.trim().length) {
 			const { _id: rid } = roomData;
 			const { name } = guestData;
+
 			await Promise.all([
 				Rooms.setFnameById(rid, name),
 				LivechatInquiry.setNameByRoomId(rid, name),
 				Subscriptions.updateDisplayNameByRoomId(rid, name),
 			]);
+
+			void notifyOnLivechatInquiryChangedByRoom(rid, 'updated', { name });
 		}
 
 		void notifyOnRoomChangedById(roomData._id);
+
+		return true;
 	}
 
 	/**

apps/meteor/app/livechat/server/lib/QueueManager.ts

@@ -1,12 +1,20 @@
 import { Apps, AppEvents } from '@rocket.chat/apps';
 import { Omnichannel } from '@rocket.chat/core-services';
-import type { ILivechatInquiryRecord, ILivechatVisitor, IMessage, IOmnichannelRoom, SelectedAgent } from '@rocket.chat/core-typings';
+import {
+	LivechatInquiryStatus,
+	type ILivechatInquiryRecord,
+	type ILivechatVisitor,
+	type IMessage,
+	type IOmnichannelRoom,
+	type SelectedAgent,
+} from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
 import { LivechatInquiry, LivechatRooms, Users } from '@rocket.chat/models';
 import { Match, check } from 'meteor/check';
 import { Meteor } from 'meteor/meteor';
 
 import { callbacks } from '../../../../lib/callbacks';
+import { notifyOnLivechatInquiryChangedById, notifyOnLivechatInquiryChanged } from '../../../lib/server/lib/notifyListener';
 import { checkServiceStatus, createLivechatRoom, createLivechatInquiry } from './Helper';
 import { RoutingManager } from './RoutingManager';
 
@@ -15,6 +23,11 @@ const logger = new Logger('QueueManager');
 export const saveQueueInquiry = async (inquiry: ILivechatInquiryRecord) => {
 	await LivechatInquiry.queueInquiry(inquiry._id);
 	await callbacks.run('livechat.afterInquiryQueued', inquiry);
+	void notifyOnLivechatInquiryChanged(inquiry, 'updated', {
+		status: LivechatInquiryStatus.QUEUED,
+		queuedAt: new Date(),
+		takenAt: undefined,
+	});
 };
 
 export const queueInquiry = async (inquiry: ILivechatInquiryRecord, defaultAgent?: SelectedAgent) => {
@@ -137,6 +150,7 @@ export const QueueManager: queueManager = {
 		if (oldInquiry) {
 			logger.debug(`Removing old inquiry (${oldInquiry._id}) for room ${rid}`);
 			await LivechatInquiry.removeByRoomId(rid);
+			void notifyOnLivechatInquiryChangedById(oldInquiry._id, 'removed');
 		}
 
 		const guest = {

apps/meteor/app/livechat/server/lib/RoutingManager.ts

@@ -11,13 +11,15 @@ import type {
 	InquiryWithAgentInfo,
 	TransferData,
 } from '@rocket.chat/core-typings';
+import { LivechatInquiryStatus } from '@rocket.chat/core-typings';
 import { License } from '@rocket.chat/license';
 import { Logger } from '@rocket.chat/logger';
 import { LivechatInquiry, LivechatRooms, Subscriptions, Rooms, Users } from '@rocket.chat/models';
 import { Match, check } from 'meteor/check';
 import { Meteor } from 'meteor/meteor';
 
 import { callbacks } from '../../../../lib/callbacks';
+import { notifyOnLivechatInquiryChangedById, notifyOnLivechatInquiryChanged } from '../../../lib/server/lib/notifyListener';
 import { settings } from '../../../settings/server';
 import {
 	createLivechatSubscription,
@@ -183,6 +185,12 @@ export const RoutingManager: Routing = {
 
 		if (shouldQueue) {
 			await LivechatInquiry.queueInquiry(inquiry._id);
+
+			void notifyOnLivechatInquiryChanged(inquiry, 'updated', {
+				status: LivechatInquiryStatus.QUEUED,
+				queuedAt: new Date(),
+				takenAt: undefined,
+			});
 		}
 
 		if (servedBy) {
@@ -192,6 +200,7 @@ export const RoutingManager: Routing = {
 		}
 
 		await dispatchInquiryQueued(inquiry);
+
 		return true;
 	},
 
@@ -250,11 +259,20 @@ export const RoutingManager: Routing = {
 		}
 
 		await LivechatInquiry.takeInquiry(_id);
+
 		const inq = await this.assignAgent(inquiry as InquiryWithAgentInfo, agent);
 		logger.info(`Inquiry ${inquiry._id} taken by agent ${agent.agentId}`);
 
 		callbacks.runAsync('livechat.afterTakeInquiry', inq, agent);
 
+		void notifyOnLivechatInquiryChangedById(inquiry._id, 'updated', {
+			status: LivechatInquiryStatus.TAKEN,
+			takenAt: new Date(),
+			defaultAgent: undefined,
+			estimatedInactivityCloseTimeAt: undefined,
+			queuedAt: undefined,
+		});
+
 		return LivechatRooms.findOneById(rid);
 	},
 
@@ -282,6 +300,7 @@ export const RoutingManager: Routing = {
 		if (defaultAgent) {
 			logger.debug(`Delegating Inquiry ${inquiry._id} to agent ${defaultAgent.username}`);
 			await LivechatInquiry.setDefaultAgentById(inquiry._id, defaultAgent);
+			void notifyOnLivechatInquiryChanged(inquiry, 'updated', { defaultAgent });
 		}
 
 		logger.debug(`Queueing inquiry ${inquiry._id}`);

apps/meteor/ee/app/livechat-enterprise/server/hooks/beforeRoutingChat.ts

@@ -1,6 +1,7 @@
 import type { ILivechatDepartment } from '@rocket.chat/core-typings';
 import { LivechatDepartment, LivechatInquiry, LivechatRooms } from '@rocket.chat/models';
 
+import { notifyOnLivechatInquiryChanged } from '../../../../../app/lib/server/lib/notifyListener';
 import { online } from '../../../../../app/livechat/server/api/lib/livechat';
 import { allowAgentSkipQueue } from '../../../../../app/livechat/server/lib/Helper';
 import { Livechat } from '../../../../../app/livechat/server/lib/LivechatTyped';
@@ -30,13 +31,22 @@ callbacks.add(
 				cbLogger.info(
 					`Inquiry ${inquiry._id} will be moved from department ${department._id} to fallback department ${department.fallbackForwardDepartment}`,
 				);
+
 				// update visitor
 				await Livechat.setDepartmentForGuest({
 					token: inquiry?.v?.token,
 					department: department.fallbackForwardDepartment,
 				});
+
 				// update inquiry
-				inquiry = (await LivechatInquiry.setDepartmentByInquiryId(inquiry._id, department.fallbackForwardDepartment)) ?? inquiry;
+				const updatedLivechatInquiry = await LivechatInquiry.setDepartmentByInquiryId(inquiry._id, department.fallbackForwardDepartment);
+
+				if (updatedLivechatInquiry) {
+					void notifyOnLivechatInquiryChanged(updatedLivechatInquiry, 'updated', { department: updatedLivechatInquiry.department });
+				}
+
+				inquiry = updatedLivechatInquiry ?? inquiry;
+
 				// update room
 				await LivechatRooms.setDepartmentByRoomId(inquiry.rid, department.fallbackForwardDepartment);
 			}