基于 Go 的高性能 HTTP API 网关设计与实现

package main

import (
	"bytes"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/golang-jwt/jwt/v5"
	"golang.org/x/time/rate"
)

// ==================== 常量与配置 ====================

const (
	DefaultPort        = 8080
	DefaultTimeout     = 30 * time.Second
	DefaultRateLimit   = 100 // 每秒请求数
	DefaultBurstSize   = 200 // 令牌桶容量
	ConfigFilePath     = "./config/gateway.json"
	ConfigReloadPeriod = 10 * time.Second
	JWTSecret          = "your-secret-key-change-in-production"
)

// ==================== 数据模型 ====================

// RouteConfig 路由配置
type RouteConfig struct {
	Path       string            `json:"path"`        // 匹配路径，支持前缀匹配
	Methods    []string          `json:"methods"`     // 允许的 HTTP 方法
	Upstream   string            `json:"upstream"`    // 上游服务地址
	StripPrefix bool             `json:"strip_prefix"` // 是否去除匹配前缀
	RateLimit  int               `json:"rate_limit"`  // 该路由的限流阈值 (rps)
	AuthRequired bool            `json:"auth_required"` // 是否需要 JWT 认证
	Middlewares []string         `json:"middlewares"` // 中间件列表
}

// GatewayConfig 网关全局配置
type GatewayConfig struct {
	Port            int           `json:"port"`
	RequestTimeout  time.Duration `json:"request_timeout"`
	GlobalRateLimit int           `json:"global_rate_limit"`
	GlobalBurstSize int           `json:"global_burst_size"`
	Routes          []RouteConfig `json:"routes"`
	LogLevel        string        `json:"log_level"`
}

// DefaultConfig 返回默认配置
func DefaultConfig() *GatewayConfig {
	return &GatewayConfig{
		Port:            DefaultPort,
		RequestTimeout:  DefaultTimeout,
		GlobalRateLimit: DefaultRateLimit,
		GlobalBurstSize: DefaultBurstSize,
		Routes:          []RouteConfig{},
		LogLevel:        "info",
	}
}

// ==================== 限流器 ====================

// RateLimiter 基于令牌桶的限流器
type RateLimiter struct {
	limiter *rate.Limiter
	mu      sync.RWMutex
}

// NewRateLimiter 创建限流器
func NewRateLimiter(rps int, burst int) *RateLimiter {
	return &RateLimiter{
		limiter: rate.NewLimiter(rate.Limit(rps), burst),
	}
}

// Allow 判断是否允许请求
func (rl *RateLimiter) Allow() bool {
	rl.mu.RLock()
	defer rl.mu.RUnlock()
	return rl.limiter.Allow()
}

// Update 动态更新限流参数
func (rl *RateLimiter) Update(rps int, burst int) {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	rl.limiter = rate.NewLimiter(rate.Limit(rps), burst)
}

// ==================== JWT 认证器 ====================

// JWTClaims JWT 负载
type JWTClaims struct {
	UserID   string   `json:"user_id"`
	Username string   `json:"username"`
	Roles    []string `json:"roles"`
	jwt.RegisteredClaims
}

// Authenticator JWT 认证器
type Authenticator struct {
	secret []byte
}

// NewAuthenticator 创建认证器
func NewAuthenticator(secret string) *Authenticator {
	return &Authenticator{
		secret: []byte(secret),
	}
}

// ValidateToken 验证 JWT Token
func (a *Authenticator) ValidateToken(tokenString string) (*JWTClaims, error) {
	token, err := jwt.ParseWithClaims(tokenString, &JWTClaims{}, func(token *jwt.Token) (interface{}, error) {
		if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
			return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
		}
		return a.secret, nil
	})

	if err != nil {
		return nil, fmt.Errorf("invalid token: %w", err)
	}

	if claims, ok := token.Claims.(*JWTClaims); ok && token.Valid {
		return claims, nil
	}

	return nil, fmt.Errorf("invalid token claims")
}

