Mirror/vega
1
0
Fork 0
mirror of https://github.com/hoornet/vega.git synced 2026-05-06 20:29:12 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add live feed subscription, timeouts on all relay fetches

Browse Source 
Global feed now uses a persistent live subscription (closeOnEose: false)
so new notes stream in real-time instead of requiring manual refresh.
Inspired by Wisp's streaming architecture.

Every fetchEvents call across the entire codebase now uses
fetchWithTimeout with groupable: false — prevents NDK from
batching/reusing stale subscriptions. This fixes Articles, DMs,
Notifications, Zaps, and Trending hanging indefinitely.

Also adds since filters on global (2h) and follow (24h) feeds
to ensure relay freshness, and fixes ArticleFeed re-fetch bug
where follows changes wiped latest tab results.
This commit is contained in:
Jure
2026-03-22 11:35:28 +01:00
parent 2e03c6ce11
commit b3e7ff7029
15 changed files with 176 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/components/article/ArticleFeed.tsx
View File

@@ -14,14 +14,20 @@ export function ArticleFeed() {
const [articles, setArticles] = useState<NDKEvent[]>([]);
const [loading, setLoading] = useState(true);
// Track follows length to avoid re-fetching latest when follows change
const followsKey = tab === "following" ? follows.join(",") : "latest";
useEffect(() => {
if (tab === "following" && follows.length === 0) return;
let cancelled = false;
setLoading(true);
const authors = tab === "following" ? follows : undefined;
fetchArticleFeed(40, authors)
.then(setArticles)
.catch(() => setArticles([]))
.finally(() => setLoading(false));
}, [tab, follows]);
.then((result) => { if (!cancelled) setArticles(result); })
.catch(() => { if (!cancelled) setArticles([]); })
.finally(() => { if (!cancelled) setLoading(false); });
return () => { cancelled = true; };
}, [followsKey]);
return (
<div className="h-full flex flex-col">

3
src/components/feed/NoteCard.tsx
View File

@@ -6,7 +6,7 @@ import { useUserStore } from "../../stores/user";
import { useMuteStore } from "../../stores/mute";
import { useUIStore } from "../../stores/ui";
import { timeAgo, shortenPubkey } from "../../lib/utils";
import { getNDK, fetchNoteById } from "../../lib/nostr";
import { getNDK, fetchNoteById, ensureConnected } from "../../lib/nostr";
import { getParentEventId } from "../../lib/threadTree";
import { NoteContent } from "./NoteContent";
import { NoteActions, LoggedOutStats } from "./NoteActions";
@@ -132,6 +132,7 @@ export function NoteCard({ event, focused, onReplyInThread }: NoteCardProps) {
return;
}
}
await ensureConnected();
const parent = await fetchNoteById(parentEventId);
if (parent) openThread(parent);
}}

4
src/components/zap/ZapHistoryView.tsx
View File

@@ -107,11 +107,11 @@ export function ZapHistoryView() {
const [tab, setTab] = useState<Tab>("received");
const [received, setReceived] = useState<NDKEvent[]>([]);
const [sent, setSent] = useState<NDKEvent[]>([]);
const [loading, setLoading] = useState(false);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
useEffect(() => {
if (!pubkey) return;
if (!pubkey) { setLoading(false); return; }
setLoading(true);
setError(null);
Promise.all([

24
src/lib/nostr/articles.ts
View File

@@ -1,5 +1,5 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage, nip19 } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind, nip19 } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, FEED_TIMEOUT, SINGLE_TIMEOUT } from "./core";
export async function publishArticle(opts: {
title: string;
@@ -45,9 +45,7 @@ export async function fetchArticle(naddr: string): Promise<NDKEvent | null> {
"#d": [identifier],
limit: 1,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
return Array.from(events)[0] ?? null;
} catch {
return null;
@@ -57,9 +55,7 @@ export async function fetchArticle(naddr: string): Promise<NDKEvent | null> {
export async function fetchAuthorArticles(pubkey: string, limit = 20): Promise<NDKEvent[]> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Article], authors: [pubkey], limit };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -67,9 +63,7 @@ export async function fetchArticleFeed(limit = 40, authors?: string[]): Promise<
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Article], limit };
if (authors && authors.length > 0) filter.authors = authors;
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -79,9 +73,7 @@ export async function searchArticles(query: string, limit = 30): Promise<NDKEven
const filter: NDKFilter & { search?: string } = isHashtag
? { kinds: [NDKKind.Article], "#t": [query.slice(1).toLowerCase()], limit }
: { kinds: [NDKKind.Article], search: query, limit };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -99,8 +91,6 @@ export async function fetchByAddr(addr: string): Promise<NDKEvent | null> {
"#d": [dTag],
limit: 1,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
return Array.from(events)[0] ?? null;
}

12
src/lib/nostr/bookmarks.ts
View File

@@ -1,12 +1,10 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, SINGLE_TIMEOUT } from "./core";
export async function fetchBookmarkList(pubkey: string): Promise<string[]> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [10003 as NDKKind], authors: [pubkey], limit: 1 };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
if (events.size === 0) return [];
const event = Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))[0];
return event.tags.filter((t) => t[0] === "e" && t[1]).map((t) => t[1]);
@@ -25,9 +23,7 @@ export async function publishBookmarkList(eventIds: string[]): Promise<void> {
export async function fetchBookmarkListFull(pubkey: string): Promise<{ eventIds: string[]; articleAddrs: string[] }> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [10003 as NDKKind], authors: [pubkey], limit: 1 };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
if (events.size === 0) return { eventIds: [], articleAddrs: [] };
const event = Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))[0];
const eventIds = event.tags.filter((t) => t[0] === "e" && t[1]).map((t) => t[1]);

