Module 05: Message Queues & Event Streams

E-commerce Order Flow (Blocking):
User clicks "Buy Now" →
├─ Wait for payment processor (2 seconds)
├─ Wait for inventory check (1 second)
├─ Wait for email send (3 seconds)
├─ Wait for shipping label (2 seconds)
└─ Total: 8 seconds (user stares at loading spinner)

Problems:
├─ Slow user experience (8 second wait)
├─ If email server crashes, entire order fails
├─ Payment processor spike = all orders delayed
└─ Cannot scale (one service slow = everything slow)

E-commerce Order Flow (Non-blocking):
User clicks "Buy Now" →
├─ Process payment (2 seconds)
├─ Add "send email" to queue (0.01 seconds)
├─ Add "check inventory" to queue (0.01 seconds)
├─ Add "create shipping label" to queue (0.01 seconds)
├─ Show "Order Confirmed!" (2 seconds total)
└─ Background workers process queue items

Benefits:
├─ Fast user experience (2 second response)
├─ If email server crashes, order still succeeds (email retried later)
├─ Each service scales independently
└─ Handles traffic spikes gracefully (queue absorbs burst)

1. User requests ride in San Francisco
2. App directly calls: Driver matching service
3. Driver matching service must:
   - Find available drivers (5 seconds)
   - Calculate ETAs (2 seconds)
   - Notify drivers (3 seconds)
4. User waits 10 seconds staring at "Finding driver..."
5. If one step fails → Entire request fails

1. User requests ride in San Francisco (0.1 seconds)
2. App immediately responds: "We're finding your driver!"
3. Request added to "ride_requests" queue
4. Background systems process queue:
   ├─ Driver matching: Processes 1000 requests/second
   ├─ ETA calculation: Processes 500 requests/second
   ├─ Push notifications: Processes 2000 requests/second
5. Driver found in 2 seconds (background)
6. User gets notification: "John is 3 minutes away"

Uber Scale:
├─ 23 million rides/day
├─ 266 rides/second average
├─ 1,500+ rides/second during peak
└─ Message queue handles all smoothly

Synchronous (Tightly Coupled):
Payment Service → Email Service
├─ If email crashes, payment fails
├─ Payment must wait for email (3 seconds)
└─ Services tied together

Asynchronous (Decoupled):
Payment Service → Queue → Email Service
├─ Payment completes instantly
├─ Email sent when service available
├─ If email crashes, retried automatically
└─ Services independent

Black Friday Without Queue:
├─ Normal: 100 orders/second
├─ Black Friday: 10,000 orders/second (100× spike)
├─ Email server capacity: 100 emails/second
├─ Result: 9,900 orders fail (email server crashes)

Black Friday With Queue:
├─ Normal: 100 orders/second → Queue → 100 emails/second
├─ Black Friday: 10,000 orders/second → Queue → 100 emails/second
├─ Queue holds: 10,000 messages, processed at steady 100/second
├─ Result: All orders succeed, emails sent within 2 minutes
└─ Queue acts as buffer/shock absorber

Without Queue:
├─ API call to payment processor fails
├─ Entire order fails
├─ User must retry manually
└─ Lost revenue from abandoned carts

With Queue:
├─ Message sent to queue
├─ Payment processor temporarily down
├─ Queue retries: 1 second, 5 seconds, 30 seconds, 5 minutes
├─ Payment processor back online
├─ Message processed successfully
└─ Order completes without user intervention

Web Server → Payment API
├─ Web server WAITS for response
├─ Payment API processes (2 seconds)
├─ Web server receives response
├─ Total time: 2 seconds (blocking)
├─ If payment API crashes: Request fails immediately
└─ Max throughput: Limited by slowest service

Web Server → Queue (0.01 seconds) → Continue serving users
Payment Worker → Reads from queue → Processes payment
├─ Web server doesn't wait
├─ Queue stores message reliably
├─ Worker processes when ready
├─ Total web server time: 0.01 seconds (non-blocking)
├─ If worker crashes: Message stays in queue, retried
└─ Max throughput: Unlimited (queue scales horizontally)

Order placed →
├─ Process payment: 2s (wait)
├─ Send confirmation email: 3s (wait)
├─ Update inventory: 1s (wait)
├─ Create shipping label: 2s (wait)
├─ Notify warehouse: 1s (wait)
Total: 9 seconds per order
Capacity: 400 orders/hour per server (slow!)

Order placed →
├─ Process payment: 2s (wait) ← Only critical step
├─ Add 4 messages to queue: 0.05s
├─ Show "Order Confirmed!": 2.05s total
Background workers:
├─ Email service: Processes 1000/second
├─ Inventory service: Processes 500/second
├─ Shipping service: Processes 800/second
├─ Warehouse service: Processes 600/second
Total: 2 seconds user wait
Capacity: 50,000+ orders/hour per server (125× faster!)
Result: 575 million orders/year handled smoothly

┌─────────────────────────────────────────────────────────────────┐
│                         Customer                                 │
│                    (200M Prime members)                          │
└───────────────────────────┬─────────────────────────────────────┘
                            │
                            │ 1. Place Order (HTTPS POST)
                            ▼
                  ┌──────────────────────┐
                  │   API Gateway         │
                  │ (AWS Managed)         │
                  │ - Rate limiting       │◄────┐
                  │ - Authentication      │     │
                  │ - Request validation  │     │
                  └──────────┬────────────┘     │
                            │                   │
              2. Synchronous critical path      │
                            │                   │
                            ▼                   │
                  ┌──────────────────────┐     │
                  │  Order Service        │     │
                  │  (Lambda/ECS)         │     │
                  │                       │     │
                  │  1. Validate order    │     │
                  │  2. Reserve inventory │     │
                  │  3. Create order ID   │     │
                  │  4. Return success    │     │ 4. Return order ID
                  └──────────┬────────────┘     │    (< 500ms response)
                            │                   │
              3. Publish to SQS queues          │
              (asynchronous, fire-and-forget)   │
                            │                   │
        ┌───────────────────┼───────────────────┼────────────┐
        │                   │                   │            │
        ▼                   ▼                   ▼            ▼
┌───────────────┐  ┌────────────────┐  ┌───────────────┐  ┌─────────────┐
│ Payment Queue │  │ Shipping Queue │  │  Email Queue  │  │Fraud Queue  │
│   (Standard)  │  │   (Standard)   │  │  (Standard)   │  │  (FIFO)     │
│               │  │                │  │               │  │             │
│ Visibility:   │  │ Visibility:    │  │ Visibility:   │  │ Exactly-    │
│ 30 seconds    │  │ 60 seconds     │  │ 120 seconds   │  │ once        │
│               │  │                │  │               │  │ delivery    │
│ Max retries:  │  │ Max retries:   │  │ Max retries:  │  │             │
│ 3 attempts    │  │ 5 attempts     │  │ 10 attempts   │  │ Critical    │
└───────┬───────┘  └────────┬───────┘  └───────┬───────┘  └──────┬──────┘
        │                   │                   │                 │
        │ Poll              │ Poll              │ Poll            │ Poll
        │ (Long polling)    │ (Long polling)    │ (Long polling)  │
        ▼                   ▼                   ▼                 ▼
┌───────────────┐  ┌────────────────┐  ┌───────────────┐  ┌─────────────┐
│Payment Worker │  │Shipping Worker │  │ Email Worker  │  │Fraud Worker │
│  (ECS Tasks)  │  │  (ECS Tasks)   │  │ (Lambda)      │  │(ECS Tasks)  │
│               │  │                │  │               │  │             │
│ Concurrency:  │  │ Concurrency:   │  │ Concurrency:  │  │Concurrency: │
│ 500 tasks     │  │ 200 tasks      │  │ 1000 Lambda   │  │ 50 tasks    │
└───────┬───────┘  └────────┬───────┘  └───────┬───────┘  └──────┬──────┘
        │                   │                   │                 │
        │ On failure        │ On failure        │ On failure      │ On max
        │ (retry 3×)        │ (retry 5×)        │ (retry 10×)     │ retries
        ▼                   ▼                   ▼                 ▼
┌───────────────┐  ┌────────────────┐  ┌───────────────┐  ┌─────────────┐
│  Payment DLQ  │  │  Shipping DLQ  │  │   Email DLQ   │  │  Fraud DLQ  │
│ (Dead Letter  │  │ (Dead Letter   │  │ (Dead Letter  │  │(Dead Letter │
│    Queue)     │  │    Queue)      │  │    Queue)     │  │   Queue)    │
│               │  │                │  │               │  │             │
│ Manual review │  │ Manual review  │  │ Manual review │  │ Manual      │
│ required      │  │ required       │  │ required      │  │ review      │
└───────────────┘  └────────────────┘  └───────────────┘  └─────────────┘
        │                   │                   │                 │
        │                   │                   │                 │
        └───────────────────┴───────────────────┴─────────────────┘
                            │
                            ▼
                  ┌──────────────────────┐
                  │  CloudWatch Alarms   │
                  │                      │
                  │  - DLQ depth > 100   │
                  │  - Processing lag    │
                  │  - Worker failures   │
                  └──────────────────────┘

POST /orders HTTP/1.1
Host: api.amazon.com
Content-Type: application/json
Authorization: Bearer <customer_token>

{
  "customerId": "CUST-123456789",
  "items": [
    {
      "asin": "B08N5WRWNW",
      "title": "Apple AirPods Pro (2nd Gen)",
      "quantity": 1,
      "price": 249.00
    }
  ],
  "shippingAddress": {
    "street": "410 Terry Ave N",
    "city": "Seattle",
    "state": "WA",
    "zip": "98109"
  },
  "paymentMethod": "CARD-****1234",
  "shippingSpeed": "PRIME_TWO_DAY"
}

import boto3
import uuid
import time
from decimal import Decimal

def lambda_handler(event, context):
    """
    Order Service - Synchronous critical path only
    Target latency: < 500ms P95
    """
    start_time = time.time()
    
    # Parse request
    body = json.loads(event['body'])
    customer_id = body['customerId']
    items = body['items']
    shipping_address = body['shippingAddress']
    
    # 1. Validate inventory (synchronous, must succeed)
    # Query DynamoDB inventory table
    dynamodb = boto3.resource('dynamodb')
    inventory_table = dynamodb.Table('Inventory')
    
    for item in items:
        asin = item['asin']
        quantity = item['quantity']
        
        # Get inventory for nearest fulfillment centers
        response = inventory_table.get_item(
            Key={
                'ASIN': asin,
                'FulfillmentCenter': get_nearest_fc(shipping_address)
            }
        )
        
        available = response['Item']['QuantityAvailable']
        if available < quantity:
            # Try next nearest fulfillment center
            # (fallback logic omitted for brevity)
            return {
                'statusCode': 400,
                'body': json.dumps({
                    'error': 'OUT_OF_STOCK',
                    'message': f'Item {asin} unavailable'
                })
            }
        
        # Reserve inventory (conditional update, prevents overselling)
        inventory_table.update_item(
            Key={
                'ASIN': asin,
                'FulfillmentCenter': get_nearest_fc(shipping_address)
            },
            UpdateExpression='SET QuantityAvailable = QuantityAvailable - :qty, '
                           'QuantityReserved = QuantityReserved + :qty',
            ConditionExpression='QuantityAvailable >= :qty',
            ExpressionAttributeValues={
                ':qty': quantity
            }
        )
    
    # 2. Create order record (synchronous)
    order_id = f"ORD-{uuid.uuid4()}"
    order_table = dynamodb.Table('Orders')
    
    order_table.put_item(
        Item={
            'OrderId': order_id,
            'CustomerId': customer_id,
            'Items': items,
            'ShippingAddress': shipping_address,
            'Status': 'PENDING_PAYMENT',
            'CreatedAt': int(time.time()),
            'TotalAmount': Decimal(str(sum(item['price'] * item['quantity'] for item in items)))
        }
    )
    
    # 3. Publish to SQS queues (asynchronous, fire-and-forget)
    sqs = boto3.client('sqs')
    
    # Payment queue (critical, Standard queue for high throughput)
    sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue',
        MessageBody=json.dumps({
            'orderId': order_id,
            'customerId': customer_id,
            'paymentMethod': body['paymentMethod'],
            'amount': sum(item['price'] * item['quantity'] for item in items),
            'currency': 'USD'
        }),
        MessageAttributes={
            'OrderType': {'StringValue': 'PRIME', 'DataType': 'String'},
            'Priority': {'StringValue': 'HIGH', 'DataType': 'String'}
        }
    )
    
    # Shipping queue (critical, Standard queue)
    sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/shipping-queue',
        MessageBody=json.dumps({
            'orderId': order_id,
            'items': items,
            'shippingAddress': shipping_address,
            'shippingSpeed': body['shippingSpeed'],
            'fulfillmentCenter': get_nearest_fc(shipping_address)
        })
    )
    
    # Email queue (non-critical, can fail without blocking order)
    sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/email-queue',
        MessageBody=json.dumps({
            'orderId': order_id,
            'customerId': customer_id,
            'type': 'ORDER_CONFIRMATION',
            'templateId': 'order-confirm-v3',
            'items': items
        })
    )
    
    # Fraud detection queue (critical, FIFO for exactly-once)
    sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/fraud-queue.fifo',
        MessageBody=json.dumps({
            'orderId': order_id,
            'customerId': customer_id,
            'amount': sum(item['price'] * item['quantity'] for item in items),
            'ipAddress': event['requestContext']['identity']['sourceIp'],
            'deviceFingerprint': event['headers'].get('X-Device-Fingerprint')
        }),
        MessageGroupId=customer_id,  # FIFO: process customer's orders in order
        MessageDeduplicationId=order_id  # FIFO: prevent duplicate processing
    )
    
    # 4. Return success to customer (< 500ms total)
    elapsed = (time.time() - start_time) * 1000
    
    return {
        'statusCode': 201,
        'headers': {
            'X-Request-Id': context.request_id,
            'X-Processing-Time-Ms': str(int(elapsed))
        },
        'body': json.dumps({
            'orderId': order_id,
            'status': 'CONFIRMED',
            'estimatedDelivery': calculate_delivery_date(body['shippingSpeed']),
            'message': 'Order confirmed! Processing payment and shipping details.'
        })
    }

def get_nearest_fc(address):
    """
    Find nearest fulfillment center based on zip code
    (Simplified - actual algorithm uses geospatial queries)
    """
    zip_to_fc = {
        '98109': 'SEA-FC-01',  # Seattle
        '10001': 'JFK-FC-03',  # New York
        '90210': 'LAX-FC-02',  # Los Angeles
        # ... 360 fulfillment centers mapped
    }
    return zip_to_fc.get(address['zip'], 'DEFAULT-FC')

def calculate_delivery_date(shipping_speed):
    """Calculate estimated delivery based on shipping speed"""
    from datetime import datetime, timedelta
    
    if shipping_speed == 'PRIME_TWO_DAY':
        return (datetime.now() + timedelta(days=2)).isoformat()
    elif shipping_speed == 'PRIME_ONE_DAY':
        return (datetime.now() + timedelta(days=1)).isoformat()
    else:
        return (datetime.now() + timedelta(days=5)).isoformat()

import boto3
import json
import time
import requests

def payment_worker():
    """
    Payment Worker - Processes payment queue messages
    Runs as ECS task with 500 concurrent instances
    """
    sqs = boto3.client('sqs')
    dynamodb = boto3.resource('dynamodb')
    orders_table = dynamodb.Table('Orders')
    
    queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue'
    
    while True:
        # Long polling (WaitTimeSeconds=20 reduces empty responses)
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,  # Batch processing for efficiency
            WaitTimeSeconds=20,  # Long polling (reduces API calls by 95%)
            VisibilityTimeout=30,  # 30 seconds to process before retry
            MessageAttributeNames=['All']
        )
        
        if 'Messages' not in response:
            # No messages available, long polling will wait 20 seconds
            continue
        
        for message in response['Messages']:
            receipt_handle = message['ReceiptHandle']
            body = json.loads(message['Body'])
            
            try:
                # Process payment
                order_id = body['orderId']
                customer_id = body['customerId']
                amount = body['amount']
                payment_method = body['paymentMethod']
                
                # Call payment gateway (Stripe, Adyen, etc.)
                payment_response = requests.post(
                    'https://payment-gateway.amazon.com/v1/charge',
                    json={
                        'customerId': customer_id,
                        'paymentMethod': payment_method,
                        'amount': amount,
                        'currency': 'USD',
                        'orderId': order_id,
                        'idempotencyKey': order_id  # Prevent duplicate charges
                    },
                    timeout=10
                )
                
                if payment_response.status_code == 200:
                    payment_data = payment_response.json()
                    
                    # Update order status
                    orders_table.update_item(
                        Key={'OrderId': order_id},
                        UpdateExpression='SET #status = :status, PaymentId = :payment_id, '
                                       'PaymentAt = :timestamp',
                        ExpressionAttributeNames={
                            '#status': 'Status'
                        },
                        ExpressionAttributeValues={
                            ':status': 'PAYMENT_COMPLETE',
                            ':payment_id': payment_data['transactionId'],
                            ':timestamp': int(time.time())
                        }
                    )
                    
                    # Delete message from queue (successful processing)
                    sqs.delete_message(
                        QueueUrl=queue_url,
                        ReceiptHandle=receipt_handle
                    )
                    
                    print(f" Payment processed: {order_id} - ${amount}")
                    
                elif payment_response.status_code == 402:
                    # Payment declined - permanent failure, move to DLQ
                    orders_table.update_item(
                        Key={'OrderId': order_id},
                        UpdateExpression='SET #status = :status, DeclineReason = :reason',
                        ExpressionAttributeNames={'#status': 'Status'},
                        ExpressionAttributeValues={
                            ':status': 'PAYMENT_DECLINED',
                            ':reason': payment_response.json().get('reason', 'Unknown')
                        }
                    )
                    
                    # Delete message (don't retry declined payments)
                    sqs.delete_message(
                        QueueUrl=queue_url,
                        ReceiptHandle=receipt_handle
                    )
                    
                    # Trigger customer notification (separate queue)
                    sqs.send_message(
                        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/email-queue',
                        MessageBody=json.dumps({
                            'orderId': order_id,
                            'customerId': customer_id,
                            'type': 'PAYMENT_DECLINED',
                            'reason': payment_response.json().get('reason')
                        })
                    )
                    
                    print(f" Payment declined: {order_id} - {payment_response.json()}")
                    
                else:
                    # Temporary failure (gateway timeout, etc.) - retry
                    # Message visibility timeout expires, returns to queue
                    # SQS will retry up to 3 times (configured on queue)
                    print(f" Payment failed (will retry): {order_id} - Status {payment_response.status_code}")
                    # Do NOT delete message - let visibility timeout expire for retry
                    
            except Exception as e:
                # Unexpected error - log and let message retry
                print(f" Error processing message: {e}")
                # Message remains in queue, retries up to maxReceiveCount
                # After 3 retries, automatically moves to Dead Letter Queue