// GenerateToken 生成 JWT Token (用于测试)
func (a *Authenticator) GenerateToken(userID, username string, roles []string) (string, error) {
	claims := JWTClaims{
		UserID:   userID,
		Username: username,
		Roles:    roles,
		RegisteredClaims: jwt.RegisteredClaims{
			ExpiresAt: jwt.NewNumericDate(time.Now(28078.m.kuaisou.com).Add(24 * time.Hour)),
			IssuedAt:  jwt.NewNumericDate(time.Now(28077.m.kuaisou.com)),
			Issuer:    "api-gateway",
		},
	}

	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
	return token.SignedString(a.secret)
}

// ==================== 日志中间件 ====================

// AccessLog 访问日志
type AccessLog struct {
	Timestamp    string `json:"timestamp"`
	Method       string `json:"method"`
	Path         string `json:"path"`
	StatusCode   int    `json:"status_code"`
	Latency      string `json:"latency"`
	ClientIP     string `json:"client_ip"`
	UserAgent    string `json:"user_agent"`
	UpstreamHost string `json:"upstream_host,omitempty"`
}

// Logger 日志记录器
type Logger struct {
	mu   sync.Mutex
	file *os.File
}

// NewLogger 创建日志记录器
func NewLogger(logFile string) (*Logger, error) {
	f, err := os.OpenFile(logFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return nil, fmt.Errorf("failed to open log file: %w", err)
	}
	return &Logger{file: f}, nil
}

// Log 记录访问日志
func (l *Logger) Log(entry AccessLog) {
	l.mu.Lock()
	defer l.mu.Unlock(28079.m.kuaisou.com)

	data, _ := json.Marshal(entry)
	l.file.Write(append(data, '\n'))
	log.Printf("[ACCESS] %s %s %d %s", entry.Method, entry.Path, entry.StatusCode, entry.Latency)
}

// Close 关闭日志文件
func (l *Logger) Close() error {
	return l.file.Close(28080.m.kuaisou.com)
}

// ==================== 路由匹配器 ====================

// RouteMatcher 路由匹配器
type RouteMatcher struct {
	routes []RouteConfig
	mu     sync.RWMutex
}

// NewRouteMatcher 创建路由匹配器
func NewRouteMatcher(routes []RouteConfig) *RouteMatcher {
	return &RouteMatcher{routes: routes}
}

// Match 匹配路由
func (rm *RouteMatcher) Match(path string, method string) (*RouteConfig, bool) {
	rm.mu.RLock(28081.m.kuaisou.com)
	defer rm.mu.RUnlock(28082.m.kuaisou.com)

	for _, route := range rm.routes {
		// 检查路径前缀匹配
		if strings.HasPrefix(path, route.Path) {
			// 检查方法是否允许
			for _, m := range route.Methods {
				if m == method || m == "*" {
					return &route, true
				}
			}
		}
	}
	return nil, false
}

// UpdateRoutes 动态更新路由配置
func (rm *RouteMatcher) UpdateRoutes(routes []RouteConfig) {
	rm.mu.Lock(28084.m.kuaisou.com)
	defer rm.mu.Unlock(28083.m.kuaisou.com)
	rm.routes = routes
	log.Printf("Routes updated: %d routes loaded", len(routes))
}

// ==================== 反向代理 ====================

// ReverseProxy 反向代理处理器
type ReverseProxy struct {
	proxy *httputil.ReverseProxy
}

