feat: add read receipts support

packages/core/src/events/edu/index.ts

@@ -6,5 +6,6 @@ export * from './base';
 
 export * from './m.typing';
 export * from './m.presence';
+export * from './m.receipt';
 
 export type MatrixEDUTypes = TypingEDU | PresenceEDU | BaseEDU;

packages/core/src/events/edu/m.receipt.ts

@@ -0,0 +1,47 @@
+import type { RoomID, UserID } from '@rocket.chat/federation-room';
+
+import type { BaseEDU } from './base';
+
+/**
+ * Typing receipt EDU as defined in the Matrix specification
+ *
+ * @see https://spec.matrix.org/latest/server-server-api/#receipts
+ */
+export interface ReceiptEDU extends BaseEDU {
+	edu_type: 'm.receipt';
+	content: Record<
+		RoomID,
+		{
+			'm.read': Record<
+				UserID,
+				{
+					data: {
+						thread_id?: string;
+						ts: number;
+					};
+					event_ids: string[];
+				}
+			>;
+		}
+	>;
+}
+
+export const isReceiptEDU = (edu: BaseEDU): edu is ReceiptEDU => {
+	return edu.edu_type === 'm.receipt';
+};
+
+export const createReceiptEDU = (roomId: RoomID, userId: UserID, eventIds: string[]): ReceiptEDU => ({
+	edu_type: 'm.receipt',
+	content: {
+		[roomId]: {
+			'm.read': {
+				[userId]: {
+					data: {
+						ts: Date.now(),
+					},
+					event_ids: eventIds,
+				},
+			},
+		},
+	},
+});

packages/federation-sdk/src/index.ts

@@ -67,6 +67,13 @@ export type HomeserverEventSignatures = {
 		last_active_ago?: number;
 		origin?: string;
 	};
