modules/storage-autobase/hyper-p2p-autobase-view-sync/index.js

require('bare-process/global')
const EventEmitter = require('bare-events')
const b4a = require('b4a')
const crypto = require('hypercore-crypto')
const { assertAutobase, attachGossip, sendGossip } = require('../../_shared/storage-gossip-base.js')
const { autobaseStats, compareViewVersion, mergeViewRecord } = require('../../_shared/autobase-base.js')

const PROTOCOL = 'autobase-view-sync/v1'

class HyperP2PAutobaseViewSync extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.topic = opts.topic || null
    this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
    this.autobase = opts.autobase || null
    this._view = { version: 0, hash: null, at: 0 }
    this._stats = { published: 0, merged: 0, gossipIn: 0, gossipOut: 0 }
    this.swarm = null
  }

  attach (autobase) {
    assertAutobase(autobase)
    this.autobase = autobase
    return this
  }

  _hashView (version) {
    return b4a.toString(crypto.hash(b4a.from(`view:${version}`)), 'hex')
  }

  publishView (version) {
    if (version < 0) throw new Error('version must be non-negative')
    const view = {
      version,
      hash: this._hashView(version),
      at: Date.now()
    }
    if (version >= this._view.version) this._view = view
    sendGossip(this, { type: 'view-sync', view })
    this._stats.published++
    this._stats.gossipOut++
    this.emit('publish', view)
    return view
  }

  mergeRemoteView (view) {
    if (!view || typeof view.version !== 'number') return false
    if (view.version < this._view.version) return false
    if (view.version === this._view.version && view.hash !== this._view.hash) return false
    this._view = { ...mergeViewRecord(this._view, view), mergedAt: Date.now() }
    this._stats.merged++
    this.emit('merged', this._view)
    return true
  }

  syncStatus (targetVersion) {
    return compareViewVersion(this._view.version, targetVersion)
  }

  snapshot () {
    return { view: this.currentView(), stats: { ...this._stats } }
  }

  currentView () {
    return { ...this._view }
  }

  isAtLeast (version) {
    return this._view.version >= version
  }

  behindBy (version) {
    if (version < 0) throw new Error('version must be non-negative')
    return Math.max(0, version - this._view.version)
  }

  viewHash () {
    return this._view.hash
  }

  ageMs () {
    if (!this._view.at) return 0
    return Math.max(0, Date.now() - this._view.at)
  }

  catchUpTo (version) {
    if (version < 0) throw new Error('version must be non-negative')
    if (this._view.version >= version) return this.currentView()
    return this.publishView(version)
  }

  isSynced (version) {
    return this._view.version >= version
  }

  resetView () {
    this._view = { version: 0, hash: null, at: 0 }
    this._stats.merged = 0
    this.emit('reset', this.currentView())
    return this.currentView()
  }

  _onGossip (d) {
    if (!d || d.type !== 'view-sync' || !d.view) return
    this._stats.gossipIn++
    this.mergeRemoteView(d.view)
  }

  getStats () {
    return autobaseStats(this._stats, PROTOCOL, { view: this._view })
  }

  async ready () {
    if (this.swarm || !this.topic) return this
    await attachGossip(this, {
      keyPair: this.keyPair,
      topic: this.topic,
      protocol: PROTOCOL,
      onmessage: (d) => this._onGossip(d)
    })
    return this
  }

  async close () {
    if (this.swarm) await this.swarm.destroy().catch(() => {})
    this.swarm = null
    this.emit('closed')
  }
}

module.exports = { HyperP2PAutobaseViewSync, PROTOCOL }