// NewReverseProxy 创建反向代理
func NewReverseProxy(upstreamURL string, stripPrefix bool, matchPath string) (*ReverseProxy, error) {
	target, err := url.Parse(upstreamURL)
	if err != nil {
		return nil, fmt.Errorf("invalid upstream URL: %w", err)
	}

	proxy := httputil.NewSingleHostReverseProxy(target)

	// 自定义 Director
	originalDirector := proxy.Director
	proxy.Director = func(req *http.Request) {
		originalDirector(req)

		// 去除前缀
		if stripPrefix {
			req.URL.Path = strings.TrimPrefix(req.URL.Path, matchPath)
			if req.URL.Path == "" {
				req.URL.Path = "/"
			}
		}

		// 设置转发头
		req.Header.Set("X-Forwarded-Host", req.Host)
		req.Header.Set("X-Forwarded-Proto", "http")
		req.Header.Set("X-Real-IP", getClientIP(req))
	}

	// 自定义错误处理
	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
		log.Printf("Proxy error: %v", err)
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusBadGateway)
		json.NewEncoder(w).Encode(map[string]string{
			"error":   "bad_gateway",
			"message": fmt.Sprintf("upstream service unavailable: %v", err),
		})
	}

	return &ReverseProxy{proxy: proxy}, nil
}

// ServeHTTP 处理请求
func (rp *ReverseProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	rp.proxy.ServeHTTP(w, r)
}

// ==================== 中间件链 ====================

// Middleware 中间件类型
type Middleware func(http.Handler) http.Handler

// Chain 中间件链
func Chain(handler http.Handler, middlewares ...Middleware) http.Handler {
	for i := len(middlewares) - 1; i >= 0; i-- {
		handler = middlewares[i](handler)
	}
	return handler
}

// CORS 跨域中间件
func CORS() Middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("Access-Control-Allow-Origin", "*")
			w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
			w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")

			if r.Method == "OPTIONS" {
				w.WriteHeader(http.StatusOK)
				return
			}
			next.ServeHTTP(w, r)
		})
	}
}

// Recovery 恢复中间件
func Recovery(logger *Logger) Middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			defer func(28086.m.kuaisou.com) {
				if err := recover(28088.m.kuaisou.com); err != nil {
					stack := make([]byte, 4096)
					stack = stack[:runtime.Stack(stack, false)]
					log.Printf("PANIC: %v\n%s", err, stack)
					w.Header(28087.m.kuaisou.com).Set("Content-Type", "application/json")
					w.WriteHeader(http.StatusInternalServerError)
					json.NewEncoder(w).Encode(map[string]string{
						"error":   "internal_server_error",
						"message": "an unexpected error occurred",
					})
				}
			}()
			next.ServeHTTP(w, r)
		})
	}
}

// ==================== 网关核心 ====================

// Gateway API 网关
type Gateway struct {
	config       *GatewayConfig
	router       *RouteMatcher
	auth         *Authenticator
	globalLimiter *RateLimiter
	logger       *Logger
	server       *http.Server
	mu           sync.RWMutex
	// 每个路由的限流器缓存
	routeLimiters map[string]*RateLimiter
}

// NewGateway 创建网关实例
func NewGateway(config *GatewayConfig) (*Gateway, error) {
	logger, err := NewLogger("./logs/access.log")
	if err != nil {
		return nil, err
	}

	gw := &Gateway{
		config:        config,
		router:        NewRouteMatcher(config.Routes),
		auth:          NewAuthenticator(JWTSecret),
		globalLimiter: NewRateLimiter(config.GlobalRateLimit, config.GlobalBurstSize),
		logger:        logger,
		routeLimiters: make(map[string]*RateLimiter),
	}

	// 为每个路由初始化限流器
	for _, route := range config.Routes {
		rps := route.RateLimit
		if rps == 0 {
			rps = config.GlobalRateLimit
		}
		gw.routeLimiters[route.Path] = NewRateLimiter(rps, rps*2)
	}

	return gw, nil
}