if __name__ == '__main__':
    payment_worker()

# SQS Queue Configuration (Terraform)
resource "aws_sqs_queue" "payment_queue" {
  name                       = "payment-queue"
  delay_seconds              = 0
  max_message_size           = 262144  # 256 KB
  message_retention_seconds  = 1209600  # 14 days
  receive_wait_time_seconds  = 20  # Long polling
  visibility_timeout_seconds = 30
  
  # Dead Letter Queue configuration
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.payment_dlq.arn
    maxReceiveCount     = 3  # After 3 failed attempts, move to DLQ
  })
  
  # CloudWatch alarms
  tags = {
    Environment = "production"
    CriticalPath = "true"
  }
}

resource "aws_sqs_queue" "payment_dlq" {
  name                      = "payment-queue-dlq"
  message_retention_seconds = 1209600  # 14 days retention
  
  tags = {
    Environment = "production"
    AlertOnMessages = "true"  # Trigger PagerDuty if messages arrive
  }
}

# CloudWatch alarm: trigger if DLQ has > 100 messages
resource "aws_cloudwatch_metric_alarm" "payment_dlq_alarm" {
  alarm_name          = "payment-dlq-high-depth"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "1"
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = "60"  # 1 minute
  statistic           = "Average"
  threshold           = "100"
  alarm_description   = "Payment DLQ has > 100 messages - manual review required"
  alarm_actions       = [aws_sns_topic.pagerduty_critical.arn]
  
  dimensions = {
    QueueName = aws_sqs_queue.payment_dlq.name
  }
}

Monthly volume:
- Orders: 575M ÷ 12 = 47.9M orders/month
- SQS sends: 47.9M × 4 queues = 191.6M sends/month
- SQS receives: 191.6M × 1.2 (includes retries) = 230M receives/month
- SQS deletes: 191.6M × 0.997 (99.7% successful) = 191M deletes/month
- Total requests: 191.6M + 230M + 191M = 612.6M requests/month

Standard queues (payment, shipping, email):
- Requests: 47.9M × 3 queues × 3 operations = 431M requests
- Cost: (431M - 1M free) × $0.40 / 1M = $172/month

FIFO queue (fraud detection):
- Requests: 47.9M × 3 operations = 143.7M requests
- Cost: (143.7M - 1M free) × $0.50 / 1M = $71.35/month

Total SQS cost: $172 + $71.35 = $243.35/month
Annual SQS cost: $243.35 × 12 = $2,920/year

Infrastructure (575M orders/year, 18.2 avg req/sec, 637 peak):
- EC2 instances: 20× r6i.2xlarge (8 vCPU, 64 GB RAM, RabbitMQ cluster)
  - Cost: $0.504/hour × 20 × 730 hours = $7,358/month
  - Rationale: High memory for queue buffering, cluster for HA
- EBS storage: 10 TB gp3 (14 days message retention)
  - Cost: $0.08/GB-month × 10,000 GB = $800/month
- Elastic Load Balancer: 1× Application Load Balancer
  - Cost: $22.50/month + $0.008/GB processed
  - Data: 6M messages × 10 KB avg = 60 GB/day = 1,800 GB/month
  - Cost: $22.50 + ($0.008 × 1,800) = $36.90/month
- VPC data transfer: $0.01/GB (inter-AZ for HA)
  - Cost: 1,800 GB × $0.01 = $18/month

Staff (3 SREs for 24/7 on-call, management, upgrades):
- Salary: $180K/year × 3 = $540K/year = $45,000/month
- Benefits (30%): $45,000 × 0.30 = $13,500/month
- Total staff: $58,500/month

Total self-hosted: $7,358 + $800 + $36.90 + $18 + $58,500 = $66,712.90/month
Annual self-hosted: $66,712.90 × 12 = $800,555/year

SQS savings: $800,555 - $2,920 = $797,635/year (99.6% cost reduction)

Timeline: Payment message processing

T+0s:   Message arrives in queue
        [Queue: 1 message visible]

T+1s:   Worker A receives message
        ReceiveMessage API call returns message
        Message becomes invisible (VisibilityTimeout=30s)
        [Queue: 0 messages visible, 1 in-flight]

T+5s:   Worker A processing payment...
        Calls payment gateway API
        [Queue: 0 messages visible, 1 in-flight]

T+10s:  Worker A crashes (out of memory, instance terminated, etc.)
        Message still invisible for 20 more seconds
        [Queue: 0 messages visible, 1 in-flight]

T+31s:  Visibility timeout expires
        Message becomes visible again
        [Queue: 1 message visible]

T+32s:  Worker B receives message (automatic retry)
        Processes payment successfully
        Calls DeleteMessage API
        [Queue: 0 messages visible]

# Too short (10 seconds): Worker needs 15s to process → message visible before completion
# → Different worker receives same message → duplicate processing
visibility_timeout_too_short = 10  #  BAD

# Optimal (30 seconds): Typical processing 5-10s, allows 20-25s for retries
visibility_timeout_optimal = 30  #  GOOD

# Too long (300 seconds): Worker crashes at 5s, message invisible for 295s
# → Retry delayed 5 minutes → customer experience degraded
visibility_timeout_too_long = 300  #  BAD (unless processing truly takes 4+ minutes)

VisibilityTimeout = (Average Processing Time × 6) + Network Latency

Example for payment processing:
- Average processing: 3 seconds
- P95 processing: 8 seconds
- Network latency: 2 seconds

VisibilityTimeout = (3s × 6) + 2s = 20 seconds

Rationale:
- 6× average covers P95 (99.7% of processing times)
- Additional 2s for network jitter
- If processing exceeds 20s, worker should call ChangeMessageVisibility API to extend

def process_large_order(message, receipt_handle):
    """
    Process order with many items (takes 45 seconds)
    Extends visibility timeout to prevent retry
    """
    sqs = boto3.client('sqs')
    queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue'
    
    # Initial visibility timeout: 30 seconds
    # Start processing
    
    time.sleep(20)  # 20 seconds elapsed
    
    # Extend visibility by another 30 seconds (now 50s total)
    sqs.change_message_visibility(
        QueueUrl=queue_url,
        ReceiptHandle=receipt_handle,
        VisibilityTimeout=30  # Extend by 30 more seconds
    )
    
    time.sleep(25)  # 45 seconds elapsed total
    
    # Processing complete, delete message
    sqs.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=receipt_handle
    )

# Create main queue with DLQ redrive policy
response = sqs.create_queue(
    QueueName='payment-queue',
    Attributes={
        'VisibilityTimeout': '30',
        'MessageRetentionPeriod': '1209600',  # 14 days
        'ReceiveMessageWaitTimeSeconds': '20',  # Long polling
        'RedrivePolicy': json.dumps({
            'deadLetterTargetArn': 'arn:aws:sqs:us-east-1:123456789012:payment-queue-dlq',
            'maxReceiveCount': '3'  # After 3 receives, move to DLQ
        })
    }
)

# Create Dead Letter Queue (no DLQ itself - manual review)
response = sqs.create_queue(
    QueueName='payment-queue-dlq',
    Attributes={
        'MessageRetentionPeriod': '1209600'  # 14 days retention
    }
)

import boto3
import json

def check_dlq_depth():
    """
    Monitor DLQ depth, alert if > 100 messages
    Runs every 1 minute via CloudWatch Events
    """
    sqs = boto3.client('sqs')
    sns = boto3.client('sns')
    
    queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue-dlq'
    
    # Get queue attributes
    response = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=['ApproximateNumberOfMessages']
    )
    
    depth = int(response['Attributes']['ApproximateNumberOfMessages'])
    
    if depth > 100:
        # Critical alert - payment processing failing
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:pagerduty-critical',
            Subject='CRITICAL: Payment DLQ depth > 100',
            Message=json.dumps({
                'severity': 'CRITICAL',
                'service': 'payment-processing',
                'metric': 'dlq_depth',
                'value': depth,
                'threshold': 100,
                'action': 'Manual review required - check payment-queue-dlq',
                'runbook': 'https://wiki.amazon.com/PaymentDLQRunbook'
            })
        )
    elif depth > 10:
        # Warning - investigate before critical
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:ops-warnings',
            Subject='WARNING: Payment DLQ depth > 10',
            Message=f'Payment DLQ has {depth} messages (threshold: 10). Investigate.'
        )

def analyze_dlq_messages():
    """
    Analyze DLQ messages to identify root cause
    Common patterns:
    - Invalid data format (fix producer)
    - External service down (retry after service restored)
    - Logic bug (fix code, redrive messages)
    """
    sqs = boto3.client('sqs')
    queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue-dlq'
    
    # Receive up to 10 messages
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        AttributeNames=['All'],
        MessageAttributeNames=['All']
    )
    
    if 'Messages' not in response:
        print("DLQ is empty ")
        return
    
    # Analyze failure patterns
    failure_types = {}
    
    for message in response['Messages']:
        body = json.loads(message['Body'])
        
        # Check for common failure patterns
        if 'paymentMethod' not in body:
            failure_type = 'MISSING_PAYMENT_METHOD'
        elif body.get('amount', 0) <= 0:
            failure_type = 'INVALID_AMOUNT'
        elif not body.get('orderId'):
            failure_type = 'MISSING_ORDER_ID'
        else:
            failure_type = 'UNKNOWN'
        
        failure_types[failure_type] = failure_types.get(failure_type, 0) + 1
    
    # Report findings
    print(f"DLQ Analysis ({len(response['Messages'])} messages sampled):")
    for failure_type, count in failure_types.items():
        print(f"  {failure_type}: {count} messages")
    
    # Example: If MISSING_PAYMENT_METHOD, fix producer code
    if 'MISSING_PAYMENT_METHOD' in failure_types:
        print("\n Action: Fix Order Service to include paymentMethod in all messages")
        print("   Affected orders:", [json.loads(msg['Body'])['orderId'] 
                                      for msg in response['Messages']
                                      if 'paymentMethod' not in json.loads(msg['Body'])])

def redrive_dlq_messages():
    """
    Move messages from DLQ back to main queue after fixing root cause
    Use RedriveAllowPolicy to prevent infinite loop (DLQ → Queue → DLQ again)
    """
    sqs = boto3.client('sqs')
    
    dlq_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue-dlq'
    main_queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/payment-queue'
    
    # Receive all messages from DLQ
    while True:
        response = sqs.receive_message(
            QueueUrl=dlq_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=1
        )
        
        if 'Messages' not in response:
            break  # DLQ empty
        
        for message in response['Messages']:
            # Send to main queue (fix applied, should succeed now)
            sqs.send_message(
                QueueUrl=main_queue_url,
                MessageBody=message['Body'],
                MessageAttributes=message.get('MessageAttributes', {})
            )
            
            # Delete from DLQ
            sqs.delete_message(
                QueueUrl=dlq_url,
                ReceiptHandle=message['ReceiptHandle']
            )
            
            print(f" Redrove message: {json.loads(message['Body'])['orderId']}")
    
    print(f" DLQ redrive complete")

# Short polling (BAD - expensive and inefficient)
while True:
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=0  # Return immediately
    )
    
    if 'Messages' in response:
        # Process messages
        pass
    else:
        # Queue empty - 95% of requests at average load
        pass
    
    time.sleep(1)  # Poll every second

# Cost analysis:
# - Requests/hour: 3,600 (one per second)
# - Requests/day: 86,400
# - Requests/month: 2,592,000
# - Empty responses: 2,462,400 (95%)
# - Cost: (2,592,000 / 1,000,000) × $0.40 = $1.04/month per worker
# - At 500 workers: $520/month
# - Annual: $6,240/year

# Long polling ( GOOD - efficient)
while True:
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20  # Wait up to 20 seconds for message
    )
    
    if 'Messages' in response:
        # Process messages
        pass
    # If no messages after 20 seconds, loop continues
    # Next receive_message waits another 20 seconds

# Cost analysis:
# - Requests/hour: 180 (one every 20 seconds if queue empty)
# - Requests/day: 4,320
# - Requests/month: 129,600 (95% reduction vs short polling)
# - Cost: (129,600 / 1,000,000) × $0.40 = $0.05/month per worker
# - At 500 workers: $25/month
# - Annual: $300/year
# 
# Savings: $6,240 - $300 = $5,940/year (95% cost reduction)

# Enable long polling at queue level (applies to all consumers)
resource "aws_sqs_queue" "payment_queue" {
  name                       = "payment-queue"
  receive_wait_time_seconds  = 20  # Long polling enabled
  
  # Workers don't need to specify WaitTimeSeconds in ReceiveMessage
  # Queue-level setting applies automatically
}

┌─────────────────────────────────────────────────────────────────┐
│                         Rider                                    │
│              (137M monthly active riders)                        │
└───────────────────────┬─────────────────────────────────────────┘
                        │
                        │ 1. Request Trip (HTTPS POST)
                        │    { from: "123 Main St", to: "456 Oak Ave" }
                        ▼
              ┌──────────────────────┐
              │   API Gateway         │
              │ - Authentication      │
              │ - Rate limiting       │◄─────┐
              │ - Request validation  │      │
              └──────────┬────────────┘      │
                        │                    │
          2. Synchronous critical path       │
                        │                    │
                        ▼                    │
              ┌──────────────────────┐      │
              │  Dispatch Service     │      │
              │  (gRPC microservice)  │      │
              │                       │      │
              │  1. Validate request  │      │
              │  2. Find drivers      │      │ 3. Return trip ID
              │     (geospatial       │      │    + estimated wait
              │      query, Redis)    │      │    (< 500ms)
              │  3. Create trip ID    │      │
              │  4. Publish event     │      │
              └──────────┬────────────┘      │
                        │                    │
         4. Publish "TripRequested" event   │
            (single SNS/EventBridge publish) │
                        │                    │
                        ▼                    │
              ┌──────────────────────┐      │
              │  SNS Topic            │      │
              │  "trip-events"        │      │
              │                       │      │
              │  Fanout to multiple   │      │
              │  subscribers          │      │
              └──────────┬────────────┘      │
                        │                    │
            ┌───────────┼────────────┬───────┴─────┬──────────┐
            │           │            │             │          │
            ▼           ▼            ▼             ▼          ▼
    ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
    │ Driver   │ │ Pricing  │ │ Fraud    │ │Analytics │ │Compliance│
    │ Notif.   │ │ Service  │ │Detection │ │ Service  │ │ Service  │
    │ (SNS→SQS)│ │ (Lambda) │ │ (Lambda) │ │(Kinesis) │ │ (Lambda) │
    └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
         │            │            │             │          │
         │ Notify     │ Calculate  │ Check for   │ Log to   │ Check
         │ 50 nearby  │ surge      │ fraud       │ Hadoop   │ driver
         │ drivers    │ pricing    │ signals     │ cluster  │ hours
         │ via FCM    │            │             │          │ limits
         │            │            │             │          │
         ▼            ▼            ▼             ▼          ▼
    [Push to     [Store in    [Flag if     [Store for   [Block if
     driver      DynamoDB     suspicious   analytics]   exceeds
     phones]     for app]     + alert]                  8hr limit]

POST /v1/trips HTTP/1.1
Host: api.uber.com
Authorization: Bearer <rider_token>
Content-Type: application/json

{
  "riderId": "RIDER-8f7a2b3c",
  "pickup": {
    "latitude": 37.7749,
    "longitude": -122.4194,
    "address": "123 Market St, San Francisco, CA"
  },
  "dropoff": {
    "latitude": 37.8044,
    "longitude": -122.2712,
    "address": "456 Broadway, Oakland, CA"
  },
  "serviceType": "UBER_X",
  "paymentMethodId": "CARD-****1234"
}

import boto3
import redis
import json
import time
from geopy.distance import geodesic

def handle_trip_request(event):
    """
    Dispatch Service - Find drivers and publish event
    Target latency: < 500ms P95
    """
    start_time = time.time()
    
    # Parse request
    body = json.loads(event['body'])
    rider_id = body['riderId']
    pickup_lat = body['pickup']['latitude']
    pickup_lon = body['pickup']['longitude']
    
    # 1. Find nearby drivers (Redis geospatial query)
    r = redis.Redis(host='dispatch-cache.redis.use1.cache.amazonaws.com', port=6379)
    
    # GEORADIUS: Find drivers within 5-mile radius (8 km)
    nearby_drivers = r.georadius(
        name='drivers:available',
        longitude=pickup_lon,
        latitude=pickup_lat,
        radius=8,
        unit='km',
        withdist=True,  # Include distance in response
        sort='ASC'  # Nearest first
    )
    
    if not nearby_drivers:
        return {
            'statusCode': 404,
            'body': json.dumps({
                'error': 'NO_DRIVERS_AVAILABLE',
                'message': 'No drivers available in your area. Please try again.'
            })
        }
    
    # 2. Create trip record (DynamoDB)
    trip_id = f"TRIP-{uuid.uuid4()}"
    dynamodb = boto3.resource('dynamodb')
    trips_table = dynamodb.Table('Trips')
    
    trips_table.put_item(
        Item={
            'TripId': trip_id,
            'RiderId': rider_id,
            'Pickup': body['pickup'],
            'Dropoff': body['dropoff'],
            'ServiceType': body['serviceType'],
            'Status': 'REQUESTED',
            'CreatedAt': int(time.time()),
            'NearbyDrivers': [driver[0].decode() for driver in nearby_drivers[:50]]  # Top 50
        }
    )
    
    # 3. Publish "TripRequested" event to SNS (asynchronous fanout)
    sns = boto3.client('sns')
    
    event_payload = {
        'eventType': 'TRIP_REQUESTED',
        'tripId': trip_id,
        'riderId': rider_id,
        'pickup': body['pickup'],
        'dropoff': body['dropoff'],
        'serviceType': body['serviceType'],
        'nearbyDrivers': [driver[0].decode() for driver in nearby_drivers[:50]],
        'estimatedDistance': calculate_distance(body['pickup'], body['dropoff']),
        'timestamp': int(time.time())
    }
    
    # Publish to SNS topic (fanout to all subscribers)
    response = sns.publish(
        TopicArn='arn:aws:sns:us-east-1:123456789012:trip-events',
        Message=json.dumps(event_payload),
        MessageAttributes={
            'eventType': {'DataType': 'String', 'StringValue': 'TRIP_REQUESTED'},
            'priority': {'DataType': 'String', 'StringValue': 'HIGH'},
            'riderId': {'DataType': 'String', 'StringValue': rider_id}
        }
    )
    
    # 4. Return immediate response to rider (< 500ms)
    elapsed = (time.time() - start_time) * 1000
    
    return {
        'statusCode': 201,
        'headers': {
            'X-Request-Id': event['requestContext']['requestId'],
            'X-Processing-Time-Ms': str(int(elapsed))
        },
        'body': json.dumps({
            'tripId': trip_id,
            'status': 'SEARCHING',
            'estimatedWait': estimate_wait_time(nearby_drivers),
            'message': 'Finding a driver...'
        })
    }

