beacon: firehose phase 2 — follows hose (kind 3) + relay harvesting

- src/hoses/event.js: extract the shared structural + schnorr-signature
  check (eventId, verifySignature); profiles hose refactored to use it.
- src/hoses/follows.js (kind 3): verify signature, derive the social-graph
  shape { pubkey, follows:[hex…], created_at, count } from p tags (64-hex,
  lowercased, deduped), latest-wins upsert (the legacy followshose blindly
  replaced — an out-of-order older list could clobber a newer one). Owns
  its indexes (pubkey, follows, created_at). Harvests relay URLs from the
  legacy relay map in kind-3 content into the relays directory via
  $setOnInsert (discovery only — never overwrites health metrics;
  best-effort, never fails ingest).
- src/indexer.js: register the follows hose; LEGACY_KINDS drops to [10002].
  The switch removes kind 3 from the legacy set automatically.
- tests: new follows-hose tests (verify/parse/harvest/canonicalize);
  plan tests updated for the now-hose-owned kind 3.

Fixes in-degree for kind-3 events the legacy raw-upsert path wrote without
a top-level follows array. npm test 34/34.

Closes #9

Author: melvincarvalho

README.md

@@ -41,7 +41,12 @@ npm run serve               # read API only
 
 v0 — MVP: indexer + read API + the regression-compatible schema, plus the DID-document resolution endpoint (`/.well-known/did/nostr/:pubkey.json` via jss's `buildDidDocument`) and a `/relays` health directory.
 
-The indexer is being reworked into composable **hoses** (`src/hoses/`). Phase 1 is the **profiles hose** (kind 0): it schnorr-verifies each event before storing it, so a relay can't inject a forged `did:nostr` profile, and owns its Mongo indexes (incl. the `content_text` search index). Follows (kind 3) and relay lists (kind 10002) still use the raw upsert path until their phases.
+The indexer is being reworked into composable **hoses** (`src/hoses/`), each owning a set of event kinds, all sharing one schnorr signature check (`src/hoses/event.js`) so a relay can't inject forged data. Which hoses run is a switch — the `HOSES` env (comma list of names; default: all).
+
+- **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) still use the raw upsert path until their phase.
 
 ## License
 

src/hoses/event.js

@@ -0,0 +1,34 @@
+// Shared Nostr event verification used by every hose.
+//
+// A hose layers kind- and content-specific rules on top of this: the checks
+// here are common to all events — well-formed pubkey/created_at/sig, an `id`
+// that matches the canonical serialization, and a valid schnorr signature.
+import { schnorr } from '@noble/curves/secp256k1.js';
+import { sha256 } from '@noble/hashes/sha2.js';
+import { hexToBytes, bytesToHex } from '@noble/hashes/utils.js';
+
+const HEX64 = /^[0-9a-f]{64}$/;
+const HEX128 = /^[0-9a-f]{128}$/;
+const enc = new TextEncoder();
+
+// NIP-01 event id: sha256 of [0, pubkey, created_at, kind, tags, content].
+export function eventId(e) {
+  const serial = JSON.stringify([0, e.pubkey, e.created_at, e.kind, e.tags || [], e.content ?? '']);
+  return bytesToHex(sha256(enc.encode(serial)));
+}
+
+/**
+ * Structural + cryptographic validation common to every event. Verifies the
+ * pubkey/created_at/sig are well-formed, the claimed `id` (if present) matches
+ * the content, and the schnorr signature checks out. Does NOT inspect `kind`
+ * or content semantics — callers add those.
+ */
+export function verifySignature(e) {
+  if (!e || typeof e.pubkey !== 'string' || !HEX64.test(e.pubkey)) return false;
+  if (!Number.isFinite(e.created_at)) return false;
+  if (typeof e.sig !== 'string' || !HEX128.test(e.sig)) return false;
+  const id = eventId(e);
+  if (e.id && e.id !== id) return false; // claimed id must match the content
+  try { return schnorr.verify(hexToBytes(e.sig), hexToBytes(id), hexToBytes(e.pubkey)); }
+  catch { return false; }
+}

src/hoses/follows.js

