diff options
Diffstat (limited to 'Homework/cs5260/widgetrequesthandler')
10 files changed, 1337 insertions, 0 deletions
diff --git a/Homework/cs5260/widgetrequesthandler/bootstrap b/Homework/cs5260/widgetrequesthandler/bootstrap new file mode 100644 index 0000000..38387b2 --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/bootstrap @@ -0,0 +1,3 @@ +#! /bin/sh +export BUN_INSTALL_CACHE_DIR=/tmp/bun/cache +exec /opt/bun --cwd $LAMBDA_TASK_ROOT /opt/runtime.ts diff --git a/Homework/cs5260/widgetrequesthandler/build.ts b/Homework/cs5260/widgetrequesthandler/build.ts new file mode 100644 index 0000000..a44e5f8 --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/build.ts @@ -0,0 +1,5 @@ +await Bun.build({ + entrypoints: ['./handler.ts'], + outdir: './dist', + target: 'bun' +}); diff --git a/Homework/cs5260/widgetrequesthandler/handler.ts b/Homework/cs5260/widgetrequesthandler/handler.ts new file mode 100644 index 0000000..6afb05c --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/handler.ts @@ -0,0 +1,89 @@ +import type { Server } from "bun"; +import { ConsoleTracingLogger, TracingLogger } from "@/shared/log"; +import { WidgetRequest } from "@/shared/schema"; +import { + ProcessorResponse, + SQSWidgetRequestProcessor, + WidgetRequestProcessor, +} from "./processors"; +import { SQSClient } from "@aws-sdk/client-sqs"; + +export class WidgetRequestHandler { + private processor: WidgetRequestProcessor; + private logger: TracingLogger; + + constructor(processor: WidgetRequestProcessor, logger: TracingLogger) { + this.processor = processor; + this.logger = logger; + } + + public async handleRequest(request: Request): Promise<Response> { + let headers = new Headers({ + "Content-Type": "application/json", + }); + let response: ProcessorResponse = { success: true, message: "success" }; + let status = 200; + + let widgetRequest; + + if (request.body) { + const body = (await request.json()) as { event: any }; + if (body) { + const event = body.event!; + this.logger.info(`received Body=(${JSON.stringify(event)})`); + widgetRequest = WidgetRequest.safeParse(event); + } + } + + if (!widgetRequest || !widgetRequest.success) { + status = 400; + response = { + success: false, + message: widgetRequest?.error + ? widgetRequest.error.toString() + : "invalid request body", + }; + } else { + try { + response = await this.processor.handle(widgetRequest.data); + } catch (e) { + status = 500; + response = { + success: false, + message: e?.toString() ?? "failed to process", + }; + } + } + + return new Response(JSON.stringify(response), { + status, + headers, + }); + } +} + +const lambdaHandler = async ( + request: Request, + _server: Server +): Promise<Response> => { + const logger = new ConsoleTracingLogger("handler"); + const sourceQueueUrl = process.env.SOURCE_QUEUE!; + const sqsClient = new SQSClient({ + region: process.env.AWS_REGION!, + }); + + const processor = new SQSWidgetRequestProcessor( + sqsClient, + sourceQueueUrl, + logger.createChild("WidgetRequestProcessor") + ); + + const handler = new WidgetRequestHandler( + processor, + logger.createChild("WidgetRequestHandler") + ); + + return await handler.handleRequest(request); +}; + +export default lambdaHandler; diff --git a/Homework/cs5260/widgetrequesthandler/lambda-test-events.txt b/Homework/cs5260/widgetrequesthandler/lambda-test-events.txt new file mode 100644 index 0000000..009342a --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/lambda-test-events.txt @@ -0,0 +1,33 @@ +-- CreateWidget +{ + "type": "create", + "requestId": "9cba45e8-ff0f-4ab1-b847-48ac3f0ea77a", + "widgetId": "7ed25344-bfde-4b78-9403-37d0a5fa59cd", + "owner": "simponic", + "label": "lambda widget test 01", + "description": "a test lambda widget", + "otherAttributes": [ + { + "name": "lambdaFn", + "value": "widget-api-handler" + } + ] +} + +-- UpdateWidget +{ + "type": "update", + "requestId": "9cba45e8-ff0f-4ab1-b847-48ac3f0ea77c", + "widgetId": "7ed25344-bfde-4b78-9403-37d0a5fa59cd", + "owner": "simponic", + "label": "lambda widget test 02", + "description": "a new updated test lambda widget" +} + +-- DeleteWidget +{ + "type": "delete", + "requestId": "9cba45e8-ff0f-4ab1-b847-48ac3f0ea77b", + "widgetId": "7ed25344-bfde-4b78-9403-37d0a5fa59cd", + "owner": "simponic" +} diff --git a/Homework/cs5260/widgetrequesthandler/processors/SQSWidgetRequestProcessor.ts b/Homework/cs5260/widgetrequesthandler/processors/SQSWidgetRequestProcessor.ts new file mode 100644 index 0000000..cf4f4c5 --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/processors/SQSWidgetRequestProcessor.ts @@ -0,0 +1,31 @@ +import { TracingLogger } from "@/shared/log"; +import { WidgetRequestT } from "@/shared/schema"; +import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs"; +import { ProcessorResponse, WidgetRequestProcessor } from "."; + +export class SQSWidgetRequestProcessor implements WidgetRequestProcessor { + private sqsClient: SQSClient; + private queueUrl: string; + private logger: TracingLogger; + + constructor(sqsClient: SQSClient, queueUrl: string, logger: TracingLogger) { + this.sqsClient = sqsClient; + this.queueUrl = queueUrl; + this.logger = logger; + } + + public async handle( + widgetRequest: WidgetRequestT, + ): Promise<ProcessorResponse> { + const MessageBody = JSON.stringify(widgetRequest); + this.logger.info(`Handling WidgetRequest=(${MessageBody})`); + + const sendMessage = new SendMessageCommand({ + QueueUrl: this.queueUrl, + MessageBody, + }); + await this.sqsClient.send(sendMessage); + + return { success: true, message: "added to queue" }; + } +} diff --git a/Homework/cs5260/widgetrequesthandler/processors/index.ts b/Homework/cs5260/widgetrequesthandler/processors/index.ts new file mode 100644 index 0000000..5ce1a0d --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/processors/index.ts @@ -0,0 +1,12 @@ +import { WidgetRequestT } from "@/shared/schema"; + +export * from "./SQSWidgetRequestProcessor.ts"; + +export interface ProcessorResponse { + success: boolean; + message: string; +} + +export interface WidgetRequestProcessor { + handle(widgetRequest: WidgetRequestT): Promise<ProcessorResponse>; +} diff --git a/Homework/cs5260/widgetrequesthandler/runtime.ts b/Homework/cs5260/widgetrequesthandler/runtime.ts new file mode 100644 index 0000000..60b0aa5 --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/runtime.ts @@ -0,0 +1,897 @@ +import type { Server, ServerWebSocket } from "bun"; +import { AwsClient } from "aws4fetch"; + +type Lambda = { + fetch: (request: Request, server: Server) => Promise<Response | undefined>; + error?: (error: unknown) => Promise<Response>; + websocket?: { + open?: (ws: ServerWebSocket) => Promise<void>; + message?: (ws: ServerWebSocket, message: string) => Promise<void>; + close?: ( + ws: ServerWebSocket, + code: number, + reason: string, + ) => Promise<void>; + }; +}; + +let requestId: string | undefined; +let traceId: string | undefined; +let functionArn: string | undefined; +let aws: AwsClient | undefined; + +let logger = console.log; + +function log(level: string, ...args: any[]): void { + if (!args.length) { + return; + } + const messages = args.map((arg) => Bun.inspect(arg).replace(/\n/g, "\r")); + if (requestId === undefined) { + logger(level, ...messages); + } else { + logger(level, `RequestId: ${requestId}`, ...messages); + } +} + +console.log = (...args: any[]) => log("INFO", ...args); +console.info = (...args: any[]) => log("INFO", ...args); +console.warn = (...args: any[]) => log("WARN", ...args); +console.error = (...args: any[]) => log("ERROR", ...args); +console.debug = (...args: any[]) => log("DEBUG", ...args); +console.trace = (...args: any[]) => log("TRACE", ...args); + +let warnings: Set<string> | undefined; + +function warnOnce(message: string, ...args: any[]): void { + if (warnings === undefined) { + warnings = new Set(); + } + if (warnings.has(message)) { + return; + } + warnings.add(message); + console.warn(message, ...args); +} + +function reset(): void { + requestId = undefined; + traceId = undefined; + warnings = undefined; +} + +function exit(...cause: any[]): never { + console.error(...cause); + process.exit(1); +} + +function env(name: string, fallback?: string): string { + const value = process.env[name] ?? fallback ?? null; + if (value === null) { + exit(`Runtime failed to find the '${name}' environment variable`); + } + return value; +} + +const runtimeUrl = new URL( + `http://${env("AWS_LAMBDA_RUNTIME_API")}/2018-06-01/`, +); + +async function fetch(url: string, options?: RequestInit): Promise<Response> { + const { href } = new URL(url, runtimeUrl); + const response = await globalThis.fetch(href, { + ...options, + timeout: false, + }); + if (!response.ok) { + exit( + `Runtime failed to send request to Lambda [status: ${response.status}]`, + ); + } + return response; +} + +async function fetchAws(url: string, options?: RequestInit): Promise<Response> { + if (aws === undefined) { + aws = new AwsClient({ + accessKeyId: env("AWS_ACCESS_KEY_ID"), + secretAccessKey: env("AWS_SECRET_ACCESS_KEY"), + sessionToken: env("AWS_SESSION_TOKEN"), + region: env("AWS_REGION"), + }); + } + return aws.fetch(url, options); +} + +type LambdaError = { + readonly errorType: string; + readonly errorMessage: string; + readonly stackTrace?: string[]; +}; + +function formatError(error: unknown): LambdaError { + if (error instanceof Error) { + return { + errorType: error.name, + errorMessage: error.message, + stackTrace: error.stack + ?.split("\n") + .filter((line) => !line.includes(" /opt/runtime.ts")), + }; + } + return { + errorType: "Error", + errorMessage: Bun.inspect(error), + }; +} + +async function sendError(type: string, cause: unknown): Promise<void> { + console.error(cause); + await fetch( + requestId === undefined + ? "runtime/init/error" + : `runtime/invocation/${requestId}/error`, + { + method: "POST", + headers: { + "Content-Type": "application/vnd.aws.lambda.error+json", + "Lambda-Runtime-Function-Error-Type": `Bun.${type}`, + }, + body: JSON.stringify(formatError(cause)), + }, + ); +} + +async function throwError(type: string, cause: unknown): Promise<never> { + await sendError(type, cause); + exit(); +} + +async function init(): Promise<Lambda> { + const handlerName = env("_HANDLER"); + const index = handlerName.lastIndexOf("."); + const fileName = handlerName.substring(0, index); + const filePath = `${env("LAMBDA_TASK_ROOT")}/${fileName}`; + let file; + try { + file = await import(filePath); + } catch (cause) { + if ( + cause instanceof Error && + cause.message.startsWith("Cannot find module") + ) { + return throwError( + "FileDoesNotExist", + `Did not find a file named '${fileName}'`, + ); + } + return throwError("InitError", cause); + } + const moduleName = handlerName.substring(index + 1) || "fetch"; + let module = file["default"] ?? file[moduleName] ?? {}; + if (typeof module === "function") { + module = { + fetch: module, + }; + } else if (typeof module === "object" && moduleName !== "fetch") { + module = { + ...module, + fetch: module[moduleName], + }; + } + const { fetch, websocket } = module; + if (typeof fetch !== "function") { + return throwError( + fetch === undefined ? "MethodDoesNotExist" : "MethodIsNotAFunction", + `${fileName} does not have a default export with a function named '${moduleName}'`, + ); + } + if (websocket === undefined) { + return module; + } + for (const name of ["open", "message", "close"]) { + const method = websocket[name]; + if (method === undefined) { + continue; + } + if (typeof method !== "function") { + return throwError( + "MethodIsNotAFunction", + `${fileName} does not have a function named '${name}' on the default 'websocket' property`, + ); + } + } + return module; +} + +type LambdaRequest<E = any> = { + readonly requestId: string; + readonly traceId: string; + readonly functionArn: string; + readonly deadlineMs: number | null; + readonly event: E; +}; + +async function receiveRequest(): Promise<LambdaRequest> { + const response = await fetch("runtime/invocation/next"); + requestId = + response.headers.get("Lambda-Runtime-Aws-Request-Id") ?? undefined; + if (requestId === undefined) { + exit("Runtime received a request without a request ID"); + } + traceId = response.headers.get("Lambda-Runtime-Trace-Id") ?? undefined; + if (traceId === undefined) { + exit("Runtime received a request without a trace ID"); + } + process.env["_X_AMZN_TRACE_ID"] = traceId; + functionArn = + response.headers.get("Lambda-Runtime-Invoked-Function-Arn") ?? undefined; + if (functionArn === undefined) { + exit("Runtime received a request without a function ARN"); + } + const deadlineMs = + parseInt(response.headers.get("Lambda-Runtime-Deadline-Ms") ?? "0") || null; + let event; + try { + event = await response.json(); + } catch (cause) { + exit("Runtime received a request with invalid JSON", cause); + } + return { + requestId, + traceId, + functionArn, + deadlineMs, + event, + }; +} + +type LambdaResponse = { + readonly statusCode: number; + readonly headers?: Record<string, string>; + readonly isBase64Encoded?: boolean; + readonly body?: string; + readonly multiValueHeaders?: Record<string, string[]>; + readonly cookies?: string[]; +}; + +async function formatResponse(response: Response): Promise<LambdaResponse> { + const statusCode = response.status; + const headers = response.headers.toJSON(); + if (statusCode === 101) { + const protocol = headers["sec-websocket-protocol"]; + if (protocol === undefined) { + return { + statusCode: 200, + }; + } + return { + statusCode: 200, + headers: { + "Sec-WebSocket-Protocol": protocol, + }, + }; + } + const mime = headers["content-type"]; + const isBase64Encoded = + !mime || + (!mime.startsWith("text/") && !mime.startsWith("application/json")); + const body = isBase64Encoded + ? Buffer.from(await response.arrayBuffer()).toString("base64") + : await response.text(); + delete headers["set-cookie"]; + const cookies = response.headers.getAll("Set-Cookie"); + if (cookies.length === 0) { + return { + statusCode, + headers, + isBase64Encoded, + body, + }; + } + return { + statusCode, + headers, + cookies, + multiValueHeaders: { + "Set-Cookie": cookies, + }, + isBase64Encoded, + body, + }; +} + +async function sendResponse(response: unknown): Promise<void> { + if (requestId === undefined) { + exit("Runtime attempted to send a response without a request ID"); + } + await fetch(`runtime/invocation/${requestId}/response`, { + method: "POST", + body: + response === null + ? null + : typeof response === "string" + ? response + : JSON.stringify(response), + }); +} + +function formatBody(body?: string, isBase64Encoded?: boolean): string | null { + if (body === undefined) { + return null; + } + if (!isBase64Encoded) { + return body; + } + return Buffer.from(body).toString("base64"); +} + +type HttpEventV1 = { + readonly requestContext: { + readonly requestId: string; + readonly domainName: string; + readonly httpMethod: string; + readonly path: string; + }; + readonly headers: Record<string, string>; + readonly multiValueHeaders?: Record<string, string[]>; + readonly queryStringParameters?: Record<string, string>; + readonly multiValueQueryStringParameters?: Record<string, string[]>; + readonly isBase64Encoded: boolean; + readonly body?: string; +}; + +function isHttpEventV1(event: any): event is HttpEventV1 { + return ( + !event.Records && + event.version !== "2.0" && + event.version !== "0" && + typeof event.requestContext === "object" + ); +} + +function formatHttpEventV1(event: HttpEventV1): Request { + const request = event.requestContext; + const headers = new Headers(); + for (const [name, values] of Object.entries(event.multiValueHeaders ?? {})) { + for (const value of values) { + headers.append(name, value); + } + } + const hostname = headers.get("Host") ?? request.domainName; + const proto = headers.get("X-Forwarded-Proto") ?? "http"; + const url = new URL(request.path, `${proto}://${hostname}/`); + for (const [name, values] of Object.entries( + event.multiValueQueryStringParameters ?? {}, + )) { + for (const value of values ?? []) { + url.searchParams.append(name, value); + } + } + return new Request(url.toString(), { + method: request.httpMethod, + headers, + body: formatBody(event.body, event.isBase64Encoded), + }); +} + +type HttpEventV2 = { + readonly version: "2.0"; + readonly requestContext: { + readonly requestId: string; + readonly domainName: string; + readonly http: { + readonly method: string; + readonly path: string; + }; + }; + readonly headers: Record<string, string>; + readonly queryStringParameters?: Record<string, string>; + readonly cookies?: string[]; + readonly isBase64Encoded: boolean; + readonly body?: string; +}; + +function isHttpEventV2(event: any): event is HttpEventV2 { + return ( + !event.Records && + event.version === "2.0" && + typeof event.requestContext === "object" + ); +} + +function formatHttpEventV2(event: HttpEventV2): Request { + const request = event.requestContext; + const headers = new Headers(); + for (const [name, values] of Object.entries(event.headers)) { + for (const value of values.split(",")) { + headers.append(name, value); + } + } + for (const [name, values] of Object.entries( + event.queryStringParameters ?? {}, + )) { + for (const value of values.split(",")) { + headers.append(name, value); + } + } + for (const cookie of event.cookies ?? []) { + headers.append("Set-Cookie", cookie); + } + const hostname = headers.get("Host") ?? request.domainName; + const proto = headers.get("X-Forwarded-Proto") ?? "http"; + const url = new URL(request.http.path, `${proto}://${hostname}/`); + return new Request(url.toString(), { + method: request.http.method, + headers, + body: formatBody(event.body, event.isBase64Encoded), + }); +} + +function isHttpEvent(event: any): boolean { + return isHttpEventV1(event) || isHttpEventV2(event); +} + +type WebSocketEvent = { + readonly headers: Record<string, string>; + readonly multiValueHeaders: Record<string, string[]>; + readonly isBase64Encoded: boolean; + readonly body?: string; + readonly requestContext: { + readonly apiId: string; + readonly requestId: string; + readonly connectionId: string; + readonly domainName: string; + readonly stage: string; + readonly identity: { + readonly sourceIp: string; + }; + } & ( + | { + readonly eventType: "CONNECT"; + } + | { + readonly eventType: "MESSAGE"; + } + | { + readonly eventType: "DISCONNECT"; + readonly disconnectStatusCode: number; + readonly disconnectReason: string; + } + ); +}; + +function isWebSocketEvent(event: any): event is WebSocketEvent { + return ( + typeof event.requestContext === "object" && + typeof event.requestContext.connectionId === "string" + ); +} + +function isWebSocketUpgrade(event: any): event is WebSocketEvent { + return ( + isWebSocketEvent(event) && event.requestContext.eventType === "CONNECT" + ); +} + +function formatWebSocketUpgrade(event: WebSocketEvent): Request { + const request = event.requestContext; + const headers = new Headers(); + headers.set("Upgrade", "websocket"); + headers.set("x-amzn-connection-id", request.connectionId); + for (const [name, values] of Object.entries(event.multiValueHeaders as any)) { + for (const value of (values as any) ?? []) { + headers.append(name, value); + } + } + const hostname = headers.get("Host") ?? request.domainName; + const proto = headers.get("X-Forwarded-Proto") ?? "http"; + const url = new URL(`${proto}://${hostname}/${request.stage}`); + return new Request(url.toString(), { + headers, + body: formatBody(event.body, event.isBase64Encoded), + }); +} + +function formatUnknownEvent(event: unknown): Request { + return new Request("https://lambda/", { + method: "POST", + body: JSON.stringify(event), + headers: { + "Content-Type": "application/json;charset=utf-8", + }, + }); +} + +function formatRequest(input: LambdaRequest): Request | undefined { + const { event, requestId, traceId, functionArn, deadlineMs } = input; + let request: Request; + if (isHttpEventV2(event)) { + request = formatHttpEventV2(event); + } else if (isHttpEventV1(event)) { + request = formatHttpEventV1(event); + } else if (isWebSocketEvent(event)) { + if (!isWebSocketUpgrade(event)) { + return undefined; + } + request = formatWebSocketUpgrade(event); + } else { + request = formatUnknownEvent(input); + } + request.headers.set("x-amzn-requestid", requestId); + request.headers.set("x-amzn-trace-id", traceId); + request.headers.set("x-amzn-function-arn", functionArn); + if (deadlineMs !== null) { + request.headers.set("x-amzn-deadline-ms", `${deadlineMs}`); + } + // @ts-ignore: Attach the original event to the Request + request.aws = event; + return request; +} + +class LambdaServer implements Server { + #lambda: Lambda; + #webSockets: Map<string, LambdaWebSocket>; + #upgrade: Response | null; + pendingRequests: number; + pendingWebSockets: number; + port: number; + hostname: string; + development: boolean; + + constructor(lambda: Lambda) { + this.#lambda = lambda; + this.#webSockets = new Map(); + this.#upgrade = null; + this.pendingRequests = 0; + this.pendingWebSockets = 0; + this.port = 80; + this.hostname = "lambda"; + this.development = false; + } + + async accept(request: LambdaRequest): Promise<unknown> { + const deadlineMs = + request.deadlineMs === null ? Date.now() + 60_000 : request.deadlineMs; + const durationMs = Math.max(1, deadlineMs - Date.now()); + let response: unknown; + try { + response = await Promise.race([ + new Promise<undefined>((resolve) => setTimeout(resolve, durationMs)), + this.#acceptRequest(request), + ]); + } catch (cause) { + await sendError("RequestError", cause); + return; + } + if (response === undefined) { + await sendError("TimeoutError", "Function timed out"); + return; + } + return response; + } + + async #acceptRequest(event: LambdaRequest): Promise<unknown> { + const request = formatRequest(event); + let response: Response | undefined; + if (request === undefined) { + await this.#acceptWebSocket(event.event); + } else { + response = await this.fetch(request); + if (response.status === 101) { + await this.#acceptWebSocket(event.event); + } + } + if (response === undefined) { + return { + statusCode: 200, + }; + } + if (!isHttpEvent(event.event)) { + return response.text(); + } + return formatResponse(response); + } + + async #acceptWebSocket(event: WebSocketEvent): Promise<void> { + const request = event.requestContext; + const { connectionId, eventType } = request; + const webSocket = this.#webSockets.get(connectionId); + if (webSocket === undefined || this.#lambda.websocket === undefined) { + return; + } + const { open, message, close } = this.#lambda.websocket; + switch (eventType) { + case "CONNECT": { + if (open) { + await open(webSocket); + } + break; + } + case "MESSAGE": { + if (message) { + const body = formatBody(event.body, event.isBase64Encoded); + if (body !== null) { + await message(webSocket, body); + } + } + break; + } + case "DISCONNECT": { + try { + if (close) { + const { disconnectStatusCode: code, disconnectReason: reason } = + request; + await close(webSocket, code, reason); + } + } finally { + this.#webSockets.delete(connectionId); + this.pendingWebSockets--; + } + break; + } + } + } + + stop(): void { + exit("Runtime exited because Server.stop() was called"); + } + + reload(options: any): void { + this.#lambda = { + fetch: options.fetch ?? this.#lambda.fetch, + error: options.error ?? this.#lambda.error, + websocket: options.websocket ?? this.#lambda.websocket, + }; + this.port = + typeof options.port === "number" + ? options.port + : typeof options.port === "string" + ? parseInt(options.port) + : this.port; + this.hostname = options.hostname ?? this.hostname; + this.development = options.development ?? this.development; + } + + async fetch(request: Request): Promise<Response> { + this.pendingRequests++; + try { + let response = await this.#lambda.fetch(request, this); + if (response instanceof Response) { + return response; + } + if (response === undefined && this.#upgrade !== null) { + return this.#upgrade; + } + throw new Error("fetch() did not return a Response"); + } catch (cause) { + console.error(cause); + if (this.#lambda.error !== undefined) { + try { + return await this.#lambda.error(cause); + } catch (cause) { + console.error(cause); + } + } + return new Response(null, { status: 500 }); + } finally { + this.pendingRequests--; + this.#upgrade = null; + } + } + + upgrade<T = undefined>( + request: Request, + options?: { + headers?: HeadersInit; + data?: T; + }, + ): boolean { + if ( + request.method === "GET" && + request.headers.get("Upgrade")?.toLowerCase() === "websocket" + ) { + this.#upgrade = new Response(null, { + status: 101, + headers: options?.headers, + }); + if ("aws" in request && isWebSocketUpgrade(request.aws)) { + const { connectionId } = request.aws.requestContext; + this.#webSockets.set( + connectionId, + new LambdaWebSocket(request.aws, options?.data), + ); + this.pendingWebSockets++; + } + return true; + } + return false; + } + + publish( + topic: string, + data: string | ArrayBuffer | ArrayBufferView, + compress?: boolean, + ): number { + let count = 0; + for (const webSocket of this.#webSockets.values()) { + count += webSocket.publish(topic, data, compress) ? 1 : 0; + } + return count; + } +} + +class LambdaWebSocket implements ServerWebSocket { + #connectionId: string; + #url: string; + #invokeArn: string; + #topics: Set<string> | null; + remoteAddress: string; + readyState: 0 | 2 | 1 | -1 | 3; + binaryType?: "arraybuffer" | "uint8array"; + data: any; + + constructor(event: WebSocketEvent, data?: any) { + const request = event.requestContext; + this.#connectionId = `${request.connectionId}`; + this.#url = `https://${request.domainName}/${request.stage}/@connections/${this.#connectionId}`; + const [region, accountId] = (functionArn ?? "").split(":").slice(3, 5); + this.#invokeArn = `arn:aws:execute-api:${region}:${accountId}:${request.apiId}/${request.stage}/*`; + this.#topics = null; + this.remoteAddress = request.identity.sourceIp; + this.readyState = 1; // WebSocket.OPEN + this.data = data; + } + + send( + data: string | ArrayBuffer | ArrayBufferView, + compress?: boolean, + ): number { + if (typeof data === "string") { + return this.sendText(data, compress); + } + if (data instanceof ArrayBuffer) { + return this.sendBinary(new Uint8Array(data), compress); + } + const buffer = new Uint8Array( + data.buffer, + data.byteOffset, + data.byteLength, + ); + return this.sendBinary(buffer, compress); + } + + sendText(data: string, compress?: boolean): number { + fetchAws(this.#url, { + method: "POST", + body: data, + }) + .then(({ status }) => { + if (status === 403) { + warnOnce( + "Failed to send WebSocket message due to insufficient IAM permissions", + `Assign the following IAM policy to ${functionArn} to fix this issue:`, + { + Version: "2012-10-17", + Statement: [ + { + Effect: "Allow", + Action: ["execute-api:Invoke"], + Resource: [this.#invokeArn], + }, + ], + }, + ); + } else { + warnOnce(`Failed to send WebSocket message due to a ${status} error`); + } + }) + .catch((error) => { + warnOnce("Failed to send WebSocket message", error); + }); + return data.length; + } + + sendBinary(data: Uint8Array, compress?: boolean): number { + warnOnce( + "Lambda does not support binary WebSocket messages", + "https://docs.aws.amazon.com/apigateway/latest/developerguide/websocket-api-develop-binary-media-types.html", + ); + const base64 = Buffer.from(data).toString("base64"); + return this.sendText(base64, compress); + } + + publish( + topic: string, + data: string | ArrayBuffer | ArrayBufferView, + compress?: boolean, + ): number { + if (this.isSubscribed(topic)) { + return this.send(data, compress); + } + return -1; + } + + publishText(topic: string, data: string, compress?: boolean): number { + if (this.isSubscribed(topic)) { + return this.sendText(data, compress); + } + return -1; + } + + publishBinary(topic: string, data: Uint8Array, compress?: boolean): number { + if (this.isSubscribed(topic)) { + return this.sendBinary(data, compress); + } + return -1; + } + + close(code?: number, reason?: string): void { + // TODO: code? reason? + fetchAws(this.#url, { + method: "DELETE", + }) + .then(({ status }) => { + if (status === 403) { + warnOnce( + "Failed to close WebSocket due to insufficient IAM permissions", + `Assign the following IAM policy to ${functionArn} to fix this issue:`, + { + Version: "2012-10-17", + Statement: [ + { + Effect: "Allow", + Action: ["execute-api:Invoke"], + Resource: [this.#invokeArn], + }, + ], + }, + ); + } else { + warnOnce(`Failed to close WebSocket due to a ${status} error`); + } + }) + .catch((error) => { + warnOnce("Failed to close WebSocket", error); + }); + this.readyState = 3; // WebSocket.CLOSED; + } + + subscribe(topic: string): void { + if (this.#topics === null) { + this.#topics = new Set(); + } + this.#topics.add(topic); + } + + unsubscribe(topic: string): void { + if (this.#topics !== null) { + this.#topics.delete(topic); + } + } + + isSubscribed(topic: string): boolean { + return this.#topics !== null && this.#topics.has(topic); + } + + cork( + callback: (ws: ServerWebSocket<undefined>) => any, + ): void | Promise<void> { + // Lambda does not support sending multiple messages at a time. + return callback(this); + } +} + +const lambda = await init(); +const server = new LambdaServer(lambda); +while (true) { + try { + const request = await receiveRequest(); + const response = await server.accept(request); + if (response !== undefined) { + await sendResponse(response); + } + } finally { + reset(); + } +} diff --git a/Homework/cs5260/widgetrequesthandler/scripts/build-layer.ts b/Homework/cs5260/widgetrequesthandler/scripts/build-layer.ts new file mode 100644 index 0000000..408e015 --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/scripts/build-layer.ts @@ -0,0 +1,105 @@ +// HACK: https://github.com/oven-sh/bun/issues/2081 +process.stdout.getWindowSize = () => [80, 80]; +process.stderr.getWindowSize = () => [80, 80]; + +import { createReadStream, createWriteStream } from "node:fs"; +import { join } from "node:path"; +import { Command, Flags } from "@oclif/core"; +import JSZip from "jszip"; + +export class BuildCommand extends Command { + static summary = "Build a custom Lambda layer for Bun."; + + static flags = { + arch: Flags.string({ + description: "The architecture type to support.", + options: ["x64", "aarch64"], + default: "aarch64", + }), + release: Flags.string({ + description: "The release of Bun to install.", + default: "latest", + }), + url: Flags.string({ + description: "A custom URL to download Bun.", + exclusive: ["release"], + }), + output: Flags.file({ + exists: false, + default: async () => "bun-lambda-layer.zip", + }), + layer: Flags.string({ + description: "The name of the Lambda layer.", + multiple: true, + default: ["bun"], + }), + region: Flags.string({ + description: "The region to publish the layer.", + multiple: true, + default: [], + }), + public: Flags.boolean({ + description: "If the layer should be public.", + default: false, + }), + }; + + async run() { + const result = await this.parse(BuildCommand); + const { flags } = result; + this.debug("Options:", flags); + const { arch, release, url, output } = flags; + const { href } = new URL( + url ?? `https://bun.sh/download/${release}/linux/${arch}?avx2=true`, + ); + this.log("Downloading...", href); + const response = await fetch(href, { + headers: { + "User-Agent": "bun-lambda", + }, + }); + if (response.url !== href) { + this.debug("Redirected URL:", response.url); + } + this.debug("Response:", response.status, response.statusText); + if (!response.ok) { + const reason = await response.text(); + this.error(reason, { exit: 1 }); + } + this.log("Extracting..."); + const buffer = await response.arrayBuffer(); + let archive; + try { + archive = await JSZip.loadAsync(buffer); + } catch (cause) { + this.debug(cause); + this.error("Failed to unzip file:", { exit: 1 }); + } + this.debug("Extracted archive:", Object.keys(archive.files)); + const bun = archive.filter( + (_, { dir, name }) => !dir && name.endsWith("bun"), + )[0]; + if (!bun) { + this.error("Failed to find executable in zip", { exit: 1 }); + } + const cwd = bun.name.split("/")[0]; + archive = archive.folder(cwd) ?? archive; + for (const filename of ["bootstrap", "runtime.ts"]) { + const path = join(__dirname, "..", filename); + archive.file(filename, createReadStream(path)); + } + this.log("Saving...", output); + archive + .generateNodeStream({ + streamFiles: true, + compression: "DEFLATE", + compressionOptions: { + level: 9, + }, + }) + .pipe(createWriteStream(output)); + this.log("Saved"); + } +} + +await BuildCommand.run(process.argv.slice(2)); diff --git a/Homework/cs5260/widgetrequesthandler/scripts/publish-layer.ts b/Homework/cs5260/widgetrequesthandler/scripts/publish-layer.ts new file mode 100644 index 0000000..b7129fc --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/scripts/publish-layer.ts @@ -0,0 +1,91 @@ +import { spawnSync } from "node:child_process"; +import { BuildCommand } from "./build-layer"; + +export class PublishCommand extends BuildCommand { + static summary = "Publish a custom Lambda layer for Bun."; + + #aws(args: string[]): string { + this.debug("$", "aws", ...args); + const { status, stdout, stderr } = spawnSync("aws", args, { + stdio: "pipe", + }); + const result = stdout.toString("utf-8").trim(); + if (status === 0) { + return result; + } + const reason = stderr.toString("utf-8").trim() || result; + throw new Error(`aws ${args.join(" ")} exited with ${status}: ${reason}`); + } + + async run() { + const { flags } = await this.parse(PublishCommand); + this.debug("Options:", flags); + try { + const version = this.#aws(["--version"]); + this.debug("AWS CLI:", version); + } catch (error) { + this.debug(error); + this.error( + "Install the `aws` CLI to continue: https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html", + { exit: 1 }, + ); + } + const { layer, region, arch, output, public: isPublic } = flags; + if (region.includes("*")) { + // prettier-ignore + const result = this.#aws([ + "ec2", + "describe-regions", + "--query", "Regions[].RegionName", + "--output", "json" + ]); + region.length = 0; + for (const name of JSON.parse(result)) { + region.push(name); + } + } else if (!region.length) { + // prettier-ignore + region.push(this.#aws([ + "configure", + "get", + "region" + ])); + } + this.log("Publishing..."); + for (const regionName of region) { + for (const layerName of layer) { + // prettier-ignore + const result = this.#aws([ + "lambda", + "publish-layer-version", + "--layer-name", layerName, + "--region", regionName, + "--description", "Bun is an incredibly fast JavaScript runtime, bundler, transpiler, and package manager.", + "--license-info", "MIT", + "--compatible-architectures", arch === "x64" ? "x86_64" : "arm64", + "--compatible-runtimes", "provided.al2", "provided", + "--zip-file", `fileb://${output}`, + "--output", "json", + ]); + const { LayerVersionArn } = JSON.parse(result); + this.log("Published", LayerVersionArn); + if (isPublic) { + // prettier-ignore + this.#aws([ + "lambda", + "add-layer-version-permission", + "--layer-name", layerName, + "--region", regionName, + "--version-number", LayerVersionArn.split(":").pop(), + "--statement-id", `${layerName}-public`, + "--action", "lambda:GetLayerVersion", + "--principal", "*", + ]); + } + } + } + this.log("Done"); + } +} + +await PublishCommand.run(process.argv.slice(2)); diff --git a/Homework/cs5260/widgetrequesthandler/t/handler.spec.ts b/Homework/cs5260/widgetrequesthandler/t/handler.spec.ts new file mode 100644 index 0000000..77513ee --- /dev/null +++ b/Homework/cs5260/widgetrequesthandler/t/handler.spec.ts @@ -0,0 +1,71 @@ +import { expect, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { WidgetRequestHandler } from "@/widgetrequesthandler/handler"; +import { VoidLogger } from "@/shared/t"; +import { WidgetRequest, WidgetRequestT } from "@/shared/schema"; +import { + ProcessorResponse, + WidgetRequestProcessor, +} from "@/widgetrequesthandler/processors"; + +const testWidget = () => + WidgetRequest.parse({ + type: "create", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + +test("when invalid body then returns 400", async () => { + const logger = new VoidLogger(); + const mockProcessor = {} as unknown as WidgetRequestProcessor; + const request = new Request("http://localhost"); + + const handler = new WidgetRequestHandler(mockProcessor, logger); + const response = await handler.handleRequest(request); + expect(response.status).toBe(400); +}); + +test("when processor fails then returns 500", async () => { + const logger = new VoidLogger(); + const mockProcessor = { + async handle(_widgetRequest: WidgetRequestT) { + throw new Error("failure"); + }, + } as unknown as WidgetRequestProcessor; + + const newWidgetRequest = testWidget(); + const request = new Request("http://localhost", { + body: JSON.stringify({ event: newWidgetRequest }), + }); + + const handler = new WidgetRequestHandler(mockProcessor, logger); + const response = await handler.handleRequest(request); + const responseBody = (await response.json()) as ProcessorResponse; + + expect(response.status).toBe(500); + expect(responseBody.success).toBe(false); +}); + +test("when processor succeeds and correct widget request then returns 200", async () => { + const logger = new VoidLogger(); + const message = randomUUID(); + const mockProcessor = { + async handle(_widgetRequest: WidgetRequestT) { + return { success: true, message }; + }, + } as unknown as WidgetRequestProcessor; + + const newWidgetRequest = testWidget(); + const request = new Request("http://localhost", { + body: JSON.stringify({ event: newWidgetRequest }), + }); + + const handler = new WidgetRequestHandler(mockProcessor, logger); + const response = await handler.handleRequest(request); + const responseBody = (await response.json()) as ProcessorResponse; + + expect(response.status).toBe(200); + expect(responseBody.success).toBe(true); + expect(responseBody.message).toBe(message); +}); |