Improve hasUserReadEvent and getUserReadUpTo realibility with threads

spec/integ/matrix-client-syncing.spec.ts

@@ -1543,6 +1543,52 @@ describe("MatrixClient syncing", () => {
                 });
             });
         });
+
+        it("only replays receipts relevant to the current context", async () => {
+            const THREAD_ID = "$unknownthread:localhost";
+
+            const receipt = {
+                type: "m.receipt",
+                room_id: "!foo:bar",
+                content: {
+                    "$event1:localhost": {
+                        [ReceiptType.Read]: {
+                            "@alice:localhost": { ts: 666, thread_id: THREAD_ID },
+                        },
+                    },
+                    "$otherevent:localhost": {
+                        [ReceiptType.Read]: {
+                            "@alice:localhost": { ts: 999, thread_id: "$otherthread:localhost" },
+                        },
+                    },
+                },
+            };
+            syncData.rooms.join[roomOne].ephemeral.events = [receipt];
+
+            httpBackend!.when("GET", "/sync").respond(200, syncData);
+            client!.startClient();
+
+            return Promise.all([httpBackend!.flushAllExpected(), awaitSyncEvent()]).then(() => {
+                const room = client?.getRoom(roomOne);
+                expect(room).toBeInstanceOf(Room);
+
+                expect(room?.cachedThreadReadReceipts.has(THREAD_ID)).toBe(true);
+
+                const thread = room!.createThread(THREAD_ID, undefined, [], true);
+
+                expect(room?.cachedThreadReadReceipts.has(THREAD_ID)).toBe(false);
+
+                const receipt = thread.getReadReceiptForUserId("@alice:localhost");
+
+                expect(receipt).toStrictEqual({
+                    data: {
+                        thread_id: "$unknownthread:localhost",
+                        ts: 666,
+                    },
+                    eventId: "$event1:localhost",
+                });
+            });
+        });
     });
 
     describe("of a room", () => {

src/@types/read_receipts.ts

@@ -54,3 +54,11 @@ export type Receipts = {
         [userId: string]: [WrappedReceipt | null, WrappedReceipt | null]; // Pair<real receipt, synthetic receipt> (both nullable)
     };
 };
+
+export type CachedReceiptStructure = {
+    eventId: string;
+    receiptType: string | ReceiptType;
+    userId: string;
+    receipt: Receipt;
+    synthetic: boolean;
+};

src/client.ts

@@ -4840,13 +4840,8 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
             $eventId: event.getId()!,
         });
 
-        const supportsThreadRR = this.canSupport.get(Feature.ThreadUnreadNotifications) !== ServerSupport.Unsupported;
-        if (supportsThreadRR && !unthreaded) {
-            const isThread = !!event.threadRootId;
-            body = {
-                ...body,
-                thread_id: isThread ? event.threadRootId : MAIN_ROOM_TIMELINE,
-            };
+        if (this.supportsExperimentalThreads() && !unthreaded) {
+            body.thread_id = event.threadRootId ? event.threadRootId : MAIN_ROOM_TIMELINE;
         }
 
         const promise = this.http.authedRequest<{}>(Method.Post, path, undefined, body || {});

src/models/room.ts

@@ -54,7 +54,13 @@ import {
     FILTER_RELATED_BY_SENDERS,
     ThreadFilterType,
 } from "./thread";
-import { MAIN_ROOM_TIMELINE, Receipt, ReceiptContent, ReceiptType } from "../@types/read_receipts";
+import {
+    CachedReceiptStructure,
+    MAIN_ROOM_TIMELINE,
+    Receipt,
+    ReceiptContent,
+    ReceiptType,
+} from "../@types/read_receipts";
 import { IStateEventWithRoomId } from "../@types/search";
 import { RelationsContainer } from "./relations-container";
 import { ReadReceipt, synthesizeReceipt } from "./read-receipt";
@@ -302,7 +308,11 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
     private txnToEvent: Record<string, MatrixEvent> = {}; // Pending in-flight requests { string: MatrixEvent }
     private notificationCounts: NotificationCount = {};
     private readonly threadNotifications = new Map<string, NotificationCount>();
-    public readonly cachedThreadReadReceipts = new Map<string, { event: MatrixEvent; synthetic: boolean }[]>();
+    public readonly cachedThreadReadReceipts = new Map<string, CachedReceiptStructure[]>();
+    // Useful to know at what point the current user has started using thread in this room
+    public oldestThreadedReceiptTs = Infinity;
+    // Important to compute compute `hasUserReadEvent` and similar method correctly
+    public unthreadedReceipts = new Map<string, Receipt>();
     private readonly timelineSets: EventTimelineSet[];
     public readonly threadsTimelineSets: EventTimelineSet[] = [];
     // any filtered timeline sets we're maintaining for this room
@@ -2721,9 +2731,18 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
                         // when the thread is created
                         this.cachedThreadReadReceipts.set(receipt.thread_id!, [
                             ...(this.cachedThreadReadReceipts.get(receipt.thread_id!) ?? []),
-                            { event, synthetic },
+                            { eventId, receiptType, userId, receipt, synthetic },
                         ]);
                     }
