From 7de31f53a4e8a4e228df04f2babe57b6bff194a3 Mon Sep 17 00:00:00 2001 From: Wes Date: Wed, 12 Aug 2026 08:30:46 -0600 Subject: [PATCH] fix(desktop): preserve live channel timelines Restore the post-subscribe refresh that closes the gap left by the live subscription's now-based cursor. Also keep a pageless window projection from replacing a populated render cache with its first live overlay event. Co-authored-by: Carl Signed-off-by: Wes --- desktop/src/features/messages/hooks.ts | 55 +++++-- .../lib/channelWindowReconciliation.ts | 12 ++ .../lib/projectChannelWindow.test.mjs | 152 +++++++++--------- .../messages/lib/projectChannelWindow.ts | 28 ---- 4 files changed, 127 insertions(+), 120 deletions(-) diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 9091121d0bf..e59cd72a919 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -1,5 +1,10 @@ import { useEffect, useEffectEvent } from "react"; -import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { + type QueryClient, + useMutation, + useQuery, + useQueryClient, +} from "@tanstack/react-query"; import { toast } from "sonner"; import { @@ -17,7 +22,6 @@ import { import { projectChannelWindowMessages, refreshChannelWindowMessages, - shouldRefreshChannelWindowAfterSubscribe, } from "@/features/messages/lib/projectChannelWindow"; import { reconcileChannelWindowMessages } from "@/features/messages/lib/channelWindowReconciliation"; import { @@ -235,26 +239,46 @@ export function useChannelWindowQuery(channel: Channel | null) { }); } +export function reconcileFetchedChannelWindow( + queryClient: QueryClient, + channelId: string, + events: Awaited>, + previousMessages: RelayEvent[], + signal: AbortSignal, +): RelayEvent[] { + // Tauri invokes cannot be canceled after dispatch. A replacement refetch can + // therefore win while this older request is still in flight. Never let that + // canceled request commit its stale page into the authoritative window. + signal.throwIfAborted(); + const windowKey = channelWindowKey(channelId); + const page = parseChannelWindowResponse(events, channelId, null); + const current = + queryClient.getQueryData(windowKey) ?? + emptyChannelWindowStore(); + const next = replaceNewestChannelWindow(current, page); + queryClient.setQueryData(windowKey, next); + return reconcileChannelWindowMessages(next, previousMessages); +} + export function useChannelMessagesQuery(channel: Channel | null) { const queryClient = useQueryClient(); const queryKey = channelMessagesKey(channel?.id ?? "none"); - const windowKey = channelWindowKey(channel?.id ?? "none"); return useQuery({ enabled: channel !== null && channel.channelType !== "forum", queryKey, - queryFn: async () => { + queryFn: async ({ signal }) => { if (!channel) throw new Error("No channel selected."); const previousMessages = queryClient.getQueryData(queryKey) ?? []; const events = await getChannelWindowEvents(channel.id); - const page = parseChannelWindowResponse(events, channel.id, null); - const current = - queryClient.getQueryData(windowKey) ?? - emptyChannelWindowStore(); - const next = replaceNewestChannelWindow(current, page); - queryClient.setQueryData(windowKey, next); - return reconcileChannelWindowMessages(next, previousMessages); + return reconcileFetchedChannelWindow( + queryClient, + channel.id, + events, + previousMessages, + signal, + ); }, staleTime: 5 * 60 * 1_000, gcTime: 60 * 60 * 1_000, @@ -382,9 +406,10 @@ export function useChannelSubscription(channel: Channel | null) { } cleanup = dispose; - if (!shouldRefreshChannelWindowAfterSubscribe(queryClient, channelId)) { - return; - } + // The live subscription starts at "now", so it cannot close the gap + // between the last page snapshot and subscription establishment. Always + // refresh after the subscription is active; freshness alone is not a + // proof that no relay events landed in that interval. void refreshNewestWindow().catch((error) => { if (!isDisposed) { console.error( @@ -406,7 +431,7 @@ export function useChannelSubscription(channel: Channel | null) { void cleanup(); } }; - }, [channelId, channelType, queryClient]); + }, [channelId, channelType]); } export function useSendMessageMutation( diff --git a/desktop/src/features/messages/lib/channelWindowReconciliation.ts b/desktop/src/features/messages/lib/channelWindowReconciliation.ts index cc2c0f034c8..f6a7e4df21f 100644 --- a/desktop/src/features/messages/lib/channelWindowReconciliation.ts +++ b/desktop/src/features/messages/lib/channelWindowReconciliation.ts @@ -28,6 +28,18 @@ export function reconcileChannelWindowMessages( messages: RelayEvent[], ) { const windowEvents = flattenChannelWindowEvents(window); + if (window.pages.length === 0) { + // A pageless window is unresolved, not authoritative. This state can exist + // briefly when the companion window query mounts beside an already-cached + // rendered timeline. Preserve that cache while admitting live events; + // otherwise the first live event projects a one-row overlay over the + // entire conversation until reload refetches page zero. + let merged = messages; + for (const event of windowEvents) { + merged = reconcileIncomingMessage(merged, event); + } + return [...merged].sort((left, right) => compareRelayOrder(right, left)); + } const authoritativeIds = new Set(windowEvents.map((event) => event.id)); const retained = retainRefetchReconciliationEvents(messages).filter( (event) => !authoritativeIds.has(event.id), diff --git a/desktop/src/features/messages/lib/projectChannelWindow.test.mjs b/desktop/src/features/messages/lib/projectChannelWindow.test.mjs index 2ca2354271e..14ec110addf 100644 --- a/desktop/src/features/messages/lib/projectChannelWindow.test.mjs +++ b/desktop/src/features/messages/lib/projectChannelWindow.test.mjs @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import { QueryClient, QueryObserver } from "@tanstack/react-query"; +import { reconcileFetchedChannelWindow } from "../hooks.ts"; import { channelMessagesKey, channelWindowKey } from "./messageQueryKeys.ts"; import { appendOlderChannelWindow, @@ -11,10 +12,8 @@ import { replaceNewestChannelWindow, } from "./channelWindowStore.ts"; import { - CHANNEL_WINDOW_FRESH_MS, projectChannelWindowMessages, refreshChannelWindowMessages, - shouldRefreshChannelWindowAfterSubscribe, } from "./projectChannelWindow.ts"; import { reconcileChannelWindowMessages } from "./channelWindowReconciliation.ts"; @@ -30,6 +29,18 @@ function event(id, createdAt) { }; } +function wirePage(rows) { + return [ + ...rows, + { + ...event("bounds", 0), + kind: 39006, + tags: [["d", "channel:head"]], + content: JSON.stringify({ has_more: false, next_cursor: null }), + }, + ]; +} + function newestPage(rows) { return { startCursor: null, @@ -276,92 +287,79 @@ test("test_live_projection_retains_pending_send_and_non_broadcast_thread_reply", ]); }); -test("test_subscribe_refresh_skips_fresh_populated_window", () => { +test("test_canceled_stale_fetch_cannot_overwrite_catch_up_window", async () => { const harness = createHarness(); - const updatedAt = harness.client.getQueryState( - harness.windowKey, - ).dataUpdatedAt; - - assert.equal( - shouldRefreshChannelWindowAfterSubscribe( - harness.client, - harness.channelId, - updatedAt + CHANNEL_WINDOW_FRESH_MS - 1, - ), - false, - ); -}); - -test("test_subscribe_refresh_runs_for_stale_window", () => { - const harness = createHarness(); - const updatedAt = harness.client.getQueryState( - harness.windowKey, - ).dataUpdatedAt; + const requests = []; + let resolveRequestStarted; + let requestStarted = new Promise((resolve) => { + resolveRequestStarted = resolve; + }); + const observer = new QueryObserver(harness.client, { + queryKey: harness.messagesKey, + queryFn: async ({ signal }) => { + const previousMessages = harness.client.getQueryData(harness.messagesKey); + let resolveFetch; + const fetch = new Promise((resolve) => { + resolveFetch = resolve; + }); + requests.push({ resolveFetch, signal }); + resolveRequestStarted(); + const events = await fetch; + return reconcileFetchedChannelWindow( + harness.client, + harness.channelId, + events, + previousMessages, + signal, + ); + }, + }); + const unsubscribe = observer.subscribe(() => {}); - assert.equal( - shouldRefreshChannelWindowAfterSubscribe( - harness.client, - harness.channelId, - updatedAt + CHANNEL_WINDOW_FRESH_MS, - ), - true, + await requestStarted; + requestStarted = new Promise((resolve) => { + resolveRequestStarted = resolve; + }); + const catchUp = refreshChannelWindowMessages( + harness.client, + harness.channelId, ); -}); + await requestStarted; -test("test_live_cache_merge_does_not_extend_window_freshness", () => { - const harness = createHarness(); - const windowUpdatedAt = harness.client.getQueryState( - harness.windowKey, - ).dataUpdatedAt; - - harness.client.setQueryData(harness.messagesKey, (messages) => [ - ...messages, - event("live-cache-only", 110), - ]); - - assert.equal( - shouldRefreshChannelWindowAfterSubscribe( - harness.client, - harness.channelId, - windowUpdatedAt + CHANNEL_WINDOW_FRESH_MS, - ), - true, + assert.equal(requests[0].signal.aborted, true); + requests[1].resolveFetch( + wirePage([event("gap", 110), event("initial", 100)]), ); -}); + await catchUp; + assert.deepEqual(contents(harness), ["initial", "gap"]); -test("test_subscribe_refresh_runs_without_a_message_query", () => { - const client = new QueryClient({ - defaultOptions: { queries: { retry: false } }, - }); + requests[0].resolveFetch(wirePage([event("initial", 100)])); + await new Promise((resolve) => setImmediate(resolve)); + appendLiveEvent(harness, event("live", 120)); - assert.equal( - shouldRefreshChannelWindowAfterSubscribe(client, "missing-channel"), - true, + assert.deepEqual(contents(harness), ["initial", "gap", "live"]); + assert.deepEqual( + flattenChannelWindowEvents( + harness.client.getQueryData(harness.windowKey), + ).map((item) => item.content), + ["initial", "gap", "live"], ); + unsubscribe(); }); -test("test_subscribe_refresh_does_not_duplicate_inflight_initial_fetch", async () => { - const client = new QueryClient({ - defaultOptions: { queries: { retry: false } }, - }); - const channelId = "pending-channel"; - const queryKey = channelMessagesKey(channelId); - let resolveFetch; - const observer = new QueryObserver(client, { - queryKey, - queryFn: () => - new Promise((resolve) => { - resolveFetch = resolve; - }), - }); - const unsubscribe = observer.subscribe(() => {}); +test("test_pageless_live_projection_preserves_cached_timeline", () => { + const harness = createHarness(); + const cached = harness.client.getQueryData(harness.messagesKey); + const pageless = emptyChannelWindowStore(); + harness.client.setQueryData(harness.windowKey, pageless); - assert.equal( - shouldRefreshChannelWindowAfterSubscribe(client, channelId), - false, + const next = mergeLiveChannelWindowEvent( + harness.client.getQueryData(harness.windowKey), + event("live", 110), ); + harness.client.setQueryData(harness.windowKey, next); + projectChannelWindowMessages(harness.client, harness.channelId); - resolveFetch([]); - await client.getQueryCache().find({ queryKey })?.promise; - unsubscribe(); + assert.deepEqual(contents(harness), ["initial", "live"]); + assert.equal(harness.client.getQueryData(harness.messagesKey)[0], cached[0]); }); diff --git a/desktop/src/features/messages/lib/projectChannelWindow.ts b/desktop/src/features/messages/lib/projectChannelWindow.ts index 2d56c096b6c..81ef3de42d0 100644 --- a/desktop/src/features/messages/lib/projectChannelWindow.ts +++ b/desktop/src/features/messages/lib/projectChannelWindow.ts @@ -8,34 +8,6 @@ import { } from "./channelWindowStore"; import { reconcileChannelWindowMessages } from "./channelWindowReconciliation"; -export const CHANNEL_WINDOW_FRESH_MS = 5 * 60_000; - -/** - * Subscription setup closes the gap between the initial page and live events, - * but revisiting a channel with a fresh page has no gap to close. Reconnects - * still refresh unconditionally at their call site. - */ -export function shouldRefreshChannelWindowAfterSubscribe( - queryClient: QueryClient, - channelId: string, - now = Date.now(), -): boolean { - const messagesState = queryClient.getQueryState( - channelMessagesKey(channelId), - ); - if (!messagesState) return true; - if (messagesState.fetchStatus === "fetching") return false; - const windowState = queryClient.getQueryState(channelWindowKey(channelId)); - if ( - messagesState.status !== "success" || - windowState?.status !== "success" || - windowState.dataUpdatedAt === 0 - ) { - return true; - } - return now - windowState.dataUpdatedAt >= CHANNEL_WINDOW_FRESH_MS; -} - /** Keep the rendered timeline cache aligned with its authoritative window. */ export function projectChannelWindowMessages( queryClient: QueryClient,