fix(desktop): stop dropping live events behind the reader clock

desktop/src/features/messages/hooks.ts

@@ -262,19 +262,19 @@ export function useChannelSubscription(channel: Channel | null) {
   });
 
   const appendMessage = useEffectEvent((event: RelayEvent) => {
-    if (!channelId) return;
+    if (!channelId) return false;
     if (event.kind === KIND_CHANNEL_THREAD_SUMMARY) {
       // Relay-pushed live badge recount — window-store overlay only, never a
       // timeline row (mirrors the page path, where 39005 is metadata).
       const parsed = parseLiveThreadSummary(event);
-      if (!parsed) return;
+      if (!parsed) return false;
       const windowKey = channelWindowKey(channelId);
       const current =
         queryClient.getQueryData<ChannelWindowStore>(windowKey) ??
         emptyChannelWindowStore();
       const next = mergeLiveThreadSummary(current, parsed.rootId, parsed.live);
       if (next !== current) queryClient.setQueryData(windowKey, next);
-      return;
+      return false;
     }
     const isTimelineRow = CHANNEL_TIMELINE_KINDS.has(event.kind);
     const threadReference = isTimelineRow
@@ -288,9 +288,9 @@ export function useChannelSubscription(channel: Channel | null) {
           (current = []) => mergeMessages(current, event),
         );
       }
-      if (!isBroadcastReply(event.tags)) return;
+      if (!isBroadcastReply(event.tags)) return false;
     }
