// repo-sync — "does my copy match the authoritative repo, and pull what is missing".
//
// ⛔ THE QUESTION NOTHING ANSWERED. This node had `conformance` (do WE appear in bsky's appview?),
// `fronts` (whose firehose are we following?) and `diag` (are our cores conflicted?). All three reported
// health on 2026-08-05 while ashatron's copy of wj62 was five days stale, because none of them asks the
// INBOUND question. The drift was found by a hand audit, and then mis-diagnosed twice, purely because
// there was no verb for "compare me to the authority".
//
// THE THREE STATES, kept distinct because they have different causes and different repairs:
// missingFromIndex — the repo has it, we never indexed it (ashatron's actual gap)
// indexedWithoutBytes — we indexed it, the bytes were refused/lost (what repair-bytes fixes)
// extraLocal — we hold it, the repo does not (a delete we never folded)
// Collapsing these into one number is how "140/140 have bytes" read as healthy on a node missing 26
// records outright.
//
// WHY getRepo AND NOT THE FIREHOSE. Boot already replays subscribeRepos(0) for a non-co-hosted DID and
// demonstrably did not recover these records. getRepo is one request carrying the signed commit and the
// whole MST: a single signature check covers every record, the set is complete by construction, and
// there is no cursor to be quietly parked at the wrong place. Measured 45,563 bytes for wj62.
//
// ⚠ `since` IS NOT USED HERE, DELIBERATELY. It used to be sent — and worked only because the membrane
// accepted and ignored it (measured 2026-08-06: byte-identical full CAR either way). Since atproto-bridge
// 0.9.45 the membrane HONOURS it and returns a genuine diff, which this node cannot apply: a partial CAR
// omits the unchanged MST nodes, so applying one needs a BLOCK store to union against and favela holds
// RECORDS. See createFetchRepo below for the full reasoning; it now refuses `since` loudly.
//
// TRUST. The verification key comes from the DID DOCUMENT, never from the host that served the CAR, for
// the reason record-fetch.js states: a key handed over by the same party as the data proves nothing.
import { readCar, loadCommit, listRecords, getBlock, verifyRepo } from 'atproto-repo/lib/repo.js'
import { verifyCommit } from 'atproto-repo/lib/commit.js'
import * as dagCbor from '@ipld/dag-cbor'
import b4a from 'b4a'
const splitKey = (key) => {
const i = key.indexOf('/')
return i > 0 ? { collection: key.slice(0, i), rkey: key.slice(i + 1) } : null
}
// ---- fetch ------------------------------------------------------------------------------------------
// Pull the authoritative repo CAR. Returns { carBytes, identity, partial } or null — fails soft like
// record-fetch.
//
// ⛔ FAVELA CANNOT CONSUME A PARTIAL CAR, AND MUST NOT ASK FOR ONE. As of atproto-bridge 0.9.45 the
// membrane HONOURS `since` and returns a real diff: the MST nodes that did not change are deliberately
// absent, so the tree cannot be walked from the root. Applying one requires a BLOCK store to union it
// with — and this node holds RECORDS, not blocks. There is nothing here to union against.
//
// Nor could a diff answer the question `diff` asks. `extraLocal` ("we hold it, the repo does not") is
// only meaningful against the COMPLETE record set; computed against a diff, every record older than
// `since` looks extra. Nothing would be deleted (spec §2 forbids it) but the report would cry wolf on
// every run, which is the failure mode that made the first per-view diff useless.
//
// So `since` is NOT sent. It was sent before — accepted and ignored by the membrane, which is why it
// appeared to work — and this is the silent regression the issue predicted, closed before it could
// happen rather than after. The parameter stays in the signature and is REFUSED loudly, because a
// caller that passes it is asking for something this cannot do and deserves to be told.
export function createFetchRepo ({ doFetch, resolveIdentity, timeoutMs = 60_000 } = {}) {
if (typeof doFetch !== 'function') throw new Error('createFetchRepo requires doFetch')
if (typeof resolveIdentity !== 'function') throw new Error('createFetchRepo requires resolveIdentity')
return async function fetchRepo ({ did, since = null } = {}) {
if (since) throw new Error('fetchRepo: `since` is not supported here — a partial CAR needs a block store to apply against, and this node holds records. See lib/repo-sync.js.')
const identity = await resolveIdentity(did)
if (identity === null) return null
const url = `${identity.pds}/xrpc/com.atproto.sync.getRepo?did=${encodeURIComponent(did)}`
try {
const res = await doFetch(url, { method: 'GET', headers: { accept: 'application/vnd.ipld.car' }, timeout: timeoutMs })
if (!res || res.ok === false) return null
const buf = await res.arrayBuffer()
const carBytes = new Uint8Array(buf)
if (!carBytes.length) return null
// Belt and braces: if a membrane ever hands us a partial CAR unasked, say so rather than letting
// parseRepo report it as a malformed repo. The two are not the same problem and the fix differs.
const partial = res.headers && typeof res.headers.get === 'function'
? res.headers.get('atproto-repo-partial') === 'true'
: false
return { carBytes, identity, partial }
} catch {
return null
}
}
}
// Head-only: the cheap staleness check. { rev, cid, active, status } or null.
export function createFetchHead ({ doFetch, resolveIdentity, timeoutMs = 20_000 } = {}) {
if (typeof doFetch !== 'function') throw new Error('createFetchHead requires doFetch')
if (typeof resolveIdentity !== 'function') throw new Error('createFetchHead requires resolveIdentity')
return async function fetchHead (did) {
const identity = await resolveIdentity(did)
if (identity === null) return null
const get = async (nsid, q) => {
try {
const res = await doFetch(`${identity.pds}/xrpc/${nsid}?${q}`, { method: 'GET', headers: { accept: 'application/json' }, timeout: timeoutMs })
if (!res || res.ok === false) return null
return await res.json()
} catch { return null }
}
const q = `did=${encodeURIComponent(did)}`
const [commit, status] = await Promise.all([
get('com.atproto.sync.getLatestCommit', q),
get('com.atproto.sync.getRepoStatus', q)
])
if (!commit) return null
return {
pds: identity.pds,
rev: commit.rev || null,
cid: commit.cid || null,
active: status ? status.active !== false : null,
status: status ? (status.status || null) : null
}
}
}
// ---- parse + verify ---------------------------------------------------------------------------------
// Decode a repo CAR and prove it before a single byte is believed. Returns null when any proof fails —
// an unverifiable repo is not a repo, and silently degrading to "index what we got" is how a hostile or
// broken host would get to rewrite our copy.
export async function parseRepo ({ did, carBytes, pubkey, partial = false }) {
// A PARTIAL CAR is not a broken one. Since atproto-bridge 0.9.45 a membrane can serve a genuine diff,
// and its MST is incomplete BY DESIGN — so verifyRepo below would fail on a missing block and report
// `malformed-repo`, sending whoever reads it hunting for corruption that is not there. Named as its own
// refusal instead. We should never receive one (createFetchRepo does not ask), which is exactly why it
// is worth catching: it would mean a membrane returned something nobody requested.
if (partial) return { ok: false, reason: 'partial-car', rev: null }
let car, commit
try {
car = await readCar(carBytes)
commit = loadCommit(car)
} catch {
return null
}
const value = commit && commit.value
if (!value) return null
// 1. the commit is for the repo we asked about
if (value.did !== did) return { ok: false, reason: 'did-mismatch', rev: value.rev || null }
// 2. signed by the key from the DID DOCUMENT
if (pubkey) {
let sigOk = false
try { sigOk = verifyCommit(value, pubkey) } catch { sigOk = false }
if (!sigOk) return { ok: false, reason: 'bad-signature', rev: value.rev || null }
}
// 3. every block re-encodes to its own CID (the wholesale form of repair-bytes' round-trip guard)
let fidelity
try { fidelity = await verifyRepo(car) } catch (e) { return { ok: false, reason: 'malformed-repo', error: (e && e.message) || String(e) } }
if (!fidelity.ok) return { ok: false, reason: 'block-cid-mismatch', mismatches: fidelity.mismatches.length }
const records = []
for (const r of listRecords(car)) {
const parts = splitKey(r.key)
if (!parts) continue
let bytes = null
try { bytes = getBlock(car, r.cid).bytes } catch { bytes = null }
if (!bytes) continue
records.push({ collection: parts.collection, rkey: parts.rkey, cid: String(r.cid), bytes, record: r.value })
}
return { ok: true, rev: value.rev || null, records, blocks: fidelity.blocks }
}
// ---- diff -------------------------------------------------------------------------------------------
// listRefs(did) -> async iterable { rkey, collection }; hasBytes(did, collection, rkey) -> boolean
//
// ⛔ COMPARE LIKE WITH LIKE. `listRefs` walks the appview's POST index, which by construction holds only
// POST_COLLECTIONS — likes, follows, reposts and the profile live in separate views it never yields. A
// first cut compared the FULL remote record set against that post-only index and so reported every like
// and follow as "missing from our index", permanently: applying them made no difference, because they
// were never going to appear in `_ref`. 19 of ashatron's 27 were this, and a diff that cries wolf on
// every run is worse than none.
//
// `tracksCollection` names what the local index can actually answer for. Everything else is counted and
// reported separately rather than silently dropped — "we cannot verify these here" is a real answer, and
// pretending they are missing or pretending they are fine are both worse.
// tracked Map<collection, { bytes: boolean }> — the views that can answer, and whether this node
// stores that collection's bytes at all (likes/follows/reposts/blocks: it does not)
// listRefsFor(did, collection) -> async iterable { rkey } from THAT collection's own view
// isIgnored(collection) -> deliberately not indexed, with a reason on file (spec §2.1)
export async function diffRepo ({ did, remote, tracked, listRefsFor, hasBytes, isIgnored = () => false }) {
const trackedMap = tracked instanceof Map ? tracked : new Map(Object.entries(tracked || {}))
const missingFromIndex = []
const indexedWithoutBytes = []
const extraLocal = []
const perCollection = {}
const ignoredByCollection = {}
const unknownByCollection = {}
// Group the authority's records by collection, and classify every one of them. A collection that is
// neither tracked nor on the ignore list is UNKNOWN — a lexicon we have never classified, which
// used to vanish through applyCommit's final `continue` without a word (spec §2).
const remoteByCollection = new Map()
for (const r of remote.records) {
if (trackedMap.has(r.collection)) {
if (!remoteByCollection.has(r.collection)) remoteByCollection.set(r.collection, new Map())
remoteByCollection.get(r.collection).set(r.rkey, r)
} else if (isIgnored(r.collection)) {
ignoredByCollection[r.collection] = (ignoredByCollection[r.collection] || 0) + 1
} else {
unknownByCollection[r.collection] = (unknownByCollection[r.collection] || 0) + 1
}
}
// Compare EVERY tracked collection, including ones the authority has none of — a collection we hold
// records for and the repo does not is exactly the extraLocal case, and skipping empty remotes would
// hide it.
for (const [collection, meta] of trackedMap) {
const remoteRk = remoteByCollection.get(collection) || new Map()
const localRk = new Set()
for await (const ref of listRefsFor(did, collection)) {
if (!ref || !ref.rkey) continue
localRk.add(ref.rkey)
if (!remoteRk.has(ref.rkey)) { extraLocal.push({ collection, rkey: ref.rkey }); continue }
// Only ask about bytes where this node is supposed to have them.
if (meta && meta.bytes && !(await hasBytes(did, collection, ref.rkey))) {
indexedWithoutBytes.push({ collection, rkey: ref.rkey })
}
}
for (const rkey of remoteRk.keys()) if (!localRk.has(rkey)) missingFromIndex.push({ collection, rkey })
if (remoteRk.size || localRk.size) perCollection[collection] = { authority: remoteRk.size, indexed: localRk.size }
}
const count = (o) => Object.values(o).reduce((a, b) => a + b, 0)
return {
did,
remoteRev: remote.rev,
remoteRecords: remote.records.length,
perCollection,
ignored: count(ignoredByCollection),
ignoredByCollection,
unknown: count(unknownByCollection),
unknownByCollection,
missingFromIndex,
indexedWithoutBytes,
extraLocal,
inSync: missingFromIndex.length === 0 && indexedWithoutBytes.length === 0 && extraLocal.length === 0
}
}
// ---- apply ------------------------------------------------------------------------------------------
// applyOps(did, ops) -> index + author bytes through the appview's own gates (unindexed records)
// putBytes(did, collection, rkey, record) -> write bytes ONLY (already-indexed records, where the
// appview's dedup gate would return before the write is retried)
//
// EXTRA-LOCAL RECORDS ARE NEVER DELETED HERE. A record we hold that the repo does not may be a genuine
// delete we failed to fold, or it may be a record the authority has not yet published — and this cannot
// tell the two apart. Deleting on that ambiguity is how the appview's straggler removal is suspected of
// having pruned the very records this exists to restore. Reported, never actioned.
export async function applyBackfill ({ did, remote, diff, applyOps, putBytes, write = false, limit = 0, onEvent = null }) {
const emit = (ev, d) => { if (onEvent) onEvent(ev, d) }
const byKey = new Map(remote.records.map((r) => [r.collection + '/' + r.rkey, r]))
const report = { indexed: 0, bytesWritten: 0, reencodeRefused: 0, errors: 0, skippedExtraLocal: diff.extraLocal.length }
let done = 0
const guarded = (r) => {
// Same guard as repair-bytes: the store re-encodes a decoded record, and what the commit signed was
// the ORIGINAL bytes. If the round trip is not byte-identical, writing would persist bytes that no
// longer match their CID.
let value, re
try { value = dagCbor.decode(r.bytes) } catch { return null }
try { re = dagCbor.encode(value) } catch { return null }
if (!b4a.equals(b4a.from(re), b4a.from(r.bytes))) return null
return value
}
for (const item of diff.missingFromIndex) {
if (limit && done >= limit) return report
const r = byKey.get(item.collection + '/' + item.rkey)
if (!r) continue
const value = guarded(r)
if (value == null) { report.reencodeRefused++; emit('reencode-refused', item); continue }
done++
if (!write) continue
try {
await applyOps(did, [{ action: 'create', collection: r.collection, rkey: r.rkey, record: value }])
report.indexed++
emit('indexed', item)
} catch (e) { report.errors++; emit('index-failed', { ...item, error: (e && e.message) || String(e) }) }
}
for (const item of diff.indexedWithoutBytes) {
if (limit && done >= limit) return report
const r = byKey.get(item.collection + '/' + item.rkey)
if (!r) continue
const value = guarded(r)
if (value == null) { report.reencodeRefused++; emit('reencode-refused', item); continue }
done++
if (!write) continue
try {
await putBytes(did, r.collection, r.rkey, value)
report.bytesWritten++
emit('bytes-written', item)
} catch (e) { report.errors++; emit('write-failed', { ...item, error: (e && e.message) || String(e) }) }
}
return report
}