refactor: move notifyListener actions out of instance-status package

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

@@ -1,6 +1,6 @@
 import { api, dbWatchersDisabled } from '@rocket.chat/core-services';
-import type { IRocketChatRecord, IRoom } from '@rocket.chat/core-typings';
-import { Rooms } from '@rocket.chat/models';
+import type { IInstanceStatus, IRocketChatRecord, IRoom } from '@rocket.chat/core-typings';
+import { InstanceStatus, Rooms } from '@rocket.chat/models';
 
 type ClientAction = 'inserted' | 'updated' | 'removed';
 
@@ -65,3 +65,20 @@ export async function notifyOnRoomChangedByUserDM<T extends IRoom>(
 		void api.broadcast('watch.rooms', { clientAction, room: item });
 	}
 }
+
+export async function notifyOnInstanceStatusChangedById<T extends IInstanceStatus>(
+	id: T['_id'],
+	clientAction: 'inserted' | 'updated' | 'removed',
+	data?: Record<string, unknown>,
+	diff?: Record<string, unknown>,
+) {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	const instanceStatus = await InstanceStatus.findOneById(id);
+
+	if (instanceStatus) {
+		void api.broadcast('watch.instanceStatus', { id, clientAction, data, diff });
+	}
+}

apps/meteor/ee/server/local-services/instance/service.ts

@@ -7,6 +7,7 @@ import EJSON from 'ejson';
 import type { BrokerNode } from 'moleculer';
 import { ServiceBroker, Transporters, Serializers } from 'moleculer';
 
+import { notifyOnInstanceStatusChangedById } from '../../../../app/lib/server/lib/notifyListener';
 import { StreamerCentral } from '../../../../server/modules/streamer/streamer.module';
 import type { IInstanceService } from '../../sdk/types/IInstanceService';
 import { getLogger } from './getLogger';
@@ -155,6 +156,10 @@ export class InstanceService extends ServiceClassInternal implements IInstanceSe
 		};
 
 		await InstanceStatus.registerInstance('rocket.chat', instance);
