Merge pull request #3659 from jianjianai/fix/ops-realtime-stats-performance

fix: 优化运维实时账号统计接口性能

backend/internal/handler/admin/ops_realtime_handler.go

@@ -1,6 +1,8 @@
 package admin
 
 import (
+	"context"
+	"errors"
 	"net/http"
 	"strconv"
 	"strings"
@@ -47,6 +49,9 @@ func (h *OpsHandler) GetConcurrencyStats(c *gin.Context) {
 
 	platform, group, account, collectedAt, err := h.opsService.GetConcurrencyStats(c.Request.Context(), platformFilter, groupID)
 	if err != nil {
+		if isOpsRealtimeRequestCanceled(c, err) {
+			return
+		}
 		response.ErrorFrom(c, err)
 		return
 	}
@@ -86,6 +91,9 @@ func (h *OpsHandler) GetUserConcurrencyStats(c *gin.Context) {
 
 	users, collectedAt, err := h.opsService.GetUserConcurrencyStats(c.Request.Context())
 	if err != nil {
+		if isOpsRealtimeRequestCanceled(c, err) {
+			return
+		}
 		response.ErrorFrom(c, err)
 		return
 	}
@@ -140,6 +148,9 @@ func (h *OpsHandler) GetAccountAvailability(c *gin.Context) {
 
 	platformStats, groupStats, accountStats, collectedAt, err := h.opsService.GetAccountAvailabilityStats(c.Request.Context(), platform, groupID)
 	if err != nil {
+		if isOpsRealtimeRequestCanceled(c, err) {
+			return
+		}
 		response.ErrorFrom(c, err)
 		return
 	}
@@ -156,6 +167,19 @@ func (h *OpsHandler) GetAccountAvailability(c *gin.Context) {
 	response.Success(c, payload)
 }
 
+func isOpsRealtimeRequestCanceled(c *gin.Context, err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.Canceled) {
+		return true
+	}
+	if c != nil && c.Request != nil && errors.Is(c.Request.Context().Err(), context.Canceled) {
+		return true
+	}
+	return strings.Contains(err.Error(), "canceling statement due to user request")
+}
+
 func parseOpsRealtimeWindow(v string) (time.Duration, string, bool) {
 	switch strings.ToLower(strings.TrimSpace(v)) {
 	case "", "1min", "1m":
@@ -236,6 +260,9 @@ func (h *OpsHandler) GetRealtimeTrafficSummary(c *gin.Context) {
 
 	summary, err := h.opsService.GetRealtimeTrafficSummary(c.Request.Context(), filter)
 	if err != nil {
+		if isOpsRealtimeRequestCanceled(c, err) {
+			return
+		}
 		response.ErrorFrom(c, err)
 		return
 	}

backend/internal/repository/account_repo.go

@@ -603,6 +603,41 @@ func (r *accountRepository) ListWithFilters(ctx context.Context, params paginati
 	return outAccounts, paginationResultFromTotal(int64(total), params), nil
 }
 
+func (r *accountRepository) ListOpsAccountsForStats(ctx context.Context, platformFilter string, groupIDFilter *int64) ([]service.Account, error) {
+	if r == nil || r.client == nil {
+		return []service.Account{}, nil
+	}
+
+	q := r.client.Account.Query()
+	if platformFilter = strings.TrimSpace(platformFilter); platformFilter != "" {
+		q = q.Where(dbaccount.PlatformEQ(platformFilter))
+	}
+	if groupIDFilter != nil && *groupIDFilter > 0 {
+		q = q.Where(dbaccount.HasAccountGroupsWith(dbaccountgroup.GroupIDEQ(*groupIDFilter)))
+	}
+
+	accounts, err := q.
+		Select(
+			dbaccount.FieldID,
+			dbaccount.FieldName,
+			dbaccount.FieldPlatform,
+			dbaccount.FieldConcurrency,
+			dbaccount.FieldLoadFactor,
+			dbaccount.FieldStatus,
+			dbaccount.FieldErrorMessage,
+			dbaccount.FieldSchedulable,
+			dbaccount.FieldRateLimitResetAt,
+			dbaccount.FieldOverloadUntil,
+			dbaccount.FieldTempUnschedulableUntil,
+		).
+		Order(dbent.Asc(dbaccount.FieldID)).
+		All(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return r.accountsToService(ctx, accounts)
+}
+
 func accountListOrder(params pagination.PaginationParams) []func(*entsql.Selector) {
 	sortBy := strings.ToLower(strings.TrimSpace(params.SortBy))
 	sortOrder := params.NormalizedSortOrder(pagination.SortOrderAsc)

backend/internal/service/ops_account_availability.go

@@ -20,7 +20,7 @@ func (s *OpsService) GetAccountAvailabilityStats(ctx context.Context, platformFi
 		return nil, nil, nil, nil, err
 	}
 
-	accounts, err := s.listAllAccountsForOps(ctx, platformFilter)
+	accounts, err := s.listAllAccountsForOps(ctx, platformFilter, groupIDFilter)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}

backend/internal/service/ops_concurrency.go

@@ -13,18 +13,29 @@ const (
 	opsConcurrencyBatchChunkSize = 200
 )
 
-func (s *OpsService) listAllAccountsForOps(ctx context.Context, platformFilter string) ([]Account, error) {
+type opsAccountStatsRepository interface {
+	ListOpsAccountsForStats(ctx context.Context, platformFilter string, groupIDFilter *int64) ([]Account, error)
+}
+
+func (s *OpsService) listAllAccountsForOps(ctx context.Context, platformFilter string, groupIDFilter *int64) ([]Account, error) {
 	if s == nil || s.accountRepo == nil {
 		return []Account{}, nil
 	}
+	if repo, ok := s.accountRepo.(opsAccountStatsRepository); ok {
+		return repo.ListOpsAccountsForStats(ctx, platformFilter, groupIDFilter)
+	}
 
 	out := make([]Account, 0, 128)
 	page := 1
+	groupID := int64(0)
+	if groupIDFilter != nil {
+		groupID = *groupIDFilter
+	}
 	for {
 		accounts, pageInfo, err := s.accountRepo.ListWithFilters(ctx, pagination.PaginationParams{
 			Page:     page,
 			PageSize: opsAccountsPageSize,
-		}, platformFilter, "", "", "", 0, "")
+		}, platformFilter, "", "", "", groupID, "")
 		if err != nil {
 			return nil, err
 		}
@@ -112,7 +123,7 @@ func (s *OpsService) GetConcurrencyStats(
 		return nil, nil, nil, nil, err
 	}
 
-	accounts, err := s.listAllAccountsForOps(ctx, platformFilter)
+	accounts, err := s.listAllAccountsForOps(ctx, platformFilter, groupIDFilter)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}