fix: dht validate if receiving stream

Author: vasco-santos

package.json

@@ -36,6 +36,7 @@
     "debug": "^4.1.0",
     "detect-node": "^2.0.4",
     "end-of-stream": "^1.4.1",
+    "err-code": "^1.1.2",
     "flatmap": "0.0.3",
     "glob": "^7.1.3",
     "ipfs-block": "~0.8.0",
@@ -85,7 +86,7 @@
     "eslint-plugin-react": "^7.11.1",
     "go-ipfs-dep": "~0.4.18",
     "gulp": "^3.9.1",
-    "interface-ipfs-core": "~0.84.3",
+    "interface-ipfs-core": "ipfs/interface-ipfs-core#fix/update-dht-responses",
     "ipfsd-ctl": "~0.40.0",
     "pull-stream": "^3.6.9",
     "stream-equal": "^1.1.1"

src/dht/findpeer.js

@@ -2,6 +2,9 @@
 
 const promisify = require('promisify-es6')
 const streamToValue = require('../utils/stream-to-value')
+const streamToValueWithTransformer = require('../utils/stream-to-value-with-transformer')
+
+const errcode = require('err-code')
 
 module.exports = (send) => {
   return promisify((peerId, opts, callback) => {
@@ -17,10 +20,51 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      // Inconsistent return values in the browser
+      if (Array.isArray(res)) {
+        res = res[0]
+      }
+
+      // Type 2 keys
+      if (res.Type !== 2) {
+        const errMsg = `key was not found (type 2)`
+
+        log.error(errMsg)
+        return callback(errcode(new Error(errMsg), 'ERR_KEY_TYPE_2_NOT_FOUND'))
+      }
+
+      const id = res.Responses[0].ID
+      const addresses = res.Responses[0].Addrs.map((addr) => {
+        // inconsistencies js / go - go does not add `/ipfs/{id}` to the address
+        if (addr.split('/ipfs/') > - 1) {
+          return addr
+        } else {
+          return `${addr}/ipfs/${id}`
+        }
+      })
+
+      const response = {
+        ...res,
+        Responses: [{
+          ID: id,
+          Addrs: addresses
+        }]
+      }
+
+      callback(null, response)
+    }
+
+    send({
       path: 'dht/findpeer',
       args: peerId,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      streamToValueWithTransformer(result, handleResult, callback)
+    })
   })
 }

src/dht/findprovs.js

@@ -17,10 +17,33 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      // Inconsistent return values in the browser vs node
+      if (Array.isArray(res)) {
+        res = res[0]
+      }
+
+      // Type 4 keys
+      if (res.Type !== 4) {
+        const errMsg = `key was not found (type 4)`
+
+        log.error(errMsg)
+        return callback(errcode(new Error(errMsg), 'ERR_KEY_TYPE_4_NOT_FOUND'))
+      }
+
+      callback(null, res)
+    }
+
+    send({
       path: 'dht/findprovs',
       args: cid,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      streamToValueWithTransformer(result, handleResult, callback)
+    })
   })
 }

src/dht/query.js

@@ -17,10 +17,20 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    send({
       path: 'dht/query',
       args: peerId,
       qs: opts
-    }, streamToValue, callback)
+    }, (err, result) => {
+      if (err) {
+        return callback(err)
+      }
+
+      if (typeof result.pipe === 'function') {
+        streamToValue(result, callback)
+      } else {
+        callback(null, result)
+      }
+    })
   })
 }

src/utils/stream-to-value-with-transformer.js

@@ -0,0 +1,18 @@
+'use strict'
+
+const streamToValue = require('./stream-to-value')
+
+function streamToValueWithTransformer (response, transformer, callback) {
+  if (typeof response.pipe === 'function') {
+    streamToValue(response, (err, res) => {
+      if (err) {
+        return callback(err)
+      }
+      transformer(res, callback)
+    })
+  } else {
+    transformer(response, callback)
+  }
+}
+
+module.exports = streamToValueWithTransformer