Skip to content

Commit 3eff2fc

Browse files
conico974conico974
committed
feat: Refactor AWS Lambda streaming support and introduce new converters
- Updated aws-lambda-streaming wrapper to utilize new converter structure. - Introduced aws-streaming converter for handling streaming responses in AWS Lambda. - Enhanced existing converters to support direct and streaming output types. - Modified core types to accommodate new converter output structure. - Updated validation logic to ensure compatibility with new streaming converter. - Added tests for aws-streaming converter and SQS revalidate functionality. - Refactored express-dev and cloudflare wrappers to align with new converter outputs. - Improved response handling in node and cloudflare-node wrappers.
1 parent 46d4792 commit 3eff2fc

27 files changed

Lines changed: 662 additions & 433 deletions

‎packages/aws/src/overrides/converters/aws-apigw-v1.ts‎

Lines changed: 21 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,12 @@ import {
33
extractHostFromHeaders,
44
removeUndefinedFromQuery,
55
} from "@opennextjs/core/overrides/converters/utils.js";
6-
import type { InternalEvent, InternalResult } from "@opennextjs/core/types/open-next.js";
6+
import type { InternalEvent } from "@opennextjs/core/types/open-next.js";
77
import type { Converter } from "@opennextjs/core/types/overrides.js";
8-
import { fromReadableStream } from "@opennextjs/core/utils/stream.js";
98
import type { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda";
109

10+
import { createBufferedStreamCreator } from "./response-stream.js";
11+
1112
function normalizeAPIGatewayProxyEventHeaders(event: APIGatewayProxyEvent): Record<string, string> {
1213
event.multiValueHeaders;
1314
const headers: Record<string, string> = {};
@@ -89,10 +90,14 @@ async function convertFromAPIGatewayProxyEvent(event: APIGatewayProxyEvent): Pro
8990
};
9091
}
9192

