API Gateway Design Pattern Unified Entry Management Strategy in Microservice Architecture（1750835439898300）

use hyperlane::*;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use serde::{Deserialize, Serialize};

// API Gateway core structure
#[derive(Clone)]
struct ApiGateway {
    service_registry: Arc<RwLock<ServiceRegistry>>,
    route_config: Arc<RwLock<RouteConfiguration>>,
    middleware_chain: MiddlewareChain,
    load_balancer: Arc<LoadBalancer>,
    circuit_breaker: Arc<CircuitBreaker>,
    rate_limiter: Arc<RateLimiter>,
}

#[derive(Clone)]
struct ServiceRegistry {
    services: HashMap<String, Vec<ServiceInstance>>,
    health_checks: HashMap<String, HealthCheckConfig>,
}

#[derive(Clone, Debug)]
struct ServiceInstance {
    id: String,
    host: String,
    port: u16,
    health_status: HealthStatus,
    last_health_check: chrono::DateTime<chrono::Utc>,
    metadata: HashMap<String, String>,
    load_factor: f64,
}

#[derive(Clone, Debug)]
enum HealthStatus {
    Healthy,
    Unhealthy,
    Unknown,
}

#[derive(Clone)]
struct HealthCheckConfig {
    endpoint: String,
    interval_seconds: u64,
    timeout_seconds: u64,
    failure_threshold: u32,
}

#[derive(Clone)]
struct RouteConfiguration {
    routes: HashMap<String, RouteRule>,
    default_route: Option<String>,
}

#[derive(Clone, Debug)]
struct RouteRule {
    path_pattern: String,
    service_name: String,
    methods: Vec<String>,
    authentication_required: bool,
    rate_limit: Option<RateLimit>,
    timeout_seconds: u64,
    retry_policy: RetryPolicy,
    transformation_rules: Vec<TransformationRule>,
}

#[derive(Clone, Debug)]
struct RateLimit {
    requests_per_minute: u32,
    burst_size: u32,
}

#[derive(Clone, Debug)]
struct RetryPolicy {
    max_attempts: u32,
    backoff_strategy: BackoffStrategy,
    retry_conditions: Vec<RetryCondition>,
}

#[derive(Clone, Debug)]
enum BackoffStrategy {
    Fixed(std::time::Duration),
    Exponential { base: std::time::Duration, max: std::time::Duration },
    Linear(std::time::Duration),
}

#[derive(Clone, Debug)]
enum RetryCondition {
    StatusCode(u16),
    Timeout,
    ConnectionError,
    ServiceUnavailable,
}

#[derive(Clone, Debug)]
struct TransformationRule {
    rule_type: TransformationType,
    source_path: String,
    target_path: String,
    transformation_logic: String,
}

#[derive(Clone, Debug)]
enum TransformationType {
    HeaderTransform,
    BodyTransform,
    QueryParamTransform,
    PathRewrite,
}

impl ApiGateway {
    fn new() -> Self {
        Self {
            service_registry: Arc::new(RwLock::new(ServiceRegistry::new())),
            route_config: Arc::new(RwLock::new(RouteConfiguration::new())),
            middleware_chain: MiddlewareChain::new(),
            load_balancer: Arc::new(LoadBalancer::new()),
            circuit_breaker: Arc::new(CircuitBreaker::new()),
            rate_limiter: Arc::new(RateLimiter::new()),
        }
    }

    async fn register_service(&self, service_name: String, instance: ServiceInstance) {
        let mut registry = self.service_registry.write().await;
        registry.services.entry(service_name.clone())
            .or_insert_with(Vec::new)
            .push(instance);

        println!("Registered service instance: {} for service: {}",
                instance.id, service_name);
    }

    async fn add_route(&self, route: RouteRule) {
        let mut config = self.route_config.write().await;
        config.routes.insert(route.path_pattern.clone(), route);
    }