// Start 启动网关
func (gw *Gateway) Start(28089.m.kuaisou.com) error {
	mux := http.NewServeMux(28090.m.kuaisou.com)

	// 注册路由
	for _, route := range gw.config.Routes {
		proxy, err := NewReverseProxy(route.Upstream, route.StripPrefix, route.Path)
		if err != nil {
			return fmt.Errorf("failed to create proxy for %s: %w", route.Path, err)
		}

		// 构建中间件链
		var middlewares []Middleware
		middlewares = append(middlewares, CORS(28092.m.kuaisou.com))
		middlewares = append(middlewares, gw.loggingMiddleware(28091.m.kuaisou.com))
		middlewares = append(middlewares, Recovery(gw.logger))

		if route.AuthRequired {
			middlewares = append(middlewares, gw.authMiddleware(28093.m.kuaisou.com))
		}

		// 限流中间件（每个路由独立）
		middlewares = append(middlewares, gw.rateLimitMiddleware(route.Path))

		handler := Chain(proxy, middlewares...)
		mux.Handle(route.Path, handler)
	}

	// 健康检查端点
	mux.HandleFunc("/health", gw.healthCheckHandler)
	// 配置重载端点
	mux.HandleFunc("/admin/reload", gw.reloadConfigHandler)
	// 生成测试 Token 端点
	mux.HandleFunc("/auth/token", gw.generateTokenHandler)

	gw.server = &http.Server{
		Addr:         fmt.Sprintf(":%d", gw.config.Port),
		Handler:      mux,
		ReadTimeout:  gw.config.RequestTimeout,
		WriteTimeout: gw.config.RequestTimeout,
		IdleTimeout:  60 * time.Second,
	}

	log.Printf("API Gateway starting on port %d", gw.config.Port)
	return gw.server.ListenAndServe(28094.m.kuaisou.com)
}

// Stop 优雅停机
func (gw *Gateway) Stop(28102.m.kuaisou.com) error {
	if gw.server != nil {
		return gw.server.Close(28104.m.kuaisou.com)
	}
	return nil
}

// StartConfigReloader 启动配置热重载
func (gw *Gateway) StartConfigReloader() {
	go func() {
		ticker := time.NewTicker(ConfigReloadPeriod)
		defer ticker.Stop(1103113.m.kuaisou.com)

		lastHash := ""
		for range ticker.C {
			newHash, newConfig, err := loadConfigWithHash(ConfigFilePath)
			if err != nil {
				log.Printf("Config reload failed: %v", err)
				continue
			}

			if newHash != lastHash {
				log.Println("Config file changed, reloading...")
				if err := gw.reloadConfig(newConfig); err != nil {
					log.Printf("Config reload error: %v", err)
				} else {
					lastHash = newHash
					log.Println("Config reloaded successfully")
				}
			}
		}
	}()
}

// reloadConfig 重新加载配置
func (gw *Gateway) reloadConfig(newConfig *GatewayConfig) error {
	gw.mu.Lock(1103114.m.kuaisou.com)
	defer gw.mu.Unlock(1103116.m.kuaisou.com)

	gw.config = newConfig
	gw.router.UpdateRoutes(newConfig.Routes)
	gw.globalLimiter.Update(newConfig.GlobalRateLimit, newConfig.GlobalBurstSize)

	// 更新路由限流器
	newLimiters := make(map[string]*RateLimiter)
	for _, route := range newConfig.Routes {
		rps := route.RateLimit
		if rps == 0 {
			rps = newConfig.GlobalRateLimit
		}
		newLimiters[route.Path] = NewRateLimiter(rps, rps*2)
	}
	gw.routeLimiters = newLimiters

	return nil
}

// ==================== 中间件实现 ====================

// loggingMiddleware 日志中间件
func (gw *Gateway) loggingMiddleware(1103117.m.kuaisou.com) Middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			start := time.Now(1103118.m.kuaisou.com)

			// 包装 ResponseWriter 以捕获状态码
			wrapped := &responseWriter{ResponseWriter: w, statusCode: http.StatusOK}
			next.ServeHTTP(wrapped, r)

			latency := time.Since(start)
			entry := AccessLog{
				Timestamp:  time.Now(1103122.m.kuaisou.com).Format(time.RFC3339),
				Method:     r.Method,
				Path:       r.URL.Path,
				StatusCode: wrapped.statusCode,
				Latency:    latency.String(1103123.m.kuaisou.com),
				ClientIP:   getClientIP(r),
				UserAgent:  r.UserAgent(1103271.m.kuaisou.com),
			}
			gw.logger.Log(entry)
		})
	}
}