92-
async function convertToApiGatewayProxyResult(result: InternalResult): Promise<APIGatewayProxyResult> {
93+
function convertToApiGatewayProxyResult(
94+
prelude: { statusCode: number; cookies: string[]; headers: Record<string, string> },
95+
body: Buffer,
96+
isBase64Encoded: boolean
97+
): APIGatewayProxyResult {
9398
const headers: Record<string, string> = {};
9499
const multiValueHeaders: Record<string, string[]> = {};
95-
Object.entries(result.headers).forEach(([key, value]) => {
100+
Object.entries(prelude.headers).forEach(([key, value]) => {
96101
if (Array.isArray(value)) {
97102
multiValueHeaders[key] = value;
98103
} else {
@@ -103,22 +108,26 @@ async function convertToApiGatewayProxyResult(result: InternalResult): Promise<A
103108
headers[key] = value;
104109
}
105110
});
106-
107-
const body = await fromReadableStream(result.body, result.isBase64Encoded);
111+
if (prelude.cookies.length > 0) {
112+
multiValueHeaders["set-cookie"] = prelude.cookies;
113+
}
108114

109115
const response: APIGatewayProxyResult = {
110-
statusCode: result.statusCode,
116+
statusCode: prelude.statusCode,
111117
headers,
112-
body,
113-
isBase64Encoded: result.isBase64Encoded,
118+
body: body.toString(isBase64Encoded ? "base64" : "utf8"),
119+
isBase64Encoded,
114120
multiValueHeaders,
115121
};
116122
debug(response);
117123
return response;
118124
}
119125

120126
export default {
121-
convertFrom: convertFromAPIGatewayProxyEvent,
122-
convertTo: convertToApiGatewayProxyResult,
127+
convertFrom: (event) => convertFromAPIGatewayProxyEvent(event as APIGatewayProxyEvent),
128+
convertTo: async () => {
129+
const { streamCreator, output } = createBufferedStreamCreator(convertToApiGatewayProxyResult);
130+
return { type: "stream" as const, streamCreator, output };
131+
},
123132
name: "aws-apigw-v1",
124-
} as Converter;
133+
} satisfies Converter;

‎packages/aws/src/overrides/converters/aws-apigw-v2.ts‎

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,12 @@ import {
55
extractHostFromHeaders,
66
removeUndefinedFromQuery,
77
} from "@opennextjs/core/overrides/converters/utils.js";
8-
import type { InternalEvent, InternalResult } from "@opennextjs/core/types/open-next.js";
8+
import type { InternalEvent } from "@opennextjs/core/types/open-next.js";
99
import type { Converter } from "@opennextjs/core/types/overrides.js";
10-
import { fromReadableStream } from "@opennextjs/core/utils/stream.js";
1110
import type { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda";
1211

12+
import { createBufferedStreamCreator } from "./response-stream.js";
13+
1314
// Not sure which one is really needed as this is not documented anywhere but server actions redirect are not working without this,
1415
// it causes a 500 error from cloudfront itself with a 'x-amzErrortype: InternalFailure' header
1516
const CloudFrontBlacklistedHeaders = [
@@ -68,7 +69,9 @@ function normalizeAPIGatewayProxyEventV2Headers(event: APIGatewayProxyEventV2):
6869
return headers;
6970
}
7071

71-
async function convertFromAPIGatewayProxyEventV2(event: APIGatewayProxyEventV2): Promise<InternalEvent> {
72+
export async function convertFromAPIGatewayProxyEventV2(
73+
event: APIGatewayProxyEventV2
74+
): Promise<InternalEvent> {
7275
const { rawPath, rawQueryString, requestContext } = event;
7376
const headers = normalizeAPIGatewayProxyEventV2Headers(event);
7477
return {
@@ -92,9 +95,13 @@ async function convertFromAPIGatewayProxyEventV2(event: APIGatewayProxyEventV2):
9295
};
9396
}
9497

95-
async function convertToApiGatewayProxyResultV2(result: InternalResult): Promise<APIGatewayProxyResultV2> {
98+
function convertToApiGatewayProxyResultV2(
99+
prelude: { statusCode: number; cookies: string[]; headers: Record<string, string> },
100+
body: Buffer,
101+
isBase64Encoded: boolean
102+
): APIGatewayProxyResultV2 {
96103
const headers: Record<string, string> = {};
97-
Object.entries(result.headers)
104+
Object.entries(prelude.headers)
98105
.map(([key, value]) => [key.toLowerCase(), value] as const)
99106
.filter(
100107
([key]) =>
@@ -110,21 +117,27 @@ async function convertToApiGatewayProxyResultV2(result: InternalResult): Promise
110117
headers[key] = Array.isArray(value) ? value.join(", ") : `${value}`;
111118
});
112119

113-
const body = await fromReadableStream(result.body, result.isBase64Encoded);
114-
115120
const response: APIGatewayProxyResultV2 = {
116-
statusCode: result.statusCode,
121+
statusCode: prelude.statusCode,
117122
headers,
118-
cookies: parseSetCookieHeader(result.headers["set-cookie"]),
119-
body,
120-
isBase64Encoded: result.isBase64Encoded,
123+
cookies:
124+
prelude.cookies.length > 0
125+
? prelude.cookies
126+
: prelude.headers["set-cookie"]
127+
? parseSetCookieHeader(prelude.headers["set-cookie"])
128+
: undefined,
129+
body: body.toString(isBase64Encoded ? "base64" : "utf8"),
130+
isBase64Encoded,
121131
};
122132
debug(response);
123133
return response;
124134
}
125135

126136
export default {
127-
convertFrom: convertFromAPIGatewayProxyEventV2,
128-
convertTo: convertToApiGatewayProxyResultV2,
137+
convertFrom: (event) => convertFromAPIGatewayProxyEventV2(event as APIGatewayProxyEventV2),
138+
convertTo: async () => {
139+
const { streamCreator, output } = createBufferedStreamCreator(convertToApiGatewayProxyResultV2);
140+
return { type: "stream" as const, streamCreator, output };
141+
},
129142
name: "aws-apigw-v2",
130-
} as Converter;
143+
} satisfies Converter;

‎packages/aws/src/overrides/converters/aws-cloudfront.ts‎

Lines changed: 64 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import { parseSetCookieHeader } from "@opennextjs/core/http/util.js";
66
import { extractHostFromHeaders } from "@opennextjs/core/overrides/converters/utils.js";
77
import type { InternalEvent, InternalResult, MiddlewareResult } from "@opennextjs/core/types/open-next.js";
88
import type { Converter } from "@opennextjs/core/types/overrides.js";
9-
import { fromReadableStream } from "@opennextjs/core/utils/stream.js";
109
import type {
1110
CloudFrontCustomOrigin,
1211
CloudFrontHeaders,
@@ -15,6 +14,8 @@ import type {
1514
CloudFrontRequestResult,
1615
} from "aws-lambda";
1716

17+
import { createBufferedStreamCreator } from "./response-stream.js";
18+
1819
const cloudfrontBlacklistedHeaders = [
1920
// Disallowed headers, see: https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/edge-function-restrictions-all.html#function-restrictions-disallowed-headers
2021
"connection",
@@ -129,67 +130,83 @@ function convertToCloudfrontHeaders(headers: Record<string, OutgoingHttpHeader>,
129130
return cloudfrontHeaders;
130131
}
131132

132-
async function convertToCloudFrontRequestResult(
133-
result: InternalResult | MiddlewareResult,
133+
async function convertMiddlewareResult(
134+
result: MiddlewareResult,
134135
originalRequest: CloudFrontRequestEvent
135136
): Promise<CloudFrontRequestResult> {
136-
if (result.type === "middleware") {
137-
const { method, clientIp, origin } = originalRequest.Records[0].cf.request;
138-
const responseHeaders = result.internalEvent.headers;
139-
140-
// Handle external rewrite
141-
142-
let customOrigin = origin?.custom as CloudFrontCustomOrigin;
143-
let host = responseHeaders.host ?? responseHeaders.Host;
144-
if (result.origin) {
145-
customOrigin = {
146-
...customOrigin,
147-
domainName: result.origin.host,
148-
port: result.origin.port ?? 443,
149-
protocol: result.origin.protocol ?? "https",
150-
customHeaders: {},
151-
};
152-
host = result.origin.host;
153-
}
154-
155-
const response: CloudFrontRequest = {
156-
clientIp,
157-
method,
158-
uri: result.internalEvent.rawPath,
159-
querystring: convertToQueryString(result.internalEvent.query).replace("?", ""),
160-
headers: convertToCloudfrontHeaders({
161-
...responseHeaders,
162-
host,
163-
}),
164-
origin: origin?.custom
165-
? {
166-
custom: customOrigin,
167-
}
168-
: origin,
137+
const { method, clientIp, origin } = originalRequest.Records[0].cf.request;
138+
const responseHeaders = result.internalEvent.headers;
139+
140+
// Handle external rewrite
141+
142+
let customOrigin = origin?.custom as CloudFrontCustomOrigin;
143+
let host = responseHeaders.host ?? responseHeaders.Host;
144+
if (result.origin) {
145+
customOrigin = {
146+
...customOrigin,
147+
domainName: result.origin.host,
148+
port: result.origin.port ?? 443,
149+
protocol: result.origin.protocol ?? "https",
150+
customHeaders: {},
169151
};
152+
host = result.origin.host;
153+
}
170154

171-
debug("response rewrite", response);
155+
const response: CloudFrontRequest = {
156+
clientIp,
157+
method,
158+
uri: result.internalEvent.rawPath,
159+
querystring: convertToQueryString(result.internalEvent.query).replace("?", ""),
160+
headers: convertToCloudfrontHeaders({
161+
...responseHeaders,
162+
host,
163+
}),
164+
origin: origin?.custom
165+
? {
166+
custom: customOrigin,
167+
}
168+
: origin,
169+
};
172170

173-
return response;
174-
}
171+
debug("response rewrite", response);
175172

176-
const body = await fromReadableStream(result.body, result.isBase64Encoded);
177-
const responseHeaders = result.headers;
173+
return response;
174+
}
178175

176+
function convertToCloudFrontRequestResult(
177+
prelude: { statusCode: number; cookies: string[]; headers: Record<string, string> },
178+
body: Buffer,
179+
isBase64Encoded: boolean
180+
): CloudFrontRequestResult {
181+
const responseHeaders = {
182+
...prelude.headers,
183+
...(prelude.cookies.length > 0 ? { "set-cookie": prelude.cookies } : {}),
184+
};
179185
const response: CloudFrontRequestResult = {
180-
status: result.statusCode.toString(),
186+
status: prelude.statusCode.toString(),
181187
statusDescription: "OK",
182188
headers: convertToCloudfrontHeaders(responseHeaders, true),
183-
bodyEncoding: result.isBase64Encoded ? "base64" : "text",
184-
body,
189+
bodyEncoding: isBase64Encoded ? "base64" : "text",
190+
body: body.toString(isBase64Encoded ? "base64" : "utf8"),
185191
};
186192

187193
debug(response);
188194
return response;
189195
}
190196

191197
export default {
192-
convertFrom: convertFromCloudFrontRequestEvent,
193-
convertTo: convertToCloudFrontRequestResult,
198+
convertFrom: (event) => convertFromCloudFrontRequestEvent(event as CloudFrontRequestEvent),
199+
convertTo: async (event) => {
200+
const { streamCreator, output } = createBufferedStreamCreator(convertToCloudFrontRequestResult);
201+
return {
202+
type: "stream" as const,
203+
streamCreator,
204+
output,
205+
data: async (result) =>
206+
result.type === "middleware"
207+
? convertMiddlewareResult(result, event as CloudFrontRequestEvent)
208+
: undefined,
209+
};
210+
},
194211
name: "aws-cloudfront",
195-
} as Converter;
212+
} satisfies Converter<InternalEvent, InternalResult | MiddlewareResult>;

‎packages/aws/src/overrides/converters/aws-streaming.ts‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
import type { Writable } from "node:stream";
2+
3+
import type { StreamCreator } from "@opennextjs/core/types/open-next.js";
4+
import type { Converter } from "@opennextjs/core/types/overrides.js";
5+
import type { APIGatewayProxyEventV2 } from "aws-lambda";
6+
7+
import { convertFromAPIGatewayProxyEventV2 } from "./aws-apigw-v2.js";
8+
9+
type StreamingContext = {
10+
responseStream: Writable & { setContentType(contentType: string): void };
11+
writable: Writable;
12+
contentEncoding: string;
13+
};
14+
15+
const converter: Converter = {
16+
convertFrom: (event) => convertFromAPIGatewayProxyEventV2(event as APIGatewayProxyEventV2),
17+
convertTo: async (_event, context) => {
18+
const { responseStream, writable, contentEncoding } = context as StreamingContext;
19+
const streamCreator: StreamCreator = {
20+
writeHeaders(prelude) {
21+
responseStream.setContentType("application/vnd.awslambda.http-integration-response");
22+
responseStream.write(
23+
JSON.stringify({
24+
...prelude,
25+
headers: {
26+
...prelude.headers,
27+
"content-encoding": contentEncoding,
28+
},
29+
})
30+
);
31+
responseStream.write(new Uint8Array(8));
32+
return writable;
33+
},
34+
retainChunks: false,
35+
};
36+
37+
return { type: "stream" as const, streamCreator };
38+
},
39+
name: "aws-streaming",
40+
};
41+
42+
export default converter;

‎packages/aws/src/overrides/converters/response-stream.ts‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
import { Writable } from "node:stream";
2+
3+
import type { StreamCreator } from "@opennextjs/core/types/open-next.js";
4+
import { isBinaryContentType } from "@opennextjs/core/utils/binary.js";
5+
6+
type Prelude = Parameters<StreamCreator["writeHeaders"]>[0];
7+
8+
export function createBufferedStreamCreator<T>(
9+
createOutput: (prelude: Prelude, body: Buffer, isBase64Encoded: boolean) => T
10+
): { streamCreator: StreamCreator; output: Promise<T> } {
11+
const { promise: output, resolve, reject } = Promise.withResolvers<T>();
12+
let prelude: Prelude | undefined;
13+
const chunks: Buffer[] = [];
14+
15+
const streamCreator: StreamCreator = {
16+
writeHeaders(value) {
17+
prelude = value;
18+
return new Writable({
19+
write(chunk, _encoding, callback) {
20+
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
21+
callback();
22+
},
23+
final(callback) {
24+
if (!prelude) {
25+
const error = new Error("Response stream finished before headers were written");
26+
reject(error);
27+
callback(error);
28+
return;
29+
}
30+
try {
31+
const isBase64Encoded =
32+
isBinaryContentType(prelude.headers["content-type"]) || !!prelude.headers["content-encoding"];
33+
resolve(createOutput(prelude, Buffer.concat(chunks), isBase64Encoded));
34+
callback();
35+
} catch (error: unknown) {
36+
reject(error);
37+
callback(error instanceof Error ? error : new Error(String(error)));
38+
}
39+
},
40+
});
41+
},
42+
retainChunks: false,
43+
};
44+
45+
return { streamCreator, output };
46+
}

0 commit comments

Comments
 (0)