def calculate_distance(pickup, dropoff):
    """Calculate distance between two coordinates"""
    return geodesic(
        (pickup['latitude'], pickup['longitude']),
        (dropoff['latitude'], dropoff['longitude'])
    ).miles

def estimate_wait_time(nearby_drivers):
    """
    Estimate wait time based on nearest driver distance
    Average Uber driver speed in city: 25 mph
    """
    if not nearby_drivers:
        return None
    
    nearest_driver_distance = nearby_drivers[0][1]  # Distance in km
    nearest_driver_distance_miles = nearest_driver_distance * 0.621371
    
    # Time = Distance / Speed (25 mph city average)
    wait_minutes = (nearest_driver_distance_miles / 25) * 60
    
    return {
        'minutes': int(wait_minutes),
        'text': f"{int(wait_minutes)} min"
    }

# SNS Topic for trip events
resource "aws_sns_topic" "trip_events" {
  name              = "trip-events"
  display_name      = "Uber Trip Events"
  fifo_topic        = false  # Standard topic (unlimited throughput)
  
  # Message retention: 14 days (if all subscribers fail)
  message_retention_seconds = 1209600
  
  tags = {
    Environment = "production"
    Service     = "dispatch"
  }
}

# Subscription 1: Driver Notification (SNS → SQS → Lambda)
resource "aws_sns_topic_subscription" "driver_notification" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.driver_notification_queue.arn
  
  # Filter: Only "TRIP_REQUESTED" events
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED"]
  })
}

# Subscription 2: Pricing Service (SNS → Lambda direct)
resource "aws_sns_topic_subscription" "pricing_service" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "lambda"
  endpoint  = aws_lambda_function.pricing_calculator.arn
  
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED", "DRIVER_ACCEPTED"]
  })
}

# Subscription 3: Fraud Detection (SNS → Lambda)
resource "aws_sns_topic_subscription" "fraud_detection" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "lambda"
  endpoint  = aws_lambda_function.fraud_detector.arn
  
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED"],
    priority = ["HIGH"]  # Only high-priority trips (high value, new rider)
  })
}

# Subscription 4: Analytics (SNS → Kinesis → S3/Hadoop)
resource "aws_sns_topic_subscription" "analytics" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "kinesis"
  endpoint  = aws_kinesis_stream.trip_analytics.arn
  
  # No filter - receive all events
}

# Subscription 5: Compliance Service (SNS → SQS → Lambda)
resource "aws_sns_topic_subscription" "compliance" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.compliance_queue.arn
  
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED", "TRIP_COMPLETED"]
  })
}

import boto3
import json
from firebase_admin import messaging

def lambda_handler(event, context):
    """
    Driver Notification Service
    Receives SNS event, notifies nearby drivers
    """
    # Parse SNS message
    for record in event['Records']:
        sns_message = json.loads(record['Sns']['Message'])
        
        trip_id = sns_message['tripId']
        pickup = sns_message['pickup']
        dropoff = sns_message['dropoff']
        nearby_drivers = sns_message['nearbyDrivers']
        
        # Notify each nearby driver (up to 50)
        for driver_id in nearby_drivers:
            # Get driver FCM token from DynamoDB
            dynamodb = boto3.resource('dynamodb')
            drivers_table = dynamodb.Table('Drivers')
            
            driver = drivers_table.get_item(Key={'DriverId': driver_id})
            fcm_token = driver['Item']['FCMToken']
            
            # Send Firebase Cloud Message
            message = messaging.Message(
                token=fcm_token,
                notification=messaging.Notification(
                    title='New trip request',
                    body=f"{pickup['address']} → {dropoff['address']}"
                ),
                data={
                    'tripId': trip_id,
                    'pickupLat': str(pickup['latitude']),
                    'pickupLon': str(pickup['longitude']),
                    'dropoffLat': str(dropoff['latitude']),
                    'dropoffLon': str(dropoff['longitude']),
                    'estimatedEarnings': calculate_earnings(pickup, dropoff)
                },
                android=messaging.AndroidConfig(
                    priority='high',  # High priority for time-sensitive notification
                    ttl=60  # 60-second TTL (trip likely accepted by then)
                )
            )
            
            try:
                response = messaging.send(message)
                print(f" Notified driver {driver_id} for trip {trip_id}")
            except Exception as e:
                print(f" Failed to notify driver {driver_id}: {e}")
                # Don't fail entire batch if one driver notification fails

import boto3
import json

def lambda_handler(event, context):
    """
    Pricing Service
    Calculates dynamic pricing based on supply/demand
    """
    # Parse SNS message
    sns_message = json.loads(event['Records'][0]['Sns']['Message'])
    
    trip_id = sns_message['tripId']
    pickup = sns_message['pickup']
    service_type = sns_message['serviceType']
    
    # Calculate supply/demand ratio
    # Query Redis for:
    # 1. Active trip requests in area (demand)
    # 2. Available drivers in area (supply)
    
    r = redis.Redis(host='pricing-cache.redis.use1.cache.amazonaws.com', port=6379)
    
    # Get geohash for pickup location (precision 6 = ~1.2 km × 0.61 km area)
    geohash = geohash_encode(pickup['latitude'], pickup['longitude'], precision=6)
    
    # Count active requests in geohash
    active_requests = r.scard(f'requests:{geohash}')
    
    # Count available drivers in geohash (GEORADIUS query)
    available_drivers = r.georadius(
        name='drivers:available',
        longitude=pickup['longitude'],
        latitude=pickup['latitude'],
        radius=2,  # 2 km
        unit='km',
        count=True
    )
    
    # Calculate surge multiplier
    if available_drivers == 0:
        surge_multiplier = 3.0  # Maximum surge
    else:
        demand_supply_ratio = active_requests / available_drivers
        
        if demand_supply_ratio < 0.5:
            surge_multiplier = 1.0  # No surge
        elif demand_supply_ratio < 1.0:
            surge_multiplier = 1.2  # Light surge
        elif demand_supply_ratio < 2.0:
            surge_multiplier = 1.5  # Moderate surge
        elif demand_supply_ratio < 3.0:
            surge_multiplier = 2.0  # High surge
        else:
            surge_multiplier = 3.0  # Maximum surge (cap at 3×)
    
    # Calculate base fare
    base_fare = get_base_fare(service_type)  # $2.50 for UberX
    per_mile = get_per_mile_rate(service_type)  # $1.75/mile
    per_minute = get_per_minute_rate(service_type)  # $0.35/minute
    
    estimated_miles = sns_message['estimatedDistance']
    estimated_minutes = estimate_duration(pickup, sns_message['dropoff'])
    
    # Total fare = (Base + Miles × Rate + Minutes × Rate) × Surge
    total_fare = (
        base_fare +
        (estimated_miles * per_mile) +
        (estimated_minutes * per_minute)
    ) * surge_multiplier
    
    # Store pricing in DynamoDB
    dynamodb = boto3.resource('dynamodb')
    pricing_table = dynamodb.Table('TripPricing')
    
    pricing_table.put_item(
        Item={
            'TripId': trip_id,
            'BaseFare': base_fare,
            'PerMile': per_mile,
            'PerMinute': per_minute,
            'EstimatedMiles': estimated_miles,
            'EstimatedMinutes': estimated_minutes,
            'SurgeMultiplier': surge_multiplier,
            'TotalFare': round(total_fare, 2),
            'CalculatedAt': int(time.time())
        }
    )
    
    print(f" Pricing calculated for trip {trip_id}: ${total_fare:.2f} ({surge_multiplier}× surge)")

import boto3
import json

def lambda_handler(event, context):
    """
    Fraud Detection Service
    ML model analyzes trip for fraud signals
    """
    sns_message = json.loads(event['Records'][0]['Sns']['Message'])
    
    trip_id = sns_message['tripId']
    rider_id = sns_message['riderId']
    
    # Get rider history from DynamoDB
    dynamodb = boto3.resource('dynamodb')
    riders_table = dynamodb.Table('Riders')
    
    rider = riders_table.get_item(Key={'RiderId': rider_id})['Item']
    
    # Fraud signals
    signals = []
    fraud_score = 0
    
    # Signal 1: New rider with high-value trip
    if rider.get('TripCount', 0) == 0 and sns_message['estimatedDistance'] > 50:
        signals.append('NEW_RIDER_HIGH_VALUE')
        fraud_score += 30
    
    # Signal 2: Multiple trips in short time (account takeover)
    recent_trips = trips_table.query(
        KeyConditionExpression='RiderId = :rider_id AND CreatedAt > :time',
        ExpressionAttributeValues={
            ':rider_id': rider_id,
            ':time': int(time.time()) - 3600  # Last hour
        }
    )['Count']
    
    if recent_trips > 5:
        signals.append('EXCESSIVE_TRIPS_HOURLY')
        fraud_score += 40
    
    # Signal 3: Payment method recently added (stolen card)
    payment_method_age = time.time() - rider.get('PaymentAddedAt', 0)
    if payment_method_age < 3600:  # Added in last hour
        signals.append('NEW_PAYMENT_METHOD')
        fraud_score += 25
    
    # Signal 4: GPS spoofing (pickup location impossible given rider's last location)
    last_location = rider.get('LastLocation')
    if last_location:
        distance_from_last = geodesic(
            (last_location['latitude'], last_location['longitude']),
            (sns_message['pickup']['latitude'], sns_message['pickup']['longitude'])
        ).miles
        
        time_since_last = time.time() - rider.get('LastLocationTime', 0)
        max_possible_distance = (time_since_last / 3600) * 60  # 60 mph max speed
        
        if distance_from_last > max_possible_distance * 1.5:
            signals.append('GPS_SPOOFING_SUSPECTED')
            fraud_score += 50
    
    # Action based on fraud score
    if fraud_score >= 70:
        # Block trip, require manual review
        trips_table.update_item(
            Key={'TripId': trip_id},
            UpdateExpression='SET #status = :status, FraudScore = :score, FraudSignals = :signals',
            ExpressionAttributeNames={'#status': 'Status'},
            ExpressionAttributeValues={
                ':status': 'BLOCKED_FRAUD',
                ':score': fraud_score,
                ':signals': signals
            }
        )
        
        # Alert fraud team
        sns = boto3.client('sns')
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:fraud-alerts',
            Subject=f'High-risk trip blocked: {trip_id}',
            Message=json.dumps({
                'tripId': trip_id,
                'riderId': rider_id,
                'fraudScore': fraud_score,
                'signals': signals
            })
        )
        
        print(f" Trip {trip_id} blocked (fraud score: {fraud_score})")
        
    elif fraud_score >= 40:
        # Flag for review but allow trip
        trips_table.update_item(
            Key={'TripId': trip_id},
            UpdateExpression='SET FraudScore = :score, FraudSignals = :signals, FlaggedForReview = :flag',
            ExpressionAttributeValues={
                ':score': fraud_score,
                ':signals': signals,
                ':flag': True
            }
        )
        
        print(f" Trip {trip_id} flagged for review (fraud score: {fraud_score})")
        
    else:
        # Low risk, allow trip
        print(f" Trip {trip_id} cleared (fraud score: {fraud_score})")

Monthly volume:
- Trips: 23M × 30 days = 690M/month
- SNS publishes: 690M publishes (one per trip)
- SNS deliveries: 690M × 5 subscribers = 3,450M deliveries/month

SNS publishes (TRIP_REQUESTED events):
- Cost: (690M - 1M free) × $0.50 / 1M = $344.50/month

SNS deliveries:
- SQS delivery (driver notification): 690M × $0.00 = $0 (free)
- Lambda delivery (pricing, fraud): 690M × 2 × $0.00 = $0 (free)
- Kinesis delivery (analytics): 690M × $0.00 = $0 (free)
- SQS delivery (compliance): 690M × $0.00 = $0 (free)

Total SNS cost: $344.50/month
Annual SNS cost: $344.50 × 12 = $4,134/year

Infrastructure (23M trips/day, 266 avg req/sec, 1,330 peak):
- EC2 instances: 30× r6i.4xlarge (16 vCPU, 128 GB RAM, RabbitMQ cluster)
  - Cost: $1.008/hour × 30 × 730 hours = $22,075/month
  - Rationale: Fanout requires high memory, 30 instances for 5× peak capacity
- EBS storage: 20 TB gp3 (message retention, 5 queues × 4 TB each)
  - Cost: $0.08/GB-month × 20,000 GB = $1,600/month
- Elastic Load Balancer: 2× Network Load Balancer (high throughput)
  - Cost: $32.40/month × 2 = $64.80/month
  - LCU charges: 1,330 req/sec peak × 0.006 LCU/req × $0.008/LCU-hour × 730 hours = $46.50/month
- VPC data transfer: $0.01/GB inter-AZ
  - Data: 115M messages × 5 KB avg = 575 GB/day = 17,250 GB/month
  - Cost: 17,250 GB × $0.01 = $172.50/month

Staff (5 SREs for 24/7 on-call, management, upgrades):
- Salary: $180K/year × 5 = $900K/year = $75,000/month
- Benefits (30%): $75,000 × 0.30 = $22,500/month
- Total staff: $97,500/month

Total self-hosted: $22,075 + $1,600 + $111.30 + $172.50 + $97,500 = $121,458.80/month
Annual self-hosted: $121,458.80 × 12 = $1,457,506/year

SNS savings: $1,457,506 - $4,134 = $1,453,372/year (99.7% cost reduction)

# EventBridge rule: Route high-value orders to special processing
{
  "source": ["uber.dispatch"],
  "detail-type": ["TRIP_REQUESTED"],
  "detail": {
    "estimatedFare": [{"numeric": [">", 100]}],  # Fare > $100
    "serviceType": ["UBER_BLACK", "UBER_LUX"]    # Premium service
  }
}
→ Target: Lambda function "high-value-trip-handler"

# EventBridge rule: Route international trips to compliance service
{
  "source": ["uber.dispatch"],
  "detail-type": ["TRIP_REQUESTED"],
  "detail": {
    "pickup": {
      "country": [{"anything-but": ["USA"]}]  # Non-US trips
    }
  }
}
→ Target: SQS queue "international-compliance-queue"

# SNS cannot do this - would require:
# 1. All subscribers receive all events
# 2. Each subscriber filters in application code
# 3. Wastes compute (Lambda invocations for filtered events)

# Fraud detection Lambda invoked for EVERY trip (23M/day)
# But only needs high-risk trips (new riders, high-value, etc.)

def lambda_handler(event, context):
    sns_message = json.loads(event['Records'][0]['Sns']['Message'])
    
    # Application-level filtering (wasteful - Lambda already invoked)
    if sns_message.get('riderTripCount', 100) > 10:
        # Experienced rider, low fraud risk, ignore
        return {'statusCode': 200, 'body': 'Skipped'}
    
    # Only 20% of trips reach here, but Lambda invoked 100% of time
    # Cost: 23M invocations/day × $0.20 / 1M = $4.60/day = $1,679/year

# Inefficient: 80% of Lambda invocations wasted (filtered in code)

# Subscribe fraud detection Lambda with filter
resource "aws_sns_topic_subscription" "fraud_detection_filtered" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "lambda"
  endpoint  = aws_lambda_function.fraud_detector.arn
  
  # Filter policy: Only invoke Lambda for high-risk trips
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED"],
    
    # New or low-trip-count riders
    riderTripCount = [{"numeric": ["<", 10]}],
    
    # OR high-value trips
    estimatedFare = [{"numeric": [">", 100]}]
  })
}

# Now Lambda only invoked for 20% of trips (4.6M/day)
# Cost: 4.6M invocations/day × $0.20 / 1M = $0.92/day = $336/year
# Savings: $1,679 - $336 = $1,343/year (80% reduction)

{
  "eventType": ["TRIP_REQUESTED"],           // Exact match (equality)
  
  "riderTripCount": [{"numeric": ["<", 10]}],  // Numeric comparison (>, <, >=, <=, =)
  
  "serviceType": [{"anything-but": ["UBER_POOL"]}],  // Negation (not equal)
  
  "pickupCity": [{"prefix": "San"}],  // Prefix match (San Francisco, San Jose, etc.)
  
  "estimatedFare": [{"numeric": [">=", 50, "<=", 150]}],  // Range (between 50 and 150)
  
  "paymentMethod": [{"exists": true}],  // Attribute exists
  
  "promoCode": [{"exists": false}]  // Attribute does not exist
}

# Driver notification: All trips (no filter)
resource "aws_sns_topic_subscription" "driver_notification" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.driver_notification_queue.arn
  
  # No filter policy - receive all TRIP_REQUESTED events
}

# Pricing: All trips (no filter)
resource "aws_sns_topic_subscription" "pricing" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "lambda"
  endpoint  = aws_lambda_function.pricing_calculator.arn
  
  # No filter - all trips need pricing
}

