From 3f2ef6046844554aca25a85314b2052cbdf43486 Mon Sep 17 00:00:00 2001 From: jianjian Date: Thu, 2 Jul 2026 20:04:47 +0000 Subject: [PATCH] fix: optimize ops realtime account stats --- .../handler/admin/ops_realtime_handler.go | 27 ++++++++++++++ backend/internal/repository/account_repo.go | 35 +++++++++++++++++++ .../service/ops_account_availability.go | 2 +- backend/internal/service/ops_concurrency.go | 17 +++++++-- 4 files changed, 77 insertions(+), 4 deletions(-) diff --git a/backend/internal/handler/admin/ops_realtime_handler.go b/backend/internal/handler/admin/ops_realtime_handler.go index c175dcd09c..ae95026329 100644 --- a/backend/internal/handler/admin/ops_realtime_handler.go +++ b/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 } diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 2a2f5e0576..19a1ff0485 100644 --- a/backend/internal/repository/account_repo.go +++ b/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) diff --git a/backend/internal/service/ops_account_availability.go b/backend/internal/service/ops_account_availability.go index da66ec4dd7..3c874aae4d 100644 --- a/backend/internal/service/ops_account_availability.go +++ b/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 } diff --git a/backend/internal/service/ops_concurrency.go b/backend/internal/service/ops_concurrency.go index 69b513af83..8837078232 100644 --- a/backend/internal/service/ops_concurrency.go +++ b/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 }