fix: Missing notifiers for priorities & some inquiry changes (#36449)

.changeset/sweet-dingos-decide.md

@@ -0,0 +1,7 @@
+---
+"@rocket.chat/meteor": patch
+"@rocket.chat/model-typings": patch
+"@rocket.chat/models": patch
+---
+
+Fixes priorities, sla changes & inquiries not being propagated when change streams were not being used

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

@@ -250,17 +250,24 @@ export const notifyOnLivechatInquiryChanged = withDbWatcherCheck(
 
 export const notifyOnLivechatInquiryChangedById = withDbWatcherCheck(
 	async (
-		id: ILivechatInquiryRecord['_id'],
+		ids: ILivechatInquiryRecord['_id'] | ILivechatInquiryRecord['_id'][],
 		clientAction: ClientAction = 'updated',
 		diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
 	): Promise<void> => {
-		const inquiry = clientAction === 'removed' ? await LivechatInquiry.trashFindOneById(id) : await LivechatInquiry.findOneById(id);
+		const eligibleIds = Array.isArray(ids) ? ids : [ids];
 
-		if (!inquiry) {
+		const items =
+			clientAction === 'removed'
+				? LivechatInquiry.trashFind({ _id: { $in: eligibleIds } })
+				: LivechatInquiry.find({ _id: { $in: eligibleIds } });
+
+		if (!items) {
 			return;
 		}
 
-		void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+		for await (const inquiry of items) {
+			void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+		}
 	},
 );
 
@@ -280,17 +287,24 @@ export const notifyOnLivechatInquiryChangedByVisitorIds = withDbWatcherCheck(
 
 export const notifyOnLivechatInquiryChangedByRoom = withDbWatcherCheck(
 	async (
-		rid: ILivechatInquiryRecord['rid'],
+		rids: ILivechatInquiryRecord['rid'] | ILivechatInquiryRecord['rid'][],
 		clientAction: ClientAction = 'updated',
 		diff?: Partial<Record<keyof ILivechatInquiryRecord, unknown> & { queuedAt: unknown; takenAt: unknown }>,
 	): Promise<void> => {
-		const inquiry = await LivechatInquiry.findOneByRoomId(rid, {});
+		const eligibleIds = Array.isArray(rids) ? rids : [rids];
 
-		if (!inquiry) {
+		const items =
+			clientAction === 'removed'
+				? LivechatInquiry.trashFind({ rid: { $in: eligibleIds } })
+				: LivechatInquiry.find({ rid: { $in: eligibleIds } });
+
+		if (!items) {
 			return;
 		}
 
-		void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+		for await (const inquiry of items) {
+			void api.broadcast('watch.inquiries', { clientAction, inquiry, diff });
+		}
 	},
 );
 

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

@@ -55,6 +55,8 @@ import {
 	notifyOnSubscriptionChangedById,
 	notifyOnSubscriptionChangedByRoomId,
 	notifyOnSubscriptionChanged,
+	notifyOnRoomChangedById,
+	notifyOnLivechatInquiryChangedByRoom,
 } from '../../../lib/server/lib/notifyListener';
 import { settings } from '../../../settings/server';
 
@@ -555,6 +557,12 @@ export const updateChatDepartment = async ({
 		Subscriptions.changeDepartmentByRoomId(rid, newDepartmentId),
 	]);
 
+	if (responses[0].modifiedCount) {
+		void notifyOnRoomChangedById(rid);
+	}
+	if (responses[1].modifiedCount) {
+		void notifyOnLivechatInquiryChangedByRoom(rid);
+	}
 	if (responses[2].modifiedCount) {
 		void notifyOnSubscriptionChangedByRoomId(rid);
 	}

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

@@ -12,6 +12,7 @@ import { callbacks } from '../../../../lib/callbacks';
 import { client, shouldRetryTransaction } from '../../../../server/database/utils';
 import {
 	notifyOnLivechatInquiryChanged,
+	notifyOnRoomChanged,
 	notifyOnRoomChangedById,
 	notifyOnSubscriptionChanged,
 } from '../../../lib/server/lib/notifyListener';
@@ -179,6 +180,9 @@ async function doCloseRoom(
 	if (!params.forceClose && removedInquiry && removedInquiry.deletedCount !== 1) {
 		throw new Error('Error removing inquiry');
 	}
+	if (removedInquiry.deletedCount) {
+		void notifyOnLivechatInquiryChanged(inquiry!, 'removed');
+	}
 
 	const updatedRoom = await LivechatRooms.closeRoomById(rid, closeData, { session });
 	if (!params.forceClose && (!updatedRoom || updatedRoom.modifiedCount !== 1)) {
@@ -207,6 +211,7 @@ async function doCloseRoom(
 		throw new Error('Error: Room not found');
 	}
 
+	void notifyOnRoomChanged(newRoom, 'updated');
 	return { room: newRoom, closedBy: closeData.closedBy, removedInquiry: inquiry };
 }
 

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

@@ -37,6 +37,7 @@ import {
 	notifyOnRoomChangedById,
 	notifyOnLivechatInquiryChanged,
 	notifyOnSubscriptionChanged,
+	notifyOnRoomChanged,
 } from '../../../lib/server/lib/notifyListener';
 import { settings } from '../../../settings/server';
 import { i18n } from '../../../utils/lib/i18n';
@@ -278,6 +279,9 @@ export async function removeOmnichannelRoom(rid: string) {
 	if (result[3]?.status === 'fulfilled' && result[3].value?.deletedCount && inquiry) {
 		void notifyOnLivechatInquiryChanged(inquiry, 'removed');
 	}
+	if (result[4]?.status === 'fulfilled' && result[4].value?.deletedCount) {
+		void notifyOnRoomChanged(room, 'removed');
+	}
 
 	for (const r of result) {
 		if (r.status === 'rejected') {

apps/meteor/app/livechat/server/roomAccessValidator.compatibility.ts

@@ -68,6 +68,7 @@ export const validators: OmnichannelRoomAccessValidator[] = [
 			],
 		};
 
+		// TODO: findone filtering if the inquiry is queued instead of checking here
 		const inquiry = await LivechatInquiry.findOne(filter, { projection: { status: 1 } });
 		return inquiry && inquiry.status === 'queued';
 	},

apps/meteor/ee/app/livechat-enterprise/server/api/lib/priorities.ts

@@ -5,6 +5,7 @@ import type { PaginatedResult } from '@rocket.chat/rest-typings';
 import { escapeRegExp } from '@rocket.chat/string-helpers';
 import type { FindOptions } from 'mongodb';
 
+import { notifyOnLivechatInquiryChangedByRoom, notifyOnRoomChanged } from '../../../../../../app/lib/server/lib/notifyListener';
 import { logger } from '../../lib/logger';
 
 type FindPriorityParams = {
@@ -24,7 +25,7 @@ export async function findPriority({
 		...(text && { $or: [{ name: new RegExp(escapeRegExp(text), 'i') }, { description: new RegExp(escapeRegExp(text), 'i') }] }),
 	};
 
-	const { cursor, totalCount } = await LivechatPriority.findPaginated(query, {
+	const { cursor, totalCount } = LivechatPriority.findPaginated(query, {
 		sort: sort || { name: 1 },
 		skip: offset,
 		limit: count,
@@ -64,7 +65,7 @@ export const updateRoomPriority = async (
 	user: Required<Pick<IUser, '_id' | 'username' | 'name'>>,
 	priorityId: string,
 ): Promise<void> => {
-	const room = await LivechatRooms.findOneById(rid, { projection: { _id: 1 } });
+	const room = await LivechatRooms.findOneById(rid);
 	if (!room) {
 		throw new Error('error-room-does-not-exist');
 	}
@@ -79,10 +80,15 @@ export const updateRoomPriority = async (
 		LivechatInquiry.setPriorityForRoom(rid, priority),
 		addPriorityChangeHistoryToRoom(room._id, user, priority),
 	]);
+
+	void notifyOnRoomChanged({ ...room, priorityId: priority._id, priorityWeight: priority.sortItem }, 'updated');
+	void notifyOnLivechatInquiryChangedByRoom(rid, 'updated');
 };
 
 export const removePriorityFromRoom = async (rid: string, user: Required<Pick<IUser, '_id' | 'username' | 'name'>>): Promise<void> => {
-	const room = await LivechatRooms.findOneById<Pick<IOmnichannelRoom, '_id'>>(rid, { projection: { _id: 1 } });
+	const room = await LivechatRooms.findOneById<Omit<IOmnichannelRoom, 'priorityId' | 'priorityWeight'>>(rid, {
+		projection: { priorityId: 0, priorityWeight: 0 },
+	});
 	if (!room) {
 		throw new Error('error-room-does-not-exist');
 	}
@@ -92,6 +98,9 @@ export const removePriorityFromRoom = async (rid: string, user: Required<Pick<IU
 		LivechatInquiry.unsetPriorityForRoom(rid),
 		addPriorityChangeHistoryToRoom(rid, user),
 	]);
+
+	void notifyOnRoomChanged(room, 'updated');
+	void notifyOnLivechatInquiryChangedByRoom(rid, 'updated');
 };
 
 const addPriorityChangeHistoryToRoom = async (

apps/meteor/ee/app/livechat-enterprise/server/api/lib/sla.ts

@@ -29,7 +29,7 @@ export async function findSLA({
 		...(text && { $or: [{ name: new RegExp(escapeRegExp(text), 'i') }, { description: new RegExp(escapeRegExp(text), 'i') }] }),
 	};
 
-	const { cursor, totalCount } = await OmnichannelServiceLevelAgreements.findPaginated(query, {
+	const { cursor, totalCount } = OmnichannelServiceLevelAgreements.findPaginated(query, {
 		sort: sort || { name: 1 },
 		skip: offset,
 		limit: count,

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

@@ -1,7 +1,7 @@
 import type { ILivechatInquiryRecord, SelectedAgent, ILivechatDepartment } from '@rocket.chat/core-typings';
 import { LivechatDepartment, LivechatInquiry, LivechatRooms } from '@rocket.chat/models';
 
-import { notifyOnLivechatInquiryChanged } from '../../../../../app/lib/server/lib/notifyListener';
+import { notifyOnLivechatInquiryChanged, notifyOnRoomChangedById } from '../../../../../app/lib/server/lib/notifyListener';
 import { allowAgentSkipQueue } from '../../../../../app/livechat/server/lib/Helper';
 import { saveQueueInquiry } from '../../../../../app/livechat/server/lib/QueueManager';
 import { setDepartmentForGuest } from '../../../../../app/livechat/server/lib/departmentsLib';
@@ -54,6 +54,8 @@ beforeRouteChat.patch(
 					void notifyOnLivechatInquiryChanged(updatedLivechatInquiry, 'updated', { department: updatedLivechatInquiry.department });
 				}
 
+				void notifyOnRoomChangedById(inquiry.rid, 'updated');
+
 				inquiry = updatedLivechatInquiry ?? inquiry;
 			}
 		}

apps/meteor/ee/app/livechat-enterprise/server/lib/SlaHelper.ts

@@ -2,17 +2,24 @@ import { Message } from '@rocket.chat/core-services';
 import type { IOmnichannelServiceLevelAgreements, IUser } from '@rocket.chat/core-typings';
 import { LivechatInquiry, LivechatRooms } from '@rocket.chat/models';
 
+import {
+	notifyOnLivechatInquiryChanged,
+	notifyOnRoomChangedById,
+	notifyOnLivechatInquiryChangedByRoom,
+} from '../../../../../app/lib/server/lib/notifyListener';
 import { callbacks } from '../../../../../lib/callbacks';
 
 export const removeSLAFromRooms = async (slaId: string, userId: string) => {
 	const extraQuery = await callbacks.run('livechat.applyRoomRestrictions', {}, { userId });
 	const openRooms = await LivechatRooms.findOpenBySlaId(slaId, { projection: { _id: 1 } }, extraQuery).toArray();
+	const openRoomIds: string[] = openRooms.map(({ _id }) => _id);
 	if (openRooms.length) {
-		const openRoomIds: string[] = openRooms.map(({ _id }) => _id);
 		await LivechatInquiry.bulkUnsetSla(openRoomIds);
+		void notifyOnLivechatInquiryChangedByRoom(openRoomIds, 'updated');
 	}
 
 	await LivechatRooms.bulkRemoveSlaFromRoomsById(slaId);
+	void notifyOnRoomChangedById(openRoomIds, 'updated');
 };
 
 export const updateInquiryQueueSla = async (roomId: string, sla: Pick<IOmnichannelServiceLevelAgreements, 'dueTimeInMinutes' | '_id'>) => {
@@ -29,6 +36,8 @@ export const updateInquiryQueueSla = async (roomId: string, sla: Pick<IOmnichann
 		slaId,
 		estimatedWaitingTimeQueue,
 	});
+
+	void notifyOnLivechatInquiryChanged({ ...inquiry, slaId, estimatedWaitingTimeQueue }, 'updated');
 };
 
 export const updateRoomSlaWeights = async (roomId: string, sla: Pick<IOmnichannelServiceLevelAgreements, 'dueTimeInMinutes' | '_id'>) => {

apps/meteor/server/services/omnichannel/queue.ts

@@ -1,10 +1,11 @@
 import { ServiceStarter } from '@rocket.chat/core-services';
-import { type InquiryWithAgentInfo, type IOmnichannelQueue } from '@rocket.chat/core-typings';
+import { LivechatInquiryStatus, type InquiryWithAgentInfo, type IOmnichannelQueue } from '@rocket.chat/core-typings';
 import { License } from '@rocket.chat/license';
 import { LivechatInquiry, LivechatRooms } from '@rocket.chat/models';
 import { tracerSpan } from '@rocket.chat/tracing';
 
 import { queueLogger } from './logger';
+import { notifyOnLivechatInquiryChangedByRoom } from '../../../app/lib/server/lib/notifyListener';
 import { getOmniChatSortQuery } from '../../../app/livechat/lib/inquiries';
 import { dispatchAgentDelegated } from '../../../app/livechat/server/lib/Helper';
 import { RoutingManager } from '../../../app/livechat/server/lib/RoutingManager';
@@ -188,6 +189,7 @@ export class OmnichannelQueue implements IOmnichannelQueue {
 					step: 'reconciliation',
 				});
 				await LivechatInquiry.removeByRoomId(roomId);
+				void notifyOnLivechatInquiryChangedByRoom(roomId, 'removed');
 				break;
 			}
 			case 'taken': {
@@ -199,6 +201,7 @@ export class OmnichannelQueue implements IOmnichannelQueue {
 				});
 				// Reconciliate served inquiries, by updating their status to taken after queue tried to pick and failed
 				await LivechatInquiry.takeInquiry(inquiryId);
+				void notifyOnLivechatInquiryChangedByRoom(roomId, 'updated', { status: LivechatInquiryStatus.TAKEN, takenAt: new Date() });
 				break;
 			}
 			case 'missing': {
@@ -209,6 +212,7 @@ export class OmnichannelQueue implements IOmnichannelQueue {
 					step: 'reconciliation',
 				});
 				await LivechatInquiry.removeByRoomId(roomId);
+				void notifyOnLivechatInquiryChangedByRoom(roomId, 'removed');
 				break;
 			}
 			default: {

packages/model-typings/src/models/ILivechatInquiryModel.ts

@@ -36,7 +36,7 @@ export interface ILivechatInquiryModel extends IBaseModel<ILivechatInquiryRecord
 	queueInquiry(inquiryId: string, lastMessage?: IMessage): Promise<ILivechatInquiryRecord | null>;
 	queueInquiryAndRemoveDefaultAgent(inquiryId: string): Promise<UpdateResult>;
 	readyInquiry(inquiryId: string): Promise<UpdateResult>;
-	changeDepartmentIdByRoomId(rid: string, department: string): Promise<void>;
+	changeDepartmentIdByRoomId(rid: string, department: string): Promise<UpdateResult>;
 	getStatus(inquiryId: string): Promise<ILivechatInquiryRecord['status'] | undefined>;
 	updateVisitorStatus(token: string, status: ILivechatInquiryRecord['v']['status']): Promise<UpdateResult>;
 	setDefaultAgentById(inquiryId: string, defaultAgent: ILivechatInquiryRecord['defaultAgent']): Promise<UpdateResult>;

packages/models/src/models/LivechatInquiry.ts

@@ -373,7 +373,7 @@ export class LivechatInquiryRaw extends BaseRaw<ILivechatInquiryRecord> implemen
 		);
 	}
 
-	async changeDepartmentIdByRoomId(rid: string, department: string): Promise<void> {
+	async changeDepartmentIdByRoomId(rid: string, department: string): Promise<UpdateResult> {
 		const query = {
 			rid,
 		};
@@ -383,7 +383,7 @@ export class LivechatInquiryRaw extends BaseRaw<ILivechatInquiryRecord> implemen
 			},
 		};
 
-		await this.updateOne(query, updateObj);
+		return this.updateOne(query, updateObj);
 	}
 
 	async getStatus(inquiryId: string): Promise<ILivechatInquiryRecord['status'] | undefined> {