Fix message ordering in threads (#2215)

src/client.ts

@@ -5197,7 +5197,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
                 const [timelineEvents, threadedEvents] = this.partitionThreadedEvents(matrixEvents);
 
                 room.addEventsToTimeline(timelineEvents, true, room.getLiveTimeline());
-                await this.processThreadEvents(room, threadedEvents);
+                await this.processThreadEvents(room, threadedEvents, true);
 
                 room.oldState.paginationToken = res.end;
                 if (res.chunk.length === 0) {
@@ -5308,7 +5308,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
             const [timelineEvents, threadedEvents] = this.partitionThreadedEvents(matrixEvents);
 
             timelineSet.addEventsToTimeline(timelineEvents, true, timeline, res.start);
-            await this.processThreadEvents(timelineSet.room, threadedEvents);
+            await this.processThreadEvents(timelineSet.room, threadedEvents, true);
 
             // there is no guarantee that the event ended up in "timeline" (we
             // might have switched to a neighbouring timeline) - so check the
@@ -5441,7 +5441,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
 
                 const timelineSet = eventTimeline.getTimelineSet();
                 timelineSet.addEventsToTimeline(timelineEvents, backwards, eventTimeline, token);
-                await this.processThreadEvents(timelineSet.room, threadedEvents);
+                await this.processThreadEvents(timelineSet.room, threadedEvents, backwards);
 
                 // if we've hit the end of the timeline, we need to stop trying to
                 // paginate. We need to keep the 'forwards' token though, to make sure
@@ -5479,7 +5479,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
 
                 eventTimeline.getTimelineSet()
                     .addEventsToTimeline(timelineEvents, backwards, eventTimeline, token);
-                await this.processThreadEvents(room, threadedEvents);
+                await this.processThreadEvents(room, threadedEvents, backwards);
 
                 // if we've hit the end of the timeline, we need to stop trying to
                 // paginate. We need to keep the 'forwards' token though, to make sure
@@ -9294,10 +9294,13 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
     /**
      * @experimental
      */
-    public async processThreadEvents(room: Room, threadedEvents: MatrixEvent[]): Promise<void> {
-        threadedEvents.sort((a, b) => a.getTs() - b.getTs());
+    public async processThreadEvents(
+        room: Room,
+        threadedEvents: MatrixEvent[],
+        toStartOfTimeline: boolean,
+    ): Promise<void> {
         for (const event of threadedEvents) {
-            await room.addThreadedEvent(event);
+            await room.addThreadedEvent(event, toStartOfTimeline);
         }
     }
 

src/event-mapper.ts

@@ -30,6 +30,12 @@ export function eventMapperFor(client: MatrixClient, options: MapperOpts): Event
 
     function mapper(plainOldJsObject: Partial<IEvent>) {
         const event = new MatrixEvent(plainOldJsObject);
+
+        const room = client.getRoom(event.getRoomId());
+        if (room?.threads.has(event.getId())) {
+            event.setThread(room.threads.get(event.getId()));
+        }
+
         if (event.isEncrypted()) {
             if (!preventReEmit) {
                 client.reEmitter.reEmit(event, [

src/models/room.ts

@@ -1406,10 +1406,10 @@ export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap>
      * Add an event to a thread's timeline. Will fire "Thread.update"
      * @experimental
      */
-    public async addThreadedEvent(event: MatrixEvent): Promise<void> {
+    public async addThreadedEvent(event: MatrixEvent, toStartOfTimeline: boolean): Promise<void> {
         let thread = this.findThreadForEvent(event);
         if (thread) {
-            thread.addEvent(event);
+            thread.addEvent(event, toStartOfTimeline);
         } else {
             const events = [event];
             let rootEvent = this.findEventById(event.threadRootId);

src/models/thread.ts

@@ -113,7 +113,7 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
         }
         this.initialiseThread(this.rootEvent);
 
-        opts?.initialEvents?.forEach(event => this.addEvent(event));
+        opts?.initialEvents?.forEach(event => this.addEvent(event, false));
 
         this.room.on(RoomEvent.LocalEchoUpdated, this.onEcho);
         this.room.on(RoomEvent.Timeline, this.onEcho);
@@ -158,7 +158,7 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
      * @param {boolean} toStartOfTimeline whether the event is being added
      * to the start (and not the end) of the timeline.
      */
-    public async addEvent(event: MatrixEvent, toStartOfTimeline = false): Promise<void> {
+    public async addEvent(event: MatrixEvent, toStartOfTimeline: boolean): Promise<void> {
         if (Thread.hasServerSideSupport === undefined) {
             await Thread.serverSupportPromise;
         }
@@ -232,22 +232,28 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
             this.setEventMetadata(event);
             this.lastEvent = event;
         }
-
-        if (!bundledRelationship && rootEvent) {
-            this.addEvent(rootEvent);
-        }
     }
 
-    public async fetchInitialEvents(): Promise<boolean> {
+    public async fetchInitialEvents(): Promise<{
+        originalEvent: MatrixEvent;
+        events: MatrixEvent[];
+        nextBatch?: string;
+        prevBatch?: string;
+    } | null> {
         if (Thread.hasServerSideSupport === undefined) {
             await Thread.serverSupportPromise;
         }
+
+        if (!Thread.hasServerSideSupport) {
+            this.initialEventsFetched = true;
+            return null;
+        }
         try {
-            await this.fetchEvents();
+            const response = await this.fetchEvents();
             this.initialEventsFetched = true;
-            return true;
+            return response;
         } catch (e) {
-            return false;
+            return null;
         }
     }
 
@@ -317,7 +323,7 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
         nextBatch?: string;
         prevBatch?: string;
     }> {
-        if (Thread.serverSupportPromise) {
+        if (Thread.hasServerSideSupport === undefined) {
             await Thread.serverSupportPromise;
         }
 
@@ -337,13 +343,13 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
         // When there's no nextBatch returned with a `from` request we have reached
         // the end of the thread, and therefore want to return an empty one
         if (!opts.to && !nextBatch) {
-            events = [originalEvent, ...events];
+            events = [...events, originalEvent];
         }
 
-        for (const event of events) {
-            await this.client.decryptEventIfNeeded(event);
+        await Promise.all(events.map(event => {
             this.setEventMetadata(event);
-        }
+            return this.client.decryptEventIfNeeded(event);
+        }));
 
         const prependEvents = !opts.direction || opts.direction === Direction.Backward;
 

src/sync.ts

@@ -320,7 +320,7 @@ export class SyncApi {
                     EventTimeline.BACKWARDS);
 
                 this.processRoomEvents(room, stateEvents, timelineEvents);
-                await this.processThreadEvents(room, threadedEvents);
+                await this.processThreadEvents(room, threadedEvents, false);
 
                 room.recalculate();
                 client.store.storeRoom(room);
@@ -1317,7 +1317,7 @@ export class SyncApi {
             const [timelineEvents, threadedEvents] = this.client.partitionThreadedEvents(events);
 
             this.processRoomEvents(room, stateEvents, timelineEvents, syncEventData.fromCache);
-            await this.processThreadEvents(room, threadedEvents);
+            await this.processThreadEvents(room, threadedEvents, false);
 
             // set summary after processing events,
             // because it will trigger a name calculation
@@ -1385,7 +1385,7 @@ export class SyncApi {
             const [timelineEvents, threadedEvents] = this.client.partitionThreadedEvents(events);
 
             this.processRoomEvents(room, stateEvents, timelineEvents);
-            await this.processThreadEvents(room, threadedEvents);
+            await this.processThreadEvents(room, threadedEvents, false);
             room.addAccountData(accountDataEvents);
 
             room.recalculate();
@@ -1730,8 +1730,12 @@ export class SyncApi {
     /**
      * @experimental
      */
-    private processThreadEvents(room: Room, threadedEvents: MatrixEvent[]): Promise<void> {
-        return this.client.processThreadEvents(room, threadedEvents);
+    private processThreadEvents(
+        room: Room,
+        threadedEvents: MatrixEvent[],
+        toStartOfTimeline: boolean,
+    ): Promise<void> {
+        return this.client.processThreadEvents(room, threadedEvents, toStartOfTimeline);
     }
 
     // extractRelatedEvents(event: MatrixEvent, events: MatrixEvent[], relatedEvents: MatrixEvent[] = []): MatrixEvent[] {