index.js

const DHT = require('dht-rpc')
const sodium = require('sodium-universal')
const c = require('compact-encoding')
const b4a = require('b4a')
const safetyCatch = require('safety-catch')
const m = require('./lib/messages')
const SocketPool = require('./lib/socket-pool')
const Persistent = require('./lib/persistent')
const Router = require('./lib/router')
const Server = require('./lib/server')
const connect = require('./lib/connect')
const { FIREWALL, BOOTSTRAP_NODES, KNOWN_NODES, COMMANDS } = require('./lib/constants')
const { hash, createKeyPair } = require('./lib/crypto')
const { decode } = require('hypercore-id-encoding')
const RawStreamSet = require('./lib/raw-stream-set')
const ConnectionPool = require('./lib/connection-pool')
const { STREAM_NOT_CONNECTED } = require('./lib/errors')

class HyperDHT extends DHT {
  constructor (opts, sig, dht_keys, keychain = {}) {
    const port = opts.port || 49737
    const bootstrap = opts.bootstrap || BOOTSTRAP_NODES
    const nodes = opts.nodes || KNOWN_NODES

    super({ ...opts, port, bootstrap, nodes, filterNode })

    const { router, persistent } = defaultCacheOpts(opts)

    this.checkedSigs = [sig]
    this.sig = sig
    this.keychain = keychain
    this.dht_keys = dht_keys
    this.defaultKeyPair = dht_keys || createKeyPair(opts.seed)
    this.listening = new Set()
    this.connectionKeepAlive = opts.connectionKeepAlive === false
      ? 0
      : opts.connectionKeepAlive || 5000

    // stats is inherited from dht-rpc so fwd the ones from there
    this.stats = { punches: { consistent: 0, random: 0, open: 0 }, ...this.stats }
    
    this._router = new Router(this, router)
    this._socketPool = new SocketPool(this, opts.host || '0.0.0.0')
    this._rawStreams = new RawStreamSet(this)
    this._persistent = null
    this._validatedLocalAddresses = new Map()

    this._lastRandomPunch = 0
    this._randomPunchInterval = 20000 // min 20s between random punches...
    this._randomPunches = 0
    this._randomPunchLimit = 1 // set to one for extra safety for now

    this.once('persistent', () => {
      this._persistent = new Persistent(this, persistent)
    })

    this.on('network-change', () => {
      for (const server of this.listening) server.refresh()
    })

    this.on('network-update', () => {
      if (!this.online) return
      for (const server of this.listening) server.notifyOnline()
    })
  }

  connect (remotePublicKey, opts) {
    return connect(this, decode(remotePublicKey), opts)
  }

  createServer (opts, onconnection) {
    if (typeof opts === 'function') return this.createServer({}, opts)
    if (opts && opts.onconnection) onconnection = opts.onconnection
    const s = new Server(this, opts)
    if (onconnection) s.on('connection', onconnection)
    return s
  }

  pool () {
    return new ConnectionPool(this)
  }

  async resume () {
    await super.resume()
    const resuming = []
    for (const server of this.listening) resuming.push(server.resume())
    await Promise.allSettled(resuming)
  }

  async suspend () {
    const suspending = []
    for (const server of this.listening) suspending.push(server.suspend())
    await Promise.allSettled(suspending)
    await super.suspend()
    await this._rawStreams.clear()
  }

  async destroy ({ force = false } = {}) {
    if (!force) {
      const closing = []
      for (const server of this.listening) closing.push(server.close())
      await Promise.allSettled(closing)
    }
    this._router.destroy()
    if (this._persistent) this._persistent.destroy()
    await this._rawStreams.clear()
    await this._socketPool.destroy()
    await super.destroy()
  }

  async validateLocalAddresses (addresses) {
    const list = []
    const socks = []
    const waiting = []

    for (const addr of addresses) {
      const { host } = addr

      if (this._validatedLocalAddresses.has(host)) {
        if (await this._validatedLocalAddresses.get(host)) {
          list.push(addr)
        }
        continue
      }

      const sock = this.udx.createSocket()
      try {
        sock.bind(0, host)
      } catch {
        this._validatedLocalAddresses.set(host, Promise.resolve(false))
        continue
      }

      socks.push(sock)

      // semi terrible heuristic until we proper fix local connections by racing them to the remote...
      const promise = new Promise(resolve => {
        sock.on('message', () => resolve(true))
        setTimeout(() => resolve(false), 500)
        sock.trySend(b4a.alloc(1), sock.address().port, addr.host)
      })

      this._validatedLocalAddresses.set(host, promise)
      waiting.push(addr)
    }

    for (const addr of waiting) {
      const { host } = addr
      if (this._validatedLocalAddresses.has(host)) {
        if (await this._validatedLocalAddresses.get(host)) {
          list.push(addr)
        }
        continue
      }
    }

    for (const sock of socks) await sock.close()

    return list
  }

  findPeer (publicKey, opts = {}) {
    const target = opts.hash === false ? publicKey : hash(publicKey)
    opts = { ...opts, map: mapFindPeer }
    return this.query({ target, command: COMMANDS.FIND_PEER, value: null }, opts)
  }

