index.js

const b4a = require('b4a')
const ReadyResource = require('ready-resource')
const debounceify = require('debounceify')
const c = require('compact-encoding')
const safetyCatch = require('safety-catch')
const hypercoreId = require('hypercore-id-encoding')
const assert = require('nanoassert')
const SignalPromise = require('signal-promise')
const CoreCoupler = require('core-coupler')
const mutexify = require('mutexify/promise')
const ProtomuxWakeup = require('protomux-wakeup')
const rrp = require('resolve-reject-promise')

const Linearizer = require('./lib/linearizer.js')
const SystemView = require('./lib/system.js')
const UpdateChanges = require('./lib/updates.js')
const messages = require('./lib/messages.js')
const Timer = require('./lib/timer.js')
const Writer = require('./lib/writer.js')
const ActiveWriters = require('./lib/active-writers.js')
const AutoWakeup = require('./lib/wakeup.js')

const FastForward = require('./lib/fast-forward.js')
const AutoStore = require('./lib/store.js')
const ApplyState = require('./lib/apply-state.js')
const { PublicApplyCalls } = require('./lib/apply-calls.js')
const boot = require('./lib/boot.js')

const inspect = Symbol.for('nodejs.util.inspect.custom')
const INTERRUPT = new Error('Apply interrupted')
const BINARY_ENCODING = c.from('binary')

const AUTOBASE_VERSION = 1

const RECOVERIES = 3
const FF_RECOVERY = 1

// default is to automatically ack
const DEFAULT_ACK_INTERVAL = 10_000
const DEFAULT_ACK_THRESHOLD = 4

const REMOTE_ADD_BATCH = 64
const MIN_FF_WAIT = 300_000 // wait at least 5min before attempting to ff again after failure

class WakeupHandler {
  constructor (base, discoveryKey) {
    this.active = true
    this.discoveryKey = discoveryKey
    this.base = base
  }

  onpeeradd (peer, session) {
    if (this.base._coupler) this.base._coupler.update(peer.stream)
  }

  onpeerremove (peer, session) {
    // do nothing
  }

  onlookup (req, peer, session) {
    const wakeup = this.base._getWakeup()
    if (wakeup.length === 0) return
    session.announce(peer, wakeup)
  }

  onannounce (wakeup, peer, session) {
    if (this.base.isFastForwarding()) return
    this.base.hintWakeup(wakeup)
  }
}

