refactor: improve system stability, performance, and code structure

This commit introduces several architectural improvements and optimizations across the codebase: - **Performance & Reliability**: - Implemented Redis pipelining in `ConversationCache.CacheMessage` to reduce network round-trips. - Added a circuit breaker to the JPush client to prevent cascading failures. - Introduced batch deletion and batch member addition capabilities in repositories. - Added message idempotency support using `client_msg_id` and a Redis-based cache. - Optimized WebSocket handling with connection limits (total and per-user) and improved error logging. - **Code Refactoring**: - Refactored `Router` to use a `RouterDeps` struct, simplifying the constructor and improving maintainability. - Unified model ID generation logic using new `id_helper.go` (supporting UUID and Snowflake). - Standardized JSON serialization/deserialization in models using `json_helper.go`. - Refactored DTO conversion logic, specifically for `UserResponse` (using functional options) and `Report` responses. - Removed redundant/deprecated DTOs like `PostDetailResponse` and `TradeItemDetailResponse`. - **Cache Improvements**: - Enhanced `LayeredCache` with `SetRaw` to avoid double-encoding when promoting values from Redis to local cache. - Added `DeleteBatch` support to the cache interface. - **Other Changes**: - Cleaned up `config.go` by removing redundant default values and explicit environment variable overrides. - Improved WebSocket registration flow to handle connection limits gracefully.
2026-05-04 13:07:03 +08:00
parent b2b55ea52d
commit ee78071d4d
65 changed files with 1293 additions and 975 deletions
--- a/internal/pkg/circuitbreaker/circuitbreaker.go
+++ b/internal/pkg/circuitbreaker/circuitbreaker.go
@@ -0,0 +1,136 @@
+package circuitbreaker
+
+import (
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+type State int
+
+const (
+	StateClosed   State = iota
+	StateOpen
+	StateHalfOpen
+)
+
+type Config struct {
+	FailureThreshold int           // 连续失败次数阈值，超过则开启熔断
+	SuccessThreshold int           // 半开状态下连续成功次数，恢复到关闭
+	Timeout          time.Duration // 熔断开启后等待多久进入半开
+	Name             string       // 熔断器名称，用于日志
+}
+
+func DefaultConfig(name string) Config {
+	return Config{
+		FailureThreshold: 5,
+		SuccessThreshold: 3,
+		Timeout:          30 * time.Second,
+		Name:             name,
+	}
+}
+
+type Breaker struct {
+	cfg    Config
+	mu     sync.Mutex
+	state  State
+	fails  int
+	wins   int
+	openAt time.Time
+}
+
+func New(cfg Config) *Breaker {
+	return &Breaker{cfg: cfg}
+}
+
+func (b *Breaker) Allow() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	switch b.state {
+	case StateClosed:
+		return true
+	case StateOpen:
+		if time.Since(b.openAt) > b.cfg.Timeout {
+			b.state = StateHalfOpen
+			b.wins = 0
+			return true
+		}
+		return false
+	case StateHalfOpen:
+		return true
+	default:
+		return true
+	}
+}
+
+func (b *Breaker) RecordSuccess() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.fails = 0
+	if b.state == StateHalfOpen {
+		b.wins++
+		if b.wins >= b.cfg.SuccessThreshold {
+			b.state = StateClosed
+			b.wins = 0
+			zap.L().Info("circuit breaker closed",
+				zap.String("breaker", b.cfg.Name),
+			)
+		}
+	}
+}
+
+func (b *Breaker) RecordFailure() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.fails++
+	if b.state == StateHalfOpen {
+		b.state = StateOpen
+		b.openAt = time.Now()
+		b.wins = 0
+		zap.L().Warn("circuit breaker re-opened from half-open",
+			zap.String("breaker", b.cfg.Name),
+		)
+		return
+	}
+	if b.fails >= b.cfg.FailureThreshold {
+		b.state = StateOpen
+		b.openAt = time.Now()
+		zap.L().Warn("circuit breaker opened",
+			zap.String("breaker", b.cfg.Name),
+			zap.Int("consecutive_failures", b.fails),
+		)
+	}
+}
+
+func (b *Breaker) State() State {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.state
+}
+
+func (b *Breaker) Execute(fn func() error) error {
+	if !b.Allow() {
+		zap.L().Debug("circuit breaker rejected request",
+			zap.String("breaker", b.cfg.Name),
+		)
+		return &OpenError{Name: b.cfg.Name}
+	}
+	if err := fn(); err != nil {
+		b.RecordFailure()
+		return err
+	}
+	b.RecordSuccess()
+	return nil
+}
+
+type OpenError struct {
+	Name string
+}
+
+func (e *OpenError) Error() string {
+	return "circuit breaker '" + e.Name + "' is open"
+}
--- a/internal/pkg/jpush/client.go
+++ b/internal/pkg/jpush/client.go
@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"time"