@@ -0,0 +1,94 @@
+// The follows hose — kind 3 (contact list / social graph).
+//
+// Stores the *derived* nostr-beacon shape `{ pubkey, follows: [hex…],
+// created_at, count }` (not the raw event): the in-degree query
+// (`followerCount`) and `enrichCounts` both rely on the top-level `follows`
+// array + `count`. As a side-effect it harvests relay URLs from the legacy
+// 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 { 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) {
+  if (!e || e.kind !== KIND) return false;
+  return verifySignature(e);
+}
+
+/** Followed pubkeys from `p` tags — 64-hex only, lowercased, deduped. */
+export function parseFollows(event) {
+  const out = new Set();
+  if (Array.isArray(event?.tags)) {
+    for (const t of event.tags) {
+      if (t?.[0] === 'p' && typeof t[1] === 'string') {
+        const hex = t[1].toLowerCase();
+        if (HEX64.test(hex)) out.add(hex);
+      }
+    }
+  }
+  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];
+}
+
+export default {
+  name: 'follows',
+  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({ 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);
+  },
+
+  /** Verify, latest-wins upsert of the derived shape, then harvest 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;
+    const follows = parseFollows(event);
+    await col.replaceOne(
+      { pubkey: event.pubkey },
+      { 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(() => {});
+    }
+    return true;
+  },
+};

src/hoses/profiles.js

@@ -10,39 +10,20 @@
 // verifies each event's schnorr signature before trusting it — beacon is an
 // identity/SSO substrate, so a malicious relay must not be able to inject a
 // forged did:nostr profile.
-import { schnorr } from '@noble/curves/secp256k1.js';
-import { sha256 } from '@noble/hashes/sha2.js';
-import { hexToBytes, bytesToHex } from '@noble/hashes/utils.js';
+import { verifySignature } from './event.js';
 import { COLLECTIONS } from '../db.js';
 
 const KIND = 0;
-const HEX64 = /^[0-9a-f]{64}$/;
-const HEX128 = /^[0-9a-f]{128}$/;
-const enc = new TextEncoder();
-
-// NIP-01 event id: sha256 of the canonical serialization
-// [0, pubkey, created_at, kind, tags, content].
-function eventId(e) {
-  const serial = JSON.stringify([0, e.pubkey, e.created_at, e.kind, e.tags || [], e.content ?? '']);
-  return bytesToHex(sha256(enc.encode(serial)));
-}
 
 /**
- * Structural + cryptographic validation of a raw kind-0 event.
- * Rejects wrong kind, malformed fields, non-JSON content, a tampered `id`,
- * and any event whose schnorr signature doesn't verify against its pubkey.
+ * Validate a raw kind-0 event: right kind, JSON-parseable content (empty is
+ * allowed, treated as {}), plus the shared structural + schnorr-signature
+ * check. Rejects forged or tampered profiles so a relay can't inject one.
  */
 export function verifyEvent(e) {
   if (!e || e.kind !== KIND) return false;
-  if (typeof e.pubkey !== 'string' || !HEX64.test(e.pubkey)) return false;
-  if (!Number.isFinite(e.created_at)) return false;
-  if (typeof e.sig !== 'string' || !HEX128.test(e.sig)) return false;
-  // kind-0 content is a JSON object; empty string is allowed (treated as {}).
   if (e.content) { try { JSON.parse(e.content); } catch { return false; } }
-  const id = eventId(e);
-  if (e.id && e.id !== id) return false; // claimed id must match the content
-  try { return schnorr.verify(hexToBytes(e.sig), hexToBytes(id), hexToBytes(e.pubkey)); }
-  catch { return false; }
+  return verifySignature(e);
 }
 
 export default {

src/indexer.js

@@ -8,19 +8,20 @@
 // double-writing against the legacy firehose processes during the migration.
 import { upsertEvent, connect } from './db.js';
 import profilesHose from './hoses/profiles.js';
+import followsHose from './hoses/follows.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];
+const ALL_HOSES = [profilesHose, followsHose];
 
-// Kinds not yet migrated to a hose: 3 (follows), 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/followshose processes.
-const LEGACY_KINDS = [3, 10002];
+// 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];
 
 /**
  * Resolve the ingest plan from the registered hoses + env switches. Pure (env

test/follows-hose.test.js

@@ -0,0 +1,79 @@
+// Follows hose: signature verification + p-tag derivation + relay harvesting.
+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, parseFollows, relayUrlsFrom, canonicalizeRelayUrl } from '../src/hoses/follows.js';
+
+const enc = new TextEncoder();
+const PRIV = hexToBytes('0000000000000000000000000000000000000000000000000000000000000003');
+const PUBKEY = bytesToHex(schnorr.getPublicKey(PRIV));
+const PK = (n) => String(n).padStart(64, '0'); // a fake 64-hex pubkey
+
+function signed({ tags = [], content = '', created_at = 1000, kind = 3 } = {}) {
+  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-3 event (empty content ok)', () => {
+  assert.equal(verifyEvent(signed()), true);
+});
+
+test('rejects the wrong kind', () => {
+  assert.equal(verifyEvent(signed({ kind: 0 })), false);
+});
+
+test('rejects a tampered event', () => {
+  const e = signed({ tags: [['p', PK(1)]] });
+  e.tags = [['p', PK(2)]]; // change 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('parseFollows: p tags only, 64-hex, lowercased, deduped', () => {
+  const e = signed({ tags: [
+    ['p', PK('a').toUpperCase()],   // uppercased -> lowercased
+    ['p', PK('a')],                 // duplicate -> deduped
+    ['p', PK('b')],
+    ['e', PK('c')],                 // non-p tag -> ignored
+    ['p', 'not-hex'],               // invalid -> dropped
+    ['p'],                          // malformed -> skipped
+  ] });
+  assert.deepEqual(parseFollows(e), [PK('a'), PK('b')]);
+});
+
+test('parseFollows: no tags -> empty', () => {
+  assert.deepEqual(parseFollows(signed()), []);
+});
+
+test('canonicalizeRelayUrl: trailing slash on origins, path preserved, junk dropped', () => {
+  assert.equal(canonicalizeRelayUrl('wss://relay.example.com'), 'wss://relay.example.com/');
+  assert.equal(canonicalizeRelayUrl('wss://relay.example.com/'), 'wss://relay.example.com/');
+  assert.equal(canonicalizeRelayUrl('wss://relay.example.com/inbox'), 'wss://relay.example.com/inbox');
+  assert.equal(canonicalizeRelayUrl('https://not-a-relay.com'), null);
+  assert.equal(canonicalizeRelayUrl('garbage'), null);
+});
+
+test('relayUrlsFrom: parses the legacy relay map, canonicalizes + dedupes', () => {
+  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/'],
+  );
+});
+
+test('relayUrlsFrom: empty / non-JSON content -> empty', () => {
+  assert.deepEqual(relayUrlsFrom(signed({ content: '' })), []);
+  assert.deepEqual(relayUrlsFrom(signed({ content: 'not json' })), []);
+});

test/indexer-plan.test.js

@@ -4,23 +4,24 @@ 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];          // mirrors today's registry
-const TWO = [profiles, follows]; // a future phase
+const ONE = [profiles];          // only profiles registered
+const TWO = [profiles, follows]; // the real registry
 
 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']);
   assert.equal(p.legacy, true);
-  assert.deepEqual(sorted(p.kinds), [0, 3, 10002]);
+  assert.deepEqual(sorted(p.kinds), [0, 10002]); // 10002 is the only legacy kind
   assert.deepEqual(p.unknown, []);
 });
 
 test('empty HOSES string falls back to the default (all on)', () => {
-  assert.deepEqual(sorted(planIngest(ONE, { HOSES: '' }).kinds), [0, 3, 10002]);
+  assert.deepEqual(sorted(planIngest(ONE, { HOSES: '' }).kinds), [0, 10002]);
 });
 
 test('INDEX_LEGACY_KINDS=0 → only enabled hoses\' kinds, no legacy', () => {
@@ -42,13 +43,13 @@ test('unknown hose names are reported and ignored', () => {
 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), [3, 10002]); // legacy fallback only
+  assert.deepEqual(sorted(p.kinds), [10002]); // legacy fallback only
 });
 
-test('a hose that owns a legacy kind removes it from the legacy set (no double-subscribe)', () => {
+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']);
-  assert.deepEqual(p.legacyKinds, [10002]); // 3 now owned by the follows hose
+  assert.deepEqual(p.legacyKinds, [10002]);
   assert.deepEqual(sorted(p.kinds), [0, 3, 10002]);
 });