package middleware import ( "fmt" "net/http" "time" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/ghttp" "github.com/gogf/gf/v2/os/gtime" "github.com/qianfree/team-api/internal/dao" do "github.com/qianfree/team-api/internal/model/do" "github.com/qianfree/team-api/internal/response" ) const ( IdempotencyHeader = "Idempotency-Key" // maxCachedBodyLen limits the response body size stored for idempotent replay. // Responses larger than this will still be deduplicated but won't cache the body. maxCachedBodyLen = 64 * 1024 // 64 KB // idempotencyTTL is how long an idempotency record is retained before it can be cleaned up. idempotencyTTL = 24 * time.Hour ) // Idempotency ensures that requests with the same Idempotency-Key // are processed only once. Subsequent requests with the same key // return the cached response. // // 幂等键按认证主体(userType + tenantID + userID)隔离:不同租户/用户即便复用同一 // 客户端 Idempotency-Key,也各自独立,避免命中彼此缓存的响应体(跨租户数据泄漏) // 或彼此误触 409。因此本中间件必须挂载在认证中间件(如 TenantAuth)之后。 func Idempotency(r *ghttp.Request) { clientKey := r.GetHeader(IdempotencyHeader) if clientKey == "" { r.Middleware.Next() return } ctx := r.Context() // 组合出按主体隔离的存储键(仍是单一字符串,不改表结构) tenantID := r.GetCtxVar(CtxKeyTenantID).Int64() userID := r.GetCtxVar(CtxKeyUserID).Int64() userType := r.GetCtxVar(CtxKeyUserType).String() // 开放平台不走 TenantAuth,身份存在 OpenPlatformAuth 注入的 typed context key 里, // 上面三个 CtxVar 全是零值。若不单独取,所有应用会共用同一个 ":0:0:" 前缀 —— // 不同租户复用同一个 Idempotency-Key 就会读到彼此缓存的响应体(跨租户数据泄漏)。 // 主体取「应用」而非「用户」:开放平台请求由应用发起,重试也以应用为单位。 if appID := GetOpenAppID(r.Context()); appID > 0 { userType = "open" tenantID = GetOpenTenantID(r.Context()) userID = appID } idempotencyKey := buildIdempotencyKey(userType, tenantID, userID, clientKey) // Check if this key has been processed before var record struct { Id int64 `json:"id" orm:"id"` Status string `json:"status" orm:"status"` ResponseBody string `json:"response_body" orm:"response_body"` } err := dao.SysIdempotencyRecords.Ctx(ctx). Where("idempotency_key", idempotencyKey). Where("expires_at > NOW()"). Scan(&record) if err != nil { // DB error: fail-closed to prevent duplicate processing response.ErrorMsg(r, 500, "系统繁忙,请稍后重试") return } if record.Id > 0 { if record.Status == "completed" && record.ResponseBody != "" { // 重放缓存响应。本中间件只挂载在 /api/* 统一响应端点上:成功响应恒为 // HTTP 200 + JSON(response.Success 使用 WriteJson),业务错误为 HTTP 422 // 且会被标记为 failed 而不进入此重放分支。因此固定回写 200 + JSON Content-Type, // 避免复用未初始化的 r.Response.Status(默认 0)以及丢失 Content-Type。 r.Response.Header().Set("Content-Type", "application/json; charset=utf-8") r.Response.WriteStatus(http.StatusOK) r.Response.Write([]byte(record.ResponseBody)) r.Exit() return } if record.Status == "processing" { writeConflictResponse(r) return } } // Insert processing record _, err = dao.SysIdempotencyRecords.Ctx(ctx).Data(do.SysIdempotencyRecords{ IdempotencyKey: idempotencyKey, Status: "processing", ExpiresAt: gtime.Now().Add(idempotencyTTL), }).Insert() if err != nil { // If insert fails (duplicate key), another request is already processing writeConflictResponse(r) return } // Capture the response r.Middleware.Next() // After processing, update the record with cached body status := "completed" statusCode := r.Response.Status if statusCode >= 400 { status = "failed" } var responseBody string if buf := r.Response.Buffer(); len(buf) > 0 && len(buf) <= maxCachedBodyLen { responseBody = string(buf) } // Update the record dao.SysIdempotencyRecords.Ctx(ctx). Where("idempotency_key", idempotencyKey). Data(do.SysIdempotencyRecords{ Status: status, ResponseBody: responseBody, }).Update() } // buildIdempotencyKey 组合出按认证主体隔离的幂等存储键。 // 加入 userType/tenantID/userID 前缀,使不同主体即便复用同一客户端 key 也互不干扰, // 杜绝跨租户命中彼此缓存响应体(数据泄漏)或误触 409。 func buildIdempotencyKey(userType string, tenantID, userID int64, clientKey string) string { return fmt.Sprintf("%s:%d:%d:%s", userType, tenantID, userID, clientKey) } // writeConflictResponse writes a 409 Conflict response for duplicate idempotency keys. func writeConflictResponse(r *ghttp.Request) { r.Response.WriteStatus(409) r.Response.WriteJson(g.Map{ "error": g.Map{ "type": "conflict", "message": "请求正在处理中,请勿重复提交", "code": "409", }, }) r.Exit() }