index.js

'use strict'

const { EventEmitter } = require('events')
const errcode = require('err-code')

const libp2pRecord = require('libp2p-record')
const { MemoryDatastore } = require('interface-datastore')
const PeerInfo = require('peer-info')

const RoutingTable = require('./routing')
const utils = require('./utils')
const c = require('./constants')
const Network = require('./network')
const contentFetching = require('./content-fetching')
const contentRouting = require('./content-routing')
const peerRouting = require('./peer-routing')
const Message = require('./message')
const Providers = require('./providers')
const RandomWalk = require('./random-walk')
const QueryManager = require('./query-manager')

const Record = libp2pRecord.Record

/**
 * A DHT implementation modeled after Kademlia with S/Kademlia modifications.
 * Original implementation in go: https://github.com/libp2p/go-libp2p-kad-dht.
 */
class KadDHT extends EventEmitter {
  /**
   * Random walk options
   * @typedef {Object} randomWalkOptions
   * @property {boolean} enabled discovery enabled (default: true)
   * @property {number} queriesPerPeriod how many queries to run per period (default: 1)
   * @property {number} interval how often to run the the random-walk process, in milliseconds (default: 300000)
   * @property {number} timeout how long to wait for the the random-walk query to run, in milliseconds (default: 30000)
   * @property {number} delay how long to wait before starting the first random walk, in milliseconds (default: 10000)
   */

  /**
   * Create a new KadDHT.
   * @param {Object} props
   * @param {Dialer} props.dialer libp2p dialer instance
   * @param {PeerInfo} props.peerInfo peer's peerInfo
   * @param {PeerStore} props.peerStore libp2p peerStore
   * @param {Object} props.registrar libp2p registrar instance
   * @param {function} props.registrar.handle
   * @param {function} props.registrar.register
   * @param {function} props.registrar.unregister
   * @param {number} props.kBucketSize k-bucket size (default 20)
   * @param {number} props.concurrency alpha concurrency of queries (default 3)
   * @param {Datastore} props.datastore datastore (default MemoryDatastore)
   * @param {object} props.validators validators object with namespace as keys and function(key, record, callback)
   * @param {object} props.selectors selectors object with namespace as keys and function(key, records)
   * @param {randomWalkOptions} options.randomWalk randomWalk options
   */
  constructor ({
    dialer,
    peerInfo,
    peerStore,
    registrar,
    datastore = new MemoryDatastore(),
    kBucketSize = c.K,
    concurrency = c.ALPHA,
    validators = {},
    selectors = {},
    randomWalk = {}
  }) {
    super()

    if (!dialer) {
      throw new Error('libp2p-kad-dht requires an instance of Dialer')
    }

    /**
     * Local reference to the libp2p dialer instance
     * @type {Dialer}
     */
    this.dialer = dialer

    /**
     * Local peer info
     * @type {PeerInfo}
     */
    this.peerInfo = peerInfo

    /**
     * Local PeerStore
     * @type {PeerStore}
     */
    this.peerStore = peerStore

    /**
     * Local peer info
     * @type {Registrar}
     */
    this.registrar = registrar

    /**
     * k-bucket size
     *
     * @type {number}
     */
    this.kBucketSize = kBucketSize

    /**
     * ALPHA concurrency at which each query path with run, defaults to 3
     * @type {number}
     */
    this.concurrency = concurrency

    /**
     * Number of disjoint query paths to use
     * This is set to `kBucketSize`/2 per the S/Kademlia paper
     * @type {number}
     */
    this.disjointPaths = Math.ceil(this.kBucketSize / 2)

    /**
     * The routing table.
     *
     * @type {RoutingTable}
     */
    this.routingTable = new RoutingTable(this.peerInfo.id, this.kBucketSize)

    /**
     * Reference to the datastore, uses an in-memory store if none given.
     *
     * @type {Datastore}
     */
    this.datastore = datastore

    /**
     * Provider management
     *
     * @type {Providers}
     */
    this.providers = new Providers(this.datastore, this.peerInfo.id)

    this.validators = {
      pk: libp2pRecord.validator.validators.pk,
      ...validators
    }

    this.selectors = {
      pk: libp2pRecord.selection.selectors.pk,
      ...selectors
    }

    this.network = new Network(this)

    this._log = utils.logger(this.peerInfo.id)

    /**
     * Random walk management
     *
     * @type {RandomWalk}
     */
    this.randomWalk = new RandomWalk(this, randomWalk)

    /**
     * Keeps track of running queries
     *
     * @type {QueryManager}
     */
    this._queryManager = new QueryManager()

    this._running = false

    // DHT components
    this.contentFetching = contentFetching(this)
    this.contentRouting = contentRouting(this)
    this.peerRouting = peerRouting(this)
  }

  /**
   * Is this DHT running.
   * @type {bool}
   */
  get isStarted () {
    return this._running
  }

