# Lambda@Edge function for request manipulation
import json
import base64
import re
from typing import Dict, Any

def handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """Lambda@Edge handler for request manipulation"""
    request = event['Records'][0]['cf']['request']
    headers = request['headers']

    # Add custom headers
    headers['x-edge-location'] = [{
        'key': 'X-Edge-Location',
        'value': event['Records'][0]['cf']['config']['distributionId']
    }]

    # Route based on device type
    user_agent = headers.get('user-agent', [{}])[0].get('value', '')
    if 'Mobile' in user_agent:
        request['uri'] = '/mobile' + request['uri']
    elif 'Tablet' in user_agent:
        request['uri'] = '/tablet' + request['uri']

    # A/B testing
    import hashlib
    visitor_id = headers.get('cookie', [{}])[0].get('value', '')
    hash_value = int(hashlib.md5(visitor_id.encode()).hexdigest(), 16)
    variant = 'A' if hash_value % 2 == 0 else 'B'
    headers['x-ab-variant'] = [{
        'key': 'X-AB-Variant',
        'value': variant
    }]

    return request

def response_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """Lambda@Edge response handler"""
    response = event['Records'][0]['cf']['response']
    headers = response['headers']

    # Add security headers
    headers['strict-transport-security'] = [{
        'key': 'Strict-Transport-Security',
        'value': 'max-age=31536000; includeSubDomains'
    }]

    headers['x-content-type-options'] = [{
        'key': 'X-Content-Type-Options',
        'value': 'nosniff'
    }]

    # Cache optimization
    uri = event['Records'][0]['cf']['request']['uri']
    if uri.startswith('/api/'):
        headers['cache-control'] = [{
            'key': 'Cache-Control',
            'value': 'no-cache, no-store, must-revalidate'
        }]
    else:
        headers['cache-control'] = [{
            'key': 'Cache-Control',
            'value': 'public, max-age=31536000'
        }]

    return response

def origin_request_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """Lambda@Edge origin request handler"""
    request = event['Records'][0]['cf']['request']

    # Add authentication headers
    request['headers']['x-api-key'] = [{
        'key': 'X-API-Key',
        'value': 'your-api-key'
    }]

    # Modify origin based on path
    if request['uri'].startswith('/graphql'):
        request['origin']['custom']['domainName'] = 'graphql.example.com'
        request['origin']['custom']['port'] = 443
        request['origin']['custom']['protocol'] = 'https'

    return request