+		void notifyOnInstanceStatusChangedById(InstanceStatus.id(), 'inserted');
+		process.on('exit', async () => {
+			void notifyOnInstanceStatusChangedById(InstanceStatus.id(), 'removed');
+		});
 
 		try {
 			const hasLicense = await License.hasModule('scalability');

apps/meteor/ee/server/startup/presence.ts

@@ -4,9 +4,15 @@ import { Accounts } from 'meteor/accounts-base';
 import { Meteor } from 'meteor/meteor';
 import { throttle } from 'underscore';
 
+import { notifyOnInstanceStatusChangedById } from '../../../app/lib/server/lib/notifyListener';
+
 // update connections count every 30 seconds
-const updateConns = throttle(function _updateConns() {
-	void InstanceStatus.updateConnections(Meteor.server.sessions.size);
+const updateConns = throttle(async () => {
+	const { modifiedCount } = await InstanceStatus.updateConnections(Meteor.server.sessions.size);
+
+	if (modifiedCount) {
+		void notifyOnInstanceStatusChangedById(InstanceStatus.id(), 'updated', { 'extraInformation.conns': Meteor.server.sessions.size });
+	}
 }, 30000);
 
 Meteor.startup(() => {
@@ -21,7 +27,7 @@ Meteor.startup(() => {
 			}
 
 			await Presence.removeConnection(session.userId, connection.id, nodeId);
-			updateConns();
+			await updateConns();
 		});
 	});
 
@@ -42,13 +48,12 @@ Meteor.startup(() => {
 
 		void (async function () {
 			await Presence.newConnection(login.user._id, login.connection.id, nodeId);
-			updateConns();
+			await updateConns();
 		})();
 	});
 
-	Accounts.onLogout((login: any): void => {
+	Accounts.onLogout(async (login: any): Promise<void> => {
 		void Presence.removeConnection(login.user._id, login.connection.id, nodeId);
-
-		updateConns();
+		await updateConns();
 	});
 });

ee/apps/ddp-streamer/src/DDPStreamer.ts

@@ -9,6 +9,7 @@ import WebSocket from 'ws';
 import { ListenersModule } from '../../../../apps/meteor/server/modules/listeners/listeners.module';
 import type { NotificationsModule } from '../../../../apps/meteor/server/modules/notifications/notifications.module';
 import { StreamerCentral } from '../../../../apps/meteor/server/modules/streamer/streamer.module';
+import { notifyOnInstanceStatusChangedById } from '../../../../apps/meteor/app/lib/server/lib/notifyListener';
 import { Client, clientMap } from './Client';
 import { events, server } from './configureServer';
 import { DDP_EVENTS } from './constants';
@@ -64,8 +65,12 @@ export class DDPStreamer extends ServiceClass {
 	}
 
 	// update connections count every 30 seconds
-	updateConnections = throttle(() => {
-		InstanceStatus.updateConnections(this.wss?.clients.size ?? 0);
+	updateConnections = throttle(async () => {
+		const { modifiedCount } = await InstanceStatus.updateConnections(this.wss?.clients.size ?? 0);
+
+		if (modifiedCount) {
+			notifyOnInstanceStatusChangedById(InstanceStatus.id(), 'updated', { 'extraInformation.conns': this.wss?.clients.size ?? 0 });
+		}
 	}, 30000);
 
 	async created(): Promise<void> {
@@ -205,7 +210,8 @@ export class DDPStreamer extends ServiceClass {
 
 			this.wss.on('connection', (ws, req) => new Client(ws, req.url !== '/websocket', req));
 
-			InstanceStatus.registerInstance('ddp-streamer', {});
+			await InstanceStatus.registerInstance('ddp-streamer', {});
+			void notifyOnInstanceStatusChangedById(InstanceStatus.id(), 'inserted');
 		} catch (err) {
 			console.error('DDPStreamer did not start correctly', err);
 		}

packages/instance-status/src/index.ts

@@ -1,8 +1,8 @@
-// import { IInstanceStatus } from '@rocket.chat/core-typings';
 import { EventEmitter } from 'events';
 
 import { InstanceStatus as InstanceStatusModel } from '@rocket.chat/models';
 import { v4 as uuidv4 } from 'uuid';
+import type { UpdateResult } from 'mongodb';
 
 const events = new EventEmitter();
 
@@ -11,21 +11,6 @@ const defaultPingInterval = parseInt(String(process.env.MULTIPLE_INSTANCES_PING_
 // if not set via env var ensures at least 3 ticks before expiring (multiple of 60s)
 const indexExpire = (parseInt(String(process.env.MULTIPLE_INSTANCES_EXPIRE)) || Math.ceil((defaultPingInterval * 3) / 60)) * 60;
 
-const dbWatchersDisabled = ['yes', 'true'].includes(String(process.env.DISABLE_DB_WATCHERS).toLowerCase());
-
-function notifyOnInstanceStatusChangedById(
-	id: string,
-	clientAction: 'inserted' | 'updated' | 'removed',
-	data?: Record<string, unknown>,
-	diff?: Record<string, unknown>,
-) {
-	if (!dbWatchersDisabled) {
-		return;
-	}
-
-	events.emit('watch.instanceStatus', { clientAction, id, data, diff });
-}
-
 let createIndexes = async () => {
 	await InstanceStatusModel.col
 		.indexes()
@@ -90,16 +75,13 @@ async function registerInstance(name: string, extraInformation: Record<string, u
 	};
 
 	try {
-		const result = await InstanceStatusModel.updateOne({ _id: ID }, instance as any, { upsert: true });
-
-		const instanceStatus = await InstanceStatusModel.findOne({ _id: ID });
+		await InstanceStatusModel.updateOne({ _id: ID }, instance as any, { upsert: true });
+		const instanceStatus = await InstanceStatusModel.findOneById(ID);
 
 		start();
 
 		events.emit('registerInstance', instanceStatus, instance);
 
-		notifyOnInstanceStatusChangedById(ID, result.upsertedId ? 'inserted' : 'updated', { ...instanceStatus });
-
 		process.on('exit', onExit);
 
 		return instanceStatus;
@@ -115,10 +97,6 @@ async function unregisterInstance() {
 
 		events.emit('unregisterInstance', ID);
 
-		if (result.deletedCount) {
-			notifyOnInstanceStatusChangedById(ID, 'removed');
-		}
-
 		process.removeListener('exit', onExit);
 
 		return result;
@@ -168,8 +146,8 @@ async function onExit() {
 	await unregisterInstance();
 }
 
-async function updateConnections(conns: number) {
-	const result = await InstanceStatusModel.updateOne(
+async function updateConnections(conns: number): Promise<UpdateResult> {
+	return InstanceStatusModel.updateOne(
 		{
 			_id: ID,
 		},
@@ -179,10 +157,6 @@ async function updateConnections(conns: number) {
 			},
 		},
 	);
-
-	if (result.modifiedCount) {
-		notifyOnInstanceStatusChangedById(ID, 'updated', undefined, { 'extraInformation.conns': conns });
-	}
 }
 
 export const InstanceStatus = {