atproto-base / lib / firehose.js

// The FIREHOSE projection — turn consecutive finalized heads into atproto `#commit` frames
// (com.atproto.sync.subscribeRepos) that the bridge/relay replicate. Deterministic like the commit
// projection: a #commit frame is a pure function of (oldRecords → newRecords), so co-owners produce
// identical frames (the seq/time the sequencer stamps are firehose-server state, not signed).
//
// A #commit carries the DIFF, not the full repo: `ops` (create/update/delete), the MINIMAL MST proof
// blocks (atproto-repo firehoseProofBlocks — created + delete-exclusion + inductive-inversion nodes)
// PLUS the changed record blocks, and `prevData` (old MST root) for inductive verification. We reuse
// atproto-store's byte-exact commitFrame/decodeCommitFrame codec (what the bridge already consumes).

import { cidForValue } from 'atproto-repo/lib/cid.js'
import { firehoseProofBlocks } from 'atproto-repo/lib/mst.js'
import { unsignedCommit, signCommit, commitCid } from 'atproto-repo/lib/commit.js'

// records: [{ key, value: recordObject }] → [{ key, cid, value }]
async function withCids (records) {
  const out = []
  for (const r of records) { const { cid, bytes } = await cidForValue(r.value); out.push({ key: r.key, cid, bytes, value: r.value }) }
  return out
}

// The atproto ops diff between two record sets (by MST key).
function diffOps (oldW, newW) {
  const oldMap = new Map(oldW.map((r) => [r.key, r.cid]))
  const newMap = new Map(newW.map((r) => [r.key, r.cid]))
  const ops = []
  for (const { key, cid } of newW) {
    const prev = oldMap.get(key)
    if (!prev) ops.push({ action: 'create', path: key, cid, prev: null })
    else if (!prev.equals(cid)) ops.push({ action: 'update', path: key, cid, prev })
  }
  for (const { key, cid } of oldW) { if (!newMap.has(key)) ops.push({ action: 'delete', path: key, cid: null, prev: cid }) }
  return ops
}

// Build the `write` object atproto-store's commitFrame() wants, for the transition oldRecords→newRecords.
// Deterministic. `prev` = previous commit CID (null = genesis), `since` = previous rev (relay gap hint).
export async function buildCommitWrite ({ oldRecords = [], newRecords, did, rev, since = null, prev = null, privateKey }) {
  const oldW = await withCids(oldRecords)
  const newW = await withCids(newRecords)
  const ops = diffOps(oldW, newW)

  // Minimal MST proof blocks + the new/old roots (firehoseProofBlocks throws if the proof is invalid).
  const { blocks, dataCid, oldRoot } = firehoseProofBlocks({
    oldRecords: oldW.map((r) => ({ key: r.key, value: r.cid })),
    newRecords: newW.map((r) => ({ key: r.key, value: r.cid })),
    ops
  })

  // The changed RECORD blocks must also travel (created/updated record content).
  const changed = new Map(blocks) // start from the MST proof block set
  for (const op of ops) {
    if (op.action === 'delete') continue
    const rec = newW.find((r) => r.key === op.path)
    changed.set(op.cid.toString(), { cid: op.cid, bytes: rec.bytes })
  }

  const signed = signCommit(unsignedCommit({ did, rev, data: dataCid, prev }), privateKey)
  const cCid = commitCid(signed)
  const { bytes: commitBytes } = await cidForValue(signed)

  return {
    commitCid: cCid,
    commitBlock: { cid: cCid, bytes: commitBytes },
    newBlocks: [...changed.values()],
    ops,
    rev,
    since,
    signedCommit: signed,
    prevData: oldRoot
  }
}