  /**
   * Start listening to incoming connections.
   * @returns {Promise<void>}
   */
  async start () {
    this._running = true
    this._queryManager.start()
    await this.network.start()

    // Start random walk, it will not run if it's disabled
    this.randomWalk.start()
  }

  /**
   * Stop accepting incoming connections and sending outgoing
   * messages.
   * @returns {Promise<void>}
   */
  stop () {
    this._running = false
    this.randomWalk.stop()
    this.providers.stop()
    this._queryManager.stop()
    return this.network.stop()
  }

  /**
   * Store the given key/value  pair in the DHT.
   * @param {Buffer} key
   * @param {Buffer} value
   * @param {Object} [options] - put options
   * @param {number} [options.minPeers] - minimum number of peers required to successfully put (default: closestPeers.length)
   * @returns {Promise<void>}
   */
  async put (key, value, options = {}) { // eslint-disable-line require-await
    return this.contentFetching.put(key, value, options)
  }

  /**
   * Get the value to the given key.
   * Times out after 1 minute by default.
   * @param {Buffer} key
   * @param {Object} [options] - get options
   * @param {number} [options.timeout] - optional timeout (default: 60000)
   * @returns {Promise<Buffer>}
   */
  async get (key, options = {}) { // eslint-disable-line require-await
    return this.contentFetching.get(key, options)
  }

  /**
   * Get the `n` values to the given key without sorting.
   * @param {Buffer} key
   * @param {number} nvals
   * @param {Object} [options] - get options
   * @param {number} [options.timeout] - optional timeout (default: 60000)
   * @returns {Promise<Array<{from: PeerId, val: Buffer}>>}
   */
  async getMany (key, nvals, options = {}) { // eslint-disable-line require-await
    return this.contentFetching.getMany(key, nvals, options)
  }

  // ----------- Content Routing

  /**
   * Announce to the network that we can provide given key's value.
   * @param {CID} key
   * @returns {Promise<void>}
   */
  async provide (key) { // eslint-disable-line require-await
    return this.contentRouting.provide(key)
  }

  /**
   * Search the dht for up to `K` providers of the given CID.
   * @param {CID} key
   * @param {Object} options - findProviders options
   * @param {number} options.timeout - how long the query should maximally run, in milliseconds (default: 60000)
   * @param {number} options.maxNumProviders - maximum number of providers to find
   * @returns {AsyncIterable<PeerInfo>}
   */
  async * findProviders (key, options = {}) {
    for await (const pInfo of this.contentRouting.findProviders(key, options)) {
      yield pInfo
    }
  }

  // ----------- Peer Routing -----------

  /**
   * Search for a peer with the given ID.
   *
   * @param {PeerId} id
   * @param {Object} options - findPeer options
   * @param {number} options.timeout - how long the query should maximally run, in milliseconds (default: 60000)
   * @returns {Promise<PeerInfo>}
   */
  async findPeer (id, options = {}) { // eslint-disable-line require-await
    return this.peerRouting.findPeer(id, options)
  }

  /**
   * Kademlia 'node lookup' operation.
   * @param {Buffer} key
   * @param {Object} [options]
   * @param {boolean} [options.shallow] shallow query (default: false)
   * @returns {AsyncIterable<PeerId>}
   */
  async * getClosestPeers (key, options = { shallow: false }) {
    for await (const pId of this.peerRouting.getClosestPeers(key, options)) {
      yield pId
    }
  }

  /**
   * Get the public key for the given peer id.
   * @param {PeerId} peer
   * @returns {Promise<PubKey>}
   */
  async getPublicKey (peer) { // eslint-disable-line require-await
    return this.peerRouting.getPublicKey(peer)
  }

  // ----------- Discovery -----------

  _peerDiscovered (peerId, multiaddrs) {
    this.emit('peer', {
      id: peerId,
      multiaddrs
    })
  }

  // ----------- Internals -----------

  /**
   * Returns the routing tables closest peers, for the key of
   * the message.
   *
   * @param {Message} msg
   * @returns {Promise<Array<PeerInfo>>}
   * @private
   */
  async _nearestPeersToQuery (msg) {
    const key = await utils.convertBuffer(msg.key)
    const ids = this.routingTable.closestPeers(key, this.kBucketSize)

    return ids.map((p) => {
      const peer = this.peerStore.get(p)
      const peerInfo = new PeerInfo(p)

      if (peer) {
        peer.protocols.forEach((p) => peerInfo.protocols.add(p))
        peer.multiaddrInfos.forEach((mi) => peerInfo.multiaddrs.add(mi.multiaddr))
      }

      return peerInfo
    })
  }

  /**
   * Get the nearest peers to the given query, but iff closer
   * than self.
   *
   * @param {Message} msg
   * @param {PeerInfo} peer
   * @returns {Promise<Array<PeerInfo>>}
   * @private
   */

