Merge pull request #8 from JavaScriptSolidServer/issue-7-profiles-hose

Firehose phase 1: validated profiles hose (kind 0) + extensible hose seam

Author: melvincarvalho

.env.example

@@ -10,3 +10,11 @@ MONGO_RELAY_DIRECTORY_COLLECTION=relays
 
 RELAYS=wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net
 PORT=3000
+
+# Firehose switch: which hoses run (comma list of names; default: all).
+# e.g. HOSES=profiles to run only the profiles hose during a phased rollout.
+# HOSES=profiles
+# Legacy raw-upsert path for kinds not yet migrated to a hose (3=follows,
+# 10002=relay lists). On by default; set to 0 to run a single hose without
+# writing collections still owned by the legacy firehose processes.
+# INDEX_LEGACY_KINDS=1

README.md

@@ -39,7 +39,9 @@ npm run serve               # read API only
 
 ## Status
 
-v0 — MVP: indexer + read API + the regression-compatible schema. DID-document resolution endpoint (`/.well-known/did/nostr/:pubkey.json` via jss's `buildDidDocument`) is the next step, per the RFC.
+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.
 
 ## License
 

package-lock.json

@@ -3,14 +3,18 @@
   "version": "0.0.1",
   "type": "module",
   "description": "did:nostr social-graph indexer (profiles + follows + relay lists) on MongoDB. Same schema as nostr-beacon for regression; DID-doc generation delegates to jss buildDidDocument per RFC.",
-  "bin": { "beacon": "index.js" },
+  "bin": {
+    "beacon": "index.js"
+  },
   "scripts": {
     "test": "node --test",
     "start": "node index.js",
     "index": "node -e \"import('./src/indexer.js').then(m => m.runIndexer())\"",
     "serve": "node -e \"import('./src/server.js').then(m => m.startServer())\""
   },
   "dependencies": {
+    "@noble/curves": "^2.2.0",
+    "@noble/hashes": "^2.2.0",
     "dotenv": "^16.4.5",
     "express": "^4.19.2",
     "mongodb": "^6.8.0"

package.json

@@ -0,0 +1,72 @@
+// The profiles hose — first concrete "hose" on the firehose indexer.
+//
+// A hose is a small, self-contained ingest module: it declares the event
+// `kinds` it wants, ensures its own Mongo indexes, and ingests one raw event
+// at a time. The indexer subscribes to the union of all hoses' kinds and
+// dispatches each event to the matching hose. Later phases add sibling hoses
+// (follows, relay lists, relay health) without touching the indexer's core.
+//
+// This hose handles kind 0 (profile metadata). Unlike a blind upsert, it
+// 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 { 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.
+ */
+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; }
+}
+
+export default {
+  name: 'profiles',
+  kinds: [KIND],
+
+  // Own the collection's indexes so a fresh deploy is fully functional:
+  // pubkey (lookup/dedupe), created_at (recency), and the text index that
+  // /api/search relies on. Tolerate a pre-existing index spec (codes 85/86).
+  async ensureIndexes(db) {
+    const col = db.collection(COLLECTIONS[KIND]);
+    const ok = (e) => { if (e?.code !== 85 && e?.code !== 86) throw e; };
+    await col.createIndex({ pubkey: 1 }).catch(ok);
+    await col.createIndex({ created_at: -1 }).catch(ok);
+    await col.createIndex({ content: 'text' }, { name: 'content_text' }).catch(ok);
+  },
+
+  /** Verify, then latest-wins upsert of the raw event. Returns true if stored. */
+  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 });
+    return true;
+  },
+};

src/hoses/profiles.js

@@ -2,24 +2,52 @@
 // Nostr is a simple protocol over WS: send ["REQ", subId, filter], receive
 // ["EVENT", subId, event]. Node 24 has a built-in global WebSocket.
 //
-// Subscribes to kind 0 (profile), 3 (follows / social graph), 10002 (relay
-// list) and upserts each raw event into Mongo.
+// Ingestion is composed of "hoses" (see src/hoses/), each owning a set of
+// event kinds. Which hoses run is a switch — the `HOSES` env (comma list of
+// names; default: all registered) — so a deploy can run a single hose without
+// double-writing against the legacy firehose processes during the migration.
 import { upsertEvent, connect } from './db.js';
+import profilesHose from './hoses/profiles.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);
 
-const KINDS = [0, 3, 10002];
+// All hoses that exist (grows each phase). `planIngest` selects which run.
+const ALL_HOSES = [profilesHose];
+
+// 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];
+
+/**
+ * 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.
+ */
+export function planIngest(allHoses = ALL_HOSES, env = process.env) {
+  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 kinds = [...new Set([...hoses.flatMap((h) => h.kinds), ...legacyKinds])];
+  return { hoses, hoseFor, kinds, legacy, legacyKinds, unknown };
+}
+
 const SUB_ID = 'beacon';
 const RECONNECT_MS = 3000;
 
