feat: ipns over pubsub

README.md

@@ -305,6 +305,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`.**
 
@@ -563,6 +564,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.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)
   - [`ipfs.name.resolve(value, [options], [callback])`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/NAME.md#nameresolve)
 
 #### Crypto and Key Management

package.json

@@ -69,7 +69,7 @@
     "execa": "^1.0.0",
     "form-data": "^2.3.3",
     "hat": "0.0.3",
-    "interface-ipfs-core": "~0.88.0",
+    "interface-ipfs-core": "ipfs/interface-ipfs-core#fix/ipns-over-pubsub-tests",
     "ipfsd-ctl": "~0.40.1",
     "ncp": "^2.0.0",
     "qs": "^6.5.2",
@@ -88,9 +88,12 @@
     "byteman": "^1.3.5",
     "cid-tool": "~0.2.0",
     "cids": "~0.5.5",
+    "class-is": "^1.1.0",
     "datastore-core": "~0.6.0",
+    "datastore-pubsub": "~0.1.1",
     "debug": "^4.1.0",
     "deep-extend": "~0.6.0",
+    "dlv": "^1.1.2",
     "err-code": "^1.1.2",
     "file-type": "^10.2.0",
     "fnv1a": "^1.0.1",
@@ -118,7 +121,7 @@
     "ipld-ethereum": "^2.0.1",
     "ipld-git": "~0.2.2",
     "ipld-zcash": "~0.1.6",
-    "ipns": "~0.3.0",
+    "ipns": "~0.4.3",
     "is-ipfs": "~0.4.7",
     "is-pull-stream": "~0.0.0",
     "is-stream": "^1.1.0",

src/cli/commands/daemon.js

@@ -27,6 +27,9 @@ module.exports = {
       })
       .option('local', {
         desc: 'Run commands locally to the daemon',
+        default: false
+      })
+      .option('enable-namesys-pubsub', {
         type: 'boolean',
         default: false
       })

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.forEach((s) => {
+          print(s)
+        })
+      }
+    })
+  }
+}

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,92 @@
+'use strict'
+
+const debug = require('debug')
+const errcode = require('err-code')
+const promisify = require('promisify-es6')
+
+const IpnsPubsubDatastore = require('../ipns/routing/pubsub-datastore')
+
+const log = debug('jsipfs:name-pubsub')
+log.error = debug('jsipfs:name-pubsub:error')
+
+// Is pubsub enabled
+const isNamePubsubEnabled = (node) => {
+  try {
+    return Boolean(getPubsubRouting(node))
+  } catch (err) {
+    return false
+  }
+}
+
+// Get pubsub from IPNS routing
+const getPubsubRouting = (node) => {
+  if (!node._ipns || !node._options.EXPERIMENTAL.ipnsPubsub) {
+    const errMsg = 'IPNS pubsub subsystem is not enabled'
+
+    throw errcode(errMsg, 'ERR_IPNS_PUBSUB_NOT_ENABLED')
+  }
+
+  // Only one store and it is pubsub
+  if (IpnsPubsubDatastore.isIpnsPubsubDatastore(node._ipns.routing)) {
+    return node._ipns.routing
+  }
+
+  // Find in tiered
+  const pubsub = (node._ipns.routing.stores || []).find(s => IpnsPubsubDatastore.isIpnsPubsubDatastore(s))
+
+  if (!pubsub) {
+    const errMsg = 'IPNS pubsub datastore not found'
+
+    throw errcode(errMsg, 'ERR_PUBSUB_DATASTORE_NOT_FOUND')
+  }
+
+  return pubsub
+}
+
+module.exports = function namePubsub (self) {
+  return {
+    /**
+     * Query the state of IPNS pubsub.
+     *
+     * @returns {Promise|void}
+     */
+    state: promisify((callback) => {
+      callback(null, {
+        enabled: isNamePubsubEnabled(self)
+      })
+    }),
+    /**
+     * Cancel a name subscription.
+     *
+     * @param {String} name subscription name.
+     * @param {function(Error)} [callback]
+     * @returns {Promise|void}
+     */
+    cancel: promisify((name, callback) => {
+      let pubsub
+      try {
+        pubsub = getPubsubRouting(self)
+      } catch (err) {
+        return callback(err)
+      }
+
+      pubsub.cancel(name, callback)
+    }),
+    /**
+     * Show current name subscriptions.
+     *
+     * @param {function(Error)} [callback]
+     * @returns {Promise|void}
+     */
+    subs: promisify((callback) => {
+      let pubsub
+      try {
+        pubsub = getPubsubRouting(self)
+      } catch (err) {
+        return callback(err)
+      }
+
+      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')
 
@@ -161,6 +162,7 @@ module.exports = function name (self) {
       }
 
       self._ipns.resolve(name, resolveOptions, callback)
-    })
+    }),
+    pubsub: namePubsub(self)
   }
 }

src/core/components/pubsub.js

@@ -8,6 +8,8 @@ const errPubsubDisabled = () => {
   return errCode(new Error('pubsub experiment is not enabled'), 'ERR_PUBSUB_DISABLED')
 }
 
+const pubsubEnabled = (options) => options.EXPERIMENTAL.pubsub || options.EXPERIMENTAL.ipnsPubsub
+
 module.exports = function pubsub (self) {
   return {
     subscribe: (topic, handler, options, callback) => {
@@ -16,7 +18,7 @@ module.exports = function pubsub (self) {
         options = {}
       }
 
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         return callback
           ? setImmediate(() => callback(errPubsubDisabled()))
           : Promise.reject(errPubsubDisabled())
@@ -37,7 +39,7 @@ module.exports = function pubsub (self) {
     },
 
     unsubscribe: (topic, handler, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         return callback
           ? setImmediate(() => callback(errPubsubDisabled()))
           : Promise.reject(errPubsubDisabled())
@@ -53,28 +55,28 @@ module.exports = function pubsub (self) {
     },
 
     publish: promisify((topic, data, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self._libp2pNode.pubsub.publish(topic, data, callback)
     }),
 
     ls: promisify((callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self._libp2pNode.pubsub.ls(callback)
     }),
 
     peers: promisify((topic, callback) => {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         return setImmediate(() => callback(errPubsubDisabled()))
       }
       self._libp2pNode.pubsub.peers(topic, callback)
     }),
 
     setMaxListeners (n) {
-      if (!self._options.EXPERIMENTAL.pubsub) {
+      if (!pubsubEnabled(self._options)) {
         throw errPubsubDisabled()
       }
       self._libp2pNode.pubsub.setMaxListeners(n)

src/core/components/start.js

@@ -2,11 +2,13 @@
 
 const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
+const get = require('dlv')
 const setImmediate = require('async/setImmediate')
 const promisify = require('promisify-es6')
 const { TieredDatastore } = require('datastore-core')
 
 const IPNS = require('../ipns')
+const PubsubDatastore = require('../ipns/routing/pubsub-datastore')
 const OfflineDatastore = require('../ipns/routing/offline-datastore')
 
 module.exports = (self) => {
@@ -41,7 +43,16 @@ module.exports = (self) => {
         // Setup online routing for IPNS with a tiered routing composed by a DHT and a Pubsub router (if properly enabled)
         const ipnsStores = []
 
-        // TODO Add IPNS pubsub if enabled
+        // Add IPNS pubsub if enabled
+        let pubsubDs
+        if (get(self._options, 'EXPERIMENTAL.ipnsPubsub', false)) {
+          const pubsub = self._libp2pNode.pubsub
+          const localDatastore = self._repo.datastore
+          const peerId = self._peerInfo.id
+
+          pubsubDs = new PubsubDatastore(pubsub, localDatastore, peerId)
+          ipnsStores.push(pubsubDs)
+        }
 
         // NOTE: IPNS routing is being replaced by the local repo datastore while the IPNS over DHT is not ready
         // When DHT is added, if local option enabled, should receive offlineDatastore as well

src/core/config.js

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

src/core/index.js

@@ -165,6 +165,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')
     }