|
| 1 | +// Active relay-health prober — refreshes the `relays` directory the /relays |
| 2 | +// page renders. Unlike the subscription hoses this is *active*: it dials every |
| 3 | +// known relay, measures reachability + latency, reads the NIP-11 info document |
| 4 | +// for auth/payment requirements, and rolls up uptime. Kind-less, so it is not a |
| 5 | +// hose — it's a self-scheduling module of its own. |
| 6 | +// |
| 7 | +// The relay universe is the `relays` collection itself, continuously seeded by |
| 8 | +// the follows/relay-lists hoses' URL harvesting. |
| 9 | +import { connect } from './db.js'; |
| 10 | +import { ensureRelayDirectoryIndex } from './hoses/relays.js'; |
| 11 | +import 'dotenv/config'; |
| 12 | + |
| 13 | +const RELAY_DIRECTORY = process.env.MONGO_RELAY_DIRECTORY_COLLECTION || 'relays'; |
| 14 | + |
| 15 | +export const USAGE = `beacon relay-health prober |
| 16 | +
|
| 17 | +Usage: node probe.js [flags] (or: node src/prober.js [flags]) |
| 18 | +
|
| 19 | +Flags (override the matching env var): |
| 20 | + --once run a single sweep and exit (env RUN_ONCE=1) |
| 21 | + --concurrency <n> relays probed in parallel (env PROBE_CONCURRENCY, default 25) |
| 22 | + --timeout <ms> per-relay connect/HTTP timeout (env PROBE_TIMEOUT, default 7000) |
| 23 | + --interval <ms> sweep interval when scheduling (env PROBE_INTERVAL, default 3600000) |
| 24 | + -h, --help show this help`; |
| 25 | + |
| 26 | +/** Parse prober CLI flags into an overlay (only keys the user passed). */ |
| 27 | +export function parseArgs(argv = process.argv.slice(2)) { |
| 28 | + const out = {}; |
| 29 | + const numNext = (i) => { const v = Number(argv[i + 1]); return Number.isFinite(v) ? v : undefined; }; |
| 30 | + for (let i = 0; i < argv.length; i++) { |
| 31 | + const a = argv[i]; |
| 32 | + if (a === '-h' || a === '--help') out.help = true; |
| 33 | + else if (a === '--once') out.once = true; |
| 34 | + else if (a === '--concurrency') { const v = numNext(i); if (v !== undefined) { out.concurrency = v; i++; } } |
| 35 | + else if (a === '--timeout') { const v = numNext(i); if (v !== undefined) { out.timeout = v; i++; } } |
| 36 | + else if (a === '--interval') { const v = numNext(i); if (v !== undefined) { out.interval = v; i++; } } |
| 37 | + } |
| 38 | + return out; |
| 39 | +} |
| 40 | + |
| 41 | +/** Resolve config from env + CLI flags (flags win). Pure — inputs passed in. */ |
| 42 | +export function proberConfig(env = process.env, argv = process.argv.slice(2)) { |
| 43 | + const flags = parseArgs(argv); |
| 44 | + const pos = (v, d) => { const n = Number(v); return Number.isFinite(n) && n > 0 ? n : d; }; |
| 45 | + return { |
| 46 | + interval: flags.interval ?? pos(env.PROBE_INTERVAL, 3_600_000), |
| 47 | + concurrency: flags.concurrency ?? pos(env.PROBE_CONCURRENCY, 25), |
| 48 | + timeout: flags.timeout ?? pos(env.PROBE_TIMEOUT, 7_000), |
| 49 | + once: flags.once || env.RUN_ONCE === '1' || env.RUN_ONCE === 'true', |
| 50 | + help: !!flags.help, |
| 51 | + }; |
| 52 | +} |
| 53 | + |
| 54 | +/** wss://… -> https://…, ws://… -> http://… (for the NIP-11 fetch). null if unparseable. */ |
| 55 | +export function relayInfoUrl(wsUrl) { |
| 56 | + try { |
| 57 | + const u = new URL(wsUrl); |
| 58 | + if (u.protocol === 'wss:') u.protocol = 'https:'; |
| 59 | + else if (u.protocol === 'ws:') u.protocol = 'http:'; |
| 60 | + else return null; |
| 61 | + return u.toString(); |
| 62 | + } catch { return null; } |
| 63 | +} |
| 64 | + |
| 65 | +/** Extract the health-relevant flags from a NIP-11 relay info document. */ |
| 66 | +export function nip11Flags(info) { |
| 67 | + const lim = (info && typeof info === 'object' && info.limitation) || {}; |
| 68 | + return { requiresAuth: !!lim.auth_required, requiresPayment: !!lim.payment_required }; |
| 69 | +} |
| 70 | + |
| 71 | +/** Open a WebSocket; resolve { online, responseTime, error }. Never rejects. */ |
| 72 | +export function checkRelay(url, timeoutMs) { |
| 73 | + return new Promise((resolve) => { |
| 74 | + const start = Date.now(); |
| 75 | + let done = false, ws, timer; |
| 76 | + const finish = (r) => { if (done) return; done = true; clearTimeout(timer); try { ws?.close(); } catch {} resolve(r); }; |
| 77 | + timer = setTimeout(() => finish({ online: false, responseTime: null, error: 'timeout' }), timeoutMs); |
| 78 | + try { ws = new WebSocket(url); } catch (e) { return finish({ online: false, responseTime: null, error: String(e?.message || e) }); } |
| 79 | + ws.addEventListener('open', () => finish({ online: true, responseTime: Date.now() - start, error: null })); |
| 80 | + ws.addEventListener('error', (e) => finish({ online: false, responseTime: null, error: e?.message || 'error' })); |
| 81 | + }); |
| 82 | +} |
| 83 | + |
| 84 | +/** Fetch + parse the NIP-11 info doc. Returns flags, or null on any failure. */ |
| 85 | +export async function fetchNip11(url, timeoutMs) { |
| 86 | + const httpUrl = relayInfoUrl(url); |
| 87 | + if (!httpUrl) return null; |
| 88 | + const ctrl = new AbortController(); |
| 89 | + const timer = setTimeout(() => ctrl.abort(), timeoutMs); |
| 90 | + try { |
| 91 | + const res = await fetch(httpUrl, { headers: { Accept: 'application/nostr+json' }, signal: ctrl.signal }); |
| 92 | + if (!res.ok) return null; |
| 93 | + return nip11Flags(await res.json()); |
| 94 | + } catch { return null; } |
| 95 | + finally { clearTimeout(timer); } |
| 96 | +} |
| 97 | + |
| 98 | +/** Probe one relay and persist the result (uptime rolled up from running totals). */ |
| 99 | +export async function probeRelay(db, relay, cfg) { |
| 100 | + const conn = await checkRelay(relay, cfg.timeout); |
| 101 | + const nip11 = conn.online ? await fetchNip11(relay, cfg.timeout) : null; |
| 102 | + await db.collection(RELAY_DIRECTORY).updateOne({ relay }, [ |
| 103 | + { |
| 104 | + $set: { |
| 105 | + lastChecked: '$$NOW', |
| 106 | + online: conn.online, |
| 107 | + responseTime: conn.responseTime, |
| 108 | + lastError: conn.error, |
| 109 | + ...(nip11 ? { requiresAuth: nip11.requiresAuth, requiresPayment: nip11.requiresPayment } : {}), |
| 110 | + checksTotal: { $add: [{ $ifNull: ['$checksTotal', 0] }, 1] }, |
| 111 | + checksOnline: { $add: [{ $ifNull: ['$checksOnline', 0] }, conn.online ? 1 : 0] }, |
| 112 | + }, |
| 113 | + }, |
| 114 | + { $set: { uptime: { $cond: [{ $gt: ['$checksTotal', 0] }, { $multiply: [{ $divide: ['$checksOnline', '$checksTotal'] }, 100] }, null] } } }, |
| 115 | + ], { upsert: true }); |
| 116 | + return conn; |
| 117 | +} |
| 118 | + |
| 119 | +// Run `fn` over `items` with at most `concurrency` in flight. |
| 120 | +async function mapPool(items, concurrency, fn) { |
| 121 | + let i = 0; |
| 122 | + const worker = async () => { while (i < items.length) { const idx = i++; await fn(items[idx], idx); } }; |
| 123 | + await Promise.all(Array.from({ length: Math.max(1, Math.min(concurrency, items.length)) }, worker)); |
| 124 | +} |
| 125 | + |
| 126 | +/** One full sweep over every relay in the directory. Returns { total, online }. */ |
| 127 | +export async function sweepOnce(db, cfg) { |
| 128 | + const docs = await db.collection(RELAY_DIRECTORY).find({}, { projection: { relay: 1, _id: 0 } }).toArray(); |
| 129 | + const relays = [...new Set(docs.map((d) => d.relay).filter(Boolean))]; |
| 130 | + let online = 0; |
| 131 | + await mapPool(relays, cfg.concurrency, async (relay) => { |
| 132 | + try { if ((await probeRelay(db, relay, cfg)).online) online++; } |
| 133 | + catch (e) { console.error(`[beacon] probe error ${relay}: ${e.message}`); } |
| 134 | + }); |
| 135 | + console.log(`[beacon] relay sweep: ${online}/${relays.length} online`); |
| 136 | + return { total: relays.length, online }; |
| 137 | +} |
| 138 | + |
| 139 | +export async function runProber() { |
| 140 | + const cfg = proberConfig(); |
| 141 | + const db = await connect(); |
| 142 | + await ensureRelayDirectoryIndex(db); |
| 143 | + if (cfg.once) { await sweepOnce(db, cfg); return { stop() {} }; } |
| 144 | + console.log(`[beacon] relay-health prober: every ${Math.round(cfg.interval / 60000)}min, concurrency ${cfg.concurrency}, timeout ${cfg.timeout}ms`); |
| 145 | + await sweepOnce(db, cfg); |
| 146 | + const timer = setInterval(() => sweepOnce(db, cfg).catch((e) => console.error('[beacon] sweep error:', e.message)), cfg.interval); |
| 147 | + const stop = () => clearInterval(timer); |
| 148 | + process.on('SIGINT', () => { stop(); process.exit(0); }); |
| 149 | + process.on('SIGTERM', () => { stop(); process.exit(0); }); |
| 150 | + return { stop }; |
| 151 | +} |
| 152 | + |
| 153 | +if (import.meta.url === `file://${process.argv[1]}`) { |
| 154 | + if (proberConfig().help) { console.log(USAGE); process.exit(0); } |
| 155 | + runProber(); |
| 156 | +} |
0 commit comments