150 lines
5.1 KiB
Go
150 lines
5.1 KiB
Go
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()
|
||
}
|