[FIX][ENTERPRISE] Race condition on Omnichannel queues

app/livechat/server/lib/Helper.js

@@ -1,6 +1,5 @@
 import { Meteor } from 'meteor/meteor';
 import { Match, check } from 'meteor/check';
-import { MongoInternals } from 'meteor/mongo';
 import { LivechatTransferEventType } from '@rocket.chat/apps-engine/definition/livechat';
 
 import { Messages, LivechatRooms, Rooms, Subscriptions, Users, LivechatInquiry, LivechatDepartment, LivechatDepartmentAgents } from '../../../models/server';
@@ -94,7 +93,7 @@ export const createLivechatInquiry = ({ rid, name, guest, message, initialStatus
 	return LivechatInquiry.insert(inquiry);
 };
 
-export const createLivechatSubscription = (rid, name, guest, agent) => {
+export const createLivechatSubscription = (rid, name, guest, agent, department) => {
 	check(rid, String);
 	check(name, String);
 	check(guest, Match.ObjectIncluding({
@@ -131,44 +130,12 @@ export const createLivechatSubscription = (rid, name, guest, agent) => {
 			token,
 			status,
 		},
+		...department && { department },
 	};
 
 	return Subscriptions.insert(subscriptionData);
 };
 
-export const createLivechatQueueView = async () => {
-	const { mongo } = MongoInternals.defaultRemoteCollectionDriver();
-
-	// recreate the view on every startup
-	const list = await mongo.db.listCollections({ name: 'view_livechat_queue_status' }).toArray();
-	if (list.length > 0) {
-		await mongo.db.dropCollection('view_livechat_queue_status');
-	}
-
-	await mongo.db.createCollection('view_livechat_queue_status', { // name of the view to create
-		viewOn: 'rocketchat_room', // name of source collection from which to create the view
-		pipeline: [
-			{
-				$match: {
-					open: true,
-					servedBy: { $exists: true },
-				},
-			},
-			{
-				$group: {
-					_id: '$servedBy._id',
-					chats: { $sum: 1 },
-				},
-			},
-			{
-				$sort: {
-					chats: 1,
-				},
-			},
-		],
-	});
-};
-
 export const removeAgentFromSubscription = (rid, { _id, username }) => {
 	const room = LivechatRooms.findOneById(rid);
 	const user = Users.findOneById(_id);
@@ -287,6 +254,7 @@ export const forwardRoomToAgent = async (room, transferData) => {
 export const updateChatDepartment = ({ rid, newDepartmentId, oldDepartmentId }) => {
 	LivechatRooms.changeDepartmentIdByRoomId(rid, newDepartmentId);
 	LivechatInquiry.changeDepartmentIdByRoomId(rid, newDepartmentId);
+	Subscriptions.changeDepartmentByRoomId(rid, newDepartmentId);
 
 	Meteor.defer(() => {
 		Apps.triggerEvent(AppEvents.IPostLivechatRoomTransferred, {

app/livechat/server/lib/RoutingManager.js

@@ -47,7 +47,6 @@ export const RoutingManager = {
 	},
 
 	async delegateInquiry(inquiry, agent) {
-		// return Room Object
 		const { department, rid } = inquiry;
 		if (!agent || (agent.username && !Users.findOneOnlineAgentByUsername(agent.username))) {
 			agent = await this.getNextAgent(department);
@@ -66,8 +65,8 @@ export const RoutingManager = {
 			username: String,
 		}));
 
-		const { rid, name, v } = inquiry;
-		if (!createLivechatSubscription(rid, name, v, agent)) {
+		const { rid, name, v, department } = inquiry;
+		if (!createLivechatSubscription(rid, name, v, agent, department)) {
 			throw new Meteor.Error('error-creating-subscription', 'Error creating subscription');
 		}
 

app/livechat/server/startup.js

@@ -8,7 +8,6 @@ import { callbacks } from '../../callbacks';
 import { settings } from '../../settings';
 import { LivechatDepartment, LivechatDepartmentAgents, LivechatInquiry } from '../../models/server';
 import { RoutingManager } from './lib/RoutingManager';
-import { createLivechatQueueView } from './lib/Helper';
 import { LivechatAgentActivityMonitor } from './statistics/LivechatAgentActivityMonitor';
 import { businessHourManager } from './business-hour';
 import { createDefaultBusinessHourIfNotExists } from './business-hour/Helper';
@@ -91,7 +90,6 @@ Meteor.startup(async () => {
 		return user;
 	}, callbacks.priority.LOW, 'cant-join-room');
 
-	createLivechatQueueView();
 
 	const monitor = new LivechatAgentActivityMonitor();
 

app/models/server/index.js

@@ -37,6 +37,7 @@ import LivechatAgentActivity from './models/LivechatAgentActivity';
 import LivechatInquiry from './models/LivechatInquiry';
 import ReadReceipts from './models/ReadReceipts';
 import LivechatExternalMessage from './models/LivechatExternalMessages';
+import OmnichannelQueue from './models/OmnichannelQueue';
 import Analytics from './models/Analytics';
 
 export { AppsLogsModel } from './models/apps-logs-model';
@@ -88,4 +89,5 @@ export {
 	LivechatExternalMessage,
 	LivechatInquiry,
 	Analytics,
+	OmnichannelQueue,
 };

app/models/server/models/OmnichannelQueue.js

@@ -0,0 +1,9 @@
+import { Base } from './_Base';
+
+export class OmnichannelQueue extends Base {
+	constructor() {
+		super('omnichannel_queue');
+	}
+}
+
+export default new OmnichannelQueue();

app/models/server/models/Subscriptions.js

@@ -32,6 +32,7 @@ export class Subscriptions extends Base {
 		this.tryEnsureIndex({ autoTranslateLanguage: 1 }, { sparse: 1 });
 		this.tryEnsureIndex({ 'userHighlights.0': 1 }, { sparse: 1 });
 		this.tryEnsureIndex({ prid: 1 });
+		this.tryEnsureIndex({ 'u._id': 1, open: 1, department: 1 });
 	}
 
 	findByRoomIds(roomIds) {
@@ -281,6 +282,19 @@ export class Subscriptions extends Base {
 		return this.update(query, update);
 	}
 
+	changeDepartmentByRoomId(rid, department) {
+		const query = {
+			rid,
+		};
+		const update = {
+			$set: {
+				department,
+			},
+		};
+
+		this.update(query, update);
+	}
+
 	findAlwaysNotifyAudioUsersByRoomId(roomId) {
 		const query = {
 			rid: roomId,

app/models/server/models/Users.js

@@ -51,6 +51,7 @@ export class Users extends Base {
 		this.tryEnsureIndex({ isRemote: 1 }, { sparse: true });
 		this.tryEnsureIndex({ 'services.saml.inResponseTo': 1 });
 		this.tryEnsureIndex({ openBusinessHours: 1 }, { sparse: true });
+		this.tryEnsureIndex({ statusLivechat: 1 }, { sparse: true });
 	}
 
 	getLoginTokensByUserId(userId) {

app/models/server/raw/LivechatInquiry.js

@@ -15,4 +15,8 @@ export class LivechatInquiryRaw extends BaseRaw {
 		};
 		return this.findOne(query, options);
 	}
+
+	getDistinctQueuedDepartments() {
+		return this.col.distinct('department', { status: 'queued' });
+	}
 }

app/models/server/raw/OmnichannelQueue.ts

@@ -0,0 +1,72 @@
+/* eslint-disable @typescript-eslint/explicit-function-return-type */
+import {
+	Collection,
+} from 'mongodb';
+
+import { BaseRaw } from './BaseRaw';
+import { IOmnichannelQueueStatus } from '../../../../definition/IOmnichannel';
+
+const UNIQUE_QUEUE_ID = 'queue';
+export class OmnichannelQueueRaw extends BaseRaw {
+	public readonly col!: Collection<IOmnichannelQueueStatus>;
+
+	initQueue() {
+		return this.col.updateOne({
+			_id: UNIQUE_QUEUE_ID,
+		}, {
+			$unset: {
+				stoppedAt: 1,
+			},
+			$set: {
+				startedAt: new Date(),
+				locked: false,
+			},
+		}, {
+			upsert: true,
+		});
+	}
+
+	stopQueue() {
+		return this.col.updateOne({
+			_id: UNIQUE_QUEUE_ID,
+		}, {
+			$set: {
+				stoppedAt: new Date(),
+				locked: false,
+			},
+		});
+	}
+
+	async lockQueue() {
+		const result = await this.col.findOneAndUpdate({
+			_id: UNIQUE_QUEUE_ID,
+			locked: false,
+		}, {
+			$set: {
+				locked: true,
+			},
+		}, {
+			sort: {
+				_id: 1,
+			},
+		});
+
+		return result.value;
+	}
+
+	async unlockQueue() {
+		const result = await this.col.findOneAndUpdate({
+			_id: UNIQUE_QUEUE_ID,
+		}, {
+			$set: {
+				locked: false,
+			},
+		}, {
+			sort: {
+				_id: 1,
+			},
+		});
+
+		return result.value;
+	}
+}

app/models/server/raw/Users.js

@@ -119,10 +119,25 @@ export class UsersRaw extends BaseRaw {
 	async getNextLeastBusyAgent(department) {
 		const aggregate = [
 			{ $match: { status: { $exists: true, $ne: 'offline' }, statusLivechat: 'available', roles: 'livechat-agent' } },
-			{ $lookup: { from: 'view_livechat_queue_status', localField: '_id', foreignField: '_id', as: 'LivechatQueueStatus' } }, // the `view_livechat_queue_status` it's a view created when the server starts
+			{ $lookup: {
+				from: 'rocketchat_subscription',
+				let: { id: '$_id' },
+				pipeline: [{
+					$match: {
+						$expr: {
+							$and: [
+								{ $eq: ['$u._id', '$$id'] },
+								{ $eq: ['$open', true] },
+								{ ...department && { $eq: ['$department', department] } },
+							],
+						},
+					},
+				}],
+				as: 'subs' },
+			},
 			{ $lookup: { from: 'rocketchat_livechat_department_agents', localField: '_id', foreignField: 'agentId', as: 'departments' } },
-			{ $project: { agentId: '$_id', username: 1, lastRoutingTime: 1, departments: 1, queueInfo: { $arrayElemAt: ['$LivechatQueueStatus', 0] } } },
-			{ $sort: { 'queueInfo.chats': 1, lastRoutingTime: 1, username: 1 } },
+			{ $project: { agentId: '$_id', username: 1, lastRoutingTime: 1, departments: 1, count: { $size: '$subs' } } },
+			{ $sort: { count: 1, lastRoutingTime: 1, username: 1 } },
 		];
 
 		if (department) {
@@ -140,25 +155,27 @@ export class UsersRaw extends BaseRaw {
 		return agent;
 	}
 
-	setLastRoutingTime(userId) {
-		const query = {
-			_id: userId,
-		};
-
-		const update = {
-			$set: {
-				lastRoutingTime: new Date(),
-			},
-		};
-
-		return this.col.updateOne(query, update);
+	async setLastRoutingTime(userId) {
+		const result = await this.col.findAndModify(
+			{ _id: userId }
+			, {
+				sort: {
+					_id: 1,
+				},
+			}, {
+				$set: {
+					lastRoutingTime: new Date(),
+				},
+			});
+		return result.value;
 	}
 
 	async getAgentAndAmountOngoingChats(userId) {
 		const aggregate = [
 			{ $match: { _id: userId, status: { $exists: true, $ne: 'offline' }, statusLivechat: 'available', roles: 'livechat-agent' } },
-			{ $lookup: { from: 'view_livechat_queue_status', localField: '_id', foreignField: '_id', as: 'LivechatQueueStatus' } },
-			{ $project: { username: 1, queueInfo: { $arrayElemAt: ['$LivechatQueueStatus', 0] } } },
+			{ $lookup: { from: 'rocketchat_subscription', localField: '_id', foreignField: 'u._id', as: 'subs' } },
+			{ $project: { agentId: '$_id', username: 1, lastAssignTime: 1, lastRoutingTime: 1, 'queueInfo.chats': { $size: '$subs' } } },
+			{ $sort: { 'queueInfo.chats': 1, lastAssignTime: 1, lastRoutingTime: 1, username: 1 } },
 		];
 
 		const [agent] = await this.col.aggregate(aggregate).toArray();

app/models/server/raw/index.ts

@@ -52,6 +52,8 @@ import LivechatBusinessHoursModel from '../models/LivechatBusinessHours';
 import { LivechatBusinessHoursRaw } from './LivechatBusinessHours';
 import ServerEventModel from '../models/ServerEvents';
 import { ServerEventsRaw } from './ServerEvents';
+import OmnichannelQueueModel from '../models/OmnichannelQueue';
+import { OmnichannelQueueRaw } from './OmnichannelQueue';
 
 export const Permissions = new PermissionsRaw(PermissionsModel.model.rawCollection());
 export const Roles = new RolesRaw(RolesModel.model.rawCollection());
@@ -80,3 +82,4 @@ export const Statistics = new StatisticsRaw(StatisticsModel.model.rawCollection(
 export const NotificationQueue = new NotificationQueueRaw(NotificationQueueModel.model.rawCollection());
 export const LivechatBusinessHours = new LivechatBusinessHoursRaw(LivechatBusinessHoursModel.model.rawCollection());
 export const ServerEvents = new ServerEventsRaw(ServerEventModel.model.rawCollection());
+export const OmnichannelQueue = new OmnichannelQueueRaw(OmnichannelQueueModel.model.rawCollection());

definition/IOmnichannel.ts

@@ -0,0 +1,6 @@
+export interface IOmnichannelQueueStatus {
+	_id: string;
+	startedAt: Date;
+	stoppedAt?: Date;
+	locked: boolean;
+}

ee/app/livechat-enterprise/server/hooks/afterTakeInquiry.js

@@ -1,6 +1,6 @@
 import { callbacks } from '../../../../../app/callbacks';
 import { settings } from '../../../../../app/settings';
-import { checkWaitingQueue, dispatchWaitingQueueStatus } from '../lib/Helper';
+import { dispatchWaitingQueueStatus } from '../lib/Helper';
 
 callbacks.add('livechat.afterTakeInquiry', async (inquiry) => {
 	if (!settings.get('Livechat_waiting_queue')) {
@@ -13,7 +13,6 @@ callbacks.add('livechat.afterTakeInquiry', async (inquiry) => {
 
 	const { department } = inquiry;
 	await dispatchWaitingQueueStatus(department);
-	await checkWaitingQueue(department);
 
 	return inquiry;
 }, callbacks.priority.MEDIUM, 'livechat-after-take-inquiry');

ee/app/livechat-enterprise/server/hooks/beforeRoutingChat.js

@@ -1,7 +1,7 @@
 import { callbacks } from '../../../../../app/callbacks';
 import { settings } from '../../../../../app/settings';
 import { LivechatInquiry } from '../../../../../app/models/server';
-import { dispatchInquiryPosition, checkWaitingQueue } from '../lib/Helper';
+import { dispatchInquiryPosition } from '../lib/Helper';
 
 callbacks.add('livechat.beforeRouteChat', async (inquiry, agent) => {
 	if (!settings.get('Livechat_waiting_queue')) {
@@ -12,7 +12,7 @@ callbacks.add('livechat.beforeRouteChat', async (inquiry, agent) => {
 		return inquiry;
 	}
 
-	const { _id, status, department } = inquiry;
+	const { _id, status } = inquiry;
 
 	if (status !== 'ready') {
 		return inquiry;
@@ -25,7 +25,5 @@ callbacks.add('livechat.beforeRouteChat', async (inquiry, agent) => {
 		dispatchInquiryPosition(inq);
 	}
 
-	await checkWaitingQueue(department);
-
 	return LivechatInquiry.findOneById(_id);
 }, callbacks.priority.HIGH, 'livechat-before-routing-chat');