// AtprotoBase — one autobase per DID whose view holds the atproto record set. The atproto commit
// chain is produced by the deterministic PROJECTION (lib/project.js) over the INDEXED view. There is
// NO signer election: signing is a pure function of the indexed view + the shared #atproto key, so it
// is indexer-agnostic (see DESIGN.md, proven by the offline suite).
//
// TYPED ops (contract-router style, like keet — NOT one `set-entry` overloaded by magic keys):
// set-record / del-record repo records (LWW by a log-derived counter)
// admit content-free admission doorbell (proof-gated, block-until-admitted)
// set-members the roster projection (admin/bootstrap only)
// start-indexing / move-indexer indexer topology (next slice)
// ban / set-epoch revocation (next slice)
//
// GOVERNANCE (this slice): proof-gated multi-writer admission + roster projection. A joiner proves
// membership (sign localKey‖base.key with its identity key) in a CONTENT-FREE optimistic doorbell
// block; apply verifies the proof resolves to a member role and addWriters it. The joiner BLOCKS
// until genuinely admitted (the golden rule — never fuse content into a pre-admission optimistic
// block, which autobase silently rolls back). Indexer topology + revocation are the next slices.
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hyperbee from 'hyperbee'
import Hypercore from 'hypercore'
import b4a from 'b4a'
import crypto from 'hypercore-crypto'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { getRepoFromCommit } from 'atproto-store/lib/read-view.js'
import { cidForValue } from 'atproto-repo/lib/cid.js'
import { readCar, loadCommit, listRecords } from 'atproto-repo'
import { projectCommit, nextRev } from './project.js'
import { buildCommitWrite } from './firehose.js'
import { revTime } from './rev.js'
// Typed ops. start-indexing/stop-indexing mirror keet's members/index.js handlers I/P EXACTLY: a
// device SELF-promotes/demotes its own writer, gated on its member being an admin (role 1 = keet
// canAdmin). NO preferFastForward, NO length-0 gate (keet promotes ACTIVE data-bearing writers —
// verified by reading keet-core 4.16 members/index.js in full). This is NOT StateBaseV3's move-indexer,
// which was a MISREAD (it demoted the creator cross-member; keet's move-indexer is a same-member
// device→device handoff — deliberately NOT ported).
const OP = { SET_RECORD: 'set-record', DEL_RECORD: 'del-record', ADMIT: 'admit', SET_MEMBERS: 'set-members', START_INDEXING: 'start-indexing', STOP_INDEXING: 'stop-indexing', BAN: 'ban' }
const MEMBERS = '__members__' // reserved view key: { identityHex: role }
const DEVICES = '__devices__' // reserved view key: { writerKeyHex: identityHex }
const REMOVED = '__removed__' // reserved view key: { identityHex: { ts } } — banned; blocks re-admission
// NOTE: ban() removeWriters a device from THIS autobase (can't write the shared view) + blocks re-admit.
// It does NOT revoke the SHARED #atproto signing key at the FEDERATION layer — a lost device still holds
// that key until a PLC rotation (a separate, unbuilt ceremony; architecture-holes #6). Document loudly.
const RKEY = (collection, rkey) => `${collection}/${rkey}`
const toBuf = (k) => (Buffer.isBuffer(k) ? k : Buffer.from(String(k), 'hex'))
// DETERMINISTIC base key (ported byte-exact from atproto-room/StateBaseV3): every co-owner that holds
// the members-only `roomSeed` derives the SAME autobase base key, so they open the SAME base with NO
// shared bootstrap. The CREATOR (lowest-key admin, or forceCreator) opens with `gsKeyPair` as its local
// writer — the deterministic genesis/bootstrap indexer; other co-owners open with a fresh writer and
// join via admission. gsNamespace makes the key roomSeed-space-specific (override for migration).
const GS_NAMESPACE = Buffer.from('atproto-base/state/v1', 'utf-8')
const ENC_NAMESPACE = Buffer.from('atproto-base/enc/v1', 'utf-8') // members-only encryption-key namespace
function deriveGsKeyPair (roomSeed, ns = GS_NAMESPACE) { return crypto.keyPair(crypto.hash([ns, toBuf(roomSeed)])) }
function deriveBaseKey (gsPublicKey) { return Hypercore.key({ version: 1, signers: [{ publicKey: gsPublicKey }] }) }
// The members-only base encryption key, derived from the roomSeed (mirrors device-node.js enc derivation:
// crypto.hash([NS('enc'), sharedKey])). Holding the roomSeed gives BOTH the base key AND the encryption
// key — a non-member has neither, so they can neither locate nor decrypt the co-owned base.
function deriveEncKey (roomSeed) { return crypto.hash([ENC_NAMESPACE, toBuf(roomSeed)]) }
// A minimal RepoStore-compatible projection (put/del/records) over atproto-base's OWN corestore. We
// don't reuse atproto-store's RepoStore because its `storage instanceof Corestore` guard fails across
// duplicated corestore module copies (atproto-store is a symlinked file: dep with its own node_modules).
// getRepoFromCommit only reads records() -> [{key, bytes}]; record bytes are canonical DAG-CBOR
// (cidForValue), so CIDs match atproto-repo's regardless of which module copy computed them.
class RepoProjection {
constructor (core) { this.core = core; this.bee = new Hyperbee(core, { keyEncoding: 'utf-8', valueEncoding: 'binary', alwaysDuplicate: false }) }
async ready () { await this.core.ready(); await this.bee.ready(); return this }
async put (collection, rkey, record) { const { bytes } = await cidForValue(record); await this.bee.put(`${collection}/${rkey}`, b4a.from(bytes)) }
async del (collection, rkey) { await this.bee.del(`${collection}/${rkey}`) }
async records () { const out = []; for await (const n of this.bee.createReadStream()) out.push({ key: String(n.key), bytes: n.value }); return out }
async close () { try { await this.bee.close() } catch {} }
}
export class AtprotoBase {
// opts: { path | store, bootstrap?, keyPair?, encryptionKey?, did, privateKey, identity?, members? }
// identity: { publicKey, secretKey } — this co-owner's membership key (ed25519, drives admission).
// members: [{ key: identityHex, role }] — the roster the creator projects (role 1=admin).
constructor (opts = {}) {
if (!opts.did) throw new Error('AtprotoBase requires opts.did')
if (!opts.privateKey) throw new Error('AtprotoBase requires opts.privateKey (the #atproto signing key)')
this._did = opts.did
this._privateKey = opts.privateKey
this._injectedStore = opts.store || null
this._path = opts.path || null
this._bootstrap = opts.bootstrap || null
this._keyPair = opts.keyPair || null
// Deterministic base key from a members-only roomSeed (co-owners share the base without a bootstrap).
this._roomSeed = opts.roomSeed ? toBuf(opts.roomSeed) : null
this._gsNamespace = opts.gsNamespace ? toBuf(opts.gsNamespace) : GS_NAMESPACE
this._gsKeyPair = this._roomSeed ? deriveGsKeyPair(this._roomSeed, this._gsNamespace) : null
this._detBaseKey = this._gsKeyPair ? deriveBaseKey(this._gsKeyPair.publicKey) : null
this._forceCreator = opts.forceCreator === true // assume-genesis heal (re-author on fresh storage)
this._forceNonCreator = opts.forceNonCreator === true // resync heal (join as a fresh non-creator writer)
// Members-only encryption: derive from the roomSeed unless an explicit key is given (migration/BYO).
this._encryptionKey = opts.encryptionKey ? toBuf(opts.encryptionKey) : (this._roomSeed ? deriveEncKey(this._roomSeed) : null)
this._identity = opts.identity ? { publicKey: toBuf(opts.identity.publicKey), secretKey: opts.identity.secretKey ? toBuf(opts.identity.secretKey) : null } : null
this._members = opts.members || null
// keet-attestation admission (the model hpp uses): injectable attest/verifyAttest (from
// atproto-room/lib/membership.js makeAttest/makeVerifyAttest — a device attested to the root
// identity). Contract (from StateBaseV3): attest(signData)->Buffer proof; verifyAttest(proof,
// signData, identity, receipt)->bool. When absent, admission falls back to raw ed25519 (identity
// key signs localKey‖base.key). All co-owners must be configured the SAME way.
this._attest = typeof opts.attest === 'function' ? opts.attest : null
this._verifyAttest = typeof opts.verifyAttest === 'function' ? opts.verifyAttest : null
this._ackInterval = opts.ackInterval || 1000
this._admitted = new Set()
this._firehose = null // FirehoseSequencer (Hypercore of #commit frames)
this._repo = null // RepoStore projection (getRepo for the bridge)
this._prevHead = { records: [], rev: null, commitCid: null, signedCommit: null } // last EMITTED head
this.store = null
this.base = null
this._ready = false
}
async ready () {
if (this._ready) return
this.store = this._injectedStore ? this._injectedStore.namespace(this._did) : new Corestore(this._path)
await this.store.ready()
const baseOpts = {
open: (s) => new Hyperbee(s.get('view'), { keyEncoding: 'utf-8', valueEncoding: 'json', extension: false }),
apply: this._apply.bind(this),
close: async (view) => { if (view) await view.close() },
valueEncoding: 'binary',
optimistic: true,
ackInterval: this._ackInterval,
ackThreshold: 4
}
// Resolve bootstrap + local writer keypair. The deterministic-base-key path (roomSeed) OVERRIDES an
// explicit bootstrap/keyPair: every co-owner opens the same base; the CREATOR (lowest-key admin, or
// forceCreator) writes as gsKeyPair (the bootstrap indexer), others open a fresh writer.
let bootstrap = this._bootstrap
let keyPair = this._keyPair
if (this._detBaseKey) {
const amCreator = this._forceCreator ? true : (this._forceNonCreator ? false : this._amCreator())
bootstrap = this._detBaseKey
keyPair = amCreator ? this._gsKeyPair : null
}
if (keyPair) baseOpts.keyPair = keyPair
if (this._encryptionKey) { baseOpts.encryptionKey = this._encryptionKey; baseOpts.encrypted = true }
this.base = new Autobase(this.store, bootstrap, baseOpts)
await this.base.ready()
// The public firehose: a plain Hypercore of #commit/#sync frames the bridge replicates + serves as
// com.atproto.sync.subscribeRepos. Separate from the autobase (a projection of it), like the membrane.
this._firehose = await new FirehoseSequencer(this.store.get({ name: 'atproto-firehose' })).ready()
// The public repo core the bridge's getRepo assembles from (a RepoStore kept in sync with the
// finalized record set on each emitFirehose, so its MST root matches the emitted signed commit).
this._repo = await new RepoProjection(this.store.get({ name: 'atproto-repo-projection' })).ready()
this._ready = true
if (this._members) this.projectMembers(this._members).catch(() => {}) // creator best-effort projection
}
// Deterministic apply — TYPED ops fold into the record view + governance. Pure: no wall-clock.
async _apply (nodes, view, host) {
for (const node of nodes) {
if (!node.value) continue
let op
try { op = JSON.parse(b4a.toString(node.value)) } catch { continue }
if (op.t === OP.ADMIT) {
const fromKey = node.from && node.from.key
if (!fromKey) continue
const fromHex = fromKey.toString('hex')
if (this._admitted.has(fromHex)) continue // idempotent — writer already admitted
const role = await this._verifyAdmission(fromKey, op.admit, view)
if (role == null) continue // proof invalid / not a member → NOT acked, block is discarded
try {
await host.addWriter(fromKey, { indexer: false }) // plain writer (N=1 posture — no auto-indexer)
this._admitted.add(fromHex)
await host.ackWriter(fromKey)
} catch {}
const drec = await view.get(DEVICES); const dev = (drec && drec.value) || {}
if (op.admit && op.admit.identity && dev[fromHex] !== op.admit.identity) { dev[fromHex] = op.admit.identity; await view.put(DEVICES, dev) }
continue
}
if (op.t === OP.SET_MEMBERS) {
const fromHex = node.from && node.from.key ? node.from.key.toString('hex') : null
const mrec = await view.get(MEMBERS); const members = (mrec && mrec.value) || {}
const drec = await view.get(DEVICES); const dev = (drec && drec.value) || {}
const authorRole = fromHex && dev[fromHex] != null ? members[dev[fromHex]] : null
const bootstrap = Object.keys(members).length === 0
if (!bootstrap && Number(authorRole) !== 1) continue // only admin (or bootstrap) may set the roster
await view.put(MEMBERS, op.members || {})
if (op.devices) await view.put(DEVICES, { ...dev, ...op.devices }) // author's own device mapping
continue
}
if (op.t === OP.START_INDEXING) {
// keet members/index.js `I`: a device self-promotes to indexer iff its member is an admin.
const fromKey = node.from && node.from.key
if (!fromKey) continue
if (await this._writerRole(view, fromKey.toString('hex')) !== 1) continue // canAdmin only
try { await host.addWriter(fromKey, { indexer: true }); await host.ackWriter(fromKey) } catch {}
continue
}
if (op.t === OP.STOP_INDEXING) {
// keet members/index.js `P`: self-demote (admin-gated). We add a removeable() guard so we never
// demote the last indexer and strand finalization (keet leaves this to policy).
const fromKey = node.from && node.from.key
if (!fromKey) continue
if (await this._writerRole(view, fromKey.toString('hex')) !== 1) continue
try { if (host.removeable(fromKey)) await host.addWriter(fromKey, { indexer: false }) } catch {}
continue
}
if (op.t === OP.BAN) {
// keet members/index.js `W` (remove-member): a moderator+ deactivates a member — mark it removed
// (blocks re-admission), drop from the roster, and removeWriter ALL its devices (removeable-gated).
// An admin can only be removed by ITSELF (keet SELF_ONLY).
const fromKey = node.from && node.from.key
const target = String(op.target || '')
if (!fromKey || !target) continue
const fromHex = fromKey.toString('hex')
const callerRole = await this._writerRole(view, fromHex)
if (callerRole == null || callerRole > 2) continue // must be admin(1) or moderator(2)
const drec = await view.get(DEVICES); const dev = (drec && drec.value) || {}
const mrec = await view.get(MEMBERS); const members = (mrec && mrec.value) || {}
const isSelf = dev[fromHex] === target
if (Number(members[target]) === 1 && !isSelf) continue // can't ban another admin (keet SELF_ONLY)
const rrec = await view.get(REMOVED); const removed = (rrec && rrec.value) || {}
removed[target] = { ts: op.ts || 0 }; await view.put(REMOVED, removed)
if (members[target] != null) { delete members[target]; await view.put(MEMBERS, members) }
for (const [writerHex, idHex] of Object.entries(dev)) {
if (idHex !== target) continue
try { const wk = b4a.from(writerHex, 'hex'); if (host.removeable(wk)) await host.removeWriter(wk) } catch {}
this._admitted.delete(writerHex)
}
continue
}
if (op.t === OP.SET_RECORD) {
const existing = await view.get(op.key)
if (existing && existing.value && typeof existing.value.w === 'number' && existing.value.w > op.w) continue // LWW
await view.put(op.key, { record: op.record, w: op.w })
} else if (op.t === OP.DEL_RECORD) {
const existing = await view.get(op.key)
if (existing && existing.value && typeof existing.value.w === 'number' && existing.value.w > op.w) continue
await view.del(op.key)
}
}
}
get writable () { return this.base ? this.base.writable : false }
// ── Governance: admission (golden rule = admit FIRST, write AFTER) ──────────────────────────────
async _buildAdmissionProof () {
if (!this._identity || !this.base || !this.base.local || !this.base.local.key || !this.base.key) return null
const signData = Buffer.concat([this.base.local.key, this.base.key]) // bound to THIS base — not replayable
if (this._attest) { // keet-attestation: a device attested to the member identity signs signData
const proof = await this._attest(signData)
if (!proof) return null
return { identity: this._identity.publicKey.toString('hex'), proof: (Buffer.isBuffer(proof) ? proof : Buffer.from(proof)).toString('hex') }
}
if (!this._identity.secretKey) return null // raw path needs the identity secret key
const sig = crypto.sign(signData, this._identity.secretKey)
return { identity: this._identity.publicKey.toString('hex'), proof: sig.toString('hex') }
}
async _verifyAdmission (writerKey, admit, view) {
if (!admit || !admit.identity || !admit.proof) return null
const signData = Buffer.concat([writerKey, this.base.key])
const identity = Buffer.from(admit.identity, 'hex')
const proof = Buffer.from(admit.proof, 'hex')
// Verify via the injected keet-attestation verifier if configured, else raw ed25519. (receipt/epoch
// = null for now — the SET_EPOCH mass-revocation is deferred, EXTRACTION-PLAN note.)
const ok = this._verifyAttest ? await this._verifyAttest(proof, signData, identity, null) : crypto.verify(signData, proof, identity)
if (!ok) return null
const rrec = await view.get(REMOVED); const removed = (rrec && rrec.value) || {}
if (removed[admit.identity]) return null // banned identity cannot (re)admit even with a valid proof
const mrec = await view.get(MEMBERS)
const members = (mrec && mrec.value) || {}
const role = members[admit.identity]
return role != null ? Number(role) : null
}
// Role of the member owning a given writer key (via __devices__ → __members__), for in-apply
// authorization of self-promote/demote (keet's memberFor(writer.key).status.canAdmin).
async _writerRole (view, writerHex) {
const drec = await view.get(DEVICES); const dev = (drec && drec.value) || {}
const idHex = dev[writerHex]
if (idHex == null) return null
const mrec = await view.get(MEMBERS); const members = (mrec && mrec.value) || {}
const role = members[idHex]
return role != null ? Number(role) : null
}
// Deterministic creator election (ported from StateBaseV3 _amCreator): the LOWEST-key eligible
// (admin/moderator) member is the bootstrap creator; hex-ordered so every co-owner agrees offline. No
// members/identity → default to creator (single-owner DID). Used only on the deterministic-base-key path.
_amCreator () {
if (!this._identity || !Array.isArray(this._members) || this._members.length === 0) return true
const selfHex = this._identity.publicKey.toString('hex').toLowerCase()
const eligible = this._members
.map((m) => ({ key: toBuf(m.key).toString('hex').toLowerCase(), role: Number(m.role) }))
.filter((m) => m.key && (m.role === 1 || m.role === 2))
.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
if (eligible.length === 0) return true
return eligible[0].key === selfHex
}
// The creator/admin projects the roster into the shared view (memberKey→role) + its own device
// mapping (writerKey→identity), so admission can resolve a joiner's role deterministically in apply.
async projectMembers (members) {
if (members) this._members = members
if (!this.base || !this.base.writable) return false
const map = {}
for (const m of (this._members || [])) { const id = toBuf(m.key).toString('hex'); map[id] = Number(m.role) || 3 }
const devices = {}
if (this._identity && this.base.local && this.base.local.key) devices[this.base.local.key.toString('hex')] = this._identity.publicKey.toString('hex')
await this.base.append(b4a.from(JSON.stringify({ t: OP.SET_MEMBERS, members: map, devices })))
await this.base.update()
return true
}
// ── Indexer topology (keet-aligned: self-promote, canAdmin-gated) ────────────────────────────────
_isLocalIndexer () { try { return this.base.isIndexer === true } catch { return false } }
// Self-promote this device to an indexer (keet `@keet/start-indexing`). MUST already be an admitted
// writer (keet: an active device) — ensureWritable() first — and this member must be an admin (role 1)
// in the roster. A current indexer must be online to apply the promotion. Grows the indexer set toward
// quorum (N=1→N=3; keep it OFF the 2-rest by granting admin to the 2nd+3rd co-owners together, exactly
// as keet gates it via canAdmin). Data-bearing writers ARE promotable (keet does this routinely).
async requestIndexer ({ timeoutMs = 20000 } = {}) {
await this.ensureWritable()
if (this._isLocalIndexer()) return true
await this.base.append(b4a.from(JSON.stringify({ t: OP.START_INDEXING }))).catch(() => {})
await this._updateBounded()
const deadline = Date.now() + timeoutMs
while (!this._isLocalIndexer() && Date.now() < deadline) { await new Promise((r) => setTimeout(r, 150)); await this._updateBounded() }
return this._isLocalIndexer()
}
// Self-demote (keet `@keet/stop-indexing`); admin-gated + removeable-guarded in apply so it never
// strands the last indexer.
async stopIndexing () {
if (!this.base || !this.base.writable) return
await this.base.append(b4a.from(JSON.stringify({ t: OP.STOP_INDEXING })))
await this.base.update()
}
// Ban a member by identity (keet remove-member): removeWriter its devices from THIS autobase + block
// re-admission. Caller must be admin/moderator (enforced in apply). ⚠ Does NOT revoke the shared
// #atproto key at bsky — that needs a PLC rotation (separate ceremony). ts stays fixed (op-carried,
// applied identically on every co-owner) so governance state converges deterministically.
async ban (targetIdentityHex) {
await this.ensureWritable()
await this.base.append(b4a.from(JSON.stringify({ t: OP.BAN, target: String(targetIdentityHex), ts: 0 })))
await this.base.update()
}
// Block until this co-owner is a genuinely ADMITTED writer, then return. No-op if already writable.
// Rings the CONTENT-FREE admission doorbell and waits; THROWS if no indexer admits us within the
// timeout (never a silent content loss). This is what makes putRecord/delRecord durable.
async ensureWritable ({ timeoutMs = 20000 } = {}) {
if (this.base && this.base.writable) return
await this._requestWriter({ timeoutMs })
if (!this.base.writable) throw new Error(`AtprotoBase: not admitted (no indexer online within ${timeoutMs}ms) — refusing to write content that would be silently rolled back`)
}
async _requestWriter ({ timeoutMs = 15000 } = {}) {
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('AtprotoBase: no identity to build an admission proof')
const ring = () => this.base.append(b4a.from(JSON.stringify({ t: OP.ADMIT, admit })), { optimistic: true }).catch(() => {})
await ring(); await this._updateBounded()
const deadline = Date.now() + timeoutMs
while (!this.base.writable && Date.now() < deadline) {
await new Promise((r) => setTimeout(r, 200))
if (!this.base.writable) await ring() // re-ring: keep a live optimistic block until an indexer acks
await this._updateBounded()
}
}
// base.update() awaits a full drain which can hang on an un-converged replica; bound the await so the
// doorbell loop keeps re-ringing while replication proceeds in the background (autobase drains on wakeup).
async _updateBounded (ms = 800) {
if (!this.base) return
let t = null
try { await Promise.race([this.base.update().catch(() => {}), new Promise((r) => { t = setTimeout(r, ms); if (t && t.unref) t.unref() })]) } finally { if (t) clearTimeout(t) }
}
// ── Records (golden rule: admit FIRST via ensureWritable, then write non-optimistically) ────────
async putRecord (collection, rkey, record) {
await this.ensureWritable()
const w = this.base.local ? this.base.local.length : 0
await this.base.append(b4a.from(JSON.stringify({ t: OP.SET_RECORD, key: RKEY(collection, rkey), record, w })))
await this.base.update()
}
async delRecord (collection, rkey) {
await this.ensureWritable()
const w = this.base.local ? this.base.local.length : 0
await this.base.append(b4a.from(JSON.stringify({ t: OP.DEL_RECORD, key: RKEY(collection, rkey), w })))
await this.base.update()
}
// The FINALIZED (durably multisig-signed) atproto record set — the input to the projection. Reads the
// view Hyperbee CHECKED OUT at the VIEW CORE's signedLength, so an un-finalized optimistic tip (a
// lagging co-owner's writes, or writes during an N=2 stall) NEVER leaks into the signed atproto chain
// (engine-level reorg-safety; see DESIGN §hard-requirements). NB: base.view.core.signedLength — NOT
// base.signedLength, which is the SYSTEM/oplog core, a different core (verified against autobase 7.28.1).
async indexedRecords () {
await this.base.update()
const signed = (this.base.view && this.base.view.core) ? this.base.view.core.signedLength : 0
if (!signed) return [] // nothing finalized yet (also avoids Hyperbee checkout(0) → full-view footgun)
const co = this.base.view.checkout(signed)
const out = []
try {
for await (const { key, value } of co.createReadStream()) {
if (String(key).startsWith('__')) continue // reserved governance keys
out.push({ key: String(key), value: value.record })
}
} finally { try { await co.close() } catch {} }
return out
}
// Drive finalization until the view core's signedLength stabilizes — so a caller can await the point
// where recent writes are durably finalized (and thus reflected in indexedRecords()/head()). The
// atproto chain intentionally advances on FINALIZATION, not on write (durability), so this lag is
// by-design. Bounded; returns once signedLength holds steady (or the timeout hits, e.g. an N=2 stall).
async settled ({ timeoutMs = 5000 } = {}) {
const deadline = Date.now() + timeoutMs
let last = -1; let stable = 0
while (Date.now() < deadline) {
try { await this.base.update() } catch {}
const s = (this.base.view && this.base.view.core) ? this.base.view.core.signedLength : 0
if (s === last) { if (++stable >= 3) return } else { stable = 0; last = s }
await new Promise((r) => setTimeout(r, 40))
}
}
async head ({ prevRev = null, prev = null } = {}) {
const records = await this.indexedRecords()
const rev = nextRev(prevRev)
return projectCommit({ records, did: this._did, rev, prev, privateKey: this._privateKey })
}
// ── Firehose (com.atproto.sync.subscribeRepos) ──────────────────────────────────────────────────
// Emit a #commit frame for the current FINALIZED head if it advanced since the last emit. Deterministic
// (buildCommitWrite is pure; the seq/time the sequencer stamps are firehose-server state). Returns the
// write, or null if the finalized set is unchanged. Call after finalization (the chain advances on
// finalization, not on write) — e.g. wire it to base 'update' in production.
async emitFirehose () {
await this.settled()
const records = await this.indexedRecords()
const rev = nextRev(this._prevHead.rev)
const write = await buildCommitWrite({
oldRecords: this._prevHead.records, newRecords: records,
did: this._did, rev, since: this._prevHead.rev, prev: this._prevHead.commitCid, privateKey: this._privateKey
})
if (write.ops.length === 0 && this._prevHead.commitCid) return null // nothing finalized-new to emit
await this._projectRepo(records) // keep the repo core in sync FIRST so getRepo() matches this commit
await this._firehose.sequenceCommit(write, { time: revTime(rev) })
this._prevHead = { records, rev, commitCid: write.commitCid, signedCommit: write.signedCommit }
return write
}
// Sync the RepoStore projection to the finalized record set (put changed, delete removed) so its MST
// root equals the emitted commit's data CID (getRepoFromCommit asserts this).
async _projectRepo (records) {
const want = new Set(records.map((r) => r.key))
for (const r of records) { const i = r.key.lastIndexOf('/'); await this._repo.put(r.key.slice(0, i), r.key.slice(i + 1), r.value) }
for (const e of await this._repo.records()) { if (!want.has(e.key)) { const i = e.key.lastIndexOf('/'); await this._repo.del(e.key.slice(0, i), e.key.slice(i + 1)) } }
}
// com.atproto.sync.getRepo — the signed CAR the bridge serves, assembled from the RepoStore + the last
// emitted signed commit (exactly PeerFront.getRepo's contract). Call emitFirehose() at least once first.
async getRepo () {
if (!this._prevHead.signedCommit) throw new Error('AtprotoBase.getRepo: no commit emitted yet — call emitFirehose() first')
return getRepoFromCommit(this._repo, this._prevHead.signedCommit)
}
// ── Migration / adopt (EXTRACTION-PLAN #5) ──────────────────────────────────────────────────────
// Adopt an existing head WITHOUT emitting: set the last-emitted head to a given commit + record set,
// so getRepo serves it and the NEXT emitFirehose CONTINUES the chain (prev = this commit, rev > this
// rev) instead of re-deriving from genesis. Load-bearing for migration: the relay's forward-only
// cursor rejects a rev regression (federation-wall). records = [{ key, value: record }].
async adoptHead ({ records, rev, commitCid, signedCommit }) {
await this._projectRepo(records)
this._prevHead = { records, rev, commitCid, signedCommit }
}
// Migration reader: import an existing atproto repo (a CAR — the live DID's getRepo, or a backup)
// into this base, then ADOPT its head so the chain continues unbroken. The base must be writable.
// Records continue-migrate byte-identically (DAG-CBOR is canonical), so CIDs are preserved.
async importRepo (carBytes) {
const car = await readCar(carBytes)
const commit = loadCommit(car) // { cid, value: signedCommit }
const records = []
for (const { key, value } of listRecords(car)) {
const i = key.lastIndexOf('/')
await this.putRecord(key.slice(0, i), key.slice(i + 1), value)
records.push({ key, value })
}
await this.settled()
await this.adoptHead({ records, rev: commit.value.rev, commitCid: commit.cid, signedCommit: commit.value })
return { rev: commit.value.rev, commitCid: commit.cid, records: records.length }
}
subscribeRepos (cursor = 0, opts) { return this._firehose.subscribe(cursor, opts) }
get firehoseKey () { return this._firehose ? this._firehose.core.key : null }
get firehoseLength () { return this._firehose ? this._firehose.length : 0 }
get repoKey () { return this._repo ? this._repo.core.key : null }
async close () {
if (this._firehose) { try { await this._firehose.close() } catch {} this._firehose = null }
if (this._repo) { try { await this._repo.close() } catch {} this._repo = null }
if (this.base) { try { await this.base.close() } catch {} this.base = null }
if (this.store && !this._injectedStore) { try { await this.store.close() } catch {} }
this._ready = false
}
}
export { crypto as _crypto } // convenience for callers minting identity keypairs in tests/tools