feat: implement backfill endpoint

packages/federation-sdk/src/repositories/event.repository.ts

@@ -1,19 +1,12 @@
 import { generateId } from '@rocket.chat/federation-core';
-import type { EventBase, EventStore } from '@rocket.chat/federation-core';
+import type { EventStore } from '@rocket.chat/federation-core';
 import {
 	type EventID,
 	Pdu,
 	PduForType,
 	PduType,
 } from '@rocket.chat/federation-room';
-import type {
-	Collection,
-	Filter,
-	FindCursor,
-	FindOptions,
-	UpdateResult,
-	WithId,
-} from 'mongodb';
+import type { Collection, FindCursor, WithId } from 'mongodb';
 import { MongoError } from 'mongodb';
 import { inject, singleton } from 'tsyringe';
 
@@ -388,4 +381,45 @@ export class EventRepository {
 			.sort({ 'event.depth': 1 })
 			.limit(limit);
 	}
+
+	async findNewestEventForBackfill(
+		roomId: string,
+		eventIds: EventID[],
+	): Promise<EventStore | null> {
+		return this.collection.findOne(
+			{
+				_id: { $in: eventIds },
+				'event.room_id': roomId,
+			},
+			{
+				sort: {
+					'event.depth': -1,
+					'event.origin_server_ts': -1,
+				},
+			},
+		);
+	}
+
+	findEventsForBackfill(
+		roomId: string,
+		depth: number,
+		originServerTs: number,
+		limit: number,
+	) {
+		return this.collection
+			.find({
+				'event.room_id': roomId,
+				$or: [
+					{ 'event.depth': { $lt: depth } },
+					{
+						'event.depth': depth,
+						'event.origin_server_ts': {
+							$lte: originServerTs,
+						},
+					},
+				],
+			})
+			.sort({ 'event.depth': -1, 'event.origin_server_ts': -1 })
+			.limit(limit);
+	}
 }

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

@@ -26,7 +26,6 @@ import type { z } from 'zod';
 import { StagingAreaQueue } from '../queues/staging-area.queue';
 import { EventStagingRepository } from '../repositories/event-staging.repository';
 import { EventRepository } from '../repositories/event.repository';
-import { KeyRepository } from '../repositories/key.repository';
 import { LockRepository } from '../repositories/lock.repository';
 import { eventSchemas } from '../utils/event-schemas';
 import { ConfigService } from './config.service';
@@ -49,7 +48,6 @@ export class EventService {
 		private readonly eventRepository: EventRepository,
 		private readonly eventStagingRepository: EventStagingRepository,
 		private readonly lockRepository: LockRepository,
-		private readonly keyRepository: KeyRepository,
 		private readonly configService: ConfigService,
 
 		private readonly stagingAreaQueue: StagingAreaQueue,
@@ -819,4 +817,57 @@ export class EventService {
 			throw error;
 		}
 	}
+
+	/**
+	 * A transaction containing the PDUs that preceded the given event(s), including the given event(s), up to the given limit.
+	 * 
+	 * Note: Though the PDU definitions require that prev_events and auth_events be limited in number, the response of backfill MUST NOT be validated on these specific restrictions.
+	 * 
+	 * Due to historical reasons, it is possible that events which were previously accepted would now be rejected by these limitations. The events should be rejected per usual by the /send, /get_missing_events, and remaining endpoints.
+
+	 */
+	async getBackfillEvents(
+		roomId: string,
+		eventIds: EventID[],
+		limit: number,
+	): Promise<{
+		origin: string;
+		origin_server_ts: number;
+		pdus: Array<Pdu>;
+	}> {
+		try {
+			const parsedLimit = Math.min(Math.max(1, limit), 100);
+
+			const newestRef = await this.eventRepository.findNewestEventForBackfill(
+				roomId,
+				eventIds,
+			);
+			if (!newestRef) {
+				throw new Error('No newest event found');
+			}
+
+			const events = await this.eventRepository
+				.findEventsForBackfill(
+					roomId,
+					newestRef.event.depth,
+					newestRef.event.origin_server_ts,
+					parsedLimit,
+				)
+				.toArray();
+
+			const pdus = events.map((eventStore) => eventStore.event);
+
+			return {
+				origin: this.configService.serverName,
+				origin_server_ts: Date.now(),
+				pdus,
+			};
+		} catch (error) {
+			this.logger.error({
+				msg: `Failed to get backfill for room ${roomId}:`,
+				err: error,
+			});
+			throw error;
+		}
+	}
 }

