4 files changed, 133 insertions, 0 deletions
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";