chore(federation): refactor edu handling

apps/meteor/ee/server/hooks/federation/index.ts

@@ -1,8 +1,10 @@
 import { api, FederationMatrix } from '@rocket.chat/core-services';
 import { isEditedMessage, type IMessage, type IRoom, type IUser } from '@rocket.chat/core-typings';
+import { License } from '@rocket.chat/license';
 import { MatrixBridgedRoom, Rooms } from '@rocket.chat/models';
 
 import notifications from '../../../../app/notifications/server/lib/Notifications';
+import { settings } from '../../../../app/settings/server';
 import { callbacks } from '../../../../lib/callbacks';
 import { afterLeaveRoomCallback } from '../../../../lib/callbacks/afterLeaveRoomCallback';
 import { afterRemoveFromRoomCallback } from '../../../../lib/callbacks/afterRemoveFromRoomCallback';
@@ -17,8 +19,6 @@
 callbacks.add('federation.afterCreateFederatedRoom', async (room, { owner, originalMemberList: members, options }) => {
 	if (FederationActions.shouldPerformFederationAction(room)) {
 		const federatedRoomId = options?.federatedRoomId;
-		// TODO: move this to the hooks folder
-		setupTypingEventListenerForRoom(room._id);
 
 		if (!federatedRoomId) {
 			// if room if exists, we don't want to create it again
@@ -227,23 +227,3 @@
 	callbacks.priority.HIGH,
 	'federation-matrix-after-create-direct-room',
 );
-
-// TODO: THIS IS NOT READY FOR PRODUCTION! IMPOSSIBLE TO ADD ONE LISTENER PER ROOM!
-const setupTypingEventListenerForRoom = (roomId: string): void => {
-	notifications.streamRoom.on(`${roomId}/user-activity`, (username, activity) => {
-		if (Array.isArray(activity) && (!activity.length || activity.includes('user-typing'))) {
-			void api.broadcast('user.typing', {
-				user: { username },
-				isTyping: activity.includes('user-typing'),
-				roomId,
-			});
-		}
-	});
-};
-
-export const setupInternalEDUEventListeners = async () => {
-	const federatedRooms = await Rooms.findFederatedRooms({ projection: { _id: 1 } }).toArray();
-	for (const room of federatedRooms) {
-		setupTypingEventListenerForRoom(room._id);
-	}
-};

apps/meteor/ee/server/index.ts

@@ -14,12 +14,6 @@ import './local-services/ldap/service';
 import './methods/getReadReceipts';
 import './patches';
 import './hooks/federation';
-import { License } from '@rocket.chat/license';
 
 export * from './apps/startup';
 export { registerEEBroker } from './startup';
-
-await License.onLicense('federation', async () => {
-	const { setupInternalEDUEventListeners } = await import('./hooks/federation');
-	await setupInternalEDUEventListeners();
-});

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

@@ -6,6 +6,7 @@
 
 import { settings } from '../../../app/settings/server';
 import { registerFederationRoutes } from '../api/federation';
+import { StreamerCentral } from '../../../server/modules/streamer/streamer.module';
 
 const logger = new Logger('Federation');
 
@@ -27,6 +28,14 @@
 		logger.debug('Starting federation-matrix service');
 		federationMatrixService = await FederationMatrix.create(InstanceStatus.id());
 
+		StreamerCentral.on('broadcast', (name, eventName, args) => {
+			if (name === 'notify-room' && eventName.endsWith('user-activity')) {
+				const [rid] = eventName.split('/');
+				const [user, activity] = args;
+				void federationMatrixService.notifyUserTyping(rid, user, activity.includes('user-typing'));
+			}
+		});
+
 		try {
 			api.registerService(federationMatrixService);
 			await registerFederationRoutes(federationMatrixService);

apps/meteor/server/modules/listeners/listeners.module.ts

@@ -151,20 +151,39 @@ export class ListenersModule {
 			notifications.notifyRoom(rid, 'videoconf', callId);
 		});
 
-		service.onEvent('presence.status', ({ user }) => this.handlePresence({ user }, notifications));
+		service.onEvent('presence.status', ({ user }) => {
+			const { _id, username, name, status, statusText, roles } = user;
+			if (!status || !username) {
+				return;
+			}
+
+			notifications.notifyUserInThisInstance(_id, 'userData', {
+				type: 'updated',
+				id: _id,
+				diff: {
+					status,
+					...(statusText && { statusText }),
+				},
+				unset: {},
+			});
+
+			notifications.notifyLoggedInThisInstance('user-status', [_id, username, STATUS_MAP[status], statusText, name, roles]);
+
+			if (_id) {
+				notifications.sendPresence(_id, username, STATUS_MAP[status], statusText);
+			}
+		});
 
 		service.onEvent('user.updateCustomStatus', (userStatus) => {
 			notifications.notifyLoggedInThisInstance('updateCustomUserStatus', {
 				userStatusData: userStatus,
 			});
 		});
 
-		service.onEvent('federation-matrix.user.typing', ({ isTyping, roomId, username }) => {
-			notifications.notifyRoom(roomId, 'user-activity', username, isTyping ? ['user-typing'] : []);
+		service.onEvent('user.activity', ({ isTyping, roomId, user }) => {
+			notifications.notifyRoom(roomId, 'user-activity', user, isTyping ? ['user-typing'] : []);
 		});
 
-		service.onEvent('federation-matrix.user.presence.status', ({ user }) => this.handlePresence({ user }, notifications));
-
 		service.onEvent('watch.messages', async ({ message }) => {
 			if (!message.rid) {
 				return;
@@ -491,30 +510,4 @@ export class ListenersModule {
 			notifications.streamRoomMessage.emit(roomId, acknowledgeMessage);
 		});
 	}
-
-	private handlePresence(
-		{ user }: { user: Pick<IUser, '_id' | 'username' | 'status' | 'statusText' | 'name' | 'roles'> },
-		notifications: NotificationsModule,
-	): void {
-		const { _id, username, name, status, statusText, roles } = user;
-		if (!status || !username) {
-			return;
-		}
-
-		notifications.notifyUserInThisInstance(_id, 'userData', {
-			type: 'updated',
-			id: _id,
-			diff: {
-				status,
-				...(statusText && { statusText }),
-			},
-			unset: {},
-		});
-
-		notifications.notifyLoggedInThisInstance('user-status', [_id, username, STATUS_MAP[status], statusText, name, roles]);
-
-		if (_id) {
-			notifications.sendPresence(_id, username, STATUS_MAP[status], statusText);
-		}
-	}
 }

ee/packages/federation-matrix/src/FederationMatrix.ts

@@ -114,24 +114,7 @@ export class FederationMatrix extends ServiceClass implements IFederationMatrixS
 		instance.homeserverServices = getAllServices();
 		MatrixMediaService.setHomeserverServices(instance.homeserverServices);
 		instance.buildMatrixHTTPRoutes();
-		instance.onEvent('user.typing', async ({ isTyping, roomId, user: { username } }): Promise<void> => {
-			if (!roomId || !username) {
-				return;
-			}
-			const externalRoomId = await MatrixBridgedRoom.getExternalRoomId(roomId);
-			if (!externalRoomId) {
-				return;
-			}
-			const localUser = await Users.findOneByUsername(username, { projection: { _id: 1 } });
-			if (!localUser) {
-				return;
-			}
-			const externalUserId = await MatrixBridgedUser.getExternalUserIdByLocalUserId(localUser._id);
-			if (!externalUserId) {
-				return;
-			}
-			void instance.homeserverServices.edu.sendTypingNotification(externalRoomId, externalUserId, isTyping);
-		});
+
 		instance.onEvent(
 			'presence.status',
 			async ({ user }: { user: Pick<IUser, '_id' | 'username' | 'status' | 'statusText' | 'name' | 'roles'> }): Promise<void> => {
@@ -1002,4 +985,24 @@ export class FederationMatrix extends ServiceClass implements IFederationMatrixS
 		}
 		await this.homeserverServices.room.setPowerLevelForUser(matrixRoomId, senderMatrixUserId, matrixUserId, powerLevel);
 	}
+
+	async notifyUserTyping(rid: string, user: string, isTyping: boolean) {
+		if (!rid || !user) {
+			return;
+		}
+		const externalRoomId = await MatrixBridgedRoom.getExternalRoomId(rid);
+		if (!externalRoomId) {
+			return;
+		}
+		const localUser = await Users.findOneByUsername(user, { projection: { _id: 1 } });
+		if (!localUser) {
+			return;
+		}
+		const externalUserId = await MatrixBridgedUser.getExternalUserIdByLocalUserId(localUser._id);
+		if (!externalUserId) {
+			return;
+		}
+
+		void this.homeserverServices.edu.sendTypingNotification(externalRoomId, externalUserId, isTyping);
+	}
 }

ee/packages/federation-matrix/src/events/edu.ts

@@ -23,13 +23,13 @@
 			}
 
 			const user = await Users.findOneById(matrixUser.uid, { projection: { _id: 1, username: 1 } });
 			if (!user || !user.username) {
 				logger.debug(`User not found for uid: ${matrixUser.uid}`);
 				return;
 			}
 
-			void api.broadcast('federation-matrix.user.typing', {
-				username: user.username,
+			void api.broadcast('user.activity', {
+				user: user.username,
 				isTyping: data.typing,
 				roomId: matrixRoom,
 			});
@@ -71,8 +71,9 @@
 			);
 
 			const { _id, username, statusText, roles, name } = user;
-			void api.broadcast('federation-matrix.user.presence.status', {
+			void api.broadcast('presence.status', {
 				user: { status, _id, username, statusText, roles, name },
+				previousStatus: undefined,
 			});
 			logger.debug(`Updated presence for user ${matrixUser.uid} to ${status} from Matrix federation`);
 		} catch (error) {

packages/core-services/src/events/Events.ts

@@ -149,12 +149,7 @@ export type EventSignatures = {
 		scope?: string;
 	}): void;
 	'user.updateCustomStatus'(userStatus: Omit<ICustomUserStatus, '_updatedAt'>): void;
-	'user.typing'(data: { user: Partial<IUser>; isTyping: boolean; roomId: string }): void;
-	'federation-matrix.user.typing'(data: { username: string; isTyping: boolean; roomId: string }): void;
-	'federation-matrix.user.presence.status'(data: {
-		user: Pick<IUser, '_id' | 'username' | 'status' | 'statusText' | 'name' | 'roles'>;
-		previousStatus?: UserStatus;
-	}): void;
+	'user.activity'(data: { user: string; isTyping: boolean; roomId: string }): void;
 	'user.video-conference'(data: {
 		userId: IUser['_id'];
 		action: string;