# Fraud detection: High-risk trips only (filtered)
resource "aws_sns_topic_subscription" "fraud_detection" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "lambda"
  endpoint  = aws_lambda_function.fraud_detector.arn
  
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED"],
    riderTripCount = [{"numeric": ["<", 10]}],  # New riders
    estimatedFare = [{"numeric": [">", 100]}]   # OR high-value
  })
}

# Compliance: International trips only (filtered)
resource "aws_sns_topic_subscription" "compliance_intl" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.compliance_intl_queue.arn
  
  filter_policy = jsonencode({
    eventType = ["TRIP_REQUESTED", "TRIP_COMPLETED"],
    pickupCountry = [{"anything-but": ["USA"]}]  # Non-US trips
  })
}

# Analytics: All events (no filter)
resource "aws_sns_topic_subscription" "analytics" {
  topic_arn = aws_sns_topic.trip_events.arn
  protocol  = "kinesis"
  endpoint  = aws_kinesis_stream.trip_analytics.arn
  
  # No filter - analyze all events
}

┌─────────────────────────────────────────────────────────────────────────┐
│                        Kafka Cluster (LinkedIn)                          │
│                     1,500 brokers, 100K topics, 7 trillion msg/day       │
└─────────────────────────────────────────────────────────────────────────┘
         │
         │ Topic: "user-activity" (all user actions: post, like, comment, share)
         │ Partitions: 1,000 (for parallel processing)
         │ Replication Factor: 3 (each partition copied to 3 brokers for durability)
         │ Retention: 7 days (replay any event from last week)
         │
         ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Topic: "user-activity"                                                   │
│ ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐│
│ │ Partition 0  │  │ Partition 1  │  │ Partition 2  │  │ Partition 999││
│ │              │  │              │  │              │  │    ...       ││
│ │ Offset 0: {} │  │ Offset 0: {} │  │ Offset 0: {} │  │              ││
│ │ Offset 1: {} │  │ Offset 1: {} │  │ Offset 1: {} │  │              ││
│ │ Offset 2: {} │  │ Offset 2: {} │  │ Offset 2: {} │  │              ││
│ │ ...          │  │ ...          │  │ ...          │  │              ││
│ │ Offset 10M:{}│  │ Offset 8M:{} │  │ Offset 12M:{}│  │              ││
│ └──────────────┘  └──────────────┘  └──────────────┘  └──────────────┘│
└─────────────────────────────────────────────────────────────────────────┘
         │                   │                   │                   │
         │ Consumer Group: "feed-builder" (100 consumers)                │
         │ Each consumer assigned specific partitions (load balancing)   │
         │                                                                │
         ├──────────────┬──────────────┬──────────────┬─────────────────┤
         ▼              ▼              ▼              ▼                 ▼
    Consumer 0     Consumer 1     Consumer 2     Consumer 3    ...  Consumer 99
    Partitions:    Partitions:    Partitions:    Partitions:        Partitions:
    0-9            10-19          20-29          30-39              990-999

Step 1: User Creates Post
┌─────────────────────┐
│  LinkedIn Member     │
│  (134M daily active) │
└─────────┬────────────┘
          │ POST /posts
          │ { "content": "Excited to announce..." }
          ▼
┌─────────────────────┐
│  Post API Service    │
│  (gRPC)              │
│                      │
│  1. Validate post    │
│  2. Store in DB      │
│  3. Publish to Kafka │
└─────────┬────────────┘
          │
          │ Kafka publish:
          │ Topic: "user-activity"
          │ Key: userId (for partitioning)
          │ Value: { "eventType": "POST_CREATED", "postId": "123", "authorId": "user-456", ... }
          ▼
┌─────────────────────────────────────────────────────────────────┐
│  Kafka Topic: "user-activity"                                    │
│  Partitions: 1,000                                               │
│  Partition determined by: hash(userId) % 1000                    │
│  → All events from same user go to same partition (ordered!)     │
└─────────┬────────────────────────────────────────────────────────┘
          │
          │ Consumed by multiple consumer groups (fanout, like SNS):
          │
    ┌─────┴──────┬──────────┬──────────┬──────────┐
    │            │          │          │          │
    ▼            ▼          ▼          ▼          ▼

┌──────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│Feed      │ │Search   │ │Analytics│ │Notific. │ │Fraud    │
│Builder   │ │Indexer  │ │Pipeline │ │Service  │ │Detectio │
│          │ │         │ │         │ │         │ │n        │
│100       │ │50       │ │20       │ │30       │ │10       │
│consumers │ │consumers│ │consumers│ │consumers│ │consumers│
└──────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘
    │            │          │          │          │
    │            │          │          │          │
    ▼            ▼          ▼          ▼          ▼
[Updates     [Indexes   [Stores in  [Pushes to  [Analyzes
 feeds of     post in    data        mobile      for spam,
 followers]   search     warehouse]  app]        fake news]
             engine]

Step 2: Feed Builder Consumer Group
┌─────────────────────────────────────────────────────────────────┐
│  Feed Builder Consumer Group (100 consumers)                     │
│                                                                  │
│  Consumer 0 (assigned partitions 0-9):                           │
│    1. Read event from Kafka (POST_CREATED)                       │
│    2. Query graph database: get followers of authorId            │
│       → Returns 930 follower IDs (average connections)           │
│    3. For each follower:                                         │
│       - Add post to their feed (Redis sorted set, score=timestamp)│
│       - Keep top 1,000 posts per user (older posts expire)       │
│    4. Commit offset to Kafka (processed successfully)            │
│                                                                  │
│  Consumer 1 (assigned partitions 10-19):                         │
│    ... same logic for different partitions ...                   │
│                                                                  │
│  If consumer crashes:                                            │
│    - Kafka reassigns partitions 0-9 to Consumer 50               │
│    - Consumer 50 starts from last committed offset               │
│    - No events lost or duplicated (exactly-once semantics)       │
└─────────────────────────────────────────────────────────────────┘

from kafka import KafkaProducer
import json
import time

# Initialize Kafka producer (reuse across requests for performance)
producer = KafkaProducer(
    bootstrap_servers=['kafka-1.linkedin.com:9092', 'kafka-2.linkedin.com:9092', 'kafka-3.linkedin.com:9092'],
    
    # Serialization: Python dict → JSON bytes
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    
    # Acknowledgment settings
    acks='all',  # Wait for all replicas to acknowledge (durability)
    # acks=1: Wait for leader only (faster but less durable)
    # acks=0: Don't wait (fastest but can lose data if broker fails)
    
    # Batching (performance optimization)
    batch_size=16384,  # 16 KB batches
    linger_ms=10,  # Wait up to 10ms to batch messages
    
    # Compression (reduce network bandwidth)
    compression_type='snappy',  # Fast compression (3× size reduction)
    # Options: none, gzip, snappy, lz4, zstd
    
    # Retries (reliability)
    retries=3,  # Retry up to 3 times on failure
    retry_backoff_ms=100,  # Wait 100ms between retries
    
    # Idempotence (prevent duplicates on retry)
    enable_idempotence=True,  # Exactly-once semantics
    
    # Timeouts
    request_timeout_ms=30000,  # 30 second timeout per request
)

def create_post(user_id, content):
    """
    Create post and publish event to Kafka
    """
    # 1. Store post in database (PostgreSQL)
    post_id = store_post_in_db(user_id, content)
    
    # 2. Publish event to Kafka
    event = {
        'eventType': 'POST_CREATED',
        'postId': post_id,
        'authorId': user_id,
        'content': content,
        'timestamp': int(time.time() * 1000),  # Milliseconds
        'visibility': 'PUBLIC',  # or CONNECTIONS_ONLY, PRIVATE
    }
    
    # Send to Kafka topic "user-activity"
    # Key = userId (ensures all user's events go to same partition → ordered)
    future = producer.send(
        topic='user-activity',
        key=user_id,  # Partition by userId (hash(userId) % partition_count)
        value=event,
        timestamp_ms=int(time.time() * 1000)  # Event timestamp (for ordering)
    )
    
    # Optional: Wait for acknowledgment (synchronous)
    # record_metadata = future.get(timeout=10)  # Blocks until ack received
    # print(f"Published to partition {record_metadata.partition} offset {record_metadata.offset}")
    
    # Asynchronous callback (non-blocking, recommended for high throughput)
    future.add_callback(on_send_success)
    future.add_errback(on_send_error)
    
    return post_id

def on_send_success(record_metadata):
    """Called when Kafka acknowledges message"""
    print(f" Published to partition {record_metadata.partition} offset {record_metadata.offset}")

def on_send_error(excp):
    """Called when publish fails (after retries)"""
    print(f" Failed to publish: {excp}")
    # Alert monitoring system (PagerDuty, Datadog, etc.)

def store_post_in_db(user_id, content):
    """Store post in PostgreSQL (details omitted)"""
    # INSERT INTO posts (user_id, content, created_at) VALUES (...)
    return "post-123456"  # Return post ID

from kafka import KafkaConsumer
import json
import redis

# Initialize Kafka consumer (one per process)
consumer = KafkaConsumer(
    'user-activity',  # Topic to consume
    
    bootstrap_servers=['kafka-1.linkedin.com:9092', 'kafka-2.linkedin.com:9092'],
    
    # Consumer group (load balancing across 100 consumers)
    group_id='feed-builder-group',
    
    # Start from beginning if no previous offset (new consumer group)
    auto_offset_reset='earliest',  # Options: earliest, latest, none
    # earliest: Start from offset 0 (replay all retained events, 7 days)
    # latest: Start from current offset (only new events)
    # none: Fail if no previous offset (require explicit offset)
    
    # Offset commit strategy
    enable_auto_commit=False,  # Manual commit (more control)
    # enable_auto_commit=True: Auto-commit every 5 seconds (simple but risky)
    
    # Deserialization: JSON bytes → Python dict
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    key_deserializer=lambda k: k.decode('utf-8') if k else None,
    
    # Max records per poll() call
    max_poll_records=500,  # Fetch 500 events per poll (batch processing)
    
    # Session timeout (consumer considered dead if no heartbeat)
    session_timeout_ms=30000,  # 30 seconds
    
    # Max poll interval (consumer considered dead if no poll() call)
    max_poll_interval_ms=300000,  # 5 minutes (allows slow processing)
)

# Redis connection for feed storage
redis_client = redis.Redis(
    host='feed-cache.redis.linkedin.com',
    port=6379,
    db=0,
    decode_responses=True
)

def consume_events():
    """
    Main consumer loop
    Runs forever, processing events from Kafka
    """
    print(f"Starting consumer, assigned partitions: {consumer.assignment()}")
    
    try:
        for message in consumer:
            # Message metadata
            partition = message.partition
            offset = message.offset
            key = message.key  # userId
            event = message.value  # Deserialized JSON
            
            print(f"Processing partition={partition} offset={offset} event={event['eventType']}")
            
            # Process event
            process_event(event)
            
            # Commit offset (manual commit for exactly-once semantics)
            consumer.commit()
            
    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()

def process_event(event):
    """
    Process POST_CREATED event: Add to follower feeds
    """
    if event['eventType'] != 'POST_CREATED':
        return  # Ignore other event types (LIKE, COMMENT, etc.)
    
    post_id = event['postId']
    author_id = event['authorId']
    timestamp = event['timestamp']
    
    # 1. Get followers from graph database (or cache)
    followers = get_followers(author_id)  # Returns list of user IDs
    
    print(f"Adding post {post_id} to feeds of {len(followers)} followers")
    
    # 2. Add post to each follower's feed (Redis sorted set)
    for follower_id in followers:
        feed_key = f"feed:{follower_id}"
        
        # ZADD: Add to sorted set with score = timestamp (chronological order)
        redis_client.zadd(
            feed_key,
            {post_id: timestamp},  # {member: score}
            nx=True  # Only add if not exists (idempotent)
        )
        
        # Keep only top 1,000 posts (remove older posts to save memory)
        redis_client.zremrangebyrank(feed_key, 0, -1001)  # Keep top 1,000 (highest scores)
        
        # Set TTL (expire feed after 7 days of inactivity)
        redis_client.expire(feed_key, 604800)  # 7 days in seconds
    
    print(f" Updated {len(followers)} feeds")

def get_followers(user_id):
    """
    Get user's followers from graph database
    (Simplified - actual implementation uses cached results)
    """
    # Query Neo4j graph database: MATCH (user {id: user_id})<-[:FOLLOWS]-(follower) RETURN follower.id
    # Returns: ['user-1', 'user-2', ..., 'user-930'] (average 930 connections)
    
    # For demo, return dummy list
    return [f"user-{i}" for i in range(1, 931)]  # 930 followers (LinkedIn average)

Topic: "user-activity"
Partitions: 10
Consumer group: "feed-builder" (10 consumers, one per partition)

Throughput per partition:
- 9M posts/day ÷ 10 partitions = 900K posts/partition/day
- 900K ÷ 86,400 seconds = 10.4 posts/second per partition

Processing time per post:
- Get 930 followers: 10ms (Redis cache hit)
- Update 930 feeds: 930 × 1ms (Redis ZADD) = 930ms
- Total: 940ms per post

Max throughput per consumer:
- 1 post / 940ms = 1.06 posts/second
- 10 consumers × 1.06 = 10.6 posts/second total

Required throughput: 104 posts/second (9M ÷ 86,400)
Actual throughput: 10.6 posts/second
 INSUFFICIENT - need 10× more partitions!

Topic: "user-activity"
Partitions: 1,000
Consumer group: "feed-builder" (100 consumers, 10 partitions each)

Throughput per partition:
- 9M posts/day ÷ 1,000 partitions = 9K posts/partition/day
- 9K ÷ 86,400 seconds = 0.104 posts/second per partition

Processing time: 940ms per post (same as above)

Max throughput per consumer (10 partitions each):
- 1 post / 940ms × 10 partitions = 10.6 posts/second per consumer
- 100 consumers × 10.6 = 1,060 posts/second total

Required throughput: 104 posts/second
Actual throughput: 1,060 posts/second
 SUFFICIENT - 10× headroom for traffic spikes!

Topic: "user-activity"
Partitions: 10,000
Consumer group: "feed-builder" (1,000 consumers, 10 partitions each)

Throughput per partition:
- 9M posts/day ÷ 10,000 partitions = 900 posts/partition/day
- 900 ÷ 86,400 seconds = 0.01 posts/second per partition

Overhead:
- Each partition stores metadata (leader, replicas, offsets): 1 KB
- 10,000 partitions × 3 replicas × 1 KB = 30 MB metadata
- Leader election on broker failure: 10,000 partitions × 5ms = 50 seconds (!)
- Consumer rebalance: Assigning 10,000 partitions to 1,000 consumers = 30 seconds

 INEFFICIENT - excessive metadata, slow rebalances, wasted resources

Partitions = (Target Throughput × Processing Time) / (Desired Headroom)

LinkedIn example:
- Target throughput: 104 posts/second (9M/day)
- Processing time: 0.940 seconds per post
- Desired headroom: 10× (for traffic spikes)

Partitions = (104 posts/sec × 0.940 sec) × 10 = 978 partitions
Round up to: 1,000 partitions 

Initial assignment (round-robin):
- Consumer 0: Partitions 0, 100, 200, 300, 400, 500, 600, 700, 800, 900 (10 partitions)
- Consumer 1: Partitions 1, 101, 201, 301, 401, 501, 601, 701, 801, 901 (10 partitions)
- Consumer 2: Partitions 2, 102, 202, 302, 402, 502, 602, 702, 802, 902 (10 partitions)
- ...
- Consumer 99: Partitions 99, 199, 299, 399, 499, 599, 699, 799, 899, 999 (10 partitions)

Each consumer processes 10 partitions (load balanced evenly)

Timeline:
T+0s:   All 100 consumers healthy, processing normally
        [1,000 partitions ÷ 100 consumers = 10 partitions each]

T+60s:  Consumer 42 crashes (out of memory, host failure, etc.)
        Kafka detects after session_timeout_ms (30 seconds)
        [99 healthy consumers remain]

