feat(admin): add egress probe filters and account status details

backend/internal/application/account/build_super_entitlement_test.go

@@ -125,7 +125,7 @@ func TestAccountViewsIncludeBuildBotFlagMetadata(t *testing.T) {
 		t.Fatalf("non-Build risk filter err = %v", err)
 	}
 	summary, err := service.Summary(ctx)
-	if err != nil || summary.Risk != 1 {
+	if err != nil || summary.Risk != 1 || summary.Providers[string(accountdomain.ProviderBuild)].Risk != 1 {
 		t.Fatalf("summary=%#v err=%v", summary, err)
 	}
 }

backend/internal/application/account/service.go

@@ -252,6 +252,9 @@ type Summary struct {
 type ProviderSummary struct {
 	Total     int64
 	Available int64
+	Recovery  RecoverySummary
+	Issues    IssueSummary
+	Risk      int64
 }
 
 type RecoverySummary struct {
@@ -282,7 +285,11 @@ func (s *Service) Summary(ctx context.Context) (Summary, error) {
 		result.Recovery.Probing += row.Probing
 		result.Issues.Disabled += row.Disabled
 		result.Issues.ReauthRequired += row.ReauthRequired
-		result.Providers[row.Provider] = ProviderSummary{Total: row.Total, Available: row.Available}
+		result.Providers[row.Provider] = ProviderSummary{
+			Total: row.Total, Available: row.Available,
+			Recovery: RecoverySummary{Cooldown: row.Cooldown, WaitingReset: row.WaitingReset, Probing: row.Probing},
+			Issues:   IssueSummary{Disabled: row.Disabled, ReauthRequired: row.ReauthRequired},
+		}
 	}
 	result.Recovering = result.Recovery.Cooldown + result.Recovery.WaitingReset + result.Recovery.Probing
 	result.Attention = result.Issues.Disabled + result.Issues.ReauthRequired
@@ -291,6 +298,9 @@ func (s *Service) Summary(ctx context.Context) (Summary, error) {
 		return Summary{}, err
 	}
 	result.Risk = int64(len(flaggedIDs))
+	build := result.Providers[string(accountdomain.ProviderBuild)]
+	build.Risk = result.Risk
+	result.Providers[string(accountdomain.ProviderBuild)] = build
 	return result, nil
 }
 

backend/internal/application/egress/assignment.go

@@ -17,15 +17,30 @@ const (
 )
 
 type RebalanceResult struct {
-	Assigned   int
-	Rebalanced int
-	Unplaced   int
+	Assigned           int
+	Rebalanced         int
+	Overflowed         int
+	Unplaced           int
+	UnplacedByProvider map[accountdomain.Provider]int
 }
 
 // RebalanceAccounts allocates only accounts that are either unbound or
 // explicitly marked auto. Manual bindings are never changed, even when their
 // node is unhealthy or over capacity.
 func (s *Service) RebalanceAccounts(ctx context.Context, autoAssign, autoBalance bool, probeInterval time.Duration) (RebalanceResult, error) {
+	return s.rebalanceAccounts(ctx, autoAssign, autoBalance, probeInterval, false)
+}
+
+// RebalanceAccountsAllowCapacityOverflow uses every eligible node before
+// leaving an automatic binding unplaced. Account capacity remains the primary
+// placement preference; once every eligible finite-capacity node is full, the
+// least-loaded node receives the overflow. This is intended for an operator's
+// immediate recovery action after a proxy outage.
+func (s *Service) RebalanceAccountsAllowCapacityOverflow(ctx context.Context, autoAssign, autoBalance bool, probeInterval time.Duration) (RebalanceResult, error) {
+	return s.rebalanceAccounts(ctx, autoAssign, autoBalance, probeInterval, true)
+}
+
+func (s *Service) rebalanceAccounts(ctx context.Context, autoAssign, autoBalance bool, probeInterval time.Duration, allowCapacityOverflow bool) (RebalanceResult, error) {
 	if s.accounts == nil {
 		return RebalanceResult{}, ErrOperationsUnavailable
 	}
@@ -47,25 +62,37 @@ func (s *Service) RebalanceAccounts(ctx context.Context, autoAssign, autoBalance
 		if err != nil {
 			return result, err
 		}
-		providerResult, providerErr := s.rebalanceProvider(ctx, provider, nodes, autoAssign, autoBalance, probeInterval, now)
+		providerResult, providerErr := s.rebalanceProvider(ctx, provider, nodes, autoAssign, autoBalance, probeInterval, now, allowCapacityOverflow)
 		result.Assigned += providerResult.Assigned
 		result.Rebalanced += providerResult.Rebalanced
+		result.Overflowed += providerResult.Overflowed
 		result.Unplaced += providerResult.Unplaced
+		for unplacedProvider, count := range providerResult.UnplacedByProvider {
+			if result.UnplacedByProvider == nil {
+				result.UnplacedByProvider = make(map[accountdomain.Provider]int)
+			}
+			result.UnplacedByProvider[unplacedProvider] += count
+		}
 		if providerErr != nil {
 			return result, providerErr
 		}
 	}
 	return result, nil
 }
 
-func (s *Service) rebalanceProvider(ctx context.Context, provider accountdomain.Provider, allNodes []domain.Node, autoAssign, autoBalance bool, probeInterval time.Duration, now time.Time) (RebalanceResult, error) {
+func (s *Service) rebalanceProvider(ctx context.Context, provider accountdomain.Provider, allNodes []domain.Node, autoAssign, autoBalance bool, probeInterval time.Duration, now time.Time, allowCapacityOverflow bool) (RebalanceResult, error) {
 	accounts, err := s.accounts.ListEgressAssignments(ctx, provider)
 	if err != nil {
 		return RebalanceResult{}, err
 	}
 	nodes := s.eligibleNodesForProvider(allNodes, provider, probeInterval, now)
 	if len(nodes) == 0 {
-		return RebalanceResult{Unplaced: countAutoAssignable(accounts, autoAssign, autoBalance)}, nil
+		unplaced := countAutoAssignable(accounts, autoAssign, autoBalance)
+		result := RebalanceResult{Unplaced: unplaced}
+		if unplaced > 0 {
+			result.UnplacedByProvider = map[accountdomain.Provider]int{provider: unplaced}
+		}
+		return result, nil
 	}
 	loads := make(map[uint64]int, len(nodes))
 	byID := make(map[uint64]domain.Node, len(nodes))
@@ -96,6 +123,11 @@ func (s *Service) rebalanceProvider(ctx context.Context, provider accountdomain.
 			continue
 		}
 		target, found := leastLoadedNode(nodes, loads)
+		overflowed := false
+		if !found && allowCapacityOverflow {
+			target, found = leastLoadedNodeIgnoringCapacity(nodes, loads)
+			overflowed = found
+		}
 		if !found {
 			result.Unplaced++
 			continue
@@ -108,6 +140,9 @@ func (s *Service) rebalanceProvider(ctx context.Context, provider accountdomain.
 		} else {
 			result.Rebalanced++
 		}
+		if overflowed {
+			result.Overflowed++
+		}
 	}
 
 	// Capacity is a placement constraint, not an optional balancing preference.
@@ -231,6 +266,17 @@ func leastLoadedNodeExcept(values []domain.Node, loads map[uint64]int, excludedI
 	return selected, found
 }
 
+func leastLoadedNodeIgnoringCapacity(values []domain.Node, loads map[uint64]int) (domain.Node, bool) {
+	var selected domain.Node
+	found := false
+	for _, value := range values {
+		if !found || loads[value.ID] < loads[selected.ID] || (loads[value.ID] == loads[selected.ID] && value.ID < selected.ID) {
+			selected, found = value, true
+		}
+	}
+	return selected, found
+}
+
 func overCapacityPair(values []domain.Node, loads map[uint64]int, blocked map[uint64]bool) (domain.Node, domain.Node, bool) {
 	ordered := append([]domain.Node(nil), values...)
 	sort.Slice(ordered, func(i, j int) bool {

backend/internal/application/egress/import_filter.go

@@ -0,0 +1,147 @@
+package egress
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	domain "github.com/chenyme/grok2api/backend/internal/domain/egress"
+)
+
+const (
+	maxSubscriptionImportLatencyMS = 60000
+	maxSubscriptionImportCountries = 50
+)
+
+type screenedSubscriptionEntry struct {
+	Entry subscriptionEntry
+	Probe *domain.ProbeResult
+}
+
+func normalizeSubscriptionImportFilter(input SubscriptionImportFilterInput) (domain.SubscriptionImportFilter, error) {
+	if input.MaxLatencyMS < 0 || input.MaxLatencyMS > maxSubscriptionImportLatencyMS {
+		return domain.SubscriptionImportFilter{}, fmt.Errorf("%w: 最大延迟必须在 0 到 %d 毫秒之间", ErrInvalidInput, maxSubscriptionImportLatencyMS)
+	}
+	if len(input.Countries) > maxSubscriptionImportCountries {
+		return domain.SubscriptionImportFilter{}, fmt.Errorf("%w: 最多选择 %d 个国家或地区", ErrInvalidInput, maxSubscriptionImportCountries)
+	}
+	countries := make(map[string]struct{}, len(input.Countries))
+	for _, raw := range input.Countries {
+		country := strings.ToUpper(strings.TrimSpace(raw))
+		if country == "" {
+			continue
+		}
+		if len(country) != 2 || !isASCIILetter(country[0]) || !isASCIILetter(country[1]) {
+			return domain.SubscriptionImportFilter{}, fmt.Errorf("%w: 国家或地区代码必须为两个字母", ErrInvalidInput)
+		}
+		countries[country] = struct{}{}
+	}
+	if len(countries) > maxSubscriptionImportCountries {
+		return domain.SubscriptionImportFilter{}, fmt.Errorf("%w: 最多选择 %d 个国家或地区", ErrInvalidInput, maxSubscriptionImportCountries)
+	}
+	values := make([]string, 0, len(countries))
+	for country := range countries {
+		values = append(values, country)
+	}
+	sort.Strings(values)
+
+	return domain.SubscriptionImportFilter{MaxLatencyMS: input.MaxLatencyMS, Countries: values}, nil
+}
+
+func isASCIILetter(value byte) bool {
+	return (value >= 'A' && value <= 'Z') || (value >= 'a' && value <= 'z')
+}
+
+func (s *Service) screenSubscriptionEntries(ctx context.Context, entries []subscriptionEntry, filter domain.SubscriptionImportFilter) ([]screenedSubscriptionEntry, int, error) {
+	if !filter.Active() {
+		values := make([]screenedSubscriptionEntry, 0, len(entries))
+		for _, entry := range entries {
+			values = append(values, screenedSubscriptionEntry{Entry: entry})
+		}
+		return values, 0, nil
+	}
+	prober, ok := s.nodeProber().(ProxyProber)
+	if !ok || prober == nil {
+		return nil, 0, ErrOperationsUnavailable
+	}
+	results := make([]screenedSubscriptionEntry, len(entries))
+	accepted := make([]bool, len(entries))
+	jobs := make(chan int)
+	var workers sync.WaitGroup
+	for range min(maxConcurrentProbes, len(entries)) {
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			for index := range jobs {
+				probe, err := prober.ProbeEgressProxy(ctx, entries[index].ProxyURL)
+				if err != nil || probe.Status != domain.ProbeStatusHealthy || !matchesSubscriptionImportFilter(filter, probe) {
+					continue
+				}
+				if probe.TestedAt.IsZero() {
+					probe.TestedAt = time.Now().UTC()
+				}
+				results[index] = screenedSubscriptionEntry{Entry: entries[index], Probe: &probe}
+				accepted[index] = true
+			}
+		}()
+	}
+	for index := range entries {
+		select {
+		case jobs <- index:
+		case <-ctx.Done():
+			close(jobs)
+			workers.Wait()
+			return nil, 0, ctx.Err()
+		}
+	}
+	close(jobs)
+	workers.Wait()
+	if err := ctx.Err(); err != nil {
+		return nil, 0, err
+	}
+	values := make([]screenedSubscriptionEntry, 0, len(entries))
+	for index, accepted := range accepted {
+		if accepted {
+			values = append(values, results[index])
+		}
+	}
+	return values, len(entries) - len(values), nil
+}
+
+func matchesSubscriptionImportFilter(filter domain.SubscriptionImportFilter, probe domain.ProbeResult) bool {
+	if probe.Status != domain.ProbeStatusHealthy {
+		return false
+	}
+	if filter.MaxLatencyMS > 0 && probe.LatencyMS > filter.MaxLatencyMS {
+		return false
+	}
+	if len(filter.Countries) > 0 {
+		country := strings.ToUpper(strings.TrimSpace(probe.ExitCountry))
+		matched := false
+		for _, allowed := range filter.Countries {
+			if allowed == country {
+				matched = true
+				break
+			}
+		}
+		if !matched {
+			return false
+		}
+	}
+	return true
+}
+
+func applyProbeResult(node *domain.Node, probe domain.ProbeResult) {
+	if node == nil {
+		return
+	}
+	node.ProbeStatus = probe.Status
+	node.LastProbedAt = &probe.TestedAt
+	node.ProbeLatencyMS = probe.LatencyMS
+	node.ExitIP = probe.ExitIP
+	node.ExitCountry = probe.ExitCountry
+	node.ProbeError = probe.Error
+}