diff --git a/backend/internal/middleware/auth.go b/backend/internal/middleware/auth.go index 9549ed8..8e54d11 100644 --- a/backend/internal/middleware/auth.go +++ b/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 diff --git a/backend/internal/middleware/rate_limit.go b/backend/internal/middleware/rate_limit.go index a5dfc9f..aec81fe 100644 --- a/backend/internal/middleware/rate_limit.go +++ b/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) { diff --git a/backend/internal/middleware/rate_limit_test.go b/backend/internal/middleware/rate_limit_test.go new file mode 100644 index 0000000..326ce7b --- /dev/null +++ b/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) + } +} diff --git a/backend/internal/middleware/request_logger.go b/backend/internal/middleware/request_logger.go index b1061f7..9a82241 100644 --- a/backend/internal/middleware/request_logger.go +++ b/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 { diff --git a/backend/internal/middleware/request_logger_test.go b/backend/internal/middleware/request_logger_test.go index c368904..05c859b 100644 --- a/backend/internal/middleware/request_logger_test.go +++ b/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) { diff --git a/backend/internal/modules/chathub/handler.go b/backend/internal/modules/chathub/handler.go index 5d9d9a8..242061b 100644 --- a/backend/internal/modules/chathub/handler.go +++ b/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" + } +} diff --git a/backend/internal/modules/chathub/handler_test.go b/backend/internal/modules/chathub/handler_test.go new file mode 100644 index 0000000..c1cd039 --- /dev/null +++ b/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) + } +} diff --git a/backend/internal/modules/chathub/hub.go b/backend/internal/modules/chathub/hub.go index a99e53b..a243ad1 100644 --- a/backend/internal/modules/chathub/hub.go +++ b/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 +} diff --git a/backend/internal/modules/chathub/hub_test.go b/backend/internal/modules/chathub/hub_test.go index 9d701e8..2b0fbfa 100644 --- a/backend/internal/modules/chathub/hub_test.go +++ b/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) } } diff --git a/backend/internal/router/router.go b/backend/internal/router/router.go index 3701623..94166b7 100644 --- a/backend/internal/router/router.go +++ b/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 { diff --git a/backend/migrations/000056_chat_admin_state_and_indexes.sql b/backend/migrations/000056_chat_admin_state_and_indexes.sql index 41ae372..f3c24c1 100644 --- a/backend/migrations/000056_chat_admin_state_and_indexes.sql +++ b/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