chore: makes instance-status works along with Moleculer lifecyle

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

@@ -1,7 +1,7 @@
 import os from 'os';
 
 import { License, ServiceClassInternal } from '@rocket.chat/core-services';
-import { InstanceStatus } from '@rocket.chat/instance-status';
+import { InstanceStatus, defaultPingInterval, indexExpire } from '@rocket.chat/instance-status';
 import { InstanceStatus as InstanceStatusRaw } from '@rocket.chat/models';
 import EJSON from 'ejson';
 import type { BrokerNode } from 'moleculer';
@@ -33,37 +33,13 @@ export class InstanceService extends ServiceClassInternal implements IInstanceSe
 
 	private transporter: Transporters.TCP | Transporters.NATS;
 
-	private isTransporterTCP = true;
-
 	private broker: ServiceBroker;
 
 	private troubleshootDisableInstanceBroadcast = false;
 
 	constructor() {
 		super();
 
-		const tx = getTransporter({ transporter: process.env.TRANSPORTER, port: process.env.TCP_PORT, extra: process.env.TRANSPORTER_EXTRA });
-		if (typeof tx === 'string') {
-			this.transporter = new Transporters.NATS({ url: tx });
-			this.isTransporterTCP = false;
-		} else {
-			this.transporter = new Transporters.TCP(tx);
-		}
-
-		if (this.isTransporterTCP) {
-			this.onEvent('watch.instanceStatus', async ({ clientAction, data }): Promise<void> => {
-				if (clientAction === 'removed') {
-					(this.broker.transit?.tx as any).nodes.disconnected(data?._id, false);
-					(this.broker.transit?.tx as any).nodes.nodes.delete(data?._id);
-					return;
-				}
-
-				if (clientAction === 'inserted' && data?.extraInformation?.tcpPort) {
-					this.connectNode(data);
-				}
-			});
-		}
-
 		this.onEvent('license.module', async ({ module, valid }) => {
 			if (module === 'scalability' && valid) {
 				await this.startBroadcast();
@@ -93,17 +69,28 @@ export class InstanceService extends ServiceClassInternal implements IInstanceSe
 	}
 
 	async created() {
+		const transporter = getTransporter({
+			transporter: process.env.TRANSPORTER,
+			port: process.env.TCP_PORT,
+			extra: process.env.TRANSPORTER_EXTRA,
+		});
+
+		const activeInstances = InstanceStatusRaw.getActiveInstancesAddress();
+
+		this.transporter =
+			typeof transporter !== 'string'
+				? new Transporters.TCP({ ...transporter, urls: activeInstances })
+				: new Transporters.NATS({ url: transporter });
+
 		this.broker = new ServiceBroker({
 			nodeID: InstanceStatus.id(),
 			transporter: this.transporter,
 			serializer: new EJSONSerializer(),
+			heartbeatInterval: defaultPingInterval,
+			heartbeatTimeout: indexExpire,
 			...getLogger(process.env),
 		});
 
-		if ((this.broker.transit?.tx as any)?.nodes?.localNode) {
-			(this.broker.transit?.tx as any).nodes.localNode.ipList = [hostIP];
-		}
-
 		this.broker.createService({
 			name: 'matrix',
 			events: {
@@ -176,31 +163,6 @@ export class InstanceService extends ServiceClassInternal implements IInstanceSe
 		this.broadcastStarted = true;
 
 		StreamerCentral.on('broadcast', this.sendBroadcast.bind(this));
-
-		if (this.isTransporterTCP) {
-			await InstanceStatusRaw.find(
-				{
-					'extraInformation.tcpPort': {
-						$exists: true,
-					},
-				},
-				{
-					sort: {
-						_createdAt: -1,
-					},
-				},
-			).forEach(this.connectNode.bind(this));
-		}
-	}
-
-	private connectNode(record: any) {
-		if (record._id === InstanceStatus.id()) {
-			return;
-		}
-
-		const { host, tcpPort } = record.extraInformation;
-
-		(this.broker?.transit?.tx as any).addOfflineNode(record._id, host, tcpPort);
 	}
 
 	private sendBroadcast(streamName: string, eventName: string, args: unknown[]) {

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

@@ -1,12 +1,13 @@
 import { Presence } from '@rocket.chat/core-services';
 import { InstanceStatus } from '@rocket.chat/instance-status';
+import { InstanceStatus as InstanceStatusModel } from '@rocket.chat/models';
 import { Accounts } from 'meteor/accounts-base';
 import { Meteor } from 'meteor/meteor';
 import { throttle } from 'underscore';
 
 // update connections count every 30 seconds
 const updateConns = throttle(function _updateConns() {
-	void InstanceStatus.updateConnections(Meteor.server.sessions.size);
+	void InstanceStatusModel.updateConnections(InstanceStatus.id(), Meteor.server.sessions.size);
 }, 30000);
 
 Meteor.startup(() => {

apps/meteor/server/database/watchCollections.ts

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

apps/meteor/server/models/raw/InstanceStatus.ts

@@ -1,6 +1,6 @@
 import type { IInstanceStatus } from '@rocket.chat/core-typings';
 import type { IInstanceStatusModel } from '@rocket.chat/model-typings';
-import type { Db } from 'mongodb';
+import type { Db, ModifyResult, UpdateResult, DeleteResult } from 'mongodb';
 
 import { BaseRaw } from './BaseRaw';
 
@@ -17,4 +17,49 @@ export class InstanceStatusRaw extends BaseRaw<IInstanceStatus> implements IInst
 	async getActiveInstanceCount(): Promise<number> {
 		return this.col.countDocuments({ _updatedAt: { $gt: new Date(Date.now() - process.uptime() * 1000 - 2000) } });
 	}
+
+	async getActiveInstancesAddress(): Promise<string[]> {
+		const instances = await this.find({}, { projection: { _id: 1, extraInformation: { host: 1, tcpPort: 1 } } }).toArray();
+		return instances.map((instance) => `${instance.extraInformation.host}:${instance.extraInformation.tcpPort}/${instance._id}`);
+	}
+
+	async removeInstanceById(_id: IInstanceStatus['_id']): Promise<DeleteResult> {
+		return this.deleteOne({ _id });
+	}
+
+	async setDocumentHeartbeat(documentId: string): Promise<UpdateResult> {
+		return this.updateOne({ _id: documentId }, { $currentDate: { _updatedAt: true } });
+	}
+
+	async upsertInstance(instance: Partial<IInstanceStatus>): Promise<ModifyResult<IInstanceStatus>> {
+		return this.findOneAndUpdate(
+			{
+				_id: instance._id,
+			},
+			{
+				$set: instance,
+				$currentDate: {
+					_createdAt: true,
+					_updatedAt: true,
+				},
+			},
+			{
+				upsert: true,
+				returnDocument: 'after',
+			},
+		);
+	}
+
+	async updateConnections(_id: IInstanceStatus['_id'], conns: number) {
+		return this.updateOne(
+			{
+				_id,
+			},
+			{
+				$set: {
+					'extraInformation.conns': conns,
+				},
+			},
+		);
+	}
 }

apps/meteor/server/startup/watchDb.ts

@@ -1,4 +1,4 @@
-import { api } from '@rocket.chat/core-services';
+import { api, dbWatchersDisabled } from '@rocket.chat/core-services';
 import { Logger } from '@rocket.chat/logger';
 import { MongoInternals } from 'meteor/mongo';
 
@@ -19,12 +19,17 @@ watcher.watch().catch((err: Error) => {
 	process.exit(1);
 });
 
-setInterval(function _checkDatabaseWatcher() {
-	if (watcher.isLastDocDelayed()) {
-		SystemLogger.error('No real time data received recently');
-	}
-}, 20000);
+if (!dbWatchersDisabled) {
+	setInterval(function _checkDatabaseWatcher() {
+		if (watcher.isLastDocDelayed()) {
+			SystemLogger.error('No real time data received recently');
+		}
+	}, 20000);
+}
 
 export function isLastDocDelayed(): boolean {
+	if (dbWatchersDisabled) {
+		return true;
+	}
 	return watcher.isLastDocDelayed();
 }

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

@@ -2,7 +2,7 @@ import crypto from 'crypto';
 
 import { MeteorService, Presence, ServiceClass } from '@rocket.chat/core-services';
 import { InstanceStatus } from '@rocket.chat/instance-status';
-import { Users } from '@rocket.chat/models';
+import { Users, InstanceStatus as InstanceStatusModel } from '@rocket.chat/models';
 import polka from 'polka';
 import { throttle } from 'underscore';
 import WebSocket from 'ws';
@@ -66,7 +66,7 @@ export class DDPStreamer extends ServiceClass {
 
 	// update connections count every 30 seconds
 	updateConnections = throttle(() => {
-		InstanceStatus.updateConnections(this.wss?.clients.size ?? 0);
+		InstanceStatusModel.updateConnections(InstanceStatus.id(), this.wss?.clients.size ?? 0);
 	}, 30000);
 
 	async created(): Promise<void> {