Sort threads by last activity

src/lib/threads.svelte.ts

@@ -1,8 +1,13 @@
-import { SimplePool } from "@nostr/tools";
+import { Relay } from "@nostr/tools";
+import type { Event } from "@nostr/tools/core";
+import type { Filter } from "@nostr/tools/filter";
 import { loadNostrUser, type NostrUser } from "@nostr/gadgets/metadata";
 import { RELAY_URL, GROUP_ID } from "$lib/config";
 import { ingestNostrUser } from "$lib/profiles.svelte";
 
+const N_OPS = 50;
+const N_REPLY_WINDOW = 200;
+
 export type ThreadData = {
   id: string;
   title: string;
@@ -30,80 +35,114 @@ async function loadProfile(pubkey: string) {
   ingestNostrUser(user);
 }
 
+function querySync(relay: Relay, filter: Filter): Promise<Event[]> {
+  return new Promise((resolve) => {
+    const events: Event[] = [];
+    const sub = relay.subscribe([filter], {
+      onevent(e) {
+        events.push(e);
+      },
+      oneose() {
+        sub.close();
+        resolve(events);
+      },
+      onclose() {
+        resolve(events);
+      },
+    });
+  });
+}
+
 export async function loadThreads() {
-  const pool = new SimplePool();
+  const relay = await Relay.connect(RELAY_URL);
   try {
-    // Step 1: fetch threads
-    const events = await pool.querySync([RELAY_URL], {
+    // Parallel: latest OPs + latest replies window
+    const [opEvents, replyEvents] = await Promise.all([
+      querySync(relay, { kinds: [11], "#h": [GROUP_ID], limit: N_OPS }),
+      querySync(relay, { kinds: [1111], "#h": [GROUP_ID], limit: N_REPLY_WINDOW }),
+    ]);
+
+    // Aggregate reply data per root thread id
+    type ReplyAgg = {
+      latestAt: number;
+      latestPubkey: string;
+      pubkeys: Set<string>;
+    };
+    const replyMap = new Map<string, ReplyAgg>();
+    for (const r of replyEvents) {
+      const rootId = r.tags.find((t) => t[0] === "E")?.[1];
+      if (!rootId) continue;
+      let agg = replyMap.get(rootId);
+      if (!agg) {
+        agg = { latestAt: 0, latestPubkey: r.pubkey, pubkeys: new Set() };
+        replyMap.set(rootId, agg);
+      }
+      agg.pubkeys.add(r.pubkey);
+      if (r.created_at > agg.latestAt) {
+        agg.latestAt = r.created_at;
+        agg.latestPubkey = r.pubkey;
+      }
+    }
+
+    // Backfill OPs that have recent replies but fell outside the OP window
+    const opIds = new Set(opEvents.map((e) => e.id));
+    const missingIds = [...replyMap.keys()].filter((id) => !opIds.has(id));
+    let extraOps: Event[] = [];
+    if (missingIds.length > 0) {
+      extraOps = await querySync(relay, {
         kinds: [11],
         "#h": [GROUP_ID],
-      limit: 50,
+        ids: missingIds,
       });
+    }
 
-    events.sort((a, b) => b.created_at - a.created_at);
-
-    threads = events.map((e) => ({
+    // Build candidate threads, computing latestAt from OP + reply window
+    const candidates: ThreadData[] = [...opEvents, ...extraOps].map((e) => {
+      const agg = replyMap.get(e.id);
+      const replierPubkeys = agg
+        ? [...agg.pubkeys].filter((p) => p !== e.pubkey).slice(0, 4)
+        : [];
+      return {
         id: e.id,
         title: e.tags.find((t) => t[0] === "title")?.[1] ?? "(untitled)",
         labels: e.tags.filter((t) => t[0] === "t" && t[1]).map((t) => t[1]),
         authorPubkey: e.pubkey,
         createdAt: e.created_at,
         replyCount: 0,
-      latestAt: e.created_at,
-      latestPubkey: e.pubkey,
-      replierPubkeys: [],
-    }));
-
-    events.forEach((e) => loadProfile(e.pubkey));
-
-    if (events.length === 0) return;
-
-    // Step 2: fetch replies for all threads in one request
-    const replies = await pool.querySync([RELAY_URL], {
-      kinds: [1111],
-      "#E": events.map((e) => e.id),
-    });
-
-    // Aggregate per-thread reply data
-    const replyMap = new Map<
-      string,
-      { count: number; latestAt: number; latestPubkey: string; pubkeys: Set<string> }
-    >();
-
-    for (const reply of replies) {
-      const rootId = reply.tags.find((t) => t[0] === "E")?.[1];
-      if (!rootId) continue;
-      if (!replyMap.has(rootId)) {
-        replyMap.set(rootId, {
-          count: 0,
-          latestAt: 0,
-          latestPubkey: reply.pubkey,
-          pubkeys: new Set(),
-        });
-      }
-      const rd = replyMap.get(rootId)!;
-      rd.count++;
-      rd.pubkeys.add(reply.pubkey);
-      if (reply.created_at > rd.latestAt) {
-        rd.latestAt = reply.created_at;
-        rd.latestPubkey = reply.pubkey;
-      }
-    }
-
-    threads = threads.map((t) => {
-      const rd = replyMap.get(t.id);
-      if (!rd) return t;
-      return {
-        ...t,
-        replyCount: rd.count,
-        latestAt: rd.latestAt,
-        latestPubkey: rd.latestPubkey,
-        replierPubkeys: [...rd.pubkeys].slice(0, 4),
+        latestAt: agg ? Math.max(e.created_at, agg.latestAt) : e.created_at,
+        latestPubkey: agg ? agg.latestPubkey : e.pubkey,
+        replierPubkeys,
       };
     });
 
-    replies.forEach((r) => loadProfile(r.pubkey));
+    // Sort by last activity, keep top N_OPS
+    candidates.sort((a, b) => b.latestAt - a.latestAt);
+    const top = candidates.slice(0, N_OPS);
+    threads = top;
+
+    // Prefetch profiles
+    for (const t of top) {
+      loadProfile(t.authorPubkey);
+      for (const p of t.replierPubkeys) loadProfile(p);
+    }
+
+    // Exact reply counts: single follow-up query bucketed by root.
+    // (Pyramid's NIP-45 COUNT ignores group storage, so we count events directly.)
+    const topIds = top.map((t) => t.id);
+    const allReplies = await querySync(relay, {
+      kinds: [1111],
+      "#h": [GROUP_ID],
+      "#E": topIds,
+      limit: 5000,
+    });
+    const countMap = new Map<string, number>();
+    for (const r of allReplies) {
+      const rootId = r.tags.find((t) => t[0] === "E")?.[1];
+      if (!rootId) continue;
+      countMap.set(rootId, (countMap.get(rootId) ?? 0) + 1);
+    }
+    threads = top.map((t) => ({ ...t, replyCount: countMap.get(t.id) ?? 0 }));
   } finally {
-    pool.close([RELAY_URL]);
+    relay.close();
   }
 }