服务端模块实现教程（HTTP/gRPC/中间件）

package server

import (
	"github.com/google/wire"
)

// ProviderSet 是 server 层的依赖注入集合。
// wire.NewSet 把构造函数注册为一个 ProviderSet，
// Wire 会根据它们的参数类型和返回值类型自动推导依赖关系。
// 修正点：新增 CORSMiddleware 的构造已并入 NewHTTPServer，无需单独 Provider。
var ProviderSet = wire.NewSet(
	NewGRPCServer,    // 创建 gRPC 服务器
	NewHTTPServer,    // 创建 HTTP 服务器（含中间件链 + 健康检查）
	NewEtcdClient,    // 创建 etcd 客户端（用于服务注册）
	NewEtcdRegistrar, // 创建 etcd 服务注册器
)

//go:build wireinject
// +build wireinject

package main

import (
	"log/slog"

	"cloud-disk/internal/biz"
	"cloud-disk/internal/conf"
	"cloud-disk/internal/data"
	"cloud-disk/internal/data/cache"
	"cloud-disk/internal/data/lock"
	"cloud-disk/internal/data/storage"
	"cloud-disk/internal/event"
	"cloud-disk/internal/server"
	"cloud-disk/internal/service"

	"github.com/go-kratos/kratos/v3"
	"github.com/google/wire"
)

// wireApp 是 Wire 注入入口。
// 参数：5 个配置对象（Server/Data/Auth/Storage/Etcd）+ logger。
// 返回：*kratos.App（应用本体）+ cleanup 清理函数 + error。
func wireApp(*conf.Server, *conf.Data, *conf.Auth, *conf.Storage, *conf.Etcd, *slog.Logger) (*kratos.App, func(), error) {
	panic(wire.Build(
		server.ProviderSet,    // HTTP/gRPC/etcd 服务器
		data.ProviderSet,      // MySQL/Redis/Kafka 数据层
		cache.ProviderSet,     // 多级缓存
		lock.ProviderSet,      // 分布式锁
		taskProviderSet,       // 定时任务管理器 + 任务服务器
		storage.ProviderSet,   // 存储后端（MinIO/本地）
		event.ProviderSet,     // Kafka 事件
		biz.ProviderSet,       // 业务用例
		service.ProviderSet,   // 服务层
		newServers,            // 把 4 个 server 组装为切片
		newApp,                // 创建 *kratos.App
		newBizCache,           // data/cache 适配 biz.Cache
		newBizLocker,          // data/lock 适配 biz.Locker
	))
}

// Code generated by Wire. DO NOT EDIT.

package main

// wireApp init kratos application.
func wireApp(confServer *conf.Server, confData *conf.Data, auth *conf.Auth, confStorage *conf.Storage, etcd *conf.Etcd, logger *slog.Logger) (*kratos.App, func(), error) {
	// 1. 创建 TokenManager（biz 层，用于 JWT 签发/校验）
	//    修正点：若 auth.JwtSecret 为空，NewTokenManager 内部会 log.Fatalf 退出，
	//    不会回退到硬编码默认值（详见 5.8）。
	tokenManager := biz.NewTokenManager(auth)

	// 2. 创建 Data（MySQL + Redis + Kafka writer），返回 cleanup 关闭函数
	dataData, cleanup, err := data.NewData(confData)
	if err != nil {
		return nil, nil, err
	}
	db := dataData.DB                  // 取出 GORM DB
	userRepo := data.NewUserRepo(db)   // 创建用户 Repo
	client := dataData.RDB             // 取出 Redis 客户端

	// 3. 多级缓存 → 适配为 biz.Cache 接口
	cacheCache := cache.NewMultiLevelCache(client)
	bizCache := newBizCache(cacheCache)

	// 4. 分布式锁 → 适配为 biz.Locker 接口
	data_Lock := confData.Lock
	lockLock := lock.NewLock(data_Lock, client)
	locker := newBizLocker(lockLock)

	// 5. biz → service 层逐层构造
	userUsecase := biz.NewUserUsecase(userRepo, tokenManager, bizCache, locker)
	userService := service.NewUserService(userUsecase)
	// ... file / recycle / share 同理，省略 ...

	// 6. 创建 HTTP / gRPC 服务器（共享同一组 service）
	grpcServer := server.NewGRPCServer(confServer, tokenManager, userService, fileService, recycleService, shareService)
	httpServer := server.NewHTTPServer(confServer, tokenManager, userService, fileService, recycleService, shareService)

	// 7. 定时任务服务器（封装为 transport.Server）
	taskManager := newTaskManager(userUsecase, recycleUsecase, bizStorage)
	scheduledTaskServer := scheduler.NewScheduledTaskServer(taskManager)

	// 8. Kafka 消费者服务器（封装为 transport.Server）
	consumerServer := event.NewConsumerServer(consumer, eventHandlerService, redisIdempotencyStore, v)

	// 9. 多 Server 合并为切片
	v2 := newServers(grpcServer, httpServer, scheduledTaskServer, consumerServer)

	// 10. etcd 客户端 + 注册器
	clientv3Client, cleanup2, err := server.NewEtcdClient(etcd)
	if err != nil {
		cleanup()           // 失败时先释放前面申请的资源
		return nil, nil, err
	}
	registrar := server.NewEtcdRegistrar(clientv3Client)

	// 11. 创建 Kratos App
	app := newApp(logger, v2, registrar)

	// 12. 返回 app + 组合 cleanup（注意调用顺序：后申请的先释放）
	return app, func() {
		cleanup2()  // 关闭 etcd
		cleanup()   // 关闭 MySQL/Redis/Kafka
	}, nil
}

