feat: Replacing chunks of files that do not increase the size of the file

Author: achingbrain

package.json

@@ -52,7 +52,9 @@
     "ipfs-unixfs": "^0.1.14",
     "ipfs-unixfs-engine": "^0.28.1",
     "promisify-es6": "^1.0.3",
-    "pull-stream": "^3.6.7"
+    "pull-paramap": "^1.2.2",
+    "pull-stream": "^3.6.7",
+    "pull-traverse": "^1.0.3"
   },
   "pre-commit": [
     "lint",

src/core/read.js

@@ -14,6 +14,7 @@ const {
   validatePath,
   traverseTo
 } = require('./utils')
+const log = require('debug')('mfs:read')
 
 const defaultOptions = {
   offset: 0,
@@ -35,6 +36,8 @@ module.exports = function mfsRead (ipfs) {
       return callback(error)
     }
 
+    log(`Reading ${path}`)
+
     waterfall([
       (done) => traverseTo(ipfs, path, {
         parents: false

src/core/utils.js

@@ -116,19 +116,19 @@ const addLink = (ipfs, options, callback) => {
 
   waterfall([
     (done) => {
-      // remove the old link if necessary
+      // Remove the old link if necessary
       DAGNode.rmLink(options.parent, options.name, done)
     },
     (parent, done) => {
-      // add the new link
+      // Add the new link to the parent
       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
+      // Persist the new parent DAGNode
       ipfs.dag.put(parent, {
         cid: new CID(parent.hash || parent.multihash)
       }, (error) => done(error, parent))
@@ -167,14 +167,13 @@ const traverseTo = (ipfs, path, options, callback) => {
             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`))
+              if (!options.parents) {
+                return done(new Error(`Cannot traverse to ${path} - '${pathSegment}' did not exist: Try again with the --parents flag`))
               }
 
               log(`Adding empty directory '${pathSegment}' to parent ${parent.name}`)

src/core/write.js

@@ -0,0 +1,15 @@
+'use strict'
+
+const unmarshal = require('ipfs-unixfs').unmarshal
+
+const findFileSize = (dagNode) => {
+  const meta = unmarshal(dagNode.data)
+
+  if (meta.blockSizes.length) {
+    return meta.blockSizes.reduce((acc, curr) => acc + curr, 0)
+  }
+
+  return meta.data.length
+}
+
+module.exports = findFileSize

src/core/write/find-file-size.js

@@ -0,0 +1,38 @@
+'use strict'
+
+const {
+  waterfall
+} = require('async')
+const pull = require('pull-stream')
+const {
+  values,
+  collect
+} = pull
+const importer = require('ipfs-unixfs-engine').importer
+
+const importNode = (ipfs, parent, fileName, buffer, options, callback) => {
+  waterfall([
+    (done) => pull(
+      values([{
+        content: buffer
+      }]),
+      importer(ipfs._ipld, {
+        progress: options.progress,
+        hashAlg: options.hash,
+        cidVersion: options.cidVersion,
+        strategy: options.strategy
+      }),
+      collect(done)
+    ),
+    (results, done) => {
+      const imported = results[0]
+
+      return done(null, {
+        size: imported.size,
+        multihash: imported.multihash
+      })
+    }
+  ], callback)
+}
+
+module.exports = importNode

src/core/write/import-node.js

@@ -0,0 +1,129 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+const CID = require('cids')
+const {
+  waterfall
+} = require('async')
+const {
+  updateMfsRoot,
+  validatePath,
+  traverseTo,
+  addLink,
+  updateTree,
+  FILE_SEPARATOR
+} = require('../utils')
+const log = require('debug')('mfs:write')
+const bs58 = require('bs58')
+const importNode = require('./import-node')
+const updateNode = require('./update-node')
+
+const defaultOptions = {
+  offset: 0, // the offset in the file to begin writing
+  length: undefined, // how many bytes from the incoming buffer to write
+  create: false, // whether to create the file if it does not exist
+  truncate: false, // whether to truncate the file first
+  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.offset < 0) {
+      return callback(new Error('cannot have negative write offset'))
+    }
+
+    if (options.length < 0) {
+      return callback(new Error('cannot have negative byte count'))
+    }
+
+    if (options.length === 0) {
+      return callback()
+    }
+
+    if (options.length) {
+      buffer = buffer.slice(0, options.length)
+    }
+
+    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) {
+              log('Updating linked DAGNode', bs58.encode(existingChild.multihash))
+              // overwrite the existing file or part of it, possibly truncating what's left
+              updateNode(ipfs, new CID(existingChild.multihash), buffer, options, next)
+            } else {
+              if (!options.create) {
+                return next(new Error('file does not exist'))
+              }
+
+              if (options.offset) {
+                // pad the start of the file with zeros
+                buffer = Buffer.concat([Buffer.alloc(options.offset, 0), buffer])
+              }
+
+              log('Importing file', fileName, buffer.length, 'bytes')
+              importNode(ipfs, containingFolder, fileName, buffer, options, next)
+            }
+          },
+
+          // Add or replace the DAGLink in the containing directory
+          (child, next) => addLink(ipfs, {
+            parent: containingFolder.node,
+            name: fileName,
+            child: {
+              multihash: child.multihash || child.hash,
+              size: child.size
+            },
+            flush: options.flush
+          }, (error, newContaingFolder) => {
+            // Store new containing folder CID
+            containingFolder.node = newContaingFolder
+
+            next(error)
+          }),
+
+          // Update the MFS tree from the containingFolder upwards
+          (next) => updateTree(ipfs, containingFolder, next),
+
+          // Update the MFS record with the new CID for the root of the tree
+          (newRoot, next) => updateMfsRoot(ipfs, newRoot.node.multihash, next)
+        ], done)
+      }
+    ], callback)
+  })
+}