chore: adding new streamer event for imported messages

apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts

@@ -1,6 +1,7 @@
 import http from 'http';
 import https from 'https';
 
+import { api } from '@rocket.chat/core-services';
 import type { IImport, MessageAttachment, IUpload } from '@rocket.chat/core-typings';
 import { Messages } from '@rocket.chat/models';
 import { Random } from '@rocket.chat/random';
@@ -80,6 +81,7 @@ export class PendingFileImporter extends Importer {
 
 		try {
 			const pendingFileMessageList = Messages.findAllImportedMessagesWithFilesToDownload();
+			const importedRoomIds = new Set<string>();
 			for await (const message of pendingFileMessageList) {
 				try {
 					const { _importFile } = message;
@@ -140,6 +142,7 @@ export class PendingFileImporter extends Importer {
 
 								await Messages.setImportFileRocketChatAttachment(_importFile.id, url, attachment);
 								await completeFile(details);
+								importedRoomIds.add(message.rid);
 							} catch (error) {
 								await completeFile(details);
 								logError(error);
@@ -150,6 +153,8 @@ export class PendingFileImporter extends Importer {
 					this.logger.error(error);
 				}
 			}
+
+			void api.broadcast('notify.importedMessages', { roomIds: Array.from(importedRoomIds) });
 		} catch (error) {
 			// If the cursor expired, restart the method
 			if (this.isCursorNotFoundError(error)) {

apps/meteor/app/importer/server/classes/ImportDataConverter.ts

@@ -715,7 +715,12 @@ export class ImportDataConverter {
 		return ImportData.getAllMessages().toArray();
 	}
 
-	async convertMessages({ beforeImportFn, afterImportFn, onErrorFn }: IConversionCallbacks = {}): Promise<void> {
+	async convertMessages({
+		beforeImportFn,
+		afterImportFn,
+		onErrorFn,
+		afterImportAllMessagesFn,
+	}: IConversionCallbacks & { afterImportAllMessagesFn?: (roomIds: string[]) => Promise<void> }): Promise<void> {
 		const rids: Array<string> = [];
 		const messages = await this.getMessagesToImport();
 
@@ -740,7 +745,6 @@ export class ImportDataConverter {
 					this._logger.warn(`Imported user not found: ${data.u._id}`);
 					throw new Error('importer-message-unknown-user');
 				}
-
 				const rid = await this.findImportedRoomId(data.rid);
 				if (!rid) {
 					throw new Error('importer-message-unknown-room');
@@ -813,6 +817,9 @@ export class ImportDataConverter {
 				this._logger.error(e);
 			}
 		}
+		if (afterImportAllMessagesFn) {
+			await afterImportAllMessagesFn(rids);
+		}
 	}
 
 	async updateRoom(room: IRoom, roomData: IImportChannel, startedByUserId: string): Promise<void> {

apps/meteor/app/importer/server/classes/Importer.ts

@@ -1,3 +1,4 @@
+import { api } from '@rocket.chat/core-services';
 import type { IImport, IImportRecord, IImportChannel, IImportUser, IImportProgress } from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
 import { Settings, ImportData, Imports } from '@rocket.chat/models';
@@ -170,6 +171,9 @@ export class Importer {
 			}
 		};
 
+		const afterImportAllMessagesFn = async (importedRoomIds: string[]): Promise<void> =>
+			api.broadcast('notify.importedMessages', { roomIds: importedRoomIds });
+
 		const afterBatchFn = async (successCount: number, errorCount: number) => {
 			if (successCount) {
 				await this.addCountCompleted(successCount);
@@ -203,7 +207,7 @@ export class Importer {
 				await this.converter.convertChannels(startedByUserId, { beforeImportFn, afterImportFn, onErrorFn });
 
 				await this.updateProgress(ProgressStep.IMPORTING_MESSAGES);
-				await this.converter.convertMessages({ afterImportFn, onErrorFn });
+				await this.converter.convertMessages({ afterImportFn, onErrorFn, afterImportAllMessagesFn });
 
 				await this.updateProgress(ProgressStep.FINISHING);
 

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

@@ -1169,6 +1169,7 @@ export default class SlackAdapter {
 
 	async processPinnedItemMessage(rocketChannel, rocketUser, slackMessage, isImporting) {
 		if (slackMessage.attachments && slackMessage.attachments[0] && slackMessage.attachments[0].text) {
+			// TODO: refactor this logic to use the service to send this system message instead of using sendMessage
 			const rocketMsgObj = {
 				rid: rocketChannel._id,
 				t: 'message_pinned',
@@ -1380,6 +1381,7 @@ export default class SlackAdapter {
 			for await (const pin of items) {
 				if (pin.message) {
 					const user = await this.rocket.findUser(pin.message.user);
+					// TODO: send this system message to the room as well (using the service)
 					const msgObj = {
 						rid,
 						t: 'message_pinned',

apps/meteor/app/ui-utils/client/lib/LegacyRoomManager.ts

@@ -171,6 +171,13 @@ const computation = Tracker.autorun(() => {
 							record.streamActive = true;
 							openedRoomsDependency.changed();
 						});
+
+					// when we receive a messages imported event we just clear the room history and fetch it again
+					Notifications.onRoom(record.rid, 'messagesImported', async () => {
+						await RoomHistoryManager.clear(record.rid);
+						await RoomHistoryManager.getMore(record.rid);
+					});
+
 					Notifications.onRoom(record.rid, 'deleteMessage', (msg) => {
 						ChatMessage.remove({ _id: msg._id });
 

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

@@ -420,6 +420,13 @@ export class ListenersModule {
 			notifications.notifyUserInThisInstance(uid, 'calendar', data);
 		});
 
+		service.onEvent('notify.importedMessages', ({ roomIds }): void => {
+			roomIds.forEach((rid) => {
+				// couldnt get TS happy by providing no data, so had to provide null
+				notifications.notifyRoomInThisInstance(rid, 'messagesImported', null);
+			});
+		});
+
 		service.onEvent('connector.statuschanged', (enabled): void => {
 			notifications.notifyLoggedInThisInstance('voip.statuschanged', enabled);
 		});

ee/packages/ddp-client/src/types/streams.ts

@@ -63,6 +63,7 @@ export interface StreamerEvents {
 		{ key: `${string}/e2e.keyRequest`; args: [unknown] },
 		{ key: `${string}/videoconf`; args: [id: string] },
 		{ key: `${string}/messagesRead`; args: [{ until: Date; tmid?: string }] },
+		{ key: `${string}/messagesImported`; args: [null] },
 		/* @deprecated over videoconf*/
 		// { key: `${string}/${string}`; args: [id: string] },
 	];

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

@@ -96,6 +96,7 @@ export type EventSignatures = {
 	'notify.updateCustomSound'(data: { soundData: ICustomSound }): void;
 	'notify.calendar'(uid: string, data: ICalendarNotification): void;
 	'notify.messagesRead'(data: { rid: string; until: Date; tmid?: string }): void;
+	'notify.importedMessages'(data: { roomIds: string[] }): void;
 	'permission.changed'(data: { clientAction: ClientAction; data: any }): void;
 	'room'(data: { action: string; room: Partial<IRoom> }): void;
 	'room.avatarUpdate'(room: Pick<IRoom, '_id' | 'avatarETag'>): void;