package server

import (
	"context"
	"encoding/json"
	"net/http"
	"strings"
	"time"

	filev1 "cloud-disk/api/file/v1"
	recyclev1 "cloud-disk/api/recycle/v1"
	sharev1 "cloud-disk/api/share/v1"
	userv1 "cloud-disk/api/user/v1"
	"cloud-disk/internal/biz"
	"cloud-disk/internal/conf"
	"cloud-disk/internal/data"
	"cloud-disk/internal/service"
	"cloud-disk/pkg/response" // 修正点：统一错误响应

	perrors "github.com/go-kratos/kratos/v3/errors"
	khttp "github.com/go-kratos/kratos/v3/transport/http"
)

// 三个手动路由的操作名常量，用于中间件白名单匹配
const (
	UserStorageOp = "/api/v1/user/storage"
	FileDeleteOp  = "/api/v1/file/delete"
	FilePreviewOp = "/api/v1/file/{id}/preview"
)

// NewHTTPServer 创建一个带有中间件链和服务路由的 HTTP 服务器。
func NewHTTPServer(c *conf.Server, tm *biz.TokenManager, us *service.UserService, fs *service.FileService, rs *service.RecycleService, ss *service.ShareService) *khttp.Server {
	// === 第一步：组装中间件链 ===
	// 修正点：在 recovery 之后、限流之前插入 CORSMiddleware，
	// 保证 OPTIONS 预检早于 JWT 被处理，避免跨域请求整体失败。
	var opts = []khttp.ServerOption{
		khttp.Middleware(
			recoveryMiddleware(),
			CORSMiddleware(c.GetCorsAllowedOrigins()...), // 修正点：按白名单配置可信源
			RateLimitMiddleware(600, time.Minute),         // 默认 600 次/分钟
			TimingMiddleware(),
			validateMiddleware(),
			CacheControlMiddleware(),
			JWTAuthMiddleware(tm,
				// === JWT 白名单：以下路径不需要登录即可访问 ===
				"/user.v1.UserService/Register",
				"/user.v1.UserService/Login",
				"/user.v1.UserService/RefreshToken",
				"/user.v1.UserService/ResetPassword",
				"/user.v1.UserService/GetSecurityQuestion",
				"/share.v1.ShareService/AccessShare",
				"/share.v1.ShareService/GetShareDetail",
				FileDeleteOp, // 手动路由加入白名单，由 handler 自行验证 token
			),
		),
		// 修正点：接入统一错误响应，避免把内部错误/堆栈泄露给前端
		khttp.ErrorEncoder(response.ErrorEncoder),
	}

	// === 第二步：根据配置设置监听参数 ===
	if c.Http.Network != "" {
		opts = append(opts, khttp.Network(c.Http.Network))
	}
	if c.Http.Addr != "" {
		opts = append(opts, khttp.Address(c.Http.Addr))
	}
	if c.Http.Timeout != nil {
		opts = append(opts, khttp.Timeout(c.Http.Timeout.AsDuration()))
	}
	srv := khttp.NewServer(opts...)

	// === 第三步：注册 proto 自动生成的 RESTful 路由 ===
	userv1.RegisterUserServiceHTTPServer(srv, us)
	filev1.RegisterFileServiceHTTPServer(srv, fs)
	recyclev1.RegisterRecycleServiceHTTPServer(srv, rs)
	sharev1.RegisterShareServiceHTTPServer(srv, ss)

	// === 第四步：手动注册特殊路由 ===
	srv.Route("/").Handle("GET", "/api/v1/user/storage", getUserStorageHandler(us))
	srv.Route("/").Handle("POST", "/api/v1/file/delete", getFileDeleteHandler(rs, tm))
	srv.Route("/").Handle("GET", "/api/v1/file/{id}/preview", getFilePreviewHandler(fs, tm))

	// === 第五步：修正点——注册健康检查端点（供 K8s/Compose 探活）===
	// liveness：只探进程；readiness：探 DB + Redis。
	srv.Route("/").Handle("GET", "/healthz", healthLivenessHandler())
	srv.Route("/").Handle("GET", "/readyz", healthReadinessHandler(dataDB, dataRDB))

	return srv
}

