feat: add 'from' property to EventStagingStore and update event handling logic

Author: ggazzo

packages/core/src/models/event.model.ts

@@ -39,6 +39,7 @@ export interface EventStore<E = Pdu> extends PersistentEventBase<E> {
 
 export interface EventStagingStore extends PersistentEventBase {
 	roomId: string;
+	from: 'join' | 'transaction';
 }
 
 export interface FetchedEvents {

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

@@ -17,6 +17,7 @@ export class EventStagingRepository {
 		eventId: EventID,
 		origin: string,
 		event: Pdu,
+		from: 'join' | 'transaction' = 'transaction',
 	): Promise<UpdateResult> {
 		// We use an upsert here to handle the case where we see the same event
 		// from the same server multiple times.
@@ -32,6 +33,7 @@ export class EventStagingRepository {
 				},
 				$set: {
 					event,
+					from,
 				},
 			},
 			{

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

@@ -40,6 +40,7 @@ import { EventEmitterService } from './event-emitter.service';
 import { EventFetcherService } from './event-fetcher.service';
 import { InviteService } from './invite.service';
 import { StateService, UnknownRoomError } from './state.service';
+import { EventStagingRepository } from '../repositories/event-staging.repository';
 
 @singleton()
 export class RoomService {
@@ -54,6 +55,7 @@ export class RoomService {
 		private readonly inviteService: InviteService,
 		private readonly eventEmitterService: EventEmitterService,
 		private readonly eventFetcherService: EventFetcherService,
+		private readonly eventStagingRepository: EventStagingRepository,
 	) {}
 
 	private validatePowerLevelChange(
@@ -929,6 +931,12 @@ export class RoomService {
 						'processing state at event',
 					);
 					await stateService._resolveStateAtEvent(missingEvent);
+					void this.eventStagingRepository.create(
+						missingEvent.eventId,
+						residentServer,
+						missingEvent.event,
+						'join',
+					);
 				}
 
 				// if partial, join event will also be partial
@@ -979,9 +987,9 @@ export class RoomService {
 		});
 
 		// try to persist the join event now, should succeed with state in place
-		await this.eventService.processIncomingPDUs(
+		void this.eventService.processIncomingPDUs(
 			residentServer || joinEventFinal.origin,
-			[joinEventFinal.event],
+			[...state, joinEventFinal.event],
 		);
 
 		return joinEventFinal.eventId;

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

@@ -84,7 +84,9 @@ export class StagingAreaService {
 
 			try {
 				await this.processDependencyStage(event);
-				await this.processAuthorizationStage(event);
+				if ('from' in event && event.from !== 'join') {
+					await this.processAuthorizationStage(event);
+				}
 				await this.stateService.handlePdu(await toEventBase(event.event));
 				await this.processNotificationStage(event);