T+90s:  Kafka triggers rebalance:
        1. Pause all consumers (stop processing)
        2. Reassign partitions across 99 consumers
           - 1,000 partitions ÷ 99 consumers = 10 partitions for 10 consumers, 11 partitions for remaining 89
        3. Resume processing

        New assignment (Consumer 42's partitions distributed):
        - Consumer 0: Partitions 0, 100, 200, 300, 400, 500, 600, 700, 800, 900 (10 partitions, unchanged)
        - Consumer 1: Partitions 1, 101, 201, 301, 401, 501, 601, 701, 801, 901, 42 (11 partitions, took partition 42)
        - Consumer 2: Partitions 2, 102, 202, 302, 402, 502, 602, 702, 802, 902, 142 (11 partitions, took partition 142)
        - ...
        - Consumer 50: Partitions 50, 150, 250, 350, 450, 550, 650, 750, 850, 950, 242 (11 partitions)
        - ... (partitions 42, 142, 242, ... distributed across consumers 1-10)

T+95s:  All consumers resumed, processing continues
        Consumer 42's partitions processed by other consumers (no data loss)

T+120s: Consumer 42 recovers (auto-scaling starts new instance)
        Joins consumer group

T+150s: Kafka triggers rebalance again:
        1. Pause all consumers
        2. Reassign partitions across 100 consumers
           - Back to 10 partitions each
        3. Resume processing

┌─────────────────────────────────────────────────────────────────────────┐
│                    Netflix Subscribers (230M)                            │
│                    Watching on 4,000+ device types                       │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                             │ HTTPS streaming (play, pause, seek events)
                             ▼
                   ┌──────────────────────┐
                   │   Playback Service    │
                   │   (Spring Boot)       │
                   │                       │
                   │   - Tracks playback   │
                   │   - Records events    │
                   │   - Publishes to      │
                   │     Kafka             │
                   └──────────┬────────────┘
                             │
                             │ Kafka produce (async, batched)
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│  Kafka Topic: "playback-events"                                          │
│  Partitions: 5,000 (sharded by subscriberId)                            │
│  Throughput: 41,667 events/second average (3.6B/day ÷ 86,400)          │
│  Peak: 150,000 events/second (Friday/Saturday nights, 3.6× average)     │
│  Retention: 24 hours (replay for reprocessing)                          │
│  Replication: 3× (durability across availability zones)                 │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                             │ Multiple consumer groups (fanout)
                             │
        ┌────────────────────┼────────────────────┬──────────────────┐
        │                    │                    │                  │
        ▼                    ▼                    ▼                  ▼
┌───────────────┐  ┌──────────────────┐  ┌──────────────┐  ┌──────────────┐
│Recommendation │  │ Continue Watching │  │   CDN        │  │ A/B Test     │
│   Pipeline    │  │    Sync          │  │   Router     │  │  Analytics   │
│               │  │                  │  │              │  │              │
│Flink Stream   │  │ Flink Stream     │  │ Flink Stream │  │ Spark Stream │
│Processing     │  │ Processing       │  │ Processing   │  │              │
│               │  │                  │  │              │  │              │
│500 tasks      │  │ 200 tasks        │  │ 100 tasks    │  │ 50 tasks     │
└───────┬───────┘  └────────┬─────────┘  └──────┬───────┘  └──────┬───────┘
        │                   │                   │                  │
        │ Update            │ Update           │ Update           │ Aggregate
        │ recommendations   │ position         │ routing          │ metrics
        │ (< 5 min)         │ (< 1 min)        │ (< 1 min)        │ (real-time)
        ▼                   ▼                   ▼                  ▼
┌───────────────┐  ┌──────────────────┐  ┌──────────────┐  ┌──────────────┐
│   Cassandra   │  │   Redis          │  │  DynamoDB    │  │  Druid       │
│   (Recs DB)   │  │   (Position      │  │  (CDN        │  │  (Metrics    │
│               │  │    Cache)        │  │   Routes)    │  │   OLAP)      │
│ 10 PB data    │  │  5 TB cache      │  │  500 GB      │  │  50 TB       │
└───────┬───────┘  └────────┬─────────┘  └──────┬───────┘  └──────┬───────┘
        │                   │                   │                  │
        │                   │                   │                  │
        ▼                   ▼                   ▼                  ▼
┌───────────────────────────────────────────────────────────────────┐
│                    Netflix API Gateway                             │
│                                                                    │
│  GET /recommendations → Read from Cassandra (< 5 min fresh)       │
│  GET /continue-watching → Read from Redis (< 1 min fresh)         │
│  GET /player/manifest → Read CDN route from DynamoDB              │
│  GET /experiments/metrics → Query Druid (real-time dashboards)    │
└───────────────────────────────────────────────────────────────────┘

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Properties;
import java.util.UUID;

public class PlaybackEventProducer {
    
    private final KafkaProducer<String, String> producer;
    private final ObjectMapper objectMapper;
    private static final String TOPIC = "playback-events";
    
    public PlaybackEventProducer() {
        Properties props = new Properties();
        
        // Kafka brokers (multiple for high availability)
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "kafka-1.netflix.com:9092,kafka-2.netflix.com:9092,kafka-3.netflix.com:9092");
        
        // Serialization
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // Acknowledgment (durability vs latency trade-off)
        props.put(ProducerConfig.ACKS_CONFIG, "1");  // Wait for leader only (faster)
        // acks=all: Wait for all replicas (safer but 2× slower, 10ms vs 5ms)
        // acks=1: Wait for leader only (Netflix's choice: 5ms P95, acceptable data loss risk)
        // acks=0: Fire-and-forget (fastest but can lose events, NOT used)
        
        // Batching (critical for high throughput)
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);  // 64 KB batches
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);  // Wait 10ms to batch
        // At 41,667 events/sec, 10ms window captures ~417 events per batch
        // Average event size: 500 bytes
        // Batch size: 417 events × 500 bytes = 208 KB (exceeds 64 KB, triggers early send)
        // Result: ~10 batches/sec per producer (4,167 events/batch actual)
        
        // Compression (reduce network bandwidth)
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");  // Fast compression
        // Event: 500 bytes → 150 bytes compressed (70% reduction)
        // Network savings: 3.6B events × 500 bytes = 1.8 TB/day → 540 GB/day (1.26 TB saved)
        
        // Buffer memory (handle traffic spikes)
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);  // 64 MB buffer
        // At 150K events/sec peak, buffer holds: 64 MB ÷ 500 bytes = 131K events = 0.87 seconds
        // Prevents blocking during short Kafka unavailability
        
        // Retries (reliability)
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);  // 100ms between retries
        
        // Idempotence (prevent duplicates on retry)
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        
        // Timeouts
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);  // 30 seconds
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);  // Block max 60s if buffer full
        
        this.producer = new KafkaProducer<>(props);
        this.objectMapper = new ObjectMapper();
    }
    
    public void publishPlayEvent(String subscriberId, String titleId, String deviceId, 
                                  long positionMillis, String sessionId) {
        try {
            // Build event
            PlaybackEvent event = PlaybackEvent.builder()
                .eventId(UUID.randomUUID().toString())
                .eventType("PLAY")
                .subscriberId(subscriberId)
                .titleId(titleId)
                .deviceId(deviceId)
                .positionMillis(positionMillis)
                .sessionId(sessionId)
                .timestamp(System.currentTimeMillis())
                .deviceType(getDeviceType(deviceId))
                .country(getCountry(subscriberId))
                .build();
            
            // Serialize to JSON
            String json = objectMapper.writeValueAsString(event);
            
            // Create Kafka record
            // Key = subscriberId (ensures all subscriber's events go to same partition)
            ProducerRecord<String, String> record = new ProducerRecord<>(
                TOPIC,           // topic
                subscriberId,    // key (partition by subscriberId for ordering)
                json            // value (JSON event)
            );
            
            // Send asynchronously (non-blocking, callback on completion)
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // Log failure (monitoring system alerts if error rate > 0.1%)
                    logger.error("Failed to publish event: {}", event.getEventId(), exception);
                    // Store in dead letter queue (S3) for later replay
                    storeInDeadLetterQueue(event);
                } else {
                    // Success (logged at DEBUG level only, too verbose for INFO)
                    logger.debug("Published event {} to partition {} offset {}", 
                        event.getEventId(), metadata.partition(), metadata.offset());
                }
            });
            
            // Return immediately (don't wait for Kafka ack, async for low latency)
            
        } catch (Exception e) {
            logger.error("Error creating playback event", e);
        }
    }
    
    public void publishPauseEvent(String subscriberId, String titleId, String deviceId,
                                   long positionMillis, String sessionId) {
        // Similar to publishPlayEvent, eventType="PAUSE"
        // Position critical for "Continue Watching" feature
    }
    
    public void publishSeekEvent(String subscriberId, String titleId, String deviceId,
                                  long fromPositionMillis, long toPositionMillis, String sessionId) {
        // Seek events indicate engagement (skip intro, rewatch scene)
        // Used by recommendation ML model (high seek rate = confusing content)
    }
    
    public void publishStopEvent(String subscriberId, String titleId, String deviceId,
                                  long positionMillis, long totalDurationMillis, String sessionId) {
        // Stop event with completion percentage
        // Completion rate critical for recommendations:
        // - < 25%: Likely poor match, downrank similar titles
        // - 25-75%: Partial interest
        // - > 75%: Strong interest, recommend similar titles
        
        try {
            PlaybackEvent event = PlaybackEvent.builder()
                .eventId(UUID.randomUUID().toString())
                .eventType("STOP")
                .subscriberId(subscriberId)
                .titleId(titleId)
                .deviceId(deviceId)
                .positionMillis(positionMillis)
                .totalDurationMillis(totalDurationMillis)
                .completionPercentage((double) positionMillis / totalDurationMillis * 100)
                .sessionId(sessionId)
                .timestamp(System.currentTimeMillis())
                .build();
            
            String json = objectMapper.writeValueAsString(event);
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, subscriberId, json);
            
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    logger.error("Failed to publish stop event: {}", event.getEventId(), exception);
                    storeInDeadLetterQueue(event);
                }
            });
            
        } catch (Exception e) {
            logger.error("Error creating stop event", e);
        }
    }
    
    private void storeInDeadLetterQueue(PlaybackEvent event) {
        // Store failed events in S3 for later replay
        // S3 path: s3://netflix-playback-dlq/year=2024/month=01/day=15/hour=14/event-{uuid}.json
        // Batch replay job runs hourly, republishes failed events
    }
    
    public void close() {
        producer.close();  // Flush remaining batched messages, close connections
    }
}

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import java.util.Properties;

public class RecommendationPipeline {
    
    public static void main(String[] args) throws Exception {
        
        // Flink execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Parallelism (500 tasks = 500 partitions processed in parallel)
        env.setParallelism(500);
        
        // Enable checkpointing (fault tolerance)
        env.enableCheckpointing(60000);  // Checkpoint every 60 seconds
        // Checkpoint stores consumer offsets + application state
        // If job crashes, restart from last checkpoint (no data loss, no duplicates)
        
        // Kafka consumer properties
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "kafka-1.netflix.com:9092,kafka-2.netflix.com:9092");
        kafkaProps.setProperty("group.id", "recommendation-pipeline");
        kafkaProps.setProperty("auto.offset.reset", "latest");  // Start from latest (not replay history)
        kafkaProps.setProperty("enable.auto.commit", "false");  // Flink manages offsets via checkpoints
        
        // Kafka source
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
            "playback-events",
            new SimpleStringSchema(),
            kafkaProps
        );
        
        // Start reading from Kafka
        DataStream<String> eventStream = env.addSource(kafkaConsumer);
        
        // Parse JSON events
        DataStream<PlaybackEvent> parsedEvents = eventStream
            .map(json -> objectMapper.readValue(json, PlaybackEvent.class))
            .filter(event -> event.getEventType().equals("STOP"))  // Only STOP events (completion data)
            .filter(event -> event.getCompletionPercentage() > 75)  // Only if watched > 75%
            .name("Parse and filter events");
        
        // Key by subscriberId (group all subscriber's events together)
        DataStream<PlaybackEvent> keyedEvents = parsedEvents
            .keyBy(event -> event.getSubscriberId());
        
        // Window: 5-minute tumbling window (aggregate events every 5 minutes)
        DataStream<RecommendationUpdate> recommendations = keyedEvents
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .process(new RecommendationFunction())
            .name("Generate recommendations");
        
        // Sink: Write to Cassandra (recommendations database)
        recommendations
            .addSink(new CassandraSink<>(
                "INSERT INTO recommendations (subscriber_id, title_id, score, updated_at) VALUES (?, ?, ?, ?)"
            ))
            .name("Write to Cassandra");
        
        // Execute Flink job
        env.execute("Netflix Recommendation Pipeline");
    }
    
    // Process function: Generate recommendations from viewing history
    public static class RecommendationFunction 
            extends ProcessWindowFunction<PlaybackEvent, RecommendationUpdate, String, TimeWindow> {
        
        @Override
        public void process(String subscriberId, Context context, 
                          Iterable<PlaybackEvent> events, 
                          Collector<RecommendationUpdate> out) {
            
            // Collect all titles watched in this 5-minute window
            List<String> watchedTitles = new ArrayList<>();
            for (PlaybackEvent event : events) {
                watchedTitles.add(event.getTitleId());
            }
            
            if (watchedTitles.isEmpty()) {
                return;  // No recommendations to update
            }
            
            // Query ML model: Get similar titles
            // Model trained on 10+ years of viewing data (collaborative filtering)
            // Input: List of watched titles
            // Output: List of recommended titles with scores
            List<Recommendation> recommendations = mlModelClient.getSimilarTitles(watchedTitles);
            
            // Emit recommendation updates
            for (Recommendation rec : recommendations) {
                out.collect(new RecommendationUpdate(
                    subscriberId,
                    rec.getTitleId(),
                    rec.getScore(),
                    System.currentTimeMillis()
                ));
            }
            
            // Log statistics
            logger.info("Generated {} recommendations for subscriber {} based on {} titles watched",
                recommendations.size(), subscriberId, watchedTitles.size());
        }
    }
}

public class ContinueWatchingPipeline {
    
    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(200);  // 200 tasks (lower parallelism, simpler processing)
        env.enableCheckpointing(30000);  // Checkpoint every 30 seconds (faster recovery)
        
        // Kafka source
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
            "playback-events",
            new SimpleStringSchema(),
            getKafkaProps("continue-watching-pipeline")
        );
        
        DataStream<String> eventStream = env.addSource(kafkaConsumer);
        
        // Parse and filter PAUSE/STOP events (position updates)
        DataStream<PlaybackEvent> positionEvents = eventStream
            .map(json -> objectMapper.readValue(json, PlaybackEvent.class))
            .filter(event -> event.getEventType().equals("PAUSE") || event.getEventType().equals("STOP"))
            .name("Parse position events");
        
        // Key by (subscriberId, titleId) - track position per title per subscriber
        DataStream<PositionUpdate> positions = positionEvents
            .keyBy(event -> event.getSubscriberId() + ":" + event.getTitleId())
            .process(new PositionTrackingFunction())
            .name("Track positions");
        
        // Sink: Write to Redis (low-latency cache)
        positions
            .addSink(new RedisSink())
            .name("Write to Redis");
        
        env.execute("Netflix Continue Watching Pipeline");
    }
    
    public static class PositionTrackingFunction 
            extends KeyedProcessFunction<String, PlaybackEvent, PositionUpdate> {
        
        // Keyed state: Store latest position for this (subscriber, title) pair
        private transient ValueState<Long> latestPositionState;
        
        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
                "latest-position",
                Long.class,
                0L  // Default: 0 milliseconds
            );
            latestPositionState = getRuntimeContext().getState(descriptor);
        }
        
        @Override
        public void processElement(PlaybackEvent event, Context context, Collector<PositionUpdate> out) 
                throws Exception {
            
            Long previousPosition = latestPositionState.value();
            Long currentPosition = event.getPositionMillis();
            
            // Only update if position advanced (ignore rewinds, seeks backward)
            if (currentPosition > previousPosition) {
                latestPositionState.update(currentPosition);
                
                out.collect(new PositionUpdate(
                    event.getSubscriberId(),
                    event.getTitleId(),
                    currentPosition,
                    event.getTotalDurationMillis(),
                    System.currentTimeMillis()
                ));
            }
        }
    }
    
    public static class RedisSink extends RichSinkFunction<PositionUpdate> {
        
        private transient RedisClient redisClient;
        
        @Override
        public void open(Configuration parameters) {
            redisClient = new RedisClient("redis-cluster.netflix.com", 6379);
        }
        
        @Override
        public void invoke(PositionUpdate position, Context context) {
            String key = String.format("continue:%s:%s", 
                position.getSubscriberId(), position.getTitleId());
            
            // Store as hash with multiple fields
            redisClient.hset(key, "position", String.valueOf(position.getPositionMillis()));
            redisClient.hset(key, "duration", String.valueOf(position.getTotalDurationMillis()));
            redisClient.hset(key, "updated_at", String.valueOf(position.getTimestamp()));
            
            // Calculate percentage
            double percentage = (double) position.getPositionMillis() / position.getTotalDurationMillis() * 100;
            redisClient.hset(key, "percentage", String.format("%.1f", percentage));
            
            // Set TTL: Expire after 30 days (unlikely to resume after 30 days)
            redisClient.expire(key, 2592000);  // 30 days in seconds
        }
    }
}

User scenario: Watch on TV, resume on mobile

T+0s:    User starts watching "Stranger Things S4E1" on TV
         Position: 0:00 (beginning)

T+10m:   User pauses at 10:23 (10 minutes 23 seconds)
         → PAUSE event published to Kafka
         → Flink processes event (< 5 seconds latency)
         → Redis updated: continue:subscriber-123:title-456 = {position: 623000, ...}

T+15m:   User opens Netflix mobile app
         → App calls API: GET /continue-watching?subscriberId=123
         → API reads from Redis: continue:subscriber-123:*
         → Returns: "Stranger Things S4E1" at 10:23 (62.3% complete)

T+16m:   User taps "Continue Watching"
         → Mobile app starts playback at position 623000ms (10:23)
         → Seamless experience: Resume exactly where left off on TV

Total latency: < 1 minute (pause on TV → visible on mobile)
Old batch system: 1 hour (hourly sync job)

Kafka Cluster:
- EC2 instances: 500× i4i.4xlarge (16 vCPU, 128 GB RAM, 3.75 TB NVMe SSD)
  - Cost: $2.074/hour × 500 × 730 hours = $757,010/month
  - Rationale: High throughput (150K events/sec peak), low latency storage (NVMe)
- Storage: 3.75 TB × 500 instances = 1,875 TB total (24-hour retention)
  - Included with instance cost (local NVMe SSD)
- Data transfer: 3.6B events × 150 bytes (compressed) = 540 GB/day = 16.2 TB/month
  - Intra-region: $0.01/GB × 16,200 GB = $162/month
  
Flink Cluster:
- EC2 instances: 1,000× r6i.2xlarge (8 vCPU, 64 GB RAM, Flink tasks)
  - Cost: $0.504/hour × 1,000 × 730 hours = $367,920/month
  - Rationale: 500 recommendation tasks + 200 continue-watching + 300 other pipelines
- State backend (RocksDB): 10 TB EBS gp3 storage (Flink checkpoints)
  - Cost: $0.08/GB-month × 10,000 GB = $800/month

Cassandra Cluster (recommendations database):
- EC2 instances: 200× i4i.2xlarge (8 vCPU, 64 GB RAM, 1.87 TB NVMe)
  - Cost: $1.037/hour × 200 × 730 hours = $151,402/month
- Storage: 1.87 TB × 200 = 374 TB (10 PB compressed with CQL compression)

Redis Cluster (continue watching cache):
- ElastiCache: 50× cache.r6g.4xlarge (16 vCPU, 104 GB RAM, 5 TB total)
  - Cost: $1.344/hour × 50 × 730 hours = $49,056/month

Staff (20 SREs for 24/7 on-call, platform management):
- Salary: $180K/year × 20 = $3.6M/year = $300,000/month
- Benefits (30%): $300,000 × 0.30 = $90,000/month
- Total staff: $390,000/month

