// lens-import.js — read an atproto AppView over XRPC. The missing arm of DESIGN.md's lens model.
//
// DESIGN.md:226 calls the *view* a replaceable lens (appview / feed generator / labeler) and :246 defines
// lens EXPORT — favela communities projected out as bsky feeds. This is the symmetric arm: lens IMPORT,
// reading a remote AppView for data our own index cannot have.
//
// WHY IT IS NEEDED. favela's appview indexes the firehoses of peers it ingests, which is a strictly
// narrower window than a public AppView's. A bsky user replying to us, or the profile of anyone outside
// that window, is not "missing" — it was never in scope. No amount of local indexing surfaces it.
//
// WHAT THIS IS NOT. It does not merge remote data into local views. Two AppViews legitimately disagree —
// different visibility, different moderation, one may block a post another serves — and atproto assumes
// exactly ONE AppView per client session (the `atproto-proxy` header names it). Aggregating invents a
// consensus neither endorses. So this returns ATTRIBUTED data for a named lens and leaves presentation to
// the caller. Counts and labels in particular must not be silently blended: bsky's counts are
// network-wide while ours are window-local (different denominators), and labels are a moderation policy,
// which atproto models as per-labeler subscription rather than aggregation.
//
// FAILS SOFT, ALWAYS. Every path returns null rather than throwing. A lens is supplementary: a thread
// must render from local data whether or not a remote AppView answers. An import that can break a page is
// worse than one that is absent.
//
// PURE + INJECTED. `doFetch` and `now` are parameters, so the whole backpressure story is unit-testable
// without a network or a sleeping clock.
const DEFAULTS = {
cacheTtlMs: 60_000,
negativeTtlMs: 15_000,
timeoutMs: 1500,
maxConcurrent: 4,
breakerThreshold: 3,
breakerCooldownMs: 60_000,
maxCacheEntries: 500
}
// Stable cache key. Params are sorted so {a,b} and {b,a} are one entry rather than two.
function keyOf (lensName, nsid, params) {
const q = Object.keys(params || {}).sort().map((k) => `${k}=${String(params[k])}`).join('&')
return `${lensName}|${nsid}|${q}`
}
// Pull the DIRECT replies out of an app.bsky.feed.getPostThread body, as plain attributed rows.
//
// A thread node is a union: #threadViewPost, #notFoundPost or #blockedPost. Only the first carries a
// post, and skipping the other two is not a detail — a blocked post is a moderation decision by THAT
// AppView, and rendering it as though we had the content would misrepresent their call as ours.
//
// Returns a normalized, RENDER-READY row rather than bsky's postView, so the presentation layer never has
// to know AppView shapes: { uri, cid, did, handle, displayName, avatar, text, createdAt, replyCount }.
// Nested descendants are deliberately dropped — one level is enough to say "there is more over there",
// and reproducing a remote AppView's whole tree inside ours starts to look like the merge this module
// exists to refuse.
export function extractThreadReplies (body, { exclude = null } = {}) {
const replies = body && body.thread && Array.isArray(body.thread.replies) ? body.thread.replies : []
const out = []
for (const node of replies) {
if (!node || typeof node !== 'object') continue
const t = node.$type
if (typeof t === 'string' && !t.endsWith('#threadViewPost')) continue // notFound / blocked
const p = node.post
if (!p || typeof p.uri !== 'string') continue
// Anything this node already indexed is OURS to render, in our tree, with our affordances. Showing it
// again under the lens would double it — this is de-duplication for display, NOT a merge of the two
// views (no counts are combined, nothing is reconciled).
if (exclude && exclude.has(p.uri)) continue
const a = p.author || {}
const rec = p.record || {}
out.push({
uri: p.uri,
cid: typeof p.cid === 'string' ? p.cid : null,
did: typeof a.did === 'string' ? a.did : null,
handle: typeof a.handle === 'string' ? a.handle : null,
displayName: typeof a.displayName === 'string' ? a.displayName : null,
avatar: typeof a.avatar === 'string' ? a.avatar : null,
text: typeof rec.text === 'string' ? rec.text : '',
createdAt: typeof rec.createdAt === 'string' ? rec.createdAt : null,
replyCount: Number.isFinite(p.replyCount) ? p.replyCount : 0
})
}
return out
}
// Pull the ANCESTOR chain out of an app.bsky.feed.getPostThread body, oldest first.
//
// This is the other half of the window problem. getPostThread nests ancestors as `thread.parent`, and
// each parent may carry its own `.parent` — so the chain is a linked list walked upward, unlike replies
// which are an array. Our own walk stops at the first parent we never indexed and marks it { missing },
// which is exactly the gap this fills.
//
// Stops at a #notFoundPost or #blockedPost, and at anything malformed, rather than skipping over it: an
// ancestor chain with a hole in the middle is not a chain, and silently closing the gap would imply the
// remaining posts are consecutive when they are not. `maxDepth` bounds a hostile or looping response.
export function extractThreadParents (body, { exclude = null, maxDepth = 20 } = {}) {
const out = []
const seen = new Set()
let node = body && body.thread ? body.thread.parent : null
for (let i = 0; i < maxDepth && node && typeof node === 'object'; i++) {
const t = node.$type
if (typeof t === 'string' && !t.endsWith('#threadViewPost')) break
const p = node.post
if (!p || typeof p.uri !== 'string') break
if (seen.has(p.uri)) break // a cycle in someone else's data must not hang our page
seen.add(p.uri)
if (!exclude || !exclude.has(p.uri)) {
const a = p.author || {}
const rec = p.record || {}
out.push({
uri: p.uri,
cid: typeof p.cid === 'string' ? p.cid : null,
did: typeof a.did === 'string' ? a.did : null,
handle: typeof a.handle === 'string' ? a.handle : null,
displayName: typeof a.displayName === 'string' ? a.displayName : null,
avatar: typeof a.avatar === 'string' ? a.avatar : null,
text: typeof rec.text === 'string' ? rec.text : '',
createdAt: typeof rec.createdAt === 'string' ? rec.createdAt : null,
replyCount: Number.isFinite(p.replyCount) ? p.replyCount : 0
})
}
node = node.parent
}
// Walked upward, so reverse to render oldest-first — the order a reader scans a conversation.
return out.reverse()
}
export function createLensReader (opts = {}) {
const cfg = { ...DEFAULTS, ...opts }
const doFetch = opts.doFetch
const now = opts.now || (() => Date.now())
if (typeof doFetch !== 'function') throw new Error('createLensReader requires doFetch(url, init)')
const cache = new Map() // key -> { at, ok, value }
const inflight = new Map() // key -> Promise (single-flight)
const breakers = new Map() // lensName -> { fails, openedAt }
const stats = { hits: 0, misses: 0, fetches: 0, failures: 0, shortCircuited: 0, coalesced: 0, timeouts: 0 }
let active = 0
const waiters = []
// A plain semaphore. Without it a burst of thread views opens one socket each; the cap makes the
// outbound load a property of THIS process rather than of how many people happen to click at once.
async function withSlot (fn) {
if (active >= cfg.maxConcurrent) await new Promise((resolve) => waiters.push(resolve))
active++
try { return await fn() } finally {
active--
const next = waiters.shift()
if (next) next()
}
}
// Per-LENS, not per-key: the failure mode being defended against is "this AppView is down", and
// per-key breakers would let a down lens be retried once per distinct thread forever.
function breakerOpen (lensName) {
const b = breakers.get(lensName)
if (!b || b.fails < cfg.breakerThreshold) return false
if (now() - b.openedAt >= cfg.breakerCooldownMs) { breakers.delete(lensName); return false }
return true
}
function recordOutcome (lensName, ok) {
if (ok) { breakers.delete(lensName); return }
const b = breakers.get(lensName) || { fails: 0, openedAt: 0 }
b.fails++
if (b.fails >= cfg.breakerThreshold) b.openedAt = now()
breakers.set(lensName, b)
}
function cacheGet (key) {
const e = cache.get(key)
if (!e) return undefined
// Negatives expire FASTER than positives: a transiently unreachable lens should recover quickly,
// but not be retried on every single page view in the meantime.
const ttl = e.ok ? cfg.cacheTtlMs : cfg.negativeTtlMs
if (now() - e.at >= ttl) { cache.delete(key); return undefined }
return e
}
function cacheSet (key, ok, value) {
if (cache.size >= cfg.maxCacheEntries) cache.clear()
cache.set(key, { at: now(), ok, value })
}
// GET <lens.base>/xrpc/<nsid>?<params>. When `lens.proxy` is set it rides as the `atproto-proxy`
// header — which is how the bridge (pds/lib/appview.js resolveAppView) is told WHICH AppView to
// forward to. That makes one configured egress reach bsky, Blacksky, or anything else, and keeps
// favela keyless: it never needs credentials for the AppView it is reading.
function urlFor (lens, nsid, params) {
const qs = Object.keys(params || {}).sort().map((k) => `${encodeURIComponent(k)}=${encodeURIComponent(params[k])}`).join('&')
return `${String(lens.base).replace(/\/$/, '')}/xrpc/${nsid}${qs ? '?' + qs : ''}`
}
async function fetchOnce (lens, nsid, params) {
const url = urlFor(lens, nsid, params)
const headers = { accept: 'application/json' }
if (lens.proxy) headers['atproto-proxy'] = lens.proxy
let timer = null
try {
const timeout = new Promise((_, reject) => { timer = setTimeout(() => reject(new Error('lens timeout')), cfg.timeoutMs) })
const res = await Promise.race([doFetch(url, { method: 'GET', headers }), timeout])
if (!res || typeof res.status !== 'number' || res.status < 200 || res.status >= 300) return null
// A body that is not JSON is a failure, not an empty result — an AppView that answers 200 with an
// HTML error page must not be cached as a valid (empty) view.
if (typeof res.json === 'function') return await res.json()
if (typeof res.body === 'string') return JSON.parse(res.body)
return res.body || null
} catch (e) {
if (String(e && e.message) === 'lens timeout') stats.timeouts++
return null
} finally { if (timer) clearTimeout(timer) }
}
// Read one XRPC method from one lens. Returns the decoded body, or null (never throws).
async function get (lens, nsid, params = {}) {
if (!lens || !lens.base || !lens.name || !nsid) return null
const key = keyOf(lens.name, nsid, params)
const cached = cacheGet(key)
if (cached !== undefined) { stats.hits++; return cached.ok ? cached.value : null }
stats.misses++
if (breakerOpen(lens.name)) { stats.shortCircuited++; return null }
// Single-flight: N concurrent viewers of the same thread share ONE request. Without this the cache
// only helps AFTER the first response, so a cold popular thread still fans out.
const pending = inflight.get(key)
if (pending) { stats.coalesced++; return pending }
const p = (async () => {
stats.fetches++
const value = await withSlot(() => fetchOnce(lens, nsid, params))
const ok = value !== null && value !== undefined
if (!ok) stats.failures++
recordOutcome(lens.name, ok)
cacheSet(key, ok, value)
return ok ? value : null
})().finally(() => inflight.delete(key))
inflight.set(key, p)
return p
}
return {
get,
stats: () => ({ ...stats, cacheSize: cache.size, inflight: inflight.size, open: [...breakers].filter(([, b]) => b.fails >= cfg.breakerThreshold).map(([n]) => n) }),
// Test/ops affordance: forget everything for a lens (e.g. after reconfiguring it).
reset (lensName = null) {
if (!lensName) { cache.clear(); breakers.clear(); return }
breakers.delete(lensName)
for (const k of [...cache.keys()]) if (k.startsWith(lensName + '|')) cache.delete(k)
}
}
}