feat: Happy path mfs.write command

Author: achingbrain

src/core/index.js

@@ -3,5 +3,6 @@
 module.exports = {
   ls: require('./ls'),
   mkdir: require('./mkdir'),
-  stat: require('./stat')
+  stat: require('./stat'),
+  write: require('./write')
 }

src/core/stat.js

@@ -71,14 +71,24 @@ module.exports = function mfsStat (ipfs) {
             (next) => ipfs.dag.get(new CID(result.hash), next),
             (result, next) => next(null, result.value),
             (node, next) => {
-              const data = unmarshal(node.data)
+              const meta = unmarshal(node.data)
+
+              let size = 0
+
+              if (meta.data && meta.data.length) {
+                size = meta.data.length
+              }
+
+              if (meta.blockSizes && meta.blockSizes.length) {
+                size = meta.blockSizes.reduce((acc, curr) => acc + curr, 0)
+              }
 
               next(null, {
                 hash: node.multihash,
-                size: data.blockSizes.reduce((acc, curr) => acc + curr, 0),
+                size: size,
                 cumulativeSize: node.size,
-                childBlocks: node.links.length,
-                type: data.type
+                childBlocks: meta.blockSizes.length,
+                type: meta.type
               })
             }
           ], done)

src/core/utils.js

@@ -1,10 +1,20 @@
 'use strict'
 
-const waterfall = require('async/waterfall')
 const Key = require('interface-datastore').Key
 const bs58 = require('bs58')
 const CID = require('cids')
 const log = require('debug')('mfs:utils')
+const UnixFS = require('ipfs-unixfs')
+const dagPb = require('ipld-dag-pb')
+const {
+  DAGNode,
+  DAGLink
+} = dagPb
+const {
+  waterfall,
+  reduce,
+  doWhilst
+} = require('async')
 
 const MFS_ROOT_KEY = new Key('/local/filesroot')
 const FILE_SEPARATOR = '/'
@@ -84,9 +94,171 @@ const updateMfsRoot = (ipfs, buffer, callback) => {
   ], (error) => callback(error, buffer))
 }
 
+const addLink = (ipfs, options, callback) => {
+  options = Object.assign({}, {
+    parent: undefined,
+    child: undefined,
+    name: undefined,
+    flush: true
+  }, options)
+
+  if (!options.parent) {
+    return callback(new Error('No parent passed to addLink'))
+  }
+
+  if (!options.child) {
+    return callback(new Error('No child passed to addLink'))
+  }
+
+  if (!options.name) {
+    return callback(new Error('No name passed to addLink'))
+  }
+
+  waterfall([
+    (done) => {
+      // remove the old link if necessary
+      DAGNode.rmLink(options.parent, options.name, done)
+    },
+    (parent, done) => {
+      // add the new link
+      DAGNode.addLink(parent, new DAGLink(options.name, options.child.size, options.child.hash || options.child.multihash), done)
+    },
+    (parent, done) => {
+      if (!options.flush) {
+        return done()
+      }
+
+      // add the new parent DAGNode
+      ipfs.dag.put(parent, {
+        cid: new CID(parent.hash || parent.multihash)
+      }, (error) => done(error, parent))
+    }
+  ], callback)
+}
+
+const traverseTo = (ipfs, path, options, callback) => {
+  options = Object.assign({}, {
+    parents: false,
+    flush: true
+  }, options)
+
+  waterfall([
+    (done) => withMfsRoot(ipfs, done),
+    (root, done) => {
+      const pathSegments = validatePath(path)
+        .split(FILE_SEPARATOR)
+        .filter(Boolean)
+
+      const trail = []
+
+      waterfall([
+        (cb) => ipfs.dag.get(root, cb),
+        (result, cb) => {
+          const rootNode = result.value
+
+          trail.push({
+            name: FILE_SEPARATOR,
+            node: rootNode,
+            parent: null
+          })
+
+          reduce(pathSegments.map((pathSegment, index) => ({pathSegment, index})), {
+            name: FILE_SEPARATOR,
+            node: rootNode,
+            parent: null
+          }, (parent, {pathSegment, index}, done) => {
+            const lastPathSegment = index === pathSegments.length - 1
+            const existingLink = parent.node.links.find(link => link.name === pathSegment)
+
+            log(`Looking for ${pathSegment} in ${parent.name}`)
+
+            if (!existingLink) {
+              if (!lastPathSegment && !options.parents) {
+                return done(new Error(`Cannot create ${path} - intermediate directory '${pathSegment}' did not exist: Try again with the --parents flag`))
+              }
+
+              log(`Adding empty directory '${pathSegment}' to parent ${parent.name}`)
+
+              return waterfall([
+                (next) => DAGNode.create(new UnixFS('directory').marshal(), [], next),
+                (emptyDirectory, next) => {
+                  addLink(ipfs, {
+                    parent: parent.node,
+                    child: emptyDirectory,
+                    name: pathSegment,
+                    flush: options.flush
+                  }, (error, updatedParent) => {
+                    parent.node = updatedParent
+
+                    next(error, {
+                      name: pathSegment,
+                      node: emptyDirectory,
+                      parent: parent
+                    })
+                  })
+                }
+              ], done)
+            }
+
+            let hash = existingLink.hash || existingLink.multihash
+
+            if (Buffer.isBuffer(hash)) {
+              hash = bs58.encode(hash)
+            }
+
+            // child existed, fetch it
+            ipfs.dag.get(hash, (error, result) => {
+              const child = {
+                name: pathSegment,
+                node: result && result.value,
+                parent: parent
+              }
+
+              trail.push(child)
+
+              done(error, child)
+            })
+          }, cb)
+        }
+      ], done)
+    }
+  ], callback)
+}
+
+const updateTree = (ipfs, child, callback) => {
+  doWhilst(
+    (next) => {
+      if (!child.parent) {
+        const lastChild = child
+        child = null
+        return next(null, lastChild)
+      }
+
+      addLink(ipfs, {
+        parent: child.parent.node,
+        child: child.node,
+        name: child.name,
+        flush: true
+      }, (error, updatedParent) => {
+        child.parent.node = updatedParent
+
+        const lastChild = child
+        child = child.parent
+
+        next(error, lastChild)
+      })
+    },
+    () => Boolean(child),
+    callback
+  )
+}
+
 module.exports = {
   validatePath,
   withMfsRoot,
   updateMfsRoot,
+  traverseTo,
+  addLink,
+  updateTree,
   FILE_SEPARATOR
 }

