feat: dynamic dag traversal

packages/verified-fetch/package.json

@@ -166,7 +166,9 @@
     "interface-datastore": "^8.3.1",
     "ipfs-unixfs-exporter": "^13.6.1",
     "ipns": "^10.0.0",
+    "it-first": "^3.0.6",
     "it-map": "^3.1.1",
+    "it-peekable": "^3.0.5",
     "it-pipe": "^3.0.1",
     "it-tar": "^6.0.5",
     "it-to-browser-readablestream": "^2.0.9",

packages/verified-fetch/src/utils/enhanced-dag-traversal.ts

@@ -0,0 +1,104 @@
+import { type ComponentLogger } from '@libp2p/interface'
+import { type ReadableStorage, exporter, type ExporterOptions } from 'ipfs-unixfs-exporter'
+import first from 'it-first'
+import peekable from 'it-peekable'
+import toBrowserReadableStream from 'it-to-browser-readablestream'
+import { type CID } from 'multiformats/cid'
+import { type ContentTypeParser } from '../types.js'
+import { getContentType } from './get-content-type.js'
+
+export interface EnhancedDagTraversalOptions extends ExporterOptions {
+  blockstore: ReadableStorage
+  cidOrPath: string | CID
+  logger: ComponentLogger
+  path: string
+  contentTypeParser?: ContentTypeParser
+}
+
+export interface EnhancedDagTraversalResponse {
+  stream: ReadableStream<Uint8Array>
+  firstChunk: Uint8Array
+}
+
+export async function enhancedDagTraversal ({
+  blockstore,
+  signal,
+  onProgress,
+  cidOrPath,
+  offset,
+  length,
+  path,
+  logger,
+  contentTypeParser
+}: EnhancedDagTraversalOptions): Promise<EnhancedDagTraversalResponse> {
+  const log = logger.forComponent('helia:verified-fetch:enhanced-dag-traversal')
+
+  const dfsEntry = await exporter(cidOrPath, blockstore, {
+    signal,
+    onProgress,
+    blockReadConcurrency: 1
+  })
+
+  const dfsIter = dfsEntry.content({
+    signal,
+    onProgress,
+    offset,
+    length
+  })
+
+  let firstChunk: Uint8Array = new Uint8Array()
+  let error: Error
+  try {
+    // Fetch the first chunk eagerly
+    const peekableIter = peekable(dfsIter)
+    const firstPeek = await peekableIter.peek()
+    if (firstPeek.done === true) {
+      throw new Error('No content found')
+    }
+    // firstChunk = await first(dfsIter)
+    firstChunk = firstPeek.value
+    peekableIter.push(firstChunk)
+  } catch (err: any) {
+    if (signal?.aborted === true) {
+      error = err
+      log.trace('Request aborted while fetching first chunk')
+    } else {
+      throw err
+    }
+  }
+
+  return {
+    stream: toBrowserReadableStream({
+      [Symbol.asyncIterator]: async function * (): AsyncGenerator<Uint8Array, void, undefined> {
+        if (error != null) {
+          throw error
+        }
+
+        // Determine content type based on the first chunk
+        const contentType = await getContentType({ bytes: firstChunk, contentTypeParser, path, log })
+
+        const isImageOrVideo = contentType.startsWith('video/') || contentType.startsWith('image/')
+        log.trace('Content type determined: %s', contentType)
+
+        const exporterEntry = isImageOrVideo
+          ? dfsEntry
+        // If not image/video, switch to a BFS iterator
+          : await exporter(cidOrPath, blockstore, {
+            signal,
+            onProgress
+          })
+
+        // continue with the BFS iterator
+        for await (const chunk of exporterEntry.content({
+          signal,
+          onProgress,
+          offset,
+          length
+        })) {
+          yield chunk
+        }
+      }
+    }),
+    firstChunk
+  }
+}

packages/verified-fetch/src/utils/get-content-type.ts

