diff --git a/examples-cloudflare/e2e/app-pages-router/package.json b/examples-cloudflare/e2e/app-pages-router/package.json index 327825a1..9c93e79d 100644 --- a/examples-cloudflare/e2e/app-pages-router/package.json +++ b/examples-cloudflare/e2e/app-pages-router/package.json @@ -9,11 +9,12 @@ "start": "next start --port 3003", "lint": "next lint", "clean": "rm -rf .turbo node_modules .next .open-next", - "build:worker:cf": "pnpm opennextjs-cloudflare build", + "build:worker": "pnpm opennextjs-cloudflare build", + "build:worker:cf": "pnpm build:worker", "preview:worker": "pnpm opennextjs-cloudflare preview", "preview": "pnpm build:worker && pnpm preview:worker", "e2e:cf": "playwright test -c e2e/playwright.config.ts", - "build:worker-turbopack": "pnpm opennextjs-cloudflare build --openNextConfigPath open-next.turbopack.config.ts", + "build:worker-turbopack": "pnpm build:worker --openNextConfigPath open-next.turbopack.config.ts", "e2e-turbopack": "playwright test -c e2e/playwright.turbopack.config.ts" }, "dependencies": { diff --git a/examples-cloudflare/e2e/app-router/package.json b/examples-cloudflare/e2e/app-router/package.json index d1924fbb..d6fc3cb5 100644 --- a/examples-cloudflare/e2e/app-router/package.json +++ b/examples-cloudflare/e2e/app-router/package.json @@ -9,7 +9,8 @@ "start": "next start --port 3001", "lint": "next lint", "clean": "rm -rf .turbo node_modules .next .open-next", - "build:worker:cf": "pnpm opennextjs-cloudflare build", + "build:worker": "pnpm opennextjs-cloudflare build", + "build:worker:cf": "pnpm build:worker", "preview:worker": "pnpm opennextjs-cloudflare preview", "preview": "pnpm build:worker && pnpm preview:worker", "e2e:cf": "playwright test -c e2e/playwright.config.ts", diff --git a/examples-cloudflare/e2e/experimental/package.json b/examples-cloudflare/e2e/experimental/package.json index a2b07c3f..2d42e6d0 100644 --- a/examples-cloudflare/e2e/experimental/package.json +++ b/examples-cloudflare/e2e/experimental/package.json @@ -8,7 +8,8 @@ "start": "next start --port 3004", "lint": "next lint", "clean": "rm -rf .turbo node_modules .next .open-next", - "build:worker:cf": "pnpm opennextjs-cloudflare build", + "build:worker": "pnpm opennextjs-cloudflare build", + "build:worker:cf": "pnpm build:worker", "preview:worker": "pnpm opennextjs-cloudflare preview", "preview": "pnpm build:worker && pnpm preview:worker", "e2e:cf": "playwright test -c e2e/playwright.config.ts" diff --git a/examples-cloudflare/e2e/pages-router/package.json b/examples-cloudflare/e2e/pages-router/package.json index 9bec254e..bbe1152b 100644 --- a/examples-cloudflare/e2e/pages-router/package.json +++ b/examples-cloudflare/e2e/pages-router/package.json @@ -9,7 +9,8 @@ "start": "next start --port 3002", "lint": "next lint", "clean": "rm -rf .turbo node_modules .next .open-next", - "build:worker:cf": "pnpm opennextjs-cloudflare build", + "build:worker": "pnpm opennextjs-cloudflare build", + "build:worker:cf": "pnpm build:worker", "preview:worker": "pnpm opennextjs-cloudflare preview", "preview": "pnpm build:worker && pnpm preview:worker", "e2e:cf": "playwright test -c e2e/playwright.config.ts", diff --git a/packages/aws/src/adapter.ts b/packages/aws/src/adapter.ts index 4496c30f..b8eea732 100644 --- a/packages/aws/src/adapter.ts +++ b/packages/aws/src/adapter.ts @@ -1,15 +1,17 @@ import { buildAdapter } from "@opennextjs/core/build/adapter.js"; import type { BuildOptions } from "@opennextjs/core/build/helper.js"; import * as buildHelper from "@opennextjs/core/build/helper.js"; +import { validateConfig } from "@opennextjs/core/build/validateConfig.js"; import type { ContentUpdater } from "@opennextjs/core/plugins/content-updater.js"; import { externalChunksPlugin, inlineRouteHandler } from "@opennextjs/core/plugins/inlineRouteHandlers.js"; +import type { BundleDefaults } from "@opennextjs/core/plugins/resolve.js"; import type { NextAdapterOutputs } from "@opennextjs/core/types/adapter.js"; -export default buildAdapter((_config, buildOpts: BuildOptions) => ({ - defaultOverrides: { +export default buildAdapter((_config, buildOpts: BuildOptions) => { + const defaultOverrides = { server: { wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", - converter: "@opennextjs/aws/overrides/converters/aws-apigw-v2.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", incrementalCache: "@opennextjs/aws/overrides/incrementalCache/s3.js", tagCache: "@opennextjs/aws/overrides/tagCache/dynamodb.js", queue: "@opennextjs/aws/overrides/queue/sqs.js", @@ -35,12 +37,22 @@ export default buildAdapter((_config, buildOpts: BuildOptions) => ({ tagCache: "@opennextjs/aws/overrides/tagCache/dynamodb-lite.js", queue: "@opennextjs/aws/overrides/queue/sqs-lite.js", }, - }, - serverBundle: { - externals: ["./middleware.mjs"], - additionalPlugins: (updater: ContentUpdater, outputs: NextAdapterOutputs) => { - const packagePath = buildHelper.getPackagePath(buildOpts); - return [inlineRouteHandler(updater, outputs, packagePath), externalChunksPlugin(outputs, packagePath)]; + } satisfies BundleDefaults; + return { + defaultOverrides, + validateConfig: (config) => { + const result = validateConfig(config, defaultOverrides); + return !result.success && result.level === "error" ? { ...result, shouldThrow: true } : result; + }, + serverBundle: { + externals: ["./middleware.mjs"], + additionalPlugins: (updater: ContentUpdater, outputs: NextAdapterOutputs) => { + const packagePath = buildHelper.getPackagePath(buildOpts); + return [ + inlineRouteHandler(updater, outputs, packagePath), + externalChunksPlugin(outputs, packagePath), + ]; + }, }, - }, -})); + }; +}); diff --git a/packages/aws/src/overrides/converters/aws-apigw-v1.ts b/packages/aws/src/overrides/converters/aws-apigw-v1.ts index 44554be3..1670d74f 100644 --- a/packages/aws/src/overrides/converters/aws-apigw-v1.ts +++ b/packages/aws/src/overrides/converters/aws-apigw-v1.ts @@ -3,11 +3,13 @@ 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, toReadableStream } from "@opennextjs/core/utils/stream.js"; +import { toReadableStream } from "@opennextjs/core/utils/stream.js"; import type { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda"; +import { createBufferedStreamCreator } from "./response-stream.js"; + function normalizeAPIGatewayProxyEventHeaders(event: APIGatewayProxyEvent): Record { const headers: Record = {}; @@ -88,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 { @@ -102,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); @@ -117,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 b053f0fc..216cd0f2 100644 --- a/packages/aws/src/overrides/converters/aws-apigw-v2.ts +++ b/packages/aws/src/overrides/converters/aws-apigw-v2.ts @@ -7,11 +7,13 @@ 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, toReadableStream } from "@opennextjs/core/utils/stream.js"; +import { toReadableStream } 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 = [ @@ -70,7 +72,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 { @@ -94,9 +98,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]) => @@ -112,21 +120,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 62db8481..01305ba9 100644 --- a/packages/aws/src/overrides/converters/aws-cloudfront.ts +++ b/packages/aws/src/overrides/converters/aws-cloudfront.ts @@ -6,7 +6,7 @@ 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, toReadableStream } from "@opennextjs/core/utils/stream.js"; +import { toReadableStream } from "@opennextjs/core/utils/stream.js"; import type { CloudFrontCustomOrigin, CloudFrontHeaders, @@ -14,6 +14,9 @@ import type { CloudFrontRequestEvent, 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", @@ -128,59 +131,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); @@ -188,7 +196,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..1341b3fe --- /dev/null +++ b/packages/aws/src/overrides/converters/aws-streaming.ts @@ -0,0 +1,90 @@ +import type { Transform, Writable } from "node:stream"; +import { finished, pipeline } from "node:stream/promises"; +import zlib from "node:zlib"; + +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 { withCompressionVary } from "../wrappers/compression.js"; + +import { convertFromAPIGatewayProxyEventV2 } from "./aws-apigw-v2.js"; + +const NULL_BODY_STATUSES = new Set([101, 103, 204, 205, 304]); + +type StreamingContext = { + responseStream: Writable & { setContentType(contentType: string): void }; + contentEncoding: string; +}; + +const converter: Converter = { + convertFrom: (event) => convertFromAPIGatewayProxyEventV2(event as APIGatewayProxyEventV2), + convertTo: async (_event, context) => { + const { responseStream, contentEncoding } = context as StreamingContext; + const { promise: output, resolve, reject } = Promise.withResolvers(); + const streamCreator: StreamCreator = { + writeHeaders(prelude) { + const existingContentEncoding = prelude.headers["content-encoding"]; + const shouldCompress = + !existingContentEncoding && + contentEncoding !== "identity" && + !NULL_BODY_STATUSES.has(prelude.statusCode); + const headers = { ...prelude.headers }; + if (shouldCompress) { + Object.assign(headers, withCompressionVary({ ...headers, "content-encoding": contentEncoding })); + delete headers["content-length"]; + } else if (!existingContentEncoding) { + headers["content-encoding"] = "identity"; + } + + responseStream.setContentType("application/vnd.awslambda.http-integration-response"); + responseStream.write( + JSON.stringify({ + statusCode: prelude.statusCode, + cookies: prelude.cookies, + headers, + }) + ); + responseStream.write(new Uint8Array(8)); + + let writable: Writable = responseStream; + let completion = finished(responseStream); + if (shouldCompress) { + const transform = createCompressionStream(contentEncoding); + writable = transform; + completion = pipeline(transform, responseStream); + } + void completion.then(resolve, reject); + return writable; + }, + }; + + return { type: "stream" as const, streamCreator, output }; + }, + name: "aws-streaming", +}; + +/** + * Creates the compression transform selected for the Lambda response. + * + * @param encoding - The negotiated response content encoding. + * @returns The corresponding Node.js compression transform. + * @throws If the encoding is not supported. + */ +function createCompressionStream(encoding: string): Transform { + switch (encoding) { + case "br": + return zlib.createBrotliCompress({ + flush: zlib.constants.BROTLI_OPERATION_FLUSH, + finishFlush: zlib.constants.BROTLI_OPERATION_FINISH, + }); + case "gzip": + return zlib.createGzip({ flush: zlib.constants.Z_SYNC_FLUSH }); + case "deflate": + return zlib.createDeflate({ flush: zlib.constants.Z_SYNC_FLUSH }); + default: + throw new Error(`Unsupported response content encoding: ${encoding}`); + } +} + +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..7fa5fdbb --- /dev/null +++ b/packages/aws/src/overrides/converters/response-stream.ts @@ -0,0 +1,60 @@ +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]; + +/** + * Creates a stream that buffers a response into a platform result. + * + * @param createOutput - Converts buffered response data to the platform result. + * @returns The response stream creator and its eventual platform output. + */ +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; + let finalized = false; + 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) { + finalized = true; + if (!prelude) { + const error = new Error("Response stream finished before headers were written"); + reject(error); + callback(error); + return; + } + try { + const isBase64Encoded = + prelude.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))); + } + }, + destroy(error, callback) { + if (!finalized) { + reject(error ?? new Error("Response stream was destroyed before it finished")); + } + callback(error); + }, + }); + }, + }; + + 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..fe52c2f8 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda-compressed.ts @@ -1,14 +1,16 @@ -import { Readable, type Transform, Writable } from "node:stream"; -import type { ReadableStream } from "node:stream/web"; +import type { Transform } from "node:stream"; +import { pipeline } from "node:stream/promises"; 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"; -import { formatWarmerResponse } from "./aws-lambda.js"; +import { formatWarmerResponse, streamResponse } from "./aws-lambda.js"; +import { selectCompressionEncoding, withCompressionVary } from "./compression.js"; + +const NULL_BODY_STATUSES = new Set([101, 103, 204, 205, 304]); const handler: WrapperHandler = async (handler, converter) => @@ -20,55 +22,22 @@ 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"] ?? ""; + const contentEncoding = selectCompressionEncoding(acceptEncoding); - let contentEncoding: string | null = null; - if (acceptEncoding?.includes("br")) { - contentEncoding = "br"; - } else if (acceptEncoding?.includes("gzip")) { - contentEncoding = "gzip"; - } else if (acceptEncoding?.includes("deflate")) { - 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); + if (directResult !== undefined) return directResult; + await streamResponse(response, withCompression(output.streamCreator, contentEncoding)); + return output.output; }; export default { @@ -77,38 +46,52 @@ 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; +/** + * Adds negotiated compression to a response stream creator. + * + * @param streamCreator - The underlying platform response stream creator. + * @param encoding - The negotiated response content encoding, if any. + * @returns A stream creator that compresses response bodies when required. + */ +function withCompression(streamCreator: StreamCreator, encoding: string | null): StreamCreator { + if (!encoding) return streamCreator; + return { + ...streamCreator, + writeHeaders(prelude) { + if (prelude.headers["content-encoding"] || NULL_BODY_STATUSES.has(prelude.statusCode)) { + return streamCreator.writeHeaders(prelude); + } + const { "content-length": _contentLength, ...headers } = prelude.headers; + const target = streamCreator.writeHeaders({ + ...prelude, + headers: withCompressionVary({ ...headers, "content-encoding": encoding }), + isBase64Encoded: true, + }); + 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; + } + void pipeline(transform, target).catch(() => undefined); + 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..2c38e971 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda-streaming.ts @@ -1,12 +1,12 @@ -import { Readable, type Writable } from "node:stream"; -import zlib from "node:zlib"; +import { finished } from "node:stream/promises"; -import { debug, 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"; +import { streamResponse } from "./aws-lambda.js"; +import { selectCompressionEncoding } from "./compression.js"; + type AwsLambdaEvent = APIGatewayProxyEventV2 | WarmerEvent; type AwsLambdaReturn = void; @@ -27,6 +27,7 @@ const handler: WrapperHandler = async (handler, converter) => if ("type" in event) { const result = await formatWarmerResponse(event); responseStream.end(Buffer.from(JSON.stringify(result)), "utf-8"); + await finished(responseStream); // disabled for now, we'll need to revisit this later if needed. //TODO: revisit that later // await globalThis.__next_route_preloader("warmerEvent"); @@ -38,67 +39,22 @@ const handler: WrapperHandler = async (handler, converter) => //Handle compression const acceptEncoding = internalEvent.headers["Accept-Encoding"] ?? internalEvent.headers["accept-encoding"] ?? ""; - let contentEncoding: string; - let compressedStream: Writable | undefined; + const contentEncoding = selectCompressionEncoding(acceptEncoding) ?? "identity"; - responseStream.on("error", (err) => { - error(err); - responseStream.end(); + const output = await converter.convertTo(event, { + responseStream, + contentEncoding, }); - - if (acceptEncoding.includes("br")) { - contentEncoding = "br"; - compressedStream = zlib.createBrotliCompress({ - flush: zlib.constants.BROTLI_OPERATION_FLUSH, - finishFlush: zlib.constants.BROTLI_OPERATION_FINISH, - }); - compressedStream.pipe(responseStream); - } else if (acceptEncoding.includes("gzip")) { - contentEncoding = "gzip"; - compressedStream = zlib.createGzip({ - flush: zlib.constants.Z_SYNC_FLUSH, - }); - compressedStream.pipe(responseStream); - } else if (acceptEncoding.includes("deflate")) { - contentEncoding = "deflate"; - compressedStream = zlib.createDeflate({ - flush: zlib.constants.Z_SYNC_FLUSH, - }); - compressedStream.pipe(responseStream); - } else { - contentEncoding = "identity"; - compressedStream = responseStream; + if (output.type === "direct") { + await output.data(await handler(internalEvent)); + return; } - 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 isUsingEdge = globalThis.isEdgeRuntime ?? false; - if (isUsingEdge) { - debug("Headers has not been set, we must be in the edge runtime"); - const stream = streamCreator.writeHeaders({ - statusCode: response.statusCode, - headers: response.headers as Record, - cookies: [], - }); - Readable.fromWeb(response.body).pipe(stream); + const response = await handler(internalEvent, { streamCreator: output.streamCreator }); + if (globalThis.isEdgeRuntime ?? false) { + await streamResponse(response, output.streamCreator); } - - // return converter.convertTo(response); + await output.output; } ) 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..bdd85e84 100644 --- a/packages/aws/src/overrides/wrappers/aws-lambda.ts +++ b/packages/aws/src/overrides/wrappers/aws-lambda.ts @@ -1,7 +1,9 @@ -import { Writable } from "node:stream"; +import { Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; import type { WarmerEvent, WarmerResponse } from "@opennextjs/core/adapters/warmer-function.js"; -import type { StreamCreator } from "@opennextjs/core/types/open-next.js"; +import { parseSetCookieHeader } from "@opennextjs/core/http/util.js"; +import type { InternalResult, 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 +26,15 @@ 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); + if (directResult !== undefined) return directResult; + await streamResponse(response, output.streamCreator); + return output.output; }; export default { @@ -52,3 +42,32 @@ export default { name: "aws-lambda", supportStreaming: false, }; + +/** + * Streams a returned response body when the handler did not write it directly. + * + * @param response - The internal response returned by the handler. + * @param streamCreator - The converter's platform response stream creator. + * @returns A promise that resolves after the returned body has been written. + */ +export async function streamResponse(response: InternalResult, streamCreator: StreamCreator): Promise { + const { "set-cookie": setCookie, ...responseHeaders } = response.headers; + const headers = Object.fromEntries( + Object.entries(responseHeaders).map(([key, value]) => [ + key, + Array.isArray(value) ? value.join(",") : value, + ]) + ); + const cookies = Array.isArray(setCookie) ? setCookie : setCookie ? parseSetCookieHeader(setCookie) : []; + const stream = streamCreator.writeHeaders({ + statusCode: response.statusCode, + headers, + cookies, + isBase64Encoded: response.isBase64Encoded, + }); + if (!response.body) { + stream.end(); + return; + } + await pipeline(Readable.fromWeb(response.body), stream); +} diff --git a/packages/aws/src/overrides/wrappers/compression.ts b/packages/aws/src/overrides/wrappers/compression.ts new file mode 100644 index 00000000..70ccebfc --- /dev/null +++ b/packages/aws/src/overrides/wrappers/compression.ts @@ -0,0 +1,43 @@ +export type CompressionEncoding = "br" | "gzip" | "deflate"; + +/** + * Selects the best supported content encoding accepted by a client. + * + * @param acceptEncoding - The request's Accept-Encoding header value. + * @returns The selected encoding, or null when compression is not acceptable. + */ +export function selectCompressionEncoding(acceptEncoding: string): CompressionEncoding | null { + const qualities = new Map(); + for (const value of acceptEncoding.split(",")) { + const [name, ...parameters] = value.trim().toLowerCase().split(";"); + if (!name) continue; + const qualityParameter = parameters.find((parameter) => parameter.trim().startsWith("q=")); + const quality = qualityParameter ? Number(qualityParameter.trim().slice(2)) : 1; + qualities.set(name, Number.isFinite(quality) ? Math.min(Math.max(quality, 0), 1) : 0); + } + + let selected: CompressionEncoding | null = null; + let selectedQuality = 0; + for (const encoding of ["br", "gzip", "deflate"] as const) { + const quality = qualities.get(encoding) ?? qualities.get("*") ?? 0; + if (quality > selectedQuality) { + selected = encoding; + selectedQuality = quality; + } + } + return selected; +} + +/** + * Adds Accept-Encoding to a response's Vary header. + * + * @param headers - Response headers to update. + * @returns A copy of the headers with the required Vary value. + */ +export function withCompressionVary(headers: Record): Record { + const vary = headers.vary; + if (vary?.split(",").some((value) => value.trim().toLowerCase() === "accept-encoding")) { + return headers; + } + return { ...headers, vary: vary ? `${vary}, Accept-Encoding` : "Accept-Encoding" }; +} diff --git a/packages/core/src/build/adapter.spec.ts b/packages/core/src/build/adapter.spec.ts index ed5925f5..e1369273 100644 --- a/packages/core/src/build/adapter.spec.ts +++ b/packages/core/src/build/adapter.spec.ts @@ -558,7 +558,7 @@ describe("buildAdapter", () => { await adapter.modifyConfig(nextConfig, { phase: "production" }); const ctx = createMockContext(); await adapter.onBuildComplete(ctx); - expect(buildOpenNextOutput).toHaveBeenCalledWith(expect.any(Object)); + expect(buildOpenNextOutput).toHaveBeenCalledWith(expect.any(Object), undefined); const fs = await import("node:fs"); expect(fs.default.writeFileSync).toHaveBeenCalledWith( expect.stringMatching(/\/\.open-next\/open-next\.output\.json$/), diff --git a/packages/core/src/build/adapter.ts b/packages/core/src/build/adapter.ts index 0ef9f62c..865d6203 100644 --- a/packages/core/src/build/adapter.ts +++ b/packages/core/src/build/adapter.ts @@ -267,7 +267,7 @@ export function buildAdapter( if (!adapterOptions.skipGenerateOutput) { const output = adapterOptions.generateOutput ? await adapterOptions.generateOutput(buildOpts) - : await buildOpenNextOutput(buildOpts); + : await buildOpenNextOutput(buildOpts, bundleDefaults); fs.writeFileSync( path.join(buildOpts.appBuildOutputPath, ".open-next", "open-next.output.json"), JSON.stringify(output) diff --git a/packages/core/src/build/createServerBundle.ts b/packages/core/src/build/createServerBundle.ts index a68f3da9..503edd00 100644 --- a/packages/core/src/build/createServerBundle.ts +++ b/packages/core/src/build/createServerBundle.ts @@ -58,7 +58,15 @@ export async function createServerBundle( const routes = fnOptions.routes; routes.forEach((route) => foundRoutes.add(route)); if (fnOptions.runtime === "edge") { - await generateEdgeBundle(name, options, fnOptions, undefined, codeCustomization.bundleDefaults?.edge); + await generateEdgeBundle( + name, + options, + fnOptions, + undefined, + fnOptions.placement === "global" + ? (codeCustomization.bundleDefaults?.global ?? codeCustomization.bundleDefaults?.edge) + : codeCustomization.bundleDefaults?.edge + ); } else { await generateBundle(name, options, fnOptions, codeCustomization, nextOutputs); } @@ -213,7 +221,10 @@ async function generateBundle( // Next.js app. const overrides = fnOptions.override ?? {}; - const defaultOverrides = codeCustomization.bundleDefaults?.server; + const defaultOverrides = + fnOptions.placement === "global" + ? (codeCustomization.bundleDefaults?.global ?? codeCustomization.bundleDefaults?.server) + : codeCustomization.bundleDefaults?.server; const disableRouting = config.middleware?.external; diff --git a/packages/core/src/build/generateOutput.spec.ts b/packages/core/src/build/generateOutput.spec.ts index 0030820d..656bd22b 100644 --- a/packages/core/src/build/generateOutput.spec.ts +++ b/packages/core/src/build/generateOutput.spec.ts @@ -14,11 +14,13 @@ vi.mock("node:fs", () => ({ default: { readdirSync: vi.fn(() => []), statSync: vi.fn(() => ({ isDirectory: () => false })), + readFileSync: vi.fn(() => "build-id"), writeFileSync: vi.fn(), existsSync: vi.fn(() => false), }, readdirSync: vi.fn(() => []), statSync: vi.fn(() => ({ isDirectory: () => false })), + readFileSync: vi.fn(() => "build-id"), writeFileSync: vi.fn(), existsSync: vi.fn(() => false), })); @@ -62,6 +64,141 @@ describe("buildOpenNextOutput", () => { expect(fs.writeFileSync).not.toHaveBeenCalled(); }); + test("uses adapter defaults for generated server metadata", async () => { + const output = await buildOpenNextOutput(createMockBuildOpts(), { + server: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + incrementalCache: "@opennextjs/aws/overrides/incrementalCache/s3.js", + tagCache: "@opennextjs/aws/overrides/tagCache/dynamodb.js", + queue: "@opennextjs/aws/overrides/queue/sqs.js", + }, + }); + + expect(output.origins.default).toMatchObject({ + streaming: true, + wrapper: "aws-lambda-streaming", + converter: "aws-streaming", + incrementalCache: "s3", + tagCache: "dynamodb", + queue: "sqs", + }); + }); + + test("detects a full-path streaming wrapper", async () => { + const opts = createMockBuildOpts(); + opts.config.default.override = { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + }; + + const output = await buildOpenNextOutput(opts); + expect(output.origins.default).toMatchObject({ + streaming: true, + wrapper: "aws-lambda-streaming", + converter: "aws-streaming", + }); + }); + + test("awaits lazy overrides when generating global edge metadata", async () => { + const opts = createMockBuildOpts(); + opts.config.functions = { + global: { + routes: ["app/page"], + runtime: "edge", + placement: "global", + override: { + wrapper: async () => { + await Promise.resolve(); + return { name: "delayed-wrapper" }; + }, + }, + }, + }; + + const output = await buildOpenNextOutput(opts, { + global: { converter: "@opennextjs/aws/overrides/converters/aws-cloudfront.js" }, + }); + expect(output.edgeFunctions.global).toMatchObject({ + wrapper: "delayed-wrapper", + converter: "aws-cloudfront", + }); + }); + + test("uses edge defaults for regional edge function metadata", async () => { + const opts = createMockBuildOpts(); + opts.config.functions = { + edge: { routes: ["app/page"], runtime: "edge" }, + }; + + const output = await buildOpenNextOutput(opts, { + edge: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda.js", + converter: "@opennextjs/aws/overrides/converters/aws-apigw-v2.js", + }, + server: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + }, + }); + + expect(output.origins.edge).toMatchObject({ + wrapper: "aws-lambda", + converter: "aws-apigw-v2", + streaming: false, + }); + }); + + test.each([ + ["edge", "aws-lambda", "aws-apigw-v2"], + ["node", "aws-lambda-streaming", "aws-streaming"], + ] as const)("falls back to %s defaults for global metadata", async (runtime, wrapper, converter) => { + const opts = createMockBuildOpts(); + opts.config.functions = { + global: { routes: ["app/page"], runtime, placement: "global" }, + }; + + const output = await buildOpenNextOutput(opts, { + edge: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda.js", + converter: "@opennextjs/aws/overrides/converters/aws-apigw-v2.js", + }, + server: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + }, + }); + + expect(output.edgeFunctions.global).toMatchObject({ wrapper, converter }); + }); + + test("uses Node defaults and adapter origin resolver for external middleware metadata", async () => { + const opts = createMockBuildOpts(); + opts.config.middleware = { external: true, runtime: "node" }; + + const output = await buildOpenNextOutput(opts, { + middleware: { originResolver: "custom-resolver" }, + }); + + expect(output.edgeFunctions.middleware).toMatchObject({ + wrapper: "node", + converter: "node", + pathResolver: "custom-resolver", + }); + }); + + test("uses edge defaults for generic external edge middleware metadata", async () => { + const opts = createMockBuildOpts(); + opts.config.middleware = { external: true }; + + const output = await buildOpenNextOutput(opts); + + expect(output.edgeFunctions.middleware).toMatchObject({ + wrapper: "dummy", + converter: "edge", + pathResolver: "pattern-env", + }); + }); + test("returns undefined revalidationFunction when disableIncrementalCache is true", async () => { const opts = createMockBuildOpts(); (opts.config as OpenNextConfig).dangerous = { disableIncrementalCache: true }; diff --git a/packages/core/src/build/generateOutput.ts b/packages/core/src/build/generateOutput.ts index e9915da6..67ae35e7 100644 --- a/packages/core/src/build/generateOutput.ts +++ b/packages/core/src/build/generateOutput.ts @@ -11,6 +11,9 @@ import type { OverrideOptions, } from "@/types/open-next"; +import { getDefaultConverterName, getDefaultWrapperName } from "../overrides/compatibility.js"; +import type { BundleDefaults, DefaultOverrides } from "../plugins/resolve.js"; + import { type BuildOptions, getBuildId } from "./helper.js"; type BaseFunction = { @@ -91,12 +94,12 @@ export interface OpenNextOutput { const indexHandler = "index.handler"; -async function canStream(opts: FunctionOptions) { +async function canStream(opts: FunctionOptions, defaults?: DefaultOverrides) { if (!opts.override?.wrapper) { - return false; + return bare(defaults?.wrapper ?? "aws-lambda") === "aws-lambda-streaming"; } if (typeof opts.override.wrapper === "string") { - return opts.override.wrapper === "aws-lambda-streaming"; + return bare(opts.override.wrapper) === "aws-lambda-streaming"; } const wrapper = await opts.override.wrapper(); return wrapper.supportStreaming; @@ -130,29 +133,34 @@ async function extractOverrideName( return overrideModule.name; } -async function extractOverrideFn(override?: DefaultOverrideOptions) { - if (!override) { - return { - wrapper: "aws-lambda", - converter: "aws-apigw-v2", - }; - } - const wrapper = await extractOverrideName("aws-lambda", override.wrapper); - const converter = await extractOverrideName("aws-apigw-v2", override.converter); +async function extractOverrideFn(override?: DefaultOverrideOptions, defaults?: DefaultOverrides) { + const configuredWrapper = typeof override?.wrapper === "string" ? bare(override.wrapper) : undefined; + const configuredConverter = typeof override?.converter === "string" ? bare(override.converter) : undefined; + const adapterWrapper = bare(defaults?.wrapper ?? "aws-lambda"); + const defaultWrapper = + (configuredConverter && getDefaultConverterName(adapterWrapper) === configuredConverter + ? adapterWrapper + : configuredConverter + ? getDefaultWrapperName(configuredConverter) + : undefined) ?? adapterWrapper; + const wrapper = await extractOverrideName(defaultWrapper, override?.wrapper); + const defaultConverter = bare( + (configuredWrapper ? getDefaultConverterName(configuredWrapper) : undefined) ?? + defaults?.converter ?? + getDefaultConverterName(wrapper) ?? + "aws-apigw-v2" + ); + const converter = await extractOverrideName(defaultConverter, override?.converter); return { wrapper, converter }; } -async function extractCommonOverride(override?: OverrideOptions) { - if (!override) { - return { - queue: "sqs", - incrementalCache: "s3", - tagCache: "dynamodb", - }; - } - const queue = await extractOverrideName("sqs", override.queue); - const incrementalCache = await extractOverrideName("s3", override.incrementalCache); - const tagCache = await extractOverrideName("dynamodb", override.tagCache); +async function extractCommonOverride(override?: OverrideOptions, defaults?: DefaultOverrides) { + const queue = await extractOverrideName(bare(defaults?.queue ?? "sqs"), override?.queue); + const incrementalCache = await extractOverrideName( + bare(defaults?.incrementalCache ?? "s3"), + override?.incrementalCache + ); + const tagCache = await extractOverrideName(bare(defaults?.tagCache ?? "dynamodb"), override?.tagCache); return { queue, incrementalCache, tagCache }; } @@ -163,31 +171,49 @@ function prefixPattern(basePath: string) { }; } -export async function buildOpenNextOutput(options: BuildOptions): Promise { +export async function buildOpenNextOutput( + options: BuildOptions, + defaultOverrides?: BundleDefaults +): Promise { const { appBuildOutputPath, config } = options; const edgeFunctions: OpenNextOutput["edgeFunctions"] = {}; const isExternalMiddleware = config.middleware?.external ?? false; if (isExternalMiddleware) { const middlewareConfig = options.config.middleware as ExternalMiddlewareConfig; + const isNodeMiddleware = middlewareConfig.runtime === "node"; + const middlewareDefaults: DefaultOverrides = { + wrapper: isNodeMiddleware ? "node" : "dummy", + converter: isNodeMiddleware ? "node" : "edge", + originResolver: "pattern-env", + ...defaultOverrides?.middleware, + }; edgeFunctions.middleware = { bundle: ".open-next/middleware", handler: "handler.handler", - pathResolver: await extractOverrideName("pattern-env", middlewareConfig.originResolver), - ...(await extractOverrideFn(middlewareConfig.override)), + pathResolver: await extractOverrideName( + bare(middlewareDefaults.originResolver ?? "pattern-env"), + middlewareConfig.originResolver + ), + ...(await extractOverrideFn(middlewareConfig.override, middlewareDefaults)), }; } // Add edge functions - Object.entries(config.functions ?? {}).forEach(async ([key, value]) => { - if (value.placement === "global") { - edgeFunctions[key] = { - bundle: `.open-next/server-functions/${key}`, - handler: indexHandler, - ...(await extractOverrideFn(value.override)), - }; - } - }); + await Promise.all( + Object.entries(config.functions ?? {}).map(async ([key, value]) => { + if (value.placement === "global") { + const functionDefaults = + defaultOverrides?.global ?? + (value.runtime === "edge" ? defaultOverrides?.edge : defaultOverrides?.server); + edgeFunctions[key] = { + bundle: `.open-next/server-functions/${key}`, + handler: indexHandler, + ...(await extractOverrideFn(value.override, functionDefaults)), + }; + } + }) + ); - const defaultOriginCanstream = await canStream(config.default); + const defaultOriginCanstream = await canStream(config.default, defaultOverrides?.server); const nextConfig = loadConfig(path.join(appBuildOutputPath, ".next")); const prefixer = prefixPattern(nextConfig.basePath ?? ""); @@ -221,24 +247,27 @@ export async function buildOpenNextOutput(options: BuildOptions): Promise { if (!value.placement || value.placement === "regional") { + const functionDefaults = value.runtime === "edge" ? defaultOverrides?.edge : defaultOverrides?.server; if (value.override?.generateDockerfile) { origins[key] = { type: "ecs", bundle: `.open-next/server-functions/${key}`, dockerfile: `.open-next/server-functions/${key}/Dockerfile`, - ...(await extractOverrideFn(value.override)), - ...(await extractCommonOverride(value.override)), + ...(await extractOverrideFn(value.override, functionDefaults)), + ...(await extractCommonOverride(value.override, functionDefaults)), }; } else { - const streaming = await canStream(value); + const streaming = await canStream(value, functionDefaults); origins[key] = { type: "function", handler: indexHandler, bundle: `.open-next/server-functions/${key}`, streaming, - ...(await extractOverrideFn(value.override)), - ...(await extractCommonOverride(value.override)), + ...(await extractOverrideFn(value.override, functionDefaults)), + ...(await extractCommonOverride(value.override, functionDefaults)), }; } } diff --git a/packages/core/src/build/validateConfig.spec.ts b/packages/core/src/build/validateConfig.spec.ts index 52bca9b1..414330bd 100644 --- a/packages/core/src/build/validateConfig.spec.ts +++ b/packages/core/src/build/validateConfig.spec.ts @@ -47,6 +47,40 @@ describe("validateConfig", () => { expect(validateConfig(config)).toEqual({ success: true }); }); + test("defaults the AWS streaming wrapper to the streaming converter", () => { + const config = { + default: { override: { wrapper: "aws-lambda-streaming" } }, + } as unknown as OpenNextConfig; + + expect(validateConfig(config)).toEqual({ success: true }); + }); + + test("rejects an explicit incompatible converter for the AWS streaming wrapper", () => { + const config = { + default: { + override: { wrapper: "aws-lambda-streaming", converter: "aws-apigw-v2" }, + }, + } as unknown as OpenNextConfig; + + const result = validateConfig(config); + expect(result.success).toBe(false); + expect(result.message).toMatch(/not compatible/); + }); + + test("pairs partial overrides before applying adapter defaults", () => { + const config = { + default: { override: { wrapper: "aws-lambda" } }, + } as unknown as OpenNextConfig; + expect( + validateConfig(config, { + server: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + }, + }) + ).toEqual({ success: true }); + }); + test.each([ ["C:\\overrides\\wrappers\\aws-lambda.cts", "C:\\overrides\\converters\\aws-apigw-v2.mts"], ["/overrides/wrappers/aws-lambda.js", "/overrides/converters/aws-apigw-v2.mjs"], @@ -71,6 +105,30 @@ describe("validateConfig", () => { expect(validateConfig(config)).toEqual({ success: true }); }); + test("preserves a compatible adapter wrapper for an ambiguous converter", () => { + const config = { + default: { override: { converter: "edge" } }, + } as unknown as OpenNextConfig; + + expect( + validateConfig(config, { + server: { + wrapper: "@opennextjs/core/overrides/wrappers/cloudflare-node.js", + converter: "@opennextjs/core/overrides/converters/edge.js", + }, + }) + ).toEqual({ success: true }); + }); + + test.each([undefined, "node"] as const)("uses compatible %s external middleware defaults", (runtime) => { + const config = { + default: {}, + middleware: { external: true, runtime }, + } as unknown as OpenNextConfig; + + expect(validateConfig(config)).toEqual({ success: true }); + }); + test("returns a descriptive issue for an unknown wrapper", () => { const config = { default: { override: { wrapper: "typo", converter: "aws-apigw-v2" } }, @@ -98,6 +156,26 @@ describe("validateConfig", () => { expect(result.message).toMatch(/Split function broken/); }); + test("returns a compatibility error instead of an earlier warning", () => { + const config = { + default: { override: { generateDockerfile: true } }, + functions: { + incompatible: { + routes: ["app/page"], + override: { + wrapper: "aws-lambda-streaming", + converter: "aws-apigw-v2", + }, + }, + }, + } as unknown as OpenNextConfig; + + const result = validateConfig(config); + expect(result.success).toBe(false); + expect(result.level).toBe("error"); + expect(result.message).toMatch(/not compatible/); + }); + test("returns shouldThrow:false for disabled incremental cache warning", () => { const config = { default: {}, diff --git a/packages/core/src/build/validateConfig.ts b/packages/core/src/build/validateConfig.ts index 25189e65..76b4541c 100644 --- a/packages/core/src/build/validateConfig.ts +++ b/packages/core/src/build/validateConfig.ts @@ -8,6 +8,9 @@ import type { SplittedFunctionOptions, } from "@/types/open-next"; +import { getDefaultConverterName, getDefaultWrapperName } from "../overrides/compatibility.js"; +import type { BundleDefaults, DefaultOverrides } from "../plugins/resolve.js"; + export type ValidateConfigResult = | { success: true } | { @@ -21,13 +24,13 @@ export type ValidateConfigResult = 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"], node: ["node"], "express-dev": ["node"], - dummy: ["dummy"], + dummy: ["dummy", "edge"], }; /** @@ -47,15 +50,31 @@ function normalizeOverrideName(value: string): string { * @param fnOptions Function options to validate. * @return The first compatibility issue, or success. */ -function validateFunctionOptions(fnOptions: FunctionOptions): ValidateConfigResult { - const wrapper = +function validateFunctionOptions( + fnOptions: FunctionOptions, + defaultOverrides?: DefaultOverrides +): ValidateConfigResult { + const configuredWrapper = typeof fnOptions.override?.wrapper === "string" ? normalizeOverrideName(fnOptions.override.wrapper) - : "aws-lambda"; - const converter = + : undefined; + const configuredConverter = typeof fnOptions.override?.converter === "string" ? normalizeOverrideName(fnOptions.override.converter) - : "aws-apigw-v2"; + : undefined; + const defaultWrapper = normalizeOverrideName(defaultOverrides?.wrapper ?? "aws-lambda"); + const wrapper = + configuredWrapper ?? + (configuredConverter && getDefaultConverterName(defaultWrapper) === configuredConverter + ? defaultWrapper + : configuredConverter + ? getDefaultWrapperName(configuredConverter) + : undefined) ?? + defaultWrapper; + const converter = + configuredConverter ?? + (configuredWrapper ? getDefaultConverterName(configuredWrapper) : undefined) ?? + normalizeOverrideName(defaultOverrides?.converter ?? getDefaultConverterName(wrapper) ?? "aws-apigw-v2"); if (fnOptions.override?.generateDockerfile && converter !== "node" && wrapper !== "node") { return { success: false, @@ -112,7 +131,8 @@ function validateFunctionOptions(fnOptions: FunctionOptions): ValidateConfigResu */ function validateSplittedFunctionOptions( fnOptions: SplittedFunctionOptions, - name: string + name: string, + defaultOverrides?: DefaultOverrides ): ValidateConfigResult { if (fnOptions.routes.length === 0) { return { @@ -138,7 +158,7 @@ function validateSplittedFunctionOptions( message: `Edge function ${name} can only have one route`, }; } - return validateFunctionOptions(fnOptions); + return validateFunctionOptions(fnOptions, defaultOverrides); } /** @@ -147,12 +167,27 @@ function validateSplittedFunctionOptions( * Fatal structural issues take precedence over compatibility warnings so warnings cannot hide an invalid build. * * @param config OpenNext configuration to validate. + * @param defaultOverrides Adapter-provided defaults used by each bundle type. * @return A fatal issue, the first nonfatal issue, or success. */ -export function validateConfig(config: OpenNextConfig): ValidateConfigResult { - const results: ValidateConfigResult[] = [validateFunctionOptions(config.default)]; +export function validateConfig( + config: OpenNextConfig, + defaultOverrides?: BundleDefaults +): ValidateConfigResult { + const results: ValidateConfigResult[] = [validateFunctionOptions(config.default, defaultOverrides?.server)]; for (const [name, fnOptions] of Object.entries(config.functions ?? {})) { - results.push(validateSplittedFunctionOptions(fnOptions, name)); + results.push( + validateSplittedFunctionOptions( + fnOptions, + name, + fnOptions.placement === "global" + ? (defaultOverrides?.global ?? + (fnOptions.runtime === "edge" ? defaultOverrides?.edge : defaultOverrides?.server)) + : fnOptions.runtime === "edge" + ? defaultOverrides?.edge + : defaultOverrides?.server + ) + ); } if (config.dangerous?.disableIncrementalCache) { results.push({ @@ -172,17 +207,25 @@ export function validateConfig(config: OpenNextConfig): ValidateConfigResult { It is safe to disable if you only use page router`, }); } - results.push(validateFunctionOptions(config.imageOptimization ?? {})); + results.push(validateFunctionOptions(config.imageOptimization ?? {}, defaultOverrides?.imageOptimization)); if (config.middleware?.external === true) { - results.push(validateFunctionOptions(config.middleware ?? {})); + const isNodeMiddleware = config.middleware.runtime === "node"; + results.push( + validateFunctionOptions(config.middleware, { + wrapper: isNodeMiddleware ? "node" : "dummy", + converter: isNodeMiddleware ? "node" : "edge", + ...defaultOverrides?.middleware, + }) + ); } //@ts-expect-error - Revalidate custom wrapper type is different - results.push(validateFunctionOptions(config.revalidate ?? {})); + results.push(validateFunctionOptions(config.revalidate ?? {}, defaultOverrides?.revalidation)); //@ts-expect-error - Warmer custom wrapper type is different - results.push(validateFunctionOptions(config.warmer ?? {})); - results.push(validateFunctionOptions(config.initializationFunction ?? {})); + results.push(validateFunctionOptions(config.warmer ?? {}, defaultOverrides?.warmer)); + results.push(validateFunctionOptions(config.initializationFunction ?? {}, defaultOverrides?.server)); return ( results.find((result) => !result.success && result.shouldThrow) ?? + results.find((result) => !result.success && result.level === "error") ?? results.find((result) => !result.success) ?? { success: true } ); } diff --git a/packages/core/src/core/requestHandler.ts b/packages/core/src/core/requestHandler.ts index aa07f1a5..6de89244 100644 --- a/packages/core/src/core/requestHandler.ts +++ b/packages/core/src/core/requestHandler.ts @@ -121,6 +121,7 @@ export async function openNextHandler( if ("type" in routingResult) { // response is used only in the streaming case if (options?.streamCreator) { + const streamCreator = options.streamCreator; const response = createServerResponse( { internalEvent, @@ -131,16 +132,26 @@ export async function openNextHandler( initialURL: internalEvent.url, }, routingResult.headers, - options.streamCreator + { + ...streamCreator, + writeHeaders: (prelude) => + streamCreator.writeHeaders({ + ...prelude, + isBase64Encoded: routingResult.isBase64Encoded, + }), + } ); response.statusCode = routingResult.statusCode; response.flushHeaders(); - const [bodyToConsume, bodyToReturn] = routingResult.body.tee(); - for await (const chunk of bodyToConsume) { - response.write(chunk); + if (routingResult.body) { + for await (const chunk of routingResult.body) { + if (!response.write(chunk)) { + await waitForDrain(response); + } + } + routingResult.body = undefined; } response.end(); - routingResult.body = bodyToReturn; } return routingResult; } @@ -181,8 +192,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 +231,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, }; @@ -233,6 +245,38 @@ export async function openNextHandler( ); } +/** + * Waits for response backpressure while detecting a closed destination. + * + * @param response - The response currently forwarding a streamed body. + * @returns A promise that resolves when writing can resume. + * @throws When the destination closes or errors before draining. + */ +function waitForDrain(response: Writable): Promise { + return new Promise((resolve, reject) => { + const cleanup = () => { + response.off("drain", onDrain); + response.off("close", onClose); + response.off("error", onError); + }; + const onDrain = () => { + cleanup(); + resolve(); + }; + const onClose = () => { + cleanup(); + reject(new Error("Response closed while waiting for backpressure")); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + response.once("drain", onDrain); + response.once("close", onClose); + response.once("error", onError); + }); +} + function getHeaders(routingResult: RoutingResult | InternalResult) { if ("type" in routingResult) { return routingResult.headers; diff --git a/packages/core/src/core/routing/util.ts b/packages/core/src/core/routing/util.ts index 13e4f034..1481fc23 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..18a2409d 100644 --- a/packages/core/src/http/openNextResponse.ts +++ b/packages/core/src/http/openNextResponse.ts @@ -4,6 +4,7 @@ import { Transform } from "node:stream"; import type { TransformCallback, Writable } from "node:stream"; import type { StreamCreator } from "@/types/open-next"; +import { isBinaryContentType } from "@/utils/binary"; import { debug } from "../adapters/logger"; @@ -18,7 +19,6 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { statusMessage = ""; headers: OutgoingHttpHeaders = {}; headersSent = false; - _chunks: Buffer[] = []; headersAlreadyFixed = false; private _cookies: string[] = []; @@ -179,6 +179,8 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse { statusCode: this.statusCode ?? 200, cookies: this._cookies, headers: parsedHeaders, + isBase64Encoded: + isBinaryContentType(parsedHeaders["content-type"]) || !!parsedHeaders["content-encoding"], }); this.pipe(this.responseStream); } @@ -255,10 +257,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 +264,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 +336,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/compatibility.ts b/packages/core/src/overrides/compatibility.ts new file mode 100644 index 00000000..404a066f --- /dev/null +++ b/packages/core/src/overrides/compatibility.ts @@ -0,0 +1,48 @@ +/** + * Returns the conventional converter for a built-in wrapper. + * + * @param wrapper - The normalized built-in wrapper name. + * @returns The paired converter name, if the wrapper is known. + */ +export function getDefaultConverterName(wrapper: string): string | undefined { + switch (wrapper) { + case "aws-lambda": + case "aws-lambda-compressed": + return "aws-apigw-v2"; + case "aws-lambda-streaming": + return "aws-streaming"; + case "cloudflare": + case "cloudflare-edge": + case "cloudflare-node": + return "edge"; + case "node": + case "express-dev": + return "node"; + case "dummy": + return "dummy"; + } +} + +/** + * Returns the conventional wrapper for a built-in converter. + * + * @param converter - The normalized built-in converter name. + * @returns The paired wrapper name, if the converter is known. + */ +export function getDefaultWrapperName(converter: string): string | undefined { + switch (converter) { + case "aws-apigw-v1": + case "aws-apigw-v2": + case "aws-cloudfront": + case "sqs-revalidate": + return "aws-lambda"; + case "aws-streaming": + return "aws-lambda-streaming"; + case "edge": + return "cloudflare-edge"; + case "node": + return "node"; + case "dummy": + return "dummy"; + } +} 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 2f4cd2ae..2d0031a7 100644 --- a/packages/core/src/overrides/converters/edge.ts +++ b/packages/core/src/overrides/converters/edge.ts @@ -1,9 +1,10 @@ -import type { ReadableStream } from "node:stream/web"; +import { Writable } from "node:stream"; +import type { ReadableStream as NodeReadableStream } from "node:stream/web"; 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"; @@ -32,7 +33,9 @@ const converter: Converter = { const shouldHaveBody = method !== "GET" && method !== "HEAD"; // Only read body for methods that should have one - const body = shouldHaveBody ? ((request.body as ReadableStream | undefined) ?? undefined) : undefined; + const body = shouldHaveBody + ? ((request.body as unknown as NodeReadableStream | undefined) ?? undefined) + : undefined; const cookieHeader = request.headers.get("cookie"); const cookies = cookieHeader ? (cookieParser.parse(cookieHeader) as Record) : {}; @@ -49,73 +52,163 @@ 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; + // Not every handler streams its response: the external middleware handler returns the + // result directly. We track whether the stream was used to know which one to return. + let isStreamed = false; + let abortResponseBody: ((reason: unknown) => Promise) | undefined; + const streamCreator: StreamCreator = { + writeHeaders(prelude) { + isStreamed = true; + 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); - } - } 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 unknown as globalThis.ReadableStream); + if (NULL_BODY_STATUSES.has(prelude.statusCode)) { + resolveOutput(new Response(null, { status: prelude.statusCode, headers: responseHeaders })); + return new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); + } + const { readable, writable } = new TransformStream(); + const writer = writable.getWriter(); + let writerClosed = false; + abortResponseBody = async (reason) => { + if (writerClosed) return; + writerClosed = true; + await writer.abort(reason); + }; + resolveOutput(new Response(readable, { status: prelude.statusCode, headers: responseHeaders })); + + return new Writable({ + write(chunk, _encoding, callback) { + writer.write(chunk).then( + () => callback(), + (error: unknown) => callback(error instanceof Error ? error : new Error(String(error))) + ); + }, + final(callback) { + writerClosed = true; + writer.close().then( + () => callback(), + (error: unknown) => callback(error instanceof Error ? error : new Error(String(error))) + ); + }, + destroy(error, callback) { + if (writerClosed) { + callback(error); + return; + } + writerClosed = true; + const close = error ? writer.abort(error) : writer.close(); + close.then( + () => callback(error), + (closeError: unknown) => + callback(closeError instanceof Error ? closeError : new Error(String(closeError))) + ); + }, + }); + }, + abortSignal, + abort: async (reason) => abortResponseBody?.(reason), + }; - return new Response(body, { - status: result.statusCode, - headers, - }); + return { + type: "stream", + streamCreator, + output, + data: async (result) => { + if ("internalEvent" in result) { + return convertMiddlewareResult(result); + } + // When the handler streamed the response, `output` already holds it. + return isStreamed ? undefined : convertInternalResult(result); + }, + }; }, name: "edge", }; +function convertInternalResult(result: InternalResult): Response { + 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 + for (const cookie of parseSetCookieHeader(value)) { + headers.append(key, cookie); + } + continue; + } + if (Array.isArray(value)) { + for (const v of value) { + headers.append(key, v); + } + } 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 ?? null) as unknown as globalThis.ReadableStream | null); + + return new Response(body, { + status: result.statusCode, + headers, + }); +} + +async function convertMiddlewareResult( + result: MiddlewareResult +): Promise { + const request = new Request(result.internalEvent.url, { + body: result.internalEvent.body as unknown 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 efe5ffb4..93979994 100644 --- a/packages/core/src/overrides/converters/node.ts +++ b/packages/core/src/overrides/converters/node.ts @@ -1,10 +1,10 @@ -import type { IncomingMessage } from "node:http"; +import type { IncomingMessage, ServerResponse } from "node:http"; import { Readable } from "node:stream"; import type { ReadableStream } from "node:stream/web"; 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"; @@ -41,12 +41,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..2fcb764b 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, { abortSignal: request.signal }); // Retrieve geo information from the cloudflare request // See https://developers.cloudflare.com/workers/runtime-apis/request @@ -43,13 +44,34 @@ const handler: WrapperHandler } } - const response = await handler(internalEvent, { + if (output.type === "direct") { + return output.data(await handler(internalEvent, { waitUntil: ctx.waitUntil.bind(ctx) })); + } + + const handlerPromise = handler(internalEvent, { + streamCreator: output.streamCreator, waitUntil: ctx.waitUntil.bind(ctx), + }).catch(async (error: unknown) => { + await output.streamCreator.abort?.(error); + throw error; }); + if (!output.output) { + const response = await handlerPromise; + return output.data?.(response); + } - const result = (await converter.convertTo(response)) as Response; - - return result; + const result = await Promise.race([ + output.output.then((value) => ({ type: "output" as const, value })), + handlerPromise.then(async (response) => ({ + type: "handler" as const, + value: await output.data?.(response), + })), + ]); + if (result.type === "output") { + ctx.waitUntil(handlerPromise.then(() => undefined)); + return result.value; + } + return result.value !== undefined ? result.value : 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..1772c4c3 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,35 @@ 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: abortSignal ?? request.signal }); + if (output.type === "direct") { + return output.data(await handler(internalEvent, { waitUntil: ctx.waitUntil.bind(ctx) })); + } - ctx.waitUntil( - handler(internalEvent, { - streamCreator, - waitUntil: ctx.waitUntil.bind(ctx), - }) - ); + const handlerPromise = handler(internalEvent, { + streamCreator: output.streamCreator, + waitUntil: ctx.waitUntil.bind(ctx), + }).catch(async (error: unknown) => { + await output.streamCreator.abort?.(error); + throw error; + }); + if (!output.output) { + const response = await handlerPromise; + return output.data?.(response); + } - return promiseResponse; + const result = await Promise.race([ + output.output.then((value) => ({ type: "output" as const, value })), + handlerPromise.then(async (response) => ({ + type: "handler" as const, + value: await output.data?.(response), + })), + ]); + if (result.type === "output") { + ctx.waitUntil(handlerPromise.then(() => undefined)); + return result.value; + } + return result.value !== undefined ? result.value : output.output; }; export default { diff --git a/packages/core/src/overrides/wrappers/dummy.ts b/packages/core/src/overrides/wrappers/dummy.ts index 8649a8c8..70065153 100644 --- a/packages/core/src/overrides/wrappers/dummy.ts +++ b/packages/core/src/overrides/wrappers/dummy.ts @@ -2,10 +2,16 @@ 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)); + } + const response = await handler(event, { ...options, streamCreator: output.streamCreator }); + const directResult = await output.data?.(response); + return directResult !== undefined ? directResult : output.output; }; export default { diff --git a/packages/core/src/overrides/wrappers/express-dev.ts b/packages/core/src/overrides/wrappers/express-dev.ts index bcc9e8ef..162ec017 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,14 @@ 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; + } + const response = await imageHandler(internalEvent, { streamCreator: output.streamCreator }); + const directResult = await output.data?.(response); + if (directResult === undefined) await output.output; }); app.all(/.*/, async (req, res) => { @@ -39,25 +39,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.spec.ts b/packages/core/src/plugins/resolve.spec.ts index 749e6e34..d9f750fc 100644 --- a/packages/core/src/plugins/resolve.spec.ts +++ b/packages/core/src/plugins/resolve.spec.ts @@ -83,6 +83,7 @@ const OVERRIDE_MODULES = [ const CORE_PKG_MODULES = [ "overrides/converters/edge.js", "overrides/converters/dummy.js", + "overrides/wrappers/cloudflare-node.js", "overrides/imageLoader/dummy.js", "overrides/originResolver/dummy.js", "overrides/proxyExternalRequest/fetch.js", @@ -91,6 +92,7 @@ const AWS_PKG_MODULES = [ "overrides/wrappers/aws-lambda.js", "overrides/wrappers/aws-lambda-streaming.js", "overrides/converters/aws-apigw-v2.js", + "overrides/converters/aws-streaming.js", "overrides/tagCache/dynamodb.js", "overrides/queue/sqs.js", "overrides/incrementalCache/s3.js", @@ -252,7 +254,7 @@ describe("openNextResolvePlugin", () => { overrides: {}, defaultOverrides: { wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", - converter: "@opennextjs/aws/overrides/converters/aws-apigw-v2.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", incrementalCache: "@opennextjs/aws/overrides/incrementalCache/s3.js", tagCache: "@opennextjs/aws/overrides/tagCache/dynamodb.js", queue: "@opennextjs/aws/overrides/queue/sqs.js", @@ -260,7 +262,7 @@ describe("openNextResolvePlugin", () => { fnName: "server", }); expect(contents).toContain(pkg("@opennextjs/aws", "wrappers/aws-lambda-streaming.js")); - expect(contents).toContain(pkg("@opennextjs/aws", "converters/aws-apigw-v2.js")); + expect(contents).toContain(pkg("@opennextjs/aws", "converters/aws-streaming.js")); expect(contents).toContain(pkg("@opennextjs/aws", "incrementalCache/s3.js")); expect(contents).toContain(pkg("@opennextjs/aws", "tagCache/dynamodb.js")); expect(contents).toContain(pkg("@opennextjs/aws", "queue/sqs.js")); @@ -268,6 +270,32 @@ describe("openNextResolvePlugin", () => { expect(contents).toContain(local("imageLoader/fs-dev.js")); }); + test("G2 - a partial AWS wrapper override selects its compatible converter", async () => { + const contents = await bundleWithPlugin({ + overrides: { wrapper: "aws-lambda" }, + defaultOverrides: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + }, + fnName: "server", + }); + expect(contents).toContain(pkg("@opennextjs/aws", "wrappers/aws-lambda.js")); + expect(contents).toContain(pkg("@opennextjs/aws", "converters/aws-apigw-v2.js")); + }); + + test("G3 - a partial AWS converter override selects its compatible wrapper", async () => { + const contents = await bundleWithPlugin({ + overrides: { converter: "aws-apigw-v2" }, + defaultOverrides: { + wrapper: "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js", + converter: "@opennextjs/aws/overrides/converters/aws-streaming.js", + }, + fnName: "server", + }); + expect(contents).toContain(pkg("@opennextjs/aws", "wrappers/aws-lambda.js")); + expect(contents).toContain(pkg("@opennextjs/aws", "converters/aws-apigw-v2.js")); + }); + test("H - bare-name user override becomes legacy relative core path", async () => { const contents = await bundleWithPlugin({ overrides: { converter: "edge" }, @@ -278,6 +306,19 @@ describe("openNextResolvePlugin", () => { expect(contents).not.toContain(local("converters/node.js")); }); + test("H2 - an ambiguous converter preserves a compatible adapter wrapper", async () => { + const contents = await bundleWithPlugin({ + overrides: { converter: "edge" }, + defaultOverrides: { + wrapper: "@opennextjs/core/overrides/wrappers/cloudflare-node.js", + converter: "@opennextjs/core/overrides/converters/edge.js", + }, + fnName: "middleware", + }); + expect(contents).toContain(pkg("@opennextjs/core", "wrappers/cloudflare-node.js")); + expect(contents).not.toContain(local("wrappers/cloudflare-edge.js")); + }); + test("I - resolvable package specifier is resolved through node_modules", async () => { await mkdir(join(root, "node_modules", "@test-pkg", "wrapper"), { recursive: true }); await writeFile( diff --git a/packages/core/src/plugins/resolve.ts b/packages/core/src/plugins/resolve.ts index 65d0a3d0..894a7055 100644 --- a/packages/core/src/plugins/resolve.ts +++ b/packages/core/src/plugins/resolve.ts @@ -16,6 +16,7 @@ import type { import type { ImageLoader, OriginResolver, Warmer } from "@/types/overrides"; import logger from "../logger.js"; +import { getDefaultConverterName, getDefaultWrapperName } from "../overrides/compatibility.js"; import { getCrossPlatformPathRegex } from "../utils/regex.js"; export interface IPluginSettings { @@ -69,6 +70,7 @@ export type BundleType = | "server" | "middleware" | "edge" + | "global" | "imageOptimization" | "revalidation" | "warmer" @@ -83,6 +85,32 @@ function isFullPath(s: string): boolean { return s.startsWith("@") || s.includes("/"); } +/** Returns the filename portion of an override name or package path. */ +function getOverrideName(value: string): string { + return value.slice(value.lastIndexOf("/") + 1).replace(/\.[^.]+$/, ""); +} + +/** Returns whether an override uses an AWS built-in name or package path. */ +function isAwsOverride(value: string): boolean { + return !isFullPath(value) || value.includes("@opennextjs/aws/overrides/"); +} + +/** Resolves a paired AWS override in the adapter package and core overrides by bare name. */ +function resolvePairedOverride( + name: string, + configuredPeer: string, + defaultPeer: string | undefined, + targetFolder: "converters" | "wrappers" +): string { + if (!name.startsWith("aws-") && name !== "sqs-revalidate") return name; + const peer = isFullPath(configuredPeer) ? configuredPeer : defaultPeer; + if (!peer || !isFullPath(peer)) return name; + return peer.replace( + /\/overrides\/(?:converters|wrappers)\/[^/]+$/, + `/overrides/${targetFolder}/${name}.js` + ); +} + /** * Turns a package specifier into a path relative to `core/resolve.js`, so that esbuild * resolves the override itself - it is the only way for it to pick up the module type @@ -118,14 +146,70 @@ export function openNextResolvePlugin({ name: "opennext-resolve", setup(build) { logger.debug(chalk.blue("OpenNext Resolve plugin"), fnName ? `for ${fnName}` : ""); + const effectiveOverrides = { ...overrides }; + if ( + typeof effectiveOverrides.wrapper === "string" && + isAwsOverride(effectiveOverrides.wrapper) && + getOverrideName(effectiveOverrides.wrapper).startsWith("aws-") + ) { + effectiveOverrides.wrapper = resolvePairedOverride( + getOverrideName(effectiveOverrides.wrapper), + effectiveOverrides.wrapper, + defaultValues?.wrapper, + "wrappers" + ) as NonNullable; + } + if ( + typeof effectiveOverrides.converter === "string" && + isAwsOverride(effectiveOverrides.converter) && + (getOverrideName(effectiveOverrides.converter).startsWith("aws-") || + getOverrideName(effectiveOverrides.converter) === "sqs-revalidate") + ) { + effectiveOverrides.converter = resolvePairedOverride( + getOverrideName(effectiveOverrides.converter), + effectiveOverrides.converter, + defaultValues?.converter, + "converters" + ) as NonNullable; + } + if (typeof effectiveOverrides.wrapper === "string" && !effectiveOverrides.converter) { + const converter = getDefaultConverterName(getOverrideName(effectiveOverrides.wrapper)); + if (converter) { + effectiveOverrides.converter = resolvePairedOverride( + converter, + effectiveOverrides.wrapper, + defaultValues?.wrapper, + "converters" + ) as NonNullable; + } + } else if (typeof effectiveOverrides.converter === "string" && !effectiveOverrides.wrapper) { + const converterName = getOverrideName(effectiveOverrides.converter); + const defaultWrapper = defaultValues?.wrapper; + if ( + typeof defaultWrapper === "string" && + getDefaultConverterName(getOverrideName(defaultWrapper)) === converterName + ) { + effectiveOverrides.wrapper = defaultWrapper as NonNullable; + } else { + const wrapper = getDefaultWrapperName(converterName); + if (wrapper) { + effectiveOverrides.wrapper = resolvePairedOverride( + wrapper, + effectiveOverrides.converter, + defaultValues?.converter, + "wrappers" + ) as NonNullable; + } + } + } // Maps the overrides folder to the specifier that should be imported instead // of the default one. Computed once, the config cannot change during a build. const redirects = new Map(); - const allKeys = new Set([...Object.keys(overrides ?? {}), ...Object.keys(defaultValues ?? {})]); + const allKeys = new Set([...Object.keys(effectiveOverrides), ...Object.keys(defaultValues ?? {})]); for (const overrideName of allKeys) { - const configValue = overrides?.[overrideName as keyof typeof overrides]; + const configValue = effectiveOverrides[overrideName as keyof typeof effectiveOverrides]; const defaultValue = defaultValues?.[overrideName as keyof typeof defaultValues]; let overrideValue = configValue ?? defaultValue; if (!overrideValue) { diff --git a/packages/core/src/types/open-next.ts b/packages/core/src/types/open-next.ts index 207d500b..3815e15a 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">; @@ -67,23 +67,18 @@ export type PartialResult = { }; export interface StreamCreator { - writeHeaders(prelude: { statusCode: number; cookies: string[]; headers: Record }): Writable; + writeHeaders(prelude: { + statusCode: number; + cookies: string[]; + headers: Record; + isBase64Encoded?: boolean; + }): Writable; // Just to fix an issue with aws lambda streaming with empty body 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; + /** Aborts an active platform response body after handler failure. */ + abort?: (reason: unknown) => void | Promise; } export type WaitUntil = (promise: Promise) => void; @@ -146,6 +141,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 0614f725..5ade0904 100644 --- a/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts +++ b/packages/tests-unit/tests/converters/aws-apigw-v1.test.ts @@ -5,10 +5,32 @@ import { fromReadableStream } from "@opennextjs/core/utils/stream.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", @@ -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"], @@ -40,7 +62,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 d121f16b..878c489a 100644 --- a/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts +++ b/packages/tests-unit/tests/converters/aws-apigw-v2.test.ts @@ -7,9 +7,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", @@ -26,7 +48,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"], @@ -42,7 +64,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 a1b76908..1e1af506 100644 --- a/packages/tests-unit/tests/converters/aws-cloudfront.test.ts +++ b/packages/tests-unit/tests/converters/aws-cloudfront.test.ts @@ -7,9 +7,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", @@ -37,7 +59,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"], @@ -62,7 +84,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": @@ -88,7 +110,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": @@ -115,7 +137,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..d7e22474 --- /dev/null +++ b/packages/tests-unit/tests/converters/aws-streaming.test.ts @@ -0,0 +1,127 @@ +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, + contentEncoding: "identity", + } + ); + 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 output.output; + + 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": "identity" }, + }); + expect(outputBody.subarray(separator + 8).toString()).toBe("hello"); + }); + + it("does not recompress an already encoded response", 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, contentEncoding: "gzip" }); + if (output.type !== "stream") throw new Error("Expected a streaming converter output"); + + const writable = output.streamCreator.writeHeaders({ + statusCode: 200, + cookies: [], + headers: { "content-encoding": "br", "content-length": "7" }, + }); + writable.end("encoded"); + await output.output; + + const outputBody = Buffer.concat(chunks); + const separator = outputBody.indexOf(Buffer.alloc(8)); + expect(JSON.parse(outputBody.subarray(0, separator).toString()).headers).toEqual({ + "content-encoding": "br", + "content-length": "7", + }); + expect(outputBody.subarray(separator + 8).toString()).toBe("encoded"); + }); + + it("compresses negotiated responses and removes their original content length", 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, contentEncoding: "gzip" }); + if (output.type !== "stream") throw new Error("Expected a streaming converter output"); + + const writable = output.streamCreator.writeHeaders({ + statusCode: 200, + cookies: [], + headers: { "content-length": "5" }, + }); + writable.end("hello"); + await output.output; + + const outputBody = Buffer.concat(chunks); + const separator = outputBody.indexOf(Buffer.alloc(8)); + expect(JSON.parse(outputBody.subarray(0, separator).toString()).headers).toEqual({ + "content-encoding": "gzip", + vary: "Accept-Encoding", + }); + expect((await import("node:zlib")).gunzipSync(outputBody.subarray(separator + 8)).toString()).toBe( + "hello" + ); + }); + + it("does not create a compressed body for a bodyless status", 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, contentEncoding: "gzip" }); + if (output.type !== "stream") throw new Error("Expected a streaming converter output"); + + output.streamCreator.writeHeaders({ statusCode: 204, cookies: [], headers: {} }).end(); + await output.output; + + const outputBody = Buffer.concat(chunks); + const separator = outputBody.indexOf(Buffer.alloc(8)); + expect(JSON.parse(outputBody.subarray(0, separator).toString()).headers).toEqual({ + "content-encoding": "identity", + }); + expect(outputBody.subarray(separator + 8)).toHaveLength(0); + }); +}); diff --git a/packages/tests-unit/tests/converters/response-stream.test.ts b/packages/tests-unit/tests/converters/response-stream.test.ts new file mode 100644 index 00000000..3e576466 --- /dev/null +++ b/packages/tests-unit/tests/converters/response-stream.test.ts @@ -0,0 +1,43 @@ +import { createBufferedStreamCreator } from "@opennextjs/aws/overrides/converters/response-stream.js"; +import { describe, expect, it } from "vitest"; + +describe("createBufferedStreamCreator", () => { + it("rejects its output when the response stream is destroyed", async () => { + const { streamCreator, output } = createBufferedStreamCreator((_prelude, body) => body); + const stream = streamCreator.writeHeaders({ statusCode: 200, cookies: [], headers: {} }); + const streamError = new Error("stream failed"); + + stream.on("error", () => undefined); + stream.destroy(streamError); + + await expect(output).rejects.toBe(streamError); + }); + + it("rejects its output when the response stream closes prematurely", async () => { + const { streamCreator, output } = createBufferedStreamCreator((_prelude, body) => body); + const stream = streamCreator.writeHeaders({ statusCode: 200, cookies: [], headers: {} }); + + stream.destroy(); + + await expect(output).rejects.toThrow("Response stream was destroyed before it finished"); + }); + + it("uses explicit binary response metadata", async () => { + const { streamCreator, output } = createBufferedStreamCreator((_prelude, body, isBase64Encoded) => ({ + body, + isBase64Encoded, + })); + const stream = streamCreator.writeHeaders({ + statusCode: 200, + cookies: [], + headers: { "content-type": "application/x-custom" }, + isBase64Encoded: true, + }); + stream.end(Buffer.from([0xff, 0x00])); + + await expect(output).resolves.toEqual({ + body: Buffer.from([0xff, 0x00]), + isBase64Encoded: true, + }); + }); +}); 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" }], + }); + }); +}); diff --git a/packages/tests-unit/tests/core/routing/util.test.ts b/packages/tests-unit/tests/core/routing/util.test.ts index 9aea1114..a7420499 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(); }); }); diff --git a/packages/tests-unit/tests/wrappers/aws-lambda-compressed.test.ts b/packages/tests-unit/tests/wrappers/aws-lambda-compressed.test.ts new file mode 100644 index 00000000..9071aa9b --- /dev/null +++ b/packages/tests-unit/tests/wrappers/aws-lambda-compressed.test.ts @@ -0,0 +1,63 @@ +import { Readable } from "node:stream"; +import { gunzipSync } from "node:zlib"; + +import converter from "@opennextjs/aws/overrides/converters/aws-apigw-v2.js"; +import wrapper from "@opennextjs/aws/overrides/wrappers/aws-lambda-compressed.js"; +import type { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda"; +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/config/index.js", () => ({})); + +describe("aws-lambda-compressed wrapper", () => { + it("returns compressed bodies as base64 and removes the original content length", async () => { + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: { "accept-encoding": "gzip" }, + requestContext: { http: { method: "GET", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + const wrapped = await wrapper.wrapper(async () => { + return { + type: "core", + statusCode: 200, + headers: { "content-type": "text/plain", "content-length": "5" }, + body: Readable.toWeb(Readable.from("hello")), + isBase64Encoded: false, + }; + }, converter); + + const response = (await wrapped(event)) as APIGatewayProxyResultV2; + + expect(response.isBase64Encoded).toBe(true); + expect(response.headers).toEqual({ + "content-type": "text/plain", + "content-encoding": "gzip", + vary: "Accept-Encoding", + }); + expect(gunzipSync(Buffer.from(response.body ?? "", "base64")).toString()).toBe("hello"); + }); + + it("finalizes a bodyless response without adding compression", async () => { + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: { "accept-encoding": "gzip" }, + requestContext: { http: { method: "HEAD", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + const wrapped = await wrapper.wrapper(async () => { + return { type: "core", statusCode: 204, headers: {}, isBase64Encoded: false }; + }, converter); + + await expect(wrapped(event)).resolves.toMatchObject({ + statusCode: 204, + body: "", + isBase64Encoded: false, + }); + }); +}); diff --git a/packages/tests-unit/tests/wrappers/aws-lambda-streaming.test.ts b/packages/tests-unit/tests/wrappers/aws-lambda-streaming.test.ts new file mode 100644 index 00000000..4493b509 --- /dev/null +++ b/packages/tests-unit/tests/wrappers/aws-lambda-streaming.test.ts @@ -0,0 +1,99 @@ +import { Writable } from "node:stream"; + +import converter from "@opennextjs/aws/overrides/converters/aws-streaming.js"; +import wrapper from "@opennextjs/aws/overrides/wrappers/aws-lambda-streaming.js"; +import type { APIGatewayProxyEventV2 } from "aws-lambda"; +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/config/index.js", () => ({})); + +describe("aws-lambda-streaming wrapper", () => { + it("waits for the Lambda response stream to finish", async () => { + let responseFinished = false; + const responseStream = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + final(callback) { + setTimeout(() => { + responseFinished = true; + callback(); + }, 10); + }, + }) as Writable & { setContentType: (contentType: string) => void }; + responseStream.setContentType = vi.fn(); + const previousAwsLambda = globalThis.awslambda; + globalThis.awslambda = { + streamifyResponse: (streamingHandler) => streamingHandler, + HttpResponseStream: previousAwsLambda?.HttpResponseStream, + }; + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: {}, + requestContext: { http: { method: "GET", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + + try { + const wrapped = await wrapper.wrapper(async (_internalEvent, options) => { + const stream = options?.streamCreator?.writeHeaders({ + statusCode: 200, + cookies: [], + headers: { "content-type": "text/plain" }, + }); + stream?.end("hello"); + return { type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }; + }, converter); + + await wrapped(event, responseStream, { callbackWaitsForEmptyEventLoop: true }); + expect(responseFinished).toBe(true); + } finally { + globalThis.awslambda = previousAwsLambda; + } + }); + + it("finalizes a bodyless response returned by an edge handler", async () => { + let responseFinished = false; + const responseStream = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + final(callback) { + responseFinished = true; + callback(); + }, + }) as Writable & { setContentType: (contentType: string) => void }; + responseStream.setContentType = vi.fn(); + const previousAwsLambda = globalThis.awslambda; + const previousEdgeRuntime = globalThis.isEdgeRuntime; + globalThis.awslambda = { + streamifyResponse: (streamingHandler) => streamingHandler, + HttpResponseStream: previousAwsLambda?.HttpResponseStream, + }; + globalThis.isEdgeRuntime = true; + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: {}, + requestContext: { http: { method: "HEAD", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + + try { + const wrapped = await wrapper.wrapper(async () => { + return { type: "core", statusCode: 204, headers: {}, isBase64Encoded: false }; + }, converter); + + await wrapped(event, responseStream, { callbackWaitsForEmptyEventLoop: true }); + expect(responseFinished).toBe(true); + } finally { + globalThis.awslambda = previousAwsLambda; + globalThis.isEdgeRuntime = previousEdgeRuntime; + } + }); +}); diff --git a/packages/tests-unit/tests/wrappers/aws-lambda.test.ts b/packages/tests-unit/tests/wrappers/aws-lambda.test.ts new file mode 100644 index 00000000..5afe0ece --- /dev/null +++ b/packages/tests-unit/tests/wrappers/aws-lambda.test.ts @@ -0,0 +1,62 @@ +import { Readable } from "node:stream"; + +import converter from "@opennextjs/aws/overrides/converters/aws-apigw-v2.js"; +import wrapper from "@opennextjs/aws/overrides/wrappers/aws-lambda.js"; +import type { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda"; +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/config/index.js", () => ({})); + +describe("aws-lambda wrapper", () => { + it("streams a body returned by an edge handler into a buffered converter", async () => { + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: {}, + requestContext: { http: { method: "GET", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + const wrapped = await wrapper.wrapper(async () => { + return { + type: "core", + statusCode: 201, + headers: { "content-type": "text/plain", "set-cookie": ["first=1", "second=2"] }, + body: Readable.toWeb(Readable.from("hello")), + isBase64Encoded: false, + }; + }, converter); + + const response = (await wrapped(event)) as APIGatewayProxyResultV2; + + expect(response).toMatchObject({ + statusCode: 201, + body: "hello", + isBase64Encoded: false, + cookies: ["first=1", "second=2"], + }); + }); + + it("finalizes a bodyless response", async () => { + const event = { + version: "2.0", + routeKey: "$default", + rawPath: "/", + rawQueryString: "", + headers: {}, + requestContext: { http: { method: "HEAD", sourceIp: "::1" } }, + isBase64Encoded: false, + } as APIGatewayProxyEventV2; + const wrapped = await wrapper.wrapper(async () => { + return { + type: "core", + statusCode: 204, + headers: {}, + isBase64Encoded: false, + }; + }, converter); + + await expect(wrapped(event)).resolves.toMatchObject({ statusCode: 204, body: "" }); + }); +}); diff --git a/packages/tests-unit/tests/wrappers/cloudflare.test.ts b/packages/tests-unit/tests/wrappers/cloudflare.test.ts new file mode 100644 index 00000000..5f655b3c --- /dev/null +++ b/packages/tests-unit/tests/wrappers/cloudflare.test.ts @@ -0,0 +1,81 @@ +import edgeConverter from "@opennextjs/core/overrides/converters/edge.js"; +import cloudflareEdge from "@opennextjs/core/overrides/wrappers/cloudflare-edge.js"; +import cloudflareNode from "@opennextjs/core/overrides/wrappers/cloudflare-node.js"; +import type { Converter } from "@opennextjs/core/types/overrides.js"; +import { describe, expect, it } from "vitest"; + +describe.each([ + ["edge", cloudflareEdge], + ["node", cloudflareNode], +])("cloudflare-%s wrapper", (_name, wrapper) => { + it("returns the response as soon as headers are available", async () => { + const pending: Promise[] = []; + const { promise: continueHandler, resolve: resolveHandler } = Promise.withResolvers(); + const wrapped = await wrapper.wrapper(async (_event, options) => { + const stream = options?.streamCreator?.writeHeaders({ + statusCode: 200, + cookies: [], + headers: { "content-type": "text/plain" }, + }); + await continueHandler; + stream?.end("hello"); + return { type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }; + }, edgeConverter); + const request = new Request("https://example.com/"); + const responsePromise = wrapped(request, {}, { waitUntil: (promise) => pending.push(promise) }); + const resolution = await Promise.race([ + responsePromise.then(() => "response"), + new Promise((resolve) => setTimeout(() => resolve("timeout"), 50)), + ]); + expect(resolution).toBe("response"); + const response = (await responsePromise) as Response; + + expect(response).toBeInstanceOf(Response); + resolveHandler(); + expect(await response.text()).toBe("hello"); + await Promise.all(pending); + }); + + it("propagates handler failures before headers are written", async () => { + const handlerError = new Error("handler failed"); + const wrapped = await wrapper.wrapper(async () => Promise.reject(handlerError), edgeConverter); + + await expect( + wrapped(new Request("https://example.com/"), {}, { waitUntil: () => undefined }) + ).rejects.toBe(handlerError); + }); + + it("aborts the response body when the handler fails after writing headers", async () => { + const pending: Promise[] = []; + const handlerError = new Error("handler failed after headers"); + const wrapped = await wrapper.wrapper(async (_event, options) => { + options?.streamCreator?.writeHeaders({ statusCode: 200, cookies: [], headers: {} }); + throw handlerError; + }, edgeConverter); + + const response = (await wrapped( + new Request("https://example.com/"), + {}, + { + waitUntil: (promise) => pending.push(promise), + } + )) as Response; + await expect(response.text()).rejects.toBe(handlerError); + await Promise.allSettled(pending); + }); + + it("provides waitUntil to handlers using a direct converter", async () => { + const waitUntil = () => undefined; + const converter = { + name: "direct", + convertFrom: edgeConverter.convertFrom, + convertTo: async () => ({ type: "direct" as const, data: async () => "response" }), + } satisfies Converter; + const wrapped = await wrapper.wrapper(async (_event, options) => { + expect(options?.waitUntil).toBeTypeOf("function"); + return { type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }; + }, converter); + + await expect(wrapped(new Request("https://example.com/"), {}, { waitUntil })).resolves.toBe("response"); + }); +}); diff --git a/packages/tests-unit/tests/wrappers/compression.test.ts b/packages/tests-unit/tests/wrappers/compression.test.ts new file mode 100644 index 00000000..86938ca6 --- /dev/null +++ b/packages/tests-unit/tests/wrappers/compression.test.ts @@ -0,0 +1,20 @@ +import { + selectCompressionEncoding, + withCompressionVary, +} from "@opennextjs/aws/overrides/wrappers/compression.js"; +import { describe, expect, it } from "vitest"; + +describe("compression helpers", () => { + it("respects quality values and disabled encodings", () => { + expect(selectCompressionEncoding("gzip;q=0, deflate;q=0.5, br;q=0.8")).toBe("br"); + expect(selectCompressionEncoding("gzip;q=0")).toBeNull(); + expect(selectCompressionEncoding("*;q=0.5")).toBe("br"); + }); + + it("adds Accept-Encoding to Vary once", () => { + expect(withCompressionVary({ vary: "RSC" })).toEqual({ vary: "RSC, Accept-Encoding" }); + expect(withCompressionVary({ vary: "RSC, accept-encoding" })).toEqual({ + vary: "RSC, accept-encoding", + }); + }); +}); diff --git a/packages/tests-unit/tests/wrappers/dummy.test.ts b/packages/tests-unit/tests/wrappers/dummy.test.ts new file mode 100644 index 00000000..2e582a17 --- /dev/null +++ b/packages/tests-unit/tests/wrappers/dummy.test.ts @@ -0,0 +1,62 @@ +import { Writable } from "node:stream"; + +import wrapper from "@opennextjs/core/overrides/wrappers/dummy.js"; +import type { Converter } from "@opennextjs/core/types/overrides.js"; +import { describe, expect, it } from "vitest"; + +describe("dummy wrapper", () => { + it("returns buffered converter output after streaming", async () => { + const converter = { + name: "buffered", + convertFrom: async (event: unknown) => event, + convertTo: async () => ({ + type: "stream" as const, + streamCreator: { writeHeaders: () => new Writable({ write: (_chunk, _encoding, done) => done() }) }, + output: Promise.resolve("platform response"), + }), + } satisfies Converter; + const wrapped = await wrapper.wrapper( + async () => ({ type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }), + converter + ); + + await expect(wrapped({ type: "core" })).resolves.toBe("platform response"); + }); + + it("returns exceptional direct mappings from streaming converters", async () => { + const converter = { + name: "exceptional", + convertFrom: async (event: unknown) => event, + convertTo: async () => ({ + type: "stream" as const, + streamCreator: { writeHeaders: () => new Writable() }, + data: async () => "direct response", + }), + } satisfies Converter; + const wrapped = await wrapper.wrapper( + async () => ({ type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }), + converter + ); + + await expect(wrapped({ type: "core" })).resolves.toBe("direct response"); + }); + + it("preserves a null direct mapping", async () => { + const converter = { + name: "null-result", + convertFrom: async (event: unknown) => event, + convertTo: async () => ({ + type: "stream" as const, + streamCreator: { writeHeaders: () => new Writable() }, + output: Promise.resolve("buffered response"), + data: async () => null, + }), + } satisfies Converter; + const wrapped = await wrapper.wrapper( + async () => ({ type: "core", statusCode: 200, headers: {}, isBase64Encoded: false }), + converter + ); + + await expect(wrapped({ type: "core" })).resolves.toBeNull(); + }); +});