feat: migrate importer to use IPLD Resolver

Author: daviddias

package.json

@@ -49,7 +49,7 @@
     "run-series": "^1.1.4"
   },
   "dependencies": {
-    "ipfs-merkle-dag": "^0.7.0",
+    "ipld-dag-pb": "^0.0.1",
     "ipfs-unixfs": "^0.1.4",
     "is-ipfs": "^0.2.0",
     "multihashes": "^0.2.2",

src/importer.js

@@ -1,56 +1,67 @@
 'use strict'
 
-const merkleDAG = require('ipfs-merkle-dag')
 const UnixFS = require('ipfs-unixfs')
 const assert = require('assert')
 const pull = require('pull-stream')
 const pushable = require('pull-pushable')
 const write = require('pull-write')
 const parallel = require('run-parallel')
+const dagPB = require('ipld-dag-pb')
+const CID = require('cids')
 
 const fsc = require('./chunker-fixed-size')
 const createAndStoreTree = require('./tree')
 
-const DAGNode = merkleDAG.DAGNode
+const DAGNode = dagPB.DAGNode
+const DAGLink = dagPB.DAGLink
 
 const CHUNK_SIZE = 262144
 
-module.exports = (dagService, options) => {
-  assert(dagService, 'Missing dagService')
+module.exports = (ipldResolver, options) => {
+  assert(ipldResolver, 'Missing IPLD Resolver')
 
   const files = []
 
   const source = pushable()
   const sink = write(
-    makeWriter(source, files, dagService),
+    makeWriter(source, files, ipldResolver),
     null,
     100,
     (err) => {
-      if (err) return source.end(err)
+      if (err) {
+        return source.end(err)
+      }
 
-      createAndStoreTree(files, dagService, source, () => {
+      createAndStoreTree(files, ipldResolver, source, () => {
         source.end()
       })
     }
   )
 
-  return {source, sink}
+  return {
+    source: source,
+    sink: sink
+  }
 }
 
-function makeWriter (source, files, dagService) {
+function makeWriter (source, files, ipldResolver) {
   return (items, cb) => {
     parallel(items.map((item) => (cb) => {
       if (!item.content) {
-        return createAndStoreDir(item, dagService, (err, node) => {
-          if (err) return cb(err)
+        return createAndStoreDir(item, ipldResolver, (err, node) => {
+          if (err) {
+            return cb(err)
+          }
           source.push(node)
           files.push(node)
           cb()
         })
       }
 
-      createAndStoreFile(item, dagService, (err, node) => {
-        if (err) return cb(err)
+      createAndStoreFile(item, ipldResolver, (err, node) => {
+        if (err) {
+          return cb(err)
+        }
         source.push(node)
         files.push(node)
         cb()
@@ -59,16 +70,21 @@ function makeWriter (source, files, dagService) {
   }
 }
 
-function createAndStoreDir (item, ds, cb) {
+function createAndStoreDir (item, ipldResolver, cb) {
   // 1. create the empty dir dag node
   // 2. write it to the dag store
 
   const d = new UnixFS('directory')
   const n = new DAGNode()
   n.data = d.marshal()
 
-  ds.put(n, (err) => {
-    if (err) return cb(err)
+  ipldResolver.put({
+    node: n,
+    cid: new CID(n.multihash())
+  }, (err) => {
+    if (err) {
+      return cb(err)
+    }
     cb(null, {
       path: item.path,
       multihash: n.multihash(),
@@ -78,7 +94,7 @@ function createAndStoreDir (item, ds, cb) {
   })
 }
 
-function createAndStoreFile (file, ds, cb) {
+function createAndStoreFile (file, ipldResolver, cb) {
   if (Buffer.isBuffer(file.content)) {
     file.content = pull.values([file.content])
   }
@@ -102,7 +118,10 @@ function createAndStoreFile (file, ds, cb) {
       const l = new UnixFS('file', Buffer(chunk))
       const n = new DAGNode(l.marshal())
 
-      ds.put(n, (err) => {
+      ipldResolver.put({
+        node: n,
+        cid: new CID(n.multihash())
+      }, (err) => {
         if (err) {
           return cb(new Error('Failed to store chunk'))
         }
@@ -116,7 +135,9 @@ function createAndStoreFile (file, ds, cb) {
       })
     }),
     pull.collect((err, leaves) => {
-      if (err) return cb(err)
+      if (err) {
+        return cb(err)
+      }
 
       if (leaves.length === 1) {
         return cb(null, {
@@ -130,18 +151,23 @@ function createAndStoreFile (file, ds, cb) {
       // create a parent node and add all the leafs
 
       const f = new UnixFS('file')
-      const n = new merkleDAG.DAGNode()
+      const n = new DAGNode()
 
       for (let leaf of leaves) {
         f.addBlockSize(leaf.leafSize)
         n.addRawLink(
-          new merkleDAG.DAGLink(leaf.Name, leaf.Size, leaf.Hash)
+          new DAGLink(leaf.Name, leaf.Size, leaf.Hash)
         )
       }
 
       n.data = f.marshal()
-      ds.put(n, (err) => {
-        if (err) return cb(err)
+      ipldResolver.put({
+        node: n,
+        cid: new CID(n.multihash())
+      }, (err) => {
+        if (err) {
+          return cb(err)
+        }
 
         cb(null, {
           path: file.path,

src/tree.js

@@ -2,12 +2,13 @@
 
 const mh = require('multihashes')
 const UnixFS = require('ipfs-unixfs')
-const merkleDAG = require('ipfs-merkle-dag')
+const CID = require('cids')
+const dagPB = require('ipld-dag-pb')
 
-const DAGLink = merkleDAG.DAGLink
-const DAGNode = merkleDAG.DAGNode
+const DAGLink = dagPB.DAGLink
+const DAGNode = dagPB.DAGNode
 
-module.exports = (files, dagService, source, cb) => {
+module.exports = (files, ipldResolver, source, cb) => {
   // file struct
   // {
   //   path: // full path
@@ -101,7 +102,10 @@ module.exports = (files, dagService, source, cb) => {
     n.data = d.marshal()
 
     pendingWrites++
-    dagService.put(n, (err) => {
+    ipldResolver.put({
+      node: n,
+      cid: new CID(n.multihash())
+    }, (err) => {
       pendingWrites--
       if (err) {
         source.push(new Error('failed to store dirNode'))

test/test-exporter.js

@@ -3,7 +3,7 @@
 
 const expect = require('chai').expect
 const BlockService = require('ipfs-block-service')
-const DAGService = require('ipfs-merkle-dag').DAGService
+const DAGService = require('ipld-resolver')
 const UnixFS = require('ipfs-unixfs')
 const fs = require('fs')
 const path = require('path')

test/test-importer.js

@@ -4,11 +4,11 @@
 const importer = require('./../src').importer
 const expect = require('chai').expect
 const BlockService = require('ipfs-block-service')
-const DAGService = require('ipfs-merkle-dag').DAGService
 const fs = require('fs')
 const path = require('path')
 const pull = require('pull-stream')
 const mh = require('multihashes')
+const IPLDResolver = require('ipld-resolver')
 
 function stringifyMh (files) {
   return files.map((file) => {
@@ -18,8 +18,8 @@ function stringifyMh (files) {
 }
 
 module.exports = function (repo) {
-  describe('importer', function () {
-    let ds
+  describe.only('importer', function () {
+    let ipldResolver
 
     const bigFile = fs.readFileSync(path.join(__dirname, '/test-data/1.2MiB.txt'))
     const smallFile = fs.readFileSync(path.join(__dirname, '/test-data/200Bytes.txt'))
@@ -30,7 +30,7 @@ module.exports = function (repo) {
 
     before(() => {
       const bs = new BlockService(repo)
-      ds = new DAGService(bs)
+      ipldResolver = new IPLDResolver(bs)
     })
 
     it('bad input', (done) => {
@@ -39,7 +39,7 @@ module.exports = function (repo) {
           path: '200Bytes.txt',
           content: 'banana'
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.onEnd((err) => {
           expect(err).to.exist
           done()
@@ -53,7 +53,7 @@ module.exports = function (repo) {
           path: '200Bytes.txt',
           content: pull.values([smallFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
           expect(stringifyMh(files)).to.be.eql([{
@@ -72,7 +72,7 @@ module.exports = function (repo) {
           path: '200Bytes.txt',
           content: smallFile
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
           expect(stringifyMh(files)).to.be.eql([{
@@ -91,7 +91,7 @@ module.exports = function (repo) {
           path: 'foo/bar/200Bytes.txt',
           content: pull.values([smallFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
           expect(files.length).to.equal(3)
@@ -129,7 +129,7 @@ module.exports = function (repo) {
           path: '1.2MiB.txt',
           content: pull.values([bigFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
           expect(stringifyMh(files)).to.be.eql([{
@@ -148,7 +148,7 @@ module.exports = function (repo) {
           path: 'foo-big/1.2MiB.txt',
           content: pull.values([bigFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
 
@@ -176,7 +176,7 @@ module.exports = function (repo) {
         pull.values([{
           path: 'empty-dir'
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
 
@@ -200,7 +200,7 @@ module.exports = function (repo) {
           path: 'pim/1.2MiB.txt',
           content: pull.values([bigFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist
 
@@ -235,7 +235,7 @@ module.exports = function (repo) {
           path: 'pam/1.2MiB.txt',
           content: pull.values([bigFile])
         }]),
-        importer(ds),
+        importer(ipldResolver),
         pull.collect((err, files) => {
           expect(err).to.not.exist