# lambda_handlers/event_processor.py
import json
import boto3
from datetime import datetime
import logging
from typing import Dict, Any

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize AWS services
dynamodb = boto3.resource('dynamodb')
s3 = boto3.client('s3')
sns = boto3.client('sns')
sqs = boto3.client('sqs')

def process_user_registration(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Process user registration events from API Gateway
    Demonstrates: API Gateway → Lambda → DynamoDB → SNS pattern
    """
    try:
        # Parse the incoming event
        body = json.loads(event.get('body', '{}'))
        user_data = {
            'user_id': body['user_id'],
            'email': body['email'],
            'name': body['name'],
            'registration_date': datetime.utcnow().isoformat(),
            'status': 'pending_verification'
        }

        # Store user in DynamoDB
        table = dynamodb.Table('users')
        table.put_item(Item=user_data)

        # Send welcome email via SNS
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:user-notifications',
            Message=json.dumps({
                'type': 'welcome_email',
                'user_data': user_data
            }),
            Subject='New User Registration'
        )

        # Queue additional processing tasks
        sqs.send_message(
            QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/user-onboarding',
            MessageBody=json.dumps({
                'user_id': user_data['user_id'],
                'task': 'setup_user_profile'
            })
        )

        logger.info(f"Successfully processed registration for user: {user_data['user_id']}")

        return {
            'statusCode': 201,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'message': 'User registered successfully',
                'user_id': user_data['user_id']
            })
        }

    except KeyError as e:
        logger.error(f"Missing required field: {str(e)}")
        return {
            'statusCode': 400,
            'body': json.dumps({'error': f'Missing required field: {str(e)}'})
        }

    except Exception as e:
        logger.error(f"Registration processing failed: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

def process_image_upload(event: Dict[str, Any], context) -> None:
    """
    Process image uploads from S3
    Demonstrates: S3 → Lambda → Rekognition → DynamoDB pattern
    """
    try:
        # Parse S3 event
        for record in event['Records']:
            bucket = record['s3']['bucket']['name']
            key = record['s3']['object']['key']

            logger.info(f"Processing image upload: {bucket}/{key}")

            # Initialize Rekognition client
            rekognition = boto3.client('rekognition')

            # Detect labels in the image
            response = rekognition.detect_labels(
                Image={
                    'S3Object': {
                        'Bucket': bucket,
                        'Name': key
                    }
                },
                MaxLabels=10,
                MinConfidence=80
            )

            # Extract metadata
            labels = [label['Name'] for label in response['Labels']]

            # Store analysis results
            table = dynamodb.Table('image_analysis')
            table.put_item(
                Item={
                    'image_key': key,
                    'bucket': bucket,
                    'labels': labels,
                    'processed_date': datetime.utcnow().isoformat(),
                    'confidence_scores': {
                        label['Name']: label['Confidence']
                        for label in response['Labels']
                    }
                }
            )

            # Generate thumbnail asynchronously
            sqs.send_message(
                QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/thumbnail-generation',
                MessageBody=json.dumps({
                    'bucket': bucket,
                    'key': key,
                    'task': 'generate_thumbnail'
                })
            )

            logger.info(f"Successfully processed image: {key}")

    except Exception as e:
        logger.error(f"Image processing failed: {str(e)}")
        # In production, you might want to send this to a dead letter queue
        raise

def process_batch_analytics(event: Dict[str, Any], context) -> None:
    """
    Process analytics data in batches
    Demonstrates: EventBridge → Lambda → Athena → S3 pattern
    """
    try:
        # Initialize services
        athena = boto3.client('athena')

        # Define the analytics query
        query = """
        SELECT
            DATE(timestamp) as date,
            COUNT(*) as total_events,
            COUNT(DISTINCT user_id) as unique_users,
            AVG(session_duration) as avg_session_duration
        FROM user_events
        WHERE timestamp >= CURRENT_DATE - INTERVAL '7' DAY
        GROUP BY DATE(timestamp)
        ORDER BY date DESC
        """

        # Execute query
        response = athena.start_query_execution(
            QueryString=query,
            QueryExecutionContext={
                'Database': 'analytics_db'
            },
            ResultConfiguration={
                'OutputLocation': 's3://analytics-results-bucket/daily-reports/'
            }
        )

        query_execution_id = response['QueryExecutionId']

        # Store query execution info for later processing
        table = dynamodb.Table('analytics_jobs')
        table.put_item(
            Item={
                'job_id': query_execution_id,
                'job_type': 'daily_analytics',
                'status': 'running',
                'created_date': datetime.utcnow().isoformat()
            }
        )

        logger.info(f"Started analytics job: {query_execution_id}")

    except Exception as e:
        logger.error(f"Analytics processing failed: {str(e)}")
        raise

# lambda_handlers/api_handler.py
import json
import boto3
from decimal import Decimal
import logging
from typing import Dict, Any, Optional
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Custom JSON encoder for DynamoDB Decimal types
class DecimalEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return super(DecimalEncoder, self).default(obj)

def create_cors_response(status_code: int, body: Dict[str, Any]) -> Dict[str, Any]:
    """Create a CORS-enabled response"""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
            'Access-Control-Allow-Headers': 'Content-Type, Authorization'
        },
        'body': json.dumps(body, cls=DecimalEncoder)
    }

def handle_options(event: Dict[str, Any], context) -> Dict[str, Any]:
    """Handle CORS preflight requests"""
    return create_cors_response(200, {'message': 'OK'})

def get_user_profile(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    GET /users/{user_id}
    Retrieve user profile with caching
    """
    try:
        # Extract user ID from path parameters
        user_id = event['pathParameters']['user_id']

        # Initialize services
        dynamodb = boto3.resource('dynamodb')
        elasticache = boto3.client('elasticache')

        # Try to get from cache first (Redis)
        cache_key = f"user_profile:{user_id}"

        try:
            # In a real implementation, you'd use a Redis client
            # This is a simplified example
            cached_data = None  # get_from_cache(cache_key)

            if cached_data:
                logger.info(f"Cache hit for user: {user_id}")
                return create_cors_response(200, cached_data)

        except Exception as cache_error:
            logger.warning(f"Cache error: {str(cache_error)}")

        # Get from DynamoDB
        table = dynamodb.Table('users')
        response = table.get_item(Key={'user_id': user_id})

        if 'Item' not in response:
            return create_cors_response(404, {'error': 'User not found'})

        user_data = response['Item']

        # Remove sensitive information
        user_data.pop('password_hash', None)
        user_data.pop('internal_notes', None)

        # Cache the result
        # set_cache(cache_key, user_data, ttl=300)  # 5 minutes

        logger.info(f"Retrieved user profile: {user_id}")
        return create_cors_response(200, user_data)

    except KeyError:
        return create_cors_response(400, {'error': 'Missing user_id parameter'})

    except ClientError as e:
        logger.error(f"DynamoDB error: {str(e)}")
        return create_cors_response(500, {'error': 'Database error'})

    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        return create_cors_response(500, {'error': 'Internal server error'})

def update_user_profile(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    PUT /users/{user_id}
    Update user profile with validation
    """
    try:
        user_id = event['pathParameters']['user_id']
        body = json.loads(event.get('body', '{}'))

        # Validate input
        allowed_fields = {'name', 'email', 'phone', 'preferences'}
        update_data = {k: v for k, v in body.items() if k in allowed_fields}

        if not update_data:
            return create_cors_response(400, {'error': 'No valid fields to update'})

        # Add timestamp
        update_data['last_modified'] = datetime.utcnow().isoformat()

        # Build DynamoDB update expression
        update_expression = "SET "
        expression_values = {}
        expression_names = {}

        for key, value in update_data.items():
            attr_name = f"#{key}"
            attr_value = f":{key}"

            update_expression += f"{attr_name} = {attr_value}, "
            expression_values[attr_value] = value
            expression_names[attr_name] = key

        update_expression = update_expression.rstrip(', ')

        # Update DynamoDB
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table('users')

        response = table.update_item(
            Key={'user_id': user_id},
            UpdateExpression=update_expression,
            ExpressionAttributeValues=expression_values,
            ExpressionAttributeNames=expression_names,
            ReturnValues='ALL_NEW',
            ConditionExpression='attribute_exists(user_id)'
        )

        updated_user = response['Attributes']

        # Invalidate cache
        # delete_from_cache(f"user_profile:{user_id}")

        # Send update notification
        sns = boto3.client('sns')
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:user-updates',
            Message=json.dumps({
                'user_id': user_id,
                'updated_fields': list(update_data.keys()),
                'timestamp': update_data['last_modified']
            })
        )

        logger.info(f"Updated user profile: {user_id}")
        return create_cors_response(200, updated_user)

    except json.JSONDecodeError:
        return create_cors_response(400, {'error': 'Invalid JSON in request body'})

    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            return create_cors_response(404, {'error': 'User not found'})
        logger.error(f"DynamoDB error: {str(e)}")
        return create_cors_response(500, {'error': 'Database error'})

    except Exception as e:
        logger.error(f"Update error: {str(e)}")
        return create_cors_response(500, {'error': 'Internal server error'})

def search_users(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    GET /users/search?q=query&limit=20&offset=0
    Search users with pagination
    """
    try:
        # Extract query parameters
        query_params = event.get('queryStringParameters') or {}
        search_query = query_params.get('q', '')
        limit = min(int(query_params.get('limit', 20)), 100)  # Max 100 results
        offset = int(query_params.get('offset', 0))

        if not search_query:
            return create_cors_response(400, {'error': 'Search query is required'})

        # Use OpenSearch for full-text search
        opensearch = boto3.client('opensearch')

        search_body = {
            "query": {
                "multi_match": {
                    "query": search_query,
                    "fields": ["name", "email", "bio"],
                    "type": "best_fields",
                    "fuzziness": "AUTO"
                }
            },
            "from": offset,
            "size": limit,
            "sort": [
                {"_score": {"order": "desc"}},
                {"created_date": {"order": "desc"}}
            ],
            "_source": ["user_id", "name", "email", "created_date", "profile_image"]
        }

        # Note: This is a simplified example. In practice, you'd use the opensearch-py library
        # search_response = opensearch_client.search(index="users", body=search_body)

        # Mock response for demonstration
        mock_results = {
            "hits": {
                "total": {"value": 42},
                "hits": [
                    {
                        "_source": {
                            "user_id": "user123",
                            "name": "John Doe",
                            "email": "[email protected]",
                            "created_date": "2024-01-15T10:30:00Z"
                        },
                        "_score": 1.5
                    }
                ]
            }
        }

        users = [hit["_source"] for hit in mock_results["hits"]["hits"]]
        total_count = mock_results["hits"]["total"]["value"]

        response_data = {
            "users": users,
            "pagination": {
                "total": total_count,
                "limit": limit,
                "offset": offset,
                "has_more": offset + limit < total_count
            }
        }

        logger.info(f"Search completed: {len(users)} results for '{search_query}'")
        return create_cors_response(200, response_data)

    except ValueError as e:
        logger.error(f"Parameter validation error: {str(e)}")
        return create_cors_response(400, {'error': 'Invalid query parameters'})

    except Exception as e:
        logger.error(f"Search error: {str(e)}")
        return create_cors_response(500, {'error': 'Search service error'})

# API Gateway routing handler
def api_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Main API handler that routes requests to appropriate functions
    """
    try:
        http_method = event['httpMethod']
        resource_path = event['resource']

        # Route based on method and path
        if http_method == 'OPTIONS':
            return handle_options(event, context)

        elif http_method == 'GET' and resource_path == '/users/{user_id}':
            return get_user_profile(event, context)

        elif http_method == 'PUT' and resource_path == '/users/{user_id}':
            return update_user_profile(event, context)

        elif http_method == 'GET' and resource_path == '/users/search':
            return search_users(event, context)

        else:
            return create_cors_response(404, {'error': 'Endpoint not found'})

    except Exception as e:
        logger.error(f"Routing error: {str(e)}")
        return create_cors_response(500, {'error': 'Internal server error'})

# lambda_handlers/stream_processor.py
import json
import base64
import boto3
from datetime import datetime, timedelta
import logging
from typing import Dict, Any, List
from collections import defaultdict

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def process_kinesis_stream(event: Dict[str, Any], context) -> None:
    """
    Process real-time data streams from Kinesis
    Demonstrates: Kinesis → Lambda → DynamoDB/S3 pattern
    """
    try:
        # Initialize AWS services
        dynamodb = boto3.resource('dynamodb')
        s3 = boto3.client('s3')
        cloudwatch = boto3.client('cloudwatch')

        # Batch processing for efficiency
        batch_writes = defaultdict(list)
        metrics_data = defaultdict(int)
        anomalies = []

        for record in event['Records']:
            # Decode Kinesis data
            payload = base64.b64decode(record['kinesis']['data'])
            data = json.loads(payload)

            # Process different event types
            event_type = data.get('event_type')

            if event_type == 'user_activity':
                process_user_activity(data, batch_writes, metrics_data)

            elif event_type == 'system_metric':
                process_system_metric(data, batch_writes, metrics_data, anomalies)

            elif event_type == 'transaction':
                process_transaction(data, batch_writes, metrics_data, anomalies)

            else:
                logger.warning(f"Unknown event type: {event_type}")

        # Batch write to DynamoDB
        write_batch_to_dynamodb(dynamodb, batch_writes)

        # Send metrics to CloudWatch
        send_metrics_to_cloudwatch(cloudwatch, metrics_data)

        # Handle anomalies
        if anomalies:
            handle_anomalies(anomalies)

        # Archive processed data to S3
        archive_processed_data(s3, event['Records'])

        logger.info(f"Processed {len(event['Records'])} records successfully")

    except Exception as e:
        logger.error(f"Stream processing failed: {str(e)}")
        raise

def process_user_activity(data: Dict[str, Any], batch_writes: Dict, metrics_data: Dict) -> None:
    """Process user activity events"""
    user_id = data['user_id']
    activity_type = data['activity_type']
    timestamp = data['timestamp']

    # Prepare DynamoDB write
    activity_record = {
        'user_id': user_id,
        'timestamp': timestamp,
        'activity_type': activity_type,
        'details': data.get('details', {}),
        'session_id': data.get('session_id'),
        'ip_address': data.get('ip_address')
    }

    batch_writes['user_activities'].append({
        'PutRequest': {'Item': activity_record}
    })

    # Update metrics
    metrics_data[f'activity_{activity_type}'] += 1
    metrics_data['total_activities'] += 1

def process_system_metric(data: Dict[str, Any], batch_writes: Dict, metrics_data: Dict, anomalies: List) -> None:
    """Process system performance metrics"""
    metric_name = data['metric_name']
    value = data['value']
    timestamp = data['timestamp']
    source = data['source']

    # Store in DynamoDB for historical analysis
    metric_record = {
        'metric_id': f"{source}#{metric_name}#{timestamp}",
        'metric_name': metric_name,
        'value': value,
        'timestamp': timestamp,
        'source': source,
        'tags': data.get('tags', {})
    }

    batch_writes['system_metrics'].append({
        'PutRequest': {'Item': metric_record}
    })

    # Anomaly detection
    if detect_metric_anomaly(metric_name, value, source):
        anomalies.append({
            'type': 'metric_anomaly',
            'metric_name': metric_name,
            'value': value,
            'source': source,
            'timestamp': timestamp,
            'severity': get_anomaly_severity(metric_name, value)
        })

    # Update CloudWatch metrics
    metrics_data[f'system_{metric_name}'] = value

def process_transaction(data: Dict[str, Any], batch_writes: Dict, metrics_data: Dict, anomalies: List) -> None:
    """Process financial transactions"""
    transaction_id = data['transaction_id']
    amount = data['amount']
    user_id = data['user_id']
    timestamp = data['timestamp']

    # Store transaction
    transaction_record = {
        'transaction_id': transaction_id,
        'user_id': user_id,
        'amount': amount,
        'currency': data['currency'],
        'timestamp': timestamp,
        'merchant': data.get('merchant'),
        'category': data.get('category'),
        'status': data.get('status', 'pending')
    }

    batch_writes['transactions'].append({
        'PutRequest': {'Item': transaction_record}
    })

    # Fraud detection
    if detect_fraud_patterns(data):
        anomalies.append({
            'type': 'fraud_alert',
            'transaction_id': transaction_id,
            'user_id': user_id,
            'amount': amount,
            'timestamp': timestamp,
            'severity': 'high',
            'reason': 'Suspicious transaction pattern detected'
        })

    # Update transaction metrics
    metrics_data['total_transactions'] += 1
    metrics_data['transaction_volume'] += amount

def detect_metric_anomaly(metric_name: str, value: float, source: str) -> bool:
    """Simple anomaly detection for system metrics"""
    # Define thresholds for different metrics
    thresholds = {
        'cpu_usage': 90.0,
        'memory_usage': 85.0,
        'disk_usage': 90.0,
        'response_time': 5000.0,  # milliseconds
        'error_rate': 5.0  # percentage
    }

    threshold = thresholds.get(metric_name)
    if threshold and value > threshold:
        return True

    # Check for sudden spikes (simplified)
    if metric_name in ['request_count', 'error_count']:
        # In production, you'd compare with historical averages
        baseline = get_metric_baseline(metric_name, source)
        if value > baseline * 3:  # 3x normal
            return True

    return False

def detect_fraud_patterns(transaction_data: Dict[str, Any]) -> bool:
    """Simple fraud detection patterns"""
    amount = transaction_data['amount']
    user_id = transaction_data['user_id']

    # Large transaction amount
    if amount > 10000:
        return True

    # Multiple rapid transactions (simplified check)
    # In production, you'd query recent transactions
    recent_transactions = get_recent_transactions(user_id)
    if len(recent_transactions) > 5:  # More than 5 in short period
        return True

    # Unusual location (would need geolocation data)
    # This is where you'd implement more sophisticated ML-based detection

    return False

def get_metric_baseline(metric_name: str, source: str) -> float:
    """Get historical baseline for metric (simplified)"""
    # In production, this would query historical data
    baselines = {
        'request_count': 1000,
        'error_count': 10,
        'response_time': 200
    }
    return baselines.get(metric_name, 100)

def get_recent_transactions(user_id: str) -> List[Dict]:
    """Get recent transactions for user (simplified)"""
    # In production, this would query DynamoDB or cache
    return []  # Simplified for demo

def get_anomaly_severity(metric_name: str, value: float) -> str:
    """Determine anomaly severity"""
    critical_metrics = ['cpu_usage', 'memory_usage', 'error_rate']
    if metric_name in critical_metrics and value > 95:
        return 'critical'
    elif value > 90:
        return 'high'
    elif value > 80:
        return 'medium'
    else:
        return 'low'

def write_batch_to_dynamodb(dynamodb, batch_writes: Dict) -> None:
    """Write batched data to DynamoDB tables"""
    for table_name, items in batch_writes.items():
        if not items:
            continue

        try:
            table = dynamodb.Table(table_name)

            # DynamoDB batch write limit is 25 items
            for i in range(0, len(items), 25):
                batch = items[i:i+25]

                response = dynamodb.batch_write_item(
                    RequestItems={
                        table_name: batch
                    }
                )

                # Handle unprocessed items
                unprocessed = response.get('UnprocessedItems', {})
                if unprocessed:
                    logger.warning(f"Unprocessed items in {table_name}: {len(unprocessed)}")
                    # In production, you'd retry unprocessed items

        except Exception as e:
            logger.error(f"Error writing to {table_name}: {str(e)}")
            raise

def send_metrics_to_cloudwatch(cloudwatch, metrics_data: Dict) -> None:
    """Send custom metrics to CloudWatch"""
    try:
        metric_data = []

        for metric_name, value in metrics_data.items():
            metric_data.append({
                'MetricName': metric_name,
                'Value': value,
                'Unit': 'Count',
                'Timestamp': datetime.utcnow()
            })

        # CloudWatch allows max 20 metrics per call
        for i in range(0, len(metric_data), 20):
            batch = metric_data[i:i+20]

            cloudwatch.put_metric_data(
                Namespace='StreamProcessing',
                MetricData=batch
            )

        logger.info(f"Sent {len(metric_data)} metrics to CloudWatch")

    except Exception as e:
        logger.error(f"Error sending metrics to CloudWatch: {str(e)}")

def handle_anomalies(anomalies: List[Dict]) -> None:
    """Handle detected anomalies"""
    try:
        sns = boto3.client('sns')

        for anomaly in anomalies:
            severity = anomaly.get('severity', 'medium')

            # Send immediate alerts for high-severity anomalies
            if severity in ['critical', 'high']:
                sns.publish(
                    TopicArn='arn:aws:sns:us-east-1:123456789012:anomaly-alerts',
                    Message=json.dumps(anomaly),
                    Subject=f'{severity.upper()} Anomaly Detected'
                )

            # Log all anomalies
            logger.warning(f"Anomaly detected: {anomaly}")

    except Exception as e:
        logger.error(f"Error handling anomalies: {str(e)}")

def archive_processed_data(s3, records: List[Dict]) -> None:
    """Archive processed records to S3 for data lake"""
    try:
        # Create archive data
        archive_data = {
            'processed_at': datetime.utcnow().isoformat(),
            'record_count': len(records),
            'records': []
        }

        for record in records:
            payload = base64.b64decode(record['kinesis']['data'])
            data = json.loads(payload)
            archive_data['records'].append(data)

        # Generate S3 key with partitioning
        now = datetime.utcnow()
        s3_key = f"processed-data/year={now.year}/month={now.month:02d}/day={now.day:02d}/hour={now.hour:02d}/{context.aws_request_id}.json"

        # Upload to S3
        s3.put_object(
            Bucket='data-lake-bucket',
            Key=s3_key,
            Body=json.dumps(archive_data),
            ContentType='application/json'
        )

        logger.info(f"Archived {len(records)} records to S3: {s3_key}")

    except Exception as e:
        logger.error(f"Error archiving data to S3: {str(e)}")

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: 'Serverless Application with Multiple Patterns'

Globals:
  Function:
    Timeout: 30
    MemorySize: 512
    Runtime: python3.9
    Environment:
      Variables:
        LOG_LEVEL: INFO
        POWERTOOLS_SERVICE_NAME: serverless-app
        POWERTOOLS_METRICS_NAMESPACE: ServerlessApp

Parameters:
  Environment:
    Type: String
    Default: dev
    AllowedValues: [dev, staging, prod]

Resources:
  # API Gateway
  ServerlessApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: !Ref Environment
      Cors:
        AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization'"
        AllowOrigin: "'*'"
      Auth:
        DefaultAuthorizer: CognitoAuthorizer
        Authorizers:
          CognitoAuthorizer:
            UserPoolArn: !GetAtt UserPool.Arn
      TracingConfig:
        TracingEnabled: true
      MethodSettings:
        - ResourcePath: "/*"
          HttpMethod: "*"
          LoggingLevel: INFO
          DataTraceEnabled: true
          MetricsEnabled: true

  # Lambda Functions
  ApiHandlerFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: lambda_handlers/
      Handler: api_handler.api_handler
      Events:
        GetUser:
          Type: Api
          Properties:
            RestApiId: !Ref ServerlessApi
            Path: /users/{user_id}
            Method: get
        UpdateUser:
          Type: Api
          Properties:
            RestApiId: !Ref ServerlessApi
            Path: /users/{user_id}
            Method: put
        SearchUsers:
          Type: Api
          Properties:
            RestApiId: !Ref ServerlessApi
            Path: /users/search
            Method: get
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UsersTable
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt UserUpdatesTopics.TopicName
      Environment:
        Variables:
          USERS_TABLE: !Ref UsersTable

  EventProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: lambda_handlers/
      Handler: event_processor.process_user_registration
      Events:
        UserRegistration:
          Type: Api
          Properties:
            RestApiId: !Ref ServerlessApi
            Path: /register
            Method: post
        ImageUpload:
          Type: S3
          Properties:
            Bucket: !Ref ImageBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: uploads/
                  - Name: suffix
                    Value: .jpg
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UsersTable
        - DynamoDBCrudPolicy:
            TableName: !Ref ImageAnalysisTable
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt UserNotificationsTopics.TopicName
        - SQSSendMessagePolicy:
            QueueName: !GetAtt UserOnboardingQueue.QueueName
        - RekognitionDetectOnlyPolicy: {}
        - S3ReadPolicy:
            BucketName: !Ref ImageBucket

  StreamProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: lambda_handlers/
      Handler: stream_processor.process_kinesis_stream
      Events:
        KinesisStream:
          Type: Kinesis
          Properties:
            Stream: !GetAtt DataStream.Arn
            StartingPosition: LATEST
            BatchSize: 100
            MaximumBatchingWindowInSeconds: 5
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UserActivitiesTable
        - DynamoDBCrudPolicy:
            TableName: !Ref SystemMetricsTable
        - DynamoDBCrudPolicy:
            TableName: !Ref TransactionsTable
        - CloudWatchPutMetricPolicy: {}
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt AnomalyAlertsTopics.TopicName
        - S3WritePolicy:
            BucketName: !Ref DataLakeBucket

  # DynamoDB Tables
  UsersTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: user_id
          AttributeType: S
      KeySchema:
        - AttributeName: user_id
          KeyType: HASH
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES
      PointInTimeRecoverySpecification:
        PointInTimeRecoveryEnabled: true
      SSESpecification:
        SSEEnabled: true
      Tags:
        - Key: Environment
          Value: !Ref Environment

  UserActivitiesTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: user_id
          AttributeType: S
        - AttributeName: timestamp
          AttributeType: S
      KeySchema:
        - AttributeName: user_id
          KeyType: HASH
        - AttributeName: timestamp
          KeyType: RANGE
      TimeToLiveSpecification:
        AttributeName: ttl
        Enabled: true
      Tags:
        - Key: Environment
          Value: !Ref Environment

  SystemMetricsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: metric_id
          AttributeType: S
      KeySchema:
        - AttributeName: metric_id
          KeyType: HASH
      TimeToLiveSpecification:
        AttributeName: ttl
        Enabled: true

  TransactionsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: transaction_id
          AttributeType: S
        - AttributeName: user_id
          AttributeType: S
      KeySchema:
        - AttributeName: transaction_id
          KeyType: HASH
      GlobalSecondaryIndexes:
        - IndexName: UserTransactionsIndex
          KeySchema:
            - AttributeName: user_id
              KeyType: HASH
          Projection:
            ProjectionType: ALL

  ImageAnalysisTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: image_key
          AttributeType: S
      KeySchema:
        - AttributeName: image_key
          KeyType: HASH

  # S3 Buckets
  ImageBucket:
    Type: AWS::S3::Bucket
    Properties:
      CorsConfiguration:
        CorsRules:
          - AllowedOrigins: ['*']
            AllowedMethods: [GET, PUT, POST, DELETE]
            AllowedHeaders: ['*']
      NotificationConfiguration:
        LambdaConfigurations:
          - Event: s3:ObjectCreated:*
            Function: !GetAtt EventProcessorFunction.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: uploads/

  DataLakeBucket:
    Type: AWS::S3::Bucket
    Properties:
      LifecycleConfiguration:
        Rules:
          - Status: Enabled
            Transitions:
              - TransitionInDays: 30
                StorageClass: STANDARD_IA
              - TransitionInDays: 90
                StorageClass: GLACIER

  # Kinesis Stream
  DataStream:
    Type: AWS::Kinesis::Stream
    Properties:
      ShardCount: 2
      RetentionPeriodHours: 168  # 7 days
      Tags:
        - Key: Environment
          Value: !Ref Environment

  # SNS Topics
  UserNotificationsTopics:
    Type: AWS::SNS::Topic
    Properties:
      DisplayName: User Notifications

  UserUpdatesTopics:
    Type: AWS::SNS::Topic
    Properties:
      DisplayName: User Updates

  AnomalyAlertsTopics:
    Type: AWS::SNS::Topic
    Properties:
      DisplayName: Anomaly Alerts

  # SQS Queues
  UserOnboardingQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeoutSeconds: 180
      MessageRetentionPeriod: 1209600  # 14 days
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt UserOnboardingDLQ.Arn
        maxReceiveCount: 3

  UserOnboardingDLQ:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600  # 14 days

  ThumbnailGenerationQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeoutSeconds: 300

  # Cognito User Pool
  UserPool:
    Type: AWS::Cognito::UserPool
    Properties:
      UserPoolName: !Sub "${AWS::StackName}-UserPool"
      AutoVerifiedAttributes:
        - email
      Policies:
        PasswordPolicy:
          MinimumLength: 8
          RequireUppercase: true
          RequireLowercase: true
          RequireNumbers: true
          RequireSymbols: true

  UserPoolClient:
    Type: AWS::Cognito::UserPoolClient
    Properties:
      UserPoolId: !Ref UserPool
      GenerateSecret: false
      ExplicitAuthFlows:
        - ALLOW_USER_PASSWORD_AUTH
        - ALLOW_REFRESH_TOKEN_AUTH

  # CloudWatch Dashboard
  ApplicationDashboard:
    Type: AWS::CloudWatch::Dashboard
    Properties:
      DashboardName: !Sub "${AWS::StackName}-Dashboard"
      DashboardBody: !Sub |
        {
          "widgets": [
            {
              "type": "metric",
              "properties": {
                "metrics": [
                  ["AWS/Lambda", "Duration", "FunctionName", "${ApiHandlerFunction}"],
                  ["AWS/Lambda", "Errors", "FunctionName", "${ApiHandlerFunction}"],
                  ["AWS/Lambda", "Invocations", "FunctionName", "${ApiHandlerFunction}"]
                ],
                "period": 300,
                "stat": "Average",
                "region": "${AWS::Region}",
                "title": "API Handler Performance"
              }
            },
            {
              "type": "metric",
              "properties": {
                "metrics": [
                  ["AWS/Kinesis", "IncomingRecords", "StreamName", "${DataStream}"],
                  ["AWS/Kinesis", "OutgoingRecords", "StreamName", "${DataStream}"]
                ],
                "period": 300,
                "stat": "Sum",
                "region": "${AWS::Region}",
                "title": "Kinesis Stream Activity"
              }
            }
          ]
        }

Outputs:
  ApiGatewayUrl:
    Description: "API Gateway endpoint URL"
    Value: !Sub "https://${ServerlessApi}.execute-api.${AWS::Region}.amazonaws.com/${Environment}/"
    Export:
      Name: !Sub "${AWS::StackName}-ApiUrl"

  UserPoolId:
    Description: "Cognito User Pool ID"
    Value: !Ref UserPool
    Export:
      Name: !Sub "${AWS::StackName}-UserPoolId"

  UserPoolClientId:
    Description: "Cognito User Pool Client ID"
    Value: !Ref UserPoolClient
    Export:
      Name: !Sub "${AWS::StackName}-UserPoolClientId"

  KinesisStreamName:
    Description: "Kinesis Stream Name"
    Value: !Ref DataStream
    Export:
      Name: !Sub "${AWS::StackName}-KinesisStream"

#!/bin/bash
# deploy.sh

set -e

# Configuration
STACK_NAME="serverless-app"
REGION="us-east-1"
ENVIRONMENT="dev"

echo "🚀 Deploying Serverless Application..."

# Install dependencies
pip install -r requirements.txt -t lambda_handlers/

# Build and deploy with SAM
sam build
sam deploy \
  --stack-name "$STACK_NAME-$ENVIRONMENT" \
  --s3-bucket "sam-deployments-$REGION" \
  --parameter-overrides Environment=$ENVIRONMENT \
  --capabilities CAPABILITY_IAM \
  --region $REGION

# Get outputs
API_URL=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME-$ENVIRONMENT" \
  --region $REGION \
  --query 'Stacks[0].Outputs[?OutputKey==`ApiGatewayUrl`].OutputValue' \
  --output text)

echo "✅ Deployment complete!"
echo "📡 API Endpoint: $API_URL"

# Run integration tests
echo "🧪 Running integration tests..."
python tests/integration_tests.py --api-url "$API_URL"

echo "🎉 All tests passed!"

# performance_optimizations.py
import json
import boto3
from functools import lru_cache
import os
from typing import List, Dict, Any

# Connection pooling for better performance
dynamodb = boto3.resource('dynamodb')
s3 = boto3.client('s3')

# Cache frequently accessed data
@lru_cache(maxsize=128)
def get_configuration(config_key: str) -> str:
    """Cache configuration values to reduce SSM calls"""
    ssm = boto3.client('ssm')
    response = ssm.get_parameter(Name=config_key, WithDecryption=True)
    return response['Parameter']['Value']

# Optimize DynamoDB queries
def batch_get_users(user_ids: List[str]) -> Dict[str, Any]:
    """Efficiently retrieve multiple users in batches"""
    table = dynamodb.Table('users')
    results = {}

    # DynamoDB batch_get_item limit is 100 items
    for i in range(0, len(user_ids), 100):
        batch = user_ids[i:i+100]
        keys = [{'user_id': uid} for uid in batch]

        response = dynamodb.batch_get_item(
            RequestItems={
                'users': {
                    'Keys': keys,
                    'ProjectionExpression': 'user_id, #name, email, created_date',
                    'ExpressionAttributeNames': {'#name': 'name'}
                }
            }
        )

        for item in response['Responses']['users']:
            results[item['user_id']] = item

    return results

# Memory-efficient processing
def process_large_dataset(s3_key: str):
    """Process large files without loading entire content into memory"""
    response = s3.get_object(Bucket='data-bucket', Key=s3_key)

    # Stream processing for large files
    for line in response['Body'].iter_lines():
        data = json.loads(line.decode('utf-8'))
        yield process_record(data)

def process_record(data):
    """Process individual record"""
    # Your record processing logic here
    return data

# Warm-up function to reduce cold starts
def warmup_handler(event, context):
    """Warm up function connections and cache"""
    # Initialize connections
    global dynamodb, s3

    # Warm up caches
    get_configuration('/app/database/connection_string')

    # Pre-load frequently accessed data
    cache_critical_data()

    return {'statusCode': 200, 'body': 'Warmed up'}

def cache_critical_data():
    """Pre-load critical data into memory"""
    # Cache lookup tables, configuration, etc.
    pass

# monitoring.py
import json
import time
from functools import wraps
from datetime import datetime
import boto3

# AWS X-Ray for distributed tracing
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

# Patch AWS SDK calls for automatic tracing
patch_all()

cloudwatch = boto3.client('cloudwatch')

def track_performance(func):
    """Decorator to track function performance"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()

        try:
            result = func(*args, **kwargs)

            # Track success metrics
            duration = (time.time() - start_time) * 1000  # milliseconds

            cloudwatch.put_metric_data(
                Namespace='ServerlessApp/Performance',
                MetricData=[
                    {
                        'MetricName': f'{func.__name__}_duration',
                        'Value': duration,
                        'Unit': 'Milliseconds'
                    },
                    {
                        'MetricName': f'{func.__name__}_success',
                        'Value': 1,
                        'Unit': 'Count'
                    }
                ]
            )

            return result

        except Exception as e:
            # Track error metrics
            cloudwatch.put_metric_data(
                Namespace='ServerlessApp/Errors',
                MetricData=[
                    {
                        'MetricName': f'{func.__name__}_error',
                        'Value': 1,
                        'Unit': 'Count',
                        'Dimensions': [
                            {
                                'Name': 'ErrorType',
                                'Value': type(e).__name__
                            }
                        ]
                    }
                ]
            )
            raise

    return wrapper

@xray_recorder.capture('database_operation')
@track_performance
def query_database(query_params):
    """Example of monitored database operation"""
    with xray_recorder.in_subsegment('dynamodb_query'):
        # Your database logic here
        pass

# cost-optimization.yaml
Resources:
  # Use scheduled scaling for predictable workloads
  ScheduledConcurrency:
    Type: AWS::Events::Rule
    Properties:
      ScheduleExpression: "cron(0 8 * * ? *)"  # 8 AM daily
      Targets:
        - Arn: !GetAtt ApiHandlerFunction.Arn
          Id: "WarmUpFunction"

  # Implement S3 lifecycle policies
  DataLakeBucket:
    Type: AWS::S3::Bucket
    Properties:
      LifecycleConfiguration:
        Rules:
          - Status: Enabled
            Transitions:
              - TransitionInDays: 30
                StorageClass: STANDARD_IA
              - TransitionInDays: 90
                StorageClass: GLACIER
              - TransitionInDays: 365
                StorageClass: DEEP_ARCHIVE
            ExpirationInDays: 2555  # 7 years

# tests/test_serverless_functions.py
import json
import pytest
import boto3
from moto import mock_dynamodb2, mock_s3, mock_sns
from lambda_handlers.api_handler import api_handler
from lambda_handlers.event_processor import process_user_registration
import os

@mock_dynamodb2
@mock_sns
class TestServerlessFunctions:

    def setup_method(self):
        """Set up test fixtures"""
        # Create mock DynamoDB table
        self.dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
        self.table = self.dynamodb.create_table(
            TableName='users',
            KeySchema=[
                {'AttributeName': 'user_id', 'KeyType': 'HASH'}
            ],
            AttributeDefinitions=[
                {'AttributeName': 'user_id', 'AttributeType': 'S'}
            ],
            BillingMode='PAY_PER_REQUEST'
        )

        # Create mock SNS topic
        self.sns = boto3.client('sns', region_name='us-east-1')
        self.topic = self.sns.create_topic(Name='user-notifications')

    def test_get_user_profile_success(self):
        """Test successful user profile retrieval"""
        # Insert test data
        self.table.put_item(
            Item={
                'user_id': 'test-user-123',
                'name': 'John Doe',
                'email': '[email protected]'
            }
        )

        # Create test event
        event = {
            'httpMethod': 'GET',
            'resource': '/users/{user_id}',
            'pathParameters': {'user_id': 'test-user-123'},
            'headers': {}
        }

        # Execute function
        response = api_handler(event, {})

        # Assert response
        assert response['statusCode'] == 200
        body = json.loads(response['body'])
        assert body['user_id'] == 'test-user-123'
        assert body['name'] == 'John Doe'

    def test_get_user_profile_not_found(self):
        """Test user profile not found scenario"""
        event = {
            'httpMethod': 'GET',
            'resource': '/users/{user_id}',
            'pathParameters': {'user_id': 'nonexistent-user'},
            'headers': {}
        }

        response = api_handler(event, {})

        assert response['statusCode'] == 404
        body = json.loads(response['body'])
        assert 'error' in body

    def test_process_user_registration(self):
        """Test user registration processing"""
        event = {
            'body': json.dumps({
                'user_id': 'new-user-123',
                'email': '[email protected]',
                'name': 'New User'
            })
        }

        response = process_user_registration(event, {})

        assert response['statusCode'] == 201

        # Verify user was created in DynamoDB
        item = self.table.get_item(Key={'user_id': 'new-user-123'})
        assert 'Item' in item
        assert item['Item']['email'] == '[email protected]'

# Integration tests
def test_end_to_end_user_flow():
    """Test complete user registration and profile flow"""
    import requests

    api_base_url = os.environ.get('API_BASE_URL', 'http://localhost:3000')

    # Register user
    registration_data = {
        'user_id': 'e2e-test-user',
        'email': '[email protected]',
        'name': 'E2E Test User'
    }

    response = requests.post(f'{api_base_url}/register', json=registration_data)
    assert response.status_code == 201

    # Retrieve user profile
    response = requests.get(f'{api_base_url}/users/e2e-test-user')
    assert response.status_code == 200

    user_data = response.json()
    assert user_data['email'] == '[email protected]'

if __name__ == '__main__':
    pytest.main([__file__])

# security.py
import jwt
from functools import wraps
import boto3
from botocore.exceptions import ClientError
import json
import os

def authenticate_request(func):
    """Decorator for JWT authentication"""
    @wraps(func)
    def wrapper(event, context):
        try:
            # Extract JWT token
            auth_header = event.get('headers', {}).get('Authorization', '')
            if not auth_header.startswith('Bearer '):
                return create_error_response(401, 'Missing or invalid authorization header')

            token = auth_header[7:]  # Remove 'Bearer ' prefix

            # Verify JWT token
            # In production, get the secret from AWS Secrets Manager
            secret = get_jwt_secret()
            payload = jwt.decode(token, secret, algorithms=['HS256'])

            # Add user info to event context
            event['user'] = payload

            return func(event, context)

        except jwt.ExpiredSignatureError:
            return create_error_response(401, 'Token expired')
        except jwt.InvalidTokenError:
            return create_error_response(401, 'Invalid token')
        except Exception as e:
            return create_error_response(500, 'Authentication error')

    return wrapper

def get_jwt_secret():
    """Retrieve JWT secret from AWS Secrets Manager"""
    secrets_manager = boto3.client('secretsmanager')
    try:
        response = secrets_manager.get_secret_value(SecretId='jwt-secret')
        return response['SecretString']
    except ClientError:
        # Fallback to environment variable for development
        return os.environ.get('JWT_SECRET', 'dev-secret-key')

def validate_input(schema):
    """Decorator for input validation"""
    def decorator(func):
        @wraps(func)
        def wrapper(event, context):
            try:
                body = json.loads(event.get('body', '{}'))

                # Validate against schema (using jsonschema library)
                from jsonschema import validate, ValidationError
                validate(body, schema)

                event['validated_body'] = body
                return func(event, context)

            except json.JSONDecodeError:
                return create_error_response(400, 'Invalid JSON')
            except ValidationError as e:
                return create_error_response(400, f'Validation error: {e.message}')

        return wrapper
    return decorator

# Input validation schemas
USER_REGISTRATION_SCHEMA = {
    "type": "object",
    "properties": {
        "user_id": {"type": "string", "minLength": 1},
        "email": {"type": "string", "format": "email"},
        "name": {"type": "string", "minLength": 1, "maxLength": 100}
    },
    "required": ["user_id", "email", "name"]
}

@authenticate_request
@validate_input(USER_REGISTRATION_SCHEMA)
def secure_user_registration(event, context):
    """Secure user registration with authentication and validation"""
    user_data = event['validated_body']
    current_user = event['user']

    # Your secure registration logic here
    pass

def create_error_response(status_code, message):
    """Create standardized error response"""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps({'error': message})
    }