WIP: kappa-next

.gitignore

@@ -1,2 +1,5 @@
 node_modules
 package-lock.json
+pnpm-lock.yaml
+yarn.lock
+SANDBOX

README.md

@@ -1,5 +1,174 @@
 # kappa-core
 
+## **kappa-core WIP rewrite** ##
+
+kappa-core is a database abstraction for append-only logs and add-only sets. A kappa-core is a container for pairs of sources and views, called *flows*. In each flow, data flows from a source into a view. Sources have a pull function that fetches new messages since the last pull. Views have a map function which is called for each batch of messages from the source.
+
+kappa-core itself is dependencyless, but this module also contains stateful source handlers for [hypercores](https://github.com/mafintosh/hypercore), [multifeed](https://github.com/kappa-core/multifeed), and [corestore](https://github.com/andrewosh/corestore).
+
+## API
+
+`const { Kappa } = require('kappa-core')`
+
+#### `const kappa = new Kappa()`
+
+Create a new kappa core.
+
+#### `kappa.use(name, source, view)`
+
+Register a flow. 
+
+* `name` (string) the name of the flow, has to be unique per kappa core
+* `source` object with properties:
+  * `open: function (flow, cb)` *(optional)* Handler to call on open. `flow` is the current flow object (see below for docs). Call `cb` when done with opening.
+  * `close: function (cb)`: *(optional)* Handler to call on close. Has to call `cb`.
+  * `pull: function (next)`: **(required)** Handler to pull new messages from the view. Should call `next` with either nothing or an object that looks like this:
+    ```javascript
+    {
+        error: Error,
+        messages: [], // array of messages
+        finished: true, // if set to false, signal that more messages are pending
+        onindexed: function (cb) {
+          // will be called when the view finished indexing
+          // call cb after the source state is updated
+          // may return a state object with, by convention, the following keys:
+          cb(null, {
+            totalBlocks: Number,
+            indexedBlocks: Number,
+            prevIndexedBlocks: Number 
+          })
+        }
+    }
+    ```
+  * `reset: function (cb)`: **(required)** Handler to reset internal state. This is called when a full reindex is necessary. This means that the next pull ought to start at the beginning.
+  * `storeVersion: function (version, cb)`: **(required)** Handler to store the flow version number.
+  * `fetchVersion: function (cb)`: **(required)** Handler to fetch the version stored with `storeVersion`.
+  * See the `SimpleState` docs below how to easily implement the `reset`, `storeVersion` and `fetchVersion` methods.
+
+* `view` object with properties:
+  * `open: function (flow, cb)` *(optional)* Handler to call on open. `flow` is the current flow object (see below for docs). Call `cb` when done with opening.
+  * `close: function (cb)`: *(optional)* Handler to call on close. Has to call `cb`.
+  * `map: function (messages, next)` **(required)** Handler for each batch of messages. Call `next` when done indexing this batch of messages.
+  * `reset: function (cb)`: **(required)** Handler to delete all indexed data. This is called by the Kappa core when a complete reindex is necessary. The `map` function will receive messages from the start on afterwards.
+  * `version: int` The view version. If the version is increased, the Kappa core will clear and restart the indexing for this view after the next reopening of the core. Defaults to `1`.
+
+Both `source` and `view` can have an `api` property with an object of functions. The functions are exposed on `kappa.view[name]` / `kappa.source[name]`. Their `this` object refers to the flow they are part of, and their first parameter is the `kappa`. Other parameters are passed through.
+
+The source has to track its state, so that subsequent calls to `pull()` do not return the same messages. Use the `onindexed` callback to update state. How to track its state is up to the source implementation. kappa-core provides a `SimpleState` helper to simplify this, see its documentation below.
+
+There are several source handlers included in kappa-core (TODO: document sources). See the tests and sources directories.
+
+#### `kappa.reset(name, cb)`
+
+Reset a specific flow, to restart indexing. This is equal to reopening the kappa-core with a changed view version for this flow.
+
+#### `kappa.ready(names, cb)`
+
+Call `cb` exactly once, after all flows with a name in the `names` array have finished processing. If `names` is empty, all flows will be awaited. This `names` is a string, the flow of this name will be awaited. If the requested flows are already ready, `cb` is called immediately.
+
+#### `kappa.pause()`
+
+Pause processing of all flows
+
+#### `kappa.resume()`
+
+Resume processing of all flows
+
+## Flow
+
+When calling `kappa.use()` a new *Flow* is created. A Flow is the combination of a source and a view - where the data flows from the source into the view. The `Flow` object is passed to sources and views in their `open` handler. It has this public API:
+
+* `flow.name`: (string) A name that uniquely identifies this flow within the Kappa core.
+* `flow.update()`: Signal to the flow that the source has new data available. Youwant to call this from a source when the source has new data. If the Kappa core is not paused, this will cause the `pull` handler to be called.
+* `flow.ready(cb)`: Calls `cb` (with no arguments) when this flow has finished processing all messages. `cb` is called immediately if the flow is already finished.
+* `flow.getState()`: Get the current indexing state. Returns an object:
+    ```javascript
+      {
+        status: 'idle' | 'running' | 'paused' | 'error',
+        error: null | Error,
+        // ... other keys as returned by the source
+        // by convention this should include the following keys:
+        totalBlocks: Number,
+        indexedBlocks: Number,
+        prevIndexedBlocks: Number
+      }
+    ```
+* `flow.view`: Object with the view's API functions
+* `flow.source`: Object with the source's API functions
+
+## SimpleState
+
+`kappa-core` exports a `SimpleState` class that can be used by sources for a simple state handling. It persists state either in-memory, and supports a [LevelDB](https://github.com/Level/level) (or compatible) option for persistence.
+
+Example:
+
+```javascript
+const { Kappa, SimpleState } = require('kappa-core')
+function createSource (opts) {
+  const state = new SimpleState({ db: opts.db })
+  return {
+    pull (next) {
+      // get your current state
+      state.get((err, state) => {
+        if (err) return next()
+        // fetch messages from your data source
+        fetchMessages(state, ({ messages, finished, nextState }) => {
+          // call next with an onindexed handler
+          next({
+            messages,
+            finished,
+            onindexed (cb) {
+              // store the new state
+              state.put(nextState, cb)
+            }
+          })
+        })
+      })
+    },
+    fetchVersion: state.fetchVersion,
+    storeVersion: state.storeVersion,
+    reset (cb) {
+      state.put('', cb)
+    }
+  }
+}
+```
+
+## Sources
+
+#### [hypercore](https://github.com/mafintosh/hypercore)
+
+```javascript
+const createHypercoreSource = require('kappa-core/sources/hypercore')
+const source = createHypercoreSource({ feed, db })
+```
+
+where `feed` is a hypercore instance and `db` is a levelup instance (for persisting state)
+
+#### [multifeed](https://github.com/kappa-core/multifeed)
+
+```javascript
+const createMultifeedSource = require('kappa-core/sources/multifeed')
+const source = createMultifeedSource({ feeds, db })
+```
+
+where `feeds` is a multifeed instance and `db` is a levelup instance (for persisting state)
+
+This source exposes an API method `feed (key)` that return a feed by key from the underlying multifeed.
+
+#### [corestore](https://github.com/andrewosh/corestore)
+
+```javascript
+const createCorestoreSource = require('kappa-core/sources/corestore')
+const source = createCorestoreSource({ store, db })
+```
+
+where `store` is a corestore instance and `db` is a levelup instance (for persisting state)
+
+This source exposes an API method `feed (key)` that return a feed by key from the underlying corestore.
+
+---
+
 > kappa-core is a minimal peer-to-peer database, based on append-only logs and materialized views.
 
 ## Introduction

index.js

@@ -1,129 +1,6 @@
-var inherits = require('inherits')
-var EventEmitter = require('events').EventEmitter
-var hypercore = require('hypercore')
-var multifeed = require('multifeed')
-var indexer = require('multifeed-index')
+const Kappa = require('./kappa')
+const SimpleState = require('./sources/util/state')
 
 module.exports = Kappa
-
-function Kappa (storage, opts) {
-  if (!(this instanceof Kappa)) return new Kappa(storage, opts)
-  if (!opts) opts = {}
-
-  this._logs = opts.multifeed || multifeed(hypercore, storage, opts)
-  this._indexes = {}
-
-  this.api = {}
-}
-
-inherits(Kappa, EventEmitter)
-
-Kappa.prototype.use = function (name, version, view) {
-  var self = this
-  if (typeof version !== 'number') {
-    view = version
-    version = undefined
-  }
-  var idx = indexer(Object.assign({}, view, {
-    log: this._logs,
-    version: version,
-    maxBatch: view.maxBatch || 10,
-    batch: view.map
-  }))
-  idx.on('error', function (err) {
-    self.emit('error', err)
-  })
-  if (view.indexed) idx.on('indexed', view.indexed)
-  this._indexes[name] = idx
-  this.api[name] = {}
-  this.api[name].ready = idx.ready.bind(idx)
-  for (var key in view.api) {
-    if (typeof view.api[key] === 'function') this.api[name][key] = view.api[key].bind(idx, this)
-    else this.api[name][key] = view.api[key]
-  }
-}
-
-Kappa.prototype.feeds = function () {
-  return this._logs.feeds()
-}
-
-Kappa.prototype.ready = function (viewNames, cb) {
-  if (typeof viewNames === 'function') {
-    cb = viewNames
-    viewNames = []
-  }
-
-  if (typeof viewNames === 'string') viewNames = [viewNames]
-  if (viewNames.length === 0) {
-    viewNames = Object.keys(this._indexes)
-  }
-
-  var pending = viewNames.length + 1
-  var self = this
-  this._logs.ready(function () {
-    for (var i = 0; i < viewNames.length; i++) {
-      self._indexes[viewNames[i]].ready(done)
-    }
-    done()
-  })
-
-  function done () {
-    if (!--pending) cb()
-  }
-}
-
-Kappa.prototype.pause = function (viewNames, cb) {
-  if (typeof viewNames === 'function') {
-    cb = viewNames
-    viewNames = []
-  }
-  cb = cb || noop
-
-  if (!viewNames) viewNames = []
-  if (typeof viewNames === 'string') viewNames = [viewNames]
-  if (viewNames.length === 0) {
-    viewNames = Object.keys(this._indexes)
-  }
-
-  var pending = viewNames.length + 1
-  var self = this
-  this._logs.ready(function () {
-    for (var i = 0; i < viewNames.length; i++) {
-      self._indexes[viewNames[i]].pause(done)
-    }
-    done()
-  })
-
-  function done () {
-    if (!--pending) cb()
-  }
-}
-
-Kappa.prototype.resume = function (viewNames) {
-  if (!viewNames) viewNames = []
-  if (typeof viewNames === 'string') viewNames = [viewNames]
-  if (viewNames.length === 0) {
-    viewNames = Object.keys(this._indexes)
-  }
-
-  var self = this
-  this._logs.ready(function () {
-    for (var i = 0; i < viewNames.length; i++) {
-      self._indexes[viewNames[i]].resume()
-    }
-  })
-}
-
-Kappa.prototype.writer = function (name, cb) {
-  this._logs.writer(name, cb)
-}
-
-Kappa.prototype.feed = function (key) {
-  return this._logs.feed(key)
-}
-
-Kappa.prototype.replicate = function (opts) {
-  return this._logs.replicate(opts)
-}
-
-function noop () {}
+module.exports.Kappa = Kappa
+module.exports.SimpleState = SimpleState