feat: simple implementation of sending transactions to all participating servers

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

@@ -13,6 +13,8 @@ import {
 import { FederationConfigService } from './federation-config.service';
 import { FederationRequestService } from './federation-request.service';
 import { SignatureVerificationService } from './signature-verification.service';
+import { StateService } from './state.service';
+import { PersistentEventBase } from '@hs/room';
 
 @singleton()
 export class FederationService {
@@ -23,6 +25,7 @@ export class FederationService {
 		@inject('FederationRequestService')
 		private readonly requestService: FederationRequestService,
 		private readonly signatureService: SignatureVerificationService,
+		private readonly stateService: StateService,
 	) {}
 
 	/**
@@ -97,7 +100,7 @@ export class FederationService {
 		transaction: Transaction,
 	): Promise<SendTransactionResponse> {
 		try {
-			const txnId = `${Date.now()}-${Math.random().toString(36).substring(2, 10)}`;
+			const txnId = Date.now().toString();
 			const uri = FederationEndpoints.sendTransaction(txnId);
 
 			return await this.requestService.put<SendTransactionResponse>(
@@ -224,4 +227,33 @@ export class FederationService {
 			throw error;
 		}
 	}
+
+	async sendEventToAllServersInRoom(event: PersistentEventBase) {
+		const servers = await this.stateService.getServersInRoom(event.roomId);
+
+		for (const server of servers) {
+			if (server === event.origin) {
+				this.logger.info(
+					`Skipping transaction to event origin: ${event.origin}`,
+				);
+				continue;
+			}
+
+			if (server === this.configService.serverName) {
+				this.logger.info(`Skipping transaction to local server: ${server}`);
+				continue;
+			}
+
+			const txn: Transaction = {
+				origin: this.configService.serverName,
+				origin_server_ts: Date.now(),
+				pdus: [event],
+				edus: [],
+			};
+
+			this.logger.info(`Sending event ${event.eventId} to server: ${server}`);
+
+			void this.sendTransaction(server, txn);
+		}
+	}
 }

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

@@ -506,4 +506,28 @@ export class StateService {
 			}))
 			.toArray();
 	}
+
+	async getMembersOfRoom(roomId: string) {
+		const stateCollection = await this.stateRepository.getCollection();
+
+		const stateMappings = await stateCollection
+			.find({ roomId, 'delta.identifier': /^m\.room\.member:/ })
+			.toArray();
+
+		const events = await this.eventRepository.findByIds(
+			stateMappings.map((stateMapping) => stateMapping.delta.eventId),
+		);
+
+		const members = events
+			.filter((event) => event.event.content?.membership === 'join')
+			.map((event) => event.event.state_key as string);
+
+		return members;
+	}
+
+	async getServersInRoom(roomId: string) {
+		return this.getMembersOfRoom(roomId).then((members) =>
+			members.filter((member) => member.split(':').pop()!),
+		);
+	}
 }