module.exports = class Autobase extends ReadyResource {
  constructor (store, bootstrap, handlers = {}) {
    if (Array.isArray(bootstrap)) bootstrap = bootstrap[0] // TODO: just a quick compat, lets remove soon

    if (bootstrap && typeof bootstrap !== 'string' && !b4a.isBuffer(bootstrap)) {
      handlers = bootstrap
      bootstrap = null
    }

    super()

    const key = bootstrap ? toKey(bootstrap) : null

    this.id = null
    this.key = key
    this.discoveryKey = null

    this.keyPair = null
    this.valueEncoding = c.from(handlers.valueEncoding || 'binary')
    this.store = store
    this.globalCache = store.globalCache || null
    this.migrated = false

    this.encrypted = handlers.encrypted || !!handlers.encryptionKey
    this.encrypt = !!handlers.encrypt
    this.encryptionKey = handlers.encryptionKey || null
    this.encryption = null

    this.local = null
    this.localWriter = null
    this.isIndexer = false

    this.activeWriters = new ActiveWriters()
    this.linearizer = null
    this.updating = false

    this.nukeTip = !!handlers.nukeTip

    this.wakeupOwner = !handlers.wakeup
    this.wakeupCapability = handlers.wakeupCapability || null
    this.wakeupProtocol = handlers.wakeup || new ProtomuxWakeup()
    this.wakeupSession = null

    this._primaryBootstrap = null

    this.fastForwardEnabled = handlers.fastForward !== false
    this.fastForwarding = null
    this.fastForwardTo = null
    this.fastForwardFailedAt = 0

    this._bootstrapWriters = [] // might contain dups, but thats ok
    this._bootstrapWritersChanged = false

    this._checkWriters = []
    this._optimistic = -1
    this._appended = 0
    this._appending = null
    this._wakeup = new AutoWakeup(this)
    this._wakeupHints = new Map()
    this._wakeupPeerBound = this._wakeupPeer.bind(this)
    this._coupler = null

    this._lock = mutexify()

    this._needsWakeup = true
    this._needsWakeupHeads = true

    this._updates = []
    this._handlers = handlers || {}
    this._warn = emitWarning.bind(this)

    this._draining = false
    this._writable = null
    this._advancing = null
    this._interrupting = false
    this._caughtup = false

    this.paused = false

    this._bump = debounceify(() => {
      this._advancing = this._advance()
      return this._advancing
    })

    this._onremotewriterchangeBound = this._onremotewriterchange.bind(this)

    this.maxSupportedVersion = AUTOBASE_VERSION // working version

    this._preopen = null

    this._hasOpen = !!this._handlers.open
    this._hasApply = !!this._handlers.apply
    this._hasOptimisticApply = !!this._handlers.optimistic
    this._hasUpdate = !!this._handlers.update
    this._hasClose = !!this._handlers.close

    this._viewStore = new AutoStore(this)
    this._applyState = null

    this.view = null
    this.core = null
    this.version = -1
    this.interrupted = null
    this.recoveries = RECOVERIES

    const {
      ackInterval = DEFAULT_ACK_INTERVAL,
      ackThreshold = DEFAULT_ACK_THRESHOLD
    } = handlers

    this._ackInterval = ackInterval
    this._ackThreshold = ackThreshold
    this._ackTickThreshold = ackThreshold
    this._ackTick = 0

    this._ackTimer = null
    this._acking = false

    this._waiting = new SignalPromise()
    this._bootRecovery = false

    this.view = this._hasOpen ? this._handlers.open(this._viewStore, new PublicApplyCalls(this)) : null
    this.core = this._viewStore.get({ name: '_system' })

    if (this.fastForwardEnabled && isObject(handlers.fastForward)) {
      this._runFastForward(new FastForward(this, handlers.fastForward.key, { verified: false })).catch(noop)
    }

    this.ready().catch(safetyCatch)
  }

  [inspect] (depth, opts) {
    let indent = ''
    if (typeof opts.indentationLvl === 'number') {
      while (indent.length < opts.indentationLvl) indent += ' '
    }

    return indent + 'Autobase { ... }'
  }

  // just compat, use .key
  get bootstrap () {
    return this.key
  }

  // TODO: compat, will be removed
  get bootstraps () {
    return [this.bootstrap]
  }

  get writable () {
    return this.localWriter !== null && !this.localWriter.isRemoved
  }

  get ackable () {
    return this.localWriter !== null && this.localWriter.isActiveIndexer
  }

  get signedLength () {
    return this.core.signedLength
  }

  get indexedLength () {
    return this._applyState ? this._applyState.indexedLength : 0
  }

  get length () {
    return this.core.length
  }

  hash () {
    return this.core.treeHash()
  }

  // deprecated, use .core.key
  getSystemKey () {
    return this.core.key
  }

  get system () {
    return this._applyState && this._applyState.system
  }

  // deprecated
  async getIndexedInfo () {
    if (this.opened === false) await this.ready()
    return this._applyState && this._applyState.system.getIndexedInfo(this._applyState.indexedLength)
  }

  _isActiveIndexer () {
    return this.localWriter ? this.localWriter.isActiveIndexer : false
  }

  replicate (isInitiator, opts) {
    const stream = this.store.replicate(isInitiator, opts)
    this.wakeupProtocol.addStream(stream)
    return stream
  }

  heads () {
    if (!this._applyState || !this._applyState.opened) return []
    const nodes = new Array(this._applyState.system.heads.length)
    for (let i = 0; i < this._applyState.system.heads.length; i++) nodes[i] = this._applyState.system.heads[i]
    return nodes.sort(compareNodes)
  }

  hintWakeup (hints) {
    if (!Array.isArray(hints)) hints = [hints]
    for (const { key, length } of hints) {
      const hex = b4a.toString(key, 'hex')
      const prev = this._wakeupHints.get(hex)
      if (!prev || length === -1 || prev < length) this._wakeupHints.set(hex, length)
    }
    this._queueBump()
  }

  _queueBump () {
    this._bump().catch(safetyCatch)
  }

  async _runPreOpen () {
    if (this._handlers.wait) await this._handlers.wait()

    await this.store.ready()

    this.keyPair = (await this._handlers.keyPair) || null

    const result = await boot(this.store, this.key, {
      encryptionKey: this.encryptionKey,
      encrypt: this.encrypt,
      keyPair: this.keyPair
    })

    const pointer = await result.local.getUserData('autobase/boot')

    if (pointer) {
      const { recoveries } = c.decode(messages.BootRecord, pointer)
      this.recoveries = recoveries
    }

    this._primaryBootstrap = result.bootstrap
    this.local = result.local

    this.key = result.bootstrap.key
    this.discoveryKey = result.bootstrap.discoveryKey
    this.id = result.bootstrap.id

    this.encryptionKey = result.encryptionKey
    if (this.encryptionKey) this.encryption = { key: this.encryptionKey }

    if (this.encrypted) {
      assert(this.encryptionKey !== null, 'Encryption key is expected')
    }

    if (this.nukeTip) await this._nukeTip()

    this.setWakeup(this.wakeupCapability || this.key, null)
  }

  async _nukeTipBatch (key, length) {
    const core = this.store.get({ key, active: false })
    await core.ready()

    const batch = core.session({ name: 'batch' })
    await batch.ready()
    if (batch.length > length) await batch.truncate(length)
    await batch.close()

    await core.close()
  }

  // TODO: not atomic atm, so more of a (very useful) debug helper
  async _nukeTip () {
    const pointer = await this.local.getUserData('autobase/boot')
    if (!pointer) return

    const boot = c.decode(messages.BootRecord, pointer)

    const tx = this.local.state.storage.write()
    tx.deleteLocalRange(b4a.from([messages.LINEARIZER_PREFIX]), b4a.from([messages.LINEARIZER_PREFIX + 1]))
    await tx.flush()

    await this._nukeTipBatch(boot.key, boot.indexedLength)

    const encryption = this.encryptionKey
      ? { key: AutoStore.getBlockKey(this.bootstrap, this.encryptionKey, '_system'), block: true }
      : null

    const core = this.store.get({ key: boot.key, encryption, active: false })
    await core.ready()

    const batch = core.session({ name: 'batch' })
    await batch.ready()

    const info = await SystemView.getIndexedInfo(batch, boot.indexedLength)
    await batch.close()

    for (const view of info.views) { // ensure any views ref'ed by system are consistent as well
      await this._nukeTipBatch(view.key, view.length)
    }

    if (boot.heads) this.hintWakeup(boot.heads)
    if (this.local.length) this.hintWakeup([{ key: this.local.key, length: this.local.length }])
  }

  setWakeup (cap, discoveryKey) {
    if (this.wakeupSession) this.wakeupSession.destroy()
    if (!discoveryKey && b4a.equals(cap, this.key)) discoveryKey = this.discoveryKey
    this.wakeupSession = this.wakeupProtocol.session(cap, new WakeupHandler(this, discoveryKey || null))
  }

  async _getMigrationPointer (key, length) {
    const encryption = this.encryptionKey
      ? { key: AutoStore.getBlockKey(this.bootstrap, this.encryptionKey, '_system'), block: true }
      : null

    const core = this.store.get({ key, active: false, encryption })
    await core.ready()

    const min = (core.manifest && core.manifest.prologue) ? core.manifest.prologue.length : 0

    for (let i = length - 1; i >= min; i--) {
      if (!(await core.has(i))) continue

      const sys = new SystemView(core, { checkout: i + 1 })
      await sys.ready()

      let good = true

      for (const v of sys.views) {
        const vc = this.store.get({ key: v.key, active: false })
        await vc.ready()
        if (vc.length < v.length) good = false
        await vc.close()
      }

      await sys.close()
      if (!good) continue

      return i + 1
    }

    return min
  }

  // migrating from 6 -> latest
  async _migrate6 (key, length) {
    const core = this.store.get({ key, active: false })
    await core.ready()
    const batch = core.session({ name: 'batch', overwrite: true, checkout: length })
    await batch.ready()
    await batch.close()
    await core.close()
  }

  // called by view-store for bootstrapping
  async _getSystemInfo () {
    const boot = await this._getBootRecord()
    if (!boot.key) return null

    const migrated = !!boot.heads

    if (migrated) { // ensure system batch is consistent on initial migration
      await this._migrate6(boot.key, boot.indexedLength)
    }

    const encryption = this.encryptionKey
      ? { key: AutoStore.getBlockKey(this.bootstrap, this.encryptionKey, '_system'), block: true }
      : null

    const core = this.store.get({ key: boot.key, encryption, active: false })
    await core.ready()

    const batch = core.session({ name: 'batch' })
    const info = await SystemView.getIndexedInfo(batch, boot.indexedLength)
    await batch.close()
    await core.close()

    if (info.version > AUTOBASE_VERSION) {
      throw new Error('Autobase upgrade required.')
    }

    // just compat
    if (migrated) {
      this.migrated = true

      for (const view of info.views) { // ensure any views ref'ed by system are consistent as well
        await this._migrate6(view.key, view.length)
      }

      if (boot.heads) this.hintWakeup(boot.heads)
      if (this.local.length) this.hintWakeup([{ key: this.local.key, length: this.local.length }])
    }

    return {
      key: boot.key,
      indexers: info.indexers,
      views: info.views
    }
  }

  // called by the apply state for bootstrapping
  async _getBootRecord () {
    await this._preopen

    const pointer = await this.local.getUserData('autobase/boot')

    const boot = pointer
      ? c.decode(messages.BootRecord, pointer)
      : { key: null, indexedLength: 0, indexersUpdated: false, fastForwarding: false, recoveries: RECOVERIES, heads: null }

    if (boot.heads) {
      const len = await this._getMigrationPointer(boot.key, boot.indexedLength)
      if (len !== boot.indexedLength) this._warn(new Error('Invalid pointer in migration, correcting (' + len + ' vs ' + boot.indexedLength + ')'))
      boot.indexedLength = len
    }

    return boot
  }

  _interrupt (reason) {
    assert(this._applyState.applying, 'Interrupt is only allowed in apply')
    this._interrupting = true
    if (reason) this.interrupted = reason
    throw INTERRUPT
  }

  async flush () {
    if (this.opened === false) await this.ready()
    await this._advancing
  }

  recouple () {
    if (this._coupler) this._coupler.destroy()
    const core = this._viewStore.getSystemCore()
    this._coupler = new CoreCoupler(core, this._wakeupPeerBound)
  }

  _updateBootstrapWriters () {
    const writers = this.linearizer.getBootstrapWriters()

    // first clear all, but without applying it for churn reasons
    for (const writer of this._bootstrapWriters) {
      writer.isBootstrap = false
      writer.isCoupled = false
    }

    // all passed are bootstraps
    for (const writer of writers) {
      writer.isCoupled = true
      writer.setBootstrap(true)
    }

    // reset activity on old ones, all should be in sync now
    for (const writer of this._bootstrapWriters) {
      if (writer.isBootstrap === false) writer.setBootstrap(false)
    }

    this._bootstrapWriters = writers
    this._bootstrapWritersChanged = false
  }

  async _openLinearizer () {
    if (this._applyState.system.bootstrapping) {
      await this._makeLinearizer(null)
      this._bootstrapLinearizer()
      return
    }

    await this._makeLinearizerFromViewState()
  }

  async _catchupApplyState () {
    if (await this._applyState.shouldMigrate()) {
      await this._migrate()
    } else {
      await this._applyState.catchup(this.linearizer)
    }

    this._caughtup = true
  }

  async _open () {
    this._preopen = this._runPreOpen()
    await this._preopen

    if (this.closing) return

    this._applyState = new ApplyState(this)

    try {
      await this._applyState.ready()
    } catch (err) {
      if (this.closing) return

      try {
        await this._applyState.close()
      } catch {}

      try {
        await this.core.ready()
      } catch {}

      this._applyState = null
      if (this.closing) return

      this._warn(new Error('Failed to boot due to: ' + err.message))

      if (this.recoveries < RECOVERIES) {
        this._bootRecovery = true
        this._queueBump()
        return
      }

      throw err
    }

    try {
      await this._openLinearizer()
      await this.core.ready()
      await this._wakeup.ready()
    } catch (err) {
      if (this.closing) return
      throw err
    }

    if (this.core.length - this._applyState.indexedLength > this._ackTickThreshold) {
      this._ackTick = this._ackTickThreshold
    }
    if (this.localWriter && this._ackInterval) {
      this._startAckTimer()
    }

    this._updateBootstrapWriters()

    this.recouple()
    this._queueFastForward()

    // queue a full bump that handles wakeup etc (not legal to wait for that here)
    this._queueBump()
  }

  async _closeLocalCores () {
    const closing = []
    if (this._primaryBootstrap) closing.push(this._primaryBootstrap.close())
    if (this.localWriter) closing.push(this._unsetLocalWriter())
    closing.push(this._closeAllActiveWriters())
    if (this.localWriter) await this.localWriter.close()
    await Promise.all(closing)
    await this.local.close()
  }

  async _close () {
    this._interrupting = true
    await Promise.resolve() // defer one tick

    if (this.wakeupSession) this.wakeupSession.destroy()
    if (this.wakeupOwner) this.wakeupProtocol.destroy()

    if (this.fastForwarding) await this.fastForwarding.close()

    if (this._coupler) this._coupler.destroy()
    this._coupler = null
    this._waiting.notify(null)

    await this.activeWriters.clear()

    const closing = this._advancing ? this._advancing.catch(safetyCatch) : null
    await this._closeLocalCores()

    if (this._ackTimer) {
      this._ackTimer.stop()
      await this._ackTimer.flush()
    }

    await this._wakeup.close()

    if (this._hasClose) await this._handlers.close(this.view)
    if (this._applyState) await this._applyState.close()

    await this._viewStore.close()
    await this.core.close()
    await this.store.close()

    if (this._writable) this._writable.resolve(false)

    await closing
  }

  _onError (err) {
    if (this.closing) return

    if (err === INTERRUPT) {
      this.emit('interrupt', this.interrupted)
      this.emit('update')
      return
    }

    this.close().catch(safetyCatch)

    // if no one is listening we should crash! we cannot rely on the EE here
    // as this is wrapped in a promise so instead of nextTick throw it
    if (ReadyResource.listenerCount(this, 'error') === 0) {
      crashSoon(err)
      return
    }

    this.emit('error', err)
  }

  async _closeWriter (w, now) {
    this.activeWriters.delete(w)
    await w.close()
  }

  async _gcWriters () {
    // just return early, why not
    if (this._checkWriters.length === 0) return

    while (this._checkWriters.length > 0) {
      const w = this._checkWriters.pop()

      // doesnt hurt
      w.updateActivity()

      if (!w.flushed()) continue

      const unqueued = this._wakeup.unqueue(w.core.key, w.core.length)

      if (!unqueued || w.isActiveIndexer) continue
      if (this.localWriter === w) continue

      await this._closeWriter(w, false)
    }

    await this._wakeup.flush()
  }

  _startAckTimer () {
    if (this._ackTimer) return
    this._ackTimer = new Timer(this._backgroundAck.bind(this), this._ackInterval)
    this._bumpAckTimer()
  }

  _bumpAckTimer () {
    if (!this._ackTimer) return
    this._ackTimer.bump()
  }

  async update () {
    if (this.opened === false) await this.ready()

    try {
      await this._bump()
      if (this._acking) await this._bump() // if acking just rebump incase it was triggered from above...
    } catch (err) {
      if (this._interrupting) return
      throw err
    }
  }

  // runs in bg, not allowed to throw
  // TODO: refactor so this only moves the writer affected to a updated set
  async _onremotewriterchange () {
    this._bumpAckTimer()

    try {
      await this._bump()
    } catch {}
  }

  _onwakeup () {
    this._needsWakeup = true
    this._queueBump()
  }

  isFastForwarding () {
    if (this.fastForwardTo !== null) return true
    return this.fastForwardEnabled && this.fastForwarding !== null
  }

  _backgroundAck () {
    return this.ack(true)
  }

  async ack (bg = false) {
    if (this.opened === false) await this.ready()
    if (this.localWriter === null || this._acking || this._interrupting || this._appending !== null) return

    if (this._applyState === null) {
      try {
        await this._bump()
      } catch {}
      if (this._applyState === null || this._interrupting) return
    }

    const applyState = this._applyState
    if (applyState.opened === false) await applyState.ready()

    const isPendingIndexer = applyState.isLocalPendingIndexer()

    // if no one is waiting for our index manifest, wait for FF before pushing an ack
    if ((!isPendingIndexer && this.isFastForwarding()) || this._interrupting) return

    const isIndexer = applyState.isLocalIndexer() || isPendingIndexer
    if (!isIndexer) return

    this._acking = true

    try {
      await this._bump()
    } catch (err) {
      if (!this._interrupting) throw err
    }

    if (this._interrupting || !this.localWriter || this.localWriter.closed) {
      this._acking = false
      return
    }

    // avoid lumping acks together due to the bump wait here
    if (this._ackTimer && bg) await this._ackTimer.asapStandalone()
    if (this._interrupting) {
      this._acking = false
      return
    }

    const alwaysWrite = isPendingIndexer || this._applyState.shouldWrite()

    if (alwaysWrite || this.linearizer.shouldAck(this.localWriter, false)) {
      try {
        if (this.localWriter && !this.localWriter.closed) await this.append(null)
      } catch (err) {
        if (!this._interrupting) throw err
      }
    }

    if (!this._interrupting) {
      this._updateAckThreshold()
      this._bumpAckTimer()
    }

    this._acking = false
  }

  async append (value, opts) {
    if (this.opened === false) await this.ready()
    if (this._interrupting) throw new Error('Autobase is closing')
    if (value && this.valueEncoding !== BINARY_ENCODING) value = normalize(this.valueEncoding, value)

    const optimistic = !!opts && !!opts.optimistic && !!value

    // we wanna allow acks so interdexers can flush
    if (!optimistic && (this.localWriter === null || (this.localWriter.isRemoved && value !== null))) {
      throw new Error('Not writable')
    }

    if (this._appending === null) this._appending = []

    if (Array.isArray(value)) {
      for (const v of value) this._append(v)
    } else {
      this._append(value)
    }

    if (optimistic) this._optimistic = this._appending.length - 1
    const target = this._appended + this._appending.length

    // await in case append is in current tick
    if (this._advancing) await this._advancing

    // bump until we've flushed the nodes
    while (this._appended < target && !this._interrupting) {
      await this._bump()
      // safety
      if (this.localWriter && this.localWriter.idle()) return
    }
  }

  _append (value) {
    // if prev value is an ack that hasnt been flushed, skip it
    if (this._appending.length > 0) {
      if (value === null) return
      if (this._appending[this._appending.length - 1] === null) {
        this._appending.pop()
      }
    }
    this._appending.push(value)
  }

  static encodeValue (value, opts = {}) {
    return c.encode(messages.OplogMessage, {
      version: AUTOBASE_VERSION,
      maxSupportedVersion: AUTOBASE_VERSION,
      digist: null,
      checkpoint: null,
      optimistic: !!opts.optimistic,
      node: {
        heads: opts.heads || [],
        batch: 1,
        value
      }
    })
  }

  static async getLocalKey (store, opts = {}) {
    const core = opts.keyPair ? store.get({ ...opts, active: false }) : store.get({ ...opts, name: 'local', active: false })
    await core.ready()
    const key = core.key
    await core.close()
    return key
  }

  static getLocalCore (store, handlers, encryptionKey) {
    const encryption = !encryptionKey ? null : { key: encryptionKey }
    const opts = { ...handlers, compat: false, active: false, exclusive: true, valueEncoding: messages.OplogMessage, encryption }
    return opts.keyPair ? store.get(opts) : store.get({ ...opts, name: 'local' })
  }

  static async getUserData (core) {
    const view = await core.getUserData('autobase/view')

    return {
      referrer: await core.getUserData('referrer'),
      view: view ? b4a.toString(view) : null
    }
  }

  static async isAutobase (core, opts = {}) {
    const block = await core.get(0, opts)
    if (!block) throw new Error('Core is empty.')
    if (!b4a.isBuffer(block)) return isAutobaseMessage(block)

    try {
      const m = c.decode(messages.OplogMessage, block)
      return isAutobaseMessage(m)
    } catch {
      return false
    }
  }

  // no guarantees where the user data is stored, just that its associated with the base
  async setUserData (key, val) {
    await this._preopen
    const core = this._primaryBootstrap === null ? this.local : this._primaryBootstrap

    await core.setUserData(key, val)
  }

  async getUserData (key) {
    await this._preopen
    const core = this._primaryBootstrap === null ? this.local : this._primaryBootstrap

    return await core.getUserData(key)
  }

  _needsLocalWriter () {
    return this.localWriter === null || this.localWriter.closed
  }

  // no guarantees about writer.isActiveIndexer property here
  async _getWriterByKey (key, len, seen, allowGC, isAdded, system) {
    assert(this._draining === true || (this.opening && !this.opened) || this._optimistic > -1)

    const release = await this._lock()

    if (this._interrupting) {
      release()
      throw new Error('Autobase is closing')
    }

    try {
      let w = this.activeWriters.get(key)

      const alreadyActive = !!w
      const sys = system || this._applyState.system
      const writerInfo = await sys.get(key)

      if (len === -1) {
        if (!allowGC && writerInfo === null) {
          if (w) w.isRemoved = !isAdded
          return null
        }

        len = writerInfo === null ? 0 : writerInfo.length
      }

      const isActive = writerInfo !== null && (isAdded || !writerInfo.isRemoved)
      const isRemoved = !isActive

      if (w) {
        w.isRemoved = isRemoved
      } else {
        w = this._makeWriter(key, len, isActive, isRemoved)
        if (!w) return null
      }

      if (isRemoved && sys.bootstrapping && b4a.equals(w.core.key, this.key)) {
        w.isRemoved = false
      }

      if (this._isLocalCore(w.core) && this._needsLocalWriter()) {
        this._setLocalWriter(w)
      }

      w.seen(seen)

      if (alreadyActive) return w

      await w.ready()

      if (this._isLocalCore(w.core) && this._needsLocalWriter()) {
        this._setLocalWriter(w)
      }

      if (allowGC && w.flushed()) {
        this._wakeup.unqueue(key, len)
        if (w !== this.localWriter) {
          await w.close()
          return w
        }
      }

      this.activeWriters.add(w)
      this._checkWriters.push(w)

      assert(w.opened)
      assert(!w.closed)

      w.updateActivity()
      return w
    } finally {
      release()
    }
  }

  _updateAll () {
    const p = []
    for (const w of this.activeWriters) p.push(w.update(null).catch(this._warn))
    return Promise.all(p)
  }

  _makeWriterCore (key) {
    if (this.closing) throw new Error('Autobase is closing')
    if (this._interrupting) throw INTERRUPT()

    const local = b4a.equals(key, this.local.key)

    const core = local
      ? this.local.session({ valueEncoding: messages.OplogMessage, encryption: this.encryption, active: false })
      : this.store.get({ key, compat: false, writable: false, valueEncoding: messages.OplogMessage, encryption: this.encryption, active: false })

    return core
  }

  _makeWriter (key, length, isActive, isRemoved) {
    const core = this._makeWriterCore(key)
    const w = new Writer(this, core, length, isRemoved)

    if (this._isLocalCore(core)) {
      if (isActive) this._setLocalWriter(w) // only set active writer
      return w
    }

    core.on('append', this._onremotewriterchangeBound)
    core.on('download', this._onremotewriterchangeBound)
    core.on('manifest', this._onremotewriterchangeBound)

    return w
  }

  _updateLinearizer (indexers, heads) {
    // only current active indexers are reset to true below
    for (const w of this.activeWriters) w.isActiveIndexer = false
    if (this.localWriter) this.localWriter.isActiveIndexer = false

    for (const writer of indexers) writer.isActiveIndexer = true

    if (this._isActiveIndexer() && !this.isIndexer) {
      this._setLocalIndexer()
    } else if (!this._isActiveIndexer() && this.isIndexer) {
      this._clearLocalIndexer()
    }

    this.linearizer = new Linearizer(indexers, { heads, writers: this.activeWriters })

    this._updateAckThreshold()
  }

  async _updateLocalWriter (sys) {
    if (this.localWriter !== null && !this.localWriter.closed) return
    await this._getWriterByKey(this.local.key, -1, 0, true, false, sys)
  }

  async _bootstrapLinearizer () {
    const bootstrap = this._makeWriter(this.key, 0, true, false)

    this.activeWriters.add(bootstrap)
    this._checkWriters.push(bootstrap)
    await bootstrap.ready()
    this._ensureWakeup(bootstrap)

    this._updateLinearizer([bootstrap], [])
  }

  async _makeLinearizer (sys) {
    if (sys === null) {
      return this._bootstrapLinearizer()
    }

    if (this.opened || await sys.hasLocal(this.local.key)) {
      await this._updateLocalWriter(sys)
    }

    const indexers = []

    for (const head of sys.indexers) {
      const writer = await this._getWriterByKey(head.key, head.length, 0, false, false, sys)
      indexers.push(writer)
    }

    if (!this._isActiveIndexer()) {
      for (const key of sys.pendingIndexers) {
        if (b4a.equals(key, this.local.key)) {
          this._setLocalIndexer()
          break
        }
      }
    }

    this._updateLinearizer(indexers, sys.heads)

    for (const { key, length } of sys.heads) {
      await this._getWriterByKey(key, length, 0, false, false, sys)
    }
  }

  async _clearWriters () {
    await this.activeWriters.clear()
    if (this.localWriter !== null) await this.localWriter.close()
    this._checkWriters = []
  }

  async _makeLinearizerFromViewState () {
    const sys = await this._applyState.getIndexedSystem()
    await this._makeLinearizer(sys)
    await sys.close()
  }

  async _recoverMaybe () {
    if (!this._applyState) {
      if (!this._bootRecovery) return
      await this._runFastForward(new FastForward(this, this.core.key, { force: true }))
      this._queueBump()
      return
    }

    this._bootRecovery = false

    const ff = await this._applyState.recoverAt()
    if (!ff || this.fastForwardTo) return

    this.fastForwardTo = ff
    this._queueBump()
  }

  async _applyFastForward () {
    if (!this.fastForwardTo.force && this.fastForwardTo.length < this.core.length + FastForward.MINIMUM) {
      this.fastForwardTo = null
      this._updateActivity()
      // still not worth it
      return
    }

    const changes = this._hasUpdate ? new UpdateChanges(this) : null
    if (changes) changes.track(this._applyState)

    // close existing state
    if (this._applyState) await this._applyState.close()

    const from = this.core.signedLength
    const store = this._viewStore.atomize()
    const views = this.fastForwardTo.views

    // mutating, prop fine as we are throwing it away immediately
    views.push({ key: this.fastForwardTo.key, length: this.fastForwardTo.length })

    const ffed = new Set()
    const migrated = !b4a.equals(this.fastForwardTo.key, this.core.key)

    for (const v of views) {
      const ref = await this._viewStore.findViewByKey(v.key, this.fastForwardTo.indexers)
      if (!ref) continue // unknown, view ignored
      ffed.add(ref)

      if (b4a.equals(ref.core.key, v.key)) {
        await ref.catchup(store.atom, v.length)
      } else {
        await this._applyFastForwardMigration(ref, v)
      }
    }

    // migrate zero length cores
    if (migrated) {
      const indexers = this.fastForwardTo.indexers
      const manifests = await this._viewStore.getIndexerManifests(indexers)

      for (const [name, ref] of this._viewStore.byName) {
        if (ffed.has(ref)) continue
        await this._migrateView(manifests, name, 0)
      }
    }

    const value = c.encode(messages.BootRecord, {
      key: this.fastForwardTo.key,
      indexedLength: this.fastForwardTo.length,
      indexersUpdated: false,
      fastForwarding: true,
      recoveries: RECOVERIES
    })

    const local = store.getLocal()
    await local.ready()
    await local.setUserData('autobase/boot', value)

    const tx = local.state.storage.write()
    // reset linearizer
    tx.deleteLocalRange(b4a.from([messages.LINEARIZER_PREFIX]), b4a.from([messages.LINEARIZER_PREFIX + 1]))
    await tx.flush()

    if (changes) changes.finalise()

    await store.flush()
    await store.close()

    const to = this.core.signedLength

    for (const ref of ffed) await ref.release()

    this.recoveries = RECOVERIES
    this.fastForwardTo = null
    this._queueFastForward()

    await this._clearWriters()

    this._applyState = new ApplyState(this)
    await this._applyState.ready()

    if (changes) await this._handlers.update(this._applyState.view, changes)

    if (await this._applyState.shouldMigrate()) {
      await this._migrate()
    } else {
      await this._makeLinearizerFromViewState()
      await this._applyState.catchup(this.linearizer)
    }

    if (!this.localWriter || this.localWriter.closed) {
      await this._updateLocalWriter(this._applyState.system)
    }

    this._caughtup = true

    this._rebooted()
    this.emit('fast-forward', to, from)
  }

  // TODO: not atomic in regards to the ff, fix that
  async _applyFastForwardMigration (ref, v) {
    const next = this.store.get(v.key)
    await next.ready()

    const prologue = next.manifest && next.manifest.prologue

    if (prologue && prologue.length > 0 && ref.core.length >= prologue.length) {
      try {
        await next.core.copyPrologue(ref.core.state)
      } catch {
        // we might be missing some nodes for this, just ignore, only an optimisation
      }
    }

    const batch = next.session({ name: 'batch', overwrite: true, checkout: v.length })
    await batch.ready()

    // remake the batch, reset from our prologue in case it replicated inbetween
    // TODO: we should really have an HC function for this

    await ref.batch.state.moveTo(batch, batch.length)
    await batch.close()

    ref.migrated(this, next)
  }

  async _migrateView (indexerManifests, name, indexedLength) {
    const ref = this._viewStore.byName.get(name)

    const core = ref.batch || ref.core
    await core.ready()

    const prologue = indexedLength === 0
      ? null
      : { length: indexedLength, hash: await core.treeHash(indexedLength) }

    const next = this._viewStore.getViewCore(indexerManifests, name, prologue)
    await next.ready()

    if (indexedLength > 0) {
      await next.core.copyPrologue(core.state)
    }

    // remake the batch, reset from our prologue in case it replicated inbetween
    // TODO: we should really have an HC function for this
    const batch = next.session({ name: 'batch', overwrite: true, checkout: indexedLength })
    await batch.ready()

    if (core.length > batch.length) {
      for (let i = batch.length; i < core.length; i++) {
        await batch.append(await core.get(i))
      }
    }

    await ref.batch.state.moveTo(batch, core.length)
    await batch.close()

    ref.migrated(this, next)

    return ref
  }

  async _migrate () {
    const length = this._applyState.indexedLength
    const system = this._applyState.system

    const info = await system.getIndexedInfo(length)
    const indexerManifests = await this._viewStore.getIndexerManifests(info.indexers)

    for (let i = 0; i < this._applyState.views.length; i++) {
      const view = this._applyState.views[i]
      const name = this._applyState.views[i].name

      const v = this._applyState.getViewFromSystem(view, info)
      const indexedLength = v ? v.length : 0

      await this._migrateView(indexerManifests, name, indexedLength)
    }

    const ref = await this._migrateView(indexerManifests, '_system', length)

    // start soft shutdown

    await this._clearWriters()
    await this._makeLinearizerFromViewState()

    await this._applyState.finalize(ref.core.key)

    this._applyState = new ApplyState(this)

    await this._applyState.ready()
    await this._applyState.catchup(this.linearizer)

    // end soft shutdown

    this._queueFastForward()

    this._rebooted()
  }

  _rebooted () {
    this.recouple()
    this._updateActivity()

    // ensure we re-evalute our state
    this._bootstrapWritersChanged = true
    this.updating = true
    this._queueBump()
  }

  _setLocalWriter (w) {
    this.localWriter = w
    if (this._ackInterval) this._startAckTimer()
  }

  _unsetLocalWriter () {
    if (!this.localWriter) return

    this._closeWriter(this.localWriter, true)
    if (this.localWriter.isActiveIndexer) this._clearLocalIndexer()

    this.localWriter = null
  }

  _setLocalIndexer () {
    assert(this.localWriter !== null)

    this.isIndexer = true
    this.emit('is-indexer')
  }

  _clearLocalIndexer () {
    assert(this.localWriter !== null)

    if (this._ackTimer) this._ackTimer.stop()

    this.isIndexer = false
    this._ackTimer = null

    this.emit('is-non-indexer')
  }

  _isLocalCore (core) {
    return core.writable && core.id === this.local.id
  }

  _addLocalHeads () {
    // not writable atm, will prop be writable in a subsequent bump tho
    if (!this.localWriter || this.localWriter.closed) return null
    // safety, localwriter is still processing, should prop be an assertion
    if (!this.localWriter.idle()) return null

    const length = this._optimistic === -1
      ? this._appending.length
      : this._optimistic || 1

    const nodes = new Array(length)
    for (let i = 0; i < length; i++) {
      const heads = this.linearizer.getHeads()
      const deps = new Set(this.linearizer.heads)
      const batch = this._appending.length - i
      const value = this._appending[i]

      const node = this.localWriter.append(value, heads, batch, deps, this.maxSupportedVersion, this._optimistic === 0)

      this.linearizer.addHead(node)
      nodes[i] = node
    }

    this._appended += length
    this._appending = length === this._appending.length ? null : this._appending.slice(length)

    if (this._optimistic > -1 && this._optimistic < length) this._optimistic = -1

    return nodes
  }

  async _addRemoteHeads () {
    let added = 0

    while (added < REMOTE_ADD_BATCH) {
      await this._updateAll()

      let advanced = 0

      for (const w of this.activeWriters) {
        let node = w.advance()
        if (node === null) continue

        advanced += node.batch

        while (true) {
          this.linearizer.addHead(node)
          if (node.batch === 1) break
          node = w.advance()
        }
      }

      if (advanced === 0) break
      added += advanced
    }

    return added
  }

  async _drain () {
    const writable = this.writable

    while (!this._interrupting && !this.paused) {
      if (this.fastForwardTo !== null) {
        await this._applyFastForward()
        continue // revaluate conditions...
      }

      // we defer this to post ready so its not blocking reading the views
      if (this._caughtup === false) {
        await this._catchupApplyState()
        continue
      }

      const remoteAdded = await this._addRemoteHeads()
      const localNodes = this._appending !== null ? this._addLocalHeads() : null

      if (this._interrupting) return

      if (remoteAdded > 0 || localNodes !== null) {
        this.updating = true
      }

      const u = this.linearizer.update()
      const indexersUpdated = u ? await this._applyState.update(u, localNodes) : false

      if (!indexersUpdated) {
        if (this._applyState.shouldFlush()) {
          await this._applyState.flush()
          this.updating = true
        }

        if (this._checkWriters.length > 0) {
          await this._gcWriters()
          continue // rerun the update loop as a writer might have been added
        }
        if (remoteAdded >= REMOTE_ADD_BATCH) continue
        break
      }

      await this._gcWriters()
      await this._migrate()
    }

    // emit state changes post drain
    if (writable !== this.writable) {
      if (this.writable && this._writable) this._writable.resolve(true)
      this.emit(writable ? 'unwritable' : 'writable')
    }
  }

  _wakeupPeer (stream) {
    if (!this.wakeupSession) return
    const wakeup = this._getWakeup()
    if (wakeup.length === 0) return
    this.wakeupSession.announceByStream(stream, wakeup)
  }

  _getWakeup () {
    const writers = []

    for (const w of this.activeWriters) {
      if (w.isActiveIndexer || w.flushed()) continue
      writers.push({ key: w.core.key, length: w.length })
    }

    return writers
  }

  async _wakeupWriter (key, length) {
    this._ensureWakeup(await this._getWriterByKey(key, -1, length, true, false, null))
  }

  // ensure wakeup on an existing writer (the writer calls this in addition to above)
  _ensureWakeup (w) {
    if (w === null || w.isBootstrap === true) return
    w.setBootstrap(true) // even if turn false at end of drain, hypercore makes them linger a bit so no churn
    this._bootstrapWriters.push(w)
    this._bootstrapWritersChanged = true
  }

  async _drainWakeup () {
    const promises = []

    // warmup all the below gets
    if (this._needsWakeup) {
      for (const { key } of this._wakeup) {
        if (this.activeWriters.has(key)) continue
        promises.push(this._applyState.system.get(key))
      }
      if (this._needsWakeupHeads) {
        for (const { key } of await this._applyState.system.heads) {
          if (this.activeWriters.has(key)) continue
          promises.push(this._applyState.system.get(key))
        }
      }
    }
    for (const [hex, length] of this._wakeupHints) {
      const key = b4a.from(hex, 'hex')
      if (length !== -1) {
        const w = this.activeWriters.get(key)
        if (w) {
          if (w.length < length) w.seen(length)
          continue
        }
      }
      promises.push(this._applyState.system.get(key))
    }

    await Promise.allSettled(promises)

    if (this._needsWakeup === true) {
      this._needsWakeup = false

      for (const { key } of this._wakeup) {
        if (this.activeWriters.has(key)) continue
        await this._wakeupWriter(key, 0)
      }

      if (this._needsWakeupHeads === true) {
        this._needsWakeupHeads = false

        for (const { key } of await this._applyState.system.heads) {
          if (this.activeWriters.has(key)) continue
          await this._wakeupWriter(key, 0)
        }
      }
    }

    for (const [hex, length] of this._wakeupHints) {
      const key = b4a.from(hex, 'hex')
      if (this.activeWriters.has(key)) continue
      if (length !== -1) {
        const info = await this._applyState.system.get(key)
        if (info && length <= info.length) continue // stale hint
      }
      await this._wakeupWriter(key, length === -1 ? 0 : length)
    }

    this._wakeupHints.clear()
  }

  pause () {
    this.paused = true
  }

  resume () {
    this.paused = false
    this._queueBump()
  }

  waitForWritable () {
    if (this.writable) return Promise.resolve(true)
    if (!this._writable) this._writable = rrp()
    return this._writable.promise
  }

  async _advance () {
    if (this.opened === false) await this.ready()
    if (this.paused || this._interrupting) return

    this._draining = true

    if (this.recoveries < FF_RECOVERY || this._bootRecovery) {
      await this._recoverMaybe()
    }

    const local = this.local.length

    try {
      await this._drain()

      // must run post drain so the linearizer is caught up
      if (this._caughtup && (this._needsWakeup === true || this._wakeupHints.size > 0)) await this._drainWakeup()

      // check if we should update local writer
      if (!this.localWriter || this.localWriter.closed) {
        await this._updateLocalWriter(this._applyState.system)
      }

      this._draining = false
    } catch (err) {
      this._onError(err)
      return
    }

    if (this._interrupting) return

    if (this.localWriter && !this.localWriter.closed) {
      if (this._applyState.isLocalPendingIndexer()) this.ack().catch(noop)
      else if (this._triggerAckAsap()) this._ackTimer.asap()
    }

    // keep bootstraps in sync with linearizer
    if (this.updating === true || this._bootstrapWritersChanged === true) {
      this._updateBootstrapWriters()
    }

    if (this.updating === true) {
      this.updating = false

      if (local !== this.local.length) this._resetAckTick()
      else this._ackTick++

      if (!this._interrupting) this.emit('update')
      this._waiting.notify(null)
    }

    if (!this._interrupting) await this._gcWriters()
  }

  _triggerAckAsap () {
    if (!this._ackTimer) return false

    // flush if threshold is reached and we are not already acking
    if (this._ackTickThreshold && !this._acking && this._ackTick >= this._ackTickThreshold) {
      if (this._ackTimer) {
        for (const w of this.linearizer.indexers) {
          if (w.core.length > w.length) return false // wait for the normal ack cycle in this case
        }
      }

      return true
    }

    return false
  }

  _queueFastForward () {
    if (!this.core.opened) return
    // should have a better way to get this
    const latestSignedLength = this.core.core.state.length

    if (!this.fastForwardEnabled || this.fastForwarding !== null || this._interrupting) return
    if (latestSignedLength - this.core.length < FastForward.MINIMUM) return
    if (this.fastForwardTo !== null) return
    if ((Date.now() - this.fastForwardFailedAt) < MIN_FF_WAIT) return

    this._runFastForward(new FastForward(this, this.core.key)).catch(noop)
  }

  _queueStaticFastForward (key) {
    if (!this.fastForwardEnabled || this.fastForwarding !== null || this._interrupting) return
    if (this.fastForwardTo !== null) return
    if ((Date.now() - this.fastForwardFailedAt) < MIN_FF_WAIT) return

    this._runFastForward(new FastForward(this, key, { verified: false })).catch(noop)
  }

  _updateActivity () {
    this.activeWriters.updateActivity()
    if (this._applyState) {
      if (this.isFastForwarding()) this._applyState.pause()
      else this._applyState.resume()
    }
  }

  async _runFastForward (ff) {
    this.fastForwarding = ff

    this._updateActivity()

    const result = await ff.upgrade()
    await ff.close()

    if (this.fastForwarding === ff) this.fastForwarding = null

    if (!result) {
      if (ff.failed) this.fastForwardFailedAt = Date.now()
      else this._queueFastForward()
      this._updateActivity()
      return
    }

    this.fastForwardFailedAt = 0
    this.fastForwardTo = result

    this._bumpAckTimer()
    this._queueBump()
  }

  async _closeAllActiveWriters () {
    for (const w of this.activeWriters) {
      if (this.localWriter === w) continue
      await this._closeWriter(w, true)
    }
  }

  // triggered from apply
  async _addWriter (key, sys) { // just compat for old version
    assert(this._applyState.applying, 'System changes are only allowed in apply')

    const writer = (await this._getWriterByKey(key, -1, 0, false, true, sys)) || this._makeWriter(key, 0, true, false)
    await writer.ready()

    if (!this.activeWriters.has(key)) {
      this.activeWriters.add(writer)
      this._checkWriters.push(writer)
      this._ensureWakeup(writer)
    }

    // fetch any nodes needed for dependents
    this._queueBump()
  }

  // triggered from apply
  _removeWriter (key) { // just compat for old version
    const w = this.activeWriters.get(key)
    if (w) w.isRemoved = true

    this._queueBump()
  }

  removeable (key) {
    return this._applyState ? this._applyState.removeable(key) : false
  }

  _updateAckThreshold () {
    if (this._ackThreshold === 0) return
    if (this._ackTimer) this._ackTimer.bau()
    this._ackTick = 0
    this._ackTickThreshold = random2over1(this.linearizer.indexers.length * this._ackThreshold)
  }

  _resetAckTick () {
    this._ackTick = 0
    if (this._ackTimer) this._ackTimer.bau()
  }

  _shiftWriter (w) {
    w.shift()
    if (w.flushed()) this._checkWriters.push(w)
  }
}

function toKey (k) {
  return b4a.isBuffer(k) ? k : hypercoreId.decode(k)
}

function isAutobaseMessage (msg) {
  return msg.checkpoint ? msg.checkpoint.length > 0 : msg.checkpoint === null
}

function compareNodes (a, b) {
  return b4a.compare(a.key, b.key)
}

function random2over1 (n) {
  return Math.floor(n + Math.random() * n)
}

function noop () {}

function crashSoon (err) {
  queueMicrotask(() => { throw err })
  throw err
}

function isObject (obj) {
  return typeof obj === 'object' && obj !== null
}

function emitWarning (err) {
  safetyCatch(err)
  this.emit('warning', err)
}

function normalize (valueEncoding, value) {
  const state = { buffer: null, start: 0, end: 0 }
  valueEncoding.preencode(state, value)
  state.buffer = b4a.allocUnsafe(state.end)
  valueEncoding.encode(state, value)
  state.start = 0
  return valueEncoding.decode(state)
}