8
src/lib/nostr/core.ts
View File

@@ -26,9 +26,13 @@ export async function fetchWithTimeout(
timeoutMs: number,
relaySet?: NDKRelaySet,
): Promise<Set<NDKEvent>> {
const opts = {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
groupable: false, // Prevent NDK from batching/reusing subscriptions
};
const promise = relaySet
? instance.fetchEvents(filter, { cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }, relaySet)
: instance.fetchEvents(filter, { cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY });
? instance.fetchEvents(filter, opts, relaySet)
: instance.fetchEvents(filter, opts);
return withTimeout(promise, timeoutMs, EMPTY_SET);
}

54
src/lib/nostr/dms.ts
View File

@@ -1,5 +1,5 @@
import { NDKEvent, NDKKind, NDKSubscriptionCacheUsage, giftWrap, giftUnwrap } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKKind, giftWrap, giftUnwrap } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, withTimeout, FEED_TIMEOUT } from "./core";
async function unwrapGiftWraps(events: NDKEvent[]): Promise<NDKEvent[]> {
const instance = getNDK();
@@ -21,21 +21,16 @@ async function unwrapGiftWraps(events: NDKEvent[]): Promise<NDKEvent[]> {
export async function fetchDMConversations(myPubkey: string): Promise<NDKEvent[]> {
const instance = getNDK();
// Fetch NIP-04 (legacy) and NIP-17 (gift-wrap) in parallel
const [nip04Received, nip04Sent, giftWraps] = await Promise.all([
instance.fetchEvents(
{ kinds: [NDKKind.EncryptedDirectMessage], "#p": [myPubkey], limit: 500 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.EncryptedDirectMessage], authors: [myPubkey], limit: 500 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.GiftWrap], "#p": [myPubkey], limit: 500 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
]);
// Fetch NIP-04 (legacy) and NIP-17 (gift-wrap) in parallel with timeouts
const [nip04Received, nip04Sent, giftWraps] = await withTimeout(
Promise.all([
fetchWithTimeout(instance, { kinds: [NDKKind.EncryptedDirectMessage], "#p": [myPubkey], limit: 500 }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.EncryptedDirectMessage], authors: [myPubkey], limit: 500 }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.GiftWrap], "#p": [myPubkey], limit: 500 }, FEED_TIMEOUT),
]),
FEED_TIMEOUT + 2000,
[new Set<NDKEvent>(), new Set<NDKEvent>(), new Set<NDKEvent>()],
);
const nip17Rumors = await unwrapGiftWraps(Array.from(giftWraps));
@@ -47,21 +42,16 @@ export async function fetchDMConversations(myPubkey: string): Promise<NDKEvent[]
export async function fetchDMThread(myPubkey: string, theirPubkey: string): Promise<NDKEvent[]> {
const instance = getNDK();
// Fetch NIP-04 and NIP-17 in parallel
const [fromThem, fromMe, giftWraps] = await Promise.all([
instance.fetchEvents(
{ kinds: [NDKKind.EncryptedDirectMessage], "#p": [myPubkey], authors: [theirPubkey], limit: 200 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.EncryptedDirectMessage], "#p": [theirPubkey], authors: [myPubkey], limit: 200 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.GiftWrap], "#p": [myPubkey], limit: 200 },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
]);
// Fetch NIP-04 and NIP-17 in parallel with timeouts
const [fromThem, fromMe, giftWraps] = await withTimeout(
Promise.all([
fetchWithTimeout(instance, { kinds: [NDKKind.EncryptedDirectMessage], "#p": [myPubkey], authors: [theirPubkey], limit: 200 }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.EncryptedDirectMessage], "#p": [theirPubkey], authors: [myPubkey], limit: 200 }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.GiftWrap], "#p": [myPubkey], limit: 200 }, FEED_TIMEOUT),
]),
FEED_TIMEOUT + 2000,
[new Set<NDKEvent>(), new Set<NDKEvent>(), new Set<NDKEvent>()],
);
// Unwrap NIP-17 and filter to only messages from/to this partner
const allRumors = await unwrapGiftWraps(Array.from(giftWraps));

