text: 'use strict'
  
  const { EventEmitter } = require('events')
  const libp2pRecord = require('libp2p-record')
  const MemoryStore = require('interface-datastore').MemoryDatastore
  const waterfall = require('async/waterfall')
  const each = require('async/each')
  const filter = require('async/filter')
  const timeout = require('async/timeout')
  const PeerId = require('peer-id')
  const PeerInfo = require('peer-info')
  const crypto = require('libp2p-crypto')
  const promiseToCallback = require('promise-to-callback')
  
  const errcode = require('err-code')
  
  const RoutingTable = require('./routing')
  const utils = require('./utils')
  const c = require('./constants')
  const Query = require('./query')
  const Network = require('./network')
  const privateApi = require('./private')
  const Providers = require('./providers')
  const Message = require('./message')
  const RandomWalk = require('./random-walk')
  const QueryManager = require('./query-manager')
  const assert = require('assert')
  
  /**
   * 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 {Switch} sw libp2p-switch instance
     * @param {object} options DHT options
     * @param {number} options.kBucketSize k-bucket size (default 20)
     * @param {number} options.concurrency alpha concurrency of queries (default 3)
     * @param {Datastore} options.datastore datastore (default MemoryDatastore)
     * @param {object} options.validators validators object with namespace as keys and function(key, record, callback)
     * @param {object} options.selectors selectors object with namespace as keys and function(key, records)
     * @param {randomWalkOptions} options.randomWalk randomWalk options
     */
    constructor (sw, options) {
      super()
      assert(sw, 'libp2p-kad-dht requires a instance of Switch')
      options = options || {}
      options.validators = options.validators || {}
      options.selectors = options.selectors || {}
  
      /**
       * Local reference to the libp2p-switch instance
       *
       * @type {Switch}
       */
      this.switch = sw
  
      /**
       * k-bucket size, defaults to 20
       *
       * @type {number}
       */
      this.kBucketSize = options.kBucketSize || c.K
  
      /**
       * ALPHA concurrency at which each query path with run, defaults to 3
       * @type {number}
       */
      this.concurrency = options.concurrency || c.ALPHA
  
      /**
       * 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 = options.datastore || new MemoryStore()
  
      /**
       * Provider management
       *
       * @type {Providers}
       */
      this.providers = new Providers(this.datastore, this.peerInfo.id)
  
      this.validators = {
        pk: libp2pRecord.validator.validators.pk,
        ...options.validators
      }
  
      this.selectors = {
        pk: libp2pRecord.selection.selectors.pk,
        ...options.selectors
      }
  
      this.network = new Network(this)
  
      this._log = utils.logger(this.peerInfo.id)
  
      // Inject private apis so we don't clutter up this file
      const pa = privateApi(this)
      Object.keys(pa).forEach((name) => { this[name] = pa[name] })
  
      /**
       * Random walk management
       *
       * @type {RandomWalk}
       */
      this.randomWalk = new RandomWalk(this, options.randomWalk)
  
      /**
       * Keeps track of running queries
       *
       * @type {QueryManager}
       */
      this._queryManager = new QueryManager()
    }
  
    /**
     * Is this DHT running.
     *
     * @type {bool}
     */
    get isStarted () {
      return this._running
    }
  
    /**
     * Start listening to incoming connections.
     *
     * @param {function(Error)} callback
     * @returns {void}
     */
    start (callback) {
      this._running = true
      this._queryManager.start()
      this.network.start((err) => {
        if (err) {
          return callback(err)
        }
  
        // Start random walk, it will not run if it's disabled
        this.randomWalk.start()
        callback()
      })
    }
  
    /**
     * Stop accepting incoming connections and sending outgoing
     * messages.
     *
     * @param {function(Error)} callback
     * @returns {void}
     */
    stop (callback) {
      this._running = false
      this.randomWalk.stop()
      this.providers.stop()
      this._queryManager.stop()
      this.network.stop(callback)
    }
  
    /**
     * Local peer (yourself)
     *
     * @type {PeerInfo}
     */
    get peerInfo () {
      return this.switch._peerInfo
    }
  
    get peerBook () {
      return this.switch._peerBook
    }
  
    /**
     * Store the given key/value  pair in the DHT.
     *
     * @param {Buffer} key
     * @param {Buffer} value
     * @param {Object} options - get options
     * @param {number} options.minPeers - minimum peers that must be put to to consider this a successful operation
     * (default: closestPeers.length)
     * @param {function(Error)} callback
     * @returns {void}
     */
    put (key, value, options, callback) {
      if (typeof options === 'function') {
        callback = options
        options = {}
      } else {
        options = options || {}
      }
  
      this._log('PutValue %b', key)
  
      waterfall([
        (cb) => utils.createPutRecord(key, value, cb),
        (rec, cb) => waterfall([
          (cb) => this._putLocal(key, rec, cb),
          (cb) => this.getClosestPeers(key, { shallow: true }, cb),
          (peers, cb) => {
            // Ensure we have a default `minPeers`
            options.minPeers = options.minPeers || peers.length
            // filter out the successful puts
            filter(peers, (peer, cb) => {
              this._putValueToPeer(key, rec, peer, (err) => {
                if (err) {
                  this._log.error('Failed to put to peer (%b): %s', peer.id, err)
                  return cb(null, false)
                }
                cb(null, true)
              })
            }, (err, results) => {
              if (err) return cb(err)
  
              // Did we put to enough peers?
              if (options.minPeers > results.length) {
                const error = errcode(new Error('Failed to put value to enough peers'), 'ERR_NOT_ENOUGH_PUT_PEERS')
                this._log.error(error)
                return cb(error)
              }
  
              cb()
            })
          }
        ], cb)
      ], callback)
    }
  
    /**
     * Get the value to the given key.
     * Times out after 1 minute.
     *
     * @param {Buffer} key
     * @param {Object} options - get options
     * @param {number} options.timeout - optional timeout (default: 60000)
     * @param {function(Error, Buffer)} callback
     * @returns {void}
     */
    get (key, options, callback) {
      if (typeof options === 'function') {
        callback = options
        options = {}
      } else {
        options = options || {}
      }
  
      if (!options.maxTimeout && !options.timeout) {
        options.timeout = c.minute // default
      } else if (options.maxTimeout && !options.timeout) { // TODO this will be deprecated in a next release
        options.timeout = options.maxTimeout
      }
  
      this._get(key, options, callback)
    }
  
    /**
     * 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)
     * @param {function(Error, Array<{from: PeerId, val: Buffer}>)} callback
     * @returns {void}
     */
    getMany (key, nvals, options, callback) {
      if (typeof options === 'function') {
        callback = options
        options = {}
      } else {
        options = options || {}
      }
  
      if (!options.maxTimeout && !options.timeout) {
        options.timeout = c.minute // default
      } else if (options.maxTimeout && !options.timeout) { // TODO this will be deprecated in a next release
        options.timeout = options.maxTimeout
      }
  
      this._log('getMany %b (%s)', key, nvals)
      let vals = []
  
      this._getLocal(key, (err, localRec) => {
        if (err && nvals === 0) {
          return callback(err)
        }
  
        if (err == null) {
          vals.push({
            val: localRec.value,
            from: this.peerInfo.id
          })
        }
  
        if (vals.length >= nvals) {
          return callback(null, vals)
        }
  
        const paths = []
        waterfall([
          (cb) => utils.convertBuffer(key, cb),
          (id, cb) => {
            const rtp = this.routingTable.closestPeers(id, this.kBucketSize)
  
            this._log('peers in rt: %d', rtp.length)
            if (rtp.length === 0) {
              const errMsg = 'Failed to lookup key! No peers from routing table!'
  
              this._log.error(errMsg)
              return cb(errcode(new Error(errMsg), 'ERR_NO_PEERS_IN_ROUTING_TABLE'))
            }
  
            // we have peers, lets do the actual query to them
            const query = new Query(this, key, (pathIndex, numPaths) => {
              // This function body runs once per disjoint path
              const pathSize = utils.pathSize(nvals - vals.length, numPaths)
              const pathVals = []
              paths.push(pathVals)
  
              // Here we return the query function to use on this particular disjoint path
              return async (peer) => {
                let rec, peers, lookupErr
                try {
                  const results = await this._getValueOrPeersAsync(peer, key)
                  rec = results.record
                  peers = results.peers
                } catch (err) {
                  // If we have an invalid record we just want to continue and fetch a new one.
                  if (err.code !== 'ERR_INVALID_RECORD') {
                    throw err
                  }
                  lookupErr = err
                }
  
                const res = { closerPeers: peers }
  
                if ((rec && rec.value) || lookupErr) {
                  pathVals.push({
                    val: rec && rec.value,
                    from: peer
                  })
                }
  
                // enough is enough
                if (pathVals.length >= pathSize) {
                  res.pathComplete = true
                }
  
                return res
              }
            })
  
            // run our query
            timeout((_cb) => {
              promiseToCallback(query.run(rtp))(_cb)
            }, options.timeout)((err, res) => {
              query.stop()
              cb(err, res)
            })
          }
        ], (err) => {
          // combine vals from each path
          vals = [].concat.apply(vals, paths).slice(0, nvals)
  
          if (err && vals.length === 0) {
            return callback(err)
          }
  
          callback(null, vals)
        })
      })
    }
  
    /**
     * Kademlia 'node lookup' operation.
     *
     * @param {Buffer} key
     * @param {Object} options
     * @param {boolean} options.shallow shallow query
     * @param {function(Error, Array<PeerId>)} callback
     * @returns {void}
     */
    getClosestPeers (key, options, callback) {
      this._log('getClosestPeers to %b', key)
  
      if (typeof options === 'function') {
        callback = options
        options = {
          shallow: false
        }
      }
  
      utils.convertBuffer(key, (err, id) => {
        if (err) {
          return callback(err)
        }
  
        const tablePeers = this.routingTable.closestPeers(id, this.kBucketSize)
  
        const q = new Query(this, key, () => {
          // There is no distinction between the disjoint paths,
          // so there are no per-path variables in this scope.
          // Just return the actual query function.
          return async (peer) => {
            const closer = await this._closerPeersSingleAsync(key, peer)
            return {
              closerPeers: closer,
              pathComplete: options.shallow ? true : undefined
            }
          }
        })
  
        promiseToCallback(q.run(tablePeers))((err, res) => {
          if (err) {
            return callback(err)
          }
  
          if (!res || !res.finalSet) {
            return callback(null, [])
          }
  
          waterfall([
            (cb) => utils.sortClosestPeers(Array.from(res.finalSet), id, cb),
            (sorted, cb) => cb(null, sorted.slice(0, this.kBucketSize))
          ], callback)
        })
      })
    }
  
    /**
     * Get the public key for the given peer id.
     *
     * @param {PeerId} peer
     * @param {function(Error, PubKey)} callback
     * @returns {void}
     */
    getPublicKey (peer, callback) {
      this._log('getPublicKey %s', peer.toB58String())
      // local check
      let info
      if (this.peerBook.has(peer)) {
        info = this.peerBook.get(peer)
  
        if (info && info.id.pubKey) {
          this._log('getPublicKey: found local copy')
          return callback(null, info.id.pubKey)
        }
      } else {
        info = this.peerBook.put(new PeerInfo(peer))
      }
      // try the node directly
      this._getPublicKeyFromNode(peer, (err, pk) => {
        if (!err) {
          info.id = new PeerId(peer.id, null, pk)
          this.peerBook.put(info)
  
          return callback(null, pk)
        }
  
        // dht directly
        const pkKey = utils.keyForPublicKey(peer)
        this.get(pkKey, (err, value) => {
          if (err) {
            return callback(err)
          }
  
          const pk = crypto.unmarshalPublicKey(value)
          info.id = new PeerId(peer, null, pk)
          this.peerBook.put(info)
  
          callback(null, pk)
        })
      })
    }
  
    /**
     * Look if we are connected to a peer with the given id.
     * Returns the `PeerInfo` for it, if found, otherwise `undefined`.
     *
     * @param {PeerId} peer
     * @param {function(Error, PeerInfo)} callback
     * @returns {void}
     */
    findPeerLocal (peer, callback) {
      this._log('findPeerLocal %s', peer.toB58String())
      this.routingTable.find(peer, (err, p) => {
        if (err) {
          return callback(err)
        }
        if (!p || !this.peerBook.has(p)) {
          return callback()
        }
        callback(null, this.peerBook.get(p))
      })
    }
  
    // ----------- Content Routing
  
    /**
     * Announce to the network that we can provide given key's value.
     *
     * @param {CID} key
     * @param {function(Error)} callback
     * @returns {void}
     */
    provide (key, callback) {
      this._log('provide: %s', key.toBaseEncodedString())
  
      const errors = []
      waterfall([
        // TODO: refactor this in method in async and remove this wrapper
        (cb) => promiseToCallback(this.providers.addProvider(key, this.peerInfo.id))(err => cb(err)),
        (cb) => this.getClosestPeers(key.buffer, cb),
        (peers, cb) => {
          const msg = new Message(Message.TYPES.ADD_PROVIDER, key.buffer, 0)
          msg.providerPeers = [this.peerInfo]
  
          each(peers, (peer, cb) => {
            this._log('putProvider %s to %s', key.toBaseEncodedString(), peer.toB58String())
            this.network.sendMessage(peer, msg, (err) => {
              if (err) errors.push(err)
              cb()
            })
          }, cb)
        }
      ], (err) => {
        if (errors.length) {
          // This should be infrequent. This means a peer we previously connected
          // to failed to exchange the provide message. If getClosestPeers was an
          // iterator, we could continue to pull until we announce to kBucketSize peers.
          err = errcode(`Failed to provide to ${errors.length} of ${this.kBucketSize} peers`, 'ERR_SOME_PROVIDES_FAILED', { errors })
        }
        callback(err)
      })
    }
  
    /**
     * 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
     * @param {function(Error, Array<PeerInfo>)} callback
     * @returns {void}
     */
    findProviders (key, options, callback) {
      if (typeof options === 'function') {
        callback = options
        options = {}
      } else {
        options = options || {}
      }
  
      if (!options.maxTimeout && !options.timeout) {
        options.timeout = c.minute // default
      } else if (options.maxTimeout && !options.timeout) { // TODO this will be deprecated in a next release
        options.timeout = options.maxTimeout
      }
  
      options.maxNumProviders = options.maxNumProviders || c.K
  
      this._log('findProviders %s', key.toBaseEncodedString())
      this._findNProviders(key, options.timeout, options.maxNumProviders, callback)
    }
  
    // ----------- 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)
     * @param {function(Error, PeerInfo)} callback
     * @returns {void}
     */
    findPeer (id, options, callback) {
      if (typeof options === 'function') {
        callback = options
        options = {}
      } else {
        options = options || {}
      }
  
      if (!options.maxTimeout && !options.timeout) {
        options.timeout = c.minute // default
      } else if (options.maxTimeout && !options.timeout) { // TODO this will be deprecated in a next release
        options.timeout = options.maxTimeout
      }
  
      this._log('findPeer %s', id.toB58String())
  
      this.findPeerLocal(id, (err, pi) => {
        if (err) {
          return callback(err)
        }
  
        // already got it
        if (pi != null) {
          this._log('found local')
          return callback(null, pi)
        }
  
        waterfall([
          (cb) => utils.convertPeerId(id, cb),
          (key, cb) => {
            const peers = this.routingTable.closestPeers(key, this.kBucketSize)
  
            if (peers.length === 0) {
              return cb(errcode(new Error('Peer lookup failed'), 'ERR_LOOKUP_FAILED'))
            }
  
            // sanity check
            const match = peers.find((p) => p.isEqual(id))
            if (match && this.peerBook.has(id)) {
              this._log('found in peerbook')
              return cb(null, this.peerBook.get(id))
            }
  
            // query the network
            const query = new Query(this, id.id, () => {
              // There is no distinction between the disjoint paths,
              // so there are no per-path variables in this scope.
              // Just return the actual query function.
              return async (peer) => {
                const msg = await this._findPeerSingleAsync(peer, id)
                const match = msg.closerPeers.find((p) => p.id.isEqual(id))
  
                // found it
                if (match) {
                  return {
                    peer: match,
                    queryComplete: true
                  }
                }
  
                return {
                  closerPeers: msg.closerPeers
                }
              }
            })
  
            timeout((_cb) => {
              promiseToCallback(query.run(peers))(_cb)
            }, options.timeout)((err, res) => {
              query.stop()
              cb(err, res)
            })
          },
          (result, cb) => {
            let success = false
            result.paths.forEach((result) => {
              if (result.success) {
                success = true
                this.peerBook.put(result.peer)
              }
            })
            this._log('findPeer %s: %s', id.toB58String(), success)
            if (!success) {
              return cb(errcode(new Error('No peer found'), 'ERR_NOT_FOUND'))
            }
            cb(null, this.peerBook.get(id))
          }
        ], callback)
      })
    }
  
    _peerDiscovered (peerInfo) {
      this.emit('peer', peerInfo)
    }
  }
  
  module.exports = KadDHT
  