add todos

Author: Ian-Nara

src/main/java/com/uid2/optout/delta/S3UploadService.java

@@ -71,7 +71,7 @@ public void uploadAndDeleteMessages(byte[] data, String s3Path, List<Message> me
             if (onSuccess != null) {
                 onSuccess.onSuccess(messages.size());
             }
-        } catch (Exception e) {
+        } catch (Exception e) { // TODO: catch specific exceptions
             LOGGER.error("s3_error: failed to upload delta or dropped requests to path={}", s3Path, e);
             throw new IOException("s3 upload failed: " + s3Path, e);
         }

src/main/java/com/uid2/optout/delta/StopReason.java

@@ -27,7 +27,7 @@ public enum StopReason {
     MESSAGE_LIMIT_EXCEEDED,
 
     /**
-     * Pre-existing manual override was set (checked at job start).
+     * Pre-existing manual override was set to DELAYED_PROCESSING (checked at job start).
      */
     MANUAL_OVERRIDE_ACTIVE,
 

src/main/java/com/uid2/optout/sqs/SqsBatchProcessor.java

@@ -83,7 +83,7 @@ public BatchProcessingResult processBatch(List<Message> messageBatch, int batchN
             List<Message> invalidMessages = identifyInvalidMessages(messageBatch, parsedBatch);
             if (!invalidMessages.isEmpty()) {
                 LOGGER.error("sqs_error: found {} invalid messages in batch {}, deleting", invalidMessages.size(), batchNumber);
-                SqsMessageOperations.deleteMessagesFromSqs(this.sqsClient, this.queueUrl, invalidMessages);
+                SqsMessageOperations.deleteMessagesFromSqs(this.sqsClient, this.queueUrl, invalidMessages); // TODO: send to a folder in the dropped requests bucket before deleting.
             }
         }
 

src/main/java/com/uid2/optout/sqs/SqsWindowReader.java

@@ -29,10 +29,10 @@ public SqsWindowReader(SqsClient sqsClient, String queueUrl, int maxMessagesPerP
                           int visibilityTimeout, int deltaWindowSeconds, int maxMessagesPerWindow) {
         this.sqsClient = sqsClient;
         this.queueUrl = queueUrl;
-        this.maxMessagesPerPoll = maxMessagesPerPoll;
-        this.visibilityTimeout = visibilityTimeout;
+        this.maxMessagesPerPoll = maxMessagesPerPoll; // 10 max 
+        this.visibilityTimeout = visibilityTimeout; // TODO: ensure we can process all messages before visibility timeout
         this.deltaWindowSeconds = deltaWindowSeconds;
-        this.maxMessagesPerWindow = maxMessagesPerWindow;
+        this.maxMessagesPerWindow = maxMessagesPerWindow; // TODO: ensure we can process all messages before visibility timeout
         this.batchProcessor = new SqsBatchProcessor(sqsClient, queueUrl, deltaWindowSeconds);
         LOGGER.info("initialized: maxMessagesPerWindow={}, maxMessagesPerPoll={}, visibilityTimeout={}, deltaWindowSeconds={}",
                 maxMessagesPerWindow, maxMessagesPerPoll, visibilityTimeout, deltaWindowSeconds);
@@ -94,7 +94,7 @@ public WindowReadResult readWindow() {
         int batchNumber = 0;
         int rawMessagesRead = 0;  // track total messages pulled from SQS
 
-        while (true) {
+        while (true) { // TODO: add a timeout to the loop
             if (windowMessages.size() >= maxMessagesPerWindow) {
                 LOGGER.warn("high_message_volume: message limit exceeded while reading window, {} messages >= limit {}", windowMessages.size(), maxMessagesPerWindow);
                 return WindowReadResult.messageLimitExceeded(windowMessages, currentWindowStart, rawMessagesRead);