full-kappa-view-ca-store.js

var sha = require('sha.js')
var umkv = require('unordered-materialized-kv')
var EventEmitter = require('events').EventEmitter

module.exports = ContentAddressableStore

function ContentAddressableStore (db, opts) {
  var kv = umkv(db)
  var events = new EventEmitter()
  opts = opts || {}

  var idx = {
    maxBatch: opts.maxBatch || 100,

    map: function (msgs, next) {
      var ops = msgs.map(function (msg) {
        var hash = sha('sha256').update(JSON.stringify(msg.value)).digest('hex')
        var msgId = msg.key + '@' + msg.seq
        return { key: hash, id: msgId, links: [] }
      })
      kv.batch(allOps, next)
    },

    indexed: function (msgs) {
      for (var i = 0; i < msgs.length; i++) {
        events.emit('update!' + msgs[i].id, msg)
      }
    },

    api: {
      get: function (core, key, cb) {
        this.ready(function () {
          kv.get(key, function (err, ids) {
            if (err) return cb(err)
            var res = []
            var pending = ids.length + 1
            for (var i = 0; i < ids.length; i++) {
              var id = ids[i]
              var feed = core._logs.feed(id.split('@')[0])
              feed.get(id.split('@')[1], done)
            }
            done()

            function done (err, msg) {
              if (err) {
                pending = Infinity
                return cb(err)
              }
              if (msg) res.push(msg)
              if (!--pending) cb(null, res)
            }
          })
        })
      },

      onUpdate: function (core, key, cb) {
        events.on('update!' + key, cb)
      }
    },

    storeState: function (state, cb) {
      db.put('state', state, cb)
    },

    fetchState: function (cb) {
      db.get('state', function (err, state) {
        if (err && err.notFound) cb()
        else if (err) cb(err)
        else cb(null, state)
      })
    },
  }
  return idx
}

kappa-view-content-store.js

var sha = require('sha.js')
var caIdx = kv(idx, function (msg, next) {
  var ops = []
  var hash = sha('sha256').update(JSON.stringify(msg.value)).digest('hex')
  var msgId = msg.key + '@' + msg.seq
  ops.push({ key: hash, id: msgId, links: [] })
  next(null, ops)
})
core.use('ca', caIdx)