Fetch events that read receipts point to if we don't have them

spec/unit/models/thread.spec.ts

@@ -775,6 +775,63 @@ describe("Thread", () => {
             });
         });
     });
+
+    describe("checkForMissingReceiptEvent", () => {
+        let previousThreadHasServerSideSupport: FeatureSupport;
+
+        beforeAll(() => {
+            previousThreadHasServerSideSupport = Thread.hasServerSideSupport;
+            Thread.hasServerSideSupport = FeatureSupport.Stable;
+            jest.useFakeTimers();
+        });
+
+        afterAll(() => {
+            Thread.hasServerSideSupport = previousThreadHasServerSideSupport;
+            jest.useRealTimers();
+        });
+
+        it("Looks up missing events referenced by read receipts", async () => {
+            const client = createClient();
+            const user = "@alice:matrix.org";
+            const roomId = "!test:room";
+            jest.spyOn(client, "getUserId").mockReturnValue(user);
+
+            const thread = await createThread(client, user, roomId);
+
+            const reactionToRoot = mkMessage({
+                event: true,
+                user,
+                room: roomId,
+                relatesTo: {
+                    event_id: thread.id,
+                    rel_type: "m.annotation",
+                    key: "",
+                },
+            });
+            const reactionEventId = reactionToRoot.getId()!;
+            await thread.room.addLiveEvents([reactionToRoot]);
+
+            thread.addReceiptToStructure(
+                reactionEventId,
+                ReceiptType.Read,
+                client.getUserId()!,
+                { ts: 0, thread_id: thread.id },
+                false,
+            );
+
+            const message1 = createThreadMessage(thread.id, user, roomId, "message1");
+            await thread.addEvent(message1, true);
+
+            // This will be called anyway by addEvent but it won't wait for it to complete,
+            // so we gut-wrench to call it again manually such that we can wait for it and
+            // not race.
+            await (thread as any).updateThreadMetadata();
+
+            jest.advanceTimersByTime(1000);
+
+            expect(client.fetchRoomEvent).toHaveBeenCalledWith(roomId, reactionEventId);
+        });
+    });
 });
 
 /**

spec/unit/utils/throttle.spec.ts

@@ -0,0 +1,96 @@
+import { throttle } from "../../../src/utils/throttle";
+
+/*
+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.
+*/
+
+describe("throttle", () => {
+    let mockFn: jest.Mock;
+    let throttledFn: ReturnType<typeof throttle>;
+
+    beforeEach(() => {
+        mockFn = jest.fn();
+        throttledFn = throttle(mockFn, 100);
+
+        jest.useFakeTimers();
+    });
+
+    afterEach(() => {
+        jest.clearAllMocks();
+
+        jest.useRealTimers();
+    });
+
+    it("should throttle multiple successive calls to a single call", () => {
+        throttledFn();
+        throttledFn();
+        throttledFn();
+
+        expect(mockFn).toHaveBeenCalledTimes(1);
+    });
+
+    it("should execute the function each time the delay elapses", () => {
+        throttledFn();
+        // should execute here (leading edge)
+        expect(mockFn).toHaveBeenCalledTimes(1);
+
+        jest.advanceTimersByTime(40);
+        throttledFn();
+        // delay hasn't elapsed yet: don't execute
+        expect(mockFn).toHaveBeenCalledTimes(1);
+
+        jest.advanceTimersByTime(40);
+        throttledFn();
+        // still hasn't elapsed
+        expect(mockFn).toHaveBeenCalledTimes(1);
+
+        jest.advanceTimersByTime(40);
+        throttledFn();
+        // delay has now elapsed
+        expect(mockFn).toHaveBeenCalledTimes(2);
+    });
+
+    it("should execute the function on leading & trailing edge", () => {
+        // call it twice...
+        throttledFn();
+        throttledFn();
+        // This should have executed the first call but not the second
+        expect(mockFn).toHaveBeenCalledTimes(1);
+
+        jest.advanceTimersByTime(110);
+        // now the second call should have been executed
+        expect(mockFn).toHaveBeenCalledTimes(2);
+    });
+
+    it("should only execute once if the function is called once", () => {
+        // call it once...
+        throttledFn();
+        // This should have executed
+        expect(mockFn).toHaveBeenCalledTimes(1);
+
+        jest.advanceTimersByTime(110);
+        // still should only have executed once
+        expect(mockFn).toHaveBeenCalledTimes(1);
+    });
+
+    it("should pass arguments to the throttled function", () => {
+        const arg1 = "arg1";
+        const arg2 = 123;
+
+        throttledFn(arg1, arg2);
+
+        expect(mockFn).toHaveBeenCalledWith(arg1, arg2);
+    });
+});

src/models/read-receipt.ts

@@ -217,7 +217,7 @@ export abstract class ReadReceipt<
         return false;
     }
 
