-- Shard 1: user_id % 4 = 0
-- Shard 2: user_id % 4 = 1
-- Shard 3: user_id % 4 = 2
-- Shard 4: user_id % 4 = 3

CREATE TABLE transactions_shard_1 (
    transaction_id UUID PRIMARY KEY,
    user_id BIGINT,
    amount DECIMAL(15,2),
    created_at TIMESTAMP,
    CONSTRAINT check_shard CHECK (user_id % 4 = 0)
);

-- Monthly partitions for transaction history
CREATE TABLE transactions_2024_01 PARTITION OF transactions
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

CREATE TABLE transactions_2024_02 PARTITION OF transactions
FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');

# Database cluster configuration
database_cluster:
  master:
    host: payment-db-master.internal
    port: 5432
    max_connections: 200
  
  read_replicas:
    - host: payment-db-replica-1.internal
      port: 5432
      weight: 50
    - host: payment-db-replica-2.internal
      port: 5432
      weight: 50
  
  connection_pool:
    min_size: 10
    max_size: 100
    timeout: 30s

# Payment Processing Service
class PaymentProcessor:
    def __init__(self):
        self.fraud_detector = FraudDetectionService()
        self.vault_service = TokenVaultService()
        self.gateway_router = PaymentGatewayRouter()
    
    async def process_payment(self, payment_request):
        # Fraud check (async)
        fraud_score = await self.fraud_detector.analyze(payment_request)
        if fraud_score > 0.8:
            return PaymentResult(status="DECLINED", reason="FRAUD_DETECTED")
        
        # Route to appropriate gateway
        gateway = self.gateway_router.select_gateway(payment_request)
        result = await gateway.charge(payment_request)
        
        return result

# Separate service for wallet operations
class WalletService:
    async def get_balance(self, user_id):
        return await self.db.fetch_balance(user_id)
    
    async def update_balance(self, user_id, amount, transaction_id):
        async with self.db.transaction():
            current_balance = await self.get_balance(user_id)
            if current_balance + amount < 0:
                raise InsufficientFundsError()
            
            await self.db.update_balance(user_id, amount, transaction_id)

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: payment-processor-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: payment-processor
  minReplicas: 5
  maxReplicas: 100
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: payment_queue_length
      target:
        type: AverageValue
        averageValue: "10"

import redis.sentinel
import json

class PaymentCache:
    def __init__(self):
        # Redis Sentinel for high availability
        self.sentinel = redis.sentinel.Sentinel([
            ('redis-sentinel-1', 26379),
            ('redis-sentinel-2', 26379),
            ('redis-sentinel-3', 26379)
        ])
        self.master = self.sentinel.master_for('payment-cache')
        self.slave = self.sentinel.slave_for('payment-cache')
    
    async def cache_payment_method(self, user_id, payment_method):
        key = f"payment_method:{user_id}"
        # Cache for 1 hour
        await self.master.setex(key, 3600, json.dumps(payment_method))
    
    async def get_cached_payment_method(self, user_id):
        key = f"payment_method:{user_id}"
        cached = await self.slave.get(key)
        return json.loads(cached) if cached else None
    
    async def cache_fraud_score(self, transaction_id, score):
        key = f"fraud_score:{transaction_id}"
        # Cache fraud scores for 24 hours
        await self.master.setex(key, 86400, str(score))

# CDN configuration for payment forms and static assets
cdn_config:
  origins:
    - domain: payment-api.company.com
      path_patterns:
        - "/static/*"
        - "/js/*"
        - "/css/*"
  
  cache_behaviors:
    - path_pattern: "/static/*"
      ttl: 86400  # 24 hours
      compress: true
    
    - path_pattern: "/api/*"
      ttl: 0  # No caching for API calls
      forward_headers: ["Authorization", "X-Request-ID"]

# Kafka cluster for payment events
kafka_cluster:
  brokers: 9
  replication_factor: 3
  partitions_per_topic: 12
  
  topics:
    payment_events:
      partitions: 24
      replication_factor: 3
      retention_ms: 604800000  # 7 days
    
    fraud_alerts:
      partitions: 12
      replication_factor: 3
      retention_ms: 2592000000  # 30 days
    
    webhook_events:
      partitions: 36
      replication_factor: 3
      retention_ms: 259200000  # 3 days

from kafka import KafkaProducer
import json

class PaymentEventProducer:
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # Wait for all replicas
            retries=3,
            batch_size=16384,
            linger_ms=10
        )
    
    async def publish_payment_event(self, event):
        # Partition by user_id for ordering
        partition_key = str(event['user_id']).encode('utf-8')
        
        self.producer.send(
            'payment_events',
            value=event,
            key=partition_key
        )

# Global regions configuration
regions:
  us_east:
    primary: true
    services: [payment_processor, fraud_detection, vault]
    database: primary_master
    
  us_west:
    primary: false
    services: [payment_processor, fraud_detection]
    database: read_replica
    
  eu_west:
    primary: false
    services: [payment_processor, fraud_detection, vault]
    database: regional_master
    compliance: [GDPR, PSD2]
    
  asia_pacific:
    primary: false
    services: [payment_processor]
    database: read_replica

