atproto-store / lib / firehose.js

// Firehose #commit frame — the atproto com.atproto.sync.subscribeRepos wire format.
//
// A frame = dag-cbor({t:'#commit', op:1})  ++  dag-cbor(body). Built to match millipds
// (repo_ops.py apply_writes) field-for-field. `seq` and `time` are firehose-server
// state supplied by the caller. `blocks` is a CAR (root = the new commit) of the
// changed blocks.

import * as dagCbor from '@ipld/dag-cbor'
import { writeCar } from 'atproto-repo/lib/repo.js'
import { commitCid } from 'atproto-repo/lib/commit.js'

// subscribeRepos frame = dag-cbor(header {t, op:1}) ++ dag-cbor(body). Each header is
// a constant byte prefix (dag-cbor canonicalises map order), so we emit/detect by it.
export const COMMIT_HEADER = dagCbor.encode({ t: '#commit', op: 1 })
export const IDENTITY_HEADER = dagCbor.encode({ t: '#identity', op: 1 })
export const ACCOUNT_HEADER = dagCbor.encode({ t: '#account', op: 1 })
export const SYNC_HEADER = dagCbor.encode({ t: '#sync', op: 1 })

function frame (header, body) {
  const bodyBytes = dagCbor.encode(body)
  const out = new Uint8Array(header.length + bodyBytes.length)
  out.set(header, 0)
  out.set(bodyBytes, header.length)
  return out
}

// Assemble a #commit frame from a RepoManager.applyWrite() result.
export async function commitFrame (write, { seq, time }) {
  const carBytes = await writeCar(write.commitCid, [write.commitBlock, ...write.newBlocks])
  return frame(COMMIT_HEADER, {
    ops: write.ops,
    seq,
    rev: write.rev,
    since: write.since,
    prev: null,
    repo: write.signedCommit.did,
    time,
    blobs: [],
    blocks: carBytes,
    commit: write.commitCid,
    rebase: false,
    tooBig: false,
    prevData: write.prevData
  })
}

// #identity — tells a relay a repo's handle/DID binding (needed for discovery).
export function identityFrame ({ seq, did, handle, time }) {
  const body = { seq, did, time }
  if (handle) body.handle = handle
  return frame(IDENTITY_HEADER, body)
}

// #account — repo hosting status (active true = "this PDS hosts this repo").
export function accountFrame ({ seq, did, active = true, status, time }) {
  const body = { seq, did, time, active }
  if (status) body.status = status
  return frame(ACCOUNT_HEADER, body)
}

// #sync (Sync 1.1) — tells a relay to RE-BASELINE the repo to `signedCommit` (re-anchor via
// getRepo) instead of applying incremental deltas. This is the sanctioned signal for a repo
// whose commit chain is discontinuous vs the relay's last-known state (e.g. a custodial→
// self-custody migration that re-authored the repo). `blocks` is a CAR carrying ONLY the
// commit block (root = its CID), ≤10KB. `rev`/`did` MUST match the commit object.
export async function syncFrame ({ seq, did, signedCommit, time }) {
  const cid = commitCid(signedCommit)
  const blocks = await writeCar(cid, [{ cid, bytes: dagCbor.encode(signedCommit) }])
  return frame(SYNC_HEADER, { seq, did, blocks, rev: signedCommit.rev, time })
}

const HEADERS = [['#commit', COMMIT_HEADER], ['#identity', IDENTITY_HEADER], ['#account', ACCOUNT_HEADER], ['#sync', SYNC_HEADER]]

// Re-stamp a frame's `seq` (relay egress): a relay aggregates many PDS firehoses and MUST
// present its own monotonic seq in each frame (the cursor contract). Decodes, replaces
// body.seq, re-encodes with the same header. Unknown frames pass through unchanged.
export function restampFrame (frameBytes, seq) {
  const d = decodeFrame(frameBytes)
  if (!d) return frameBytes
  const header = d.t === '#commit' ? COMMIT_HEADER : d.t === '#identity' ? IDENTITY_HEADER : d.t === '#account' ? ACCOUNT_HEADER : SYNC_HEADER
  d.body.seq = seq
  const bodyBytes = dagCbor.encode(d.body)
  const out = new Uint8Array(header.length + bodyBytes.length)
  out.set(header, 0)
  out.set(bodyBytes, header.length)
  return out
}

// Split + decode ANY known subscribeRepos frame → { t, body } (null if unknown).
export function decodeFrame (frameBytes) {
  for (const [t, header] of HEADERS) {
    if (frameBytes.length >= header.length &&
        Buffer.from(frameBytes.subarray(0, header.length)).equals(Buffer.from(header))) {
      return { t, body: dagCbor.decode(frameBytes.subarray(header.length)) }
    }
  }
  return null
}

// Split + decode a #commit frame. Returns { header, body } or null if not a #commit.
export function decodeCommitFrame (frameBytes) {
  const prefix = frameBytes.subarray(0, COMMIT_HEADER.length)
  if (!Buffer.from(prefix).equals(Buffer.from(COMMIT_HEADER))) return null
  return {
    header: { t: '#commit', op: 1 },
    body: dagCbor.decode(frameBytes.subarray(COMMIT_HEADER.length))
  }
}