Merge pull request #1954 from PR0M3TH3AN/codex/refactor-subscription-logic-and-add-logging

Centralize relay subscriptions and route profile fetches through metadata service
This commit is contained in:
thePR0M3TH3AN authored and GitHub committed 2026-01-29 16:32:23 -05:00
commit 8779cd2524
8 files changed
+554 -47

No files matched your search

+25 -23
View File
@@ -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}:`,
+32 -22
View File
@@ -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);
+64
View File
@@ -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;
+18 -2
View File
@@ -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) {
+43
View File
@@ -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);
}
}
},
});
}
+250
View File
@@ -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();
+63
View File
@@ -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;
+59
View File
@@ -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;