fix: read a reply thread once when it opens

resources/js/components/commit-thread.tsx

@@ -399,6 +399,8 @@ export function useThread(
         (lifetime.current ??= new AbortController()).signal;
     // When the replies shown were read, on the server's clock, to ask what changed since after missing updates.
     const readSince = useRef(Math.min(serverNow - 5_000, readAt()));
+    // When the thread read the replies shown, by the browser's clock, so listening catches up only on what came after.
+    const repliesReadAt = useRef<number | undefined>(undefined);
     const [isReplying, setIsReplying] = useState(false);
 
     const [previousInitial, setPreviousInitial] = useState(initial);
@@ -415,9 +417,26 @@ export function useThread(
         }
     }
 
-    const applyPage = (loaded: ThreadCommit[], cursor: string | null) => {
+    // Replies the page brought were read with it, whatever the thread read before, once they show.
+    const onInitialShown = useEffectEvent(() => {
+        readSince.current = Math.min(serverNow - 5_000, readAt());
+        repliesReadAt.current = undefined;
+    });
+    useEffect(() => {
+        if (initial !== undefined && replies === initial.data) {
+            onInitialShown();
+        }
+    }, [initial, replies]);
+
+    const applyPage = (
+        loaded: ThreadCommit[],
+        cursor: string | null,
+        startedAt?: number,
+    ) => {
         // Read over however many requests, so dated from before the first of them could have started.
-        readSince.current = readAt() - 10_000;
+        readSince.current =
+            startedAt === undefined ? readAt() - 10_000 : readAt(startedAt);
+        repliesReadAt.current = startedAt;
         setThread((current) => ({
             ...current,
             replies: loaded,
@@ -485,6 +504,7 @@ export function useThread(
                   : replies.length + extra;
 
         const request = ++replyRequest.current;
+        const startedAt = Date.now();
         setIsLoading(true);
 
         return fetchRepliesUntil(
@@ -497,7 +517,7 @@ export function useThread(
         )
             .then((page) => {
                 if (!signal.aborted && request === replyRequest.current) {
-                    applyPage(page.data, page.next_cursor);
+                    applyPage(page.data, page.next_cursor, startedAt);
                 }
             })
             .catch(() => undefined)
@@ -550,6 +570,7 @@ export function useThread(
         let isCurrent = true;
         const request = ++replyRequest.current;
         const signal = readsLifetime();
+        const startedAt = Date.now();
         setIsLoading(true);
 
         fetchRepliesUntil(
@@ -562,7 +583,7 @@ export function useThread(
         )
             .then((page) => {
                 if (isCurrent && request === replyRequest.current) {
-                    onRepliesLoaded(page.data, page.next_cursor);
+                    onRepliesLoaded(page.data, page.next_cursor, startedAt);
                 }
             })
             .catch(() => undefined)
@@ -787,6 +808,7 @@ export function useThread(
                         refresh.request();
                     }
                 }),
+            repliesReadAt.current,
         );
 
         // Muting, blocking or unmuting someone, or being blocked, changes which replies the viewer sees, including ones
@@ -819,10 +841,12 @@ export function useThread(
                       userChannel(viewerId),
                       { relationships: recoverWhenFree },
                       isPreview ? undefined : recoverWhenFree,
+                      repliesReadAt.current,
                   );
 
         return () => {
             isListening = false;
+            repliesReadAt.current = undefined;
             stop();
             stopRelationships();
             refresh.cancel();

resources/js/lib/commit-changes.ts

@@ -22,10 +22,11 @@ export function setServerTime(serverNow: number): void {
 }
 
 /**
- * Now, on the server's clock, minus the margin: when something just read was read, to ask what changed since.
+ * Now, or the given time on the browser's clock, on the server's clock minus the margin: when something read then was
+ * read, to ask what changed since.
  */
-export function readAt(): number {
-    return Date.now() + serverOffset - margin;
+export function readAt(at = Date.now()): number {
+    return at + serverOffset - margin;
 }
 
 type Waiting = {

resources/js/lib/realtime.ts

@@ -142,12 +142,14 @@ function entryFor(channel: RealtimeChannel, socket: RealtimeConnection) {
 
 /**
  * Listens for events on a channel, calling `onSync` whenever the channel is joined, at first and after every
- * reconnection, since updates sent in between were never heard. Returns the function that stops listening.
+ * reconnection, since updates sent in between were never heard. `readAt` is when what it listens for was read, if after
+ * the page. Returns the function that stops listening.
  */
 export function listen(
     channel: RealtimeChannel,
     events: Record<string, RealtimeHandler>,
     onSync?: () => void,
+    readAt?: number,
 ): () => void {
     const socket = connection;
 
@@ -186,7 +188,7 @@ export function listen(
 
         if (
             entry.isSubscribed &&
-            mayHaveMissed(entry, Object.keys(events), pageRead)
+            mayHaveMissed(entry, Object.keys(events), readAt ?? pageRead)
         ) {
             queueMicrotask(sync);
         }

tests/Unit/Realtime.test.ts

@@ -327,6 +327,33 @@ void test('a listener joining a channel listened to since before the page was re
     assert.equal(busy, 1);
 });
 
+void test('a listener for what was read after the page, since the channel was joined, catches up only on what was heard since', async () => {
+    mock.timers.enable({ apis: ['setTimeout', 'Date'], now: 1_000_000 });
+    const fake = fakeConnection();
+    connectRealtime(fake.connection);
+    pageReadAt(Date.now());
+    mock.timers.tick(1_000);
+    listen(userChannel('a'), { notifications: () => {} });
+    fake.setStatus('connected');
+    fake.joined.get('private-users.a')?.join();
+    mock.timers.tick(1_000);
+    const readAt = Date.now();
+    mock.timers.tick(500);
+    let quiet = 0;
+    let busy = 0;
+    let withThePage = 0;
+
+    listen(userChannel('a'), { relationships: () => {} }, () => quiet++, readAt);
+    fake.joined.get('private-users.a')?.emit('account');
+    listen(userChannel('a'), { account: () => {} }, () => busy++, readAt);
+    listen(userChannel('a'), { relationships: () => {} }, () => withThePage++);
+    await Promise.resolve();
+
+    assert.equal(quiet, 0);
+    assert.equal(busy, 1);
+    assert.equal(withThePage, 1, 'what came with the page was read before the channel was joined');
+});
+
 void test('a page coming back from history makes every listener joining it catch up, however late, and reads live props again', async () => {
     mock.timers.enable({ apis: ['setTimeout', 'Date'], now: 1_000_000 });
     const fake = fakeConnection();

tests/Unit/ThreadRecovery.test.tsx

@@ -25,7 +25,7 @@ vi.doMock('@inertiajs/react', async () => ({
     router: { restore: () => undefined, on: () => () => {}, reload: () => {} },
 }));
 
-const { connectRealtime } = await import('../../resources/js/lib/realtime');
+const { commitChannel, connectRealtime, listen, userChannel } = await import('../../resources/js/lib/realtime');
 const { useThread } = await import('../../resources/js/components/commit-thread');
 
 const reply = (id: string) =>
@@ -40,8 +40,8 @@ let thread: ReturnType<typeof useThread>;
 
 type Initial = { data: ThreadCommit[]; next_cursor: null; prev_cursor: null };
 
-function Probe(props: { initial: Initial }) {
-    thread = useThread(root, props.initial, true);
+function Probe(props: { initial?: Initial }) {
+    thread = useThread(root, props.initial, props.initial !== undefined);
 
     return null;
 }
@@ -98,7 +98,7 @@ async function settle() {
     }
 }
 
-let props = { initial: initialOf(['r1', 'r2']) };
+let props: { initial?: Initial } = { initial: initialOf(['r1', 'r2']) };
 
 const shownIds = () => thread.replies?.map((shown) => shown.id);
 
@@ -152,3 +152,63 @@ test('catching up on many changes reads again only what was loaded, leaving a la
     assert.equal(thread.hasMore, true);
     assert.ok(replyReads.length - before <= 2, `read ${replyReads.length - before} pages`);
 });
+
+test('opening a thread reads its replies once, though the channels it listens on were joined before', async () => {
+    props = {};
+    server = ['r1', 'r2'];
+    listen(userChannel('me'), {});
+    listen(commitChannel('root'), {});
+    joinAll(fake);
+    await vi.advanceTimersByTimeAsync(1_000);
+    void harness.render(Probe, props);
+    thread.toggle();
+    await settle();
+
+    assert.deepEqual(shownIds(), ['r1', 'r2']);
+    assert.equal(replyReads.length, 1);
+});
+
+test('a reply posted while a slow thread was opening shows once its replies land', async () => {
+    props = {};
+    server = ['r1'];
+    listen(userChannel('me'), {});
+    listen(commitChannel('root'), {});
+    joinAll(fake);
+    await vi.advanceTimersByTimeAsync(1_000);
+    const answer = globalThis.fetch;
+    let postedAt = Infinity;
+    let isHeld = false;
+    let land = () => {};
+    vi.stubGlobal('fetch', async (input: string, init?: RequestInit) => {
+        const url = new URL(input, 'http://localhost');
+
+        if (url.pathname === '/commits/changes') {
+            const replies = Number(url.searchParams.get('since')) <= postedAt ? ['r2'] : [];
+
+            return Response.json({ changes: { root: { changed: replies.length > 0, replies, truncated: false } } });
+        }
+
+        const response = answer(input, init);
+
+        if (isHeld || !url.searchParams.has('limit')) {
+            return response;
+        }
+
+        isHeld = true;
+
+        return new Promise<Response>((resolve) => {
+            land = () => resolve(response);
+        });
+    });
+    void harness.render(Probe, props);
+    thread.toggle();
+    await vi.advanceTimersByTimeAsync(1_000);
+    server = ['r1', 'r2'];
+    postedAt = Date.now();
+    fake.joined.get('commits.root')?.emit('visibility', { ids: ['r2'] });
+    await vi.advanceTimersByTimeAsync(20_000);
+    land();
+    await settle();
+
+    assert.deepEqual(shownIds(), ['r1', 'r2']);
+});