feat: add a cloudflare-streaming wrapper (#642)

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/createServerBundle.ts

@@ -117,7 +117,6 @@ async function generateBundle(
   //       `node_modules` inside `.next/standalone`, and others inside
   //       `.next/standalone/package/path` (ie. `.next`, `server.js`).
   //       We need to output the handler file inside the package path.
-  const isMonorepo = monorepoRoot !== appPath;
   const packagePath = path.relative(monorepoRoot, appBuildOutputPath);
   fs.mkdirSync(path.join(outputPath, packagePath), { recursive: true });
 
@@ -244,6 +243,7 @@ async function generateBundle(
     options,
   );
 
+  const isMonorepo = monorepoRoot !== appPath;
   if (isMonorepo) {
     addMonorepoEntrypoint(outputPath, packagePath);
   }
@@ -301,7 +301,7 @@ function addMonorepoEntrypoint(outputPath: string, packagePath: string) {
   const packagePosixPath = packagePath.split(path.sep).join(path.posix.sep);
   fs.writeFileSync(
     path.join(outputPath, "index.mjs"),
-    [`export * from "./${packagePosixPath}/index.mjs";`].join(""),
+    `export * from "./${packagePosixPath}/index.mjs";`,
   );
 }
 

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: [],
 };
@@ -99,7 +100,7 @@ export function validateConfig(config: OpenNextConfig) {
   }
   if (config.dangerous?.disableTagCache) {
     logger.warn(
-      `You've disabled tag cache. 
+      `You've disabled tag cache.
        This means that revalidatePath and revalidateTag from next/cache will not work.
        It is safe to disable if you only use page router`,
     );

packages/open-next/src/overrides/converters/aws-apigw-v2.ts

@@ -11,7 +11,8 @@ import { debug } from "../../adapters/logger";
 import { convertToQuery } from "../../core/routing/util";
 import { removeUndefinedFromQuery } from "./utils";
 
-// Not sure which one is reallly needed as this is not documented anywhere but server actions redirect are not working without this, it causes a 500 error from cloudfront itself with a 'x-amzErrortype: InternalFailure' header
+// Not sure which one is really needed as this is not documented anywhere but server actions redirect are not working without this,
+// it causes a 500 error from cloudfront itself with a 'x-amzErrortype: InternalFailure' header
 const CloudFrontBlacklistedHeaders = [
   "connection",
   "expect",

packages/open-next/src/overrides/converters/edge.ts

@@ -16,7 +16,9 @@ const converter: Converter<
   InternalResult | ({ type: "middleware" } & MiddlewareOutputEvent)
 > = {
   convertFrom: async (event: Request) => {
-    const searchParams = new URL(event.url).searchParams;
+    const url = new URL(event.url);
+
+    const searchParams = url.searchParams;
     const query: Record<string, string | string[]> = {};
     for (const [key, value] of searchParams.entries()) {
       if (key in query) {
@@ -29,13 +31,13 @@ const converter: Converter<
         query[key] = value;
       }
     }
-    //Transform body into Buffer
+    // Transform body into Buffer
     const body = await event.arrayBuffer();
     const headers: Record<string, string> = {};
     event.headers.forEach((value, key) => {
       headers[key] = value;
     });
-    const rawPath = new URL(event.url).pathname;
+    const rawPath = url.pathname;
     const method = event.method;
     const shouldHaveBody = method !== "GET" && method !== "HEAD";
     const cookies: Record<string, string> = Object.fromEntries(
@@ -100,9 +102,10 @@ const converter: Converter<
     for (const [key, value] of Object.entries(result.headers)) {
       headers.set(key, Array.isArray(value) ? value.join(",") : value);
     }
+
     return new Response(result.body as ReadableStream, {
       status: result.statusCode,
-      headers: headers,
+      headers,
     });
   },
   name: "edge",

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/overrides/wrappers/cloudflare.ts

@@ -41,12 +41,14 @@ const handler: WrapperHandler<
     // Retrieve geo information from the cloudflare request
     // See https://developers.cloudflare.com/workers/runtime-apis/request
     // Note: This code could be moved to a cloudflare specific converter when one is created.
-    const cfProperties = (request as any).cf as Record<string, string | null>;
+    const cfProperties = (request as any).cf as
+      | Record<string, string | null>
+      | undefined;
     for (const [propName, headerName] of Object.entries(
       cfPropNameToHeaderName,
     )) {
-      const propValue = cfProperties[propName];
-      if (propValue !== null) {
+      const propValue = cfProperties?.[propName];
+      if (propValue != null) {
         internalEvent.headers[headerName] = propValue;
       }
     }

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 =