feat: message reactions

.env.template

@@ -0,0 +1,24 @@
+MONGODB_URI=
+DATABASE_NAME=
+DATABASE_POOL_SIZE=
+
+SERVER_NAME=
+SERVER_VERSION=
+SERVER_PORT=
+SERVER_BASE_URL=
+SERVER_HOST=
+
+MATRIX_SERVER_NAME=
+MATRIX_DOMAIN=
+MATRIX_KEY_REFRESH_INTERVAL=
+
+CONFIG_FOLDER=
+
+HOMESERVER_CONFIG_DNS_SERVERS=
+
+LOG_LEVEL=
+NODE_ENV=
+
+DEBUG=
+
+NODE_TLS_REJECT_UNAUTHORIZED=

packages/federation-sdk/src/index.ts

@@ -132,6 +132,29 @@ export type HomeserverEventSignatures = {
 			membership: Membership;
 		};
 	};
+	'homeserver.matrix.reaction': {
+		event_id: string;
+		room_id: string;
+		sender: string;
+		origin_server_ts: number;
+		content: {
+			'm.relates_to': {
+				rel_type: 'm.annotation';
+				event_id: string;
+				key: string;
+			};
+		};
+	};
+	'homeserver.matrix.redaction': {
+		event_id: string;
+		room_id: string;
+		sender: string;
+		origin_server_ts: number;
+		redacts: string;
+		content: {
+			reason?: string;
+		};
+	};
 };
 
 export function getAllServices(): HomeserverServices {

packages/federation-sdk/src/queues/staging-area.queue.ts

@@ -15,11 +15,20 @@ type QueueHandler = (item: StagingAreaEventType) => Promise<void>;
 @singleton()
 export class StagingAreaQueue {
 	private queue: StagingAreaEventType[] = [];
+	private priorityQueue: StagingAreaEventType[] = [];
 	private handlers: QueueHandler[] = [];
 	private processing = false;
 
 	enqueue(item: StagingAreaEventType): void {
-		this.queue.push(item);
+		// If this is a continuation of processing (has metadata.state), add to priority queue
+		if (
+			item.metadata?.state &&
+			item.metadata.state !== 'pending_dependencies'
+		) {
+			this.priorityQueue.push(item);
+		} else {
+			this.queue.push(item);
+		}
 		this.processQueue();
 	}
 
@@ -32,15 +41,16 @@ export class StagingAreaQueue {
 	}
 
 	private async processQueue(): Promise<void> {
-		if (this.processing || this.queue.length === 0) {
+		if (this.processing) {
 			return;
 		}
 
 		this.processing = true;
 
 		try {
-			while (this.queue.length > 0) {
-				const item = this.queue.shift();
+			while (this.priorityQueue.length > 0 || this.queue.length > 0) {
+				// Process priority queue first (events in mid-processing)
+				const item = this.priorityQueue.shift() || this.queue.shift();
 				if (!item) continue;
 
 				for (const handler of this.handlers) {
@@ -49,6 +59,11 @@ export class StagingAreaQueue {
 			}
 		} finally {
 			this.processing = false;
+
+			// Check if new items were added while processing
+			if (this.priorityQueue.length > 0 || this.queue.length > 0) {
+				this.processQueue();
+			}
 		}
 	}
 }

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

@@ -49,7 +49,6 @@ export const AppConfigSchema = z.object({
 
 export class ConfigService {
 	private config: AppConfig;
-	private fileConfig: Partial<AppConfig> = {};
 	private logger = createLogger('ConfigService');
 
 	constructor(values: AppConfig) {

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

@@ -284,15 +284,22 @@ export class FederationService {
 			}
 
 			const txn: Transaction = {
-				origin: 'rc1.tunnel.dev.rocket.chat', //this.configService.serverName,
+				origin: this.configService.serverName,
 				origin_server_ts: Date.now(),
 				pdus: [event.event],
 				edus: [],
 			};
 
 			this.logger.info(`Sending event ${event.eventId} to server: ${server}`);
 
-			void this.sendTransaction(server, txn);
+			try {
+				await this.sendTransaction(server, txn);
+			} catch (error) {
+				this.logger.error(
+					`Failed to send event ${event.eventId} to server: ${server}`,
+					error,
+				);
+			}
 		}
 	}
 }

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

@@ -1,8 +1,3 @@
-import {
-	type ReactionAuthEvents,
-	type ReactionEvent,
-	reactionEvent,
-} from '@hs/core';
 import {
 	type MessageAuthEvents,
 	type RoomMessageEvent,
@@ -17,16 +12,20 @@ import {
 	redactionEvent,
 } from '@hs/core';
 import { createLogger } from '@hs/core';
-import { generateId } from '@hs/core';
 import { signEvent } from '@hs/core';
+import {
+	type PersistentEventBase,
+	PersistentEventFactory,
+	type RoomVersion,
+} from '@hs/room';
 import { inject } from 'tsyringe';
 import { singleton } from 'tsyringe';
+import type { EventRepository } from '../repositories/event.repository';
 import type { ConfigService } from './config.service';
 import { EventService, EventType } from './event.service';
 import { FederationService } from './federation.service';
 import type { RoomService } from './room.service';
-import { StateService } from './state.service';
-import { PersistentEventBase, PersistentEventFactory } from '@hs/room';
+import type { StateService } from './state.service';
 
 @singleton()
 export class MessageService {
@@ -39,6 +38,8 @@ export class MessageService {
 		private readonly federationService: FederationService,
 		@inject('RoomService') private readonly roomService: RoomService,
 		@inject('StateService') private readonly stateService: StateService,
+		@inject('EventRepository')
+		private readonly eventRepository: EventRepository,
 	) {}
 
 	async sendMessage(
@@ -82,76 +83,67 @@ export class MessageService {
 		eventId: string,
 		emoji: string,
 		senderUserId: string,
-		targetServer: string,
-	): Promise<SignedEvent<ReactionEvent>> {
+	): Promise<string> {
 		const isTombstoned = await this.roomService.isRoomTombstoned(roomId);
 		if (isTombstoned) {
 			this.logger.warn(
-				`Attempted to send message to a tombstoned room: ${roomId}`,
+				`Attempted to react to a message in a tombstoned room: ${roomId}`,
 			);
 			throw new ForbiddenError(
 				'Cannot react to a message in a tombstoned room',
 			);
 		}
 
-		const serverName = this.configService.getServerConfig().name;
-		const signingKey = await this.configService.getSigningKey();
+		const roomInfo = await this.stateService.getRoomInformation(roomId);
 
-		const latestEventDoc = await this.eventService.getLastEventForRoom(roomId);
-		const prevEvents = latestEventDoc ? [latestEventDoc._id] : [];
-
-		const authEvents = await this.eventService.getAuthEventIds(
-			EventType.REACTION,
-			{ roomId, senderId: senderUserId },
+		const reactionEvent = PersistentEventFactory.newReactionEvent(
+			roomId,
+			senderUserId,
+			eventId,
+			emoji,
+			roomInfo.room_version as RoomVersion,
 		);
 
-		const currentDepth = latestEventDoc?.event?.depth ?? 0;
-		const newDepth = currentDepth + 1;
+		await this.stateService.addAuthEvents(reactionEvent);
 
-		const authEventsMap: ReactionAuthEvents = {
-			'm.room.create':
-				authEvents.find((event) => event.type === EventType.CREATE)?._id || '',
-			'm.room.power_levels':
-				authEvents.find((event) => event.type === EventType.POWER_LEVELS)
-					?._id || '',
-			'm.room.member':
-				authEvents.find((event) => event.type === EventType.MEMBER)?._id || '',
-		};
+		await this.stateService.addPrevEvents(reactionEvent);
 
-		const { state_key, ...eventForSigning } = reactionEvent({
-			roomId,
-			sender: senderUserId,
-			auth_events: authEventsMap,
-			prev_events: prevEvents,
-			depth: newDepth,
-			content: {
-				'm.relates_to': {
-					rel_type: 'm.annotation',
-					event_id: eventId,
-					key: emoji,
-				},
-			},
-			origin: serverName,
-			ts: Date.now(),
-		});
+		await this.stateService.signEvent(reactionEvent);
 
-		const signedEvent = await signEvent(
-			eventForSigning,
-			Array.isArray(signingKey) ? signingKey[0] : signingKey,
-			serverName,
-		);
+		await this.stateService.persistTimelineEvent(reactionEvent);
 
-		this.logger.debug(signedEvent);
+		void this.federationService.sendEventToAllServersInRoom(reactionEvent);
 
-		await this.federationService.sendEvent(targetServer, signedEvent);
+		return reactionEvent.eventId;
+	}
 
-		await this.eventService.insertEvent(signedEvent, eventId);
+	async unsetReaction(
+		roomId: string,
+		eventIdReactedTo: string,
+		_emoji: string,
+		senderUserId: string,
+	): Promise<string> {
+		const roomInfo = await this.stateService.getRoomInformation(roomId);
 
-		this.logger.info(
-			`Sent reaction $emojito $targetServerfor event $eventId- $generateId(${signedEvent})`,
+		const redactionEvent = PersistentEventFactory.newRedactionEvent(
+			roomId,
+			senderUserId,
+			eventIdReactedTo,
+			'Unsetting reaction',
+			roomInfo.room_version as RoomVersion,
 		);
 
-		return signedEvent;
+		await this.stateService.addAuthEvents(redactionEvent);
+
+		await this.stateService.addPrevEvents(redactionEvent);
+
+		await this.stateService.signEvent(redactionEvent);
+
+		await this.stateService.persistTimelineEvent(redactionEvent);
+
+		void this.federationService.sendEventToAllServersInRoom(redactionEvent);
+
+		return redactionEvent.eventId;
 	}
 
 	async updateMessage(

packages/federation-sdk/src/services/staging-area.service.ts

@@ -4,14 +4,14 @@ import type { StagingAreaEventType } from '../queues/staging-area.queue';
 import { StagingAreaQueue } from '../queues/staging-area.queue';
 
 import { createLogger } from '@hs/core';
+import { Pdu, PersistentEventFactory } from '@hs/room';
 import { Lock } from '../utils/lock.decorator';
 import { EventAuthorizationService } from './event-authorization.service';
+import { EventEmitterService } from './event-emitter.service';
 import { EventStateService } from './event-state.service';
 import { EventService } from './event.service';
 import { EventType } from './event.service';
 import { MissingEventService } from './missing-event.service';
-import { EventEmitterService } from './event-emitter.service';
-import { Pdu, PersistentEventFactory } from '@hs/room';
 import { StateService } from './state.service';
 
 // ProcessingState indicates where in the flow an event is
@@ -47,9 +47,7 @@ export class StagingAreaService {
 		private readonly eventStateService: EventStateService,
 		private readonly eventEmitterService: EventEmitterService,
 		private readonly stateService: StateService,
-	) {
-		this.processQueue();
-	}
+	) {}
 
 	addEventToQueue(event: StagingAreaEventType) {
 		const extendedEvent: ExtendedStagingEvent = {
@@ -70,22 +68,13 @@ export class StagingAreaService {
 		this.logger.debug(`Added event ${event.eventId} to processing queue`);
 	}
 
-	private async processQueue() {
-		setInterval(async () => {
-			const event = this.stagingAreaQueue.dequeue();
-			if (event) {
-				await this.processEvent(event);
-			}
-		}, 100);
-	}
-
 	extractEventsFromIncomingPDU(pdu: StagingAreaEventType) {
 		const authEvents = pdu.event.auth_events || [];
 		const prevEvents = pdu.event.prev_events || [];
 		return [...authEvents, ...prevEvents];
 	}
 
-	@Lock({ timeout: 10000, keyPath: 'event.room_id' })
+	// @Lock({ timeout: 10000, keyPath: 'event.room_id' })
 	async processEvent(event: StagingAreaEventType & { metadata?: any }) {
 		const eventId = event.eventId;
 		const trackedEvent = this.processingEvents.get(eventId);
@@ -347,7 +336,9 @@ export class StagingAreaService {
 	private async processNotificationStage(event: StagingAreaEventType) {
 		const eventId = event.eventId;
 		const trackedEvent = this.processingEvents.get(eventId);
-		if (!trackedEvent) return;
+		if (!trackedEvent) {
+			return;
+		}
 
 		try {
 			this.logger.debug(`Notifying clients about event ${eventId}`);
@@ -365,6 +356,35 @@ export class StagingAreaService {
 						},
 					});
 					break;
+				case EventType.REACTION: {
+					this.eventEmitterService.emit('homeserver.matrix.reaction', {
+						event_id: event.eventId,
+						room_id: event.roomId,
+						sender: event.event.sender,
+						origin_server_ts: event.event.origin_server_ts,
+						content: event.event.content as {
+							'm.relates_to': {
+								rel_type: 'm.annotation';
+								event_id: string;
+								key: string;
+							};
+						},
+					});
+					break;
+				}
+				case EventType.REDACTION: {
+					this.eventEmitterService.emit('homeserver.matrix.redaction', {
+						event_id: event.eventId,
+						room_id: event.roomId,
+						sender: event.event.sender,
+						origin_server_ts: event.event.origin_server_ts,
+						redacts: (event.event as any).redacts,
+						content: {
+							reason: event.event.content?.reason as string | undefined,
+						},
+					});
+					break;
+				}
 				default:
 					this.logger.warn(
 						`Unknown event type: ${event.event.type} for emitterService for now`,