From 3eff2fc045a68480cd921b48d5f80efe6c31c37d Mon Sep 17 00:00:00 2001 From: Nicolas Dorseuil Date: Tue, 14 Jul 2026 16:43:38 +0200 Subject: [PATCH 1/2] feat: Refactor AWS Lambda streaming support and introduce new converters - Updated aws-lambda-streaming wrapper to utilize new converter structure. - Introduced aws-streaming converter for handling streaming responses in AWS Lambda. - Enhanced existing converters to support direct and streaming output types. - Modified core types to accommodate new converter output structure. - Updated validation logic to ensure compatibility with new streaming converter. - Added tests for aws-streaming converter and SQS revalidate functionality. - Refactored express-dev and cloudflare wrappers to align with new converter outputs. - Improved response handling in node and cloudflare-node wrappers. --- .../src/overrides/converters/aws-apigw-v1.ts | 33 ++-- .../src/overrides/converters/aws-apigw-v2.ts | 41 +++-- .../overrides/converters/aws-cloudfront.ts | 111 ++++++----- .../src/overrides/converters/aws-streaming.ts | 42 +++++ .../overrides/converters/response-stream.ts | 46 +++++ .../overrides/converters/sqs-revalidate.ts | 15 +- .../wrappers/aws-lambda-compressed.ts | 121 +++++------- .../wrappers/aws-lambda-streaming.ts | 37 ++-- .../aws/src/overrides/wrappers/aws-lambda.ts | 31 +--- packages/core/src/build/generateOutput.ts | 5 +- packages/core/src/build/validateConfig.ts | 8 +- .../core/src/overrides/converters/dummy.ts | 8 +- .../core/src/overrides/converters/edge.ts | 172 ++++++++++++------ .../core/src/overrides/converters/node.ts | 32 +++- .../src/overrides/wrappers/cloudflare-edge.ts | 12 +- .../src/overrides/wrappers/cloudflare-node.ts | 101 +--------- packages/core/src/overrides/wrappers/dummy.ts | 8 +- .../src/overrides/wrappers/express-dev.ts | 40 ++-- packages/core/src/overrides/wrappers/node.ts | 27 +-- packages/core/src/plugins/resolve.ts | 6 +- packages/core/src/types/open-next.ts | 1 + packages/core/src/types/overrides.ts | 31 +++- .../tests/converters/aws-apigw-v1.test.ts | 28 ++- .../tests/converters/aws-apigw-v2.test.ts | 28 ++- .../tests/converters/aws-cloudfront.test.ts | 32 +++- .../tests/converters/aws-streaming.test.ts | 54 ++++++ .../tests/converters/sqs-revalidate.test.ts | 25 +++ 27 files changed, 662 insertions(+), 433 deletions(-) create mode 100644 packages/aws/src/overrides/converters/aws-streaming.ts create mode 100644 packages/aws/src/overrides/converters/response-stream.ts create mode 100644 packages/tests-unit/tests/converters/aws-streaming.test.ts create mode 100644 packages/tests-unit/tests/converters/sqs-revalidate.test.ts diff --git a/packages/aws/src/overrides/converters/aws-apigw-v1.ts b/packages/aws/src/overrides/converters/aws-apigw-v1.ts index 7f769aa0..b3ad1b6d 100644 --- a/packages/aws/src/overrides/converters/aws-apigw-v1.ts +++ b/packages/aws/src/overrides/converters/aws-apigw-v1.ts @@ -3,11 +3,12 @@ import { extractHostFromHeaders, removeUndefinedFromQuery, } from "@opennextjs/core/overrides/converters/utils.js"; -import type { InternalEvent, InternalResult } from "@opennextjs/core/types/open-next.js"; +import type { InternalEvent } from "@opennextjs/core/types/open-next.js"; import type { Converter } from "@opennextjs/core/types/overrides.js"; -import { fromReadableStream } from "@opennextjs/core/utils/stream.js"; import type { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda"; +import { createBufferedStreamCreator } from "./response-stream.js"; + function normalizeAPIGatewayProxyEventHeaders(event: APIGatewayProxyEvent): Record { event.multiValueHeaders; const headers: Record = {}; @@ -89,10 +90,14 @@ async function convertFromAPIGatewayProxyEvent(event: APIGatewayProxyEvent): Pro }; } -async function convertToApiGatewayProxyResult(result: InternalResult): Promise { +function convertToApiGatewayProxyResult( + prelude: { statusCode: number; cookies: string[]; headers: Record }, + body: Buffer, + isBase64Encoded: boolean +): APIGatewayProxyResult { const headers: Record = {}; const multiValueHeaders: Record = {}; - Object.entries(result.headers).forEach(([key, value]) => { + Object.entries(prelude.headers).forEach(([key, value]) => { if (Array.isArray(value)) { multiValueHeaders[key] = value; } else { @@ -103,14 +108,15 @@ async function convertToApiGatewayProxyResult(result: InternalResult): Promise 0) { + multiValueHeaders["set-cookie"] = prelude.cookies; + } const response: APIGatewayProxyResult = { - statusCode: result.statusCode, + statusCode: prelude.statusCode, headers, - body, - isBase64Encoded: result.isBase64Encoded, + body: body.toString(isBase64Encoded ? "base64" : "utf8"), + isBase64Encoded, multiValueHeaders, }; debug(response); @@ -118,7 +124,10 @@ async function convertToApiGatewayProxyResult(result: InternalResult): Promise convertFromAPIGatewayProxyEvent(event as APIGatewayProxyEvent), + convertTo: async () => { + const { streamCreator, output } = createBufferedStreamCreator(convertToApiGatewayProxyResult); + return { type: "stream" as const, streamCreator, output }; + }, name: "aws-apigw-v1", -} as Converter; +} satisfies Converter; diff --git a/packages/aws/src/overrides/converters/aws-apigw-v2.ts b/packages/aws/src/overrides/converters/aws-apigw-v2.ts index 036498a7..2c24f488 100644 --- a/packages/aws/src/overrides/converters/aws-apigw-v2.ts +++ b/packages/aws/src/overrides/converters/aws-apigw-v2.ts @@ -5,11 +5,12 @@ import { extractHostFromHeaders, removeUndefinedFromQuery, } from "@opennextjs/core/overrides/converters/utils.js"; -import type { InternalEvent, InternalResult } from "@opennextjs/core/types/open-next.js"; +import type { InternalEvent } from "@opennextjs/core/types/open-next.js"; import type { Converter } from "@opennextjs/core/types/overrides.js"; -import { fromReadableStream } from "@opennextjs/core/utils/stream.js"; import type { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda"; +import { createBufferedStreamCreator } from "./response-stream.js"; + // 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 = [ @@ -68,7 +69,9 @@ function normalizeAPIGatewayProxyEventV2Headers(event: APIGatewayProxyEventV2): return headers; } -async function convertFromAPIGatewayProxyEventV2(event: APIGatewayProxyEventV2): Promise { +export async function convertFromAPIGatewayProxyEventV2( + event: APIGatewayProxyEventV2 +): Promise { const { rawPath, rawQueryString, requestContext } = event; const headers = normalizeAPIGatewayProxyEventV2Headers(event); return { @@ -92,9 +95,13 @@ async function convertFromAPIGatewayProxyEventV2(event: APIGatewayProxyEventV2): }; } -async function convertToApiGatewayProxyResultV2(result: InternalResult): Promise { +function convertToApiGatewayProxyResultV2( + prelude: { statusCode: number; cookies: string[]; headers: Record }, + body: Buffer, + isBase64Encoded: boolean +): APIGatewayProxyResultV2 { const headers: Record = {}; - Object.entries(result.headers) + Object.entries(prelude.headers) .map(([key, value]) => [key.toLowerCase(), value] as const) .filter( ([key]) => @@ -110,21 +117,27 @@ async function convertToApiGatewayProxyResultV2(result: InternalResult): Promise headers[key] = Array.isArray(value) ? value.join(", ") : `${value}`; }); - const body = await fromReadableStream(result.body, result.isBase64Encoded); - const response: APIGatewayProxyResultV2 = { - statusCode: result.statusCode, + statusCode: prelude.statusCode, headers, - cookies: parseSetCookieHeader(result.headers["set-cookie"]), - body, - isBase64Encoded: result.isBase64Encoded, + cookies: + prelude.cookies.length > 0 + ? prelude.cookies + : prelude.headers["set-cookie"] + ? parseSetCookieHeader(prelude.headers["set-cookie"]) + : undefined, + body: body.toString(isBase64Encoded ? "base64" : "utf8"), + isBase64Encoded, }; debug(response); return response; } export default { - convertFrom: convertFromAPIGatewayProxyEventV2, - convertTo: convertToApiGatewayProxyResultV2, + convertFrom: (event) => convertFromAPIGatewayProxyEventV2(event as APIGatewayProxyEventV2), + convertTo: async () => { + const { streamCreator, output } = createBufferedStreamCreator(convertToApiGatewayProxyResultV2); + return { type: "stream" as const, streamCreator, output }; + }, name: "aws-apigw-v2", -} as Converter; +} satisfies Converter; diff --git a/packages/aws/src/overrides/converters/aws-cloudfront.ts b/packages/aws/src/overrides/converters/aws-cloudfront.ts index 10f30941..b9676e8b 100644 --- a/packages/aws/src/overrides/converters/aws-cloudfront.ts +++ b/packages/aws/src/overrides/converters/aws-cloudfront.ts @@ -6,7 +6,6 @@ import { parseSetCookieHeader } from "@opennextjs/core/http/util.js"; import { extractHostFromHeaders } from "@opennextjs/core/overrides/converters/utils.js"; import type { InternalEvent, InternalResult, MiddlewareResult } from "@opennextjs/core/types/open-next.js"; import type { Converter } from "@opennextjs/core/types/overrides.js"; -import { fromReadableStream } from "@opennextjs/core/utils/stream.js"; import type { CloudFrontCustomOrigin, CloudFrontHeaders, @@ -15,6 +14,8 @@ import type { CloudFrontRequestResult, } from "aws-lambda"; +import { createBufferedStreamCreator } from "./response-stream.js"; + const cloudfrontBlacklistedHeaders = [ // Disallowed headers, see: https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/edge-function-restrictions-all.html#function-restrictions-disallowed-headers "connection", @@ -129,59 +130,64 @@ function convertToCloudfrontHeaders(headers: Record, return cloudfrontHeaders; } -async function convertToCloudFrontRequestResult( - result: InternalResult | MiddlewareResult, +async function convertMiddlewareResult( + result: MiddlewareResult, originalRequest: CloudFrontRequestEvent ): Promise { - if (result.type === "middleware") { - const { method, clientIp, origin } = originalRequest.Records[0].cf.request; - const responseHeaders = result.internalEvent.headers; - - // Handle external rewrite - - let customOrigin = origin?.custom as CloudFrontCustomOrigin; - let host = responseHeaders.host ?? responseHeaders.Host; - if (result.origin) { - customOrigin = { - ...customOrigin, - domainName: result.origin.host, - port: result.origin.port ?? 443, - protocol: result.origin.protocol ?? "https", - customHeaders: {}, - }; - host = result.origin.host; - } - - const response: CloudFrontRequest = { - clientIp, - method, - uri: result.internalEvent.rawPath, - querystring: convertToQueryString(result.internalEvent.query).replace("?", ""), - headers: convertToCloudfrontHeaders({ - ...responseHeaders, - host, - }), - origin: origin?.custom - ? { - custom: customOrigin, - } - : origin, + const { method, clientIp, origin } = originalRequest.Records[0].cf.request; + const responseHeaders = result.internalEvent.headers; + + // Handle external rewrite + + let customOrigin = origin?.custom as CloudFrontCustomOrigin; + let host = responseHeaders.host ?? responseHeaders.Host; + if (result.origin) { + customOrigin = { + ...customOrigin, + domainName: result.origin.host, + port: result.origin.port ?? 443, + protocol: result.origin.protocol ?? "https", + customHeaders: {}, }; + host = result.origin.host; + } - debug("response rewrite", response); + const response: CloudFrontRequest = { + clientIp, + method, + uri: result.internalEvent.rawPath, + querystring: convertToQueryString(result.internalEvent.query).replace("?", ""), + headers: convertToCloudfrontHeaders({ + ...responseHeaders, + host, + }), + origin: origin?.custom + ? { + custom: customOrigin, + } + : origin, + }; - return response; - } + debug("response rewrite", response); - const body = await fromReadableStream(result.body, result.isBase64Encoded); - const responseHeaders = result.headers; + return response; +} +function convertToCloudFrontRequestResult( + prelude: { statusCode: number; cookies: string[]; headers: Record }, + body: Buffer, + isBase64Encoded: boolean +): CloudFrontRequestResult { + const responseHeaders = { + ...prelude.headers, + ...(prelude.cookies.length > 0 ? { "set-cookie": prelude.cookies } : {}), + }; const response: CloudFrontRequestResult = { - status: result.statusCode.toString(), + status: prelude.statusCode.toString(), statusDescription: "OK", headers: convertToCloudfrontHeaders(responseHeaders, true), - bodyEncoding: result.isBase64Encoded ? "base64" : "text", - body, + bodyEncoding: isBase64Encoded ? "base64" : "text", + body: body.toString(isBase64Encoded ? "base64" : "utf8"), }; debug(response); @@ -189,7 +195,18 @@ async function convertToCloudFrontRequestResult( } export default { - convertFrom: convertFromCloudFrontRequestEvent, - convertTo: convertToCloudFrontRequestResult, + convertFrom: (event) => convertFromCloudFrontRequestEvent(event as CloudFrontRequestEvent), + convertTo: async (event) => { + const { streamCreator, output } = createBufferedStreamCreator(convertToCloudFrontRequestResult); + return { + type: "stream" as const, + streamCreator, + output, + data: async (result) => + result.type === "middleware" + ? convertMiddlewareResult(result, event as CloudFrontRequestEvent) + : undefined, + }; + }, name: "aws-cloudfront", -} as Converter; +} satisfies Converter; diff --git a/packages/aws/src/overrides/converters/aws-streaming.ts b/packages/aws/src/overrides/converters/aws-streaming.ts new file mode 100644 index 00000000..b962e755 --- /dev/null +++ b/packages/aws/src/overrides/converters/aws-streaming.ts @@ -0,0 +1,42 @@ +import type { Writable } from "node:stream"; + +import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; +import type { Converter } from "@opennextjs/core/types/overrides.js"; +import type { APIGatewayProxyEventV2 } from "aws-lambda"; + +import { convertFromAPIGatewayProxyEventV2 } from "./aws-apigw-v2.js"; + +type StreamingContext = { + responseStream: Writable & { setContentType(contentType: string): void }; + writable: Writable; + contentEncoding: string; +}; + +const converter: Converter = { + convertFrom: (event) => convertFromAPIGatewayProxyEventV2(event as APIGatewayProxyEventV2), + convertTo: async (_event, context) => { + const { responseStream, writable, contentEncoding } = context as StreamingContext; + const streamCreator: StreamCreator = { + writeHeaders(prelude) { + responseStream.setContentType("application/vnd.awslambda.http-integration-response"); + responseStream.write( + JSON.stringify({ + ...prelude, + headers: { + ...prelude.headers, + "content-encoding": contentEncoding, + }, + }) + ); + responseStream.write(new Uint8Array(8)); + return writable; + }, + retainChunks: false, + }; + + return { type: "stream" as const, streamCreator }; + }, + name: "aws-streaming", +}; + +export default converter; diff --git a/packages/aws/src/overrides/converters/response-stream.ts b/packages/aws/src/overrides/converters/response-stream.ts new file mode 100644 index 00000000..393b11e6 --- /dev/null +++ b/packages/aws/src/overrides/converters/response-stream.ts @@ -0,0 +1,46 @@ +import { Writable } from "node:stream"; + +import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; +import { isBinaryContentType } from "@opennextjs/core/utils/binary.js"; + +type Prelude = Parameters[0]; + +export function createBufferedStreamCreator( + createOutput: (prelude: Prelude, body: Buffer, isBase64Encoded: boolean) => T +): { streamCreator: StreamCreator; output: Promise } { + const { promise: output, resolve, reject } = Promise.withResolvers(); + let prelude: Prelude | undefined; + const chunks: Buffer[] = []; + + const streamCreator: StreamCreator = { + writeHeaders(value) { + prelude = value; + return new Writable({ + write(chunk, _encoding, callback) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + callback(); + }, + final(callback) { + if (!prelude) { + const error = new Error("Response stream finished before headers were written"); + reject(error); + callback(error); + return; + } + try { + const isBase64Encoded = + isBinaryContentType(prelude.headers["content-type"]) || !!prelude.headers["content-encoding"]; + resolve(createOutput(prelude, Buffer.concat(chunks), isBase64Encoded)); + callback(); + } catch (error: unknown) { + reject(error); + callback(error instanceof Error ? error : new Error(String(error))); + } + }, + }); + }, + retainChunks: false, + }; + + return { streamCreator, output }; +} diff --git a/packages/aws/src/overrides/converters/sqs-revalidate.ts b/packages/aws/src/overrides/converters/sqs-revalidate.ts index 02e0117e..e5936a9c 100644 --- a/packages/aws/src/overrides/converters/sqs-revalidate.ts +++ b/packages/aws/src/overrides/converters/sqs-revalidate.ts @@ -14,12 +14,17 @@ const converter: Converter = { records, }); }, - convertTo(revalidateEvent) { + convertTo() { return Promise.resolve({ - type: "revalidate", - batchItemFailures: revalidateEvent.records.map((record) => ({ - itemIdentifier: record.id, - })), + type: "direct" as const, + data(revalidateEvent) { + return Promise.resolve({ + type: "revalidate", + batchItemFailures: revalidateEvent.records.map((record) => ({ + itemIdentifier: record.id, + })), + }); + }, }); }, name: "sqs-revalidate", diff --git a/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts b/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts index 743449c4..43bb81be 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts @@ -1,9 +1,7 @@ -import { Readable, type Transform, Writable } from "node:stream"; -import type { ReadableStream } from "node:stream/web"; +import type { Transform } from "node:stream"; import zlib from "node:zlib"; -import { error } from "@opennextjs/core/adapters/logger.js"; -import type { InternalResult, StreamCreator } from "@opennextjs/core/types/open-next.js"; +import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; import type { WrapperHandler } from "@opennextjs/core/types/overrides.js"; import type { AwsLambdaEvent, AwsLambdaReturn } from "../../types/aws-lambda.js"; @@ -20,35 +18,13 @@ const handler: WrapperHandler = } const internalEvent = await converter.convertFrom(lambdaEvent); - // This is a workaround - // https://github.com/opennextjs/opennextjs-aws/blob/e9b37fd44eb856eb8ae73168bf455ff85dd8b285/packages/open-next/src/overrides/wrappers/aws-lambda.ts#L49-L53 - const fakeStream: StreamCreator = { - writeHeaders: () => { - return new Writable({ - write: (_chunk, _encoding, callback) => { - callback(); - }, - }); - }, - }; - - const handlerResponse = await handler(internalEvent, { - streamCreator: fakeStream, - }); - - // Check if response is already compressed - // The handlers response headers are lowercase - const alreadyEncoded = handlerResponse.headers["content-encoding"] ?? ""; - - // Return early here if the response is already compressed - if (alreadyEncoded) { - return converter.convertTo(handlerResponse, lambdaEvent); + const output = await converter.convertTo(lambdaEvent); + if (output.type === "direct") { + return output.data(await handler(internalEvent)); } - // We compress the body if the client accepts it const acceptEncoding = internalEvent.headers["accept-encoding"] ?? internalEvent.headers["Accept-Encoding"] ?? ""; - let contentEncoding: string | null = null; if (acceptEncoding?.includes("br")) { contentEncoding = "br"; @@ -58,17 +34,11 @@ const handler: WrapperHandler = contentEncoding = "deflate"; } - const response: InternalResult = { - ...handlerResponse, - body: compressBody(handlerResponse.body, contentEncoding), - headers: { - ...handlerResponse.headers, - ...(contentEncoding ? { "content-encoding": contentEncoding } : {}), - }, - isBase64Encoded: !!contentEncoding || handlerResponse.isBase64Encoded, - }; - - return converter.convertTo(response, lambdaEvent); + const response = await handler(internalEvent, { + streamCreator: withCompression(output.streamCreator, contentEncoding), + }); + const directResult = await output.data?.(response); + return directResult ?? output.output; }; export default { @@ -77,38 +47,43 @@ export default { supportStreaming: false, }; -function compressBody(body: ReadableStream, encoding: string | null) { - // If no encoding is specified, return original body - if (!encoding) return body; - try { - const readable = Readable.fromWeb(body); - let transform: Transform; +function withCompression(streamCreator: StreamCreator, encoding: string | null): StreamCreator { + if (!encoding) return streamCreator; + return { + ...streamCreator, + writeHeaders(prelude) { + if (prelude.headers["content-encoding"]) { + return streamCreator.writeHeaders(prelude); + } + const target = streamCreator.writeHeaders({ + ...prelude, + headers: { ...prelude.headers, "content-encoding": encoding }, + }); + let transform: Transform; - switch (encoding) { - case "br": - const quality = Number(process.env.BROTLI_QUALITY); - transform = zlib.createBrotliCompress({ - params: { - // This is a compromise between speed and compression ratio. - // The default one will most likely timeout an AWS Lambda with default configuration on large bodies (>6mb). - // Therefore we set it to 6, which is a good compromise. - [zlib.constants.BROTLI_PARAM_QUALITY]: Number.isNaN(quality) ? 6 : quality, - }, - }); - break; - case "gzip": - transform = zlib.createGzip(); - break; - case "deflate": - transform = zlib.createDeflate(); - break; - default: - return body; - } - return Readable.toWeb(readable.pipe(transform)); - } catch (e) { - error("Error compressing body:", e); - // Fall back to no compression on error - return body; - } + switch (encoding) { + case "br": + const quality = Number(process.env.BROTLI_QUALITY); + transform = zlib.createBrotliCompress({ + params: { + // This is a compromise between speed and compression ratio. + // The default one will most likely timeout an AWS Lambda with default configuration on large bodies (>6mb). + // Therefore we set it to 6, which is a good compromise. + [zlib.constants.BROTLI_PARAM_QUALITY]: Number.isNaN(quality) ? 6 : quality, + }, + }); + break; + case "gzip": + transform = zlib.createGzip(); + break; + case "deflate": + transform = zlib.createDeflate(); + break; + default: + return target; + } + transform.pipe(target); + return transform; + }, + }; } diff --git a/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts b/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts index e0ba8a2f..ad75069e 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts @@ -1,9 +1,8 @@ import { Readable, type Writable } from "node:stream"; import zlib from "node:zlib"; -import { debug, error } from "@opennextjs/core/adapters/logger.js"; +import { error } from "@opennextjs/core/adapters/logger.js"; import type { WarmerEvent, WarmerResponse } from "@opennextjs/core/adapters/warmer-function.js"; -import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; import type { Wrapper, WrapperHandler } from "@opennextjs/core/types/overrides.js"; import type { APIGatewayProxyEventV2 } from "aws-lambda"; @@ -70,35 +69,25 @@ const handler: WrapperHandler = async (handler, converter) => compressedStream = responseStream; } - const streamCreator: StreamCreator = { - writeHeaders: (_prelude) => { - responseStream.setContentType("application/vnd.awslambda.http-integration-response"); - _prelude.headers["content-encoding"] = contentEncoding; - - const prelude = JSON.stringify(_prelude); - - responseStream.write(prelude); - - responseStream.write(new Uint8Array(8)); - - return compressedStream ?? responseStream; - }, - }; - - const response = await handler(internalEvent, { streamCreator }); + const output = await converter.convertTo(event, { + responseStream, + writable: compressedStream ?? responseStream, + contentEncoding, + }); + if (output.type === "direct") { + await output.data(await handler(internalEvent)); + return; + } - const isUsingEdge = globalThis.isEdgeRuntime ?? false; - if (isUsingEdge) { - debug("Headers has not been set, we must be in the edge runtime"); - const stream = streamCreator.writeHeaders({ + const response = await handler(internalEvent, { streamCreator: output.streamCreator }); + if (globalThis.isEdgeRuntime ?? false) { + const stream = output.streamCreator.writeHeaders({ statusCode: response.statusCode, headers: response.headers as Record, cookies: [], }); Readable.fromWeb(response.body).pipe(stream); } - - // return converter.convertTo(response); } ) as (...args: unknown[]) => unknown; diff --git a/packages/aws/src/overrides/wrappers/aws-lambda.ts b/packages/aws/src/overrides/wrappers/aws-lambda.ts index ec6323d3..8319f440 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda.ts @@ -1,7 +1,4 @@ -import { Writable } from "node:stream"; - import type { WarmerEvent, WarmerResponse } from "@opennextjs/core/adapters/warmer-function.js"; -import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; import type { WrapperHandler } from "@opennextjs/core/types/overrides.js"; import type { AwsLambdaEvent, AwsLambdaReturn } from "../../types/aws-lambda.js"; @@ -24,27 +21,13 @@ const handler: WrapperHandler = } const internalEvent = await converter.convertFrom(lambdaEvent); - - //TODO: create a simple reproduction and open an issue in the node repo - //This is a workaround, there is an issue in node that causes node to crash silently if the OpenNextNodeResponse stream is not consumed - //This does not happen everytime, it's probably caused by suspended component in ssr (either via or loading.tsx) - //Everyone that wish to create their own wrapper without a StreamCreator should implement this workaround - //This is not necessary if the underlying handler does not use OpenNextNodeResponse (At the moment, OpenNextNodeResponse is used by the node runtime servers and the image server) - const fakeStream: StreamCreator = { - writeHeaders: () => { - return new Writable({ - write: (_chunk, _encoding, callback) => { - callback(); - }, - }); - }, - }; - - const response = await handler(internalEvent, { - streamCreator: fakeStream, - }); - - return converter.convertTo(response, lambdaEvent); + const output = await converter.convertTo(lambdaEvent); + if (output.type === "direct") { + return output.data(await handler(internalEvent)); + } + const response = await handler(internalEvent, { streamCreator: output.streamCreator }); + const directResult = await output.data?.(response); + return directResult ?? output.output; }; export default { diff --git a/packages/core/src/build/generateOutput.ts b/packages/core/src/build/generateOutput.ts index 79716cdc..89039d44 100644 --- a/packages/core/src/build/generateOutput.ts +++ b/packages/core/src/build/generateOutput.ts @@ -124,7 +124,10 @@ async function extractOverrideFn(override?: DefaultOverrideOptions) { }; } const wrapper = await extractOverrideName("aws-lambda", override.wrapper); - const converter = await extractOverrideName("aws-apigw-v2", override.converter); + const converter = await extractOverrideName( + wrapper === "aws-lambda-streaming" ? "aws-streaming" : "aws-apigw-v2", + override.converter + ); return { wrapper, converter }; } diff --git a/packages/core/src/build/validateConfig.ts b/packages/core/src/build/validateConfig.ts index 22779d08..266c0332 100644 --- a/packages/core/src/build/validateConfig.ts +++ b/packages/core/src/build/validateConfig.ts @@ -11,7 +11,7 @@ import logger from "../logger.js"; const compatibilityMatrix: Record = { "aws-lambda": ["aws-apigw-v1", "aws-apigw-v2", "aws-cloudfront", "sqs-revalidate"], "aws-lambda-compressed": ["aws-apigw-v2"], - "aws-lambda-streaming": ["aws-apigw-v2"], + "aws-lambda-streaming": ["aws-streaming"], cloudflare: ["edge"], "cloudflare-edge": ["edge"], "cloudflare-node": ["edge"], @@ -23,7 +23,11 @@ const compatibilityMatrix: Record = { function validateFunctionOptions(fnOptions: FunctionOptions) { const wrapper = typeof fnOptions.override?.wrapper === "string" ? fnOptions.override.wrapper : "aws-lambda"; const converter = - typeof fnOptions.override?.converter === "string" ? fnOptions.override.converter : "aws-apigw-v2"; + typeof fnOptions.override?.converter === "string" + ? fnOptions.override.converter + : wrapper === "aws-lambda-streaming" + ? "aws-streaming" + : "aws-apigw-v2"; if (fnOptions.override?.generateDockerfile && converter !== "node" && wrapper !== "node") { logger.warn( "You've specified generateDockerfile without node converter and wrapper. Without custom converter and wrapper the dockerfile will not work" diff --git a/packages/core/src/overrides/converters/dummy.ts b/packages/core/src/overrides/converters/dummy.ts index 3d666d99..5c973806 100644 --- a/packages/core/src/overrides/converters/dummy.ts +++ b/packages/core/src/overrides/converters/dummy.ts @@ -12,10 +12,12 @@ const converter: Converter = { original: event, }); }, - convertTo(internalResult) { + convertTo() { return Promise.resolve({ - type: "dummy", - original: internalResult, + type: "direct" as const, + data(result) { + return Promise.resolve(result); + }, }); }, name: "dummy", diff --git a/packages/core/src/overrides/converters/edge.ts b/packages/core/src/overrides/converters/edge.ts index 5261283d..92fb5821 100644 --- a/packages/core/src/overrides/converters/edge.ts +++ b/packages/core/src/overrides/converters/edge.ts @@ -1,9 +1,9 @@ import { Buffer } from "node:buffer"; +import { Writable } from "node:stream"; import cookieParser from "cookie"; -import { parseSetCookieHeader } from "@/http/util"; -import type { InternalEvent, InternalResult, MiddlewareResult } from "@/types/open-next"; +import type { InternalEvent, InternalResult, MiddlewareResult, StreamCreator } from "@/types/open-next"; import type { Converter } from "@/types/overrides"; import { getQueryFromSearchParams } from "./utils.js"; @@ -49,71 +49,123 @@ const converter: Converter = { cookies, }; }, - convertTo: async (result) => { - if ("internalEvent" in result) { - const request = new Request(result.internalEvent.url, { - body: result.internalEvent.body as BodyInit | undefined, - method: result.internalEvent.method, - headers: { - ...result.internalEvent.headers, - "x-forwarded-host": result.internalEvent.headers.host, - }, - }); - - if (globalThis.__dangerous_ON_edge_converter_returns_request === true) { - if (result.initialResponse) { - return { - initialResponse: result.initialResponse, - request, - }; + convertTo: async (event, context) => { + const request = event as Request; + const url = new URL(request.url); + const { promise: output, resolve: resolveOutput } = Promise.withResolvers(); + const abortSignal = (context as { abortSignal?: AbortSignal } | undefined)?.abortSignal; + const streamCreator: StreamCreator = { + writeHeaders(prelude) { + const responseHeaders = new Headers(prelude.headers); + for (const cookie of prelude.cookies) { + responseHeaders.append("Set-Cookie", cookie); } - return request; - } - - const cfCache = - (result.isISR || result.internalEvent.rawPath.startsWith("/_next/image")) && - process.env.DISABLE_CACHE !== "true" - ? { cacheEverything: true } - : {}; - - //TODO: we need to handle the PPR case here as well. - // We'll revisit this when we'll look at making StreamCreator mandatory. - return fetch(request, { - // This is a hack to make sure that the response is cached by Cloudflare - // See https://developers.cloudflare.com/workers/examples/cache-using-fetch/#caching-html-resources - // @ts-expect-error - This is a Cloudflare specific option - cf: cfCache, - }); - } - const headers = new Headers(); - for (const [key, value] of Object.entries(result.headers)) { - if (key === "set-cookie" && typeof value === "string") { - // If the value is a string, we need to parse it into an array - // This is the case for middleware direct result - const cookies = parseSetCookieHeader(value); - for (const cookie of cookies) { - headers.append(key, cookie); + + // TODO(vicb): this is a workaround to make PPR work with `wrangler dev` + // See https://github.com/cloudflare/workers-sdk/issues/8004 + if (url.hostname === "localhost") { + responseHeaders.set("Content-Encoding", "identity"); } - continue; - } - if (Array.isArray(value)) { - for (const v of value) { - headers.append(key, v); + + if (NULL_BODY_STATUSES.has(prelude.statusCode)) { + resolveOutput(new Response(null, { status: prelude.statusCode, headers: responseHeaders })); + return new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); } - } else { - headers.set(key, value); - } - } - // We should not return a body for statusCode's that doesn't allow bodies - const body = NULL_BODY_STATUSES.has(result.statusCode) ? null : (result.body as ReadableStream); + let controller: ReadableStreamDefaultController; + const readable = new ReadableStream({ + start(value) { + controller = value; + }, + }); + resolveOutput(new Response(readable, { status: prelude.statusCode, headers: responseHeaders })); + + return new Writable({ + write(chunk, _encoding, callback) { + try { + controller.enqueue(chunk); + callback(); + } catch (error: unknown) { + callback(error instanceof Error ? error : new Error(String(error))); + } + }, + final(callback) { + controller.close(); + callback(); + }, + destroy(error, callback) { + if (error) { + controller.error(error); + } else { + try { + controller.close(); + } catch { + // Ignore an already closed stream. + } + } + callback(error); + }, + }); + }, + abortSignal, + retainChunks: false, + }; - return new Response(body, { - status: result.statusCode, - headers, - }); + return { + type: "stream", + streamCreator, + output, + data: async (result) => { + if (!("internalEvent" in result)) { + return undefined; + } + return convertMiddlewareResult(result); + }, + }; }, name: "edge", }; +async function convertMiddlewareResult( + result: MiddlewareResult +): Promise { + const request = new Request(result.internalEvent.url, { + body: result.internalEvent.body as BodyInit | undefined, + method: result.internalEvent.method, + headers: { + ...result.internalEvent.headers, + "x-forwarded-host": result.internalEvent.headers.host, + }, + }); + + if (globalThis.__dangerous_ON_edge_converter_returns_request === true) { + if (result.initialResponse) { + return { + initialResponse: result.initialResponse, + request, + }; + } + return request; + } + + const cfCache = + (result.isISR || result.internalEvent.rawPath.startsWith("/_next/image")) && + process.env.DISABLE_CACHE !== "true" + ? { cacheEverything: true } + : {}; + + //TODO: we need to handle the PPR case here as well. + // We'll revisit this when we'll look at making StreamCreator mandatory. + return fetch(request, { + // This is a hack to make sure that the response is cached by Cloudflare + // See https://developers.cloudflare.com/workers/examples/cache-using-fetch/#caching-html-resources + // @ts-expect-error - This is a Cloudflare specific option + cf: cfCache, + }); +} + export default converter; diff --git a/packages/core/src/overrides/converters/node.ts b/packages/core/src/overrides/converters/node.ts index 60c5fffa..4c5bfac8 100644 --- a/packages/core/src/overrides/converters/node.ts +++ b/packages/core/src/overrides/converters/node.ts @@ -1,8 +1,8 @@ -import type { IncomingMessage } from "node:http"; +import type { IncomingMessage, ServerResponse } from "node:http"; import cookieParser from "cookie"; -import type { InternalResult } from "@/types/open-next"; +import type { StreamCreator } from "@/types/open-next"; import type { Converter } from "@/types/overrides"; import { extractHostFromHeaders, getQueryFromSearchParams } from "./utils.js"; @@ -46,12 +46,28 @@ const converter: Converter = { cookies, }; }, - // Nothing to do here, it's streaming - convertTo: async (internalResult: InternalResult) => ({ - body: internalResult.body, - headers: internalResult.headers, - statusCode: internalResult.statusCode, - }), + convertTo: async (_event, context) => { + const res = context as ServerResponse; + const abortController = new AbortController(); + const streamCreator: StreamCreator = { + writeHeaders: (prelude) => { + res.setHeader("Set-Cookie", prelude.cookies); + res.writeHead(prelude.statusCode, prelude.headers); + res.flushHeaders(); + return res; + }, + abortSignal: abortController.signal, + }; + + res.on("close", () => { + abortController.abort(); + }); + + return { + type: "stream", + streamCreator, + }; + }, name: "node", }; diff --git a/packages/core/src/overrides/wrappers/cloudflare-edge.ts b/packages/core/src/overrides/wrappers/cloudflare-edge.ts index 60f7e8c6..c61debe8 100644 --- a/packages/core/src/overrides/wrappers/cloudflare-edge.ts +++ b/packages/core/src/overrides/wrappers/cloudflare-edge.ts @@ -30,6 +30,7 @@ const handler: WrapperHandler } const internalEvent = await converter.convertFrom(request); + const output = await converter.convertTo(request); // Retrieve geo information from the cloudflare request // See https://developers.cloudflare.com/workers/runtime-apis/request @@ -43,13 +44,16 @@ const handler: WrapperHandler } } + if (output.type === "direct") { + return output.data(await handler(internalEvent, { waitUntil: ctx.waitUntil.bind(ctx) })); + } + const response = await handler(internalEvent, { + streamCreator: output.streamCreator, waitUntil: ctx.waitUntil.bind(ctx), }); - - const result = (await converter.convertTo(response)) as Response; - - return result; + const directResult = await output.data?.(response); + return directResult ?? output.output; }; export default { diff --git a/packages/core/src/overrides/wrappers/cloudflare-node.ts b/packages/core/src/overrides/wrappers/cloudflare-node.ts index 4f90d895..a53f4d23 100644 --- a/packages/core/src/overrides/wrappers/cloudflare-node.ts +++ b/packages/core/src/overrides/wrappers/cloudflare-node.ts @@ -1,11 +1,6 @@ -import { Writable } from "node:stream"; - -import type { InternalEvent, InternalResult, StreamCreator } from "@/types/open-next"; +import type { InternalEvent, InternalResult } from "@/types/open-next"; import type { Wrapper, WrapperHandler } from "@/types/overrides"; -// Response with null body status (101, 204, 205, or 304) cannot have a body. -const NULL_BODY_STATUSES = new Set([101, 204, 205, 304]); - const handler: WrapperHandler = async (handler, converter) => async (...args: unknown[]): Promise => { @@ -25,101 +20,19 @@ const handler: WrapperHandler = } const internalEvent = await converter.convertFrom(request); - const url = new URL(request.url); - - const { promise: promiseResponse, resolve: resolveResponse } = Promise.withResolvers(); - - const streamCreator: StreamCreator = { - writeHeaders(prelude: { - statusCode: number; - cookies: string[]; - headers: Record; - }): Writable { - const { statusCode, cookies, headers } = prelude; - - const responseHeaders = new Headers(headers); - for (const cookie of cookies) { - responseHeaders.append("Set-Cookie", cookie); - } - - // TODO(vicb): this is a workaround to make PPR work with `wrangler dev` - // See https://github.com/cloudflare/workers-sdk/issues/8004 - if (url.hostname === "localhost") { - responseHeaders.set("Content-Encoding", "identity"); - } - - // Optimize: skip ReadableStream creation for null body statuses - if (NULL_BODY_STATUSES.has(statusCode)) { - const response = new Response(null, { - status: statusCode, - headers: responseHeaders, - }); - resolveResponse(response); - - // Return a no-op Writable that discards all data - return new Writable({ - write(chunk, encoding, callback) { - callback(); - }, - }); - } - - let controller: ReadableStreamDefaultController; - const readable = new ReadableStream({ - start(c) { - controller = c; - }, - }); - - const response = new Response(readable, { - status: statusCode, - headers: responseHeaders, - }); - resolveResponse(response); - - return new Writable({ - write(chunk, encoding, callback) { - try { - controller.enqueue(chunk); - } catch (e: unknown) { - return callback(e instanceof Error ? e : new Error(String(e))); - } - callback(); - }, - final(callback) { - controller.close(); - callback(); - }, - destroy(error, callback) { - if (error) { - controller.error(error); - } else { - try { - controller.close(); - } catch { - // Ignore "This ReadableStream is closed" error - } - } - callback(error); - }, - }); - }, - // This is for passing along the original abort signal from the initial Request you retrieve in your worker - // Ensures that the response we pass to NextServer is aborted if the request is aborted - // By doing this `request.signal.onabort` will work in route handlers - abortSignal: abortSignal, - // There is no need to retain the chunks that were pushed to the response stream. - retainChunks: false, - }; + const output = await converter.convertTo(request, { abortSignal }); + if (output.type === "direct") { + return output.data(await handler(internalEvent)); + } ctx.waitUntil( handler(internalEvent, { - streamCreator, + streamCreator: output.streamCreator, waitUntil: ctx.waitUntil.bind(ctx), }) ); - return promiseResponse; + return output.output; }; export default { diff --git a/packages/core/src/overrides/wrappers/dummy.ts b/packages/core/src/overrides/wrappers/dummy.ts index 8649a8c8..dbbdb9f3 100644 --- a/packages/core/src/overrides/wrappers/dummy.ts +++ b/packages/core/src/overrides/wrappers/dummy.ts @@ -2,10 +2,14 @@ import type { InternalEvent } from "@/types/open-next"; import type { OpenNextHandlerOptions, Wrapper, WrapperHandler } from "@/types/overrides"; const dummyWrapper: WrapperHandler = - async (handler, _converter) => + async (handler, converter) => async (...args: unknown[]): Promise => { const [event, options] = args as [InternalEvent, OpenNextHandlerOptions | undefined]; - return await handler(event, options); + const output = await converter.convertTo(event, options); + if (output.type === "direct") { + return output.data(await handler(event, options)); + } + return handler(event, { ...options, streamCreator: output.streamCreator }); }; export default { diff --git a/packages/core/src/overrides/wrappers/express-dev.ts b/packages/core/src/overrides/wrappers/express-dev.ts index bcc9e8ef..996cb1cf 100644 --- a/packages/core/src/overrides/wrappers/express-dev.ts +++ b/packages/core/src/overrides/wrappers/express-dev.ts @@ -3,7 +3,6 @@ import path from "node:path"; import express from "express"; import { NextConfig } from "@/config/index"; -import type { StreamCreator } from "@/types/open-next.js"; import type { WrapperHandler } from "@/types/overrides.js"; import { getMonorepoRelativePath } from "@/utils/normalize-path"; @@ -22,13 +21,12 @@ const wrapper: WrapperHandler = async (handler, converter) => { app.all(`${NextConfig.basePath ?? ""}/_next/image`, async (req, res) => { const internalEvent = await converter.convertFrom(req); - const streamCreator: StreamCreator = { - writeHeaders: (prelude) => { - res.writeHead(prelude.statusCode, prelude.headers); - return res; - }, - }; - await imageHandler(internalEvent, { streamCreator }); + const output = await converter.convertTo(req, res); + if (output.type === "direct") { + await output.data(await imageHandler(internalEvent)); + return; + } + await imageHandler(internalEvent, { streamCreator: output.streamCreator }); }); app.all(/.*/, async (req, res) => { @@ -39,25 +37,13 @@ const wrapper: WrapperHandler = async (handler, converter) => { req.headers["x-forwarded-proto"] = req.protocol; } const internalEvent = await converter.convertFrom(req); - - const abortController = new AbortController(); - - const streamCreator: StreamCreator = { - writeHeaders: (prelude) => { - res.setHeader("Set-Cookie", prelude.cookies); - res.writeHead(prelude.statusCode, prelude.headers); - res.flushHeaders(); - return res; - }, - onFinish: () => {}, - abortSignal: abortController.signal, - }; - - res.on("close", () => { - abortController.abort(); - }); - - await handler(internalEvent, { streamCreator }); + const output = await converter.convertTo(req, res); + if (output.type === "direct") { + await output.data(await handler(internalEvent)); + return; + } + await handler(internalEvent, { streamCreator: output.streamCreator }); + await output.output; }); const server = app.listen(Number.parseInt(process.env.PORT ?? "3000", 10), () => { diff --git a/packages/core/src/overrides/wrappers/node.ts b/packages/core/src/overrides/wrappers/node.ts index c1432bf6..77ab85bf 100644 --- a/packages/core/src/overrides/wrappers/node.ts +++ b/packages/core/src/overrides/wrappers/node.ts @@ -1,6 +1,5 @@ import { createServer } from "node:http"; -import type { StreamCreator } from "@/types/open-next"; import type { Wrapper, WrapperHandler } from "@/types/overrides"; import { debug, error } from "../../adapters/logger"; @@ -8,22 +7,7 @@ import { debug, error } from "../../adapters/logger"; const wrapper: WrapperHandler = async (handler, converter) => { const server = createServer(async (req, res) => { const internalEvent = await converter.convertFrom(req); - - const abortController = new AbortController(); - - const streamCreator: StreamCreator = { - writeHeaders: (prelude) => { - res.setHeader("Set-Cookie", prelude.cookies); - res.writeHead(prelude.statusCode, prelude.headers); - res.flushHeaders(); - return res; - }, - abortSignal: abortController.signal, - }; - - res.on("close", () => { - abortController.abort(); - }); + const output = await converter.convertTo(req, res); if (internalEvent.rawPath === "/__health") { res.writeHead(200, { @@ -31,9 +15,12 @@ const wrapper: WrapperHandler = async (handler, converter) => { }); res.end("OK"); } else { - await handler(internalEvent, { - streamCreator, - }); + if (output.type === "direct") { + await output.data(await handler(internalEvent)); + } else { + await handler(internalEvent, { streamCreator: output.streamCreator }); + await output.output; + } } }); diff --git a/packages/core/src/plugins/resolve.ts b/packages/core/src/plugins/resolve.ts index 1a8521a5..df844d47 100644 --- a/packages/core/src/plugins/resolve.ts +++ b/packages/core/src/plugins/resolve.ts @@ -81,7 +81,11 @@ export function openNextResolvePlugin({ overrides, fnName }: IPluginSettings): P logger.debug(chalk.blue("OpenNext Resolve plugin"), fnName ? `for ${fnName}` : ""); build.onLoad({ filter: getCrossPlatformPathRegex("core/resolve.js") }, async (args) => { let contents = await readFile(args.path, "utf-8"); - const overridesEntries = Object.entries(overrides ?? {}); + const resolvedOverrides = { ...overrides }; + if (resolvedOverrides.wrapper === "aws-lambda-streaming" && !resolvedOverrides.converter) { + resolvedOverrides.converter = "aws-streaming"; + } + const overridesEntries = Object.entries(resolvedOverrides); for (let [overrideName, overrideValue] of overridesEntries) { if (!overrideValue) { continue; diff --git a/packages/core/src/types/open-next.ts b/packages/core/src/types/open-next.ts index 8216e491..298300bb 100644 --- a/packages/core/src/types/open-next.ts +++ b/packages/core/src/types/open-next.ts @@ -146,6 +146,7 @@ export type IncludedConverter = | "aws-apigw-v2" | "aws-apigw-v1" | "aws-cloudfront" + | "aws-streaming" | "edge" | "node" | "sqs-revalidate" diff --git a/packages/core/src/types/overrides.ts b/packages/core/src/types/overrides.ts index 84589959..a5cd7e1c 100644 --- a/packages/core/src/types/overrides.ts +++ b/packages/core/src/types/overrides.ts @@ -225,12 +225,41 @@ export type OpenNextHandler< R extends BaseEventOrResult = InternalResult, > = (event: E, options?: OpenNextHandlerOptions) => Promise; +export type ConverterOutput = + | { + type: "stream"; + streamCreator: StreamCreator; + /** + * A platform-specific result produced once the response stream completes. + * It is omitted by transports that write directly to their response. + */ + output?: Promise; + /** + * Handles exceptional non-streamed results (for example middleware rewrites). + */ + data?: (result: R) => Promise; + } + | { + type: "direct"; + /** + * Converts a handler result that has no HTTP response stream into the + * platform's native return value. + */ + data: (result: R) => Promise; + }; + export type Converter< E extends BaseEventOrResult = InternalEvent, R extends BaseEventOrResult = InternalResult, > = BaseOverride & { convertFrom: (event: unknown) => Promise; - convertTo: (result: R, originalRequest?: unknown) => Promise; + /** + * Creates the output for an invocation before the handler runs. + * + * `context` is supplied by the wrapper and contains transport-specific + * response objects such as a Node ServerResponse or Lambda response stream. + */ + convertTo: (event: unknown, context?: unknown) => Promise>; }; export type Warmer = BaseOverride & { diff --git a/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts b/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts index 25378578..697b81a8 100644 --- a/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts +++ b/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts @@ -4,10 +4,32 @@ import converter from "@opennextjs/aws/overrides/converters/aws-apigw-v1.js"; import type { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda"; import { describe, it, expect } from "vitest"; +async function convertResponse(result: { + body: ReadableStream; + headers: Record; + statusCode: number; +}) { + const output = await converter.convertTo({}); + if (output.type !== "stream" || !output.output) { + throw new Error("Expected a streaming converter output"); + } + const stream = output.streamCreator.writeHeaders({ + statusCode: result.statusCode, + headers: result.headers as Record, + cookies: [], + }); + await new Promise((resolve, reject) => { + stream.on("finish", resolve); + stream.on("error", reject); + Readable.fromWeb(result.body).pipe(stream); + }); + return output.output; +} + describe("convertTo", () => { describe("AWS API Gateway v2 Result", () => { it("Should parse the headers", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { "content-type": "application/json", @@ -24,7 +46,7 @@ describe("convertTo", () => { }); it("Should parse the headers with arrays", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { test: ["test1", "test2"], @@ -39,7 +61,7 @@ describe("convertTo", () => { }); it("Should parse single and array headers", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { single: "test", diff --git a/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts b/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts index 35fbe83b..ea2484ae 100644 --- a/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts +++ b/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts @@ -6,9 +6,31 @@ import { vi, describe, it, expect } from "vitest"; vi.mock("@/config/index.js", () => ({})); +async function convertResponse(result: { + body: ReadableStream; + headers: Record; + statusCode: number; +}) { + const output = await converter.convertTo({}); + if (output.type !== "stream" || !output.output) { + throw new Error("Expected a streaming converter output"); + } + const stream = output.streamCreator.writeHeaders({ + statusCode: result.statusCode, + headers: result.headers as Record, + cookies: [], + }); + await new Promise((resolve, reject) => { + stream.on("finish", resolve); + stream.on("error", reject); + Readable.fromWeb(result.body).pipe(stream); + }); + return output.output; +} + describe("convertTo", () => { it("Should parse the headers", async () => { - const response = await converter.convertTo({ + const response = await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { "content-type": "application/json", @@ -25,7 +47,7 @@ describe("convertTo", () => { }); it("Should parse the headers with arrays", async () => { - const response = await converter.convertTo({ + const response = await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { test: ["test1", "test2"], @@ -41,7 +63,7 @@ describe("convertTo", () => { describe("blacklisted headers", () => { it("should remove all blacklisted headers from the response", async () => { - const response = await converter.convertTo({ + const response = await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { Connection: "keep-alive", diff --git a/packages/tests-unit/tests/converters/aws-cloudfront.test.ts b/packages/tests-unit/tests/converters/aws-cloudfront.test.ts index e93ed6e8..77bfee04 100644 --- a/packages/tests-unit/tests/converters/aws-cloudfront.test.ts +++ b/packages/tests-unit/tests/converters/aws-cloudfront.test.ts @@ -6,9 +6,31 @@ import { vi, describe, it, expect } from "vitest"; vi.mock("@/config/index.js", () => ({})); +async function convertResponse(result: { + body: ReadableStream; + headers: Record; + statusCode: number; +}) { + const output = await converter.convertTo({}); + if (output.type !== "stream" || !output.output) { + throw new Error("Expected a streaming converter output"); + } + const stream = output.streamCreator.writeHeaders({ + statusCode: result.statusCode, + headers: result.headers as Record, + cookies: [], + }); + await new Promise((resolve, reject) => { + stream.on("finish", resolve); + stream.on("error", reject); + Readable.fromWeb(result.body).pipe(stream); + }); + return output.output; +} + describe("convertTo", () => { it("Should parse the headers", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { "content-type": "application/json", @@ -36,7 +58,7 @@ describe("convertTo", () => { }); it("Should parse the headers with arrays", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { test: ["test1", "test2"], @@ -61,7 +83,7 @@ describe("convertTo", () => { }); it("Should parse the headers with cookies", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { "set-cookie": @@ -87,7 +109,7 @@ describe("convertTo", () => { }); it("Should parse the headers with cookies + expires", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { "set-cookie": @@ -114,7 +136,7 @@ describe("convertTo", () => { describe("blacklisted headers", () => { it("should remove all blacklisted or read-only headers from the response", async () => { - const response = (await converter.convertTo({ + const response = (await convertResponse({ body: Readable.toWeb(Readable.from(Buffer.from(""))), headers: { Connection: "keep-alive", diff --git a/packages/tests-unit/tests/converters/aws-streaming.test.ts b/packages/tests-unit/tests/converters/aws-streaming.test.ts new file mode 100644 index 00000000..b5e7afc3 --- /dev/null +++ b/packages/tests-unit/tests/converters/aws-streaming.test.ts @@ -0,0 +1,54 @@ +import { PassThrough } from "node:stream"; + +import converter from "@opennextjs/aws/overrides/converters/aws-streaming.js"; +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/config/index.js", () => ({})); + +describe("aws-streaming converter", () => { + it("writes the Lambda integration prelude and streams the body", async () => { + const responseStream = new PassThrough() as PassThrough & { + setContentType: (contentType: string) => void; + }; + responseStream.setContentType = vi.fn(); + const chunks: Buffer[] = []; + responseStream.on("data", (chunk: Buffer) => chunks.push(chunk)); + + const output = await converter.convertTo( + {}, + { + responseStream, + writable: responseStream, + contentEncoding: "gzip", + } + ); + expect(output.type).toBe("stream"); + if (output.type !== "stream") { + throw new Error("Expected a streaming converter output"); + } + + const writable = output.streamCreator.writeHeaders({ + statusCode: 201, + cookies: ["session=abc"], + headers: { "content-type": "text/plain" }, + }); + writable.end("hello"); + await new Promise((resolve) => responseStream.on("end", resolve)); + + expect(responseStream.setContentType).toHaveBeenCalledWith( + "application/vnd.awslambda.http-integration-response" + ); + const outputBody = Buffer.concat(chunks); + const separator = outputBody.indexOf(Buffer.alloc(8)); + expect(separator).toBeGreaterThan(0); + expect(JSON.parse(outputBody.subarray(0, separator).toString())).toEqual({ + statusCode: 201, + cookies: ["session=abc"], + headers: { + "content-type": "text/plain", + "content-encoding": "gzip", + }, + }); + expect(outputBody.subarray(separator + 8).toString()).toBe("hello"); + }); +}); diff --git a/packages/tests-unit/tests/converters/sqs-revalidate.test.ts b/packages/tests-unit/tests/converters/sqs-revalidate.test.ts new file mode 100644 index 00000000..fe161f70 --- /dev/null +++ b/packages/tests-unit/tests/converters/sqs-revalidate.test.ts @@ -0,0 +1,25 @@ +import converter from "@opennextjs/aws/overrides/converters/sqs-revalidate.js"; +import { describe, expect, it } from "vitest"; + +describe("sqs-revalidate converter", () => { + it("returns a direct finalizer for failed records", async () => { + const output = await converter.convertTo({}); + expect(output.type).toBe("direct"); + if (output.type !== "direct") { + throw new Error("Expected a direct converter output"); + } + + await expect( + output.data({ + type: "revalidate", + records: [ + { host: "example.com", url: "/first", id: "one" }, + { host: "example.com", url: "/second", id: "two" }, + ], + }) + ).resolves.toEqual({ + type: "revalidate", + batchItemFailures: [{ itemIdentifier: "one" }, { itemIdentifier: "two" }], + }); + }); +}); From 113b5a03c25980a598994e4dc9082d396eabc143 Mon Sep 17 00:00:00 2001 From: Nicolas Dorseuil Date: Tue, 14 Jul 2026 16:43:51 +0200 Subject: [PATCH 2/2] refactor: Remove retainChunks option and update response handling for improved memory management --- .../src/overrides/converters/aws-streaming.ts | 1 - .../overrides/converters/response-stream.ts | 1 - .../wrappers/aws-lambda-streaming.ts | 2 +- packages/core/src/core/requestHandler.ts | 19 +++++++++++-------- packages/core/src/core/routing/util.ts | 11 ----------- packages/core/src/http/openNextResponse.ts | 12 ------------ .../core/src/overrides/converters/edge.ts | 1 - packages/core/src/types/open-next.ts | 14 +------------- .../tests/core/routing/util.test.ts | 9 +++------ 9 files changed, 16 insertions(+), 54 deletions(-) diff --git a/packages/aws/src/overrides/converters/aws-streaming.ts b/packages/aws/src/overrides/converters/aws-streaming.ts index b962e755..812dce6d 100644 --- a/packages/aws/src/overrides/converters/aws-streaming.ts +++ b/packages/aws/src/overrides/converters/aws-streaming.ts @@ -31,7 +31,6 @@ const converter: Converter = { responseStream.write(new Uint8Array(8)); return writable; }, - retainChunks: false, }; return { type: "stream" as const, streamCreator }; diff --git a/packages/aws/src/overrides/converters/response-stream.ts b/packages/aws/src/overrides/converters/response-stream.ts index 393b11e6..d1324ddf 100644 --- a/packages/aws/src/overrides/converters/response-stream.ts +++ b/packages/aws/src/overrides/converters/response-stream.ts @@ -39,7 +39,6 @@ export function createBufferedStreamCreator( }, }); }, - retainChunks: false, }; return { streamCreator, output }; diff --git a/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts b/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts index ad75069e..a10abd96 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts @@ -80,7 +80,7 @@ const handler: WrapperHandler = async (handler, converter) => } const response = await handler(internalEvent, { streamCreator: output.streamCreator }); - if (globalThis.isEdgeRuntime ?? false) { + if ((globalThis.isEdgeRuntime ?? false) && response.body) { const stream = output.streamCreator.writeHeaders({ statusCode: response.statusCode, headers: response.headers as Record, diff --git a/packages/core/src/core/requestHandler.ts b/packages/core/src/core/requestHandler.ts index aa07f1a5..b9260891 100644 --- a/packages/core/src/core/requestHandler.ts +++ b/packages/core/src/core/requestHandler.ts @@ -135,12 +135,14 @@ export async function openNextHandler( ); response.statusCode = routingResult.statusCode; response.flushHeaders(); - const [bodyToConsume, bodyToReturn] = routingResult.body.tee(); - for await (const chunk of bodyToConsume) { - response.write(chunk); + if (routingResult.body) { + const [bodyToConsume, bodyToReturn] = routingResult.body.tee(); + for await (const chunk of bodyToConsume) { + response.write(chunk); + } + routingResult.body = bodyToReturn; } response.end(); - routingResult.body = bodyToReturn; } return routingResult; } @@ -181,8 +183,10 @@ export async function openNextHandler( if (routingResult.initialResponse) { res.statusCode = routingResult.initialResponse.statusCode; res.flushHeaders(); - for await (const chunk of routingResult.initialResponse.body) { - res.write(chunk); + if (routingResult.initialResponse.body) { + for await (const chunk of routingResult.initialResponse.body) { + res.write(chunk); + } } //We create a special response for the PPR resume request @@ -218,13 +222,12 @@ export async function openNextHandler( }); //#endOverride - const { statusCode, headers: responseHeaders, isBase64Encoded, body } = convertRes(res); + const { statusCode, headers: responseHeaders, isBase64Encoded } = convertRes(res); const internalResult = { type: internalEvent.type, statusCode, headers: responseHeaders, - body, isBase64Encoded, }; diff --git a/packages/core/src/core/routing/util.ts b/packages/core/src/core/routing/util.ts index 0d4ead50..ffe29031 100644 --- a/packages/core/src/core/routing/util.ts +++ b/packages/core/src/core/routing/util.ts @@ -104,21 +104,10 @@ export function convertRes(res: OpenNextNodeResponse): InternalResult { // Probably some kind of race condition const headers = parseHeaders(res.getFixedHeaders()); const isBase64Encoded = isBinaryContentType(headers["content-type"]) || !!headers["content-encoding"]; - const body = new ReadableStream({ - pull(controller) { - if (!res._chunks || res._chunks.length === 0) { - controller.close(); - return; - } - - controller.enqueue(res._chunks.shift()); - }, - }); return { type: "core", statusCode, headers, - body, isBase64Encoded, }; } diff --git a/packages/core/src/http/openNextResponse.ts b/packages/core/src/http/openNextResponse.ts index b76cc271..dec7779a 100644 --- a/packages/core/src/http/openNextResponse.ts +++ b/packages/core/src/http/openNextResponse.ts @@ -18,7 +18,6 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { statusMessage = ""; headers: OutgoingHttpHeaders = {}; headersSent = false; - _chunks: Buffer[] = []; headersAlreadyFixed = false; private _cookies: string[] = []; @@ -255,10 +254,6 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { return this.headers; } - getBody() { - return Buffer.concat(this._chunks); - } - private _internalWrite(chunk: Buffer | string, encoding: BufferEncoding) { // When encoding === 'buffer', chunk is already a Buffer // and does not need to be converted again. @@ -266,10 +261,6 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { // official type definition const buffer = encoding === "buffer" ? (chunk as Buffer) : Buffer.from(chunk, encoding); this.bodyLength += buffer.length; - if (this.streamCreator?.retainChunks !== false) { - // Avoid keeping chunks around when the `StreamCreator` supports it to save memory - this._chunks.push(buffer); - } // No need to pass the encoding for buffers this.push(buffer); this.streamCreator?.onWrite?.(); @@ -342,9 +333,6 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { } send() { - for (const chunk of this._chunks) { - this.write(chunk); - } this.end(); } diff --git a/packages/core/src/overrides/converters/edge.ts b/packages/core/src/overrides/converters/edge.ts index 92fb5821..cf00cf56 100644 --- a/packages/core/src/overrides/converters/edge.ts +++ b/packages/core/src/overrides/converters/edge.ts @@ -112,7 +112,6 @@ const converter: Converter = { }); }, abortSignal, - retainChunks: false, }; return { diff --git a/packages/core/src/types/open-next.ts b/packages/core/src/types/open-next.ts index 298300bb..5fe43b9e 100644 --- a/packages/core/src/types/open-next.ts +++ b/packages/core/src/types/open-next.ts @@ -43,7 +43,7 @@ export type MiddlewareEvent = InternalEvent & { export type InternalResult = { statusCode: number; headers: Record; - body: ReadableStream; + body?: ReadableStream; isBase64Encoded: boolean; rewriteStatusCode?: number; } & BaseEventOrResult<"core">; @@ -72,18 +72,6 @@ export interface StreamCreator { onWrite?: () => void; onFinish?: (length: number) => void; abortSignal?: AbortSignal; - /** - * Normally there is no need to retain the chunks that have been pushed to the response stream. - * - * However some implementations use a fake `StreamCreator` and expect the chunks to be retained. - * When your stream controller implementation doesn't need to retain the chunk, you can set this - * to `false` to reduce memory usage. - * - * @see https://github.com/opennextjs/opennextjs-aws/blob/main/packages/open-next/src/overrides/wrappers/aws-lambda.ts - * - * @default true for backward compatibility. - */ - retainChunks?: boolean; } export type WaitUntil = (promise: Promise) => void; diff --git a/packages/tests-unit/tests/core/routing/util.test.ts b/packages/tests-unit/tests/core/routing/util.test.ts index e09eaa8e..80daf99d 100644 --- a/packages/tests-unit/tests/core/routing/util.test.ts +++ b/packages/tests-unit/tests/core/routing/util.test.ts @@ -49,9 +49,6 @@ function createResponse(res: Partial) { return { statusCode: res.statusCode, getFixedHeaders: () => res.headers ?? {}, - body: res.body ?? "", - getBody: () => Buffer.from(res.body ?? ""), - _chunks: res.body ? [Buffer.from(res.body)] : [], }; } @@ -237,7 +234,7 @@ describe("convertRes", () => { isBase64Encoded: false, }) ); - expect(await fromReadableStream(result.body)).toEqual(res.getBody().toString()); + expect(result.body).toBeUndefined(); }); it("convert a response with default status code", async () => { @@ -260,7 +257,7 @@ describe("convertRes", () => { isBase64Encoded: false, }) ); - expect(await fromReadableStream(result.body)).toEqual(res.getBody().toString()); + expect(result.body).toBeUndefined(); }); it("convert a response with base64 encoding", async () => { @@ -283,7 +280,7 @@ describe("convertRes", () => { isBase64Encoded: true, }) ); - expect(await fromReadableStream(result.body)).toEqual(res.getBody().toString()); + expect(result.body).toBeUndefined(); }); });