feat(enterprise): Implement cold storage archiving for read receipts

.changeset/tender-seas-sparkle.md

@@ -0,0 +1,13 @@
+---
+'@rocket.chat/meteor': minor
+---
+
+Introduces Cold Storage Archiving for Read Receipts to improve performance and scalability in large deployments.
+
+Enterprise workspaces can now archive older read receipts into a dedicated cold storage collection, reducing the size of the primary read receipts dataset and improving query performance in environments with high message volumes.
+
+This feature is disabled by default and can be enabled through the new setting:
+
+**Message → Read Receipts → Enable Read Receipts Cold Storage**
+
+This feature is especially recommended for deployments with high message throughput and long data retention requirements, where reducing the size of hot collections significantly improves overall system responsiveness.

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

@@ -1,6 +1,6 @@
 import { api } from '@rocket.chat/core-services';
 import type { IRoom } from '@rocket.chat/core-typings';
-import { Messages, Rooms, Subscriptions, ReadReceipts, Users } from '@rocket.chat/models';
+import { Messages, Rooms, Subscriptions, ReadReceipts, ReadReceiptsArchive } from '@rocket.chat/models';
 
 import { deleteRoom } from './deleteRoom';
 import { NOTIFICATION_ATTACHMENT_COLOR } from '../../../../lib/constants';
