chore: use single message event (RocketChat#31527)

Co-authored-by: Pierre Lehnen <[email protected]>
parampavar · Jan 24, 2024 · 3b4176d · 3b4176d
1 parent 9310e84
commit 3b4176d
Show file tree

Hide file tree

Showing 22 changed files with 91 additions and 123 deletions.
diff --git a/apps/meteor/app/autotranslate/server/autotranslate.ts b/apps/meteor/app/autotranslate/server/autotranslate.ts
@@ -1,4 +1,3 @@
-import { api } from '@rocket.chat/core-services';
 import type {
 	IMessage,
 	IRoom,
@@ -16,7 +15,7 @@ import _ from 'underscore';
 
 import { callbacks } from '../../../lib/callbacks';
 import { isTruthy } from '../../../lib/isTruthy';
-import { broadcastMessageSentEvent } from '../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../server/modules/watchers/lib/messages';
 import { Markdown } from '../../markdown/server';
 import { settings } from '../../settings/server';
 
@@ -333,9 +332,8 @@ export abstract class AutoTranslate {
 	}
 
 	private notifyTranslatedMessage(messageId: string): void {
-		void broadcastMessageSentEvent({
+		void broadcastMessageFromData({
 			id: messageId,
-			broadcastCallback: (message) => api.broadcast('message.sent', message),
 		});
 	}
 

diff --git a/apps/meteor/app/discussion/server/hooks/propagateDiscussionMetadata.ts b/apps/meteor/app/discussion/server/hooks/propagateDiscussionMetadata.ts
@@ -1,20 +1,18 @@
-import { api } from '@rocket.chat/core-services';
 import type { IRoom } from '@rocket.chat/core-typings';
 import { Messages, Rooms } from '@rocket.chat/models';
 
 import { callbacks } from '../../../../lib/callbacks';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { deleteRoom } from '../../../lib/server/functions/deleteRoom';
 
 const updateAndNotifyParentRoomWithParentMessage = async (room: IRoom): Promise<void> => {
 	const { value: parentMessage } = await Messages.refreshDiscussionMetadata(room);
 	if (!parentMessage) {
 		return;
 	}
-	void broadcastMessageSentEvent({
+	void broadcastMessageFromData({
 		id: parentMessage._id,
 		data: parentMessage,
-		broadcastCallback: (message) => api.broadcast('message.sent', message),
 	});
 };
 

diff --git a/apps/meteor/app/federation/server/endpoints/dispatch.js b/apps/meteor/app/federation/server/endpoints/dispatch.js
@@ -3,7 +3,7 @@ import { eventTypes } from '@rocket.chat/core-typings';
 import { FederationServers, FederationRoomEvents, Rooms, Messages, Subscriptions, Users, ReadReceipts } from '@rocket.chat/models';
 import EJSON from 'ejson';
 
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { API } from '../../../api/server';
 import { FileUpload } from '../../../file-upload/server';
 import { deleteRoom } from '../../../lib/server/functions/deleteRoom';
@@ -284,10 +284,9 @@ const eventHandlers = {
 				}
 			}
 			if (messageForNotification) {
-				void broadcastMessageSentEvent({
+				void broadcastMessageFromData({
 					id: messageForNotification._id,
 					data: messageForNotification,
-					broadcastCallback: (message) => api.broadcast('message.sent', message),
 				});
 			}
 		}
@@ -316,14 +315,13 @@ const eventHandlers = {
 			} else {
 				// Update the message
 				await Messages.updateOne({ _id: persistedMessage._id }, { $set: { msg: message.msg, federation: message.federation } });
-				void broadcastMessageSentEvent({
+				void broadcastMessageFromData({
 					id: persistedMessage._id,
 					data: {
 						...persistedMessage,
 						msg: message.msg,
 						federation: message.federation,
 					},
-					broadcastCallback: (message) => api.broadcast('message.sent', message),
 				});
 			}
 		}
@@ -387,7 +385,7 @@ const eventHandlers = {
 
 			// Update the property
 			await Messages.updateOne({ _id: messageId }, { $set: { [`reactions.${reaction}`]: reactionObj } });
-			void broadcastMessageSentEvent({
+			void broadcastMessageFromData({
 				id: persistedMessage._id,
 				data: {
 					...persistedMessage,
@@ -396,7 +394,6 @@ const eventHandlers = {
 						[reaction]: reactionObj,
 					},
 				},
-				broadcastCallback: (message) => api.broadcast('message.sent', message),
 			});
 		}
 
@@ -446,7 +443,7 @@ const eventHandlers = {
 				// Otherwise, update the property
 				await Messages.updateOne({ _id: messageId }, { $set: { [`reactions.${reaction}`]: reactionObj } });
 			}
-			void broadcastMessageSentEvent({
+			void broadcastMessageFromData({
 				id: persistedMessage._id,
 				data: {
 					...persistedMessage,
@@ -455,7 +452,6 @@ const eventHandlers = {
 						[reaction]: reactionObj,
 					},
 				},
-				broadcastCallback: (message) => api.broadcast('message.sent', message),
 			});
 		}
 

diff --git a/apps/meteor/app/lib/server/functions/deleteMessage.ts b/apps/meteor/app/lib/server/functions/deleteMessage.ts
@@ -5,7 +5,7 @@ import { Meteor } from 'meteor/meteor';
 
 import { Apps } from '../../../../ee/server/apps';
 import { callbacks } from '../../../../lib/callbacks';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { canDeleteMessageAsync } from '../../../authorization/server/functions/canDeleteMessage';
 import { FileUpload } from '../../../file-upload/server';
 import { settings } from '../../../settings/server';
@@ -90,9 +90,8 @@ export async function deleteMessage(message: IMessage, user: IUser): Promise<voi
 	await callbacks.run('afterDeleteMessage', deletedMsg, room);
 
 	if (keepHistory || showDeletedStatus) {
-		void broadcastMessageSentEvent({
+		void broadcastMessageFromData({
 			id: message._id,
-			broadcastCallback: (message) => api.broadcast('message.sent', message),
 		});
 	}
 

diff --git a/apps/meteor/app/lib/server/functions/sendMessage.ts b/apps/meteor/app/lib/server/functions/sendMessage.ts
@@ -1,4 +1,4 @@
-import { Message, api } from '@rocket.chat/core-services';
+import { Message } 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';
@@ -7,7 +7,7 @@ import { Apps } from '../../../../ee/server/apps';
 import { callbacks } from '../../../../lib/callbacks';
 import { isRelativeURL } from '../../../../lib/utils/isRelativeURL';
 import { isURL } from '../../../../lib/utils/isURL';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+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';
@@ -285,9 +285,8 @@ export const sendMessage = async function (user: any, message: any, room: any, u
 
 	// Execute all callbacks
 	await callbacks.run('afterSaveMessage', message, room);
-	void broadcastMessageSentEvent({
+	void broadcastMessageFromData({
 		id: message._id,
-		broadcastCallback: (message) => api.broadcast('message.sent', message),
 	});
 	return message;
 };
diff --git a/apps/meteor/app/lib/server/functions/updateMessage.ts b/apps/meteor/app/lib/server/functions/updateMessage.ts
@@ -1,11 +1,11 @@
-import { Message, api } from '@rocket.chat/core-services';
+import { Message } from '@rocket.chat/core-services';
 import type { IEditedMessage, IMessage, IUser, AtLeast } from '@rocket.chat/core-typings';
 import { Messages, Rooms } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
 import { Apps } from '../../../../ee/server/apps';
 import { callbacks } from '../../../../lib/callbacks';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { settings } from '../../../settings/server';
 import { parseUrlsInMessage } from './parseUrlsInMessage';
 
@@ -85,10 +85,9 @@ export const updateMessage = async function (
 		const msg = await Messages.findOneById(_id);
 		if (msg) {
 			await callbacks.run('afterSaveMessage', msg, room, user._id);
-			void broadcastMessageSentEvent({
+			void broadcastMessageFromData({
 				id: msg._id,
 				data: msg,
-				broadcastCallback: (message) => api.broadcast('message.sent', message),
 			});
 		}
 	});

diff --git a/apps/meteor/app/message-pin/server/pinMessage.ts b/apps/meteor/app/message-pin/server/pinMessage.ts
@@ -1,4 +1,4 @@
-import { Message, api } from '@rocket.chat/core-services';
+import { Message } from '@rocket.chat/core-services';
 import { isQuoteAttachment, isRegisterUser } from '@rocket.chat/core-typings';
 import type { IMessage, MessageAttachment, MessageQuoteAttachment } from '@rocket.chat/core-typings';
 import { Messages, Rooms, Subscriptions, Users, ReadReceipts } from '@rocket.chat/models';
@@ -8,7 +8,7 @@ import { Meteor } from 'meteor/meteor';
 
 import { Apps, AppEvents } from '../../../ee/server/apps/orchestrator';
 import { isTruthy } from '../../../lib/isTruthy';
-import { broadcastMessageSentEvent } from '../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../server/modules/watchers/lib/messages';
 import { canAccessRoomAsync, roomAccessAttributes } from '../../authorization/server';
 import { hasPermissionAsync } from '../../authorization/server/functions/hasPermission';
 import { isTheLastMessage } from '../../lib/server/functions/isTheLastMessage';
@@ -222,9 +222,8 @@ Meteor.methods<ServerMethods>({
 		if (settings.get('Message_Read_Receipt_Store_Users')) {
 			await ReadReceipts.setPinnedByMessageId(originalMessage._id, originalMessage.pinned);
 		}
-		void broadcastMessageSentEvent({
+		void broadcastMessageFromData({
 			id: message._id,
-			broadcastCallback: (message) => api.broadcast('message.sent', message),
 		});
 
 		return true;

diff --git a/apps/meteor/app/message-star/server/starMessage.ts b/apps/meteor/app/message-star/server/starMessage.ts
@@ -1,11 +1,10 @@
-import { api } from '@rocket.chat/core-services';
 import type { IMessage } from '@rocket.chat/core-typings';
 import { Messages, Subscriptions, Rooms } from '@rocket.chat/models';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
 import { Apps, AppEvents } from '../../../ee/server/apps/orchestrator';
-import { broadcastMessageSentEvent } from '../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../server/modules/watchers/lib/messages';
 import { canAccessRoomAsync, roomAccessAttributes } from '../../authorization/server';
 import { isTheLastMessage } from '../../lib/server/functions/isTheLastMessage';
 import { settings } from '../../settings/server';
@@ -62,9 +61,8 @@ Meteor.methods<ServerMethods>({
 
 		await Messages.updateUserStarById(message._id, uid, message.starred);
 
-		void broadcastMessageSentEvent({
+		void broadcastMessageFromData({
 			id: message._id,
-			broadcastCallback: (message) => api.broadcast('message.sent', message),
 		});
 
 		return true;

diff --git a/apps/meteor/app/reactions/server/setReaction.ts b/apps/meteor/app/reactions/server/setReaction.ts
@@ -8,7 +8,7 @@ import _ from 'underscore';
 import { AppEvents, Apps } from '../../../ee/server/apps/orchestrator';
 import { callbacks } from '../../../lib/callbacks';
 import { i18n } from '../../../server/lib/i18n';
-import { broadcastMessageSentEvent } from '../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../server/modules/watchers/lib/messages';
 import { canAccessRoomAsync } from '../../authorization/server';
 import { hasPermissionAsync } from '../../authorization/server/functions/hasPermission';
 import { emoji } from '../../emoji/server';
@@ -108,9 +108,8 @@ async function setReaction(room: IRoom, user: IUser, message: IMessage, reaction
 
 	await Apps.triggerEvent(AppEvents.IPostMessageReacted, message, user, reaction, isReacted);
 
-	void broadcastMessageSentEvent({
+	void broadcastMessageFromData({
 		id: message._id,
-		broadcastCallback: (message) => api.broadcast('message.sent', message),
 	});
 }
 

diff --git a/apps/meteor/app/threads/server/hooks/aftersavemessage.ts b/apps/meteor/app/threads/server/hooks/aftersavemessage.ts
@@ -1,11 +1,10 @@
-import { api } from '@rocket.chat/core-services';
 import type { IMessage, IRoom } from '@rocket.chat/core-typings';
 import { isEditedMessage } from '@rocket.chat/core-typings';
 import { Messages } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
 import { callbacks } from '../../../../lib/callbacks';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 import { updateThreadUsersSubscriptions, getMentions } from '../../../lib/server/lib/notifyUsersOnMessage';
 import { sendMessageNotifications } from '../../../lib/server/lib/sendNotificationsOnMessage';
 import { settings } from '../../../settings/server';
@@ -63,9 +62,8 @@ export async function processThreads(message: IMessage, room: IRoom) {
 	await notifyUsersOnReply(message, replies, room);
 	await metaData(message, parentMessage, replies);
 	await notification(message, room, replies);
-	void broadcastMessageSentEvent({
+	void broadcastMessageFromData({
 		id: message.tmid,
-		broadcastCallback: (message) => api.broadcast('message.sent', message),
 	});
 
 	return message;

diff --git a/apps/meteor/ee/server/lib/message-read-receipt/ReadReceipt.js b/apps/meteor/ee/server/lib/message-read-receipt/ReadReceipt.js
@@ -5,7 +5,7 @@ import { Random } from '@rocket.chat/random';
 import { settings } from '../../../../app/settings/server';
 import { SystemLogger } from '../../../../server/lib/logger/system';
 import { roomCoordinator } from '../../../../server/lib/rooms/roomCoordinator';
-import { broadcastMessageSentEvent } from '../../../../server/modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../../../server/modules/watchers/lib/messages';
 
 // debounced function by roomId, so multiple calls within 2 seconds to same roomId runs only once
 const list = {};
@@ -68,9 +68,8 @@ export const ReadReceipt = {
 		if (isUserAlone) {
 			const result = await Messages.setAsReadById(message._id);
 			if (result.modifiedCount > 0) {
-				void broadcastMessageSentEvent({
+				void broadcastMessageFromData({
 					id: message._id,
-					broadcastCallback: (message) => api.broadcast('message.sent', message),
 				});
 			}
 		}

diff --git a/apps/meteor/server/features/EmailInbox/EmailInbox_Incoming.ts b/apps/meteor/server/features/EmailInbox/EmailInbox_Incoming.ts
@@ -1,4 +1,3 @@
-import { api } from '@rocket.chat/core-services';
 import type {
 	ILivechatVisitor,
 	IOmnichannelRoom,
@@ -17,7 +16,7 @@ import { Livechat as LivechatTyped } from '../../../app/livechat/server/lib/Live
 import { QueueManager } from '../../../app/livechat/server/lib/QueueManager';
 import { settings } from '../../../app/settings/server';
 import { i18n } from '../../lib/i18n';
-import { broadcastMessageSentEvent } from '../../modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../modules/watchers/lib/messages';
 import { logger } from './logger';
 
 type FileAttachment = VideoAttachmentProps & ImageAttachmentProps & AudioAttachmentProps;
@@ -238,9 +237,8 @@ export async function onEmailReceived(email: ParsedMail, inbox: string, departme
 				},
 			);
 			room && (await LivechatRooms.updateEmailThreadByRoomId(room._id, thread));
-			void broadcastMessageSentEvent({
+			void broadcastMessageFromData({
 				id: msgId,
-				broadcastCallback: (message) => api.broadcast('message.sent', message),
 			});
 		})
 		.catch((err) => {

diff --git a/apps/meteor/server/features/EmailInbox/EmailInbox_Outgoing.ts b/apps/meteor/server/features/EmailInbox/EmailInbox_Outgoing.ts
@@ -1,4 +1,3 @@
-import { api } from '@rocket.chat/core-services';
 import { isIMessageInbox } from '@rocket.chat/core-typings';
 import type { IEmailInbox, IUser, IMessage, IOmnichannelRoom, SlashCommandCallbackParams } from '@rocket.chat/core-typings';
 import { Messages, Uploads, LivechatRooms, Rooms, Users } from '@rocket.chat/models';
@@ -11,7 +10,7 @@ import { settings } from '../../../app/settings/server';
 import { slashCommands } from '../../../app/utils/server/slashCommand';
 import { callbacks } from '../../../lib/callbacks';
 import { i18n } from '../../lib/i18n';
-import { broadcastMessageSentEvent } from '../../modules/watchers/lib/messages';
+import { broadcastMessageFromData } from '../../modules/watchers/lib/messages';
 import { inboxes } from './EmailInbox';
 import type { Inbox } from './EmailInbox';
 import { logger } from './logger';
@@ -172,9 +171,8 @@ slashCommands.add({
 				},
 			},
 		);
-		void broadcastMessageSentEvent({
+		void broadcastMessageFromData({
 			id: message._id,
-			broadcastCallback: (message) => api.broadcast('message.sent', message),
 		});
 
 		return sendSuccessReplyMessage({

diff --git a/apps/meteor/server/modules/listeners/listeners.module.ts b/apps/meteor/server/modules/listeners/listeners.module.ts
@@ -1,7 +1,7 @@
 import type { AppStatus } from '@rocket.chat/apps-engine/definition/AppStatus';
 import type { ISetting as AppsSetting } from '@rocket.chat/apps-engine/definition/settings';
 import type { IServiceClass } from '@rocket.chat/core-services';
-import { EnterpriseSettings, listenToMessageSentEvent } from '@rocket.chat/core-services';
+import { EnterpriseSettings } from '@rocket.chat/core-services';
 import { isSettingColor, isSettingEnterprise } from '@rocket.chat/core-typings';
 import type { IUser, IRoom, VideoConference, ISetting, IOmnichannelRoom } from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
@@ -158,7 +158,7 @@ export class ListenersModule {
 			});
 		});
 
-		listenToMessageSentEvent(service, async (message) => {
+		service.onEvent('watch.messages', async ({ message }) => {
 			if (!message.rid) {
 				return;
 			}