    async fn handle_request(&self, ctx: &mut Context) -> Result<(), GatewayError> {
        let request_path = ctx.get_request_path().await;
        let request_method = ctx.get_request_method().await;

        // Find matching route
        let route = self.find_matching_route(&request_path, &request_method).await
            .ok_or(GatewayError::RouteNotFound)?;

        // Apply rate limiting
        self.apply_rate_limiting(ctx, &route).await?;

        // Apply authentication if required
        if route.authentication_required {
            self.authenticate_request(ctx).await?;
        }

        // Apply request transformations
        self.apply_request_transformations(ctx, &route).await?;

        // Select target service instance
        let service_instance = self.select_service_instance(&route.service_name).await?;

        // Forward request with circuit breaker protection
        let response = self.forward_request_with_protection(ctx, &service_instance, &route).await?;

        // Apply response transformations
        self.apply_response_transformations(ctx, &route, response).await?;

        Ok(())
    }

    async fn find_matching_route(&self, path: &str, method: &str) -> Option<RouteRule> {
        let config = self.route_config.read().await;

        for (pattern, route) in &config.routes {
            if self.path_matches(path, pattern) && route.methods.contains(&method.to_string()) {
                return Some(route.clone());
            }
        }

        None
    }

    fn path_matches(&self, path: &str, pattern: &str) -> bool {
        // Simplified path matching - in production, use regex or more sophisticated matching
        if pattern.contains("*") {
            let prefix = pattern.trim_end_matches("*");
            path.starts_with(prefix)
        } else if pattern.contains("{") {
            // Handle path parameters like /users/{id}
            let pattern_parts: Vec<&str> = pattern.split('/').collect();
            let path_parts: Vec<&str> = path.split('/').collect();

            if pattern_parts.len() != path_parts.len() {
                return false;
            }

            for (pattern_part, path_part) in pattern_parts.iter().zip(path_parts.iter()) {
                if !pattern_part.starts_with('{') && pattern_part != path_part {
                    return false;
                }
            }

            true
        } else {
            path == pattern
        }
    }

    async fn apply_rate_limiting(&self, ctx: &Context, route: &RouteRule) -> Result<(), GatewayError> {
        if let Some(rate_limit) = &route.rate_limit {
            let client_id = self.extract_client_id(ctx).await;

            if !self.rate_limiter.allow_request(&client_id, rate_limit).await {
                return Err(GatewayError::RateLimitExceeded);
            }
        }

        Ok(())
    }

    async fn extract_client_id(&self, ctx: &Context) -> String {
        // Extract client ID from various sources
        let headers = ctx.get_request_headers().await;

        if let Some(api_key) = headers.get("x-api-key") {
            return api_key.clone();
        }

        if let Some(auth_header) = headers.get("authorization") {
            if let Some(token) = auth_header.strip_prefix("Bearer ") {
                return format!("token:{}", token);
            }
        }

        // Fallback to IP address
        ctx.get_socket_addr_or_default_string().await
    }

    async fn authenticate_request(&self, ctx: &Context) -> Result<(), GatewayError> {
        let headers = ctx.get_request_headers().await;

        if let Some(auth_header) = headers.get("authorization") {
            if let Some(token) = auth_header.strip_prefix("Bearer ") {
                // Validate JWT token or API key
                if self.validate_token(token).await {
                    return Ok(());
                }
            }
        }

        Err(GatewayError::Unauthorized)
    }

    async fn validate_token(&self, token: &str) -> bool {
        // Simplified token validation
        // In production, validate JWT signature, expiration, etc.
        !token.is_empty() && token.len() > 10
    }

    async fn apply_request_transformations(&self, ctx: &mut Context, route: &RouteRule) -> Result<(), GatewayError> {
        for transformation in &route.transformation_rules {
            match transformation.rule_type {
                TransformationType::HeaderTransform => {
                    self.transform_headers(ctx, transformation).await?;
                }
                TransformationType::PathRewrite => {
                    self.rewrite_path(ctx, transformation).await?;
                }
                TransformationType::QueryParamTransform => {
                    self.transform_query_params(ctx, transformation).await?;
                }
                _ => {}
            }
        }

        Ok(())
    }

    async fn transform_headers(&self, ctx: &mut Context, transformation: &TransformationRule) -> Result<(), GatewayError> {
        // Add service identification headers
        ctx.set_request_header("X-Gateway-Service", &transformation.target_path).await;
        ctx.set_request_header("X-Gateway-Timestamp", &chrono::Utc::now().timestamp().to_string()).await;

        Ok(())
    }