@@ -143,13 +143,9 @@ export async function cleanRoomHistory({
 		selectedMessageIds,
 	);
 
-	if (!limit) {
-		const uids = await Users.findByUsernames(fromUsers, { projection: { _id: 1 } })
-			.map((user) => user._id)
-			.toArray();
-		await ReadReceipts.removeByIdPinnedTimestampLimitAndUsers(rid, excludePinned, ignoreDiscussion, ts, uids, ignoreThreads);
-	} else if (selectedMessageIds) {
+	if (limit && selectedMessageIds) {
 		await ReadReceipts.removeByMessageIds(selectedMessageIds);
+		await ReadReceiptsArchive.removeByMessageIds(selectedMessageIds);
 	}
 
 	if (count) {

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

@@ -1,7 +1,7 @@
 import { AppEvents, Apps } from '@rocket.chat/apps';
 import { api, Message } from '@rocket.chat/core-services';
 import { isThreadMessage, type AtLeast, type IMessage, type IRoom, type IThreadMessage, type IUser } from '@rocket.chat/core-typings';
-import { Messages, Rooms, Uploads, Users, ReadReceipts, Subscriptions } from '@rocket.chat/models';
+import { Messages, Rooms, Uploads, Users, ReadReceipts, ReadReceiptsArchive, Subscriptions } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
 import { callbacks } from '../../../../server/lib/callbacks';
@@ -69,6 +69,7 @@ export async function deleteMessage(message: IMessage, user: IUser): Promise<voi
 			await Messages.removeById(message._id);
 		}
 		await ReadReceipts.removeByMessageId(message._id);
+		await ReadReceiptsArchive.removeByMessageId(message._id);
 
 		for (const file of files) {
 			file?._id && (await FileUpload.getStore('Uploads').deleteById(file._id));

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

@@ -10,6 +10,7 @@ import {
 	Subscriptions,
 	Users,
 	ReadReceipts,
+	ReadReceiptsArchive,
 	LivechatUnitMonitors,
 	ModerationReports,
 } from '@rocket.chat/models';
@@ -87,6 +88,7 @@ export async function deleteUser(userId: string, confirmRelinquish = false, dele
 
 				await Messages.removeByUserId(userId);
 				await ReadReceipts.removeByUserId(userId);
+				await ReadReceiptsArchive.removeByUserId(userId);
 
 				await ModerationReports.hideMessageReportsByUserId(
 					userId,

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

@@ -1,5 +1,5 @@
 import type { IRoom } from '@rocket.chat/core-typings';
-import { Messages, Rooms, Subscriptions, ReadReceipts, Team } from '@rocket.chat/models';
+import { Messages, Rooms, Subscriptions, ReadReceipts, ReadReceiptsArchive, Team } from '@rocket.chat/models';
 
 import type { SubscribedRoomsForUserWithDetails } from './getRoomsWithSingleOwner';
 import { addUserRolesAsync } from '../../../../server/lib/roles/addUserRoles';
@@ -36,14 +36,15 @@ const bulkRoomCleanUp = async (rids: string[]) => {
 	// no bulk deletion for files
 	await Promise.all(rids.map((rid) => FileUpload.removeFilesByRoomId(rid)));
 
-	const [, , , deletedRoomIds] = await Promise.all([
+	const [, , , , deletedRoomIds] = await Promise.all([
 		Subscriptions.removeByRoomIds(rids, {
 			async onTrash(doc) {
 				void notifyOnSubscriptionChanged(doc, 'removed');
 			},
 		}),
 		Messages.removeByRoomIds(rids),
 		ReadReceipts.removeByRoomIds(rids),
+		ReadReceiptsArchive.removeByRoomIds(rids),
 		bulkTeamCleanup(rids),
 	]);
 

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

@@ -7,6 +7,7 @@ import {
 	LivechatRooms,
 	Messages,
 	ReadReceipts,
+	ReadReceiptsArchive,
 	Subscriptions,
 	LivechatContacts,
 	Users,
@@ -120,6 +121,7 @@ async function cleanGuestHistory(_id: string) {
 			FileUpload.removeFilesByRoomId(room._id),
 			Messages.removeByRoomId(room._id),
 			ReadReceipts.removeByRoomId(room._id),
+			ReadReceiptsArchive.removeByRoomId(room._id),
 		]);
 	}
 

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

@@ -20,6 +20,7 @@ import {
 	Subscriptions,
 	Users,
 	ReadReceipts,
+	ReadReceiptsArchive,
 } from '@rocket.chat/models';
 import { Meteor } from 'meteor/meteor';
 
@@ -293,6 +294,7 @@ export async function removeOmnichannelRoom(rid: string) {
 		}),
 		LivechatInquiry.removeByRoomId(rid),
 		LivechatRooms.removeById(rid),
+		ReadReceiptsArchive.removeByRoomId(rid),
 	]);
 
 	if (result[3]?.status === 'fulfilled' && result[3].value?.deletedCount && inquiry) {

apps/meteor/app/message-pin/server/pinMessage.ts

@@ -3,7 +3,7 @@ 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 type { ServerMethods } from '@rocket.chat/ddp-client';
-import { Messages, Rooms, Subscriptions, Users, ReadReceipts } from '@rocket.chat/models';
+import { Messages, Rooms, Subscriptions, Users } from '@rocket.chat/models';
 import { isTruthy } from '@rocket.chat/tools';
 import { check } from 'meteor/check';
 import { Meteor } from 'meteor/meteor';
@@ -91,9 +91,6 @@ export async function pinMessage(message: IMessage, userId: string, pinnedAt?: D
 	originalMessage = await Message.beforeSave({ message: originalMessage, room, user: me });
 
 	await Messages.setPinnedByIdAndUserId(originalMessage._id, originalMessage.pinnedBy, originalMessage.pinned);
-	if (settings.get('Message_Read_Receipt_Store_Users')) {
-		await ReadReceipts.setPinnedByMessageId(originalMessage._id, originalMessage.pinned);
-	}
 	if (isTheLastMessage(room, originalMessage)) {
 		await Rooms.setLastMessagePinned(room._id, originalMessage.pinnedBy, originalMessage.pinned);
 	}
@@ -192,9 +189,6 @@ export const unpinMessage = async (userId: string, message: IMessage) => {
 	await Apps.self?.triggerEvent(AppEvents.IPostMessagePinned, originalMessage, me, originalMessage.pinned);
 
 	await Messages.setPinnedByIdAndUserId(originalMessage._id, originalMessage.pinnedBy, originalMessage.pinned);
-	if (settings.get('Message_Read_Receipt_Store_Users')) {
-		await ReadReceipts.setPinnedByMessageId(originalMessage._id, originalMessage.pinned);
-	}
 	void notifyOnMessageChange({
 		id: message._id,
 	});

apps/meteor/app/slackbridge/server/SlackAdapter.ts

@@ -9,7 +9,7 @@ import https from 'https';
 import url from 'url';
 
 import { Message } from '@rocket.chat/core-services';
-import { Messages, Rooms, Users, ReadReceipts } from '@rocket.chat/models';
+import { Messages, Rooms, Users } from '@rocket.chat/models';
 import { App as SlackApp } from '@slack/bolt';
 import { RTMClient } from '@slack/rtm-api';
 import { Meteor } from 'meteor/meteor';
@@ -1201,9 +1201,6 @@ export default class SlackAdapter {
 			if (!isImporting && slackMessage.attachments[0].channel_id && slackMessage.attachments[0].ts) {
 				const messageId = this.createSlackMessageId(slackMessage.attachments[0].ts, slackMessage.attachments[0].channel_id);
 				await Messages.setPinnedByIdAndUserId(messageId, rocketMsgObj.u, true, new Date(parseInt(slackMessage.ts.split('.')[0]) * 1000));
-				if (settings.get('Message_Read_Receipt_Store_Users')) {
-					await ReadReceipts.setPinnedByMessageId(messageId, true);
-				}
 			}
 
 			return rocketMsgObj;
@@ -1412,9 +1409,6 @@ export default class SlackAdapter {
 
 					const messageId = this.createSlackMessageId(pin.message.ts, pin.channel);
 					await Messages.setPinnedByIdAndUserId(messageId, msgObj.u, true, new Date(parseInt(pin.message.ts.split('.')[0]) * 1000));
-					if (settings.get('Message_Read_Receipt_Store_Users')) {
-						await ReadReceipts.setPinnedByMessageId(messageId, true);
-					}
 				}
 			}
 		}

apps/meteor/app/threads/server/functions.ts

@@ -1,6 +1,6 @@
 import type { IMessage, IRoom, IUser } from '@rocket.chat/core-typings';
 import { isEditedMessage } from '@rocket.chat/core-typings';
-import { Messages, Subscriptions, ReadReceipts, NotificationQueue } from '@rocket.chat/models';
+import { Messages, Subscriptions, NotificationQueue } from '@rocket.chat/models';
 
 import { callbacks } from '../../../server/lib/callbacks';
 import {
@@ -41,10 +41,7 @@ export async function reply({ tmid }: { tmid?: string }, message: IMessage, pare
 	// Notify message mentioned users and highlights
 	const mentionedUsers = [...new Set([...mentionIds, ...highlightsUids])];
 
-	const promises = [
-		ReadReceipts.setAsThreadById(tmid),
-		Subscriptions.addUnreadThreadByRoomIdAndUserIds(rid, threadFollowersUids, tmid, notifyOptions),
-	];
+	const promises = [Subscriptions.addUnreadThreadByRoomIdAndUserIds(rid, threadFollowersUids, tmid, notifyOptions)];
 
 	if (mentionedUsers.length) {
 		promises.push(Subscriptions.addUnreadThreadByRoomIdAndUserIds(rid, mentionedUsers, tmid, { userMention: true }));

apps/meteor/definition/IRoomTypeConfig.ts

@@ -1,14 +1,4 @@
-import type {
-	IRoom,
-	RoomType,
-	IUser,
-	IMessage,
-	IReadReceipt,
-	ValueOf,
-	AtLeast,
-	ISubscription,
-	IOmnichannelRoom,
-} from '@rocket.chat/core-typings';
+import type { IRoom, RoomType, IUser, IMessage, ValueOf, AtLeast, ISubscription, IOmnichannelRoom } from '@rocket.chat/core-typings';
 import type { Keys as IconName } from '@rocket.chat/icons';
 import type { IRouterPaths, RouteName } from '@rocket.chat/ui-contexts';
 
@@ -107,7 +97,6 @@ export interface IRoomTypeServerDirectives {
 	) => Promise<{ title: string | undefined; text: string; name: string | undefined }>;
 	getMsgSender: (message: IMessage) => Promise<IUser | null>;
 	includeInRoomSearch: () => boolean;
-	getReadReceiptsExtraData: (message: IMessage) => Partial<IReadReceipt>;
 	includeInDashboard: () => boolean;
 	roomFind?: (rid: string) => Promise<IRoom | undefined> | Promise<IOmnichannelRoom | null> | IRoom | undefined;
 }

apps/meteor/ee/app/message-read-receipt/server/hooks/afterDeleteRoom.ts

@@ -1,11 +1,12 @@
-import { ReadReceipts } from '@rocket.chat/models';
+import { ReadReceipts, ReadReceiptsArchive } from '@rocket.chat/models';
 
 import { callbacks } from '../../../../../server/lib/callbacks';
 
 callbacks.add(
 	'afterDeleteRoom',
 	async (rid) => {
 		await ReadReceipts.removeByRoomId(rid);
+		await ReadReceiptsArchive.removeByRoomId(rid);
 		return rid;
 	},
 	callbacks.priority.LOW,