New: Use database change streams when available

app/api/server/v1/settings.js

@@ -6,7 +6,7 @@ import _ from 'underscore';
 import { Settings } from '../../../models/server';
 import { hasPermission } from '../../../authorization';
 import { API } from '../api';
-import { SettingsEvents } from '../../../settings/server';
+import { SettingsEvents, settings } from '../../../settings/server';
 
 const fetchSettings = (query, sort, offset, count, fields) => {
 	const settings = Settings.find(query, {
@@ -146,6 +146,10 @@ API.v1.addRoute('settings/:_id', { authRequired: true }, {
 				value: Match.Any,
 			});
 			if (Settings.updateValueNotHiddenById(this.urlParams._id, this.bodyParams.value)) {
+				settings.storeSettingValue({
+					_id: this.urlParams._id,
+					value: this.bodyParams.value,
+				});
 				return API.v1.success();
 			}
 

app/assets/server/assets.js

@@ -7,8 +7,7 @@ import _ from 'underscore';
 import sizeOf from 'image-size';
 import sharp from 'sharp';
 
-import { settings } from '../../settings';
-import { Settings } from '../../models';
+import { settings } from '../../settings/server';
 import { getURL } from '../../utils/lib/getURL';
 import { mime } from '../../utils/lib/mimeTypes';
 import { hasPermission } from '../../authorization';
@@ -354,19 +353,7 @@ for (const key of Object.keys(assets)) {
 	addAssetToSetting(key, value);
 }
 
-Settings.find().observe({
-	added(record) {
-		return RocketChatAssets.processAsset(record._id, record.value);
-	},
-
-	changed(record) {
-		return RocketChatAssets.processAsset(record._id, record.value);
-	},
-
-	removed(record) {
-		return RocketChatAssets.processAsset(record._id, undefined);
-	},
-});
+settings.get(/^Assets_/, (key, value) => RocketChatAssets.processAsset(key, value));
 
 Meteor.startup(function() {
 	return Meteor.setTimeout(function() {

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

@@ -12,7 +12,7 @@ Permissions.on('change', ({ clientAction, id, data, diff }) => {
 	switch (clientAction) {
 		case 'updated':
 		case 'inserted':
-			data = data || Permissions.findOneById(id);
+			data = data ?? Permissions.findOneById(id);
 			break;
 
 		case 'removed':

app/dolphin/lib/common.js

@@ -5,7 +5,6 @@ import { ServiceConfiguration } from 'meteor/service-configuration';
 import { settings } from '../../settings';
 import { CustomOAuth } from '../../custom-oauth';
 import { callbacks } from '../../callbacks';
-import { Settings } from '../../models';
 
 const config = {
 	serverURL: '',
@@ -31,15 +30,9 @@ function DolphinOnCreateUser(options, user) {
 
 if (Meteor.isServer) {
 	Meteor.startup(() =>
-		Settings.find({ _id: 'Accounts_OAuth_Dolphin_URL' }).observe({
-			added() {
-				config.serverURL = settings.get('Accounts_OAuth_Dolphin_URL');
-				return Dolphin.configure(config);
-			},
-			changed() {
-				config.serverURL = settings.get('Accounts_OAuth_Dolphin_URL');
-				return Dolphin.configure(config);
-			},
+		settings.get('Accounts_OAuth_Dolphin_URL', (key, value) => {
+			config.serverURL = value;
+			return Dolphin.configure(config);
 		}),
 	);
 

app/integrations/server/lib/triggerHandler.js

@@ -22,19 +22,26 @@ integrations.triggerHandler = new class RocketChatIntegrationHandler {
 		this.compiledScripts = {};
 		this.triggers = {};
 
-		Models.Integrations.find({ type: 'webhook-outgoing' }).observe({
-			added: (record) => {
-				this.addIntegration(record);
-			},
-
-			changed: (record) => {
-				this.removeIntegration(record);
-				this.addIntegration(record);
-			},
+		Models.Integrations.find({ type: 'webhook-outgoing' }).fetch().forEach((data) => this.addIntegration(data));
 
-			removed: (record) => {
-				this.removeIntegration(record);
-			},
+		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;
+			}
 		});
 	}
 

app/lib/server/startup/userDataStream.js

@@ -1,10 +1,73 @@
+import { MongoInternals } from 'meteor/mongo';
+
 import { Users } from '../../../models/server';
 import { Notifications } from '../../../notifications/server';
 
+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();
+
+	// 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);
+			}
+		}
+		return {
+			stop() {
+				if (cbs) {
+					cbs.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);
+				});
+			}
+		}
+	};
+}
+
 Users.on('change', ({ clientAction, id, data, diff }) => {
 	switch (clientAction) {
 		case 'updated':
 			Notifications.notifyUserInThisInstance(id, 'userData', { diff, type: clientAction });
+
+			if (disableOplog) {
+				processOnChange(diff, id);
+			}
+
 			break;
 		case 'inserted':
 			Notifications.notifyUserInThisInstance(id, 'userData', { data, type: clientAction });

app/metrics/server/index.js

@@ -1,6 +1,7 @@
 import { metrics } from './lib/metrics';
 import StatsTracker from './lib/statsTracker';
 
+import './lib/collectMetrics';
 import './callbacksMetrics';
 
 export {

app/metrics/server/lib/collectMetrics.js

@@ -0,0 +1,177 @@
+import http from 'http';
+
+import client from 'prom-client';
+import connect from 'connect';
+import _ from 'underscore';
+import gcStats from 'prometheus-gc-stats';
+import { Meteor } from 'meteor/meteor';
+import { Facts } from 'meteor/facts-base';
+
+import { Info, getOplogInfo } from '../../../utils/server';
+import { Migrations } from '../../../migrations';
+import { settings } from '../../../settings';
+import { Statistics } from '../../../models';
+import { metrics } from './metrics';
+
+Facts.incrementServerFact = function(pkg, fact, increment) {
+	metrics.meteorFacts.inc({ pkg, fact }, increment);
+};
+
+const setPrometheusData = async () => {
+	metrics.info.set({
+		version: Info.version,
+		unique_id: settings.get('uniqueID'),
+		site_url: settings.get('Site_Url'),
+	}, 1);
+
+	const sessions = Array.from(Meteor.server.sessions.values());
+	const authenticatedSessions = sessions.filter((s) => s.userId);
+	metrics.ddpSessions.set(Meteor.server.sessions.size);
+	metrics.ddpAuthenticatedSessions.set(authenticatedSessions.length);
+	metrics.ddpConnectedUsers.set(_.unique(authenticatedSessions.map((s) => s.userId)).length);
+
+	const statistics = Statistics.findLast();
+	if (!statistics) {
+		return;
+	}
+
+	metrics.version.set({ version: statistics.version }, 1);
+	metrics.migration.set(Migrations._getControl().version);
+	metrics.instanceCount.set(statistics.instanceCount);
+	metrics.oplogEnabled.set({ enabled: statistics.oplogEnabled }, 1);
+
+	// User statistics
+	metrics.totalUsers.set(statistics.totalUsers);
+	metrics.activeUsers.set(statistics.activeUsers);
+	metrics.nonActiveUsers.set(statistics.nonActiveUsers);
+	metrics.onlineUsers.set(statistics.onlineUsers);
+	metrics.awayUsers.set(statistics.awayUsers);
+	metrics.offlineUsers.set(statistics.offlineUsers);
+
+	// Room statistics
+	metrics.totalRooms.set(statistics.totalRooms);
+	metrics.totalChannels.set(statistics.totalChannels);
+	metrics.totalPrivateGroups.set(statistics.totalPrivateGroups);
+	metrics.totalDirect.set(statistics.totalDirect);
+	metrics.totalLivechat.set(statistics.totalLivechat);
+
+	// Message statistics
+	metrics.totalMessages.set(statistics.totalMessages);
+	metrics.totalChannelMessages.set(statistics.totalChannelMessages);
+	metrics.totalPrivateGroupMessages.set(statistics.totalPrivateGroupMessages);
+	metrics.totalDirectMessages.set(statistics.totalDirectMessages);
+	metrics.totalLivechatMessages.set(statistics.totalLivechatMessages);
+
+	const oplogQueue = getOplogInfo().mongo._oplogHandle?._entryQueue?.length || 0;
+	metrics.oplogQueue.set(oplogQueue);
+
+	metrics.pushQueue.set(statistics.pushQueue || 0);
+};
+
+const app = connect();
+
+// const compression = require('compression');
+// app.use(compression());
+
+app.use('/metrics', (req, res) => {
+	res.setHeader('Content-Type', 'text/plain');
+	const data = client.register.metrics();
+
+	metrics.metricsRequests.inc();
+	metrics.metricsSize.set(data.length);
+
+	res.end(data);
+});
+
+app.use('/', (req, res) => {
+	const html = `<html>
+		<head>
+			<title>Rocket.Chat Prometheus Exporter</title>
+		</head>
+		<body>
+			<h1>Rocket.Chat Prometheus Exporter</h1>
+			<p><a href="/metrics">Metrics</a></p>
+		</body>
+	</html>`;
+
+	res.write(html);
+	res.end();
+});
+
+const server = http.createServer(app);
+
+let timer;
+let resetTimer;
+let defaultMetricsInitiated = false;
+let gcStatsInitiated = false;
+const was = {
+	enabled: false,
+	port: 9458,
+	resetInterval: 0,
+	collectGC: false,
+};
+const updatePrometheusConfig = async () => {
+	const is = {
+		port: process.env.PROMETHEUS_PORT || settings.get('Prometheus_Port'),
+		enabled: settings.get('Prometheus_Enabled'),
+		resetInterval: settings.get('Prometheus_Reset_Interval'),
+		collectGC: settings.get('Prometheus_Garbage_Collector'),
+	};
+
+	if (Object.values(is).some((s) => s == null)) {
+		return;
+	}
+
+	if (Object.entries(is).every(([k, v]) => v === was[k])) {
+		return;
+	}
+
+	if (!is.enabled) {
+		if (was.enabled) {
+			console.log('Disabling Prometheus');
+			server.close();
+			Meteor.clearInterval(timer);
+		}
+		Object.assign(was, is);
+		return;
+	}
+
+	console.log('Configuring Prometheus', is);
+
+	if (!was.enabled) {
+		server.listen({
+			port: is.port,
+			host: process.env.BIND_IP || '0.0.0.0',
+		});
+
+		timer = Meteor.setInterval(setPrometheusData, 5000);
+	}
+
+	Meteor.clearInterval(resetTimer);
+	if (is.resetInterval) {
+		resetTimer = Meteor.setInterval(() => {
+			client.register.getMetricsAsArray().forEach((metric) => { metric.hashMap = {}; });
+		}, is.resetInterval);
+	}
+
+	// Prevent exceptions on calling those methods twice since
+	// it's not possible to stop them to be able to restart
+	try {
+		if (defaultMetricsInitiated === false) {
+			defaultMetricsInitiated = true;
+			client.collectDefaultMetrics();
+		}
+		if (is.collectGC && gcStatsInitiated === false) {
+			gcStatsInitiated = true;
+			gcStats()();
+		}
+	} catch (error) {
+		console.error(error);
+	}
+
+	Object.assign(was, is);
+};
+
+Meteor.startup(async () => {
+	settings.get(/^Prometheus_.+/, updatePrometheusConfig);
+});