  async _betterPeersToQuery (msg, peer) {
    this._log('betterPeersToQuery')
    const closer = await this._nearestPeersToQuery(msg)

    return closer.filter((closer) => {
      if (this._isSelf(closer.id)) {
        // Should bail, not sure
        this._log.error('trying to return self as closer')
        return false
      }

      return !closer.id.isEqual(peer.id)
    })
  }

  /**
   * Try to fetch a given record by from the local datastore.
   * Returns the record iff it is still valid, meaning
   * - it was either authored by this node, or
   * - it was received less than `MAX_RECORD_AGE` ago.
   *
   * @param {Buffer} key
   * @returns {Promise<Record>}
   * @private
   */

  async _checkLocalDatastore (key) {
    this._log('checkLocalDatastore: %b', key)
    const dsKey = utils.bufferToKey(key)

    // Fetch value from ds
    let rawRecord
    try {
      rawRecord = await this.datastore.get(dsKey)
    } catch (err) {
      if (err.code === 'ERR_NOT_FOUND') {
        return undefined
      }
      throw err
    }

    // Create record from the returned bytes
    const record = Record.deserialize(rawRecord)

    if (!record) {
      throw errcode('Invalid record', 'ERR_INVALID_RECORD')
    }

    // Check validity: compare time received with max record age
    if (record.timeReceived == null ||
      utils.now() - record.timeReceived > c.MAX_RECORD_AGE) {
      // If record is bad delete it and return
      await this.datastore.delete(dsKey)
      return undefined
    }

    // Record is valid
    return record
  }

  /**
   * Add the peer to the routing table and update it in the peerStore.
   *
   * @param {PeerInfo} peer
   * @returns {Promise<void>}
   * @private
   */

  async _add (peer) {
    await this.routingTable.add(peer.id)
  }

  /**
   * Verify a record without searching the DHT.
   *
   * @param {Record} record
   * @returns {Promise<void>}
   * @private
   */

  async _verifyRecordLocally (record) {
    this._log('verifyRecordLocally')

    await libp2pRecord.validator.verifyRecord(this.validators, record)
  }

  /**
   * Is the given peer id our PeerId?
   *
   * @param {PeerId} other
   * @returns {bool}
   *
   * @private
   */

  _isSelf (other) {
    return other && this.peerInfo.id.id.equals(other.id)
  }

  /**
   * Store the given key/value pair at the peer `target`.
   *
   * @param {Buffer} key
   * @param {Buffer} rec - encoded record
   * @param {PeerId} target
   * @returns {Promise<void>}
   *
   * @private
   */

  async _putValueToPeer (key, rec, target) {
    const msg = new Message(Message.TYPES.PUT_VALUE, key, 0)
    msg.record = rec

    const resp = await this.network.sendRequest(target, msg)

    if (!resp.record.value.equals(Record.deserialize(rec).value)) {
      throw errcode(new Error('value not put correctly'), 'ERR_PUT_VALUE_INVALID')
    }
  }

  /**
   * Query a particular peer for the value for the given key.
   * It will either return the value or a list of closer peers.
   *
   * Note: The peerStore is updated with new addresses found for the given peer.
   *
   * @param {PeerId} peer
   * @param {Buffer} key
   * @returns {Promise<{Record, Array<PeerInfo}>}
   * @private
   */

  async _getValueOrPeers (peer, key) {
    const msg = await this._getValueSingle(peer, key)

    const peers = msg.closerPeers
    const record = msg.record

    if (record) {
      // We have a record
      try {
        await this._verifyRecordOnline(record)
      } catch (err) {
        const errMsg = 'invalid record received, discarded'
        this._log(errMsg)
        throw errcode(new Error(errMsg), 'ERR_INVALID_RECORD')
      }

      return { record, peers }
    }

    if (peers.length > 0) {
      return { peers }
    }

    throw errcode(new Error('Not found'), 'ERR_NOT_FOUND')
  }

  /**
   * Get a value via rpc call for the given parameters.
   *
   * @param {PeerId} peer
   * @param {Buffer} key
   * @returns {Promise<Message>}
   * @private
   */

  async _getValueSingle (peer, key) { // eslint-disable-line require-await
    const msg = new Message(Message.TYPES.GET_VALUE, key, 0)
    return this.network.sendRequest(peer, msg)
  }

  /**
   * Verify a record, fetching missing public keys from the network.
   * Calls back with an error if the record is invalid.
   *
   * @param {Record} record
   * @returns {Promise<void>}
   * @private
   */

  async _verifyRecordOnline (record) {
    await libp2pRecord.validator.verifyRecord(this.validators, record)
  }
}

module.exports = KadDHT
module.exports.multicodec = c.PROTOCOL_DHT