packages/homeserver/src/controllers/federation/transactions.controller.ts

@@ -7,6 +7,10 @@ import {
 import { Elysia } from 'elysia';
 import { container } from 'tsyringe';
 import {
+	BackfillErrorResponseDto,
+	BackfillParamsDto,
+	BackfillQueryDto,
+	BackfillResponseDto,
 	ErrorResponseDto,
 	GetEventErrorResponseDto,
 	GetEventParamsDto,
@@ -83,5 +87,55 @@ export const transactionsPlugin = (app: Elysia) => {
 					description: 'Get an event',
 				},
 			},
+		)
+		.get(
+			'/_matrix/federation/v1/backfill/:roomId',
+			async ({ params, query, set }) => {
+				try {
+					const limit = query.limit;
+					const eventIdParam = query.v;
+					if (!eventIdParam) {
+						set.status = 400;
+						return {
+							errcode: 'M_BAD_REQUEST',
+							error: 'Event ID must be provided in v query parameter',
+						};
+					}
+
+					const eventIds = Array.isArray(eventIdParam)
+						? eventIdParam
+						: [eventIdParam];
+
+					return eventService.getBackfillEvents(
+						params.roomId,
+						eventIds as EventID[],
+						limit,
+					);
+				} catch {
+					set.status = 500;
+					return {
+						errcode: 'M_UNKNOWN',
+						error: 'Failed to get backfill events',
+					};
+				}
+			},
+			{
+				params: BackfillParamsDto,
+				query: BackfillQueryDto,
+				response: {
+					200: BackfillResponseDto,
+					400: BackfillErrorResponseDto,
+					401: BackfillErrorResponseDto,
+					403: BackfillErrorResponseDto,
+					404: BackfillErrorResponseDto,
+					500: BackfillErrorResponseDto,
+				},
+				detail: {
+					tags: ['Federation'],
+					summary: 'Backfill room events',
+					description:
+						'Retrieves a sliding-window history of previous PDUs that occurred in the given room',
+				},
+			},
 		);
 };

packages/homeserver/src/dtos/common/event.dto.ts

@@ -1,4 +1,4 @@
-import { t } from 'elysia';
+import { TSchema, t } from 'elysia';
 import {
 	DepthDto,
 	RoomIdDto,
@@ -16,6 +16,12 @@ export const EventSignatureDto = t.Record(
 	{ description: 'Event signatures by server and key ID' },
 );
 
+type TOptional = ReturnType<typeof t.Optional>;
+
+function HiddenOptional<T extends TOptional>(schema: T): T {
+	return schema;
+}
+
 export const EventBaseDto = t.Object({
 	type: t.String({ description: 'Event type' }),
 	content: t.Record(t.String(), t.Any(), { description: 'Event content' }),
@@ -26,8 +32,8 @@ export const EventBaseDto = t.Object({
 	prev_events: t.Array(t.String(), {
 		description: 'Previous events in the room',
 	}),
+	origin: HiddenOptional(t.Optional(t.String())),
 	auth_events: t.Array(t.String(), { description: 'Authorization events' }),
-	origin: t.String({ description: 'Origin server' }),
 	hashes: EventHashDto,
 	signatures: EventSignatureDto,
 	unsigned: t.Optional(

packages/homeserver/src/dtos/federation/backfill.dto.ts

@@ -0,0 +1,22 @@
+import { t } from 'elysia';
+import { EventBaseDto } from '../common/event.dto';
+
+export const BackfillParamsDto = t.Object({
+	roomId: t.String({ minLength: 1 }),
+});
+
+export const BackfillQueryDto = t.Object({
+	limit: t.Number({ minimum: 1, maximum: 100 }),
+	v: t.Union([t.String(), t.Array(t.String())]),
+});
+
+export const BackfillResponseDto = t.Object({
+	origin: t.String(),
+	origin_server_ts: t.Number(),
+	pdus: t.Array(EventBaseDto),
+});
+
+export const BackfillErrorResponseDto = t.Object({
+	errcode: t.String(),
+	error: t.String(),
+});

packages/homeserver/src/dtos/index.ts

@@ -4,6 +4,7 @@ export * from './common/event.dto';
 export * from './common/validation.dto';
 
 // Federation DTOs
+export * from './federation/backfill.dto';
 export * from './federation/invite.dto';
 export * from './federation/profiles.dto';
 export * from './federation/send-join.dto';