refactor(middleware): use public bucket handler directly

package.json

@@ -42,7 +42,7 @@
     "@ucanto/client": "^9.0.1",
     "@ucanto/principal": "^9.0.1",
     "@ucanto/transport": "^9.1.1",
-    "@web3-storage/blob-fetcher": "^2.4.4-rc.0",
+    "@web3-storage/blob-fetcher": "^2.4.4-rc.1",
     "@web3-storage/capabilities": "^17.4.1",
     "@web3-storage/gateway-lib": "^5.1.2",
     "dagula": "^8.0.0",

src/middleware/withCarBlockHandler.js

@@ -3,11 +3,12 @@
 
 import { CAR_CODE } from '../constants.js'
 import { HttpError } from '@web3-storage/gateway-lib/util'
+// @ts-expect-error no types
+import httpRangeParse from 'http-range-parse'
 import { base58btc } from 'multiformats/bases/base58'
-import { parseRange, toResponse } from './utils.js'
 
 /**
- * @import { R2Bucket, KVNamespace, RateLimit, R2Range } from '@cloudflare/workers-types'
+ * @import { R2Bucket, KVNamespace, RateLimit } from '@cloudflare/workers-types'
  * @import {
  *   IpfsUrlContext,
  *   Middleware,
@@ -18,6 +19,8 @@ import { parseRange, toResponse } from './utils.js'
  * @import { Environment } from './withCarBlockHandler.types.js'
  */
 