Total real-time: $757,010 + $367,920 + $151,402 + $49,056 + $390,000 = $1,715,388/month
Annual: $1,715,388 × 12 = $20,584,656/year

Hadoop Cluster (daily batch jobs):
- EC2 instances: 2,000× r6i.4xlarge (16 vCPU, 128 GB RAM, MapReduce workers)
  - Cost: $1.008/hour × 2,000 × 730 hours = $1,471,680/month
  - Rationale: Batch processing requires more compute (reprocess all data daily)
- Storage: S3 for raw events (3.6B events × 500 bytes × 30 days retention)
  - Data: 3.6B × 500 bytes = 1.8 TB/day × 30 days = 54 TB
  - Cost: $0.023/GB-month × 54,000 GB = $1,242/month
- Data transfer: 1.8 TB/day × 30 days = 54 TB/month (S3 → Hadoop)
  - Cost: $0.09/GB × 54,000 GB = $4,860/month

Cassandra Cluster (same as real-time):
- Cost: $151,402/month

Staff (30 data engineers + SREs for Hadoop maintenance, slower iteration):
- Salary: $180K/year × 30 = $5.4M/year = $450,000/month
- Benefits: $450,000 × 0.30 = $135,000/month
- Total staff: $585,000/month

Total batch: $1,471,680 + $1,242 + $4,860 + $151,402 + $585,000 = $2,214,184/month
Annual: $2,214,184 × 12 = $26,570,208/year

Cost comparison:
- Real-time (Kafka + Flink): $20.58M/year
- Batch (Hadoop): $26.57M/year
- Savings: $5.99M/year (23% cheaper with real-time!)

Plus intangible benefits:
- Real-time recommendations: 80% viewing from recommendations (up from 60% with batch)
- Continue watching: < 1 min sync (down from 1 hour batch)
- A/B test velocity: 250 simultaneous experiments (up from 10 with daily batch)
- Faster time-to-market: Real-time dashboards enable same-day decisions

Topic: "playback-events"
Retention: 24 hours (current Netflix setting for production cost optimization)

To replay 30 days:
- Increase retention to 30 days (temporary, for reprocessing)
- Storage required: 3.6B events/day × 30 days × 150 bytes (compressed) = 16.2 TB
- Cost: 16.2 TB × $0.08/GB-month (EBS gp3) = $1,296/month
- Similar cost to S3, but faster replay (Kafka sequential reads vs S3 random reads)

Reprocessing steps:
1. Deploy new Flink job with updated ML model
2. Set auto.offset.reset='earliest' (replay from 30 days ago)
3. Process events at full speed (150K events/sec with 500 tasks)
4. Duration: 3.6B events/day × 30 days ÷ 150K events/sec = 720K seconds = 200 hours
   Wait, that's 8.3 days! Too slow.
   
Optimization: Scale out to 5,000 Flink tasks (10× parallelism)
- Throughput: 150K events/sec × 10 = 1.5M events/sec
- Duration: 3.6B × 30 ÷ 1.5M = 72K seconds = 20 hours
- Cost: 5,000 tasks × $0.504/hour × 20 hours = $50,400 (one-time)

Result: 20-hour reprocessing vs 8-hour Hadoop batch
- Kafka slightly slower for full reprocessing (sequential reads, network overhead)
- But Kafka enables real-time + batch workloads on same infrastructure (dual-purpose)
- Hadoop requires separate infrastructure (real-time + batch = 2× cost)

Topic: "playback-events"
Retention: 24 hours (production)
- Enables replay for bug fixes (<24 hour window)
- Minimizes storage cost ($1,296/month × 1/30 = $43/month for 1-day retention)

Topic: "playback-events-archive" (copy of playback-events)
Retention: 365 days
Partitions: 100 (lower parallelism, archive workload not latency-sensitive)
Purpose: Long-term replay for major model updates
Cost: 3.6B events/day × 365 days × 150 bytes = 197 TB × $0.08/GB = $15,760/month

Total Kafka storage: $43/month (prod) + $15,760/month (archive) = $15,803/month
vs S3 archive: 197 TB × $0.023/GB = $4,531/month (71% cheaper than Kafka for long-term storage)

Hybrid strategy (Netflix actual):
- Kafka: 24-hour retention for real-time + short-term replay ($43/month)
- S3: Archive to S3 after 24 hours for long-term replay ($4,531/month)
- Total: $43 + $4,531 = $4,574/month (saves $11,229/month vs Kafka-only long-term storage)

Best of both worlds: Real-time with Kafka, long-term archive with S3

Kafka latency: 5-10ms produce + 10-20ms consume + network = 50-100ms P95
Acceptable for analytics, too slow for chat (users expect < 100ms end-to-end)

WebSocket connection overhead with Kafka:
- User A sends message
- API server publishes to Kafka (10ms)
- Message processed by consumer (20ms)
- Consumer pushes to WebSocket server (10ms)
- WebSocket server sends to User B (5ms)
Total: 45ms server-side + 50ms network = 95ms P95

Under load (100K msg/sec):
- Kafka queue builds up (1-2 second lag)
- Users see messages delayed by 2+ seconds
- Chat feels slow, users think app broken

SNS latency: 20ms publish + 50-200ms fanout = 70-220ms P95
Plus WebSocket delivery: +50ms
Total: 120-270ms (unacceptable for chat)

┌─────────────────────────────────────────────────────────────────────────┐
│                    Slack User A (sender)                                 │
│                    Desktop app, WebSocket connected                      │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                             │ 1. Send message (WebSocket frame)
                             │    { "type": "message", "channel": "C123", "text": "Hello!" }
                             ▼
                   ┌──────────────────────┐
                   │   WebSocket Server    │
                   │   (Node.js cluster)   │
                   │                       │
                   │   - 10,000 servers    │
                   │   - 2,000 connections │
                   │     per server        │
                   │   - 20M total users   │
                   └──────────┬────────────┘
                             │
                             │ 2. Authenticate, validate
                             ▼
                   ┌──────────────────────┐
                   │   Message API         │
                   │   (gRPC service)      │
                   │                       │
                   │   1. Assign msg ID    │
                   │   2. Dual-write:      │
                   └──────────┬────────────┘
                             │
            ┌────────────────┴────────────────┐
            │                                 │
            │ 3a. Redis Pub/Sub (fast path)  │ 3b. Kafka (durable path)
            │     - Instant delivery          │     - Persistent storage
            │     - No durability guarantee   │     - Guaranteed durability
            ▼                                 ▼
┌────────────────────────────┐    ┌────────────────────────────┐
│  Redis Pub/Sub Cluster      │    │   Kafka Cluster            │
│  (in-memory messaging)      │    │   (durable message log)    │
│                            │    │                            │
│  PUBLISH channel:C123       │    │   Topic: "slack-messages"  │
│  → Instantly broadcasts to  │    │   Retention: 30 days       │
│     all subscribers         │    │   Replication: 3×          │
│                            │    │                            │
│  Latency: < 1ms             │    │   Latency: 5-10ms         │
│  Durability: None (RAM)     │    │   Durability: Guaranteed   │
└────────────┬───────────────┘    └────────────┬───────────────┘
            │                                 │
            │ 4. Subscribers receive         │ 5. Consumers persist
            │    instantly                   │    to MySQL
            │                                 │
    ┌───────┴─────────┐                     ▼
    │                 │              ┌──────────────────┐
    ▼                 ▼              │  Message DB      │
┌──────────┐    ┌──────────┐        │  (MySQL sharded) │
│WebSocket │    │WebSocket │        │                  │
│Server 1  │    │Server 2  │        │  1. Messages     │
│          │    │          │        │  2. Threads      │
│Subscribed│    │Subscribed│        │  3. Reactions    │
│to C123   │    │to C123   │        └──────────────────┘
│          │    │          │
│User B    │    │User C    │
│connected │    │connected │
└────┬─────┘    └────┬─────┘
     │               │
     │ 6. Push over  │
     │    WebSocket  │
     ▼               ▼
┌──────────┐    ┌──────────┐
│ Slack    │    │ Slack    │
│ User B   │    │ User C   │
│(receiver)│    │(receiver)│
└──────────┘    └──────────┘

Total latency: 1ms API + 1ms Redis + 5ms network = ~7ms P95 (real-time!)

Dual-write pattern:

┌──────────────────────────────────────────────────────────────┐
│                    Message API Service                        │
│                                                               │
│  def send_message(channel_id, text, user_id):                │
│                                                               │
│      # 1. Generate message ID                                │
│      msg_id = generate_snowflake_id()  # Twitter Snowflake   │
│                                                               │
│      # 2. Write to Redis (fast path, best-effort)            │
│      redis.publish(f"channel:{channel_id}", json.dumps({     │
│          "id": msg_id,                                        │
│          "channel": channel_id,                              │
│          "text": text,                                        │
│          "user": user_id,                                     │
│          "timestamp": time.time()                            │
│      }))                                                      │
│      # → Instant delivery to online users (< 1ms)            │
│                                                               │
│      # 3. Write to Kafka (durable path, guaranteed)          │
│      kafka_producer.send("slack-messages", {                 │
│          "id": msg_id,                                        │
│          "channel": channel_id,                              │
│          "text": text,                                        │
│          "user": user_id,                                     │
│          "timestamp": time.time()                            │
│      })                                                       │
│      # → Durable storage for offline users, search, audit    │
│                                                               │
│      return msg_id                                           │
└──────────────────────────────────────────────────────────────┘

import redis
import json
import time

# Redis connection pool (reuse across requests)
redis_pool = redis.ConnectionPool(
    host='redis-cluster.slack.com',
    port=6379,
    max_connections=1000,  # Connection pool for high concurrency
    socket_keepalive=True,
    socket_keepalive_options={
        socket.TCP_KEEPIDLE: 60,
        socket.TCP_KEEPINTVL: 10,
        socket.TCP_KEEPCNT: 3
    }
)

redis_client = redis.Redis(connection_pool=redis_pool)

def send_message(channel_id, text, user_id):
    """
    Send message to Slack channel
    Dual-write: Redis (instant) + Kafka (durable)
    """
    
    # 1. Generate message ID (Twitter Snowflake: timestamp + machine ID + sequence)
    msg_id = generate_snowflake_id()
    timestamp_ms = int(time.time() * 1000)
    
    # 2. Build message payload
    message = {
        'id': msg_id,
        'channel_id': channel_id,
        'user_id': user_id,
        'text': text,
        'timestamp': timestamp_ms,
        'type': 'message',
        'subtype': None,  # Regular message (vs 'bot_message', 'file_share', etc.)
        'thread_ts': None  # Root message (not a thread reply)
    }
    
    # 3. Publish to Redis Pub/Sub (fast path, < 1ms)
    try:
        channel_key = f"channel:{channel_id}"
        redis_client.publish(channel_key, json.dumps(message))
        # PUBLISH returns number of subscribers that received message
        # But we don't wait for response (fire-and-forget for speed)
    except redis.RedisError as e:
        # Redis failure - non-critical (message still goes to Kafka)
        logger.error(f"Redis publish failed: {e}")
        # Metric: increment redis_publish_errors counter
        metrics.increment('redis.publish.errors')
    
    # 4. Publish to Kafka (durable path, 5-10ms)
    try:
        kafka_producer.send(
            topic='slack-messages',
            key=channel_id,  # Partition by channel (all channel messages in order)
            value=message,
            timestamp_ms=timestamp_ms
        )
        # Async send - don't wait for Kafka ack (API returns quickly)
    except Exception as e:
        # Kafka failure - CRITICAL (message lost if not persisted)
        logger.critical(f"Kafka publish failed: {e}")
        metrics.increment('kafka.publish.errors')
        # Store in dead letter queue (S3) for manual recovery
        store_in_dlq(message)
    
    # 5. Return immediately (don't wait for Kafka ack)
    return {
        'ok': True,
        'channel': channel_id,
        'ts': str(timestamp_ms / 1000),  # Timestamp in seconds (Slack API format)
        'message': message
    }

def generate_snowflake_id():
    """
    Twitter Snowflake ID:
    - 41 bits: Timestamp (milliseconds since custom epoch)
    - 10 bits: Machine ID (1024 machines max)
    - 12 bits: Sequence number (4096 per millisecond per machine)
    
    Result: Globally unique, time-sortable, 64-bit integer
    """
    epoch = 1577836800000  # 2020-01-01 00:00:00 UTC (Slack custom epoch)
    timestamp = int(time.time() * 1000) - epoch
    machine_id = get_machine_id()  # 0-1023
    sequence = get_next_sequence()  # 0-4095 (reset every millisecond)
    
    # Bit shifting: timestamp (41 bits) | machine_id (10 bits) | sequence (12 bits)
    snowflake_id = (timestamp << 22) | (machine_id << 12) | sequence
    return snowflake_id

import asyncio
import websockets
import redis
import json

class WebSocketServer:
    
    def __init__(self):
        # Redis Pub/Sub client (separate from regular Redis client)
        self.redis_pubsub = redis.Redis(
            host='redis-cluster.slack.com',
            port=6379,
            decode_responses=True  # Auto-decode bytes to strings
        ).pubsub()
        
        # Track connected clients: {user_id: {websocket, subscribed_channels}}
        self.clients = {}
        
    async def handle_client(self, websocket, path):
        """
        Handle WebSocket connection from Slack client
        """
        user_id = None
        
        try:
            # 1. Authenticate connection
            auth_message = await websocket.recv()
            auth_data = json.loads(auth_message)
            user_id = self.authenticate(auth_data['token'])
            
            if not user_id:
                await websocket.send(json.dumps({'error': 'authentication_failed'}))
                return
            
            # 2. Register client
            self.clients[user_id] = {
                'websocket': websocket,
                'subscribed_channels': set()
            }
            
            # 3. Subscribe to user's channels (typically 75 channels per user)
            user_channels = self.get_user_channels(user_id)  # Query from MySQL
            
            for channel_id in user_channels:
                await self.subscribe_channel(user_id, channel_id)
            
            # 4. Send acknowledgment
            await websocket.send(json.dumps({
                'type': 'hello',
                'user_id': user_id,
                'channels': list(user_channels)
            }))
            
            # 5. Main message loop (handle incoming messages from client)
            async for message in websocket:
                data = json.loads(message)
                
                if data['type'] == 'message':
                    # User sending message → forward to Message API
                    await self.handle_send_message(user_id, data)
                    
                elif data['type'] == 'subscribe':
                    # User joined new channel → subscribe to Redis channel
                    await self.subscribe_channel(user_id, data['channel_id'])
                    
                elif data['type'] == 'unsubscribe':
                    # User left channel → unsubscribe from Redis channel
                    await self.unsubscribe_channel(user_id, data['channel_id'])
                    
        except websockets.exceptions.ConnectionClosed:
            # Client disconnected (closed browser, network failure, etc.)
            logger.info(f"Client {user_id} disconnected")
            
        finally:
            # Cleanup: Unsubscribe from all channels, remove from clients dict
            if user_id and user_id in self.clients:
                for channel_id in self.clients[user_id]['subscribed_channels']:
                    await self.unsubscribe_channel(user_id, channel_id)
                del self.clients[user_id]
    
    async def subscribe_channel(self, user_id, channel_id):
        """
        Subscribe user to Redis Pub/Sub channel
        """
        if user_id not in self.clients:
            return
        
        channel_key = f"channel:{channel_id}"
        
        # Subscribe to Redis channel (blocks until message received)
        self.redis_pubsub.subscribe(channel_key)
        
        # Track subscription
        self.clients[user_id]['subscribed_channels'].add(channel_id)
        
        logger.info(f"User {user_id} subscribed to channel {channel_id}")
    
    async def redis_listener(self):
        """
        Background task: Listen for Redis Pub/Sub messages
        Forward to WebSocket clients
        """
        while True:
            try:
                # Blocking call: Wait for message from Redis
                message = self.redis_pubsub.get_message(ignore_subscribe_messages=True)
                
                if message and message['type'] == 'message':
                    # Parse message
                    channel_key = message['channel']  # "channel:C123"
                    channel_id = channel_key.split(':')[1]
                    data = json.loads(message['data'])
                    
                    # Forward to all clients subscribed to this channel
                    for user_id, client_info in self.clients.items():
                        if channel_id in client_info['subscribed_channels']:
                            websocket = client_info['websocket']
                            
                            try:
                                await websocket.send(json.dumps({
                                    'type': 'message',
                                    'channel': channel_id,
                                    'message': data
                                }))
                            except websockets.exceptions.ConnectionClosed:
                                # Client disconnected, will be cleaned up in handle_client
                                pass
                    
                await asyncio.sleep(0.001)  # 1ms sleep (prevent tight loop)
                
            except Exception as e:
                logger.error(f"Redis listener error: {e}")
                await asyncio.sleep(1)  # Back off on error
    
    async def handle_send_message(self, user_id, data):
        """
        User sending message → forward to Message API
        """
        # Call Message API (gRPC)
        response = await message_api_client.send_message(
            channel_id=data['channel'],
            text=data['text'],
            user_id=user_id
        )
        
        # Message API handles Redis publish (we'll receive via Redis Pub/Sub listener)
    
    def get_user_channels(self, user_id):
        """
        Get user's channels from MySQL
        Average user: 75 channels (public channels + DMs + private channels)
        """
        # Query: SELECT channel_id FROM memberships WHERE user_id = ?
        # Returns: ['C123', 'C456', 'D789', ...]
        return ['C123', 'C456', 'D789']  # Placeholder

# Start WebSocket server
async def main():
    server = WebSocketServer()
    
    # Start Redis listener in background
    asyncio.create_task(server.redis_listener())
    
    # Start WebSocket server on port 443 (WSS)
    async with websockets.serve(server.handle_client, "0.0.0.0", 443):
        await asyncio.Future()  # Run forever

if __name__ == '__main__':
    asyncio.run(main())

Redis Cluster (Pub/Sub):
- ElastiCache: 50× cache.r6g.4xlarge (16 vCPU, 104 GB RAM)
  - Cost: $1.344/hour × 50 × 730 hours = $49,056/month
  - Rationale: High memory for connection buffers, low latency (sub-millisecond)
  - Throughput: 100K msg/sec per instance × 50 = 5M msg/sec capacity
  - Actual load: 350K msg/sec peak (14× headroom)