+	"with_you/internal/pkg/circuitbreaker"
+
 	"go.uber.org/zap"
 )

@@ -29,6 +31,7 @@ type Client struct {
 	httpClient    *http.Client
 	logger        *zap.Logger
 	lastRateLimit *RateLimitInfo
+	breaker       *circuitbreaker.Breaker
 }

 func NewClient(appKey, masterSecret string, production bool, logger *zap.Logger) *Client {
@@ -48,6 +51,12 @@ func NewClient(appKey, masterSecret string, production bool, logger *zap.Logger)
 		production:   production,
 		httpClient:   &http.Client{Timeout: 30 * time.Second, Transport: transport},
 		logger:       logger,
+		breaker: circuitbreaker.New(circuitbreaker.Config{
+			FailureThreshold: 5,
+			SuccessThreshold: 3,
+			Timeout:          30 * time.Second,
+			Name:             "jpush",
+		}),
 	}
 }

@@ -259,7 +268,15 @@ func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *N
 		},
 	}

-	result, _, err := c.Push(payload)
+	var result *PushResponse
+	err := c.breaker.Execute(func() error {
+		resp, _, pushErr := c.Push(payload)
+		if pushErr != nil {
+			return pushErr
+		}
+		result = resp
+		return nil
+	})
 	if err != nil {
 		c.logger.Error("jpush push by registration_ids failed",
 			zap.Int("count", len(registrationIDs)),
--- a/internal/pkg/response/response.go
+++ b/internal/pkg/response/response.go
@@ -15,13 +15,6 @@ type Response struct {
 	Data    any    `json:"data,omitempty"`
 }

-// ResponseSnakeCase 统一响应结构(snake_case)
-type ResponseSnakeCase struct {
-	Code    int    `json:"code"`
-	Message string `json:"message"`
-	Data    any    `json:"data,omitempty"`
-}
-
 // Success 成功响应
 func Success(c *gin.Context, data any) {
 	c.JSON(http.StatusOK, Response{
--- a/internal/pkg/ws/hub.go
+++ b/internal/pkg/ws/hub.go
@@ -2,6 +2,7 @@ package ws

 import (
 	"encoding/json"
+	"errors"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -10,8 +11,12 @@ import (
 )

 const (
-	defaultUserBufferSize = 128
-	maxReplayEvents       = 200
+	defaultUserBufferSize    = 128
+	maxReplayEvents          = 200
+	groupPushConcurrency     = 64
+	groupPushQueueSize       = 512
+	maxTotalConnections      = 100000
+	maxConnectionsPerUser    = 10
 )

 // Event 表示一个WebSocket事件
@@ -31,6 +36,9 @@ type Client struct {
 	Quit   chan struct{}
 }

+// ErrConnectionLimit 连接数限制错误
+var ErrConnectionLimit = errors.New("connection limit reached")
+
 // Message 表示WebSocket消息
 type Message struct {
 	Type    string          `json:"type"`
@@ -73,14 +81,15 @@ type ConnectHandler func(userID string, isFirstConnection bool)

 // Hub WebSocket连接管理器
 type Hub struct {
-	seq uint64
+	seq       uint64
+	connCount atomic.Int64

 	mu                 sync.RWMutex
-	clients            map[string]map[uint64]*Client     // userID -> clientID -> Client
-	history            map[string][]Event                // userID -> []Event (用于断线重连历史回放)
-	subscribers        map[string]map[uint64]*subscriber // userID -> subscriberID -> subscriber
-	disconnectHandlers []DisconnectHandler               // 断开连接事件处理器列表
-	connectHandlers    []ConnectHandler                  // 连接事件处理器列表
+	clients            map[string]map[uint64]*Client
+	history            map[string][]Event
+	subscribers        map[string]map[uint64]*subscriber
+	disconnectHandlers []DisconnectHandler
+	connectHandlers    []ConnectHandler
 }

 // NewHub 创建WebSocket Hub
@@ -99,11 +108,19 @@ func (h *Hub) NextID() uint64 {
 	return atomic.AddUint64(&h.seq, 1)
 }

-// Register 注册客户端连接
-func (h *Hub) Register(client *Client) []Event {
+// Register 注册客户端连接，返回回放事件和可能的错误（连接数超限）
+func (h *Hub) Register(client *Client) ([]Event, error) {
+	if h.connCount.Load() >= int64(maxTotalConnections) {
+		return nil, ErrConnectionLimit
+	}
+
 	h.mu.Lock()

-	// 检查是否是该用户的第一个连接（之前完全离线）
+	if len(h.clients[client.UserID]) >= maxConnectionsPerUser {
+		h.mu.Unlock()
+		return nil, ErrConnectionLimit
+	}
+
 	_, existed := h.clients[client.UserID]
 	isFirstConnection := !existed

@@ -111,8 +128,8 @@ func (h *Hub) Register(client *Client) []Event {
 		h.clients[client.UserID] = make(map[uint64]*Client)
 	}
 	h.clients[client.UserID][client.ID] = client
+	h.connCount.Add(1)

-	// 获取历史事件用于回放
 	replay := make([]Event, 0)
 	for _, e := range h.history[client.UserID] {
 		replay = append(replay, e)
@@ -122,15 +139,14 @@ func (h *Hub) Register(client *Client) []Event {
 		zap.String("user_id", client.UserID),
 		zap.Uint64("client_id", client.ID),
 		zap.Int("total_clients", len(h.clients[client.UserID])),
+		zap.Int64("total_connections", h.connCount.Load()),
 		zap.Bool("is_first_connection", isFirstConnection),
 	)

-	// 复制处理器列表避免锁竞争
 	handlers := make([]ConnectHandler, len(h.connectHandlers))
 	copy(handlers, h.connectHandlers)
 	h.mu.Unlock()

-	// 异步调用连接处理器，避免阻塞 Register
 	if len(handlers) > 0 {
 		go func() {
 			for _, handler := range handlers {
@@ -139,7 +155,7 @@ func (h *Hub) Register(client *Client) []Event {
 		}()
 	}

-	return replay
+	return replay, nil
 }

 // Unregister 注销客户端连接
@@ -151,6 +167,7 @@ func (h *Hub) Unregister(client *Client) {
 		if _, exists := userClients[client.ID]; exists {
 			delete(userClients, client.ID)
 			close(client.Quit)
+			h.connCount.Add(-1)
 			remainingCount = len(userClients)

 			if remainingCount == 0 {
@@ -187,6 +204,20 @@ func (h *Hub) HasClients(userID string) bool {
 	return len(h.clients[userID]) > 0
 }

+// FilterOnline 批量检查用户在线状态，一次加锁返回在线/离线列表
+func (h *Hub) FilterOnline(userIDs []string) (online []string, offline []string) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for _, uid := range userIDs {
+		if len(h.clients[uid]) > 0 {
+			online = append(online, uid)
+		} else {
+			offline = append(offline, uid)
+		}
+	}
+	return
+}
+
 // OnDisconnect 注册断开连接事件处理器
 // 当用户断开 WebSocket 连接时，会调用所有注册的处理器
 // 处理器会收到用户ID和该用户剩余的连接数
@@ -224,10 +255,65 @@ func (h *Hub) PublishToUser(userID string, eventType string, payload any) Event
 	return ev
 }

-// PublishToUsers 向多个用户发送事件
+// PublishToUsers 向多个用户发送事件（共享一次序列化）
 func (h *Hub) PublishToUsers(userIDs []string, eventType string, payload any) {
+	if len(userIDs) == 0 {
+		return
+	}
+
+	ev := Event{
+		ID:      h.NextID(),
+		Type:    eventType,
+		TS:      time.Now().UnixMilli(),
+		Payload: payload,
+	}
+
+	msg := ResponseMessage{
+		EventID: ev.ID,
+		Type:    eventType,
+		TS:      ev.TS,
+		Payload: payload,
+	}
+	data, err := json.Marshal(msg)
+	if err != nil {
+		zap.L().Error("Failed to marshal WebSocket message",
+			zap.String("event_type", eventType),
+			zap.Error(err),
+		)
+		return
+	}
+
 	for _, uid := range userIDs {
-		h.PublishToUser(uid, eventType, payload)
+		h.publishPreMarshaled(uid, ev, data)
+	}
+}
+
+// publishPreMarshaled 使用预序列化的数据发送事件
+func (h *Hub) publishPreMarshaled(userID string, ev Event, data []byte) {
+	h.mu.Lock()
+	history := append(h.history[userID], ev)
+	if len(history) > maxReplayEvents {
+		history = history[len(history)-maxReplayEvents:]
+	}
+	h.history[userID] = history
+
+	targets := make([]*Client, 0, len(h.clients[userID]))
+	for _, c := range h.clients[userID] {
+		targets = append(targets, c)
+	}
+	h.mu.Unlock()
+
+	for _, c := range targets {
+		select {
+		case <-c.Quit:
+		case c.Send <- data:
+		default:
+			zap.L().Warn("WebSocket client buffer full, closing slow client",
+				zap.String("user_id", userID),
+				zap.Uint64("client_id", c.ID),
+			)
+			close(c.Quit)
+		}
 	}
 }

@@ -338,19 +424,16 @@ func (h *Hub) publish(userID string, ev Event) {
 		return
 	}

-	// 非阻塞发送，慢消费者丢弃消息
 	for _, c := range targets {
 		select {
 		case <-c.Quit:
-			// 客户端已断开
 		case c.Send <- data:
-			// 消息已发送
 		default:
-			// 发送缓冲区满，丢弃消息
-			zap.L().Warn("WebSocket client buffer full, dropping message",
+			zap.L().Warn("WebSocket client buffer full, closing slow client",
 				zap.String("user_id", userID),
 				zap.Uint64("client_id", c.ID),
 			)
+			close(c.Quit)
 		}
 	}
 }
@@ -387,6 +470,81 @@ func (h *Hub) Broadcast(eventType string, payload any) {
 	h.PublishToUsers(userIDs, eventType, payload)
 }

+// PublishToUsersConcurrent 并发推送消息给多个用户（适用于群聊等大列表场景）
+// 使用有界并发控制，避免 goroutine 爆炸
+func (h *Hub) PublishToUsersConcurrent(userIDs []string, eventType string, payload any) {
+	if len(userIDs) == 0 {
+		return
+	}
+
+	ev := Event{
+		ID:      h.NextID(),
+		Type:    eventType,
+		TS:      time.Now().UnixMilli(),
+		Payload: payload,
+	}
+
+	msg := ResponseMessage{
+		EventID: ev.ID,
+		Type:    eventType,
+		TS:      ev.TS,
+		Payload: payload,
+	}
+	data, err := json.Marshal(msg)
+	if err != nil {
+		zap.L().Error("Failed to marshal WebSocket message for batch push",
+			zap.String("event_type", eventType),
+			zap.Error(err),
+		)
+		return
+	}
+
+	sem := make(chan struct{}, groupPushConcurrency)
+	var wg sync.WaitGroup
+	wg.Add(len(userIDs))
+
+	for _, uid := range userIDs {
+		sem <- struct{}{}
+		go func(userID string) {
+			defer wg.Done()
+			defer func() { <-sem }()
+
+			h.mu.RLock()
+			targets := make([]*Client, 0, len(h.clients[userID]))
+			for _, c := range h.clients[userID] {
+				targets = append(targets, c)
+			}
+			h.mu.RUnlock()
+
+			for _, c := range targets {
+				select {
+				case <-c.Quit:
+				case c.Send <- data:
+				default:
+					zap.L().Warn("WebSocket client buffer full, closing slow client",
+						zap.String("user_id", userID),
+						zap.Uint64("client_id", c.ID),
+					)
+					close(c.Quit)
+				}
+			}
+		}(uid)
+	}
+
+	wg.Wait()
+
+	// Store history for each user for reconnection replay
+	h.mu.Lock()
+	for _, uid := range userIDs {
+		history := append(h.history[uid], ev)
+		if len(history) > maxReplayEvents {
+			history = history[len(history)-maxReplayEvents:]
+		}
+		h.history[uid] = history
+	}
+	h.mu.Unlock()
+}
+
 // GetOnlineUsers 获取在线用户列表
 func (h *Hub) GetOnlineUsers() []string {
 	h.mu.RLock()
@@ -406,6 +564,29 @@ func (h *Hub) GetOnlineCount() int {
 	return len(h.clients)
 }

+// GetConnectionCount 获取总连接数
+func (h *Hub) GetConnectionCount() int64 {
+	return h.connCount.Load()
+}
+
+// CloseAllConnections 关闭所有WebSocket连接（用于优雅关闭）
+func (h *Hub) CloseAllConnections() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	for _, userClients := range h.clients {
+		for _, c := range userClients {
+			select {
+			case <-c.Quit:
+			default:
+				close(c.Quit)
+			}
+		}
+	}
+	h.clients = make(map[string]map[uint64]*Client)
+	h.connCount.Store(0)
+}
+
 // Subscribe 订阅事件
 func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Event) {
 	subID := h.NextID()