fix(presence): update connections on heartbeat and remove them when stale

.changeset/spicy-nails-design.md

@@ -0,0 +1,10 @@
+---
+"@rocket.chat/meteor": patch
+"@rocket.chat/core-services": patch
+"@rocket.chat/model-typings": patch
+"@rocket.chat/models": patch
+"@rocket.chat/ddp-streamer": patch
+"@rocket.chat/presence": patch
+---
+
+Fixes user status inaccuracy by refreshing active connections and filtering out the stale ones.

apps/meteor/client/meteor/overrides/ddpOverREST.ts

@@ -6,11 +6,13 @@ import { getUserId } from '../../lib/user';
 
 const bypassMethods: string[] = ['setUserStatus', 'logout'];
 
-const shouldBypass = ({ msg, method, params }: Meteor.IDDPMessage): boolean => {
-	if (msg !== 'method') {
+const shouldBypass = (message: Meteor.IDDPMessage): message is Exclude<Meteor.IDDPMessage, Meteor.IDDPMethodMessage> => {
+	if (message.msg !== 'method') {
 		return true;
 	}
 
+	const { method, params } = message;
+
 	if (method === 'login' && params[0]?.resume) {
 		return true;
 	}
@@ -29,6 +31,9 @@ const shouldBypass = ({ msg, method, params }: Meteor.IDDPMessage): boolean => {
 const withDDPOverREST = (_send: (this: Meteor.IMeteorConnection, message: Meteor.IDDPMessage, ...args: unknown[]) => void) => {
 	return function _sendOverREST(this: Meteor.IMeteorConnection, message: Meteor.IDDPMessage, ...args: unknown[]): void {
 		if (shouldBypass(message)) {
+			if (message.msg === 'ping') {
+				sdk.call('UserPresence:ping');
+			}
 			return _send.call(this, message, ...args);
 		}
 

apps/meteor/definition/externals/meteor/meteor.d.ts

@@ -47,13 +47,19 @@ declare module 'meteor/meteor' {
 			twoFactorChecked: boolean | undefined;
 		}
 
-		interface IDDPMessage {
+		interface IDDPPingMessage {
+			msg: 'ping';
+		}
+
+		interface IDDPMethodMessage {
 			msg: 'method';
 			method: string;
 			params: EJSON[];
 			id: string;
 		}
 
+		type IDDPMessage = IDDPPingMessage | IDDPMethodMessage;
+
 		interface IDDPUpdatedMessage {
 			msg: 'updated';
 			methods: string[];

apps/meteor/server/methods/userPresence.ts

@@ -9,6 +9,7 @@ declare module '@rocket.chat/ddp-client' {
 		'UserPresence:setDefaultStatus'(status: UserStatus): boolean | undefined;
 		'UserPresence:online'(): boolean | undefined;
 		'UserPresence:away'(): boolean | undefined;
+		'UserPresence:ping'(): boolean | undefined;
 	}
 }
 
@@ -25,13 +26,20 @@ Meteor.methods<ServerMethods>({
 		if (!userId || !connection) {
 			return;
 		}
-		return Presence.setConnectionStatus(userId, UserStatus.ONLINE, connection.id);
+		return Presence.setConnectionStatus(userId, connection.id, UserStatus.ONLINE);
 	},
 	'UserPresence:away'() {
 		const { userId, connection } = this;
 		if (!userId || !connection) {
 			return;
 		}
-		return Presence.setConnectionStatus(userId, UserStatus.AWAY, connection.id);
+		return Presence.setConnectionStatus(userId, connection.id, UserStatus.AWAY);
+	},
+	'UserPresence:ping'() {
+		const { connection, userId } = this;
+		if (!userId || !connection) {
+			return;
+		}
+		return Presence.setConnectionStatus(userId, connection.id);
 	},
 });

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

@@ -215,6 +215,14 @@ export class DDPStreamer extends ServiceClass {
 		server.on(DDP_EVENTS.CONNECTED, ({ connection }) => {
 			this.api?.broadcast('socket.connected', connection);
 		});
+
+		server.on(DDP_EVENTS.PING, (client: Client): void => {
+			const { connection, userId } = client;
+			if (!userId) {
+				return;
+			}
+			Presence.setConnectionStatus(userId, connection.id);
+		});
 	}
 
 	async started(): Promise<void> {

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

@@ -121,14 +121,21 @@ server.methods({
 		if (!userId) {
 			return;
 		}
-		return Presence.setConnectionStatus(userId, UserStatus.ONLINE, session);
+		return Presence.setConnectionStatus(userId, session, UserStatus.ONLINE);
 	},
 	'UserPresence:away'() {
 		const { userId, session } = this;
 		if (!userId) {
 			return;
 		}
-		return Presence.setConnectionStatus(userId, UserStatus.AWAY, session);
+		return Presence.setConnectionStatus(userId, session, UserStatus.AWAY);
+	},
+	'UserPresence:ping'() {
+		const { userId, session } = this;
+		if (!userId) {
+			return;
+		}
+		return Presence.setConnectionStatus(userId, session);
 	},
 	'setUserStatus'(status, statusText) {
 		const { userId } = this;

ee/packages/presence/src/Presence.ts

@@ -4,7 +4,7 @@ import type { IUser } from '@rocket.chat/core-typings';
 import { UserStatus } from '@rocket.chat/core-typings';
 import { Settings, Users, UsersSessions } from '@rocket.chat/models';
 
-import { processPresenceAndStatus } from './lib/processConnectionStatus';
+import { processPresenceAndStatus } from './processPresenceAndStatus';
 
 const MAX_CONNECTIONS = 200;
 
@@ -203,7 +203,7 @@ export class Presence extends ServiceClass implements IPresence {
 		return !!result.modifiedCount;
 	}
 
-	async setConnectionStatus(uid: string, status: UserStatus, session: string): Promise<boolean> {
+	async setConnectionStatus(uid: string, session: string, status?: UserStatus): Promise<boolean> {
 		const result = await UsersSessions.updateConnectionStatusById(uid, session, status);
 
 		await this.updateUserPresence(uid);

ee/packages/presence/src/lib/processConnectionStatus.ts

@@ -1,4 +1,3 @@
-import type { IUserSessionConnection } from '@rocket.chat/core-typings';
 import { UserStatus } from '@rocket.chat/core-typings';
 
 /**
@@ -13,35 +12,3 @@ export const processConnectionStatus = (current: UserStatus, status: UserStatus)
 	}
 	return current;
 };
-
-/**
- * Defines user's status based on presence and connection status
- */
-export const processStatus = (statusConnection: UserStatus, statusDefault: UserStatus): UserStatus => {
-	if (statusConnection === UserStatus.OFFLINE) {
-		return statusConnection;
-	}
-
-	if (statusDefault === UserStatus.ONLINE) {
-		return statusConnection;
-	}
-
-	return statusDefault;
-};
-
-/**
- * Defines user's status and connection status based on user's connections and default status
- */
-export const processPresenceAndStatus = (
-	userSessions: IUserSessionConnection[] = [],
-	statusDefault = UserStatus.ONLINE,
-): { status: UserStatus; statusConnection: UserStatus } => {
-	const statusConnection = userSessions.map((s) => s.status).reduce(processConnectionStatus, UserStatus.OFFLINE);
-
-	const status = processStatus(statusConnection, statusDefault);
-
-	return {
-		status,
-		statusConnection,
-	};
-};

ee/packages/presence/src/lib/processStatus.ts

@@ -0,0 +1,16 @@
+import { UserStatus } from '@rocket.chat/core-typings';
+
+/**
+ * Defines user's status based on presence and connection status
+ */
+export const processStatus = (statusConnection: UserStatus, statusDefault: UserStatus): UserStatus => {
+	if (statusConnection === UserStatus.OFFLINE) {
+		return statusConnection;
+	}
+
+	if (statusDefault === UserStatus.ONLINE) {
+		return statusConnection;
+	}
+
+	return statusDefault;
+};

ee/packages/presence/src/processPresenceAndStatus.ts

@@ -0,0 +1,31 @@
+import type { IUserSessionConnection } from '@rocket.chat/core-typings';
+import { UserStatus } from '@rocket.chat/core-typings';
+
+import { processConnectionStatus } from './lib/processConnectionStatus';
+import { processStatus } from './lib/processStatus';
+
+const isAtMostFiveMinutesAgo = (userSession: IUserSessionConnection): boolean => {
+	const now = Date.now();
+	const diff = now - userSession._updatedAt.getTime();
+	return diff <= 300_000; // 5 minutes in milliseconds
+};
+
+/**
+ * Defines user's status and connection status based on user's connections and default status
+ */
+export const processPresenceAndStatus = (
+	userSessions: IUserSessionConnection[] = [],
+	statusDefault = UserStatus.ONLINE,
+): { status: UserStatus; statusConnection: UserStatus } => {
+	const statusConnection = userSessions
+		.filter(isAtMostFiveMinutesAgo)
+		.map((s) => s.status)
+		.reduce(processConnectionStatus, UserStatus.OFFLINE);
+
+	const status = processStatus(statusConnection, statusDefault);
+
+	return {
+		status,
+		statusConnection,
+	};
+};