// The lens role — a favela community's bsky custom feed, served keylessly by the gateway (hyper-social #80 G2).
//
// Replaces favela-web's lens (stopped since 2026-10-11, and the reason the loser-club feed was down). Same
// public contract, same selection rules (favela-web lib/feedgen.js), different sources:
//
// WHO the community room's members minus bans — favela-community, READ-ONLY, opened by the membrane on
// its own store and swarm (membrane.communityRooms()). Watched: a roster change re-applies at once.
// WHAT each member's app.bsky.feed.post records, read from the repo the membrane already FRONTS. Indexed
// per member and re-read only when that member's served head moves — checked on each append to the
// relay's aggregate firehose. Never on bsky's request: a front's listRecords is a full scan.
// HOW newest record createdAt first; ONLY TID rkeys (one bad rkey — loser-club's own 'self.hello' —
// makes bsky reject the whole response with InvalidFeedResponse); `cursor` a string or ABSENT.
//
// Served through the membrane's XRPC extensions, so did.json advertises #bsky_fg and bsky's AppView reaches
// it at pds.7rnx.net through the community's PLC #bsky_fg — nothing outside this process changes.
//
// Labels (member labels, #80): with `labels` (lib/labels.js) the lens serves queryLabels/subscribeLabels from
// the verified label log and hands each community DID's front to it; with `memberLabels`
// (lib/member-labels.js) every roster change is projected into `<feed>-member` labels. Without them the
// label surfaces answer a valid empty set, so the community's advertised labeler is valid but silent.
const GENERATOR = 'app.bsky.feed.generator'
const POST = 'app.bsky.feed.post'
const FLOOR = '0000-01-01T00:00:00.000Z' // favela-web's createdAt for a record that has none
const TID_RE = /^[234567abcdefghij][234567abcdefghijklmnopqrstuvwxyz]{12}$/
// Opaque cursor: hex of the URI-encoded sort key — pure ASCII, so no TextEncoder (Bare has none without a
// polyfill). favela-web returned the raw key, NUL bytes and all.
const toCursor = (key) => [...encodeURIComponent(key)].map((c) => c.charCodeAt(0).toString(16).padStart(2, '0')).join('')
const fromCursor = (c) => {
if (!/^([0-9a-f]{2})+$/.test(String(c || ''))) return null
let s = ''
for (let i = 0; i < c.length; i += 2) s += String.fromCharCode(parseInt(c.slice(i, i + 2), 16))
try { return decodeURIComponent(s) } catch { return null }
}
const parseFeedUri = (uri) => {
const m = /^at:\/\/([^/]+)\/([^/]+)\/([^/]+)$/.exec(String(uri || ''))
return m ? { did: m[1], collection: m[2], rkey: m[3] } : null
}
// favela-web's slugRkey: the feed id → the generator record's rkey.
export function slugRkey (id) {
const s = String(id || '').toLowerCase().replace(/[^a-z0-9._~-]+/g, '-').replace(/^[-.]+|[-.]+$/g, '').slice(0, 64)
return s || 'feed'
}
/**
* @param {object} opts
* @param {object} opts.membrane the membrane role (startMembrane's return): extensions, peers, relay,
* metrics(), communityRooms()
* @param {Array<{id, roomKey}>} opts.feeds e.g. [{ id: 'loser-club', roomKey: '91ce4dbc…' }]
*/
export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = 1000, frontTimeoutMs = 20000, labels = null, memberLabels = null } = {}) {
const index = new Map() // did -> { rev, rows: [{ key, did, rkey, createdAt }], at }
const state = feeds.map((f) => ({ ...f, rkey: slugRkey(f.id), communityDid: null, curation: 'roster', members: [], unwatch: null, error: null }))
let rooms = null
let appendTimer = null
let queue = Promise.resolve()
let stopped = false
const errors = new Map() // did -> last indexing error
const fronted = (did) => !!(membrane.peers && membrane.peers.isFronted && membrane.peers.isFronted(did))
const servedRevs = () => {
const out = new Map()
for (const d of (membrane.metrics().dids || [])) out.set(d.did, d.rev)
return out
}
// Read one member's posts from its fronted repo. Serialized: one rebuild at a time keeps the load on the
// membrane's fronts bounded however many appends arrive.
function rebuild (did, rev) {
queue = queue.then(async () => {
if (stopped) return
try {
const front = await membrane.peers.front(did, { timeout: frontTimeoutMs })
const rows = []
let cursor = null
do {
const page = await front.listRecords(POST, { limit: 100, cursor })
for (const r of page.records || []) {
const rkey = String(r.uri || '').split('/').pop()
const v = r.value || {}
if (v.$type && v.$type !== POST) continue
if (!TID_RE.test(rkey)) continue
const createdAt = typeof v.createdAt === 'string' ? v.createdAt : FLOOR
rows.push({ key: `${createdAt}\u0000${did}\u0000${rkey}`, did, rkey, createdAt })
}
cursor = page.cursor || null
} while (cursor)
index.set(did, { rev, rows, at: Date.now() })
errors.delete(did)
log({ msg: 'lens: indexed member', did, rev, posts: rows.length })
} catch (err) {
errors.set(did, (err && err.message) || String(err))
log({ msg: 'lens: could not index member', did, error: errors.get(did) })
}
})
return queue
}
// Re-read every member (of every feed) whose served head differs from the one indexed.
function reconcile () {
const revs = servedRevs()
const wanted = new Set()
for (const f of state) for (const did of f.members) wanted.add(did)
for (const did of wanted) {
if (!fronted(did)) continue
const rev = revs.get(did) || null
const have = index.get(did)
if (!have || have.rev !== rev) rebuild(did, rev)
}
return queue
}
const onAppend = () => {
clearTimeout(appendTimer)
appendTimer = setTimeout(() => { reconcile() }, debounceMs)
}
function rowsFor (f) {
const all = []
for (const did of f.members) {
const e = index.get(did)
if (e) all.push(...e.rows)
}
return all.sort((a, b) => (a.key < b.key ? 1 : a.key > b.key ? -1 : 0)) // newest first, as favela-web
}
const findFeed = (did, rkey) => state.find((f) => f.communityDid === did && f.rkey === rkey) || null
const feedUri = (f) => `at://${f.communityDid}/${GENERATOR}/${f.rkey}`
async function getFeedSkeleton ({ query }) {
const p = parseFeedUri(query.get('feed'))
const f = p && p.collection === GENERATOR ? findFeed(p.did, p.rkey) : null
if (!f) return { status: 400, body: { error: 'UnsupportedAlgorithm', message: 'Unsupported algorithm' } }
const limit = Math.max(1, Math.min(Number(query.get('limit')) || 50, 100))
const after = query.get('cursor') ? fromCursor(query.get('cursor')) : null
let rows = rowsFor(f)
if (after != null) rows = rows.filter((r) => r.key < after)
const page = rows.slice(0, limit)
const body = { feed: page.map((r) => ({ post: `at://${r.did}/${POST}/${r.rkey}` })) }
// atproto: a string or ABSENT — `cursor: null` fails bsky's response validation for the whole feed.
if (rows.length > limit) body.cursor = toCursor(page[page.length - 1].key)
return { status: 200, body }
}
async function describeFeedGenerator ({ query }) {
const want = query.get('did')
const served = state.filter((f) => f.communityDid && (!want || f.communityDid === want))
// The lexicon requires `did`: the asked-for community, or the first served one.
const did = want || (served[0] && served[0].communityDid) || null
return { status: 200, body: { did, feeds: served.filter((f) => f.communityDid === did).map((f) => ({ uri: feedUri(f) })) } }
}
// No labels module: a valid empty set, and an empty stream held open until the client goes.
async function queryLabels () { return { status: 200, body: { labels: [] } } }
function subscribeLabels (ws) {
ws.on('error', () => {})
ws.on('close', () => {})
}
const XRPC = {
'app.bsky.feed.getFeedSkeleton': getFeedSkeleton,
'app.bsky.feed.describeFeedGenerator': describeFeedGenerator,
'com.atproto.label.queryLabels': queryLabels,
...(labels ? labels.xrpc : {})
}
const WS = labels ? { ...labels.ws } : { '/xrpc/com.atproto.label.subscribeLabels': subscribeLabels }
// The community DID is the labeler. A front that already held its label core before the labels module
// started is never announced to it by the router (onLabels fires on OPENING one), so hand it over here.
const watchLabeler = (did) => {
if (!labels || !did) return
membrane.peers.front(did, { timeout: frontTimeoutMs }).then((fr) => labels.watch(did, fr)).catch(() => {})
}
// ⛔ A LABELER COMING ONLINE IS AN EVENT, NOT ONLY A ROSTER CHANGE. Measured 2026-10-14: the projector's one
// run was at boot, against an hpp that was not yet a labeler; hpp came back with `labeler: true` and began
// announcing a label core, and with an unchanged roster nothing ever re-ran. The router's onLabels fires
// when a front first opens a DID's label core, so re-project the feeds whose community DID that is. Safe to
// repeat: label-intent is idempotent (an unchanged state writes nothing).
let unLabels = null
const onLabelerOnline = (did) => {
if (stopped || !memberLabels) return
for (const f of state) {
if (f.communityDid !== did) continue
memberLabels.apply({ id: f.id, communityDid: f.communityDid, members: [...f.members] }).catch(() => {})
}
}
async function start () {
if (labels) await labels.start()
if (memberLabels && membrane.peers && typeof membrane.peers.onLabels === 'function') {
unLabels = membrane.peers.onLabels((did) => onLabelerOnline(did))
}
rooms = await membrane.communityRooms()
for (const f of state) {
f.unwatch = await rooms.watch(f.roomKey, async (members) => {
f.members = [...members]
// meta:did and curation replicate with the room; re-read them on every change until they land.
try { f.communityDid = (await rooms.communityDid(f.roomKey)) || f.communityDid } catch {}
try { const c = await rooms.curation(f.roomKey); f.curation = (c && c.curator) || 'roster' } catch {}
if (f.curation !== 'roster') log({ msg: 'lens: curation is not roster — serving roster (only roster is implemented)', feed: f.id, curation: f.curation })
log({ msg: 'lens: roster', feed: f.id, members: f.members.length, communityDid: f.communityDid })
watchLabeler(f.communityDid)
// Not awaited: a slow or absent labeler socket must not hold up the feed. It reports its own outcome.
if (memberLabels) memberLabels.apply({ id: f.id, communityDid: f.communityDid, members: [...f.members] }).catch(() => {})
await reconcile()
})
}
if (membrane.relay && membrane.relay.seq && membrane.relay.seq.core) membrane.relay.seq.core.on('append', onAppend)
for (const [nsid, fn] of Object.entries(XRPC)) membrane.extensions.xrpc.set(nsid, fn)
for (const [path, fn] of Object.entries(WS)) membrane.extensions.ws.set(path, fn)
await queue
}
async function stop () {
stopped = true
clearTimeout(appendTimer)
if (unLabels) { try { unLabels() } catch {} ; unLabels = null }
for (const nsid of Object.keys(XRPC)) membrane.extensions.xrpc.delete(nsid)
for (const path of Object.keys(WS)) membrane.extensions.ws.delete(path)
if (membrane.relay && membrane.relay.seq && membrane.relay.seq.core && membrane.relay.seq.core.off) membrane.relay.seq.core.off('append', onAppend)
for (const f of state) { if (f.unwatch) { try { await f.unwatch() } catch {} } }
if (memberLabels) { try { await memberLabels.close() } catch {} }
if (labels) { try { await labels.stop() } catch {} }
}
function metrics () {
return {
labels: labels ? labels.metrics() : null,
memberLabels: memberLabels ? memberLabels.metrics() : null,
feeds: state.map((f) => {
const rows = rowsFor(f)
return {
id: f.id,
roomKey: f.roomKey,
communityDid: f.communityDid,
feedUri: f.communityDid ? feedUri(f) : null,
curation: f.curation,
members: f.members.length,
unfronted: f.members.filter((d) => !fronted(d)),
posts: rows.length,
newest: rows.length ? rows[0].createdAt : null,
perMember: f.members.map((d) => {
const e = index.get(d)
return { did: d, posts: e ? e.rows.length : 0, indexedAt: e ? e.at : null, error: errors.get(d) || null }
})
}
})
}
}
return {
start,
stop,
metrics,
status: () => ({ ok: true, lens: true, feeds: state.map((f) => ({ id: f.id, communityDid: f.communityDid, members: f.members.length })) }),
handlers: {},
steps: [['lens', () => stop()]]
}
}
export default createLens