  lookup (target, opts = {}) {
    opts = { ...opts, map: mapLookup }
    return this.query({ target, command: COMMANDS.LOOKUP, value: null }, opts)
  }

  lookupAndUnannounce (target, keyPair, opts = {}) {
    const unannounces = []
    const dht = this
    const userCommit = opts.commit || noop
    const signUnannounce = opts.signUnannounce || Persistent.signUnannounce

    if (this._persistent !== null) { // unlink self
      this._persistent.unannounce(target, keyPair.publicKey)
    }

    opts = { ...opts, map, commit }
    return this.query({ target, command: COMMANDS.LOOKUP, value: null }, opts)

    async function commit (reply, dht, query) {
      await Promise.all(unannounces) // can never fail, caught below
      return userCommit(reply, dht, query)
    }

    function map (reply) {
      const data = mapLookup(reply)

      if (!data || !data.token) return data

      let found = data.peers.length >= 20
      for (let i = 0; !found && i < data.peers.length; i++) {
        found = b4a.equals(data.peers[i].publicKey, keyPair.publicKey)
      }

      if (!found) return data

      if (!data.from.id) return data

      unannounces.push(
        dht._requestUnannounce(
          keyPair,
          dht,
          target,
          data.token,
          data.from,
          signUnannounce
        ).catch(safetyCatch)
      )

      return data
    }
  }

  unannounce (target, keyPair, opts = {}) {
    return this.lookupAndUnannounce(target, keyPair, opts).finished()
  }

  announce (target, keyPair, relayAddresses, opts = {}) {
    const signAnnounce = opts.signAnnounce || Persistent.signAnnounce

    opts = { ...opts, commit }

    return opts.clear
      ? this.lookupAndUnannounce(target, keyPair, opts)
      : this.lookup(target, opts)

    function commit (reply, dht) {
      return dht._requestAnnounce(
        keyPair,
        dht,
        target,
        reply.token,
        reply.from,
        relayAddresses,
        signAnnounce
      )
    }
  }

  async immutableGet (target, opts = {}) {
    opts = { ...opts, map: mapImmutable }

    const query = this.query({ target, command: COMMANDS.IMMUTABLE_GET, value: null }, opts)
    const check = b4a.allocUnsafe(32)

    for await (const node of query) {
      const { value } = node
      sodium.crypto_generichash(check, value)
      if (b4a.equals(check, target)) return node
    }

    return null
  }

  async immutablePut (value, opts = {}) {
    const target = b4a.allocUnsafe(32)
    sodium.crypto_generichash(target, value)

    opts = {
      ...opts,
      map: mapImmutable,
      commit (reply, dht) {
        return dht.request({ token: reply.token, target, command: COMMANDS.IMMUTABLE_PUT, value }, reply.from)
      }
    }

    const query = this.query({ target, command: COMMANDS.IMMUTABLE_GET, value: null }, opts)
    await query.finished()

    return { hash: target, closestNodes: query.closestNodes }
  }

  async mutableGet (publicKey, opts = {}) {
    let refresh = opts.refresh || null
    let signed = null
    let result = null

    opts = { ...opts, map: mapMutable, commit: refresh ? commit : null }

    const target = b4a.allocUnsafe(32)
    sodium.crypto_generichash(target, publicKey)

    const userSeq = opts.seq || 0
    const query = this.query({ target, command: COMMANDS.MUTABLE_GET, value: c.encode(c.uint, userSeq) }, opts)
    const latest = opts.latest !== false

    for await (const node of query) {
      if (result && node.seq <= result.seq) continue
      if (node.seq < userSeq || !Persistent.verifyMutable(node.signature, node.seq, node.value, publicKey)) continue
      if (!latest) return node
      if (!result || node.seq > result.seq) result = node
    }

    return result

    function commit (reply, dht) {
      if (!signed && result && refresh) {
        if (refresh(result)) {
          signed = c.encode(m.mutablePutRequest, {
            publicKey,
            seq: result.seq,
            value: result.value,
            signature: result.signature
          })
        } else {
          refresh = null
        }
      }

      return signed ? dht.request({ token: reply.token, target, command: COMMANDS.MUTABLE_PUT, value: signed }, reply.from) : Promise.resolve(null)
    }
  }

  async mutablePut (keyPair, value, opts = {}) {
    const signMutable = opts.signMutable || Persistent.signMutable

    const target = b4a.allocUnsafe(32)
    sodium.crypto_generichash(target, keyPair.publicKey)

    const seq = opts.seq || 0
    const signature = await signMutable(seq, value, keyPair)

    const signed = c.encode(m.mutablePutRequest, {
      publicKey: keyPair.publicKey,
      seq,
      value,
      signature
    })

    opts = {
      ...opts,
      map: mapMutable,
      commit (reply, dht) {
        return dht.request({ token: reply.token, target, command: COMMANDS.MUTABLE_PUT, value: signed }, reply.from)
      }
    }

    // use seq = 0, for the query part here, as we don't care about the actual values
    const query = this.query({ target, command: COMMANDS.MUTABLE_GET, value: c.encode(c.uint, 0) }, opts)
    await query.finished()

    return { publicKey: keyPair.publicKey, closestNodes: query.closestNodes, seq, signature }
  }

