Updated changes from Mike's comments

Author: zalilias

serverless-message-processing/README.md

@@ -4,8 +4,8 @@
 An adaptable pattern for message processing using AWS serverless services, featuring error handling and automatic recovery mechanisms.
 
 ## Core Components
-- API Gateway (message ingestion)
-- SQS Queues (main + DLQs)
+- Amazon API Gateway (message ingestion)
+- Amazon SQS Queues (main + DLQs)
 - Lambda Functions (processing + recovery)
 
 
@@ -24,6 +24,17 @@ An adaptable pattern for message processing using AWS serverless services, featu
 
 ## Deployment
 # Build the SAM application
+
+1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository:
+    ``` 
+    git clone https://github.com/aws-samples/serverless-patterns
+    ```
+2. Change directory to the pattern directory:
+    ```
+    cd serverless-patterns/serverless-messaging-redrive
+    ```
+
+# Build the SAM application
 The ```sam build ``` command prepares an application for subsequent steps in the developer workflow, such as local testing or deploying to AWS.
 
    ```

serverless-message-processing/functions/decision_maker/app.py

@@ -66,7 +66,7 @@ def validate_fixed_email(email):
     Returns:
         bool: True if valid email format, False otherwise
     """
-    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
+    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\$'
     return bool(re.match(email_pattern, email))
 
 def lambda_handler(event, context):
@@ -93,15 +93,17 @@ def lambda_handler(event, context):
         event: Lambda event object containing SQS messages
         context: Lambda context object
     Returns:
-        dict: Processing summary with counts
+        dict: Processing summary with counts and batchItemFailures
     """
     processed_count = 0
     fixed_count = 0
     fatal_count = 0
+    failed_message_ids = []
 
     logger.info(f"Starting to process batch of {len(event['Records'])} messages")
 
     for record in event['Records']:
+        original_message_id = "unknown"
         try:
             # Parse the failed message
             message = json.loads(record['body'])
@@ -158,6 +160,8 @@ def lambda_handler(event, context):
 
         except Exception as e:
             logger.error(f"Error processing message {original_message_id}: {str(e)}")
+            # Add message ID to failed messages list
+            failed_message_ids.append(record['messageId'])
             try:
                 error_message = {
                     'originalMessage': record['body'],
@@ -171,22 +175,25 @@ def lambda_handler(event, context):
                 fatal_count += 1
             except Exception as fatal_e:
                 logger.critical(f"Could not send to fatal DLQ: {str(fatal_e)}")
-                raise
 
     # Execution summary
     logger.info(f"""
     === Execution Summary ===
     Messages Processed: {processed_count}
     Messages Fixed: {fixed_count}
     Messages Fatal: {fatal_count}
+    Messages Failed: {len(failed_message_ids)}
     ========================
     """)
 
-    return {
-        'statusCode': 200,
-        'body': json.dumps({
+    # Return both the processing info and the batch failures for SQS
+    result = {
+        'batchItemFailures': [{"itemIdentifier": message_id} for message_id in failed_message_ids],
+        'processingInfo': {
             'processed': processed_count,
             'fixed': fixed_count,
             'fatal': fatal_count
-        })
+        }
     }
+    
+    return result

serverless-message-processing/functions/processor/app.py

@@ -14,7 +14,7 @@ def validate_email(email):
     Returns:
         bool: True if valid email format, False otherwise
     """
-    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\$'
+    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
     if not bool(re.match(email_pattern, email)):
         logger.error(f"Invalid email format: {email}")
         return False

serverless-message-processing/template.yaml

@@ -1,6 +1,6 @@
 AWSTemplateFormatVersion: '2010-09-09'
 Transform: AWS::Serverless-2016-10-31
-Description: Automated Serverless Messaging Redrive
+Description: Automated Serverless Messaging Redrive (uksb-1tthgi812) (tag:serverless-message-processing)
 
 Parameters:
   ProcessorMemorySize:
@@ -136,7 +136,7 @@ Resources:
     Type: AWS::Serverless::Function
     Properties:
       Handler: app.lambda_handler
-      Runtime: python3.11
+      Runtime: python3.13
       CodeUri: ./functions/processor/
       MemorySize: !Ref ProcessorMemorySize
       Timeout: !Ref LambdaTimeout
@@ -147,6 +147,9 @@ Resources:
           Properties:
             Queue: !GetAtt MainQueue.Arn
             BatchSize: 10
+            FunctionResponseTypes:
+              - ReportBatchItemFailures
+
       Environment:
         Variables:
           MAIN_QUEUE_URL: !Ref MainQueue
@@ -156,7 +159,7 @@ Resources:
     Type: AWS::Serverless::Function
     Properties:
       Handler: app.lambda_handler
-      Runtime: python3.11
+      Runtime: python3.13
       CodeUri: ./functions/decision_maker/
       MemorySize: !Ref DecisionMakerMemorySize
       Timeout: !Ref LambdaTimeout
@@ -167,9 +170,11 @@ Resources:
           Properties:
             Queue: !GetAtt AutomatedDLQ.Arn
             BatchSize: 10
+            FunctionResponseTypes:
+              - ReportBatchItemFailures
+
+
       Policies:
-        - SQSPollerPolicy:
-            QueueName: !GetAtt MainQueue.QueueName
         - SQSPollerPolicy:
             QueueName: !GetAtt FatalDLQ.QueueName
         - SQSSendMessagePolicy: