diff options
| author | Elizabeth Alexander Hunt <me@liz.coffee> | 2026-07-02 11:55:17 -0700 |
|---|---|---|
| committer | Elizabeth Alexander Hunt <me@liz.coffee> | 2026-07-02 11:55:17 -0700 |
| commit | 6bf4b90c90f15f4ab60833bddf5b5756d1a6b1f6 (patch) | |
| tree | ed97e39ec77c5231ffd2c394493e68d00ddac5a4 /Homework/cs5260 | |
| download | misc-undergrad-6bf4b90c90f15f4ab60833bddf5b5756d1a6b1f6.tar.gz misc-undergrad-6bf4b90c90f15f4ab60833bddf5b5756d1a6b1f6.zip | |
Diffstat (limited to 'Homework/cs5260')
55 files changed, 3167 insertions, 0 deletions
diff --git a/Homework/cs5260/Dockerfile.consumer b/Homework/cs5260/Dockerfile.consumer new file mode 100644 index 0000000..2f33170 --- /dev/null +++ b/Homework/cs5260/Dockerfile.consumer @@ -0,0 +1,9 @@ +FROM oven/bun:1.0.7 + +COPY . /app + +WORKDIR /app + +RUN bun install + +CMD /app/consumer_entrypoint.sh diff --git a/Homework/cs5260/Dockerfile.lambda b/Homework/cs5260/Dockerfile.lambda new file mode 100644 index 0000000..c91bd04 --- /dev/null +++ b/Homework/cs5260/Dockerfile.lambda @@ -0,0 +1,9 @@ +FROM oven/bun:1.0.7 + +COPY . /app + +WORKDIR /app + +RUN bun install + +CMD /app/lambda_entrypoint.sh diff --git a/Homework/cs5260/README.md b/Homework/cs5260/README.md new file mode 100644 index 0000000..1196082 --- /dev/null +++ b/Homework/cs5260/README.md @@ -0,0 +1,50 @@ +# cs5260 consumer + +This project uses the [Bun TS Runtime](https://bun.sh/) + +To install dependencies: + +```bash +bun install +``` + +To run: + +```bash +bun run index.ts +``` + +With the arguments: +``` +--sourcebucket: the bucket to poll for widget requests +--storagebucket: the bucket to store and modify widgets according to received requests +--storagetable: the DDB table to store and modify widgets according to received requests +--storageprefix: the prefix to store S3 objects in (only required if storagebucket is given) +``` + +And either one of storagebucket or storagetable must be given. + +Unit tests: +```bash +bun test +``` + +## Lambda Handler +There is a current bug in bun 1.0.8, the latest version: [https://github.com/oven-sh/bun/issues/7028], +which is not present in v1.0.7 and earlier. Thus, I recommend using `asdf` to manage the bun version in +this part of the project. + +To deploy the lambda handler, first build and publish the Bun runtime layer on your +AWS account: +```bash +bun run build-layer --arch aarch64 \ + --release 1.0.7 --output ./bun-lambda-layer.zip +bun run publish-layer --region us-east-1 \ + --arch aarch64 --release 1.0.7 \ + --output ./bun-lambda-layer.zip +``` + +Then, create a Lambda function with runtime "custom bootstrap on amazon linux 2" and +name "widget-api-handler", and handler to "handler.fetch" (found in "Code"). + +In the Lambda console then attach the layer "bun". diff --git a/Homework/cs5260/bun.lockb b/Homework/cs5260/bun.lockb Binary files differnew file mode 100644 index 0000000..6c173bf --- /dev/null +++ b/Homework/cs5260/bun.lockb diff --git a/Homework/cs5260/consumer/Loop.ts b/Homework/cs5260/consumer/Loop.ts new file mode 100644 index 0000000..e8e88d2 --- /dev/null +++ b/Homework/cs5260/consumer/Loop.ts @@ -0,0 +1,47 @@ +import { TracingLogger } from "./log"; +import { WidgetRequestProcessor } from "./processor"; +import { WidgetRequestRetriever } from "./retrievers"; +import { WidgetRequestT } from "./schema"; + +const sleepFor = (ms: number) => new Promise((res) => setTimeout(res, ms)); + +export const MainLoop = async ( + retrievers: WidgetRequestRetriever[], + processors: WidgetRequestProcessor[], + maxPollAttempts: number, + pollIntervalMs: number, + logger: TracingLogger, +) => { + let pollAttemptsWithNoResults = 0; + while (pollAttemptsWithNoResults < maxPollAttempts) { + logger.info(`polling...`); + + let requests: WidgetRequestT[] = []; + for (const retriever of retrievers) { + const retrievalRequests = await retriever.refreshRequests(); + requests = requests.concat(retrievalRequests); + } + logger.info(`received Requests=(${requests.length}) requests...`); + + for (const request of requests) { + for (const processor of processors) { + await processor.processRequest(request); + } + } + logger.info( + `finished processing Requests=(${requests.length}) requests...`, + ); + + if (requests.length === 0) { + pollAttemptsWithNoResults++; + logger.info( + `sleeping for Interval=(${pollIntervalMs})ms: (${pollAttemptsWithNoResults} / ${maxPollAttempts})`, + ); + await sleepFor(pollIntervalMs); + } else { + pollAttemptsWithNoResults = 0; + } + } + + logger.info(`no requests after Attempts=(${maxPollAttempts})...`); +}; diff --git a/Homework/cs5260/consumer/args.ts b/Homework/cs5260/consumer/args.ts new file mode 100644 index 0000000..0b9a0bd --- /dev/null +++ b/Homework/cs5260/consumer/args.ts @@ -0,0 +1,13 @@ +import parseArgs from "minimist"; + +export type Args = { + _: any; + storagetable?: string; + storagebucket?: string; + storageprefix?: string; + + sourcebucket?: string; + sourcequeue?: string; +}; + +export const args = parseArgs(process.argv.slice(2)) as Args; diff --git a/Homework/cs5260/consumer/dao/DynamoWidgetDAO.ts b/Homework/cs5260/consumer/dao/DynamoWidgetDAO.ts new file mode 100644 index 0000000..a90788f --- /dev/null +++ b/Homework/cs5260/consumer/dao/DynamoWidgetDAO.ts @@ -0,0 +1,100 @@ +import { Widget, WidgetT } from "@/shared/schema"; +import { WidgetDAO } from "."; +import { TracingLogger } from "@/consumer/log"; +import { + AttributeValue, + DeleteItemCommand, + DynamoDBClient, + GetItemCommand, + PutItemCommand, +} from "@aws-sdk/client-dynamodb"; + +export class DynamoWidgetDAO implements WidgetDAO { + private table: string; + private ddbClient: DynamoDBClient; + private logger: TracingLogger; + + constructor(table: string, ddbClient: DynamoDBClient, logger: TracingLogger) { + this.table = table; + this.ddbClient = ddbClient; + this.logger = logger; + } + + public async retrieve(widget: WidgetT) { + const getReq = new GetItemCommand({ + TableName: this.table, + Key: { + id: { + S: widget.id, + }, + }, + }); + + const response = await this.ddbClient.send(getReq); + if (response.Item) { + this.logger.info(`sucessfully retrieved Widget=(${widget.id})`); + return this.deflate(response.Item); + } + } + + public async save(widget: WidgetT) { + const putReq = new PutItemCommand({ + TableName: this.table, + Item: this.inflate(widget), + }); + + await this.ddbClient.send(putReq); + + this.logger.info(`sucessfully saved Widget=(${widget.id})`); + return widget; + } + + public async delete(widget: WidgetT) { + const deleteReq = new DeleteItemCommand({ + TableName: this.table, + Key: { + id: { S: widget.id }, + }, + }); + + await this.ddbClient.send(deleteReq); + + this.logger.info(`sucessfully deleted Widget=(${widget.id})`); + return widget; + } + + private inflate(widget: WidgetT): Record<string, AttributeValue> { + const additionalAttributes = + widget.otherAttributes?.reduce((acc, { name, value }) => { + acc[name] = { S: value }; + return acc; + }, {} as Record<string, { S: string }>) ?? {}; + if (widget.label) additionalAttributes["label"] = { S: widget.label }; + if (widget.description) + additionalAttributes["description"] = { S: widget.description }; + + return { + id: { S: widget.id }, + owner: { S: widget.owner }, + ...additionalAttributes, + }; + } + + private deflate(item: Record<string, AttributeValue>): WidgetT { + const respWidget = Object.keys(item).reduce((widge, key: string) => { + const attributeValue = item![key]; + if (["id", "owner", "label", "description"].includes(key)) { + widge[key] = attributeValue.S; + return widge; + } + + const attribute = { name: key, value: attributeValue.S! }; + widge["otherAttributes"] = widge["otherAttributes"]?.concat([ + attribute, + ]) ?? [attribute]; + + return widge; + }, {} as Record<string, any>); + return Widget.parse(respWidget); + } +} diff --git a/Homework/cs5260/consumer/dao/S3WidgetDao.ts b/Homework/cs5260/consumer/dao/S3WidgetDao.ts new file mode 100644 index 0000000..d5aee72 --- /dev/null +++ b/Homework/cs5260/consumer/dao/S3WidgetDao.ts @@ -0,0 +1,93 @@ +import { WidgetT, Widget } from "@/shared/schema"; +import { WidgetDAO } from "."; +import { + DeleteObjectCommand, + GetObjectCommand, + NoSuchKey, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3"; +import { S3WidgetKeyGenerator } from "./keys"; +import { TracingLogger } from "@/consumer/log"; + +export class S3WidgetDAO implements WidgetDAO { + private bucket: string; + private s3Client: S3Client; + private widgetKeyGenerator: S3WidgetKeyGenerator; + private logger: TracingLogger; + + constructor( + bucket: string, + s3Client: S3Client, + widgetKeyGenerator: S3WidgetKeyGenerator, + logger: TracingLogger + ) { + this.bucket = bucket; + this.s3Client = s3Client; + this.widgetKeyGenerator = widgetKeyGenerator; + this.logger = logger; + } + + public async retrieve(widget: WidgetT) { + const key = this.widgetKeyGenerator.fromWidget(widget); + + this.logger.info( + `getting Widget=(${widget.id}) in Bucket=(${this.bucket})` + ); + + const retrievalRequest = new GetObjectCommand({ + Bucket: this.bucket, + Key: key, + }); + + try { + const objectResp = await this.s3Client.send(retrievalRequest); + const textRepr = await objectResp.Body?.transformToString(); + + if (textRepr) { + this.logger.info(`retrieved WidgetObject=(${widget.id}) from S3`); + return Widget.parse(JSON.parse(textRepr)); + } + } catch (e: unknown) { + if (e instanceof NoSuchKey) { + return; + } + throw e; + } + } + + public async save(widget: WidgetT) { + const key = this.widgetKeyGenerator.fromWidget(widget); + + this.logger.info(`saving Widget=(${widget.id}) in Bucket=(${this.bucket})`); + + const putReq = new PutObjectCommand({ + Bucket: this.bucket, + Key: key, + Body: JSON.stringify(widget, null, 2), + }); + + await this.s3Client.send(putReq); + + this.logger.info(`successfully saved Widget=(${widget.id}) in S3`); + return widget; + } + + public async delete(widget: WidgetT) { + const key = this.widgetKeyGenerator.fromWidget(widget); + + this.logger.info( + `deleting Widget=(${widget.id}) in Bucket=(${this.bucket})` + ); + + const deleteReq = new DeleteObjectCommand({ + Key: key, + Bucket: this.bucket, + }); + + await this.s3Client.send(deleteReq); + + this.logger.info(`successfully deleted Widget=(${widget.id}) in S3`); + return widget; + } +} diff --git a/Homework/cs5260/consumer/dao/WidgetDAO.ts b/Homework/cs5260/consumer/dao/WidgetDAO.ts new file mode 100644 index 0000000..2b7f3d6 --- /dev/null +++ b/Homework/cs5260/consumer/dao/WidgetDAO.ts @@ -0,0 +1,7 @@ +import { WidgetT } from "@/shared/schema"; + +export interface WidgetDAO { + retrieve(widget: WidgetT): Promise<WidgetT | undefined>; + save(widget: WidgetT): Promise<WidgetT>; + delete(widget: WidgetT): Promise<WidgetT>; +} diff --git a/Homework/cs5260/consumer/dao/index.ts b/Homework/cs5260/consumer/dao/index.ts new file mode 100644 index 0000000..82e158c --- /dev/null +++ b/Homework/cs5260/consumer/dao/index.ts @@ -0,0 +1,3 @@ +export * from "./WidgetDAO"; +export * from "./S3WidgetDao"; +export * from "./DynamoWidgetDAO"; diff --git a/Homework/cs5260/consumer/dao/keys/S3WidgetKeyGenerator.ts b/Homework/cs5260/consumer/dao/keys/S3WidgetKeyGenerator.ts new file mode 100644 index 0000000..2e4c254 --- /dev/null +++ b/Homework/cs5260/consumer/dao/keys/S3WidgetKeyGenerator.ts @@ -0,0 +1,43 @@ +import { WidgetT } from "@/shared/schema"; + +export class S3WidgetKeyGenerator { + private prefix: string = ""; + private owner: string = ""; + private widgetId: string = ""; + + public withPrefix(prefix: string): S3WidgetKeyGenerator { + this.prefix = prefix; + return this; + } + + public withOwner(owner: string): S3WidgetKeyGenerator { + this.owner = owner; + return this; + } + + public withWidgetId(widgetId: string): S3WidgetKeyGenerator { + this.widgetId = widgetId; + return this; + } + + public build(): string { + const parts = [ + this.prefix, + this.owner.replaceAll(" ", "-").toLowerCase(), + this.widgetId, + ]; + if (parts.some((part) => !part || !part.match(/^[a-zA-Z0-9!\-_.*'()]+$/))) { + throw new Error("Invalid Key Specification"); + } + return parts.join("/"); + } + + public fromWidget(widget: WidgetT) { + const keyBuilder = new S3WidgetKeyGenerator(); + return keyBuilder + .withPrefix(this.prefix) + .withOwner(widget.owner) + .withWidgetId(widget.id) + .build(); + } +} diff --git a/Homework/cs5260/consumer/dao/keys/index.ts b/Homework/cs5260/consumer/dao/keys/index.ts new file mode 100644 index 0000000..c7c8467 --- /dev/null +++ b/Homework/cs5260/consumer/dao/keys/index.ts @@ -0,0 +1 @@ +export * from "./S3WidgetKeyGenerator"; diff --git a/Homework/cs5260/consumer/index.ts b/Homework/cs5260/consumer/index.ts new file mode 100644 index 0000000..f8ce640 --- /dev/null +++ b/Homework/cs5260/consumer/index.ts @@ -0,0 +1,113 @@ +import { S3Client } from "@aws-sdk/client-s3"; +import { SQSClient } from "@aws-sdk/client-sqs"; +import { DynamoDBClient } from "@aws-sdk/client-dynamodb"; + +import { Args, args } from "@/consumer/args"; +import { DynamoWidgetDAO, S3WidgetDAO } from "@/consumer/dao"; +import { S3WidgetKeyGenerator } from "@/consumer/dao/keys"; +import { ConsoleTracingLogger } from "@/shared/log"; +import { ConsumerRequestProcessor, WidgetRequestProcessor } from "@/consumer/processor"; +import { + S3WidgetRequestRetriever, + WidgetRequestRetriever, + SQSWidgetRequestRetriever, +} from "@/consumer/retrievers"; +import { MainLoop } from "@/consumer/Loop"; + +const main = async (args: Args) => { + const mainLogger = new ConsoleTracingLogger("main"); + + const s3Client = new S3Client(); + const sqsClient = new SQSClient(); + const dynamoClient = new DynamoDBClient(); + + const retrievers: WidgetRequestRetriever[] = []; + if (!args.sourcebucket && !args.sourcequeue) { + throw new Error("Must specify source to poll."); + } + + if (args.sourcebucket) { + mainLogger.info( + `constructing s3 retriever for Bucket=(${args.sourcebucket})`, + ); + + const retriever = new S3WidgetRequestRetriever( + args.sourcebucket, + s3Client, + mainLogger.createChild("S3WidgetRequestRetriever"), + ); + retrievers.push(retriever); + } + + if (args.sourcequeue) { + mainLogger.info( + `constructing sqs retriever for Bucket=(${args.sourcequeue})`, + ); + + const retriever = new SQSWidgetRequestRetriever( + args.sourcequeue, + sqsClient, + mainLogger.createChild("SQSWidgetRequestRetriever"), + ); + retrievers.push(retriever); + } + + if (!args.storagebucket && !args.storagetable) { + throw new Error("Must specify a target data store for Widgets"); + } + + const processors: WidgetRequestProcessor[] = []; + + if (args.storagebucket) { + if (!args.storageprefix) { + throw new Error("must specify widget prefix location for storage bucket"); + } + + mainLogger.info( + `constructing s3 processor for Bucket=(${args.storagebucket})`, + ); + + const s3KeyGenerator = new S3WidgetKeyGenerator().withPrefix( + args.storageprefix, + ); + const s3WidgetDao = new S3WidgetDAO( + args.storagebucket, + s3Client, + s3KeyGenerator, + mainLogger.createChild("S3WidgetDAO"), + ); + + const s3RequestProcessor = new ConsumerRequestProcessor( + s3WidgetDao, + mainLogger.createChild("ConsumerRequestProcessor => (S3)"), + ); + + processors.push(s3RequestProcessor); + } + + if (args.storagetable) { + mainLogger.info( + `constructing dynamo processor for Table=(${args.storagetable})`, + ); + + const dynamoWidgetDao = new DynamoWidgetDAO( + args.storagetable, + dynamoClient, + mainLogger.createChild("DynamoWidgetDAO"), + ); + + const dynamoRequestProcessor = new ConsumerRequestProcessor( + dynamoWidgetDao, + mainLogger.createChild("ConsumerRequestProcessor => (Dynamo)"), + ); + + processors.push(dynamoRequestProcessor); + } + + const maxPollAttempts = 15; + await MainLoop(retrievers, processors, maxPollAttempts, 200, mainLogger); + + mainLogger.info("EXITING!"); +}; + +main(args); diff --git a/Homework/cs5260/consumer/processor/ConsumerRequestProcessor.ts b/Homework/cs5260/consumer/processor/ConsumerRequestProcessor.ts new file mode 100644 index 0000000..87e66c1 --- /dev/null +++ b/Homework/cs5260/consumer/processor/ConsumerRequestProcessor.ts @@ -0,0 +1,133 @@ +import { WidgetDAO } from "@/consumer/dao"; +import { TracingLogger } from "@/consumer/log"; +import { Widget, WidgetT, WidgetRequestT } from "@/shared/schema"; + +export class ConsumerRequestProcessor implements WidgetRequestProcessor { + private widgetDao: WidgetDAO; + private logger: TracingLogger; + + constructor(widgetDao: WidgetDAO, logger: TracingLogger) { + this.widgetDao = widgetDao; + this.logger = logger; + } + + public async processRequest(widgetRequest: WidgetRequestT) { + this.logger.info( + `processing RequestId=(${widgetRequest.requestId}) with WidgetRequestType=(${widgetRequest.type})` + ); + switch (widgetRequest.type) { + case "create": + await this.createWidget(widgetRequest); + break; + case "delete": + await this.deleteWidget(widgetRequest); + break; + case "update": + await this.updateWidget(widgetRequest); + break; + default: + this.logger.warn( + `received WidgetRequestType=(${widgetRequest.type})... don't know what to do!` + ); + break; + } + } + + private async createWidget(widgetRequest: WidgetRequestT) { + this.logger.info(`creating new Widget=(${widgetRequest.widgetId})`); + const widget = this.widgetFromWidgetRequest(widgetRequest); + + const currentWidget = await this.widgetDao.retrieve(widget); + + if (!currentWidget) { + await this.widgetDao.save(widget); + } else { + this.logger.warn( + `received creation request for Widget=(${widget.id}) but it already exists... skipping` + ); + } + } + + private async deleteWidget(widgetRequest: WidgetRequestT) { + this.logger.info(`deleting Widget=(${widgetRequest.widgetId})`); + const widget = this.widgetFromWidgetRequest(widgetRequest); + + const currentWidget = await this.widgetDao.retrieve(widget); + + if (currentWidget) { + await this.widgetDao.delete(widget); + } else { + this.logger.warn( + `received deletion request for Widget=(${widget.id}) but it did not exist... skipping` + ); + } + } + + private async updateWidget(widgetRequest: WidgetRequestT) { + this.logger.info(`updating Widget=(${widgetRequest.widgetId})`); + const widget = this.widgetFromWidgetRequest(widgetRequest); + + const currentWidget = await this.widgetDao.retrieve(widget); + if (!currentWidget) { + this.logger.warn( + `attempted to retrieve Widget=(${widgetRequest.widgetId}) to update, but it doesn't exist! skipping...` + ); + return; + } + + const immutableKeys = ["id", "owner"] as unknown as (keyof WidgetT)[]; + + for (const key of Object.keys(widget) as unknown as (keyof WidgetT)[]) { + const isImmutable = immutableKeys.includes(key); + if (!isImmutable && widget[key] === "") { + delete widget[key]; + delete currentWidget[key]; + } else if (isImmutable && widget[key] !== currentWidget[key]) { + this.logger.warn( + `cannot change ${key} on Widget=(${widget.id}). skipping...` + ); + return; + } + } + + const otherAttributeMap = new Map<string, string>(); + [ + ...(currentWidget.otherAttributes ?? []), + ...(widget.otherAttributes ?? []), + ].forEach(({ name, value }) => { + if (value != null) otherAttributeMap.set(name, value); + if (value == "") otherAttributeMap.delete(name); + }); + const otherAttributes = Array.from(otherAttributeMap.entries()).map( + ([name, value]) => { + return { name, value }; + } + ); + + const newWidget: WidgetT = { + ...currentWidget, + ...widget, + }; + + if (otherAttributes.length) { + newWidget.otherAttributes = otherAttributes; + } + + await this.widgetDao.save(newWidget); + } + + private widgetFromWidgetRequest(widgetRequest: WidgetRequestT): WidgetT { + const widget: Record<string, any> = { + id: widgetRequest.widgetId, + owner: widgetRequest.owner, + }; + + if (widgetRequest.label !== undefined) widget.label = widgetRequest.label; + if (widgetRequest.description !== undefined) + widget.description = widgetRequest.description; + if (widgetRequest.otherAttributes?.length) + widget.otherAttributes = widgetRequest.otherAttributes; + + return Widget.parse(widget); + } +} diff --git a/Homework/cs5260/consumer/processor/WidgetRequestProcessor.ts b/Homework/cs5260/consumer/processor/WidgetRequestProcessor.ts new file mode 100644 index 0000000..fbb96ac --- /dev/null +++ b/Homework/cs5260/consumer/processor/WidgetRequestProcessor.ts @@ -0,0 +1,5 @@ +import { WidgetRequestT } from "@/shared/schema"; + +export interface WidgetRequestProcessor { + processRequest(widgetRequest: WidgetRequestT): Promise<void>; +} diff --git a/Homework/cs5260/consumer/processor/index.ts b/Homework/cs5260/consumer/processor/index.ts new file mode 100644 index 0000000..983da1b --- /dev/null +++ b/Homework/cs5260/consumer/processor/index.ts @@ -0,0 +1,2 @@ +export * from "./ConsumerRequestProcessor"; +export * from "./WidgetRequestProcessor"; diff --git a/Homework/cs5260/consumer/retrievers/S3WidgetRequestRetriever.ts b/Homework/cs5260/consumer/retrievers/S3WidgetRequestRetriever.ts new file mode 100644 index 0000000..786e0a6 --- /dev/null +++ b/Homework/cs5260/consumer/retrievers/S3WidgetRequestRetriever.ts @@ -0,0 +1,64 @@ +import { + DeleteObjectCommand, + GetObjectCommand, + ListObjectsV2Command, + S3Client, +} from "@aws-sdk/client-s3"; +import { WidgetRequestRetriever } from "."; +import { WidgetRequest } from "@/shared/schema"; +import { TracingLogger } from "@/consumer/log"; + +export class S3WidgetRequestRetriever implements WidgetRequestRetriever { + static LIST_KEY_MAX = 1; + private bucket: string; + private s3Client: S3Client; + private logger: TracingLogger; + + constructor(bucket: string, s3Client: S3Client, logger: TracingLogger) { + this.bucket = bucket; + this.s3Client = s3Client; + this.logger = logger; + } + + private async getWidgetRequest(key: string) { + const getCommand = new GetObjectCommand({ + Bucket: this.bucket, + Key: key, + }); + + const response = await this.s3Client.send(getCommand); + const body = await response.Body?.transformToString(); + + if (body && body.length) return WidgetRequest.parse(JSON.parse(body)); + } + + public async refreshRequests() { + // ListObjectsV2 returns keys in ascending lexicographical order + const listCommand = new ListObjectsV2Command({ + MaxKeys: S3WidgetRequestRetriever.LIST_KEY_MAX, + Bucket: this.bucket, + }); + const response = await this.s3Client.send(listCommand); + + const key = response.Contents && response.Contents[0]?.Key; + + if (!key) { + this.logger.info(`No more keys in Bucket=(${this.bucket}).`); + return []; + } + + const widgetRequest = await this.getWidgetRequest(key); + + const deleteWidgetRequest = new DeleteObjectCommand({ + Key: key, + Bucket: this.bucket, + }); + await this.s3Client.send(deleteWidgetRequest); + + if (!widgetRequest) { + this.logger.warn(`No valid widget at Key=(${key}).`); + return []; + } + return [widgetRequest]; + } +} diff --git a/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts b/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts new file mode 100644 index 0000000..40a837b --- /dev/null +++ b/Homework/cs5260/consumer/retrievers/SQSWidgetRequestRetriever.ts @@ -0,0 +1,60 @@ +import { + DeleteMessageBatchCommand, + ReceiveMessageCommand, + SQSClient, +} from "@aws-sdk/client-sqs"; +import { WidgetRequestRetriever } from "."; +import { TracingLogger } from "@/shared/log"; +import { WidgetRequest, WidgetRequestT } from "@/shared/schema"; + +export class SQSWidgetRequestRetriever implements WidgetRequestRetriever { + static MESSAGES_PER_POLL = 10; + private queueUrl: string; + private sqsClient: SQSClient; + private logger: TracingLogger; + + constructor(queueUrl: string, sqsClient: SQSClient, logger: TracingLogger) { + this.queueUrl = queueUrl; + this.sqsClient = sqsClient; + this.logger = logger; + } + + public async refreshRequests() { + const retrieveMessagesRequest = new ReceiveMessageCommand({ + QueueUrl: this.queueUrl, + MaxNumberOfMessages: SQSWidgetRequestRetriever.MESSAGES_PER_POLL, + }); + + this.logger.info( + `looking for Messages=(${SQSWidgetRequestRetriever.MESSAGES_PER_POLL}) WidgetRequest's from Queue=(${this.queueUrl})`, + ); + const response = await this.sqsClient.send(retrieveMessagesRequest); + if (!response.Messages) { + this.logger.info(`Queue=(${this.queueUrl}) returned no messages!`); + return []; + } + + const widgetRequests = response.Messages.map((message) => { + if (message.Body) return WidgetRequest.parse(JSON.parse(message.Body)); + }).filter((x) => x) as WidgetRequestT[]; + this.logger.info( + `found WidgetRequests=(${widgetRequests.length}) requests`, + ); + + const messageDeleteInputs = response.Messages.map( + ({ ReceiptHandle, MessageId }) => { + return { Id: MessageId, ReceiptHandle }; + }, + ); + this.logger.info( + `requesting deletion of Inputs=(${messageDeleteInputs.length})`, + ); + const deleteRequestCommand = new DeleteMessageBatchCommand({ + QueueUrl: this.queueUrl, + Entries: messageDeleteInputs, + }); + await this.sqsClient.send(deleteRequestCommand); + + return widgetRequests; + } +} diff --git a/Homework/cs5260/consumer/retrievers/WidgetRequestRetriever.ts b/Homework/cs5260/consumer/retrievers/WidgetRequestRetriever.ts new file mode 100644 index 0000000..f5f5775 --- /dev/null +++ b/Homework/cs5260/consumer/retrievers/WidgetRequestRetriever.ts @@ -0,0 +1,6 @@ +import { WidgetRequestT } from "@/shared/schema"; + +export interface WidgetRequestRetriever { + /* get new requests and delete them */ + refreshRequests(): Promise<WidgetRequestT[]>; +} diff --git a/Homework/cs5260/consumer/retrievers/index.ts b/Homework/cs5260/consumer/retrievers/index.ts new file mode 100644 index 0000000..1bfc014 --- /dev/null +++ b/Homework/cs5260/consumer/retrievers/index.ts @@ -0,0 +1,3 @@ +export * from "./WidgetRequestRetriever"; +export * from "./S3WidgetRequestRetriever"; +export * from "./SQSWidgetRequestRetriever"; diff --git a/Homework/cs5260/consumer/t/ConsumerRequestProcessor.spec.ts b/Homework/cs5260/consumer/t/ConsumerRequestProcessor.spec.ts new file mode 100644 index 0000000..28d785b --- /dev/null +++ b/Homework/cs5260/consumer/t/ConsumerRequestProcessor.spec.ts @@ -0,0 +1,200 @@ +import { randomUUID } from "node:crypto"; +import { Widget, WidgetRequest, WidgetT } from "@/shared/schema"; +import { expect, mock, test } from "bun:test"; +import { ConsumerRequestProcessor } from "@/consumer/processor"; +import { VoidLogger } from "@/shared/t"; + +const widgetDAOFactory = (): any => { + return { + retrieve: mock(async (widget: WidgetT) => { + return widget; + }), + save: mock(async (widget: WidgetT) => { + return widget; + }), + delete: mock(async (widget: WidgetT) => { + return widget; + }), + }; +}; + +test("when create request then warns when already existant and not save", async () => { + const logger = new VoidLogger(); + logger.warn = mock((_s) => {}); + + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (widget: WidgetT) => widget); + + const newWidgetRequest = WidgetRequest.parse({ + type: "create", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(newWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save).toHaveBeenCalledTimes(0); + expect(logger.warn).toHaveBeenCalledTimes(1); +}); + +test("when create request then saves in dao if not exist", async () => { + const logger = new VoidLogger(); + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (_widget: WidgetT) => { + return undefined; + }); + + const newWidgetRequest = WidgetRequest.parse({ + type: "create", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(newWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save).toHaveBeenCalledTimes(1); +}); + +test("when delete request then fails and warns if not exists", async () => { + const logger = new VoidLogger(); + logger.warn = mock((_s) => {}); + + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (_widget: WidgetT) => { + return undefined; + }); + + const deleteWidgetRequest = WidgetRequest.parse({ + type: "delete", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(deleteWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.delete).toHaveBeenCalledTimes(0); + expect(logger.warn).toHaveBeenCalledTimes(1); +}); + +test("when delete request then success if exists", async () => { + const logger = new VoidLogger(); + + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (widget: WidgetT) => { + return widget; + }); + + const deleteWidgetRequest = WidgetRequest.parse({ + type: "delete", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(deleteWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.delete).toHaveBeenCalledTimes(1); +}); + +test("when update request then deletes empty string values", async () => { + const logger = new VoidLogger(); + + const testWidget = Widget.parse({ + id: "test-widget", + owner: "simponic", + label: "current label", + }); + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (_widget: WidgetT) => testWidget); + + const updateWidgetRequest = WidgetRequest.parse({ + type: "update", + requestId: randomUUID(), + owner: "simponic", + widgetId: testWidget.id, + label: "", + }); + + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(updateWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save.mock.calls[0][0]).toEqual( + Widget.parse({ id: "test-widget", owner: "simponic" }), // "label" has been removed + ); +}); + +test("when update attempts to overwrite owner then update fails and warns", async () => { + const logger = new VoidLogger(); + logger.warn = mock((_s) => {}); + + const testWidget = Widget.parse({ + id: "test-widget", + owner: "simponic", + label: "current label", + }); + + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (_widget: WidgetT) => testWidget); + + const newWidgetRequest = WidgetRequest.parse({ + type: "update", + requestId: randomUUID(), + owner: "newOwner", + widgetId: testWidget.id, + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(newWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save).toHaveBeenCalledTimes(0); + expect(logger.warn).toHaveBeenCalledTimes(1); +}); + +test("when update mutable attribute field unspecified then not overwritten", async () => { + const logger = new VoidLogger(); + + const testWidget = Widget.parse({ + id: "test-widget", + owner: "simponic", + label: "non-overwritten label", + }); + + const mockWidgetDAO = widgetDAOFactory(); + mockWidgetDAO.retrieve = mock(async (_widget: WidgetT) => testWidget); + + const newWidgetRequest = WidgetRequest.parse({ + type: "update", + requestId: randomUUID(), + owner: testWidget.owner, + widgetId: testWidget.id, + description: "", + }); + const processor = new ConsumerRequestProcessor(mockWidgetDAO, logger); + + await processor.processRequest(newWidgetRequest); + + expect(mockWidgetDAO.retrieve).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save).toHaveBeenCalledTimes(1); + expect(mockWidgetDAO.save.mock.calls[0][0]).toEqual( + Widget.parse({ + id: "test-widget", + owner: "simponic", + label: "non-overwritten label", + }), + ); +}); diff --git a/Homework/cs5260/consumer/t/DynamoWidgetDAO.spec.ts b/Homework/cs5260/consumer/t/DynamoWidgetDAO.spec.ts new file mode 100644 index 0000000..01ea185 --- /dev/null +++ b/Homework/cs5260/consumer/t/DynamoWidgetDAO.spec.ts @@ -0,0 +1,117 @@ +import { Widget } from "@/shared/schema"; +import { expect, test, mock } from "bun:test"; +import { + DeleteItemCommand, + DynamoDBClient, + GetItemCommand, + PutItemCommand, +} from "@aws-sdk/client-dynamodb"; +import { DynamoWidgetDAO } from "@/consumer/dao"; +import { VoidLogger } from "@/shared/t"; + +const testTable = "widgets-t"; +const testWidget = Widget.parse({ + id: "testWidgetId", + owner: "simponic", + label: "A Widget", + description: "Nothing to see here!", + otherAttributes: [{ name: "length", value: "20cm" }], +}); + +const mockGetItemSend = mock(async (req: GetItemCommand) => { + if (req.input.Key!.id.S! === testWidget.id) { + return { + Item: { + id: { S: testWidget.id }, + owner: { S: testWidget.owner }, + label: { S: testWidget.label }, + description: { S: testWidget.description }, + length: { S: "20cm" }, + }, + }; + } + return {}; +}); + +const expectMockPutItem = (field: string, value: string) => + mock(async (req: PutItemCommand) => { + if (req.input.Item![field].S === value) { + return { Attributes: true }; + } + throw new Error("invalid field"); + }); + +const mockDeleteItemSuccessful = mock(async (req: DeleteItemCommand) => { + if (req.input!.Key!.id.S === testWidget.id) { + return { $metadata: true }; + } +}); + +test("when retrieve dynamo dao item then inflates and returns retrieved item", async () => { + const mockClient = { send: mockGetItemSend }; + const logger = new VoidLogger(); + const dao = new DynamoWidgetDAO( + testTable, + mockClient as unknown as DynamoDBClient, + logger, + ); + + const resp = await dao.retrieve(testWidget); + expect(resp).toEqual(testWidget); +}); + +test("when retrieve dynamo dao item not exist then returns undefined", async () => { + const mockClient = { send: mockGetItemSend }; + const logger = new VoidLogger(); + const dao = new DynamoWidgetDAO( + testTable, + mockClient as unknown as DynamoDBClient, + logger, + ); + + const resp = await dao.retrieve({ + id: "this-widget-doesnt-exist", + owner: "simponic", + }); + + expect(resp).toBeUndefined(); +}); + +test("when dynamo dao save then inflates object and returns", async () => { + const mockClient = { + send: expectMockPutItem("new_attribute", "exists"), + }; + const logger = new VoidLogger(); + const dao = new DynamoWidgetDAO( + testTable, + mockClient as unknown as DynamoDBClient, + logger, + ); + + const newWidget = Widget.parse({ + ...testWidget, + otherAttributes: [...testWidget.otherAttributes!], + }); + newWidget.otherAttributes!.push({ name: "new_attribute", value: "exists" }); + + const resp = await dao.save(newWidget); + + expect(resp).toEqual(newWidget); +}); + +test("when dynamo dao delete then returns deleted ", async () => { + const mockClient = { + send: mockDeleteItemSuccessful, + }; + const logger = new VoidLogger(); + + const dao = new DynamoWidgetDAO( + testTable, + mockClient as unknown as DynamoDBClient, + logger, + ); + + const resp = await dao.delete(testWidget); + + expect(resp).toEqual(testWidget); +}); diff --git a/Homework/cs5260/consumer/t/Loop.spec.ts b/Homework/cs5260/consumer/t/Loop.spec.ts new file mode 100644 index 0000000..44ba517 --- /dev/null +++ b/Homework/cs5260/consumer/t/Loop.spec.ts @@ -0,0 +1,48 @@ +import { MainLoop } from "@/consumer/Loop"; +import { WidgetRequestProcessor } from "@/consumer/processor"; +import { WidgetRequestRetriever } from "@/consumer/retrievers"; +import { expect, mock, test } from "bun:test"; +import { VoidLogger } from "@/shared/t"; +import { WidgetRequest } from "@/shared/schema"; +import { randomUUID } from "node:crypto"; + +test("main loop exits after max poll attempts", async () => { + const logger = new VoidLogger(); + + const maxPollAttempts = 3; + const timesToGiveData = 5; + let timesGivenData = 0; + + const mockRetriever: WidgetRequestRetriever = { + refreshRequests: mock(async () => { + if (timesGivenData < timesToGiveData) { + timesGivenData++; + return [ + WidgetRequest.parse({ + type: "create", + owner: "simponic", + widgetId: randomUUID(), + requestId: randomUUID(), + }), + ]; + } + return []; + }), + }; + const mockProcessor: WidgetRequestProcessor = { + processRequest: mock(async (_r) => {}), + }; + + await MainLoop( + [mockRetriever], + [mockProcessor], + maxPollAttempts, + 200, + logger, + ); + + expect(mockRetriever.refreshRequests).toHaveBeenCalledTimes( + timesGivenData + maxPollAttempts, + ); + expect(mockProcessor.processRequest).toHaveBeenCalledTimes(timesGivenData); +}); diff --git a/Homework/cs5260/consumer/t/S3WidgetDao.spec.ts b/Homework/cs5260/consumer/t/S3WidgetDao.spec.ts new file mode 100644 index 0000000..5eb4602 --- /dev/null +++ b/Homework/cs5260/consumer/t/S3WidgetDao.spec.ts @@ -0,0 +1,115 @@ +import { Widget } from "@/shared/schema"; +import { expect, test, mock } from "bun:test"; +import { S3WidgetKeyGenerator } from "@/consumer/dao/keys"; +import { + DeleteObjectCommand, + GetObjectCommand, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3"; +import { S3WidgetDAO } from "@/consumer/dao"; +import { VoidLogger } from "@/shared/t"; + +const testBucket = "test-bucket"; +const testWidget = Widget.parse({ + id: "testWidgetId", + owner: "simponic", + label: "A Widget", + description: "Nothing to see here!", +}); + +const testKeyGenerator = new S3WidgetKeyGenerator().withPrefix("test-widgets"); + +const mockedGetSend = mock(async (req: GetObjectCommand) => { + if (req.input.Key == "test-widgets/simponic/testWidgetId") { + return { + Body: { + transformToString: () => { + return JSON.stringify(testWidget); + }, + }, + }; + } + return {}; +}); + +const mockedPutSendSuccessful = mock(async (req: PutObjectCommand) => { + if (req.input.Key == "test-widgets/simponic/testWidgetId") { + return { + ChecksumSHA256: "laksdjf-lkejfoaj", + }; + } +}); + +const mockedDeleteSendSuccessful = mock(async (req: DeleteObjectCommand) => { + if (req.input.Key == "test-widgets/simponic/testWidgetId") { + return { + $metadata: true, + }; + } +}); + +test("when given widget then retrieves from s3", async () => { + const mockedClient = { send: mockedGetSend }; + const logger = new VoidLogger(); + + const dao = new S3WidgetDAO( + testBucket, + mockedClient as unknown as S3Client, + testKeyGenerator, + logger, + ); + + const resp = await dao.retrieve({ id: "testWidgetId", owner: "simponic" }); + + expect(resp?.label).toEqual(testWidget.label); + expect(resp?.id).toEqual(testWidget.id); +}); + +test("when cannot retrieve widget then returns undefined", async () => { + const mockedClient = { send: mockedGetSend }; + const logger = new VoidLogger(); + + const dao = new S3WidgetDAO( + testBucket, + mockedClient as unknown as S3Client, + testKeyGenerator, + logger, + ); + + const resp = await dao.retrieve({ + id: "thisWidgetDoesNotExist", + owner: "simponic", + }); + + expect(resp).toBeUndefined(); +}); + +test("when given widget then puts in s3 and returns", async () => { + const mockedClient = { send: mockedPutSendSuccessful }; + const logger = new VoidLogger(); + const dao = new S3WidgetDAO( + testBucket, + mockedClient as unknown as S3Client, + testKeyGenerator, + logger, + ); + + const resp = await dao.save(testWidget); + expect(resp).toEqual(testWidget); +}); + +test("when given widget then deletes in s3 and returns", async () => { + const mockedClient = { send: mockedDeleteSendSuccessful }; + const logger = new VoidLogger(); + + const dao = new S3WidgetDAO( + testBucket, + mockedClient as unknown as S3Client, + testKeyGenerator, + logger, + ); + + const resp = await dao.delete(testWidget); + expect(resp).toEqual(testWidget); +}); diff --git a/Homework/cs5260/consumer/t/S3WidgetKeyGenerator.spec.ts b/Homework/cs5260/consumer/t/S3WidgetKeyGenerator.spec.ts new file mode 100644 index 0000000..642c7a4 --- /dev/null +++ b/Homework/cs5260/consumer/t/S3WidgetKeyGenerator.spec.ts @@ -0,0 +1,48 @@ +import { expect, test } from "bun:test"; +import { S3WidgetKeyGenerator } from "@/consumer/dao/keys"; +import { Widget } from "@/shared/schema"; + +test("builds keys correctly", () => { + const generator = new S3WidgetKeyGenerator() + .withPrefix("test") + .withWidgetId("test") + .withOwner("simponic"); + + expect(generator.build()).toEqual("test/simponic/test"); +}); + +test("when owner given spaces then replaces with dashes and lowercases", () => { + const generator = new S3WidgetKeyGenerator() + .withPrefix("widgets") + .withWidgetId("test") + .withOwner("simPOnic test"); + + expect(generator.build()).toEqual("widgets/simponic-test/test"); +}); + +test("when not all specifications given then throws", () => { + const generator = new S3WidgetKeyGenerator() + .withPrefix("test") + .withWidgetId("test"); + + expect(() => generator.build()).toThrow(); +}); + +test("when invalid keys given then throws", () => { + const generator = new S3WidgetKeyGenerator() + .withPrefix("test") + .withOwner("simponic/simponic") + .withWidgetId("test"); + + expect(() => generator.build()).toThrow(); +}); + +test("when keys correct then builds from widget", () => { + const generator = new S3WidgetKeyGenerator().withPrefix("test-widgets"); + + const testWidget = Widget.parse({ owner: "simponic", id: "test-id" }); + + expect(generator.fromWidget(testWidget)).toEqual( + "test-widgets/simponic/test-id" + ); +}); diff --git a/Homework/cs5260/consumer/t/S3WidgetRequestRetriever.spec.ts b/Homework/cs5260/consumer/t/S3WidgetRequestRetriever.spec.ts new file mode 100644 index 0000000..9807875 --- /dev/null +++ b/Homework/cs5260/consumer/t/S3WidgetRequestRetriever.spec.ts @@ -0,0 +1,172 @@ +import { S3WidgetRequestRetriever } from "@/consumer/retrievers"; +import { WidgetRequest } from "@/shared/schema"; +import { + DeleteObjectCommand, + GetObjectCommand, + ListObjectsV2Command, + S3Client, +} from "@aws-sdk/client-s3"; +import { expect, mock, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { VoidLogger } from "@/shared/t"; + +test("when widget request in S3 then reads, deletes, and returns", async () => { + const logger = new VoidLogger(); + const widgetRequest = WidgetRequest.parse({ + type: "create", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + + const key = "test-key"; + + const deleteKeyMock = mock(() => true); + const listKeyMock = mock(() => { + return { + Contents: [{ Key: key }], + }; + }); + const getKeyMock = mock(() => { + return { + Body: { + transformToString: async () => JSON.stringify(widgetRequest), + }, + }; + }); + + const mockS3Client = { + send: async ( + req: ListObjectsV2Command | DeleteObjectCommand | GetObjectCommand, + ) => { + if (req instanceof ListObjectsV2Command) { + return listKeyMock(); + } + + if (req instanceof DeleteObjectCommand) { + return deleteKeyMock(); + } + + if (req instanceof GetObjectCommand) { + if (req.input.Key === key) { + return getKeyMock(); + } + } + }, + }; + + const retriever = new S3WidgetRequestRetriever( + "test-bucket", + mockS3Client as unknown as S3Client, + logger, + ); + + expect(await retriever.refreshRequests()).toEqual([widgetRequest]); + expect(getKeyMock).toHaveBeenCalledTimes(1); + expect(listKeyMock).toHaveBeenCalledTimes(1); + expect(deleteKeyMock).toHaveBeenCalledTimes(1); +}); + +test("when empty widget request in S3 then deletes and returns empty", async () => { + const logger = new VoidLogger(); + const key = "test-key"; + + const deleteKeyMock = mock(() => true); + const listKeyMock = mock(() => { + return { + Contents: [{ Key: key }], + }; + }); + const getKeyMock = mock(() => { + return { + Body: { + transformToString: async () => "", + }, + }; + }); + + const mockS3Client = { + send: mock( + async ( + req: ListObjectsV2Command | DeleteObjectCommand | GetObjectCommand, + ) => { + if (req instanceof ListObjectsV2Command) { + return listKeyMock(); + } + + if (req instanceof DeleteObjectCommand) { + return deleteKeyMock(); + } + + if (req instanceof GetObjectCommand) { + if (req.input.Key === key) { + return getKeyMock(); + } + } + }, + ), + }; + + const retriever = new S3WidgetRequestRetriever( + "test-bucket", + mockS3Client as unknown as S3Client, + logger, + ); + + expect(await retriever.refreshRequests()).toBeEmpty(); + expect(getKeyMock).toHaveBeenCalledTimes(1); + expect(listKeyMock).toHaveBeenCalledTimes(1); + expect(deleteKeyMock).toHaveBeenCalledTimes(1); +}); + +test("when no keys then returns empty", async () => { + const logger = new VoidLogger(); + const key = "test-key"; + + const deleteKeyMock = mock(() => true); + const listKeyMock = mock(() => { + return { + Contents: [], + }; + }); + const getKeyMock = mock(() => { + return { + Body: { + transformToString: async () => "", + }, + }; + }); + + const mockS3Client = { + send: mock( + async ( + req: ListObjectsV2Command | DeleteObjectCommand | GetObjectCommand, + ) => { + if (req instanceof ListObjectsV2Command) { + return listKeyMock(); + } + + if (req instanceof DeleteObjectCommand) { + return deleteKeyMock(); + } + + if (req instanceof GetObjectCommand) { + if (req.input.Key === key) { + return getKeyMock(); + } + } + }, + ), + }; + + const retriever = new S3WidgetRequestRetriever( + "test-bucket", + mockS3Client as unknown as S3Client, + logger, + ); + + expect(await retriever.refreshRequests()).toEqual([]); + expect(listKeyMock).toHaveBeenCalledTimes(1); + expect(getKeyMock).toHaveBeenCalledTimes(0); + expect(deleteKeyMock).toHaveBeenCalledTimes(0); +}); diff --git a/Homework/cs5260/consumer/t/SQSWidgetRequestRetriever.spec.ts b/Homework/cs5260/consumer/t/SQSWidgetRequestRetriever.spec.ts new file mode 100644 index 0000000..cae6a65 --- /dev/null +++ b/Homework/cs5260/consumer/t/SQSWidgetRequestRetriever.spec.ts @@ -0,0 +1,95 @@ +import { SQSWidgetRequestRetriever } from "@/consumer/retrievers"; +import { WidgetRequest } from "@/shared/schema"; +import { + DeleteMessageBatchCommand, + ReceiveMessageCommand, + SQSClient, +} from "@aws-sdk/client-sqs"; +import { expect, mock, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { VoidLogger } from "@/shared/t"; + +test("when widget request in SQS then reads, deletes, and returns", async () => { + const logger = new VoidLogger(); + const widgetRequest = WidgetRequest.parse({ + type: "create", + requestId: randomUUID(), + owner: "simponic", + widgetId: randomUUID(), + }); + + const queueUrl = "https://asdf.fdsa.asdf"; + + const deleteMessagesBatch = mock(() => true); + const getMessages = mock(() => { + return { + Messages: [ + { + Body: JSON.stringify(widgetRequest), + MessageId: "fdsa", + ReceiptHandle: "asdf", + }, + ], + }; + }); + + const mockSQSClient = { + send: async (req: ReceiveMessageCommand | DeleteMessageBatchCommand) => { + if (req instanceof ReceiveMessageCommand) { + if (req.input.QueueUrl === queueUrl) return getMessages(); + } + if (req instanceof DeleteMessageBatchCommand) { + expect( + req.input.Entries?.find( + (x) => x.Id === "fdsa" && x.ReceiptHandle === "asdf", + ), + ).toBeTruthy(); + + if (req.input.QueueUrl === queueUrl) return deleteMessagesBatch(); + } + return; + }, + }; + + const retriever = new SQSWidgetRequestRetriever( + queueUrl, + mockSQSClient as unknown as SQSClient, + logger, + ); + + expect(await retriever.refreshRequests()).toEqual([widgetRequest]); + expect(deleteMessagesBatch).toHaveBeenCalledTimes(1); + expect(getMessages).toHaveBeenCalledTimes(1); +}); + +test("when empty widget request in SQS then returns empty", async () => { + const logger = new VoidLogger(); + const queueUrl = "https://asdf.fdsa.asdf"; + + const getMessages = mock(() => { + return {}; + }); + const deleteMessagesBatch = mock(() => true); + + const mockSQSClient = { + send: async (req: ReceiveMessageCommand | DeleteMessageBatchCommand) => { + if (req instanceof ReceiveMessageCommand) { + if (req.input.QueueUrl === queueUrl) return getMessages(); + } + if (req instanceof DeleteMessageBatchCommand) { + if (req.input.QueueUrl === queueUrl) return deleteMessagesBatch(); + } + return; + }, + }; + + const retriever = new SQSWidgetRequestRetriever( + queueUrl, + mockSQSClient as unknown as SQSClient, + logger, + ); + + expect(await retriever.refreshRequests()).toBeEmpty(); + expect(getMessages).toHaveBeenCalledTimes(1); + expect(deleteMessagesBatch).toHaveBeenCalledTimes(0); +}); diff --git a/Homework/cs5260/consumer/t/index.ts b/Homework/cs5260/consumer/t/index.ts new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/Homework/cs5260/consumer/t/index.ts diff --git a/Homework/cs5260/consumer_entrypoint.sh b/Homework/cs5260/consumer_entrypoint.sh new file mode 100644 index 0000000..a8dd6ce --- /dev/null +++ b/Homework/cs5260/consumer_entrypoint.sh @@ -0,0 +1,8 @@ +#!/bin/bash +bun run consumer/index.ts \ + --sourcequeue=$SOURCE_QUEUE \ + --storagetable=$STORAGE_TABLE \ + --storagebucket=$STORAGE_BUCKET \ + --storageprefix=$STORAGE_PREFIX \ + 2>&1 \ + | tee "runlogs/consumer-$(date).log" diff --git a/Homework/cs5260/docker-compose.yml b/Homework/cs5260/docker-compose.yml new file mode 100644 index 0000000..f6ed2e7 --- /dev/null +++ b/Homework/cs5260/docker-compose.yml @@ -0,0 +1,48 @@ +version: '3' + +services: + consumer: + depends_on: + - ecs-local-endpoints + build: + dockerfile: Dockerfile.consumer + networks: + credentials_network: + ipv4_address: "169.254.170.3" + volumes: + - ./runlogs:/app/runlogs:rw + env_file: + - .env + + lambda: + depends_on: + - ecs-local-endpoints + build: + dockerfile: Dockerfile.lambda + networks: + credentials_network: + ipv4_address: "169.254.170.4" + volumes: + - ./runlogs:/app/runlogs:rw + env_file: + - .env + + ecs-local-endpoints: + image: amazon/amazon-ecs-local-container-endpoints + volumes: + - /var/run:/var/run + - $HOME/.aws/:/home/.aws/ + environment: + HOME: "/home" + AWS_PROFILE: "default" + networks: + credentials_network: + ipv4_address: "169.254.170.2" + +networks: + credentials_network: + driver: bridge + ipam: + config: + - subnet: "169.254.170.0/24" + gateway: 169.254.170.1 diff --git a/Homework/cs5260/lambda_entrypoint.sh b/Homework/cs5260/lambda_entrypoint.sh new file mode 100644 index 0000000..71264cb --- /dev/null +++ b/Homework/cs5260/lambda_entrypoint.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +bun run widgetrequesthandler/handler.ts \ + 2>&1 \ + | tee "runlogs/lambda-$(date).log" diff --git a/Homework/cs5260/package.json b/Homework/cs5260/package.json new file mode 100644 index 0000000..5656a11 --- /dev/null +++ b/Homework/cs5260/package.json @@ -0,0 +1,30 @@ +{ + "name": "consume", + "module": "index.ts", + "type": "module", + "devDependencies": { + "bun-types": "latest", + "json-schema-to-typescript": "^13.1.1", + "@types/minimist": "^1.2.4", + "@oclif/plugin-plugins": "^3.5.0", + "jszip": "^3.10.1", + "oclif": "^3.6.5", + "prettier": "^2.8.2" + }, + "peerDependencies": { + "typescript": "^5.0.0" + }, + "dependencies": { + "@aws-sdk/client-dynamodb": "^3.431.0", + "@aws-sdk/client-s3": "^3.431.0", + "@aws-sdk/client-sqs": "^3.441.0", + "aws4fetch": "^1.0.17", + "minimist": "^1.2.8", + "zod": "^3.22.4" + }, + "scripts": { + "build-layer": "bun run widgetrequesthandler/scripts/build-layer.ts", + "publish-layer": "bun run widgetrequesthandler/scripts/publish-layer.ts", + "publish-lambda": "./publish_lambda.sh" + } +} diff --git a/Homework/cs5260/producer/producer.jar b/Homework/cs5260/producer/producer.jar Binary files differnew file mode 100644 index 0000000..06d2941 --- /dev/null +++ b/Homework/cs5260/producer/producer.jar diff --git a/Homework/cs5260/publish_lambda.sh b/Homework/cs5260/publish_lambda.sh new file mode 100644 index 0000000..e2ad000 --- /dev/null +++ b/Homework/cs5260/publish_lambda.sh @@ -0,0 +1,30 @@ +#!/bin/sh +export AWS_REGION=us-east-1 + +LAMBDA_NAME=widget-api-handler + +echo "building..." +cd widgetrequesthandler +bun run build.ts + +echo "zipping..." +cd dist +zip -r widgetrequesthandler.zip * + +echo "pushing to $LAMBDA_NAME..." +aws lambda update-function-code \ + --region $AWS_REGION \ + --function-name $LAMBDA_NAME \ + --zip-file fileb://widgetrequesthandler.zip \ + > /dev/null + +echo "setting env vars..." +export $(cat ../../.env | xargs) + +sleep 3 +aws lambda update-function-configuration \ + --function-name $LAMBDA_NAME \ + --environment "Variables={SOURCE_QUEUE=$SOURCE_QUEUE}" \ + > /dev/null + +echo "done!" diff --git a/Homework/cs5260/shared/log/ConsoleLogger.ts b/Homework/cs5260/shared/log/ConsoleLogger.ts new file mode 100644 index 0000000..4d594a8 --- /dev/null +++ b/Homework/cs5260/shared/log/ConsoleLogger.ts @@ -0,0 +1,29 @@ +import { TracingLogger } from "."; + +export class ConsoleTracingLogger implements TracingLogger { + private prefix: string; + + constructor(prefix: string) { + this.prefix = prefix; + } + + private makePrefix(level: "info" | "warn" | "error"): string { + return `[${new Date().toISOString()}] ${level} (${this.prefix}) > `; + } + + public info(log: string) { + console.log(this.makePrefix("info") + log); + } + + public warn(log: string) { + console.warn(this.makePrefix("warn") + log); + } + + public error(log: string) { + console.error(this.makePrefix("error") + log); + } + + public createChild(prefix: string) { + return new ConsoleTracingLogger(`${this.prefix} -> ${prefix}`); + } +} diff --git a/Homework/cs5260/shared/log/TracingLogger.ts b/Homework/cs5260/shared/log/TracingLogger.ts new file mode 100644 index 0000000..4765480 --- /dev/null +++ b/Homework/cs5260/shared/log/TracingLogger.ts @@ -0,0 +1,7 @@ +export interface TracingLogger { + info(log: string): void; + warn(log: string): void; + error(log: string): void; + + createChild(prefix: string): TracingLogger; +} diff --git a/Homework/cs5260/shared/log/index.ts b/Homework/cs5260/shared/log/index.ts new file mode 100644 index 0000000..d1bd68d --- /dev/null +++ b/Homework/cs5260/shared/log/index.ts @@ -0,0 +1,2 @@ +export * from "./TracingLogger"; +export * from "./ConsoleLogger"; diff --git a/Homework/cs5260/shared/schema/README.md b/Homework/cs5260/shared/schema/README.md new file mode 100644 index 0000000..6c5d0a0 --- /dev/null +++ b/Homework/cs5260/shared/schema/README.md @@ -0,0 +1,3 @@ +to compile TypeScript types + +`json-schema-to-zod --source=<schema.json> > <schema>.ts` diff --git a/Homework/cs5260/shared/schema/Widget.ts b/Homework/cs5260/shared/schema/Widget.ts new file mode 100644 index 0000000..e7f09d1 --- /dev/null +++ b/Homework/cs5260/shared/schema/Widget.ts @@ -0,0 +1,13 @@ +import { z } from "zod"; + +export const Widget = z.object({ + id: z.string(), + owner: z.string().regex(new RegExp("[A-Za-z ]+")), + label: z.string().optional(), + description: z.string().optional(), + otherAttributes: z + .array(z.object({ name: z.string(), value: z.string() })) + .optional(), +}); + +export type WidgetT = z.infer<typeof Widget>; diff --git a/Homework/cs5260/shared/schema/WidgetRequest.json b/Homework/cs5260/shared/schema/WidgetRequest.json new file mode 100644 index 0000000..a378949 --- /dev/null +++ b/Homework/cs5260/shared/schema/WidgetRequest.json @@ -0,0 +1,44 @@ +{ + "$schema": "http://json-schema.org/draft-04/schema#", + "type": "object", + "properties": { + "type": { + "type": "string", + "pattern": "create|delete|update" + }, + "requestId": { + "type": "string" + }, + "widgetId": { + "type": "string" + }, + "owner": { + "type": "string", + "pattern": "[A-Za-z ]+" + }, + "label": { + "type": "string" + }, + "description": { + "type": "string" + }, + "otherAttributes": { + "type": "array", + "items": [ + { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "value": { + "type": "string" + } + }, + "required": ["name", "value"] + } + ] + } + }, + "required": ["type", "requestId", "widgetId", "owner"] +} diff --git a/Homework/cs5260/shared/schema/WidgetRequest.ts b/Homework/cs5260/shared/schema/WidgetRequest.ts new file mode 100644 index 0000000..11259cf --- /dev/null +++ b/Homework/cs5260/shared/schema/WidgetRequest.ts @@ -0,0 +1,16 @@ +import { z } from "zod"; + +// generated from json-schema-to-zod +export const WidgetRequest = z.object({ + type: z.string().regex(new RegExp("create|delete|update")), + requestId: z.string(), + widgetId: z.string(), + owner: z.string().regex(new RegExp("[A-Za-z ]+")), + label: z.string().optional(), + description: z.string().optional(), + otherAttributes: z + .array(z.object({ name: z.string(), value: z.string() })) + .optional(), +}); + +export type WidgetRequestT = z.infer<typeof WidgetRequest>; diff --git a/Homework/cs5260/shared/schema/index.ts b/Homework/cs5260/shared/schema/index.ts new file mode 100644 index 0000000..44cdfd7 --- /dev/null +++ b/Homework/cs5260/shared/schema/index.ts @@ -0,0 +1,2 @@ +export * from "./WidgetRequest.ts"; +export * from "./Widget"; diff --git a/Homework/cs5260/shared/t/VoidLogger.ts b/Homework/cs5260/shared/t/VoidLogger.ts new file mode 100644 index 0000000..4133104 --- /dev/null +++ b/Homework/cs5260/shared/t/VoidLogger.ts @@ -0,0 +1,11 @@ +import { TracingLogger } from "../log"; + +export class VoidLogger implements TracingLogger { + public info(_log: string) {} + public warn(_log: string) {} + public error(_log: string) {} + + public createChild(_prefix: string) { + return new VoidLogger(); + } +} diff --git a/Homework/cs5260/shared/t/index.ts b/Homework/cs5260/shared/t/index.ts new file mode 100644 index 0000000..93f45ce --- /dev/null +++ b/Homework/cs5260/shared/t/index.ts @@ -0,0 +1 @@ +export * from "./VoidLogger"; diff --git a/Homework/cs5260/tsconfig.json b/Homework/cs5260/tsconfig.json new file mode 100644 index 0000000..fbc6fef --- /dev/null +++ b/Homework/cs5260/tsconfig.json @@ -0,0 +1,25 @@ +{ + "compilerOptions": { + "lib": ["ESNext"], + "module": "esnext", + "target": "esnext", + "moduleResolution": "bundler", + "moduleDetection": "force", + "allowImportingTsExtensions": true, + "noEmit": true, + "composite": true, + "strict": true, + "downlevelIteration": true, + "skipLibCheck": true, + "jsx": "react-jsx", + "allowSyntheticDefaultImports": true, + "forceConsistentCasingInFileNames": true, + "allowJs": true, + "types": [ + "bun-types" // add Bun global + ], + "paths": { + "@/*": ["./*"] + } + } +} 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); +}); |
