feat: sync with go-ipfs 0.5 (#3013)

New features:

- `ipfs.dht.get` accepts string or buffer
- http-client `ipfs.dht.get` returns a buffer to match core api

Breaking changes:

- ipfs.ls remove all options in core api, but keep `long` in http api and cli same as go-ipfs
  - remove sort from the cli and ignore go-ipfs `U` option
  - enable http api mfs tests
- key.gen defaults, to rsa and 2048
- pin.list only supports streaming responses
- -U argument to `files.ls` has been removed
- dht put uses body to send data instead of query string 

Co-authored-by: Alex Potsides <alex@achingbrain.net>

docs/core-api/DHT.md

@@ -126,7 +126,7 @@ A great source of [examples][] can be found in the tests for this API.
 
 | Name | Type | Description |
 | ---- | ---- | ----------- |
-| key | Buffer | The key associated with the value to find |
+| key | `Buffer` or `string` | The key associated with the value to find |
 
 ### Options
 

docs/core-api/FILES.md

@@ -916,7 +916,6 @@ An optional object which may have the following keys:
 
 | Name | Type | Default | Description |
 | ---- | ---- | ------- | ----------- |
-| sort | `boolean` | `false` | If true entries will be sorted by filename |
 | timeout | `Number` | `undefined` | A timeout in ms |
 | signal | [AbortSignal][] | `undefined` |  Can be used to cancel any long running requests started as a result of this call |
 

docs/core-api/KEY.md

@@ -289,4 +289,4 @@ A great source of [examples][] can be found in the tests for this API.
 
 [examples]: https://github.com/ipfs/js-ipfs/blob/master/packages/interface-ipfs-core/src/key
 [cid]: https://www.npmjs.com/package/cids
-[AbortSignal]: https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal
+[AbortSignal]: https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal

packages/interface-ipfs-core/src/bootstrap/rm.js

@@ -61,7 +61,7 @@ module.exports = (common, options) => {
       const rmRes = await ipfs.bootstrap.rm(null, { all: true })
       const removedPeers = rmRes.Peers
 
-      expect(removedPeers).to.eql(addedPeers)
+      expect(removedPeers.sort()).to.deep.equal(addedPeers.sort())
     })
   })
 }

packages/interface-ipfs-core/src/dht/get.js

@@ -1,10 +1,9 @@
 /* eslint-env mocha */
 'use strict'
 
-const { Buffer } = require('buffer')
 const { getDescribe, getIt, expect } = require('../utils/mocha')
 const testTimeout = require('../utils/test-timeout')
