From 2e07ddb35f62124bebe5f7431e1453f56f250484 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Mon, 19 Jan 2026 08:12:03 -0500 Subject: [PATCH] Add DM-specific nostr relay service --- js/services/dmNostrService.js | 567 ++++++++++++++++++++++++++++++++++ 1 file changed, 567 insertions(+) create mode 100644 js/services/dmNostrService.js diff --git a/js/services/dmNostrService.js b/js/services/dmNostrService.js new file mode 100644 index 00000000..00fc3eea --- /dev/null +++ b/js/services/dmNostrService.js @@ -0,0 +1,567 @@ +import { ensureNostrTools, resolveSimplePoolConstructor, shimLegacySimplePoolMethods } from "../nostr/toolkit.js"; +import { publishEventToRelay } from "../nostrPublish.js"; +import logger from "../utils/logger.js"; + +const DEFAULT_BACKOFF = { + baseMs: 1000, + maxMs: 30000, + jitterRatio: 0.3, +}; + +const MAX_SEEN_EVENT_IDS = 5000; + +function normalizeRelayList(relays) { + if (!Array.isArray(relays)) { + return []; + } + + return Array.from( + new Set( + relays + .map((relay) => (typeof relay === "string" ? relay.trim() : "")) + .filter(Boolean), + ), + ); +} + +function normalizePubkey(value) { + if (typeof value !== "string") { + return ""; + } + return value.trim().toLowerCase(); +} + +function normalizeLogger(candidate) { + if ( + candidate && + typeof candidate === "object" && + candidate.dev && + candidate.user + ) { + return candidate; + } + + return logger; +} + +function clampNumber(value, minimum, maximum) { + if (!Number.isFinite(value)) { + return minimum; + } + + return Math.min(Math.max(value, minimum), maximum); +} + +function resolveBackoffConfig(custom = {}) { + return { + baseMs: clampNumber(custom.baseMs ?? DEFAULT_BACKOFF.baseMs, 250, 60000), + maxMs: clampNumber(custom.maxMs ?? DEFAULT_BACKOFF.maxMs, 1000, 300000), + jitterRatio: clampNumber(custom.jitterRatio ?? DEFAULT_BACKOFF.jitterRatio, 0, 1), + }; +} + +function applyJitter(value, ratio) { + if (!Number.isFinite(value) || value <= 0) { + return 0; + } + if (!Number.isFinite(ratio) || ratio <= 0) { + return value; + } + const jitter = value * ratio; + return Math.round(value + (Math.random() * jitter - jitter / 2)); +} + +class DmNostrService { + constructor({ + relays = [], + preferredRelays = [], + backoff = {}, + log = null, + } = {}) { + this.relays = normalizeRelayList(relays); + this.preferredRelays = normalizeRelayList(preferredRelays); + this.pool = null; + this.poolPromise = null; + this.logger = normalizeLogger(log || logger); + this.backoffConfig = resolveBackoffConfig(backoff); + this.relayStates = new Map(); + this.relayWatermarks = new Map(); + this.relayTimers = new Map(); + this.seenEventIds = new Set(); + this.seenEventOrder = []; + this.subscriptions = new Map(); + this.activeFilters = []; + this.actorPubkey = ""; + this.handlers = { + onEvent: null, + onEose: null, + onError: null, + }; + } + + async ensurePool() { + if (this.pool) { + return this.pool; + } + + if (this.poolPromise) { + return this.poolPromise; + } + + this.poolPromise = (async () => { + const tools = await ensureNostrTools(); + const SimplePool = resolveSimplePoolConstructor(tools); + if (typeof SimplePool !== "function") { + throw new Error("nostr-tools SimplePool unavailable"); + } + + const pool = new SimplePool(); + shimLegacySimplePoolMethods(pool); + this.pool = pool; + this.poolPromise = null; + return pool; + })(); + + try { + return await this.poolPromise; + } catch (error) { + this.poolPromise = null; + this.logger.user.warn("[dmNostrService] Failed to initialize pool.", error); + throw error; + } + } + + getRelayStatus(relayUrl) { + const relay = this.relayStates.get(relayUrl); + return relay ? { ...relay } : null; + } + + listRelayStatuses() { + return Array.from(this.relayStates.entries()).map(([url, state]) => ({ + url, + ...state, + })); + } + + async connect({ relays = null } = {}) { + if (Array.isArray(relays)) { + this.relays = normalizeRelayList(relays); + } + + if (!this.relays.length) { + this.logger.user.warn("[dmNostrService] No relays configured for DM service."); + return; + } + + await this.ensurePool(); + + await Promise.all( + this.relays.map(async (relayUrl) => { + await this.connectRelay(relayUrl); + }), + ); + } + + async connectRelay(relayUrl) { + if (!this.pool) { + await this.ensurePool(); + } + + if (!relayUrl) { + return; + } + + const relayState = this.ensureRelayState(relayUrl); + relayState.lastConnectAttempt = Date.now(); + + if (typeof this.pool.ensureRelay === "function") { + try { + const relay = await this.pool.ensureRelay(relayUrl); + relayState.connected = Boolean(relay); + if (relay) { + this.attachRelayListeners(relayUrl, relay); + this.recordRelaySuccess(relayUrl); + } + } catch (error) { + this.recordRelayFailure(relayUrl, error); + this.scheduleReconnect(relayUrl, "connect-failed"); + } + } else { + relayState.connected = true; + this.recordRelaySuccess(relayUrl); + } + } + + attachRelayListeners(relayUrl, relay) { + if (!relay || typeof relay.on !== "function") { + return; + } + + try { + relay.on("connect", () => { + this.recordRelaySuccess(relayUrl); + }); + relay.on("disconnect", () => { + this.recordRelayFailure(relayUrl, new Error("relay-disconnect")); + this.scheduleReconnect(relayUrl, "relay-disconnect"); + }); + relay.on("error", (error) => { + this.recordRelayFailure(relayUrl, error); + this.scheduleReconnect(relayUrl, "relay-error"); + }); + } catch (error) { + this.logger.dev.warn("[dmNostrService] Failed to attach relay listeners.", error); + } + } + + ensureRelayState(relayUrl) { + if (!this.relayStates.has(relayUrl)) { + this.relayStates.set(relayUrl, { + connected: false, + okCount: 0, + failCount: 0, + lastOkAt: 0, + lastFailAt: 0, + lastConnectAttempt: 0, + backoffMs: 0, + attempts: 0, + }); + } + + return this.relayStates.get(relayUrl); + } + + recordRelaySuccess(relayUrl) { + const relayState = this.ensureRelayState(relayUrl); + relayState.connected = true; + relayState.okCount += 1; + relayState.lastOkAt = Date.now(); + relayState.backoffMs = 0; + relayState.attempts = 0; + } + + recordRelayFailure(relayUrl, error) { + const relayState = this.ensureRelayState(relayUrl); + relayState.connected = false; + relayState.failCount += 1; + relayState.lastFailAt = Date.now(); + + this.logger.dev.warn("[dmNostrService] Relay failure", { + relayUrl, + error, + }); + } + + scheduleReconnect(relayUrl, reason) { + const relayState = this.ensureRelayState(relayUrl); + relayState.attempts += 1; + + const backoffBase = this.backoffConfig.baseMs * 2 ** (relayState.attempts - 1); + const backoffMs = applyJitter( + Math.min(backoffBase, this.backoffConfig.maxMs), + this.backoffConfig.jitterRatio, + ); + + relayState.backoffMs = backoffMs; + + if (this.relayTimers.has(relayUrl)) { + clearTimeout(this.relayTimers.get(relayUrl)); + } + + this.logger.dev.info("[dmNostrService] Scheduling relay reconnect", { + relayUrl, + reason, + backoffMs, + }); + + const timer = setTimeout(() => { + this.relayTimers.delete(relayUrl); + this.resubscribeRelay(relayUrl); + }, backoffMs); + + this.relayTimers.set(relayUrl, timer); + } + + buildFilters({ actorPubkey, authors = [], recipients = [] } = {}) { + const normalizedActor = normalizePubkey(actorPubkey || this.actorPubkey); + const authorSet = new Set( + [normalizedActor, ...authors].map(normalizePubkey).filter(Boolean), + ); + const recipientSet = new Set( + [normalizedActor, ...recipients].map(normalizePubkey).filter(Boolean), + ); + + const filters = []; + + if (authorSet.size) { + filters.push({ + kinds: [4], + authors: Array.from(authorSet), + }); + } + + if (recipientSet.size) { + filters.push({ + kinds: [4], + "#p": Array.from(recipientSet), + }); + } + + return filters; + } + + updateWatermark(relayUrl, event) { + if (!event || typeof event.created_at !== "number") { + return; + } + + const current = this.relayWatermarks.get(relayUrl) || 0; + if (event.created_at > current) { + this.relayWatermarks.set(relayUrl, event.created_at); + } + } + + hasSeenEvent(eventId) { + if (!eventId) { + return false; + } + return this.seenEventIds.has(eventId); + } + + rememberEvent(eventId) { + if (!eventId || this.seenEventIds.has(eventId)) { + return; + } + + this.seenEventIds.add(eventId); + this.seenEventOrder.push(eventId); + + if (this.seenEventOrder.length > MAX_SEEN_EVENT_IDS) { + const oldest = this.seenEventOrder.shift(); + if (oldest) { + this.seenEventIds.delete(oldest); + } + } + } + + shouldProcessEvent(relayUrl, event) { + if (!event || typeof event.id !== "string") { + return false; + } + + if (this.hasSeenEvent(event.id)) { + return false; + } + + const watermark = this.relayWatermarks.get(relayUrl) || 0; + if (typeof event.created_at === "number" && event.created_at < watermark) { + return false; + } + + return true; + } + + async subscribe({ + actorPubkey, + authors, + recipients, + relays, + onEvent, + onEose, + onError, + } = {}) { + if (typeof onEvent === "function") { + this.handlers.onEvent = onEvent; + } + if (typeof onEose === "function") { + this.handlers.onEose = onEose; + } + if (typeof onError === "function") { + this.handlers.onError = onError; + } + + if (typeof actorPubkey === "string" && actorPubkey.trim()) { + this.actorPubkey = normalizePubkey(actorPubkey); + } + + if (Array.isArray(relays)) { + this.relays = normalizeRelayList(relays); + } + + if (!this.relays.length) { + this.logger.user.warn("[dmNostrService] Cannot subscribe without relays."); + return; + } + + this.activeFilters = this.buildFilters({ + actorPubkey: this.actorPubkey, + authors, + recipients, + }); + + if (!this.activeFilters.length) { + this.logger.user.warn("[dmNostrService] Cannot subscribe without filters."); + return; + } + + await this.connect(); + + this.relays.forEach((relayUrl) => { + this.subscribeRelay(relayUrl); + }); + } + + subscribeRelay(relayUrl) { + if (!this.pool || !relayUrl) { + return; + } + + const since = this.relayWatermarks.get(relayUrl); + const filters = this.activeFilters.map((filter) => { + if (typeof since === "number" && since > 0) { + return { + ...filter, + since: Math.max(0, since - 1), + }; + } + return { ...filter }; + }); + + if (this.subscriptions.has(relayUrl)) { + this.unsubscribeRelay(relayUrl); + } + + let subscription; + try { + subscription = this.pool.sub([relayUrl], filters); + } catch (error) { + this.recordRelayFailure(relayUrl, error); + this.scheduleReconnect(relayUrl, "subscribe-failed"); + return; + } + + if (!subscription || typeof subscription.on !== "function") { + this.recordRelayFailure(relayUrl, new Error("subscription-unavailable")); + this.scheduleReconnect(relayUrl, "subscribe-unavailable"); + return; + } + + subscription.on("event", (event) => { + if (!this.shouldProcessEvent(relayUrl, event)) { + return; + } + + this.rememberEvent(event.id); + this.updateWatermark(relayUrl, event); + this.recordRelaySuccess(relayUrl); + + if (typeof this.handlers.onEvent === "function") { + try { + this.handlers.onEvent(event, { relayUrl }); + } catch (error) { + this.logger.dev.warn("[dmNostrService] onEvent handler failed.", error); + } + } + }); + + subscription.on("eose", () => { + if (typeof this.handlers.onEose === "function") { + try { + this.handlers.onEose({ relayUrl }); + } catch (error) { + this.logger.dev.warn("[dmNostrService] onEose handler failed.", error); + } + } + }); + + subscription.on("close", () => { + this.recordRelayFailure(relayUrl, new Error("subscription-closed")); + this.scheduleReconnect(relayUrl, "subscription-closed"); + }); + + this.subscriptions.set(relayUrl, subscription); + } + + resubscribeRelay(relayUrl) { + if (!relayUrl) { + return; + } + + this.subscribeRelay(relayUrl); + } + + unsubscribeRelay(relayUrl) { + const subscription = this.subscriptions.get(relayUrl); + if (subscription && typeof subscription.unsub === "function") { + try { + subscription.unsub(); + } catch (error) { + this.logger.dev.warn("[dmNostrService] Failed to unsubscribe relay.", error); + } + } + this.subscriptions.delete(relayUrl); + } + + stop() { + for (const relayUrl of this.subscriptions.keys()) { + this.unsubscribeRelay(relayUrl); + } + + for (const timer of this.relayTimers.values()) { + clearTimeout(timer); + } + + this.relayTimers.clear(); + } + + async publishSignedEvent(signedEvent, { relays = null } = {}) { + if (!signedEvent || typeof signedEvent !== "object") { + throw new Error("signed-event-required"); + } + + if (signedEvent.kind !== 4) { + throw new Error("invalid-event-kind"); + } + + const relayList = normalizeRelayList( + Array.isArray(relays) && relays.length + ? relays + : this.preferredRelays.length + ? this.preferredRelays + : this.relays, + ); + + if (!relayList.length) { + throw new Error("no-relays-configured"); + } + + await this.ensurePool(); + + const results = await Promise.all( + relayList.map(async (relayUrl) => { + let result; + try { + result = await publishEventToRelay(this.pool, relayUrl, signedEvent); + } catch (error) { + this.recordRelayFailure(relayUrl, error); + return { success: false, error, relayUrl }; + } + + if (result?.success) { + this.recordRelaySuccess(relayUrl); + } else { + this.recordRelayFailure( + relayUrl, + result?.error || new Error("publish-failed"), + ); + } + + return { ...result, relayUrl }; + }), + ); + + return results; + } +} + +export default DmNostrService;