package server

import (
	"cloud-disk/internal/biz"
	"cloud-disk/internal/conf"
	"cloud-disk/internal/service"

	"github.com/go-kratos/kratos/v3/middleware/recovery"
	"github.com/go-kratos/kratos/v3/transport/grpc"
)

// NewGRPCServer 使用给定的配置和服务创建一个新的 gRPC 服务器。
func NewGRPCServer(c *conf.Server, tm *biz.TokenManager, us *service.UserService, fs *service.FileService, rs *service.RecycleService, ss *service.ShareService) *grpc.Server {
	var opts = []grpc.ServerOption{
		grpc.Middleware(
			recovery.Recovery(),
			// 修正点：启用 gRPC 业务时此处追加 JWTAuthMiddleware(tm, whitelist...)，
			// 与 HTTP 共用同一套白名单与校验逻辑，避免 gRPC 入口裸奔。
		),
	}
	if c.Grpc.Network != "" {
		opts = append(opts, grpc.Network(c.Grpc.Network))
	}
	if c.Grpc.Addr != "" {
		opts = append(opts, grpc.Address(c.Grpc.Addr))
	}
	if c.Grpc.Timeout != nil {
		opts = append(opts, grpc.Timeout(c.Grpc.Timeout.AsDuration()))
	}
	srv := grpc.NewServer(opts...)

	// 当前版本只搭骨架，未注册具体 gRPC 服务
	_ = tm
	_ = us
	return srv
}

// 修正点：自定义 context key 类型，避免与其他包裸字符串 key 冲突
type ctxKey string

const (
	ctxKeyUserID   ctxKey = "user_id"
	ctxKeyUsername ctxKey = "username"
	ctxKeyToken    ctxKey = "token"
)

// JWTAuthMiddleware 返回一个验证 JWT token 的中间件。
func JWTAuthMiddleware(tokenManager *biz.TokenManager, whitelist ...string) middleware.Middleware {
	whitelistMap := make(map[string]bool, len(whitelist))
	for _, p := range whitelist {
		whitelistMap[p] = true
	}

	return func(handler middleware.Handler) middleware.Handler {
		return func(ctx context.Context, req interface{}) (interface{}, error) {
			// === 第一步：白名单检查 ===
			if tr, ok := transport.FromServerContext(ctx); ok {
				op := tr.Operation()
				if whitelistMap[op] {
					return handler(ctx, req)
				}
				if ht, ok := tr.(interface{ Request() *http.Request }); ok {
					if whitelistMap[ht.Request().URL.Path] {
						return handler(ctx, req)
					}
				}
			}

			// === 第二步：从 Authorization 头提取 Bearer token ===
			var tokenStr string
			if tr, ok := transport.FromServerContext(ctx); ok {
				header := tr.RequestHeader()
				auth := header.Get("Authorization")
				if strings.HasPrefix(auth, "Bearer ") {
					tokenStr = strings.TrimPrefix(auth, "Bearer ")
				}
			}
			if tokenStr == "" {
				return nil, errors.Unauthorized("MISSING_TOKEN", "缺少认证令牌")
			}

			// === 第三步：校验 token（传入 ctx，便于内部查 Redis/DB）===
			// 修正点：签名 + 过期 + 黑名单 + 版本号，均在这一步完成。
			claims, err := tokenManager.VerifyToken(ctx, tokenStr)
			if err != nil {
				return nil, err
			}

			// === 第四步：用自定义 key 把用户信息注入 context ===
			// 修正点：ctxKey 自定义类型，下游用 CtxUserID/CtxUsername/CtxToken 读取。
			ctx = context.WithValue(ctx, ctxKeyUserID, claims.UserID)
			ctx = context.WithValue(ctx, ctxKeyUsername, claims.Username)
			ctx = context.WithValue(ctx, ctxKeyToken, tokenStr)

			return handler(ctx, req)
		}
	}
}

