Partial replication changes

index.js

@@ -2,7 +2,7 @@ const path = require('path')
 const pull = require('pull-stream')
 const toPull = require('push-stream-to-pull-stream')
 const EBT = require('epidemic-broadcast-trees')
-const isFeed = require('ssb-ref').isFeed
+const ref = require('ssb-ref')
 const Store = require('lossy-store')
 const toUrlFriendly = require('base64-url').escape
 const getSeverity = require('ssb-network-errors')
@@ -21,21 +21,23 @@ exports.version = '1.0.0'
 
 exports.manifest = {
   replicate: 'duplex',
+  replicateFormat: 'duplex',
   request: 'sync',
   block: 'sync',
-  peerStatus: 'sync'
+  peerStatus: 'sync',
+  clock: 'async'
 }
 
 exports.permissions = {
   anonymous: {
-    allow: ['replicate']
+    allow: ['replicate', 'replicateFormat', 'clock']
   }
 }
 
 // there was a bug that caused some peers
 // to request things that weren't feeds.
 // this is fixed, so just ignore anything that isn't a feed.
-function cleanClock (clock) {
+function cleanClock (clock, isFeed) {
   for (const k in clock) {
     if (!isFeed(k)) {
       delete clock[k]
@@ -44,48 +46,105 @@ function cleanClock (clock) {
 }
 
 exports.init = function (sbot, config) {
-  const dir = config.path ? path.join(config.path, 'ebt') : null
-  const store = Store(dir, null, toUrlFriendly)
-
-  const ebt = EBT({
-    logging: config.ebt && config.ebt.logging,
-    id: sbot.id,
-    getClock (id, cb) {
-      store.ensure(id, function () {
-        const clock = store.get(id) || {}
-        cleanClock(clock)
-        cb(null, clock)
-      })
-    },
-    setClock (id, clock) {
-      cleanClock(clock, 'non-feed key when saving clock')
-      store.set(id, clock)
-    },
-    getAt (pair, cb) {
-      sbot.getAtSequence([pair.id, pair.sequence], (err, data) => {
-        cb(err, data ? data.value : null)
-      })
-    },
-    append (msg, cb) {
-      sbot.add(msg, (err, msg) => {
-        cb(err && err.fatal ? err : null, msg)
-      })
-    },
-    isFeed: isFeed
-  })
+  const formats = {
+    'classic': {
+      // used in request, block, cleanClock, sbot.post
+      isFeed: ref.isFeed,
+      getAtSequence(sbot, pair, cb) {
+        sbot.getAtSequence([pair.id, pair.sequence], (err, msg) => {
+          cb(err, msg ? msg.value : null)
+        })
+      },
+      appendMsg(sbot, msgVal, cb) {
+        sbot.add(msgVal, (err, msg) => {
+          cb(err && err.fatal ? err : null, msg)
+        })
+      },
+
+      // used in ebt:stream to distinguish between messages and notes
+      isMsg(msgVal) {
+        return Number.isInteger(msgVal.sequence) && msgVal.sequence > 0 &&
+          ref.isFeed(msgVal.author) && msgVal.content
+      },
+      // used in ebt:events
+      getMsgAuthor(msgVal) {
+        return msgVal.author
+      },
+      // used in ebt:events
+      getMsgSequence(msgVal) {
+        return msgVal.sequence
+      },
+    }
+  }
+
+  const ebts = {}
+  function addEBT(formatName) {
+    const dirName = 'ebt' + (formatName === 'classic' ? '' : formatName)
+    const dir = config.path ? path.join(config.path, dirName) : null
+    const store = Store(dir, null, toUrlFriendly)
+
+    const format = formats[formatName]
+
+    const ebt = EBT(Object.assign({
+      logging: config.ebt && config.ebt.logging,
+      id: sbot.id,
+      getClock (id, cb) {
+        store.ensure(id, function () {
+          const clock = store.get(id) || {}
+          cleanClock(clock, format.isFeed)
+          cb(null, clock)
+        })
+      },
+      setClock (id, clock) {
+        cleanClock(clock, format.isFeed)
+        store.set(id, clock)
+      },
+      getAt (pair, cb) {
+        format.getAtSequence(sbot, pair, cb)
+      },
+      append (msgVal, cb) {
+        format.appendMsg(sbot, msgVal, cb)
+      }
+    }, format))
+
+    ebts[formatName] = ebt
+  }
+
+  function getEBT(formatName) {
+    const ebt = ebts[formatName]
+    if (!ebt)
+      throw new Error('Unknown format' + formatName)
+
+    return ebt
+  }
+
+  addEBT('classic')
 
   const initialized = DeferredPromise()
 
   sbot.getVectorClock((err, clock) => {
     if (err) console.warn('Failed to getVectorClock in ssb-ebt because:', err)
-    ebt.state.clock = clock || {}
-    ebt.update()
+    for (let formatName in ebts) {
+      const format = formats[formatName]
+      const ebt = ebts[formatName]
+
+      validClock = {}
+      for (let k in clock)
+        if (format.isFeed(k))
+          validClock[k] = clock[k]
+
+      ebt.state.clock = validClock
+      ebt.update()
+    }
     initialized.resolve()
   })
 
   sbot.post((msg) => {
     initialized.promise.then(() => {
-      ebt.onAppend(msg.value)
+      for (let format in ebts) {
+        if (formats[format].isFeed(msg.value.author))
+          ebts[format].onAppend(msg.value)
+      }
     })
   })
 
@@ -94,6 +153,7 @@ exports.init = function (sbot, config) {
   if (sbot.progress) {
     hook(sbot.progress, function (fn) {
       const _progress = fn()
+      const ebt = ebts['classic']
       const ebtProg = ebt.progress()
       if (ebtProg.target) _progress.ebt = ebtProg
       return _progress
@@ -104,29 +164,48 @@ exports.init = function (sbot, config) {
     if (rpc.id === sbot.id) return // ssb-client connecting to ssb-server
     if (isClient) {
       initialized.promise.then(() => {
-        const opts = { version: 3 }
-        const local = toPull.duplex(ebt.createStream(rpc.id, opts.version, true))
-        const remote = rpc.ebt.replicate(opts, (networkError) => {
-          if (networkError && getSeverity(networkError) >= 3) {
-            console.error('rpc.ebt.replicate exception:', networkError)
-          }
-        })
-        pull(local, remote, local)
+        for (let format in ebts) {
+          const ebt = ebts[format]
+          const opts = { version: 3, format }
+          const local = toPull.duplex(ebt.createStream(rpc.id, opts.version, true))
+
+          // for backwards compatibility we always replicate classic
+          // feeds using existing replicate RPC
+          const replicate = (format === 'classic' ? rpc.ebt.replicate : rpc.ebt.replicateFormat)
+
+          const remote = replicate(opts, (networkError) => {
+            if (networkError && getSeverity(networkError) >= 3) {
+              console.error('rpc.ebt.replicate exception:', networkError)
+            }
+          })
+          pull(local, remote, local)
+        }
       })
     }
   })
 
-  function request (destFeedId, requesting) {
+  function request(destFeedId, requesting, formatName) {
     initialized.promise.then(() => {
-      if (!isFeed(destFeedId)) return
-      ebt.request(destFeedId, requesting)
+      formatName = formatName || 'classic'
+      const format = formats[formatName]
+
+      if (!(format && format.isFeed(destFeedId))) return
+
+      ebts[formatName].request(destFeedId, requesting)
     })
   }
 
-  function block (origFeedId, destFeedId, blocking) {
+  function block(origFeedId, destFeedId, blocking, formatName) {
     initialized.promise.then(() => {
-      if (!isFeed(origFeedId)) return
-      if (!isFeed(destFeedId)) return
+      formatName = formatName || 'classic'
+      const format = formats[formatName]
+
+      if (!format) return
+      if (!format.isFeed(origFeedId)) return
+      if (!format.isFeed(destFeedId)) return
+
+      const ebt = ebts[formatName]
+
       if (blocking) {
         ebt.block(origFeedId, destFeedId, true)
       } else if (
@@ -139,11 +218,14 @@ exports.init = function (sbot, config) {
     })
   }
 
-  function replicate (opts) {
+  function replicateFormat(opts) {
     if (opts.version !== 3) {
       throw new Error('expected ebt.replicate({version: 3})')
     }
 
+    let formatName = opts.format || 'classic'
+    const ebt = getEBT(formatName)
+
     var deferred = pullDefer.duplex()
     initialized.promise.then(() => {
       // `this` refers to the remote peer who called this muxrpc API
@@ -152,14 +234,18 @@ exports.init = function (sbot, config) {
     return deferred
   }
 
-  // get replication status for feeds for this id.
-  function peerStatus (id) {
+  // get replication status for feeds for this id
+  function peerStatus(id, formatName) {
     id = id || sbot.id
+    formatName = formatName || 'classic'
+    const ebt = getEBT(formatName)
+
     const data = {
       id: id,
       seq: ebt.state.clock[id],
       peers: {}
     }
+
     for (const k in ebt.state.peers) {
       const peer = ebt.state.peers[k]
       if (peer.clock[id] != null ||
@@ -171,13 +257,43 @@ exports.init = function (sbot, config) {
         }
       }
     }
+
     return data
   }
 
+  function clock(formatName, cb) {
+    if (!cb) {
+      cb = formatName
+      formatName = 'classic'
+    }
+
+    initialized.promise.then(() => {
+      const ebt = getEBT(formatName)
+      cb(null, ebt.state.clock)
+    })
+  }
+
+  function setClockForSlicedReplication(format, feed, sequence) {
+    initialized.promise.then(() => {
+      const ebt = getEBT(format)
+      ebt.state.clock[feed] = sequence
+    })
+  }
+
+  function registerFormat(formatName, methods) {
+    formats[formatName] = methods
+    addEBT(formatName)
+  }
+
   return {
     request,
     block,
-    replicate,
-    peerStatus
+    replicate: replicateFormat,
+    replicateFormat,
+    peerStatus,
+    clock,
+    setClockForSlicedReplication,
+    registerFormat,
+    formats
   }
 }

package.json

@@ -37,10 +37,13 @@
     "rimraf": "^2.7.1",
     "rng": "^0.2.2",
     "secret-stack": "^6.4.0",
+    "ssb-bendy-butt": "^0.12.2",
     "ssb-client": "^4.9.0",
     "ssb-db": "^19.2.0",
+    "ssb-db2": "^2.4.0",
     "ssb-generate": "^1.0.1",
     "ssb-keys": "^8.1.0",
+    "ssb-uri2": "^1.5.2",
     "ssb-validate": "^4.1.4",
     "standardx": "^7.0.0",
     "tap-spec": "^5.0.0",