feat: add a cloudflare-streaming wrapper

Author: vicb

.changeset/ten-trainers-boil.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/aws": minor
+---
+
+feat: add a cloudflare-streaming wrapper

packages/open-next/src/build/validateConfig.ts

@@ -17,6 +17,7 @@ const compatibilityMatrix: Record<IncludedWrapper, IncludedConverter[]> = {
   ],
   "aws-lambda-streaming": ["aws-apigw-v2"],
   cloudflare: ["edge"],
+  "cloudflare-streaming": ["edge"],
   node: ["node"],
   dummy: [],
 };

packages/open-next/src/overrides/wrappers/cloudflare-streaming.ts

@@ -0,0 +1,77 @@
+import type { InternalEvent, InternalResult } from "types/open-next";
+import type { WrapperHandler } from "types/overrides";
+
+import { Writable } from "node:stream";
+import type { StreamCreator } from "http/index";
+import type { MiddlewareOutputEvent } from "../../core/routingHandler";
+
+const handler: WrapperHandler<
+  InternalEvent,
+  InternalResult | ({ type: "middleware" } & MiddlewareOutputEvent)
+> =
+  async (handler, converter) =>
+  async (
+    request: Request,
+    env: Record<string, string>,
+    ctx: any,
+  ): Promise<Response> => {
+    globalThis.process = process;
+    globalThis.openNextWaitUntil = ctx.waitUntil.bind(ctx);
+
+    // Set the environment variables
+    // Cloudflare suggests to not override the process.env object but instead apply the values to it
+    for (const [key, value] of Object.entries(env)) {
+      if (typeof value === "string") {
+        process.env[key] = value;
+      }
+    }
+
+    const internalEvent = await converter.convertFrom(request);
+
+    // TODO:
+    // The edge converter populate event.url with the url including the origin.
+    // This is required for middleware to keep track of the protocol (i.e. http with wrangler dev).
+    // However the server expects that the origin is not included.
+    const url = new URL(internalEvent.url);
+    (internalEvent.url as string) = url.href.slice(url.origin.length);
+
+    const { promise: promiseResponse, resolve: resolveResponse } =
+      Promise.withResolvers<Response>();
+
+    const streamCreator: StreamCreator = {
+      writeHeaders(prelude: {
+        statusCode: number;
+        cookies: string[];
+        headers: Record<string, string>;
+      }): Writable {
+        const { statusCode, cookies, headers } = prelude;
+
+        const responseHeaders = new Headers(headers);
+        for (const cookie of cookies) {
+          responseHeaders.append("Set-Cookie", cookie);
+        }
+
+        const { readable, writable } = new TransformStream();
+        const response = new Response(readable, {
+          status: statusCode,
+          headers: responseHeaders,
+        });
+        resolveResponse(response);
+
+        return Writable.fromWeb(writable);
+      },
+      onWrite: () => {},
+      onFinish: (_length: number) => {},
+    };
+
+    ctx.waitUntil(handler(internalEvent, streamCreator));
+
+    return promiseResponse;
+  };
+
+export default {
+  wrapper: handler,
+  name: "cloudflare-streaming",
+  supportStreaming: true,
+  edgeRuntime: true,
+};

packages/open-next/src/types/open-next.ts

@@ -81,6 +81,7 @@ export type IncludedWrapper =
   | "aws-lambda-streaming"
   | "node"
   | "cloudflare"
+  | "cloudflare-streaming"
   | "dummy";
 
 export type IncludedConverter =