[NEW][Enterprise] Micro services (#19000)

Co-authored-by: Rodrigo Nascimento <rodrigoknascimento@gmail.com>
Co-authored-by: Alan Sikora <alansikora@gmail.com>

.eslintignore

@@ -18,3 +18,4 @@ public/pdf.worker.min.js
 public/workers/**/*
 imports/client/
 !/.storybook/
+ee/server/services/dist/**

.github/workflows/build_and_test.yml

@@ -487,3 +487,40 @@ jobs:
 
         docker build -t ${IMAGE}:develop .
         docker push ${IMAGE}:develop
+
+  services-image-build:
+    runs-on: ubuntu-latest
+    needs: deploy
+    if: github.event.pull_request.head.repo.full_name == github.repository
+
+    strategy:
+      matrix:
+        service: ["account", "authorization", "ddp-streamer", "presence", "stream-hub"]
+
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Use Node.js 12.18.4
+      uses: actions/setup-node@v1
+      with:
+        node-version: "12.18.4"
+
+    - name: Login to DockerHub
+      uses: docker/login-action@v1
+      with:
+        username: ${{ secrets.DOCKER_USER }}
+        password: ${{ secrets.DOCKER_PASS }}
+
+    - name: Build Docker images
+      run: |
+        npm i
+
+        cd ./ee/server/services
+        npm i
+        npm run build
+
+        echo "Building Docker image for service: ${{ matrix.service }}"
+
+        docker build --build-arg SERVICE=${{ matrix.service }} -t rocketchat/${{ matrix.service }}-service .
+
+        docker push rocketchat/${{ matrix.service }}-service

.meteorignore

@@ -0,0 +1 @@
+ee/server/services

.scripts/start.js

@@ -19,13 +19,17 @@ const isPortTaken = (port) => new Promise((resolve, reject) => {
 		.listen(port);
 });
 
-const waitPortRelease = (port) => new Promise((resolve, reject) => {
+const waitPortRelease = (port, count = 0) => new Promise((resolve, reject) => {
 	isPortTaken(port).then((taken) => {
 		if (!taken) {
 			return resolve();
 		}
+		if (count > 60) {
+			return reject();
+		}
+		console.log('Port', port, 'not release, waiting 1s...');
 		setTimeout(() => {
-			waitPortRelease(port).then(resolve).catch(reject);
+			waitPortRelease(port, ++count).then(resolve).catch(reject);
 		}, 1000);
 	});
 });
@@ -77,7 +81,10 @@ function startProcess(opts, callback) {
 
 		processes.splice(processes.indexOf(proc), 1);
 
-		processes.forEach((p) => p.kill());
+		processes.forEach((p) => {
+			console.log('Killing process', p.pid);
+			p.kill();
+		});
 
 		if (processes.length === 0) {
 			waitPortRelease(appOptions.env.PORT).then(() => {

app/api/server/v1/settings.js

@@ -7,6 +7,7 @@ import { Settings } from '../../../models/server';
 import { hasPermission } from '../../../authorization';
 import { API } from '../api';
 import { SettingsEvents, settings } from '../../../settings/server';
+import { setValue } from '../../../settings/server/raw';
 
 const fetchSettings = (query, sort, offset, count, fields) => {
 	const settings = Settings.find(query, {
@@ -150,6 +151,7 @@ API.v1.addRoute('settings/:_id', { authRequired: true }, {
 					_id: this.urlParams._id,
 					value: this.bodyParams.value,
 				});
+				setValue(this.urlParams._id, this.bodyParams.value);
 				return API.v1.success();
 			}
 

app/apps/server/bridges/messages.js

@@ -1,9 +1,8 @@
-import { Random } from 'meteor/random';
-
 import { Messages, Users, Subscriptions } from '../../../models/server';
-import { Notifications } from '../../../notifications';
 import { updateMessage } from '../../../lib/server/functions/updateMessage';
 import { executeSendMessage } from '../../../lib/server/methods/sendMessage';
+import { api } from '../../../../server/sdk/api';
+import notifications from '../../../notifications/server/lib/Notifications';
 
 export class AppMessageBridge {
 	constructor(orch) {
@@ -52,10 +51,8 @@ export class AppMessageBridge {
 			return;
 		}
 
-		Notifications.notifyUser(user.id, 'message', {
+		api.broadcast('stream.ephemeralMessage', user.id, msg.rid, {
 			...msg,
-			_id: Random.id(),
-			ts: new Date(),
 		});
 	}
 
@@ -67,11 +64,6 @@ export class AppMessageBridge {
 		}
 
 		const msg = this.orch.getConverters().get('messages').convertAppMessage(message);
-		const rmsg = Object.assign(msg, {
-			_id: Random.id(),
-			rid: room.id,
-			ts: new Date(),
-		});
 
 		const users = Subscriptions.findByRoomIdWhenUserIdExists(room.id, { fields: { 'u._id': 1 } })
 			.fetch()
@@ -80,14 +72,16 @@ export class AppMessageBridge {
 		Users.findByIds(users, { fields: { _id: 1 } })
 			.fetch()
 			.forEach(({ _id }) =>
-				Notifications.notifyUser(_id, 'message', rmsg),
+				api.broadcast('stream.ephemeralMessage', _id, room.id, {
+					...msg,
+				}),
 			);
 	}
 
 	async typing({ scope, id, username, isTyping }) {
 		switch (scope) {
 			case 'room':
-				Notifications.notifyRoom(id, 'typing', username, isTyping);
+				notifications.notifyRoom(id, 'typing', username, isTyping);
 				return;
 			default:
 				throw new Error('Unrecognized typing scope provided');

app/apps/server/communication/websockets.js

@@ -1,6 +1,7 @@
-import { Meteor } from 'meteor/meteor';
 import { AppStatusUtils } from '@rocket.chat/apps-engine/definition/AppStatus';
 
+import notifications from '../../../notifications/server/lib/Notifications';
+
 export const AppEvents = Object.freeze({
 	APP_ADDED: 'app/added',
 	APP_REMOVED: 'app/removed',
@@ -101,18 +102,10 @@ export class AppServerListener {
 
 export class AppServerNotifier {
 	constructor(orch) {
-		this.engineStreamer = new Meteor.Streamer('apps-engine', { retransmit: false });
-		this.engineStreamer.serverOnly = true;
-		this.engineStreamer.allowRead('none');
-		this.engineStreamer.allowEmit('all');
-		this.engineStreamer.allowWrite('none');
+		this.engineStreamer = notifications.streamAppsEngine;
 
 		// This is used to broadcast to the web clients
-		this.clientStreamer = new Meteor.Streamer('apps', { retransmit: false });
-		this.clientStreamer.serverOnly = true;
-		this.clientStreamer.allowRead('all');
-		this.clientStreamer.allowEmit('all');
-		this.clientStreamer.allowWrite('none');
+		this.clientStreamer = notifications.streamApps;
 
 		this.received = new Map();
 		this.listener = new AppServerListener(orch, this.engineStreamer, this.clientStreamer, this.received);

app/authentication/server/lib/restrictLoginAttempts.ts

@@ -18,10 +18,10 @@ export const isValidLoginAttemptByIp = async (ip: string): Promise<boolean> => {
 		return true;
 	}
 
-	const lastLogin = await Sessions.findLastLoginByIp(ip);
+	const lastLogin = await Sessions.findLastLoginByIp(ip) as {loginAt?: Date} | undefined;
 	let failedAttemptsSinceLastLogin;
 
-	if (!lastLogin) {
+	if (!lastLogin || !lastLogin.loginAt) {
 		failedAttemptsSinceLastLogin = await ServerEvents.countFailedAttemptsByIp(ip);
 	} else {
 		failedAttemptsSinceLastLogin = await ServerEvents.countFailedAttemptsByIpSince(ip, new Date(lastLogin.loginAt));

app/authorization/lib/AuthorizationUtils.ts

@@ -1,15 +1,13 @@
-import { Meteor } from 'meteor/meteor';
-
 const restrictedRolePermissions = new Map();
 
 export const AuthorizationUtils = class {
-	static addRolePermissionWhiteList(roleId: string, list: [string]): void {
+	static addRolePermissionWhiteList(roleId: string, list: string[]): void {
 		if (!roleId) {
-			throw new Meteor.Error('invalid-param');
+			throw new Error('invalid-param');
 		}
 
 		if (!list) {
-			throw new Meteor.Error('invalid-param');
+			throw new Error('invalid-param');
 		}
 
 		if (!restrictedRolePermissions.has(roleId)) {
@@ -25,7 +23,7 @@ export const AuthorizationUtils = class {
 
 	static isPermissionRestrictedForRole(permissionId: string, roleId: string): boolean {
 		if (!roleId || !permissionId) {
-			throw new Meteor.Error('invalid-param');
+			throw new Error('invalid-param');
 		}
 
 		if (!restrictedRolePermissions.has(roleId)) {
@@ -40,9 +38,9 @@ export const AuthorizationUtils = class {
 		return !rules.has(permissionId);
 	}
 
-	static isPermissionRestrictedForRoleList(permissionId: string, roleList: [string]): boolean {
+	static isPermissionRestrictedForRoleList(permissionId: string, roleList: string[]): boolean {
 		if (!roleList || !permissionId) {
-			throw new Meteor.Error('invalid-param');
+			throw new Error('invalid-param');
 		}
 
 		for (const roleId of roleList) {

app/authorization/server/functions/canAccessRoom.js

@@ -1,40 +0,0 @@
-import { hasPermissionAsync } from './hasPermission';
-import { Subscriptions } from '../../../models/server/raw';
-import { getValue } from '../../../settings/server/raw';
-
-export const roomAccessValidators = [
-	async function(room, user = {}) {
-		if (room && room.t === 'c') {
-			const anonymous = await getValue('Accounts_AllowAnonymousRead');
-			if (!user._id && anonymous === true) {
-				return true;
-			}
-
-			return hasPermissionAsync(user._id, 'view-c-room');
-		}
-	},
-	async function(room, user) {
-		if (!room || !user) {
-			return;
-		}
-
-		const exists = await Subscriptions.countByRoomIdAndUserId(room._id, user._id);
-		if (exists) {
-			return true;
-		}
-	},
-];
-
-export const canAccessRoomAsync = async (room, user, extraData) => {
-	for (let i = 0, total = roomAccessValidators.length; i < total; i++) {
-		// eslint-disable-next-line no-await-in-loop
-		const permitted = await roomAccessValidators[i](room, user, extraData);
-		if (permitted) {
-			return true;
-		}
-	}
-};
-
-export const canAccessRoom = (room, user, extraData) => Promise.await(canAccessRoomAsync(room, user, extraData));
-
-export const addRoomAccessValidator = (validator) => roomAccessValidators.push(validator.bind(this));

app/authorization/server/functions/canAccessRoom.ts

@@ -0,0 +1,8 @@
+import { Promise } from 'meteor/promise';
+
+import { Authorization } from '../../../../server/sdk';
+import { IAuthorization } from '../../../../server/sdk/types/IAuthorization';
+
+export const canAccessRoomAsync = Authorization.canAccessRoom;
+
+export const canAccessRoom = (...args: Parameters<IAuthorization['canAccessRoom']>): boolean => Promise.await(canAccessRoomAsync(...args));

app/authorization/server/functions/hasPermission.js

@@ -1,63 +1,8 @@
-import mem from 'mem';
+import { Authorization } from '../../../../server/sdk';
 
-import { Permissions, Users, Subscriptions } from '../../../models/server/raw';
-import { AuthorizationUtils } from '../../lib/AuthorizationUtils';
-
-const rolesHasPermission = mem(async (permission, roles) => {
-	if (AuthorizationUtils.isPermissionRestrictedForRoleList(permission, roles)) {
-		return false;
-	}
-
-	const result = await Permissions.findOne({ _id: permission, roles: { $in: roles } }, { projection: { _id: 1 } });
-	return !!result;
-}, {
-	cacheKey: JSON.stringify,
-	...process.env.TEST_MODE === 'true' && { maxAge: 1 },
-});
-
-const getRoles = mem(async (uid, scope) => {
-	const { roles: userRoles = [] } = await Users.findOne({ _id: uid }, { projection: { roles: 1 } });
-	const { roles: subscriptionsRoles = [] } = (scope && await Subscriptions.findOne({ rid: scope, 'u._id': uid }, { projection: { roles: 1 } })) || {};
-	return [...userRoles, ...subscriptionsRoles].sort((a, b) => a.localeCompare(b));
-}, { maxAge: 1000, cacheKey: JSON.stringify });
-
-export const clearCache = () => {
-	mem.clear(getRoles);
-	mem.clear(rolesHasPermission);
-};
-
-async function atLeastOne(uid, permissions = [], scope) {
-	const sortedRoles = await getRoles(uid, scope);
-	for (const permission of permissions) {
-		if (await rolesHasPermission(permission, sortedRoles)) { // eslint-disable-line
-			return true;
-		}
-	}
-
-	return false;
-}
-
-async function all(uid, permissions = [], scope) {
-	const sortedRoles = await getRoles(uid, scope);
-	for (const permission of permissions) {
-		if (!await rolesHasPermission(permission, sortedRoles)) { // eslint-disable-line
-			return false;
-		}
-	}
-
-	return true;
-}
-
-function _hasPermission(userId, permissions, scope, strategy) {
-	if (!userId) {
-		return false;
-	}
-	return strategy(userId, [].concat(permissions), scope);
-}
-
-export const hasAllPermissionAsync = async (userId, permissions, scope) => _hasPermission(userId, permissions, scope, all);
-export const hasPermissionAsync = async (userId, permissionId, scope) => _hasPermission(userId, permissionId, scope, all);
-export const hasAtLeastOnePermissionAsync = async (userId, permissions, scope) => _hasPermission(userId, permissions, scope, atLeastOne);
+export const hasAllPermissionAsync = async (userId, permissions, scope) => Authorization.hasAllPermission(userId, permissions, scope);
+export const hasPermissionAsync = async (userId, permissionId, scope) => Authorization.hasPermission(userId, permissionId, scope);
+export const hasAtLeastOnePermissionAsync = async (userId, permissions, scope) => Authorization.hasAtLeastOnePermission(userId, permissions, scope);
 
 export const hasAllPermission = (userId, permissions, scope) => Promise.await(hasAllPermissionAsync(userId, permissions, scope));
 export const hasPermission = (userId, permissionId, scope) => Promise.await(hasPermissionAsync(userId, permissionId, scope));

app/authorization/server/index.js

@@ -1,6 +1,5 @@
 import { addUserRoles } from './functions/addUserRoles';
 import {
-	addRoomAccessValidator,
 	canAccessRoom,
 	roomAccessValidators,
 } from './functions/canAccessRoom';
@@ -32,7 +31,6 @@ export {
 	removeUserFromRoles,
 	canSendMessage,
 	validateRoomMessagePermissions,
-	addRoomAccessValidator,
 	roomAccessValidators,
 	addUserRoles,
 	canAccessRoom,

app/authorization/server/lib/streamer.js

@@ -1,5 +0,0 @@
-import { Meteor } from 'meteor/meteor';
-
-export const rolesStreamer = new Meteor.Streamer('roles');
-rolesStreamer.allowWrite('none');
-rolesStreamer.allowRead('logged');

app/authorization/server/methods/addUserToRole.js

@@ -3,8 +3,8 @@ import _ from 'underscore';
 
 import { Users, Roles } from '../../../models/server';
 import { settings } from '../../../settings/server';
-import { Notifications } from '../../../notifications/server';
 import { hasPermission } from '../functions/hasPermission';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	'authorization:addUserToRole'(roleName, username, scope) {
@@ -50,7 +50,7 @@ Meteor.methods({
 		const add = Roles.addUserRoles(user._id, roleName, scope);
 
 		if (settings.get('UI_DisplayRoles')) {
-			Notifications.notifyLogged('roles-change', {
+			api.broadcast('user.roleUpdate', {
 				type: 'added',
 				_id: roleName,
 				u: {

app/authorization/server/methods/deleteRole.js

@@ -2,7 +2,6 @@ import { Meteor } from 'meteor/meteor';
 
 import * as Models from '../../../models/server';
 import { hasPermission } from '../functions/hasPermission';
-import { rolesStreamer } from '../lib/streamer';
 
 Meteor.methods({
 	'authorization:deleteRole'(roleName) {
@@ -36,13 +35,6 @@ Meteor.methods({
 			});
 		}
 
-		const removed = Models.Roles.remove(role.name);
-		if (removed) {
-			rolesStreamer.emit('roles', {
-				type: 'removed',
-				name: roleName,
-			});
-		}
-		return removed;
+		return Models.Roles.remove(role.name);
 	},
 });

app/authorization/server/methods/removeUserFromRole.js

@@ -3,8 +3,8 @@ import _ from 'underscore';
 
 import { Roles } from '../../../models/server';
 import { settings } from '../../../settings/server';
-import { Notifications } from '../../../notifications/server';
 import { hasPermission } from '../functions/hasPermission';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	'authorization:removeUserFromRole'(roleName, username, scope) {
@@ -55,7 +55,7 @@ Meteor.methods({
 
 		const remove = Roles.removeUserRoles(user._id, roleName, scope);
 		if (settings.get('UI_DisplayRoles')) {
-			Notifications.notifyLogged('roles-change', {
+			api.broadcast('user.roleUpdate', {
 				type: 'removed',
 				_id: roleName,
 				u: {

app/authorization/server/methods/saveRole.js

@@ -2,9 +2,8 @@ import { Meteor } from 'meteor/meteor';
 
 import { Roles } from '../../../models/server';
 import { settings } from '../../../settings/server';
-import { Notifications } from '../../../notifications/server';
 import { hasPermission } from '../functions/hasPermission';
-import { rolesStreamer } from '../lib/streamer';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	'authorization:saveRole'(roleData) {
@@ -27,15 +26,11 @@ Meteor.methods({
 
 		const update = Roles.createOrUpdate(roleData.name, roleData.scope, roleData.description, false, roleData.mandatory2fa);
 		if (settings.get('UI_DisplayRoles')) {
-			Notifications.notifyLogged('roles-change', {
+			api.broadcast('user.roleUpdate', {
 				type: 'changed',
 				_id: roleData.name,
 			});
 		}
-		rolesStreamer.emit('roles', {
-			type: 'changed',
-			...roleData,
-		});
 		return update;
 	},
 });

app/authorization/server/startup.js

@@ -4,7 +4,6 @@ import { Meteor } from 'meteor/meteor';
 import { Roles, Permissions, Settings } from '../../models/server';
 import { settings } from '../../settings/server';
 import { getSettingPermissionId, CONSTANTS } from '../lib';
-import { clearCache } from './functions/hasPermission';
 
 Meteor.startup(function() {
 	// Note:
@@ -224,12 +223,4 @@ Meteor.startup(function() {
 	};
 
 	settings.onload('*', createPermissionForAddedSetting);
-
-	Roles.on('change', ({ diff }) => {
-		if (diff && Object.keys(diff).length === 1 && diff._updatedAt) {
-			// avoid useless changes
-			return;
-		}
-		clearCache();
-	});
 });

app/authorization/server/streamer/permissions/emitter.js

@@ -1,45 +0,0 @@
-import Settings from '../../../../models/server/models/Settings';
-import { Notifications } from '../../../../notifications/server';
-import { CONSTANTS } from '../../../lib';
-import Permissions from '../../../../models/server/models/Permissions';
-import { clearCache } from '../../functions/hasPermission';
-
-Permissions.on('change', ({ clientAction, id, data, diff }) => {
-	if (diff && Object.keys(diff).length === 1 && diff._updatedAt) {
-		// avoid useless changes
-		return;
-	}
-	switch (clientAction) {
-		case 'updated':
-		case 'inserted':
-			data = data ?? Permissions.findOneById(id);
-			break;
-
-		case 'removed':
-			data = { _id: id };
-			break;
-	}
-
-	clearCache();
-
-	Notifications.notifyLoggedInThisInstance(
-		'permissions-changed',
-		clientAction,
-		data,
-	);
-
-	if (data.level && data.level === CONSTANTS.SETTINGS_LEVEL) {
-		// if the permission changes, the effect on the visible settings depends on the role affected.
-		// The selected-settings-based consumers have to react accordingly and either add or remove the
-		// setting from the user's collection
-		const setting = Settings.findOneNotHiddenById(data.settingId);
-		if (!setting) {
-			return;
-		}
-		Notifications.notifyLoggedInThisInstance(
-			'private-settings-changed',
-			'updated',
-			setting,
-		);
-	}
-});

app/authorization/server/streamer/permissions/index.js

@@ -1,7 +1,6 @@
 import { Meteor } from 'meteor/meteor';
 
 import Permissions from '../../../../models/server/models/Permissions';
-import './emitter';
 
 Meteor.methods({
 	'permissions/get'(updatedAt) {

app/discussion/server/authorization.js

@@ -1,10 +0,0 @@
-import { Meteor } from 'meteor/meteor';
-
-import { addRoomAccessValidator, canAccessRoom } from '../../authorization';
-import { Rooms } from '../../models';
-
-Meteor.startup(() => {
-	addRoomAccessValidator(function(room, user) {
-		return room && room.prid && canAccessRoom(Rooms.findOne(room.prid), user);
-	});
-});

app/discussion/server/index.js

@@ -1,5 +1,4 @@
 import './config';
-import './authorization';
 import './permissions';
 
 import './hooks/propagateDiscussionMetadata';

app/emoji-custom/server/methods/deleteEmojiCustom.js

@@ -1,27 +1,24 @@
 import { Meteor } from 'meteor/meteor';
 
+import { api } from '../../../../server/sdk/api';
 import { hasPermission } from '../../../authorization';
 import { EmojiCustom } from '../../../models';
-import { Notifications } from '../../../notifications';
 import { RocketChatFileEmojiCustomInstance } from '../startup/emoji-custom';
 
 Meteor.methods({
 	deleteEmojiCustom(emojiID) {
-		let emoji = null;
-
-		if (hasPermission(this.userId, 'manage-emoji')) {
-			emoji = EmojiCustom.findOneById(emojiID);
-		} else {
+		if (!hasPermission(this.userId, 'manage-emoji')) {
 			throw new Meteor.Error('not_authorized');
 		}
 
+		const emoji = EmojiCustom.findOneById(emojiID);
 		if (emoji == null) {
 			throw new Meteor.Error('Custom_Emoji_Error_Invalid_Emoji', 'Invalid emoji', { method: 'deleteEmojiCustom' });
 		}
 
 		RocketChatFileEmojiCustomInstance.deleteFile(encodeURIComponent(`${ emoji.name }.${ emoji.extension }`));
 		EmojiCustom.removeById(emojiID);
-		Notifications.notifyLogged('deleteEmojiCustom', { emojiData: emoji });
+		api.broadcast('emoji.deleteCustom', emoji);
 
 		return true;
 	},

app/emoji-custom/server/methods/insertOrUpdateEmoji.js

@@ -4,9 +4,9 @@ import s from 'underscore.string';
 import limax from 'limax';
 
 import { hasPermission } from '../../../authorization';
-import { Notifications } from '../../../notifications';
 import { EmojiCustom } from '../../../models';
 import { RocketChatFileEmojiCustomInstance } from '../startup/emoji-custom';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	insertOrUpdateEmoji(emojiData) {
@@ -73,7 +73,7 @@ Meteor.methods({
 
 			const _id = EmojiCustom.create(createEmoji);
 
-			Notifications.notifyLogged('updateEmojiCustom', { emojiData: createEmoji });
+			api.broadcast('emoji.updateCustom', createEmoji);
 
 			return _id;
 		}
@@ -107,7 +107,7 @@ Meteor.methods({
 			EmojiCustom.setAliases(emojiData._id, []);
 		}
 
-		Notifications.notifyLogged('updateEmojiCustom', { emojiData });
+		api.broadcast('emoji.updateCustom', emojiData);
 
 		return true;
 	},

app/emoji-custom/server/methods/uploadEmojiCustom.js

@@ -1,10 +1,10 @@
 import { Meteor } from 'meteor/meteor';
 import limax from 'limax';
 
-import { Notifications } from '../../../notifications';
 import { hasPermission } from '../../../authorization';
 import { RocketChatFile } from '../../../file';
 import { RocketChatFileEmojiCustomInstance } from '../startup/emoji-custom';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	uploadEmojiCustom(binaryContent, contentType, emojiData) {
@@ -21,7 +21,7 @@ Meteor.methods({
 		RocketChatFileEmojiCustomInstance.deleteFile(encodeURIComponent(`${ emojiData.name }.${ emojiData.extension }`));
 		const ws = RocketChatFileEmojiCustomInstance.createWriteStream(encodeURIComponent(`${ emojiData.name }.${ emojiData.extension }`), contentType);
 		ws.on('end', Meteor.bindEnvironment(() =>
-			Meteor.setTimeout(() => Notifications.notifyLogged('updateEmojiCustom', { emojiData }), 500),
+			Meteor.setTimeout(() => api.broadcast('emoji.updateCustom', emojiData), 500),
 		));
 
 		rs.pipe(ws);

app/google-vision/server/googlevision.js

@@ -1,12 +1,11 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { settings } from '../../settings';
 import { callbacks } from '../../callbacks';
-import { Notifications } from '../../notifications';
 import { Uploads, Settings, Users, Messages } from '../../models';
 import { FileUpload } from '../../file-upload';
+import { api } from '../../../server/sdk/api';
 
 class GoogleVision {
 	constructor() {
@@ -67,10 +66,7 @@ class GoogleVision {
 						FileUpload.getStore('Uploads').deleteById(file._id);
 						const user = Users.findOneById(message.u && message.u._id);
 						if (user) {
-							Notifications.notifyUser(user._id, 'message', {
-								_id: Random.id(),
-								rid: message.rid,
-								ts: new Date(),
+							api.broadcast('notify.ephemeralMessage', user._id, message.rid, {
 								msg: TAPi18n.__('Adult_images_are_not_allowed', {}, user.language),
 							});
 						}

app/importer/server/classes/ImporterWebsocket.js

@@ -1,11 +1,8 @@
-import { Meteor } from 'meteor/meteor';
+import notifications from '../../../notifications/server/lib/Notifications';
 
 class ImporterWebsocketDef {
 	constructor() {
-		this.streamer = new Meteor.Streamer('importers', { retransmit: false });
-		this.streamer.allowRead('all');
-		this.streamer.allowEmit('all');
-		this.streamer.allowWrite('none');
+		this.streamer = notifications.streamImporters;
 	}
 
 	/**

app/integrations/server/index.js

@@ -11,7 +11,6 @@ import './methods/outgoing/deleteOutgoingIntegration';
 import './methods/clearIntegrationHistory';
 import './api/api';
 import './lib/triggerHandler';
-import './streamer';
 import './triggers';
 
 export { mountIntegrationQueryBasedOnPermissions, mountIntegrationHistoryQueryBasedOnPermissions } from './lib/mountQueriesBasedOnPermission';

app/integrations/server/lib/triggerHandler.js

@@ -23,26 +23,6 @@ integrations.triggerHandler = new class RocketChatIntegrationHandler {
 		this.triggers = {};
 
 		Models.Integrations.find({ type: 'webhook-outgoing' }).fetch().forEach((data) => this.addIntegration(data));
-
-		Models.Integrations.on('change', ({ clientAction, id, data }) => {
-			switch (clientAction) {
-				case 'inserted':
-					if (data.type === 'webhook-outgoing') {
-						this.addIntegration(data);
-					}
-					break;
-				case 'updated':
-					data = data ?? Models.Integrations.findOneById(id);
-					if (data.type === 'webhook-outgoing') {
-						this.removeIntegration(data);
-						this.addIntegration(data);
-					}
-					break;
-				case 'removed':
-					this.removeIntegration({ _id: id });
-					break;
-			}
-		});
 	}
 
 	addIntegration(record) {
@@ -826,3 +806,5 @@ integrations.triggerHandler = new class RocketChatIntegrationHandler {
 		this.executeTriggerUrl(history.url, integration, { event, message, room, owner, user });
 	}
 }();
+
+export { integrations };

app/integrations/server/methods/clearIntegrationHistory.js

@@ -2,7 +2,7 @@ import { Meteor } from 'meteor/meteor';
 
 import { hasPermission } from '../../../authorization';
 import { IntegrationHistory, Integrations } from '../../../models';
-import { integrationHistoryStreamer } from '../streamer';
+import notifications from '../../../notifications/server/lib/Notifications';
 
 Meteor.methods({
 	clearIntegrationHistory(integrationId) {
@@ -22,7 +22,7 @@ Meteor.methods({
 
 		IntegrationHistory.removeByIntegrationId(integrationId);
 
-		integrationHistoryStreamer.emit(integrationId, { type: 'removed' });
+		notifications.streamIntegrationHistory.emit(integrationId, { type: 'removed' });
 
 		return true;
 	},

app/integrations/server/streamer.js

@@ -1,30 +0,0 @@
-import { Meteor } from 'meteor/meteor';
-
-import { hasAtLeastOnePermission } from '../../authorization/server';
-import { IntegrationHistory } from '../../models/server';
-
-export const integrationHistoryStreamer = new Meteor.Streamer('integrationHistory');
-integrationHistoryStreamer.allowWrite('none');
-integrationHistoryStreamer.allowRead(function() {
-	return this.userId && hasAtLeastOnePermission(this.userId, [
-		'manage-outgoing-integrations',
-		'manage-own-outgoing-integrations',
-	]);
-});
-
-IntegrationHistory.on('change', ({ clientAction, id, data, diff }) => {
-	switch (clientAction) {
-		case 'updated': {
-			const history = IntegrationHistory.findOneById(id, { fields: { 'integration._id': 1 } });
-			if (!history && !history.integration) {
-				return;
-			}
-			integrationHistoryStreamer.emit(history.integration._id, { id, diff, type: clientAction });
-			break;
-		}
-		case 'inserted': {
-			integrationHistoryStreamer.emit(data.integration._id, { data, type: clientAction });
-			break;
-		}
-	}
-});

app/ldap/server/sync.js

@@ -8,13 +8,13 @@ import LDAP from './ldap';
 import { callbacks } from '../../callbacks/server';
 import { RocketChatFile } from '../../file';
 import { settings } from '../../settings';
-import { Notifications } from '../../notifications';
 import { Users, Roles, Rooms, Subscriptions } from '../../models';
 import { Logger } from '../../logger';
 import { _setRealName, _setUsername } from '../../lib';
 import { templateVarHandler } from '../../utils';
 import { FileUpload } from '../../file-upload';
 import { addUserToRoom, removeUserFromRoom, createRoom } from '../../lib/server/functions';
+import { api } from '../../../server/sdk/api';
 
 
 export const logger = new Logger('LDAPSync', {});
@@ -264,7 +264,7 @@ export function mapLdapGroupsToUserRoles(ldap, ldapUser, user) {
 
 		const del = Roles.removeUserRoles(user._id, roleName);
 		if (settings.get('UI_DisplayRoles') && del) {
-			Notifications.notifyLogged('roles-change', {
+			api.broadcast('user.roleUpdate', {
 				type: 'removed',
 				_id: roleName,
 				u: {
@@ -365,7 +365,7 @@ function syncUserAvatar(user, ldapUser) {
 			fileStore.insert(file, rs, (err, result) => {
 				Meteor.setTimeout(function() {
 					Users.setAvatarData(user._id, 'ldap', result.etag);
-					Notifications.notifyLogged('updateAvatar', { username: user.username, etag: result.etag });
+					api.broadcast('user.avatarUpdate', { username: user.username, avatarETag: result.etag });
 				}, 500);
 			});
 		});
@@ -412,7 +412,7 @@ export function syncUserData(user, ldapUser, ldap) {
 		for (const roleName of userRoles) {
 			const add = Roles.addUserRoles(user._id, roleName);
 			if (settings.get('UI_DisplayRoles') && add) {
-				Notifications.notifyLogged('roles-change', {
+				api.broadcast('user.roleUpdate', {
 					type: 'added',
 					_id: roleName,
 					u: {

app/lib/server/functions/deleteUser.js

@@ -4,11 +4,11 @@ import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import { FileUpload } from '../../../file-upload/server';
 import { Users, Subscriptions, Messages, Rooms, Integrations, FederationServers } from '../../../models/server';
 import { settings } from '../../../settings/server';
-import { Notifications } from '../../../notifications/server';
 import { updateGroupDMsName } from './updateGroupDMsName';
 import { relinquishRoomOwnerships } from './relinquishRoomOwnerships';
 import { getSubscribedRoomsForUserWithDetails, shouldRemoveOrChangeOwner } from './getRoomsWithSingleOwner';
 import { getUserSingleOwnedRooms } from './getUserSingleOwnedRooms';
+import { api } from '../../../../server/sdk/api';
 
 export const deleteUser = function(userId, confirmRelinquish = false) {
 	const user = Users.findOneById(userId, {
@@ -65,7 +65,7 @@ export const deleteUser = function(userId, confirmRelinquish = false) {
 		}
 
 		Integrations.disableByUserId(userId); // Disables all the integrations which rely on the user being deleted.
-		Notifications.notifyLogged('Users:Deleted', { userId });
+		api.broadcast('user.deleted', user);
 	}
 
 	// Remove user from users database

app/lib/server/functions/setRealName.js

@@ -3,9 +3,9 @@ import s from 'underscore.string';
 
 import { Users } from '../../../models/server';
 import { settings } from '../../../settings';
-import { Notifications } from '../../../notifications';
 import { hasPermission } from '../../../authorization';
 import { RateLimiter } from '../lib';
+import { api } from '../../../../server/sdk/api';
 
 export const _setRealName = function(userId, name, fullUser) {
 	name = s.trim(name);
@@ -30,7 +30,7 @@ export const _setRealName = function(userId, name, fullUser) {
 	user.name = name;
 
 	if (settings.get('UI_Use_Real_Name') === true) {
-		Notifications.notifyLogged('Users:NameChanged', {
+		api.broadcast('user.nameChanged', {
 			_id: user._id,
 			name: user.name,
 			username: user.username,

app/lib/server/functions/setRoomAvatar.js

@@ -2,8 +2,8 @@ import { Meteor } from 'meteor/meteor';
 
 import { RocketChatFile } from '../../../file';
 import { FileUpload } from '../../../file-upload';
-import { Notifications } from '../../../notifications';
 import { Rooms, Avatars, Messages } from '../../../models/server';
+import { api } from '../../../../server/sdk/api';
 
 export const setRoomAvatar = function(rid, dataURI, user) {
 	const fileStore = FileUpload.getStore('Avatars');
@@ -13,7 +13,7 @@ export const setRoomAvatar = function(rid, dataURI, user) {
 	if (!dataURI) {
 		fileStore.deleteByRoomId(rid);
 		Messages.createRoomSettingsChangedWithTypeRoomIdMessageAndUser('room_changed_avatar', rid, '', user);
-		Notifications.notifyLogged('updateAvatar', { rid });
+		api.broadcast('room.avatarUpdate', { rid });
 
 		return Rooms.unsetAvatarData(rid);
 	}
@@ -40,7 +40,7 @@ export const setRoomAvatar = function(rid, dataURI, user) {
 			}
 			Rooms.setAvatarData(rid, 'upload', result.etag);
 			Messages.createRoomSettingsChangedWithTypeRoomIdMessageAndUser('room_changed_avatar', rid, '', user);
-			Notifications.notifyLogged('updateAvatar', { rid, etag: result.etag });
+			api.broadcast('room.avatarUpdate', { rid, avatarETag: result.etag });
 		}, 500);
 	});
 };

app/lib/server/functions/setStatusText.js

@@ -1,19 +1,11 @@
 import { Meteor } from 'meteor/meteor';
 import s from 'underscore.string';
 
-import { Users } from '../../../models';
+import { Users } from '../../../models/server';
 import { Users as UsersRaw } from '../../../models/server/raw';
-import { Notifications } from '../../../notifications';
-import { hasPermission } from '../../../authorization';
+import { hasPermission } from '../../../authorization/server';
 import { RateLimiter } from '../lib';
-
-// mirror of object in /imports/startup/client/listenActiveUsers.js - keep updated
-const STATUS_MAP = {
-	offline: 0,
-	online: 1,
-	away: 2,
-	busy: 3,
-};
+import { api } from '../../../../server/sdk/api';
 
 export const _setStatusTextPromise = async function(userId, statusText) {
 	if (!userId) { return false; }
@@ -28,12 +20,8 @@ export const _setStatusTextPromise = async function(userId, statusText) {
 
 	await UsersRaw.updateStatusText(user._id, statusText);
 
-	Notifications.notifyLogged('user-status', [
-		user._id,
-		user.username,
-		STATUS_MAP[user.status],
-		statusText,
-	]);
+	const { _id, username, status } = user;
+	api.broadcast('userpresence', { user: { _id, username, status, statusText } });
 
 	return true;
 };
@@ -59,12 +47,8 @@ export const _setStatusText = function(userId, statusText) {
 	Users.updateStatusText(user._id, statusText);
 	user.statusText = statusText;
 
-	Notifications.notifyLogged('user-status', [
-		user._id,
-		user.username,
-		STATUS_MAP[user.status],
-		statusText,
-	]);
+	const { _id, username, status } = user;
+	api.broadcast('userpresence', { user: { _id, username, status, statusText } });
 
 	return true;
 };

app/lib/server/functions/setUserAvatar.js

@@ -4,7 +4,7 @@ import { HTTP } from 'meteor/http';
 import { RocketChatFile } from '../../../file';
 import { FileUpload } from '../../../file-upload';
 import { Users } from '../../../models';
-import { Notifications } from '../../../notifications';
+import { api } from '../../../../server/sdk/api';
 
 export const setUserAvatar = function(user, dataURI, contentType, service) {
 	let encoding;
@@ -64,7 +64,7 @@ export const setUserAvatar = function(user, dataURI, contentType, service) {
 	fileStore.insert(file, buffer, (err, result) => {
 		Meteor.setTimeout(function() {
 			Users.setAvatarData(user._id, service, result.etag);
-			Notifications.notifyLogged('updateAvatar', { username: user.username, etag: result.etag });
+			api.broadcast('user.avatarUpdate', { username: user.username, avatarETag: result.etag });
 		}, 500);
 	});
 };

app/lib/server/functions/setUsername.js

@@ -6,8 +6,8 @@ import { settings } from '../../../settings';
 import { Users, Invites } from '../../../models/server';
 import { hasPermission } from '../../../authorization';
 import { RateLimiter } from '../lib';
-import { Notifications } from '../../../notifications/server';
 import { addUserToRoom } from './addUserToRoom';
+import { api } from '../../../../server/sdk/api';
 
 import { checkUsernameAvailability, setUserAvatar, getAvatarSuggestionForUser } from '.';
 
@@ -76,7 +76,7 @@ export const _setUsername = function(userId, u, fullUser) {
 		}
 	}
 
-	Notifications.notifyLogged('Users:NameChanged', {
+	api.broadcast('user.nameChanged', {
 		_id: user._id,
 		name: user.name,
 		username: user.username,

app/lib/server/index.js

@@ -6,7 +6,6 @@ import './startup/settings';
 import './startup/settingsOnLoadCdnPrefix';
 import './startup/settingsOnLoadDirectReply';
 import './startup/settingsOnLoadSMTP';
-import './startup/userDataStream';
 import '../lib/MessageTypes';
 import '../startup';
 import '../startup/defaultRoomTypes';

app/lib/server/lib/msgStream.js

@@ -1,57 +1,3 @@
-import { Meteor } from 'meteor/meteor';
-import { DDPCommon } from 'meteor/ddp-common';
+import notifications from '../../../notifications/server/lib/Notifications';
 
-const changedPayload = function(collection, id, fields) {
-	return DDPCommon.stringifyDDP({
-		msg: 'changed',
-		collection,
-		id,
-		fields,
-	});
-};
-
-const send = function(self, msg) {
-	if (!self.socket) {
-		return;
-	}
-	self.socket.send(msg);
-};
-
-class MessageStream extends Meteor.Streamer {
-	getSubscriptionByUserIdAndRoomId(userId, rid) {
-		return this.subscriptions.find((sub) => sub.eventName === rid && sub.subscription.userId === userId);
-	}
-
-	_publish(publication, eventName, options) {
-		super._publish(publication, eventName, options);
-		const uid = Meteor.userId();
-
-		const userEvent = (clientAction, { rid }) => {
-			switch (clientAction) {
-				case 'removed':
-					this.removeListener(uid, userEvent);
-					this.removeSubscription(this.getSubscriptionByUserIdAndRoomId(uid, rid), eventName);
-					break;
-			}
-		};
-		this.on(uid, userEvent);
-	}
-
-	mymessage = (eventName, args) => {
-		const subscriptions = this.subscriptionsByEventName[eventName];
-		if (!Array.isArray(subscriptions)) {
-			return;
-		}
-		subscriptions.forEach(({ subscription }) => {
-			const options = this.isEmitAllowed(subscription, eventName, args);
-			if (options) {
-				send(subscription._session, changedPayload(this.subscriptionName, 'id', {
-					eventName,
-					args: [args, options],
-				}));
-			}
-		});
-	}
-}
-
-export const msgStream = new MessageStream('room-messages');
+export const msgStream = notifications.streamRoomMessage;

app/lib/server/methods/addUsersToRoom.js

@@ -1,12 +1,11 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { Rooms, Subscriptions, Users } from '../../../models';
 import { hasPermission } from '../../../authorization';
 import { addUserToRoom } from '../functions';
-import { Notifications } from '../../../notifications';
+import { api } from '../../../../server/sdk/api';
 
 Meteor.methods({
 	addUsersToRoom(data = {}) {
@@ -73,10 +72,7 @@ Meteor.methods({
 			if (!subscription) {
 				addUserToRoom(data.rid, newUser, user);
 			} else {
-				Notifications.notifyUser(userId, 'message', {
-					_id: Random.id(),
-					rid: data.rid,
-					ts: new Date(),
+				api.broadcast('notify.ephemeralMessage', userId, data.rid, {
 					msg: TAPi18n.__('Username_is_already_in_here', {
 						postProcess: 'sprintf',
 						sprintf: [newUser.username],

app/lib/server/methods/filterATAllTag.js

@@ -1,13 +1,12 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import _ from 'underscore';
 import moment from 'moment';
 
 import { hasPermission } from '../../../authorization';
 import { callbacks } from '../../../callbacks';
-import { Notifications } from '../../../notifications';
 import { Users } from '../../../models';
+import { api } from '../../../../server/sdk/api';
 
 callbacks.add('beforeSaveMessage', function(message) {
 	// If the message was edited, or is older than 60 seconds (imported)
@@ -27,10 +26,7 @@ callbacks.add('beforeSaveMessage', function(message) {
 
 			// Add a notification to the chat, informing the user that this
 			// action is not allowed.
-			Notifications.notifyUser(message.u._id, 'message', {
-				_id: Random.id(),
-				rid: message.rid,
-				ts: new Date(),
+			api.broadcast('notify.ephemeralMessage', message.u._id, message.rid, {
 				msg: TAPi18n.__('error-action-not-allowed', { action }, language),
 			});
 

app/lib/server/methods/filterATHereTag.js

@@ -1,13 +1,12 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import _ from 'underscore';
 import moment from 'moment';
 
 import { hasPermission } from '../../../authorization';
 import { callbacks } from '../../../callbacks';
-import { Notifications } from '../../../notifications';
 import { Users } from '../../../models';
+import { api } from '../../../../server/sdk/api';
 
 callbacks.add('beforeSaveMessage', function(message) {
 	// If the message was edited, or is older than 60 seconds (imported)
@@ -26,10 +25,7 @@ callbacks.add('beforeSaveMessage', function(message) {
 
 			// Add a notification to the chat, informing the user that this
 			// action is not allowed.
-			Notifications.notifyUser(message.u._id, 'message', {
-				_id: Random.id(),
-				rid: message.rid,
-				ts: new Date(),
+			api.broadcast('notify.ephemeralMessage', message.u._id, message.rid, {
 				msg: TAPi18n.__('error-action-not-allowed', { action }, language),
 			});
 

app/lib/server/methods/sendMessage.js

@@ -1,19 +1,18 @@
 import { Meteor } from 'meteor/meteor';
 import { check } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import moment from 'moment';
 
 import { hasPermission } from '../../../authorization';
 import { metrics } from '../../../metrics';
 import { settings } from '../../../settings';
-import { Notifications } from '../../../notifications';
 import { messageProperties } from '../../../ui-utils';
 import { Users, Messages } from '../../../models';
 import { sendMessage } from '../functions';
 import { RateLimiter } from '../lib';
 import { canSendMessage } from '../../../authorization/server';
 import { SystemLogger } from '../../../logger/server';
+import { api } from '../../../../server/sdk/api';
 
 export function executeSendMessage(uid, message) {
 	if (message.tshow && !message.tmid) {
@@ -81,10 +80,7 @@ export function executeSendMessage(uid, message) {
 		SystemLogger.error('Error sending message:', error);
 
 		const errorMessage = typeof error === 'string' ? error : error.error || error.message;
-		Notifications.notifyUser(uid, 'message', {
-			_id: Random.id(),
-			rid: message.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', uid, message.rid, {
 			msg: TAPi18n.__(errorMessage, {}, user.language),
 		});
 

app/lib/server/startup/userDataStream.js

@@ -1,104 +0,0 @@
-import { MongoInternals } from 'meteor/mongo';
-
-import { Users } from '../../../models/server';
-import { Notifications } from '../../../notifications/server';
-import loginServiceConfiguration from '../../../models/server/models/LoginServiceConfiguration';
-
-let processOnChange;
-// eslint-disable-next-line no-undef
-const disableOplog = Package['disable-oplog'];
-
-if (disableOplog) {
-	// Stores the callbacks for the disconnection reactivity bellow
-	const userCallbacks = new Map();
-	const serviceConfigCallbacks = new Set();
-
-	// Overrides the native observe changes to prevent database polling and stores the callbacks
-	// for the users' tokens to re-implement the reactivity based on our database listeners
-	const { mongo } = MongoInternals.defaultRemoteCollectionDriver();
-	MongoInternals.Connection.prototype._observeChanges = function({ collectionName, selector, options = {} }, _ordered, callbacks) {
-		// console.error('Connection.Collection.prototype._observeChanges', collectionName, selector, options);
-		let cbs;
-		if (callbacks?.added) {
-			const records = Promise.await(mongo.rawCollection(collectionName).find(selector, { projection: options.fields }).toArray());
-			for (const { _id, ...fields } of records) {
-				callbacks.added(_id, fields);
-			}
-
-			if (collectionName === 'users' && selector['services.resume.loginTokens.hashedToken']) {
-				cbs = userCallbacks.get(selector._id) || new Set();
-				cbs.add({
-					hashedToken: selector['services.resume.loginTokens.hashedToken'],
-					callbacks,
-				});
-				userCallbacks.set(selector._id, cbs);
-			}
-		}
-
-		if (collectionName === 'meteor_accounts_loginServiceConfiguration') {
-			serviceConfigCallbacks.add(callbacks);
-		}
-
-		return {
-			stop() {
-				if (cbs) {
-					cbs.delete(callbacks);
-				}
-				serviceConfigCallbacks.delete(callbacks);
-			},
-		};
-	};
-
-	// Re-implement meteor's reactivity that uses observe to disconnect sessions when the token
-	// associated was removed
-	processOnChange = (diff, id) => {
-		const loginTokens = diff['services.resume.loginTokens'];
-		if (loginTokens) {
-			const tokens = loginTokens.map(({ hashedToken }) => hashedToken);
-
-			const cbs = userCallbacks.get(id);
-			if (cbs) {
-				[...cbs].filter(({ hashedToken }) => !tokens.includes(hashedToken)).forEach((item) => {
-					item.callbacks.removed(id);
-					cbs.delete(item);
-				});
-			}
-		}
-	};
-
-	loginServiceConfiguration.on('change', ({ clientAction, id, data, diff }) => {
-		switch (clientAction) {
-			case 'inserted':
-			case 'updated':
-				const record = { ...data || diff };
-				delete record.secret;
-				serviceConfigCallbacks.forEach((callbacks) => {
-					callbacks[clientAction === 'inserted' ? 'added' : 'changed']?.(id, record);
-				});
-				break;
-			case 'removed':
-				serviceConfigCallbacks.forEach((callbacks) => {
-					callbacks.removed?.(id);
-				});
-		}
-	});
-}
-
-Users.on('change', ({ clientAction, id, data, diff, unset }) => {
-	switch (clientAction) {
-		case 'updated':
-			Notifications.notifyUserInThisInstance(id, 'userData', { diff, unset, type: clientAction });
-
-			if (disableOplog) {
-				processOnChange(diff, id);
-			}
-
-			break;
-		case 'inserted':
-			Notifications.notifyUserInThisInstance(id, 'userData', { data, type: clientAction });
-			break;
-		case 'removed':
-			Notifications.notifyUserInThisInstance(id, 'userData', { id, type: clientAction });
-			break;
-	}
-});

app/livechat/server/index.js

@@ -80,8 +80,6 @@ import './lib/routing/External';
 import './lib/routing/ManualSelection';
 import './lib/routing/AutoSelection';
 import './lib/stream/agentStatus';
-import './lib/stream/departmentAgents';
-import './lib/stream/queueManager';
 import './sendMessageBySMS';
 import './api';
 import './api/rest';

app/livechat/server/lib/Helper.js

@@ -8,6 +8,7 @@ import { RoutingManager } from './RoutingManager';
 import { callbacks } from '../../../callbacks/server';
 import { settings } from '../../../settings';
 import { Apps, AppEvents } from '../../../apps/server';
+import notifications from '../../../notifications/server/lib/Notifications';
 
 export const createLivechatRoom = (rid, name, guest, roomInfo = {}, extraData = {}) => {
 	check(rid, String);
@@ -192,7 +193,7 @@ export const normalizeAgent = (agentId) => {
 export const dispatchAgentDelegated = (rid, agentId) => {
 	const agent = normalizeAgent(agentId);
 
-	Livechat.stream.emit(rid, {
+	notifications.streamLivechatRoom.emit(rid, {
 		type: 'agentData',
 		data: agent,
 	});

app/livechat/server/lib/Livechat.js

@@ -38,6 +38,7 @@ import { FileUpload } from '../../../file-upload/server';
 import { normalizeTransferredByData, parseAgentCustomFields, updateDepartmentAgents } from './Helper';
 import { Apps, AppEvents } from '../../../apps/server';
 import { businessHourManager } from '../business-hour';
+import notifications from '../../../notifications/server/lib/Notifications';
 
 export const Livechat = {
 	Analytics,
@@ -1107,7 +1108,7 @@ export const Livechat = {
 		}
 
 		LivechatRooms.findOpenByAgent(userId).forEach((room) => {
-			Livechat.stream.emit(room._id, {
+			notifications.streamLivechatRoom.emit(room._id, {
 				type: 'agentStatus',
 				status,
 			});
@@ -1123,7 +1124,7 @@ export const Livechat = {
 	},
 
 	notifyRoomVisitorChange(roomId, visitor) {
-		Livechat.stream.emit(roomId, {
+		notifications.streamLivechatRoom.emit(roomId, {
 			type: 'visitorData',
 			visitor,
 		});
@@ -1156,22 +1157,6 @@ export const Livechat = {
 	},
 };
 
-Livechat.stream = new Meteor.Streamer('livechat-room');
-
-Livechat.stream.allowRead((roomId, extraData) => {
-	const room = LivechatRooms.findOneById(roomId);
-
-	if (!room) {
-		console.warn(`Invalid eventName: "${ roomId }"`);
-		return false;
-	}
-
-	if (room.t === 'l' && extraData && extraData.visitorToken && room.v.token === extraData.visitorToken) {
-		return true;
-	}
-	return false;
-});
-
 settings.get('Livechat_history_monitor_type', (key, value) => {
 	Livechat.historyMonitorType = value;
 });

app/livechat/server/lib/stream/agentStatus.ts

@@ -2,9 +2,8 @@ import { Meteor } from 'meteor/meteor';
 
 import { Livechat } from '../Livechat';
 import { settings } from '../../../../settings/server';
-import { Users } from '../../../../models/server';
 
-let monitorAgents = false;
+export let monitorAgents = false;
 let actionTimeout = 60000;
 let action = 'none';
 let comment = '';
@@ -28,7 +27,7 @@ settings.get('Livechat_agent_leave_comment', (_key, value) => {
 	comment = value;
 });
 
-const onlineAgents = {
+export const onlineAgents = {
 	users: new Set(),
 	queue: new Map(),
 
@@ -74,36 +73,3 @@ const onlineAgents = {
 		}
 	}),
 };
-
-Users.on('change', ({ clientAction, id, diff }) => {
-	if (!monitorAgents) {
-		return;
-	}
-
-	if (clientAction !== 'removed' && diff && !diff.status && !diff.statusLivechat) {
-		return;
-	}
-
-	switch (clientAction) {
-		case 'updated':
-		case 'inserted':
-			const agent = Users.findOneAgentById(id, {
-				fields: {
-					status: 1,
-					statusLivechat: 1,
-				},
-			});
-			const serviceOnline = agent && agent.status !== 'offline' && agent.statusLivechat === 'available';
-
-			if (serviceOnline) {
-				return onlineAgents.add(id);
-			}
-
-			onlineAgents.remove(id);
-
-			break;
-		case 'removed':
-			onlineAgents.remove(id);
-			break;
-	}
-});

app/livechat/server/lib/stream/departmentAgents.js

@@ -1,29 +0,0 @@
-import { LivechatDepartmentAgents } from '../../../../models/server';
-import { Notifications } from '../../../../notifications';
-
-const fields = { agentId: 1, departmentId: 1 };
-
-const emitNotification = (action, payload = {}) => {
-	const { agentId = null } = payload;
-	if (!agentId) {
-		return;
-	}
-
-	Notifications.notifyUserInThisInstance(agentId, 'departmentAgentData', {
-		action,
-		...payload,
-	});
-};
-
-LivechatDepartmentAgents.on('change', ({ clientAction, id }) => {
-	switch (clientAction) {
-		case 'inserted':
-		case 'updated':
-			emitNotification(clientAction, LivechatDepartmentAgents.findOneById(id, { fields }));
-			break;
-
-		case 'removed':
-			emitNotification(clientAction, LivechatDepartmentAgents.trashFindOneById(id, { fields }));
-			break;
-	}
-});

app/livechat/server/lib/stream/queueManager.js

@@ -1,52 +0,0 @@
-import { Meteor } from 'meteor/meteor';
-
-import { hasPermission } from '../../../../authorization/server';
-import { LivechatInquiry } from '../../../../models/server';
-import { LIVECHAT_INQUIRY_QUEUE_STREAM_OBSERVER } from '../../../lib/stream/constants';
-import { RoutingManager } from '../RoutingManager';
-
-const queueDataStreamer = new Meteor.Streamer(LIVECHAT_INQUIRY_QUEUE_STREAM_OBSERVER);
-queueDataStreamer.allowWrite('none');
-queueDataStreamer.allowRead(function() {
-	return this.userId ? hasPermission(this.userId, 'view-l-room') : false;
-});
-
-const emitQueueDataEvent = (event, data) => queueDataStreamer.emitWithoutBroadcast(event, data);
-const mountDataToEmit = (type, data) => ({ type, ...data });
-
-LivechatInquiry.on('change', ({ clientAction, id: _id, data: record }) => {
-	if (RoutingManager.getConfig().autoAssignAgent) {
-		return;
-	}
-
-	switch (clientAction) {
-		case 'inserted':
-			emitQueueDataEvent(_id, { ...record, clientAction });
-			if (record && record.department) {
-				return emitQueueDataEvent(`department/${ record.department }`, mountDataToEmit('added', record));
-			}
-			emitQueueDataEvent('public', mountDataToEmit('added', record));
-			break;
-		case 'updated':
-			const isUpdatingDepartment = record && record.department;
-			const updatedRecord = LivechatInquiry.findOneById(_id);
-			emitQueueDataEvent(_id, { ...updatedRecord, clientAction });
-			if (updatedRecord && !updatedRecord.department) {
-				return emitQueueDataEvent('public', mountDataToEmit('changed', updatedRecord));
-			}
-			if (isUpdatingDepartment) {
-				emitQueueDataEvent('public', mountDataToEmit('changed', updatedRecord));
-			}
-			emitQueueDataEvent(`department/${ updatedRecord.department }`, mountDataToEmit('changed', updatedRecord));
-			break;
-
-		case 'removed':
-			const removedRecord = LivechatInquiry.trashFindOneById(_id);
-			emitQueueDataEvent(_id, { _id, clientAction });
-			if (removedRecord && removedRecord.department) {
-				return emitQueueDataEvent(`department/${ removedRecord.department }`, mountDataToEmit('removed', { _id }));
-			}
-			emitQueueDataEvent('public', mountDataToEmit('removed', { _id }));
-			break;
-	}
-});

app/livechat/server/roomAccessValidator.compatibility.js

@@ -0,0 +1,51 @@
+import { LivechatRooms } from '../../models';
+import { hasPermission, hasRole } from '../../authorization';
+import { LivechatDepartment, LivechatDepartmentAgents, LivechatInquiry } from '../../models/server';
+import { RoutingManager } from './lib/RoutingManager';
+
+export const validators = [
+	function(room, user) {
+		return hasPermission(user._id, 'view-livechat-rooms');
+	},
+	function(room, user) {
+		const { _id: userId } = user;
+		const { servedBy: { _id: agentId } = {} } = room;
+		return userId === agentId || (!room.open && hasPermission(user._id, 'view-livechat-room-closed-by-another-agent'));
+	},
+	function(room, user, extraData) {
+		if (extraData && extraData.rid) {
+			room = LivechatRooms.findOneById(extraData.rid);
+		}
+		return extraData && extraData.visitorToken && room.v && room.v.token === extraData.visitorToken;
+	},
+	function(room, user) {
+		const { previewRoom } = RoutingManager.getConfig();
+		if (!previewRoom) {
+			return;
+		}
+
+		let departmentIds;
+		if (!hasRole(user._id, 'livechat-manager')) {
+			const departmentAgents = LivechatDepartmentAgents.findByAgentId(user._id).fetch().map((d) => d.departmentId);
+			departmentIds = LivechatDepartment.find({ _id: { $in: departmentAgents }, enabled: true }).fetch().map((d) => d._id);
+		}
+
+		const filter = {
+			rid: room._id,
+			...departmentIds && departmentIds.length > 0 && { department: { $in: departmentIds } },
+		};
+
+		const inquiry = LivechatInquiry.findOne(filter, { fields: { status: 1 } });
+		return inquiry && inquiry.status === 'queued';
+	},
+	function(room, user) {
+		if (!room.departmentId || room.open) {
+			return;
+		}
+		const agentOfDepartment = LivechatDepartmentAgents.findOneByAgentIdAndDepartmentId(user._id, room.departmentId);
+		if (!agentOfDepartment) {
+			return;
+		}
+		return hasPermission(user._id, 'view-livechat-room-closed-same-department');
+	},
+];

app/livechat/server/roomAccessValidator.internalService.ts

@@ -0,0 +1,22 @@
+import { ServiceClass } from '../../../server/sdk/types/ServiceClass';
+import { IAuthorizationLivechat } from '../../../server/sdk/types/IAuthorizationLivechat';
+import { validators } from './roomAccessValidator.compatibility';
+import { api } from '../../../server/sdk/api';
+import { IRoom } from '../../../definition/IRoom';
+import { IUser } from '../../../definition/IUser';
+
+class AuthorizationLivechat extends ServiceClass implements IAuthorizationLivechat {
+	protected name = 'authorization-livechat';
+
+	async canAccessRoom(room: Partial<IRoom>, user: Pick<IUser, '_id'>, extraData?: object): Promise<boolean> {
+		for (const validator of validators) {
+			if (validator(room, user, extraData)) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+}
+
+api.registerService(new AuthorizationLivechat());

app/livechat/server/roomType.js

@@ -31,10 +31,6 @@ class LivechatRoomTypeServer extends LivechatRoomType {
 		const { token } = message;
 		return { token };
 	}
-
-	isEmitAllowed() {
-		return true;
-	}
 }
 
 roomTypes.add(new LivechatRoomTypeServer());

app/livechat/server/startup.js

@@ -3,76 +3,18 @@ import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { roomTypes } from '../../utils';
 import { LivechatRooms } from '../../models';
-import { hasPermission, hasRole, addRoomAccessValidator } from '../../authorization';
 import { callbacks } from '../../callbacks';
 import { settings } from '../../settings';
-import { LivechatDepartment, LivechatDepartmentAgents, LivechatInquiry } from '../../models/server';
-import { RoutingManager } from './lib/RoutingManager';
 import { LivechatAgentActivityMonitor } from './statistics/LivechatAgentActivityMonitor';
 import { businessHourManager } from './business-hour';
 import { createDefaultBusinessHourIfNotExists } from './business-hour/Helper';
+import { hasPermission } from '../../authorization/server';
 
-function allowAccessClosedRoomOfSameDepartment(room, user) {
-	if (!room || !user || room.t !== 'l' || !room.departmentId || room.open) {
-		return;
-	}
-	const agentOfDepartment = LivechatDepartmentAgents.findOneByAgentIdAndDepartmentId(user._id, room.departmentId);
-	if (!agentOfDepartment) {
-		return;
-	}
-	return hasPermission(user._id, 'view-livechat-room-closed-same-department');
-}
+import './roomAccessValidator.internalService';
 
 Meteor.startup(async () => {
 	roomTypes.setRoomFind('l', (_id) => LivechatRooms.findOneById(_id));
 
-	addRoomAccessValidator(function(room, user) {
-		return room && room.t === 'l' && user && hasPermission(user._id, 'view-livechat-rooms');
-	});
-
-	addRoomAccessValidator(function(room, user) {
-		if (!room || !user || room.t !== 'l') {
-			return;
-		}
-		const { _id: userId } = user;
-		const { servedBy: { _id: agentId } = {} } = room;
-		return userId === agentId || (!room.open && hasPermission(user._id, 'view-livechat-room-closed-by-another-agent'));
-	});
-
-	addRoomAccessValidator(function(room, user, extraData) {
-		if (!room && extraData && extraData.rid) {
-			room = LivechatRooms.findOneById(extraData.rid);
-		}
-		return room && room.t === 'l' && extraData && extraData.visitorToken && room.v && room.v.token === extraData.visitorToken;
-	});
-
-	addRoomAccessValidator(function(room, user) {
-		const { previewRoom } = RoutingManager.getConfig();
-		if (!previewRoom) {
-			return;
-		}
-
-		if (!user || !room || room.t !== 'l') {
-			return;
-		}
-
-		let departmentIds;
-		if (!hasRole(user._id, 'livechat-manager')) {
-			const departmentAgents = LivechatDepartmentAgents.findByAgentId(user._id).fetch().map((d) => d.departmentId);
-			departmentIds = LivechatDepartment.find({ _id: { $in: departmentAgents }, enabled: true }).fetch().map((d) => d._id);
-		}
-
-		const filter = {
-			rid: room._id,
-			...departmentIds && departmentIds.length > 0 && { department: { $in: departmentIds } },
-		};
-
-		const inquiry = LivechatInquiry.findOne(filter, { fields: { status: 1 } });
-		return inquiry && inquiry.status === 'queued';
-	});
-
-	addRoomAccessValidator(allowAccessClosedRoomOfSameDepartment);
-
 	callbacks.add('beforeLeaveRoom', function(user, room) {
 		if (room.t !== 'l') {
 			return user;

app/logger/server/streamer.js

@@ -6,7 +6,7 @@ import { EJSON } from 'meteor/ejson';
 import { Log } from 'meteor/logging';
 
 import { settings } from '../../settings';
-import { hasPermission } from '../../authorization/server';
+import notifications from '../../notifications/server/lib/Notifications';
 
 export const processString = function(string, date) {
 	let obj;
@@ -52,17 +52,17 @@ export const StdOut = new class extends EventEmitter {
 	}
 }();
 
-const stdoutStreamer = new Meteor.Streamer('stdout');
-stdoutStreamer.allowWrite('none');
-stdoutStreamer.allowRead(function() {
-	return this.userId ? hasPermission(this.userId, 'view-logs') : false;
-});
-
 Meteor.startup(() => {
 	const handler = (string, item) => {
-		stdoutStreamer.emitWithoutBroadcast('stdout', {
+		// TODO having this as 'emitWithoutBroadcast' will not sent this data to ddp-streamer, so this data
+		// won't be available when using micro services.
+		notifications.streamStdout.emitWithoutBroadcast('stdout', {
 			...item,
 		});
 	};
-	StdOut.on('write', handler);
+
+	// do not emit to StdOut if moleculer log level set to debug because it creates an infinite loop
+	if (String(process.env.MOLECULER_LOG_LEVEL).toLowerCase() !== 'debug') {
+		StdOut.on('write', handler);
+	}
 });

app/mentions/server/server.js

@@ -1,13 +1,12 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import _ from 'underscore';
 
 import MentionsServer from './Mentions';
 import { settings } from '../../settings';
 import { callbacks } from '../../callbacks';
-import { Notifications } from '../../notifications';
 import { Users, Subscriptions, Rooms } from '../../models';
+import { api } from '../../../server/sdk/api';
 
 const mention = new MentionsServer({
 	pattern: () => settings.get('UTF8_Names_Validation'),
@@ -21,12 +20,8 @@ const mention = new MentionsServer({
 		const { language } = this.getUser(sender._id);
 		const msg = TAPi18n.__('Group_mentions_disabled_x_members', { total: this.messageMaxAll }, language);
 
-		Notifications.notifyUser(sender._id, 'message', {
-			_id: Random.id(),
-			rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', sender._id, rid, {
 			msg,
-			groupable: false,
 		});
 
 		// Also throw to stop propagation of 'sendMessage'.

app/meteor-accounts-saml/server/lib/Utils.ts

@@ -393,7 +393,7 @@ export class SAMLUtils {
 		return mainValue;
 	}
 
-	public static convertArrayBufferToString(buffer: ArrayBuffer, encoding = 'utf8'): string {
+	public static convertArrayBufferToString(buffer: ArrayBuffer, encoding: BufferEncoding = 'utf8'): string {
 		return Buffer.from(buffer).toString(encoding);
 	}
 

app/models/server/models/LivechatRooms.js

@@ -137,7 +137,7 @@ export class LivechatRooms extends Base {
 		return this.find(query, options);
 	}
 
-	findOneById(_id, fields) {
+	findOneById(_id, fields = {}) {
 		const options = {};
 
 		if (fields) {

app/models/server/models/Subscriptions.js

@@ -453,7 +453,7 @@ export class Subscriptions extends Base {
 	}
 
 	// FIND ONE
-	findOneByRoomIdAndUserId(roomId, userId, options) {
+	findOneByRoomIdAndUserId(roomId, userId, options = {}) {
 		const query = {
 			rid: roomId,
 			'u._id': userId,
@@ -561,7 +561,7 @@ export class Subscriptions extends Base {
 		return this.find(query, options);
 	}
 
-	findByRoomIdAndNotUserId(roomId, userId, options) {
+	findByRoomIdAndNotUserId(roomId, userId, options = {}) {
 		const query = {
 			rid: roomId,
 			'u._id': {

app/models/server/models/Users.js

@@ -608,7 +608,7 @@ export class Users extends Base {
 		return this.findOne(query, options);
 	}
 
-	findOneById(userId, options) {
+	findOneById(userId, options = {}) {
 		const query = { _id: userId };
 
 		return this.findOne(query, options);

app/models/server/models/_BaseDb.js

@@ -48,6 +48,14 @@ export class BaseDb extends EventEmitter {
 
 		this.wrapModel();
 
+		if (!process.env.DISABLE_DB_WATCH) {
+			this.initDbWatch();
+		}
+
+		this.tryEnsureIndex({ _updatedAt: 1 }, options._updatedAtIndexOptions);
+	}
+
+	initDbWatch() {
 		const _oplogHandle = Promise.await(getOplogHandle());
 
 		// When someone start listening for changes we start oplog if available
@@ -85,8 +93,6 @@ export class BaseDb extends EventEmitter {
 		if (_oplogHandle) {
 			this.on('newListener', handleListener);
 		}
-
-		this.tryEnsureIndex({ _updatedAt: 1 }, options._updatedAtIndexOptions);
 	}
 
 	get baseName() {

app/models/server/models/_oplogHandle.ts

@@ -1,13 +1,13 @@
 import { Meteor } from 'meteor/meteor';
 import { Promise } from 'meteor/promise';
-import { MongoInternals } from 'meteor/mongo';
+import { MongoInternals, OplogHandle } from 'meteor/mongo';
 import semver from 'semver';
 import s from 'underscore.string';
 import { MongoClient, Cursor, Timestamp, Db } from 'mongodb';
 
 import { urlParser } from './_oplogUrlParser';
 
-class OplogHandle {
+class CustomOplogHandle {
 	dbName: string;
 
 	client: MongoClient;
@@ -43,7 +43,7 @@ class OplogHandle {
 		return true;
 	}
 
-	async start(): Promise<OplogHandle> {
+	async start(): Promise<CustomOplogHandle> {
 		this.usingChangeStream = await this.isChangeStreamAvailable();
 		const oplogUrl = this.usingChangeStream ? process.env.MONGO_URL : process.env.MONGO_OPLOG_URL;
 
@@ -70,7 +70,7 @@ class OplogHandle {
 		this.client = new MongoClient(oplogUrl, {
 			useUnifiedTopology: true,
 			useNewUrlParser: true,
-			...!this.usingChangeStream && { poolSize: 1 },
+			poolSize: this.usingChangeStream ? 15 : 1,
 		});
 
 		await this.client.connect();
@@ -83,6 +83,10 @@ class OplogHandle {
 		return this;
 	}
 
+	async stop(): Promise<void> {
+		return this.client?.close();
+	}
+
 	async startOplog(): Promise<void> {
 		const isMasterDoc = await this.db.admin().command({ ismaster: 1 });
 		if (!isMasterDoc || !isMasterDoc.setName) {
@@ -152,7 +156,10 @@ class OplogHandle {
 							// o: event.fullDocument,
 							o: {
 								$set: event.updateDescription.updatedFields,
-								$unset: event.updateDescription.removedFields,
+								$unset: event.updateDescription.removedFields.reduce((obj, field) => {
+									obj[field as string] = true;
+									return obj;
+								}, {} as Record<string, true>),
 							},
 						},
 					});
@@ -174,25 +181,32 @@ class OplogHandle {
 	}
 }
 
-let oplogHandle: Promise<OplogHandle>;
+let oplogHandle: Promise<CustomOplogHandle>;
 
-// @ts-ignore
-// eslint-disable-next-line no-undef
-if (Package['disable-oplog']) {
-	try {
-		oplogHandle = Promise.await(new OplogHandle().start());
-	} catch (e) {
-		console.error(e.message);
+if (!process.env.DISABLE_DB_WATCH) {
+	// @ts-ignore
+	// eslint-disable-next-line no-undef
+	if (Package['disable-oplog']) {
+		try {
+			oplogHandle = Promise.await(new CustomOplogHandle().start());
+		} catch (e) {
+			console.error(e.message);
+		}
 	}
 }
 
-export const getOplogHandle = async (): Promise<OplogHandle | undefined> => {
+export const getOplogHandle = async (): Promise<OplogHandle | CustomOplogHandle | undefined> => {
+	if (process.env.DISABLE_DB_WATCH) {
+		return;
+	}
+
 	if (oplogHandle) {
 		return oplogHandle;
 	}
 
 	const { mongo } = MongoInternals.defaultRemoteCollectionDriver();
-	if (mongo._oplogHandle?.onOplogEntry) {
-		return mongo._oplogHandle;
+	if (!mongo._oplogHandle?.onOplogEntry) {
+		return;
 	}
+	return mongo._oplogHandle;
 };

app/models/server/raw/BaseRaw.js

@@ -1,52 +0,0 @@
-export class BaseRaw {
-	constructor(col) {
-		this.col = col;
-	}
-
-	_ensureDefaultFields(options) {
-		if (!this.defaultFields) {
-			return options;
-		}
-
-		if (!options) {
-			return { projection: this.defaultFields };
-		}
-
-		// TODO: change all places using "fields" for raw models and remove the additional condition here
-		if ((options.projection != null && Object.keys(options.projection).length > 0)
-		|| (options.fields != null && Object.keys(options.fields).length > 0)) {
-			return options;
-		}
-
-		return {
-			...options,
-			projection: this.defaultFields,
-		};
-	}
-
-	findOneById(_id, options = {}) {
-		return this.findOne({ _id }, options);
-	}
-
-	findOne(query = {}, options = {}) {
-		const optionsDef = this._ensureDefaultFields(options);
-		return this.col.findOne(query, optionsDef);
-	}
-
-	findUsersInRoles() {
-		throw new Error('overwrite-function', 'You must overwrite this function in the extended classes');
-	}
-
-	find(query = {}, options = {}) {
-		const optionsDef = this._ensureDefaultFields(options);
-		return this.col.find(query, optionsDef);
-	}
-
-	update(...args) {
-		return this.col.update(...args);
-	}
-
-	removeById(_id) {
-		return this.col.deleteOne({ _id });
-	}
-}

app/models/server/raw/BaseRaw.ts

@@ -0,0 +1,94 @@
+import { Collection, FindOneOptions, Cursor, WriteOpResult, DeleteWriteOpResultObject, FilterQuery, UpdateQuery, UpdateOneOptions } from 'mongodb';
+
+interface ITrash {
+	__collection__: string;
+}
+
+export interface IBaseRaw<T> {
+	col: Collection<T>;
+}
+
+const baseName = 'rocketchat_';
+
+export class BaseRaw<T> implements IBaseRaw<T> {
+	public defaultFields?: Record<string, 1 | 0>;
+
+	protected name: string;
+
+	constructor(
+		public readonly col: Collection<T>,
+		public readonly trash?: Collection<T>,
+	) {
+		this.name = this.col.collectionName.replace(baseName, '');
+	}
+
+	_ensureDefaultFields<T>(options: FindOneOptions<T>): FindOneOptions<T> {
+		if (!this.defaultFields) {
+			return options;
+		}
+
+		if (!options) {
+			return { projection: this.defaultFields };
+		}
+
+		// TODO: change all places using "fields" for raw models and remove the additional condition here
+		if ((options.projection != null && Object.keys(options.projection).length > 0)
+		|| (options.fields != null && Object.keys(options.fields).length > 0)) {
+			return options;
+		}
+
+		return {
+			...options,
+			projection: this.defaultFields,
+		};
+	}
+
+	async findOneById(_id: string, options: FindOneOptions<T> = {}): Promise<T | undefined> {
+		return this.findOne({ _id }, options);
+	}
+
+	async findOne(query = {}, options: FindOneOptions<T> = {}): Promise<T | undefined> {
+		const optionsDef = this._ensureDefaultFields<T>(options);
+
+		if (typeof query === 'string') {
+			return this.findOneById(query, options);
+		}
+
+		return await this.col.findOne<T>(query, optionsDef) ?? undefined;
+	}
+
+	findUsersInRoles(): void {
+		throw new Error('[overwrite-function] You must overwrite this function in the extended classes');
+	}
+
+	find(query = {}, options: FindOneOptions<T> = {}): Cursor<T> {
+		const optionsDef = this._ensureDefaultFields(options);
+		return this.col.find(query, optionsDef);
+	}
+
+	update(filter: FilterQuery<T>, update: UpdateQuery<T> | Partial<T>, options?: UpdateOneOptions & { multi?: boolean }): Promise<WriteOpResult> {
+		return this.col.update(filter, update, options);
+	}
+
+	removeById(_id: string): Promise<DeleteWriteOpResultObject> {
+		const query: object = { _id };
+		return this.col.deleteOne(query);
+	}
+
+	// Trash
+	trashFind(query: FilterQuery<T & ITrash>, options: FindOneOptions<T>): Cursor<T> | undefined {
+		return this.trash?.find<T>({
+			__collection__: this.name,
+			...query,
+		}, options);
+	}
+
+	async trashFindOneById(_id: string, options: FindOneOptions<T> = {}): Promise<T | undefined> {
+		const query: object = {
+			_id,
+			__collection__: this.name,
+		};
+
+		return await this.trash?.findOne<T>(query, options) ?? undefined;
+	}
+}

app/models/server/raw/InstanceStatus.ts

@@ -0,0 +1,4 @@
+import { BaseRaw } from './BaseRaw';
+import { IInstanceStatus } from '../../../../definition/IInstanceStatus';
+
+export class InstanceStatusRaw extends BaseRaw<IInstanceStatus> {}

app/models/server/raw/IntegrationHistory.ts

@@ -0,0 +1,4 @@
+import { BaseRaw } from './BaseRaw';
+import { IIntegrationHistory } from '../../../../definition/IIntegrationHistory';
+
+export class IntegrationHistoryRaw extends BaseRaw<IIntegrationHistory> {}

app/models/server/raw/LivechatBusinessHours.ts

@@ -17,10 +17,10 @@ export interface IWorkHoursCronJobsWrapper {
 	finish: IWorkHoursCronJobsItem[];
 }
 
-export class LivechatBusinessHoursRaw extends BaseRaw {
+export class LivechatBusinessHoursRaw extends BaseRaw<ILivechatBusinessHour> {
 	public readonly col!: Collection<ILivechatBusinessHour>;
 
-	findOneDefaultBusinessHour(options?: any): Promise<ILivechatBusinessHour> {
+	findOneDefaultBusinessHour(options?: any): Promise<ILivechatBusinessHour | undefined> {
 		return this.findOne({ type: LivechatBusinessHourTypes.DEFAULT }, options);
 	}
 

app/models/server/raw/LoginServiceConfiguration.ts

@@ -0,0 +1,4 @@
+import { BaseRaw } from './BaseRaw';
+import { ILoginServiceConfiguration } from '../../../../definition/ILoginServiceConfiguration';
+
+export class LoginServiceConfigurationRaw extends BaseRaw<ILoginServiceConfiguration> {}

app/models/server/raw/NotificationQueue.ts

@@ -7,7 +7,7 @@ import {
 import { BaseRaw } from './BaseRaw';
 import { INotification } from '../../../../definition/INotification';
 
-export class NotificationQueueRaw extends BaseRaw {
+export class NotificationQueueRaw extends BaseRaw<INotification> {
 	public readonly col!: Collection<INotification>;
 
 	unsetSendingById(_id: string) {

app/models/server/raw/OmnichannelQueue.ts

@@ -1,15 +1,9 @@
 /* eslint-disable @typescript-eslint/explicit-function-return-type */
-import {
-	Collection,
-} from 'mongodb';
-
 import { BaseRaw } from './BaseRaw';
 import { IOmnichannelQueueStatus } from '../../../../definition/IOmnichannel';
 
 const UNIQUE_QUEUE_ID = 'queue';
-export class OmnichannelQueueRaw extends BaseRaw {
-	public readonly col!: Collection<IOmnichannelQueueStatus>;
-
+export class OmnichannelQueueRaw extends BaseRaw<IOmnichannelQueueStatus> {
 	initQueue() {
 		return this.col.updateOne({
 			_id: UNIQUE_QUEUE_ID,

app/models/server/raw/Permissions.js

@@ -1,4 +0,0 @@
-import { BaseRaw } from './BaseRaw';
-
-export class PermissionsRaw extends BaseRaw {
-}

app/models/server/raw/Permissions.ts

@@ -0,0 +1,5 @@
+import { BaseRaw } from './BaseRaw';
+import { IPermission } from '../../../../definition/IPermission';
+
+export class PermissionsRaw extends BaseRaw<IPermission> {
+}

app/models/server/raw/Roles.js

@@ -1,8 +1,12 @@
 import { BaseRaw } from './BaseRaw';
 
-import * as Models from './index';
-
 export class RolesRaw extends BaseRaw {
+	constructor(col, trash, models) {
+		super(col, trash);
+
+		this.models = models;
+	}
+
 	async isUserInRoles(userId, roles, scope) {
 		roles = [].concat(roles);
 
@@ -12,7 +16,7 @@ export class RolesRaw extends BaseRaw {
 			// eslint-disable-next-line no-await-in-loop
 			const role = await this.findOne({ _id: roleName });
 			const roleScope = (role && role.scope) || 'Users';
-			const model = Models[roleScope];
+			const model = this.models[roleScope];
 
 			// eslint-disable-next-line no-await-in-loop
 			const permitted = await (model && model.isUserInRole && model.isUserInRole(userId, roleName, scope));

app/models/server/raw/ServerEvents.ts

@@ -4,7 +4,7 @@ import { BaseRaw } from './BaseRaw';
 import { IServerEvent, IServerEventType } from '../../../../definition/IServerEvent';
 import { IUser } from '../../../../definition/IUser';
 
-export class ServerEventsRaw extends BaseRaw {
+export class ServerEventsRaw extends BaseRaw<IServerEvent> {
 	public readonly col!: Collection<IServerEvent>;
 
 	async insertOne(data: Omit<IServerEvent, '_id'>): Promise<any> {

app/models/server/raw/Settings.js

@@ -1,37 +0,0 @@
-import { BaseRaw } from './BaseRaw';
-
-export class SettingsRaw extends BaseRaw {
-	async getValueById(_id) {
-		const setting = await this.col.findOne({ _id }, { projection: { value: 1 } });
-
-		return setting.value;
-	}
-
-	findByIds(_id = []) {
-		_id = [].concat(_id);
-
-		const query = {
-			_id: {
-				$in: _id,
-			},
-		};
-
-		return this.find(query);
-	}
-
-	updateValueById(_id, value) {
-		const query = {
-			blocked: { $ne: true },
-			value: { $ne: value },
-			_id,
-		};
-
-		const update = {
-			$set: {
-				value,
-			},
-		};
-
-		return this.col.update(query, update);
-	}
-}

app/models/server/raw/Settings.ts

@@ -0,0 +1,53 @@
+import { Cursor, WriteOpResult } from 'mongodb';
+
+import { BaseRaw } from './BaseRaw';
+import { ISetting } from '../../../../definition/ISetting';
+
+type T = ISetting;
+
+export class SettingsRaw extends BaseRaw<T> {
+	async getValueById(_id: string): Promise<ISetting['value'] | undefined> {
+		const setting = await this.findOne({ _id }, { projection: { value: 1 } });
+
+		return setting?.value;
+	}
+
+	findOneNotHiddenById(_id: string): Promise<T | undefined> {
+		const query = {
+			_id,
+			hidden: { $ne: true },
+		};
+
+		return this.findOne(query);
+	}
+
+	findByIds(_id: string[] | string = []): Cursor<T> {
+		if (typeof _id === 'string') {
+			_id = [_id];
+		}
+
+		const query = {
+			_id: {
+				$in: _id,
+			},
+		};
+
+		return this.find(query);
+	}
+
+	updateValueById(_id: string, value: any): Promise<WriteOpResult> {
+		const query = {
+			blocked: { $ne: true },
+			value: { $ne: value },
+			_id,
+		};
+
+		const update = {
+			$set: {
+				value,
+			},
+		};
+
+		return this.update(query, update);
+	}
+}

app/models/server/raw/Subscriptions.js

@@ -1,57 +0,0 @@
-import { BaseRaw } from './BaseRaw';
-
-export class SubscriptionsRaw extends BaseRaw {
-	findOneByRoomIdAndUserId(rid, uid, options) {
-		const query = {
-			rid,
-			'u._id': uid,
-		};
-
-		return this.col.findOne(query, options);
-	}
-
-	countByRoomIdAndUserId(rid, uid) {
-		const query = {
-			rid,
-			'u._id': uid,
-		};
-
-		const cursor = this.col.find(query);
-
-		return cursor.count();
-	}
-
-	isUserInRole(uid, roleName, rid) {
-		if (rid == null) {
-			return;
-		}
-
-		const query = {
-			'u._id': uid,
-			rid,
-			roles: roleName,
-		};
-
-		return this.findOne(query, { fields: { roles: 1 } });
-	}
-
-	setAsReadByRoomIdAndUserId(rid, uid, alert = false) {
-		const query = {
-			rid,
-			'u._id': uid,
-		};
-
-		const update = {
-			$set: {
-				open: true,
-				alert,
-				unread: 0,
-				userMentions: 0,
-				groupMentions: 0,
-				ls: new Date(),
-			},
-		};
-
-		return this.col.update(query, update);
-	}
-}

app/models/server/raw/Subscriptions.ts

@@ -0,0 +1,72 @@
+import { FindOneOptions, Cursor, UpdateQuery, FilterQuery } from 'mongodb';
+
+import { BaseRaw } from './BaseRaw';
+import { ISubscription } from '../../../../definition/ISubscription';
+
+type T = ISubscription;
+export class SubscriptionsRaw extends BaseRaw<T> {
+	findOneByRoomIdAndUserId(rid: string, uid: string, options: FindOneOptions<T> = {}): Promise<T | undefined> {
+		const query = {
+			rid,
+			'u._id': uid,
+		};
+
+		return this.findOne(query, options);
+	}
+
+	findByRoomIdAndNotUserId(roomId: string, userId: string, options: FindOneOptions<T> = {}): Cursor<T> {
+		const query = {
+			rid: roomId,
+			'u._id': {
+				$ne: userId,
+			},
+		};
+
+		return this.find(query, options);
+	}
+
+	countByRoomIdAndUserId(rid: string, uid: string): Promise<number> {
+		const query = {
+			rid,
+			'u._id': uid,
+		};
+
+		const cursor = this.find(query, { projection: { _id: 0 } });
+
+		return cursor.count();
+	}
+
+	async isUserInRole(uid: string, roleName: string, rid: string): Promise<T | undefined> {
+		if (rid == null) {
+			return;
+		}
+
+		const query = {
+			'u._id': uid,
+			rid,
+			roles: roleName,
+		};
+
+		return this.findOne(query, { projection: { roles: 1 } });
+	}
+
+	setAsReadByRoomIdAndUserId(rid: string, uid: string, alert = false, options: FindOneOptions<T> = {}): ReturnType<BaseRaw<T>['update']> {
+		const query: FilterQuery<T> = {
+			rid,
+			'u._id': uid,
+		};
+
+		const update: UpdateQuery<T> = {
+			$set: {
+				open: true,
+				alert,
+				unread: 0,
+				userMentions: 0,
+				groupMentions: 0,
+				ls: new Date(),
+			},
+		};
+
+		return this.update(query, update, options);
+	}
+}

app/models/server/raw/Users.js

@@ -27,6 +27,15 @@ export class UsersRaw extends BaseRaw {
 		return this.findOne(query, options);
 	}
 
+	findOneAgentById(_id, options) {
+		const query = {
+			_id,
+			roles: 'livechat-agent',
+		};
+
+		return this.findOne(query, options);
+	}
+
 	findUsersInRolesWithQuery(roles, query, options) {
 		roles = [].concat(roles);
 
@@ -48,6 +57,15 @@ export class UsersRaw extends BaseRaw {
 		return this.findOne(query, options);
 	}
 
+	findOneByIdAndLoginHashedToken(_id, token, options = {}) {
+		const query = {
+			_id,
+			'services.resume.loginTokens.hashedToken': token,
+		};
+
+		return this.findOne(query, options);
+	}
+
 	findByActiveUsersExcept(searchTerm, exceptions, options, searchFields, extraQuery = [], { startsWith = false, endsWith = false } = {}) {
 		if (exceptions == null) { exceptions = []; }
 		if (options == null) { options = {}; }

app/models/server/raw/UsersSessions.ts

@@ -0,0 +1,4 @@
+import { BaseRaw } from './BaseRaw';
+import { IUserSession } from '../../../../definition/IUserSession';
+
+export class UsersSessionsRaw extends BaseRaw<IUserSession> {}

app/models/server/raw/index.ts

@@ -51,35 +51,93 @@ import { NotificationQueueRaw } from './NotificationQueue';
 import LivechatBusinessHoursModel from '../models/LivechatBusinessHours';
 import { LivechatBusinessHoursRaw } from './LivechatBusinessHours';
 import ServerEventModel from '../models/ServerEvents';
+import { UsersSessionsRaw } from './UsersSessions';
+import UsersSessionsModel from '../models/UsersSessions';
 import { ServerEventsRaw } from './ServerEvents';
+import { trash } from '../models/_BaseDb';
+import { initWatchers } from '../../../../server/modules/watchers/watchers.module';
+import LoginServiceConfigurationModel from '../models/LoginServiceConfiguration';
+import { LoginServiceConfigurationRaw } from './LoginServiceConfiguration';
+import { InstanceStatusRaw } from './InstanceStatus';
+import InstanceStatusModel from '../models/InstanceStatus';
+import { IntegrationHistoryRaw } from './IntegrationHistory';
+import IntegrationHistoryModel from '../models/IntegrationHistory';
 import OmnichannelQueueModel from '../models/OmnichannelQueue';
 import { OmnichannelQueueRaw } from './OmnichannelQueue';
 
-export const Permissions = new PermissionsRaw(PermissionsModel.model.rawCollection());
-export const Roles = new RolesRaw(RolesModel.model.rawCollection());
-export const Subscriptions = new SubscriptionsRaw(SubscriptionsModel.model.rawCollection());
-export const Settings = new SettingsRaw(SettingsModel.model.rawCollection());
-export const Users = new UsersRaw(UsersModel.model.rawCollection());
-export const Sessions = new SessionsRaw(SessionsModel.model.rawCollection());
-export const Rooms = new RoomsRaw(RoomsModel.model.rawCollection());
-export const LivechatCustomField = new LivechatCustomFieldRaw(LivechatCustomFieldModel.model.rawCollection());
-export const LivechatTrigger = new LivechatTriggerRaw(LivechatTriggerModel.model.rawCollection());
-export const LivechatDepartment = new LivechatDepartmentRaw(LivechatDepartmentModel.model.rawCollection());
-export const LivechatDepartmentAgents = new LivechatDepartmentAgentsRaw(LivechatDepartmentAgentsModel.model.rawCollection());
-export const LivechatRooms = new LivechatRoomsRaw(LivechatRoomsModel.model.rawCollection());
-export const Messages = new MessagesRaw(MessagesModel.model.rawCollection());
-export const LivechatExternalMessage = new LivechatExternalMessageRaw(LivechatExternalMessagesModel.model.rawCollection());
-export const LivechatVisitors = new LivechatVisitorsRaw(LivechatVisitorsModel.model.rawCollection());
-export const LivechatInquiry = new LivechatInquiryRaw(LivechatInquiryModel.model.rawCollection());
-export const Integrations = new IntegrationsRaw(IntegrationsModel.model.rawCollection());
-export const EmojiCustom = new EmojiCustomRaw(EmojiCustomModel.model.rawCollection());
-export const WebdavAccounts = new WebdavAccountsRaw(WebdavAccountsModel.model.rawCollection());
-export const OAuthApps = new OAuthAppsRaw(OAuthAppsModel.model.rawCollection());
-export const CustomSounds = new CustomSoundsRaw(CustomSoundsModel.model.rawCollection());
-export const CustomUserStatus = new CustomUserStatusRaw(CustomUserStatusModel.model.rawCollection());
-export const LivechatAgentActivity = new LivechatAgentActivityRaw(LivechatAgentActivityModel.model.rawCollection());
-export const Statistics = new StatisticsRaw(StatisticsModel.model.rawCollection());
-export const NotificationQueue = new NotificationQueueRaw(NotificationQueueModel.model.rawCollection());
-export const LivechatBusinessHours = new LivechatBusinessHoursRaw(LivechatBusinessHoursModel.model.rawCollection());
-export const ServerEvents = new ServerEventsRaw(ServerEventModel.model.rawCollection());
-export const OmnichannelQueue = new OmnichannelQueueRaw(OmnichannelQueueModel.model.rawCollection());
+const trashCollection = trash.rawCollection();
+
+export const Permissions = new PermissionsRaw(PermissionsModel.model.rawCollection(), trashCollection);
+export const Subscriptions = new SubscriptionsRaw(SubscriptionsModel.model.rawCollection(), trashCollection);
+export const Settings = new SettingsRaw(SettingsModel.model.rawCollection(), trashCollection);
+export const Users = new UsersRaw(UsersModel.model.rawCollection(), trashCollection);
+export const Rooms = new RoomsRaw(RoomsModel.model.rawCollection(), trashCollection);
+export const LivechatCustomField = new LivechatCustomFieldRaw(LivechatCustomFieldModel.model.rawCollection(), trashCollection);
+export const LivechatTrigger = new LivechatTriggerRaw(LivechatTriggerModel.model.rawCollection(), trashCollection);
+export const LivechatDepartment = new LivechatDepartmentRaw(LivechatDepartmentModel.model.rawCollection(), trashCollection);
+export const LivechatDepartmentAgents = new LivechatDepartmentAgentsRaw(LivechatDepartmentAgentsModel.model.rawCollection(), trashCollection);
+export const LivechatRooms = new LivechatRoomsRaw(LivechatRoomsModel.model.rawCollection(), trashCollection);
+export const Messages = new MessagesRaw(MessagesModel.model.rawCollection(), trashCollection);
+export const LivechatExternalMessage = new LivechatExternalMessageRaw(LivechatExternalMessagesModel.model.rawCollection(), trashCollection);
+export const LivechatVisitors = new LivechatVisitorsRaw(LivechatVisitorsModel.model.rawCollection(), trashCollection);
+export const LivechatInquiry = new LivechatInquiryRaw(LivechatInquiryModel.model.rawCollection(), trashCollection);
+export const Integrations = new IntegrationsRaw(IntegrationsModel.model.rawCollection(), trashCollection);
+export const EmojiCustom = new EmojiCustomRaw(EmojiCustomModel.model.rawCollection(), trashCollection);
+export const WebdavAccounts = new WebdavAccountsRaw(WebdavAccountsModel.model.rawCollection(), trashCollection);
+export const OAuthApps = new OAuthAppsRaw(OAuthAppsModel.model.rawCollection(), trashCollection);
+export const CustomSounds = new CustomSoundsRaw(CustomSoundsModel.model.rawCollection(), trashCollection);
+export const CustomUserStatus = new CustomUserStatusRaw(CustomUserStatusModel.model.rawCollection(), trashCollection);
+export const LivechatAgentActivity = new LivechatAgentActivityRaw(LivechatAgentActivityModel.model.rawCollection(), trashCollection);
+export const Statistics = new StatisticsRaw(StatisticsModel.model.rawCollection(), trashCollection);
+export const NotificationQueue = new NotificationQueueRaw(NotificationQueueModel.model.rawCollection(), trashCollection);
+export const LivechatBusinessHours = new LivechatBusinessHoursRaw(LivechatBusinessHoursModel.model.rawCollection(), trashCollection);
+export const ServerEvents = new ServerEventsRaw(ServerEventModel.model.rawCollection(), trashCollection);
+export const Roles = new RolesRaw(RolesModel.model.rawCollection(), trashCollection, { Users, Subscriptions });
+export const UsersSessions = new UsersSessionsRaw(UsersSessionsModel.model.rawCollection(), trashCollection);
+export const LoginServiceConfiguration = new LoginServiceConfigurationRaw(LoginServiceConfigurationModel.model.rawCollection(), trashCollection);
+export const InstanceStatus = new InstanceStatusRaw(InstanceStatusModel.model.rawCollection(), trashCollection);
+export const IntegrationHistory = new IntegrationHistoryRaw(IntegrationHistoryModel.model.rawCollection(), trashCollection);
+export const Sessions = new SessionsRaw(SessionsModel.model.rawCollection(), trashCollection);
+export const OmnichannelQueue = new OmnichannelQueueRaw(OmnichannelQueueModel.model.rawCollection(), trashCollection);
+
+const map = {
+	[Messages.col.collectionName]: MessagesModel,
+	[Users.col.collectionName]: UsersModel,
+	[Subscriptions.col.collectionName]: SubscriptionsModel,
+	[Settings.col.collectionName]: SettingsModel,
+	[Roles.col.collectionName]: RolesModel,
+	[Permissions.col.collectionName]: PermissionsModel,
+	[LivechatInquiry.col.collectionName]: LivechatInquiryModel,
+	[LivechatDepartmentAgents.col.collectionName]: LivechatDepartmentAgentsModel,
+	[UsersSessions.col.collectionName]: UsersSessionsModel,
+	[Rooms.col.collectionName]: RoomsModel,
+	[LoginServiceConfiguration.col.collectionName]: LoginServiceConfigurationModel,
+	[InstanceStatus.col.collectionName]: InstanceStatusModel,
+	[IntegrationHistory.col.collectionName]: IntegrationHistoryModel,
+	[Integrations.col.collectionName]: IntegrationsModel,
+};
+
+!process.env.DISABLE_DB_WATCH && initWatchers({
+	Messages,
+	Users,
+	Subscriptions,
+	Settings,
+	LivechatInquiry,
+	LivechatDepartmentAgents,
+	UsersSessions,
+	Permissions,
+	Roles,
+	Rooms,
+	LoginServiceConfiguration,
+	InstanceStatus,
+	IntegrationHistory,
+	Integrations,
+}, (model, fn) => {
+	const meteorModel = map[model.col.collectionName];
+
+	if (!meteorModel) {
+		return;
+	}
+
+	meteorModel.on('change', fn);
+});

app/notifications/server/lib/Notifications.js

@@ -1,212 +0,0 @@
-import { Meteor } from 'meteor/meteor';
-import { DDPCommon } from 'meteor/ddp-common';
-
-import { WEB_RTC_EVENTS } from '../../../webrtc';
-import { Subscriptions, Rooms } from '../../../models/server';
-import { settings } from '../../../settings/server';
-
-const changedPayload = function(collection, id, fields) {
-	return DDPCommon.stringifyDDP({
-		msg: 'changed',
-		collection,
-		id,
-		fields,
-	});
-};
-const send = function(self, msg) {
-	if (!self.socket) {
-		return;
-	}
-	self.socket.send(msg);
-};
-class RoomStreamer extends Meteor.Streamer {
-	_publish(publication, eventName, options) {
-		super._publish(publication, eventName, options);
-		const uid = Meteor.userId();
-		if (/rooms-changed/.test(eventName)) {
-			const roomEvent = (...args) => send(publication._session, changedPayload(this.subscriptionName, 'id', {
-				eventName: `${ uid }/rooms-changed`,
-				args,
-			}));
-			const rooms = Subscriptions.find({ 'u._id': uid }, { fields: { rid: 1 } }).fetch();
-			rooms.forEach(({ rid }) => {
-				this.on(rid, roomEvent);
-			});
-
-			const userEvent = (clientAction, { rid }) => {
-				switch (clientAction) {
-					case 'inserted':
-						rooms.push({ rid });
-						this.on(rid, roomEvent);
-
-						// after a subscription is added need to emit the room again
-						roomEvent('inserted', Rooms.findOneById(rid));
-						break;
-
-					case 'removed':
-						this.removeListener(rid, roomEvent);
-						break;
-				}
-			};
-			this.on(uid, userEvent);
-
-			publication.onStop(() => {
-				this.removeListener(uid, userEvent);
-				rooms.forEach(({ rid }) => this.removeListener(rid, roomEvent));
-			});
-		}
-	}
-}
-
-class Notifications {
-	constructor() {
-		const self = this;
-		this.debug = false;
-		this.notifyUser = this.notifyUser.bind(this);
-		this.streamAll = new Meteor.Streamer('notify-all');
-		this.streamLogged = new Meteor.Streamer('notify-logged');
-		this.streamRoom = new Meteor.Streamer('notify-room');
-		this.streamRoomUsers = new Meteor.Streamer('notify-room-users');
-		this.streamUser = new RoomStreamer('notify-user');
-		this.streamAll.allowWrite('none');
-		this.streamLogged.allowWrite('none');
-		this.streamRoom.allowWrite('none');
-		this.streamRoomUsers.allowWrite(function(eventName, ...args) {
-			const [roomId, e] = eventName.split('/');
-			// const user = Meteor.users.findOne(this.userId, {
-			// 	fields: {
-			// 		username: 1
-			// 	}
-			// });
-			if (Subscriptions.findOneByRoomIdAndUserId(roomId, this.userId) != null) {
-				const subscriptions = Subscriptions.findByRoomIdAndNotUserId(roomId, this.userId).fetch();
-				subscriptions.forEach((subscription) => self.notifyUser(subscription.u._id, e, ...args));
-			}
-			return false;
-		});
-		this.streamUser.allowWrite('logged');
-		this.streamAll.allowRead('all');
-		this.streamLogged.allowRead('logged');
-		this.streamRoom.allowRead(function(eventName, extraData) {
-			const [roomId] = eventName.split('/');
-			const room = Rooms.findOneById(roomId);
-			if (!room) {
-				console.warn(`Invalid streamRoom eventName: "${ eventName }"`);
-				return false;
-			}
-			if (room.t === 'l' && extraData && extraData.token && room.v.token === extraData.token) {
-				return true;
-			}
-			if (this.userId == null) {
-				return false;
-			}
-			const subscription = Subscriptions.findOneByRoomIdAndUserId(roomId, this.userId, { fields: { _id: 1 } });
-			return subscription != null;
-		});
-		this.streamRoomUsers.allowRead('none');
-		this.streamUser.allowRead(function(eventName) {
-			const [userId] = eventName.split('/');
-			return (this.userId != null) && this.userId === userId;
-		});
-	}
-
-	notifyAll(eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyAll', [eventName, ...args]);
-		}
-		args.unshift(eventName);
-		return this.streamAll.emit.apply(this.streamAll, args);
-	}
-
-	notifyLogged(eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyLogged', [eventName, ...args]);
-		}
-		args.unshift(eventName);
-		return this.streamLogged.emit.apply(this.streamLogged, args);
-	}
-
-	notifyRoom(room, eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyRoom', [room, eventName, ...args]);
-		}
-		args.unshift(`${ room }/${ eventName }`);
-		return this.streamRoom.emit.apply(this.streamRoom, args);
-	}
-
-	notifyUser(userId, eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyUser', [userId, eventName, ...args]);
-		}
-		args.unshift(`${ userId }/${ eventName }`);
-		return this.streamUser.emit.apply(this.streamUser, args);
-	}
-
-	notifyAllInThisInstance(eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyAll', [eventName, ...args]);
-		}
-		args.unshift(eventName);
-		return this.streamAll.emitWithoutBroadcast.apply(this.streamAll, args);
-	}
-
-	notifyLoggedInThisInstance(eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyLogged', [eventName, ...args]);
-		}
-		args.unshift(eventName);
-		return this.streamLogged.emitWithoutBroadcast.apply(this.streamLogged, args);
-	}
-
-	notifyRoomInThisInstance(room, eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyRoomAndBroadcast', [room, eventName, ...args]);
-		}
-		args.unshift(`${ room }/${ eventName }`);
-		return this.streamRoom.emitWithoutBroadcast.apply(this.streamRoom, args);
-	}
-
-	notifyUserInThisInstance(userId, eventName, ...args) {
-		if (this.debug === true) {
-			console.log('notifyUserAndBroadcast', [userId, eventName, ...args]);
-		}
-		args.unshift(`${ userId }/${ eventName }`);
-		return this.streamUser.emitWithoutBroadcast.apply(this.streamUser, args);
-	}
-}
-
-const notifications = new Notifications();
-
-notifications.streamRoom.allowWrite(function(eventName, username, typing, extraData) {
-	const [roomId, e] = eventName.split('/');
-
-	if (isNaN(e) ? e === WEB_RTC_EVENTS.WEB_RTC : parseFloat(e) === WEB_RTC_EVENTS.WEB_RTC) {
-		return true;
-	}
-
-	if (e === 'typing') {
-		const key = settings.get('UI_Use_Real_Name') ? 'name' : 'username';
-		// typing from livechat widget
-		if (extraData && extraData.token) {
-			const room = Rooms.findOneById(roomId);
-			if (room && room.t === 'l' && room.v.token === extraData.token) {
-				return true;
-			}
-		}
-
-		const user = Meteor.users.findOne(this.userId, {
-			fields: {
-				[key]: 1,
-			},
-		});
-
-		if (!user) {
-			return false;
-		}
-
-		return user[key] === username;
-	}
-	return false;
-});
-
-export default notifications;

app/notifications/server/lib/Notifications.ts

@@ -0,0 +1,54 @@
+import { Meteor } from 'meteor/meteor';
+import { Promise } from 'meteor/promise';
+import { DDPCommon } from 'meteor/ddp-common';
+
+import { NotificationsModule } from '../../../../server/modules/notifications/notifications.module';
+import { Streamer, StreamerCentral } from '../../../../server/modules/streamer/streamer.module';
+import { api } from '../../../../server/sdk/api';
+import {
+	Subscriptions as SubscriptionsRaw,
+	Rooms as RoomsRaw,
+	Users as UsersRaw,
+	Settings as SettingsRaw,
+} from '../../../models/server/raw';
+
+// TODO: Replace this in favor of the api.broadcast
+StreamerCentral.on('broadcast', (name, eventName, args) => {
+	api.broadcast('stream', [
+		name,
+		eventName,
+		args,
+	]);
+});
+
+export class Stream extends Streamer {
+	registerPublication(name: string, fn: (eventName: string, options: boolean | {useCollection?: boolean; args?: any}) => void): void {
+		Meteor.publish(name, function(eventName, options) {
+			return Promise.await(fn.call(this, eventName, options));
+		});
+	}
+
+	registerMethod(methods: Record<string, (eventName: string, ...args: any[]) => any>): void {
+		Meteor.methods(methods);
+	}
+
+	changedPayload(collection: string, id: string, fields: Record<string, any>): string | false {
+		return DDPCommon.stringifyDDP({
+			msg: 'changed',
+			collection,
+			id,
+			fields,
+		});
+	}
+}
+
+const notifications = new NotificationsModule(Stream);
+
+notifications.configure({
+	Rooms: RoomsRaw,
+	Subscriptions: SubscriptionsRaw,
+	Users: UsersRaw,
+	Settings: SettingsRaw,
+});
+
+export default notifications;

app/reactions/server/setReaction.js

@@ -1,14 +1,13 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 import _ from 'underscore';
 
 import { Messages, EmojiCustom, Rooms } from '../../models';
-import { Notifications } from '../../notifications';
 import { callbacks } from '../../callbacks';
 import { emoji } from '../../emoji';
 import { isTheLastMessage, msgStream } from '../../lib';
 import { hasPermission } from '../../authorization/server/functions/hasPermission';
+import { api } from '../../../server/sdk/api';
 
 const removeUserReaction = (message, reaction, username) => {
 	message.reactions[reaction].usernames.splice(message.reactions[reaction].usernames.indexOf(username), 1);
@@ -112,10 +111,7 @@ Meteor.methods({
 			return Promise.await(executeSetReaction(reaction, messageId, shouldReact));
 		} catch (e) {
 			if (e.error === 'error-not-allowed' && e.reason && e.details && e.details.rid) {
-				Notifications.notifyUser(Meteor.userId(), 'message', {
-					_id: Random.id(),
-					rid: e.details.rid,
-					ts: new Date(),
+				api.broadcast('notify.ephemeralMessage', Meteor.userId(), e.details.rid, {
 					msg: e.reason,
 				});
 

app/search/server/events/events.js

@@ -2,7 +2,6 @@ import _ from 'underscore';
 
 import { settings } from '../../../settings/server';
 import { callbacks } from '../../../callbacks/server';
-import { Users, Rooms } from '../../../models/server';
 import { searchProviderService } from '../service/providerService';
 import SearchLogger from '../logger/logger';
 
@@ -19,61 +18,27 @@ class EventService {
 	}
 }
 
-const eventService = new EventService();
+export const searchEventService = new EventService();
 
 /**
  * Listen to message changes via Hooks
  */
 function afterSaveMessage(m) {
-	eventService.promoteEvent('message.save', m._id, m);
+	searchEventService.promoteEvent('message.save', m._id, m);
 	return m;
 }
 
 function afterDeleteMessage(m) {
-	eventService.promoteEvent('message.delete', m._id);
+	searchEventService.promoteEvent('message.delete', m._id);
 	return m;
 }
 
-/**
- * Listen to user and room changes via cursor
- */
-function onUsersChange({ clientAction, id, data }) {
-	switch (clientAction) {
-		case 'updated':
-		case 'inserted':
-			const user = data ?? Users.findOneById(id);
-			eventService.promoteEvent('user.save', id, user);
-			break;
-
-		case 'removed':
-			eventService.promoteEvent('user.delete', id);
-			break;
-	}
-}
-
-function onRoomsChange({ clientAction, id, data }) {
-	switch (clientAction) {
-		case 'updated':
-		case 'inserted':
-			const room = data ?? Rooms.findOneById(id);
-			eventService.promoteEvent('room.save', id, room);
-			break;
-
-		case 'removed':
-			eventService.promoteEvent('room.delete', id);
-			break;
-	}
-}
 
 settings.get('Search.Provider', _.debounce(() => {
 	if (searchProviderService.activeProvider?.on) {
-		Users.on('change', onUsersChange);
-		Rooms.on('change', onRoomsChange);
 		callbacks.add('afterSaveMessage', afterSaveMessage, callbacks.priority.MEDIUM, 'search-events');
 		callbacks.add('afterDeleteMessage', afterDeleteMessage, callbacks.priority.MEDIUM, 'search-events-delete');
 	} else {
-		Users.removeListener('change', onUsersChange);
-		Rooms.removeListener('change', onRoomsChange);
 		callbacks.remove('afterSaveMessage', 'search-events');
 		callbacks.remove('afterDeleteMessage', 'search-events-delete');
 	}

app/search/server/index.js

@@ -3,6 +3,7 @@ import { searchProviderService } from './service/providerService.js';
 import './service/validationService.js';
 import './events/events.js';
 import './provider/defaultProvider.js';
+import './search.internalService';
 
 
 export {

app/search/server/search.internalService.ts

@@ -0,0 +1,45 @@
+import _ from 'underscore';
+
+import { Users } from '../../models/server';
+import { settings } from '../../settings/server';
+import { searchProviderService } from './service/providerService';
+import { ServiceClass } from '../../../server/sdk/types/ServiceClass';
+import { api } from '../../../server/sdk/api';
+import { searchEventService } from './events/events';
+
+class Search extends ServiceClass {
+	protected name = 'search';
+
+	constructor() {
+		super();
+
+		this.onEvent('watch.users', async ({ clientAction, data, id }) => {
+			if (clientAction === 'removed') {
+				searchEventService.promoteEvent('user.delete', id, undefined);
+				return;
+			}
+
+			const user = data ?? Users.findOneById(id);
+			searchEventService.promoteEvent('user.save', id, user);
+		});
+
+		this.onEvent('watch.rooms', async ({ clientAction, room }) => {
+			if (clientAction === 'removed') {
+				searchEventService.promoteEvent('room.delete', room._id, undefined);
+				return;
+			}
+
+			searchEventService.promoteEvent('room.save', room._id, room);
+		});
+	}
+}
+
+const service = new Search();
+
+settings.get('Search.Provider', _.debounce(() => {
+	if (searchProviderService.activeProvider?.on) {
+		api.registerService(service);
+	} else {
+		api.destroyService(service);
+	}
+}, 1000));

app/settings/client/lib/settings.ts

@@ -2,7 +2,8 @@ import { Meteor } from 'meteor/meteor';
 import { ReactiveDict } from 'meteor/reactive-dict';
 
 import { PublicSettingsCachedCollection } from '../../../../client/lib/settings/PublicSettingsCachedCollection';
-import { SettingsBase, SettingValue } from '../../lib/settings';
+import { SettingsBase } from '../../lib/settings';
+import { SettingValue } from '../../../../definition/ISetting';
 
 class Settings extends SettingsBase {
 	cachedCollection = PublicSettingsCachedCollection.get()

app/settings/lib/settings.ts

@@ -1,9 +1,8 @@
 import { Meteor } from 'meteor/meteor';
 import _ from 'underscore';
 
-export type SettingValueMultiSelect = Array<{key: string; i18nLabel: string}>
-export type SettingValueRoomPick = Array<{_id: string; name: string}> | string
-export type SettingValue = string | boolean | number | SettingValueMultiSelect | undefined;
+import { SettingValue } from '../../../definition/ISetting';
+
 export type SettingComposedValue = {key: string; value: SettingValue};
 export type SettingCallback = (key: string, value: SettingValue, initialLoad?: boolean) => void;
 

app/settings/server/functions/settings.ts

@@ -3,9 +3,10 @@ import { EventEmitter } from 'events';
 import { Meteor } from 'meteor/meteor';
 import _ from 'underscore';
 
-import { SettingsBase, SettingValue } from '../../lib/settings';
+import { SettingsBase } from '../../lib/settings';
 import SettingsModel from '../../../models/server/models/Settings';
-import { setValue, updateValue } from '../raw';
+import { updateValue } from '../raw';
+import { ISetting, SettingValue } from '../../../../definition/ISetting';
 
 const blockedSettings = new Set<string>();
 const hiddenSettings = new Set<string>();
@@ -64,44 +65,8 @@ const overrideSetting = (_id: string, value: SettingValue, options: ISettingAddO
 	return value;
 };
 
-export interface ISettingAddOptions {
-	_id?: string;
-	type?: 'group' | 'boolean' | 'int' | 'string' | 'asset' | 'code' | 'select' | 'password' | 'action' | 'relativeUrl' | 'language' | 'date' | 'color' | 'font' | 'roomPick' | 'multiSelect';
-	editor?: string;
-	packageEditor?: string;
-	packageValue?: SettingValue;
-	valueSource?: string;
-	hidden?: boolean;
-	blocked?: boolean;
-	requiredOnWizard?: boolean;
-	secret?: boolean;
-	sorter?: number;
-	i18nLabel?: string;
-	i18nDescription?: string;
-	autocomplete?: boolean;
+export interface ISettingAddOptions extends Partial<ISetting> {
 	force?: boolean;
-	group?: string;
-	section?: string;
-	enableQuery?: any;
-	processEnvValue?: SettingValue;
-	meteorSettingsValue?: SettingValue;
-	value?: SettingValue;
-	ts?: Date;
-	multiline?: boolean;
-	values?: Array<ISettingSelectOption>;
-	public?: boolean;
-	enterprise?: boolean;
-	modules?: Array<string>;
-	invalidValue?: SettingValue;
-}
-export interface ISettingSelectOption {
-	key: string;
-	i18nLabel: string;
-}
-export interface ISettingRecord extends ISettingAddOptions {
-	_id: string;
-	env: boolean;
-	value: SettingValue;
 }
 
 export interface ISettingAddGroupOptions {
@@ -362,7 +327,7 @@ class Settings extends SettingsBase {
 	/*
 	* Change a setting value on the Meteor.settings object
 	*/
-	storeSettingValue(record: ISettingRecord, initialLoad: boolean): void {
+	storeSettingValue(record: ISetting, initialLoad: boolean): void {
 		const newData = {
 			value: record.value,
 		};
@@ -380,7 +345,7 @@ class Settings extends SettingsBase {
 	/*
 	* Remove a setting value on the Meteor.settings object
 	*/
-	removeSettingValue(record: ISettingRecord, initialLoad: boolean): void {
+	removeSettingValue(record: ISetting, initialLoad: boolean): void {
 		SettingsEvents.emit('remove-setting-value', record);
 
 		delete Meteor.settings[record._id];
@@ -396,28 +361,12 @@ class Settings extends SettingsBase {
 	*/
 	init(): void {
 		this.initialLoad = true;
-		SettingsModel.find().fetch().forEach((record: ISettingRecord) => {
+		SettingsModel.find().fetch().forEach((record: ISetting) => {
 			this.storeSettingValue(record, this.initialLoad);
 			updateValue(record._id, { value: record.value });
 		});
 		this.initialLoad = false;
 		this.afterInitialLoad.forEach((fn) => fn(Meteor.settings));
-
-		SettingsModel.on('change', ({ clientAction, id, data }) => {
-			switch (clientAction) {
-				case 'inserted':
-				case 'updated':
-					data = data ?? SettingsModel.findOneById(id);
-					this.storeSettingValue(data, this.initialLoad);
-					updateValue(id, { value: data.value });
-					break;
-				case 'removed':
-					data = SettingsModel.trashFindOneById(id);
-					this.removeSettingValue(data, this.initialLoad);
-					setValue(id, undefined);
-					break;
-			}
-		});
 	}
 
 	onAfterInitialLoad(fn: (settings: Meteor.Settings) => void): void {

app/settings/server/raw.js

@@ -1,4 +1,4 @@
-import { Settings } from '../../models/server/models/Settings';
+import Settings from '../../models/server/models/Settings';
 
 const cache = new Map();
 

app/slashcommands-archiveroom/server/server.js

@@ -1,11 +1,10 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { Rooms, Messages } from '../../models';
 import { slashCommands } from '../../utils';
-import { Notifications } from '../../notifications';
+import { api } from '../../../server/sdk/api';
 
 function Archive(command, params, item) {
 	if (command !== 'archive' || !Match.test(params, String)) {
@@ -26,10 +25,7 @@ function Archive(command, params, item) {
 	const user = Meteor.users.findOne(Meteor.userId());
 
 	if (!room) {
-		return Notifications.notifyUser(Meteor.userId(), 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 			msg: TAPi18n.__('Channel_doesnt_exist', {
 				postProcess: 'sprintf',
 				sprintf: [channel],
@@ -43,10 +39,7 @@ function Archive(command, params, item) {
 	}
 
 	if (room.archived) {
-		Notifications.notifyUser(Meteor.userId(), 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 			msg: TAPi18n.__('Duplicate_archived_channel_name', {
 				postProcess: 'sprintf',
 				sprintf: [channel],
@@ -57,10 +50,7 @@ function Archive(command, params, item) {
 	Meteor.call('archiveRoom', room._id);
 
 	Messages.createRoomArchivedByRoomIdAndUser(room._id, Meteor.user());
-	Notifications.notifyUser(Meteor.userId(), 'message', {
-		_id: Random.id(),
-		rid: item.rid,
-		ts: new Date(),
+	api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 		msg: TAPi18n.__('Channel_Archived', {
 			postProcess: 'sprintf',
 			sprintf: [channel],

app/slashcommands-create/server/server.js

@@ -1,12 +1,11 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { settings } from '../../settings';
-import { Notifications } from '../../notifications';
 import { Rooms } from '../../models';
 import { slashCommands } from '../../utils';
+import { api } from '../../../server/sdk/api';
 
 function Create(command, params, item) {
 	function getParams(str) {
@@ -36,10 +35,7 @@ function Create(command, params, item) {
 	const user = Meteor.users.findOne(Meteor.userId());
 	const room = Rooms.findOneByName(channel);
 	if (room != null) {
-		Notifications.notifyUser(Meteor.userId(), 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 			msg: TAPi18n.__('Channel_already_exist', {
 				postProcess: 'sprintf',
 				sprintf: [channel],

app/slashcommands-help/server/server.js

@@ -1,9 +1,8 @@
 import { Meteor } from 'meteor/meteor';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { slashCommands } from '../../utils';
-import { Notifications } from '../../notifications';
+import { api } from '../../../server/sdk/api';
 
 /*
 * Help is a named function that will replace /join commands
@@ -39,10 +38,7 @@ slashCommands.add('help', function Help(command, params, item) {
 	},
 	];
 	keys.forEach((key) => {
-		Notifications.notifyUser(Meteor.userId(), 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 			msg: TAPi18n.__(Object.keys(key)[0], {
 				postProcess: 'sprintf',
 				sprintf: [key[Object.keys(key)[0]]],

app/slashcommands-hide/server/hide.js

@@ -1,11 +1,10 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { Rooms, Subscriptions } from '../../models';
-import { Notifications } from '../../notifications';
 import { slashCommands } from '../../utils';
+import { api } from '../../../server/sdk/api';
 
 /*
 * Hide is a named function that will replace /hide commands
@@ -29,10 +28,7 @@ function Hide(command, param, item) {
 		});
 
 		if (!roomObject) {
-			return Notifications.notifyUser(user._id, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			api.broadcast('notify.ephemeralMessage', user._id, item.rid, {
 				msg: TAPi18n.__('Channel_doesnt_exist', {
 					postProcess: 'sprintf',
 					sprintf: [room],
@@ -41,10 +37,7 @@ function Hide(command, param, item) {
 		}
 
 		if (!Subscriptions.findOneByRoomIdAndUserId(room._id, user._id, { fields: { _id: 1 } })) {
-			return Notifications.notifyUser(user._id, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			return api.broadcast('notify.ephemeralMessage', user._id, item.rid, {
 				msg: TAPi18n.__('error-logged-user-not-in-room', {
 					postProcess: 'sprintf',
 					sprintf: [room],
@@ -56,10 +49,7 @@ function Hide(command, param, item) {
 
 	Meteor.call('hideRoom', rid, (error) => {
 		if (error) {
-			return Notifications.notifyUser(user._id, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			return api.broadcast('notify.ephemeralMessage', user._id, item.rid, {
 				msg: TAPi18n.__(error, null, user.language),
 			});
 		}

app/slashcommands-invite/server/server.js

@@ -1,11 +1,10 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
-import { Notifications } from '../../notifications';
 import { slashCommands } from '../../utils';
 import { Subscriptions } from '../../models';
+import { api } from '../../../server/sdk/api';
 
 /*
 * Invite is a named function that will replace /invite commands
@@ -30,10 +29,7 @@ function Invite(command, params, item) {
 	const userId = Meteor.userId();
 	const currentUser = Meteor.users.findOne(userId);
 	if (users.count() === 0) {
-		Notifications.notifyUser(userId, 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 			msg: TAPi18n.__('User_doesnt_exist', {
 				postProcess: 'sprintf',
 				sprintf: [usernames.join(' @')],
@@ -46,10 +42,7 @@ function Invite(command, params, item) {
 		if (subscription == null) {
 			return true;
 		}
-		Notifications.notifyUser(userId, 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 			msg: TAPi18n.__('Username_is_already_in_here', {
 				postProcess: 'sprintf',
 				sprintf: [user.username],
@@ -66,17 +59,11 @@ function Invite(command, params, item) {
 			});
 		} catch ({ error }) {
 			if (error === 'cant-invite-for-direct-room') {
-				Notifications.notifyUser(userId, 'message', {
-					_id: Random.id(),
-					rid: item.rid,
-					ts: new Date(),
+				api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 					msg: TAPi18n.__('Cannot_invite_users_to_direct_rooms', null, currentUser.language),
 				});
 			} else {
-				Notifications.notifyUser(userId, 'message', {
-					_id: Random.id(),
-					rid: item.rid,
-					ts: new Date(),
+				api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 					msg: TAPi18n.__(error, null, currentUser.language),
 				});
 			}

app/slashcommands-inviteall/server/server.js

@@ -4,13 +4,12 @@
  */
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { Rooms, Subscriptions } from '../../models';
 import { slashCommands } from '../../utils';
 import { settings } from '../../settings';
-import { Notifications } from '../../notifications';
+import { api } from '../../../server/sdk/api';
 
 function inviteAll(type) {
 	return function inviteAll(command, params, item) {
@@ -30,10 +29,7 @@ function inviteAll(type) {
 		const targetChannel = type === 'from' ? Rooms.findOneById(item.rid) : Rooms.findOneByName(channel);
 
 		if (!baseChannel) {
-			return Notifications.notifyUser(userId, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			return api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 				msg: TAPi18n.__('Channel_doesnt_exist', {
 					postProcess: 'sprintf',
 					sprintf: [channel],
@@ -52,10 +48,7 @@ function inviteAll(type) {
 
 			if (!targetChannel && ['c', 'p'].indexOf(baseChannel.t) > -1) {
 				Meteor.call(baseChannel.t === 'c' ? 'createChannel' : 'createPrivateGroup', channel, users);
-				Notifications.notifyUser(userId, 'message', {
-					_id: Random.id(),
-					rid: item.rid,
-					ts: new Date(),
+				api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 					msg: TAPi18n.__('Channel_created', {
 						postProcess: 'sprintf',
 						sprintf: [channel],
@@ -67,18 +60,12 @@ function inviteAll(type) {
 					users,
 				});
 			}
-			return Notifications.notifyUser(userId, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			return api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 				msg: TAPi18n.__('Users_added', null, currentUser.language),
 			});
 		} catch (e) {
 			const msg = e.error === 'cant-invite-for-direct-room' ? 'Cannot_invite_users_to_direct_rooms' : e.error;
-			Notifications.notifyUser(userId, 'message', {
-				_id: Random.id(),
-				rid: item.rid,
-				ts: new Date(),
+			api.broadcast('notify.ephemeralMessage', userId, item.rid, {
 				msg: TAPi18n.__(msg, null, currentUser.language),
 			});
 		}

app/slashcommands-join/server/server.js

@@ -1,11 +1,10 @@
 import { Meteor } from 'meteor/meteor';
 import { Match } from 'meteor/check';
-import { Random } from 'meteor/random';
 import { TAPi18n } from 'meteor/rocketchat:tap-i18n';
 
 import { Rooms, Subscriptions } from '../../models';
-import { Notifications } from '../../notifications';
 import { slashCommands } from '../../utils';
+import { api } from '../../../server/sdk/api';
 
 function Join(command, params, item) {
 	if (command !== 'join' || !Match.test(params, String)) {
@@ -19,10 +18,7 @@ function Join(command, params, item) {
 	const user = Meteor.users.findOne(Meteor.userId());
 	const room = Rooms.findOneByNameAndType(channel, 'c');
 	if (!room) {
-		Notifications.notifyUser(Meteor.userId(), 'message', {
-			_id: Random.id(),
-			rid: item.rid,
-			ts: new Date(),
+		api.broadcast('notify.ephemeralMessage', Meteor.userId(), item.rid, {
 			msg: TAPi18n.__('Channel_doesnt_exist', {
 				postProcess: 'sprintf',
 				sprintf: [channel],