60
src/lib/nostr/engagement.ts
View File

@@ -1,5 +1,5 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, withTimeout, FEED_TIMEOUT, SINGLE_TIMEOUT } from "./core";
export async function publishReaction(eventId: string, eventPubkey: string, reaction = "+"): Promise<void> {
const instance = getNDK();
@@ -17,34 +17,22 @@ export async function publishReaction(eventId: string, eventPubkey: string, reac
export async function fetchReactionCount(eventId: string): Promise<number> {
const instance = getNDK();
const filter: NDKFilter = {
kinds: [NDKKind.Reaction],
"#e": [eventId],
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const filter: NDKFilter = { kinds: [NDKKind.Reaction], "#e": [eventId] };
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
return events.size;
}
export async function fetchReplyCount(eventId: string): Promise<number> {
const instance = getNDK();
const filter: NDKFilter = {
kinds: [NDKKind.Text],
"#e": [eventId],
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const filter: NDKFilter = { kinds: [NDKKind.Text], "#e": [eventId] };
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
return events.size;
}
export async function fetchZapCount(eventId: string): Promise<{ count: number; totalSats: number }> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Zap], "#e": [eventId] };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
let totalSats = 0;
for (const event of events) {
const desc = event.tags.find((t) => t[0] === "description")?.[1];
@@ -71,20 +59,15 @@ export async function fetchBatchEngagement(eventIds: string[]): Promise<Map<stri
for (let i = 0; i < eventIds.length; i += chunkSize) {
const chunk = eventIds.slice(i, i + chunkSize);
const [reactions, replies, zaps] = await Promise.all([
instance.fetchEvents(
{ kinds: [NDKKind.Reaction], "#e": chunk },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.Text], "#e": chunk },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
instance.fetchEvents(
{ kinds: [NDKKind.Zap], "#e": chunk },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
),
]);
const [reactions, replies, zaps] = await withTimeout(
Promise.all([
fetchWithTimeout(instance, { kinds: [NDKKind.Reaction], "#e": chunk }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.Text], "#e": chunk }, FEED_TIMEOUT),
fetchWithTimeout(instance, { kinds: [NDKKind.Zap], "#e": chunk }, FEED_TIMEOUT),
]),
FEED_TIMEOUT + 2000, // outer timeout slightly longer than inner
[new Set<NDKEvent>(), new Set<NDKEvent>(), new Set<NDKEvent>()],
);
for (const event of reactions) {
const eTag = event.tags.find((t) => t[0] === "e")?.[1];
@@ -117,18 +100,15 @@ export async function fetchBatchEngagement(eventIds: string[]): Promise<Map<stri
export async function fetchZapsReceived(pubkey: string, limit = 50): Promise<NDKEvent[]> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Zap], "#p": [pubkey], limit };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
// Zap queries can be slow — give relays more time
const events = await fetchWithTimeout(instance, filter, 12000);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
export async function fetchZapsSent(pubkey: string, limit = 50): Promise<NDKEvent[]> {
const instance = getNDK();
// Zap receipts (kind 9735) with uppercase P tag = the sender's pubkey
// #P (uppercase) is poorly supported; also try finding zap requests we authored
const filter: NDKFilter = { kinds: [NDKKind.Zap], "#P": [pubkey], limit };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, 12000);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}

