chore: decompose modules and reuse import logic for expanding files

License: MIT
Signed-off-by: achingbrain <[email protected]>

Author: achingbrain

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

@@ -6,35 +6,34 @@ const values = require('pull-stream/sources/values')
 const collect = require('pull-stream/sinks/collect')
 const importer = require('ipfs-unixfs-engine').importer
 const {
-  limitStreamBytes
+  loadNode
 } = require('../utils')
 
-const importNode = (ipfs, parent, fileName, source, options, callback) => {
+const defaultOptions = {
+  progress: undefined,
+  hash: undefined,
+  cidVersion: undefined,
+  strategy: undefined
+}
+
+const importStream = (ipfs, source, options, callback) => {
+  options = Object.assign({}, defaultOptions, options)
+
   waterfall([
-    (done) => pull(
+    (cb) => pull(
       values([{
-        content: pull(
-          source,
-          limitStreamBytes(options.length)
-        )
+        content: pull(source)
       }]),
       importer(ipfs._ipld, {
         progress: options.progress,
         hashAlg: options.hash,
         cidVersion: options.cidVersion,
         strategy: options.strategy
       }),
-      collect(done)
+      collect(cb)
     ),
-    (results, done) => {
-      const imported = results[0]
-
-      done(null, {
-        size: imported.size,
-        multihash: imported.multihash
-      })
-    }
+    (results, cb) => loadNode(ipfs, results[0], cb)
   ], callback)
 }
 
-module.exports = importNode
+module.exports = importStream

src/core/write/import-node.js

@@ -9,6 +9,7 @@ const {
   traverseTo,
   addLink,
   updateTree,
+  limitStreamBytes,
   FILE_SEPARATOR
 } = require('../utils')
 const values = require('pull-stream/sources/values')
@@ -22,6 +23,7 @@ const isNode = require('detect-node')
 const fileReaderStream = require('filereader-stream')
 const isPullStream = require('is-pull-stream')
 const cat = require('pull-cat')
+const pull = require('pull-stream/pull')
 
 let fs
 
@@ -165,8 +167,13 @@ module.exports = function mfsWrite (ipfs) {
                 ])
               }
 
+              source = pull(
+                source,
+                limitStreamBytes(options.length)
+              )
+
               log('Importing file', fileName)
-              importNode(ipfs, containingFolder, fileName, source, options, next)
+              importNode(ipfs, source, options, next)
             }
           },
 

src/core/write/index.js

@@ -6,34 +6,25 @@ const {
 } = UnixFs
 const pull = require('pull-stream/pull')
 const cat = require('pull-cat')
-const values = require('pull-stream/sources/values')
 const collect = require('pull-stream/sinks/collect')
 const pushable = require('pull-pushable')
 const map = require('pull-stream/throughs/map')
-const asyncMap = require('pull-stream/throughs/async-map')
 const filter = require('pull-stream/throughs/filter')
-const paramap = require('pull-paramap')
-const {
-  leafFirst
-} = require('pull-traverse')
 const waterfall = require('async/waterfall')
 const parallel = require('async/parallel')
-const findFileSize = require('./find-file-size')
 const {
-  DAGNode,
   DAGLink
 } = require('ipld-dag-pb')
 const log = require('debug')('mfs:write:update-node')
-const bs58 = require('bs58')
 const {
   limitStreamBytes,
-  addLink,
   createNode,
   zeros,
   loadNode,
   MAX_CHUNK_SIZE
 } = require('../utils')
