atproto-store / lib / repo-store.js

// RepoStore — the Holepunch-native home for an atproto repo.
//
// Records live in a CLASSIC Hyperbee (matches our production stack — NOT hyperbee2/
// autobee, which are experimental; see the ecosystem audit): key = `collection/rkey`,
// value = the record's canonical DAG-CBOR bytes. Single-writer (one author key);
// multi-device is handled by sharing the mnemonic-derived key. The atproto-repo
// module is the *bridge* that reads this store and emits MST/CAR/commits on demand —
// this layer stores nothing atproto-specific beyond the records themselves.

import Corestore from 'corestore'
import { isCorestore } from './corestore.js'
import Hyperbee from 'hyperbee'
import * as dagCbor from '@ipld/dag-cbor'
import b4a from 'b4a'

// Range upper bound for a `collection/` prefix scan. '/' (0x2f) + 1 = '0' (0x30),
// and all atproto rkey chars sort >= '2' > '/', so `collection/…` < `collection0`.
function prefixUpper (prefix) {
  return prefix.slice(0, -1) + String.fromCharCode(prefix.charCodeAt(prefix.length - 1) + 1)
}

export class RepoStore {
  constructor (storage, opts = {}) {
    // Behavioural check, NOT `instanceof` — a second copy of corestore in a bundled tree makes a real
    // store look like a path and Corestore then path.join()s an object. See lib/corestore.js.
    this._ownStore = !isCorestore(storage)
    this._store = this._ownStore ? new Corestore(storage) : storage
    this._name = opts.name || 'atproto-repo'
    this._key = opts.key || null // open a specific core by KEY (a replica of another instance's repo)
    this.bee = null
    this.core = null
  }

  static key (collection, rkey) { return `${collection}/${rkey}` }

  // The underlying Corestore — so the PDS can open sibling cores (e.g. the firehose
  // sequence log) on the SAME store rather than a second Corestore over the same dir.
  get corestore () { return this._store }

  async ready () {
    // By KEY = replicate another instance's repo (read-only over the swarm); by NAME = our own.
    this.core = this._key ? this._store.get({ key: this._key }) : this._store.get({ name: this._name })
    await this.core.ready()
    // alwaysDuplicate:false enables Hyperbee's byte-identical value-dedup — a re-put of an unchanged
    // (key,value) becomes a no-op READ instead of appending a fresh B-tree path. Without it (the default
    // TRUE), redundant puts (e.g. a projection re-putting the whole record set on a timer) balloon this
    // append-only core to millions of dead blocks. Defense-in-depth under every caller.
    this.bee = new Hyperbee(this.core, { keyEncoding: 'utf-8', valueEncoding: 'binary', alwaysDuplicate: false })
    await this.bee.ready()
    return this
  }

  async put (collection, rkey, record) {
    await this.bee.put(RepoStore.key(collection, rkey), b4a.from(dagCbor.encode(record)))
  }

  async get (collection, rkey) {
    const node = await this.bee.get(RepoStore.key(collection, rkey))
    return node ? dagCbor.decode(node.value) : null
  }

  // Raw stored bytes + decoded record (the bridge CIDs the exact bytes).
  async getRaw (collection, rkey) {
    const node = await this.bee.get(RepoStore.key(collection, rkey))
    return node ? { bytes: node.value, record: dagCbor.decode(node.value) } : null
  }

  async del (collection, rkey) {
    await this.bee.del(RepoStore.key(collection, rkey))
  }

  // Yield { rkey, record } for one collection, in key order.
  async * list (collection) {
    const gte = `${collection}/`
    const lt = prefixUpper(gte)
    for await (const node of this.bee.createReadStream({ gte, lt })) {
      yield { rkey: node.key.slice(gte.length), record: dagCbor.decode(node.value) }
    }
  }

  // All records as { key, bytes, record } in key order. `bytes` is the canonical
  // DAG-CBOR — the bridge CIDs these directly to build the MST (byte-exact).
  async records () {
    const out = []
    for await (const node of this.bee.createReadStream()) {
      out.push({ key: node.key, bytes: node.value, record: dagCbor.decode(node.value) })
    }
    return out
  }

  // Persist the commit head ({ rev, commitCid, dataCid }) as core USER DATA — NOT a record,
  // so it never appears in records()/the MST. This is what makes the head stable across a
  // restart: the rev is restored, not re-minted (which would forge a phantom commit).
  async saveHead (head) {
    await this.core.setUserData('atproto-head', b4a.from(JSON.stringify(head)))
  }

  async loadHead () {
    const v = await this.core.getUserData('atproto-head')
    return v ? JSON.parse(b4a.toString(v)) : null
  }

  async close () {
    if (this.bee) await this.bee.close()
    if (this._ownStore) await this._store.close()
  }
}