regression: federated messages out of order

apps/meteor/app/api/server/v1/chat.ts

@@ -434,7 +434,7 @@ API.v1.addRoute(
 			}
 
 			const sent = await applyAirGappedRestrictionsValidation(() =>
-				executeSendMessage(this.userId, this.bodyParams.message as Pick<IMessage, 'rid'>, this.bodyParams.previewUrls),
+				executeSendMessage(this.userId, this.bodyParams.message as Pick<IMessage, 'rid'>, { previewUrls: this.bodyParams.previewUrls }),
 			);
 			const [message] = await normalizeMessagesForUser([sent], this.userId);
 

apps/meteor/app/lib/server/methods/sendMessage.ts

@@ -19,7 +19,21 @@ import { settings } from '../../../settings/server';
 import { sendMessage } from '../functions/sendMessage';
 import { RateLimiter } from '../lib';
 
-export async function executeSendMessage(uid: IUser['_id'], message: AtLeast<IMessage, 'rid'>, previewUrls?: string[]) {
+/**
+ *
+ * @param uid
+ * @param message
+ * @param extraInfo
+ *   - ts: The timestamp of the message. the message object already has a ts, but this value is validated and only a window of 10 seconds is allowed to be used. this value overrides the message.ts value without validation.
+ *
+ *
+ * @returns
+ */
+export async function executeSendMessage(
+	uid: IUser['_id'],
+	message: AtLeast<IMessage, 'rid'>,
+	extraInfo?: { ts?: Date; previewUrls?: string[] },
+) {
 	if (message.tshow && !message.tmid) {
 		throw new Meteor.Error('invalid-params', 'tshow provided but missing tmid', {
 			method: 'sendMessage',
@@ -32,19 +46,26 @@ export async function executeSendMessage(uid: IUser['_id'], message: AtLeast<IMe
 		});
 	}
 
-	if (message.ts) {
+	const now = new Date();
+
+	message.ts ??= now;
+
+	if (extraInfo?.ts) {
+		message.ts = extraInfo.ts;
+	}
+
+	if (![now, extraInfo?.ts].filter(Boolean).includes(message.ts)) {
 		const tsDiff = Math.abs(moment(message.ts).diff(Date.now()));
 		if (tsDiff > 60000) {
 			throw new Meteor.Error('error-message-ts-out-of-sync', 'Message timestamp is out of sync', {
 				method: 'sendMessage',
 				message_ts: message.ts,
 				server_ts: new Date().getTime(),
 			});
-		} else if (tsDiff > 10000) {
+		}
+		if (tsDiff > 10000) {
 			message.ts = new Date();
 		}
-	} else {
-		message.ts = new Date();
 	}
 
 	if (message.msg) {
@@ -90,7 +111,7 @@ export async function executeSendMessage(uid: IUser['_id'], message: AtLeast<IMe
 		}
 
 		metrics.messagesSent.inc(); // TODO This line needs to be moved to it's proper place. See the comments on: https://github.com/RocketChat/Rocket.Chat/pull/5736
-		return await sendMessage(user, message, room, false, previewUrls);
+		return await sendMessage(user, message, room, false, extraInfo?.previewUrls);
 	} catch (err: any) {
 		SystemLogger.error({ msg: 'Error sending message:', err });
 
@@ -147,7 +168,7 @@ Meteor.methods<ServerMethods>({
 		}
 
 		try {
-			return await applyAirGappedRestrictionsValidation(() => executeSendMessage(uid, message, previewUrls));
+			return await applyAirGappedRestrictionsValidation(() => executeSendMessage(uid, message, { previewUrls }));
 		} catch (error: any) {
 			if (['error-not-allowed', 'restricted-workspace'].includes(error.error || error.message)) {
 				throw new Meteor.Error(error.error || error.message, error.reason, {

apps/meteor/server/services/messages/service.ts

@@ -95,6 +95,7 @@ export class MessageService extends ServiceClassInternal implements IMessageServ
 		files,
 		attachments,
 		thread,
+		ts,
 	}: {
 		fromId: string;
 		rid: string;
@@ -108,23 +109,28 @@ export class MessageService extends ServiceClassInternal implements IMessageServ
 		files?: IMessage['files'];
 		attachments?: IMessage['attachments'];
 		thread?: { tmid: string; tshow: boolean };
+		ts: Date;
 	}): Promise<IMessage> {
-		return executeSendMessage(fromId, {
-			rid,
-			msg,
-			...thread,
-			federation: {
-				eventId: federation_event_id,
-				version: 1,
+		return executeSendMessage(
+			fromId,
+			{
+				rid,
+				msg,
+				...thread,
+				federation: {
+					eventId: federation_event_id,
+					version: 1,
+				},
+				...(file && { file }),
+				...(files && { files }),
+				...(attachments && { attachments }),
+				...(e2e_content && {
+					t: 'e2e',
+					content: e2e_content,
+				}),
 			},
-			...(file && { file }),
-			...(files && { files }),
-			...(attachments && { attachments }),
-			...(e2e_content && {
-				t: 'e2e',
-				content: e2e_content,
-			}),
-		});
+			{ ts },
+		);
 	}
 
 	async sendMessageWithValidation(user: IUser, message: Partial<IMessage>, room: Partial<IRoom>, upsert = false): Promise<IMessage> {

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

@@ -225,6 +225,7 @@ export function message() {
 					msg: formatted,
 					federation_event_id: eventId,
 					thread,
+					ts: new Date(event.origin_server_ts),
 				});
 				return;
 			}
@@ -242,7 +243,7 @@ export function message() {
 					eventId,
 					thread,
 				);
-				await Message.saveMessageFromFederation(result);
+				await Message.saveMessageFromFederation({ ...result, ts: new Date(event.origin_server_ts) });
 			} else {
 				const formatted = toInternalMessageFormat({
 					rawMessage: messageBody,
@@ -257,6 +258,7 @@ export function message() {
 					msg: formatted,
 					federation_event_id: eventId,
 					thread,
+					ts: new Date(event.origin_server_ts),
 				});
 			}
 		} catch (error) {
@@ -359,6 +361,7 @@ export function message() {
 					},
 					federation_event_id: eventId,
 					thread,
+					ts: new Date(event.origin_server_ts),
 				});
 				return;
 			}
@@ -372,6 +375,7 @@ export function message() {
 				},
 				federation_event_id: eventId,
 				thread,
+				ts: new Date(event.origin_server_ts),
 			});
 		} catch (error) {
 			logger.error(error, 'Error processing Matrix message:');

packages/core-services/src/types/IMessageService.ts

@@ -19,6 +19,7 @@ export interface IMessageService {
 		files,
 		attachments,
 		thread,
+		ts,
 	}: {
 		fromId: string;
 		rid: string;
@@ -32,6 +33,7 @@ export interface IMessageService {
 		files?: IMessage['files'];
 		attachments?: IMessage['attachments'];
 		thread?: { tmid: string; tshow: boolean };
+		ts: Date;
 	}): Promise<IMessage>;
 	saveSystemMessageAndNotifyUser<T = IMessage>(
 		type: MessageTypesValues,