8
src/lib/nostr/muting.ts
View File

@@ -1,12 +1,10 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, SINGLE_TIMEOUT } from "./core";
export async function fetchMuteList(pubkey: string): Promise<string[]> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [10000 as NDKKind], authors: [pubkey], limit: 1 };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
if (events.size === 0) return [];
const event = Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))[0];
return event.tags.filter((t) => t[0] === "p" && t[1]).map((t) => t[1]);

7
src/lib/nostr/notes.ts
View File

@@ -4,7 +4,9 @@ import { fetchUserRelayList } from "./relays";
export async function fetchGlobalFeed(limit: number = 50): Promise<NDKEvent[]> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Text], limit };
// Ask for notes from the last 2 hours to ensure freshness
const since = Math.floor(Date.now() / 1000) - 2 * 3600;
const filter: NDKFilter = { kinds: [NDKKind.Text], limit, since };
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -12,7 +14,8 @@ export async function fetchGlobalFeed(limit: number = 50): Promise<NDKEvent[]> {
export async function fetchFollowFeed(pubkeys: string[], limit = 80): Promise<NDKEvent[]> {
if (pubkeys.length === 0) return [];
const instance = getNDK();
const filter: NDKFilter = { kinds: [NDKKind.Text], authors: pubkeys, limit };
const since = Math.floor(Date.now() / 1000) - 24 * 3600; // last 24h for follows
const filter: NDKFilter = { kinds: [NDKKind.Text], authors: pubkeys, limit, since };
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}

6
src/lib/nostr/relays.ts
View File

@@ -1,12 +1,12 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, SINGLE_TIMEOUT } from "./core";
export interface UserRelayList { read: string[]; write: string[]; }
export async function fetchUserRelayList(pubkey: string): Promise<UserRelayList> {
const instance = getNDK();
const filter: NDKFilter = { kinds: [10002 as NDKKind], authors: [pubkey], limit: 1 };
const events = await instance.fetchEvents(filter, { cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY });
const events = await fetchWithTimeout(instance, filter, SINGLE_TIMEOUT);
if (events.size === 0) return { read: [], write: [] };
const event = Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))[0];
const read: string[] = [], write: string[] = [];

28
src/lib/nostr/search.ts
View File

@@ -1,6 +1,6 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage, NDKUser } from "@nostr-dev-kit/ndk";
import { NDKEvent, NDKFilter, NDKKind, NDKUser } from "@nostr-dev-kit/ndk";
import { type ParsedSearch, matchesHasFilter } from "../search";
import { getNDK } from "./core";
import { getNDK, fetchWithTimeout, FEED_TIMEOUT } from "./core";
export async function searchNotes(query: string, limit = 50): Promise<NDKEvent[]> {
const instance = getNDK();
@@ -8,9 +8,7 @@ export async function searchNotes(query: string, limit = 50): Promise<NDKEvent[]
const filter: NDKFilter & { search?: string } = isHashtag
? { kinds: [NDKKind.Text], "#t": [query.slice(1).toLowerCase()], limit }
: { kinds: [NDKKind.Text], search: query, limit };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -21,9 +19,7 @@ export async function searchUsers(query: string, limit = 20): Promise<NDKEvent[]
search: query,
limit,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events);
}
@@ -119,24 +115,14 @@ export async function advancedSearch(parsed: ParsedSearch, limit = 50): Promise<
return filter;
};
const opts = { cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY };
// Wrap fetchEvents with a timeout — NDK can hang forever if no relay supports the filter
const fetchWithTimeout = (filter: NDKFilter & { search?: string }, timeoutMs = 8000): Promise<Set<NDKEvent>> => {
return Promise.race([
instance.fetchEvents(filter, opts),
new Promise<Set<NDKEvent>>((resolve) => setTimeout(() => resolve(new Set()), timeoutMs)),
]);
};
const noteFilter = noteKinds.length > 0 ? buildFilter(noteKinds) : null;
const articleFilter = articleKinds.length > 0 ? buildFilter(articleKinds) : null;
const shouldSearchUsers = (!hasKindFilter || parsed.kinds.includes(0)) && hasSearch && !hasHashtags;
const [noteEvents, articleEvents, userEvents] = await Promise.all([
noteFilter ? fetchWithTimeout(noteFilter) : Promise.resolve(new Set<NDKEvent>()),
articleFilter ? fetchWithTimeout(articleFilter) : Promise.resolve(new Set<NDKEvent>()),
shouldSearchUsers ? fetchWithTimeout({ kinds: [NDKKind.Metadata], search: searchText, limit: 20 } as NDKFilter & { search: string }) : Promise.resolve(new Set<NDKEvent>()),
noteFilter ? fetchWithTimeout(instance, noteFilter, FEED_TIMEOUT) : Promise.resolve(new Set<NDKEvent>()),
articleFilter ? fetchWithTimeout(instance, articleFilter, FEED_TIMEOUT) : Promise.resolve(new Set<NDKEvent>()),
shouldSearchUsers ? fetchWithTimeout(instance, { kinds: [NDKKind.Metadata], search: searchText, limit: 20 } as NDKFilter & { search: string }, FEED_TIMEOUT) : Promise.resolve(new Set<NDKEvent>()),
]);
let notes = Array.from(noteEvents);

