Optimize desktop channel live subscriptions

Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>

desktop/src/features/messages/hooks.ts

@@ -17,6 +17,12 @@ import {
 } from "@/features/messages/lib/threading";
 import { splitOutgoingTags } from "@/features/messages/lib/imetaMediaMarkdown";
 import { relayClient } from "@/shared/api/relayClient";
+import { buildChannelLiveFilter } from "@/shared/api/relayChannelFilters";
+import {
+  frontierBridge,
+  newestEventFrontier,
+  type EventFrontier,
+} from "@/shared/api/relayFrontierBridge";
 import { customEmojiQueryKey } from "@/features/custom-emoji/hooks";
 import { reactionEmojiUrl } from "@/shared/api/customEmoji";
 import type { CustomEmoji } from "@/shared/lib/remarkCustomEmoji";
@@ -269,23 +275,56 @@ export function useChannelSubscription(channel: Channel | null) {
   const queryClient = useQueryClient();
   const channelId = channel?.id ?? null;
   const channelType = channel?.channelType ?? null;
-  const syncLatestHistory = useEffectEvent(async () => {
-    if (!channelId) {
-      return;
-    }
+  const bridgeFromFrontier = useEffectEvent(
+    async (
+      capturedFrontier?: EventFrontier | null,
+      isActive: () => boolean = () => true,
+    ) => {
+      if (!channelId) {
+        return;
+      }
 
-    const history = await relayClient.fetchChannelHistory(
-      channelId,
-      CHANNEL_HISTORY_LIMIT,
-    );
+      const queryKey = channelMessagesKey(channelId);
+      const current = queryClient.getQueryData<RelayEvent[]>(queryKey) ?? [];
+      const frontier = capturedFrontier ?? newestEventFrontier(current);
 
-    queryClient.setQueryData<RelayEvent[]>(
-      channelMessagesKey(channelId),
-      (current = []) => mergeTimelineHistoryMessages(current, history),
-    );
+      if (!frontier) {
+        const history = await relayClient.fetchChannelHistory(
+          channelId,
+          CHANNEL_HISTORY_LIMIT,
+        );
+        if (!isActive()) {
+          return;
+        }
+        queryClient.setQueryData<RelayEvent[]>(queryKey, (latest = []) =>
+          mergeTimelineHistoryMessages(latest, history),
+        );
+        void backfillAuxForMessages(queryClient, channelId, history);
+        return;
+      }
 
-    void backfillAuxForMessages(queryClient, channelId, history);
-  });
+      const bridgedEvents: RelayEvent[] = [];
+      await frontierBridge({
+        frontier,
+        isActive,
+        knownEventIds: new Set(current.map((event) => event.id)),
+        targetFilter: buildChannelLiveFilter(channelId),
+        requestHistory: (filter) => relayClient.fetchEvents(filter),
+        onEvent: (event) => {
+          bridgedEvents.push(event);
+        },
+      });
+
+      if (bridgedEvents.length === 0 || !isActive()) {
+        return;
+      }
+
+      queryClient.setQueryData<RelayEvent[]>(queryKey, (latest = []) =>
+        mergeTimelineHistoryMessages(latest, bridgedEvents),
+      );
+      void backfillAuxForMessages(queryClient, channelId, bridgedEvents);
+    },
+  );
 
   const appendMessage = useEffectEvent((event: RelayEvent) => {
     if (!channelId) {
@@ -327,10 +366,10 @@ export function useChannelSubscription(channel: Channel | null) {
     let isDisposed = false;
     let cleanup: (() => Promise<void>) | undefined;
     const disposeReconnectListener = relayClient.subscribeToReconnects(() => {
-      void syncLatestHistory().catch((error) => {
+      void bridgeFromFrontier(undefined, () => !isDisposed).catch((error) => {
         if (!isDisposed) {
           console.error(
-            "Failed to refresh channel history after reconnecting",
+            "Failed to bridge channel history after reconnecting",
             channelId,
             error,
           );
@@ -338,6 +377,11 @@ export function useChannelSubscription(channel: Channel | null) {
       });
     });
 
+    const queryKey = channelMessagesKey(channelId);
+    const subscriptionFrontier = newestEventFrontier(
+      queryClient.getQueryData<RelayEvent[]>(queryKey) ?? [],
+    );
+
     relayClient
       .subscribeToChannel(channelId, (event) => {
         if (!isDisposed) {
@@ -351,15 +395,21 @@ export function useChannelSubscription(channel: Channel | null) {
         }
 
         cleanup = dispose;
-        // No post-subscribe history refetch: useChannelMessagesQuery already
-        // loaded the latest CHANNEL_HISTORY_LIMIT events, and the live
-        // subscription itself backfills up to 50 most-recent events via its
-        // initial REQ (buildChannelFilter(id, 50)). Both write into the same
-        // channelMessagesKey cache, so any window between the two REQs is
-        // covered by the live sub's overlap unless >50 messages land in
-        // <1s — vanishingly rare in practice. The reconnect listener above
-        // still bridges gaps from connection drops, where the gap *is*
-        // unbounded.
+        // The live sub is registered with limit:0, so channel switches no
+        // longer replay the same 50 historical events that the query path just
+        // loaded. Once EOSE confirms the live sub is installed, explicitly
+        // bridge from the cache frontier to close the history→live race.
+        void bridgeFromFrontier(subscriptionFrontier, () => !isDisposed).catch(
+          (error) => {
+            if (!isDisposed) {
+              console.error(
+                "Failed to bridge channel history",
+                channelId,
+                error,
+              );
+            }
+          },
+        );
       })
       .catch((error) => {
         console.error("Failed to subscribe to channel", channelId, error);
@@ -372,7 +422,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,6 +4,7 @@ import test from "node:test";
 import {
   buildChannelAuxDeletionFilter,
   buildChannelAuxFilter,
+  buildChannelLiveFilter,
   buildChannelReactionAuxFilter,
   buildChannelStructuralAuxFilter,
 } from "./relayChannelFilters.ts";
@@ -43,3 +44,9 @@ test("buildChannelStructuralAuxFilter excludes reactions", () => {
   assert.deepEqual(filter["#e"], IDS);
   assert.equal("#h" in filter, false);
 });
+
+test("buildChannelLiveFilter subscribes with limit 0 to avoid historical replay", () => {
+  const filter = buildChannelLiveFilter(CHANNEL);
+  assert.equal(filter.limit, 0);
+  assert.deepEqual(filter["#h"], [CHANNEL]);
+});

desktop/src/shared/api/relayChannelFilters.ts

@@ -40,6 +40,12 @@ export function buildChannelFilter(
   return filter;
 }
 
+export function buildChannelLiveFilter(
+  channelId: string,
+): RelaySubscriptionFilter {
+  return buildChannelFilter(channelId, 0);
+}
+
 /**
  * History filter for cold-load and scrollback: message kinds *only*, so the
  * `limit` budget buys visible message depth. Auxiliary events (reactions,

desktop/src/shared/api/relayClientSession.ts

@@ -22,8 +22,8 @@ import {
 import {
   AUX_BACKFILL_CHUNK_SIZE,
   buildChannelAuxDeletionFilter,
-  buildChannelFilter,
   buildChannelHistoryFilter,
+  buildChannelLiveFilter,
   buildChannelMentionFilter,
   buildGlobalStreamFilter,
 } from "@/shared/api/relayChannelFilters";
@@ -330,7 +330,7 @@ export class RelayClient {
     channelId: string,
     onEvent: (event: RelayEvent) => void,
   ) {
-    return this.subscribe(buildChannelFilter(channelId, 50), onEvent);
+    return this.subscribe(buildChannelLiveFilter(channelId), onEvent);
   }
 
   /**

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

@@ -0,0 +1,145 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  buildFrontierBridgeFilter,
+  frontierBridge,
+  isEventAfterFrontier,
+  newestEventFrontier,
+  shouldApplyBridgeEvent,
+} from "./relayFrontierBridge.ts";
+import { buildChannelFilter } from "./relayChannelFilters.ts";
+import { mergeTimelineHistoryMessages } from "../../features/messages/lib/messageQueryKeys.ts";
+
+function event(id, createdAt, kind = 9) {
+  return {
+    id,
+    pubkey: "pubkey",
+    created_at: createdAt,
+    kind,
+    tags: [["h", "channel-1"]],
+    content: "",
+    sig: "sig",
+  };
+}
+
+function verifyBridge({ before, bridgePage, atomic }) {
+  const frontier = newestEventFrontier(before);
+  const delivered = [];
+
+  return frontierBridge({
+    frontier,
+    targetFilter: buildChannelFilter("channel-1", 0),
+    knownEventIds: new Set(before.map((ev) => ev.id)),
+    requestHistory: async () => bridgePage,
+    onEvent: (ev) => delivered.push(ev),
+  }).then(() => {
+    const bridged = mergeTimelineHistoryMessages(before, delivered);
+    const fresh = mergeTimelineHistoryMessages(before, atomic);
+    assert.deepEqual(
+      bridged.map((ev) => ev.id),
+      fresh.map((ev) => ev.id),
+    );
+  });
+}
+
+test("newestEventFrontier uses the composite (created_at, id) frontier", () => {
+  const frontier = newestEventFrontier([
+    event("b", 10),
+    event("a", 11),
+    event("c", 11),
+  ]);
+
+  assert.deepEqual(frontier, { createdAt: 11, eventId: "c" });
+});
+
+test("frontier bridge filter keeps since inclusive for same-second ties", () => {
+  const targetFilter = { ...buildChannelFilter("channel-1", 0), since: 90 };
+  const filter = buildFrontierBridgeFilter({
+    frontier: { createdAt: 100, eventId: "b" },
+    targetFilter,
+  });
+
+  assert.equal(filter.limit, 500);
+  assert.equal(filter.since, 100);
+  assert.deepEqual(filter["#h"], ["channel-1"]);
+  assert.deepEqual(filter.kinds, targetFilter.kinds);
+});
+
+test("frontier comparison orders dense-second events by id", () => {
+  const frontier = { createdAt: 100, eventId: "b" };
+
+  assert.equal(isEventAfterFrontier(event("a", 100), frontier), false);
+  assert.equal(isEventAfterFrontier(event("b", 100), frontier), false);
+  assert.equal(isEventAfterFrontier(event("c", 100), frontier), true);
+  assert.equal(isEventAfterFrontier(event("a", 101), frontier), true);
+});
+
+test("bridge application keeps unknown same-second events that sort before the frontier", () => {
+  const frontier = { createdAt: 100, eventId: "m" };
+  const knownEventIds = new Set(["m"]);
+
+  assert.equal(
+    shouldApplyBridgeEvent({
+      event: event("a", 100),
+      frontier,
+      knownEventIds,
+    }),
+    true,
+  );
+  assert.equal(
+    shouldApplyBridgeEvent({
+      event: event("m", 100),
+      frontier,
+      knownEventIds,
+    }),
+    false,
+  );
+});
+
+test("verify_bridge: live-only bridge equals fresh atomic fetch across dense-second and aux events", async () => {
+  const before = [event("a", 100), event("b", 100), event("m", 101)];
+  const afterSameSecondLowerId = event("l", 101);
+  const afterSameSecond = event("z", 101);
+  const afterNextSecond = event("n", 102);
+  const aux = event("aux", 103, 7);
+  const duplicateAtFrontier = event("m", 101);
+
+  await verifyBridge({
+    before,
+    // Relay history arrives newest-first; the bridge must sort/apply into the
+    // same timeline cache projection as a fresh atomic read.
+    bridgePage: [
+      aux,
+      afterNextSecond,
+      duplicateAtFrontier,
+      afterSameSecond,
+      afterSameSecondLowerId,
+    ],
+    atomic: [
+      duplicateAtFrontier,
+      afterSameSecondLowerId,
+      afterSameSecond,
+      afterNextSecond,
+      aux,
+    ],
+  });
+});
+
+test("frontierBridge stops applying when the owner is disposed", async () => {
+  const delivered = [];
+  let active = true;
+
+  await frontierBridge({
+    frontier: { createdAt: 100, eventId: "a" },
+    targetFilter: buildChannelFilter("channel-1", 0),
+    requestHistory: async () => {
+      active = false;
+      return [event("b", 101)];
+    },
+    isActive: () => active,
+    onEvent: (ev) => delivered.push(ev),
+  });
+
+  assert.deepEqual(delivered, []);
+});

desktop/src/shared/api/relayFrontierBridge.ts

@@ -0,0 +1,135 @@
+import { sortEvents } from "@/shared/api/relayClientShared";
+import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared";
+import type { RelayEvent } from "@/shared/api/types";
+
+export type EventFrontier = {
+  createdAt: number;
+  eventId: string;
+};
+
+export const FRONTIER_BRIDGE_PAGE_LIMIT = 500;
+
+export function compareEventToFrontier(
+  event: RelayEvent,
+  frontier: EventFrontier,
+) {
+  if (event.created_at !== frontier.createdAt) {
+    return event.created_at - frontier.createdAt;
+  }
+  return event.id.localeCompare(frontier.eventId);
+}
+
+export function isEventAfterFrontier(
+  event: RelayEvent,
+  frontier: EventFrontier,
+) {
+  return compareEventToFrontier(event, frontier) > 0;
+}
+
+export function shouldApplyBridgeEvent({
+  event,
+  frontier,
+  knownEventIds,
+}: {
+  event: RelayEvent;
+  frontier: EventFrontier;
+  knownEventIds: ReadonlySet<string>;
+}) {
+  if (event.created_at > frontier.createdAt) {
+    return true;
+  }
+
+  // NIP-01 `since` is second-granularity. Keep unknown events in the frontier
+  // second even if their id sorts before the frontier id; they may have landed
+  // in the history→live race and a strict `(created_at,id) > frontier` filter
+  // would silently drop them. Known ids are harmless duplicates.
+  return (
+    event.created_at === frontier.createdAt && !knownEventIds.has(event.id)
+  );
+}
+
+export function newestEventFrontier(
+  events: RelayEvent[],
+): EventFrontier | null {
+  let frontier: EventFrontier | null = null;
+
+  for (const event of events) {
+    if (!frontier || isEventAfterFrontier(event, frontier)) {
+      frontier = {
+        createdAt: event.created_at,
+        eventId: event.id,
+      };
+    }
+  }
+
+  return frontier;
+}
+
+export function buildFrontierBridgeFilter({
+  frontier,
+  limit = FRONTIER_BRIDGE_PAGE_LIMIT,
+  targetFilter,
+}: {
+  frontier: EventFrontier;
+  limit?: number;
+  targetFilter: RelaySubscriptionFilter;
+}): RelaySubscriptionFilter {
+  return {
+    ...targetFilter,
+    limit,
+    since:
+      targetFilter.since === undefined
+        ? frontier.createdAt
+        : Math.max(targetFilter.since, frontier.createdAt),
+  };
+}
+
+/**
+ * Fetch and apply the explicit gap bridge after a live subscription is ready.
+ *
+ * The relay's NIP-01 `since` cursor is second-granularity, so the request is
+ * intentionally inclusive at `frontier.createdAt`; the client then drops known
+ * duplicates and keeps unknown frontier-second events too. That closes the
+ * history→live race even when a new same-second event sorts before the current
+ * frontier id, while normal id-dedupe keeps duplicate delivery harmless.
+ */
+export async function frontierBridge({
+  frontier,
+  isActive = () => true,
+  knownEventIds,
+  limit,
+  onEvent,
+  requestHistory,
+  targetFilter,
+}: {
+  frontier: EventFrontier | null;
+  isActive?: () => boolean;
+  limit?: number;
+  knownEventIds?: ReadonlySet<string>;
+  onEvent: (event: RelayEvent) => void;
+  requestHistory: (filter: RelaySubscriptionFilter) => Promise<RelayEvent[]>;
+  targetFilter: RelaySubscriptionFilter;
+}) {
+  if (!frontier) {
+    return;
+  }
+
+  const knownIds = knownEventIds ?? new Set<string>();
+
+  const events = await requestHistory(
+    buildFrontierBridgeFilter({ frontier, limit, targetFilter }),
+  );
+
+  if (!isActive()) {
+    return;
+  }
+
+  for (const event of sortEvents(events).filter((event) =>
+    shouldApplyBridgeEvent({ event, frontier, knownEventIds: knownIds }),
+  )) {
+    if (!isActive()) {
+      return;
+    }
+    onEvent(event);
+  }
+}