feat: ipns over pubsub

Author: vasco-santos

README.md

@@ -304,6 +304,7 @@ Configure remote preload nodes. The remote will preload content added on this no
 Enable and configure experimental features.
 
 - `pubsub` (boolean): Enable libp2p pub-sub. (Default: `false`)
+- `ipnsPubsub` (boolean): Enable pub-sub on IPNS. (Default: `false`)
 - `sharding` (boolean): Enable directory sharding. Directories that have many child objects will be represented by multiple DAG nodes instead of just one. It can improve lookup performance when a directory has several thousand files or more. (Default: `false`)
 - `dht` (boolean): Enable KadDHT. **This is currently not interoperable with `go-ipfs`.**
 
@@ -554,6 +555,9 @@ The core API is grouped into several areas:
 - [name](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md)
   - [`ipfs.name.publish(value, [options], [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#namepublish)
   - [`ipfs.name.resolve(value, [options], [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#nameresolve)
+  - [`ipfs.name.pubsub.cancel(arg, [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#namepubsubcancel)
+  - [`ipfs.name.pubsub.state([callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#namepubsubstate)
+  - [`ipfs.name.pubsub.subs([callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#namepubsubsubs)
 
 #### Crypto and Key Management
 

package.json

@@ -72,7 +72,7 @@
     "form-data": "^2.3.2",
     "hat": "0.0.3",
     "interface-ipfs-core": "~0.78.0",
-    "ipfsd-ctl": "~0.39.1",
+    "ipfsd-ctl": "~0.39.2",
     "mocha": "^5.2.0",
     "ncp": "^2.0.0",
     "nexpect": "~0.5.0",
@@ -86,13 +86,16 @@
   "dependencies": {
     "@nodeutils/defaults-deep": "^1.1.0",
     "async": "^2.6.1",
+    "base32.js": "~0.1.0",
     "big.js": "^5.1.2",
     "binary-querystring": "~0.1.2",
     "bl": "^2.0.1",
     "boom": "^7.2.0",
     "bs58": "^4.0.1",
     "byteman": "^1.3.5",
     "cids": "~0.5.3",
+    "datastore-core": "~0.4.0",
+    "datastore-pubsub": "~0.0.2",
     "debug": "^3.1.0",
     "err-code": "^1.1.2",
     "file-type": "^8.1.0",

src/cli/commands/daemon.js

@@ -25,6 +25,10 @@ module.exports = {
         type: 'boolean',
         default: false
       })
+      .option('enable-namesys-pubsub', {
+        type: 'boolean',
+        default: false
+      })
   },
 
   handler (argv) {

src/cli/commands/name/pubsub.js

@@ -0,0 +1,18 @@
+'use strict'
+
+/*
+Manage and inspect the state of the IPNS pubsub resolver.
+Note: this command is experimental and subject to change as the system is refined.
+*/
+module.exports = {
+  command: 'pubsub',
+
+  description: 'IPNS pubsub management.',
+
+  builder (yargs) {
+    return yargs.commandDir('pubsub')
+  },
+
+  handler (argv) {
+  }
+}

src/cli/commands/name/pubsub/cancel.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const print = require('../../../utils').print
+
+module.exports = {
+  command: 'cancel <name>',
+
+  describe: 'Cancel a name subscription.',
+
+  handler (argv) {
+    argv.ipfs.name.pubsub.cancel(argv.name, (err, result) => {
+      if (err) {
+        throw err
+      } else {
+        print(result.canceled ? 'canceled' : 'no subscription')
+      }
+    })
+  }
+}

src/cli/commands/name/pubsub/state.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const print = require('../../../utils').print
+
+module.exports = {
+  command: 'state',
+
+  describe: 'Query the state of IPNS pubsub.',
+
+  handler (argv) {
+    argv.ipfs.name.pubsub.state((err, result) => {
+      if (err) {
+        throw err
+      } else {
+        print(result.enabled ? 'enabled' : 'disabled')
+      }
+    })
+  }
+}

src/cli/commands/name/pubsub/subs.js

@@ -0,0 +1,21 @@
+'use strict'
+
+const print = require('../../../utils').print
+
+module.exports = {
+  command: 'subs',
+
+  describe: 'Show current name subscriptions.',
+
+  handler (argv) {
+    argv.ipfs.name.pubsub.subs((err, result) => {
+      if (err) {
+        throw err
+      } else {
+        result.strings.forEach((s) => {
+          print(s)
+        })
+      }
+    })
+  }
+}

src/core/components/init.js

@@ -7,6 +7,9 @@ const promisify = require('promisify-es6')
 const defaultConfig = require('../runtime/config-nodejs.js')
 const Keychain = require('libp2p-keychain')
 
+const IPNS = require('../ipns')
+const OfflineDatastore = require('../ipns/routing/offline-datastore')
+
 const addDefaultAssets = require('./init-assets')
 
 module.exports = function init (self) {
@@ -103,6 +106,13 @@ module.exports = function init (self) {
           cb(null, true)
         }
       },
+      // Setup offline routing for IPNS. This is primarily used for offline ipns modifications, such as the initializeKeyspace feature.
+      (_, cb) => {
+        const offlineDatastore = new OfflineDatastore(self._repo)
+
+        self._ipns = new IPNS(offlineDatastore, self)
+        cb(null, true)
+      },
       // add empty unixfs dir object (go-ipfs assumes this exists)
       (_, cb) => {
         if (opts.emptyRepo) {

src/core/components/libp2p.js

@@ -45,7 +45,7 @@ module.exports = function libp2p (self) {
               },
               EXPERIMENTAL: {
                 dht: get(opts.options, 'EXPERIMENTAL.dht', false),
-                pubsub: get(opts.options, 'EXPERIMENTAL.pubsub', false)
+                pubsub: get(opts.options, 'EXPERIMENTAL.pubsub', false) || get(opts.options, 'EXPERIMENTAL.ipnsPubsub', false)
               }
             },
             connectionManager: get(opts.options, 'connectionManager',

src/core/components/name-pubsub.js

@@ -0,0 +1,60 @@
+'use strict'
+
+const debug = require('debug')
+const errcode = require('err-code')
+const promisify = require('promisify-es6')
+
+const log = debug('jsipfs:name-pubsub')
+log.error = debug('jsipfs:name-pubsub:error')
+
+const isNamePubsubEnabled = (node) => (
+  node._options.EXPERIMENTAL.ipnsPubsub && node._libp2pNode._floodSub
+)
+
+module.exports = function namePubsub (self) {
+  return {
+    /**
+     * Query the state of IPNS pubsub.
+     *
+     * @returns {Promise|void}
+     */
+    state: promisify((callback) => {
+      callback(null, {
+        enabled: Boolean(isNamePubsubEnabled(self))
+      })
+    }),
+    /**
+     * Cancel a name subscription.
+     *
+     * @param {String} name subscription name.
+     * @param {function(Error)} [callback]
+     * @returns {Promise|void}
+     */
+    cancel: promisify((name, callback) => {
+      if (!isNamePubsubEnabled(self)) {
+        const errMsg = 'IPNS pubsub subsystem is not enabled'
+
+        log.error(errMsg)
+        return callback(errcode(errMsg, 'ERR_IPNS_PS_NOT_ENABLED'))
+      }
+
+      self._ipns.pubsub.cancel(name, callback)
+    }),
+    /**
+     * Show current name subscriptions.
+     *
+     * @param {function(Error)} [callback]
+     * @returns {Promise|void}
+     */
+    subs: promisify((callback) => {
+      if (!isNamePubsubEnabled(self)) {
+        const errMsg = 'IPNS pubsub subsystem is not enabled'
+
+        log.error(errMsg)
+        return callback(errcode(errMsg, 'ERR_IPNS_PS_NOT_ENABLED'))
+      }
+
+      self._ipns.pubsub.getSubscriptions(callback)
+    })
+  }
+}

src/core/components/name.js

@@ -11,6 +11,7 @@ const errcode = require('err-code')
 const log = debug('jsipfs:name')
 log.error = debug('jsipfs:name:error')
 
+const namePubsub = require('./name-pubsub')
 const utils = require('../utils')
 const path = require('../ipns/path')
 
@@ -128,7 +129,7 @@ module.exports = function name (self) {
       const nocache = options.nocache && options.nocache.toString() === 'true'
       const recursive = options.recursive && options.recursive.toString() === 'true'
 
-      const local = true // TODO ROUTING - use self._options.local
+      const local = self._options.local
 
       if (!self.isOnline() && !local) {
         const errMsg = utils.OFFLINE_ERROR
@@ -157,11 +158,11 @@ module.exports = function name (self) {
 
       const resolveOptions = {
         nocache,
-        recursive,
-        local
+        recursive
       }
 
-      self._ipns.resolve(name, self._peerInfo.id, resolveOptions, callback)
-    })
+      self._ipns.resolve(name, resolveOptions, callback)
+    }),
+    pubsub: namePubsub(self)
   }
 }

src/core/components/start.js

@@ -2,8 +2,14 @@
 
 const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
+const get = require('lodash/get')
 const setImmediate = require('async/setImmediate')
 const promisify = require('promisify-es6')
+const { TieredDatastore } = require('datastore-core')
+
+const IPNS = require('../ipns')
+const Pubsub = require('../ipns/routing/pubsub')
+const OfflineDatastore = require('../ipns/routing/offline-datastore')
 
 module.exports = (self) => {
   return promisify((callback) => {
@@ -34,6 +40,25 @@ module.exports = (self) => {
       },
       (cb) => self.libp2p.start(cb),
       (cb) => {
+        // Setup online routing for IPNS with a tiered routing composed by a DHT and a Pubsub router (if properly enabled)
+        const ipnsStores = []
+
+        // Add IPNS pubsub if enabled
+        let pubsub
+        if (get(self._options, 'EXPERIMENTAL.ipnsPubsub', false)) {
+          pubsub = new Pubsub(self)
+
+          ipnsStores.push(pubsub)
+        }
+
+        // NOTE: Until the IPNS over DHT is not ready, it is being replaced by the local repo datastore
+        // When DHT is added, If local option enabled, should receive offlineDatastore as well
+        const offlineDatastore = new OfflineDatastore(self._repo)
+        ipnsStores.push(offlineDatastore)
+
+        const routing = new TieredDatastore(ipnsStores)
+        self._ipns = new IPNS(routing, self, pubsub)
+
         self._bitswap = new Bitswap(
           self._libp2pNode,
           self._repo.blocks,

src/core/config.js

@@ -28,6 +28,7 @@ const schema = Joi.object().keys({
   }).allow(null),
   EXPERIMENTAL: Joi.object().keys({
     pubsub: Joi.boolean(),
+    namesysPubsub: Joi.boolean(),
     sharding: Joi.boolean(),
     dht: Joi.boolean()
   }).allow(null),

src/core/index.js

@@ -20,7 +20,7 @@ const EventEmitter = require('events')
 const config = require('./config')
 const boot = require('./boot')
 const components = require('./components')
-const IPNS = require('./ipns')
+
 // replaced by repo-browser when running in the browser
 const defaultRepo = require('./runtime/repo-nodejs')
 const preload = require('./preload')
@@ -89,7 +89,7 @@ class IPFS extends EventEmitter {
     this._ipld = new Ipld(this._blockService)
     this._preload = preload(this)
     this._mfsPreload = mfsPreload(this)
-    this._ipns = new IPNS(null, this)
+    this._ipns = undefined
 
     // IPFS Core exposed components
     //   - for booting up a node
@@ -127,6 +127,9 @@ class IPFS extends EventEmitter {
     if (this._options.EXPERIMENTAL.pubsub) {
       this.log('EXPERIMENTAL pubsub is enabled')
     }
+    if (this._options.EXPERIMENTAL.ipnsPubsub) {
+      this.log('EXPERIMENTAL IPNS pubsub is enabled')
+    }
     if (this._options.EXPERIMENTAL.sharding) {
       this.log('EXPERIMENTAL sharding is enabled')
     }

src/core/ipns/index.js

@@ -16,11 +16,12 @@ const path = require('./path')
 const defaultRecordTtl = 60 * 1000
 
 class IPNS {
-  constructor (routing, ipfs) {
+  constructor (routing, ipfs, pubsub) {
     this.publisher = new IpnsPublisher(routing, ipfs._repo)
     this.republisher = new IpnsRepublisher(this.publisher, ipfs)
     this.resolver = new IpnsResolver(routing, ipfs._repo)
     this.cache = new Receptacle({ max: 1000 }) // Create an LRU cache with max 1000 items
+    this.pubsub = pubsub
   }
 
   // Publish
@@ -53,7 +54,7 @@ class IPNS {
   }
 
   // Resolve
-  resolve (name, peerId, options, callback) {
+  resolve (name, options, callback) {
     // If recursive, we should not try to get the cached value
     if (!options.nocache && !options.recursive) {
       // Try to get the record from cache
@@ -67,7 +68,7 @@ class IPNS {
       }
     }
 
-    this.resolver.resolve(name, peerId, options, (err, result) => {
+    this.resolver.resolve(name, options, (err, result) => {
       if (err) {
         log.error(err)
         return callback(err)

src/core/ipns/path.js

@@ -13,13 +13,13 @@ const resolvePath = (ipfsNode, name, callback) => {
   if (isIPFS.ipnsPath(name)) {
     log(`resolve ipns path ${name}`)
 
-    const local = true // TODO ROUTING - use self._options.local
+    const local = ipfsNode._options.local
 
     const options = {
       local: local
     }
 
-    return ipfsNode._ipns.resolve(name, ipfsNode._peerInfo.id, options, callback)
+    return ipfsNode._ipns.resolve(name, options, callback)
   }
 
   // ipfs path