improve(relay):优化代码

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

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 {

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

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 {

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 {

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)

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)

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)

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)

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)

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)

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"`

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: