[FIX][ENTERPRISE] Omnichannel real time data on micro services (#26703)

Co-authored-by: Diego Sampaio <chinello@gmail.com>

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

@@ -21,11 +21,11 @@ import { callbacks } from '../../../../lib/callbacks';
 import { Logger } from '../../../logger';
 import { settings } from '../../../settings/server';
 import { Apps, AppEvents } from '../../../apps/server';
-import notifications from '../../../notifications/server/lib/Notifications';
 import { sendNotification } from '../../../lib/server';
 import { sendMessage } from '../../../lib/server/functions/sendMessage';
 import { queueInquiry, saveQueueInquiry } from './QueueManager';
 import { validateEmail as validatorFunc } from '../../../../lib/emailValidator';
+import { api } from '../../../../server/sdk/api';
 
 const logger = new Logger('LivechatHelper');
 
@@ -268,7 +268,7 @@ export const normalizeAgent = (agentId) => {
 export const dispatchAgentDelegated = (rid, agentId) => {
 	const agent = normalizeAgent(agentId);
 
-	notifications.streamLivechatRoom.emit(rid, {
+	api.broadcast('omnichannel.room', rid, {
 		type: 'agentData',
 		data: agent,
 	});

apps/meteor/app/livechat/server/lib/Livechat.js

@@ -37,10 +37,10 @@ import { FileUpload } from '../../../file-upload/server';
 import { normalizeTransferredByData, parseAgentCustomFields, updateDepartmentAgents, validateEmail } from './Helper';
 import { Apps, AppEvents } from '../../../apps/server';
 import { businessHourManager } from '../business-hour';
-import notifications from '../../../notifications/server/lib/Notifications';
 import { addUserRoles } from '../../../../server/lib/roles/addUserRoles';
 import { removeUserFromRoles } from '../../../../server/lib/roles/removeUserFromRoles';
 import { VideoConf } from '../../../../server/sdk';
+import { api } from '../../../../server/sdk/api';
 
 const logger = new Logger('Livechat');
 
@@ -1373,7 +1373,7 @@ export const Livechat = {
 		}
 
 		LivechatRooms.findOpenByAgent(userId).forEach((room) => {
-			notifications.streamLivechatRoom.emit(room._id, {
+			api.broadcast('omnichannel.room', room._id, {
 				type: 'agentStatus',
 				status,
 			});
@@ -1389,7 +1389,7 @@ export const Livechat = {
 	},
 
 	notifyRoomVisitorChange(roomId, visitor) {
-		notifications.streamLivechatRoom.emit(roomId, {
+		api.broadcast('omnichannel.room', roomId, {
 			type: 'visitorData',
 			visitor,
 		});

apps/meteor/ee/app/livechat-enterprise/server/lib/Helper.js

@@ -12,9 +12,9 @@ import { Users, LivechatInquiry, LivechatRooms, Messages } from '../../../../../
 import { settings } from '../../../../../app/settings/server';
 import { RoutingManager } from '../../../../../app/livechat/server/lib/RoutingManager';
 import { dispatchAgentDelegated } from '../../../../../app/livechat/server/lib/Helper';
-import notifications from '../../../../../app/notifications/server/lib/Notifications';
 import { logger, helperLogger } from './logger';
 import { OmnichannelQueueInactivityMonitor } from './QueueInactivityMonitor';
+import { api } from '../../../../../server/sdk/api';
 
 export const getMaxNumberSimultaneousChat = async ({ agentId, departmentId }) => {
 	if (departmentId) {
@@ -81,7 +81,7 @@ export const dispatchInquiryPosition = async (inquiry, queueInfo) => {
 	const { position, department } = inquiry;
 	const data = await normalizeQueueInfo({ position, queueInfo, department });
 	const propagateInquiryPosition = Meteor.bindEnvironment((inquiry) => {
-		notifications.streamLivechatRoom.emit(inquiry.rid, {
+		api.broadcast('omnichannel.room', inquiry.rid, {
 			type: 'queueData',
 			data,
 		});

apps/meteor/ee/server/lib/registerServiceModels.ts

@@ -1,4 +1,11 @@
-import type { Db } from 'mongodb';
+import type { Collection, Db } from 'mongodb';
+import type {
+	ILivechatDepartmentAgents,
+	ILivechatInquiryRecord,
+	ISetting,
+	ISubscription,
+	RocketChatRecordDeleted,
+} from '@rocket.chat/core-typings';
 import { registerModel } from '@rocket.chat/models';
 
 import { RolesRaw } from '../../../server/models/raw/Roles';
@@ -20,21 +27,30 @@ import { IntegrationsRaw } from '../../../server/models/raw/Integrations';
 import { EmailInboxRaw } from '../../../server/models/raw/EmailInbox';
 import { PbxEventsRaw } from '../../../server/models/raw/PbxEvents';
 
-// TODO add trash param to model instances
-export const registerServiceModels = (db: Db): void => {
+// TODO add trash param to appropiate model instances
+export function registerServiceModels(db: Db, trash?: Collection): void {
 	registerModel('IRolesModel', () => new RolesRaw(db));
 	registerModel('IRoomsModel', () => new RoomsRaw(db));
-	registerModel('ISettingsModel', () => new SettingsRaw(db));
-	registerModel('ISubscriptionsModel', () => new SubscriptionsRaw(db));
+	registerModel('ISettingsModel', () => new SettingsRaw(db, trash as unknown as Collection<RocketChatRecordDeleted<ISetting>>));
+	registerModel(
+		'ISubscriptionsModel',
+		() => new SubscriptionsRaw(db, trash as unknown as Collection<RocketChatRecordDeleted<ISubscription>>),
+	);
 	registerModel('ITeamModel', () => new TeamRaw(db));
 	registerModel('ITeamMemberModel', () => new TeamMemberRaw(db));
 	registerModel('IUsersModel', () => new UsersRaw(db));
 
 	// @ts-ignore-error
 	registerModel('IMessagesModel', () => new MessagesRaw(db));
 
-	registerModel('ILivechatInquiryModel', () => new LivechatInquiryRaw(db));
-	registerModel('ILivechatDepartmentAgentsModel', () => new LivechatDepartmentAgentsRaw(db));
+	registerModel(
+		'ILivechatInquiryModel',
+		() => new LivechatInquiryRaw(db, trash as unknown as Collection<RocketChatRecordDeleted<ILivechatInquiryRecord>>),
+	);
+	registerModel(
+		'ILivechatDepartmentAgentsModel',
+		() => new LivechatDepartmentAgentsRaw(db, trash as unknown as Collection<RocketChatRecordDeleted<ILivechatDepartmentAgents>>),
+	);
 	registerModel('IUsersSessionsModel', () => new UsersSessionsRaw(db));
 	registerModel('IPermissionsModel', () => new PermissionsRaw(db));
 	registerModel('ILoginServiceConfigurationModel', () => new LoginServiceConfigurationRaw(db));
@@ -43,4 +59,4 @@ export const registerServiceModels = (db: Db): void => {
 	registerModel('IIntegrationsModel', () => new IntegrationsRaw(db));
 	registerModel('IEmailInboxModel', () => new EmailInboxRaw(db));
 	registerModel('IPbxEventsModel', () => new PbxEventsRaw(db));
-};
+}

apps/meteor/ee/server/services/authorization/service.ts

@@ -1,12 +1,16 @@
+import type { Document } from 'mongodb';
+
 import '../../startup/broker';
 
 import { api } from '../../../../server/sdk/api';
 import { Authorization } from '../../../../server/services/authorization/service';
-import { getConnection } from '../mongo';
+import { Collections, getCollection, getConnection } from '../mongo';
 import { registerServiceModels } from '../../lib/registerServiceModels';
 
-getConnection().then((db) => {
-	registerServiceModels(db);
+getConnection().then(async (db) => {
+	const trash = await getCollection<Document>(Collections.Trash);
+
+	registerServiceModels(db, trash);
 
 	api.registerService(new Authorization(db));
 });

apps/meteor/ee/server/services/mongo.ts

@@ -9,7 +9,7 @@ export enum Collections {
 	Subscriptions = 'rocketchat_subscription',
 	UserSession = 'usersSessions',
 	User = 'users',
-	Trash = 'rocketchat_trash',
+	Trash = 'rocketchat__trash',
 	Messages = 'rocketchat_message',
 	Rooms = 'rocketchat_room',
 	Settings = 'rocketchat_settings',
@@ -40,6 +40,7 @@ export const getConnection = ((): ((options?: MongoClientOptions) => Promise<Db>
 				db = c.db(name);
 			});
 		}
+
 		// if getConnection was called multiple times before it was connected, wait for the connection
 		return (await client).db(name);
 	};

apps/meteor/ee/server/services/package.json

@@ -68,6 +68,6 @@
 		"typescript": "~4.5.5"
 	},
 	"volta": {
-		"extends": "../../package.json"
+		"extends": "../../../package.json"
 	}
 }

apps/meteor/ee/server/services/stream-hub/service.ts

@@ -1,11 +1,15 @@
+import type { Document } from 'mongodb';
+
 import '../../startup/broker';
 
 import { api } from '../../../../server/sdk/api';
-import { getConnection } from '../mongo';
+import { Collections, getCollection, getConnection } from '../mongo';
 import { registerServiceModels } from '../../lib/registerServiceModels';
 
 getConnection().then(async (db) => {
-	registerServiceModels(db);
+	const trash = await getCollection<Document>(Collections.Trash);
+
+	registerServiceModels(db, trash);
 
 	// need to import StreamHub service after models are registered
 	const { StreamHub } = await import('./StreamHub');

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

@@ -345,5 +345,8 @@ export class ListenersModule {
 		service.onEvent('connector.statuschanged', (enabled): void => {
 			notifications.notifyLoggedInThisInstance('voip.statuschanged', enabled);
 		});
+		service.onEvent('omnichannel.room', (roomId, data): void => {
+			notifications.streamLivechatRoom.emitWithoutBroadcast(roomId, data);
+		});
 	}
 }

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

@@ -124,6 +124,10 @@ export type EventSignatures = {
 		diff?: undefined | Record<string, any>;
 		id: string;
 	}): void;
+	'omnichannel.room'(
+		roomId: string,
+		data: { type: 'agentStatus'; status: string } | { type: 'queueData' | 'agentData'; data: { [k: string]: unknown } },
+	): void;
 
 	// Send all events from here
 	'voip.events'(userId: string, data: VoipEventDataSignature): void;

ee/apps/ddp-streamer/src/streams.ts

@@ -1,12 +1,16 @@
+import type { Document } from 'mongodb';
+
 import { NotificationsModule } from '../../../../apps/meteor/server/modules/notifications/notifications.module';
 import { Stream } from './Streamer';
-import { getConnection } from '../../../../apps/meteor/ee/server/services/mongo';
+import { Collections, getCollection, getConnection } from '../../../../apps/meteor/ee/server/services/mongo';
 import { registerServiceModels } from '../../../../apps/meteor/ee/server/lib/registerServiceModels';
 
 export const notifications = new NotificationsModule(Stream);
 
-getConnection().then((db) => {
-	registerServiceModels(db);
+getConnection().then(async (db) => {
+	const trash = await getCollection<Document>(Collections.Trash);
+
+	registerServiceModels(db, trash);
 
 	notifications.configure();
 });