  onrequest (req) {
    switch (req.command) {
      case COMMANDS.PEER_HANDSHAKE: {
        this._router.onpeerhandshake(req)
        return true
      }
      case COMMANDS.PEER_HOLEPUNCH: {
        this._router.onpeerholepunch(req)
        return true
      }
    }

    if (this._persistent === null) return false

    switch (req.command) {
      case COMMANDS.FIND_PEER: {
        this._persistent.onfindpeer(req)
        return true
      }
      case COMMANDS.LOOKUP: {
        this._persistent.onlookup(req)
        return true
      }
      case COMMANDS.ANNOUNCE: {
        this._persistent.onannounce(req)
        return true
      }
      case COMMANDS.UNANNOUNCE: {
        this._persistent.onunannounce(req)
        return true
      }
      case COMMANDS.MUTABLE_PUT: {
        this._persistent.onmutableput(req)
        return true
      }
      case COMMANDS.MUTABLE_GET: {
        this._persistent.onmutableget(req)
        return true
      }
      case COMMANDS.IMMUTABLE_PUT: {
        this._persistent.onimmutableput(req)
        return true
      }
      case COMMANDS.IMMUTABLE_GET: {
        this._persistent.onimmutableget(req)
        return true
      }
    }

    return false
  }

  static keyPair (seed) {
    return createKeyPair(seed)
  }

  static hash (data) {
    return hash(data)
  }

  static connectRawStream (encryptedStream, rawStream, remoteId) {
    const stream = encryptedStream.rawStream

    if (!stream.connected) throw STREAM_NOT_CONNECTED()

    rawStream.connect(
      stream.socket,
      remoteId,
      stream.remotePort,
      stream.remoteHost
    )
  }

  createRawStream (opts) {
    return this._rawStreams.add(opts)
  }

  async _requestAnnounce (keyPair, dht, target, token, from, relayAddresses, sign) {
    const ann = {
      peer: {
        publicKey: keyPair.publicKey,
        relayAddresses: relayAddresses || []
      },
      refresh: null,
      signature: null
    }

    ann.signature = await sign(target, token, from.id, ann, keyPair)

    const value = c.encode(m.announce, ann)

    return dht.request({
      token,
      target,
      command: COMMANDS.ANNOUNCE,
      value
    }, from)
  }

  async _requestUnannounce (keyPair, dht, target, token, from, sign) {
    const unann = {
      peer: {
        publicKey: keyPair.publicKey,
        relayAddresses: []
      },
      signature: null
    }

    unann.signature = await sign(target, token, from.id, unann, keyPair)

    const value = c.encode(m.announce, unann)

    return dht.request({
      token,
      target,
      command: COMMANDS.UNANNOUNCE,
      value
    }, from)
  }
}

HyperDHT.BOOTSTRAP = BOOTSTRAP_NODES
HyperDHT.FIREWALL = FIREWALL

module.exports = HyperDHT

function mapLookup (node) {
  if (!node.value) return null

  try {
    return {
      token: node.token,
      from: node.from,
      to: node.to,
      peers: c.decode(m.peers, node.value)
    }
  } catch {
    return null
  }
}

function mapFindPeer (node) {
  if (!node.value) return null

  try {
    return {
      token: node.token,
      from: node.from,
      to: node.to,
      peer: c.decode(m.peer, node.value)
    }
  } catch {
    return null
  }
}

function mapImmutable (node) {
  if (!node.value) return null

  return {
    token: node.token,
    from: node.from,
    to: node.to,
    value: node.value
  }
}

function mapMutable (node) {
  if (!node.value) return null

  try {
    const { seq, value, signature } = c.decode(m.mutableGetResponse, node.value)

    return {
      token: node.token,
      from: node.from,
      to: node.to,
      seq,
      value,
      signature
    }
  } catch {
    return null
  }
}

function noop () {}

function filterNode (node) {
  // always skip these testnet nodes that got mixed in by accident, until they get updated
  return !(node.port === 49738 && (node.host === '134.209.28.98' || node.host === '167.99.142.185')) &&
    !(node.port === 9400 && node.host === '35.233.47.252') && !(node.host === '150.136.142.116')
}

const defaultMaxSize = 65536
const defaultMaxAge = 20 * 60 * 1000 // 20 minutes

function defaultCacheOpts (opts) {
  const maxSize = opts.maxSize || defaultMaxSize
  const maxAge = opts.maxAge || defaultMaxAge

  return {
    router: {
      forwards: { maxSize, maxAge }
    },
    persistent: {
      records: { maxSize, maxAge },
      refreshes: { maxSize, maxAge },
      mutables: {
        maxSize: maxSize / 2 | 0,
        maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
      },
      immutables: {
        maxSize: maxSize / 2 | 0,
        maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
      }
    }
  }
}