-const importer = require('ipfs-unixfs-engine').importer
+const importNode = require('./import-node')
+const updateNodeBytes = require('./update-tree')
 
 const updateNode = (ipfs, cidToUpdate, source, options, callback) => {
   let offset = options.offset || 0
@@ -50,7 +41,8 @@ const updateNode = (ipfs, cidToUpdate, source, options, callback) => {
   waterfall([
     (done) => loadNode(ipfs, cidToUpdate, done),
     (node, done) => {
-      const fileSize = findFileSize(node)
+      const meta = unmarshal(node.data)
+      const fileSize = meta.fileSize()
 
       log(`Updating bytes ${streamStart}-${streamEnd} of ${fileSize} bytes from ${cidToUpdate.toBaseEncodedString()} with source`)
 
@@ -118,10 +110,10 @@ const updateNode = (ipfs, cidToUpdate, source, options, callback) => {
           // wait for both streams to end
           parallel([
             // set up pull stream for replacing bytes
-            (cb) => updateNodeBytes(node, fileSize, updateSource, cb),
+            (cb) => updateNodeBytes(ipfs, node, fileSize, streamStart, streamEnd, updateSource, options, cb),
 
             // setup pull stream for appending bytes
-            (cb) => appendNodeBytes(appendSource, cb)
+            (cb) => importNode(ipfs, appendSource, options, cb)
           ], next)
         },
         ([updatedNode, appendedNode], next) => {
@@ -143,8 +135,8 @@ const updateNode = (ipfs, cidToUpdate, source, options, callback) => {
                 return createNode(ipfs, newMeta.marshal(), [], options, next)
               } else {
                 // We expanded one DAGNode into two so create a tree
-                const link1 = new DAGLink('', updatedMeta.fileSize(), updatedNode.multihash)
-                const link2 = new DAGLink('', appendedMeta.fileSize(), appendedNode.multihash)
+                const link1 = new DAGLink('', updatedNode.data.length, updatedNode.multihash)
+                const link2 = new DAGLink('', appendedNode.data.length, appendedNode.multihash)
 
                 const newMeta = new UnixFs(updatedMeta.type)
                 newMeta.addBlockSize(updatedMeta.fileSize())
@@ -156,219 +148,21 @@ const updateNode = (ipfs, cidToUpdate, source, options, callback) => {
             }
 
             // if we added new bytes, add them to the root node of the original file
-            // this is consistent with the go implementation but probably broken
+            // this is consistent with the go implementation but probably not the right thing to do
 
             // update UnixFs metadata on the root node
             updatedMeta.addBlockSize(appendedMeta.fileSize())
 
-            return waterfall([
-              (cb) => DAGNode.create(updatedMeta.marshal(), updatedNode.links, cb),
-              (newNode, cb) => addLink(ipfs, {
-                parent: newNode,
-                child: appendedNode
-              }, cb)
-            ], next)
+            return createNode(ipfs, updatedMeta.marshal(), updatedNode.links.concat(
+              new DAGLink('', appendedNode.data.length, appendedNode.multihash)
+            ), options, next)
           }
 
           next(null, updatedNode)
         }
       ], done)
     }
   ], callback)
