fix(desktop): preserve live channel timelines

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<ReturnType<typeof getChannelWindowEvents>>,
+  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<ChannelWindowStore>(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<RelayEvent[]>(queryKey) ?? [];
       const events = await getChannelWindowEvents(channel.id);
-      const page = parseChannelWindowResponse(events, channel.id, null);
-      const current =
-        queryClient.getQueryData<ChannelWindowStore>(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(

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),

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]);
 });

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,