fix: OTR messages not working when running microservices (#32230)

.changeset/strange-countries-visit.md

@@ -0,0 +1,6 @@
+---
+"@rocket.chat/meteor": patch
+"@rocket.chat/core-services": patch
+---
+
+Fixed a problem that caused OTR Session messages' to not being transmitted from one peer to another when running Rocket.Chat as microservices. This was caused by a legacy streamer that tried to use the websocket directly, which works on monolith but doesn't on microservices, cause these events are routed through DDP Streamer service.

apps/meteor/app/lib/server/functions/sendMessage.ts

@@ -1,5 +1,5 @@
 import { Apps } from '@rocket.chat/apps';
-import { Message } from '@rocket.chat/core-services';
+import { Message, api } from '@rocket.chat/core-services';
 import type { IMessage, IRoom } from '@rocket.chat/core-typings';
 import { Messages } from '@rocket.chat/models';
 import { Match, check } from 'meteor/check';
@@ -10,7 +10,6 @@ import { isURL } from '../../../../lib/utils/isURL';
 import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { hasPermissionAsync } from '../../../authorization/server/functions/hasPermission';
 import { FileUpload } from '../../../file-upload/server';
-import notifications from '../../../notifications/server/lib/Notifications';
 import { settings } from '../../../settings/server';
 import { parseUrlsInMessage } from './parseUrlsInMessage';
 
@@ -216,7 +215,7 @@ export const sendMessage = async function (user: any, message: any, room: any, u
 	prepareMessageObject(message, room._id, user);
 
 	if (message.t === 'otr') {
-		notifications.streamRoomMessage.emit(message.rid, message, user, room);
+		void api.broadcast('otrMessage', { roomId: message.rid, message, user, room });
 		return message;
 	}
 

apps/meteor/app/otr/server/methods/updateOTRAck.ts

@@ -1,9 +1,8 @@
+import { api } from '@rocket.chat/core-services';
 import type { IOTRMessage } from '@rocket.chat/core-typings';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
-import notifications from '../../../notifications/server/lib/Notifications';
-
 declare module '@rocket.chat/ui-contexts' {
 	// eslint-disable-next-line @typescript-eslint/naming-convention
 	interface ServerMethods {
@@ -16,7 +15,7 @@ Meteor.methods<ServerMethods>({
 		if (!Meteor.userId()) {
 			throw new Meteor.Error('error-invalid-user', 'Invalid user', { method: 'updateOTRAck' });
 		}
-		const acknowlegedMessage: IOTRMessage = { ...message, otrAck: ack };
-		notifications.streamRoomMessage.emit(message.rid, acknowlegedMessage);
+		const acknowledgeMessage: IOTRMessage = { ...message, otrAck: ack };
+		void api.broadcast('otrAckUpdate', { roomId: message.rid, acknowledgeMessage });
 	},
 });

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

@@ -3,7 +3,7 @@ import type { ISetting as AppsSetting } from '@rocket.chat/apps-engine/definitio
 import type { IServiceClass } from '@rocket.chat/core-services';
 import { EnterpriseSettings } from '@rocket.chat/core-services';
 import { isSettingColor, isSettingEnterprise, UserStatus } from '@rocket.chat/core-typings';
-import type { IUser, IRoom, VideoConference, ISetting, IOmnichannelRoom } from '@rocket.chat/core-typings';
+import type { IUser, IRoom, VideoConference, ISetting, IOmnichannelRoom, IMessage, IOTRMessage } from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
 import { parse } from '@rocket.chat/message-parser';
 
@@ -485,5 +485,12 @@ export class ListenersModule {
 			notifications.streamApps.emitWithoutBroadcast('actions/changed');
 			notifications.streamApps.emitWithoutBroadcast('apps', ['actions/changed', []]);
 		});
+
+		service.onEvent('otrMessage', ({ roomId, message, user, room }: { roomId: string; message: IMessage; user: IUser; room: IRoom }) => {
+			notifications.streamRoomMessage.emit(roomId, message, user, room);
+		});
+		service.onEvent('otrAckUpdate', ({ roomId, acknowledgeMessage }: { roomId: string; acknowledgeMessage: IOTRMessage }) => {
+			notifications.streamRoomMessage.emit(roomId, acknowledgeMessage);
+		});
 	}
 }

packages/core-services/src/events/Events.ts

@@ -33,6 +33,7 @@ import type {
 	LicenseLimitKind,
 	ICustomUserStatus,
 	IWebdavAccount,
+	IOTRMessage,
 } from '@rocket.chat/core-typings';
 import type * as UiKit from '@rocket.chat/ui-kit';
 
@@ -301,4 +302,6 @@ export type EventSignatures = {
 	'command.updated'(command: string): void;
 	'command.removed'(command: string): void;
 	'actions.changed'(): void;
+	'otrMessage'(data: { roomId: string; message: IMessage; room: IRoom; user: IUser }): void;
+	'otrAckUpdate'(data: { roomId: string; acknowledgeMessage: IOTRMessage }): void;
 };