feat: add walkUnixfsPath to emit nodes for each path segment

index.d.ts

@@ -38,6 +38,10 @@ export interface IDagula {
    * Get UnixFS files and directories.
    */
   getUnixfs (path: CID|string, options?: AbortOptions): Promise<UnixFSEntry>
+  /**
+   * Emit nodes for all path segements and get UnixFS files and directories
+   */
+  walkUnixfsPath (path: CID|string, options?: AbortOptions): Promise<UnixFSEntry>
 }
 
 export declare class Dagula implements IDagula {
@@ -54,6 +58,10 @@ export declare class Dagula implements IDagula {
    * Get UnixFS files and directories.
    */
   getUnixfs (path: CID|string, options?: AbortOptions): Promise<UnixFSEntry>
+  /**
+   * Emit nodes for all path segements and get UnixFS files and directories
+   */
+  walkUnixfsPath (path: CID|string, options?: AbortOptions): Promise<UnixFSEntry>
   /**
    * Create a new Dagula instance from the passed libp2p Network interface.
    */

index.js

@@ -7,7 +7,7 @@ import * as dagCbor from '@ipld/dag-cbor'
 import * as dagJson from '@ipld/dag-json'
 import * as Block from 'multiformats/block'
 import { sha256 as hasher } from 'multiformats/hashes/sha2'
-import { exporter } from 'ipfs-unixfs-exporter'
+import { exporter, walkPath } from 'ipfs-unixfs-exporter'
 import { transform } from 'streaming-iterables'
 import { BitswapFetcher } from './bitswap-fetcher.js'
 
@@ -131,4 +131,24 @@ export class Dagula {
     // @ts-ignore exporter requires Blockstore but only uses `get`
     return exporter(path, blockstore, { signal: options.signal })
   }
+
+  /**
+   * @param {string|import('multiformats').CID} path
+   * @param {{ signal?: AbortSignal }} [options]
+   */
+  async * walkUnixfsPath (path, options = {}) {
+    log('walking unixfs %s', path)
+    const blockstore = {
+      /**
+       * @param {CID} cid
+       * @param {{ signal?: AbortSignal }} [options]
+       */
+      get: async (cid, options) => {
+        const block = await this.getBlock(cid, options)
+        return block.bytes
+      }
+    }
+
+    yield * walkPath(path, blockstore, { signal: options.signal })
+  }
 }

package.json

@@ -47,6 +47,7 @@
   "devDependencies": {
     "ava": "^4.3.1",
     "blockstore-core": "^1.0.5",
+    "ipfs-unixfs": "^11.0.0",
     "miniswap": "^2.0.0",
     "standard": "^17.0.0",
     "uint8arrays": "^3.0.0"

test.js

@@ -6,6 +6,8 @@ import { mplex } from '@libp2p/mplex'
 import { MemoryBlockstore } from 'blockstore-core/memory'
 import { fromString, toString } from 'uint8arrays'
 import * as raw from 'multiformats/codecs/raw'
+import * as dagPB from '@ipld/dag-pb'
+import { UnixFS } from 'ipfs-unixfs'
 import { sha256 } from 'multiformats/hashes/sha2'
 import { CID } from 'multiformats/cid'
 import { Miniswap, BITSWAP_PROTOCOL } from 'miniswap'
@@ -41,6 +43,41 @@ test('should fetch a single CID', async t => {
   }
 })
 
+test('should walk a unixfs path', async t => {
+  // create blockstore and add data
+  const serverBlockstore = new MemoryBlockstore()
+  const data = fromString(`TEST DATA ${Date.now()}`)
+  const hash = await sha256.digest(data)
+  const cid = CID.create(1, raw.code, hash)
+  await serverBlockstore.put(cid, data)
+  const linkName = 'foo'
+  const dirData = new UnixFS({ type: 'directory' }).marshal()
+  const dirBytes = dagPB.encode(dagPB.prepare({ Data: dirData, Links: [{ Name: linkName, Hash: cid }] }))
+  const dirCid = CID.create(1, dagPB.code, await sha256.digest(dirBytes))
+  await serverBlockstore.put(dirCid, dirBytes)
+
+  const server = await createLibp2p({
+    addresses: { listen: ['/ip4/127.0.0.1/tcp/0/ws'] },
+    transports: [webSockets()],
+    streamMuxers: [mplex()],
+    connectionEncryption: [noise()]
+  })
+
+  const miniswap = new Miniswap(serverBlockstore)
+  server.handle(BITSWAP_PROTOCOL, miniswap.handler)
+
+  await server.start()
+
+  const libp2p = await getLibp2p()
+  const dagula = await Dagula.fromNetwork(libp2p, { peer: server.getMultiaddrs()[0] })
+  const entries = []
+  for await (const entry of dagula.walkUnixfsPath(`${dirCid}/${linkName}`)) {
+    entries.push(entry)
+  }
+  t.is(entries.at(0).cid.toString(), dirCid.toString())
+  t.is(entries.at(1).cid.toString(), cid.toString())
+})
+
 test('should abort a fetch', async t => {
   const server = await createLibp2p({
     addresses: { listen: ['/ip4/127.0.0.1/tcp/0/ws'] },