Merge pull request #12 from JavaScriptSolidServer/issue-11-relaylists-hose

Firehose phase 3: relay-lists hose (kind 10002) — retires the last legacy kind

Author: melvincarvalho

README.md

@@ -45,8 +45,9 @@ The indexer is being reworked into composable **hoses** (`src/hoses/`), each own
 
 - **Profiles** (kind 0) — verifies + stores the raw event latest-wins, owns its indexes incl. the `content_text` search index.
 - **Follows** (kind 3) — verifies + stores the derived social-graph shape `{ pubkey, follows:[hex…], count }` (so in-degree works), and harvests relay URLs from legacy kind-3 content into the `relays` directory.
+- **Relay lists** (kind 10002) — verifies + stores the raw event latest-wins (the shape the DID document's `service` entries are built from), and harvests the `r`-tag URLs into the `relays` directory (the canonical relay-URL source).
 
-Relay lists (kind 10002) still use the raw upsert path until their phase.
+Every kind is now hose-owned; the legacy raw-upsert fallback is empty.
 
 ## License
 

src/hoses/follows.js

@@ -7,11 +7,11 @@
 // relay map some clients still put in kind-3 `content`, seeding the `relays`
 // directory (discovery only — it never overwrites health metrics).
 import { verifySignature } from './event.js';
+import { harvestRelays, ensureRelayDirectoryIndex } from './relays.js';
 import { COLLECTIONS } from '../db.js';
 
 const KIND = 3;
 const HEX64 = /^[0-9a-f]{64}$/;
-const RELAY_DIRECTORY = process.env.MONGO_RELAY_DIRECTORY_COLLECTION || 'relays';
 
 /** Validate a kind-3 event. Unlike kind 0, content is optional / free-form. */
 export function verifyEvent(e) {
@@ -33,26 +33,13 @@ export function parseFollows(event) {
   return [...out];
 }
 
-// Trailing slash on bare origins (so `wss://r.com` and `wss://r.com/` dedupe);
-// keep an explicit path as-is. Non-ws(s) or unparseable URLs are dropped.
-export function canonicalizeRelayUrl(url) {
-  try {
-    const u = new URL(url);
-    if (u.protocol !== 'wss:' && u.protocol !== 'ws:') return null;
-    if (u.pathname === '' || u.pathname === '/') return `${u.protocol}//${u.host}/`;
-    return url;
-  } catch { return null; }
-}
-
 /** Relay URLs from the legacy NIP-65-ish relay map in kind-3 content. */
 export function relayUrlsFrom(event) {
   if (!event?.content) return [];
   let map;
   try { map = JSON.parse(event.content); } catch { return []; }
   if (!map || typeof map !== 'object') return [];
-  const out = new Set();
-  for (const k of Object.keys(map)) { const u = canonicalizeRelayUrl(k); if (u) out.add(u); }
-  return [...out];
+  return Object.keys(map);
 }
 
 export default {
@@ -65,7 +52,7 @@ export default {
     await col.createIndex({ pubkey: 1 }).catch(ok);
     await col.createIndex({ follows: 1 }).catch(ok); // reverse lookup / in-degree
     await col.createIndex({ created_at: -1 }).catch(ok);
-    await db.collection(RELAY_DIRECTORY).createIndex({ relay: 1 }).catch(ok);
+    await ensureRelayDirectoryIndex(db);
   },
 
   /** Verify, latest-wins upsert of the derived shape, then harvest relays. */
@@ -80,15 +67,7 @@ export default {
       { pubkey: event.pubkey, follows, created_at: event.created_at, count: follows.length },
       { upsert: true },
     );
-    // Discovery side-effect: seed newly-seen relay URLs without touching any
-    // existing health fields. Best-effort — never fail ingest on it.
-    const urls = relayUrlsFrom(event);
-    if (urls.length) {
-      await db.collection(RELAY_DIRECTORY).bulkWrite(
-        urls.map((relay) => ({ updateOne: { filter: { relay }, update: { $setOnInsert: { relay } }, upsert: true } })),
-        { ordered: false },
-      ).catch(() => {});
-    }
+    await harvestRelays(db, relayUrlsFrom(event)); // discovery; best-effort
     return true;
   },
 };

src/hoses/relaylists.js

@@ -0,0 +1,47 @@
+// The relay-lists hose — kind 10002 (NIP-65 relay list metadata).
+//
+// Stores the raw event latest-wins in `relay_lists` (the shape diddoc reads to
+// build the DID document's `service`/`Relay` entries). As a side-effect it
+// harvests the `r`-tag URLs into the `relays` directory — the canonical source
+// of relay URLs, which the relay-health prober (phase 4) checks.
+import { verifySignature } from './event.js';
+import { harvestRelays, ensureRelayDirectoryIndex } from './relays.js';
+import { COLLECTIONS } from '../db.js';
+
+const KIND = 10002;
+
+/** Validate a kind-10002 event. Content is free-form; relays live in `r` tags. */
+export function verifyEvent(e) {
+  if (!e || e.kind !== KIND) return false;
+  return verifySignature(e);
+}
+
+/** Relay URLs from `r` tags (NIP-65). */
+export function relayUrlsFrom(event) {
+  if (!Array.isArray(event?.tags)) return [];
+  return event.tags.filter((t) => t?.[0] === 'r' && typeof t[1] === 'string').map((t) => t[1]);
+}
+
+export default {
+  name: 'relaylists',
+  kinds: [KIND],
+
+  async ensureIndexes(db) {
+    const ok = (e) => { if (e?.code !== 85 && e?.code !== 86) throw e; };
+    const col = db.collection(COLLECTIONS[KIND]);
+    await col.createIndex({ pubkey: 1 }).catch(ok);
+    await col.createIndex({ created_at: -1 }).catch(ok);
+    await ensureRelayDirectoryIndex(db);
+  },
+
+  /** Verify, latest-wins upsert of the raw event, then harvest its relays. */
+  async ingest(event, db) {
+    if (!verifyEvent(event)) return false;
+    const col = db.collection(COLLECTIONS[KIND]);
+    const existing = await col.findOne({ pubkey: event.pubkey }, { projection: { created_at: 1 } });
+    if (existing && existing.created_at >= event.created_at) return false;
+    await col.updateOne({ pubkey: event.pubkey }, { $set: { ...event } }, { upsert: true });
+    await harvestRelays(db, relayUrlsFrom(event)); // discovery; best-effort
+    return true;
+  },
+};

src/hoses/relays.js

@@ -0,0 +1,36 @@
+// Shared relay-directory helpers, used by any hose that discovers relay URLs
+// (follows harvests them from legacy kind-3 content; relay-lists from kind-3…
+// sorry, kind-10002 `r` tags). Discovery only seeds the URL — it never touches
+// the health metrics the relay-health prober owns.
+const RELAY_DIRECTORY = process.env.MONGO_RELAY_DIRECTORY_COLLECTION || 'relays';
+
+// Trailing slash on bare origins (so `wss://r.com` and `wss://r.com/` dedupe);
+// keep an explicit path as-is. Non-ws(s) or unparseable URLs return null.
+export function canonicalizeRelayUrl(url) {
+  try {
+    const u = new URL(url);
+    if (u.protocol !== 'wss:' && u.protocol !== 'ws:') return null;
+    if (u.pathname === '' || u.pathname === '/') return `${u.protocol}//${u.host}/`;
+    return url;
+  } catch { return null; }
+}
+
+export async function ensureRelayDirectoryIndex(db) {
+  await db.collection(RELAY_DIRECTORY).createIndex({ relay: 1 })
+    .catch((e) => { if (e?.code !== 85 && e?.code !== 86) throw e; });
+}
+
+/**
+ * Best-effort discovery: seed newly-seen relay URLs into the directory without
+ * touching any existing fields ($setOnInsert). Canonicalizes + dedupes first.
+ * Never throws — discovery must never fail an ingest. Returns the count seeded.
+ */
+export async function harvestRelays(db, urls) {
+  const uniq = [...new Set((urls || []).map(canonicalizeRelayUrl).filter(Boolean))];
+  if (!uniq.length) return 0;
+  await db.collection(RELAY_DIRECTORY).bulkWrite(
+    uniq.map((relay) => ({ updateOne: { filter: { relay }, update: { $setOnInsert: { relay } }, upsert: true } })),
+    { ordered: false },
+  ).catch(() => {});
+  return uniq.length;
+}

src/indexer.js

@@ -9,33 +9,32 @@
 import { upsertEvent, connect } from './db.js';
 import profilesHose from './hoses/profiles.js';
 import followsHose from './hoses/follows.js';
+import relaylistsHose from './hoses/relaylists.js';
 import 'dotenv/config';
 
 const RELAYS = (process.env.RELAYS || 'wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net')
   .split(',').map((s) => s.trim()).filter(Boolean);
 
 // All hoses that exist (grows each phase). `planIngest` selects which run.
-const ALL_HOSES = [profilesHose, followsHose];
-
-// Kinds not yet migrated to a hose: 10002 (relay lists). They use the legacy
-// raw-upsert path, on by default for local parity. Set INDEX_LEGACY_KINDS=0 to
-// turn it off so a single-hose deploy never writes collections still owned by
-// the legacy firehose processes.
-const LEGACY_KINDS = [10002];
+// Every event kind now has a hose, so there are no legacy kinds left — the
+// raw-upsert fallback and its INDEX_LEGACY_KINDS switch remain only as a safety
+// net for any kind a future phase adds before its hose lands.
+const ALL_HOSES = [profilesHose, followsHose, relaylistsHose];
+const LEGACY_KINDS = [];
 
 /**
- * Resolve the ingest plan from the registered hoses + env switches. Pure (env
- * passed in) so it's unit-testable. Returns the active hoses, a kind→hose
- * lookup, the union of kinds to subscribe to, and the legacy-fallback state.
+ * Resolve the ingest plan from the registered hoses + env switches. Pure (all
+ * inputs passed in) so it's unit-testable. Returns the active hoses, a
+ * kind→hose lookup, the union of kinds to subscribe to, and the legacy state.
  */
-export function planIngest(allHoses = ALL_HOSES, env = process.env) {
+export function planIngest(allHoses = ALL_HOSES, env = process.env, legacyKindsAll = LEGACY_KINDS) {
   const want = (env.HOSES || allHoses.map((h) => h.name).join(','))
     .split(',').map((s) => s.trim()).filter(Boolean);
   const hoses = allHoses.filter((h) => want.includes(h.name));
   const unknown = want.filter((n) => !allHoses.some((h) => h.name === n));
   const legacy = env.INDEX_LEGACY_KINDS !== '0' && env.INDEX_LEGACY_KINDS !== 'false';
   const hoseFor = (kind) => hoses.find((h) => h.kinds.includes(kind));
-  const legacyKinds = legacy ? LEGACY_KINDS.filter((k) => !hoseFor(k)) : [];
+  const legacyKinds = legacy ? legacyKindsAll.filter((k) => !hoseFor(k)) : [];
   const kinds = [...new Set([...hoses.flatMap((h) => h.kinds), ...legacyKinds])];
   return { hoses, hoseFor, kinds, legacy, legacyKinds, unknown };
 }

test/follows-hose.test.js

@@ -4,7 +4,8 @@ import assert from 'node:assert/strict';
 import { schnorr } from '@noble/curves/secp256k1.js';
 import { sha256 } from '@noble/hashes/sha2.js';
 import { hexToBytes, bytesToHex } from '@noble/hashes/utils.js';
-import { verifyEvent, parseFollows, relayUrlsFrom, canonicalizeRelayUrl } from '../src/hoses/follows.js';
+import { verifyEvent, parseFollows, relayUrlsFrom } from '../src/hoses/follows.js';
+import { canonicalizeRelayUrl } from '../src/hoses/relays.js';
 
 const enc = new TextEncoder();
 const PRIV = hexToBytes('0000000000000000000000000000000000000000000000000000000000000003');
@@ -60,17 +61,13 @@ test('canonicalizeRelayUrl: trailing slash on origins, path preserved, junk drop
   assert.equal(canonicalizeRelayUrl('garbage'), null);
 });
 
-test('relayUrlsFrom: parses the legacy relay map, canonicalizes + dedupes', () => {
+test('relayUrlsFrom: returns the raw relay-map keys (canon/dedup is harvestRelays\' job)', () => {
   const content = JSON.stringify({
     'wss://relay.example.com': { read: true, write: true },
-    'wss://relay.example.com/': { read: true, write: false }, // dedupes with above
     'wss://other.example.com/inbox': { read: true, write: true },
-    'https://bad.example.com': {}, // dropped (not ws/wss)
   });
-  assert.deepEqual(
-    relayUrlsFrom(signed({ content })).sort(),
-    ['wss://other.example.com/inbox', 'wss://relay.example.com/'],
-  );
+  assert.deepEqual(relayUrlsFrom(signed({ content })),
+    ['wss://relay.example.com', 'wss://other.example.com/inbox']);
 });
 
 test('relayUrlsFrom: empty / non-JSON content -> empty', () => {

test/indexer-plan.test.js

@@ -1,59 +1,60 @@
 // The HOSES / INDEX_LEGACY_KINDS switch: planIngest selects which hoses run
-// and which kinds get subscribed. Pure function — test with fake hoses.
+// and which kinds get subscribed. Pure function — test with fake hoses and an
+// explicit legacy-kind set (so these stay stable as real hoses are added).
 import test from 'node:test';
 import assert from 'node:assert/strict';
 import { planIngest } from '../src/indexer.js';
 
-// Fakes — the real LEGACY_KINDS is [10002] (kinds 0 and 3 are now hose-owned).
 const profiles = { name: 'profiles', kinds: [0] };
 const follows = { name: 'follows', kinds: [3] };
-const ONE = [profiles];          // only profiles registered
-const TWO = [profiles, follows]; // the real registry
+const ALL = [profiles, follows];
+const LEGACY = [10002]; // a kind with no hose yet, for these scenarios
 
 const sorted = (a) => [...a].sort((x, y) => x - y);
 
 test('default (no env): all registered hoses on + legacy kinds', () => {
-  const p = planIngest(ONE, {});
-  assert.deepEqual(p.hoses.map((h) => h.name), ['profiles']);
+  const p = planIngest(ALL, {}, LEGACY);
+  assert.deepEqual(p.hoses.map((h) => h.name), ['profiles', 'follows']);
   assert.equal(p.legacy, true);
-  assert.deepEqual(sorted(p.kinds), [0, 10002]); // 10002 is the only legacy kind
+  assert.deepEqual(sorted(p.kinds), [0, 3, 10002]);
   assert.deepEqual(p.unknown, []);
 });
 
 test('empty HOSES string falls back to the default (all on)', () => {
-  assert.deepEqual(sorted(planIngest(ONE, { HOSES: '' }).kinds), [0, 10002]);
+  assert.deepEqual(sorted(planIngest(ALL, { HOSES: '' }, LEGACY).kinds), [0, 3, 10002]);
+});
+
+test('HOSES selects a subset', () => {
+  const p = planIngest(ALL, { HOSES: 'profiles' }, LEGACY);
+  assert.deepEqual(p.hoses.map((h) => h.name), ['profiles']);
+  assert.deepEqual(sorted(p.kinds), [0, 10002]); // profiles + legacy
 });
 
 test('INDEX_LEGACY_KINDS=0 → only enabled hoses\' kinds, no legacy', () => {
-  const p = planIngest(ONE, { HOSES: 'profiles', INDEX_LEGACY_KINDS: '0' });
+  const p = planIngest(ALL, { HOSES: 'profiles', INDEX_LEGACY_KINDS: '0' }, LEGACY);
   assert.equal(p.legacy, false);
   assert.deepEqual(p.kinds, [0]);
 });
 
 test('INDEX_LEGACY_KINDS=false is also off', () => {
-  assert.equal(planIngest(ONE, { INDEX_LEGACY_KINDS: 'false' }).legacy, false);
+  assert.equal(planIngest(ALL, { INDEX_LEGACY_KINDS: 'false' }, LEGACY).legacy, false);
 });
 
 test('unknown hose names are reported and ignored', () => {
-  const p = planIngest(ONE, { HOSES: 'profiles,nope' });
+  const p = planIngest(ALL, { HOSES: 'profiles,nope' }, LEGACY);
   assert.deepEqual(p.hoses.map((h) => h.name), ['profiles']);
   assert.deepEqual(p.unknown, ['nope']);
 });
 
-test('selecting only an unknown hose leaves no hose kinds (legacy still applies)', () => {
-  const p = planIngest(ONE, { HOSES: 'nope' });
-  assert.deepEqual(p.hoses, []);
-  assert.deepEqual(sorted(p.kinds), [10002]); // legacy fallback only
-});
-
-test('the real two-hose registry: profiles + follows own 0 and 3, legacy = 10002', () => {
-  const p = planIngest(TWO, {}); // follows owns kind 3
-  assert.deepEqual(p.hoses.map((h) => h.name), ['profiles', 'follows']);
+test('a hose that owns a legacy kind removes it from the legacy set', () => {
+  // follows owns kind 3 — if 3 were also "legacy" it must not double-subscribe.
+  const p = planIngest(ALL, {}, [3, 10002]);
   assert.deepEqual(p.legacyKinds, [10002]);
   assert.deepEqual(sorted(p.kinds), [0, 3, 10002]);
 });
 
-test('single-hose deploy: HOSES=follows + no legacy → just kind 3', () => {
-  const p = planIngest(TWO, { HOSES: 'follows', INDEX_LEGACY_KINDS: '0' });
-  assert.deepEqual(p.kinds, [3]);
+test('no legacy kinds (the real phase-3 state): just the hose kinds', () => {
+  const p = planIngest(ALL, {}, []);
+  assert.deepEqual(p.legacyKinds, []);
+  assert.deepEqual(sorted(p.kinds), [0, 3]);
 });

test/relaylists-hose.test.js

@@ -0,0 +1,50 @@
+// Relay-lists hose: signature verification + r-tag extraction.
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { schnorr } from '@noble/curves/secp256k1.js';
+import { sha256 } from '@noble/hashes/sha2.js';
+import { hexToBytes, bytesToHex } from '@noble/hashes/utils.js';
+import { verifyEvent, relayUrlsFrom } from '../src/hoses/relaylists.js';
+
+const enc = new TextEncoder();
+const PRIV = hexToBytes('0000000000000000000000000000000000000000000000000000000000000004');
+const PUBKEY = bytesToHex(schnorr.getPublicKey(PRIV));
+
+function signed({ tags = [], content = '', created_at = 1000, kind = 10002 } = {}) {
+  const base = { pubkey: PUBKEY, created_at, kind, tags, content };
+  const id = bytesToHex(sha256(enc.encode(JSON.stringify([0, base.pubkey, base.created_at, base.kind, base.tags, base.content]))));
+  const sig = bytesToHex(schnorr.sign(hexToBytes(id), PRIV));
+  return { ...base, id, sig };
+}
+
+test('accepts a validly-signed kind-10002 event', () => {
+  assert.equal(verifyEvent(signed({ tags: [['r', 'wss://relay.example.com']] })), true);
+});
+
+test('rejects the wrong kind', () => {
+  assert.equal(verifyEvent(signed({ kind: 3 })), false);
+});
+
+test('rejects a tampered event', () => {
+  const e = signed({ tags: [['r', 'wss://relay.example.com']] });
+  e.tags = [['r', 'wss://evil.example.com']]; // changed after signing
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects malformed input without throwing', () => {
+  for (const bad of [null, undefined, {}, 42, 'x']) assert.equal(verifyEvent(bad), false);
+});
+
+test('relayUrlsFrom: pulls r-tag URLs, ignores other tags', () => {
+  const e = signed({ tags: [
+    ['r', 'wss://relay.example.com'],
+    ['r', 'wss://other.example.com/', 'read'], // marker is fine
+    ['L', 'pink.momostr'],                     // non-r tag ignored
+    ['r'],                                      // malformed skipped
+  ] });
+  assert.deepEqual(relayUrlsFrom(e), ['wss://relay.example.com', 'wss://other.example.com/']);
+});
+
+test('relayUrlsFrom: no tags -> empty', () => {
+  assert.deepEqual(relayUrlsFrom(signed()), []);
+});