refactor: Replace Notifications in favor of sdk.stream;

apps/meteor/app/e2e/client/rocketchat.e2e.room.js

@@ -7,7 +7,6 @@ import { RoomManager } from '../../../client/lib/RoomManager';
 import { roomCoordinator } from '../../../client/lib/rooms/roomCoordinator';
 import { RoomSettingsEnum } from '../../../definition/IRoomTypeConfig';
 import { ChatRoom, Subscriptions, Messages } from '../../models/client';
-import { Notifications } from '../../notifications/client';
 import { sdk } from '../../utils/client/lib/SDKClient';
 import { E2ERoomState } from './E2ERoomState';
 import {
@@ -240,7 +239,7 @@ export class E2ERoom extends Emitter {
 
 			this.setState(E2ERoomState.WAITING_KEYS);
 			this.log('Requesting room key');
-			Notifications.notifyUsersOfRoom(this.roomId, 'e2ekeyRequest', this.roomId, room.e2eKeyId);
+			sdk.publish('notify-room-users', [`${this.roomId}/e2ekeyRequest`, this.roomId, room.e2eKeyId]);
 		} catch (error) {
 			// this.error = error;
 			this.setState(E2ERoomState.ERROR);

apps/meteor/app/emoji-custom/client/index.ts

@@ -1,3 +1 @@
 import './lib/emojiCustom';
-import './notifications/deleteEmojiCustom';
-import './notifications/updateEmojiCustom';

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

@@ -9,7 +9,6 @@ import { fireGlobalEvent } from '../../../../client/lib/utils/fireGlobalEvent';
 import { getConfig } from '../../../../client/lib/utils/getConfig';
 import { callbacks } from '../../../../lib/callbacks';
 import { CachedChatRoom, ChatMessage, ChatSubscription, CachedChatSubscription } from '../../../models/client';
-import { Notifications } from '../../../notifications/client';
 import { sdk } from '../../../utils/client/lib/SDKClient';
 import { upsertMessage, RoomHistoryManager } from './RoomHistoryManager';
 import { mainReady } from './mainReady';
@@ -37,8 +36,8 @@ function close(typeName: string) {
 	if (openedRooms[typeName]) {
 		if (openedRooms[typeName].rid) {
 			sdk.stop('room-messages', openedRooms[typeName].rid);
-			Notifications.unRoom(openedRooms[typeName].rid, 'deleteMessage');
-			Notifications.unRoom(openedRooms[typeName].rid, 'deleteMessageBulk');
+			sdk.stop('notify-room', `${openedRooms[typeName].rid}/deleteMessage`);
+			sdk.stop('notify-room', `${openedRooms[typeName].rid}/deleteMessageBulk`);
 		}
 
 		openedRooms[typeName].ready = false;
@@ -135,20 +134,21 @@ const computation = Tracker.autorun(() => {
 						});
 
 					// when we receive a messages imported event we just clear the room history and fetch it again
-					Notifications.onRoom(record.rid, 'messagesImported', async () => {
+					sdk.stream('notify-room', [`${record.rid}/messagesImported`], async () => {
 						await RoomHistoryManager.clear(record.rid);
 						await RoomHistoryManager.getMore(record.rid);
 					});
 
-					Notifications.onRoom(record.rid, 'deleteMessage', (msg) => {
+					sdk.stream('notify-room', [`${record.rid}/deleteMessage`], (msg) => {
 						ChatMessage.remove({ _id: msg._id });
 
 						// remove thread refenrece from deleted message
 						ChatMessage.update({ tmid: msg._id }, { $unset: { tmid: 1 } }, { multi: true });
 					});
-					Notifications.onRoom(
-						record.rid,
-						'deleteMessageBulk',
+
+					sdk.stream(
+						'notify-room',
+						[`${record.rid}/deleteMessageBulk`],
 						({ rid, ts, excludePinned, ignoreDiscussion, users, ids, showDeletedStatus }) => {
 							const query: Mongo.Selector<IMessage> = { rid };
 
@@ -177,7 +177,8 @@ const computation = Tracker.autorun(() => {
 							return ChatMessage.remove(query);
 						},
 					);
-					Notifications.onRoom(record.rid, 'messagesRead', ({ tmid, until }) => {
+
+					sdk.stream('notify-room', [`${record.rid}/messagesRead`], ({ tmid, until }) => {
 						if (tmid) {
 							return ChatMessage.update(
 								{

apps/meteor/app/ui/client/lib/UserAction.ts

@@ -3,8 +3,8 @@ import { debounce } from 'lodash';
 import { Meteor } from 'meteor/meteor';
 import { ReactiveDict } from 'meteor/reactive-dict';
 
-import { Notifications } from '../../../notifications/client';
 import { settings } from '../../../settings/client';
+import { sdk } from '../../../utils/client/lib/SDKClient';
 
 const TIMEOUT = 15000;
 const RENEW = TIMEOUT / 3;
@@ -38,7 +38,7 @@ const shownName = function (user: IUser | null | undefined): string | undefined
 
 const emitActivities = debounce(async (rid: string, extras: IExtras): Promise<void> => {
 	const activities = roomActivities.get(extras?.tmid || rid) || new Set();
-	Notifications.notifyRoom(rid, USER_ACTIVITY, shownName(Meteor.user() as unknown as IUser), [...activities], extras);
+	sdk.publish('notify-room', [`${rid}/${USER_ACTIVITY}`, shownName(Meteor.user() as unknown as IUser), [...activities], extras]);
 }, 500);
 
 function handleStreamAction(rid: string, username: string, activityTypes: string[], extras?: IExtras): void {
@@ -65,10 +65,11 @@ function handleStreamAction(rid: string, username: string, activityTypes: string
 	performingUsers.set(rid, roomActivities);
 }
 export const UserAction = new (class {
-	addStream(rid: string): void {
+	addStream(rid: string): () => void {
 		if (rooms.get(rid)) {
-			return;
+			throw new Error('UserAction - addStream should only be called once per room');
 		}
+
 		const handler = function (username: string, activityType: string[], extras?: object): void {
 			const user = Meteor.users.findOne(Meteor.userId() || undefined, {
 				fields: { name: 1, username: 1 },
@@ -79,7 +80,15 @@ export const UserAction = new (class {
 			handleStreamAction(rid, username, activityType, extras);
 		};
 		rooms.set(rid, handler);
-		Notifications.onRoom(rid, USER_ACTIVITY, handler);
+
+		const { stop } = sdk.stream('notify-room', [`${rid}/${USER_ACTIVITY}`], handler);
+		return () => {
+			if (!rooms.get(rid)) {
+				return;
+			}
+			stop();
+			rooms.delete(rid);
+		};
 	}
 
 	performContinuously(rid: string, activityType: string, extras: IExtras = {}): void {
@@ -156,15 +165,6 @@ export const UserAction = new (class {
 		void emitActivities(rid, extras);
 	}
 
-	cancel(rid: string): void {
-		if (!rooms.get(rid)) {
-			return;
-		}
-
-		Notifications.unRoom(rid, USER_ACTIVITY);
-		rooms.delete(rid);
-	}
-
 	get(roomId: string): IRoomActivity | undefined {
 		return performingUsers.get(roomId);
 	}

apps/meteor/app/user-status/client/index.ts

@@ -1,5 +1,3 @@
-import './notifications/deleteCustomUserStatus';
-import './notifications/updateCustomUserStatus';
 import './lib/customUserStatus';
 
 export { userStatus } from './lib/userStatus';

apps/meteor/client/lib/settings/PrivateSettingsCachedCollection.ts

@@ -1,7 +1,7 @@
 import type { ISetting } from '@rocket.chat/core-typings';
 
-import { Notifications } from '../../../app/notifications/client';
 import { CachedCollection } from '../../../app/ui-cached-collection/client';
+import { sdk } from '../../../app/utils/client/lib/SDKClient';
 
 class PrivateSettingsCachedCollection extends CachedCollection<ISetting> {
 	constructor() {
@@ -12,7 +12,7 @@ class PrivateSettingsCachedCollection extends CachedCollection<ISetting> {
 	}
 
 	async setupListener(): Promise<void> {
-		Notifications.onLogged(this.eventName as 'private-settings-changed', async (t: string, { _id, ...record }: { _id: string }) => {
+		sdk.stream('notify-logged', [this.eventName as 'private-settings-changed'], async (t: string, { _id, ...record }: { _id: string }) => {
 			this.log('record received', t, { _id, ...record });
 			this.collection.upsert({ _id }, record);
 			this.sync();

apps/meteor/client/lib/userData.ts

@@ -2,7 +2,6 @@ import type { ILivechatAgent, IUser, Serialized } from '@rocket.chat/core-typing
 import { ReactiveVar } from 'meteor/reactive-var';
 
 import { Users } from '../../app/models/client';
-import { Notifications } from '../../app/notifications/client';
 import { sdk } from '../../app/utils/client/lib/SDKClient';
 
 export const isSyncReady = new ReactiveVar(false);
@@ -60,7 +59,7 @@ export const synchronizeUserData = async (uid: IUser['_id']): Promise<RawUserDat
 
 	cancel?.();
 
-	const result = Notifications.onUser('userData', (data) => {
+	const result = sdk.stream('notify-user', [`${uid}/userData`], (data) => {
 		switch (data.type) {
 			case 'inserted':
 				// eslint-disable-next-line @typescript-eslint/no-unused-vars

apps/meteor/client/providers/EmojiPickerProvider.tsx → apps/meteor/client/providers/EmojiPickerProvider/EmojiPickerProvider.tsx

@@ -2,10 +2,11 @@ import { useDebouncedState, useLocalStorage } from '@rocket.chat/fuselage-hooks'
 import type { ReactNode, ReactElement, ContextType } from 'react';
 import React, { useState, useCallback, useMemo, useEffect } from 'react';
 
-import type { EmojiByCategory } from '../../app/emoji/client';
-import { emoji, getFrequentEmoji, updateRecent, createEmojiList, createPickerEmojis, CUSTOM_CATEGORY } from '../../app/emoji/client';
-import { EmojiPickerContext } from '../contexts/EmojiPickerContext';
-import EmojiPicker from '../views/composer/EmojiPicker/EmojiPicker';
+import type { EmojiByCategory } from '../../../app/emoji/client';
+import { emoji, getFrequentEmoji, updateRecent, createEmojiList, createPickerEmojis, CUSTOM_CATEGORY } from '../../../app/emoji/client';
+import { EmojiPickerContext } from '../../contexts/EmojiPickerContext';
+import EmojiPicker from '../../views/composer/EmojiPicker/EmojiPicker';
+import { useUpdateCustomEmoji } from './useUpdateCustomEmoji';
 
 const DEFAULT_ITEMS_LIMIT = 90;
 
@@ -23,6 +24,8 @@ const EmojiPickerProvider = ({ children }: { children: ReactNode }): ReactElemen
 		getFrequentEmoji(frequentEmojis.map(([emoji]) => emoji)),
 	);
 
+	useUpdateCustomEmoji();
+
 	const addFrequentEmojis = useCallback(
 		(emoji: string) => {
 			const empty: [string, number][] = frequentEmojis.some(([emojiName]) => emojiName === emoji) ? [] : [[emoji, 0]];

apps/meteor/client/providers/EmojiPickerProvider/index.ts

@@ -0,0 +1 @@
+export { default } from './EmojiPickerProvider';

apps/meteor/client/providers/EmojiPickerProvider/useUpdateCustomEmoji.ts

@@ -0,0 +1,17 @@
+import { useStream } from '@rocket.chat/ui-contexts';
+import { useEffect } from 'react';
+
+import { updateEmojiCustom, deleteEmojiCustom } from '../../../app/emoji-custom/client/lib/emojiCustom';
+
+export const useUpdateCustomEmoji = () => {
+	const notify = useStream('notify-logged');
+	useEffect(() => {
+		const unsubUpdate = notify('updateEmojiCustom', (data) => updateEmojiCustom(data.emojiData));
+		const unsubDelete = notify('deleteEmojiCustom', (data) => deleteEmojiCustom(data.emojiData));
+
+		return () => {
+			unsubUpdate();
+			unsubDelete();
+		};
+	}, [notify]);
+};

apps/meteor/client/providers/UserProvider/UserProvider.tsx

@@ -13,8 +13,11 @@ import { afterLogoutCleanUpCallback } from '../../../lib/callbacks/afterLogoutCl
 import { useReactiveValue } from '../../hooks/useReactiveValue';
 import { createReactiveSubscriptionFactory } from '../../lib/createReactiveSubscriptionFactory';
 import { useCreateFontStyleElement } from '../../views/account/accessibility/hooks/useCreateFontStyleElement';
+import { useDeleteUser } from './hooks/useDeleteUser';
 import { useEmailVerificationWarning } from './hooks/useEmailVerificationWarning';
 import { useLDAPAndCrowdCollisionWarning } from './hooks/useLDAPAndCrowdCollisionWarning';
+import { useUpdateAvatar } from './hooks/useUpdateAvatar';
+import { useUpdateCustomUserStatus } from './hooks/useUpdateCustomUserStatus';
 
 const getUserId = (): string | null => Meteor.userId();
 
@@ -78,6 +81,10 @@ const UserProvider = ({ children }: UserProviderProps): ReactElement => {
 	useLDAPAndCrowdCollisionWarning();
 	useEmailVerificationWarning(user ?? undefined);
 
+	useUpdateCustomUserStatus();
+	useDeleteUser();
+	useUpdateAvatar();
+
 	const contextValue = useMemo(
 		(): ContextType<typeof UserContext> => ({
 			userId,

apps/meteor/client/providers/UserProvider/hooks/useDeleteUser.ts

@@ -0,0 +1,32 @@
+import { useStream } from '@rocket.chat/ui-contexts';
+import { useEffect } from 'react';
+
+import { ChatMessage } from '../../../../app/models/client';
+
+export const useDeleteUser = () => {
+	const notify = useStream('notify-logged');
+
+	useEffect(() => {
+		return notify('Users:Deleted', ({ userId, messageErasureType, replaceByUser }) => {
+			if (messageErasureType === 'Unlink' && replaceByUser) {
+				return ChatMessage.update(
+					{
+						'u._id': userId,
+					},
+					{
+						$set: {
+							'alias': replaceByUser.alias,
+							'u._id': replaceByUser._id,
+							'u.username': replaceByUser.username,
+							'u.name': undefined,
+						},
+					},
+					{ multi: true },
+				);
+			}
+			ChatMessage.remove({
+				'u._id': userId,
+			});
+		});
+	}, [notify]);
+};

apps/meteor/client/providers/UserProvider/hooks/useUpdateAvatar.ts

@@ -0,0 +1,15 @@
+import { useStream } from '@rocket.chat/ui-contexts';
+import { Meteor } from 'meteor/meteor';
+import { useEffect } from 'react';
+
+export const useUpdateAvatar = () => {
+	const notify = useStream('notify-logged');
+	useEffect(() => {
+		return notify('updateAvatar', (data) => {
+			if ('username' in data) {
+				const { username, etag } = data;
+				username && Meteor.users.update({ username }, { $set: { avatarETag: etag } });
+			}
+		});
+	}, [notify]);
+};

apps/meteor/client/providers/UserProvider/hooks/useUpdateCustomUserStatus.ts

@@ -0,0 +1,17 @@
+import { useStream } from '@rocket.chat/ui-contexts';
+import { useEffect } from 'react';
+
+import { updateCustomUserStatus, deleteCustomUserStatus } from '../../../../app/user-status/client/lib/customUserStatus';
+
+export const useUpdateCustomUserStatus = () => {
+	const notify = useStream('notify-logged');
+	useEffect(() => {
+		const unsubUpdate = notify('updateCustomUserStatus', (data) => updateCustomUserStatus(data.userStatusData));
+		const unsubDelete = notify('deleteCustomUserStatus', (data) => deleteCustomUserStatus(data.userStatusData));
+
+		return () => {
+			unsubUpdate();
+			unsubDelete();
+		};
+	}, [notify]);
+};