-    if (!isTimelineRow && !CHANNEL_AUX_KINDS.has(event.kind)) return;
+    if (!isTimelineRow && !CHANNEL_AUX_KINDS.has(event.kind)) return false;
     if (!isTimelineRow) {
       queryClient.setQueriesData<RelayEvent[]>(
         { queryKey: ["thread-replies", channelId] },
@@ -305,7 +305,6 @@ export function useChannelSubscription(channel: Channel | null) {
     const next = mergeLiveChannelWindowEvent(current, event, isTimelineRow);
     if (next !== current) {
       queryClient.setQueryData(windowKey, next);
-      projectChannelWindowMessages(queryClient, channelId);
     }
 
     if (event.kind === KIND_SYSTEM_MESSAGE) {
@@ -328,6 +327,7 @@ export function useChannelSubscription(channel: Channel | null) {
         // Non-JSON system message — ignore.
       }
     }
+    return next !== current;
   });
 
   // Notify the relay client which channel is currently visible so its live
@@ -348,6 +348,7 @@ export function useChannelSubscription(channel: Channel | null) {
 
     let isDisposed = false;
     let cleanup: (() => Promise<void>) | undefined;
+    let shouldProjectOnFlush = false;
     const disposeReconnectListener = relayClient.subscribeToReconnects(() => {
       void refreshNewestWindow().catch((error) => {
         if (!isDisposed) {
@@ -361,11 +362,22 @@ export function useChannelSubscription(channel: Channel | null) {
     });
 
     relayClient
-      .subscribeToChannelLive(channelId, (event) => {
-        if (!isDisposed) {
-          appendMessage(event);
-        }
-      })
+      .subscribeToChannelLive(
+        channelId,
+        (event) => {
+          if (!isDisposed) {
+            shouldProjectOnFlush = appendMessage(event) || shouldProjectOnFlush;
+          }
+        },
+        {
+          onFlush: () => {
+            if (!isDisposed && shouldProjectOnFlush) {
+              projectChannelWindowMessages(queryClient, channelId);
+            }
+            shouldProjectOnFlush = false;
+          },
+        },
+      )
       .then((dispose) => {
         if (isDisposed) {
           void dispose();
@@ -394,7 +406,7 @@ export function useChannelSubscription(channel: Channel | null) {
         void cleanup();
       }
     };
-  }, [channelId, channelType]);
+  }, [channelId, channelType, queryClient]);
 }
 
 export function useSendMessageMutation(

desktop/src/shared/api/relayChannelFilters.test.mjs

@@ -4,9 +4,16 @@ import test from "node:test";
 import {
   buildChannelAuxDeletionFilter,
   buildChannelAuxFilter,
+  buildChannelLiveFilter,
   buildChannelReactionAuxFilter,
   buildChannelStructuralAuxFilter,
 } from "./relayChannelFilters.ts";
+import {
+  CHANNEL_EVENT_KINDS,
+  KIND_CHANNEL_THREAD_SUMMARY,
+} from "../constants/kinds.ts";
+import { shouldPageReconnectReplay } from "./relayReconnectReplay.ts";
+import { handleRelayClosed } from "./relayClosedRecovery.ts";
 
 const CHANNEL = "36411e44-0e2d-4cfe-bd6e-567eb169db9f";
 const IDS = [
@@ -43,3 +50,89 @@ test("buildChannelStructuralAuxFilter excludes reactions", () => {
   assert.deepEqual(filter["#e"], IDS);
   assert.equal("#h" in filter, false);
 });
+
+test("buildChannelLiveFilter is replay-bounded without a reader-clock since", () => {
+  const filter = buildChannelLiveFilter(CHANNEL);
+
+  assert.equal("since" in filter, false);
+  assert.ok(filter.limit > 0);
+  assert.equal(shouldPageReconnectReplay(filter), true);
+  assert.deepEqual(filter["#h"], [CHANNEL]);
+  assert.deepEqual(filter.kinds, [
+    ...CHANNEL_EVENT_KINDS,
+    KIND_CHANNEL_THREAD_SUMMARY,
+  ]);
+});
+
+test("CLOSED retry pages a gap larger than the live limit from last-seen author time", async () => {
+  const originalWindow = globalThis.window;
+  const originalDateNow = Date.now;
+  let retry;
+  globalThis.window = {
+    setTimeout: (callback) => {
+      retry = callback;
+      return 1;
+    },
+    clearTimeout: () => {},
+  };
+  Date.now = () => 1_100_000;
+
+  try {
+    const liveFilter = buildChannelLiveFilter(CHANNEL);
+    const recoveredEvents = Array.from(
+      { length: liveFilter.limit + 1 },
+      (_, index) => ({
+        id: String(index).padStart(64, "0"),
+        pubkey: "a".repeat(64),
+        created_at: 1_001 + index,
+        kind: 9,
+        tags: [["h", CHANNEL]],
+        content: `recovered ${index}`,
+        sig: "b".repeat(128),
+      }),
+    );
+    const delivered = [];
+    const sentFilters = [];
+    let requestedFilter;
+    let resolveReplay;
+    const replayed = new Promise((resolve) => {
+      resolveReplay = resolve;
+    });
+    const subscription = {
+      mode: "live",
+      filter: liveFilter,
+      onEvent: (event) => delivered.push(event),
+      lastSeenCreatedAt: 1_000,
+    };
+    const subscriptions = new Map([["live-closed", subscription]]);
+
+    handleRelayClosed({
+      subscriptions,
+      subId: "live-closed",
+      message: "error: temporary relay failure",
+      sendReq: async (_subId, filter) => {
+        sentFilters.push(filter);
+      },
+      requestHistory: async (filter) => {
+        requestedFilter = filter;
+        resolveReplay();
+        return recoveredEvents;
+      },
+      replaySubscriptionEvent: (_subId, event) => delivered.push(event),
+    });
+
+    assert.equal(typeof retry, "function");
+    retry();
+    await replayed;
+    await Promise.resolve();
+
+    assert.equal(sentFilters.length, 1);
+    assert.equal(sentFilters[0], liveFilter);
+    assert.equal(requestedFilter.since, 995);
+    assert.equal(requestedFilter.limit, 500);
+    assert.equal(delivered.length, liveFilter.limit + 1);
+  } finally {
+    globalThis.window = originalWindow;
+    Date.now = originalDateNow;
+  }
+});

desktop/src/shared/api/relayChannelFilters.ts

@@ -3,6 +3,7 @@ import {
   CHANNEL_EVENT_KINDS,
   CHANNEL_TIMELINE_CONTENT_KINDS,
   HOME_MENTION_EVENT_KINDS,
+  KIND_CHANNEL_THREAD_SUMMARY,
   KIND_DELETION,
   KIND_NIP29_DELETE_EVENT,
   KIND_REACTION,
@@ -17,6 +18,29 @@ import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared";
 export const AUX_BACKFILL_CHUNK_SIZE = 100;
 export const MAX_HISTORICAL_LIMIT = 10_000;
 
+/**
+ * Live window-store subscription for an open channel.
+ *
+ * Deliberately omits `since`: deriving it from the reader's clock permanently
+ * drops accepted events whose author clock is behind. Initial replay volume is
+ * bounded with `limit: 50` instead; `limit` does not constrain future fan-out.
+ * Keep the limit above zero because `shouldPageReconnectReplay` uses that as
+ * the gate for paged reconnect recovery. The authoritative HTTP window read
+ * owns history depth, so a larger WebSocket replay only adds serialized load.
+ */
+export function buildChannelLiveFilter(
+  channelId: string,
+): RelaySubscriptionFilter {
+  return {
+    // 39005 rides only this window-store subscription — not
+    // CHANNEL_EVENT_KINDS, whose other consumers (unread tracking,
+    // timeline-cache merges) must never see summary overlays.
+    kinds: [...CHANNEL_EVENT_KINDS, KIND_CHANNEL_THREAD_SUMMARY],
+    "#h": [channelId],
+    limit: 50,
+  };
+}
+
 /**
  * Live-subscription filter for an open channel: the broad
  * {@link CHANNEL_EVENT_KINDS} set so the tail delivers reactions/edits/

desktop/src/shared/api/relayClientSession.ts

@@ -9,8 +9,6 @@ import {
   KIND_STREAM_MESSAGE,
   KIND_TYPING_INDICATOR,
   KIND_USER_STATUS,
-  CHANNEL_EVENT_KINDS,
-  KIND_CHANNEL_THREAD_SUMMARY,
 } from "@/shared/constants/kinds";
 import {
   getTextPayload,
@@ -23,6 +21,7 @@ import {
   buildChannelAuxDeletionFilter,
   buildChannelFilter,
   buildChannelHistoryFilter,
+  buildChannelLiveFilter,
   buildChannelMentionFilter,
   buildGlobalStreamFilter,
 } from "@/shared/api/relayChannelFilters";
@@ -80,6 +79,8 @@ export const BACKOFF_RESET_STABLE_MS = 60_000;
 const STALL_CHECK_INTERVAL_MS = 10_000;
 const STALL_IDLE_TIMEOUT_MS = 60_000;
 
+type LiveCallbacks = Partial<Record<"onFlush", () => void>>;
+
 export class RelayClient {
   private wsId: number | null = null;
   private relayUrl: string | null = null;
@@ -356,23 +357,13 @@ export class RelayClient {
     return this.subscribe(buildChannelFilter(channelId, 50), onEvent);
   }
 
-  /** Subscribe to channel rows and aux starting now, with no history replay. */
+  /** Subscribe to channel rows and aux, with one callback per buffered flush. */
   async subscribeToChannelLive(
     channelId: string,
     onEvent: (event: RelayEvent) => void,
+    options?: LiveCallbacks,
   ) {
-    return this.subscribe(
-      {
-        // 39005 rides only this window-store subscription — not
-        // CHANNEL_EVENT_KINDS, whose other consumers (unread tracking,
-        // timeline-cache merges) must never see summary overlays.
-        kinds: [...CHANNEL_EVENT_KINDS, KIND_CHANNEL_THREAD_SUMMARY],
-        "#h": [channelId],
-        limit: 1000,
-        since: Math.floor(Date.now() / 1_000),
-      },
-      onEvent,
-    );
+    return this.subscribe(buildChannelLiveFilter(channelId), onEvent, options);
   }
 
   /**
@@ -600,6 +591,7 @@ export class RelayClient {
   private async subscribe(
     filter: RelaySubscriptionFilter,
     onEvent: (event: RelayEvent) => void,
+    options?: LiveCallbacks,
   ) {
     await this.ensureConnected();
 
@@ -622,6 +614,7 @@ export class RelayClient {
       filter,
       onEvent,
       resolveReady,
+      ...options,
     });
 
     try {
@@ -829,6 +822,8 @@ export class RelayClient {
             ["REQ", subId, filter],
             "Failed to restore relay subscription after CLOSED.",
           ),
+        requestHistory: (filter) => this.requestHistory(filter),
+        replaySubscriptionEvent: this.handleEvent.bind(this),
       });
       return;
     }
@@ -884,14 +879,19 @@ export class RelayClient {
     this.flushTimeout = null;
     const buffer = this.eventBuffer;
     this.eventBuffer = [];
+    const flushCallbacks = new Set<() => void>();
 
     // Re-lookup: subscriptions removed during batch window are intentionally skipped.
     for (const { subId, event } of buffer) {
       const subscription = this.subscriptions.get(subId);
       if (subscription?.mode === "live") {
         subscription.onEvent(event);
+        const callback = (subscription as typeof subscription & LiveCallbacks)
+          .onFlush;
+        if (callback) flushCallbacks.add(callback);
       }
     }
+    for (const callback of flushCallbacks) callback();
   }
 
   private handleEose(subId: string) {
@@ -943,14 +943,14 @@ export class RelayClient {
 
     return false;
   }
-
   private async replayLiveSubscriptions() {
     const generation = this.connectionGeneration;
     try {
       await replayLiveSubscriptions({
         subscriptions: this.subscriptions,
         sendRaw: (payload) => this.sendRaw(payload),
         requestHistory: (filter) => this.requestHistory(filter),
+        replaySubscriptionEvent: this.handleEvent.bind(this),
         visibleChannelId: this.visibleChannelId,
         isActive: () => this.connectionGeneration === generation,
       });