class CrossRegionReplication:
    def __init__(self):
        self.regions = {
            'us-east-1': {'primary': True, 'endpoint': 'payment-db-us-east.rds.amazonaws.com'},
            'eu-west-1': {'primary': False, 'endpoint': 'payment-db-eu-west.rds.amazonaws.com'},
            'ap-southeast-1': {'primary': False, 'endpoint': 'payment-db-ap-se.rds.amazonaws.com'}
        }
    
    async def replicate_transaction(self, transaction):
        # Write to primary region first
        primary_result = await self.write_to_primary(transaction)
        
        # Async replication to other regions
        replication_tasks = []
        for region, config in self.regions.items():
            if not config['primary']:
                task = self.replicate_to_region(transaction, region)
                replication_tasks.append(task)
        
        # Don't wait for replication to complete
        asyncio.create_task(asyncio.gather(*replication_tasks))
        
        return primary_result

load_balancer:
  type: application
  scheme: internet-facing
  
  listeners:
    - port: 443
      protocol: HTTPS
      ssl_certificate: payment-service-cert
      
      rules:
        - path: "/api/v1/payments/*"
          target_group: payment-processor-tg
          
        - path: "/api/v1/webhooks/*"
          target_group: webhook-handler-tg
  
  target_groups:
    payment_processor_tg:
      protocol: HTTP
      port: 8080
      health_check:
        path: "/health"
        interval: 30
        timeout: 5
        healthy_threshold: 2
        unhealthy_threshold: 3

import asyncio
from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
    
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpenError("Circuit breaker is open")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e
    
    def _should_attempt_reset(self):
        return (datetime.now() - self.last_failure_time) > timedelta(seconds=self.timeout)
    
    def _on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage in payment gateway
class PaymentGateway:
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(failure_threshold=3, timeout=30)
    
    async def process_payment(self, payment_data):
        return await self.circuit_breaker.call(self._make_payment_request, payment_data)

import asyncpg
import asyncio

class DatabasePool:
    def __init__(self):
        self.pool = None
    
    async def initialize(self):
        self.pool = await asyncpg.create_pool(
            host='payment-db.internal',
            port=5432,
            user='payment_service',
            password='secure_password',
            database='payments',
            min_size=20,
            max_size=100,
            command_timeout=30,
            server_settings={
                'application_name': 'payment_service',
                'jit': 'off'  # Disable JIT for consistent performance
            }
        )
    
    async def execute_transaction(self, query, *args):
        async with self.pool.acquire() as connection:
            async with connection.transaction():
                return await connection.fetch(query, *args)

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncPaymentProcessor:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=50)
        self.semaphore = asyncio.Semaphore(100)  # Limit concurrent operations
    
    async def process_batch_payments(self, payments):
        async with self.semaphore:
            tasks = []
            for payment in payments:
                task = self.process_single_payment(payment)
                tasks.append(task)
            
            # Process up to 10 payments concurrently
            results = []
            for i in range(0, len(tasks), 10):
                batch = tasks[i:i+10]
                batch_results = await asyncio.gather(*batch, return_exceptions=True)
                results.extend(batch_results)
            
            return results
    
    async def process_single_payment(self, payment):
        # CPU-intensive fraud detection in thread pool
        fraud_score = await asyncio.get_event_loop().run_in_executor(
            self.executor, self.calculate_fraud_score, payment
        )
        
        if fraud_score > 0.8:
            return {"status": "declined", "reason": "fraud"}
        
        # I/O bound payment gateway call
        return await self.charge_payment_gateway(payment)

from prometheus_client import Counter, Histogram, Gauge

# Payment processing metrics
payment_requests_total = Counter('payment_requests_total', 'Total payment requests', ['status', 'gateway'])
payment_processing_duration = Histogram('payment_processing_seconds', 'Payment processing duration')
active_connections = Gauge('database_connections_active', 'Active database connections')
fraud_detection_duration = Histogram('fraud_detection_seconds', 'Fraud detection duration')

class MetricsCollector:
    @staticmethod
    def record_payment_request(status, gateway):
        payment_requests_total.labels(status=status, gateway=gateway).inc()
    
    @staticmethod
    def record_processing_time(duration):
        payment_processing_duration.observe(duration)
    
    @staticmethod
    def update_active_connections(count):
        active_connections.set(count)

from fastapi import FastAPI, HTTPException
import asyncio

app = FastAPI()

class HealthChecker:
    def __init__(self):
        self.db_pool = DatabasePool()
        self.redis_client = RedisClient()
        self.kafka_producer = KafkaProducer()
    
    async def check_database(self):
        try:
            await self.db_pool.execute("SELECT 1")
            return {"status": "healthy", "response_time": "< 100ms"}
        except Exception as e:
            return {"status": "unhealthy", "error": str(e)}
    
    async def check_redis(self):
        try:
            await self.redis_client.ping()
            return {"status": "healthy"}
        except Exception as e:
            return {"status": "unhealthy", "error": str(e)}
    
    async def check_kafka(self):
        try:
            # Check if we can produce a test message
            await self.kafka_producer.send('health_check', {'test': True})
            return {"status": "healthy"}
        except Exception as e:
            return {"status": "unhealthy", "error": str(e)}

