// favela appview index — folds atproto posts into a LOCAL Hyperbee for query (timeline, author feed).
//
// FAVELA-LOCAL by design (§7.5): every peer derives its OWN view from the replicated firehoses; this
// index is a rebuildable CACHE, never replicated. Bare-native (hypercore/hyperbee, no shims).
//
// This is a per-peer atproto FEED GENERATOR: the "firehose" is each DID's FirehoseSequencer, the index
// is this Hyperbee, and getTimeline is the (locally-hydrated) getFeedSkeleton. It ingests two ways:
// - indexPeer(peer): a SNAPSHOT reconcile from the current repo (listRecords). Catches adopted/
// historical records that were imported in bulk (not as per-record firehose commits), and REMOVES
// entries whose record is no longer present (delete-correctness for the baseline).
// - applyCommit(did, body): INCREMENTAL per-commit ingest from a #commit frame — create/update ->
// putPost (record read from the frame's own CAR, race-free), delete -> delPost. This is the live tail.
//
// Keys — ISO createdAt sorts lexicographically = chronologically; reverse:true = newest first:
// timeline sub: `${createdAt}\x00${did}\x00${rkey}` -> { did, rkey, record, createdAt }
// author sub: `${did}\x00${createdAt}\x00${rkey}` -> { did, rkey, record, createdAt }
// ref sub: `${did}\x00${rkey}` -> { createdAt } (reverse lookup: lets delPost
// reconstruct the createdAt-bearing timeline/author keys, and lets an UPDATE that moved createdAt
// clear its stale keys — a delete op carries no record, so the createdAt must be remembered.)
import Hyperbee from 'hyperbee'
import { displayHandle } from './handles.js'
import crypto from 'hypercore-crypto'
import b4a from 'b4a'
import { POST_COLLECTIONS } from './scopes.js'
import { isIgnoredCollection } from './hosted-dids.js'
import { parseQuery, matches, score, postText, actorText } from './search.js'
import * as dagCbor from '@ipld/dag-cbor'
// Byte length our copy WOULD serialize to — used only to report a divergence from the canonical bytes,
// never to decide a write. Returns -1 if the record cannot be encoded, so the comparison simply reads as
// "differs" rather than throwing inside the ingest path.
function recEncodedLength (record) {
try { return dagCbor.encode(record).length } catch { return -1 }
}
function deepEq (a, b) {
if (a === b) return true
if (a == null || b == null || typeof a !== typeof b) return false
if (typeof a !== 'object') return false
if (Array.isArray(a) !== Array.isArray(b)) return false
if (Array.isArray(a)) return a.length === b.length && a.every((x, i) => deepEq(x, b[i]))
const ka = Object.keys(a); const kb = Object.keys(b)
return ka.length === kb.length && ka.every((k) => deepEq(a[k], b[k]))
}
// Is `ours` a strictly LOSSY version of `theirs` — every field we have, they have identically, and they
// carry at least one field we dropped? That is the exact shape of the 2026-07-30 outage: hpp's record
// carried `langs: ['en']`, ours was identical without it, and re-encoding ours over the canonical file
// invalidated the committed pointer.
//
// This is deliberately NARROW. A genuine EDIT changes a value, so it is not a subset and still writes —
// which matters, because for a peer whose records no local hpp holds we are the only writer, and an
// edited profile that never updates its bytes is its own bug (caught by appview-ingest/-dedup).
export function isLossySubset (ours, theirs) {
if (!ours || !theirs || typeof ours !== 'object' || typeof theirs !== 'object') return false
const ko = Object.keys(ours); const kt = Object.keys(theirs)
if (kt.length <= ko.length) return false
return ko.every((k) => k in theirs && deepEq(ours[k], theirs[k]))
}
// The atproto profile record. One per repo, rkey always 'self'.
// What this build knows how to index. BUMP THIS when the ingest filter widens, and every DID gets ONE
// firehose replay from seq 0 to pick up what the previous build discarded.
//
// This exists because the fast path is self-fulfilling. bin/favela-web.js decides `coHosted` by asking
// whether the local shared store already holds records for a DID — but favela's OWN putPost writes
// ingested records into that store, so the first ingest makes it non-empty and the DID is treated as
// co-hosted forever after, tailing from the current end and never replaying history. e2jev backfilled
// from 0 exactly once, while applyCommit still dropped profiles, and could never do it again: its
// profile commit sat in firehose history that nothing would re-read, and its profile record was absent
// from the local store, so the store-based baseline had nothing to fold either. A dead end reachable
// only by widening the filter, which is precisely when it matters.
//
// 1: post collections only
// 2: + app.bsky.actor.profile
// 3: + app.bsky.feed.like, app.bsky.graph.follow
// 4: + reply threading (post records' reply.parent indexed)
// 5: + app.bsky.feed.repost
// 6: + app.bsky.graph.block — a block is a PUBLIC record and federates, so a block made on ANOTHER
// device (or by a peer) must be visible here. Until this it was tracked only in node-local
// bookkeeping, so a block made elsewhere simply did not exist for this node.
export const INGEST_VERSION = 6
const PROFILE_COLLECTION = 'app.bsky.actor.profile'
const PROFILE_RKEY = 'self'
const LIKE_COLLECTION = 'app.bsky.feed.like'
const FOLLOW_COLLECTION = 'app.bsky.graph.follow'
const REPOST_COLLECTION = 'app.bsky.feed.repost'
const BLOCK_COLLECTION = 'app.bsky.graph.block'
const PROF_TTL_MS = 30 * 1000
const SEP = '\x00'
// The author-feed filters, verbatim from app.bsky.feed.getAuthorFeed's knownValues (verified against the
// lexicon 2026-07-30). Kept as the bsky set rather than a favela-specific one so the profile tabs mean
// the same thing they do elsewhere in atproto, and so a lens import can pass the value straight through.
export const AUTHOR_FILTERS = new Set([
'posts_with_replies', // default — everything
'posts_no_replies', // top-level only
'posts_with_media', // images or video
'posts_and_author_threads',// top-level, plus replies to the author's OWN posts (self-threads)
'posts_with_video'
])
// Which media an embed carries. Mirrors render.js's embedImages for the two shapes we actually write
// (app.bsky.embed.images, and the .media side of recordWithMedia), plus video for the video filter.
function embedMedia (record) {
const e = record && record.embed
if (!e) return { images: false, video: false }
const m = (e.$type === 'app.bsky.embed.recordWithMedia' && e.media) ? e.media : e
return {
images: !!(m && m.$type === 'app.bsky.embed.images' && Array.isArray(m.images) && m.images.length),
video: !!(m && m.$type === 'app.bsky.embed.video')
}
}
// Every real post in a thread, flat, in render order. Moderation (withMod) works on a `.posts` array and
// mutates each entry in place, so flattening for it leaves the TREE carrying the moderated objects —
// labels/mutes apply to a reply nested four levels down exactly as they do on the timeline. `missing`
// ancestor markers are not posts and are skipped.
export function threadPostList (thread) {
if (!thread || !thread.post) return []
const out = [thread.post]
for (const a of thread.ancestors || []) if (a && !a.missing) out.push(a)
const walk = (nodes) => { for (const n of nodes || []) { if (n && n.post) { out.push(n.post); walk(n.replies) } } }
walk(thread.replies)
return out
}
const FLOOR = '0000-01-01T00:00:00.000Z' // sorts before any real ISO createdAt
// Normalize an author scope (Set | array | null) to a Set of DIDs, or null (= no filter / global view).
// An empty scope stays null rather than "match nothing": a signed-in viewer with zero known communities
// falls back to the global timeline instead of an empty page.
function normAuthors (authors) {
if (!authors) return null
const set = authors instanceof Set ? authors : new Set(authors)
return set.size ? set : null
}
// Extract hashtags/topics from a post: the structured self-`tags` array + richtext #tag facets + the
// #hashtags users type in text. Normalized (lowercase, de-#, deduped). This is the topic-feed index —
// the atproto feed-generator "filter" (DESIGN §5 native local search) applied locally per-peer.
const TAG_RE = /(?:^|\s)#([\p{L}\p{N}_]+)/gu
function normTag (t) { const s = String(t || '').replace(/^#/, '').trim().toLowerCase(); return s || null }
export function extractTags (record) {
if (!record) return []
const out = new Set()
if (Array.isArray(record.tags)) for (const t of record.tags) { const n = normTag(t); if (n) out.add(n) }
if (Array.isArray(record.facets)) {
for (const f of record.facets) {
for (const feat of (f && Array.isArray(f.features) ? f.features : [])) {
if (feat && feat.$type === 'app.bsky.richtext.facet#tag' && feat.tag) { const n = normTag(feat.tag); if (n) out.add(n) }
}
}
}
const text = typeof record.text === 'string' ? record.text : ''
for (const m of text.matchAll(TAG_RE)) { const n = normTag(m[1]); if (n) out.add(n) }
return [...out]
}
// Deterministic content signature of a record. indexPeer re-folds the FULL repo on every baseline/#sync,
// and Hyperbee sub-bees do NOT honor alwaysDuplicate:false (a re-put ALWAYS appends — verified), so putPost
// must skip the write entirely when the record is unchanged (sig match). That gate is what stops the index
// ballooning: without it, every reconnect re-appends every record's timeline/author/tags/ref entries.
function recSig (record) { return b4a.toString(crypto.data(b4a.from(JSON.stringify(record || null))), 'hex') }
function sameTags (a, b) {
const x = Array.isArray(a) ? a : []
const y = Array.isArray(b) ? b : []
if (x.length !== y.length) return false
for (let i = 0; i < x.length; i++) if (x[i] !== y[i]) return false
return true
}
export class Appview {
// corestore: a Corestore (favela's own copy — the index touches NO atproto-store cores, so there is
// no cross-copy instanceof gotcha here). name: the index core's namespace.
constructor (corestore, { name = 'favela-appview', getRecordStore = null, handleResolver = null, isForeign = () => true } = {}) {
// Optional: resolves + bidirectionally VERIFIES a DID's handle for bylines (see _hydrate).
this._handles = handleResolver
// Is this DID one a local hpp owns (roster) or a foreign peer's? Gates whether we may CREATE record
// bytes — see the guard in putPost. Defaults to "everything is foreign", which is the pre-existing
// behaviour: a caller that does not know its roster keeps writing exactly as before.
this._isForeign = typeof isForeign === 'function' ? isForeign : () => true
this.bee = new Hyperbee(corestore.get({ name }), { keyEncoding: 'utf-8', valueEncoding: 'json' })
this._timeline = this.bee.sub('timeline')
this._author = this.bee.sub('author')
this._tags = this.bee.sub('tags')
this._ref = this.bee.sub('ref')
// Profiles are indexed like posts because that is how atproto works: an AppView consumes the
// firehose and indexes app.bsky.actor.profile alongside app.bsky.feed.post, then serves feeds with
// the author ALREADY HYDRATED (app.bsky.feed.defs#postView carries a #profileViewBasic). Clients
// never resolve authors themselves. Before this, profile commits arrived on the firehose and were
// dropped by the POST_COLLECTIONS filter in applyCommit, so every node could render its OWN profile
// and nobody else's — remote authors showed as a raw DID forever.
this._profiles = this.bee.sub('profiles')
// blocks: `<did>\x00<subject>` -> {…}; blockRkey: `<did>\x00<rkey>` -> {subject}, so a delete op
// (which carries no record) can find what it removed — the same pair follows uses.
this._blocks = this.bee.sub('blocks')
this._blockRkey = this.bee.sub('blockrkey')
// Per-DID bookkeeping: which INGEST_VERSION this DID's history has been folded at.
this._meta = this.bee.sub('meta')
// ---- the ingest-time collection ledger (spec/convergence.md §2, org #33) --------------------------
//
// applyCommit used to end with a bare `continue` for every collection it had no branch for, so a NEW
// LEXICON appearing in a repo we follow vanished without a word. That is the same silence, in a fourth
// costume, that let a node sit broken for five days under four green checks: the code was working
// exactly as written, and what it discarded was never anybody's to see.
//
// `<collection>` -> { collection, class: 'unknown'|'ignored', count, firstSeen, lastSeen, dids[], sample }
//
// Keyed by COLLECTION, not by record, so the ledger grows with distinct lexicons (two today) and never
// with traffic — a repo pushing a million unknown records adds one key.
//
// PERSISTED, not in-memory, because the question outlives the process: an operator who restarts must
// not thereby lose the fact that something unclassified has been arriving. Deletion story: this is part
// of the appview index, which is a rebuildable local CACHE (see the header) — a rebuild re-derives the
// ledger from re-ingest, and nothing here is replicated or authoritative for anyone else.
this._collections = this.bee.sub('collections')
// Log an unknown collection ONCE per process. Without this a repo carrying a lexicon we do not know
// emits one warning per op forever, and a log that floods is read by nobody — which would reproduce
// the very failure this closes.
this._loggedUnknown = new Set()
// Social graph. Both are keyed so that the natural read is a RANGE SCAN and every write is
// idempotent — the same (liker, rkey) or (follower, subject) always lands on the same key.
//
// Deliberately NOT stored as counters. A counter is an increment, and increments are not idempotent
// under replay: the INGEST_VERSION backfill re-reads a peer's whole firehose, so every like would be
// counted again on every version bump. Counting keys in a range is O(n) per post instead of O(1), but
// it is CORRECT under replay, which a counter is not without a separate dedup ledger.
this._likes = this.bee.sub('likes') // <subjectUri>\x00<likerDid>\x00<rkey> -> { did, rkey, subject }
this._reposts = this.bee.sub('reposts') // <subjectUri>\x00<did>\x00<rkey> -> { did, rkey, subject }
this._repostedBy = this.bee.sub('repostedby') // <did>\x00<rkey> -> subjectUri (deletes carry only rkey)
this._likedBy = this.bee.sub('likedby') // <likerDid>\x00<rkey> -> subjectUri (for deletes, which carry only rkey)
this._follows = this.bee.sub('follows') // <followerDid>\x00<subjectDid> -> { did, subject, createdAt }
this._followers = this.bee.sub('followers') // <subjectDid>\x00<followerDid> -> same (reverse index)
this._followRkey = this.bee.sub('followrk')
// Reply threading. A reply is an ordinary post whose record carries reply: { root, parent } as
// strongRefs, so there is no new COLLECTION to ingest — only an extra index off the parent uri.
// Keyed parent-first + createdAt so a thread reads as one forward-chron range scan.
this._replies = this.bee.sub('replies') // <parentUri>\x00<createdAt>\x00<did>\x00<rkey> -> pointer
this._replyOf = this.bee.sub('replyof') // <did>\x00<rkey> -> parentUri (deletes carry only rkey) // <followerDid>\x00<rkey> -> subjectDid (deletes carry only rkey)
// A timeline is a handful of distinct authors repeated many times, and _hydrate runs per post.
// Memoized briefly so a 50-post page is a few reads, not fifty.
this._profCache = new Map()
// getRecordStore(did) -> OverlayRepoStore: the SHARED content-addressed record byte store, ONE physical
// copy per machine at ATPROTO_OVERLAY_ROOT/<roomKey16> — the SAME files hpp writes (identical layout +
// byte-exact DAG-CBOR). The index below holds only POINTERS { did, collection, rkey, createdAt }; the
// record BYTES live once in that store, never duplicated into this Hyperbee. Null => records unresolved.
this._getRecordStore = getRecordStore
}
async ready () { await this.bee.ready(); return this }
// Resolve a record's bytes from the shared store by (did, collection, rkey). null if no store / absent.
async _resolveRecord (did, collection, rkey) {
if (!this._getRecordStore) return null
try { const s = await this._getRecordStore(did); return s ? await s.get(collection, rkey) : null } catch { return null }
}
// The post an embed.record / embed.recordWithMedia points at, as { did, rkey, record, author } — or null
// when we do not hold it (a quote of a post outside this node's window is normal, and saying nothing is
// better than rendering an empty box). Never recurses: the returned view carries no `quoted` of its own.
async _resolveQuoted (record) {
const e = record && record.embed
if (!e) return null
const ref = e.$type === 'app.bsky.embed.record'
? e.record
: (e.$type === 'app.bsky.embed.recordWithMedia' && e.record ? (e.record.record || e.record) : null)
const uri = ref && typeof ref.uri === 'string' ? ref.uri : null
if (!uri) return null
const parts = uri.replace(/^at:\/\//, '').split('/')
if (parts.length !== 3 || !parts[0].startsWith('did:')) return null
const [did, collection, rkey] = parts
const rec = await this._resolveRecord(did, collection, rkey)
if (!rec) return null
const prof = await this.getProfile(did)
const h = this._handles ? await this._handles.resolve(did) : null
return {
did,
rkey,
record: rec,
author: {
did,
displayName: (prof && prof.displayName) || null,
avatar: (prof && prof.avatar) || null,
handle: displayHandle(h),
handleVerified: !!(h && h.verified)
}
}
}
// SEARCH — a scan over the existing subs, not an inverted index. See lib/search.js for why that is an
// honest trade at this corpus size and what to do when it stops being one. `scanCap` bounds the work so
// a query can never walk an unbounded index: it is a COST ceiling, and hitting it is reported
// (`truncated`) rather than silently returning a short list that looks complete.
async searchPosts (q, { limit = 25, viewerDid = null, scanCap = 5000 } = {}) {
const query = parseQuery(q)
if (!query.terms.length && !query.phrases.length) return { posts: [], cursor: null, truncated: false }
const hits = []
let scanned = 0
// Newest first, so an early exit keeps the most recent matches rather than an arbitrary slice.
for await (const e of this._timeline.createReadStream({ reverse: true })) {
if (++scanned > scanCap) break
const v = e && e.value
if (!v) continue
const rec = await this._resolveRecord(v.did, v.collection || 'app.bsky.feed.post', v.rkey)
if (!rec) continue
const text = postText(rec)
if (!matches(text, query)) continue
hits.push({ v, s: score(text, query) })
if (hits.length >= limit * 4) break // enough to rank meaningfully without scanning to the end
}
hits.sort((a, b) => b.s - a.s)
const posts = []
for (const h of hits.slice(0, limit)) {
const p = await this._hydrate(h.v, viewerDid)
if (p) posts.push(p)
}
return { posts, cursor: null, truncated: scanned > scanCap }
}
// People this node knows about, by display name, handle, DID or bio.
async searchActors (q, { limit = 10 } = {}) {
const query = parseQuery(q)
if (!query.terms.length && !query.phrases.length) return []
const hits = []
try {
for await (const e of this._profiles.createReadStream()) {
const did = String(e.key)
const prof = e.value || {}
const text = actorText(did, prof)
if (!matches(text, query)) continue
hits.push({ did, profile: prof, s: score(text, query) })
}
} catch { /* an unreadable profile sub must not fail the whole search */ }
hits.sort((a, b) => b.s - a.s)
return hits.slice(0, limit)
}
// Hydrate a pointer { did, collection, rkey, createdAt } into { did, rkey, record, createdAt } for reads.
async _hydrate (v, viewerDid = null) {
const record = await this._resolveRecord(v.did, v.collection || 'app.bsky.feed.post', v.rkey)
// Author rides the view, atproto-style — the render layer does no lookups of its own.
const prof = await this.getProfile(v.did)
// The HANDLE is not in the profile record — atproto keeps it in the DID document (alsoKnownAs), so
// it needs resolution, and resolution alone is not enough: alsoKnownAs is a CLAIM by the DID. The
// resolver verifies it bidirectionally and yields 'handle.invalid' when it does not check out, so an
// impostor DID claiming someone else's handle can never render as that person. Optional: with no
// resolver wired the byline falls back to a short DID rather than inventing a name.
const h = this._handles ? await this._handles.resolve(v.did) : null
const author = {
did: v.did,
displayName: (prof && prof.displayName) || null,
avatar: (prof && prof.avatar) || null,
handle: displayHandle(h),
handleVerified: !!(h && h.verified)
}
const uri = `at://${v.did}/${v.collection || 'app.bsky.feed.post'}/${v.rkey}`
const likeCount = await this.likeCount(uri)
// Viewer state rides the view, as it does in atproto (postView carries a #viewerState). The render
// layer must not have to ask "did I like this" per post.
const viewerLiked = viewerDid ? await this.hasLiked(viewerDid, uri) : false
const replyCount = await this.replyCount(uri)
const repostCount = await this.repostCount(uri)
const viewerReposted = viewerDid ? await this.hasReposted(viewerDid, uri) : false
// A QUOTE POST embeds a strongRef, not the quoted content. Until this, embed.record was rendered as
// NOTHING — a quote showed the commentary with the post it was about invisible, which reads as a
// non-sequitur. Resolved here for the same reason `author` is: the render layer does no lookups.
// One level only, deliberately — a quote of a quote of a quote would recurse a timeline into a tree.
const quoted = await this._resolveQuoted(record)
return { did: v.did, rkey: v.rkey, record, createdAt: v.createdAt, editedAt: v.editedAt || null, author, likeCount, replyCount, repostCount, uri, viewerLiked, viewerReposted, quoted }
}
_tlKey (createdAt, did, rkey) { return `${createdAt}${SEP}${did}${SEP}${rkey}` }
_auKey (did, createdAt, rkey) { return `${did}${SEP}${createdAt}${SEP}${rkey}` }
_tagKey (tag, createdAt, did, rkey) { return `${tag}${SEP}${createdAt}${SEP}${did}${SEP}${rkey}` }
_refKey (did, rkey) { return `${did}${SEP}${rkey}` }
// Every TRACKED collection, and whether this node stores its BYTES.
//
// ⛔ The bytes flag is not cosmetic. putLike/putRepost/putFollow/putBlock index a record without ever
// writing it to the record store — only posts and the profile get bytes. A convergence check that
// asked "do we have the bytes?" for a like would report a gap on every like this node has ever
// indexed, which is worse than the blind spot it replaced. See spec/convergence.md §2.
static get TRACKED () {
const m = new Map()
for (const c of POST_COLLECTIONS) m.set(c, { bytes: true, view: 'post' })
m.set(PROFILE_COLLECTION, { bytes: true, view: 'profile' })
m.set(LIKE_COLLECTION, { bytes: false, view: 'like' })
m.set(REPOST_COLLECTION, { bytes: false, view: 'repost' })
m.set(FOLLOW_COLLECTION, { bytes: false, view: 'follow' })
m.set(BLOCK_COLLECTION, { bytes: false, view: 'block' })
return m
}
// Refs this node has indexed for ONE collection, from the view that actually owns it.
//
// Each of the five non-post views keeps a `<did><SEP><rkey>` reverse index precisely so a record can be
// located by its rkey (it is what del*() uses); walking that prefix is the same question convergence
// asks. The profile is a single record per DID, keyed by did alone, so its rkey is always 'self'.
async * indexedRefsFor (did, collection) {
const meta = Appview.TRACKED.get(collection)
if (!meta) return
if (meta.view === 'post') {
for await (const ref of this.indexedRefs(did)) if (ref.collection === collection) yield ref
return
}
if (meta.view === 'profile') {
const p = await this._profiles.get(did)
if (p && p.value) yield { did, rkey: PROFILE_RKEY, collection }
return
}
const bee = { like: this._likedBy, repost: this._repostedBy, follow: this._followRkey, block: this._blockRkey }[meta.view]
if (!bee) return
const lo = `${did}${SEP}`
const hi = `${did}\x01`
for await (const { key } of bee.createReadStream({ gte: lo, lt: hi })) {
const i = key.indexOf(SEP)
if (i < 0) continue
yield { did, rkey: key.slice(i + 1), collection }
}
}
// Every record this node has INDEXED for a DID, as { did, rkey, collection, createdAt }.
//
// Exposed because "indexed" and "has bytes" are separate states here — the create-gate above may
// refuse byte authorship while still indexing — and lib/repair-bytes.js needs to walk the first set to
// find the gap in the second. Reaching into `_ref` from a caller would couple it to the key encoding.
async * indexedRefs (did) {
const lo = `${did}${SEP}`
const hi = `${did}\x01`
for await (const { key, value } of this._ref.createReadStream({ gte: lo, lt: hi })) {
const i = key.indexOf(SEP)
if (i < 0) continue
yield {
did,
rkey: key.slice(i + 1),
collection: (value && value.collection) || null,
createdAt: (value && value.createdAt) || null
}
}
}
// Index one post into both views. Idempotent by (did, rkey). If a prior index of the SAME (did, rkey)
// used a different createdAt (an edit that moved the timestamp), its stale timeline/author keys are
// removed first — otherwise the reverse-chron key would orphan.
// `edited` is set ONLY by applyCommit, from the firehose op's action. It is deliberately NOT
// inferred from "a prior entry exists": indexPeer re-puts every record on a snapshot reconcile, so
// inferring would mark the whole repo edited on any re-index. An edit is a claim about history and
// must come from the log that records history.
//
// Stored in the index (derived, rebuildable) rather than in the record — spec §10.5.2: atproto's
// $orig/$hist pattern exists because atproto has no history, and this substrate has some.
async putPost ({ did, collection = 'app.bsky.feed.post', rkey, record, edited = false }) {
const createdAt = (record && record.createdAt) || FLOOR
const tags = extractTags(record)
const refKey = this._refKey(did, rkey)
const prior = await this._ref.get(refKey)
const priorVal = prior && prior.value
// DEDUP GATE (the balloon fix): if this exact record is already indexed, skip the whole write. A blind
// re-put would append dead blocks — Hyperbee subs don't dedup — and indexPeer re-folds the full repo on
// every baseline/#sync, so an ungated re-index balloons the log on every reconnect.
const sig = recSig(record)
if (priorVal && priorVal.sig === sig) return
// Record BYTES go to the SHARED content-addressed store (one copy per machine). The index holds only a
// POINTER, never the record.
//
// NEVER LET A LOSSY COPY CLOBBER A RICHER ONE. This used to call put() unconditionally, on the
// assumption that our re-encode is "byte-identical to hpp's, so writing an already-present record is
// idempotent". It is not, and on 2026-07-30 that assumption froze wj62's repo for ~42h:
//
// hpp wrote the record WITH langs (390 bytes) and published a pointer to that hash
// -> we re-encoded our decoded copy WITHOUT langs (380 bytes) over the same file
// -> the converged pointer now referenced bytes that existed on NO node
// -> the signer could not resolve them, so the head never advanced again, and two posts
// ("we workin", "moar torren") were never published anywhere.
//
// These files are shared with hpp (ATPROTO_OVERLAY_ROOT is its record dir) and hpp is the AUTHORITY:
// its PointerSubstrate hashes the bytes it writes and commits that hash to the autobase. We write no
// pointer, so we must never invalidate one — but we DO remain the only writer for a peer whose records
// no local hpp holds, so a blanket "never overwrite" would freeze every ingested profile at its
// first-seen version. Refusing only a LOSSY overwrite threads both.
if (this._getRecordStore) {
try {
const s = await this._getRecordStore(did)
if (s) {
// getRaw returns { bytes, record } (or null) — NOT a buffer.
const existing = await s.getRaw(collection, rkey)
// ⛔ THE HOLE THE LOSSY GUARD BELOW COULD NOT SEE. isLossySubset can only compare against a file
// that is ALREADY on this disk. When the file is absent it had nothing to compare, so it wrote
// unconditionally — and "absent" is exactly the state hpp is in between publishing a pointer and
// this node materialising the bytes. Our re-encode then lands first, hashes differently from the
// pointer hpp already committed, and the signer can never resolve it: the head stops advancing and
// NOTHING federates. That is the 2026-07-31 recurrence (392-byte pointer, 382-byte file, `langs`
// dropped) — the same defect as 2026-07-30, through the one branch the fix did not cover.
//
// For a DID on our ROSTER, hpp is the AUTHORITY for these bytes and the only writer that may
// create them. Absence means "not here YET", never "ours to author". For a foreign peer no local
// hpp holds, we remain the only writer, so the original behaviour is preserved — which is what
// keeps ingested profiles from freezing at their first-seen version.
if (existing == null && !this._isForeign(did)) {
console.error(JSON.stringify({
msg: 'appview: refusing to CREATE bytes for a roster DID — hpp owns these and may already have published a pointer',
did,
collection,
rkey,
ourBytes: recEncodedLength(record)
}))
} else if (existing == null) await s.put(collection, rkey, record)
else if (!isLossySubset(record, existing.record)) await s.put(collection, rkey, record)
else {
// Divergence is now LOUD. Silence here is what let a clobbered record sit undetected for two
// days while every surface reported health. We keep the canonical bytes and say so, naming the
// dropped fields — that is what identifies the write path that lost them.
console.error(JSON.stringify({
msg: 'appview: our copy is LOSSY vs canonical bytes — refusing to overwrite',
did,
collection,
rkey,
canonicalBytes: existing.bytes.length,
ourBytes: recEncodedLength(record),
droppedFields: Object.keys(existing.record).filter((k) => !(k in record))
}))
}
}
} catch {}
}
// The record changed (or is new). Clear the OLD keys only when createdAt or the tag set MOVED — those
// define the key layout; if unchanged, the puts below overwrite in place (a text-only edit). Skipping
// the del avoids a tombstone append (del always appends, regardless of alwaysDuplicate).
if (priorVal && (priorVal.createdAt !== createdAt || !sameTags(priorVal.tags, tags))) await this._removeEntries(did, rkey, priorVal)
// Once edited, ALWAYS edited: a later unrelated re-put (re-index, snapshot reconcile) must not
// erase the fact. First edit wins the timestamp — it is "this has been edited", not "last edited".
const editedAt = (priorVal && priorVal.editedAt) || (edited ? new Date().toISOString() : null)
const val = { did, collection, rkey, createdAt } // POINTER only — record bytes live in the shared store
if (editedAt) val.editedAt = editedAt
await this._timeline.put(this._tlKey(createdAt, did, rkey), val)
await this._author.put(this._auKey(did, createdAt, rkey), val)
for (const tag of tags) await this._tags.put(this._tagKey(tag, createdAt, did, rkey), val)
// Thread edge, when this post is a reply. parent.uri is what a thread hangs off; root is not indexed
// because a thread view walks DOWN from a post, and bsky's own getPostThread does the same.
const parentUri = record && record.reply && record.reply.parent && record.reply.parent.uri
if (typeof parentUri === 'string' && parentUri.startsWith('at://')) {
await this._replies.put(`${parentUri}${SEP}${createdAt}${SEP}${did}${SEP}${rkey}`, val)
await this._replyOf.put(refKey, { parentUri, createdAt })
}
await this._ref.put(refKey, { createdAt, tags, sig, collection })
}
// Remove every index entry for a (did, rkey) given its stored ref { createdAt, tags }.
async _removeEntries (did, rkey, ref) {
const createdAt = ref.createdAt || FLOOR
await this._timeline.del(this._tlKey(createdAt, did, rkey))
await this._author.del(this._auKey(did, createdAt, rkey))
// Drop the thread edge too, or a deleted reply leaves a phantom in its parent's thread.
try {
const r = await this._replyOf.get(this._refKey(did, rkey))
if (r && r.value && r.value.parentUri) {
await this._replies.del(`${r.value.parentUri}${SEP}${r.value.createdAt || createdAt}${SEP}${did}${SEP}${rkey}`)
await this._replyOf.del(this._refKey(did, rkey))
}
} catch {}
for (const tag of (Array.isArray(ref.tags) ? ref.tags : [])) await this._tags.del(this._tagKey(tag, createdAt, did, rkey))
}
// Remove a post from all views (timeline/author/tags). No-op if not indexed.
async delPost ({ did, rkey }) {
const refKey = this._refKey(did, rkey)
const prior = await this._ref.get(refKey)
if (!prior || !prior.value) return false
const ref = prior.value
if (this._getRecordStore && ref.collection) { try { const s = await this._getRecordStore(did); if (s) await s.del(ref.collection, rkey) } catch {} }
await this._removeEntries(did, rkey, ref)
await this._ref.del(refKey)
return true
}
// Index a peer's profile record. Same two-layer shape as putPost: the record BYTES go to the shared
// content-addressed overlay store (one physical copy per machine, byte-identical to hpp's, so this is
// also how a remote profile ends up REPLICATED into hyper-overlay rather than living only on its home
// node) and the index keeps a small projection for hydration. Dedup-gated by content signature for the
// same reason putPost is: Hyperbee subs do not dedup, and a re-fold on every reconnect would balloon.
//
// `avatar` is stored as the RAW blob ref, not a cid string — normalizing it is the render layer's job
// (it owns blobCid and the four shapes a ref arrives in), and the appview should not grow a second
// copy of that knowledge.
async putProfile ({ did, record }) {
if (!did || !record) return false
const sig = recSig(record)
const prior = await this._profiles.get(did)
if (prior && prior.value && prior.value.sig === sig) return false
if (this._getRecordStore) { try { const s = await this._getRecordStore(did); if (s) await s.put(PROFILE_COLLECTION, PROFILE_RKEY, record) } catch {} }
await this._profiles.put(did, {
did,
displayName: typeof record.displayName === 'string' ? record.displayName : null,
description: typeof record.description === 'string' ? record.description : null,
avatar: record.avatar || null,
sig
})
this._profCache.delete(did)
return true
}
async delProfile (did) {
const prior = await this._profiles.get(did)
if (!prior || !prior.value) return false
if (this._getRecordStore) { try { const s = await this._getRecordStore(did); if (s) await s.del(PROFILE_COLLECTION, PROFILE_RKEY) } catch {} }
await this._profiles.del(did)
this._profCache.delete(did)
return true
}
// The projected profile for a DID, or null. Memoized briefly (see _profCache).
async getProfile (did) {
if (!did) return null
const now = Date.now()
const hit = this._profCache.get(did)
if (hit && now - hit.at < PROF_TTL_MS) return hit.value
let value = null
try { const r = await this._profiles.get(did); value = (r && r.value) || null } catch { value = null }
if (this._profCache.size > 500) this._profCache.clear()
this._profCache.set(did, { at: now, value })
return value
}
// Replies to a post, forward-chron (a conversation reads oldest-first, unlike a timeline).
async getReplies (uri, { limit = 50, viewerDid = null } = {}) {
const out = []
if (!uri) return out
try {
for await (const { value } of this._replies.createReadStream({ gte: `${uri}${SEP}`, lt: `${uri}\x01`, limit })) {
if (value) out.push(await this._hydrate(value, viewerDid))
}
} catch {}
return out
}
// Parse an at:// post URI into its parts. Returns null for anything that is not one, so a malformed
// permalink is a 404 rather than a throw.
_parseUri (uri) {
if (typeof uri !== 'string' || !uri.startsWith('at://')) return null
const rest = uri.slice('at://'.length)
const i = rest.indexOf('/')
if (i < 0) return null
const did = rest.slice(0, i)
const tail = rest.slice(i + 1)
const j = tail.lastIndexOf('/')
if (j < 0) return null
const collection = tail.slice(0, j)
const rkey = tail.slice(j + 1)
if (!did || !collection || !rkey) return null
return { did, collection, rkey }
}
// POINT lookup of one post. The permalink route used to find a post by scanning the author's last 100
// and matching rkey, so ANY older post 404'd on its own permalink. `_ref` already maps
// (did, rkey) -> { createdAt, collection }, which is the same answer in one get and has no horizon.
async getPost (uri, { viewerDid = null } = {}) {
const p = this._parseUri(uri)
if (!p) return null
const ref = await this._ref.get(this._refKey(p.did, p.rkey))
if (!ref || !ref.value) return null
return this._hydrate({ did: p.did, collection: ref.value.collection || p.collection, rkey: p.rkey, createdAt: ref.value.createdAt }, viewerDid)
}
// A thread in BOTH directions, mirroring app.bsky.feed.getPostThread and its two knobs: `depth` walks
// DOWN (descendants), `parentHeight` walks UP (ancestors). Same defaults as the lexicon (6 / 80).
//
// NB the comment on the reply index used to say root is not indexed "because a thread view walks DOWN
// from a post, and bsky's own getPostThread does the same". bsky walks BOTH — parentHeight is an
// ancestor walk — so that premise was wrong. Ancestors are reachable anyway via `_replyOf` (child ->
// parent) one hop at a time, which is what this does; indexing root would only buy an O(1) jump to the
// top of a thread.
//
// An ancestor that is NOT in this node's window stops the walk and is reported as { missing, uri }
// rather than dropped. "The rest of this thread is outside our firehose" is true and worth rendering,
// and it is precisely the hole a lens import would later fill; truncating silently would instead look
// like the thread simply started there.
async getPostThread (uri, { depth = 6, parentHeight = 80, viewerDid = null, maxNodes = 200 } = {}) {
const post = await this.getPost(uri, { viewerDid })
if (!post) return null
const ancestors = []
const seen = new Set([uri])
let cur = uri
for (let i = 0; i < parentHeight; i++) {
const p = this._parseUri(cur)
if (!p) break
const r = await this._replyOf.get(this._refKey(p.did, p.rkey))
const parentUri = r && r.value && r.value.parentUri
// `seen` guards a malformed cycle (a reply chain that points back into itself). Data we did not
// author can be arbitrary, and a cycle here would hang the request forever.
if (!parentUri || seen.has(parentUri)) break
seen.add(parentUri)
const parent = await this.getPost(parentUri, { viewerDid })
if (!parent) { ancestors.unshift({ missing: true, uri: parentUri }); break }
ancestors.unshift(parent)
cur = parentUri
}
// Descendants. maxNodes bounds the TOTAL, not the per-level fan-out, so one very wide thread cannot
// pin the event loop of a single-process SSR gateway.
let nodes = 0
const walk = async (parentUri, d) => {
if (d >= depth || nodes >= maxNodes) return []
const kids = await this.getReplies(parentUri, { viewerDid })
const out = []
for (const kid of kids) {
if (nodes >= maxNodes) break
nodes++
out.push({ post: kid, replies: await walk(kid.uri, d + 1) })
}
return out
}
const replies = await walk(uri, 0)
return { post, ancestors, replies, truncated: nodes >= maxNodes }
}
// Count only — the timeline wants a number, not the posts. Range scan, idempotent under replay for the
// same reason likeCount is.
async replyCount (uri) {
if (!uri) return 0
let n = 0
try { for await (const _ of this._replies.createReadStream({ gte: `${uri}${SEP}`, lt: `${uri}\x01` })) n++ } catch { return 0 }
return n
}
// ── social graph ────────────────────────────────────────────────────────────────────────────────
// A like's subject is a strong ref { uri, cid }; the uri is the at:// of the liked post. A delete
// commit carries only the rkey, so _likedBy maps (liker, rkey) -> subject to make removal possible.
async putLike ({ did, rkey, record }) {
const uri = record && record.subject && record.subject.uri
if (!did || !rkey || typeof uri !== 'string' || !uri.startsWith('at://')) return false
await this._likes.put(`${uri}${SEP}${did}${SEP}${rkey}`, { did, rkey, subject: uri, createdAt: (record && record.createdAt) || null })
await this._likedBy.put(`${did}${SEP}${rkey}`, { subject: uri })
return true
}
async delLike ({ did, rkey }) {
const r = await this._likedBy.get(`${did}${SEP}${rkey}`)
if (!r || !r.value || !r.value.subject) return false
await this._likes.del(`${r.value.subject}${SEP}${did}${SEP}${rkey}`)
await this._likedBy.del(`${did}${SEP}${rkey}`)
return true
}
// How many distinct (liker, rkey) entries point at this post. Range scan — see the note on counters.
async likeCount (uri) {
if (!uri) return 0
let n = 0
try { for await (const _ of this._likes.createReadStream({ gte: `${uri}${SEP}`, lt: `${uri}\x01` })) n++ } catch { return 0 }
return n
}
// The rkey of this DID's like of that post, or null. A toggle needs it: unliking is a DELETE of the
// liker's OWN record, and only the liker's repo knows its rkey — it is not derivable from the subject.
async likeRkey (did, uri) {
if (!did || !uri) return null
try {
for await (const { value } of this._likes.createReadStream({ gte: `${uri}${SEP}${did}${SEP}`, lt: `${uri}${SEP}${did}\x01` })) {
if (value && value.rkey) return value.rkey
}
} catch {}
return null
}
// The rkey of this DID's follow of that subject, or null. Same reason as likeRkey.
async followRkey (did, subject) {
if (!did || !subject) return null
try { const r = await this._follows.get(`${did}${SEP}${subject}`); return (r && r.value && r.value.rkey) || null } catch { return null }
}
// Did this DID like that post? Used to render the viewer's own like state.
async hasLiked (did, uri) {
if (!did || !uri) return false
try {
for await (const { value } of this._likes.createReadStream({ gte: `${uri}${SEP}${did}${SEP}`, lt: `${uri}${SEP}${did}\x01` })) { if (value) return true }
} catch {}
return false
}
// A repost's subject is the same strongRef shape a like uses, so this mirrors putLike exactly —
// including storing keys rather than a counter, so a replay cannot inflate the count.
async putRepost ({ did, rkey, record }) {
const uri = record && record.subject && record.subject.uri
if (!did || !rkey || typeof uri !== 'string' || !uri.startsWith('at://')) return false
await this._reposts.put(`${uri}${SEP}${did}${SEP}${rkey}`, { did, rkey, subject: uri, createdAt: (record && record.createdAt) || null })
await this._repostedBy.put(`${did}${SEP}${rkey}`, { subject: uri })
return true
}
async delRepost ({ did, rkey }) {
const r = await this._repostedBy.get(`${did}${SEP}${rkey}`)
if (!r || !r.value || !r.value.subject) return false
await this._reposts.del(`${r.value.subject}${SEP}${did}${SEP}${rkey}`)
await this._repostedBy.del(`${did}${SEP}${rkey}`)
return true
}
async repostCount (uri) {
if (!uri) return 0
let n = 0
try { for await (const _ of this._reposts.createReadStream({ gte: `${uri}${SEP}`, lt: `${uri}\x01` })) n++ } catch { return 0 }
return n
}
async hasReposted (did, uri) {
if (!did || !uri) return false
try { for await (const { value } of this._reposts.createReadStream({ gte: `${uri}${SEP}${did}${SEP}`, lt: `${uri}${SEP}${did}\x01` })) { if (value) return true } } catch {}
return false
}
// Needed to UNrepost: the delete targets our own record, whose rkey only our repo knows.
async repostRkey (did, uri) {
if (!did || !uri) return null
try {
for await (const { value } of this._reposts.createReadStream({ gte: `${uri}${SEP}${did}${SEP}`, lt: `${uri}${SEP}${did}\x01` })) { if (value && value.rkey) return value.rkey }
} catch {}
return null
}
async putFollow ({ did, rkey, record }) {
const subject = record && record.subject
if (!did || !rkey || typeof subject !== 'string' || !subject.startsWith('did:')) return false
const val = { did, subject, rkey, createdAt: (record && record.createdAt) || null }
await this._follows.put(`${did}${SEP}${subject}`, val)
await this._followers.put(`${subject}${SEP}${did}`, val)
await this._followRkey.put(`${did}${SEP}${rkey}`, { subject })
return true
}
async delFollow ({ did, rkey }) {
const r = await this._followRkey.get(`${did}${SEP}${rkey}`)
if (!r || !r.value || !r.value.subject) return false
const subject = r.value.subject
await this._follows.del(`${did}${SEP}${subject}`)
await this._followers.del(`${subject}${SEP}${did}`)
await this._followRkey.del(`${did}${SEP}${rkey}`)
return true
}
// A BLOCK is public and federates by design (unlike a mute, which is private and node-local). Indexed
// so a block made on another device — or by a peer — is visible here rather than only in this node's
// own bookkeeping.
async putBlock ({ did, rkey, record }) {
const subject = record && record.subject
if (!did || !rkey || typeof subject !== 'string' || !subject.startsWith('did:')) return false
await this._blocks.put(`${did}${SEP}${subject}`, { did, subject, rkey, createdAt: (record && record.createdAt) || null })
await this._blockRkey.put(`${did}${SEP}${rkey}`, { subject })
return true
}
async delBlock ({ did, rkey }) {
const r = await this._blockRkey.get(`${did}${SEP}${rkey}`)
if (!r || !r.value || !r.value.subject) return false
await this._blocks.del(`${did}${SEP}${r.value.subject}`)
await this._blockRkey.del(`${did}${SEP}${rkey}`)
return true
}
// Every DID this DID blocks, from ANY device — the point of indexing the record rather than trusting
// local bookkeeping alone.
async blocking (did) {
const out = new Set()
if (!did) return out
try { for await (const { value } of this._blocks.createReadStream({ gte: `${did}${SEP}`, lt: `${did}\x01` })) if (value && value.subject) out.add(value.subject) } catch {}
return out
}
// The rkey of the block record, so an unblock can delete the right one wherever it was created.
async blockRkeyFor (did, subject) {
if (!did || !subject) return null
try { const r = await this._blocks.get(`${did}${SEP}${subject}`); return (r && r.value && r.value.rkey) || null } catch { return null }
}
// The DIDs this DID follows. This is the author set a Following feed scopes on — getTimeline already
// takes `authors`, so a following feed is this list handed to the existing filter.
async following (did) {
const out = []
if (!did) return out
try { for await (const { value } of this._follows.createReadStream({ gte: `${did}${SEP}`, lt: `${did}\x01` })) if (value && value.subject) out.push(value.subject) } catch {}
return out
}
async followers (did) {
const out = []
if (!did) return out
try { for await (const { value } of this._followers.createReadStream({ gte: `${did}${SEP}`, lt: `${did}\x01` })) if (value && value.did) out.push(value.did) } catch {}
return out
}
// ---- collection classification at ingest (spec/convergence.md §2, org #33) ------------------------
//
// The fallthrough of applyCommit, made EXPLICIT. Three outcomes, and the difference between them is the
// whole point:
// TRACKED — never reaches here; a view indexed it above.
// IGNORED-BY-POLICY — a decision on file with a stated reason (lib/hosted-dids.js). Counted, not logged.
// UNKNOWN — a lexicon nobody has classified. An operator event, logged once per collection.
//
// Returns the class so a caller can act on it; null for a malformed op.
//
// ⛔ NEVER THROWS. This sits in the live ingest path. A ledger write that failed must not cost us a
// commit: a dropped counter is a worse report, a dropped commit is a data gap, and they are not close.
async noteUnhandled (did, collection, rkey = null) {
if (typeof collection !== 'string' || collection.length === 0) return null
// A TRACKED collection reaching a fallthrough means a MALFORMED record (e.g. a profile at an rkey other
// than `self`), not an unclassified lexicon. Recording it as UNKNOWN would send an operator to
// hosted-dids.js to classify something already classified. Not this ledger's question.
if (Appview.TRACKED.has(collection)) return 'tracked'
const cls = isIgnoredCollection(collection) ? 'ignored' : 'unknown'
if (cls === 'unknown' && !this._loggedUnknown.has(collection)) {
this._loggedUnknown.add(collection)
console.warn(JSON.stringify({
msg: 'ingest: UNKNOWN collection — not tracked by any view and not on the ignore list',
collection,
did: did || null,
rkey: rkey || null,
action: 'classify it: add a view, or add it to IGNORED_COLLECTIONS in lib/hosted-dids.js with a reason (spec/convergence.md §2.1)'
}))
}
try {
const now = new Date().toISOString()
const prior = await this._collections.get(collection)
const v = (prior && prior.value) || null
// Bounded: the entry records WHICH repos are sending this, but must not become a list of every DID
// on a large fleet. Twenty is enough to recognise a pattern; the count carries the magnitude.
const dids = new Set((v && Array.isArray(v.dids)) ? v.dids : [])
if (did && dids.size < 20) dids.add(did)
await this._collections.put(collection, {
collection,
class: cls,
count: ((v && v.count) || 0) + 1,
firstSeen: (v && v.firstSeen) || now,
lastSeen: now,
dids: [...dids],
sample: (v && v.sample) || (rkey ? { did: did || null, rkey } : null)
})
} catch { /* diagnostics must never cost an ingest */ }
return cls
}
// What has actually ARRIVED that no view indexes. Deliberately distinct from `diff`'s unknown count,
// which is derived from the AUTHORITY's repo at one instant: this is the live firehose's own record, and
// it survives a restart. A collection can appear here and not there (the authority deleted it since) or
// there and not here (we have never been online for one of its commits), and both are worth knowing.
// UNKNOWN sorts first — it is the class that should make someone look.
async collectionLedger () {
const out = []
try {
for await (const { value } of this._collections.createReadStream()) if (value && value.collection) out.push(value)
} catch { return out }
out.sort((a, b) => (a.class === b.class ? (b.count || 0) - (a.count || 0) : (a.class === 'unknown' ? -1 : 1)))
return out
}
async unknownCollections () { return (await this.collectionLedger()).filter((e) => e.class === 'unknown') }
// The INGEST_VERSION this DID's history has been folded at (0 = never recorded, i.e. pre-versioning).
async ingestVersion (did) {
try { const r = await this._meta.get('ingest:' + did); return (r && r.value && r.value.v) || 0 } catch { return 0 }
}
// True when this build indexes more than the build that last folded this DID's history — so its
// firehose must be replayed from 0 once. Fails OPEN (true): a replay is idempotent thanks to the
// recSig dedup gate, whereas wrongly skipping it leaves records permanently missing.
async needsBackfill (did) {
try { return (await this.ingestVersion(did)) < INGEST_VERSION } catch { return true }
}
// Record that this DID's history has been folded at the current version. Called only once the replay
// has actually caught up, never on merely STARTING one — marking early would make a crash mid-replay
// look like a completed backfill, which is the one outcome that is silently wrong forever.
async markBackfilled (did, v = INGEST_VERSION) {
try { await this._meta.put('ingest:' + did, { v, at: Date.now() }); return true } catch { return false }
}
// SNAPSHOT reconcile of one peer's current repo into the index (baseline). Folds every favela post
// collection (public + local — see scopes.js) via listRecords, AND removes previously-indexed records
// for this DID that are no longer present (delete-correctness a pure add-only fold can't do). Returns
// the count of records indexed (present).
async indexPeer (peer) {
const did = peer.did
const desired = new Map() // rkey -> { collection, record }
// A noRepo peer (co-hosted DID) has NO repo core — its records live in the SHARED overlay store, which
// is the authoritative view (bypasses membrane fork/desync). Source the baseline from there. Every other
// peer keeps sourcing from its own repo core (listRecords) — unchanged. Discriminate on peer._noRepo so
// the store-write of ordinary peers (for read-hydration) never becomes an aliased second source of truth.
let fromStore = false
if (peer._noRepo && this._getRecordStore) {
try {
const store = await this._getRecordStore(did)
const recs = store ? await store.records() : []
fromStore = true // authoritative: a noRepo peer has no repo core to fall back to
for (const r of recs) {
const i = r.key.indexOf('/')
if (i <= 0) continue
const collection = r.key.slice(0, i)
if (POST_COLLECTIONS.includes(collection)) desired.set(r.key.slice(i + 1), { collection, record: r.record })
else if (collection === PROFILE_COLLECTION && r.key.slice(i + 1) === PROFILE_RKEY) await this.putProfile({ did, record: r.record })
else if (collection === LIKE_COLLECTION) await this.putLike({ did, rkey: r.key.slice(i + 1), record: r.record })
else if (collection === REPOST_COLLECTION) await this.putRepost({ did, rkey: r.key.slice(i + 1), record: r.record })
else if (collection === FOLLOW_COLLECTION) await this.putFollow({ did, rkey: r.key.slice(i + 1), record: r.record })
else if (collection === BLOCK_COLLECTION) await this.putBlock({ did, rkey: r.key.slice(i + 1), record: r.record })
// The SNAPSHOT path had the same silent tail as applyCommit's (org #33): this `else if` chain
// simply ended, so a lexicon in the store that no branch matched was dropped without a word.
// The other baseline path below cannot have this bug — it asks for named collections one at a
// time and never enumerates — which also means it can never DISCOVER an unknown one. This branch
// is the only place a snapshot sees the whole repo, so it is the only place that can report.
else await this.noteUnhandled(did, collection, r.key.slice(i + 1))
}
} catch {}
}
if (!fromStore) {
// ⛔ `peer.listRecords` RETURNS { records, cursor }, NOT AN ARRAY (atproto-did-peer 0.2.10, per the
// lexicon). Every iteration below read the wrapper. This path is currently unreachable in
// favela-web — bin/favela-web.js:539 makes EVERY DID noRepo, so `fromStore` is always taken — which
// is the only reason it never threw. It is still wrong, and it is the branch a store-open failure
// falls into, so the failure mode would have been a TypeError at exactly the moment the store was
// already misbehaving. Normalised once here rather than at three call sites.
const listOf = async (coll) => {
const r = await peer.listRecords(coll)
return Array.isArray(r) ? r : ((r && r.records) || [])
}
// Baseline the profile too — a peer that has not edited it since we connected emits no commit, so
// the fold is the only place a first-time author's profile is picked up.
try {
for (const r of await listOf(PROFILE_COLLECTION)) {
const rkey = String(r.uri || r.key || '').split('/').pop()
if (rkey === PROFILE_RKEY && r.value) await this.putProfile({ did, record: r.value })
}
} catch {}
for (const [coll, put] of [[LIKE_COLLECTION, 'putLike'], [FOLLOW_COLLECTION, 'putFollow'], [REPOST_COLLECTION, 'putRepost'], [BLOCK_COLLECTION, 'putBlock']]) {
try {
for (const r of await listOf(coll)) {
const rkey = String(r.uri || r.key || '').split('/').pop()
if (rkey && r.value) await this[put]({ did, rkey, record: r.value })
}
} catch {}
}
for (const collection of POST_COLLECTIONS) {
for (const r of await listOf(collection)) { // [{ uri|key, cid, value }]
const path = r.uri || r.key || ''
const rkey = path.split('/').pop()
if (rkey) desired.set(rkey, { collection, record: r.value })
}
}
}
// Remove stragglers: indexed rkeys for this DID that are no longer in the repo.
const lo = `${did}${SEP}`
const hi = `${did}\x01`
const stale = []
for await (const { key } of this._ref.createReadStream({ gte: lo, lt: hi })) {
const i = key.indexOf(SEP)
const rkey = i >= 0 ? key.slice(i + 1) : null
if (rkey && !desired.has(rkey)) stale.push(rkey)
}
for (const rkey of stale) await this.delPost({ did, rkey })
// Add / update present records.
for (const [rkey, { collection, record }] of desired) await this.putPost({ did, collection, rkey, record })
return desired.size
}
// INCREMENTAL ingest of normalized commit ops [{ action, collection, rkey, record? }] — decoded by
// DidPeer.commitOps from a #commit frame (the record is read from the frame's own CAR, race-free).
// create/update -> putPost, delete -> delPost, for favela post collections only. Returns ops applied.
async applyCommit (did, ops) {
if (!Array.isArray(ops)) return 0
let n = 0
for (const op of ops) {
// Profile commits ride the SAME firehose as posts and used to be discarded here. The record is read
// from the frame's own CAR, so indexing it needs no fetch — the bytes already arrived.
if (op.collection === PROFILE_COLLECTION) {
if (op.action === 'delete') { if (await this.delProfile(did)) n++ ; continue }
if ((op.action === 'create' || op.action === 'update') && op.record) { if (await this.putProfile({ did, record: op.record })) n++ }
continue
}
if (op.collection === LIKE_COLLECTION) {
if (op.action === 'delete') { if (await this.delLike({ did, rkey: op.rkey })) n++ ; continue }
if (op.record && await this.putLike({ did, rkey: op.rkey, record: op.record })) n++
continue
}
if (op.collection === REPOST_COLLECTION) {
if (op.action === 'delete') { if (await this.delRepost({ did, rkey: op.rkey })) n++ ; continue }
if (op.record && await this.putRepost({ did, rkey: op.rkey, record: op.record })) n++
continue
}
if (op.collection === FOLLOW_COLLECTION) {
if (op.action === 'delete') { if (await this.delFollow({ did, rkey: op.rkey })) n++ ; continue }
if (op.record && await this.putFollow({ did, rkey: op.rkey, record: op.record })) n++
continue
}
if (op.collection === BLOCK_COLLECTION) {
if (op.action === 'delete') { if (await this.delBlock({ did, rkey: op.rkey })) n++ ; continue }
if (op.record && await this.putBlock({ did, rkey: op.rkey, record: op.record })) n++
continue
}
// ⛔ NOT A BARE `continue` (org #33). Anything reaching here is either IGNORED-BY-POLICY or UNKNOWN,
// and collapsing those two is exactly the mistake spec §2 exists to prevent: one is a decision, the
// other is a lexicon nobody has looked at. Both are recorded; only the second warns.
if (!POST_COLLECTIONS.includes(op.collection)) { await this.noteUnhandled(did, op.collection, op.rkey || null); continue }
if (op.action === 'delete') { await this.delPost({ did, rkey: op.rkey }); n++; continue }
if ((op.action === 'create' || op.action === 'update') && op.record) {
// op.action comes off the #commit frame, so 'update' IS the firehose's own statement that
// this rkey was edited — the prev chain, read rather than re-implemented (spec §10.5.2).
await this.putPost({ did, collection: op.collection, rkey: op.rkey, record: op.record, edited: op.action === 'update' }); n++
}
}
return n
}
// Reverse-chron timeline. cursor = the last key from a prior page (exclusive).
// `authors` (optional Set|array of DIDs) scopes the timeline to those authors — this is the PER-VIEWER
// view (DESIGN §7.5): a signed-in viewer sees the union of the member DIDs of the communities they
// belong to, computed per-request by lib/communities.js. Omitted/empty → the global node timeline
// (anonymous / unscoped). The filter is applied over the reverse-chron stream so paging still yields
// `limit` matching posts; because the index only holds the node's rostered DIDs, this is naturally the
// INTERSECTION of the viewer's communities with what this node serves.
async getTimeline ({ limit = 30, cursor = null, authors = null, viewerDid = null } = {}) {
const allow = normAuthors(authors)
const posts = []
let lastKey = null
// When filtering we may scan past `limit` index entries to collect `limit` matches — keep streaming
// until we have enough (or the stream ends), advancing the cursor to the last SCANNED key.
const stream = this._timeline.createReadStream({ ...(cursor ? { lt: cursor } : {}), reverse: true, ...(allow ? {} : { limit }) })
for await (const { key, value } of stream) {
lastKey = key
if (allow && !allow.has(value.did)) continue
posts.push(await this._hydrate(value, viewerDid))
if (posts.length >= limit) break
}
// More pages remain iff we filled this one (unfiltered: exactly limit; filtered: we broke on limit).
return { posts, cursor: posts.length >= limit ? lastKey : null }
}
// Reverse-chron feed for one author. All of a DID's keys live in [`${did}\x00`, `${did}\x01`).
// reverse pagination: keep the DID floor as gte, move the ceiling (lt) down to the last key seen.
//
// `filter` mirrors app.bsky.feed.getAuthorFeed's knownValues, so the profile tabs are the same split
// bsky makes rather than one we invented: posts_with_replies (default), posts_no_replies,
// posts_with_media, posts_and_author_threads, posts_with_video. An unknown value degrades to the
// default instead of erroring — a filter is a view preference, not an assertion, and a bad query
// string should not 500 a profile.
//
// Filtering happens OVER the reverse-chron stream (the timeline's pattern), so a page still yields
// `limit` matches rather than "limit scanned, some discarded" — otherwise a reply-heavy profile shows
// three posts under posts_no_replies and looks empty.
async getAuthorFeed (did, { limit = 30, cursor = null, viewerDid = null, filter = 'posts_with_replies' } = {}) {
const f = AUTHOR_FILTERS.has(filter) ? filter : 'posts_with_replies'
const unfiltered = f === 'posts_with_replies'
const needsRecord = f === 'posts_with_media' || f === 'posts_with_video'
const gte = `${did}${SEP}`
const lt = cursor || `${did}\x01`
const posts = []
let lastKey = null
const stream = this._author.createReadStream({ gte, lt, reverse: true, ...(unfiltered ? { limit } : {}) })
for await (const { key, value } of stream) {
lastKey = key
if (!unfiltered) {
// Reply-ness is one cheap get off the child->parent index — decided BEFORE hydrating, since
// hydration costs a record read plus profile and three counts per post.
const r = await this._replyOf.get(this._refKey(value.did, value.rkey))
const parentUri = r && r.value && r.value.parentUri
if (f === 'posts_no_replies' && parentUri) continue
if (f === 'posts_and_author_threads' && parentUri) {
// A self-thread only. Replying to someone ELSE is a conversation, not the author's own thread.
const p = this._parseUri(parentUri)
if (!p || p.did !== did) continue
}
if (needsRecord) {
const record = await this._resolveRecord(value.did, value.collection || 'app.bsky.feed.post', value.rkey)
const m = embedMedia(record)
if (f === 'posts_with_media' && !(m.images || m.video)) continue
if (f === 'posts_with_video' && !m.video) continue
}
}
posts.push(await this._hydrate(value, viewerDid))
if (posts.length >= limit) break
}
return { posts, cursor: posts.length >= limit ? lastKey : null }
}
// Reverse-chron feed for one topic/hashtag (a custom feed = the atproto feed-generator "filter",
// run locally). All of a tag's keys live in [`${tag}\x00`, `${tag}\x01`); pagination mirrors the
// author feed (move the `lt` ceiling down to the last key seen).
async getTagFeed (tag, { limit = 30, cursor = null, viewerDid = null } = {}) {
const t = normTag(tag)
if (!t) return { posts: [], cursor: null }
const gte = `${t}${SEP}`
const lt = cursor || `${t}\x01`
const posts = []
let lastKey = null
for await (const { key, value } of this._tags.createReadStream({ gte, lt, reverse: true, limit })) {
posts.push(await this._hydrate(value, viewerDid)); lastKey = key
}
return { posts, cursor: posts.length === limit ? lastKey : null }
}
async close () { await this.bee.close() }
}