    async fn rewrite_path(&self, ctx: &mut Context, transformation: &TransformationRule) -> Result<(), GatewayError> {
        let current_path = ctx.get_request_path().await;
        let new_path = current_path.replace(&transformation.source_path, &transformation.target_path);
        ctx.set_request_path(&new_path).await;

        Ok(())
    }

    async fn transform_query_params(&self, ctx: &mut Context, _transformation: &TransformationRule) -> Result<(), GatewayError> {
        // Add gateway-specific query parameters
        let mut query_params = ctx.get_query_params().await;
        query_params.insert("gateway_version".to_string(), "1.0".to_string());

        Ok(())
    }

    async fn select_service_instance(&self, service_name: &str) -> Result<ServiceInstance, GatewayError> {
        let registry = self.service_registry.read().await;

        if let Some(instances) = registry.services.get(service_name) {
            let healthy_instances: Vec<&ServiceInstance> = instances.iter()
                .filter(|instance| matches!(instance.health_status, HealthStatus::Healthy))
                .collect();

            if healthy_instances.is_empty() {
                return Err(GatewayError::NoHealthyInstances);
            }

            // Use load balancer to select instance
            self.load_balancer.select_instance(&healthy_instances).await
                .ok_or(GatewayError::LoadBalancingFailed)
        } else {
            Err(GatewayError::ServiceNotFound)
        }
    }

    async fn forward_request_with_protection(&self, ctx: &Context, instance: &ServiceInstance, route: &RouteRule) -> Result<ServiceResponse, GatewayError> {
        // Check circuit breaker
        if !self.circuit_breaker.allow_request(&instance.id).await {
            return Err(GatewayError::CircuitBreakerOpen);
        }

        let mut attempts = 0;
        let mut last_error = None;

        while attempts < route.retry_policy.max_attempts {
            attempts += 1;

            match self.forward_request(ctx, instance).await {
                Ok(response) => {
                    self.circuit_breaker.record_success(&instance.id).await;
                    return Ok(response);
                }
                Err(error) => {
                    self.circuit_breaker.record_failure(&instance.id).await;
                    last_error = Some(error);

                    if attempts < route.retry_policy.max_attempts {
                        let delay = self.calculate_retry_delay(&route.retry_policy, attempts);
                        tokio::time::sleep(delay).await;
                    }
                }
            }
        }

        Err(last_error.unwrap_or(GatewayError::RequestFailed))
    }

    async fn forward_request(&self, ctx: &Context, instance: &ServiceInstance) -> Result<ServiceResponse, GatewayError> {
        // Simulate HTTP request forwarding
        let target_url = format!("http://{}:{}{}", instance.host, instance.port, ctx.get_request_path().await);

        // In a real implementation, use an HTTP client like reqwest
        tokio::time::sleep(std::time::Duration::from_millis(50)).await; // Simulate network delay

        Ok(ServiceResponse {
            status_code: 200,
            headers: HashMap::new(),
            body: serde_json::json!({
                "message": "Response from service",
                "instance_id": instance.id,
                "timestamp": chrono::Utc::now().timestamp()
            }),
        })
    }

    fn calculate_retry_delay(&self, policy: &RetryPolicy, attempt: u32) -> std::time::Duration {
        match &policy.backoff_strategy {
            BackoffStrategy::Fixed(duration) => *duration,
            BackoffStrategy::Exponential { base, max } => {
                let delay = *base * 2_u32.pow(attempt - 1);
                std::cmp::min(delay, *max)
            }
            BackoffStrategy::Linear(duration) => *duration * attempt,
        }
    }

    async fn apply_response_transformations(&self, ctx: &mut Context, _route: &RouteRule, response: ServiceResponse) -> Result<(), GatewayError> {
        // Set response status and headers
        ctx.set_response_status_code(response.status_code).await;

        for (key, value) in response.headers {
            ctx.set_response_header(&key, &value).await;
        }

        // Add gateway headers
        ctx.set_response_header("X-Gateway-Response-Time", &chrono::Utc::now().timestamp().to_string()).await;
        ctx.set_response_header("X-Powered-By", "Hyperlane-Gateway").await;

        // Set response body
        ctx.set_response_body(serde_json::to_string(&response.body).unwrap()).await;

        Ok(())
    }
}