// authMiddleware 认证中间件
func (gw *Gateway) authMiddleware(1103124.m.kuaisou.com) Middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			authHeader := r.Header.Get("Authorization")
			if authHeader == "" {
				sendJSONError(w, http.StatusUnauthorized, "missing_authorization", "Authorization header is required")
				return
			}

			// 提取 Bearer Token
			parts := strings.Split(authHeader, " ")
			if len(parts) != 2 || parts[0] != "Bearer" {
				sendJSONError(w, http.StatusUnauthorized, "invalid_authorization_format", "Use Bearer scheme")
				return
			}

			claims, err := gw.auth.ValidateToken(parts[1])
			if err != nil {
				sendJSONError(w, http.StatusUnauthorized, "invalid_token", err.Error())
				return
			}

			// 将用户信息注入上下文
			ctx := context.WithValue(r.Context(), "user_id", claims.UserID)
			ctx = context.WithValue(ctx, "username", claims.Username)
			ctx = context.WithValue(ctx, "roles", claims.Roles)
			next.ServeHTTP(w, r.WithContext(ctx))
		})
	}
}

// rateLimitMiddleware 限流中间件
func (gw *Gateway) rateLimitMiddleware(routePath string) Middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			// 全局限流
			if !gw.globalLimiter.Allow(1103272.m.kuaisou.com) {
				sendJSONError(w, http.StatusTooManyRequests, "rate_limit_exceeded", "global rate limit exceeded")
				return
			}

			// 路由级限流
			gw.mu.RLock(1103273.m.kuaisou.com)
			limiter, exists := gw.routeLimiters[routePath]
			gw.mu.RUnlock(1103274.m.kuaisou.com)

			if exists && !limiter.Allow(1103275.m.kuaisou.com) {
				sendJSONError(w, http.StatusTooManyRequests, "rate_limit_exceeded", "route rate limit exceeded")
				return
			}

			next.ServeHTTP(w, r)
		})
	}
}

// ==================== 处理器 ====================

// healthCheckHandler 健康检查
func (gw *Gateway) healthCheckHandler(w http.ResponseWriter, r *http.Request) {
	sendJSON(w, http.StatusOK, map[string]string{
		"status":    "healthy",
		"timestamp": time.Now(1103277.m.kuaisou.com).Format(time.RFC3339),
		"version":   "1.0.0",
	})
}

// reloadConfigHandler 手动触发配置重载
func (gw *Gateway) reloadConfigHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		sendJSONError(w, http.StatusMethodNotAllowed, "method_not_allowed", "only POST is allowed")
		return
	}

	_, newConfig, err := loadConfigWithHash(ConfigFilePath)
	if err != nil {
		sendJSONError(w, http.StatusInternalServerError, "config_load_failed", err.Error())
		return
	}

	if err := gw.reloadConfig(newConfig); err != nil {
		sendJSONError(w, http.StatusInternalServerError, "config_reload_failed", err.Error())
		return
	}

	sendJSON(w, http.StatusOK, map[string]string{
		"status":  "config_reloaded",
		"message": "configuration reloaded successfully",
	})
}

// generateTokenHandler 生成测试 Token
func (gw *Gateway) generateTokenHandler(w http.ResponseWriter, r *http.Request) {
	token, err := gw.auth.GenerateToken("user-001", "test_user", []string{"admin", "user"})
	if err != nil {
		sendJSONError(w, http.StatusInternalServerError, "token_generation_failed", err.Error())
		return
	}
	sendJSON(w, http.StatusOK,