Updated code based upon recommendations

Author: zalilias

serverless-message-processing/README.md

@@ -11,38 +11,47 @@ An adaptable pattern for message processing using AWS serverless services, featu
 ## Basic Flow
 1. Messages enter through API Gateway
 2. Main queue receives messages
-3. Processor Lambda handles messages
-4. Failed messages route to DLQs
-5. Decision maker attempts an automated recovery
+3. Lambda function polls the main queue using Event Source Mappings (ESMs) and handles the messages. 
+   Read more about how Lambda synchronously processes queue messages in this [documentation.](https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html)
+4. Failed messages route to DLQs -In this case a failed message would be a malformed email, however this can be adapted to other use cases.
+5. Decision maker attempts an automated recovery -In this sample, we remediate common email malform issues including whitespace and typos in domain extensions.
 
 ## Deployment
 # 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.
+
    ```
 sam build
    ```
 # Deploy the application
+The ```sam deploy``` command deploys an application to the AWS Cloud using AWS CloudFormation. The ```--guided``` option is to have the AWS SAM CLI use prompts to guide you through the deployment.
+
    ```
 sam deploy --guided
    ```
 
 ## Key Features
 - Automatic retry mechanism
 - Segregation of recoverable/fatal errors
-- Extensible processing logic
+- Processing logic with the potential for points of adaptation
 
 ## API Reference
 # Send Message
-   ```
 
-POST /message
-Content-Type: application/json
-   ```
+The following is an example API call that you can try with your own endpoint.
+
    ```
-{
-    "messageType": "TYPE_A|TYPE_B|TYPE_C",
-    "payload": {},
-    "timestamp": "ISO8601_TIMESTAMP"
-}
+
+curl -X POST \
+  'https://\${endpoint}/prod/message' \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "messageType": "TYPE_A",
+    "body": {
+        "email": "user@@example.com"
+    },
+    "timestamp": "2023-11-22T10:30:00Z"
+}'
    ```
 
 
@@ -54,5 +63,12 @@ Content-Type: application/json
 - Monitoring requirements
 - API Design
 
-## Note
-This is a sample pattern. Adapt security, scaling, and processing logic according to your requirements.
+## Cleanup
+1. Delete the SAM template
+```
+sam delete
+```
+2. Confirm the stack has been deleted
+```
+aws cloudformation list-stacks --query "StackSummaries[?contains(StackName,'STACK_NAME')].StackStatus"
+```

serverless-message-processing/example-pattern.json