-    private getLatestReceipt(userId: string, ignoreSynthesized: boolean): WrappedReceipt | null {
+    protected getLatestReceipt(userId: string, ignoreSynthesized: boolean): WrappedReceipt | null {
         // XXX: This is very very ugly and I hope I won't have to ever add a new
         // receipt type here again. IMHO this should be done by the server in
         // some more intelligent manner or the client should just use timestamps

src/models/thread.ts

@@ -29,6 +29,7 @@ import { logger } from "../logger";
 import { ReadReceipt } from "./read-receipt";
 import { CachedReceiptStructure, Receipt, ReceiptType } from "../@types/read_receipts";
 import { Feature, ServerSupport } from "../feature";
+import { throttle } from "../utils/throttle";
 
 export enum ThreadEvent {
     New = "Thread.new",
@@ -140,6 +141,8 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
      */
     public replayEvents: MatrixEvent[] | null = [];
 
+    private isCheckingForMissingReceiptEvent = false;
+
     public constructor(public readonly id: string, public rootEvent: MatrixEvent | undefined, opts: IThreadOpts) {
         super();
 
@@ -624,6 +627,8 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
             }
         }
 
+        this.checkForMissingReceiptEvent();
+
         this.emit(ThreadEvent.Update, this);
     }
 
@@ -656,6 +661,49 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
         }
     }
 
+    /**
+     * Workaround for servers that don't support MSC3981: if we can't see the event that the threaded
+     * read receipt is pointing to, fetch it by event ID so that we can use it in our unread calculation.
+     * This is mainly for situations where the read receipt is pointing to a relation event that we don't
+     * have because it's too far back in the scrollback for us to have paginated it and it doesn't come down
+     * with the thread events due to lack of MSC3981 support.
+     *
+     * If you are reading this in the future and MSC3981 is merged and widely supported then you, future
+     * space person, get the honour of building a large pyre to incinerate this piece of code and revel
+     * in the delicious Typescript fumes.
+     */
+    private checkForMissingReceiptEvent = throttle((): void => {
+        if (this.isCheckingForMissingReceiptEvent) return;
+        this.isCheckingForMissingReceiptEvent = true;
+
+        (async (): Promise<void> => {
+            try {
+                const myThreadedReceipt = this.getLatestReceipt(this.client.getUserId()!, true);
+                // We check there are messages in the timeline as it doesn't really make sense to check for missing
+                // events if there are no events in the thread at all (yet).
+                if (myThreadedReceipt && this.timeline.length > 0 && !this.findEventById(myThreadedReceipt.eventId)) {
+                    logger.info(
+                        `Found threaded read receipt in thread ${this.id} referencing unknown event ${myThreadedReceipt.eventId}: attempting to fetch event`,
+                    );
+
+                    try {
+                        const rawEvent = await this.client.fetchRoomEvent(this.roomId, myThreadedReceipt.eventId);
+                        const ev = new MatrixEvent(rawEvent);
+                        await this.client.decryptEventIfNeeded(ev);
+                        this.insertEventIntoTimeline(ev);
+                        logger.info(
+                            `Found and inserted event for read receipt: ${myThreadedReceipt.eventId} in thread ${this.id}`,
+                        );
+                    } catch (e) {
+                        logger.warn("Failed to fetch event referenced by read receipt", e);
+                    }
+                }
+            } finally {
+                this.isCheckingForMissingReceiptEvent = false;
+            }
+        })();
+    }, 1000);
+
     public setEventMetadata(event: Optional<MatrixEvent>): void {
         if (event) {
             EventTimeline.setEventMetadata(event, this.roomState, false);

src/utils/throttle.ts

@@ -0,0 +1,48 @@
+/*
+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.
+*/
+
+type ThrottleFunction = (...args: any[]) => void;
+
+/**
+ * Throttles a function to only be called once per `delay` milliseconds.
+ * Simple, dependency-free replacement for lodash's throttle. This version executes
+ * on the leading and trailing edge.
+ * @param fn - The function to throttle.
+ * @param delay - The delay in milliseconds.
+ * @returns The throttled function.
+ */
+export function throttle(fn: ThrottleFunction, delay: number): ThrottleFunction {
+    let lastExecutionTime = 0;
+    let timeout: ReturnType<typeof setTimeout>;
+
+    return (...args: any[]) => {
+        const currentTime = Date.now();
+
+        if (currentTime - lastExecutionTime < delay) {
+            if (timeout !== undefined) {
+                clearTimeout(timeout);
+            }
+
+            timeout = setTimeout(() => {
+                lastExecutionTime = currentTime;
+                fn(...args);
+            }, delay);
+        } else {
+            lastExecutionTime = currentTime;
+            fn(...args);
+        }
+    };
+}