Clear notifications when we can infer read status from receipts (#3139)

spec/integ/matrix-client-unread-notifications.spec.ts

@@ -0,0 +1,290 @@
+/*
+Copyright 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.
+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 "fake-indexeddb/auto";
+
+import HttpBackend from "matrix-mock-request";
+
+import { Category, ISyncResponse, MatrixClient, NotificationCountType, Room } from "../../src";
+import { TestClient } from "../TestClient";
+
+describe("MatrixClient syncing", () => {
+    const userA = "@alice:localhost";
+    const userB = "@bob:localhost";
+
+    const selfUserId = userA;
+    const selfAccessToken = "aseukfgwef";
+
+    let client: MatrixClient | undefined;
+    let httpBackend: HttpBackend | undefined;
+
+    const setupTestClient = (): [MatrixClient, HttpBackend] => {
+        const testClient = new TestClient(selfUserId, "DEVICE", selfAccessToken);
+        const httpBackend = testClient.httpBackend;
+        const client = testClient.client;
+        httpBackend!.when("GET", "/versions").respond(200, {});
+        httpBackend!.when("GET", "/pushrules").respond(200, {});
+        httpBackend!.when("POST", "/filter").respond(200, { filter_id: "a filter id" });
+        return [client, httpBackend];
+    };
+
+    beforeEach(() => {
+        [client, httpBackend] = setupTestClient();
+    });
+
+    afterEach(() => {
+        httpBackend!.verifyNoOutstandingExpectation();
+        client!.stopClient();
+        return httpBackend!.stop();
+    });
+
+    describe("Stuck unread notifications integration tests", () => {
+        const ROOM_ID = "!room:localhost";
+
+        const syncData = getSampleStuckNotificationSyncResponse(ROOM_ID);
+
+        it("resets notifications if the last event originates from the logged in user", async () => {
+            httpBackend!
+                .when("GET", "/sync")
+                .check((req) => {
+                    expect(req.queryParams!.filter).toEqual("a filter id");
+                })
+                .respond(200, syncData);
+
+            client!.store.getSavedSyncToken = jest.fn().mockResolvedValue("this-is-a-token");
+            client!.startClient({ initialSyncLimit: 1 });
+
+            await httpBackend!.flushAllExpected();
+
+            const room = client?.getRoom(ROOM_ID);
+
+            expect(room).toBeInstanceOf(Room);
+            expect(room?.getUnreadNotificationCount(NotificationCountType.Total)).toBe(0);
+        });
+    });
+
+    function getSampleStuckNotificationSyncResponse(roomId: string): Partial<ISyncResponse> {
+        return {
+            next_batch: "batch_token",
+            rooms: {
+                [Category.Join]: {
+                    [roomId]: {
+                        timeline: {
+                            events: [
+                                {
+                                    content: {
+                                        creator: userB,
+                                        room_version: "9",
+                                    },
+                                    origin_server_ts: 1,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.create",
+                                    event_id: "$event1",
+                                },
+                                {
+                                    content: {
+                                        avatar_url: "",
+                                        displayname: userB,
+                                        membership: "join",
+                                    },
+                                    origin_server_ts: 2,
+                                    sender: userB,
+                                    state_key: userB,
+                                    type: "m.room.member",
+                                    event_id: "$event2",
+                                },
+                                {
+                                    content: {
+                                        ban: 50,
+                                        events: {
+                                            "m.room.avatar": 50,
+                                            "m.room.canonical_alias": 50,
+                                            "m.room.encryption": 100,
+                                            "m.room.history_visibility": 100,
+                                            "m.room.name": 50,
+                                            "m.room.power_levels": 100,
+                                            "m.room.server_acl": 100,
+                                            "m.room.tombstone": 100,
+                                        },
+                                        events_default: 0,
+                                        historical: 100,
+                                        invite: 0,
+                                        kick: 50,
+                                        redact: 50,
+                                        state_default: 50,
+                                        users: {
+                                            [userA]: 100,
+                                            [userB]: 100,
+                                        },
+                                        users_default: 0,
+                                    },
+                                    origin_server_ts: 3,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.power_levels",
+                                    event_id: "$event3",
+                                },
+                                {
+                                    content: {
+                                        join_rule: "invite",
+                                    },
+                                    origin_server_ts: 4,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.join_rules",
+                                    event_id: "$event4",
+                                },
+                                {
+                                    content: {
+                                        history_visibility: "shared",
+                                    },
+                                    origin_server_ts: 5,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.history_visibility",
+                                    event_id: "$event5",
+                                },
+                                {
+                                    content: {
+                                        guest_access: "can_join",
+                                    },
+                                    origin_server_ts: 6,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.guest_access",
+                                    unsigned: {
+                                        age: 1651569,
+                                    },
+                                    event_id: "$event6",
+                                },
+                                {
+                                    content: {
+                                        algorithm: "m.megolm.v1.aes-sha2",
+                                    },
+                                    origin_server_ts: 7,
+                                    sender: userB,
+                                    state_key: "",
+                                    type: "m.room.encryption",
+                                    event_id: "$event7",
+                                },
+                                {
+                                    content: {
+                                        avatar_url: "",
+                                        displayname: userA,
+                                        is_direct: true,
+                                        membership: "invite",
+                                    },
+                                    origin_server_ts: 8,
+                                    sender: userB,
+                                    state_key: userA,
+                                    type: "m.room.member",
+                                    event_id: "$event8",
+                                },
+                                {
+                                    content: {
+                                        msgtype: "m.text",
+                                        body: "hello",
+                                    },
+                                    origin_server_ts: 9,
+                                    sender: userB,
+                                    type: "m.room.message",
+                                    event_id: "$event9",
+                                },
+                                {
+                                    content: {
+                                        avatar_url: "",
+                                        displayname: userA,
+                                        membership: "join",
+                                    },
+                                    origin_server_ts: 10,
+                                    sender: userA,
+                                    state_key: userA,
+                                    type: "m.room.member",
+                                    event_id: "$event10",
+                                },
+                                {
+                                    content: {
+                                        msgtype: "m.text",
+                                        body: "world",
+                                    },
+                                    origin_server_ts: 11,
+                                    sender: userA,
+                                    type: "m.room.message",
+                                    event_id: "$event11",
+                                },
+                            ],
+                            prev_batch: "123",
+                            limited: false,
+                        },
+                        state: {
+                            events: [],
+                        },
+                        account_data: {
+                            events: [
+                                {
+                                    type: "m.fully_read",
+                                    content: {
+                                        event_id: "$dER5V1RCMxzAhHXQJoMjqyuoxpPtK2X6hCb9T8Jg2wU",
+                                    },
+                                },
+                            ],
+                        },
+                        ephemeral: {
+                            events: [
+                                {
+                                    type: "m.receipt",
+                                    content: {
+                                        $event9: {
+                                            "m.read": {
+                                                [userA]: {
+                                                    ts: 100,
+                                                },
+                                            },
+                                            "m.read.private": {
+                                                [userA]: {
+                                                    ts: 100,
+                                                },
+                                            },
+                                        },
+                                        dER5V1RCMxzAhHXQJoMjqyuoxpPtK2X6hCb9T8Jg2wU: {
+                                            "m.read": {
+                                                [userB]: {
+                                                    ts: 666,
+                                                },
+                                            },
+                                        },
+                                    },
+                                },
+                            ],
+                        },
+                        unread_notifications: {
+                            notification_count: 1,
+                            highlight_count: 0,
+                        },
+                        summary: {
+                            "m.joined_member_count": 2,
+                            "m.invited_member_count": 0,
+                            "m.heroes": [userB],
+                        },
+                    },
+                },
+                [Category.Leave]: {},
+                [Category.Invite]: {},
+            },
+        };
+    }
+});

spec/test-utils/webrtc.ts

@@ -450,6 +450,10 @@ export class MockCallMatrixClient extends TypedEventEmitter<EmittedEvents, Emitt
         ]
     >();
 
+    public isInitialSyncComplete(): boolean {
+        return false;
+    }
+
     public getMediaHandler(): MediaHandler {
         return this.mediaHandler.typed();
     }

spec/unit/models/thread.spec.ts

@@ -82,6 +82,7 @@ describe("Thread", () => {
         beforeEach(() => {
             client = getMockClientWithEventEmitter({
                 ...mockClientMethodsUser(),
+                isInitialSyncComplete: jest.fn().mockReturnValue(false),
                 getRoom: jest.fn().mockImplementation(() => room),
                 decryptEventIfNeeded: jest.fn().mockResolvedValue(void 0),
                 supportsThreads: jest.fn().mockReturnValue(true),
@@ -193,6 +194,7 @@ describe("Thread", () => {
         beforeEach(() => {
             client = getMockClientWithEventEmitter({
                 ...mockClientMethodsUser(),
+                isInitialSyncComplete: jest.fn().mockReturnValue(false),
                 getRoom: jest.fn().mockImplementation(() => room),
                 decryptEventIfNeeded: jest.fn().mockResolvedValue(void 0),
                 supportsThreads: jest.fn().mockReturnValue(true),

spec/unit/notifications.spec.ts

@@ -53,6 +53,7 @@ describe("fixNotificationCountOnDecryption", () => {
     beforeEach(() => {
         mockClient = getMockClientWithEventEmitter({
             ...mockClientMethodsUser(),
+            isInitialSyncComplete: jest.fn().mockReturnValue(false),
             getPushActionsForEvent: jest.fn().mockReturnValue(mkPushAction(true, true)),
             getRoom: jest.fn().mockImplementation(() => room),
             decryptEventIfNeeded: jest.fn().mockResolvedValue(void 0),

spec/unit/room.spec.ts

@@ -3307,6 +3307,7 @@ describe("Room", function () {
         beforeEach(() => {
             client = getMockClientWithEventEmitter({
                 ...mockClientMethodsUser(),
+                isInitialSyncComplete: jest.fn().mockReturnValue(false),
                 supportsThreads: jest.fn().mockReturnValue(true),
             });
         });

src/client.ts

@@ -1307,6 +1307,8 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
             this.on(ClientEvent.Sync, this.startCallEventHandler);
         }
 
+        this.on(ClientEvent.Sync, this.fixupRoomNotifications);
+
         this.timelineSupport = Boolean(opts.timelineSupport);
 
         this.cryptoStore = opts.cryptoStore;
@@ -6817,6 +6819,31 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         }
     };
 
+    /**
+     * Once the client has been initialised, we want to clear notifications we
+     * know for a fact should be here.
+     * This issue should also be addressed on synapse's side and is tracked as part
+     * of https://github.com/matrix-org/synapse/issues/14837
+     *
+     * We consider a room or a thread as fully read if the current user has sent
+     * the last event in the live timeline of that context and if the read receipt
+     * we have on record matches.
+     */
+    private fixupRoomNotifications = (): void => {
+        if (this.isInitialSyncComplete()) {
+            const unreadRooms = (this.getRooms() ?? []).filter((room) => {
+                return room.getUnreadNotificationCount(NotificationCountType.Total) > 0;
+            });
+
+            for (const room of unreadRooms) {
+                const currentUserId = this.getSafeUserId();
+                room.fixupNotifications(currentUserId);
+            }
+
+            this.off(ClientEvent.Sync, this.fixupRoomNotifications);
+        }
+    };
+
     /**
      * @returns Promise which resolves: ITurnServerResponse object
      * @returns Rejects: with an error response.

src/models/read-receipt.ts

@@ -25,6 +25,7 @@ import * as utils from "../utils";
 import { MatrixEvent } from "./event";
 import { EventType } from "../@types/event";
 import { EventTimelineSet } from "./event-timeline-set";
+import { NotificationCountType } from "./room";
 
 export function synthesizeReceipt(userId: string, event: MatrixEvent, receiptType: ReceiptType): MatrixEvent {
     return new MatrixEvent({
@@ -219,6 +220,29 @@ export abstract class ReadReceipt<
 
     public abstract addReceipt(event: MatrixEvent, synthetic: boolean): void;
 
+    public abstract setUnread(type: NotificationCountType, count: number): void;
+
+    /**
+     * This issue should also be addressed on synapse's side and is tracked as part
+     * of https://github.com/matrix-org/synapse/issues/14837
+     *
+     * Retrieves the read receipt for the logged in user and checks if it matches
+     * the last event in the room and whether that event originated from the logged
+     * in user.
+     * Under those conditions we can consider the context as read. This is useful
+     * because we never send read receipts against our own events
+     * @param userId - the logged in user
+     */
+    public fixupNotifications(userId: string): void {
+        const receipt = this.getReadReceiptForUserId(userId, false);
+
+        const lastEvent = this.timeline[this.timeline.length - 1];
+        if (lastEvent && receipt?.eventId === lastEvent.getId() && userId === lastEvent.getSender()) {
+            this.setUnread(NotificationCountType.Total, 0);
+            this.setUnread(NotificationCountType.Highlight, 0);
+        }
+    }
+
     /**
      * Add a temporary local-echo receipt to the room to reflect in the
      * client the fact that we've sent one.

src/models/room.ts

@@ -1412,6 +1412,10 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
         this.emit(RoomEvent.UnreadNotifications, this.notificationCounts);
     }
 
+    public setUnread(type: NotificationCountType, count: number): void {
+        return this.setUnreadNotificationCount(type, count);
+    }
+
     public setSummary(summary: IRoomSummary): void {
         const heroes = summary["m.heroes"];
         const joinedCount = summary["m.joined_member_count"];
@@ -2762,6 +2766,21 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
                             receipt,
                             synthetic,
                         );
+
+                        // If the read receipt sent for the logged in user matches
+                        // the last event of the live timeline, then we know for a fact
+                        // that the user has read that message.
+                        // We can mark the room as read and not wait for the local echo
+                        // from synapse
+                        // This needs to be done after the initial sync as we do not want this
+                        // logic to run whilst the room is being initialised
+                        if (this.client.isInitialSyncComplete() && userId === this.client.getUserId()) {
+                            const lastEvent = receiptDestination.timeline[receiptDestination.timeline.length - 1];
+                            if (lastEvent && eventId === lastEvent.getId() && userId === lastEvent.getSender()) {
+                                receiptDestination.setUnread(NotificationCountType.Total, 0);
+                                receiptDestination.setUnread(NotificationCountType.Highlight, 0);
+                            }
+                        }
                     } else {
                         // The thread does not exist locally, keep the read receipt
                         // in a cache locally, and re-apply  the `addReceipt` logic
@@ -3374,6 +3393,29 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
     public getLastUnthreadedReceiptFor(userId: string): Receipt | undefined {
         return this.unthreadedReceipts.get(userId);
     }
+
+    /**
+     * This issue should also be addressed on synapse's side and is tracked as part
+     * of https://github.com/matrix-org/synapse/issues/14837
+     *
+     *
+     * We consider a room  fully read if the current user has sent
+     * the last event in the live timeline of that context and if the read receipt
+     * we have on record matches.
+     * This also detects all unread threads and applies the same logic to those
+     * contexts
+     */
+    public fixupNotifications(userId: string): void {
+        super.fixupNotifications(userId);
+
+        const unreadThreads = this.getThreads().filter(
+            (thread) => this.getThreadUnreadNotificationCount(thread.id, NotificationCountType.Total) > 0,
+        );
+
+        for (const thread of unreadThreads) {
+            thread.fixupNotifications(userId);
+        }
+    }
 }
 
 // a map from current event status to a list of allowed next statuses

src/models/thread.ts

@@ -22,7 +22,7 @@ import { RelationType } from "../@types/event";
 import { IThreadBundledRelationship, MatrixEvent, MatrixEventEvent } from "./event";
 import { Direction, EventTimeline } from "./event-timeline";
 import { EventTimelineSet, EventTimelineSetHandlerMap } from "./event-timeline-set";
-import { Room, RoomEvent } from "./room";
+import { NotificationCountType, Room, RoomEvent } from "./room";
 import { RoomState } from "./room-state";
 import { ServerControlledNamespacedValue } from "../NamespacedValue";
 import { logger } from "../logger";
@@ -638,6 +638,10 @@ export class Thread extends ReadReceipt<EmittedEvents, EventHandlerMap> {
 
         return super.hasUserReadEvent(userId, eventId);
     }
+
+    public setUnread(type: NotificationCountType, count: number): void {
+        return this.room.setThreadUnreadNotificationCount(this.id, type, count);
+    }
 }
 
 export const FILTER_RELATED_BY_SENDERS = new ServerControlledNamespacedValue(