From b22230bddbadd5dd89057d44c01d2660c285d368 Mon Sep 17 00:00:00 2001
From: Nitin Goyal <nitingoyal.dev@gmail.com>
Date: Tue, 29 Sep 2026 21:12:11 +0530
Subject: [PATCH] feat(deploy): add DLQ, log retention, and existing
 destination bucket to CloudFormation (#18)

- Add ProcessorDeadLetterQueue SQS resource and RedrivePolicy to ProcessorQueue with configurable SqsMaxReceiveCount
- Add explicit AWS::Logs::LogGroup resource with configurable LogRetentionDays
- Parameterize LambdaMemorySize (default 512MB)
- Remove internal DestinationBucket creation and replace with DasDestinationBucketName parameter for existing data lake bucket integration
- Update IAM execution role and DAS_OUTPUT_BUCKET to use DasDestinationBucketName
- Add stack outputs for ProcessorDLQArn and ProcessorDLQUrl
---
 deploy/cloudformation/lambda.yaml | 66 +++++++++++++++++++++++--------
 1 file changed, 49 insertions(+), 17 deletions(-)

diff --git a/deploy/cloudformation/lambda.yaml b/deploy/cloudformation/lambda.yaml
index 93ae874..7cc938f 100644
--- a/deploy/cloudformation/lambda.yaml
+++ b/deploy/cloudformation/lambda.yaml
@@ -10,6 +10,10 @@ Parameters:
     Type: String
     Description: 'ARN of the source S3 Bucket containing the encrypted DAS logs.'
 
+  DasDestinationBucketName:
+    Type: String
+    Description: 'Name of the existing destination S3 Bucket for processed Parquet files (DAS_OUTPUT_BUCKET).'
+
   KmsKeyArn:
     Type: String
     Description: 'ARN of the KMS Key used to encrypt the DAS logs.'
@@ -27,20 +31,32 @@ Parameters:
     Default: 'default'
     Description: 'Name of the JSON DSL filter to apply (DAS_FILTER_NAME).'
 
+  LambdaMemorySize:
+    Type: Number
+    Default: 512
+    AllowedValues: [256, 512, 1024, 1536, 2048, 3072]
+    Description: 'Memory size in MB allocated to the Lambda function.'
+
+  LogRetentionDays:
+    Type: Number
+    Default: 30
+    AllowedValues: [1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1096, 1827, 2192, 2557, 2922, 3288, 3653]
+    Description: 'Number of days to retain CloudWatch Logs for the Lambda function.'
+
+  SqsMaxReceiveCount:
+    Type: Number
+    Default: 5
+    MinValue: 1
+    MaxValue: 1000
+    Description: 'The number of times a message is delivered to the processor queue before being moved to the dead-letter queue.'
+
 Resources:
-  # 1. Destination S3 Bucket (Parquet Data Lake)
-  DestinationBucket:
-    Type: AWS::S3::Bucket
+  # 1. Dead-Letter Queue for Fanout
+  ProcessorDeadLetterQueue:
+    Type: AWS::SQS::Queue
     Properties:
-      BucketEncryption:
-        ServerSideEncryptionConfiguration:
-          - ServerSideEncryptionByDefault:
-              SSEAlgorithm: AES256
-      PublicAccessBlockConfiguration:
-        BlockPublicAcls: true
-        BlockPublicPolicy: true
-        IgnorePublicAcls: true
-        RestrictPublicBuckets: true
+      MessageRetentionPeriod: 1209600 # 14 days
+      KmsMasterKeyId: alias/aws/sqs # SSE encryption for DLQ
 
   # 2. SQS Queue for Fanout
   ProcessorQueue:
@@ -49,6 +65,9 @@ Resources:
       VisibilityTimeout: 300
       MessageRetentionPeriod: 1209600 # 14 days
       KmsMasterKeyId: alias/aws/sqs # SSE encryption for queue
+      RedrivePolicy:
+        deadLetterTargetArn: !GetAtt ProcessorDeadLetterQueue.Arn
+        maxReceiveCount: !Ref SqsMaxReceiveCount
 
   # 3. SQS Queue Policy allowing SNS to publish
   ProcessorQueuePolicy:
@@ -102,7 +121,7 @@ Resources:
               - Effect: Allow
                 Action:
                   - s3:PutObject
-                Resource: !Sub '${DestinationBucket.Arn}/*'
+                Resource: !Sub 'arn:aws:s3:::${DasDestinationBucketName}/*'
               - Effect: Allow
                 Action:
                   - kms:Decrypt
@@ -123,16 +142,23 @@ Resources:
         ImageUri: !Ref LambdaImageUri
       Role: !GetAtt LambdaExecutionRole.Arn
       Timeout: 120
-      MemorySize: 512
+      MemorySize: !Ref LambdaMemorySize
       Environment:
         Variables:
           DAS_FILTER_NAME: !Ref FilterName
           DAS_KMS_REGION_NAME: !Ref AWS::Region
           DAS_RDS_RESOURCE_ID: !Ref RdsResourceId
           # Output bucket for Parquet files
-          DAS_OUTPUT_BUCKET: !Ref DestinationBucket
+          DAS_OUTPUT_BUCKET: !Ref DasDestinationBucketName
+
+  # 7. CloudWatch Log Group with retention policy
+  ProcessorLogGroup:
+    Type: AWS::Logs::LogGroup
+    Properties:
+      LogGroupName: !Sub '/aws/lambda/${ProcessorFunction}'
+      RetentionInDays: !Ref LogRetentionDays
 
-  # 7. SQS Event Source Mapping to Lambda
+  # 8. SQS Event Source Mapping to Lambda
   EventSourceMapping:
     Type: AWS::Lambda::EventSourceMapping
     Properties:
@@ -146,7 +172,13 @@ Resources:
 Outputs:
   DestinationBucketName:
     Description: 'S3 Data Lake Bucket for Parquet files'
-    Value: !Ref DestinationBucket
+    Value: !Ref DasDestinationBucketName
   ProcessorFunctionName:
     Description: 'Lambda Function Name'
     Value: !Ref ProcessorFunction
+  ProcessorDLQArn:
+    Description: 'ARN of the Dead Letter Queue for failed DAS records'
+    Value: !GetAtt ProcessorDeadLetterQueue.Arn
+  ProcessorDLQUrl:
+    Description: 'URL of the Dead Letter Queue'
+    Value: !Ref ProcessorDeadLetterQueue