feat: presence sync for voice and video calls

Author: ricardogarim

apps/meteor/server/services/media-call/service.ts

@@ -1,4 +1,4 @@
-import { api, ServiceClassInternal, type IMediaCallService, Authorization } from '@rocket.chat/core-services';
+import { api, Presence, ServiceClassInternal, type IMediaCallService, Authorization } from '@rocket.chat/core-services';
 import type {
 	IMediaCall,
 	IUser,
@@ -7,6 +7,7 @@ import type {
 	CallHistoryItemState,
 	IExternalMediaCallHistoryItem,
 } from '@rocket.chat/core-typings';
+import { UserStatus } from '@rocket.chat/core-typings';
 import { callServer, type IMediaCallServerSettings, getSignalsForExistingCall } from '@rocket.chat/media-calls';
 import type {
 	CallFeature,
@@ -34,6 +35,8 @@ export class MediaCallService extends ServiceClassInternal implements IMediaCall
 		super();
 		callServer.emitter.on('signalRequest', ({ toUid, signal }) => this.sendSignal(toUid, signal));
 		callServer.emitter.on('callUpdated', (params) => api.broadcast('media-call.updated', params));
+		callServer.emitter.on('callActivated', ({ uids }) => this.setPresenceForUsers(uids));
+		callServer.emitter.on('callEnded', ({ uids }) => this.clearPresenceForUsers(uids));
 		callServer.emitter.on('historyUpdate', ({ callId }) => setImmediate(() => this.saveCallToHistory(callId)));
 		callServer.emitter.on('pushNotificationRequest', ({ callId, event }) => sendVoipPushNotification(callId, event));
 		this.onEvent('media-call.updated', (params) => callServer.receiveCallUpdate(params));
@@ -258,8 +261,8 @@ export class MediaCallService extends ServiceClassInternal implements IMediaCall
 		}
 	}
 
-	private getLanguageForUser(user: IUser): string {
-		return user.language || settings.get('Language') || 'en';
+	private getLanguageForUser(language?: string): string {
+		return language || settings.get('Language') || 'en';
 	}
 
 	private async sendHistoryMessage(call: IMediaCall, room: IRoom): Promise<void> {
@@ -274,7 +277,7 @@ export class MediaCallService extends ServiceClassInternal implements IMediaCall
 		const skipNotifications = state !== 'not-answered' || call.hangupReason === 'rejected';
 		const i18nKey = callStateToTranslationKey(state).i18n?.key;
 
-		const msg = i18nKey ? i18n.t(i18nKey, { lng: this.getLanguageForUser(user) }) : '';
+		const msg = i18nKey ? i18n.t(i18nKey, { lng: this.getLanguageForUser(user.language) }) : '';
 		const duration = this.getCallDuration(call);
 
 		const record = getHistoryMessagePayload(state, duration, call._id, msg);
@@ -366,6 +369,31 @@ export class MediaCallService extends ServiceClassInternal implements IMediaCall
 		};
 	}
 
+	private async setPresenceForUsers(uids: IUser['_id'][]): Promise<void> {
+		await Promise.all(
+			uids.map(async (uid) => {
+				try {
+					const user = await Users.findOneById(uid, { projection: { language: 1 } });
+					await Presence.setActiveState(uid, {
+						statusDefault: UserStatus.BUSY,
+						statusText: i18n.t('Presence_status_on_a_call', { lng: this.getLanguageForUser(user?.language) }),
+						statusSource: 'internal',
+					});
+				} catch (err) {
+					logger.error({ msg: 'Failed to set presence for user on call', uid, err });
+				}
+			}),
+		);
+	}
+
+	private async clearPresenceForUsers(uids: IUser['_id'][]): Promise<void> {
+		await Promise.all(
+			uids.map((uid) =>
+				Presence.endActiveState(uid).catch((err) => logger.error({ msg: 'Failed to clear presence for user after call', uid, err })),
+			),
+		);
+	}
+
 	private async sendSignal(toUid: IUser['_id'], signal: ServerMediaSignal): Promise<void> {
 		void api.broadcast('user.media-signal', { userId: toUid, signal });
 	}

apps/meteor/server/services/video-conference/service.ts

@@ -2,7 +2,7 @@ import { Apps } from '@rocket.chat/apps';
 import type { AppVideoConfProviderManager } from '@rocket.chat/apps/dist/server/managers/AppVideoConfProviderManager';
 import type { VideoConfData, VideoConfDataExtended } from '@rocket.chat/apps-engine/definition/videoConfProviders';
 import type { IVideoConfService, VideoConferenceJoinOptions } from '@rocket.chat/core-services';
