text: const after = require('after-all')
  const lexint = require('lexicographic-integer')
  const collect = require('stream-collector')
  const through = require('through2')
  const pump = require('pump')
  const from = require('from2')
  const mutexify = require('mutexify')
  const cuid = require('cuid')
  const logs = require('level-logs')
  const events = require('events')
  const util = require('util')
  const enumerate = require('level-enumerate')
  const replicate = require('./lib/replicate')
  const messages = require('./lib/messages')
  const hash = require('./lib/hash')
  const encoder = require('./lib/encode')
  const defined = require('defined')
  const parallel = require('run-parallel')
  const waterfall = require('run-waterfall')
  
  const ID = '!!id'
  const CHANGES = '!changes!'
  const NODES = '!nodes!'
  const HEADS = '!heads!'
  
  const INVALID_SIGNATURE = new Error('Invalid signature')
  const CHECKSUM_MISMATCH = new Error('Checksum mismatch')
  const INVALID_LOG = new Error('Invalid log sequence')
  
  INVALID_LOG.notFound = true
  INVALID_LOG.status = 404
  
  const noop = function () {}
  
  const Hyperlog = function (db, opts) {
    if (!(this instanceof Hyperlog)) return new Hyperlog(db, opts)
    if (!opts) opts = {}
  
    events.EventEmitter.call(this)
  
    this.id = defined(opts.id, null)
    this.enumerate = enumerate(db, { prefix: 'enum' })
    this.db = db
    this.logs = logs(db, { prefix: 'logs', valueEncoding: messages.Entry })
    this.lock = defined(opts.lock, mutexify())
    this.changes = 0
    this.setMaxListeners(0)
    this.valueEncoding = defined(opts.valueEncoding, opts.encoding, 'binary')
    this.identity = defined(opts.identity, null)
    this.verify = defined(opts.verify, null)
    this.sign = defined(opts.sign, null)
    this.hash = defined(opts.hash, hash)
    this.asyncHash = defined(opts.asyncHash, null)
  
    // Retrieve this hyperlog instance's unique ID.
    const self = this
    const getId = defined(opts.getId, function (cb) {
      db.get(ID, { valueEncoding: 'utf-8' }, function (_, id) {
        if (id) return cb(null, id)
        id = cuid()
        db.put(ID, id, function () {
          cb(null, id)
        })
      })
    })
  
    // Startup logic to..
    // 1. Determine & record the largest change # in the db.
    // 2. Determine this hyperlog db's local ID.
    //
    // This is behind a lock in order to ensure that no hyperlog operations
    // can be performed -- these two values MUST be known before any
    // hyperlog usage may occur.
    this.lock(function (release) {
      collect(db.createKeyStream({ gt: CHANGES, lt: CHANGES + '~', reverse: true, limit: 1 }), function (_, keys) {
        self.changes = Math.max(self.changes, keys && keys.length ? lexint.unpack(keys[0].split('!').pop(), 'hex') : 0)
        if (self.id) return release()
        getId(function (_, id) {
          self.id = id || cuid()
          release()
        })
      })
    })
  }
  
  util.inherits(Hyperlog, events.EventEmitter)
  
  // Call callback 'cb' once the hyperlog is ready for use (knows some
  // fundamental properties about itself from the leveldb). If it's already
  // ready, cb is called immediately.
  Hyperlog.prototype.ready = function (cb) {
    if (this.id) return cb()
    this.lock(function (release) {
      release()
      cb()
    })
  }
  
  // Returns a readable stream of all hyperlog heads. That is, all nodes that no
  // other nodes link to.
  Hyperlog.prototype.heads = function (opts, cb) {
    const self = this
    if (!opts) opts = {}
    if (typeof opts === 'function') {
      cb = opts
      opts = {}
    }
  
    const rs = this.db.createValueStream({
      gt: HEADS,
      lt: HEADS + '~',
      valueEncoding: 'utf-8'
    })
  
    const format = through.obj(function (key, enc, cb) {
      self.get(key, opts, cb)
    })
  
    return collect(pump(rs, format), cb)
  }
  
  // Retrieve a single, specific node, given its key.
  Hyperlog.prototype.get = function (key, opts, cb) {
    if (!opts) opts = {}
    if (typeof opts === 'function') {
      cb = opts
      opts = {}
    }
    const self = this
    this.db.get(NODES + key, { valueEncoding: 'binary' }, function (err, buf) {
      if (err) return cb(err)
      const node = messages.Node.decode(buf)
      node.value = encoder.decode(node.value, opts.valueEncoding || self.valueEncoding)
      cb(null, node)
    })
  }
  
  // Utility function to be used in a nodes.reduce() to determine the largest
  // change # present.
  const maxChange = function (max, cur) {
    return Math.max(max, cur.change)
  }
  
  // Consumes either a string or a hyperlog node and returns its key.
  const toKey = function (link) {
    return typeof link !== 'string' ? link.key : link
  }
  
  // Adds a new hyperlog node to an existing array of leveldb batch insertions.
  // This includes performing crypto signing and verification.
  // Performs deduplication; returns the existing node if alreay present in the hyperlog.
  const addBatchAndDedupe = function (dag, node, logLinks, batch, opts, cb) {
    if (opts.hash && node.key !== opts.hash) return cb(CHECKSUM_MISMATCH)
    if (opts.seq && node.seq !== opts.seq) return cb(INVALID_LOG)
  
    const log = {
      change: node.change,
      node: node.key,
      links: logLinks
    }
  
    const onclone = function (clone) {
      if (!opts.log) return cb(null, clone, [])
      batch.push({ type: 'put', key: dag.logs.key(node.log, node.seq), value: messages.Entry.encode(log) })
      cb(null, clone)
    }
  
    const done = function () {
      dag.get(node.key, { valueEncoding: 'binary' }, function (_, clone) {
        // This node already exists somewhere in the hyperlog; add it to the
        // log's append-only log, but don't insert it again.
        if (clone) return onclone(clone)
  
        const links = node.links
        for (let i = 0; i < links.length; i++) batch.push({ type: 'del', key: HEADS + links[i] })
        batch.push({ type: 'put', key: CHANGES + lexint.pack(node.change, 'hex'), value: node.key })
        batch.push({ type: 'put', key: NODES + node.key, value: messages.Node.encode(node) })
        batch.push({ type: 'put', key: HEADS + node.key, value: node.key })
        batch.push({ type: 'put', key: dag.logs.key(node.log, node.seq), value: messages.Entry.encode(log) })
  
        cb(null, node)
      })
    }
  
    // Local node; sign it.
    if (node.log === dag.id) {
      if (!dag.sign || node.signature) return done()
      dag.sign(node, function (err, sig) {
        if (err) return cb(err)
        if (!node.identity) node.identity = dag.identity
        node.signature = sig
        done()
      })
    // Remote node; verify it.
    } else {
      if (!dag.verify) return done()
      dag.verify(node, function (err, valid) {
        if (err) return cb(err)
        if (!valid) return cb(INVALID_SIGNATURE)
        done()
      })
    }
  }
  
  const getLinks = function (dag, id, links, cb) {
    const logLinks = []
    const nextLink = function () {
      const cb = next()
      return function (err, link) {
        if (err) return cb(err)
        if (link.log !== id && logLinks.indexOf(link.log) === -1) logLinks.push(link.log)
        cb(null)
      }
    }
    const next = after(function (err) {
      if (err) cb(err)
      else cb(null, logLinks)
    })
  
    for (let i = 0; i < links.length; i++) {
      dag.get(links[i], nextLink())
    }
  }
  
  // Produce a readable stream of all nodes added from this point onward, in
  // topographic order.
  const createLiveStream = function (dag, opts) {
    let since = opts.since || 0
    let limit = opts.limit || -1
    let wait = null
  
    const read = function (size, cb) {
      if (dag.changes <= since) {
        wait = cb
        return
      }
  
      if (!limit) return cb(null, null)
  
      dag.db.get(CHANGES + lexint.pack(since + 1, 'hex'), function (err, hash) {
        if (err) return cb(err)
        dag.get(hash, opts, function (err, node) {
          if (err) return cb(err)
          since = node.change
          if (limit !== -1) limit--
          cb(null, node)
        })
      })
    }
  
    const kick = function () {
      if (!wait) return
      const cb = wait
      wait = null
      read(0, cb)
    }
  
    dag.on('add', kick)
    dag.ready(kick)
  
    const rs = from.obj(read)
  
    rs.once('close', function () {
      dag.removeListener('add', kick)
    })
  
    return rs
  }
  
  // Produce a readable stream of nodes in the hyperlog, in topographic order.
  Hyperlog.prototype.createReadStream = function (opts) {
    if (!opts) opts = {}
    if (opts.tail) {
      opts.since = this.changes
    }
    if (opts.live) return createLiveStream(this, opts)
  
    const self = this
    const since = opts.since || 0
    const until = opts.until || 0
  
    const keys = this.db.createValueStream({
      gt: CHANGES + lexint.pack(since, 'hex'),
      lt: CHANGES + (until ? lexint.pack(until, 'hex') : '~'),
      valueEncoding: 'utf-8',
      reverse: opts.reverse,
      limit: opts.limit
    })
  
    const get = function (key, enc, cb) {
      self.get(key, opts, cb)
    }
  
    return pump(keys, through.obj(get))
  }
  
  Hyperlog.prototype.replicate =
  Hyperlog.prototype.createReplicationStream = function (opts) {
    return replicate(this, opts)
  }
  
  Hyperlog.prototype.add = function (links, value, opts, cb) {
    if (typeof opts === 'function') {
      cb = opts
      opts = {}
    }
    if (!cb) cb = noop
    this.batch([{ links: links, value: value }], opts, function (err, nodes) {
      if (err) cb(err)
      else cb(null, nodes[0])
    })
  }
  
  Hyperlog.prototype.batch = function (docs, opts, cb) {
    // 0. preamble
    if (typeof opts === 'function') {
      cb = opts
      opts = {}
    }
    if (!cb) cb = noop
    if (!opts) opts = {}
  
    // Bail asynchronously; nothing to add.
    if (docs.length === 0) return process.nextTick(function () { cb(null, []) })
  
    const self = this
    const id = opts.log || self.id
    opts.log = id
  
    const logLinks = {}
    let lockRelease = null
    let latestSeq
  
    // Bubble up errors on non-batch (1 element) calls.
    let bubbleUpErrors = false
    if (docs.length === 1) {
      bubbleUpErrors = true
    }
  
    // 1. construct initial hyperlog "node" for each of "docs"
    const nodes = docs.map(function (doc) {
      return constructInitialNode(doc, opts)
    })
  
    // 2. emit all preadd events
    nodes.forEach(function (node) {
      self.emit('preadd', node)
    })
  
    waterfall([
      // 3. lock the hyperlog (if needed)
      // 4. wait until the hyperlog is 'ready'
      // 5. retrieve the seq# of this hyperlog's head
      lockAndGetSeqNumber,
  
      // 3. hash (async/sync) all nodes
      // 4. retrieve + set 'getLinks' for each node
      function (seq, release, done) {
        lockRelease = release
        latestSeq = seq
  
        hashNodesAndFindLinks(nodes, done)
      },
  
      // 8. dedupe the node against the params AND the hyperlog (in sequence)
      function (nodes, done) {
        dedupeNodes(nodes, latestSeq, done)
      },
  
      // 9. create each node's leveldb batch operation object
      function (nodes, done) {
        computeBatchNodeOperations(nodes, done)
      },
  
      // 10. perform the leveldb batch op
      function (nodes, batchOps, done) {
        self.db.batch(batchOps, function (err) {
          if (err) {
            nodes.forEach(rejectNode)
            return done(err)
          }
          done(null, nodes)
        })
      },
  
      // 11. update the hyperlog's change#
      // 12. emit all add/reject events
      function (nodes, done) {
        self.changes = nodes.reduce(maxChange, self.changes)
        done(null, nodes)
      }
    ], function (err, nodes) {
      // release lock, if necessary
      if (lockRelease) return lockRelease(onUnlocked, err)
      onUnlocked(err)
  
      function onUnlocked (err) {
        // Error; all nodes were rejected.
        if (err) return cb(err)
  
        // Emit add events.
        nodes.forEach(function (node) {
          self.emit('add', node)
        })
  
        cb(null, nodes)
      }
    })
  
    function rejectNode (node) {
      self.emit('reject', node)
    }
  
    // Hashes and finds links for the given nodes. If some nodes fail to hash to
    // have their links found, they are rejected and not returned in the results.
    function hashNodesAndFindLinks (nodes, done) {
      const goodNodes = []
  
      parallel(
        nodes.map(function (node) {
          return function (done) {
            hashNode(node, function (err, key) {
              if (err) {
                rejectNode(node)
                return done(bubbleUpErrors ? err : null)
              }
              node.key = key
  
              getLinks(self, id, node.links, function (err, links) {
                if (err) {
                  rejectNode(node)
                  return done(bubbleUpErrors ? err : null)
                }
                logLinks[node.key] = links
  
                if (!node.log) node.log = self.id
  
                goodNodes.push(node)
                done()
              })
            })
          }
        }),
        function (err) {
          done(err, goodNodes)
        }
      )
    }
  
    function lockAndGetSeqNumber (done) {
      if (opts.release) onlocked(opts.release)
      else self.lock(onlocked)
  
      function onlocked (release) {
        self.ready(function () {
          self.logs.head(id, function (err, seq) {
            if (err) return release(cb, err)
            done(null, seq, release)
          })
        })
      }
    }
  
    function dedupeNodes (nodes, seq, done) {
      const goodNodes = []
  
      const added = nodes.length > 1 ? {} : null
      let seqIdx = 1
      let changeIdx = 1
  
      waterfall(
        nodes.map(function (node) {
          return function (done) {
            dedupeNode(node, done)
          }
        }),
        function (err) {
          done(err, goodNodes)
        }
      )
  
      function dedupeNode (node, done) {
        // Check if the to-be-added node already exists in the hyperlog.
        self.get(node.key, function (_, clone) {
          // It already exists
          if (clone) {
            node.seq = seq + (seqIdx++)
            node.change = clone.change
          // It already exists; it was added in this batch op earlier on.
          } else if (added && added[node.key]) {
            node.seq = added[node.key].seq
            node.change = added[node.key].change
            rejectNode(node)
            return done()
          } else {
            // new node across all logs
            node.seq = seq + (seqIdx++)
            node.change = self.changes + (changeIdx++)
          }
  
          if (added) added[node.key] = node
  
          goodNodes.push(node)
  
          done()
        })
      }
    }
  
    function computeBatchNodeOperations (nodes, done) {
      let batch = []
      const goodNodes = []
  
      waterfall(
        nodes.map(function (node) {
          return function (done) {
            computeNodeBatchOp(node, function (err, ops) {
              if (err) {
                rejectNode(node)
                return done(bubbleUpErrors ? err : null)
              }
              batch = batch.concat(ops)
              goodNodes.push(node)
              done()
            })
          }
        }),
        function (err) {
          if (err) return done(err)
          done(null, nodes, batch)
        }
      )
  
      // Create a new leveldb batch operation for this node.
      function computeNodeBatchOp (node, done) {
        const batch = []
        const links = logLinks[node.key]
        addBatchAndDedupe(self, node, links, batch, opts, function (err, newNode) {
          if (err) return done(err)
          newNode.value = encoder.decode(newNode.value, opts.valueEncoding || self.valueEncoding)
          done(null, batch)
        })
      }
    }
  
    function constructInitialNode (doc, opts) {
      let links = doc.links || []
      if (!Array.isArray(links)) links = [links]
      links = links.map(toKey)
  
      const encodedValue = encoder.encode(doc.value, opts.valueEncoding || self.valueEncoding)
      return {
        log: opts.log || self.id,
        key: null,
        identity: doc.identity || opts.identity || null,
        signature: opts.signature || null,
        value: encodedValue,
        links: links
      }
    }
  
    function hashNode (node, done) {
      if (self.asyncHash) {
        self.asyncHash(node.links, node.value, done)
      } else {
        const key = self.hash(node.links, node.value)
        done(null, key)
      }
    }
  }
  
  Hyperlog.prototype.append = function (value, opts, cb) {
    if (typeof opts === 'function') {
      cb = opts
      opts = {}
    }
    if (!cb) cb = noop
    if (!opts) opts = {}
    const self = this
  
    this.lock(function (release) {
      self.heads(function (err, heads) {
        if (err) return release(cb, err)
        opts.release = release
        self.add(heads, value, opts, cb)
      })
    })
  }
  
  module.exports = Hyperlog
  