Thread store and event parsing to create threads

spec/unit/models/thread.spec.ts

@@ -0,0 +1,105 @@
+import { MatrixEvent } from "../../../src/models/event";
+import { Thread } from "../../../src/models/thread";
+
+const mockMessage = (id: string, body: string, sender: string, replyId?: string) => {
+    const opts = {
+        content: {
+            body,
+        },
+        event_id: id,
+        origin_server_ts: 1628082832517,
+        sender: sender,
+        type: "m.room.message",
+        room_id: "!room123:hs1",
+    };
+
+    if (replyId) {
+        opts.content["m.relates_to"] = {
+            "m.in_reply_to": {
+                "event_id": replyId,
+            },
+        };
+    }
+
+    return new MatrixEvent(opts);
+};
+
+const mockAnnotation = (id: string, body: string, sender: string, replyId: string) => {
+    return new MatrixEvent({
+        "content": {
+            "m.relates_to": {
+                "event_id": replyId,
+                "key": body,
+                "rel_type": "m.annotation",
+            },
+        },
+        "origin_server_ts": 1628084947352,
+        "sender": sender,
+        "type": "m.reaction",
+        "event_id": id,
+        "room_id": "!room123:hs1",
+    });
+};
+
+const mockThread = () => {
+    const events = [
+        mockMessage("event1", "Hello", "alice"),
+        mockMessage("event2", "Bonjour", "bob", "event1"),
+        mockMessage("event3", "How are you?", "bob", "event2"),
+    ];
+
+    return new Thread(events);
+};
+
+describe('Thread', () => {
+    it('should count participants', () => {
+        const thread = mockThread();
+
+        expect(thread.participants.size).toBe(2);
+
+        thread.addEvent(mockMessage("event4", "Ça va?", "bob", "event2"));
+        thread.addEvent(mockMessage("event5", "Cześć", "charlie", "event2"));
+
+        expect(thread.participants.size).toBe(3);
+    });
+
+    it('should store reference to root and tails', () => {
+        const thread = mockThread();
+        expect(thread.id).toBe("event1");
+        expect(thread.tail.size).toBe(1);
+
+        thread.addEvent(mockMessage("event4", "Ça va?", "bob", "event2"));
+        expect(thread.tail.size).toBe(2);
+
+        thread.addEvent(mockMessage("event5", "Ça va?", "bob", "event4"));
+        expect(thread.tail.size).toBe(2);
+
+        thread.addEvent(mockMessage("event6", "Ça va?", "bob", "event1"));
+        expect(thread.tail.size).toBe(3);
+    });
+
+    it('should only count message events', () => {
+        const thread = mockThread();
+        expect(thread.length).toBe(3);
+
+        thread.addEvent(mockMessage("event4", "Ça va?", "bob", "event2"));
+        expect(thread.length).toBe(4);
+
+        const reaction = mockAnnotation("event6", "✅", "bob", "event2");
+
+        thread.addEvent(reaction);
+
+        expect(thread.length).toBe(4);
+        expect(thread.eventTimeline.length).toBe(5);
+    });
+
+    it('tails event can only be m.room.message', () => {
+        const thread = mockThread();
+        expect(thread.length).toBe(3);
+
+        const reaction = mockAnnotation("event10", "✅", "bob", "event2");
+        thread.addEvent(reaction);
+
+        expect(Array.from(thread.tail)[0]).toBe("event3");
+    });
+});

src/models/event.ts

