// FeedRegistry — the watched, co-ownable list of feeds (favela-web/spec/community-ux.md). One class, three
// roles (same shape as CommunityRoom):
// - OWNER (create): genesis writer + sole indexer; publishes/removes feed entries.
// - MOD (open + admitted): a co-writer, admitted by the owner (addMod) — so a registry can be co-owned.
// - READER (openReadonly): KEYLESS — opens by the PUBLIC bootstrapKey, holds no key, never a writer;
// replicates + lists + WATCHES. This is the reactor path (favela-web / membrane / hpp / blind-relay).
//
// Entries linearize into a plaintext Hyperbee view (feed-schema.js). It is the source of truth: the mint
// writes an entry; the reactors watch it and serve/front/index/seed the feed with no per-feed config.
//
// hyper-rooms `Room` is imported by SUBPATH (skips the room-pairing/blind-pairing dep the read/write paths
// here don't use — mod admission is key-exchange, below), matching CommunityRoom.
import { Room } from 'hyper-rooms/lib/room.js'
import { feedRegistrySchema, isDid, isRkey, FEED_PREFIX } from './feed-schema.js'
const normKey = (k) => (typeof k === 'string' ? k.toLowerCase() : k)
export class FeedRegistry {
constructor (room) { this._room = room }
// OWNER — create a fresh registry. This peer is the genesis writer AND the sole indexer (N=1, the safe
// default per the autobase quorum rule — never auto-grow to N=2). Grow indexers only via addIndexer.
static async create (corestore, { namespace = 'feed-registry', name = null } = {}) {
const room = new Room(corestore, { schema: feedRegistrySchema, bootstrapKey: null, namespace })
await room.ready()
const fr = new FeedRegistry(room)
if (name) await fr._room.append({ type: 'set-meta', key: 'name', value: name }).catch(() => {})
return fr
}
// MOD — open an EXISTING registry to co-write. Writable only AFTER the owner addMods this peer's localKey
// (waitWritable blocks until then). Until admitted it can still READ.
static async open (corestore, roomKey, { namespace } = {}) {
const key = normKey(roomKey)
if (!/^[0-9a-f]{64}$/.test(String(key))) throw new Error('roomKey must be a 32-byte hex bootstrap key')
const room = new Room(corestore, { schema: feedRegistrySchema, bootstrapKey: key, namespace: namespace || `feed-registry:${key}` })
await room.ready()
return new FeedRegistry(room)
}
// READER — keyless read-only open (semantically identical to open(); named for intent). The reactors use
// this: they never call a write method, so are never admitted and hold no key.
static openReadonly (corestore, roomKey, opts) { return FeedRegistry.open(corestore, roomKey, opts) }
// ── identity / replication surface ──
get key () { return this._room.keyHex } // PUBLIC bootstrap key — share to reactors/co-mods
get localKey () { return this._room.localKeyHex } // THIS peer's writer key — give to the owner for addMod
get discoveryKey () { return this._room.discoveryKey }
get writable () { return this._room.writable }
get corestore () { return this._room.corestore }
get base () { return this._room.base }
get view () { return this._room.view }
async update () { await this._room.base.update() }
async close () { await this._room.close() }
// Block until this peer has been admitted as a writer (autopass whenWritable pattern). Returns false on
// timeout. A mod calls this after handing its localKey to the owner, before writing.
waitWritable ({ timeout = 30000 } = {}) {
const base = this._room.base
if (base.writable) return Promise.resolve(true)
return new Promise((resolve) => {
let done = false
const finish = (v) => { if (done) return; done = true; base.off('update', check); clearTimeout(timer); resolve(v) }
const check = () => { if (base.writable) finish(true) }
const timer = setTimeout(() => finish(false), timeout)
base.on('update', check)
})
}
// ── reads (keyless-safe) ──
// Every registered feed, as the stored entry objects { did, rkey, name, curator, params, roomKey, handle,
// createdBy, at }. This is what a reactor iterates: index / front / serve / seed each.
async feeds () {
await this._room.base.update()
const out = []
for await (const { value } of this._room.view.createReadStream({ gte: FEED_PREFIX, lt: FEED_PREFIX + '\xff' })) out.push(value)
return out
}
async feed (did, rkey) {
await this._room.base.update()
const node = await this._room.view.get(`${FEED_PREFIX}${did}/${rkey}`)
return (node && node.value) || null
}
// ── writes (owner/mods; throw if not admitted) ──
// Publish (or update — last-writer-wins on the key) a feed. did = the repo that HOLDS the feed record
// (your own hpp for a light feed, or a co-owned community DID). curator/params = the contract favela-web
// runs; roomKey = the roster CommunityRoom (only for community/roster feeds). Idempotent per (did, rkey).
async addFeed ({ did, rkey, name = null, curator, params = {}, roomKey = null, handle = null, by = null } = {}) {
this._assertWritable('addFeed')
if (!isDid(did)) throw new Error('addFeed requires a valid did (the feed record home repo)')
if (!isRkey(rkey)) throw new Error('addFeed requires a valid rkey')
if (typeof curator !== 'string' || !curator) throw new Error('addFeed requires a curator name')
if (params != null && (typeof params !== 'object' || Array.isArray(params))) throw new Error('addFeed params must be an object')
await this._room.append({ type: 'add-feed', did, rkey, name, curator, params: params || {}, roomKey, handle, by, at: Date.now() })
}
async removeFeed (did, rkey) {
this._assertWritable('removeFeed')
if (!isDid(did) || !isRkey(rkey)) throw new Error('removeFeed requires a valid did + rkey')
await this._room.append({ type: 'remove-feed', did, rkey })
}
// ── admission (owner adds a co-writer) — same model + warnings as CommunityRoom.addMod ──
// Default role 'writer' = a plain writer (non-indexer, the SAFE topology: owner stays sole indexer N=1, so
// the mod's ops still linearize via ackInterval without the N=2 quorum trap). role 'owner' promotes to
// INDEXER — only to reach an ODD indexer count ≥3 on independent always-on boxes.
async addMod (writerKeyHex, { role = 'writer', userId = null, addedBy = null } = {}) {
this._assertWritable('addMod')
if (!/^[0-9a-f]{64}$/i.test(String(writerKeyHex))) throw new Error('addMod requires the mod\'s 64-hex localKey')
if (!['owner', 'writer', 'mirror'].includes(role)) throw new Error('role must be owner, writer, or mirror')
await this._room.addWriter(userId || String(writerKeyHex).toLowerCase(), { writerKey: String(writerKeyHex).toLowerCase(), role, addedBy })
}
async removeMod (userId) { this._assertWritable('removeMod'); await this._room.removeWriter(userId) }
_assertWritable (op) { if (!this._room.writable) throw new Error(`${op}: not a writer — this peer holds no key / has not been admitted (reactors are keyless by design)`) }
}
// ── read-manager (reactors) ─────────────────────────────────────
// Opens + maintains a KEYLESS read replica of a registry on a shared corestore + swarm, answers feeds(), and
// WATCHES for changes. This is the reactor primitive: favela-web / membrane / hpp / blind-relay each wrap a
// FeedRegistries and react to onChange(feeds[]) — index / front / serve / seed the new feed. No per-feed config.
export class FeedRegistries {
constructor (corestore, swarm) {
this._store = corestore
this._swarm = swarm
this._registries = new Map() // roomKeyHex -> Promise<FeedRegistry>
// Replicate our corestore on every swarm connection (joining a discoveryKey only DISCOVERS; without
// this the connection forms but no cores flow and the list never syncs). Mirrors CommunityRooms.
if (swarm) swarm.on('connection', (conn) => { conn.on('error', () => {}); this._store.replicate(conn) })
}
registry (roomKeyHex) {
const key = normKey(roomKeyHex)
if (this._registries.has(key)) return this._registries.get(key)
const p = (async () => {
const fr = await FeedRegistry.openReadonly(this._store, key)
if (this._swarm) this._swarm.join(fr.discoveryKey, { client: true, server: true })
return fr
})().catch((e) => { this._registries.delete(key); throw e })
this._registries.set(key, p)
return p
}
async feeds (roomKeyHex) { return (await this.registry(roomKeyHex)).feeds() }
// Watch a registry for changes. onChange(feeds[]) fires (debounced) on every linearized-view advance — i.e.
// whenever a feed add/remove replicates in — PLUS once immediately as a baseline. This is what a reactor
// uses to pick up feeds published AFTER it started, with NO restart. Serialized + coalesced; onChange
// errors are swallowed (a flaky reactor must never break replication). Returns an async unwatch fn.
async watch (roomKeyHex, onChange, { debounceMs = 300 } = {}) {
const fr = await this.registry(roomKeyHex)
const base = fr.base
let timer = null
let running = false
let pending = false
const run = async () => {
if (running) { pending = true; return }
running = true
try { await onChange(await fr.feeds()) } catch {}
running = false
if (pending) { pending = false; schedule() }
}
const schedule = () => { clearTimeout(timer); timer = setTimeout(run, debounceMs) }
base.on('update', schedule)
schedule() // baseline: emit the current feed set now
return async () => { clearTimeout(timer); base.off('update', schedule) }
}
async close () {
for (const p of this._registries.values()) { try { await (await p).close() } catch {} }
this._registries.clear()
}
}
export default { FeedRegistry, FeedRegistries }