src/core/write.js

@@ -0,0 +1,125 @@
+'use strict'
+
+const importer = require('ipfs-unixfs-engine').importer
+const promisify = require('promisify-es6')
+const CID = require('cids')
+const pull = require('pull-stream')
+const {
+  collect,
+  values
+} = pull
+const {
+  waterfall
+} = require('async')
+const {
+  updateMfsRoot,
+  validatePath,
+  traverseTo,
+  addLink,
+  updateTree,
+  FILE_SEPARATOR
+} = require('./utils')
+
+const defaultOptions = {
+  offset: 0,
+  count: undefined,
+  create: false,
+  truncate: false,
+  length: undefined,
+  rawLeaves: false,
+  cidVersion: undefined,
+  hash: undefined,
+  parents: false,
+  progress: undefined,
+  strategy: 'balanced',
+  flush: true
+}
+
+module.exports = function mfsWrite (ipfs) {
+  return promisify((path, buffer, options, callback) => {
+    if (typeof options === 'function') {
+      callback = options
+      options = {}
+    }
+
+    options = Object.assign({}, defaultOptions, options)
+
+    try {
+      path = validatePath(path)
+    } catch (error) {
+      return callback(error)
+    }
+
+    if (options.count === 0) {
+      return callback()
+    }
+
+    if (options.count) {
+      buffer = buffer.slice(0, options.count)
+    }
+
+    const parts = path
+      .split(FILE_SEPARATOR)
+      .filter(Boolean)
+    const fileName = parts.pop()
+
+    waterfall([
+      // walk the mfs tree to the containing folder node
+      (done) => traverseTo(ipfs, `${FILE_SEPARATOR}${parts.join(FILE_SEPARATOR)}`, options, done),
+      (containingFolder, done) => {
+        waterfall([
+          (next) => {
+            const existingChild = containingFolder.node.links.reduce((last, child) => {
+              if (child.name === fileName) {
+                return child
+              }
+
+              return last
+            }, null)
+
+            if (existingChild) {
+              // overwrite the existing file or part of it
+              return next(new Error('Not implemented yet!'))
+            } else {
+              // import the file to IPFS and add it as a child of the containing directory
+              return pull(
+                values([{
+                  content: buffer
+                }]),
+                importer(ipfs._ipld, {
+                  progress: options.progress,
+                  hashAlg: options.hash,
+                  cidVersion: options.cidVersion,
+                  strategy: options.strategy
+                }),
+                collect(next)
+              )
+            }
+          },
+          // load the DAGNode corresponding to the newly added/updated file
+          (results, next) => ipfs.dag.get(new CID(results[0].multihash), next),
+          (result, next) => {
+            // link the newly added DAGNode to the containing older
+            waterfall([
+              (cb) => addLink(ipfs, {
+                parent: containingFolder.node,
+                child: result.value,
+                name: fileName
+              }, cb),
+              (newContainingFolder, cb) => {
+                containingFolder.node = newContainingFolder
+
+                // update all the parent node CIDs
+                updateTree(ipfs, containingFolder, cb)
+              }
+            ], next)
+          },
+          (result, next) => {
+            // update new MFS root CID
+            updateMfsRoot(ipfs, result.node.multihash, next)
+          }
+        ], done)
+      }
+    ], callback)
+  })
+}

test/fixtures/index.js

@@ -8,11 +8,7 @@ const {
   race,
   waterfall
 } = require('async')
-const {
-  ls,
-  mkdir,
-  stat
-} = require('../../src/core')
+const core = require('../../src/core')
 
 const createMfs = promisify((cb) => {
   let node = ipfs.createNode({
@@ -25,12 +21,17 @@ const createMfs = promisify((cb) => {
       (next) => node.once('ready', next)
     ], (error) => done(error, node)),
     (node, done) => {
-      done(null, {
-        ls: ls(node),
-        mkdir: mkdir(node),
-        stat: stat(node),
-        node: node
-      })
+      const mfs = {
+        node
+      }
+
+      for (let key in core) {
+        if (core.hasOwnProperty(key)) {
+          mfs[key] = core[key](node)
+        }
+      }
+
+      done(null, mfs)
     }
   ], cb)
 })

test/fixtures/large-file.jpg

@@ -0,0 +1 @@
+Hello world!