diff --git a/js/nostr/subscriptionManager.js b/js/nostr/subscriptionManager.js new file mode 100644 index 00000000..fcabb19d --- /dev/null +++ b/js/nostr/subscriptionManager.js @@ -0,0 +1,194 @@ +// SubscriptionManager (L1) — the single chokepoint for relay reads. +// +// Per docs/architecture-refactor.md, NOTHING outside this module should call +// pool.sub / pool.list directly. It generalizes relaySubscriptionService by: +// - defaulting to a bounded, health-gated relay set (so callers stop fanning +// every query across a user's full ~20-relay NIP-65 list), +// - deduping one-shot list() fetches that are identical and in-flight, +// - keeping a registry of active live subscriptions and re-issuing them on +// relay reconnect (instead of every subsystem re-subscribing on its own, +// which was the reconnect-storm amplifier). +// +// Callers are expected to pass BATCHED filters (one filter covering many ids, +// e.g. {kinds:[1984], "#e":[...all]}) — never one subscription per item. + +import { relaySubscriptionService } from "../services/relaySubscriptionService.js"; +import { devLogger, userLogger } from "../utils/logger.js"; + +const DEFAULT_LIST_TIMEOUT_MS = 8000; + +function normalizeRelays(relays) { + if (!Array.isArray(relays)) return []; + const seen = new Set(); + const out = []; + for (const r of relays) { + const url = typeof r === "string" ? r.trim() : ""; + if (url && !seen.has(url)) { + seen.add(url); + out.push(url); + } + } + return out; +} + +function stableSignature(value) { + if (Array.isArray(value)) { + return `[${value.map(stableSignature).join(",")}]`; + } + if (value && typeof value === "object") { + const keys = Object.keys(value).sort(); + return `{${keys.map((k) => `${JSON.stringify(k)}:${stableSignature(value[k])}`).join(",")}}`; + } + return JSON.stringify(value); +} + +/** + * @param {object} deps + * @param {() => object|null} deps.getPool returns the relay pool (pool.sub/pool.list) + * @param {() => string[]} deps.getDefaultRelays bounded, health-gated read relays + * @param {object} [deps.subscriptions] relaySubscriptionService (injectable for tests) + * @param {object} [deps.logger] + */ +export function createSubscriptionManager({ + getPool, + getDefaultRelays, + subscriptions = relaySubscriptionService, + logger = { dev: devLogger, user: userLogger }, +} = {}) { + // Active live subscriptions, keyed by logical `key`, for reconnect re-issue. + const active = new Map(); + // In-flight one-shot list() promises, keyed by signature, for dedup. + const inflightLists = new Map(); + + const resolveRelays = (relays) => { + const explicit = normalizeRelays(relays); + if (explicit.length) return explicit; + try { + return normalizeRelays(getDefaultRelays?.() || []); + } catch (error) { + logger.dev?.warn?.("[subscriptionManager] getDefaultRelays threw", error); + return []; + } + }; + + /** + * Open (or update) a live subscription. Deduped by `key`. Returns a handle. + */ + function subscribe({ key, filters, relays, onEvent, onEose, onClose, label = "sub" }) { + if (!key || typeof key !== "string") { + logger.dev?.warn?.("[subscriptionManager] subscribe requires a string key"); + return { update() {}, close() {} }; + } + const pool = getPool?.(); + const record = { key, filters, onEvent, onEose, onClose, label }; + active.set(key, record); + + const open = (nextFilters, nextRelays) => { + record.filters = nextFilters; + if (nextRelays !== undefined) record.relays = nextRelays; + const relayList = resolveRelays(record.relays); + subscriptions.ensureSubscription({ + key, + pool, + relays: relayList, + filters: record.filters, + label, + onEvent, + onEose, + onClose, + }); + }; + + open(filters, relays); + + return { + update({ filters: nextFilters, relays: nextRelays } = {}) { + open(nextFilters ?? record.filters, nextRelays); + }, + close() { + active.delete(key); + subscriptions.stopSubscription(key, "handle-close"); + }, + }; + } + + /** + * One-shot fetch (resolves on EOSE). Deduped while identical requests are + * in-flight, health-gated, with a timeout fallback so a dead relay set can't + * hang the caller forever. + */ + function list({ filters, relays, timeoutMs = DEFAULT_LIST_TIMEOUT_MS } = {}) { + const pool = getPool?.(); + if (!pool || typeof pool.list !== "function") { + logger.dev?.warn?.("[subscriptionManager] pool.list unavailable"); + return Promise.resolve([]); + } + const relayList = resolveRelays(relays); + if (!relayList.length) return Promise.resolve([]); + + const signature = stableSignature({ relays: relayList.slice().sort(), filters }); + const existing = inflightLists.get(signature); + if (existing) return existing; + + const promise = (async () => { + try { + const result = await Promise.race([ + Promise.resolve(pool.list(relayList, filters)), + new Promise((resolve) => setTimeout(() => resolve(null), timeoutMs)), + ]); + return Array.isArray(result) ? result : []; + } catch (error) { + logger.dev?.warn?.("[subscriptionManager] list() failed", error); + return []; + } finally { + inflightLists.delete(signature); + } + })(); + + inflightLists.set(signature, promise); + return promise; + } + + /** + * Re-issue all active live subscriptions. Call after a relay (re)connects so + * subscriptions dropped by a flaky relay are restored — once, centrally — + * instead of each subsystem re-subscribing independently. + */ + function handleReconnect() { + const pool = getPool?.(); + for (const record of active.values()) { + // The old subscription was dropped by the disconnected relay, but + // relaySubscriptionService still holds it (same signature) and would + // reuse the dead handle. Stop it first to force a fresh REQ. + subscriptions.stopSubscription(record.key, "reconnect"); + const relayList = resolveRelays(record.relays); + subscriptions.ensureSubscription({ + key: record.key, + pool, + relays: relayList, + filters: record.filters, + label: record.label, + onEvent: record.onEvent, + onEose: record.onEose, + onClose: record.onClose, + }); + } + } + + function closeAll(reason = "close-all") { + for (const key of Array.from(active.keys())) { + subscriptions.stopSubscription(key, reason); + } + active.clear(); + inflightLists.clear(); + } + + return { + subscribe, + list, + handleReconnect, + closeAll, + // introspection for tests/telemetry + getActiveKeys: () => Array.from(active.keys()), + }; +} diff --git a/js/services/relaySubscriptionService.js b/js/services/relaySubscriptionService.js index b5457449..ccdbebc2 100644 --- a/js/services/relaySubscriptionService.js +++ b/js/services/relaySubscriptionService.js @@ -89,7 +89,7 @@ function buildSignature({ relays, filters }) { }); } -class RelaySubscriptionService { +export class RelaySubscriptionService { constructor() { this.subscriptions = new Map(); } diff --git a/package.json b/package.json index 2e4aa53b..ad541440 100644 --- a/package.json +++ b/package.json @@ -17,7 +17,7 @@ "ci:beacon": "npm run build:beacon", "format": "prettier --cache --write css/tailwind.source.css css/tokens.css css/docs.css \"docs/**/*.html\" tailwind.config.cjs README.md CONTRIBUTING.md", "audit": "node scripts/daily-design-system-audit.mjs", - "lint": "npm run lint:css && npm run lint:hex && npm run lint:inline-styles && npm run lint:tokens && npm run lint:tailwind-brackets && npm run lint:tailwind-colors && npm run lint:file-size && npm run lint:innerhtml && npm run lint:assets && npm run lint:sw-compat", + "lint": "npm run lint:css && npm run lint:hex && npm run lint:inline-styles && npm run lint:tokens && npm run lint:tailwind-brackets && npm run lint:tailwind-colors && npm run lint:file-size && npm run lint:innerhtml && npm run lint:assets && npm run lint:sw-compat && npm run lint:pool-access", "lint:file-size": "node scripts/check-file-size.mjs", "lint:innerhtml": "node scripts/check-innerhtml.mjs", "lint:css": "stylelint css/tailwind.source.css css/tokens.css css/docs.css", @@ -59,7 +59,8 @@ "torch:lock": "npm run --prefix torch lock:lock", "torch:health": "npm run --prefix torch lock:health", "torch:memory:list": "node torch/bin/torch-lock.mjs list-memories", - "torch:memory:inspect": "node torch/bin/torch-lock.mjs inspect-memory" + "torch:memory:inspect": "node torch/bin/torch-lock.mjs inspect-memory", + "lint:pool-access": "node scripts/check-direct-pool-access.mjs" }, "devDependencies": { "@playwright/test": "^1.58.2", diff --git a/scripts/check-direct-pool-access.mjs b/scripts/check-direct-pool-access.mjs new file mode 100644 index 00000000..7e0d73b6 --- /dev/null +++ b/scripts/check-direct-pool-access.mjs @@ -0,0 +1,98 @@ +#!/usr/bin/env node + +/** + * Lint: enforce the SubscriptionManager (L1) chokepoint. + * + * Per docs/architecture-refactor.md, relay reads must go through + * js/nostr/subscriptionManager.js — nothing else should call pool.sub / + * pool.list / legacySub / legacyList / subscribeMap directly. + * + * Files that still do are GRANDFATHERED below; the allowlist must only SHRINK as + * phases migrate subsystems. A new file using direct pool access fails the lint; + * an allowlisted file is reported as a remaining migration target. + * + * Usage: + * node scripts/check-direct-pool-access.mjs # enforce (exit 1) + * node scripts/check-direct-pool-access.mjs --report # report only (exit 0) + */ + +import { readFileSync, readdirSync, statSync } from "node:fs"; +import { join, relative } from "node:path"; + +const ROOT = new URL("..", import.meta.url).pathname.replace(/\/$/, ""); +const SOURCE_DIR = join(ROOT, "js"); + +// Patterns that indicate a direct relay-read outside L1. +const PATTERNS = [/\.pool\.sub\(/, /\.pool\.list\(/, /\blegacySub\(/, /\blegacyList\(/, /\bsubscribeMap\(/]; + +// The only modules allowed to touch the pool directly. +const L1_FILES = new Set([ + "js/nostr/subscriptionManager.js", + "js/services/relaySubscriptionService.js", + "js/nostr/toolkit.js", // vendored pool shim (defines legacySub/legacyList) + "js/nostr/managers/ConnectionManager.js", // owns the pool lifecycle / health probe +]); + +// Grandfathered: not yet migrated. SHRINK this list as phases land; do not grow. +const GRANDFATHERED = new Set([ + "js/adminListStore.js", + "js/app.js", + "js/channelProfile.js", + "js/feedEngine/watchHistoryFeed.js", + "js/nostr/client.js", + "js/reactionCounter.js", + "js/relayManager.js", + "js/searchView.js", + "js/services/authService.js", + "js/services/dmNostrService.js", + "js/services/moderationService.js", + "js/services/nostrService.js", +]); + +function collectJsFiles(dir) { + const out = []; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + const full = join(dir, entry.name); + if (entry.isDirectory()) out.push(...collectJsFiles(full)); + else if (entry.isFile() && entry.name.endsWith(".js")) out.push(full); + } + return out; +} + +const reportOnly = process.argv.includes("--report"); +const violations = []; +const remaining = []; + +for (const filePath of collectJsFiles(SOURCE_DIR)) { + const rel = relative(ROOT, filePath); + if (rel.endsWith(".min.js") || L1_FILES.has(rel)) continue; + const content = readFileSync(filePath, "utf8"); + if (!PATTERNS.some((re) => re.test(content))) continue; + if (GRANDFATHERED.has(rel)) remaining.push(rel); + else violations.push(rel); +} + +// Allowlist entries that no longer need to be there (already migrated) — nudge. +const stale = [...GRANDFATHERED].filter( + (rel) => !remaining.includes(rel), +); + +if (remaining.length) { + console.log(`\n${remaining.length} file(s) still using direct pool access (migration targets):`); + for (const r of remaining.sort()) console.log(` • ${r}`); +} +if (stale.length) { + console.log(`\n${stale.length} allowlist entr(y/ies) can be REMOVED (already migrated):`); + for (const r of stale.sort()) console.log(` ✓ ${r}`); +} + +if (violations.length) { + console.error(`\n${violations.length} NEW direct pool access violation(s):`); + for (const v of violations.sort()) console.error(` ✗ ${v}`); + console.error( + "\nRoute relay reads through js/nostr/subscriptionManager.js (subscribe/list).", + ); + if (!reportOnly) process.exit(1); +} else { + console.log("\nNo new direct pool access. (Allowlist must only shrink.)"); +} diff --git a/tests/nostr/subscription-manager.test.mjs b/tests/nostr/subscription-manager.test.mjs new file mode 100644 index 00000000..5c42f2b6 --- /dev/null +++ b/tests/nostr/subscription-manager.test.mjs @@ -0,0 +1,97 @@ +// Scenario tests for the L1 SubscriptionManager (docs/architecture-refactor.md). +// Uses a fake pool + the REAL relaySubscriptionService, asserting externally +// observable transport behavior (REQ counts, relay targeting), not internals. + +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { createSubscriptionManager } from "../../js/nostr/subscriptionManager.js"; +import { RelaySubscriptionService } from "../../js/services/relaySubscriptionService.js"; + +function makeFakePool() { + let subCalls = 0; + let listCalls = 0; + const lastSubRelays = []; + return { + subCalls: () => subCalls, + listCalls: () => listCalls, + lastSubRelays: () => lastSubRelays, + sub(relays) { + subCalls += 1; + lastSubRelays.length = 0; + lastSubRelays.push(...relays); + const handlers = {}; + return { + on: (ev, cb) => { + handlers[ev] = cb; + }, + unsub: () => {}, + }; + }, + list() { + listCalls += 1; + return Promise.resolve([{ id: "evt-1" }]); + }, + }; +} + +function makeManager(pool, defaultRelays = ["wss://core1", "wss://core2"]) { + return createSubscriptionManager({ + getPool: () => pool, + getDefaultRelays: () => defaultRelays, + subscriptions: new RelaySubscriptionService(), // fresh per manager (isolation) + logger: { dev: { warn() {}, log() {} }, user: { warn() {} } }, + }); +} + +// SCN-subman-dedup: identical (key + filters) subscribe calls share ONE REQ. +test("subscribe dedups by key — identical calls do not reopen", () => { + const pool = makeFakePool(); + const m = makeManager(pool); + const f = [{ kinds: [1984], "#e": ["a".repeat(64)] }]; + m.subscribe({ key: "reports", filters: f }); + m.subscribe({ key: "reports", filters: f }); + assert.equal(pool.subCalls(), 1, "same key+filters must reuse one subscription"); +}); + +// SCN-subman-default-relays: with no relays given, uses the bounded core set. +test("subscribe targets the health-gated default relay set when none given", () => { + const pool = makeFakePool(); + const m = makeManager(pool, ["wss://core1", "wss://core2"]); + m.subscribe({ key: "videos", filters: [{ kinds: [30078] }] }); + assert.deepEqual(pool.lastSubRelays(), ["wss://core1", "wss://core2"]); +}); + +// SCN-subman-batched-update: updating the id set re-REQs once (not per id). +test("handle.update re-issues a single batched subscription", () => { + const pool = makeFakePool(); + const m = makeManager(pool); + const h = m.subscribe({ key: "reports", filters: [{ kinds: [1984], "#e": ["a".repeat(64)] }] }); + h.update({ filters: [{ kinds: [1984], "#e": ["a".repeat(64), "b".repeat(64)] }] }); + assert.equal(pool.subCalls(), 2, "one initial + one updated REQ — still O(1), not per-id"); +}); + +// SCN-subman-list-dedup: identical in-flight list() requests share one REQ. +test("list dedups identical in-flight requests", async () => { + const pool = makeFakePool(); + const m = makeManager(pool); + const filters = [{ kinds: [0], authors: ["a".repeat(64)] }]; + const [r1, r2] = await Promise.all([m.list({ filters }), m.list({ filters })]); + assert.equal(pool.listCalls(), 1, "concurrent identical lists must coalesce to one REQ"); + assert.deepEqual(r1, r2); +}); + +// SCN-subman-reconnect: reconnect re-issues active subs once, centrally. +test("handleReconnect re-issues active subscriptions (and not closed ones)", () => { + const pool = makeFakePool(); + const m = makeManager(pool); + m.subscribe({ key: "reports", filters: [{ kinds: [1984] }] }); + const vids = m.subscribe({ key: "videos", filters: [{ kinds: [30078] }] }); + assert.equal(pool.subCalls(), 2); + + vids.close(); // closed subs must NOT come back on reconnect + m.handleReconnect(); + + // Only "reports" should be re-issued -> exactly one more sub call. + assert.equal(pool.subCalls(), 3, "only the still-active sub is re-issued on reconnect"); + assert.deepEqual(m.getActiveKeys(), ["reports"]); +});