beacon: HOSES switch + gate legacy 3/10002 fallback

Make the firehose a switch so a deploy can run a single hose without
double-writing against the legacy firehose/followshose processes during
the phased migration.

- planIngest(allHoses, env): pure, testable selector. HOSES env (comma
  list of hose names; default: all registered) picks active hoses;
  INDEX_LEGACY_KINDS (default on) gates the raw-upsert fallback for kinds
  not yet migrated to a hose (3, 10002). Subscription kinds = union of
  enabled hoses' kinds + legacy kinds, with a hose-owned kind removed from
  the legacy set so nothing is double-subscribed.
- runIndexer: use the plan; warn on unknown hose names / empty kind set.
- .env.example: document HOSES + INDEX_LEGACY_KINDS.
- test/indexer-plan.test.js: 8 cases (defaults, legacy off, unknown names,
  single-hose deploy, hose-owned legacy kind removal).

Default behaviour unchanged: profiles hose + legacy 3/10002.

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

src/indexer.js

@@ -2,29 +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);
 
-// Registered hoses (phase 1: profiles only). Each owns its kinds + ingest.
-// Kinds with no hose fall back to the legacy raw upsert until their own phase.
-const HOSES = [profilesHose];
-const hoseFor = (kind) => HOSES.find((h) => h.kinds.includes(kind));
-const KINDS = [...new Set([...HOSES.flatMap((h) => h.kinds), 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());
@@ -40,18 +63,24 @@ function connectRelay(url, onEvent) {
 
 export async function runIndexer() {
   const db = await connect();
-  for (const h of HOSES) await h.ensureIndexes(db);
-  console.log(`[beacon] indexing kinds ${KINDS.join(',')} from ${RELAYS.length} relays`);
+  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 {
       const hose = hoseFor(event.kind);
-      const stored = hose ? await hose.ingest(event, db) : await upsertEvent(event);
+      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] 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); });

test/indexer-plan.test.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]);
+});