WebSocket Servers (Node.js):
- EC2 instances: 10,000× c6i.2xlarge (8 vCPU, 16 GB RAM)
  - Cost: $0.34/hour × 10,000 × 730 hours = $2,482,000/month
  - Rationale: 2,000 connections per server, CPU-intensive (WebSocket framing, JSON parsing)
  - Connections: 10,000 servers × 2,000 = 20M concurrent connections

Kafka Cluster (durable storage, not real-time path):
- EC2 instances: 100× i4i.2xlarge (8 vCPU, 64 GB RAM, 1.87 TB NVMe)
  - Cost: $1.037/hour × 100 × 730 hours = $75,701/month
  - Purpose: Message persistence, offline user queue, search indexing

Staff (15 SREs for 24/7 on-call, infrastructure management):
- Salary: $180K/year × 15 = $2.7M/year = $225,000/month
- Benefits (30%): $225,000 × 0.30 = $67,500/month
- Total staff: $292,500/month

Total Redis Pub/Sub approach: $49,056 + $2,482,000 + $75,701 + $292,500 = $2,899,257/month
Annual: $2,899,257 × 12 = $34,791,084/year

Kafka Cluster (scaled for real-time latency):
- EC2 instances: 500× i4i.4xlarge (16 vCPU, 128 GB RAM, 3.75 TB NVMe)
  - Cost: $2.074/hour × 500 × 730 hours = $757,010/month
  - Rationale: 5× larger cluster to achieve <10ms latency at 350K msg/sec peak
  - Kafka latency increases with queue depth; need overprovisioning for low latency

WebSocket Servers (consume from Kafka, push to clients):
- EC2 instances: 10,000× c6i.2xlarge (same as Redis approach)
  - Cost: $2,482,000/month
  - Same connection capacity (20M connections)

Kafka Consumer workers (poll Kafka, push to WebSocket servers):
- EC2 instances: 500× c6i.xlarge (4 vCPU, 8 GB RAM)
  - Cost: $0.17/hour × 500 × 730 hours = $62,050/month
  - Purpose: Bridge Kafka (pull) to WebSocket (push)

Staff (20 SREs - more complex architecture, troubleshooting latency):
- Salary: $180K/year × 20 = $3.6M/year = $300,000/month
- Benefits (30%): $90,000/month
- Total staff: $390,000/month

Total Kafka-only: $757,010 + $2,482,000 + $62,050 + $390,000 = $3,691,060/month
Annual: $3,691,060 × 12 = $44,292,720/year

Cost comparison:
- Redis Pub/Sub + Kafka: $34.79M/year
- Kafka-only: $44.29M/year
- Savings: $9.50M/year (21% cheaper with Redis Pub/Sub)

Latency comparison:
- Redis Pub/Sub: 8-20ms P95 end-to-end
- Kafka-only: 50-100ms P95 (3-5× slower)

User experience:
- Redis Pub/Sub: Instant messaging, feels real-time (< 20ms)
- Kafka-only: Noticeable delay, feels laggy (> 50ms)

Problem:
- User A sends message while User B offline
- Redis publishes message (0 subscribers)
- Message lost (not stored)
- User B reconnects, never sees message

Slack's solution: Kafka backup
1. Message API writes to both Redis (fast) + Kafka (durable)
2. User B reconnects → WebSocket server queries Kafka for missed messages
3. Fetch messages where timestamp > user's last_seen_timestamp
4. Deliver missed messages over WebSocket (catch-up)
5. Subscribe to Redis Pub/Sub for new messages (real-time)

Code:
async def handle_reconnect(user_id, last_seen_timestamp):
    # 1. Fetch missed messages from Kafka
    missed_messages = kafka_consumer.fetch_messages(
        user_id=user_id,
        since=last_seen_timestamp,
        limit=1000  # Max 1,000 missed messages per reconnect
    )
    
    # 2. Send missed messages over WebSocket (catch-up)
    for message in missed_messages:
        await websocket.send(json.dumps(message))
    
    # 3. Subscribe to Redis Pub/Sub for real-time messages
    await subscribe_to_user_channels(user_id)

Problem:
- Redis publishes message to channel:C123
- WebSocket server subscribed but network glitch (packet loss)
- Server never receives message
- User never sees message (silent failure)

Slack's solution: Client-side acknowledgment
1. Message API assigns unique ID to each message
2. Client receives message, sends ACK back to server
3. If no ACK received within 5 seconds, server queries Kafka and resends
4. Client deduplicates based on message ID (idempotency)

Code:
# Server: Track pending ACKs
pending_acks = {}  # {message_id: (timestamp, retries)}

async def send_message_to_client(websocket, message):
    # Send message
    await websocket.send(json.dumps(message))
    
    # Track for ACK
    pending_acks[message['id']] = (time.time(), 0)
    
    # Wait 5 seconds for ACK
    await asyncio.sleep(5)
    
    if message['id'] in pending_acks:
        # No ACK received, resend from Kafka
        kafka_message = kafka_consumer.fetch_by_id(message['id'])
        await websocket.send(json.dumps(kafka_message))
        pending_acks[message['id']] = (time.time(), 1)  # Increment retry count

# Client: Send ACK after receiving message
def on_message_received(message):
    # Display message in UI
    display_message(message)
    
    # Send ACK to server
    websocket.send(json.dumps({
        'type': 'ack',
        'message_id': message['id']
    }))

Problem:
- User A sends: "Hello" (message 1)
- User A sends: "How are you?" (message 2)
- Redis Pub/Sub doesn't guarantee order
- User B receives: "How are you?" then "Hello" (reversed!)

Slack's solution: Snowflake IDs (timestamp-based)
1. Message API assigns Snowflake ID (timestamp in first 41 bits)
2. Client sorts messages by ID (equivalent to sorting by timestamp)
3. If message arrives out of order, client re-sorts before displaying
4. Guarantees chronological display even if network reorders

Code:
# Client: Sort messages by Snowflake ID before displaying
message_buffer = []

def on_message_received(message):
    # Add to buffer
    message_buffer.append(message)
    
    # Sort by Snowflake ID (ascending = oldest first)
    message_buffer.sort(key=lambda m: m['id'])
    
    # Display all messages in order
    for msg in message_buffer:
        display_message(msg)
    
    # Clear buffer
    message_buffer = []

Problem:
- Large channel: 50,000 members (company all-hands)
- All 50,000 subscribed to channel:C123
- Redis broadcasts to 50,000 WebSocket servers (assuming 1 subscriber per server)
- Network bandwidth: 50,000 × 1 KB message = 50 MB per message
- At 100 messages/minute: 5 GB/minute network egress from Redis

Slack's solution: Hierarchical fan-out
1. Redis publishes to 10 "zone" servers (regional clusters)
2. Each zone server forwards to 5,000 WebSocket servers in its region
3. Total fan-out: 10 (Redis → zone) + 5,000 (zone → WebSocket) = 5,010 messages
4. vs 50,000 messages (Redis → WebSocket direct)
5. Reduces Redis network egress by 90%

Architecture:
Redis → Zone 1 (5,000 servers)
      → Zone 2 (5,000 servers)
      → ...
      → Zone 10 (5,000 servers)

CREATE TABLE bookings (
    id VARCHAR(36) PRIMARY KEY,
    listing_id VARCHAR(36),
    guest_id VARCHAR(36),
    host_id VARCHAR(36),
    check_in DATE,
    check_out DATE,
    status VARCHAR(20),  -- 'pending', 'confirmed', 'cancelled', 'completed'
    total_price DECIMAL(10, 2),
    created_at TIMESTAMP,
    updated_at TIMESTAMP
);

-- Example record:
-- id: 'BK123', status: 'confirmed', total_price: 467.00, updated_at: '2024-01-15 10:30:00'

Event Log (immutable, append-only):

Event 1: BookingRequested
{
  "eventId": "EVT-001",
  "eventType": "BookingRequested",
  "bookingId": "BK123",
  "listingId": "LST-456",
  "guestId": "USR-789",
  "checkIn": "2024-02-01",
  "checkOut": "2024-02-03",
  "originalPrice": 500.00,
  "timestamp": "2024-01-15T10:00:00Z"
}

Event 2: CouponApplied
{
  "eventId": "EVT-002",
  "eventType": "CouponApplied",
  "bookingId": "BK123",
  "couponCode": "SPRING20",
  "discount": 33.00,
  "newPrice": 467.00,
  "timestamp": "2024-01-15T10:01:00Z"
}

Event 3: PaymentProcessed
{
  "eventId": "EVT-003",
  "eventType": "PaymentProcessed",
  "bookingId": "BK123",
  "paymentMethod": "CARD-****1234",
  "amount": 467.00,
  "transactionId": "TXN-ABC123",
  "timestamp": "2024-01-15T10:02:00Z"
}

Event 4: BookingConfirmed
{
  "eventId": "EVT-004",
  "eventType": "BookingConfirmed",
  "bookingId": "BK123",
  "confirmationCode": "ABCDEF",
  "timestamp": "2024-01-15T10:03:00Z"
}

Current State (derived by replaying events 1-4):
{
  "bookingId": "BK123",
  "status": "confirmed",
  "confirmationCode": "ABCDEF",
  "totalPrice": 467.00,
  "paymentStatus": "paid"
}

from kafka import KafkaProducer
import json
import uuid
import time

class BookingEventStore:
    """
    Event store for booking events
    All events appended to Kafka topic (immutable log)
    """
    
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=['kafka-1.airbnb.com:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # Wait for all replicas (durability critical)
            enable_idempotence=True  # Exactly-once semantics
        )
        self.topic = 'booking-events'
    
    def request_booking(self, listing_id, guest_id, check_in, check_out, price):
        """
        Step 1: Guest requests booking
        """
        booking_id = f"BK-{uuid.uuid4()}"
        
        event = {
            'eventId': f"EVT-{uuid.uuid4()}",
            'eventType': 'BookingRequested',
            'bookingId': booking_id,
            'listingId': listing_id,
            'guestId': guest_id,
            'checkIn': check_in,
            'checkOut': check_out,
            'originalPrice': price,
            'timestamp': int(time.time() * 1000)
        }
        
        # Append to event log (Kafka)
        # Key = bookingId (all events for same booking go to same partition → ordered)
        self.producer.send(
            topic=self.topic,
            key=booking_id,
            value=event
        )
        
        return booking_id
    
    def apply_coupon(self, booking_id, coupon_code, discount):
        """
        Step 2: Apply coupon discount
        """
        event = {
            'eventId': f"EVT-{uuid.uuid4()}",
            'eventType': 'CouponApplied',
            'bookingId': booking_id,
            'couponCode': coupon_code,
            'discount': discount,
            'timestamp': int(time.time() * 1000)
        }
        
        self.producer.send(
            topic=self.topic,
            key=booking_id,
            value=event
        )
    
    def process_payment(self, booking_id, payment_method, amount, transaction_id):
        """
        Step 3: Process payment
        """
        event = {
            'eventId': f"EVT-{uuid.uuid4()}",
            'eventType': 'PaymentProcessed',
            'bookingId': booking_id,
            'paymentMethod': payment_method,
            'amount': amount,
            'transactionId': transaction_id,
            'timestamp': int(time.time() * 1000)
        }
        
        self.producer.send(
            topic=self.topic,
            key=booking_id,
            value=event
        )
    
    def confirm_booking(self, booking_id, confirmation_code):
        """
        Step 4: Confirm booking
        """
        event = {
            'eventId': f"EVT-{uuid.uuid4()}",
            'eventType': 'BookingConfirmed',
            'bookingId': booking_id,
            'confirmationCode': confirmation_code,
            'timestamp': int(time.time() * 1000)
        }
        
        self.producer.send(
            topic=self.topic,
            key=booking_id,
            value=event
        )
    
    def cancel_booking(self, booking_id, cancelled_by, reason):
        """
        Step 5 (optional): Cancel booking
        """
        event = {
            'eventId': f"EVT-{uuid.uuid4()}",
            'eventType': 'BookingCancelled',
            'bookingId': booking_id,
            'cancelledBy': cancelled_by,  # 'guest' or 'host'
            'reason': reason,
            'timestamp': int(time.time() * 1000)
        }
        
        self.producer.send(
            topic=self.topic,
            key=booking_id,
            value=event
        )
    
    def get_booking_history(self, booking_id):
        """
        Retrieve all events for booking (replay from event log)
        """
        # In production: Query Kafka directly or read from materialized view
        # For demo: Return example events
        return [
            {'eventType': 'BookingRequested', 'timestamp': 1705315200000, ...},
            {'eventType': 'CouponApplied', 'timestamp': 1705315260000, ...},
            {'eventType': 'PaymentProcessed', 'timestamp': 1705315320000, ...},
            {'eventType': 'BookingConfirmed', 'timestamp': 1705315380000, ...}
        ]
    
    def get_current_state(self, booking_id):
        """
        Derive current state by replaying all events
        """
        events = self.get_booking_history(booking_id)
        
        # Initialize state
        state = {
            'bookingId': booking_id,
            'status': 'unknown',
            'price': 0,
            'paymentStatus': 'unpaid'
        }
        
        # Replay events (apply each event to state)
        for event in events:
            if event['eventType'] == 'BookingRequested':
                state['status'] = 'requested'
                state['price'] = event['originalPrice']
                state['listingId'] = event['listingId']
                state['guestId'] = event['guestId']
                
            elif event['eventType'] == 'CouponApplied':
                state['price'] -= event['discount']
                state['couponCode'] = event['couponCode']
                
            elif event['eventType'] == 'PaymentProcessed':
                state['paymentStatus'] = 'paid'
                state['transactionId'] = event['transactionId']
                
            elif event['eventType'] == 'BookingConfirmed':
                state['status'] = 'confirmed'
                state['confirmationCode'] = event['confirmationCode']
                
            elif event['eventType'] == 'BookingCancelled':
                state['status'] = 'cancelled'
                state['cancelledBy'] = event['cancelledBy']
                state['cancellationReason'] = event['reason']
        
        return state

Booking with 1M events (extreme case: many cancellations, repricing):
- Replay time: 1M events × 0.1ms = 100 seconds (unacceptable)
- Query API: GET /bookings/BK123 must return < 100ms

class BookingMaterializedView:
    """
    Snapshot of booking state (updated incrementally as events arrive)
    Stored in fast database (Redis, DynamoDB)
    """
    
    def __init__(self):
        self.redis = redis.Redis(host='redis-cluster.airbnb.com', port=6379)
    
    def update_from_event(self, event):
        """
        Update materialized view when new event arrives
        Consumer subscribes to Kafka, calls this method for each event
        """
        booking_id = event['bookingId']
        
        # Get current snapshot from Redis
        snapshot_json = self.redis.get(f"booking:{booking_id}")
        if snapshot_json:
            snapshot = json.loads(snapshot_json)
        else:
            # Initialize snapshot
            snapshot = {
                'bookingId': booking_id,
                'status': 'unknown',
                'price': 0,
                'eventCount': 0
            }
        
        # Apply event to snapshot (same logic as get_current_state)
        if event['eventType'] == 'BookingRequested':
            snapshot['status'] = 'requested'
            snapshot['price'] = event['originalPrice']
            snapshot['listingId'] = event['listingId']
            
        elif event['eventType'] == 'CouponApplied':
            snapshot['price'] -= event['discount']
            
        elif event['eventType'] == 'PaymentProcessed':
            snapshot['paymentStatus'] = 'paid'
            
        elif event['eventType'] == 'BookingConfirmed':
            snapshot['status'] = 'confirmed'
            snapshot['confirmationCode'] = event['confirmationCode']
            
        elif event['eventType'] == 'BookingCancelled':
            snapshot['status'] = 'cancelled'
        
        # Increment event counter
        snapshot['eventCount'] += 1
        snapshot['lastUpdated'] = event['timestamp']
        
        # Store updated snapshot in Redis
        self.redis.set(f"booking:{booking_id}", json.dumps(snapshot))
    
    def get_snapshot(self, booking_id):
        """
        Fast query: Read snapshot from Redis (< 1ms)
        """
        snapshot_json = self.redis.get(f"booking:{booking_id}")
        if snapshot_json:
            return json.loads(snapshot_json)
        
        # Fallback: Replay events if snapshot not found
        # (rare case: Redis cache miss, new booking)
        return BookingEventStore().get_current_state(booking_id)

Query: GET /bookings/BK123

Without materialized view (replay 1,000 events):
- Kafka fetch: 50ms (read 1,000 events from partition)
- Replay: 1,000 events × 0.1ms = 100ms
- Total: 150ms P95

With materialized view (Redis snapshot):
- Redis GET: 1ms
- Total: 1ms P95 (150× faster)

Trade-off:
- Storage: Redis stores 150M snapshots × 1 KB = 150 GB (~$12,000/month ElastiCache)
- Consistency: Eventually consistent (snapshot updated async via Kafka consumer, 100-500ms lag)

Kafka event log:
- Events: 150M bookings/year × 8 events average = 1.2B events/year
- Storage: 1.2B events × 2 KB = 2.4 TB/year
- Kafka retention: 1 year (audit compliance)
- Cost: 2.4 TB × $0.08/GB-month = $196/month = $2,352/year

Redis materialized views:
- Snapshots: 150M bookings × 1 KB = 150 GB
- ElastiCache r6g.4xlarge: 104 GB RAM × 2 instances = 208 GB capacity
- Cost: $1.344/hour × 2 × 730 hours = $1,962/month = $23,544/year

Alternative (traditional database without event sourcing):
- MySQL bookings table: 150M rows × 500 bytes = 75 GB
- RDS db.r6g.4xlarge: 128 GB RAM
- Cost: $1.632/hour × 730 hours = $1,191/month = $14,292/year
- Plus S3 backups: 75 GB × 12 months × $0.023/GB = $21/year
- Total: $14,313/year

Event sourcing cost: $2,352 + $23,544 = $25,896/year
Traditional cost: $14,313/year
Additional cost: $11,583/year (81% more expensive)