-import { api, ServiceClassInternal, Room } from '@rocket.chat/core-services';
+import { api, Presence, ServiceClassInternal, Room } from '@rocket.chat/core-services';
 import type {
 	IDirectVideoConference,
 	ILivechatVideoConference,
@@ -25,6 +25,7 @@ import type {
 	IVoIPVideoConference,
 } from '@rocket.chat/core-typings';
 import {
+	UserStatus,
 	VideoConferenceStatus,
 	isDirectVideoConference,
 	isGroupVideoConference,
@@ -322,8 +323,8 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 			throw new Error('Invalid User');
 		}
 
-		const user = await Users.findOneById<Required<Pick<IUser, '_id' | 'username' | 'name' | 'avatarETag'>>>(userId, {
-			projection: { username: 1, name: 1, avatarETag: 1 },
+		const user = await Users.findOneById<Required<Pick<IUser, '_id' | 'username' | 'name' | 'avatarETag' | 'language'>>>(userId, {
+			projection: { username: 1, name: 1, avatarETag: 1, language: 1 },
 		});
 		if (!user) {
 			throw new Error('Invalid User');
@@ -334,6 +335,7 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 			username: user.username,
 			name: user.name,
 			avatarETag: user.avatarETag,
+			language: user.language,
 			ts: ts || new Date(),
 		});
 	}
@@ -502,6 +504,15 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 		}
 
 		await VideoConferenceModel.setDataById(call._id, { endedAt: new Date(), status: VideoConferenceStatus.ENDED });
+
+		await Promise.all(
+			call.users.map((user) =>
+				Presence.endActiveState(user._id).catch((err) =>
+					logger.error({ msg: 'Failed to clear presence for user after video conference', uid: user._id, err }),
+				),
+			),
+		);
+
 		await this.runVideoConferenceChangedEvent(call._id);
 		this.notifyVideoConfUpdate(call.rid, call._id);
 
@@ -576,6 +587,10 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 		return message._id;
 	}
 
