add cron every minute to look expired status

Author: ricardogarim

ee/packages/presence/src/Presence.ts

@@ -2,6 +2,7 @@ import type { IPresence, IBrokerNode } from '@rocket.chat/core-services';
 import { License, ServiceClass, Settings } from '@rocket.chat/core-services';
 import type { IUser } from '@rocket.chat/core-typings';
 import { UserStatus } from '@rocket.chat/core-typings';
+import { cronJobs } from '@rocket.chat/cron';
 import { Users, UsersSessions } from '@rocket.chat/models';
 
 import { PresenceReaper } from './lib/PresenceReaper';
@@ -98,6 +99,25 @@ export class Presence extends ServiceClass implements IPresence {
 		} catch (e: unknown) {
 			// ignore
 		}
+
+		// TODO: Agenda default lockLifetime is 10 min. This job executes in ms and runs every 1 min.
+		// If an instance crashes mid-execution, expired statuses stay locked for up to 10 min.
+		// Reduce lockLifetime to 60s once @rocket.chat/agenda exposes the option in its typed API.
+		await cronJobs.add('presence-status-expiration', '* * * * *', () => this.processExpiredStatuses());
+	}
+
+	private async processExpiredStatuses(batchSize = 100): Promise<void> {
+		const expiredUsers = await Users.findExpiredStatuses(batchSize).toArray();
+
+		if (expiredUsers.length === 0) {
+			return;
+		}
+
+		await Promise.allSettled(expiredUsers.map(({ _id }) => this.endActiveState(_id)));
+
+		if (expiredUsers.length === batchSize) {
+			return this.processExpiredStatuses(batchSize);
+		}
 	}
 
 	private async handleReaperUpdates(userIds: string[]): Promise<void> {

packages/model-typings/src/models/IUsersModel.ts

@@ -172,6 +172,8 @@ export interface IUsersModel extends IBaseModel<IUser> {
 
 	updateStatusText(_id: IUser['_id'], statusText: string, options?: UpdateOptions): Promise<UpdateResult>;
 
+	findExpiredStatuses(limit: number): FindCursor<Pick<IUser, '_id'>>;
+
 	findOneForPresenceEngine(userId: IUser['_id']): Promise<IUser | null>;
 
 	updatePresenceAndStatus(userId: IUser['_id'], values: Record<string, unknown>, clear?: string[]): Promise<IUser | null>;
@@ -242,8 +244,6 @@ export interface IUsersModel extends IBaseModel<IUser> {
 		}: { statusDefault?: UserStatus; status: UserStatus; statusConnection: UserStatus; statusText?: string },
 	): Promise<UpdateResult>;
 
-	updateStatusAndStatusDefault(userId: string, status: UserStatus, statusDefault: UserStatus): Promise<UpdateResult>;
-
 	setFederationAvatarUrlById(userId: IUser['_id'], federationAvatarUrl: string): Promise<UpdateResult>;
 
 	setFederationAvatarUrlById(userId: IUser['_id'], federationAvatarUrl: string): Promise<UpdateResult>;

packages/models/src/models/Users.ts

@@ -1097,6 +1097,10 @@ export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IU
 		return this.updateOne({ _id }, update, { session: options?.session });
 	}
 
+	findExpiredStatuses(limit: number) {
+		return this.find({ statusExpiresAt: { $lt: new Date() } }, { projection: { _id: 1 }, limit });
+	}
+
 	findOneForPresenceEngine(userId: IUser['_id']) {
 		return this.findOneById(userId, {
 			projection: {