func CtxUserID(ctx context.Context) uint64 {
	if id, ok := ctx.Value(ctxKeyUserID).(uint64); ok {
		return id
	}
	return 0
}
func CtxUsername(ctx context.Context) string {
	if name, ok := ctx.Value(ctxKeyUsername).(string); ok {
		return name
	}
	return ""
}
func CtxToken(ctx context.Context) string {
	if t, ok := ctx.Value(ctxKeyToken).(string); ok {
		return t
	}
	return ""
}

// recoveryMiddleware 从 panic 中恢复并返回 500 错误。
// 必须放在中间件链的最外层，确保能捕获内层所有 panic。
func recoveryMiddleware() middleware.Middleware {
	return func(handler middleware.Handler) middleware.Handler {
		return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
			// defer + recover 是 Go 标准 panic 捕获模式
			defer func() {
				if r := recover(); r != nil {
					if e, ok := r.(error); ok {
						err = e
					} else {
						// 修正点：对外统一回 InternalServer，不把 r 直接序列化出去
						err = errors.InternalServer("PANIC", "internal server error")
					}
					// 记录 panic 详情，便于事后排查
					log.Error("panic recovered", "err", r)
				}
			}()

			reply, err = handler(ctx, req)
			return
		}
	}
}

// TimingMiddleware 记录请求耗时用于性能监控。
func TimingMiddleware() middleware.Middleware {
	return func(handler middleware.Handler) middleware.Handler {
		return func(ctx context.Context, req interface{}) (interface{}, error) {
			start := time.Now()

			op := "unknown"
			if tr, ok := transport.FromServerContext(ctx); ok {
				op = tr.Operation()
			}

			reply, err := handler(ctx, req)
			duration := time.Since(start)

			// 修正点：高频/健康检查端点可跳过，或改为 metrics 上报
			if op != "/healthz" && op != "/readyz" {
				log.Info("api.timing",
					"op", op,
					"duration_ms", duration.Milliseconds(),
					"err", err,
				)
			}
			return reply, err
		}
	}
}

func newApp(logger *slog.Logger, servers []transport.Server, registrar registry.Registrar) *kratos.App {
	opts := []kratos.Option{
		kratos.ID(id),
		kratos.Name(Name),
		kratos.Version(Version),
		kratos.Metadata(map[string]string{}),
		kratos.Logger(logger),
		kratos.Server(servers...),
	}
	if registrar != nil {
		opts = append(opts, kratos.Registrar(registrar))
	}
	return kratos.New(opts...)
}

func newServers(
	grpcServer *kgrpc.Server,
	httpServer *khttp.Server,
	scheduledTaskServer *scheduler.ScheduledTaskServer,
	consumerServer *event.ConsumerServer,
) []transport.Server {
	return []transport.Server{grpcServer, httpServer, scheduledTaskServer, consumerServer}
}

