atproto-store / lib / repo-manager.js

// RepoManager — the WRITE path + commit head-state over a RepoStore.
//
// Holds the current commit head { rev, commitCid, dataCid }. Each applyWrite mutates
// the store, rebuilds the MST, mints a new monotonic rev, signs a new commit, and
// returns the diff needed to emit a firehose #commit (see firehose.js).
//
// `newBlocks` is the MINIMAL Sync-v1.1 covering proof (slice 3b): the MST-inversion
// proof set (`created | deletion_proof | inductive_proof_nodes`) plus the new record —
// NOT the whole tree (which would trip a relay's `tooBig`). See firehoseProofBlocks
// in repo/lib/mst.js. Prior record CIDs travel in each op's `prev` so a cold relay can
// invert with no old-tree state.

import * as dagCbor from '@ipld/dag-cbor'
import { buildMst, firehoseProofBlocks } from 'atproto-repo/lib/mst.js'
import { unsignedCommit, signCommit, commitCid } from 'atproto-repo/lib/commit.js'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { tidNow } from './tid.js'

export class RepoManager {
  constructor (store, { did, privateKey, clock } = {}) {
    this.store = store
    this.did = did
    this.privateKey = privateKey
    this._clock = clock // optional injectable micros clock for tests
    this.rev = null
    this.commitCid = null
    this.dataCid = null
  }

  _nextRev () { return this._clock ? tidNow(0, this._clock) : tidNow() }

  // Build the current head. On restart, RESTORE the persisted rev (recomputing the commit
  // from records is deterministic, so the head is byte-stable); only mint a fresh rev for a
  // genuinely new repo (no saved head). Re-minting on every open forged a "phantom commit"
  // with a new rev/CID on each restart — see docs/ARCHITECTURE-AUDIT.md.
  async init () {
    const saved = await this.store.loadHead()
    const rev = saved ? saved.rev : this._nextRev()
    const { commitCid: cc, dataCid } = await this._commitCurrent(rev)
    this.rev = rev; this.commitCid = cc; this.dataCid = dataCid
    if (!saved) await this.store.saveHead({ rev, commitCid: cc.toString(), dataCid: dataCid.toString() })
    return this
  }

  async _commitCurrent (rev) {
    const recs = await this.store.records()
    const mstRecords = recs.map(r => ({ key: r.key, value: cidForBytesSync(r.bytes) }))
    const { root: dataCid, ns } = buildMst(mstRecords)
    const signed = signCommit(unsignedCommit({ did: this.did, rev, data: dataCid, prev: null }), this.privateKey)
    return { signed, commitCid: commitCid(signed), dataCid, ns }
  }

  // Apply one write and advance the head. Returns everything firehose.js needs.
  //   op = { action: 'create'|'update'|'delete', collection, rkey, record? }
  async applyWrite ({ action, collection, rkey, record }) {
    const prevRev = this.rev
    const prevData = this.dataCid
    const path = `${collection}/${rkey}`

    // Capture the OLD record set + the prior value at this path — both are inputs to
    // the MST-inversion proof (the prior CID lets a cold relay invert the commit).
    const oldRecs = await this.store.records()
    const existing = await this.store.getRaw(collection, rkey)
    const priorCid = existing ? cidForBytesSync(existing.bytes) : null

    let opCid = null
    if (action === 'create' || action === 'update') {
      await this.store.put(collection, rkey, record)
      const raw = await this.store.getRaw(collection, rkey)
      opCid = cidForBytesSync(raw.bytes)
    } else if (action === 'delete') {
      await this.store.del(collection, rkey)
    } else {
      throw new Error(`invalid action: ${action}`)
    }

    const rev = this._nextRev()
    const { signed, commitCid: cc, dataCid } = await this._commitCurrent(rev)

    // Minimal Sync-v1.1 covering proof (created | deletion_proof | inductive_proof).
    const newRecs = await this.store.records()
    const toMst = (recs) => recs.map(r => ({ key: r.key, value: cidForBytesSync(r.bytes) }))
    const proofOp = { action, path, cid: opCid, prev: priorCid }
    const proof = firehoseProofBlocks({ oldRecords: toMst(oldRecs), newRecords: toMst(newRecs), ops: [proofOp] })

    // Invariants: the independently-rebuilt proof trees must match the signed commit.
    if (!proof.dataCid.equals(dataCid)) throw new Error('proof dataCid != signed commit data')
    if (!proof.oldRoot.equals(prevData)) throw new Error('proof oldRoot != prevData')

    const blocks = [...proof.blocks.values()]
    if (opCid) {
      const raw = await this.store.getRaw(collection, rkey)
      blocks.push({ cid: opCid, bytes: raw.bytes })
    }

    // Wire op: {action, path, cid} per the repoOp lexicon; `prev` carried for
    // update/delete so a relay can invert without holding the old tree.
    const op = { action, path, cid: opCid }
    if (action === 'update' || action === 'delete') op.prev = priorCid

    this.rev = rev; this.commitCid = cc; this.dataCid = dataCid
    await this.store.saveHead({ rev, commitCid: cc.toString(), dataCid: dataCid.toString() })

    return {
      commitCid: cc,
      commitBlock: { cid: cc, bytes: dagCbor.encode(signed) },
      signedCommit: signed,
      dataCid,
      prevData,
      rev,
      since: prevRev,
      ops: [op],
      newBlocks: blocks
    }
  }
}