@@ -1,17 +1,13 @@
 {
-    "title": "Step Functions to Athena",
+    "title": "Serverless Messaging Redrive",
     "description": "Create a Step Functions workflow to query Amazon Athena.",
     "language": "Python",
     "level": "200",
     "framework": "SAM",
     "introBox": {
       "headline": "How it works",
       "text": [
-        "This sample project demonstrates how to use a serverless solution for processing and fixing malformed messages using SQS queues and Lambda functions",
-        "The system automatically handles message validation, applies fixes where possible, and routes messages to appropriate queues based on their fixability.",
-        "It has built-in error handling and detailed logging, it provides a robust framework for message processing that can be easily extended for specific business needs.",
-        "This pattern uses AWS Lambda for processing, multiple SQS queues for message routing, and includes 2 dead-letter queue (DLQ) for messages requiring human intervention or for auto-remediation."
-      ]
+        "This sample project demonstrates how to use a serverless solution for processing and fixing malformed messages using SQS queues and Lambda functions"      ]
     },
     "gitHub": {
       "template": {

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

@@ -15,61 +15,72 @@
 MAIN_QUEUE_URL = os.environ['MAIN_QUEUE_URL']
 FATAL_DLQ_URL = os.environ['FATAL_DLQ_URL']
 
-# Email validation pattern
-EMAIL_PATTERN = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
-
 def fix_email(email):
     """
     Attempt to fix common email format issues
-    Can be amended to other scenarios e.g. Downstream issues
+    Args:
+        email: String containing malformed email
+    Returns:
+        str: Fixed email or original if unfixable
     """
-    # Remove multiple @ symbols, keep the last one
-    if email.count('@') > 1:
-        parts = email.split('@')
-        email = f"{parts[0]}@{parts[-1]}"
-    
-    # Remove spaces
-    email = email.strip().replace(' ', '')
-    
-    # Fix common typos in domain extensions
-    common_fixes = {
-        '.con': '.com',
-        '.vom': '.com',
-        '.comm': '.com',
-        '.orgg': '.org',
-        '.nett': '.net'
-    }
-    
-    for wrong, right in common_fixes.items():
-        if email.endswith(wrong):
-            email = email[:-len(wrong)] + right
-    
-    return email
+    try:
+        logger.info(f"Starting email fix attempt for: {email}")
+        
+        # Remove whitespace
+        email = email.strip()
+        
+        # Handle multiple @ symbols
+        if email.count('@') > 1:
+            parts = email.split('@')
+            email = f"{parts[0]}@{parts[-1]}"
+            logger.info(f"Fixed multiple @ symbols. Result: {email}")
+        
+        # Common domain typo fixes
+        domain_fixes = {
+            '.con': '.com',
+            '.vom': '.com',
+            '.comm': '.com',
+            '.orgg': '.org',
+            '.nett': '.net',
+            '.ckm': '.com',
+            '.cm': '.com'
+        }
+        
+        original_email = email
+        for wrong, right in domain_fixes.items():
+            if email.endswith(wrong):
+                email = email[:-len(wrong)] + right
+                logger.info(f"Fixed domain from {wrong} to {right}. Before: {original_email}, After: {email}")
+                break
+                
+        return email
+    except Exception as e:
+        logger.error(f"Error fixing email: {str(e)}")
+        return None
 
-def can_fix_email(message):
+def validate_fixed_email(email):
     """
-    Check if the email in the message can be fixed
+    Validate fixed email format.
+    Args:
+        email: String containing fixed email
+    Returns:
+        bool: True if valid email format, False otherwise
     """
-    if 'email' not in message:
-        return False
-        
-    email = message['email']
-    fixed_email = fix_email(email)
-    
-    return bool(re.match(EMAIL_PATTERN, fixed_email))
-
+    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):
     """
-    Processes messages from a DLQ that have already failed to be automatically processed, 
-    and attempts automated remediation and redelivery of the messages back to the main queue. 
-    If no suitable fixes can be applied, messages end up in a fatal DLQ where the typical 
-    approach of human intervention is required. 
-
+    Processes messages from a DLQ that have failed validation, 
+    attempting to fix common email format issues.
+    If fixed successfully, messages are sent back to the main queue.
+    If unfixable, messages are sent to a fatal DLQ.
+    
     Flow:
-    1. Attempt to fix message
-    2. If fixable -> Main Queue
-    3. If unfixable -> Fatal DLQ
+    1. Extract email from failed message
+    2. Attempt to fix common issues
+    3. If fixed → Main Queue
+    4. If unfixable → Fatal DLQ
     
     Extension points:
     1. Add more sophisticated routing logic- including a delay queue 
@@ -78,75 +89,104 @@ def lambda_handler(event, context):
     4. Implement retry mechanisms
     5. Add monitoring and metrics
 
-   """
+    Args:
+        event: Lambda event object containing SQS messages
+        context: Lambda context object
+    Returns:
+        dict: Processing summary with counts
+    """
     processed_count = 0
+    fixed_count = 0
+    fatal_count = 0
+
+    logger.info(f"Starting to process batch of {len(event['Records'])} messages")
 
     for record in event['Records']:
         try:
-            # Parse the message body
+            # Parse the failed message
             message = json.loads(record['body'])
-            original_message_id = record.get('messageId', 'unknown')
+            original_message_id = message.get('messageId', 'unknown')
 
-            logger.info(f"Processing failed message: {original_message_id}")
+            # Detailed message content logging
+            logger.info(f"Processing message content: {json.dumps(message, indent=2)}")
 
-           
-                
-
-            # Option A: Try to fix malformed email
+            # Check if message has already been remediated
+            if 'remediation' in message:
+                logger.info("Message already remediated, skipping processing")
+                continue
 
-            if can_fix_email(message) and not re.match(EMAIL_PATTERN, message['email']):
-                fixed_email = fix_email(message['email'])
-                logger.info(f"Fixed email from '{message['email']}' to '{fixed_email}'")
+            # Extract email from payload
+            if 'payload' in message and 'email' in message['payload']:
+                original_email = message['payload']['email']
 
-                # Update the message with fixed email
-                message['email'] = fixed_email
-                message['emailWasFixed'] = True
+                # Attempt to fix email
+                fixed_email = fix_email(original_email)
 
-                # Send back to main queue
-                sqs.send_message(
-                    QueueUrl=MAIN_QUEUE_URL,
-                    MessageBody=json.dumps(message)
-                )
-              
-                logger.info(f"Sent fixed message back to main queue: {original_message_id}")   
-
-            # Option B: Cannot fix - send to fatal DLQ
+                if fixed_email and validate_fixed_email(fixed_email):
+                    # Update message with fixed email
+                    message['payload']['email'] = fixed_email
+                    message['remediation'] = {
+                        'original_email': original_email,
+                        'fixed_email': fixed_email,
+                        'timestamp': context.invoked_function_arn
+                    }
+                    
+                    # Send back to main queue
+                    sqs.send_message(
+                        QueueUrl=MAIN_QUEUE_URL,
+                        MessageBody=json.dumps(message)
+                    )
+                    fixed_count += 1
+                else:
+                    # Send to fatal DLQ if unfixable
+                    message['failureReason'] = 'Email could not be remediated'
+                    sqs.send_message(
+                        QueueUrl=FATAL_DLQ_URL,
+                        MessageBody=json.dumps(message)
+                    )
+                    fatal_count += 1
             else:
-                logger.warning(f"Message cannot be fixed, sending to fatal DLQ: {original_message_id}")
-                
-                # Add failure reason if not present
-                if 'failureReason' not in message:
-                    message['failureReason'] = 'Unrecoverable error - could not fix message'
-                
-                # Send to fatal DLQ
+                # Send to fatal DLQ if message structure is invalid
+                message['failureReason'] = 'Invalid message structure - missing email in payload'
                 sqs.send_message(
                     QueueUrl=FATAL_DLQ_URL,
                     MessageBody=json.dumps(message)
                 )
-             
+                fatal_count += 1
+            
             processed_count += 1
 
         except Exception as e:
             logger.error(f"Error processing message {original_message_id}: {str(e)}")
-            # If we can't process the decision, send to fatal DLQ
             try:
                 error_message = {
                     'originalMessage': record['body'],
-                    'failureReason': f"Decision maker error: {str(e)}",
+                    'failureReason': f"Remediation error: {str(e)}",
                     'timestamp': context.invoked_function_arn
                 }
                 sqs.send_message(
                     QueueUrl=FATAL_DLQ_URL,
                     MessageBody=json.dumps(error_message)
                 )
-               
+                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}
+    ========================
+    """)
+
     return {
         'statusCode': 200,
         'body': json.dumps({
-            'processedMessages': processed_count
+            'processed': processed_count,
+            'fixed': fixed_count,
+            'fatal': fatal_count
         })
     }