func main() {
	flag.Parse()

	// === 1. 初始化 logger（结构化日志，slog）===
	logger := log.NewLogger(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
		AddSource: true,
		Level:     slog.LevelInfo,
	})).With(
		slog.String("service.id", id),
		slog.String("service.name", Name),
		slog.String("service.version", Version),
	)
	log.SetDefault(logger)

	// === 2. 加载配置（支持环境变量展开，envsource）===
	c := config.New(
		config.WithSource(
			conf.NewEnvExpandSource(file.NewSource(flagconf)),
		),
	)
	defer c.Close()
	if err := c.Load(); err != nil {
		panic(err)
	}

	var bc conf.Bootstrap
	if err := c.Scan(&bc); err != nil {
		panic(err)
	}

	// === 3. Wire 依赖注入：构造 App + cleanup ===
	app, cleanup, err := wireApp(bc.Server, bc.Data, bc.Auth, bc.Storage, bc.Etcd, logger)
	if err != nil {
		panic(err)
	}
	defer cleanup() // 确保资源释放（即使 panic 也会执行）

	// === 4. 启动并等待停止信号（优雅启停）===
	// 修正点：K8s 部署时 terminationGracePeriodSeconds 需 ≥ 在途请求最长耗时，
	// 否则 kubelet 在优雅窗口结束前发 SIGKILL，在途请求被强杀。
	if err := app.Run(); err != nil {
		panic(err)
	}
}

// NewEtcdClient 根据配置创建 etcd 客户端。
func NewEtcdClient(c *conf.Etcd) (*clientv3.Client, func(), error) {
	// 防御性检查：配置为空或没有 endpoint，直接返回 nil
	if c == nil || len(c.Endpoints) == 0 {
		slog.Warn("etcd 配置为空，跳过 etcd 客户端创建")
		return nil, func() {}, nil
	}
	// ... 构造 clientv3.Config ...
	client, err := clientv3.New(cfg)
	if err != nil {
		return nil, nil, err
	}
	return client, func() { _ = client.Close() }, nil
}

// NewEtcdRegistrar 创建 etcd 服务注册器。client 为 nil 时返回 nil。
func NewEtcdRegistrar(client *clientv3.Client) registry.Registrar {
	if client == nil {
		return nil
	}
	return etcd.New(
		client,
		etcd.Namespace("/microservices"),
		etcd.RegisterTTL(15 * time.Second),
	)
}

