Improve hasUserReadEvent and getUserReadUpTo realibility with threads

spec/unit/models/thread.spec.ts

@@ -1,5 +1,5 @@
 /*
-Copyright 2022 The Matrix.org Foundation C.I.C.
+Copyright 2022 - 2023 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.
@@ -19,8 +19,12 @@ import { Room } from "../../../src/models/room";
 import { Thread, THREAD_RELATION_TYPE, ThreadEvent } from "../../../src/models/thread";
 import { mkThread } from "../../test-utils/thread";
 import { TestClient } from "../../TestClient";
-import { emitPromise, mkMessage } from "../../test-utils/test-utils";
-import { EventStatus } from "../../../src";
+import { emitPromise, mkMessage, mock } from "../../test-utils/test-utils";
+import { EventStatus, MatrixEvent } from "../../../src";
+import { ReceiptType } from "../../../src/@types/read_receipts";
+import { getMockClientWithEventEmitter, mockClientMethodsUser } from "../../test-utils/client";
+import { ReEmitter } from "../../../src/ReEmitter";
+import { Feature, ServerSupport } from "../../../src/feature";
 
 describe("Thread", () => {
     describe("constructor", () => {
@@ -71,17 +75,54 @@ describe("Thread", () => {
     });
 
     describe("hasUserReadEvent", () => {
-        const myUserId = "@bob:example.org";
+        let myUserId: string;
         let client: MatrixClient;
         let room: Room;
 
         beforeEach(() => {
-            const testClient = new TestClient(myUserId, "DEVICE", "ACCESS_TOKEN", undefined, {
-                timelineSupport: false,
+            client = getMockClientWithEventEmitter({
+                ...mockClientMethodsUser(),
+                getRoom: jest.fn().mockImplementation(() => room),
+                decryptEventIfNeeded: jest.fn().mockResolvedValue(void 0),
+                supportsExperimentalThreads: jest.fn().mockReturnValue(true),
+            });
+            client.reEmitter = mock(ReEmitter, "ReEmitter");
+            client.canSupport = new Map();
+            Object.keys(Feature).forEach((feature) => {
+                client.canSupport.set(feature as Feature, ServerSupport.Stable);
             });
-            client = testClient.client;
+
+            myUserId = client.getUserId()!;
+
             room = new Room("123", client, myUserId);
 
+            const receipt = new MatrixEvent({
+                type: "m.receipt",
+                room_id: "!foo:bar",
+                content: {
+                    // first threaded receipt
+                    "$event0:localhost": {
+                        [ReceiptType.Read]: {
+                            [client.getUserId()!]: { ts: 100, thread_id: "$threadId:localhost" },
+                        },
+                    },
+                    // last unthreaded receipt
+                    "$event1:localhost": {
+                        [ReceiptType.Read]: {
+                            [client.getUserId()!]: { ts: 200 },
+                            ["@alice:example.org"]: { ts: 200 },
+                        },
+                    },
+                    // last threaded receipt
+                    "$event2:localhost": {
+                        [ReceiptType.Read]: {
+                            [client.getUserId()!]: { ts: 300, thread_id: "$threadId" },
+                        },
+                    },
+                },
+            });
+            room.addReceipt(receipt);
+
             jest.spyOn(client, "getRoom").mockReturnValue(room);
         });
 
@@ -98,19 +139,148 @@ describe("Thread", () => {
                 length: 2,
             });
 
+            // The event is automatically considered read as the current user is the sender
             expect(thread.hasUserReadEvent(myUserId, events.at(-1)!.getId() ?? "")).toBeTruthy();
         });
 
         it("considers other events with no RR as unread", () => {
+            const { thread, events } = mkThread({
+                room,
+                client,
+                authorId: myUserId,
+                participantUserIds: [myUserId],
+                length: 25,
+                ts: 190,
+            });
+
+            // Before alice's last unthreaded receipt
+            expect(thread.hasUserReadEvent("@alice:example.org", events.at(1)!.getId() ?? "")).toBeTruthy();
+
+            // After alice's last unthreaded receipt
+            expect(thread.hasUserReadEvent("@alice:example.org", events.at(-1)!.getId() ?? "")).toBeFalsy();
+        });
+
+        it("considers event as read if there's a more recent unthreaded receipt", () => {
             const { thread, events } = mkThread({
                 room,
                 client,
                 authorId: myUserId,
                 participantUserIds: ["@alice:example.org"],
                 length: 2,
+                ts: 150, // before the latest unthreaded receipt
             });
+            expect(thread.hasUserReadEvent(client.getUserId()!, events.at(-1)!.getId() ?? "")).toBe(true);
+        });
 
-            expect(thread.hasUserReadEvent("@alice:example.org", events.at(-1)!.getId() ?? "")).toBeFalsy();
+        it("considers event as unread if there's no more recent unthreaded receipt", () => {
+            const { thread, events } = mkThread({
+                room,
+                client,
+                authorId: myUserId,
+                participantUserIds: ["@alice:example.org"],
+                length: 2,
+                ts: 1000,
+            });
+            expect(thread.hasUserReadEvent(client.getUserId()!, events.at(-1)!.getId() ?? "")).toBe(false);
+        });
+    });
+
+    describe("getEventReadUpTo", () => {
+        let myUserId: string;
+        let client: MatrixClient;
+        let room: Room;
+
+        beforeEach(() => {
+            client = getMockClientWithEventEmitter({
+                ...mockClientMethodsUser(),
+                getRoom: jest.fn().mockImplementation(() => room),
+                decryptEventIfNeeded: jest.fn().mockResolvedValue(void 0),
+                supportsExperimentalThreads: jest.fn().mockReturnValue(true),
+            });
+            client.reEmitter = mock(ReEmitter, "ReEmitter");
+            client.canSupport = new Map();
+            Object.keys(Feature).forEach((feature) => {
+                client.canSupport.set(feature as Feature, ServerSupport.Stable);
+            });
+
+            myUserId = client.getUserId()!;
+
+            room = new Room("123", client, myUserId);
+
+            jest.spyOn(client, "getRoom").mockReturnValue(room);
+        });
+
+        afterAll(() => {
+            jest.resetAllMocks();
+        });
+
+        it("uses unthreaded receipt to figure out read up to", () => {
+            const receipt = new MatrixEvent({
+                type: "m.receipt",
+                room_id: "!foo:bar",
+                content: {
+                    // last unthreaded receipt
+                    "$event1:localhost": {
+                        [ReceiptType.Read]: {
+                            ["@alice:example.org"]: { ts: 200 },
+                        },
+                    },
+                },
+            });
+            room.addReceipt(receipt);
+
+            const { thread, events } = mkThread({
+                room,
+                client,
+                authorId: myUserId,
+                participantUserIds: [myUserId],
+                length: 25,
+                ts: 190,
+            });
+
+            // The 10th event has been read, as alice's last unthreaded receipt is at ts 200
+            // and `mkThread` increment every thread response by 1ms.
+            expect(thread.getEventReadUpTo("@alice:example.org")).toBe(events.at(9)!.getId());
+        });
+
+        it("considers thread created before the first threaded receipt to be read", () => {
+            const receipt = new MatrixEvent({
+                type: "m.receipt",
+                room_id: "!foo:bar",
+                content: {
+                    // last unthreaded receipt
+                    "$event1:localhost": {
+                        [ReceiptType.Read]: {
+                            [myUserId]: { ts: 200, thread_id: "$threadId" },
+                        },
+                    },
+                },
+            });
+            room.addReceipt(receipt);
+
+            const { thread, events } = mkThread({
+                room,
+                client,
+                authorId: "@alice:example.org",
+                participantUserIds: ["@alice:example.org"],
+                length: 2,
+                ts: 10,
+            });
+
+            // This is marked as read as it is before alice's first threaded receipt...
+            expect(thread.getEventReadUpTo(myUserId)).toBe(events.at(-1)!.getId());
+
+            const { thread: thread2 } = mkThread({
+                room,
+                client,
+                authorId: "@alice:example.org",
+                participantUserIds: ["@alice:example.org"],
+                length: 2,
+                ts: 1000,
+            });
+
+            // Nothing has been read, this thread is after the first threaded receipt...
+            expect(thread2.getEventReadUpTo(myUserId)).toBe(null);
         });
     });
 });

spec/unit/notifications.spec.ts

@@ -1,5 +1,5 @@
 /*
-Copyright 2022 The Matrix.org Foundation C.I.C.
+Copyright 2022 - 2023 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.
@@ -14,6 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+import { ReceiptType } from "../../src/@types/read_receipts";
 import { Feature, ServerSupport } from "../../src/feature";
 import {
     EventType,
@@ -64,6 +65,30 @@ describe("fixNotificationCountOnDecryption", () => {
         });
 
         room = new Room(ROOM_ID, mockClient, mockClient.getUserId() ?? "");
+
+        const receipt = new MatrixEvent({
+            type: "m.receipt",
+            room_id: "!foo:bar",
+            content: {
+                "$event0:localhost": {
+                    [ReceiptType.Read]: {
+                        [mockClient.getUserId()!]: { ts: 123 },
+                    },
+                },
+                "$event1:localhost": {
+                    [ReceiptType.Read]: {
+                        [mockClient.getUserId()!]: { ts: 666, thread_id: THREAD_ID },
+                    },
+                },
+                "$otherevent:localhost": {
+                    [ReceiptType.Read]: {
+                        [mockClient.getUserId()!]: { ts: 999, thread_id: "$otherthread:localhost" },
+                    },
+                },
+            },
+        });
+        room.addReceipt(receipt);
+
         room.setUnreadNotificationCount(NotificationCountType.Total, 1);
         room.setUnreadNotificationCount(NotificationCountType.Highlight, 0);
 
@@ -75,6 +100,7 @@ describe("fixNotificationCountOnDecryption", () => {
                     body: "Hello world!",
                 },
                 event: true,
+                ts: 1234,
             },
             mockClient,
         );
@@ -90,6 +116,7 @@ describe("fixNotificationCountOnDecryption", () => {
                 "msgtype": MsgType.Text,
                 "body": "Thread reply",
             },
+            ts: 5678,
             event: true,
         });
         room.createThread(THREAD_ID, event, [threadEvent], false);
@@ -155,6 +182,7 @@ describe("fixNotificationCountOnDecryption", () => {
                 "msgtype": MsgType.Text,
                 "body": "Thread reply",
             },
+            ts: 8901,
             event: true,
         });
 

src/models/room.ts

@@ -1,5 +1,5 @@
 /*
-Copyright 2015 - 2022 The Matrix.org Foundation C.I.C.
+Copyright 2015 - 2023 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.
@@ -309,6 +309,13 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
     private notificationCounts: NotificationCount = {};
     private readonly threadNotifications = new Map<string, NotificationCount>();
     public readonly cachedThreadReadReceipts = new Map<string, CachedReceiptStructure[]>();
+    // Useful to know at what point the current user has started using threads in this room
+    private oldestThreadedReceiptTs = Infinity;
+    /**
+     * Keeping a record of the lastest unthread receipts per user
+     * This is useful in determining whether a user has read a thread or not
+     */
+    private unthreadedReceipts = new Map<string, Receipt>();
     private readonly timelineSets: EventTimelineSet[];
     public readonly threadsTimelineSets: EventTimelineSet[] = [];
     // any filtered timeline sets we're maintaining for this room
@@ -2727,6 +2734,17 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
                             { eventId, receiptType, userId, receipt, synthetic },
                         ]);
                     }
+
+                    const me = this.client.getUserId();
+                    // Track the time of the current user's oldest threaded receipt in the room.
+                    if (userId === me && !receiptForMainTimeline && receipt.ts < this.oldestThreadedReceiptTs) {
+                        this.oldestThreadedReceiptTs = receipt.ts;
+                    }
+
+                    // Track each user's unthreaded read receipt.
+                    if (!receipt.thread_id && receipt.ts > (this.unthreadedReceipts.get(userId)?.ts ?? 0)) {
+                        this.unthreadedReceipts.set(userId, receipt);
+                    }
                 });
             });
         });
@@ -3277,6 +3295,24 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
         }
         event.applyVisibilityEvent(visibilityChange);
     }
+
+    /**
+     * Find when a client has gained thread capabilities by inspecting the oldest
+     * threaded receipt
+     * @returns the timestamp of the oldest threaded receipt
+     */
+    public getOldestThreadedReceiptTs(): number {
+        return this.oldestThreadedReceiptTs;
+    }
+
+    /**
+     * Returns the most receipt unthreaded receipt for a given user
+     * @param userId - the MxID of the User
+     * @returns an unthreaded Receipt
+     */
+    public getLastUnthreadedReceiptFor(userId: string): Receipt | undefined {
+        return this.unthreadedReceipts.get(userId);
+    }
 }
 
 // a map from current event status to a list of allowed next statuses