+/** @typedef {{ offset: number, length?: number } | { offset?: number, length: number } | { suffix: number }} Range */
+
 /**
  * Middleware that will serve CAR files if a CAR codec is found in the path
  * CID. If the CID is not a CAR CID it delegates to the next middleware.
@@ -89,7 +92,7 @@ export async function handleCarBlock (request, env, ctx) {
     })
   }
 
-  /** @type {R2Range|undefined} */
+  /** @type {Range|undefined} */
   let range
   if (request.headers.has('range')) {
     try {
@@ -105,13 +108,32 @@ export async function handleCarBlock (request, env, ctx) {
   }
   if (!obj) throw new HttpError('CAR not found', { status: 404 })
 
-  return toResponse(obj, range, new Headers({
+  const status = range ? 206 : 200
+  const headers = new Headers({
     'Content-Type': 'application/vnd.ipld.car; version=1;',
     'X-Content-Type-Options': 'nosniff',
     'Cache-Control': 'public, max-age=29030400, immutable',
     'Content-Disposition': `attachment; filename="${dataCid}.car"`,
     Etag: etag
-  }), (body, length) => body.pipeThrough(new FixedLengthStream(length)))
+  })
+
+  let contentLength = obj.size
+  if (range) {
+    let first, last
+    if ('suffix' in range) {
+      first = obj.size - range.suffix
+      last = obj.size - 1
+    } else {
+      first = range.offset || 0
+      last = range.length != null ? first + range.length - 1 : obj.size - 1
+    }
+    headers.set('Content-Range', `bytes ${first}-${last}/${obj.size}`)
+    contentLength = last - first + 1
+  }
+  headers.set('Content-Length', contentLength.toString())
+
+  // @ts-expect-error ReadableStream types incompatible
+  return new Response(obj.body.pipeThrough(new FixedLengthStream(contentLength)), { status, headers })
 }
 
 /** @param {import('multiformats').UnknownLink} cid */
@@ -122,3 +144,18 @@ const toBlobKey = digest => {
   const mhStr = base58btc.encode(digest.bytes)
   return `${mhStr}/${mhStr}.blob`
 }
+
+/**
+ * Convert a HTTP Range header to a range object.
+ * @param {string} value
+ * @returns {Range}
+ */
+function parseRange (value) {
+  const result = httpRangeParse(value)
+  if (result.ranges) throw new Error('Multipart ranges not supported')
+  const { unit, first, last, suffix } = result
+  if (unit !== 'bytes') throw new Error(`Unsupported range unit: ${unit}`)
+  return suffix != null
+    ? { suffix }
+    : { offset: first, length: last != null ? last - first + 1 : undefined }
+}

src/middleware/withCarParkFetch.js

@@ -1,5 +1,34 @@
 import { withSimpleSpan } from '@web3-storage/blob-fetcher/tracing/tracing'
-import { parseRange, toResponse } from './utils.js'
+import { createHandler } from '@web3-storage/public-bucket/server'
+// eslint-disable-next-line
+import * as BucketAPI from '@web3-storage/public-bucket'
+
+/** @implements {BucketAPI.Bucket} */
+export class TraceBucket {
+  #bucket
+
+  /**
+   *
+   * @param {BucketAPI.Bucket} bucket
+   */
+  constructor (bucket) {
+    this.#bucket = bucket
+  }
+
+  /** @param {string} key */
+  head (key) {
+    return withSimpleSpan('bucket.head', this.#bucket.head, this.#bucket)(key)
+  }
+
+  /**
+   * @param {string} key
+   * @param {BucketAPI.GetOptions} [options]
+   */
+  get (key, options) {
+    return withSimpleSpan('bucket.get', this.#bucket.get, this.#bucket)(key, options)
+  }
+}
+
 /**
  * @import {
  *   Middleware,
@@ -11,6 +40,12 @@ import { parseRange, toResponse } from './utils.js'
  * } from './withCarParkFetch.types.js'
  */
 
+/**
+ * 20MiB should allow the worker to process ~4-5 concurrent requests that
+ * require a batch at the maximum size.
+ */
+const MAX_BATCH_SIZE = 20 * 1024 * 1024
+
 /**
  * Adds {@link CarParkFetchContext.fetch} to the context. This version of fetch
  * will pull directly from R2 CARPARK when present
@@ -23,39 +58,20 @@ export function withCarParkFetch (handler) {
     if (!env.CARPARK_PUBLIC_BUCKET_URL) {
       return handler(request, env, { ...ctx, fetch: globalThis.fetch })
     }
+    const bucket = new TraceBucket(/** @type {import('@web3-storage/public-bucket').Bucket} */ (env.CARPARK))
+    const bucketHandler = createHandler({ bucket, maxBatchSize: MAX_BATCH_SIZE })
+
     /**
      *
      * @param {globalThis.RequestInfo | URL} input
      * @param {globalThis.RequestInit} [init]
      * @returns {Promise<globalThis.Response>}
      */
     const fetch = async (input, init) => {
-      const urlString = input instanceof Request ? input.url : input instanceof URL ? input.toString() : input
+      const request = input instanceof Request ? input : new Request(input, init)
       // check whether request is going to CARPARK
-      if (env.CARPARK_PUBLIC_BUCKET_URL && urlString.startsWith(env.CARPARK_PUBLIC_BUCKET_URL)) {
-        // extract carpark key from request
-        let key = urlString.replace(env.CARPARK_PUBLIC_BUCKET_URL, '')
-        key = key[0] === '/' ? key.slice(1) : key
-        // extract headers from request
-        const headers = input instanceof Request ? input.headers : init?.headers || {}
-        // extract range header
-        const rangeHeader = (new Headers(headers)).get('Range')
-
-        // extract range if present from range header
-        /** @type {import('@cloudflare/workers-types').R2Range|undefined} */
-        let range
-        if (rangeHeader) {
-          try {
-            range = parseRange(rangeHeader)
-          } catch (err) {
-            return globalThis.fetch(input, init)
-          }
-        }
-        // fetch directly from carpark
-        const resp = await withSimpleSpan('carPark.get', env.CARPARK.get, env.CARPARK)(key, { range })
-
-        // return a fetch response object from the CARPARK response
-        return resp == null ? new Response(null, { status: 404 }) : toResponse(resp, range)
+      if (env.CARPARK_PUBLIC_BUCKET_URL && request.url.startsWith(env.CARPARK_PUBLIC_BUCKET_URL)) {
+        return bucketHandler(request)
       }
       return globalThis.fetch(input, init)
     }

tsconfig.json

@@ -15,7 +15,6 @@
     "skipLibCheck": true,
     "resolveJsonModule": true,
     "lib": ["ES2022", "DOM", "DOM.Iterable"],
-    "types": ["@cloudflare/workers-types"],
     "target": "ES2022",
     "sourceMap": true
   }