fix: dht validate if receiving stream

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

@@ -3,6 +3,8 @@
 const promisify = require('promisify-es6')
 const streamToValue = require('../utils/stream-to-value')
 
+const errcode = require('err-code')
+
 module.exports = (send) => {
   return promisify((peerId, opts, callback) => {
     if (typeof opts === 'function' && !callback) {
@@ -17,10 +19,60 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    const handleResult = (res, callback) => {
+      // TODO verify - Inconsistent return values in the browser vs node
+      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) => {
+        // TODO inconsistencies node / browser ... go?
+        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)
+      }
+
+      if (typeof result.pipe === 'function') {
+        streamToValue(result, (err, res) => {
+          if (err) {
+            return callback(err)
+          }
+          handleResult(res, callback)
+        })
+      } else {
+        handleResult(result, callback)
+      }
+    })
   })
 }

src/dht/findprovs.js

@@ -17,10 +17,44 @@ module.exports = (send) => {
       opts = {}
     }
 
-    send.andTransform({
+    // TODO create util for the 3?
+    const handleResult = (res, callback) => {
+      // TODO VERIFY - 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) => {
+      // TODO utility
+      if (err) {
+        return callback(err)
+      }
+
+      if (typeof result.pipe === 'function') {
+        streamToValue(result, (err, res) => {
+          if (err) {
+            return callback(err)
+          }
+          handleResult(res, callback)
+        })
+      } else {
+        handleResult(result, 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)
+      }
+    })
   })
 }