+	private getLanguageForUser(language?: string): string {
+		return language || settings.get('Language') || 'en';
+	}
+
 	private async validateProvider(providerName: string): Promise<void> {
 		const manager = await this.getProviderManager();
 		const configured = await manager.isFullyConfigured(providerName).catch(() => false);
@@ -1067,7 +1082,14 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 
 	private async addUserToCall(
 		call: Optional<VideoConference, 'providerData'>,
-		{ _id, username, name, avatarETag, ts }: AtLeast<Required<IUser>, '_id' | 'username' | 'name' | 'avatarETag'> & { ts?: Date },
+		{
+			_id,
+			username,
+			name,
+			avatarETag,
+			language,
+			ts,
+		}: AtLeast<Required<IUser>, '_id' | 'username' | 'name' | 'avatarETag'> & { language?: string; ts?: Date },
 	): Promise<void> {
 		// If the call has a discussion, ensure the user is subscribed to it;
 		// This is done even if the user has already joined the call before, so they can be added back if they had left the discussion.
@@ -1081,6 +1103,12 @@ export class VideoConfService extends ServiceClassInternal implements IVideoConf
 
 		await VideoConferenceModel.addUserById(call._id, { _id, username, name, avatarETag, ts });
 
+		await Presence.setActiveState(_id, {
+			statusDefault: UserStatus.BUSY,
+			statusText: i18n.t('Presence_status_in_a_meeting', { lng: this.getLanguageForUser(language) }),
+			statusSource: 'internal',
+		}).catch((err) => logger.error({ msg: 'Failed to set presence for user joining video conference', uid: _id, err }));
+
 		if (call.type === 'direct') {
 			return this.updateDirectCall(call as IDirectVideoConference, _id);
 		}

apps/meteor/tests/end-to-end/apps/presence-state.ts

@@ -1,6 +1,6 @@
 import type { App } from '@rocket.chat/core-typings';
 import { expect } from 'chai';
-import { after, before, describe, it } from 'mocha';
+import { after, afterEach, before, describe, it } from 'mocha';
 
 import { getCredentials, request, credentials } from '../../data/api-data';
 import { appPresenceStateTest } from '../../data/apps/app-packages';
@@ -22,6 +22,15 @@ import { IS_EE } from '../../e2e/config/constants';
 
 	after(() => cleanupApps());
 
+	afterEach(async () => {
+		const user = await getUserByUsername(adminUsername);
+		await request
+			.post(apps(`/public/${app.id}/end-active-state`))
+			.set(credentials)
+			.send({ userId: user._id })
+			.expect(200);
+	});
+
 	describe('[setActiveState]', () => {
 		it('should set the user presence with status text and source', async () => {
 			const user = await getUserByUsername(adminUsername);
@@ -44,10 +53,20 @@ import { IS_EE } from '../../e2e/config/constants';
 	});
 
 	describe('[endActiveState]', () => {
-		it('should restore the user presence to previous state', async () => {
+		it('should restore the displaced same-priority state and reset once the queue is exhausted', async () => {
 			const user = await getUserByUsername(adminUsername);
 
-			// First set an active state
+			await request
+				.post(apps(`/public/${app.id}/set-active-state`))
+				.set(credentials)
+				.send({
+					userId: user._id,
+					statusDefault: 'busy',
+					statusText: 'In a meeting',
+					statusSource: 'internal',
+				})
+				.expect(200);
+
 			await request
 				.post(apps(`/public/${app.id}/set-active-state`))
 				.set(credentials)
@@ -59,16 +78,25 @@ import { IS_EE } from '../../e2e/config/constants';
 				})
 				.expect(200);
 
-			// Then end it
 			await request
 				.post(apps(`/public/${app.id}/end-active-state`))
 				.set(credentials)
 				.send({ userId: user._id })
 				.expect(200);
 
-			const updatedUser = await getUserByUsername(adminUsername);
-			expect(updatedUser.statusText).to.not.equal('On a call');
-			expect(updatedUser.statusSource).to.not.equal('internal');
+			const restoredUser = await getUserByUsername(adminUsername);
+			expect(restoredUser.statusText).to.equal('In a meeting');
+			expect(restoredUser.statusSource).to.equal('internal');
+
+			await request
+				.post(apps(`/public/${app.id}/end-active-state`))
+				.set(credentials)
+				.send({ userId: user._id })
+				.expect(200);
+
+			const clearedUser = await getUserByUsername(adminUsername);
+			expect(clearedUser.statusText).to.equal('');
+			expect(clearedUser.statusSource).to.be.undefined;
 		});
 	});
 });

ee/packages/media-calls/src/definition/IMediaCallServer.ts

@@ -9,6 +9,8 @@ export type VoipPushNotificationEventType = 'new' | 'answer' | 'end';
 
 export type MediaCallServerEvents = {
 	callUpdated: { callId: string; dtmf?: ClientMediaSignalBody<'dtmf'> };
+	callActivated: { callId: string; uids: IUser['_id'][] };
+	callEnded: { callId: string; uids: IUser['_id'][] };
 	signalRequest: { toUid: IUser['_id']; signal: ServerMediaSignal };
 	historyUpdate: { callId: string };
 	pushNotificationRequest: { callId: string; event: VoipPushNotificationEventType };

ee/packages/media-calls/src/server/CallDirector.ts

@@ -38,6 +38,7 @@ class MediaCallDirector {
 		if (modified) {
 			await actorAgent.onCallEnded(call._id);
 			await actorAgent.oppositeAgent?.onCallEnded(call._id);
+			getMediaCallServer().emitter.emit('callEnded', { callId: call._id, uids: call.uids });
 		}
 	}
 
@@ -55,6 +56,7 @@ class MediaCallDirector {
 
 		logger.info({ msg: 'Call was flagged as active', callId: call._id });
 		this.scheduleExpirationCheckByCallId(call._id);
+		getMediaCallServer().emitter.emit('callActivated', { callId: call._id, uids: call.uids });
 		return actorAgent.oppositeAgent?.onCallActive(call._id);
 	}
 

ee/packages/presence/src/Presence.ts

@@ -17,6 +17,7 @@ type PresenceUser = Pick<
 	IUser,
 	| '_id'
 	| 'username'
+	| 'type'
 	| 'roles'
 	| 'status'
 	| 'statusDefault'
@@ -326,11 +327,12 @@ export class Presence extends ServiceClass implements IPresence {
 	}
 
 	/**
-	 * Ends the current active claim. Restores previous if valid, otherwise
-	 * falls back to system-managed.
+	 * Ends the current active claim. Restores previous if valid, otherwise falls back to
+	 * system-managed. When `expectedSource` is given, only ends the claim if that source still
+	 * owns the display (so a caller can clear its own claim without clobbering one that took over).
 	 */
-	async endActiveState(userId: string): Promise<boolean> {
-		return this.updatePresenceAndReschedule(userId, { type: 'endActive' });
+	async endActiveState(userId: string, expectedSource?: IUser['statusSource']): Promise<boolean> {
+		return this.updatePresenceAndReschedule(userId, { type: 'endActive', ...(expectedSource && { expectedSource }) });
 	}
 
 	/**
@@ -358,6 +360,7 @@ export class Presence extends ServiceClass implements IPresence {
 				? await Users.findOneById<PresenceUser>(uidOrUser, {
 						projection: {
 							username: 1,
+							type: 1,
 							roles: 1,
 							status: 1,
 							statusDefault: 1,