```terraform
# CloudFront distribution with Lambda@Edge
resource "aws_cloudfront_distribution" "main" {
  origin {
    domain_name = aws_s3_bucket.website.bucket_regional_domain_name
    origin_id   = "S3-Website"

    s3_origin_config {
      origin_access_identity = aws_cloudfront_origin_access_identity.website.cloudfront_access_identity_path
    }
  }

  origin {
    domain_name = "api.example.com"
    origin_id   = "API-Origin"

    custom_origin_config {
      http_port              = 80
      https_port             = 443
      origin_protocol_policy = "https-only"
      origin_ssl_protocols   = ["TLSv1.2"]
    }
  }

  enabled             = true
  is_ipv6_enabled     = true
  default_root_object = "index.html"
  price_class         = "PriceClass_All"

  aliases = ["www.example.com"]

  default_cache_behavior {
    allowed_methods        = ["GET", "HEAD", "OPTIONS"]
    cached_methods         = ["GET", "HEAD"]
    target_origin_id       = "S3-Website"
    viewer_protocol_policy = "redirect-to-https"
    min_ttl                = 0
    default_ttl            = 86400
    max_ttl                = 31536000
    compress               = true

    lambda_function_association {
      event_type   = "viewer-request"
      lambda_arn   = aws_lambda_edge_function.request.qualified_arn
      include_body = false
    }

    lambda_function_association {
      event_type   = "viewer-response"
      lambda_arn   = aws_lambda_edge_function.response.qualified_arn
      include_body = false
    }
  }

  ordered_cache_behavior {
    path_pattern           = "/api/*"
    allowed_methods        = ["GET", "HEAD", "OPTIONS", "PUT", "POST", "PATCH", "DELETE"]
    cached_methods         = ["GET", "HEAD"]
    target_origin_id       = "API-Origin"
    viewer_protocol_policy = "https-only"
    min_ttl                = 0
    default_ttl            = 0
    max_ttl                = 300

    lambda_function_association {
      event_type   = "origin-request"
      lambda_arn   = aws_lambda_edge_function.origin_request.qualified_arn
      include_body = false
    }
  }

  restrictions {
    geo_restriction {
      restriction_type = "none"
    }
  }

  viewer_certificate {
    acm_certificate_arn      = aws_acm_certificate.main.arn
    ssl_support_method       = "sni-only"
    minimum_protocol_version = "TLSv1.2_2021"
  }

  tags = {
    Environment = var.environment
  }
}

// Cloudflare Worker for edge computing
export default {
  async fetch(request, env, ctx) {
    const url = new URL(request.url);

    // A/B testing
    const visitorId = request.headers.get('cf-visitor-id');
    const variant = this.getVariant(visitorId);

    // Route based on variant
    if (variant === 'A') {
      return this.handleVariantA(request, env);
    } else {
      return this.handleVariantB(request, env);
    }
  },

  getVariant(visitorId) {
    // Simple hash-based A/B testing
    let hash = 0;
    for (let i = 0; i < visitorId.length; i++) {
      hash = ((hash << 5) - hash) + visitorId.charCodeAt(i);
      hash |= 0;
    }
    return hash % 2 === 0 ? 'A' : 'B';
  },

  async handleVariantA(request, env) {
    // Variant A implementation
    return new Response('Variant A', {
      headers: { 'Content-Type': 'text/html' }
    });
  },

  async handleVariantB(request, env) {
    // Variant B implementation
    return new Response('Variant B', {
      headers: { 'Content-Type': 'text/html' }
    });
  }
};

// Edge caching with KV
export class EdgeCache &#123;
  constructor(kv) &#123;
    this.kv = kv;
  &#125;

  async get(key) &#123;
    const value = await this.kv.get(key, &#123; type: 'json' &#125;);
    return value;
  &#125;

  async set(key, value, ttl = 3600) &#123;
    await this.kv.put(key, JSON.stringify(value), &#123;
      expirationTtl: ttl
    &#125;);
  &#125;

  async invalidate(key) &#123;
    await this.kv.delete(key);
  &#125;

  async getOrSet(key, factory, ttl = 3600) &#123;
    let value = await this.get(key);
    if (value === null) &#123;
      value = await factory();
      await this.set(key, value, ttl);
    &#125;
    return value;
  &#125;
&#125;

// Edge rate limiting
export class EdgeRateLimiter {
  constructor(kv) {
    this.kv = kv;
  }

  async isAllowed(key, limit, windowSeconds) {
    const now = Math.floor(Date.now() / 1000);
    const windowKey = `rate:${key}:${Math.floor(now / windowSeconds)}`;

    const current = await this.kv.get(windowKey, { type: 'json' }) || { count: 0 };

    if (current.count >= limit) {
      return false;
    }

    current.count += 1;
    await this.kv.put(windowKey, JSON.stringify(current), {
      expirationTtl: windowSeconds * 2
    });

    return true;
  }
}

# IoT edge computing with AWS Greengrass
import boto3
import json
from typing import Dict, Any, Callable
from dataclasses import dataclass

@dataclass
class EdgeConfig:
    device_id: str
    region: str
    local_mqtt_port: int = 8883
    cloud_endpoint: str = ""

class IoTEdgeManager:
    """IoT edge computing manager"""

    def __init__(self, config: EdgeConfig):
        self.config = config
        self.iot = boto3.client('iot', region_name=config.region)
        self.greengrass = boto3.client('greengrassv2', region_name=config.region)

    def deploy_to_edge(self, component_name: str, 
                      component_version: str) -> str:
        """Deploy component to edge device"""
        response = self.greengrass.create_deployment(
            targetArn=self.config.device_id,
            deploymentName=f"deploy-{component_name}",
            components={
                component_name: {
                    "componentVersion": component_version,
                    "configurationUpdate": {
                        "merge": json.dumps({
                            "runtime": {
                                "maxMemory": 256
                            }
                        })
                    }
                }
            }
        )

        return response['deploymentId']

    def get_edge_metrics(self) -> Dict[str, Any]:
        """Get edge device metrics"""
        iot_data = boto3.client('iot-data', region_name=self.config.region)

        response = iot_data.publish(
            topic=f"edge/{self.config.device_id}/metrics",
            qos=1,
            payload=json.dumps({
                'request': 'get_metrics'
            })
        )

        # In production, subscribe to response topic
        return {'status': 'request_sent'}

    def process_edge_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Process data at the edge"""
        # Apply edge ML model
        processed = self._apply_edge_model(data)

        # Filter and aggregate
        aggregated = self._aggregate_data(processed)

        # Send to cloud
        self._send_to_cloud(aggregated)

        return aggregated

    def _apply_edge_model(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Apply ML model at the edge"""
        # Simplified edge ML processing
        features = data.get('features', [])
        prediction = sum(features) / len(features) if features else 0

        return {
            'device_id': self.config.device_id,
            'prediction': prediction,
            'features': features,
            'timestamp': data.get('timestamp')
        }

    def _aggregate_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Aggregate edge data"""
        return {
            'device_id': data['device_id'],
            'aggregated_prediction': data['prediction'],
            'data_points': 1,
            'timestamp': data.get('timestamp')
        }

    def _send_to_cloud(self, data: Dict[str, Any]):
        """Send aggregated data to cloud"""
        iot_data = boto3.client('iot-data', region_name=self.config.region)

        iot_data.publish(
            topic=f"edge/{self.config.device_id}/cloud",
            qos=1,
            payload=json.dumps(data)
        )

# Edge data synchronization
import boto3
import json
from typing import Dict, Any, List
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class SyncConfig:
    device_id: str
    sync_interval: int = 300  # seconds
    conflict_resolution: str = "last-write-wins"
    max_retries: int = 3

class EdgeSyncManager:
    """Edge data synchronization manager"""

    def __init__(self, config: SyncConfig):
        self.config = config
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table('edge-sync')

    def sync_to_cloud(self, local_data: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Sync local data to cloud"""
        synced = 0
        failed = 0

        for item in local_data:
            try:
                self._put_item(item)
                synced += 1
            except Exception as e:
                print(f"Sync failed for item: {e}")
                failed += 1

        return {
            'synced': synced,
            'failed': failed,
            'timestamp': datetime.utcnow().isoformat()
        }

    def sync_from_cloud(self, last_sync: datetime) -> List[Dict[str, Any]]:
        """Sync data from cloud"""
        response = self.table.query(
            IndexName='device-sync-index',
            KeyConditionExpression='device_id = :device_id AND last_updated > :last_sync',
            ExpressionAttributeValues={
                ':device_id': self.config.device_id,
                ':last_sync': last_sync.isoformat()
            }
        )

        return response.get('Items', [])

    def _put_item(self, item: Dict[str, Any]):
        """Put item to DynamoDB"""
        self.table.put_item(
            Item={
                'device_id': self.config.device_id,
                'item_id': item['id'],
                'data': item,
                'last_updated': datetime.utcnow().isoformat(),
                'sync_status': 'synced'
            }
        )

    def resolve_conflicts(self, local_items: List[Dict[str, Any]], 
                         cloud_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Resolve sync conflicts"""
        resolved = []
        local_map = {item['id']: item for item in local_items}
        cloud_map = {item['id']: item for item in cloud_items}

        all_ids = set(list(local_map.keys()) + list(cloud_map.keys()))

        for item_id in all_ids:
            local = local_map.get(item_id)
            cloud = cloud_map.get(item_id)

            if local and cloud:
                # Conflict - use resolution strategy
                if self.config.conflict_resolution == "last-write-wins":
                    local_time = local.get('last_updated', '')
                    cloud_time = cloud.get('last_updated', '')
                    resolved.append(local if local_time > cloud_time else cloud)
            elif local:
                resolved.append(local)
            else:
                resolved.append(cloud)

        return resolved