@@ -190,6 +190,10 @@ export class MatrixEvent extends EventEmitter {
      */
     private txnId: string = null;
 
+    /** A reference to the event ID making the root of the thread
+     */
+    private threadRoot: string = null;
+
     /* Set an approximate timestamp for the event relative the local clock.
      * This will inherently be approximate because it doesn't take into account
      * the time between the server putting the 'age' field on the event as it sent
@@ -380,6 +384,14 @@ export class MatrixEvent extends EventEmitter {
         return this.event.content || {};
     }
 
+    /**
+     * Get the event ID of the replied event
+     */
+    public get replyEventId(): string {
+        const relations = this.getWireContent()["m.relates_to"];
+        return relations?.["m.in_reply_to"]?.["event_id"];
+    }
+
     /**
      * Get the previous event content JSON. This will only return something for
      * state events which exist in the timeline.
@@ -1265,6 +1277,14 @@ export class MatrixEvent extends EventEmitter {
     public getTxnId(): string | undefined {
         return this.txnId;
     }
+
+    public setThreadRoot(threadRoot: string): void {
+        this.threadRoot = threadRoot;
+    }
+
+    public getThreadRoot(): string {
+        return this.threadRoot;
+    }
 }
 
 /* REDACT_KEEP_KEYS gives the keys we keep when an event is redacted

src/models/room.ts

@@ -35,6 +35,7 @@ import { IRoomVersionsCapability, MatrixClient, PendingEventOrdering, RoomVersio
 import { ResizeMethod } from "../@types/partials";
 import { Filter } from "../filter";
 import { RoomState } from "./room-state";
+import { Thread } from "./thread";
 
 // These constants are used as sane defaults when the homeserver doesn't support
 // the m.room_versions capability. In practice, KNOWN_SAFE_ROOM_VERSION should be
@@ -145,6 +146,8 @@ export class Room extends EventEmitter {
     public oldState: RoomState;
     public currentState: RoomState;
 
+    private threads = new Map<string, Thread>();
+
     /**
      * Construct a new Room.
      *
@@ -1047,6 +1050,74 @@ export class Room extends EventEmitter {
             events, toStartOfTimeline,
             timeline, paginationToken,
         );
+        this.parseEventsForThread(events, timeline);
+    }
+
+    public async parseEventsForThread(
+        events: MatrixEvent[],
+        timeline: EventTimeline,
+    ): Promise<void> {
+        for (let i = events.length - 1; i >= 0; i--) {
+            if (events[i].replyEventId) {
+                const thread = this.findThreadByTailEvent(events[i].replyEventId);
+                if (thread) {
+                    thread.addEvent(events[i]);
+                } else {
+                    const replyChain = await this.getReplyChain(timeline.getTimelineSet(), [events[i]]);
+
+                    const rootId = replyChain[0].getId();
+                    const thread = this.threads.get(rootId);
+                    if (thread) {
+                        replyChain.forEach(event => thread.addEvent(event));
+                    } else {
+                        this.threads.set(
+                            rootId,
+                            new Thread(replyChain),
+                        );
+                    }
+                }
+            }
+        }
+    }
+
+    public findThreadByTailEvent(eventId: string): Thread {
+        return Array.from(this.threads.values()).find(thread => {
+            return thread.tail.has(eventId);
+        });
+    }
+
+    /**
+     * Build the reply chain starting from the bottom up
+     */
+    public async getReplyChain(
+        timelinetSet: EventTimelineSet,
+        events: MatrixEvent[] = [],
+    ): Promise<MatrixEvent[]> {
+        const parentEvent = await this.getEvent(events[0].replyEventId);
+        if (parentEvent.replyEventId) {
+            return this.getReplyChain(timelinetSet, [parentEvent, ...events]);
+        } else {
+            return [parentEvent, ...events];
+        }
+    }
+
+    /**
+     * Retrieve an event with the option to get it from the already
+     * loaded event timeline set
+     */
+    public async getEvent(eventId: string, timelineSet?: EventTimelineSet): Promise<MatrixEvent> {
+        const parentEvent = timelineSet?.findEventById(eventId);
+        if (parentEvent) {
+            return parentEvent;
+        }
+
+        const response = await this.client.http.authedRequest(
+            undefined,
+            "GET",
+            `/rooms/${this.roomId}/event/${eventId}`,
+        );
+
+        return new MatrixEvent(response);
     }
 
     /**

src/models/thread.ts

@@ -0,0 +1,101 @@
+/*
+Copyright 2021 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { EventType } from "../@types/event";
+import { MatrixEvent } from "./event";
+
+export class Thread {
+    private root: string;
+    public tail = new Set<string>();
+    private events = new Map<string, MatrixEvent>();
+    private _messageCount = 0;
+
+    constructor(events: MatrixEvent[] = []) {
+        events.forEach(event => this.addEvent(event));
+    }
+
+    /**
+     * Add an event to the thread and updates
+     * the tail/root references if needed
+     * @param event The event to add
+     */
+    public addEvent(event: MatrixEvent): void {
+        if (this.events.has(event.getId())) {
+            return;
+        }
+
+        const isRoomMessage = event.getType() === EventType.RoomMessage;
+
+        if (this.tail.has(event.replyEventId)) {
+            this.tail.delete(event.replyEventId);
+        }
+        this.tail.add(event.getId());
+
+        if (!event.replyEventId && isRoomMessage) {
+            this.root = event.getId();
+            this.events.forEach(event => event.setThreadRoot(this.root));
+        }
+
+        if (isRoomMessage) {
+            this._messageCount++;
+        }
+
+        this.events.set(event.getId(), event);
+
+        if (this.root) {
+            event.setThreadRoot(this.root);
+        }
+    }
+
+    /**
+     * A sorted list of events to display
+     */
+    public get eventTimeline(): MatrixEvent[] {
+        return Array.from(this.events.values())
+            .sort((a, b) => a.getTs() - b.getTs());
+    }
+
+    /**
+     * The thread ID, which is the same as the root event ID
+     */
+    public get id(): string {
+        return this.root;
+    }
+
+    public get rootEvent(): MatrixEvent {
+        return this.events.get(this.root);
+    }
+
+    /**
+     * The number of messages in the thread
+     */
+    public get length(): number {
+        return this._messageCount;
+    }
+
+    /**
+     * A set of mxid participating to the thread
+     */
+    public get participants(): Set<string> {
+        const participants = new Set<string>();
+        this.events.forEach(event => {
+            if (event.getType() === EventType.RoomMessage) {
+                participants.add(event.getSender());
+            }
+        });
+        return participants;
+    }
+}