hyper-vdrive / lib / storage.cjs

// Step 1 — atomic pointers.
//
// T3 kept index->hash in a JSON side-file, so a crash between the tree write and
// the flush orphaned the tree. Fix: the pointer IS the block value. Every stored
// record carries a 1-byte tag:
//
//   0x00 | <original bytes>     literal (bee blocks, anything we don't hold)
//   0x01 | <32-byte hash>       pointer (bytes live in the content store)
//
// Now the pointer lands in the same CoreTX as the tree node — same atom, same
// flush — and deleteBlock/deleteBlockRange/truncate free it with no extra work.
// Cost: 1 byte per literal block, 33 bytes per virtualised block.

// Dual-runtime: pgv3 and pgh-sidecar run under Bare, the tests and CLI under Node.
// Static requires on both branches so bundlers can still see the dependency.
const { isBare } = require('which-runtime')
const fs = isBare ? require('bare-fs') : require('fs')
const crypto = require('hypercore-crypto')
const CorestoreStorage = require('hypercore-storage')

const TAG_LITERAL = 0x00
const TAG_POINTER = 0x01

const stats = {
  putRedirected: 0,
  putStored: 0,
  getRedirected: 0,
  getStored: 0,
  bytesServed: 0,
  hashFailures: 0,
  missingBytes: 0
}

class ContentStore {
  constructor () {
    this.locations = new Map()
    this._fds = new Map()
  }

  add (hash, file, offset, length) {
    if (!this.locations.has(hash)) this.locations.set(hash, { file, offset, length })
  }

  has (hash) {
    return this.locations.has(hash)
  }

  read (hash, { verify = true } = {}) {
    const loc = this.locations.get(hash)
    if (!loc) {
      stats.missingBytes++
      return null
    }
    let fd = this._fds.get(loc.file)
    if (fd === undefined) {
      fd = fs.openSync(loc.file, 'r')
      this._fds.set(loc.file, fd)
    }
    const buf = Buffer.allocUnsafe(loc.length)
    fs.readSync(fd, buf, 0, loc.length, loc.offset)
    if (verify && crypto.data(buf).toString('hex') !== hash) {
      stats.hashFailures++
      return null
    }
    stats.bytesServed += buf.length
    return buf
  }

  close () {
    for (const fd of this._fds.values()) fs.closeSync(fd)
    this._fds.clear()
  }
}

function delegate (target, overrides) {
  return new Proxy(target, {
    get (t, prop) {
      if (prop === '__unwrapped') return t
      if (overrides && prop in overrides) return overrides[prop]
      const v = Reflect.get(t, prop)
      return typeof v === 'function' ? v.bind(t) : v
    },
    set (t, prop, value) {
      t[prop] = value
      return true
    }
  })
}

function build (content) {
  function decode (record) {
    if (record === null) return null
    if (record[0] === TAG_POINTER) {
      stats.getRedirected++
      return content.read(record.subarray(1).toString('hex'))
    }
    stats.getStored++
    return record.subarray(1)
  }

  function wrapRX (rx) {
    return delegate(rx, {
      getBlock (index) {
        return rx.getBlock(index).then(decode)
      }
    })
  }

  function wrapTX (tx) {
    return delegate(tx, {
      putBlock (index, data) {
        const digest = crypto.data(data)
        if (content.has(digest.toString('hex'))) {
          stats.putRedirected++
          const rec = Buffer.allocUnsafe(33)
          rec[0] = TAG_POINTER
          digest.copy(rec, 1)
          return tx.putBlock(index, rec)
        }
        stats.putStored++
        const rec = Buffer.allocUnsafe(data.byteLength + 1)
        rec[0] = TAG_LITERAL
        data.copy(rec, 1)
        return tx.putBlock(index, rec)
      }
    })
  }

  function wrapCoreStorage (inner) {
    if (!inner) return inner
    return delegate(inner, {
      read: (...args) => wrapRX(inner.read(...args)),
      write: (...args) => wrapTX(inner.write(...args)),
      snapshot: (...args) => wrapCoreStorage(inner.snapshot(...args)),
      atomize: (...args) => wrapCoreStorage(inner.atomize(...args))
    })
  }

  return class RedirectingCorestoreStorage extends CorestoreStorage {
    async resumeCore (discoveryKey) {
      return wrapCoreStorage(await super.resumeCore(discoveryKey))
    }

    async createCore (data) {
      return wrapCoreStorage(await super.createCore(data))
    }
  }
}

module.exports = { ContentStore, build, stats, TAG_LITERAL, TAG_POINTER }