Extend logic for local notification processing to threads (matrix-org…

…#4111)

* Move code for processing our own receipts to Room

This is some code to process our own receipts and recalculate our
notification counts.

There was no reason for this to be in client. Room is still rather
large, but at least it makes somewhat more sense there.

Moving as a refactor before I start work on it.

* Add test for the client-side e2e notifications code

* Extend logic for local notification processing to threads

There's collection of logic for for processing receipts and recomputing
notifications for encrypted rooms, but we didn't do the same for threads.
As a reasult, when I tried pulling some of the logic over in
matrix-org#4106
clearing notifications on threads just broke.

This extends the logic of reprocessing local notifications when a receipt
arrives to threads.

Based on matrix-org#4109

* simplify object literal

* Add tests & null guard

* Remove unused imports

* Add another skipped test

* Unused import

* enable tests

* Fix thread support nightmare

* Try this way

* Unused import

* Comment the bear trap

* expand comment

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

@@ -39,6 +39,7 @@ import {
     IndexedDBStore,
     RelationType,
     EventType,
+    MatrixEventEvent,
 } from "../../src";
 import { ReceiptType } from "../../src/@types/read_receipts";
 import { UNREAD_THREAD_NOTIFICATIONS } from "../../src/@types/sync";
@@ -1800,7 +1801,7 @@ describe("MatrixClient syncing", () => {
                 expect(room.getRoomUnreadNotificationCount(NotificationCountType.Total)).toBe(0);
             });
 
-            it("should recalculate highlights on receipt for encrypted rooms", async () => {
+            it("should recalculate highlights on unthreaded receipt for encrypted rooms", async () => {
                 const myUserId = client!.getUserId()!;
 
                 const firstEventId = syncData.rooms.join[roomId].timeline.events[1].event_id;
@@ -1850,6 +1851,235 @@ describe("MatrixClient syncing", () => {
                 // the room should now have one highlight since our receipt was before the ping message
                 expect(room.getRoomUnreadNotificationCount(NotificationCountType.Highlight)).toBe(1);
             });
+
+            it("should recalculate highlights on main thread receipt for encrypted rooms", async () => {
+                const myUserId = client!.getUserId()!;
+
+                const firstEventId = syncData.rooms.join[roomId].timeline.events[1].event_id;
+
+                // add a receipt for the first event in the room (let's say the user has already read that one)
+                syncData.rooms.join[roomId].ephemeral.events = [
+                    {
+                        content: {
+                            [firstEventId]: {
+                                "m.read": {
+                                    [myUserId]: { ts: 1, thread_id: "main" },
+                                },
+                            },
+                        },
+                        type: "m.receipt",
+                    },
+                ];
+
+                // Now add a highlighting event after that receipt
+                const pingEvent = utils.mkMessage({
+                    room: roomId,
+                    user: otherUserId,
+                    msg: client?.getUserId() + " ping",
+                }) as IRoomEvent;
+                syncData.rooms.join[roomId].timeline.events.push(pingEvent);
+
+                // fudge this to make it a highlight
+                client!.getPushActionsForEvent = (ev: MatrixEvent): IActionsObject | null => {
+                    if (ev.getId() === pingEvent.event_id) {
+                        return {
+                            notify: true,
+                            tweaks: {
+                                highlight: true,
+                            },
+                        };
+                    }
+                    return null;
+                };
+
+                httpBackend!.when("GET", "/sync").respond(200, syncData);
+                client!.startClient();
+
+                await Promise.all([httpBackend!.flushAllExpected(), awaitSyncEvent()]);
+
+                const room = client!.getRoom(roomId)!;
+                expect(room).toBeInstanceOf(Room);
+                // the room should now have one highlight since our receipt was before the ping message
+                expect(room.getRoomUnreadNotificationCount(NotificationCountType.Highlight)).toBe(1);
+            });
+
+            describe("notification processing in threads", () => {
+                let threadEvent1: IRoomEvent;
+                let threadEvent2: IRoomEvent;
+                let firstEventId: string;
+
+                beforeEach(() => {
+                    firstEventId = syncData.rooms.join[roomId].timeline.events[1].event_id;
+
+                    // Add a threaded event off of the first event
+                    threadEvent1 = utils.mkEvent({
+                        type: EventType.RoomMessage,
+                        user: otherUserId,
+                        room: roomId,
+                        ts: 500,
+                        content: {
+                            "body": "first thread response",
+                            "m.relates_to": {
+                                "event_id": firstEventId,
+                                "m.in_reply_to": {
+                                    event_id: firstEventId,
+                                },
+                                "rel_type": "io.element.thread",
+                            },
+                        },
+                    }) as IRoomEvent;
+                    syncData.rooms.join[roomId].timeline.events.push(threadEvent1);
+
+                    // ...and another
+                    threadEvent2 = utils.mkEvent({
+                        type: EventType.RoomMessage,
+                        user: otherUserId,
+                        room: roomId,
+                        ts: 1500,
+                        content: {
+                            "body": "second thread response",
+                            "m.relates_to": {
+                                "event_id": firstEventId,
+                                "m.in_reply_to": {
+                                    event_id: firstEventId,
+                                },
+                                "rel_type": "io.element.thread",
+                            },
+                        },
+                    }) as IRoomEvent;
+                    syncData.rooms.join[roomId].timeline.events.push(threadEvent2);
+
+                    // fudge to make these highlights
+                    client!.getPushActionsForEvent = (ev: MatrixEvent): IActionsObject | null => {
+                        if ([threadEvent1.event_id, threadEvent2.event_id].includes(ev.getId()!)) {
+                            return {
+                                notify: true,
+                                tweaks: {
+                                    highlight: true,
+                                },
+                            };
+                        }
+                        return null;
+                    };
+                });
+
+                it("checks threads with notifications on unthreaded receipts", async () => {
+                    const myUserId = client!.getUserId()!;
+
+                    // add a receipt for a random, ficticious thread, otherwise the client will
+                    // think that the thread is before any threaded receipts and ignore it.
+                    syncData.rooms.join[roomId].ephemeral.events = [
+                        {
+                            content: {
+                                [firstEventId]: {
+                                    "m.read": {
+                                        [myUserId]: { ts: 1, thread_id: "some_other_thread" },
+                                    },
+                                },
+                            },
+                            type: "m.receipt",
+                        },
+                    ];
+
+                    httpBackend!.when("GET", "/sync").respond(200, syncData);
+                    client!.startClient({ threadSupport: true });
+
+                    await Promise.all([httpBackend!.flushAllExpected(), awaitSyncEvent()]);
+
+                    const room = client!.getRoom(roomId)!;
+
+                    // pretend that the client has decrypted an event to trigger it to compute
+                    // local notifications
+                    client?.emit(MatrixEventEvent.Decrypted, room.findEventById(firstEventId)!);
+                    client?.emit(MatrixEventEvent.Decrypted, room.findEventById(threadEvent1.event_id)!);
+                    client?.emit(MatrixEventEvent.Decrypted, room.findEventById(threadEvent2.event_id)!);
+
+                    expect(room).toBeInstanceOf(Room);
+
+                    // we should now have one highlight: the unread message that pings
+                    expect(
+                        room.getThreadUnreadNotificationCount(firstEventId, NotificationCountType.Highlight),
+                    ).toEqual(2);
+
+                    const syncData2 = {
+                        rooms: {
+                            join: {
+                                [roomId]: {
+                                    ephemeral: {
+                                        events: [
+                                            {
+                                                content: {
+                                                    [firstEventId]: {
+                                                        "m.read": {
+                                                            [myUserId]: { ts: 1 },
+                                                        },
+                                                    },
+                                                },
+                                                type: "m.receipt",
+                                            },
+                                        ],
+                                    },
+                                },
+                            },
+                        },
+                    } as unknown as ISyncResponse;
+
+                    httpBackend!.when("GET", "/sync").respond(200, syncData2);
+
+                    await Promise.all([httpBackend!.flush("/sync", 1), utils.syncPromise(client!)]);
+
+                    expect(room.getRoomUnreadNotificationCount(NotificationCountType.Highlight)).toBe(0);
+                });
+
+                it("should recalculate highlights on threaded receipt for encrypted rooms", async () => {
+                    const myUserId = client!.getUserId()!;
+
+                    // add a receipt for the first message in the threadm leaving the second one unread
+                    syncData.rooms.join[roomId].ephemeral.events = [
+                        {
+                            content: {
+                                [threadEvent1.event_id]: {
+                                    "m.read": {
+                                        [myUserId]: { ts: 1, thread_id: firstEventId },
+                                    },
+                                },
+                            },
+                            type: "m.receipt",
+                        },
+                    ];
+
+                    // fudge to make both thread replies highlights
+                    client!.getPushActionsForEvent = (ev: MatrixEvent): IActionsObject | null => {
+                        if ([threadEvent1.event_id, threadEvent2.event_id].includes(ev.getId()!)) {
+                            return {
+                                notify: true,
+                                tweaks: {
+                                    highlight: true,
+                                },
+                            };
+                        }
+                        return null;
+                    };
+
+                    httpBackend!.when("GET", "/sync").respond(200, syncData);
+                    client!.startClient({ threadSupport: true });
+
+                    await Promise.all([httpBackend!.flushAllExpected(), awaitSyncEvent()]);
+
+                    const room = client!.getRoom(roomId)!;
+                    expect(room).toBeInstanceOf(Room);
+
+                    // pretend that the client has decrypted an event to trigger it to compute
+                    // local notifications
+                    client?.emit(MatrixEventEvent.Decrypted, room.findEventById(firstEventId)!);
+
+                    // the room should now have one highlight: the second thread message
+
+                    expect(room.getThreadUnreadNotificationCount(firstEventId, NotificationCountType.Highlight)).toBe(
+                        1,
+                    );
+                });
+            });
         });
     });
 

src/@types/event.ts

@@ -136,6 +136,10 @@ export enum RelationType {
     Annotation = "m.annotation",
     Replace = "m.replace",
     Reference = "m.reference",
+
+    // Don't use this yet: it's only the stable version. The code still assumes we support the unstable prefix and,
+    // moreover, our tests currently use the unstable prefix. Use THREAD_RELATION_TYPE.name.
+    // Once we support *only* the stable prefix, THREAD_RELATION_TYPE can die and we can switch to this.
     Thread = "m.thread",
 }
 

src/models/room.ts

@@ -1323,46 +1323,84 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
         // This fixes https://github.com/vector-im/element-web/issues/9421
 
         // Figure out if we've read something or if it's just informational
+        // We need to work out what threads we've just recieved receipts for, so we
+        // know which ones to update. If we've received an unthreaded receipt, we'll
+        // need to update all threads.
+        let threadIds: string[] = [];
+        let hasUnthreadedReceipt = false;
+
         const content = event.getContent();
-        const isSelf =
-            Object.keys(content).filter((eid) => {
-                for (const [key, value] of Object.entries(content[eid])) {
-                    if (!utils.isSupportedReceiptType(key)) continue;
-                    if (!value) continue;
 
-                    if (Object.keys(value).includes(this.client.getUserId()!)) return true;
+        for (const receiptGroup of Object.values(content)) {
+            for (const [receiptType, userReceipt] of Object.entries(receiptGroup)) {
+                if (!utils.isSupportedReceiptType(receiptType)) continue;
+                if (!userReceipt) continue;
+
+                for (const [userId, singleReceipt] of Object.entries(userReceipt)) {
+                    if (!singleReceipt || typeof singleReceipt !== "object") continue;
+                    const typedSingleReceipt = singleReceipt as Record<string, any>;
+                    if (userId !== this.client.getUserId()) continue;
+                    if (typedSingleReceipt.thread_id === undefined) {
+                        hasUnthreadedReceipt = true;
+                    } else if (typeof typedSingleReceipt.thread_id === "string") {
+                        threadIds.push(typedSingleReceipt.thread_id);
+                    }
                 }
+            }
+        }
 
-                return false;
-            }).length > 0;
+        if (hasUnthreadedReceipt) {
+            // If we have an unthreaded receipt, we need to update any threads that have a notification
+            // in them (because we know the receipt can't go backwards so we don't need to check any with
+            // no notifications: the number can only decrease from a receipt).
+            threadIds = this.getThreads()
+                .filter(
+                    (thread) =>
+                        this.getThreadUnreadNotificationCount(thread.id, NotificationCountType.Total) > 0 ||
+                        this.getThreadUnreadNotificationCount(thread.id, NotificationCountType.Highlight) > 0,
+                )
+                .map((thread) => thread.id);
+            threadIds.push("main");
+        }
+
+        for (const threadId of threadIds) {
+            // Work backwards to determine how many events are unread. We also set
+            // a limit for how back we'll look to avoid spinning CPU for too long.
+            // If we hit the limit, we assume the count is unchanged.
+            const maxHistory = 20;
+            const timeline = threadId === "main" ? this.getLiveTimeline() : this.getThread(threadId)?.liveTimeline;
+
+            if (!timeline) {
+                logger.warn(`Couldn't find timeline for thread ID ${threadId} in room ${this.roomId}`);
+                continue;
+            }
 
-        if (!isSelf) return;
+            const events = timeline.getEvents();
 
-        // Work backwards to determine how many events are unread. We also set
-        // a limit for how back we'll look to avoid spinning CPU for too long.
-        // If we hit the limit, we assume the count is unchanged.
-        const maxHistory = 20;
-        const events = this.getLiveTimeline().getEvents();
+            let highlightCount = 0;
 
-        let highlightCount = 0;
+            for (let i = events.length - 1; i >= 0; i--) {
+                if (i === events.length - maxHistory) return; // limit reached
 
-        for (let i = events.length - 1; i >= 0; i--) {
-            if (i === events.length - maxHistory) return; // limit reached
+                const event = events[i];
 
-            const event = events[i];
+                if (this.hasUserReadEvent(this.client.getUserId()!, event.getId()!)) {
+                    // If the user has read the event, then the counting is done.
+                    break;
+                }
 
-            if (this.hasUserReadEvent(this.client.getUserId()!, event.getId()!)) {
-                // If the user has read the event, then the counting is done.
-                break;
+                const pushActions = this.client.getPushActionsForEvent(event);
+                highlightCount += pushActions?.tweaks?.highlight ? 1 : 0;
             }
 
-            const pushActions = this.client.getPushActionsForEvent(event);
-            highlightCount += pushActions?.tweaks?.highlight ? 1 : 0;
+            // Note: we don't need to handle 'total' notifications because the counts
+            // will come from the server.
+            if (threadId === "main") {
+                this.setUnreadNotificationCount(NotificationCountType.Highlight, highlightCount);
+            } else {
+                this.setThreadUnreadNotificationCount(threadId, NotificationCountType.Highlight, highlightCount);
+            }
         }
-
-        // Note: we don't need to handle 'total' notifications because the counts
-        // will come from the server.
-        this.setUnreadNotificationCount(NotificationCountType.Highlight, highlightCount);
     }
 
     /**

src/models/thread.ts

@@ -709,7 +709,7 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
      * Return last reply to the thread, if known.
      */
     public lastReply(
-        matches: (ev: MatrixEvent) => boolean = (ev): boolean => ev.isRelation(RelationType.Thread),
+        matches: (ev: MatrixEvent) => boolean = (ev): boolean => ev.isRelation(THREAD_RELATION_TYPE.name),
     ): MatrixEvent | null {
         for (let i = this.timeline.length - 1; i >= 0; i--) {
             const event = this.timeline[i];