Value gained:
- Compliance: Avoid $5M+ GDPR fines (complete audit trail)
- Debugging: Saved 500 engineering hours/year ($90K at $180K salary)
- Refunds: Identified $127,500 pricing bug (would've gone unnoticed)
- BI insights: 15% conversion improvement via funnel optimization ($150M revenue impact at $1B GMV)

ROI: ($5M + $90K + $127K + $150M) ÷ $11.6K = 13,362× return

Writes (create payment):
- Frequency: 75M/day = 868 requests/second average
- Latency requirement: < 500ms P95 (customer waiting at checkout)
- Data: Simple (payment amount, currency, customer, payment method)
- Consistency: Strong (payment must succeed exactly once, no duplicates)

Reads (retrieve payment, list payments, search):
- Frequency: 500M/day = 5,787 requests/second (6.7× more reads than writes)
- Latency requirement: < 100ms P95 (dashboard queries, webhook delivery confirmation)
- Data: Complex (payment with related charges, refunds, disputes, balance transactions)
- Consistency: Eventual (okay if dashboard shows payment 1-2 seconds after creation)
- Queries: Diverse (by customer, by date range, by status, by amount range, full-text search)

Problem with single database:
- Write-optimized (normalized schema): Fast writes, slow complex queries
- Read-optimized (denormalized views): Fast queries, slow writes (multiple table updates)
- Can't optimize for both simultaneously

Architecture:

┌─────────────────────────────────────────────────────────────────┐
│                        Stripe API                                │
│                                                                  │
│  POST /v1/charges (write command)                               │
│  ↓                                                               │
│  Command Model (write-optimized)                                │
│  - Normalized schema (payments, customers, methods tables)      │
│  - PostgreSQL (ACID transactions, strong consistency)           │
│  - Write latency: 50ms P95                                      │
│                                                                  │
│  GET /v1/charges (read query)                                   │
│  ↓                                                               │
│  Query Model (read-optimized)                                   │
│  - Denormalized views (charges with embedded customer data)     │
│  - Elasticsearch (full-text search, aggregations)               │
│  - Read latency: 10ms P95                                       │
│                                                                  │
│  Event Bus: Kafka                                               │
│  - Command model publishes events after write                   │
│  - Query model subscribes, updates denormalized views           │
│  - Eventual consistency: 100-500ms lag                          │
└─────────────────────────────────────────────────────────────────┘

import psycopg2
import uuid
import time
from kafka import KafkaProducer

class ChargeCommandModel:
    """
    Write model: Optimized for fast writes, strong consistency
    Normalized schema in PostgreSQL
    """
    
    def __init__(self):
        self.db = psycopg2.connect("host=postgres.stripe.com dbname=payments")
        self.kafka_producer = KafkaProducer(
            bootstrap_servers=['kafka-1.stripe.com:9092'],
            acks='all',
            enable_idempotence=True
        )
    
    def create_charge(self, amount, currency, customer_id, payment_method_id):
        """
        Write path: Create charge (command)
        """
        cursor = self.db.cursor()
        
        try:
            # Generate charge ID
            charge_id = f"ch_{uuid.uuid4().hex}"
            timestamp = int(time.time())
            
            # 1. Insert into charges table (write-optimized, normalized)
            cursor.execute("""
                INSERT INTO charges (
                    id, amount, currency, customer_id, payment_method_id, 
                    status, created_at
                ) VALUES (%s, %s, %s, %s, %s, %s, %s)
            """, (charge_id, amount, currency, customer_id, payment_method_id, 
                  'pending', timestamp))
            
            # 2. Call payment processor (Visa, Mastercard, etc.)
            payment_result = self.process_with_card_network(
                amount, currency, payment_method_id
            )
            
            if payment_result['status'] == 'succeeded':
                # Update charge status
                cursor.execute("""
                    UPDATE charges 
                    SET status = 'succeeded', transaction_id = %s
                    WHERE id = %s
                """, (payment_result['transaction_id'], charge_id))
            else:
                # Payment failed
                cursor.execute("""
                    UPDATE charges 
                    SET status = 'failed', failure_code = %s, failure_message = %s
                    WHERE id = %s
                """, (payment_result['error_code'], payment_result['error_message'], charge_id))
            
            # 3. Commit transaction (ACID guarantees)
            self.db.commit()
            
            # 4. Publish event to Kafka (after commit succeeds)
            event = {
                'eventId': f"evt_{uuid.uuid4().hex}",
                'eventType': 'ChargeCreated',
                'chargeId': charge_id,
                'amount': amount,
                'currency': currency,
                'customerId': customer_id,
                'status': payment_result['status'],
                'timestamp': timestamp
            }
            
            self.kafka_producer.send(
                topic='payment-events',
                key=charge_id,
                value=json.dumps(event).encode('utf-8')
            )
            
            return {
                'id': charge_id,
                'status': payment_result['status'],
                'amount': amount,
                'currency': currency
            }
            
        except Exception as e:
            # Rollback on error
            self.db.rollback()
            raise e
        finally:
            cursor.close()
    
    def process_with_card_network(self, amount, currency, payment_method_id):
        """
        Call Visa/Mastercard API to process payment
        (Simplified - actual implementation handles 3DS, fraud checks, etc.)
        """
        # Simulate payment processing
        return {
            'status': 'succeeded',
            'transaction_id': f"txn_{uuid.uuid4().hex}"
        }

-- Charges table (core entity)
CREATE TABLE charges (
    id VARCHAR(64) PRIMARY KEY,
    amount BIGINT NOT NULL,  -- Amount in cents (e.g., 5000 = $50.00)
    currency VARCHAR(3) NOT NULL,  -- ISO 4217 (USD, EUR, GBP)
    customer_id VARCHAR(64) REFERENCES customers(id),
    payment_method_id VARCHAR(64) REFERENCES payment_methods(id),
    status VARCHAR(20) NOT NULL,  -- pending, succeeded, failed
    transaction_id VARCHAR(64),  -- Card network transaction ID
    failure_code VARCHAR(64),
    failure_message TEXT,
    created_at INTEGER NOT NULL,  -- Unix timestamp
    INDEX idx_customer (customer_id),
    INDEX idx_created_at (created_at)
);

-- Customers table (separate, normalized)
CREATE TABLE customers (
    id VARCHAR(64) PRIMARY KEY,
    email VARCHAR(255) NOT NULL,
    name VARCHAR(255),
    created_at INTEGER NOT NULL
);

-- Payment methods table (separate, normalized)
CREATE TABLE payment_methods (
    id VARCHAR(64) PRIMARY KEY,
    customer_id VARCHAR(64) REFERENCES customers(id),
    type VARCHAR(20),  -- card, bank_account, etc.
    card_last4 VARCHAR(4),
    card_brand VARCHAR(20),  -- visa, mastercard, amex
    created_at INTEGER NOT NULL
);

from elasticsearch import Elasticsearch

class ChargeQueryModel:
    """
    Read model: Optimized for fast queries, eventual consistency
    Denormalized documents in Elasticsearch
    """
    
    def __init__(self):
        self.es = Elasticsearch(['https://elasticsearch.stripe.com:9200'])
    
    def get_charge(self, charge_id):
        """
        Read path: Retrieve charge (query)
        Returns denormalized document (charge + customer + payment method embedded)
        """
        result = self.es.get(index='charges', id=charge_id)
        return result['_source']
    
    def list_charges(self, customer_id=None, status=None, limit=10):
        """
        List charges with filters
        Elasticsearch enables fast queries on any field
        """
        query = {'bool': {'must': []}}
        
        if customer_id:
            query['bool']['must'].append({'term': {'customerId': customer_id}})
        if status:
            query['bool']['must'].append({'term': {'status': status}})
        
        result = self.es.search(
            index='charges',
            body={'query': query, 'size': limit, 'sort': [{'createdAt': 'desc'}]}
        )
        
        return [hit['_source'] for hit in result['hits']['hits']]
    
    def search_charges(self, query_string):
        """
        Full-text search across all fields
        Example: "john@example.com" finds all charges for that customer
        """
        result = self.es.search(
            index='charges',
            body={
                'query': {
                    'multi_match': {
                        'query': query_string,
                        'fields': ['customerId', 'customerEmail', 'customerName', 'chargeId']
                    }
                },
                'size': 100
            }
        )
        
        return [hit['_source'] for hit in result['hits']['hits']]

{
  "chargeId": "ch_abc123",
  "amount": 5000,
  "currency": "usd",
  "status": "succeeded",
  "transactionId": "txn_xyz789",
  "createdAt": 1705315200,
  
  // Customer data embedded (denormalized)
  "customerId": "cus_def456",
  "customerEmail": "john@example.com",
  "customerName": "John Doe",
  
  // Payment method embedded (denormalized)
  "paymentMethodId": "pm_ghi789",
  "paymentMethodType": "card",
  "cardLast4": "4242",
  "cardBrand": "visa"
}

from kafka import KafkaConsumer
from elasticsearch import Elasticsearch
import json

class ChargeProjector:
    """
    Projector: Subscribes to Kafka events, updates Elasticsearch
    Synchronizes read model (Elasticsearch) with write model (PostgreSQL)
    """
    
    def __init__(self):
        self.kafka_consumer = KafkaConsumer(
            'payment-events',
            bootstrap_servers=['kafka-1.stripe.com:9092'],
            group_id='elasticsearch-projector',
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )
        self.es = Elasticsearch(['https://elasticsearch.stripe.com:9200'])
        self.pg = psycopg2.connect("host=postgres.stripe.com dbname=payments")
    
    def run(self):
        """
        Main loop: Consume events, update Elasticsearch
        """
        for message in self.kafka_consumer:
            event = message.value
            
            if event['eventType'] == 'ChargeCreated':
                self.handle_charge_created(event)
            elif event['eventType'] == 'ChargeUpdated':
                self.handle_charge_updated(event)
            elif event['eventType'] == 'CustomerUpdated':
                self.handle_customer_updated(event)
    
    def handle_charge_created(self, event):
        """
        New charge created: Fetch from PostgreSQL, index in Elasticsearch
        """
        charge_id = event['chargeId']
        
        # Fetch full charge data from PostgreSQL (with JOINs)
        cursor = self.pg.cursor()
        cursor.execute("""
            SELECT 
                c.id, c.amount, c.currency, c.status, c.transaction_id, c.created_at,
                cust.id as customer_id, cust.email, cust.name,
                pm.id as payment_method_id, pm.type, pm.card_last4, pm.card_brand
            FROM charges c
            JOIN customers cust ON c.customer_id = cust.id
            JOIN payment_methods pm ON c.payment_method_id = pm.id
            WHERE c.id = %s
        """, (charge_id,))
        
        row = cursor.fetchone()
        
        # Build denormalized document
        doc = {
            'chargeId': row[0],
            'amount': row[1],
            'currency': row[2],
            'status': row[3],
            'transactionId': row[4],
            'createdAt': row[5],
            'customerId': row[6],
            'customerEmail': row[7],
            'customerName': row[8],
            'paymentMethodId': row[9],
            'paymentMethodType': row[10],
            'cardLast4': row[11],
            'cardBrand': row[12]
        }
        
        # Index in Elasticsearch
        self.es.index(index='charges', id=charge_id, document=doc)
        
        print(f" Indexed charge {charge_id} in Elasticsearch")
    
    def handle_customer_updated(self, event):
        """
        Customer email changed: Update all charges for that customer
        """
        customer_id = event['customerId']
        new_email = event['newEmail']
        
        # Find all charges for this customer in Elasticsearch
        result = self.es.search(
            index='charges',
            body={'query': {'term': {'customerId': customer_id}}, 'size': 10000}
        )
        
        # Update each charge document
        for hit in result['hits']['hits']:
            charge_id = hit['_id']
            self.es.update(
                index='charges',
                id=charge_id,
                body={'doc': {'customerEmail': new_email}}
            )
        
        print(f" Updated {len(result['hits']['hits'])} charges for customer {customer_id}")

Timeline: Create charge

T+0ms:   API receives POST /v1/charges
         ↓
T+50ms:  Write to PostgreSQL (command model)
         - INSERT INTO charges
         - UPDATE charges SET status='succeeded'
         - COMMIT transaction
         ↓
T+60ms:  Publish event to Kafka
         - Topic: payment-events
         - Event: ChargeCreated
         ↓
T+100ms: Kafka consumer receives event
         - ChargeProjector.handle_charge_created()
         ↓
T+150ms: Query PostgreSQL (fetch denormalized data)
         - SELECT charges JOIN customers JOIN payment_methods
         ↓
T+200ms: Index in Elasticsearch (query model)
         - es.index(index='charges', document={...})
         ↓
T+210ms: Elasticsearch query now returns new charge
         - GET /v1/charges/ch_abc123 (fast, < 10ms)

Total lag: 210ms (command model → query model)

CQRS implementation:
- PostgreSQL (write model): 10× db.r6g.4xlarge ($1.632/hour × 10 × 730) = $11,914/month
- Elasticsearch (read model): 50× i3.xlarge.elasticsearch ($0.28/hour × 50 × 730) = $10,220/month
- Kafka (event bus): 20× i4i.xlarge ($0.519/hour × 20 × 730) = $7,577/month
- Total: $29,711/month = $356,532/year

Traditional single database:
- PostgreSQL (read + write): 100× db.r6g.4xlarge ($1.632/hour × 100 × 730) = $119,136/month
  - Requires 10× more capacity due to:
    - Complex JOIN queries on read path (slow, CPU-intensive)
    - Full-text search (slow, table scans)
    - Aggregations (slow, requires scanning millions of rows)
- Total: $119,136/month = $1,429,632/year

Savings: $1,429,632 - $356,532 = $1,073,100/year (75% cost reduction)

Plus operational benefits:
- Elasticsearch downtime: Writes unaffected (PostgreSQL continues accepting charges)
- PostgreSQL downtime: Reads unaffected (Elasticsearch serves queries from cached data)
- Independent scaling: Add read capacity without impacting write latency

Throughput: 10,000 events/sec × 1 KB = 10 MB/sec

Shards required: 10 MB/sec ÷ 1 MB/sec = 10 shards

Cost:
- Shard-hours: 10 shards × 730 hours = 7,300 shard-hours × $0.015 = $109.50/month
- PUT requests: 10,000 events/sec × 2.6M seconds/month = 26B requests
  - Cost: 26,000 million × $0.014 / 1M = $364/month
- Total: $109.50 + $364 = $473.50/month

Data volume: 10 MB/sec × 2.6M seconds = 26 TB/month
Cost per GB: $473.50 ÷ 26,000 GB = $0.018/GB

Throughput: 10 MB/sec (same as Kinesis example)

Throughput units: 10 MB/sec ÷ 1 MB/sec = 10 TUs

Cost:
- TU-hours: 10 TUs × 730 hours = 7,300 TU-hours × $0.0279 = $203.67/month
- Events: 26B events/month × $0.028 / 1M = $728/month
- Total: $203.67 + $728 = $931.67/month

Cost per GB: $931.67 ÷ 26,000 GB = $0.036/GB (2× more expensive than Kinesis)

Throughput: 10 MB/sec = 26 TB/month (same)

Cost:
- Ingress: 26,000 GB × $0.04 = $1,040/month
- Egress: 26,000 GB × $0.08 = $2,080/month (1 consumer)
- Total: $1,040 + $2,080 = $3,120/month

Cost per GB: $3,120 ÷ 26,000 GB = $0.12/GB (6.7× more expensive than Kinesis)

With 5 consumers (fanout):
- Egress: 26,000 GB × 5 × $0.08 = $10,400/month
- Total: $1,040 + $10,400 = $11,440/month (24× more expensive)

AWS Kinesis:
- Shards: 1 (0.01 MB/sec well under 1 MB/sec limit)
- Cost: 1 shard × 730 hours × $0.015 = $10.95/month
- PUT requests: 30 GB/month ÷ 1 KB = 30M requests × $0.014 / 1M = $0.42/month
- Total: $11.37/month ($0.38/GB)

Azure Event Hubs:
- TUs: 1 (minimum)
- Cost: 1 TU × 730 hours × $0.0279 = $20.37/month
- Events: 30M events × $0.028 / 1M = $0.84/month
- Total: $21.21/month ($0.71/GB)

GCP Pub/Sub:
- Ingress: 30 GB × $0.04 = $1.20/month
- Egress: 30 GB × $0.08 = $2.40/month
- Total: $3.60/month ($0.12/GB)

Winner: GCP Pub/Sub (3.2× cheaper than Kinesis, 5.9× cheaper than Event Hubs)

AWS Kinesis:
- Shards: 2 (1.2 MB/sec peak needs 2 shards)
- Cost: 2 shards × 730 hours × $0.015 = $21.90/month
- PUT requests: 3,000 GB/month ÷ 1 KB = 3B requests × $0.014 / 1M = $42/month
- Total: $63.90/month ($0.021/GB)
- Note: Multiple consumers read from same shards (no additional egress cost)

Azure Event Hubs:
- TUs: 2
- Cost: 2 TUs × 730 hours × $0.0279 = $40.74/month
- Events: 3B events × $0.028 / 1M = $84/month
- Total: $124.74/month ($0.042/GB)

GCP Pub/Sub:
- Ingress: 3,000 GB × $0.04 = $120/month
- Egress: 3,000 GB × 3 consumers × $0.08 = $720/month
- Total: $840/month ($0.28/GB)

Winner: AWS Kinesis (13× cheaper than Pub/Sub, 2× cheaper than Event Hubs)

AWS Kinesis:
- Shards: 12 (11.6 MB/sec peak needs 12 shards)
- Cost: 12 shards × 730 hours × $0.015 = $131.40/month
- PUT requests: 30,000 GB/month ÷ 1 KB = 30B requests × $0.014 / 1M = $420/month
- Total: $551.40/month ($0.018/GB)

Azure Event Hubs:
- TUs: 12
- Cost: 12 TUs × 730 hours × $0.0279 = $244.44/month
- Events: 30B events × $0.028 / 1M = $840/month
- Total: $1,084.44/month ($0.036/GB)

GCP Pub/Sub:
- Ingress: 30,000 GB × $0.04 = $1,200/month
- Egress: 30,000 GB × 10 consumers × $0.08 = $24,000/month
- Total: $25,200/month ($0.84/GB)

Winner: AWS Kinesis (45.7× cheaper than Pub/Sub, 2× cheaper than Event Hubs)