diff --git a/resources/js/components/commit-thread.tsx b/resources/js/components/commit-thread.tsx index 4c36788f0..ef4f4422b 100644 --- a/resources/js/components/commit-thread.tsx +++ b/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(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(); diff --git a/resources/js/lib/commit-changes.ts b/resources/js/lib/commit-changes.ts index f82dc773e..492ac46b8 100644 --- a/resources/js/lib/commit-changes.ts +++ b/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 = { diff --git a/resources/js/lib/realtime.ts b/resources/js/lib/realtime.ts index 51e494827..3315bf94e 100644 --- a/resources/js/lib/realtime.ts +++ b/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, 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); } diff --git a/tests/Unit/Realtime.test.ts b/tests/Unit/Realtime.test.ts index a53793816..b500c417a 100644 --- a/tests/Unit/Realtime.test.ts +++ b/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(); diff --git a/tests/Unit/ThreadRecovery.test.tsx b/tests/Unit/ThreadRecovery.test.tsx index 6f059d16a..79b38223e 100644 --- a/tests/Unit/ThreadRecovery.test.tsx +++ b/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; 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((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']); +});