Dispatch thread events

src/models/event.ts

@@ -30,6 +30,7 @@ import { deepSortedObjectEntries } from "../utils";
 import { RoomMember } from "./room-member";
 import { Thread } from "./thread";
 import { IActionsObject } from '../pushprocessor';
+import { ReEmitter } from '../ReEmitter';
 
 /**
  * Enum for event statuses.
@@ -217,6 +218,8 @@ export class MatrixEvent extends EventEmitter {
      */
     public verificationRequest = null;
 
+    private readonly reEmitter: ReEmitter;
+
     /**
      * Construct a Matrix Event object
      * @constructor
@@ -266,6 +269,7 @@ export class MatrixEvent extends EventEmitter {
 
         this.txnId = event.txn_id || null;
         this.localTimestamp = Date.now() - this.getAge();
+        this.reEmitter = new ReEmitter(this);
     }
 
     /**
@@ -1287,6 +1291,7 @@ export class MatrixEvent extends EventEmitter {
 
     public setThread(thread: Thread): void {
         this.thread = thread;
+        this.reEmitter.reEmit(thread, ["Thread.ready", "Thread.update"]);
     }
 
     public getThread(): Thread {

src/models/room.ts

@@ -866,7 +866,20 @@ export class Room extends EventEmitter {
      * @return {?module:models/event.MatrixEvent} the given event, or undefined if unknown
      */
     public findEventById(eventId: string): MatrixEvent | undefined {
-        return this.getUnfilteredTimelineSet().findEventById(eventId);
+        let event = this.getUnfilteredTimelineSet().findEventById(eventId);
+
+        if (event) {
+            return event;
+        } else {
+            const threads = this.getThreads();
+            for (let i = 0; i < threads.length; i++) {
+                const thread = threads[i];
+                event = thread.findEventById(eventId);
+                if (event) {
+                    return event;
+                }
+            }
+        }
     }
 
     /**
@@ -1261,6 +1274,24 @@ export class Room extends EventEmitter {
         }
     }
 
+    public addThreadedEvent(event: MatrixEvent): void {
+        if (event.getUnsigned().transaction_id) {
+            const existingEvent = this.txnToEvent[event.getUnsigned().transaction_id];
+            if (existingEvent) {
+                // remote echo of an event we sent earlier
+                this.handleRemoteEcho(event, existingEvent);
+            }
+        }
+
+        let thread = this.findEventById(event.replyEventId)?.getThread();
+        if (thread) {
+            thread.addEvent(event);
+        } else {
+            thread = new Thread([event], this, this.client);
+            this.addThread(thread);
+        }
+    }
+
     /**
      * Add an event to the end of this room's live timelines. Will fire
      * "Room.timeline".

src/models/thread.ts

@@ -46,7 +46,7 @@ export class Thread extends EventEmitter {
      * the tail/root references if needed
      * @param event The event to add
      */
-    public addEvent(event: MatrixEvent): void {
+    public async addEvent(event: MatrixEvent): Promise<void> {
         if (this.events.has(event.getId()) || event.status !== null) {
             return;
         }
@@ -62,25 +62,26 @@ export class Thread extends EventEmitter {
 
         event.setThread(this);
         this.events.set(event.getId(), event);
-        this._timelineSet.addEventToTimeline(
-            event,
-            this._timelineSet.getLiveTimeline(),
-            false,
-            false,
-        );
+        this._timelineSet.addLiveEvent(event);
 
         if (this.ready) {
             this.client.decryptEventIfNeeded(event, {});
             this.emit("Thread.update", this);
+        } else {
+            this.emit("Thread.update", this);
         }
     }
 
     public async fetchReplyChain(): Promise<void> {
         if (!this.ready) {
-            const mxEvent = await this.fetchEventById(
-                this.rootEvent.getRoomId(),
-                this.rootEvent.replyEventId,
-            );
+            let mxEvent = this.room.findEventById(this.rootEvent.replyEventId);
+            if (!mxEvent) {
+                mxEvent = await this.fetchEventById(
+                    this.rootEvent.getRoomId(),
+                    this.rootEvent.replyEventId,
+                );
+            }
+
             this.addEvent(mxEvent);
             if (mxEvent.replyEventId) {
                 await this.fetchReplyChain();
@@ -101,7 +102,7 @@ export class Thread extends EventEmitter {
         this.decrypted = true;
     }
 
-    public async fetchEventById(roomId: string, eventId: string): Promise<MatrixEvent> {
+    private async fetchEventById(roomId: string, eventId: string): Promise<MatrixEvent> {
         const response = await this.client.http.authedRequest(
             undefined,
             "GET",
@@ -110,8 +111,12 @@ export class Thread extends EventEmitter {
         return new MatrixEvent(response);
     }
 
+    public findEventById(eventId: string) {
+        return this.events.get(eventId);
+    }
+
     public get ready(): boolean {
-        return this.rootEvent.replyEventId === undefined && this.decrypted;
+        return this.rootEvent.replyEventId === undefined;
     }
 
     /**

src/sync.ts

@@ -301,7 +301,7 @@ export class SyncApi {
                 client.store.storeRoom(room);
                 client.emit("Room", room);
 
-                this.processEventsForNotifs(room, timelineEvents);
+                this.processEventsForNotifs(room, events);
             });
             return rooms;
         });
@@ -1314,7 +1314,7 @@ export class SyncApi {
                 client.emit("Room", room);
             }
 
-            this.processEventsForNotifs(room, timelineEvents);
+            this.processEventsForNotifs(room, events);
 
             const processRoomEvent = async (e) => {
                 client.emit("event", e);
@@ -1370,7 +1370,7 @@ export class SyncApi {
                 client.emit("Room", room);
             }
 
-            this.processEventsForNotifs(room, timelineEvents);
+            this.processEventsForNotifs(room, events);
 
             stateEvents.forEach(function(e) {
                 client.emit("event", e);
@@ -1702,13 +1702,7 @@ export class SyncApi {
 
     private processThreadEvents(room: Room, threadedEvents: MatrixEvent[]): void {
         threadedEvents.forEach(event => {
-            let thread = room.findEventById(event.replyEventId).getThread();
-            if (thread) {
-                thread.addEvent(event);
-            } else {
-                thread = new Thread([event], room, this.client);
-                room.addThread(thread);
-            }
+            room.addThreadedEvent(event);
         });
     }