-
-  function appendNodeBytes (source, callback) {
-    waterfall([
-      (cb) => pull(
-        values([{
-          content: pull(source)
-        }]),
-        importer(ipfs._ipld, {
-          progress: options.progress,
-          hashAlg: options.hash,
-          cidVersion: options.cidVersion,
-          strategy: options.strategy
-        }),
-        collect(cb)
-      ),
-      (results, cb) => loadNode(ipfs, results[0], cb)
-    ], callback)
-  }
-
-  function updateNodeBytes (node, fileSize, source, callback) {
-    waterfall([
-      (cb) => pull(
-        source,
-        asyncMap((buffer, done) => {
-          // Find the DAGNodes that contain the data at the specified offset/length
-          // Merge the data and create new DAGNodes with the merged data
-          // Keep a record of the new CIDs and update the tree
-
-          pull(
-            leafFirst({
-              parent: null,
-              link: null,
-              index: null,
-              node,
-              nodeStart: streamPosition,
-              nodeEnd: fileSize
-            }, findDAGNodesWithRequestedData),
-            paramap(updateNodeData(buffer)),
-            filter(Boolean),
-            asyncMap((link, next) => {
-              if (!link.parent || link.index === undefined) {
-                return next(null, link)
-              }
-
-              // Create a new list of links
-              const links = link.parent.node.links.map((existingLink, index) => {
-                if (index === link.index) {
-                  return new DAGLink('', link.size, link.multihash)
-                }
-
-                return existingLink
-              })
-
-              // Update node's parent
-              waterfall([
-                // Create a DAGNode with the new data
-                (cb) => createNode(ipfs, link.parent.node.data, links, options, cb),
-                (newNode, cb) => {
-                  link.parent.node = newNode
-
-                  cb(null, link)
-                }
-              ], next)
-            }),
-            collect((error, results) => {
-              let updatedRoot
-
-              if (!error) {
-                updatedRoot = results[0]
-
-                while (updatedRoot.parent) {
-                  updatedRoot = updatedRoot.parent
-                }
-
-                if (updatedRoot.node) {
-                  updatedRoot = updatedRoot.node
-                }
-              }
-
-              offset += buffer.length
-
-              log(`Updated root is ${bs58.encode(updatedRoot.multihash)}`)
-
-              done(error, updatedRoot)
-            })
-          )
-        }),
-        collect((error, results) => cb(error, results && results[0]))
-      ),
-      (updatedNodeCID, cb) => loadNode(ipfs, updatedNodeCID, cb)
-    ], callback)
-  }
-
-  // Where we currently are in the existing file
-  let streamPosition = 0
-
-  // Returns a pull stream that will load the data from the children of the passed node
-  function findDAGNodesWithRequestedData ({ node }) {
-    const meta = unmarshal(node.data)
-
-    log(`Node links ${node.links.length}, block sizes ${meta.blockSizes}`)
-
-    const parent = {
-      node: node
-    }
-
-    // work out which child nodes contain the requested data
-    const filteredLinks = node.links
-      .map((link, index) => {
-        const child = {
-          parent,
-          link: link,
-          index: index,
-          nodeStart: streamPosition,
-          nodeEnd: streamPosition + meta.blockSizes[index]
-        }
-
-        streamPosition = child.nodeEnd
-
-        return child
-      })
-      .filter((child, index) => {
-        log('child.nodeStart', child.nodeStart, 'child.nodeEnd', child.nodeEnd, 'streamStart', streamStart, 'streamEnd', streamEnd)
-
-        return (streamStart >= child.nodeStart && streamStart < child.nodeEnd) || // child has begin byte
-          (streamEnd > child.nodeStart && streamEnd <= child.nodeEnd) || // child has end byte
-          (streamStart < child.nodeStart && streamEnd > child.nodeEnd) // child is between begin and end bytes
-      })
-
-    if (filteredLinks.length) {
-      // move stream position to the first node we're going to return data from
-      streamPosition = filteredLinks[0].nodeStart
-
-      log(`Updating links with index(es) ${filteredLinks.map(link => link.index).join(',')}`)
-    } else {
-      log(`No links to update`)
-    }
-
-    return pull(
-      values(filteredLinks),
-      paramap((child, cb) => {
-        loadNode(ipfs, child.link, (error, node) => {
-          cb(error, Object.assign({}, child, {
-            node
-          }))
-        })
-      })
-    )
-  }
-
-  function updateNodeData (newContent) {
-    return ({ parent, link, nodeStart, node, index }, done) => {
-      if (!node || !node.data) {
-        return done()
-      }
-
-      const meta = unmarshal(node.data)
-
-      if (!meta || !meta.data || !meta.data.length) {
-        return done()
-      }
-
-      const targetStart = streamStart - nodeStart
-      const sourceStart = 0
-      let sourceEnd = streamEnd - streamStart
-
-      if (meta.data.length < sourceEnd) {
-        // we need to write to another DAGNode so increment the streamStart
-        // by the number of bytes from buffer we've written
-        streamStart += meta.data.length
-      }
-
-      const newData = Buffer.from(meta.data)
-      newContent.copy(newData, targetStart, sourceStart, sourceEnd)
-
-      const nodeData = new UnixFs(meta.type, newData).marshal()
-
-      waterfall([
-        // Create a DAGNode with the new data
-        (cb) => createNode(ipfs, nodeData, [], options, cb),
-        (newNode, cb) => {
-          log(`Created DAGNode with new data with hash ${bs58.encode(newNode.multihash)} to replace ${bs58.encode(node.multihash)}`)
-
-          // Store the CID and friends so we can update it's parent's links
-          cb(null, {
-            parent: parent,
-            index: index,
-            multihash: newNode.multihash,
-            size: newNode.size
-          })
-        }
-      ], done)
-    }
-  }
 }
 
 module.exports = updateNode