+	'homeserver.matrix.receipt': {
+		room_id: string;
+		user_id: string;
+		event_ids: string[];
+		ts: number;
+		thread_id?: string;
+	};
 	'homeserver.matrix.encryption': {
 		event_id: EventID;
 		event: PduForType<'m.room.encryption'>;

packages/federation-sdk/src/sdk.ts

@@ -266,4 +266,8 @@ export class FederationSDK {
 	sendPresenceUpdateToRooms(...args: Parameters<typeof this.eduService.sendPresenceUpdateToRooms>) {
 		return this.eduService.sendPresenceUpdateToRooms(...args);
 	}
+
+	sendReadReceipt(...args: Parameters<typeof this.eduService.sendReadReceipt>) {
+		return this.eduService.sendReadReceipt(...args);
+	}
 }

packages/federation-sdk/src/services/config.service.ts

@@ -28,6 +28,7 @@ export interface AppConfig {
 	edu: {
 		processTyping: boolean;
 		processPresence: boolean;
+		processReceipt?: boolean;
 	};
 	userCheckTimeoutMs?: number;
 	networkCheckTimeoutMs?: number;
@@ -59,6 +60,7 @@ export const AppConfigSchema = z.object({
 	edu: z.object({
 		processTyping: z.boolean(),
 		processPresence: z.boolean(),
+		processReceipt: z.boolean().optional(),
 	}),
 	networkCheckTimeoutMs: z.number().int().min(1000, 'Network check timeout must be at least 1000ms').default(5000).optional(),
 	userCheckTimeoutMs: z.number().int().min(1000, 'User check timeout must be at least 1000ms').default(10000).optional(),

packages/federation-sdk/src/services/edu.service.ts

@@ -1,4 +1,4 @@
-import type { PresenceUpdate } from '@rocket.chat/federation-core';
+import type { PresenceUpdate, ReceiptEDU } from '@rocket.chat/federation-core';
 import { createPresenceEDU, createTypingEDU, createLogger } from '@rocket.chat/federation-core';
 import { RoomID } from '@rocket.chat/federation-room';
 import { singleton } from 'tsyringe';
@@ -67,4 +67,53 @@ export class EduService {
 			throw error;
 		}
 	}
+
+	async sendReadReceipt({
+		roomId,
+		userId,
+		eventIds,
+		threadId,
+	}: {
+		roomId: RoomID;
+		userId: string;
+		eventIds: string[];
+		threadId?: string;
+	}): Promise<void> {
+		try {
+			const origin = this.configService.serverName;
+			const receiptEDU: ReceiptEDU = {
+				edu_type: 'm.receipt',
+				content: {
+					[roomId]: {
+						'm.read': {
+							[userId]: {
+								data: {
+									ts: Date.now(),
+									thread_id: threadId,
+								},
+								event_ids: eventIds,
+							},
+						},
+					},
+				},
+			};
+
+			this.logger.debug(
+				`Sending read receipt for user ${userId} in room ${roomId} for events ${eventIds.join(', ')} to all servers in room`,
+			);
+
+			const servers = await this.stateService.getServersInRoom(roomId);
+			const uniqueServers = servers.filter((server) => server !== origin);
+
+			await this.federationService.sendEDUToServers([receiptEDU], uniqueServers);
+
+			this.logger.debug(`Sent read receipt to ${uniqueServers.length} unique servers for room ${roomId}`);
+		} catch (error) {
+			this.logger.error({
+				msg: 'Failed to send read receipt',
+				err: error,
+			});
+			throw error;
+		}
+	}
 }

packages/federation-sdk/src/services/event.service.ts

@@ -3,11 +3,20 @@ import type {
 	EventStagingStore,
 	PresenceEDU,
 	RoomPowerLevelsEvent,
+	ReceiptEDU,
 	TypingEDU,
 	RedactionEvent,
 	EventStore,
 } from '@rocket.chat/federation-core';
-import { isPresenceEDU, isTypingEDU, generateId, pruneEventDict, checkSignAndHashes, createLogger } from '@rocket.chat/federation-core';
+import {
+	isPresenceEDU,
+	isReceiptEDU,
+	isTypingEDU,
+	generateId,
+	pruneEventDict,
+	checkSignAndHashes,
+	createLogger,
+} from '@rocket.chat/federation-core';
 import {
 	type EventID,
 	type Pdu,
@@ -279,6 +288,9 @@ export class EventService {
 		if (isPresenceEDU(edu)) {
 			await this.processPresenceEDU(edu, origin);
 		}
+		if (isReceiptEDU(edu)) {
+			await this.processReceiptEDU(edu);
+		}
 	}
 
 	private async processTypingEDU(typingEDU: TypingEDU, origin?: string): Promise<void> {
@@ -338,6 +350,57 @@ export class EventService {
 		}
 	}
 
+	private async processReceiptEDU(receiptEDU: ReceiptEDU): Promise<void> {
+		const config = this.configService.getConfig('edu');
+
+		// turned on by default, so if undefined we should process receipts
+		if (config.processReceipt === false) {
+			return;
+		}
+
+		const { content } = receiptEDU;
+
+		if (!content || typeof content !== 'object') {
+			this.logger.warn('Invalid receipt EDU content, missing or invalid content');
+			return;
+		}
+
+		// Loop through each room in the receipt EDU
+		for await (const [roomId, roomReceipts] of Object.entries(content)) {
+			if (!roomReceipts?.['m.read']) {
+				this.logger.warn(`Invalid receipt EDU for room ${roomId}, missing m.read`);
+				continue;
+			}
+
+			const readReceipts = roomReceipts['m.read'];
+
+			// Loop through each user's read receipt in this room
+			for await (const [userId, receiptData] of Object.entries(readReceipts)) {
+				if (!receiptData?.event_ids || !Array.isArray(receiptData.event_ids)) {
+					this.logger.warn(`Invalid receipt data for user ${userId} in room ${roomId}, missing event_ids`);
+					continue;
+				}
+
+				const { event_ids, data } = receiptData;
+
+				const threadId = (data as { ts: number; thread_id?: string }).thread_id;
+				this.logger.debug(
+					`Processing read receipt for room ${roomId}: ${userId} read events ${event_ids.join(', ')}${
+						threadId ? ` in thread ${threadId}` : ''
+					}`,
+				);
+
+				await this.eventEmitterService.emit('homeserver.matrix.receipt', {
+					room_id: roomId,
+					user_id: userId,
+					event_ids,
+					ts: data.ts,
+					thread_id: threadId,
+				});
+			}
+		}
+	}
+
 	private validateEventByType(event: Pdu): string[] {
 		const errors: string[] = [];
 

packages/homeserver/src/homeserver.module.ts

@@ -75,6 +75,7 @@ export async function setup() {
 		edu: {
 			processTyping: process.env.EDU_PROCESS_TYPING !== 'false',
 			processPresence: process.env.EDU_PROCESS_PRESENCE === 'true',
+			processReceipt: process.env.EDU_PROCESS_RECEIPT !== 'false',
 		},
 	});