#[derive(Debug)]
enum GatewayError {
    RouteNotFound,
    ServiceNotFound,
    NoHealthyInstances,
    LoadBalancingFailed,
    RateLimitExceeded,
    Unauthorized,
    CircuitBreakerOpen,
    RequestFailed,
    TransformationFailed,
}

#[derive(Debug)]
struct ServiceResponse {
    status_code: u16,
    headers: HashMap<String, String>,
    body: serde_json::Value,
}

// Supporting components
struct LoadBalancer {
    strategy: LoadBalancingStrategy,
}

enum LoadBalancingStrategy {
    RoundRobin,
    LeastConnections,
    WeightedRandom,
}

impl LoadBalancer {
    fn new() -> Self {
        Self {
            strategy: LoadBalancingStrategy::RoundRobin,
        }
    }

    async fn select_instance(&self, instances: &[&ServiceInstance]) -> Option<ServiceInstance> {
        if instances.is_empty() {
            return None;
        }

        match self.strategy {
            LoadBalancingStrategy::RoundRobin => {
                // Simplified round-robin selection
                Some(instances[0].clone())
            }
            LoadBalancingStrategy::LeastConnections => {
                // Select instance with lowest load factor
                instances.iter()
                    .min_by(|a, b| a.load_factor.partial_cmp(&b.load_factor).unwrap())
                    .map(|&instance| instance.clone())
            }
            LoadBalancingStrategy::WeightedRandom => {
                // Weighted random selection based on load factor
                Some(instances[rand::random::<usize>() % instances.len()].clone())
            }
        }
    }
}

struct CircuitBreaker {
    states: Arc<RwLock<HashMap<String, CircuitBreakerState>>>,
}

#[derive(Clone)]
struct CircuitBreakerState {
    state: CircuitState,
    failure_count: u32,
    last_failure_time: Option<chrono::DateTime<chrono::Utc>>,
    success_count: u32,
}

#[derive(Clone)]
enum CircuitState {
    Closed,
    Open,
    HalfOpen,
}

