diff --git a/internal/logic/common/verify_code.go b/internal/logic/common/verify_code.go index 6a8ce27..33863ae 100644 --- a/internal/logic/common/verify_code.go +++ b/internal/logic/common/verify_code.go @@ -50,9 +50,13 @@ func SendVerifyCode(ctx context.Context, email string, purpose VerifyPurpose) er if err == nil && hourlyCount.Int() > 5 { return NewBusinessError(consts.CodeRateLimitExceeded, "每小时最多发送5次验证码") } - // Always ensure TTL is set (idempotent: EXPIRE resets existing TTL or sets missing one) - if _, err := g.Redis().Do(ctx, "EXPIRE", hourlyKey, 3600); err != nil { - g.Log().Warningf(ctx, "failed to set hourly rate limit TTL: %v", err) + // Set the 1-hour window TTL only when it is missing. This keeps a fixed + // window (not a sliding one) while still repairing an orphaned key left by + // a crash between INCR and EXPIRE (TTL < 0 means no expiry or key absent). + if ttl, err := g.Redis().Do(ctx, "TTL", hourlyKey); err == nil && ttl.Int() < 0 { + if _, err := g.Redis().Do(ctx, "EXPIRE", hourlyKey, 3600); err != nil { + g.Log().Warningf(ctx, "failed to set hourly rate limit TTL: %v", err) + } } // Generate 6-digit code diff --git a/internal/middleware/idempotency.go b/internal/middleware/idempotency.go index e5ef29d..05dc9b6 100644 --- a/internal/middleware/idempotency.go +++ b/internal/middleware/idempotency.go @@ -1,8 +1,12 @@ package middleware import ( + "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" @@ -14,6 +18,8 @@ const ( // 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 @@ -47,7 +53,12 @@ func Idempotency(r *ghttp.Request) { if record.Id > 0 { if record.Status == "completed" && record.ResponseBody != "" { - r.Response.WriteStatus(r.Response.Status) + // 重放缓存响应。本中间件只挂载在 /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 @@ -62,6 +73,7 @@ func Idempotency(r *ghttp.Request) { _, err = dao.SysIdempotencyRecords.Ctx(ctx).Data(do.SysIdempotencyRecords{ IdempotencyKey: idempotencyKey, Status: "processing", + ExpiresAt: gtime.Now().Add(idempotencyTTL), }).Insert() if err != nil { diff --git a/relay/channel/claude/response.go b/relay/channel/claude/response.go index 1a91c83..37295a7 100644 --- a/relay/channel/claude/response.go +++ b/relay/channel/claude/response.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/common" @@ -64,8 +63,8 @@ func (a *Adaptor) handleStreamToOpenAI(ctx context.Context, resp *http.Response, } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) @@ -382,8 +381,8 @@ func (a *Adaptor) handleClaudeNativeStream(ctx context.Context, resp *http.Respo } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() reader := bufio.NewReaderSize(resp.Body, 64*1024) var usage dto.ClaudeUsage diff --git a/relay/channel/coze/adaptor.go b/relay/channel/coze/adaptor.go index 06d41ca..1e2e811 100644 --- a/relay/channel/coze/adaptor.go +++ b/relay/channel/coze/adaptor.go @@ -9,7 +9,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/channel/openai" @@ -131,8 +130,8 @@ func (a *Adaptor) DoResponse(ctx context.Context, resp *http.Response, info *com // handleStreamResponse 流式模式:逐事件读取 Coze SSE,转换为 OpenAI SSE 格式输出 func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() streamStatus := info.StreamStatus if streamStatus == nil { diff --git a/relay/channel/dify/adaptor.go b/relay/channel/dify/adaptor.go index 9330c6c..038ab22 100644 --- a/relay/channel/dify/adaptor.go +++ b/relay/channel/dify/adaptor.go @@ -9,7 +9,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/channel/openai" @@ -186,8 +185,8 @@ func (a *Adaptor) handleNonStreamResponse(ctx context.Context, resp *http.Respon // handleStreamResponse 处理 Dify streaming 模式 SSE 响应 func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() streamStatus := info.StreamStatus if streamStatus == nil { diff --git a/relay/channel/gemini/response.go b/relay/channel/gemini/response.go index dba371f..18c41de 100644 --- a/relay/channel/gemini/response.go +++ b/relay/channel/gemini/response.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/common" @@ -205,8 +204,8 @@ func (a *Adaptor) handleStreamToOpenAI(ctx context.Context, resp *http.Response, } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() isCA := a.isCodeAssistActive() scanner := bufio.NewScanner(resp.Body) diff --git a/relay/channel/ollama/adaptor.go b/relay/channel/ollama/adaptor.go index 11b9026..56ceddf 100644 --- a/relay/channel/ollama/adaptor.go +++ b/relay/channel/ollama/adaptor.go @@ -9,7 +9,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/channel/openai" @@ -214,8 +213,8 @@ func (a *Adaptor) handleChatStreamResponse(ctx context.Context, resp *http.Respo } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) @@ -362,8 +361,8 @@ func (a *Adaptor) handleGenerateStreamResponse(ctx context.Context, resp *http.R } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) diff --git a/relay/channel/openai/claude_response.go b/relay/channel/openai/claude_response.go index 6785cc5..f60586c 100644 --- a/relay/channel/openai/claude_response.go +++ b/relay/channel/openai/claude_response.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/common" @@ -79,8 +78,8 @@ func handleClaudeInboundStream(ctx context.Context, resp *http.Response, info *c } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) diff --git a/relay/channel/openai/gemini_response.go b/relay/channel/openai/gemini_response.go index 652c5d0..1bb7318 100644 --- a/relay/channel/openai/gemini_response.go +++ b/relay/channel/openai/gemini_response.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/common" @@ -64,8 +63,8 @@ func handleGeminiInboundStream(ctx context.Context, resp *http.Response, info *c } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) diff --git a/relay/channel/openai/responses.go b/relay/channel/openai/responses.go index d5fa995..e3f494d 100644 --- a/relay/channel/openai/responses.go +++ b/relay/channel/openai/responses.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/qianfree/team-api/relay/common" @@ -171,8 +170,8 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R } helper.SetEventStreamHeaders(writer) - var writeMu sync.Mutex - defer helper.PingTicker(writer, 15*time.Second, &writeMu)() + writer = helper.NewSafeWriter(writer) + defer helper.PingTicker(writer, 15*time.Second)() scanner := bufio.NewScanner(resp.Body) buf := make([]byte, 0, 64*1024) diff --git a/relay/channel/tencent/adaptor.go b/relay/channel/tencent/adaptor.go index 9d2355a..6bea976 100644 --- a/relay/channel/tencent/adaptor.go +++ b/relay/channel/tencent/adaptor.go @@ -146,14 +146,14 @@ func (a *Adaptor) DoRequest(ctx context.Context, info *common.RelayInfo, request override.MergeHeaderOverrides(httpReq.Header, hdrOverrides) } + // 时间戳始终设置(代理网关场景上游也可能要求该头);签名仅在持有 secretKey 时计算。 + timestamp := time.Now().Unix() + httpReq.Header.Set("X-TC-Timestamp", fmt.Sprintf("%d", timestamp)) + // 计算 TC3-HMAC-SHA256 签名(需要完整请求体) if a.secretKey != "" { contentType := httpReq.Header.Get("Content-Type") host := httpReq.Header.Get("Host") - timestamp := time.Now().Unix() - - // 更新时间戳(确保签名和头一致) - httpReq.Header.Set("X-TC-Timestamp", fmt.Sprintf("%d", timestamp)) authorization := sign(a.secretID, a.secretKey, serviceName, host, contentType, bodyBytes, timestamp) httpReq.Header.Set("Authorization", authorization) diff --git a/relay/dto/openai_responses.go b/relay/dto/openai_responses.go index 0a6ac43..c1951c2 100644 --- a/relay/dto/openai_responses.go +++ b/relay/dto/openai_responses.go @@ -145,8 +145,11 @@ type ResponsesOutput struct { CallID string `json:"call_id,omitempty"` Name string `json:"name,omitempty"` Arguments string `json:"arguments,omitempty"` - // web_search_call - Action *ResponsesWebSearchAction `json:"action,omitempty"` + // action 字段为多态:web_search_call / shell_call / local_shell_call / apply_patch_call + // 各类型的 action 结构不同(见 ResponsesWebSearchAction / ResponsesShellAction / + // ResponsesPatchAction)。此处用 RawMessage 原样透传,避免多个字段共用同一 json tag + // 导致 encoding/json 在序列化和反序列化时静默丢弃全部 action 字段。 + Action json.RawMessage `json:"action,omitempty"` // file_search_call Queries []string `json:"queries,omitempty"` Results []ResponsesFileSearchResult `json:"results,omitempty"` @@ -161,10 +164,6 @@ type ResponsesOutput struct { Code string `json:"code,omitempty"` ContainerID string `json:"container_id,omitempty"` Outputs []ResponsesCodeInterpreterOutput `json:"outputs,omitempty"` - // shell_call / local_shell_call - ShellAction *ResponsesShellAction `json:"action,omitempty"` - // apply_patch_call - PatchAction *ResponsesPatchAction `json:"action,omitempty"` // mcp_call ServerLabel string `json:"server_label,omitempty"` Output string `json:"output,omitempty"` diff --git a/relay/helper/stream.go b/relay/helper/stream.go index 20c0c8a..9e8a0a6 100644 --- a/relay/helper/stream.go +++ b/relay/helper/stream.go @@ -116,10 +116,55 @@ func EstimateTokens(text string) int { return (len(text) + 3) / 4 } +// SafeWriter 包装 http.ResponseWriter,用互斥锁串行化 Write/WriteHeader/Flush, +// 使保活 ping goroutine 与主循环可以安全地并发写同一个 ResponseWriter。 +// +// 注意:fmt.Fprintf 对单个 SSE 帧只产生一次 Write 调用,因此每个 SSE 帧在锁内是原子写入的; +// Flush 单独加锁,与其他写入交错也不会破坏帧的字节完整性。 +type SafeWriter struct { + w http.ResponseWriter + mu sync.Mutex +} + +// NewSafeWriter 创建一个并发安全的 ResponseWriter 包装器。 +func NewSafeWriter(w http.ResponseWriter) *SafeWriter { + return &SafeWriter{w: w} +} + +func (s *SafeWriter) Header() http.Header { + return s.w.Header() +} + +func (s *SafeWriter) WriteHeader(statusCode int) { + s.mu.Lock() + defer s.mu.Unlock() + s.w.WriteHeader(statusCode) +} + +func (s *SafeWriter) Write(b []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.w.Write(b) +} + +// Flush 实现 http.Flusher,刷新底层 writer(若支持)。 +func (s *SafeWriter) Flush() { + s.mu.Lock() + defer s.mu.Unlock() + if f, ok := s.w.(http.Flusher); ok { + f.Flush() + } +} + +var ( + _ http.ResponseWriter = (*SafeWriter)(nil) + _ http.Flusher = (*SafeWriter)(nil) +) + // PingTicker 在后台定期发送 SSE 保活注释 -// mu 用于与主循环的写操作串行化,避免并发写 ResponseWriter +// 调用方必须传入并发安全的 writer(如 SafeWriter),以避免与主循环并发写 ResponseWriter // 返回一个 stop 函数用于停止 goroutine -func PingTicker(w http.ResponseWriter, interval time.Duration, mu *sync.Mutex) (stop func()) { +func PingTicker(w http.ResponseWriter, interval time.Duration) (stop func()) { ticker := time.NewTicker(interval) done := make(chan struct{}) @@ -128,10 +173,7 @@ func PingTicker(w http.ResponseWriter, interval time.Duration, mu *sync.Mutex) ( for { select { case <-ticker.C: - mu.Lock() - err := WriteSSEPing(w) - mu.Unlock() - if err != nil { + if err := WriteSSEPing(w); err != nil { return } case <-done: