Batch useAuthor and useEventStats to reduce concurrent REQs

Both hooks now collect requests within a 50ms window and flush them as a single relay query, matching the pattern used in Agora. useAuthor: Instead of N individual kind:0 queries (one per post author), all pubkeys within a 50ms window are batched into a single query with authors: [pk1, pk2, ...]. Results are seeded into the individual ['author', pubkey] cache entries. useEventStats: Instead of 2 REQs per post (e-tag + q-tag), all event IDs within a 50ms window are batched into 2 total queries. Results are grouped by referenced event ID and seeded into individual caches. Also restored the correct reqRouter pattern (all read relays with eoseTimeout race) matching Agora's approach. Co-authored-by: shakespeare.diy <assistant@shakespeare.diy>
2026-02-18 00:21:28 -06:00
parent 9ca82c3431
commit 7eafd14ae3
3 changed files with 252 additions and 59 deletions
@@ -44,7 +44,8 @@ const NostrProvider: React.FC<NostrProviderProps> = (props) => {
      reqRouter(filters: NostrFilter[]) {
        const routes = new Map<string, NostrFilter[]>();

-        // Route to all read relays
+        // Route to all read relays — eoseTimeout races them so the
+        // first relay to finish wins while the rest are cut off.
        const readRelays = effectiveRelays.current.relays
          .filter(r => r.read)
          .map(r => r.url);
@@ -56,14 +57,12 @@ const NostrProvider: React.FC<NostrProviderProps> = (props) => {
        return routes;
      },
      eventRouter(_event: NostrEvent) {
-        // Get write relays from effective relays
+        // Publish to all write relays for maximum reach
        const writeRelays = effectiveRelays.current.relays
          .filter(r => r.write)
          .map(r => r.url);

-        const allRelays = new Set<string>(writeRelays);
-
-        return [...allRelays];
+        return [...new Set(writeRelays)];
      },
      // Resolve queries quickly once any relay sends EOSE, instead of
      // waiting for every relay to finish. This is the single biggest
@@ -1,34 +1,130 @@
 import { type NostrEvent, type NostrMetadata, NSchema as n } from '@nostrify/nostrify';
 import { useNostr } from '@nostrify/react';
-import { useQuery } from '@tanstack/react-query';
+import { useQuery, useQueryClient } from '@tanstack/react-query';
+
+// ---------------------------------------------------------------------------
+// Batching layer — collects individual pubkey requests within a 50 ms window
+// and resolves them all with a single relay query.
+// ---------------------------------------------------------------------------
+
+type NostrPool = ReturnType<typeof useNostr>['nostr'];
+
+interface PendingRequest {
+  resolve: (data: { event?: NostrEvent; metadata?: NostrMetadata }) => void;
+  reject: (err: Error) => void;
+}
+
+/** Map of pubkey → pending resolvers, flushed every 50 ms. */
+const pendingBatch = new Map<string, PendingRequest[]>();
+let flushTimer: ReturnType<typeof setTimeout> | null = null;
+let currentPool: NostrPool | null = null;
+let currentQueryClient: ReturnType<typeof useQueryClient> | null = null;
+
+function flushBatch() {
+  flushTimer = null;
+  const pool = currentPool;
+  const qc = currentQueryClient;
+  if (!pool || !qc) return;
+
+  // Snapshot & clear
+  const batch = new Map(pendingBatch);
+  pendingBatch.clear();
+
+  const pubkeys = [...batch.keys()];
+  if (pubkeys.length === 0) return;
+
+  (async () => {
+    try {
+      const events = await pool.query(
+        [{ kinds: [0], authors: pubkeys, limit: pubkeys.length }],
+        { signal: AbortSignal.timeout(5000) },
+      );
+
+      // Index results by pubkey
+      const byPubkey = new Map<string, NostrEvent>();
+      for (const event of events) {
+        const existing = byPubkey.get(event.pubkey);
+        if (!existing || event.created_at > existing.created_at) {
+          byPubkey.set(event.pubkey, event);
+        }
+      }
+
+      // Resolve each pending request
+      for (const [pk, resolvers] of batch) {
+        const event = byPubkey.get(pk);
+        let result: { event?: NostrEvent; metadata?: NostrMetadata };
+
+        if (event) {
+          let metadata: NostrMetadata | undefined;
+          try {
+            metadata = n.json().pipe(n.metadata()).parse(event.content);
+          } catch {
+            // unparseable
+          }
+          result = { event, metadata };
+        } else {
+          result = {};
+        }
+
+        // Seed individual cache
+        qc.setQueryData(['author', pk], result);
+
+        for (const r of resolvers) {
+          r.resolve(result);
+        }
+      }
+    } catch (err) {
+      // Reject all pending
+      for (const resolvers of batch.values()) {
+        for (const r of resolvers) {
+          r.reject(err instanceof Error ? err : new Error(String(err)));
+        }
+      }
+    }
+  })();
+}
+
+function fetchAuthorBatched(
+  pubkey: string,
+  pool: NostrPool,
+  queryClient: ReturnType<typeof useQueryClient>,
+): Promise<{ event?: NostrEvent; metadata?: NostrMetadata }> {
+  // Keep refs current so the flush uses the latest pool
+  currentPool = pool;
+  currentQueryClient = queryClient;
+
+  return new Promise((resolve, reject) => {
+    const existing = pendingBatch.get(pubkey);
+    if (existing) {
+      existing.push({ resolve, reject });
+    } else {
+      pendingBatch.set(pubkey, [{ resolve, reject }]);
+    }
+
+    // Schedule a flush if not already pending
+    if (!flushTimer) {
+      flushTimer = setTimeout(flushBatch, 50);
+    }
+  });
+}
+
+// ---------------------------------------------------------------------------
+// Hook
+// ---------------------------------------------------------------------------

 export function useAuthor(pubkey: string | undefined) {
  const { nostr } = useNostr();
+  const queryClient = useQueryClient();

  return useQuery<{ event?: NostrEvent; metadata?: NostrMetadata }>({
    queryKey: ['author', pubkey ?? ''],
-    queryFn: async ({ signal }) => {
+    queryFn: async () => {
      if (!pubkey) {
        return {};
      }
-
-      const [event] = await nostr.query(
-        [{ kinds: [0], authors: [pubkey!], limit: 1 }],
-        { signal: AbortSignal.any([signal, AbortSignal.timeout(1500)]) },
-      );
-
-      if (!event) {
-        throw new Error('No event found');
-      }
-
-      try {
-        const metadata = n.json().pipe(n.metadata()).parse(event.content);
-        return { metadata, event };
-      } catch {
-        return { event };
-      }
+      return fetchAuthorBatched(pubkey, nostr, queryClient);
    },
-    staleTime: 5 * 60 * 1000, // Keep cached data fresh for 5 minutes
-    retry: 3,
+    staleTime: 5 * 60 * 1000,
+    retry: 1,
  });
 }
@@ -1,5 +1,5 @@
 import { useNostr } from '@nostrify/react';
-import { useQuery } from '@tanstack/react-query';
+import { useQuery, useQueryClient } from '@tanstack/react-query';
 import type { NostrEvent } from '@nostrify/nostrify';

 export interface TrendingTag {
@@ -117,67 +117,165 @@ function parseBolt11Amount(bolt11: string): number {
  }
 }

-/** Counts engagement (replies, reposts, quotes, reactions, zaps) for a given event. */
-export function useEventStats(eventId: string | undefined) {
-  const { nostr } = useNostr();
+// ---------------------------------------------------------------------------
+// Batched event stats — collects individual event ID requests within a 50 ms
+// window and resolves them all with a single pair of relay queries.
+// ---------------------------------------------------------------------------

-  return useQuery({
-    queryKey: ['event-stats', eventId ?? ''],
-    queryFn: async ({ signal }) => {
-      if (!eventId) return { replies: 0, reposts: 0, quotes: 0, reactions: 0, zapAmount: 0, reactionEmojis: [] as string[] };
+type NostrPool = ReturnType<typeof useNostr>['nostr'];

-      const timeout = AbortSignal.timeout(5000);
-      const combined = AbortSignal.any([signal, timeout]);
+export interface EventStats {
+  replies: number;
+  reposts: number;
+  quotes: number;
+  reactions: number;
+  zapAmount: number;
+  reactionEmojis: string[];
+}

-      // Two queries: one for e-tag interactions, one for q-tag quotes
+const EMPTY_STATS: EventStats = { replies: 0, reposts: 0, quotes: 0, reactions: 0, zapAmount: 0, reactionEmojis: [] };
+
+interface PendingStatsRequest {
+  resolve: (data: EventStats) => void;
+  reject: (err: Error) => void;
+}
+
+const pendingStatsBatch = new Map<string, PendingStatsRequest[]>();
+let statsFlushTimer: ReturnType<typeof setTimeout> | null = null;
+let statsPool: NostrPool | null = null;
+let statsQueryClient: ReturnType<typeof useQueryClient> | null = null;
+
+function flushStatsBatch() {
+  statsFlushTimer = null;
+  const pool = statsPool;
+  const qc = statsQueryClient;
+  if (!pool || !qc) return;
+
+  const batch = new Map(pendingStatsBatch);
+  pendingStatsBatch.clear();
+
+  const eventIds = [...batch.keys()];
+  if (eventIds.length === 0) return;
+
+  (async () => {
+    try {
+      // Two batched queries for ALL pending event IDs at once
      const [eTagEvents, qTagEvents] = await Promise.all([
-        nostr.query(
-          [{ kinds: [1, 6, 7, 9735], '#e': [eventId], limit: 200 }],
-          { signal: combined },
+        pool.query(
+          [{ kinds: [1, 6, 7, 9735], '#e': eventIds, limit: eventIds.length * 50 }],
+          { signal: AbortSignal.timeout(5000) },
        ),
-        nostr.query(
-          [{ kinds: [1], '#q': [eventId], limit: 50 }],
-          { signal: combined },
+        pool.query(
+          [{ kinds: [1], '#q': eventIds, limit: eventIds.length * 10 }],
+          { signal: AbortSignal.timeout(5000) },
        ),
      ]);

-      let replies = 0;
-      let reposts = 0;
-      let reactions = 0;
-      let zapAmount = 0;
-      const reactionEmojiSet = new Set<string>();
+      // Group results by referenced event ID
+      const statsMap = new Map<string, {
+        replies: number; reposts: number; reactions: number; zapAmount: number;
+        reactionEmojis: Set<string>; quotes: number;
+      }>();

+      // Initialize
+      for (const id of eventIds) {
+        statsMap.set(id, { replies: 0, reposts: 0, reactions: 0, zapAmount: 0, reactionEmojis: new Set(), quotes: 0 });
+      }
+
+      // Process e-tag events
      for (const e of eTagEvents) {
+        const refId = e.tags.find(([name]) => name === 'e')?.[1];
+        if (!refId) continue;
+        const entry = statsMap.get(refId);
+        if (!entry) continue;
+
        switch (e.kind) {
-          case 1: replies++; break;
-          case 6: reposts++; break;
+          case 1: entry.replies++; break;
+          case 6: entry.reposts++; break;
          case 7: {
-            reactions++;
-            // Extract the emoji from the reaction content (kind 7 events use content for the emoji)
+            entry.reactions++;
            const emoji = e.content.trim();
            if (emoji === '+' || emoji === '') {
-              reactionEmojiSet.add('👍');
+              entry.reactionEmojis.add('👍');
            } else if (emoji !== '-') {
-              reactionEmojiSet.add(emoji);
+              entry.reactionEmojis.add(emoji);
            }
            break;
          }
          case 9735: {
            const msats = extractZapAmount(e);
-            if (msats > 0) {
-              zapAmount += Math.floor(msats / 1000);
-            }
+            if (msats > 0) entry.zapAmount += Math.floor(msats / 1000);
            break;
          }
        }
      }

-      const quotes = qTagEvents.length;
+      // Process q-tag events (quotes)
+      for (const e of qTagEvents) {
+        const refId = e.tags.find(([name]) => name === 'q')?.[1];
+        if (!refId) continue;
+        const entry = statsMap.get(refId);
+        if (entry) entry.quotes++;
+      }

-      return { replies, reposts, quotes, reactions, zapAmount, reactionEmojis: Array.from(reactionEmojiSet) };
+      // Resolve all pending requests
+      for (const [id, resolvers] of batch) {
+        const raw = statsMap.get(id);
+        const result: EventStats = raw
+          ? { replies: raw.replies, reposts: raw.reposts, quotes: raw.quotes, reactions: raw.reactions, zapAmount: raw.zapAmount, reactionEmojis: [...raw.reactionEmojis] }
+          : { ...EMPTY_STATS };
+
+        qc.setQueryData(['event-stats', id], result);
+
+        for (const r of resolvers) {
+          r.resolve(result);
+        }
+      }
+    } catch (err) {
+      for (const resolvers of batch.values()) {
+        for (const r of resolvers) {
+          r.reject(err instanceof Error ? err : new Error(String(err)));
+        }
+      }
+    }
+  })();
+}
+
+function fetchStatsBatched(
+  eventId: string,
+  pool: NostrPool,
+  queryClient: ReturnType<typeof useQueryClient>,
+): Promise<EventStats> {
+  statsPool = pool;
+  statsQueryClient = queryClient;
+
+  return new Promise((resolve, reject) => {
+    const existing = pendingStatsBatch.get(eventId);
+    if (existing) {
+      existing.push({ resolve, reject });
+    } else {
+      pendingStatsBatch.set(eventId, [{ resolve, reject }]);
+    }
+
+    if (!statsFlushTimer) {
+      statsFlushTimer = setTimeout(flushStatsBatch, 50);
+    }
+  });
+}
+
+/** Counts engagement (replies, reposts, quotes, reactions, zaps) for a given event. */
+export function useEventStats(eventId: string | undefined) {
+  const { nostr } = useNostr();
+  const queryClient = useQueryClient();
+
+  return useQuery<EventStats>({
+    queryKey: ['event-stats', eventId ?? ''],
+    queryFn: async () => {
+      if (!eventId) return { ...EMPTY_STATS };
+      return fetchStatsBatched(eventId, nostr, queryClient);
    },
    enabled: !!eventId,
    staleTime: 60 * 1000,
-    placeholderData: (prev) => prev, // Keep showing previous counts while refetching
+    placeholderData: (prev) => prev,
  });
 }