impl CircuitBreaker {
    fn new() -> Self {
        Self {
            states: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    async fn allow_request(&self, service_id: &str) -> bool {
        let states = self.states.read().await;

        if let Some(state) = states.get(service_id) {
            match state.state {
                CircuitState::Closed => true,
                CircuitState::Open => {
                    // Check if enough time has passed to try half-open
                    if let Some(last_failure) = state.last_failure_time {
                        chrono::Utc::now().signed_duration_since(last_failure).num_seconds() > 60
                    } else {
                        false
                    }
                }
                CircuitState::HalfOpen => state.success_count < 3, // Allow limited requests
            }
        } else {
            true // Default to allowing requests for new services
        }
    }

    async fn record_success(&self, service_id: &str) {
        let mut states = self.states.write().await;
        let state = states.entry(service_id.to_string()).or_insert_with(|| CircuitBreakerState {
            state: CircuitState::Closed,
            failure_count: 0,
            last_failure_time: None,
            success_count: 0,
        });

        state.success_count += 1;
        state.failure_count = 0;

        if matches!(state.state, CircuitState::HalfOpen) && state.success_count >= 3 {
            state.state = CircuitState::Closed;
        }
    }

    async fn record_failure(&self, service_id: &str) {
        let mut states = self.states.write().await;
        let state = states.entry(service_id.to_string()).or_insert_with(|| CircuitBreakerState {
            state: CircuitState::Closed,
            failure_count: 0,
            last_failure_time: None,
            success_count: 0,
        });

        state.failure_count += 1;
        state.last_failure_time = Some(chrono::Utc::now());
        state.success_count = 0;

        if state.failure_count >= 5 {
            state.state = CircuitState::Open;
        }
    }
}

struct RateLimiter {
    buckets: Arc<RwLock<HashMap<String, TokenBucket>>>,
}

struct TokenBucket {
    tokens: f64,
    last_refill: chrono::DateTime<chrono::Utc>,
    capacity: f64,
    refill_rate: f64,
}

impl RateLimiter {
    fn new() -> Self {
        Self {
            buckets: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    async fn allow_request(&self, client_id: &str, rate_limit: &RateLimit) -> bool {
        let mut buckets = self.buckets.write().await;
        let bucket = buckets.entry(client_id.to_string()).or_insert_with(|| TokenBucket {
            tokens: rate_limit.requests_per_minute as f64,
            last_refill: chrono::Utc::now(),
            capacity: rate_limit.requests_per_minute as f64,
            refill_rate: rate_limit.requests_per_minute as f64 / 60.0, // tokens per second
        });

        // Refill tokens based on elapsed time
        let now = chrono::Utc::now();
        let elapsed_seconds = now.signed_duration_since(bucket.last_refill).num_seconds() as f64;
        bucket.tokens = (bucket.tokens + elapsed_seconds * bucket.refill_rate).min(bucket.capacity);
        bucket.last_refill = now;

        // Check if request can be allowed
        if bucket.tokens >= 1.0 {
            bucket.tokens -= 1.0;
            true
        } else {
            false
        }
    }
}

impl ServiceRegistry {
    fn new() -> Self {
        Self {
            services: HashMap::new(),
            health_checks: HashMap::new(),
        }
    }
}

impl RouteConfiguration {
    fn new() -> Self {
        Self {
            routes: HashMap::new(),
            default_route: None,
        }
    }
}

struct MiddlewareChain;

impl MiddlewareChain {
    fn new() -> Self {
        Self
    }
}

static API_GATEWAY: once_cell::sync::Lazy<ApiGateway> =
    once_cell::sync::Lazy::new(|| {
        let gateway = ApiGateway::new();

        // Initialize with some default routes and services
        tokio::spawn(async move {
            // This would be done during application startup
        });

        gateway
    });

#[get]
async fn gateway_proxy_endpoint(ctx: Context) {
    let mut ctx = ctx;

    match API_GATEWAY.handle_request(&mut ctx).await {
        Ok(()) => {
            // Request was successfully processed and response was set
        }
        Err(error) => {
            let (status_code, error_message) = match error {
                GatewayError::RouteNotFound => (404, "Route not found"),
                GatewayError::ServiceNotFound => (503, "Service not available"),
                GatewayError::NoHealthyInstances => (503, "No healthy service instances"),
                GatewayError::RateLimitExceeded => (429, "Rate limit exceeded"),
                GatewayError::Unauthorized => (401, "Unauthorized"),
                GatewayError::CircuitBreakerOpen => (503, "Service temporarily unavailable"),
                _ => (500, "Internal gateway error"),
            };

            ctx.set_response_status_code(status_code)
                .await
                .set_response_header(CONTENT_TYPE, APPLICATION_JSON)
                .await
                .set_response_body(format!(r#"{{"error": "{}"}}"#, error_message))
                .await;
        }
    }
}

#[post]
async fn register_service_endpoint(ctx: Context) {
    let request_body: Vec<u8> = ctx.get_request_body().await;
    let registration_request: ServiceRegistrationRequest = serde_json::from_slice(&request_body).unwrap();

    let instance = ServiceInstance {
        id: registration_request.instance_id,
        host: registration_request.host,
        port: registration_request.port,
        health_status: HealthStatus::Unknown,
        last_health_check: chrono::Utc::now(),
        metadata: registration_request.metadata,
        load_factor: 0.0,
    };

    API_GATEWAY.register_service(registration_request.service_name, instance).await;

    ctx.set_response_status_code(201)
        .await
        .set_response_header(CONTENT_TYPE, APPLICATION_JSON)
        .await
        .set_response_body(r#"{"status": "registered"}"#)
        .await;
}

#[derive(Deserialize)]
struct ServiceRegistrationRequest {
    service_name: String,
    instance_id: String,
    host: String,
    port: u16,
    metadata: HashMap<String, String>,
}

#[get]
async fn gateway_health_endpoint(ctx: Context) {
    let health_status = serde_json::json!({
        "status": "healthy",
        "timestamp": chrono::Utc::now().timestamp(),
        "version": "1.0.0",
        "services_registered": 5,
        "active_routes": 12,
        "requests_processed": 150000
    });

    ctx.set_response_status_code(200)
        .await
        .set_response_header(CONTENT_TYPE, APPLICATION_JSON)
        .await
        .set_response_body(serde_json::to_string(&health_status).unwrap())
        .await;
}