-const drain = require('it-drain')
+const all = require('it-all')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
 /**
@@ -15,9 +14,7 @@ module.exports = (common, options) => {
   const describe = getDescribe(options)
   const it = getIt(options)
 
-  describe.skip('.dht.get', function () {
-    this.timeout(80 * 1000)
-
+  describe('.dht.get', function () {
     let nodeA
     let nodeB
 
@@ -30,12 +27,10 @@ module.exports = (common, options) => {
     after(() => common.clean())
 
     it('should respect timeout option when getting a value from the DHT', async () => {
-      const key = Buffer.from('/ipfs/QmUNLLsPACCz1vLxQVkXqqLX5R1X345qqfHbsf67hvA3Nn')
-      const data = Buffer.from('data')
-
-      await drain(nodeA.dht.put(key, data, { verbose: true }))
+      const [data] = await all(nodeA.add('should put a value to the DHT'))
+      const publish = await nodeA.name.publish(data.cid)
 
-      await testTimeout(() => nodeB.dht.get(key, {
+      await testTimeout(() => nodeB.dht.get(`/ipns/${publish.name}`, {
         timeout: 1
       }))
     })
@@ -46,13 +41,11 @@ module.exports = (common, options) => {
     })
 
     it('should get a value after it was put on another node', async () => {
-      const key = Buffer.from('/ipfs/QmUNLLsPACCz1vLxQVkXqqLX5R1X345qqfHbsf67hvA3Nn')
-      const value = Buffer.from('data')
-
-      await drain(nodeB.dht.put(key, value))
-      const result = await nodeA.dht.get(key)
+      const [data] = await all(nodeA.add('should put a value to the DHT'))
+      const publish = await nodeA.name.publish(data.cid)
+      const record = await nodeA.dht.get(`/ipns/${publish.name}`)
 
-      expect(result).to.eql(value)
+      expect(record.toString()).to.contain(data.cid.toString())
     })
   })
 }

packages/interface-ipfs-core/src/dht/put.js

@@ -2,10 +2,10 @@
 'use strict'
 
 const { Buffer } = require('buffer')
-const { getDescribe, getIt } = require('../utils/mocha')
-const drain = require('it-drain')
+const { getDescribe, getIt, expect } = require('../utils/mocha')
 const testTimeout = require('../utils/test-timeout')
 const CID = require('cids')
+const all = require('it-all')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
 /**
@@ -16,10 +16,7 @@ module.exports = (common, options) => {
   const describe = getDescribe(options)
   const it = getIt(options)
 
-  // TODO unskip this after go-ipfs 0.5.0 ships interface is going to change
-  describe.skip('.dht.put', function () {
-    this.timeout(80 * 1000)
-
+  describe('.dht.put', function () {
     let nodeA
     let nodeB
 
@@ -38,12 +35,13 @@ module.exports = (common, options) => {
     })
 
     it('should put a value to the DHT', async function () {
-      this.timeout(80 * 1000)
-
-      const key = Buffer.from('/ipfs/QmUNLLsPACCz1vLxQVkXqqLX5R1X345qqfHbsf67hvA3Nn')
-      const data = Buffer.from('data')
-
-      await drain(nodeA.dht.put(key, data, { verbose: true }))
+      const [data] = await all(nodeA.add('should put a value to the DHT'))
+      const publish = await nodeA.name.publish(data.cid)
+      const record = await nodeA.dht.get(`/ipns/${publish.name}`)
+      const value = await all(nodeA.dht.put(`/ipns/${publish.name}`, record, { verbose: true }))
+      expect(value).to.has.length(3)
+      expect(value[2].id.toString()).to.be.equal(nodeB.peerId.id)
+      expect(value[2].type).to.be.equal(5)
     })
   })
 }

packages/interface-ipfs-core/src/pubsub/publish.js

@@ -4,7 +4,7 @@
 const { Buffer } = require('buffer')
 const { nanoid } = require('nanoid')
 const { getTopic } = require('./utils')
-const { getDescribe, getIt } = require('../utils/mocha')
+const { getDescribe, getIt, expect } = require('../utils/mocha')
 const testTimeout = require('../utils/test-timeout')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
@@ -38,6 +38,11 @@ module.exports = (common, options) => {
       return ipfs.pubsub.publish(topic, 'hello friend')
     })
 
+    it('should fail with undefined msg', async () => {
+      const topic = getTopic()
+      await expect(ipfs.pubsub.publish(topic)).to.eventually.rejectedWith('argument "data" is required')
+    })
+
     it('should publish message from buffer', () => {
       const topic = getTopic()
       return ipfs.pubsub.publish(topic, Buffer.from(nanoid()))

packages/interface-ipfs-core/src/pubsub/subscribe.js

@@ -2,12 +2,14 @@
 'use strict'
 
 const { Buffer } = require('buffer')
+const { nanoid } = require('nanoid')
 const pushable = require('it-pushable')
 const all = require('it-all')
 const { waitForPeers, getTopic } = require('./utils')
 const { getDescribe, getIt, expect } = require('../utils/mocha')
 const delay = require('delay')
-const { isWebWorker } = require('ipfs-utils/src/env')
+const AbortController = require('abort-controller')
+const { isWebWorker, isNode } = require('ipfs-utils/src/env')
 
 /** @typedef { import("ipfsd-ctl/src/factory") } Factory */
 /**
@@ -163,6 +165,66 @@ module.exports = (common, options) => {
         return ipfs1.swarm.connect(ipfs2Addr)
       })
 
+      it('should receive messages from a different node with floodsub', async function () {
+        if (!isNode) {
+          return this.skip()
+        }
+        const expectedString = 'should receive messages from a different node with floodsub'
+        const topic = `floodsub-${nanoid()}`
+        const ipfs1 = (await common.spawn({
+          ipfsOptions: {
+            config: {
+              Pubsub: {
+                Router: 'floodsub'
+              }
+            }
+          }
+        })).api
+        const ipfs2 = (await common.spawn({
+          type: isWebWorker ? 'go' : undefined,
+          ipfsOptions: {
+            config: {
+              Pubsub: {
+                Router: 'floodsub'
+              }
+            }
+          }
+        })).api
+        await ipfs1.swarm.connect(ipfs2.peerId.addresses[0])
+
+        const msgStream1 = pushable()
+        const msgStream2 = pushable()
+
+        const sub1 = msg => {
+          msgStream1.push(msg)
+          msgStream1.end()
+        }
+        const sub2 = msg => {
+          msgStream2.push(msg)
+          msgStream2.end()
+        }
+
+        const abort1 = new AbortController()
+        const abort2 = new AbortController()
+        await Promise.all([
+          ipfs1.pubsub.subscribe(topic, sub1, { signal: abort1.signal }),
+          ipfs2.pubsub.subscribe(topic, sub2, { signal: abort2.signal })
+        ])
+
+        await waitForPeers(ipfs2, topic, [ipfs1.peerId.id], 30000)
+        await ipfs2.pubsub.publish(topic, Buffer.from(expectedString))
+
+        const [sub1Msg] = await all(msgStream1)
+        expect(sub1Msg.data.toString()).to.be.eql(expectedString)
+        expect(sub1Msg.from).to.eql(ipfs2.peerId.id)
+
+        const [sub2Msg] = await all(msgStream2)
+        expect(sub2Msg.data.toString()).to.be.eql(expectedString)
+        expect(sub2Msg.from).to.eql(ipfs2.peerId.id)
+        abort1.abort()
+        abort2.abort()
+      })
+
       it('should receive messages from a different node', async () => {
         const expectedString = 'hello from the other side'
 
@@ -184,7 +246,7 @@ module.exports = (common, options) => {
         ])
 
         await waitForPeers(ipfs2, topic, [ipfs1.peerId.id], 30000)
-
+        await delay(5000) // gossipsub need this delay https://github.com/libp2p/go-libp2p-pubsub/issues/331
         await ipfs2.pubsub.publish(topic, Buffer.from(expectedString))
 
         const [sub1Msg] = await all(msgStream1)
@@ -218,7 +280,7 @@ module.exports = (common, options) => {
         ])
 
         await waitForPeers(ipfs2, topic, [ipfs1.peerId.id], 30000)
-
+        await delay(5000) // gossipsub need this delay https://github.com/libp2p/go-libp2p-pubsub/issues/331
         await ipfs2.pubsub.publish(topic, buffer)
 
         const [sub1Msg] = await all(msgStream1)
@@ -256,7 +318,7 @@ module.exports = (common, options) => {
         ])
 
         await waitForPeers(ipfs2, topic, [ipfs1.peerId.id], 30000)
-
+        await delay(5000) // gossipsub need this delay https://github.com/libp2p/go-libp2p-pubsub/issues/331
         outbox.forEach(msg => ipfs2.pubsub.publish(topic, Buffer.from(msg)))
 
         const sub1Msgs = await all(msgStream1)
@@ -290,7 +352,7 @@ module.exports = (common, options) => {
         ])
 
         await waitForPeers(ipfs1, topic, [ipfs2.peerId.id], 30000)
-
+        await delay(5000) // gossipsub need this delay https://github.com/libp2p/go-libp2p-pubsub/issues/331
         const startTime = new Date().getTime()
 
         for (let i = 0; i < count; i++) {

packages/ipfs-http-client/package.json

@@ -70,7 +70,7 @@
   "devDependencies": {
     "aegir": "^22.0.0",
     "cross-env": "^7.0.0",
-    "go-ipfs-dep": "0.4.23-3",
+    "go-ipfs-dep": "^0.5.1",
     "interface-ipfs-core": "^0.135.1",
     "ipfsd-ctl": "^4.1.1",
     "it-all": "^1.0.1",

packages/ipfs-http-client/src/dht/get.js

@@ -1,30 +1,25 @@
 'use strict'
 
 const { Buffer } = require('buffer')
-const encodeBufferURIComponent = require('../lib/encode-buffer-uri-component')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
 const { Value } = require('./response-types')
 
 module.exports = configure(api => {
   return async function get (key, options = {}) {
-    if (!Buffer.isBuffer(key)) {
-      throw new Error('invalid key')
-    }
-
     const res = await api.post('dht/get', {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        key: encodeBufferURIComponent(key),
+        arg: Buffer.isBuffer(key) ? key.toString() : key,
         ...options
       }),
       headers: options.headers
     })
 
     for await (const message of res.ndjson()) {
       if (message.Type === Value) {
-        return message.Extra
+        return Buffer.from(message.Extra, 'base64')
       }
     }
 

packages/ipfs-http-client/src/dht/put.js

@@ -5,20 +5,20 @@ const multiaddr = require('multiaddr')
 const toCamel = require('../lib/object-to-camel')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
+const multipartRequest = require('../lib/multipart-request')
 
 module.exports = configure(api => {
   return async function * put (key, value, options = {}) {
     const res = await api.post('dht/put', {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: [
-          key,
-          value
-        ],
+        arg: key,
         ...options
       }),
-      headers: options.headers
+      ...(
+        await multipartRequest(value, options.headers)
+      )
     })
 
     for await (let message of res.ndjson()) {

packages/ipfs-http-client/src/files/ls.js

@@ -17,12 +17,8 @@ module.exports = configure(api => {
       signal: options.signal,
       searchParams: toUrlSearchParams({
         arg: CID.isCID(path) ? `/ipfs/${path}` : path,
-
-        // TODO the args below are not in the go-ipfs or interface core docs
-        long: options.long == null ? true : options.long,
-
-        // TODO: remove after go-ipfs 0.5 is released
-        l: options.long == null ? true : options.long,
+        // default long to true, diverges from go-ipfs where its false by default
+        long: true,
         ...options,
         stream: true
       }),

packages/ipfs-http-client/src/lib/multipart-request.js

@@ -7,7 +7,7 @@ const modeToString = require('../lib/mode-to-string')
 const mtimeToObject = require('../lib/mtime-to-object')
 const merge = require('merge-options').bind({ ignoreUndefined: true })
 
-async function multipartRequest (source, abortController, headers = {}, boundary = `-----------------------------${nanoid()}`) {
+async function multipartRequest (source = '', abortController, headers = {}, boundary = `-----------------------------${nanoid()}`) {
   async function * streamFiles (source) {
     try {
       let index = 0