+
+                    const me = this.client.getSafeUserId();
+                    if (userId === me && !receiptForMainTimeline && receipt.ts < this.oldestThreadedReceiptTs) {
+                        this.oldestThreadedReceiptTs = receipt.ts;
+                    }
+
+                    if (!receipt.thread_id && receipt.ts > (this.unthreadedReceipts.get(userId)?.ts ?? 0)) {
+                        this.unthreadedReceipts.set(userId, receipt);
+                    }
                 });
             });
         });

src/models/thread.ts

@@ -27,7 +27,7 @@ import { RoomState } from "./room-state";
 import { ServerControlledNamespacedValue } from "../NamespacedValue";
 import { logger } from "../logger";
 import { ReadReceipt } from "./read-receipt";
-import { Receipt, ReceiptContent, ReceiptType } from "../@types/read_receipts";
+import { CachedReceiptStructure, ReceiptType } from "../@types/read_receipts";
 
 export enum ThreadEvent {
     New = "Thread.new",
@@ -50,7 +50,7 @@ interface IThreadOpts {
     room: Room;
     client: MatrixClient;
     pendingEventOrdering?: PendingEventOrdering;
-    receipts?: { event: MatrixEvent; synthetic: boolean }[];
+    receipts?: CachedReceiptStructure[];
 }
 
 export enum FeatureSupport {
@@ -317,17 +317,9 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
      * and apply them to the current thread
      * @param receipts - A collection of the receipts cached from initial sync
      */
-    private processReceipts(receipts: { event: MatrixEvent; synthetic: boolean }[] = []): void {
-        for (const { event, synthetic } of receipts) {
-            const content = event.getContent<ReceiptContent>();
-            Object.keys(content).forEach((eventId: string) => {
-                Object.keys(content[eventId]).forEach((receiptType: ReceiptType | string) => {
-                    Object.keys(content[eventId][receiptType]).forEach((userId: string) => {
-                        const receipt = content[eventId][receiptType][userId] as Receipt;
-                        this.addReceiptToStructure(eventId, receiptType as ReceiptType, userId, receipt, synthetic);
-                    });
-                });
-            });
+    private processReceipts(receipts: CachedReceiptStructure[] = []): void {
+        for (const { eventId, receiptType, userId, receipt, synthetic } of receipts) {
+            this.addReceiptToStructure(eventId, receiptType as ReceiptType, userId, receipt, synthetic);
         }
     }
 
@@ -512,8 +504,42 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
         throw new Error("Unsupported function on the thread model");
     }
 
+    public getEventReadUpTo(userId: string, ignoreSynthesized?: boolean): string | null {
+        const isCurrentUser = userId === this.client.getUserId();
+        if (isCurrentUser && this.lastReply()) {
+            // If a thread last activity is prior the first read receipt sent in a thread
+            // we want to consider that this thread has been read
+            const beforeFirstThreadedReceipt = (this.lastReply()?.getTs() ?? 0) < this.room.oldestThreadedReceiptTs;
+            if (beforeFirstThreadedReceipt) {
+                return this.timeline.at(-1)?.getId() ?? null;
+            }
+        }
+
+        const readUpToId = super.getEventReadUpTo(userId, ignoreSynthesized);
+
+        // Checking whether the unthreaded read receipt for that user is more recent
+        // than the read receipt inside that thread.
+        if (isCurrentUser && this.lastReply()) {
+            const unthreadedReceiptTs = this.room.unthreadedReceipts.get(userId)?.ts ?? Infinity;
+            for (let i = this.timeline?.length - 1; i >= 0; --i) {
+                const ev = this.timeline[i];
+                if (ev.getTs() > unthreadedReceiptTs) return ev.getId() ?? readUpToId;
+                if (ev.getId() === readUpToId) return readUpToId;
+            }
+        }
+
+        return readUpToId;
+    }
+
     public hasUserReadEvent(userId: string, eventId: string): boolean {
         if (userId === this.client.getUserId()) {
+            const beforeFirstThreadedReceipt = (this.lastReply()?.getTs() ?? 0) < this.room.oldestThreadedReceiptTs;
+            const unthreadedReceiptTs = this.room.unthreadedReceipts.get(userId)?.ts ?? 0;
+            const beforeLastUnthreadedReceipt = (this?.lastReply()?.getTs() ?? 0) < unthreadedReceiptTs;
+            if (beforeFirstThreadedReceipt || beforeLastUnthreadedReceipt) {
+                return true;
+            }
+
             const publicReadReceipt = this.getReadReceiptForUserId(userId, false, ReceiptType.Read);
             const privateReadReceipt = this.getReadReceiptForUserId(userId, false, ReceiptType.ReadPrivate);
             const hasUnreads = this.room.getThreadUnreadNotificationCount(this.id, NotificationCountType.Total) > 0;