// The favela gateway — the public-facing bridges as ROLES of one process (hyper-social #80).
//
// membrane atproto-bridge's keyless bsky membrane (pds.7rnx.net, *.7rnx.net → PDS_PORT)
// fediverse ap-bridge, the ActivityPub bridge (ap.7rnx.net → AP_PORT, admin → AP_ADMIN_PORT)
// (G2: lens — the bsky custom feed; G3: seeder — holepunch blind-peer)
//
// ⛔ EVERY ROLE KEEPS ITS OWN STORES AND SWARM IDENTITY. The membrane's `swarm-keypair.json` is what
// hyper-proto-peers allowlist; the fediverse role's `held/swarm-key` public key IS the published
// bridgeKey registrants find it by. The gateway is pointed at the existing data dirs and opens
// nothing of its own, so moving a bridge into it changes no identity anywhere.
//
// ⛔ A ROLE THAT FAILS TO START IS REPORTED, NOT FATAL. The membrane serving bsky is worth more than a
// gateway that refuses to boot because the fediverse role's key file moved. Only when NO requested
// role started does start throw: a gateway serving nothing should not look healthy to its supervisor.
//
// Nothing here owns the process: no control socket, no exit, no signal handlers. bin/favela-gateway.js
// owns those, and serves `handlers` on the gateway's one control socket.
import { startMembrane } from 'atproto-bridge/pds/lib/membrane.js'
import { startApBridge } from 'ap-bridge/lib/bridge.js'
import os from 'bare-os'
import { createProbes } from './probes.js'
import { createLens } from './lens.js'
import { createLabels } from './labels.js'
import { createMemberLabels } from './member-labels.js'
import { Client } from 'hyper-svc'
import { createWatcher } from './watch.js'
import { renderStatusPage } from './panel.js'
import { renderPublicPage } from './public.js'
const ROLES = {
membrane: async (cfg, opts) => startMembrane(cfg, opts),
// The fediverse role's admin surface (:3401, authenticated) also serves the gateway's status page.
fediverse: async (cfg, opts) => startApBridge(cfg, { ...opts, adminPages: opts.adminPages }),
// The lens serves THROUGH the membrane (its XRPC extensions, its store, its swarm, its fronts), so it
// needs the membrane role up and started first.
lens: async (cfg, opts, running) => {
if (!running.membrane) throw new Error('the lens needs the membrane role (start it before the lens)')
// Labels need the state dir (the gateway's own label log, whose seq bsky's cursor counts). The member
// projector needs the labeler DID's hpp socket on this host (LENS_LABELER_SOCK) — hpp signs, we state.
const labels = opts.stateDir ? createLabels({ peers: running.membrane.peers, stateDir: opts.stateDir, log: opts.log }) : null
const memberLabels = cfg && cfg.labelerSock ? createMemberLabels({ connect: () => Client.connectUds(cfg.labelerSock), log: opts.log }) : null
const lens = createLens({ membrane: running.membrane, feeds: (cfg && cfg.feeds) || [], log: opts.log, labels, memberLabels })
await lens.start()
// ⛔ bsky → favela notifications (hyper-social #82, I2b): interactions on BOUND DIDs' posts go into the identity's
// observation feed — the one the fediverse role already writes for it. Needs that role; never blocks the lens.
let ingest = null
let rebindTimer = null
let onAppend = null
const relayCore = running.membrane.relay && running.membrane.relay.seq && running.membrane.relay.seq.core
if (running.fediverse && typeof running.fediverse.observationsForFeed === 'function' && !(cfg && cfg.bskyInbound === false)) {
const { BskyIngest } = await import('./bsky-ingest.js')
const { createBareSocket } = await import('./spacedust-socket.js')
ingest = new BskyIngest({ membrane: running.membrane, fediverse: running.fediverse, fetch: opts.fetch || globalThis.fetch || null, createSocket: createBareSocket, log: opts.log })
ingest.start().catch((err) => opts.log({ msg: 'bsky ingest did not start', error: (err && err.message) || String(err) }))
// Re-bind when any fronted repo commits: a new net.favela.feed/self, or a new DID. Debounced, no poll.
if (relayCore) {
onAppend = () => { clearTimeout(rebindTimer); rebindTimer = setTimeout(() => { ingest.rebind().catch(() => {}) }, 5000) }
relayCore.on('append', onAppend)
}
}
return {
...lens,
metrics: () => ({ ...lens.metrics(), bskyIngest: ingest ? ingest.metrics() : null }),
stop: async () => {
clearTimeout(rebindTimer)
if (relayCore && onAppend && relayCore.off) { try { relayCore.off('append', onAppend) } catch {} }
if (ingest) { try { ingest.stop() } catch {} }
await lens.stop()
}
}
}
}
const defaultLog = (o) => console.log(JSON.stringify(o))
/**
* @param {object} opts
* @param {string[]} opts.roles in start order; stopped in reverse
* @param {object} [opts.membrane] atproto-bridge loadConfig() shape
* @param {object} [opts.fediverse] ap-bridge assertUsable(loadConfig()) shape
* @param {Function} [opts.log]
* @param {Array} [opts.bootstrap] DHT bootstrap for every role's swarm (tests pass [])
* @param {string} [opts.version]
*/
export async function startGateway ({ roles, log = defaultLog, bootstrap = null, version = null, fetch = null, stateDir = null, watch = false, ...configs }) {
for (const name of roles) if (!ROLES[name]) throw new Error('unknown role: ' + name)
const startedAt = Date.now()
// Admin pages for the fediverse role's authenticated surface; filled once every role has started.
const adminPages = new Map()
const running = {}
const down = {}
for (const name of roles) {
const roleLog = (o) => log({ role: name, ...o })
try {
// A config the entry could not load arrives as an Error: that role is down, for that reason.
if (configs[name] instanceof Error) throw configs[name]
running[name] = await ROLES[name](configs[name], { log: roleLog, bootstrap, version, adminPages, stateDir, fetch }, running)
log({ msg: 'role up', role: name })
} catch (err) {
down[name] = (err && err.message) || String(err)
log({ msg: 'role DOWN — the other roles keep serving', role: name, error: down[name] })
}
}
// Stop in reverse start order, each step named by its role so a slow shutdown says whose it was.
const steps = []
for (const name of [...roles].reverse()) {
if (!running[name]) continue
for (const [step, fn] of running[name].steps) steps.push([name + ':' + step, fn])
}
if (!Object.keys(running).length) {
// Release whatever a half-started role left behind before refusing.
for (const [, fn] of steps) { try { await fn() } catch {} }
throw new Error('no role started: ' + JSON.stringify(down))
}
const status = async () => {
const out = {}
for (const name of roles) {
if (running[name]) out[name] = await running[name].status()
else out[name] = { down: true, error: down[name] }
}
return { ok: true, version, roles: out }
}
// bsky's own view of the membrane's host. Only when the membrane runs: it is the role bsky reaches.
const membraneCfg = configs.membrane && !(configs.membrane instanceof Error) ? configs.membrane : null
const hostname = membraneCfg
? (membraneCfg.crawlHostname || (/^did:web:/.test(membraneCfg.serverDid || '') ? membraneCfg.serverDid.slice('did:web:'.length) : null))
: null
const probes = running.membrane && hostname ? createProbes({ hostname, fetch }) : null
/**
* One answer for the panel and the CLI. Each role reports from its own local state; bsky's view comes
* from the probes (cached; `refresh: true` forces a new probe). A role that is down is named with its
* error and has no section — never a stale one.
*/
const metrics = async (params = {}) => {
const mem = os.memoryUsage()
const out = {
at: Date.now(),
version,
process: { uptimeS: Math.round((Date.now() - startedAt) / 1000), rss: mem.rss, heapUsed: mem.heapUsed, heapTotal: mem.heapTotal },
roles: {},
membrane: null,
fediverse: null,
lens: null,
bsky: null
}
for (const name of roles) out.roles[name] = running[name] ? { up: true } : { up: false, error: down[name] }
if (running.membrane) out.membrane = running.membrane.metrics()
if (running.fediverse) out.fediverse = await running.fediverse.metrics()
if (running.lens) out.lens = running.lens.metrics()
if (probes) {
const dids = out.membrane.dids.map((d) => ({ did: d.did, rev: d.rev }))
const feeds = out.lens ? out.lens.feeds.filter((f) => f.feedUri).map((f) => f.feedUri) : []
out.bsky = await probes.get({ dids, emitted: out.membrane.relay.emitted, feeds }, { force: !!params.refresh })
}
if (watcher) out.alerts = watcher.current()
return out
}
// The watcher: checks on its own and keeps alert state on disk (lib/watch.js). Its metrics calls skip
// `alerts` (it is computing them). Off unless asked, so tests and one-off runs do not schedule checks.
let watcher = null
if (watch) {
watcher = createWatcher({
metrics: async (p) => { const m = await metrics(p); delete m.alerts; return m },
stateFile: stateDir ? stateDir.replace(/\/$/, '') + '/alerts.json' : null,
log: (o) => log({ role: 'alerts', ...o })
})
watcher.start()
steps.unshift(['alerts', () => watcher.stop()])
}
// ONE verb map. The roles' verbs do not overlap except `status`, which the gateway answers for all
// of them; a collision would be a bug, so it is refused here rather than silently shadowed.
const handlers = { status, metrics }
// Open alerts, recently resolved, last check. `refresh: true` checks now (fresh probes) and answers after.
handlers.alerts = async (params = {}) => {
if (!watcher) return { ok: false, error: 'the alert watcher is not running' }
if (params.refresh) await watcher.checkNow()
return watcher.current()
}
for (const name of roles) {
if (!running[name]) continue
for (const [verb, fn] of Object.entries(running[name].handlers)) {
if (verb === 'status' || verb === 'metrics' || verb === 'alerts') continue
if (handlers[verb]) throw new Error(`verb ${verb} is claimed by two roles`)
handlers[verb] = fn
}
}
// The status page on the admin surface: both bridges' health, alerts first (lib/panel.js).
adminPages.set('/admin/status', async () => renderStatusPage(await metrics({})))
// The PUBLIC page (lib/public.js) on the membrane's own host only: https://<host>/ and /status. Any
// other host (a *.7rnx.net handle host) is declined and falls through to what it served before.
if (running.membrane && running.membrane.extensions && running.membrane.extensions.http && hostname) {
const page = async ({ host }) => {
if (host !== hostname) return null
return { status: 200, headers: { 'content-type': 'text/html; charset=utf-8', 'cache-control': 'no-store' }, body: renderPublicPage(await metrics({})) }
}
running.membrane.extensions.http.set('/', page)
running.membrane.extensions.http.set('/status', page)
steps.unshift(['public-page', () => { running.membrane.extensions.http.delete('/'); running.membrane.extensions.http.delete('/status') }])
}
// The verbs this gateway serves, for `favela-gateway help`. release-ready is the process owner's.
handlers.verbs = async () => ({ verbs: [...Object.keys(handlers), 'release-ready'].sort() })
return { roles: running, down, handlers, status, metrics, adminPages, steps }
}
export default startGateway