// CORSMiddleware 按白名单处理跨域，正确处理 OPTIONS 预检。
// allowedOrigins 来自配置（如 https://cloud.example.com），绝不接受 "*"。
func CORSMiddleware(allowedOrigins ...string) middleware.Middleware {
	allowSet := make(map[string]bool, len(allowedOrigins))
	for _, o := range allowedOrigins {
		allowSet[o] = true
	}
	return func(handler middleware.Handler) middleware.Handler {
		return func(ctx context.Context, req interface{}) (interface{}, error) {
			if tr, ok := transport.FromServerContext(ctx); ok {
				if ht, ok := tr.(interface{ Request() *http.Request }); ok {
					r := ht.Request()
					origin := r.Header.Get("Origin")
					// 同源或没有 Origin（非浏览器请求）直接放行
					if origin == "" || allowSet[origin] {
						if h, ok := tr.(khttp.Transporter); ok {
							h.ReplyHeader().Set("Access-Control-Allow-Origin", origin)
							// 修正点：仅在可信源开启凭据，绝不与 "*" 共存
							if allowSet[origin] {
								h.ReplyHeader().Set("Access-Control-Allow-Credentials", "true")
							}
							h.ReplyHeader().Set("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
							h.ReplyHeader().Set("Access-Control-Allow-Headers", "Authorization,Content-Type")
							h.ReplyHeader().Set("Access-Control-Max-Age", "600")
						}
						// 预检请求直接短路返回 204，不再进入 JWT 等后续中间件
						if r.Method == http.MethodOptions {
							return nil, nil
						}
					}
				}
			}
			return handler(ctx, req)
		}
	}
}

// healthLivenessHandler 存活探针：进程能响应即视为活着。
func healthLivenessHandler() func(ctx khttp.Context) error {
	return func(ctx khttp.Context) error {
		return ctx.Result(http.StatusOK, map[string]string{"status": "ok"})
	}
}

// healthReadinessHandler 就绪探针：探 DB + Redis 真实可用。
func healthReadinessHandler(db *gorm.DB, rdb *redis.Client) func(ctx khttp.Context) error {
	return func(ctx khttp.Context) error {
		// 修正点：readiness 必须探真实依赖，否则"进程活但 DB 挂"仍接流量会大面积失败
		if db != nil {
			sqlDB, _ := db.DB()
			if sqlDB == nil || sqlDB.Ping() != nil {
				return ctx.Result(http.StatusServiceUnavailable, map[string]string{"status": "db unavailable"})
			}
		}
		if rdb != nil {
			if rdb.Ping(ctx.Request().Context()).Err() != nil {
				return ctx.Result(http.StatusServiceUnavailable, map[string]string{"status": "redis unavailable"})
			}
		}
		return ctx.Result(http.StatusOK, map[string]string{"status": "ok"})
	}
}

// fromError 将错误转为统一响应。
// 修正点：非 Kratos 错误不再返回 err.Error()，避免泄露 SQL/堆栈/路径。
func fromError(err error) *Response {
	if err == nil {
		return Success(nil)
	}
	se := errors.FromError(err)
	if se != nil {
		return &Response{Code: int(se.Code), Message: se.Message}
	}
	// 未知错误：统一回内部错误，详情进日志由调用方记录
	return &Response{Code: http.StatusInternalServerError, Message: "internal server error"}
}

// NewTokenManager 生产化：未配置强密钥直接启动失败。
func NewTokenManager(c *conf.Auth) *TokenManager {
	expire := 24 * time.Hour
	refreshExpire := 7 * 24 * time.Hour
	secret := ""
	if c != nil {
		if c.JwtExpire != nil {
			expire = c.JwtExpire.AsDuration()
		}
		if c.JwtRefreshExpire != nil {
			refreshExpire = c.JwtRefreshExpire.AsDuration()
		}
		secret = c.JwtSecret
	}
	// 修正点：绝不回退到硬编码默认值，否则任何人可用默认密钥伪造令牌
	if len(secret) < 32 {
		log.Fatalf("auth.jwt_secret 必须配置且长度≥32，否则拒绝启动（fail-fast）")
	}
	return &TokenManager{
		secret:          []byte(secret),
		expire:          expire,
		refreshExpire:   refreshExpire,
		blacklistPrefix: "blacklist:token:",
	}
}

// NewData 中替换 AutoMigrate 为版本化迁移（示意）。
func NewData(c *conf.Data) (*Data, func(), error) {
	db, err := NewGormDB(c.Database)
	if err != nil {
		return nil, nil, err
	}
	// 修正点：生产用版本化迁移，而非 db.AutoMigrate。
	// 例如在 main 或 CI 中执行：
	//   migrate -path migrations -database "$DSN" up
	// 此处仅做迁移后连接可用性校验。
	if err := db.Exec("SELECT 1").Error; err != nil {
		return nil, nil, fmt.Errorf("db not ready: %w", err)
	}
	// ... redis / kafka 初始化、cleanup ...
}

http {
    # 性能与超时（略）

    # 修正点：安全响应头
    # 1) 移除已废弃的 X-XSS-Protection（现代浏览器已忽略，且可能引入漏洞）
    # 2) 新增 Content-Security-Policy，限制脚本/连接来源
    add_header X-Content-Type-Options nosniff always;
    add_header X-Frame-Options DENY always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
    add_header Content-Security-Policy "default-src 'self'; img-src 'self' data:; object-src 'none'" always;

    server {
        listen 443 ssl http2;
        # 修正点：TLS 在 nginx 终止，并启用 HSTS 强制 HTTPS
        ssl_certificate /etc/nginx/ssl/cert.pem;
        ssl_certificate_key /etc/nginx/ssl/key.pem;
        ssl_protocols TLSv1.2 TLSv1.3;
        add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;

        # 修正点：一般 /api/ 限制请求体（防超大 JSON 打爆内存），上传路由单独放开
        location /api/ {
            client_max_body_size 2m;          # 普通接口 2MB 上限
            proxy_pass http://backend:8000/api/;
            # ... proxy_set_header / 超时（略）...
        }
        # 上传走独立路由，放宽到对象存储单文件上限
        location /api/v1/file/upload {
            client_max_body_size 1024m;       # 与后端/MinIO 上限对齐
            proxy_request_buffering off;      # 流式上传，不缓冲到磁盘
            proxy_pass http://backend:8000/api/v1/file/upload;
        }
    }
}