atproto-store / lib / blob-store.js

// BlobStore — content-addressed blob storage for the PDS (images, avatars).
//
// atproto blobs are opaque bytes addressed by a CIDv1/raw/sha256 CID; a record references
// one via a blob ref ({$type:'blob', ref:{$link:<cid>}, mimeType, size}), and
// com.atproto.sync.getBlob serves it back by CID. Content-addressed, so it's shared across
// accounts on this PDS (the same bytes upload once). Backed by ONE classic Hyperbee on the
// shared corestore — same substrate as RepoStore — value framed as [4-byte headerLen] +
// JSON{mimeType,size} + raw bytes, so a single binary bee holds bytes + their mime type.
//
// (hyperblobs would be the heavier "right" tool for very large blobs; a Hyperbee is ample
// for avatars/images and adds no dependency. Revisit if blob sizes grow.)

import Corestore from 'corestore'
import { isCorestore } from './corestore.js'
import Hyperbee from 'hyperbee'
import b4a from 'b4a'
import { cidForBlobSync } from 'atproto-repo/lib/cid.js'

export class BlobStore {
  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 || 'pds-blobs'
    this.bee = null
  }

  async ready () {
    const core = this._store.get({ name: this._name })
    await core.ready()
    // alwaysDuplicate:false → re-uploading identical bytes at the same CID is a no-op, not a re-append of
    // the full blob value (content-addressed key is stable, so the value is byte-identical). See repo-store.js.
    this.bee = new Hyperbee(core, { keyEncoding: 'utf-8', valueEncoding: 'binary', alwaysDuplicate: false })
    await this.bee.ready()
    return this
  }

  // Store bytes; returns { cid (CID), mimeType, size }. Idempotent (content-addressed).
  async put (bytes, mimeType = 'application/octet-stream') {
    const buf = b4a.isBuffer(bytes) ? bytes : b4a.from(bytes)
    const cid = cidForBlobSync(buf)
    const header = b4a.from(JSON.stringify({ mimeType, size: buf.length }), 'utf-8')
    const len = b4a.allocUnsafe(4)
    len.writeUInt32BE(header.length, 0)
    await this.bee.put(cid.toString(), b4a.concat([len, header, buf]))
    return { cid, mimeType, size: buf.length }
  }

  // Fetch by CID string; returns { mimeType, size, bytes } or null.
  async get (cidStr) {
    const node = await this.bee.get(cidStr)
    if (!node) return null
    const frame = node.value
    const headerLen = frame.readUInt32BE(0)
    const header = JSON.parse(b4a.toString(frame.subarray(4, 4 + headerLen), 'utf-8'))
    return { mimeType: header.mimeType, size: header.size, bytes: frame.subarray(4 + headerLen) }
  }

  async has (cidStr) { return !!(await this.bee.get(cidStr)) }

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