feat: add shamefullyPatchR2Buckets

Author: pi0

README.md

@@ -44,6 +44,7 @@ You can configure additional options using `cloudflareDev: { }` in `nitro.config
 - `persistDir`: Sets the persist dir (default `.wrangler/state/v3`).
 - `configPath`: Sets a custom path for `wrangler.toml` file.
 - `silent`: Hide initial banner.
+- `shamefullyPatchR2Buckets`: Add workaround for https://github.com/cloudflare/workers-sdk/issues/5360
 
 ## Development
 

examples/nitro/nitro.config.ts

@@ -3,4 +3,7 @@ import nitroCloudflareBindings from "nitro-cloudflare-dev";
 // https://nitro.unjs.io/config
 export default defineNitroConfig({
   modules: [nitroCloudflareBindings],
+  cloudflareDev: {
+    shamefullyPatchR2Buckets: true,
+  },
 });

package.json

@@ -41,6 +41,7 @@
     "pkg-types": "^1.0.3"
   },
   "devDependencies": {
+    "@cloudflare/workers-types": "^4.20240320.1",
     "@types/node": "^20.11.30",
     "changelogen": "^0.5.5",
     "eslint": "^8.57.0",

pnpm-lock.yaml

@@ -14,7 +14,9 @@ declare module "nitropack" {
     cloudflareDev?: {
       configPath?: string;
       persistDir?: string;
-      silent: boolean;
+      silent?: boolean;
+      /** workaround for https://github.com/cloudflare/workers-sdk/issues/5360 */
+      shamefullyPatchR2Buckets?: boolean;
     };
   }
 }
@@ -68,6 +70,8 @@ async function nitroModule(nitro: Nitro) {
   // Share config to the runtime
   nitro.options.runtimeConfig.wrangler = {
     ...nitro.options.runtimeConfig.wrangler,
+    shamefullyPatchR2Buckets:
+      nitro.options.cloudflareDev?.shamefullyPatchR2Buckets,
     configPath,
     persistDir,
   };

src/index.ts

@@ -64,21 +64,35 @@ async function _getPlatformProxy() {
   })) as typeof import("wrangler");
 
   const runtimeConfig: {
-    wrangler: { configPath: string; persistDir: string };
+    wrangler: {
+      configPath: string;
+      persistDir: string;
+      shamefullyPatchR2Buckets?: boolean;
+    };
   } = useRuntimeConfig();
 
   const proxy = await getPlatformProxy({
     configPath: runtimeConfig.wrangler.configPath,
     persist: { path: runtimeConfig.wrangler.persistDir },
   });
 
+  if (runtimeConfig.wrangler.shamefullyPatchR2Buckets) {
+    const { patchR2Bucket } = await import("./r2-patch");
+    for (const [key, binding] of Object.entries(proxy.env) as [string, any][]) {
+      if (binding.createMultipartUpload) {
+        // console.log("Patching R2Bucket", key);
+        proxy.env[key] = patchR2Bucket(binding);
+      }
+    }
+  }
+
   return proxy;
 }
 
 function _createStubProxy(): PlatformProxy {
   return {
     env: {},
-    cf: {},
+    cf: {} as any,
     ctx: {
       waitUntil() {},
       passThroughOnException() {},

src/runtime/plugin.dev.ts

@@ -0,0 +1,51 @@
+import type { R2Bucket } from "@cloudflare/workers-types";
+/**
+ * Workaround for https://github.com/cloudflare/workers-sdk/issues/5360
+ */
+export function patchR2Bucket(bucket: R2Bucket) {
+  let _mutex: Promise<any> | undefined;
+
+  const _get = bucket.get.bind(bucket);
+
+  async function getAndRead(...args: Parameters<R2Bucket["get"]>) {
+    const obj = await _get(...args);
+    if (!obj) {
+      return obj;
+    }
+    const chunks: any[] = [];
+    for await (const chunk of obj.body) {
+      chunks.push(chunk);
+    }
+    const body = new ReadableStream({
+      start(controller) {
+        for (const chunk of chunks) {
+          controller.enqueue(chunk);
+        }
+        controller.close();
+      },
+    });
+    return { ...obj, body };
+  }
+
+  async function get(...args: Parameters<R2Bucket["get"]>) {
+    if (_mutex) {
+      await _mutex;
+    }
+    try {
+      _mutex = getAndRead(...args);
+      const obj = await _mutex;
+      return obj;
+    } finally {
+      _mutex = undefined;
+    }
+  }
+
+  return new Proxy(bucket, {
+    get(target, prop) {
+      if (prop === "get") {
+        return get;
+      }
+      return Reflect.get(target, prop);
+    },
+  });
+}

src/runtime/r2-patch.ts

@@ -3,3 +3,7 @@ compatibility_date = "2022-07-12"
 kv_namespaces = [
   { binding = "KV", id = "<KV_ID>" }
 ]
+
+r2_buckets = [
+  { binding = "BLOB", bucket_name = "default" },
+]