New serverless pattern - Serverless Messaging Redrive

Author: zalilias

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

@@ -0,0 +1,127 @@
+import json
+import boto3
+import os
+import logging
+
+# Set up logging
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+# Initialize AWS clients
+sqs = boto3.client('sqs')
+
+# Environment variables
+MAIN_QUEUE_URL = os.environ['MAIN_QUEUE_URL']
+FATAL_DLQ_URL = os.environ['FATAL_DLQ_URL']
+
+def can_fix_message(message):
+    """
+    Determine if a message can be fixed automatically.
+    
+    Extension points:
+    1. Add validation for specific message formats
+    2. Implement business-specific fix rules
+    3. Add data transformation logic
+    4. Implement retry strategies
+    5. Add validation against external systems
+    """
+    try:
+        # Basic message validation
+        # Add your validation logic here
+        return True
+    except Exception as e:
+        logger.error(f"Validation error: {str(e)}")
+        return False
+
+def fix_message(message):
+    """
+    Apply fixes to the message.
+    
+    Extension points:
+    1. Add data normalization
+    2. Implement field-specific fixes
+    3. Add data enrichment
+    4. Implement format conversion
+    5. Add validation rules
+    """
+    try:
+        fixed_message = message.copy()
+        # Add your fix logic here
+        fixed_message['wasFixed'] = True
+        return fixed_message
+    except Exception as e:
+        logger.error(f"Fix error: {str(e)}")
+        return None
+
+def lambda_handler(event, context):
+    """
+    Process messages and route them based on fixability.
+    
+    Flow:
+    1. Attempt to fix message
+    2. If fixable -> Main Queue
+    3. If unfixable -> Fatal DLQ
+    
+    Extension points:
+    1. Add more sophisticated routing logic
+    2. Implement custom error handling
+    3. Add message transformation
+    4. Implement retry mechanisms
+    5. Add monitoring and metrics
+    """
+    processed_count = 0
+    
+    for record in event['Records']:
+        message_id = 'unknown'  # Initialize message_id with default value
+        
+        try:
+            message = json.loads(record['body'])
+            message_id = record.get('messageId', 'unknown')
+            
+            logger.info(f"Processing message: {message_id}")
+            
+            if can_fix_message(message):
+                fixed_message = fix_message(message)
+                if fixed_message:
+                    # Send to main queue
+                    sqs.send_message(
+                        QueueUrl=MAIN_QUEUE_URL,
+                        MessageBody=json.dumps(fixed_message)
+                    )
+                    logger.info(f"Fixed message sent to main queue: {message_id}")
+                else:
+                    raise ValueError("Message fix failed")
+            else:
+                # Send to fatal DLQ
+                message['failureReason'] = 'Message cannot be automatically fixed'
+                sqs.send_message(
+                    QueueUrl=FATAL_DLQ_URL,
+                    MessageBody=json.dumps(message)
+                )
+                logger.warning(f"Message sent to fatal DLQ: {message_id}")
+            
+            processed_count += 1
+            
+        except Exception as e:
+            logger.error(f"Error processing message {message_id}: {str(e)}")
+            try:
+                error_message = {
+                    'originalMessage': record['body'],
+                    'failureReason': str(e),
+                    'timestamp': context.invoked_function_arn
+                }
+                sqs.send_message(
+                    QueueUrl=FATAL_DLQ_URL,
+                    MessageBody=json.dumps(error_message)
+                )
+                logger.error(f"Error message sent to fatal DLQ: {message_id}")
+            except Exception as fatal_e:
+                logger.critical(f"Fatal DLQ error: {str(fatal_e)}")
+                raise
+
+    return {
+        'statusCode': 200,
+        'body': json.dumps({
+            'processedMessages': processed_count
+        })
+    }

serverless-message-processing/functions/decision_maker/requirements.txt

@@ -0,0 +1,2 @@
+boto3==1.26.137
+jsonschema==4.17.3

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

@@ -0,0 +1,121 @@
+import json
+import logging
+
+# Set up logging
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+def validate_message_structure(message):
+    """
+    Validate message structure and required fields.
+    Args:
+        message: Dictionary containing message data
+    Returns:
+        bool: True if valid message structure, False otherwise
+    """
+    required_fields = ['messageType', 'payload', 'timestamp']
+    return all(field in message for field in required_fields)
+
+def process_message(message):
+    """
+    Process the message content.
+    Args:
+        message: Dictionary containing message data
+    Returns:
+        bool: True if processing successful, False otherwise
+    """
+    try:
+        # Validate message structure
+        if not validate_message_structure(message):
+            logger.error("Message missing required fields")
+            raise ValueError("Invalid message structure")
+
+        message_type = message['messageType']
+        payload = message['payload']
+        
+        # Validate message type
+        valid_types = ['TYPE_A', 'TYPE_B', 'TYPE_C']
+        if message_type not in valid_types:
+            logger.error(f"Invalid message type: {message_type}")
+            raise ValueError(f"Invalid message type: {message_type}")
+
+        # Check for downstream system status
+        if 'systemStatus' in message and message['systemStatus'].lower() == 'unavailable':
+            logger.error("Target system is unavailable")
+            raise ValueError("DOWNSTREAM_ERROR: Target system unavailable")
+
+        # Process the message based on type
+        logger.info(f"Processing message type: {message_type}")
+        
+        # Add type-specific processing logic here
+        if message_type == 'TYPE_A':
+            # Process TYPE_A messages
+            pass
+        elif message_type == 'TYPE_B':
+            # Process TYPE_B messages
+            pass
+        elif message_type == 'TYPE_C':
+            # Process TYPE_C messages
+            pass
+
+        return True
+
+    except Exception as e:
+        logger.error(f"Error processing message: {str(e)}")
+        raise
+
+def lambda_handler(event, context):
+    """
+    Main Lambda handler function.
+    Args:
+        event: Lambda event object
+        context: Lambda context object
+    Returns:
+        dict: Response object
+    """
+    logger.info(f"Processing {len(event['Records'])} messages")
+    
+    processed_count = 0
+    failed_count = 0
+    downstream_errors = 0
+
+    for record in event['Records']:
+        try:
+            # Parse the message body
+            message = json.loads(record['body'])
+            
+            # Process the message
+            if process_message(message):
+                processed_count += 1
+                logger.info(f"Successfully processed message: {message.get('messageId', 'unknown')}")
+            else:
+                failed_count += 1
+                logger.warning(f"Message processing returned False: {message.get('messageId', 'unknown')}")
+
+        except json.JSONDecodeError as e:
+            failed_count += 1
+            logger.error(f"Invalid JSON in message: {str(e)}")
+            raise
+
+        except ValueError as e:
+            if "DOWNSTREAM_ERROR" in str(e):
+                downstream_errors += 1
+                logger.error("Downstream error detected")
+                raise
+            failed_count += 1
+            logger.error(f"Validation error: {str(e)}")
+            raise
+
+        except Exception as e:
+            failed_count += 1
+            logger.error(f"Unexpected error processing message: {str(e)}")
+            raise
+
+    return {
+        'statusCode': 200,
+        'body': json.dumps({
+            'processed': processed_count,
+            'failed': failed_count,
+            'downstream_errors': downstream_errors
+        })
+    }

serverless-message-processing/functions/processor/requirements.txt

@@ -0,0 +1,2 @@
+boto3==1.26.137
+jsonschema==4.17.3