@@ -0,0 +1,36 @@
+import { type Logger } from '@libp2p/interface'
+import { type ContentTypeParser } from '../types.js'
+import { isPromise } from './type-guards.js'
+
+export interface GetContentTypeOptions {
+  bytes: Uint8Array
+  path: string
+  defaultContentType?: string
+  contentTypeParser?: ContentTypeParser
+  log: Logger
+}
+
+export async function getContentType ({ bytes, contentTypeParser, path, log, defaultContentType = 'application/octet-stream' }: GetContentTypeOptions): Promise<string> {
+  let contentType: string | undefined
+
+  if (contentTypeParser != null) {
+    try {
+      let fileName = path.split('/').pop()?.trim()
+      fileName = fileName === '' ? undefined : fileName
+      const parsed = contentTypeParser(bytes, fileName)
+
+      if (isPromise(parsed)) {
+        const result = await parsed
+
+        if (result != null) {
+          contentType = result
+        }
+      } else if (parsed != null) {
+        contentType = parsed
+      }
+    } catch (err) {
+      log.error('error parsing content type', err)
+    }
+  }
+  return contentType ?? defaultContentType
+}

packages/verified-fetch/src/utils/set-content-type.ts

@@ -1,38 +1,18 @@
 import { type Logger } from '@libp2p/interface'
 import { type ContentTypeParser } from '../types.js'
-import { isPromise } from './type-guards.js'
+import { getContentType } from './get-content-type.js'
 
 export interface SetContentTypeOptions {
   bytes: Uint8Array
   path: string
   response: Response
   defaultContentType?: string
-  contentTypeParser: ContentTypeParser | undefined
+  contentTypeParser?: ContentTypeParser
   log: Logger
 }
 
 export async function setContentType ({ bytes, path, response, contentTypeParser, log, defaultContentType = 'application/octet-stream' }: SetContentTypeOptions): Promise<void> {
-  let contentType: string | undefined
-
-  if (contentTypeParser != null) {
-    try {
-      let fileName = path.split('/').pop()?.trim()
-      fileName = fileName === '' ? undefined : fileName
-      const parsed = contentTypeParser(bytes, fileName)
-
-      if (isPromise(parsed)) {
-        const result = await parsed
-
-        if (result != null) {
-          contentType = result
-        }
-      } else if (parsed != null) {
-        contentType = parsed
-      }
-    } catch (err) {
-      log.error('error parsing content type', err)
-    }
-  }
+  const contentType = await getContentType({ bytes, contentTypeParser, path, log, defaultContentType })
   log.trace('setting content type to "%s"', contentType ?? defaultContentType)
   response.headers.set('content-type', contentType ?? defaultContentType)
 }

packages/verified-fetch/src/verified-fetch.ts

@@ -19,11 +19,11 @@ import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import { ByteRangeContext } from './utils/byte-range-context.js'
 import { dagCborToSafeJSON } from './utils/dag-cbor-to-safe-json.js'
+import { enhancedDagTraversal } from './utils/enhanced-dag-traversal.js'
 import { getContentDispositionFilename } from './utils/get-content-disposition-filename.js'
 import { getETag } from './utils/get-e-tag.js'
 import { getPeerIdFromString } from './utils/get-peer-id-from-string.js'
 import { getResolvedAcceptHeader } from './utils/get-resolved-accept-header.js'
-import { getStreamFromAsyncIterable } from './utils/get-stream-from-async-iterable.js'
 import { tarStream } from './utils/get-tar-stream.js'
 import { getRedirectResponse } from './utils/handle-redirects.js'
 import { parseResource } from './utils/parse-resource.js'
@@ -375,22 +375,16 @@ export class VerifiedFetch {
     this.log.trace('calling exporter for %c/%s with offset=%o & length=%o', resolvedCID, path, offset, length)
 
     try {
-      const entry = await exporter(resolvedCID, this.helia.blockstore, {
-        signal: options?.signal,
-        onProgress: options?.onProgress
-      })
-
-      const asyncIter = entry.content({
+      const { firstChunk, stream } = await enhancedDagTraversal({
+        blockstore: this.helia.blockstore,
         signal: options?.signal,
         onProgress: options?.onProgress,
+        cidOrPath: resolvedCID,
         offset,
-        length
-      })
-      this.log('got async iterator for %c/%s', cid, path)
-
-      const { stream, firstChunk } = await getStreamFromAsyncIterable(asyncIter, path ?? '', this.helia.logger, {
-        onProgress: options?.onProgress,
-        signal: options?.signal
+        length,
+        path,
+        logger: this.helia.logger,
+        contentTypeParser: this.contentTypeParser
       })
       byteRangeContext.setBody(stream)
       // if not a valid range request, okRangeRequest will call okResponse
@@ -399,6 +393,7 @@ export class VerifiedFetch {
       })
 
       await setContentType({ bytes: firstChunk, path, response, contentTypeParser: this.contentTypeParser, log: this.log })
+
       setIpfsRoots(response, ipfsRoots)
 
       return response