From a196e95127d12764ce85ff138d0ccf564d918d45 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Thu, 29 Jan 2026 16:32:11 -0500 Subject: [PATCH] Centralize relay list subscriptions --- js/app.js | 48 ++--- js/channelProfile.js | 54 +++-- js/services/hashtagPreferencesService.js | 64 ++++++ js/services/nostrService.js | 20 +- js/services/profileMetadataService.js | 43 ++++ js/services/relaySubscriptionService.js | 250 +++++++++++++++++++++++ js/subscriptions.js | 63 ++++++ js/userBlocks.js | 59 ++++++ 8 files changed, 554 insertions(+), 47 deletions(-) create mode 100644 js/services/relaySubscriptionService.js diff --git a/js/app.js b/js/app.js index c3c586f9..50123bd6 100644 --- a/js/app.js +++ b/js/app.js @@ -64,6 +64,10 @@ import getAuthProvider, { import hashtagPreferences, { HASHTAG_PREFERENCES_EVENTS, } from "./services/hashtagPreferencesService.js"; +import { + fetchProfileMetadata, + ensureProfileMetadataSubscription, +} from "./services/profileMetadataService.js"; import { getSidebarLoadingMarkup } from "./sidebarLoading.js"; import { subscriptions } from "./subscriptions.js"; import { @@ -9561,45 +9565,43 @@ class Application { Array.isArray(nostrClient?.relays) && nostrClient.relays.length ? nostrClient.relays : null; - if (!relayList || !nostrClient?.pool || typeof nostrClient.pool.list !== "function") { + if (!relayList) { return; } - const events = await nostrClient.pool.list(relayList, [ - { kinds: [0], authors: [normalized], limit: 1 }, - ]); + const profileEntry = await fetchProfileMetadata(normalized, { + nostr: nostrClient, + relays: relayList, + logger: devLogger, + }); if (this.modalCreatorProfileRequestToken !== requestToken) { return; } - const newest = Array.isArray(events) - ? events.reduce((latest, event) => { - if (!event || typeof event !== "object") { - return latest; - } - const eventPubkey = this.normalizeHexPubkey(event.pubkey); - if (eventPubkey && eventPubkey !== normalized) { - return latest; - } - const createdAt = Number.isFinite(event.created_at) ? event.created_at : 0; - if (!latest || createdAt > latest.createdAt) { - return { createdAt, event }; - } - return latest; - }, null) - : null; - - if (!newest || !newest.event) { + if (!profileEntry?.event) { if (this.modalCreatorProfileRequestToken === requestToken) { this.modalCreatorProfileRequestToken = null; } return; } + ensureProfileMetadataSubscription({ + pubkey: normalized, + nostr: nostrClient, + relays: relayList, + onProfile: ({ profile }) => { + if (typeof this.setProfileCacheEntry === "function" && profile) { + this.setProfileCacheEntry(normalized, profile); + } + }, + }); + let parsed = null; try { - parsed = newest.event.content ? JSON.parse(newest.event.content) : null; + parsed = profileEntry.event.content + ? JSON.parse(profileEntry.event.content) + : null; } catch (error) { devLogger.warn( `[Application] Failed to parse creator profile content for ${normalized}:`, diff --git a/js/channelProfile.js b/js/channelProfile.js index fbdcc161..83cf0913 100644 --- a/js/channelProfile.js +++ b/js/channelProfile.js @@ -16,6 +16,10 @@ import { createChannelProfileMenuPanel } from "./ui/components/videoMenuRenderer import { ALLOW_NSFW_CONTENT } from "./config.js"; import { sanitizeProfileMediaUrl } from "./utils/profileMedia.js"; import moderationService from "./services/moderationService.js"; +import { + fetchProfileMetadata, + ensureProfileMetadataSubscription, +} from "./services/profileMetadataService.js"; import { createModerationStage } from "./feedEngine/stages.js"; import { DEFAULT_AUTOPLAY_BLOCK_THRESHOLD, @@ -5099,31 +5103,20 @@ async function fetchChannelProfileFromRelays(pubkey) { } try { - const events = await pool.list(relayUrls, [ - { kinds: [0], authors: [pubkey], limit: 1 } - ]); + const profileEntry = await fetchProfileMetadata(pubkey, { + nostr: nostrClient, + relays: relayUrls, + logger: devLogger, + }); - let newestEvent = null; - for (const event of Array.isArray(events) ? events : []) { - if (!event || !event.content) { - continue; - } - if (!newestEvent || event.created_at > newestEvent.created_at) { - newestEvent = event; - } + if (!profileEntry) { + return { event: null, profile: {} }; } - if (newestEvent?.content) { - try { - const meta = JSON.parse(newestEvent.content); - return { event: newestEvent, profile: meta }; - } catch (error) { - userLogger.warn("Failed to parse channel metadata payload:", error); - return { event: newestEvent, profile: {} }; - } - } - - return { event: newestEvent, profile: {} }; + return { + event: profileEntry.event || null, + profile: profileEntry.profile || {}, + }; } catch (error) { throw error; } @@ -5213,6 +5206,23 @@ async function loadUserProfile(pubkey) { npub: currentChannelNpub, loadToken }); + + ensureProfileMetadataSubscription({ + pubkey, + nostr: nostrClient, + onProfile: ({ profile, event }) => { + if (loadToken !== currentProfileLoadToken) { + return; + } + applyChannelProfileMetadata({ + profile, + event, + pubkey, + npub: currentChannelNpub, + loadToken, + }); + }, + }); } catch (error) { if (loadToken === currentProfileLoadToken) { userLogger.error("Failed to fetch user profile data:", error); diff --git a/js/services/hashtagPreferencesService.js b/js/services/hashtagPreferencesService.js index 0cef2e22..bf674d80 100644 --- a/js/services/hashtagPreferencesService.js +++ b/js/services/hashtagPreferencesService.js @@ -20,6 +20,7 @@ import { profileCache } from "../state/profileCache.js"; import { DEFAULT_RELAY_URLS } from "../nostr/toolkit.js"; import { relayManager } from "../relayManager.js"; import { runNip07WithRetry, NIP07_PRIORITY } from "../nostr/nip07Permissions.js"; +import { relaySubscriptionService } from "./relaySubscriptionService.js"; const LOG_PREFIX = "[HashtagPreferences]"; const HASHTAG_IDENTIFIER = "bitvid:tag-preferences"; @@ -243,6 +244,7 @@ class HashtagPreferencesService { this.decryptRetryTimeoutId = null; this.loadPromise = null; this.loadingPubkey = null; + this.subscriptionKey = null; profileCache.subscribe((event, detail) => { if (event === "profileChanged") { @@ -264,6 +266,7 @@ class HashtagPreferencesService { } reset() { + this.stopHashtagPreferencesSubscription(); this.interests = new Set(); this.disinterests = new Set(); this.eventId = null; @@ -280,6 +283,56 @@ class HashtagPreferencesService { this.emitChange("reset"); } + stopHashtagPreferencesSubscription() { + if (this.subscriptionKey) { + relaySubscriptionService.stopSubscription(this.subscriptionKey, "reset"); + this.subscriptionKey = null; + } + } + + ensureHashtagPreferencesSubscription(pubkey, relays) { + const normalized = normalizeHexPubkey(pubkey); + if (!normalized || !nostrClient?.pool) { + return null; + } + + const schema = getNostrEventSchema(NOTE_TYPES.HASHTAG_PREFERENCES); + const canonicalKind = schema?.kind ?? 30015; + const legacyKind = 30005; + const kinds = canonicalKind === legacyKind + ? [canonicalKind] + : [canonicalKind, legacyKind]; + + const filters = kinds.map((kind) => ({ + kinds: [kind], + authors: [normalized], + "#d": [HASHTAG_IDENTIFIER], + })); + + const key = `hashtags:${normalized}`; + this.subscriptionKey = key; + + return relaySubscriptionService.ensureSubscription({ + key, + pool: nostrClient.pool, + relays, + filters, + label: "hashtag-preferences", + onEvent: (event) => this.handleHashtagPreferencesEvent(event), + }); + } + + handleHashtagPreferencesEvent(event) { + const normalized = normalizeHexPubkey(event?.pubkey); + if (!normalized || normalized !== this.activePubkey) { + return; + } + + this.load(normalized, { allowPermissionPrompt: false }).catch((error) => { + devLogger.warn(`${LOG_PREFIX} Failed to refresh after hashtag event`, error); + }); + } + saveToCache() { const interests = this.getInterests(); const disinterests = this.getDisinterests(); @@ -564,6 +617,17 @@ class HashtagPreferencesService { } return; } + if (!nostrClient?.pool && typeof nostrClient?.ensurePool === "function") { + try { + await nostrClient.ensurePool(); + } catch (error) { + devLogger.warn( + `${LOG_PREFIX} Failed to initialize relay pool for subscription`, + error, + ); + } + } + this.ensureHashtagPreferencesSubscription(normalized, relays); const schema = getNostrEventSchema(NOTE_TYPES.HASHTAG_PREFERENCES); const canonicalKind = schema?.kind ?? 30015; diff --git a/js/services/nostrService.js b/js/services/nostrService.js index 01316bf1..2a2a36c9 100644 --- a/js/services/nostrService.js +++ b/js/services/nostrService.js @@ -614,6 +614,7 @@ export class NostrService { this.dmMessageIndex = new Map(); this.dmSubscription = null; this.dmActorPubkey = null; + this.dmSubscriptionRelays = null; this.dmHydratedFromSnapshot = false; this.dmNotificationManager = new DmNotificationManager({ logger: userLogger, @@ -1111,13 +1112,24 @@ export class NostrService { if (this.dmActorPubkey === normalizedActor) { devLogger.log( "[nostrService] Reusing existing DM subscription for active actor.", - { actorPubkey: normalizedActor }, + { + actorPubkey: normalizedActor, + relays: Array.isArray(this.dmSubscriptionRelays) + ? this.dmSubscriptionRelays + : [], + }, ); return this.dmSubscription; } devLogger.log( "[nostrService] Replacing existing DM subscription for new actor.", - { previous: this.dmActorPubkey, next: normalizedActor }, + { + previous: this.dmActorPubkey, + next: normalizedActor, + previousRelays: Array.isArray(this.dmSubscriptionRelays) + ? this.dmSubscriptionRelays + : [], + }, ); this.stopDirectMessageSubscription(); } @@ -1222,6 +1234,9 @@ export class NostrService { this.dmSubscription = subscription; this.dmActorPubkey = normalizedActor; + this.dmSubscriptionRelays = Array.isArray(resolvedRelays) + ? resolvedRelays.slice() + : []; devLogger.log("[nostrService] DM subscription activated.", { actorPubkey: normalizedActor, relays: resolvedRelays, @@ -1257,6 +1272,7 @@ export class NostrService { } } this.dmSubscription = null; + this.dmSubscriptionRelays = null; } async persistDirectMessageSnapshot(actorPubkey = null, messages = null) { diff --git a/js/services/profileMetadataService.js b/js/services/profileMetadataService.js index 84513613..84b11405 100644 --- a/js/services/profileMetadataService.js +++ b/js/services/profileMetadataService.js @@ -1,5 +1,6 @@ import { nostrClient } from "../nostrClientFacade.js"; import { devLogger } from "../utils/logger.js"; +import { relaySubscriptionService } from "./relaySubscriptionService.js"; const DEFAULT_PROFILE_IMAGE = "assets/svg/default-profile.svg"; const HEX64_REGEX = /^[0-9a-f]{64}$/i; @@ -248,3 +249,45 @@ export async function fetchProfileMetadata(pubkey, options = {}) { return results.get(normalized) || null; } + +export function ensureProfileMetadataSubscription({ + pubkey, + nostr = nostrClient, + relays = null, + logger = devLogger, + defaultProfileImage = DEFAULT_PROFILE_IMAGE, + onProfile, +} = {}) { + const normalized = normalizePubkey(pubkey); + if (!normalized || !nostr?.pool) { + return null; + } + + const relayUrls = resolveRelays(relays, nostr); + const filters = [{ kinds: [0], authors: [normalized], limit: 1 }]; + const key = `profile:${normalized}`; + + return relaySubscriptionService.ensureSubscription({ + key, + pool: nostr.pool, + relays: relayUrls, + filters, + label: "profile-metadata", + onEvent: (event) => { + if (!event || event.pubkey !== normalized) { + return; + } + const profile = parseProfileEvent(event, defaultProfileImage); + if (!profile) { + return; + } + if (typeof onProfile === "function") { + try { + onProfile({ profile, event }); + } catch (error) { + logger.warn("[profileMetadata] onProfile callback threw", error); + } + } + }, + }); +} diff --git a/js/services/relaySubscriptionService.js b/js/services/relaySubscriptionService.js new file mode 100644 index 00000000..b5457449 --- /dev/null +++ b/js/services/relaySubscriptionService.js @@ -0,0 +1,250 @@ +import { devLogger, userLogger } from "../utils/logger.js"; + +function normalizeRelayList(relays) { + if (!Array.isArray(relays)) { + return []; + } + + const seen = new Set(); + const normalized = []; + relays + .map((relay) => (typeof relay === "string" ? relay.trim() : "")) + .filter(Boolean) + .forEach((relay) => { + if (!seen.has(relay)) { + seen.add(relay); + normalized.push(relay); + } + }); + + normalized.sort(); + return normalized; +} + +function normalizeFilterValue(value) { + if (Array.isArray(value)) { + const normalizedArray = value.map((entry) => normalizeFilterValue(entry)); + const isPrimitiveArray = normalizedArray.every( + (entry) => typeof entry === "string" || typeof entry === "number", + ); + if (isPrimitiveArray) { + return normalizedArray + .slice() + .sort((a, b) => String(a).localeCompare(String(b))); + } + return normalizedArray; + } + + if (value && typeof value === "object") { + return normalizeFilter(value); + } + + return value; +} + +function normalizeFilter(filter) { + if (!filter || typeof filter !== "object") { + return {}; + } + + const entries = Object.entries(filter) + .filter(([key]) => typeof key === "string") + .sort(([a], [b]) => a.localeCompare(b)); + + const normalized = {}; + entries.forEach(([key, value]) => { + normalized[key] = normalizeFilterValue(value); + }); + + return normalized; +} + +function normalizeFilters(filters) { + if (!Array.isArray(filters)) { + return []; + } + + return filters.map((filter) => normalizeFilter(filter)); +} + +function stableStringify(value) { + if (Array.isArray(value)) { + return `[${value.map((entry) => stableStringify(entry)).join(",")}]`; + } + + if (value && typeof value === "object") { + const keys = Object.keys(value).sort(); + return `{${keys + .map((key) => `${JSON.stringify(key)}:${stableStringify(value[key])}`) + .join(",")}}`; + } + + return JSON.stringify(value); +} + +function buildSignature({ relays, filters }) { + return stableStringify({ + relays: normalizeRelayList(relays), + filters: normalizeFilters(filters), + }); +} + +class RelaySubscriptionService { + constructor() { + this.subscriptions = new Map(); + } + + ensureSubscription({ + key, + pool, + relays, + filters, + label = "subscription", + onEvent, + onEose, + onClose, + } = {}) { + if (!key || typeof key !== "string") { + return null; + } + + if (!pool || typeof pool.sub !== "function") { + devLogger.warn("[relaySubscriptions] Pool unavailable; skipping subscription", { + key, + label, + }); + return null; + } + + const signature = buildSignature({ relays, filters }); + const existing = this.subscriptions.get(key); + + if (existing && existing.signature === signature && existing.subscription) { + devLogger.log("[relaySubscriptions] Reusing existing subscription", { + key, + label, + relays: existing.relays, + }); + return existing.subscription; + } + + if (existing && existing.unsubscribe) { + devLogger.warn("[relaySubscriptions] Replacing existing subscription", { + key, + label, + }); + try { + existing.unsubscribe(); + } catch (error) { + userLogger.warn("[relaySubscriptions] Failed to unsubscribe existing subscription", error); + } + } + + const relayList = normalizeRelayList(relays); + const filterList = normalizeFilters(filters); + + if (!relayList.length || !filterList.length) { + devLogger.warn("[relaySubscriptions] Missing relays or filters; skipping subscription", { + key, + label, + relays: relayList, + filters: filterList, + }); + return null; + } + + let subscription; + try { + subscription = pool.sub(relayList, filterList); + } catch (error) { + userLogger.warn("[relaySubscriptions] Failed to create relay subscription", error); + return null; + } + + const unsubscribe = () => { + try { + if (typeof subscription?.unsub === "function") { + subscription.unsub(); + } + } catch (error) { + userLogger.warn("[relaySubscriptions] Failed to unsubscribe relay subscription", error); + } + }; + + if (subscription && typeof subscription.on === "function") { + if (typeof onEvent === "function") { + subscription.on("event", (event) => { + try { + onEvent(event); + } catch (error) { + userLogger.warn("[relaySubscriptions] Event handler threw", error); + } + }); + } + if (typeof onEose === "function") { + subscription.on("eose", () => { + try { + onEose(); + } catch (error) { + userLogger.warn("[relaySubscriptions] EOSE handler threw", error); + } + }); + } + if (typeof onClose === "function") { + subscription.on("close", (reasons) => { + try { + onClose(reasons); + } catch (error) { + userLogger.warn("[relaySubscriptions] Close handler threw", error); + } + }); + } + } + + this.subscriptions.set(key, { + signature, + subscription, + unsubscribe, + label, + relays: relayList, + filters: filterList, + }); + + devLogger.log("[relaySubscriptions] Subscription activated", { + key, + label, + relays: relayList, + filters: filterList, + }); + + return subscription; + } + + stopSubscription(key, reason = "") { + if (!key || typeof key !== "string") { + return; + } + + const existing = this.subscriptions.get(key); + if (!existing) { + return; + } + + try { + existing.unsubscribe?.(); + } finally { + this.subscriptions.delete(key); + devLogger.log("[relaySubscriptions] Subscription stopped", { + key, + label: existing.label, + reason, + }); + } + } + + getSignature({ relays, filters }) { + return buildSignature({ relays, filters }); + } +} + +export const relaySubscriptionService = new RelaySubscriptionService(); diff --git a/js/subscriptions.js b/js/subscriptions.js index 64fc3659..a660f9ea 100644 --- a/js/subscriptions.js +++ b/js/subscriptions.js @@ -32,6 +32,7 @@ import moderationService from "./services/moderationService.js"; import nostrService from "./services/nostrService.js"; import { profileCache } from "./state/profileCache.js"; import { runNip07WithRetry, NIP07_PRIORITY } from "./nostr/nip07Permissions.js"; +import { relaySubscriptionService } from "./services/relaySubscriptionService.js"; const SUBSCRIPTION_SET_KIND = getNostrEventSchema(NOTE_TYPES.SUBSCRIPTION_LIST)?.kind ?? 30000; @@ -342,6 +343,7 @@ class SubscriptionsManager { this.emitter = new TinyEventEmitter(); this.decryptRetryTimeoutId = null; this.ensureNostrServiceListener(); + this.subscriptionKey = null; profileCache.subscribe((event, detail) => { if (event === "profileChanged") { @@ -463,6 +465,17 @@ class SubscriptionsManager { } return; } + if (!nostrClient?.pool && typeof nostrClient?.ensurePool === "function") { + try { + await nostrClient.ensurePool(); + } catch (error) { + devLogger.warn( + "[SubscriptionsManager] Failed to initialize relay pool for subscription", + error, + ); + } + } + this.ensureSubscriptionListSubscription(normalizedUserPubkey, relayUrls); // Use incremental fetch helper const cachedSnapshot = parseCachedSubscriptionSnapshot( @@ -674,6 +687,7 @@ class SubscriptionsManager { } reset() { + this.stopSubscriptionListSubscription(); this.subscribedPubkeys.clear(); this.subsEventId = null; this.subsEventCreatedAt = null; @@ -690,6 +704,55 @@ class SubscriptionsManager { this.emitter.emit("change", { action: "reset", subscribedPubkeys: [] }); } + stopSubscriptionListSubscription() { + if (this.subscriptionKey) { + relaySubscriptionService.stopSubscription(this.subscriptionKey, "reset"); + this.subscriptionKey = null; + } + } + + ensureSubscriptionListSubscription(userPubkey, relays) { + const normalized = normalizeHexPubkey(userPubkey); + if (!normalized || !nostrClient?.pool) { + return null; + } + + const key = `subscriptions:${normalized}`; + this.subscriptionKey = key; + + const filters = [ + { + kinds: [SUBSCRIPTION_SET_KIND], + authors: [normalized], + "#d": [SUBSCRIPTION_LIST_IDENTIFIER], + }, + ]; + + return relaySubscriptionService.ensureSubscription({ + key, + pool: nostrClient.pool, + relays, + filters, + label: "subscription-list", + onEvent: (event) => this.handleSubscriptionListEvent(event), + }); + } + + handleSubscriptionListEvent(event) { + const normalized = normalizeHexPubkey(event?.pubkey); + if (!normalized || normalized !== this.currentUserPubkey) { + return; + } + + this.updateFromRelays(normalized, { allowPermissionPrompt: false }) + .catch((error) => { + devLogger.warn( + "[SubscriptionsManager] Failed to refresh after subscription list event", + error, + ); + }); + } + async ensureLoaded(actorHex, options = {}) { devLogger.log("[SubscriptionsManager] ensureLoaded start", actorHex); const normalizedActor = normalizeHexPubkey(actorHex) || actorHex; diff --git a/js/userBlocks.js b/js/userBlocks.js index e0c28723..e82d9b92 100644 --- a/js/userBlocks.js +++ b/js/userBlocks.js @@ -21,6 +21,7 @@ import { profileCache } from "./state/profileCache.js"; import { DEFAULT_RELAY_URLS } from "./nostr/toolkit.js"; import { relayManager } from "./relayManager.js"; import { runNip07WithRetry, NIP07_PRIORITY } from "./nostr/nip07Permissions.js"; +import { relaySubscriptionService } from "./services/relaySubscriptionService.js"; class TinyEventEmitter { constructor() { @@ -682,6 +683,7 @@ class UserBlockListManager { this.decryptRetryTimeoutId = null; this.loadPromise = null; this.loadingPubkey = null; + this.blockListSubscriptionKey = null; profileCache.subscribe((event, detail) => { if (event === "profileChanged") { @@ -696,6 +698,7 @@ class UserBlockListManager { } reset() { + this.stopBlockListSubscription(); this.blockedPubkeys.clear(); this.blockEventId = null; this.blockEventCreatedAt = null; @@ -711,6 +714,54 @@ class UserBlockListManager { this.loadingPubkey = null; } + stopBlockListSubscription() { + if (this.blockListSubscriptionKey) { + relaySubscriptionService.stopSubscription( + this.blockListSubscriptionKey, + "reset", + ); + this.blockListSubscriptionKey = null; + } + } + + ensureBlockListSubscription(userPubkey, relays) { + const normalized = normalizeHex(userPubkey); + if (!normalized || !nostrClient?.pool) { + return null; + } + + const key = `block-list:${normalized}`; + this.blockListSubscriptionKey = key; + + const filters = [ + { kinds: [10000], authors: [normalized] }, + { kinds: [30002], authors: [normalized], "#d": [BLOCK_LIST_IDENTIFIER] }, + ]; + + return relaySubscriptionService.ensureSubscription({ + key, + pool: nostrClient.pool, + relays, + filters, + label: "block-list", + onEvent: (event) => this.handleBlockListEvent(event), + }); + } + + handleBlockListEvent(event) { + const normalized = normalizeHex(event?.pubkey); + if (!normalized || normalized !== this.activePubkey) { + return; + } + if (!isUserBlockListEvent(event)) { + return; + } + + this.loadBlocks(normalized).catch((error) => { + devLogger.warn("[UserBlockList] Failed to refresh after block list event", error); + }); + } + on(eventName, handler) { return this.emitter.on(eventName, handler); } @@ -844,6 +895,14 @@ class UserBlockListManager { const readRelays = relayManager.getReadRelayUrls(); const relays = readRelays.length > 0 ? readRelays : Array.from(DEFAULT_RELAY_URLS); emitStatus({ status: "loading", relays }); + if (!nostrClient?.pool && typeof nostrClient?.ensurePool === "function") { + try { + await nostrClient.ensurePool(); + } catch (error) { + devLogger.warn("[UserBlockList] Failed to initialize relay pool for subscription", error); + } + } + this.ensureBlockListSubscription(normalized, relays); const localBlocks = this._loadLocal(normalized); const hasLocalData = !!localBlocks;