@app.get("/health")
async def health_check():
    checker = HealthChecker()
    
    checks = await asyncio.gather(
        checker.check_database(),
        checker.check_redis(),
        checker.check_kafka(),
        return_exceptions=True
    )
    
    db_health, redis_health, kafka_health = checks
    
    overall_status = "healthy"
    if any(check.get("status") == "unhealthy" for check in checks):
        overall_status = "unhealthy"
    
    return {
        "status": overall_status,
        "timestamp": datetime.utcnow().isoformat(),
        "checks": {
            "database": db_health,
            "redis": redis_health,
            "kafka": kafka_health
        }
    }

class CapacityPlanner:
    def __init__(self):
        self.base_tps = 1000  # Transactions per second
        self.peak_multiplier = 10  # Black Friday multiplier
        self.safety_margin = 1.5
    
    def calculate_required_capacity(self):
        peak_tps = self.base_tps * self.peak_multiplier
        required_tps = peak_tps * self.safety_margin
        
        # Each instance can handle 100 TPS
        required_instances = math.ceil(required_tps / 100)
        
        # Database connections (10 per instance)
        required_db_connections = required_instances * 10
        
        # Redis memory (1GB per 1000 TPS)
        required_redis_memory = math.ceil(required_tps / 1000)
        
        return {
            "required_instances": required_instances,
            "required_db_connections": required_db_connections,
            "required_redis_memory_gb": required_redis_memory,
            "estimated_cost_per_hour": required_instances * 0.50  # $0.50 per instance
        }

# CloudWatch scheduled scaling for known traffic patterns
scheduled_scaling:
  - name: "morning_rush"
    schedule: "0 8 * * MON-FRI"  # 8 AM weekdays
    min_capacity: 20
    max_capacity: 50
    target_capacity: 30
  
  - name: "evening_peak"
    schedule: "0 18 * * *"  # 6 PM daily
    min_capacity: 25
    max_capacity: 60
    target_capacity: 40
  
  - name: "weekend_low"
    schedule: "0 2 * * SAT,SUN"  # 2 AM weekends
    min_capacity: 10
    max_capacity: 25
    target_capacity: 15

# Reactive scaling based on metrics
reactive_scaling:
  scale_up:
    - metric: "CPUUtilization"
      threshold: 70
      duration: 300  # 5 minutes
      action: "increase_capacity_by_20_percent"
    
    - metric: "payment_queue_length"
      threshold: 100
      duration: 60  # 1 minute
      action: "increase_capacity_by_50_percent"
  
  scale_down:
    - metric: "CPUUtilization"
      threshold: 30
      duration: 900  # 15 minutes
      action: "decrease_capacity_by_10_percent"

class CostOptimizer:
    def __init__(self):
        self.instance_types = {
            't3.medium': {'cpu': 2, 'memory': 4, 'cost_per_hour': 0.0416},
            't3.large': {'cpu': 2, 'memory': 8, 'cost_per_hour': 0.0832},
            'c5.large': {'cpu': 2, 'memory': 4, 'cost_per_hour': 0.085},
            'c5.xlarge': {'cpu': 4, 'memory': 8, 'cost_per_hour': 0.17}
        }
    
    def recommend_instance_type(self, cpu_requirement, memory_requirement, tps_requirement):
        suitable_instances = []
        
        for instance_type, specs in self.instance_types.items():
            if specs['cpu'] >= cpu_requirement and specs['memory'] >= memory_requirement:
                # Estimate TPS capacity (rough calculation)
                estimated_tps = specs['cpu'] * 50  # 50 TPS per CPU
                
                if estimated_tps >= tps_requirement:
                    cost_per_tps = specs['cost_per_hour'] / estimated_tps
                    suitable_instances.append({
                        'type': instance_type,
                        'cost_per_hour': specs['cost_per_hour'],
                        'cost_per_tps': cost_per_tps,
                        'estimated_tps': estimated_tps
                    })
        
        # Sort by cost efficiency
        return sorted(suitable_instances, key=lambda x: x['cost_per_tps'])

# Reserved capacity for baseline load
reserved_instances:
  payment_processor:
    instance_type: "c5.large"
    count: 10
    term: "1_year"
    payment_option: "partial_upfront"
    estimated_savings: "40%"
  
  database:
    instance_type: "r5.xlarge"
    count: 2
    term: "3_year"
    payment_option: "all_upfront"
    estimated_savings: "60%"

# Spot instances for batch processing
spot_instances:
  fraud_analysis:
    instance_type: "m5.large"
    max_price: "$0.05"
    interruption_handling: "graceful_shutdown"