refactor: users out of db watcher (#32567)

apps/meteor/app/2fa/server/functions/resetTOTP.ts

@@ -4,6 +4,7 @@ import { Meteor } from 'meteor/meteor';
 
 import { i18n } from '../../../../server/lib/i18n';
 import { isUserIdFederated } from '../../../../server/lib/isUserIdFederated';
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 import * as Mailer from '../../../mailer/server/api';
 import { settings } from '../../../settings/server';
 
@@ -68,6 +69,14 @@ export async function resetTOTP(userId: string, notifyUser = false): Promise<boo
 
 	if (result?.modifiedCount === 1) {
 		await Users.unsetLoginTokens(userId);
+
+		void notifyOnUserChange({
+			clientAction: 'updated',
+			id: userId,
+			diff: {
+				'services.resume.loginTokens': [],
+			},
+		});
 		return true;
 	}
 

apps/meteor/app/2fa/server/methods/validateTempToken.ts

@@ -2,6 +2,7 @@ import { Users } from '@rocket.chat/models';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
+import { notifyOnUserChangeAsync } from '../../../lib/server/lib/notifyListener';
 import { TOTP } from '../lib/totp';
 
 declare module '@rocket.chat/ui-contexts' {
@@ -43,11 +44,25 @@ Meteor.methods<ServerMethods>({
 
 		// Once the TOTP is validated we logout all other clients
 		const { 'x-auth-token': xAuthToken } = this.connection?.httpHeaders ?? {};
-		if (xAuthToken) {
+		if (xAuthToken && this.userId) {
 			const hashedToken = Accounts._hashLoginToken(xAuthToken);
 
-			if (!(await Users.removeNonPATLoginTokensExcept(this.userId, hashedToken))) {
-				throw new Meteor.Error('error-logging-out-other-clients', 'Error logging out other clients');
+			const { modifiedCount } = await Users.removeNonPATLoginTokensExcept(this.userId, hashedToken);
+
+			if (modifiedCount > 0) {
+				// TODO this can be optmized so places that care about loginTokens being removed are invoked directly
+				// instead of having to listen to every watch.users event
+				void notifyOnUserChangeAsync(async () => {
+					if (!this.userId) {
+						return;
+					}
+					const userTokens = await Users.findOneById(this.userId, { projection: { 'services.resume.loginTokens': 1 } });
+					return {
+						clientAction: 'updated',
+						id: this.userId,
+						diff: { 'services.resume.loginTokens': userTokens?.services?.resume?.loginTokens },
+					};
+				});
 			}
 		}
 

apps/meteor/app/api/server/api.ts

@@ -14,9 +14,11 @@ import { Restivus } from 'meteor/rocketchat:restivus';
 import _ from 'underscore';
 
 import { isObject } from '../../../lib/utils/isObject';
+import { getNestedProp } from '../../../server/lib/getNestedProp';
 import { getRestPayload } from '../../../server/lib/logger/logPayloads';
 import { checkCodeForUser } from '../../2fa/server/code';
 import { hasPermissionAsync } from '../../authorization/server/functions/hasPermission';
+import { notifyOnUserChangeAsync } from '../../lib/server/lib/notifyListener';
 import { metrics } from '../../metrics/server';
 import { settings } from '../../settings/server';
 import { getDefaultUserFields } from '../../utils/server/functions/getDefaultUserFields';
@@ -848,6 +850,19 @@ export class APIClass<TBasePath extends string = ''> extends Restivus {
 				},
 			);
 
+			// TODO this can be optmized so places that care about loginTokens being removed are invoked directly
+			// instead of having to listen to every watch.users event
+			void notifyOnUserChangeAsync(async () => {
+				const userTokens = await Users.findOneById(this.user._id, { projection: { [tokenPath]: 1 } });
+				if (!userTokens) {
+					return;
+				}
+
+				const diff = { [tokenPath]: getNestedProp(userTokens, tokenPath) };
+
+				return { clientAction: 'updated', id: this.user._id, diff };
+			});
+
 			const response = {
 				status: 'success',
 				data: {

apps/meteor/app/api/server/v1/users.ts

@@ -43,6 +43,7 @@ import { setStatusText } from '../../../lib/server/functions/setStatusText';
 import { setUserAvatar } from '../../../lib/server/functions/setUserAvatar';
 import { setUsernameWithValidation } from '../../../lib/server/functions/setUsername';
 import { validateCustomFields } from '../../../lib/server/functions/validateCustomFields';
+import { notifyOnUserChange, notifyOnUserChangeAsync } from '../../../lib/server/lib/notifyListener';
 import { generateAccessToken } from '../../../lib/server/methods/createToken';
 import { settings } from '../../../settings/server';
 import { getURL } from '../../../utils/server/getURL';
@@ -387,7 +388,8 @@ API.v1.addRoute(
 			const lastLoggedIn = new Date();
 			lastLoggedIn.setDate(lastLoggedIn.getDate() - daysIdle);
 
-			const count = (await Users.setActiveNotLoggedInAfterWithRole(lastLoggedIn, role, false)).modifiedCount;
+			// since we're deactiving users that are not logged in, there is no need to send data through WS
+			const { modifiedCount: count } = await Users.setActiveNotLoggedInAfterWithRole(lastLoggedIn, role, false);
 
 			return API.v1.success({
 				count,
@@ -861,14 +863,31 @@ API.v1.addRoute(
 
 			// When 2FA is enable we logout all other clients
 			const xAuthToken = this.request.headers['x-auth-token'] as string;
-			if (xAuthToken) {
-				const hashedToken = Accounts._hashLoginToken(xAuthToken);
+			if (!xAuthToken) {
+				return API.v1.success();
+			}
 
-				if (!(await Users.removeNonPATLoginTokensExcept(this.userId, hashedToken))) {
-					throw new MeteorError('error-logging-out-other-clients', 'Error logging out other clients');
-				}
+			const hashedToken = Accounts._hashLoginToken(xAuthToken);
+
+			if (!(await Users.removeNonPATLoginTokensExcept(this.userId, hashedToken))) {
+				throw new MeteorError('error-logging-out-other-clients', 'Error logging out other clients');
 			}
 
+			// TODO this can be optmized so places that care about loginTokens being removed are invoked directly
+			// instead of having to listen to every watch.users event
+			void notifyOnUserChangeAsync(async () => {
+				const userTokens = await Users.findOneById(this.userId, { projection: { 'services.resume.loginTokens': 1 } });
+				if (!userTokens) {
+					return;
+				}
+
+				return {
+					clientAction: 'updated',
+					id: this.user._id,
+					diff: { 'services.resume.loginTokens': userTokens.services?.resume?.loginTokens },
+				};
+			});
+
 			return API.v1.success();
 		},
 	},
@@ -1021,6 +1040,12 @@ API.v1.addRoute(
 
 			const me = (await Users.findOneById(this.userId, { projection: { 'services.resume.loginTokens': 1 } })) as Pick<IUser, 'services'>;
 
+			void notifyOnUserChange({
+				clientAction: 'updated',
+				id: this.userId,
+				diff: { 'services.resume.loginTokens': me.services?.resume?.loginTokens },
+			});
+
 			const token = me.services?.resume?.loginTokens?.find((token) => token.hashedToken === hashedToken);
 
 			const tokenExpires =
@@ -1172,6 +1197,8 @@ API.v1.addRoute(
 				throw new Meteor.Error('error-invalid-user-id', 'Invalid user id');
 			}
 
+			void notifyOnUserChange({ clientAction: 'updated', id: this.userId, diff: { 'services.resume.loginTokens': [] } });
+
 			return API.v1.success({
 				message: `User ${userId} has been logged out!`,
 			});
@@ -1242,6 +1269,8 @@ API.v1.addRoute(
 				return API.v1.unauthorized();
 			}
 
+			// TODO refactor to not update the user twice (one inside of `setStatusText` and then later just the status + statusDefault)
+
 			if (this.bodyParams.message || this.bodyParams.message === '') {
 				await setStatusText(user._id, this.bodyParams.message);
 			}

apps/meteor/app/api/server/v1/voip/omnichannel.ts

@@ -3,6 +3,7 @@ import type { IUser, IVoipExtensionWithAgentInfo } from '@rocket.chat/core-typin
 import { Users } from '@rocket.chat/models';
 import { Match, check } from 'meteor/check';
 
+import { notifyOnUserChange } from '../../../../lib/server/lib/notifyListener';
 import { API } from '../../api';
 import { getPaginationItems } from '../../helpers/getPaginationItems';
 import { logger } from './logger';
@@ -79,6 +80,15 @@ API.v1.addRoute(
 
 			try {
 				await Users.setExtension(user._id, extension);
+
+				void notifyOnUserChange({
+					clientAction: 'updated',
+					id: user._id,
+					diff: {
+						extension,
+					},
+				});
+
 				return API.v1.success();
 			} catch (e) {
 				logger.error({ msg: 'Extension already in use' });
@@ -150,6 +160,15 @@ API.v1.addRoute(
 
 			logger.debug(`Removing extension association for user ${user._id} (extension was ${user.extension})`);
 			await Users.unsetExtension(user._id);
+
+			void notifyOnUserChange({
+				clientAction: 'updated',
+				id: user._id,
+				diff: {
+					extension: null,
+				},
+			});
+
 			return API.v1.success();
 		},
 	},

apps/meteor/app/apps/server/bridges/users.ts

@@ -11,6 +11,7 @@ import { deleteUser } from '../../../lib/server/functions/deleteUser';
 import { getUserCreatedByApp } from '../../../lib/server/functions/getUserCreatedByApp';
 import { setUserActiveStatus } from '../../../lib/server/functions/setUserActiveStatus';
 import { setUserAvatar } from '../../../lib/server/functions/setUserAvatar';
+import { notifyOnUserChange, notifyOnUserChangeById } from '../../../lib/server/lib/notifyListener';
 
 export class AppUserBridge extends UserBridge {
 	constructor(private readonly orch: IAppServerOrchestrator) {
@@ -97,6 +98,8 @@ export class AppUserBridge extends UserBridge {
 				throw new Error('Creating normal users is currently not supported');
 		}
 
+		void notifyOnUserChangeById({ clientAction: 'inserted', id: user._id });
+
 		return user._id;
 	}
 
@@ -137,6 +140,8 @@ export class AppUserBridge extends UserBridge {
 
 		await Users.updateOne({ _id: user.id }, { $set: fields as any });
 
+		void notifyOnUserChange({ clientAction: 'updated', id: user.id, diff: fields });
+
 		return true;
 	}
 

apps/meteor/app/crowd/server/crowd.ts

@@ -10,6 +10,7 @@ import { crowdIntervalValuesToCronMap } from '../../../server/settings/crowd';
 import { deleteUser } from '../../lib/server/functions/deleteUser';
 import { _setRealName } from '../../lib/server/functions/setRealName';
 import { setUserActiveStatus } from '../../lib/server/functions/setUserActiveStatus';
+import { notifyOnUserChange, notifyOnUserChangeById, notifyOnUserChangeAsync } from '../../lib/server/lib/notifyListener';
 import { settings } from '../../settings/server';
 import { logger } from './logger';
 
@@ -215,6 +216,15 @@ export class CROWD {
 			},
 		);
 
+		void notifyOnUserChange({
+			clientAction: 'updated',
+			id,
+			diff: {
+				...user,
+				...(crowdUser.displayname && { name: crowdUser.displayname }),
+			},
+		});
+
 		await setUserActiveStatus(id, crowdUser.active);
 	}
 
@@ -312,6 +322,21 @@ export class CROWD {
 				},
 			);
 
+			// TODO this can be optmized so places that care about loginTokens being removed are invoked directly
+			// instead of having to listen to every watch.users event
+			void notifyOnUserChangeAsync(async () => {
+				const userTokens = await Users.findOneById(crowdUser._id, { projection: { 'services.resume.loginTokens': 1 } });
+				if (!userTokens) {
+					return;
+				}
+
+				return {
+					clientAction: 'updated',
+					id: crowdUser._id,
+					diff: { 'services.resume.loginTokens': userTokens.services?.resume?.loginTokens },
+				};
+			});
+
 			await this.syncDataToUser(crowdUser, user._id);
 
 			return {
@@ -324,6 +349,8 @@ export class CROWD {
 		try {
 			crowdUser._id = await Accounts.createUserAsync(crowdUser);
 
+			void notifyOnUserChangeById({ clientAction: 'inserted', id: crowdUser._id });
+
 			// sync the user data
 			await this.syncDataToUser(crowdUser, crowdUser._id);
 

apps/meteor/app/custom-oauth/server/custom_oauth_server.js

@@ -11,6 +11,7 @@ import _ from 'underscore';
 
 import { callbacks } from '../../../lib/callbacks';
 import { isURL } from '../../../lib/utils/isURL';
+import { notifyOnUserChange } from '../../lib/server/lib/notifyListener';
 import { registerAccessTokenService } from '../../lib/server/oauth/oauth';
 import { settings } from '../../settings/server';
 import { normalizers, fromTemplate, renameInvalidProperties } from './transform_helpers';
@@ -374,6 +375,8 @@ export class CustomOAuth {
 				};
 
 				await Users.update({ _id: user._id }, update);
+
+				void notifyOnUserChange({ clientAction: 'updated', id: user._id, diff: update });
 			}
 		});
 

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

@@ -28,6 +28,7 @@ import { generateUsernameSuggestion } from '../../../lib/server/functions/getUse
 import { insertMessage } from '../../../lib/server/functions/insertMessage';
 import { saveUserIdentity } from '../../../lib/server/functions/saveUserIdentity';
 import { setUserActiveStatus } from '../../../lib/server/functions/setUserActiveStatus';
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 import { createChannelMethod } from '../../../lib/server/methods/createChannel';
 import { createPrivateGroupMethod } from '../../../lib/server/methods/createPrivateGroup';
 import { getValidRoomName } from '../../../utils/server/lib/getValidRoomName';
@@ -250,6 +251,9 @@ export class ImportDataConverter {
 
 	async updateUser(existingUser: IUser, userData: IImportUser): Promise<void> {
 		const { _id } = existingUser;
+		if (!_id) {
+			return;
+		}
 
 		userData._id = _id;
 
@@ -297,10 +301,12 @@ export class ImportDataConverter {
 
 		// Deleted users are 'inactive' users in Rocket.Chat
 		if (userData.deleted && existingUser?.active) {
-			userData._id && (await setUserActiveStatus(userData._id, false, true));
+			await setUserActiveStatus(_id, false, true);
 		} else if (userData.deleted === false && existingUser?.active === false) {
-			userData._id && (await setUserActiveStatus(userData._id, true));
+			await setUserActiveStatus(_id, true);
 		}
+
+		void notifyOnUserChange({ clientAction: 'updated', id: _id, diff: updateData.$set });
 	}
 
 	private async hashPassword(password: string): Promise<string> {

apps/meteor/app/irc/server/irc-bridge/peerHandlers/disconnected.js

@@ -1,5 +1,7 @@
 import { Users } from '@rocket.chat/models';
 
+import { notifyOnUserChange } from '../../../../lib/server/lib/notifyListener';
+
 export default async function handleQUIT(args) {
 	const user = await Users.findOne({
 		'profile.irc.nick': args.nick,
@@ -13,4 +15,6 @@ export default async function handleQUIT(args) {
 			},
 		},
 	);
+
+	void notifyOnUserChange({ id: user._id, clientAction: 'updated', diff: { status: 'offline' } });
 }

apps/meteor/app/irc/server/irc-bridge/peerHandlers/nickChanged.js

@@ -1,5 +1,7 @@
 import { Users } from '@rocket.chat/models';
 
+import { notifyOnUserChange } from '../../../../lib/server/lib/notifyListener';
+
 export default async function handleNickChanged(args) {
 	const user = await Users.findOne({
 		'profile.irc.nick': args.nick,
@@ -21,4 +23,6 @@ export default async function handleNickChanged(args) {
 			},
 		},
 	);
+
+	void notifyOnUserChange({ clientAction: 'updated', id: user._id, diff: { name: args.newNick } });
 }

apps/meteor/app/irc/server/irc-bridge/peerHandlers/userRegistered.js

@@ -1,5 +1,7 @@
 import { Users } from '@rocket.chat/models';
 
+import { notifyOnUserChange } from '../../../../lib/server/lib/notifyListener';
+
 export default async function handleUserRegistered(args) {
 	// Check if there is an user with the given username
 	let user = await Users.findOne({
@@ -28,6 +30,8 @@ export default async function handleUserRegistered(args) {
 		};
 
 		user = await Users.create(userToInsert);
+
+		void notifyOnUserChange({ id: user._id, clientAction: 'inserted', data: user });
 	} else {
 		// ...otherwise, log the user in and update the information
 		this.log(`Logging in ${args.username} with nick: ${args.nick}`);
@@ -43,5 +47,7 @@ export default async function handleUserRegistered(args) {
 				},
 			},
 		);
+
+		void notifyOnUserChange({ id: user._id, clientAction: 'updated', diff: { status: 'online' } });
 	}
 }

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

@@ -19,7 +19,12 @@ import { callbacks } from '../../../../lib/callbacks';
 import { i18n } from '../../../../server/lib/i18n';
 import { FileUpload } from '../../../file-upload/server';
 import { settings } from '../../../settings/server';
-import { notifyOnRoomChangedById, notifyOnIntegrationChangedByUserId, notifyOnLivechatDepartmentAgentChanged } from '../lib/notifyListener';
+import {
+	notifyOnRoomChangedById,
+	notifyOnIntegrationChangedByUserId,
+	notifyOnLivechatDepartmentAgentChanged,
+	notifyOnUserChange,
+} from '../lib/notifyListener';
 import { getSubscribedRoomsForUserWithDetails, shouldRemoveOrChangeOwner } from './getRoomsWithSingleOwner';
 import { getUserSingleOwnedRooms } from './getUserSingleOwnedRooms';
 import { relinquishRoomOwnerships } from './relinquishRoomOwnerships';
@@ -156,5 +161,7 @@ export async function deleteUser(userId: string, confirmRelinquish = false, dele
 	// Refresh the servers list
 	await FederationServers.refreshServers();
 
+	void notifyOnUserChange({ clientAction: 'removed', id: user._id });
+
 	await callbacks.run('afterDeleteUser', user);
 }

apps/meteor/app/lib/server/functions/saveUser.js

@@ -16,6 +16,7 @@ import { settings } from '../../../settings/server';
 import { safeGetMeteorUser } from '../../../utils/server/functions/safeGetMeteorUser';
 import { validateEmailDomain } from '../lib';
 import { generatePassword } from '../lib/generatePassword';
+import { notifyOnUserChangeById, notifyOnUserChange } from '../lib/notifyListener';
 import { passwordPolicy } from '../lib/passwordPolicy';
 import { checkEmailAvailability } from './checkEmailAvailability';
 import { checkUsernameAvailability } from './checkUsernameAvailability';
@@ -329,6 +330,8 @@ const saveNewUser = async function (userData, sendPassword) {
 		}
 	}
 
+	void notifyOnUserChangeById({ clientAction: 'inserted', id: _id });
+
 	return _id;
 };
 
@@ -432,7 +435,7 @@ export const saveUser = async function (userId, userData) {
 	await Users.updateOne({ _id: userData._id }, updateUser);
 
 	// App IPostUserUpdated event hook
-	const userUpdated = await Users.findOneById(userId);
+	const userUpdated = await Users.findOneById(userData._id);
 
 	await callbacks.run('afterSaveUser', {
 		user: userUpdated,
@@ -449,5 +452,17 @@ export const saveUser = async function (userId, userData) {
 		await _sendUserEmail(settings.get('Password_Changed_Email_Subject'), passwordChangedHtml, userData);
 	}
 
+	if (typeof userData.verified === 'boolean') {
+		delete userData.verified;
+	}
+	void notifyOnUserChange({
+		clientAction: 'updated',
+		id: userData._id,
+		diff: {
+			...userData,
+			emails: userUpdated.emails,
+		},
+	});
+
 	return true;
 };

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

@@ -8,7 +8,7 @@ import { Meteor } from 'meteor/meteor';
 import { callbacks } from '../../../../lib/callbacks';
 import * as Mailer from '../../../mailer/server/api';
 import { settings } from '../../../settings/server';
-import { notifyOnRoomChangedById, notifyOnRoomChangedByUserDM } from '../lib/notifyListener';
+import { notifyOnRoomChangedById, notifyOnRoomChangedByUserDM, notifyOnUserChange } from '../lib/notifyListener';
 import { closeOmnichannelConversations } from './closeOmnichannelConversations';
 import { shouldRemoveOrChangeOwner, getSubscribedRoomsForUserWithDetails } from './getRoomsWithSingleOwner';
 import { getUserSingleOwnedRooms } from './getUserSingleOwnedRooms';
@@ -107,11 +107,16 @@ export async function setUserActiveStatus(userId: string, active: boolean, confi
 	if (active === false) {
 		await Users.unsetLoginTokens(userId);
 		await Rooms.setDmReadOnlyByUserId(userId, undefined, true, false);
+
+		void notifyOnUserChange({ clientAction: 'updated', id: userId, diff: { 'services.resume.loginTokens': [], active } });
 		void notifyOnRoomChangedByUserDM(userId);
 	} else {
 		await Users.unsetReason(userId);
+
+		void notifyOnUserChange({ clientAction: 'updated', id: userId, diff: { active } });
 		await reactivateDirectConversations(userId);
 	}
+
 	if (active && !settings.get('Accounts_Send_Email_When_Activating')) {
 		return true;
 	}

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

@@ -10,6 +10,7 @@ import { SystemLogger } from '../../../../server/lib/logger/system';
 import { hasPermissionAsync } from '../../../authorization/server/functions/hasPermission';
 import { settings } from '../../../settings/server';
 import { RateLimiter } from '../lib';
+import { notifyOnUserChange } from '../lib/notifyListener';
 import { addUserToRoom } from './addUserToRoom';
 import { checkUsernameAvailability } from './checkUsernameAvailability';
 import { getAvatarSuggestionForUser } from './getAvatarSuggestionForUser';
@@ -67,6 +68,8 @@ export const setUsernameWithValidation = async (userId: string, username: string
 		await joinDefaultChannels(user._id, joinDefaultChannelsSilenced);
 		setImmediate(async () => callbacks.run('afterCreateUser', user));
 	}
+
+	void notifyOnUserChange({ clientAction: 'updated', id: user._id, diff: { username } });
 };
 
 export const _setUsername = async function (userId: string, u: string, fullUser: IUser): Promise<unknown> {

apps/meteor/app/lib/server/lib/notifyListener.ts

@@ -16,6 +16,7 @@ import type {
 	IIntegrationHistory,
 	AtLeast,
 	ISettingColor,
+	IUser,
 } from '@rocket.chat/core-typings';
 import {
 	Rooms,
@@ -28,6 +29,7 @@ import {
 	IntegrationHistory,
 	LivechatInquiry,
 	LivechatDepartmentAgents,
+	Users,
 } from '@rocket.chat/models';
 
 type ClientAction = 'inserted' | 'updated' | 'removed';
@@ -426,7 +428,6 @@ export async function notifyOnSettingChanged(
 	if (!dbWatchersDisabled) {
 		return;
 	}
-
 	void api.broadcast('watch.settings', { clientAction, setting });
 }
 
@@ -434,7 +435,6 @@ export async function notifyOnSettingChangedById(id: ISetting['_id'], clientActi
 	if (!dbWatchersDisabled) {
 		return;
 	}
-
 	const item = clientAction === 'removed' ? await Settings.trashFindOneById(id) : await Settings.findOneById(id);
 
 	if (!item) {
@@ -443,3 +443,61 @@ export async function notifyOnSettingChangedById(id: ISetting['_id'], clientActi
 
 	void api.broadcast('watch.settings', { clientAction, setting: item });
 }
+
+type NotifyUserChange = {
+	id: IUser['_id'];
+	clientAction: 'inserted' | 'removed' | 'updated';
+	data?: IUser;
+	diff?: Record<string, any>;
+	unset?: Record<string, number>;
+};
+
+export async function notifyOnUserChange({ clientAction, id, data, diff, unset }: NotifyUserChange) {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+	if (clientAction === 'removed') {
+		void api.broadcast('watch.users', { clientAction, id });
+		return;
+	}
+	if (clientAction === 'inserted') {
+		void api.broadcast('watch.users', { clientAction, id, data: data! });
+		return;
+	}
+
+	void api.broadcast('watch.users', { clientAction, diff: diff!, unset: unset || {}, id });
+}
+
+/**
+ * Calls the callback only if DB Watchers are disabled
+ */
+export async function notifyOnUserChangeAsync(cb: () => Promise<NotifyUserChange | NotifyUserChange[] | void>) {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const result = await cb();
+	if (!result) {
+		return;
+	}
+
+	if (Array.isArray(result)) {
+		result.forEach((n) => notifyOnUserChange(n));
+		return;
+	}
+
+	return notifyOnUserChange(result);
+}
+
+// TODO this may be only useful on 'inserted'
+export async function notifyOnUserChangeById({ clientAction, id }: { id: IUser['_id']; clientAction: 'inserted' | 'removed' | 'updated' }) {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+	const user = await Users.findOneById(id);
+	if (!user) {
+		return;
+	}
+
+	void notifyOnUserChange({ id, clientAction, data: user });
+}

apps/meteor/app/livechat/server/business-hour/AbstractBusinessHour.ts

@@ -5,6 +5,7 @@ import moment from 'moment-timezone';
 import type { UpdateFilter } from 'mongodb';
 
 import type { IWorkHoursCronJobsWrapper } from '../../../../server/models/raw/LivechatBusinessHours';
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 
 export interface IBusinessHourBehavior {
 	findHoursToCreateJobs(): Promise<IWorkHoursCronJobsWrapper[]>;
@@ -49,7 +50,7 @@ export abstract class AbstractBusinessHourBehavior {
 	}
 
 	async changeAgentActiveStatus(agentId: string, status: ILivechatAgentStatus): Promise<any> {
-		return this.UsersRepository.setLivechatStatusIf(
+		const result = await this.UsersRepository.setLivechatStatusIf(
 			agentId,
 			status,
 			// Why this works: statusDefault is the property set when a user manually changes their status
@@ -57,6 +58,16 @@ export abstract class AbstractBusinessHourBehavior {
 			{ livechatStatusSystemModified: true, statusDefault: { $ne: 'offline' } },
 			{ livechatStatusSystemModified: true },
 		);
+
+		if (result.modifiedCount > 0) {
+			void notifyOnUserChange({
+				clientAction: 'updated',
+				id: agentId,
+				diff: { statusLivechat: 'available', livechatStatusSystemModified: true },
+			});
+		}
+
+		return result;
 	}
 }
 

apps/meteor/app/livechat/server/business-hour/BusinessHourManager.ts

@@ -5,6 +5,7 @@ import { LivechatBusinessHours, LivechatDepartment, Users } from '@rocket.chat/m
 import moment from 'moment';
 
 import { callbacks } from '../../../../lib/callbacks';
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 import { settings } from '../../../settings/server';
 import { businessHourLogger } from '../lib/logger';
 import type { IBusinessHourBehavior, IBusinessHourType } from './AbstractBusinessHour';
@@ -126,7 +127,12 @@ export class BusinessHourManager {
 			return this.behavior.changeAgentActiveStatus(agentId, 'available');
 		}
 
-		return Users.setLivechatStatusActiveBasedOnBusinessHours(agentId);
+		const result = await Users.setLivechatStatusActiveBasedOnBusinessHours(agentId);
+		if (result.updatedCount > 0) {
+			void notifyOnUserChange({ clientAction: 'updated', id: agentId, diff: { statusLivechat: 'available ' } });
+		}
+
+		return result;
 	}
 
 	async restartCronJobsIfNecessary(): Promise<void> {

apps/meteor/app/livechat/server/business-hour/Helper.ts

@@ -1,8 +1,9 @@
 import type { ILivechatBusinessHour } from '@rocket.chat/core-typings';
-import { LivechatBusinessHourTypes } from '@rocket.chat/core-typings';
+import { ILivechatAgentStatus, LivechatBusinessHourTypes } from '@rocket.chat/core-typings';
 import { LivechatBusinessHours, Users } from '@rocket.chat/models';
 import moment from 'moment';
 
+import { notifyOnUserChangeAsync } from '../../../lib/server/lib/notifyListener';
 import { businessHourLogger } from '../lib/logger';
 import { createDefaultBusinessHourRow } from './LivechatBusinessHours';
 import { filterBusinessHoursThatMustBeOpened } from './filterBusinessHoursThatMustBeOpened';
@@ -32,13 +33,14 @@ export const openBusinessHourDefault = async (): Promise<void> => {
 			active: 1,
 		},
 	});
+
 	const businessHoursToOpenIds = (await filterBusinessHoursThatMustBeOpened(activeBusinessHours)).map((businessHour) => businessHour._id);
 	businessHourLogger.debug({ msg: 'Opening default business hours', businessHoursToOpenIds });
 	await Users.openAgentsBusinessHoursByBusinessHourId(businessHoursToOpenIds);
 	if (businessHoursToOpenIds.length) {
-		await Users.makeAgentsWithinBusinessHourAvailable();
+		await makeOnlineAgentsAvailable();
 	}
-	await Users.updateLivechatStatusBasedOnBusinessHours();
+	await makeAgentsUnavailableBasedOnBusinessHour();
 };
 
 export const createDefaultBusinessHourIfNotExists = async (): Promise<void> => {
@@ -46,3 +48,55 @@ export const createDefaultBusinessHourIfNotExists = async (): Promise<void> => {
 		await LivechatBusinessHours.insertOne(createDefaultBusinessHourRow());
 	}
 };
+
+export async function makeAgentsUnavailableBasedOnBusinessHour(agentIds: string[] | null = null) {
+	const results = await Users.findAgentsAvailableWithoutBusinessHours(agentIds).toArray();
+
+	const update = await Users.updateLivechatStatusByAgentIds(
+		results.map(({ _id }) => _id),
+		ILivechatAgentStatus.NOT_AVAILABLE,
+	);
+
+	if (update.modifiedCount === 0) {
+		return;
+	}
+
+	void notifyOnUserChangeAsync(async () =>
+		results.map(({ _id, openBusinessHours }) => {
+			return {
+				id: _id,
+				clientAction: 'updated',
+				diff: {
+					statusLivechat: 'not-available',
+					openBusinessHours,
+				},
+			};
+		}),
+	);
+}
+
+export async function makeOnlineAgentsAvailable(agentIds: string[] | null = null) {
+	const results = await Users.findOnlineButNotAvailableAgents(agentIds).toArray();
+
+	const update = await Users.updateLivechatStatusByAgentIds(
+		results.map(({ _id }) => _id),
+		ILivechatAgentStatus.AVAILABLE,
+	);
+
+	if (update.modifiedCount === 0) {
+		return;
+	}
+
+	void notifyOnUserChangeAsync(async () =>
+		results.map(({ _id, openBusinessHours }) => {
+			return {
+				id: _id,
+				clientAction: 'updated',
+				diff: {
+					statusLivechat: 'available',
+					openBusinessHours,
+				},
+			};
+		}),
+	);
+}

apps/meteor/app/livechat/server/business-hour/Single.ts

@@ -1,10 +1,11 @@
 import { ILivechatAgentStatus, LivechatBusinessHourTypes } from '@rocket.chat/core-typings';
 import { LivechatBusinessHours, Users } from '@rocket.chat/models';
 
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 import { businessHourLogger } from '../lib/logger';
 import type { IBusinessHourBehavior } from './AbstractBusinessHour';
 import { AbstractBusinessHourBehavior } from './AbstractBusinessHour';
-import { filterBusinessHoursThatMustBeOpened, openBusinessHourDefault } from './Helper';
+import { filterBusinessHoursThatMustBeOpened, makeAgentsUnavailableBasedOnBusinessHour, openBusinessHourDefault } from './Helper';
 
 export class SingleBusinessHourBehavior extends AbstractBusinessHourBehavior implements IBusinessHourBehavior {
 	async openBusinessHoursByDayAndHour(): Promise<void> {
@@ -18,7 +19,8 @@ export class SingleBusinessHourBehavior extends AbstractBusinessHourBehavior imp
 			})
 		).map((businessHour) => businessHour._id);
 		await this.UsersRepository.closeAgentsBusinessHoursByBusinessHourIds(businessHoursIds);
-		await this.UsersRepository.updateLivechatStatusBasedOnBusinessHours();
+
+		await makeAgentsUnavailableBasedOnBusinessHour();
 	}
 
 	async onStartBusinessHours(): Promise<void> {
@@ -41,7 +43,19 @@ export class SingleBusinessHourBehavior extends AbstractBusinessHourBehavior imp
 				agentId,
 				newStatus: ILivechatAgentStatus.NOT_AVAILABLE,
 			});
-			await Users.setLivechatStatus(agentId, ILivechatAgentStatus.NOT_AVAILABLE);
+
+			const { modifiedCount } = await Users.setLivechatStatus(agentId, ILivechatAgentStatus.NOT_AVAILABLE);
+			if (modifiedCount > 0) {
+				void notifyOnUserChange({
+					id: agentId,
+					clientAction: 'updated',
+					diff: {
+						statusLivechat: ILivechatAgentStatus.NOT_AVAILABLE,
+						livechatStatusSystemModified: false,
+					},
+				});
+			}
+
 			return;
 		}
 

apps/meteor/app/livechat/server/business-hour/closeBusinessHour.ts

@@ -3,6 +3,7 @@ import { Users } from '@rocket.chat/models';
 import { makeFunction } from '@rocket.chat/patch-injection';
 
 import { businessHourLogger } from '../lib/logger';
+import { makeAgentsUnavailableBasedOnBusinessHour } from './Helper';
 import { getAgentIdsForBusinessHour } from './getAgentIdsForBusinessHour';
 
 export const closeBusinessHourByAgentIds = async (
@@ -16,7 +17,8 @@ export const closeBusinessHourByAgentIds = async (
 		top10AgentIds: agentIds.slice(0, 10),
 	});
 	await Users.removeBusinessHourByAgentIds(agentIds, businessHourId);
-	await Users.updateLivechatStatusBasedOnBusinessHours();
+
+	await makeAgentsUnavailableBasedOnBusinessHour();
 };
 
 export const closeBusinessHour = makeFunction(async (businessHour: Pick<ILivechatBusinessHour, '_id' | 'type'>): Promise<void> => {

apps/meteor/app/livechat/server/hooks/afterAgentRemoved.ts

@@ -1,18 +1,32 @@
 import { LivechatDepartment, Users, LivechatDepartmentAgents, LivechatVisitors } from '@rocket.chat/models';
 
 import { callbacks } from '../../../../lib/callbacks';
-import { notifyOnLivechatDepartmentAgentChanged } from '../../../lib/server/lib/notifyListener';
+import { notifyOnLivechatDepartmentAgentChanged, notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
 
 callbacks.add('livechat.afterAgentRemoved', async ({ agent }) => {
 	const departments = await LivechatDepartmentAgents.findByAgentId(agent._id).toArray();
 
-	const [, { deletedCount }] = await Promise.all([
+	const [{ modifiedCount }, { deletedCount }] = await Promise.all([
 		Users.removeAgent(agent._id),
 		LivechatDepartmentAgents.removeByAgentId(agent._id),
 		agent.username && LivechatVisitors.removeContactManagerByUsername(agent.username),
 		departments.length && LivechatDepartment.decreaseNumberOfAgentsByIds(departments.map(({ departmentId }) => departmentId)),
 	]);
 
+	if (modifiedCount > 0) {
+		void notifyOnUserChange({
+			id: agent._id,
+			clientAction: 'updated',
+			diff: {
+				operator: false,
+				livechat: null,
+				statusLivechat: null,
+				extension: null,
+				openBusinessHours: null,
+			},
+		});
+	}
+
 	if (deletedCount > 0) {
 		departments.forEach((depAgent) => {
 			void notifyOnLivechatDepartmentAgentChanged(

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

@@ -63,6 +63,7 @@ import {
 	notifyOnRoomChangedById,
 	notifyOnLivechatInquiryChangedByToken,
 	notifyOnLivechatDepartmentAgentChangedByDepartmentId,
+	notifyOnUserChange,
 } from '../../../lib/server/lib/notifyListener';
 import * as Mailer from '../../../mailer/server/api';
 import { metrics } from '../../../metrics/server';
@@ -1308,9 +1309,18 @@ class LivechatClass {
 	}
 
 	async setUserStatusLivechatIf(userId: string, status: ILivechatAgentStatus, condition?: Filter<IUser>, fields?: AKeyOf<ILivechatAgent>) {
-		const user = await Users.setLivechatStatusIf(userId, status, condition, fields);
+		const result = await Users.setLivechatStatusIf(userId, status, condition, fields);
+
+		if (result.modifiedCount > 0) {
+			void notifyOnUserChange({
+				id: userId,
+				clientAction: 'updated',
+				diff: { ...fields, statusLivechat: status },
+			});
+		}
+
 		callbacks.runAsync('livechat.setUserStatusLivechat', { userId, status });
-		return user;
+		return result;
 	}
 
 	async returnRoomAsInquiry(room: IOmnichannelRoom, departmentId?: string, overrideTransferData: any = {}) {
@@ -1692,6 +1702,18 @@ class LivechatClass {
 	async setUserStatusLivechat(userId: string, status: ILivechatAgentStatus) {
 		const user = await Users.setLivechatStatus(userId, status);
 		callbacks.runAsync('livechat.setUserStatusLivechat', { userId, status });
+
+		if (user.modifiedCount > 0) {
+			void notifyOnUserChange({
+				id: userId,
+				clientAction: 'updated',
+				diff: {
+					statusLivechat: status,
+					livechatStatusSystemModified: false,
+				},
+			});
+		}
+
 		return user;
 	}
 

apps/meteor/app/livechat/server/startup.ts

@@ -10,6 +10,7 @@ import { beforeLeaveRoomCallback } from '../../../lib/callbacks/beforeLeaveRoomC
 import { i18n } from '../../../server/lib/i18n';
 import { roomCoordinator } from '../../../server/lib/rooms/roomCoordinator';
 import { hasPermissionAsync } from '../../authorization/server/functions/hasPermission';
+import { notifyOnUserChange } from '../../lib/server/lib/notifyListener';
 import { settings } from '../../settings/server';
 import { businessHourManager } from './business-hour';
 import { createDefaultBusinessHourIfNotExists } from './business-hour/Helper';
@@ -86,15 +87,25 @@ Meteor.startup(async () => {
 	});
 
 	// Remove when accounts.onLogout is async
-	Accounts.onLogout(
-		({ user }: { user: IUser }) =>
-			user?.roles?.includes('livechat-agent') &&
-			!user?.roles?.includes('bot') &&
-			void LivechatTyped.setUserStatusLivechatIf(
-				user._id,
-				ILivechatAgentStatus.NOT_AVAILABLE,
-				{},
-				{ livechatStatusSystemModified: true },
-			).catch(),
-	);
+	Accounts.onLogout(({ user }: { user: IUser }) => {
+		if (!user?.roles?.includes('livechat-agent') || user?.roles?.includes('bot')) {
+			return;
+		}
+
+		void LivechatTyped.setUserStatusLivechatIf(
+			user._id,
+			ILivechatAgentStatus.NOT_AVAILABLE,
+			{},
+			{ livechatStatusSystemModified: true },
+		).catch();
+
+		void notifyOnUserChange({
+			id: user._id,
+			clientAction: 'updated',
+			diff: {
+				statusLivechat: ILivechatAgentStatus.NOT_AVAILABLE,
+				livechatStatusSystemModified: true,
+			},
+		});
+	});
 });

apps/meteor/app/version-check/server/methods/banner_dismiss.ts

@@ -2,6 +2,8 @@ import { Users } from '@rocket.chat/models';
 import type { ServerMethods } from '@rocket.chat/ui-contexts';
 import { Meteor } from 'meteor/meteor';
 
+import { notifyOnUserChange } from '../../../lib/server/lib/notifyListener';
+
 declare module '@rocket.chat/ui-contexts' {
 	// eslint-disable-next-line @typescript-eslint/naming-convention
 	interface ServerMethods {
@@ -17,5 +19,13 @@ Meteor.methods<ServerMethods>({
 		}
 
 		await Users.setBannerReadById(userId, id);
+
+		void notifyOnUserChange({
+			id: userId,
+			clientAction: 'updated',
+			diff: {
+				[`banners.${id}.read`]: true,
+			},
+		});
 	},
 });

apps/meteor/ee/app/livechat-enterprise/server/business-hour/Custom.ts

@@ -5,7 +5,10 @@ import { LivechatDepartment, LivechatDepartmentAgents, Users } from '@rocket.cha
 import { businessHourManager } from '../../../../../app/livechat/server/business-hour';
 import type { IBusinessHourType } from '../../../../../app/livechat/server/business-hour/AbstractBusinessHour';
 import { AbstractBusinessHourType } from '../../../../../app/livechat/server/business-hour/AbstractBusinessHour';
-import { filterBusinessHoursThatMustBeOpened } from '../../../../../app/livechat/server/business-hour/Helper';
+import {
+	filterBusinessHoursThatMustBeOpened,
+	makeAgentsUnavailableBasedOnBusinessHour,
+} from '../../../../../app/livechat/server/business-hour/Helper';
 import { bhLogger } from '../lib/logger';
 
 type IBusinessHoursExtraProperties = {
@@ -79,7 +82,8 @@ class CustomBusinessHour extends AbstractBusinessHourType implements IBusinessHo
 		await this.BusinessHourRepository.removeById(businessHourId);
 		await this.removeBusinessHourFromAgents(businessHourId);
 		await LivechatDepartment.removeBusinessHourFromDepartmentsByBusinessHourId(businessHourId);
-		this.UsersRepository.updateLivechatStatusBasedOnBusinessHours();
+
+		await makeAgentsUnavailableBasedOnBusinessHour();
 	}
 
 	private async removeBusinessHourFromAgents(businessHourId: string): Promise<void> {
@@ -147,7 +151,8 @@ class CustomBusinessHour extends AbstractBusinessHourType implements IBusinessHo
 		).map((dept) => dept.agentId);
 
 		await Users.removeBusinessHourByAgentIds(agentsConnectedToDefaultBH, defaultBusinessHour._id);
-		await Users.updateLivechatStatusBasedOnBusinessHours();
+
+		await makeAgentsUnavailableBasedOnBusinessHour();
 	}
 
 	private async addBusinessHourToDepartmentsIfNeeded(businessHourId: string, departmentsToAdd: string[]): Promise<void> {

apps/meteor/ee/app/livechat-enterprise/server/business-hour/Helper.ts

@@ -2,6 +2,10 @@ import type { ILivechatBusinessHour } from '@rocket.chat/core-typings';
 import { LivechatBusinessHourTypes } from '@rocket.chat/core-typings';
 import { LivechatDepartment, LivechatDepartmentAgents, Users } from '@rocket.chat/models';
 
+import {
+	makeAgentsUnavailableBasedOnBusinessHour,
+	makeOnlineAgentsAvailable,
+} from '../../../../../app/livechat/server/business-hour/Helper';
 import { getAgentIdsForBusinessHour } from '../../../../../app/livechat/server/business-hour/getAgentIdsForBusinessHour';
 import { businessHourLogger } from '../../../../../app/livechat/server/lib/logger';
 
@@ -33,10 +37,10 @@ export const openBusinessHour = async (
 		top10AgentIds: agentIds.slice(0, 10),
 	});
 	await Users.addBusinessHourByAgentIds(agentIds, businessHour._id);
-	await Users.makeAgentsWithinBusinessHourAvailable(agentIds);
+	await makeOnlineAgentsAvailable(agentIds);
 
 	if (updateLivechatStatus) {
-		await Users.updateLivechatStatusBasedOnBusinessHours();
+		await makeAgentsUnavailableBasedOnBusinessHour();
 	}
 };
 
@@ -45,5 +49,5 @@ export const removeBusinessHourByAgentIds = async (agentIds: string[], businessH
 		return;
 	}
 	await Users.removeBusinessHourByAgentIds(agentIds, businessHourId);
-	await Users.updateLivechatStatusBasedOnBusinessHours();
+	await makeAgentsUnavailableBasedOnBusinessHour();
 };

apps/meteor/ee/app/livechat-enterprise/server/business-hour/Multiple.ts

@@ -9,6 +9,8 @@ import { AbstractBusinessHourBehavior } from '../../../../../app/livechat/server
 import {
 	filterBusinessHoursThatMustBeOpened,
 	filterBusinessHoursThatMustBeOpenedByDay,
+	makeOnlineAgentsAvailable,
+	makeAgentsUnavailableBasedOnBusinessHour,
 } from '../../../../../app/livechat/server/business-hour/Helper';
 import { closeBusinessHour } from '../../../../../app/livechat/server/business-hour/closeBusinessHour';
 import { settings } from '../../../../../app/settings/server';
@@ -34,7 +36,10 @@ export class MultipleBusinessHoursBehavior extends AbstractBusinessHourBehavior
 
 	async onStartBusinessHours(): Promise<void> {
 		await this.UsersRepository.removeBusinessHoursFromAllUsers();
-		await this.UsersRepository.updateLivechatStatusBasedOnBusinessHours();
+
+		// TODO is this required? since we're calling `this.openBusinessHour(businessHour)` later on, which will call this again (kinda)
+		await makeAgentsUnavailableBasedOnBusinessHour();
+
 		const currentTime = moment.utc(moment().utc().format('dddd:HH:mm'), 'dddd:HH:mm');
 		const day = currentTime.format('dddd');
 		const activeBusinessHours = await this.BusinessHourRepository.findActiveAndOpenBusinessHoursByDay(day, {
@@ -118,7 +123,7 @@ export class MultipleBusinessHoursBehavior extends AbstractBusinessHourBehavior
 			}
 
 			await this.UsersRepository.addBusinessHourByAgentIds(agentsId, defaultBusinessHour._id);
-			await this.UsersRepository.makeAgentsWithinBusinessHourAvailable(agentsId);
+			await makeOnlineAgentsAvailable(agentsId);
 
 			return options;
 		}
@@ -139,7 +144,7 @@ export class MultipleBusinessHoursBehavior extends AbstractBusinessHourBehavior
 		}
 
 		await this.UsersRepository.addBusinessHourByAgentIds(agentsId, businessHour._id);
-		await this.UsersRepository.makeAgentsWithinBusinessHourAvailable(agentsId);
+		await makeOnlineAgentsAvailable(agentsId);
 
 		return options;
 	}
@@ -208,11 +213,13 @@ export class MultipleBusinessHoursBehavior extends AbstractBusinessHourBehavior
 		if (!settings.get('Livechat_enable_business_hours')) {
 			return;
 		}
+
 		const businessHourToOpen = await filterBusinessHoursThatMustBeOpened([businessHour, defaultBH]);
 		for await (const bh of businessHourToOpen) {
 			await openBusinessHour(bh, false);
 		}
-		await Users.updateLivechatStatusBasedOnBusinessHours();
+
+		await makeAgentsUnavailableBasedOnBusinessHour();
 		await businessHourManager.restartCronJobsIfNecessary();
 	}
 
@@ -228,7 +235,7 @@ export class MultipleBusinessHoursBehavior extends AbstractBusinessHourBehavior
 	async onNewAgentCreated(agentId: string): Promise<void> {
 		await this.applyAnyOpenBusinessHourToAgent(agentId);
 
-		await Users.updateLivechatStatusBasedOnBusinessHours([agentId]);
+		await makeAgentsUnavailableBasedOnBusinessHour([agentId]);
 	}
 
 	private async applyAnyOpenBusinessHourToAgent(agentId: string): Promise<void> {

apps/meteor/ee/server/lib/ldap/copyCustomFieldsLDAP.ts

@@ -2,7 +2,7 @@ import type { IImportUser, ILDAPEntry } from '@rocket.chat/core-typings';
 import type { Logger } from '@rocket.chat/logger';
 
 import { templateVarHandler } from '../../../../app/utils/lib/templateVarHandler';
-import { getNestedProp } from './getNestedProp';
+import { getNestedProp } from '../../../../server/lib/getNestedProp';
 import { replacesNestedValues } from './replacesNestedValues';
 
 export const copyCustomFieldsLDAP = (

apps/meteor/server/database/watchCollections.ts

@@ -29,10 +29,11 @@ const onlyCollections = DBWATCHER_ONLY_COLLECTIONS.split(',')
 	.filter(Boolean);
 
 export function getWatchCollections(): string[] {
-	const collections = [Users.getCollectionName(), InstanceStatus.getCollectionName(), Subscriptions.getCollectionName()];
+	const collections = [InstanceStatus.getCollectionName(), Subscriptions.getCollectionName()];
 
 	// add back to the list of collections in case db watchers are enabled
 	if (!dbWatchersDisabled) {
+		collections.push(Users.getCollectionName());
 		collections.push(Messages.getCollectionName());
 		collections.push(LivechatInquiry.getCollectionName());
 		collections.push(Roles.getCollectionName());

apps/meteor/server/lib/sendMessagesToAdmins.ts

@@ -1,6 +1,7 @@
 import type { IUser, IMessage } from '@rocket.chat/core-typings';
 import { Roles, Users } from '@rocket.chat/models';
 
+import { notifyOnUserChangeAsync } from '../../app/lib/server/lib/notifyListener';
 import { executeSendMessage } from '../../app/lib/server/methods/sendMessage';
 import { createDirectMessage } from '../methods/createDirectMessage';
 import { SystemLogger } from './logger/system';
@@ -40,6 +41,8 @@ export async function sendMessagesToAdmins({
 
 	const users = await (await Roles.findUsersInRole('admin')).toArray();
 
+	const notifyAdmins: string[] = [];
+
 	for await (const adminUser of users) {
 		if (fromUser) {
 			try {
@@ -53,6 +56,29 @@ export async function sendMessagesToAdmins({
 			}
 		}
 
-		await Promise.all((await getData<Banner>(banners, adminUser)).map((banner) => Users.addBannerById(adminUser._id, banner)));
+		const updates = await Promise.all(
+			(await getData<Banner>(banners, adminUser)).map((banner) => Users.addBannerById(adminUser._id, banner)),
+		);
+
+		const hasUpdated = updates.some(({ modifiedCount }) => modifiedCount > 0);
+		if (hasUpdated) {
+			notifyAdmins.push(adminUser._id);
+		}
 	}
+
+	if (notifyAdmins.length === 0) {
+		return;
+	}
+
+	void notifyOnUserChangeAsync(async () => {
+		const results = await Users.findByIds<Pick<IUser, '_id' | 'banners'>>(notifyAdmins, { projection: { banners: 1 } }).toArray();
+
+		return results.map(({ _id, banners }) => ({
+			id: _id,
+			clientAction: 'updated',
+			diff: {
+				banners,
+			},
+		}));
+	});
 }

apps/meteor/server/methods/saveUserPreferences.ts

@@ -5,6 +5,7 @@ import type { ThemePreference } from '@rocket.chat/ui-theming/src/types/themes';
 import { Match, check } from 'meteor/check';
 import { Meteor } from 'meteor/meteor';
 
+import { notifyOnUserChange } from '../../app/lib/server/lib/notifyListener';
 import { settings as rcSettings } from '../../app/settings/server';
 
 type UserPreferences = {
@@ -94,8 +95,8 @@ export const saveUserPreferences = async (settings: Partial<UserPreferences>, us
 		mentionsWithSymbol: Match.Optional(Boolean),
 	};
 	check(settings, Match.ObjectIncluding(keys));
-	const user = await Users.findOneById(userId);
 
+	const user = await Users.findOneById(userId);
 	if (!user) {
 		return;
 	}
@@ -128,6 +129,21 @@ export const saveUserPreferences = async (settings: Partial<UserPreferences>, us
 
 	await Users.setPreferences(user._id, settings);
 
+	const diff = (Object.keys(settings) as (keyof UserPreferences)[]).reduce<Record<string, any>>((data, key) => {
+		data[`settings.preferences.${key}`] = settings[key];
+
+		return data;
+	}, {});
+
+	void notifyOnUserChange({
+		id: user._id,
+		clientAction: 'updated',
+		diff: {
+			...diff,
+			...(settings.language != null && { language: settings.language }),
+		},
+	});
+
 	// propagate changed notification preferences
 	setImmediate(async () => {
 		if (settings.desktopNotifications && oldDesktopNotifications !== settings.desktopNotifications) {

apps/meteor/server/models/raw/Users.js

@@ -968,9 +968,9 @@ export class UsersRaw extends BaseRaw {
 		return this.updateMany(query, update);
 	}
 
-	makeAgentsWithinBusinessHourAvailable(agentIds) {
+	findOnlineButNotAvailableAgents(userIds) {
 		const query = {
-			...(agentIds && { _id: { $in: agentIds } }),
+			...(userIds && { _id: { $in: userIds } }),
 			roles: 'livechat-agent',
 			// Exclude away users
 			status: 'online',
@@ -978,13 +978,7 @@ export class UsersRaw extends BaseRaw {
 			statusLivechat: 'not-available',
 		};
 
-		const update = {
-			$set: {
-				statusLivechat: 'available',
-			},
-		};
-
-		return this.updateMany(query, update);
+		return this.find(query);
 	}
 
 	removeBusinessHourByAgentIds(agentIds = [], businessHourId) {
@@ -1044,24 +1038,21 @@ export class UsersRaw extends BaseRaw {
 		return this.updateMany(query, update);
 	}
 
-	updateLivechatStatusBasedOnBusinessHours(userIds = []) {
-		const query = {
-			$or: [{ openBusinessHours: { $exists: false } }, { openBusinessHours: { $size: 0 } }],
-			$and: [{ roles: 'livechat-agent' }, { roles: { $ne: 'bot' } }],
-			// exclude deactivated users
-			active: true,
-			// Avoid unnecessary updates
-			statusLivechat: 'available',
-			...(Array.isArray(userIds) && userIds.length > 0 && { _id: { $in: userIds } }),
-		};
-
-		const update = {
-			$set: {
-				statusLivechat: 'not-available',
+	findAgentsAvailableWithoutBusinessHours(userIds = []) {
+		return this.find(
+			{
+				$or: [{ openBusinessHours: { $exists: false } }, { openBusinessHours: { $size: 0 } }],
+				$and: [{ roles: 'livechat-agent' }, { roles: { $ne: 'bot' } }],
+				// exclude deactivated users
+				active: true,
+				// Avoid unnecessary updates
+				statusLivechat: 'available',
+				...(Array.isArray(userIds) && userIds.length > 0 && { _id: { $in: userIds } }),
 			},
-		};
-
-		return this.updateMany(query, update);
+			{
+				projection: { openBusinessHours: 1 },
+			},
+		);
 	}
 
 	setLivechatStatusActiveBasedOnBusinessHours(userId) {
@@ -3074,4 +3065,17 @@ export class UsersRaw extends BaseRaw {
 	countByRole(role) {
 		return this.col.countDocuments({ roles: role });
 	}
+
+	updateLivechatStatusByAgentIds(userIds, status) {
+		return this.updateMany(
+			{
+				_id: { $in: userIds },
+			},
+			{
+				$set: {
+					statusLivechat: status,
+				},
+			},
+		);
+	}
 }

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

@@ -157,6 +157,16 @@ export class ListenersModule {
 				return;
 			}
 
+			notifications.notifyUserInThisInstance(_id, 'userData', {
+				type: 'updated',
+				id: _id,
+				diff: {
+					status,
+					...(statusText && { statusText }),
+				},
+				unset: {},
+			});
+
 			notifications.notifyLoggedInThisInstance('user-status', [_id, username, STATUS_MAP[status], statusText, name, roles]);
 
 			if (_id) {

packages/model-typings/src/models/IUsersModel.ts

@@ -134,8 +134,6 @@ export interface IUsersModel extends IBaseModel<IUser> {
 
 	addBusinessHourByAgentIds(agentIds: string[], businessHourId: string): any;
 
-	makeAgentsWithinBusinessHourAvailable(agentIds?: string[]): Promise<UpdateResult | Document>;
-
 	removeBusinessHourByAgentIds(agentIds: any, businessHourId: any): any;
 
 	openBusinessHourToAgentsWithoutDepartment(agentIdsWithDepartment: any, businessHourId: any): any;
@@ -144,8 +142,6 @@ export interface IUsersModel extends IBaseModel<IUser> {
 
 	closeAgentsBusinessHoursByBusinessHourIds(businessHourIds: any): any;
 
-	updateLivechatStatusBasedOnBusinessHours(userIds?: any): any;
-
 	setLivechatStatusActiveBasedOnBusinessHours(userId: any): any;
 
 	isAgentWithinBusinessHours(agentId: string): Promise<boolean>;
@@ -166,7 +162,7 @@ export interface IUsersModel extends IBaseModel<IUser> {
 
 	isUserInRoleScope(uid: IUser['_id']): Promise<boolean>;
 
-	addBannerById(_id: any, banner: any): any;
+	addBannerById(_id: IUser['_id'], banner: any): Promise<UpdateResult>;
 
 	findOneByAgentUsername(username: any, options: any): any;
 
@@ -393,4 +389,7 @@ export interface IUsersModel extends IBaseModel<IUser> {
 	countByRole(roleName: string): Promise<number>;
 	removeEmailCodeOfUserId(userId: string): Promise<UpdateResult>;
 	incrementInvalidEmailCodeAttempt(userId: string): Promise<ModifyResult<IUser>>;
+	findOnlineButNotAvailableAgents(userIds: string[] | null): FindCursor<Pick<ILivechatAgent, '_id' | 'openBusinessHours'>>;
+	findAgentsAvailableWithoutBusinessHours(userIds: string[] | null): FindCursor<Pick<ILivechatAgent, '_id' | 'openBusinessHours'>>;
+	updateLivechatStatusByAgentIds(userIds: string[], status: ILivechatAgentStatus): Promise<UpdateResult>;
 }