17
src/lib/nostr/social.ts
View File

@@ -1,5 +1,5 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, FEED_TIMEOUT } from "./core";
export async function publishProfile(fields: {
name?: string;
@@ -47,9 +47,7 @@ export async function fetchFollowSuggestions(myFollows: string[]): Promise<{ pub
for (let i = 0; i < myFollows.length; i += batchSize) {
const batch = myFollows.slice(i, i + batchSize);
const filter: NDKFilter = { kinds: [3 as NDKKind], authors: batch, limit: batch.length };
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
allContactEvents.push(...Array.from(events));
}
@@ -76,9 +74,10 @@ export async function fetchFollowSuggestions(myFollows: string[]): Promise<{ pub
export async function fetchMentions(pubkey: string, since: number, limit = 50): Promise<NDKEvent[]> {
const instance = getNDK();
const events = await instance.fetchEvents(
const events = await fetchWithTimeout(
instance,
{ kinds: [NDKKind.Text], "#p": [pubkey], since, limit },
{ cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY }
FEED_TIMEOUT,
);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -91,8 +90,6 @@ export async function fetchNewFollowers(pubkey: string, since: number, limit = 2
since,
limit,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}

12
src/lib/nostr/trending.ts
View File

@@ -1,5 +1,5 @@
import { NDKEvent, NDKFilter, NDKKind, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { getNDK } from "./core";
import { NDKEvent, NDKFilter, NDKKind } from "@nostr-dev-kit/ndk";
import { getNDK, fetchWithTimeout, FEED_TIMEOUT } from "./core";
export async function fetchTrendingCandidates(limit = 200, sinceHours = 24): Promise<NDKEvent[]> {
const instance = getNDK();
@@ -9,9 +9,7 @@ export async function fetchTrendingCandidates(limit = 200, sinceHours = 24): Pro
since,
limit,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
return Array.from(events).sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0));
}
@@ -23,9 +21,7 @@ export async function fetchTrendingHashtags(limit = 15): Promise<{ tag: string;
since,
limit: 500,
};
const events = await instance.fetchEvents(filter, {
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
const events = await fetchWithTimeout(instance, filter, FEED_TIMEOUT);
const counts = new Map<string, number>();
for (const event of events) {

81
src/stores/feed.ts
View File

@@ -1,12 +1,16 @@
import { create } from "zustand";
import { NDKEvent } from "@nostr-dev-kit/ndk";
import { NDKEvent, NDKFilter, NDKKind, NDKSubscription, NDKSubscriptionCacheUsage } from "@nostr-dev-kit/ndk";
import { connectToRelays, ensureConnected, resetNDK, fetchGlobalFeed, fetchBatchEngagement, fetchTrendingCandidates, getNDK } from "../lib/nostr";
import { dbLoadFeed, dbSaveNotes } from "../lib/db";
import { diagWrapFetch, logDiag, startRelaySnapshots, getRelayStates } from "../lib/feedDiagnostics";
const TRENDING_CACHE_KEY = "wrystr_trending_cache";
const TRENDING_TTL = 10 * 60 * 1000; // 10 minutes
const MAX_FEED_SIZE = 200;
// Live subscription handle — persists across store calls
let liveSub: NDKSubscription | null = null;
let saveTimer: ReturnType<typeof setTimeout> | null = null;
interface FeedState {
notes: NDKEvent[];
@@ -19,6 +23,7 @@ interface FeedState {
connect: () => Promise<void>;
loadCachedFeed: () => Promise<void>;
loadFeed: () => Promise<void>;
startLiveFeed: () => void;
loadTrendingFeed: (force?: boolean) => Promise<void>;
setFocusedNoteIndex: (n: number) => void;
}
@@ -73,7 +78,11 @@ export const useFeedStore = create<FeedState>((set, get) => ({
resetNDK().then(() => {
if (getNDK().pool?.relays) {
const after = Array.from(getNDK().pool.relays.values());
if (after.some((r) => r.connected)) set({ connected: true });
if (after.some((r) => r.connected)) {
set({ connected: true });
// Restart live sub after NDK reset
get().startLiveFeed();
}
}
}).catch(() => {});
} else {
@@ -90,7 +99,7 @@ export const useFeedStore = create<FeedState>((set, get) => ({
loadCachedFeed: async () => {
try {
const rawNotes = await dbLoadFeed(200);
const rawNotes = await dbLoadFeed(MAX_FEED_SIZE);
if (rawNotes.length === 0) return;
const ndk = getNDK();
const events = rawNotes.map((raw) => new NDKEvent(ndk, JSON.parse(raw)));
@@ -100,6 +109,9 @@ export const useFeedStore = create<FeedState>((set, get) => ({
}
},
/**
* One-shot feed fetch — loads initial batch, then starts live subscription.
*/
loadFeed: async () => {
if (get().loading) return;
set({ loading: true, error: null });
@@ -107,23 +119,78 @@ export const useFeedStore = create<FeedState>((set, get) => ({
await ensureConnected();
const fresh = await diagWrapFetch("global_fetch", () => fetchGlobalFeed(80));
// Merge with currently displayed notes so cached notes aren't lost
// if the relay returns fewer results than the cache had.
// Merge with currently displayed notes
const freshIds = new Set(fresh.map((n) => n.id));
const kept = get().notes.filter((n) => !freshIds.has(n.id));
const merged = [...fresh, ...kept]
.sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))
.slice(0, 200);
.slice(0, MAX_FEED_SIZE);
set({ notes: merged, loading: false, focusedNoteIndex: -1 });
// Persist fresh notes to SQLite (fire-and-forget)
dbSaveNotes(fresh.map((e) => JSON.stringify(e.rawEvent())));
// Start live subscription after initial load
get().startLiveFeed();
} catch (err) {
set({ error: `Feed failed: ${err}`, loading: false });
}
},
/**
* Start a persistent live subscription for new notes.
* New events stream in and are prepended to the feed in real time.
*/
startLiveFeed: () => {
// Close existing subscription if any
if (liveSub) {
try { liveSub.stop(); } catch { /* ignore */ }
liveSub = null;
}
const ndk = getNDK();
const since = Math.floor(Date.now() / 1000);
const filter: NDKFilter = { kinds: [NDKKind.Text], since, limit: 20 };
const sub = ndk.subscribe(filter, {
closeOnEose: false, // Keep subscription open — this is the key difference
groupable: false,
cacheUsage: NDKSubscriptionCacheUsage.ONLY_RELAY,
});
sub.on("event", (event: NDKEvent) => {
const current = get().notes;
// Deduplicate
if (current.some((n) => n.id === event.id)) return;
const updated = [event, ...current]
.sort((a, b) => (b.created_at ?? 0) - (a.created_at ?? 0))
.slice(0, MAX_FEED_SIZE);
set({ notes: updated });
// Debounced save to SQLite — batch saves every 5s
if (!saveTimer) {
saveTimer = setTimeout(() => {
saveTimer = null;
const toSave = get().notes.slice(0, 20);
dbSaveNotes(toSave.map((e) => JSON.stringify(e.rawEvent())));
}, 5000);
}
});
sub.on("eose", () => {
logDiag({
ts: new Date().toISOString(),
action: "live_feed_eose",
details: "Live subscription received EOSE — now streaming new events",
});
});
liveSub = sub;
console.log("[Wrystr] Live feed subscription started");
},
loadTrendingFeed: async (force?: boolean) => {
if (get().trendingLoading) return;