-function connectRelay(url, onEvent) {
+function connectRelay(url, kinds, onEvent) {
   let ws, closed = false, timer;
   const open = () => {
     if (closed) return;
     ws = new WebSocket(url);
-    ws.addEventListener('open', () => ws.send(JSON.stringify(['REQ', SUB_ID, { kinds: KINDS }])));
+    ws.addEventListener('open', () => ws.send(JSON.stringify(['REQ', SUB_ID, { kinds }])));
     ws.addEventListener('message', (m) => {
       try {
         const msg = JSON.parse(typeof m.data === 'string' ? m.data : m.data.toString());
@@ -34,16 +62,25 @@ function connectRelay(url, onEvent) {
 }
 
 export async function runIndexer() {
-  await connect();
-  console.log(`[beacon] indexing kinds ${KINDS.join(',')} from ${RELAYS.length} relays`);
+  const db = await connect();
+  const { hoses, hoseFor, kinds, legacy, unknown } = planIngest();
+  if (unknown.length) console.warn(`[beacon] unknown hose(s) in HOSES, ignored: ${unknown.join(', ')}`);
+  if (!kinds.length) { console.warn('[beacon] no hoses enabled and no legacy kinds — nothing to index'); return { stop() {} }; }
+  for (const h of hoses) await h.ensureIndexes(db);
+  console.log(`[beacon] hoses: [${hoses.map((h) => h.name).join(', ') || 'none'}]  legacy kinds: ${legacy ? 'on' : 'off'}  subscribing kinds ${kinds.join(',')} from ${RELAYS.length} relays`);
   const onEvent = async (event) => {
     try {
-      if (await upsertEvent(event)) console.log(`[beacon] kind ${event.kind}  ${String(event.pubkey).slice(0, 12)}…`);
+      const hose = hoseFor(event.kind);
+      let stored;
+      if (hose) stored = await hose.ingest(event, db);
+      else if (legacy) stored = await upsertEvent(event);
+      else return; // not an enabled kind
+      if (stored) console.log(`[beacon] kind ${event.kind}  ${String(event.pubkey).slice(0, 12)}…`);
     } catch (e) {
-      console.error('[beacon] upsert error:', e.message);
+      console.error('[beacon] ingest error:', e.message);
     }
   };
-  const stoppers = RELAYS.map((url) => connectRelay(url, onEvent));
+  const stoppers = RELAYS.map((url) => connectRelay(url, kinds, onEvent));
   const stop = () => stoppers.forEach((s) => s());
   process.on('SIGINT', () => { stop(); process.exit(0); });
   process.on('SIGTERM', () => { stop(); process.exit(0); });

src/indexer.js

@@ -0,0 +1,58 @@
+// The HOSES / INDEX_LEGACY_KINDS switch: planIngest selects which hoses run
+// and which kinds get subscribed. Pure function — test with fake hoses.
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { planIngest } from '../src/indexer.js';
+
+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 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(p.unknown, []);
+});
+
+test('empty HOSES string falls back to the default (all on)', () => {
+  assert.deepEqual(sorted(planIngest(ONE, { HOSES: '' }).kinds), [0, 3, 10002]);
+});
+
+test('INDEX_LEGACY_KINDS=0 → only enabled hoses\' kinds, no legacy', () => {
+  const p = planIngest(ONE, { HOSES: 'profiles', INDEX_LEGACY_KINDS: '0' });
+  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);
+});
+
+test('unknown hose names are reported and ignored', () => {
+  const p = planIngest(ONE, { HOSES: 'profiles,nope' });
+  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), [3, 10002]); // legacy fallback only
+});
+
+test('a hose that owns a legacy kind removes it from the legacy set (no double-subscribe)', () => {
+  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(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/indexer-plan.test.js

@@ -0,0 +1,68 @@
+// Profiles hose: validation + signature verification.
+// We sign real kind-0 events with a known key, then assert accept/reject.
+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 } from '../src/hoses/profiles.js';
+
+const enc = new TextEncoder();
+const PRIV = hexToBytes('0000000000000000000000000000000000000000000000000000000000000001');
+const PUBKEY = bytesToHex(schnorr.getPublicKey(PRIV));
+
+// Build a properly-signed kind-0 event.
+function signed({ content = '{"name":"alice"}', created_at = 1000, kind = 0 } = {}) {
+  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-0 event', () => {
+  assert.equal(verifyEvent(signed()), true);
+});
+
+test('accepts empty content (treated as {})', () => {
+  assert.equal(verifyEvent(signed({ content: '' })), true);
+});
+
+test('rejects tampered content (sig no longer matches)', () => {
+  const e = signed();
+  e.content = '{"name":"mallory"}';
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects a tampered id', () => {
+  const e = signed();
+  e.id = 'ff' + e.id.slice(2);
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects a forged/garbage signature', () => {
+  const e = signed();
+  e.sig = 'f'.repeat(128);
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects non-hex pubkey', () => {
+  const e = signed();
+  e.pubkey = 'npub1xxx';
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects the wrong kind', () => {
+  assert.equal(verifyEvent(signed({ kind: 3 })), false);
+});
+
+test('rejects non-JSON content', () => {
+  // Re-sign so the sig is valid but content is not JSON — must still reject.
+  const e = signed({ content: 'not json{' });
+  assert.equal(verifyEvent(e), false);
+});
+
+test('rejects malformed input without throwing', () => {
+  for (const bad of [null, undefined, {}, { kind: 0 }, 42, 'x']) {
+    assert.equal(verifyEvent(bad), false);
+  }
+});