fix(chat): 优化后台限流与 SSE 日志

backend/internal/middleware/auth.go

@@ -52,6 +52,18 @@ func extractToken(c *gin.Context) string {
 	return c.Query("token")
 }
 
+func extractAdminToken(c *gin.Context) (string, string) {
+	if tokenText := extractBearerToken(c); tokenText != "" {
+		return tokenText, "bearer"
+	}
+	if cookieToken, err := c.Cookie(AdminAccessCookieName); err == nil {
+		if tokenText := strings.TrimSpace(cookieToken); tokenText != "" {
+			return tokenText, "cookie"
+		}
+	}
+	return "", "none"
+}
+
 func Auth(jwtManager *auth.JWTManager, validate UserTokenValidatorFunc) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		tokenText := extractToken(c)
@@ -87,16 +99,9 @@ func Auth(jwtManager *auth.JWTManager, validate UserTokenValidatorFunc) gin.Hand
 
 func AdminAuth(jwtManager *auth.JWTManager, validate AdminTokenValidatorFunc) gin.HandlerFunc {
 	return func(c *gin.Context) {
-		tokenText := extractBearerToken(c)
-		tokenSource := "bearer"
+		tokenText, tokenSource := extractAdminToken(c)
 		if tokenText == "" {
-			if cookieToken, err := c.Cookie(AdminAccessCookieName); err == nil {
-				tokenText = strings.TrimSpace(cookieToken)
-				tokenSource = "cookie"
-			}
-		}
-		if tokenText == "" {
-			RecordAdminAuthFailure(c, "missing", "none", 0, 0)
+			RecordAdminAuthFailure(c, "missing", tokenSource, 0, 0)
 			response.Unauthorized(c, "缺少后台访问令牌")
 			c.Abort()
 			return

backend/internal/middleware/rate_limit.go

@@ -7,10 +7,14 @@ import (
 	"sync"
 	"time"
 
+	"hfb_sys/backend/internal/modules/auth"
+
 	"github.com/gin-gonic/gin"
 	"github.com/redis/go-redis/v9"
 )
 
+type rateLimitKeyFunc func(c *gin.Context) string
+
 type rateLimitBucket struct {
 	count    int
 	resetAt  time.Time
@@ -32,6 +36,16 @@ type redisRateLimiter struct {
 }
 
 func RateLimitPerMinute(limit int, rdb *redis.Client) gin.HandlerFunc {
+	return rateLimitPerMinute(limit, rdb, ipRateLimitKey)
+}
+
+// AdminAwareRateLimitPerMinute isolates authenticated admin traffic by admin ID.
+// Login and other requests without a valid admin access token fall back to IP.
+func AdminAwareRateLimitPerMinute(limit int, rdb *redis.Client, jwtManager *auth.JWTManager) gin.HandlerFunc {
+	return rateLimitPerMinute(limit, rdb, adminOrIPRateLimitKey(jwtManager))
+}
+
+func rateLimitPerMinute(limit int, rdb *redis.Client, keyFunc rateLimitKeyFunc) gin.HandlerFunc {
 	if limit <= 0 {
 		return func(c *gin.Context) {
 			c.Next()
@@ -48,34 +62,62 @@ func RateLimitPerMinute(limit int, rdb *redis.Client) gin.HandlerFunc {
 			fallback: limiter,
 			limit:    limit,
 			window:   time.Minute,
-		}).handle
+		}).handle(keyFunc)
 	}
-	return limiter.handle
+	return limiter.handle(keyFunc)
 }
 
-func (l *redisRateLimiter) handle(c *gin.Context) {
-	now := time.Now()
-	key := c.ClientIP()
-	allowed, resetAt, err := l.allow(c.Request.Context(), key, now)
-	if err != nil {
-		allowed, resetAt = l.fallback.allow(key, now)
+func (l *redisRateLimiter) handle(keyFunc rateLimitKeyFunc) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		now := time.Now()
+		key := keyFunc(c)
+		allowed, resetAt, err := l.allow(c.Request.Context(), key, now)
+		if err != nil {
+			allowed, resetAt = l.fallback.allow(key, now)
+		}
+		if !allowed {
+			writeRateLimited(c, now, resetAt)
+			return
+		}
+		c.Next()
 	}
-	if !allowed {
-		writeRateLimited(c, now, resetAt)
-		return
-	}
-	c.Next()
 }
 
-func (l *rateLimiter) handle(c *gin.Context) {
-	now := time.Now()
-	key := c.ClientIP()
-	allowed, resetAt := l.allow(key, now)
-	if !allowed {
-		writeRateLimited(c, now, resetAt)
-		return
+func (l *rateLimiter) handle(keyFunc rateLimitKeyFunc) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		now := time.Now()
+		key := keyFunc(c)
+		allowed, resetAt := l.allow(key, now)
+		if !allowed {
+			writeRateLimited(c, now, resetAt)
+			return
+		}
+		c.Next()
+	}
+}
+
+func ipRateLimitKey(c *gin.Context) string {
+	return "ip:" + c.ClientIP()
+}
+
+func adminOrIPRateLimitKey(jwtManager *auth.JWTManager) rateLimitKeyFunc {
+	return func(c *gin.Context) string {
+		if value, ok := c.Get(ContextAdminID); ok {
+			if adminID, ok := value.(uint64); ok && adminID != 0 {
+				return "admin:" + strconv.FormatUint(adminID, 10)
+			}
+		}
+		if jwtManager != nil {
+			tokenText, _ := extractAdminToken(c)
+			if tokenText != "" {
+				claims, err := jwtManager.ParseSubject(tokenText, "access", "admin")
+				if err == nil && claims.UserID != 0 {
+					return "admin:" + strconv.FormatUint(claims.UserID, 10)
+				}
+			}
+		}
+		return ipRateLimitKey(c)
 	}
-	c.Next()
 }
 
 func (l *redisRateLimiter) allow(ctx context.Context, key string, now time.Time) (bool, time.Time, error) {

backend/internal/middleware/rate_limit_test.go

@@ -0,0 +1,91 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"hfb_sys/backend/internal/modules/auth"
+
+	"github.com/gin-gonic/gin"
+)
+
+func runRateLimitedRequest(router *gin.Engine, bearerToken string) int {
+	req := httptest.NewRequest(http.MethodGet, "/ping", nil)
+	if bearerToken != "" {
+		req.Header.Set("Authorization", "Bearer "+bearerToken)
+	}
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, req)
+	return recorder.Code
+}
+
+func newRateLimitRouter(limit int, rdb interface{}, handler gin.HandlerFunc) *gin.Engine {
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	router.Use(handler)
+	router.GET("/ping", func(c *gin.Context) {
+		c.Status(http.StatusOK)
+	})
+	return router
+}
+
+func adminAccessToken(t *testing.T, jwtManager *auth.JWTManager, adminID uint64, phone string) string {
+	t.Helper()
+	pair, err := jwtManager.GenerateSubjectPairWithVersion(adminID, phone, "admin", 0)
+	if err != nil {
+		t.Fatalf("生成后台令牌失败: %v", err)
+	}
+	return pair.AccessToken
+}
+
+func TestAdminAwareRateLimitIsolatesAdminsBehindSharedIP(t *testing.T) {
+	jwtManager := auth.NewJWTManager("test-secret")
+	adminA := adminAccessToken(t, jwtManager, 101, "13800000001")
+	adminB := adminAccessToken(t, jwtManager, 102, "13800000002")
+	router := newRateLimitRouter(2, nil, AdminAwareRateLimitPerMinute(2, nil, jwtManager))
+
+	for i := 0; i < 2; i++ {
+		if code := runRateLimitedRequest(router, adminA); code != http.StatusOK {
+			t.Fatalf("客服A 第 %d 次请求 = %d, want 200", i+1, code)
+		}
+	}
+	if code := runRateLimitedRequest(router, adminA); code != http.StatusTooManyRequests {
+		t.Fatalf("客服A 超额请求 = %d, want 429", code)
+	}
+	if code := runRateLimitedRequest(router, adminB); code != http.StatusOK {
+		t.Fatalf("同出口IP的客服B 应不受客服A影响, got %d, want 200", code)
+	}
+}
+
+func TestAdminAwareRateLimitFallsBackToIP(t *testing.T) {
+	jwtManager := auth.NewJWTManager("test-secret")
+	userPair, err := jwtManager.GenerateSubjectPairWithVersion(201, "13900000001", "user", 0)
+	if err != nil {
+		t.Fatalf("生成用户令牌失败: %v", err)
+	}
+	router := newRateLimitRouter(2, nil, AdminAwareRateLimitPerMinute(2, nil, jwtManager))
+
+	for i := 0; i < 2; i++ {
+		if code := runRateLimitedRequest(router, ""); code != http.StatusOK {
+			t.Fatalf("匿名请求 第 %d 次 = %d, want 200", i+1, code)
+		}
+	}
+	if code := runRateLimitedRequest(router, ""); code != http.StatusTooManyRequests {
+		t.Fatalf("匿名请求超额 = %d, want 429", code)
+	}
+	// 用户令牌无法按 admin 身份解析，应与匿名共享同一个 IP 桶。
+	if code := runRateLimitedRequest(router, userPair.AccessToken); code != http.StatusTooManyRequests {
+		t.Fatalf("用户令牌应回退到IP维度, got %d, want 429", code)
+	}
+}
+
+func TestLegacyRateLimitPerMinuteKeepsIPKey(t *testing.T) {
+	router := newRateLimitRouter(1, nil, RateLimitPerMinute(1, nil))
+	if code := runRateLimitedRequest(router, ""); code != http.StatusOK {
+		t.Fatalf("首次匿名请求 = %d, want 200", code)
+	}
+	if code := runRateLimitedRequest(router, ""); code != http.StatusTooManyRequests {
+		t.Fatalf("第二次匿名请求 = %d, want 429", code)
+	}
+}

backend/internal/middleware/request_logger.go

@@ -108,16 +108,25 @@ func isPaymentNotifyPath(path string) bool {
 }
 
 // shouldSkipHTTPLog 判断是否为无需记录的普通请求。
-func shouldSkipHTTPLog(_, _ string, status int, latencyMs float64) bool {
+func shouldSkipHTTPLog(path, route string, status int, latencyMs float64) bool {
 	if status >= 500 || status == http.StatusTooManyRequests {
 		return false
 	}
+	// SSE 的请求耗时就是连接存活时间，由 chathub 单独记录连接生命周期。
+	if isChatSSEPath(path, route) {
+		return true
+	}
 	if latencyMs >= slowRequestThresholdMs {
 		return false
 	}
 	return true
 }
 
+func isChatSSEPath(path, route string) bool {
+	return path == "/api/chats/events" || path == "/api/admin/chats/events" ||
+		route == "/api/chats/events" || route == "/api/admin/chats/events"
+}
+
 func meaningfulAuthFailure(c *gin.Context) bool {
 	value, ok := c.Get(ContextAuthFailureReason)
 	if !ok {

backend/internal/middleware/request_logger_test.go

@@ -19,6 +19,10 @@ func TestShouldSkipHTTPLog(t *testing.T) {
 		{name: "轮询 500 不跳过", path: "/api/wallet/balance", status: 500, latencyMs: 1, wantSkip: false},
 		{name: "限流请求不跳过", path: "/api/auth/sms", status: 429, latencyMs: 1, wantSkip: false},
 		{name: "慢请求不跳过", path: "/api/orders", status: 200, latencyMs: 500, wantSkip: false},
+		{name: "用户 SSE 长连接跳过", path: "/api/chats/events", route: "/api/chats/events", status: 200, latencyMs: 60_000, wantSkip: true},
+		{name: "后台 SSE 长连接跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 200, latencyMs: 600_000, wantSkip: true},
+		{name: "SSE 服务端错误不跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 500, latencyMs: 600_000, wantSkip: false},
+		{name: "SSE 限流不跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 429, latencyMs: 1, wantSkip: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

backend/internal/modules/chathub/handler.go

@@ -1,6 +1,7 @@
 package chathub
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -8,16 +9,21 @@ import (
 	"hfb_sys/backend/pkg/response"
 
 	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
 )
 
 const heartbeatInterval = 30 * time.Second
 
 type Handler struct {
-	hub *Hub
+	hub    *Hub
+	logger *zap.Logger
 }
 
-func NewHandler(hub *Hub) *Handler {
-	return &Handler{hub: hub}
+func NewHandler(hub *Hub, logger *zap.Logger) *Handler {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	return &Handler{hub: hub, logger: logger}
 }
 
 // UserEvents 处理用户端 SSE 连接: GET /api/chats/events
@@ -51,8 +57,23 @@ func (h *Handler) AdminEvents(c *gin.Context) {
 }
 
 func (h *Handler) serveSSE(c *gin.Context, pType string, pID uint64) {
+	startedAt := time.Now()
+	disconnectReason := "handler_completed"
+	var disconnectErr error
 	ch := h.hub.Subscribe(pType, pID)
-	defer h.hub.Unsubscribe(pType, pID, ch)
+	h.logger.Info("SSE 连接建立", h.connectionLogFields(c, pType, pID)...)
+	defer func() {
+		h.hub.Unsubscribe(pType, pID, ch)
+		fields := h.connectionLogFields(c, pType, pID)
+		fields = append(fields,
+			zap.String("disconnect_reason", disconnectReason),
+			zap.Float64("lifetime_ms", float64(time.Since(startedAt).Microseconds())/1000),
+		)
+		if disconnectErr != nil {
+			fields = append(fields, zap.Error(disconnectErr))
+		}
+		h.logger.Info("SSE 连接断开", fields...)
+	}()
 
 	c.Header("Content-Type", "text/event-stream")
 	c.Header("Cache-Control", "no-cache")
@@ -60,7 +81,11 @@ func (h *Handler) serveSSE(c *gin.Context, pType string, pID uint64) {
 	c.Header("X-Accel-Buffering", "no")
 
 	// 发送初始连接确认
-	fmt.Fprintf(c.Writer, "event: connected\ndata: {\"ok\":true}\n\n")
+	if _, err := fmt.Fprintf(c.Writer, "event: connected\ndata: {\"ok\":true}\n\n"); err != nil {
+		disconnectReason = "initial_write_error"
+		disconnectErr = err
+		return
+	}
 	c.Writer.Flush()
 
 	heartbeat := time.NewTicker(heartbeatInterval)
@@ -71,17 +96,55 @@ func (h *Handler) serveSSE(c *gin.Context, pType string, pID uint64) {
 	for {
 		select {
 		case <-clientGone:
+			disconnectReason = contextDisconnectReason(c.Request.Context().Err())
 			return
 		case <-heartbeat.C:
-			fmt.Fprintf(c.Writer, ":heartbeat\n\n")
+			if _, err := fmt.Fprintf(c.Writer, ":heartbeat\n\n"); err != nil {
+				disconnectReason = "heartbeat_write_error"
+				disconnectErr = err
+				return
+			}
 			c.Writer.Flush()
 		case event, ok := <-ch:
 			if !ok {
+				disconnectReason = "subscription_closed"
 				return
 			}
 			data := MarshalEvent(event)
-			fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event.Type, data)
+			if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event.Type, data); err != nil {
+				disconnectReason = "event_write_error"
+				disconnectErr = err
+				return
+			}
 			c.Writer.Flush()
 		}
 	}
 }
+
+func (h *Handler) connectionLogFields(c *gin.Context, pType string, pID uint64) []zap.Field {
+	fields := []zap.Field{
+		zap.String("request_id", middleware.GetRequestID(c)),
+		zap.String("principal_type", pType),
+		zap.Uint64("principal_id", pID),
+		zap.String("client_ip", c.ClientIP()),
+		zap.Int("active_sse_connections", h.hub.OnlineCount()),
+		zap.Int("active_principal_type_connections", h.hub.OnlineCountByType(pType)),
+	}
+	if pType == "admin" {
+		fields = append(fields, zap.Uint64("admin_id", pID))
+	} else {
+		fields = append(fields, zap.Uint64("user_id", pID))
+	}
+	return fields
+}
+
+func contextDisconnectReason(err error) string {
+	switch err {
+	case context.Canceled:
+		return "context_canceled"
+	case context.DeadlineExceeded:
+		return "context_deadline_exceeded"
+	default:
+		return "context_closed"
+	}
+}

backend/internal/modules/chathub/handler_test.go

@@ -0,0 +1,60 @@
+package chathub
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"hfb_sys/backend/internal/middleware"
+
+	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
+)
+
+func TestSSELifecycleLogsConnectionCountReasonAndLifetime(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	core, observed := observer.New(zap.InfoLevel)
+	hub := NewHub(nil)
+	handler := NewHandler(hub, zap.New(core))
+
+	requestContext, cancel := context.WithCancel(context.Background())
+	cancel()
+	request := httptest.NewRequest(http.MethodGet, "/api/admin/chats/events", nil).WithContext(requestContext)
+	request.RemoteAddr = "42.49.131.69:34567"
+	recorder := httptest.NewRecorder()
+	ginContext, _ := gin.CreateTestContext(recorder)
+	ginContext.Request = request
+	ginContext.Set(middleware.ContextAdminID, uint64(88))
+	ginContext.Set(middleware.ContextRequestID, "sse-test-request")
+
+	handler.AdminEvents(ginContext)
+
+	connected := observed.FilterMessage("SSE 连接建立").All()
+	if len(connected) != 1 {
+		t.Fatalf("connection logs = %d, want 1", len(connected))
+	}
+	connectedFields := connected[0].ContextMap()
+	if connectedFields["admin_id"] != uint64(88) || connectedFields["active_sse_connections"] != int64(1) {
+		t.Fatalf("unexpected connection fields: %v", connectedFields)
+	}
+
+	disconnected := observed.FilterMessage("SSE 连接断开").All()
+	if len(disconnected) != 1 {
+		t.Fatalf("disconnection logs = %d, want 1", len(disconnected))
+	}
+	disconnectedFields := disconnected[0].ContextMap()
+	if disconnectedFields["disconnect_reason"] != "context_canceled" {
+		t.Fatalf("disconnect reason = %v, want context_canceled", disconnectedFields["disconnect_reason"])
+	}
+	if disconnectedFields["active_sse_connections"] != int64(0) {
+		t.Fatalf("active connections after disconnect = %v, want 0", disconnectedFields["active_sse_connections"])
+	}
+	if _, ok := disconnectedFields["lifetime_ms"]; !ok {
+		t.Fatalf("missing lifetime_ms field: %v", disconnectedFields)
+	}
+	if got := hub.OnlineCount(); got != 0 {
+		t.Fatalf("online connection count = %d, want 0", got)
+	}
+}

backend/internal/modules/chathub/hub.go

@@ -190,3 +190,16 @@ func (h *Hub) OnlineCount() int {
 	}
 	return count
 }
+
+// OnlineCountByType 返回当前进程内指定主体类型的 SSE 连接数。
+func (h *Hub) OnlineCountByType(pType string) int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	count := 0
+	for key, clients := range h.clients {
+		if key.Type == pType {
+			count += len(clients)
+		}
+	}
+	return count
+}

backend/internal/modules/chathub/hub_test.go

@@ -7,6 +7,7 @@ func TestDisconnectUserClosesAllConnections(t *testing.T) {
 	first := hub.Subscribe("user", 7)
 	second := hub.Subscribe("user", 7)
 	hub.Subscribe("user", 8)
+	hub.Subscribe("admin", 9)
 
 	hub.DisconnectUser(7)
 	for name, ch := range map[string]<-chan *ChatEvent{"first": first, "second": second} {
@@ -19,7 +20,13 @@ func TestDisconnectUserClosesAllConnections(t *testing.T) {
 			t.Fatalf("%s connection did not close immediately", name)
 		}
 	}
-	if got := hub.OnlineCount(); got != 1 {
-		t.Fatalf("online connection count = %d, want 1", got)
+	if got := hub.OnlineCount(); got != 2 {
+		t.Fatalf("online connection count = %d, want 2", got)
+	}
+	if got := hub.OnlineCountByType("user"); got != 1 {
+		t.Fatalf("online user connection count = %d, want 1", got)
+	}
+	if got := hub.OnlineCountByType("admin"); got != 1 {
+		t.Fatalf("online admin connection count = %d, want 1", got)
 	}
 }

backend/internal/router/router.go

@@ -60,6 +60,7 @@ func New(cfg config.Config, deps Dependencies, logger *zap.Logger) *gin.Engine {
 	engine.Use(middleware.RequestID())
 	engine.Use(middleware.RequestLogger(logger))
 	engine.Use(middleware.Recovery(logger))
+	jwtManager := auth.NewJWTManager(cfg.JWTSecret)
 
 	health := handler.NewHealthHandler()
 	engine.GET("/health", health.Check)
@@ -69,7 +70,7 @@ func New(cfg config.Config, deps Dependencies, logger *zap.Logger) *gin.Engine {
 		engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
 	}
 	if cfg.RateLimit.Enabled {
-		engine.Use(middleware.RateLimitPerMinute(cfg.RateLimit.RequestsPerMinute, deps.Redis))
+		engine.Use(middleware.AdminAwareRateLimitPerMinute(cfg.RateLimit.RequestsPerMinute, deps.Redis, jwtManager))
 	}
 
 	// 业务字段加密器（实名/收款账号）：主密钥 + legacy 回退，用于密钥轮换期间透明解出旧密文。
@@ -112,7 +113,6 @@ func New(cfg config.Config, deps Dependencies, logger *zap.Logger) *gin.Engine {
 		logger.Debug("开发环境使用模拟业务字段加密器")
 	}
 
-	jwtManager := auth.NewJWTManager(cfg.JWTSecret)
 	var userRepo *auth.UserRepository
 	if deps.DB != nil {
 		userRepo = auth.NewUserRepository(deps.DB)
@@ -285,7 +285,7 @@ func New(cfg config.Config, deps Dependencies, logger *zap.Logger) *gin.Engine {
 	chatHandler := chat.NewHandler(chatService)
 	var chatHubHandler *chathub.Handler
 	if chatHub != nil {
-		chatHubHandler = chathub.NewHandler(chatHub)
+		chatHubHandler = chathub.NewHandler(chatHub, logger)
 	}
 	var disputeRepo *dispute.Repository
 	if deps.DB != nil {

backend/migrations/000056_chat_admin_state_and_indexes.sql

@@ -35,11 +35,14 @@ WHERE cp.participant_type = 'admin'
   AND (cp.remark <> '' OR cp.last_read_at IS NOT NULL)
 ON DUPLICATE KEY UPDATE
   remark = VALUES(remark),
-  last_read_message_id = GREATEST(last_read_message_id, VALUES(last_read_message_id)),
+  last_read_message_id = GREATEST(
+    chat_admin_conversation_states.last_read_message_id,
+    VALUES(last_read_message_id)
+  ),
   last_read_at = CASE
-    WHEN last_read_at IS NULL THEN VALUES(last_read_at)
-    WHEN VALUES(last_read_at) IS NULL THEN last_read_at
-    ELSE GREATEST(last_read_at, VALUES(last_read_at))
+    WHEN chat_admin_conversation_states.last_read_at IS NULL THEN VALUES(last_read_at)
+    WHEN VALUES(last_read_at) IS NULL THEN chat_admin_conversation_states.last_read_at
+    ELSE GREATEST(chat_admin_conversation_states.last_read_at, VALUES(last_read_at))
   END;
 
 ALTER TABLE chat_conversations