mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
refactor(service): 纯移动拆分 admin_service.go(4409→642行)
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,965 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dbent "github.com/Wei-Shaw/sub2api/ent"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/antigravity"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/claude"
|
||||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/geminicli"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/openai"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
|
||||
)
|
||||
|
||||
// Group management implementations
|
||||
func (s *adminServiceImpl) ListGroups(ctx context.Context, page, pageSize int, platform, status, search string, isExclusive *bool, sortBy, sortOrder string) ([]Group, int64, error) {
|
||||
params := pagination.PaginationParams{Page: page, PageSize: pageSize, SortBy: sortBy, SortOrder: sortOrder}
|
||||
groups, result, err := s.groupRepo.ListWithFilters(ctx, params, platform, status, search, isExclusive)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return groups, result.Total, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetAllGroups(ctx context.Context) ([]Group, error) {
|
||||
return s.groupRepo.ListActive(ctx)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetAllGroupsByPlatform(ctx context.Context, platform string) ([]Group, error) {
|
||||
return s.groupRepo.ListActiveByPlatform(ctx, platform)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetAllGroupsIncludingInactive(ctx context.Context) ([]Group, error) {
|
||||
// ListWithFilters with empty status = no status filter, so active + disabled groups are returned.
|
||||
// PageSize 10000 is intentionally large; group count is O(dozens) in practice.
|
||||
groups, _, err := s.groupRepo.ListWithFilters(ctx, pagination.PaginationParams{Page: 1, PageSize: 10000}, "", "", "", nil)
|
||||
return groups, err
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetGroup(ctx context.Context, id int64) (*Group, error) {
|
||||
return s.groupRepo.GetByID(ctx, id)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetGroupModelsListCandidates(ctx context.Context, id int64, platform string) ([]string, error) {
|
||||
platform = strings.TrimSpace(platform)
|
||||
if id > 0 {
|
||||
group, err := s.groupRepo.GetByIDLite(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if platform == "" {
|
||||
platform = group.Platform
|
||||
}
|
||||
}
|
||||
if platform == "" {
|
||||
platform = PlatformAnthropic
|
||||
}
|
||||
|
||||
candidates := defaultModelsListCandidateIDs(platform)
|
||||
if id <= 0 || s.accountRepo == nil {
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
accounts, err := s.accountRepo.ListSchedulableByGroupID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
seen := make(map[string]struct{}, len(candidates))
|
||||
for _, model := range candidates {
|
||||
seen[model] = struct{}{}
|
||||
}
|
||||
for _, acc := range accounts {
|
||||
if acc.Platform != platform {
|
||||
continue
|
||||
}
|
||||
for model := range acc.GetModelMapping() {
|
||||
model = strings.TrimSpace(model)
|
||||
if model == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[model]; ok {
|
||||
continue
|
||||
}
|
||||
seen[model] = struct{}{}
|
||||
candidates = append(candidates, model)
|
||||
}
|
||||
}
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
func defaultModelsListCandidateIDs(platform string) []string {
|
||||
switch platform {
|
||||
case PlatformOpenAI:
|
||||
return openai.DefaultModelIDs()
|
||||
case PlatformGemini:
|
||||
ids := make([]string, 0, len(geminicli.DefaultModels))
|
||||
for _, model := range geminicli.DefaultModels {
|
||||
ids = append(ids, model.ID)
|
||||
}
|
||||
return ids
|
||||
case PlatformAntigravity:
|
||||
models := antigravity.DefaultModels()
|
||||
ids := make([]string, 0, len(models))
|
||||
for _, model := range models {
|
||||
ids = append(ids, model.ID)
|
||||
}
|
||||
return ids
|
||||
case PlatformGrok:
|
||||
return xai.DefaultModelIDs()
|
||||
default:
|
||||
ids := make([]string, 0, len(claude.DefaultModels))
|
||||
for _, model := range claude.DefaultModels {
|
||||
ids = append(ids, model.ID)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
}
|
||||
|
||||
func defaultAllowImageGenerationForPlatform(platform string) bool {
|
||||
// Grok image and video generation routes share the legacy image-generation gate.
|
||||
// Older clients send the false zero value, so Grok groups must default enabled.
|
||||
return platform == PlatformGrok
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) CreateGroup(ctx context.Context, input *CreateGroupInput) (*Group, error) {
|
||||
if input.RateMultiplier <= 0 {
|
||||
return nil, errors.New("rate_multiplier must be > 0")
|
||||
}
|
||||
|
||||
platform := input.Platform
|
||||
if platform == "" {
|
||||
platform = PlatformAnthropic
|
||||
}
|
||||
|
||||
subscriptionType := input.SubscriptionType
|
||||
if subscriptionType == "" {
|
||||
subscriptionType = SubscriptionTypeStandard
|
||||
}
|
||||
|
||||
// 限额字段:nil/负数 表示"无限制",0 表示"不允许用量",正数表示具体限额
|
||||
dailyLimit := normalizeLimit(input.DailyLimitUSD)
|
||||
weeklyLimit := normalizeLimit(input.WeeklyLimitUSD)
|
||||
monthlyLimit := normalizeLimit(input.MonthlyLimitUSD)
|
||||
|
||||
// 图片价格:负数表示清除(使用默认价格),0 保留(表示免费)
|
||||
imagePrice1K := normalizePrice(input.ImagePrice1K)
|
||||
imagePrice2K := normalizePrice(input.ImagePrice2K)
|
||||
imagePrice4K := normalizePrice(input.ImagePrice4K)
|
||||
imageRateMultiplier := 1.0
|
||||
if input.ImageRateMultiplier != nil {
|
||||
if *input.ImageRateMultiplier < 0 {
|
||||
return nil, errors.New("image_rate_multiplier must be >= 0")
|
||||
}
|
||||
imageRateMultiplier = *input.ImageRateMultiplier
|
||||
}
|
||||
batchImageDiscountMultiplier := defaultBatchImageDiscountMultiplier
|
||||
if input.BatchImageDiscountMultiplier != nil {
|
||||
if *input.BatchImageDiscountMultiplier < 0 {
|
||||
return nil, errors.New("batch_image_discount_multiplier must be >= 0")
|
||||
}
|
||||
batchImageDiscountMultiplier = *input.BatchImageDiscountMultiplier
|
||||
}
|
||||
batchImageHoldMultiplier := defaultBatchImageHoldMultiplier
|
||||
if input.BatchImageHoldMultiplier != nil {
|
||||
if *input.BatchImageHoldMultiplier < 0 {
|
||||
return nil, errors.New("batch_image_hold_multiplier must be >= 0")
|
||||
}
|
||||
batchImageHoldMultiplier = *input.BatchImageHoldMultiplier
|
||||
}
|
||||
// 不变式:hold 比例 >= discount 比例。否则批量任务成功率足够高时
|
||||
// 实际成本会超过冻结额,结算永远失败、用户冻结余额无法解冻。
|
||||
if batchImageHoldMultiplier < batchImageDiscountMultiplier {
|
||||
return nil, errors.New("batch_image_hold_multiplier must be >= batch_image_discount_multiplier")
|
||||
}
|
||||
|
||||
peakRateMultiplier := 1.0
|
||||
if input.PeakRateMultiplier != nil {
|
||||
peakRateMultiplier = *input.PeakRateMultiplier
|
||||
}
|
||||
// 先归一化(非订阅分组清空高峰配置、清洗停用状态下的脏字段)再校验,与 UpdateGroup 同一收口。
|
||||
peakRateEnabled, peakStart, peakEnd, peakRateMultiplier := NormalizePeakRateConfig(subscriptionType, input.PeakRateEnabled, input.PeakStart, input.PeakEnd, peakRateMultiplier)
|
||||
if err := ValidatePeakRateConfig(subscriptionType, peakRateEnabled, peakStart, peakEnd, peakRateMultiplier); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 校验降级分组
|
||||
if input.FallbackGroupID != nil {
|
||||
if err := s.validateFallbackGroup(ctx, 0, *input.FallbackGroupID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
fallbackOnInvalidRequest := input.FallbackGroupIDOnInvalidRequest
|
||||
if fallbackOnInvalidRequest != nil && *fallbackOnInvalidRequest <= 0 {
|
||||
fallbackOnInvalidRequest = nil
|
||||
}
|
||||
// 校验无效请求兜底分组
|
||||
if fallbackOnInvalidRequest != nil {
|
||||
if err := s.validateFallbackGroupOnInvalidRequest(ctx, 0, platform, subscriptionType, *fallbackOnInvalidRequest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// MCPXMLInject:默认为 true,仅当显式传入 false 时关闭
|
||||
mcpXMLInject := true
|
||||
if input.MCPXMLInject != nil {
|
||||
mcpXMLInject = *input.MCPXMLInject
|
||||
}
|
||||
|
||||
allowImageGeneration := input.AllowImageGeneration || defaultAllowImageGenerationForPlatform(platform)
|
||||
allowBatchImageGeneration := input.AllowBatchImageGeneration && allowImageGeneration && platform == PlatformGemini
|
||||
|
||||
// 如果指定了复制账号的源分组,先获取账号 ID 列表
|
||||
var accountIDsToCopy []int64
|
||||
if len(input.CopyAccountsFromGroupIDs) > 0 {
|
||||
// 去重源分组 IDs
|
||||
seen := make(map[int64]struct{})
|
||||
uniqueSourceGroupIDs := make([]int64, 0, len(input.CopyAccountsFromGroupIDs))
|
||||
for _, srcGroupID := range input.CopyAccountsFromGroupIDs {
|
||||
if _, exists := seen[srcGroupID]; !exists {
|
||||
seen[srcGroupID] = struct{}{}
|
||||
uniqueSourceGroupIDs = append(uniqueSourceGroupIDs, srcGroupID)
|
||||
}
|
||||
}
|
||||
|
||||
// 校验源分组的平台是否与新分组一致
|
||||
for _, srcGroupID := range uniqueSourceGroupIDs {
|
||||
srcGroup, err := s.groupRepo.GetByIDLite(ctx, srcGroupID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("source group %d not found: %w", srcGroupID, err)
|
||||
}
|
||||
if srcGroup.Platform != platform {
|
||||
return nil, fmt.Errorf("source group %d platform mismatch: expected %s, got %s", srcGroupID, platform, srcGroup.Platform)
|
||||
}
|
||||
}
|
||||
|
||||
// 获取所有源分组的账号(去重)
|
||||
var err error
|
||||
accountIDsToCopy, err = s.groupRepo.GetAccountIDsByGroupIDs(ctx, uniqueSourceGroupIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get accounts from source groups: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
group := &Group{
|
||||
Name: input.Name,
|
||||
Description: input.Description,
|
||||
Platform: platform,
|
||||
RateMultiplier: input.RateMultiplier,
|
||||
IsExclusive: input.IsExclusive,
|
||||
Status: StatusActive,
|
||||
SubscriptionType: subscriptionType,
|
||||
DailyLimitUSD: dailyLimit,
|
||||
WeeklyLimitUSD: weeklyLimit,
|
||||
MonthlyLimitUSD: monthlyLimit,
|
||||
AllowImageGeneration: allowImageGeneration,
|
||||
AllowBatchImageGeneration: allowBatchImageGeneration,
|
||||
ImageRateIndependent: input.ImageRateIndependent,
|
||||
ImageRateMultiplier: imageRateMultiplier,
|
||||
BatchImageDiscountMultiplier: batchImageDiscountMultiplier,
|
||||
BatchImageHoldMultiplier: batchImageHoldMultiplier,
|
||||
PeakRateEnabled: peakRateEnabled,
|
||||
PeakStart: peakStart,
|
||||
PeakEnd: peakEnd,
|
||||
PeakRateMultiplier: peakRateMultiplier,
|
||||
ImagePrice1K: imagePrice1K,
|
||||
ImagePrice2K: imagePrice2K,
|
||||
ImagePrice4K: imagePrice4K,
|
||||
ClaudeCodeOnly: input.ClaudeCodeOnly,
|
||||
FallbackGroupID: input.FallbackGroupID,
|
||||
FallbackGroupIDOnInvalidRequest: fallbackOnInvalidRequest,
|
||||
ModelRouting: input.ModelRouting,
|
||||
MCPXMLInject: mcpXMLInject,
|
||||
SupportedModelScopes: input.SupportedModelScopes,
|
||||
AllowMessagesDispatch: input.AllowMessagesDispatch,
|
||||
RequireOAuthOnly: input.RequireOAuthOnly,
|
||||
RequirePrivacySet: input.RequirePrivacySet,
|
||||
DefaultMappedModel: input.DefaultMappedModel,
|
||||
MessagesDispatchModelConfig: normalizeOpenAIMessagesDispatchModelConfig(input.MessagesDispatchModelConfig),
|
||||
ModelsListConfig: normalizeGroupModelsListConfig(input.ModelsListConfig),
|
||||
RPMLimit: input.RPMLimit,
|
||||
}
|
||||
sanitizeGroupMessagesDispatchFields(group)
|
||||
if err := s.groupRepo.Create(ctx, group); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// require_oauth_only: 过滤掉 apikey 类型账号
|
||||
if group.RequireOAuthOnly && (group.Platform == PlatformOpenAI || group.Platform == PlatformAntigravity || group.Platform == PlatformAnthropic || group.Platform == PlatformGemini || group.Platform == PlatformGrok) && len(accountIDsToCopy) > 0 {
|
||||
accounts, err := s.accountRepo.GetByIDs(ctx, accountIDsToCopy)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch accounts for oauth filter: %w", err)
|
||||
}
|
||||
oauthIDs := make(map[int64]struct{}, len(accounts))
|
||||
for _, acc := range accounts {
|
||||
if acc.Type != AccountTypeAPIKey {
|
||||
oauthIDs[acc.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
var filtered []int64
|
||||
for _, aid := range accountIDsToCopy {
|
||||
if _, ok := oauthIDs[aid]; ok {
|
||||
filtered = append(filtered, aid)
|
||||
}
|
||||
}
|
||||
accountIDsToCopy = filtered
|
||||
}
|
||||
|
||||
// 如果有需要复制的账号,绑定到新分组
|
||||
if len(accountIDsToCopy) > 0 {
|
||||
if err := s.groupRepo.BindAccountsToGroup(ctx, group.ID, accountIDsToCopy); err != nil {
|
||||
return nil, fmt.Errorf("failed to bind accounts to new group: %w", err)
|
||||
}
|
||||
group.AccountCount = int64(len(accountIDsToCopy))
|
||||
}
|
||||
|
||||
return group, nil
|
||||
}
|
||||
|
||||
// normalizeLimit 将负数转换为 nil(表示无限制),0 保留(表示限额为零)
|
||||
func normalizeLimit(limit *float64) *float64 {
|
||||
if limit == nil || *limit < 0 {
|
||||
return nil
|
||||
}
|
||||
return limit
|
||||
}
|
||||
|
||||
// normalizePrice 将负数转换为 nil(表示使用默认价格),0 保留(表示免费)
|
||||
func normalizePrice(price *float64) *float64 {
|
||||
if price == nil || *price < 0 {
|
||||
return nil
|
||||
}
|
||||
return price
|
||||
}
|
||||
|
||||
// validateFallbackGroup 校验降级分组的有效性
|
||||
// currentGroupID: 当前分组 ID(新建时为 0)
|
||||
// fallbackGroupID: 降级分组 ID
|
||||
func (s *adminServiceImpl) validateFallbackGroup(ctx context.Context, currentGroupID, fallbackGroupID int64) error {
|
||||
// 不能将自己设置为降级分组
|
||||
if currentGroupID > 0 && currentGroupID == fallbackGroupID {
|
||||
return fmt.Errorf("cannot set self as fallback group")
|
||||
}
|
||||
|
||||
visited := map[int64]struct{}{}
|
||||
nextID := fallbackGroupID
|
||||
for {
|
||||
if _, seen := visited[nextID]; seen {
|
||||
return fmt.Errorf("fallback group cycle detected")
|
||||
}
|
||||
visited[nextID] = struct{}{}
|
||||
if currentGroupID > 0 && nextID == currentGroupID {
|
||||
return fmt.Errorf("fallback group cycle detected")
|
||||
}
|
||||
|
||||
// 检查降级分组是否存在
|
||||
fallbackGroup, err := s.groupRepo.GetByIDLite(ctx, nextID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("fallback group not found: %w", err)
|
||||
}
|
||||
|
||||
// 降级分组不能启用 claude_code_only,否则会造成死循环
|
||||
if nextID == fallbackGroupID && fallbackGroup.ClaudeCodeOnly {
|
||||
return fmt.Errorf("fallback group cannot have claude_code_only enabled")
|
||||
}
|
||||
|
||||
if fallbackGroup.FallbackGroupID == nil {
|
||||
return nil
|
||||
}
|
||||
nextID = *fallbackGroup.FallbackGroupID
|
||||
}
|
||||
}
|
||||
|
||||
// validateFallbackGroupOnInvalidRequest 校验无效请求兜底分组的有效性
|
||||
// currentGroupID: 当前分组 ID(新建时为 0)
|
||||
// platform/subscriptionType: 当前分组的有效平台/订阅类型
|
||||
// fallbackGroupID: 兜底分组 ID
|
||||
func (s *adminServiceImpl) validateFallbackGroupOnInvalidRequest(ctx context.Context, currentGroupID int64, platform, subscriptionType string, fallbackGroupID int64) error {
|
||||
if platform != PlatformAnthropic && platform != PlatformAntigravity {
|
||||
return fmt.Errorf("invalid request fallback only supported for anthropic or antigravity groups")
|
||||
}
|
||||
if subscriptionType == SubscriptionTypeSubscription {
|
||||
return fmt.Errorf("subscription groups cannot set invalid request fallback")
|
||||
}
|
||||
if currentGroupID > 0 && currentGroupID == fallbackGroupID {
|
||||
return fmt.Errorf("cannot set self as invalid request fallback group")
|
||||
}
|
||||
|
||||
fallbackGroup, err := s.groupRepo.GetByIDLite(ctx, fallbackGroupID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("fallback group not found: %w", err)
|
||||
}
|
||||
if fallbackGroup.Platform != PlatformAnthropic {
|
||||
return fmt.Errorf("fallback group must be anthropic platform")
|
||||
}
|
||||
if fallbackGroup.SubscriptionType == SubscriptionTypeSubscription {
|
||||
return fmt.Errorf("fallback group cannot be subscription type")
|
||||
}
|
||||
if fallbackGroup.FallbackGroupIDOnInvalidRequest != nil {
|
||||
return fmt.Errorf("fallback group cannot have invalid request fallback configured")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *UpdateGroupInput) (*Group, error) {
|
||||
group, err := s.groupRepo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if input.Name != "" {
|
||||
group.Name = input.Name
|
||||
}
|
||||
if input.Description != nil {
|
||||
group.Description = *input.Description
|
||||
}
|
||||
if input.Platform != "" {
|
||||
group.Platform = input.Platform
|
||||
}
|
||||
if input.RateMultiplier != nil {
|
||||
if *input.RateMultiplier <= 0 {
|
||||
return nil, errors.New("rate_multiplier must be > 0")
|
||||
}
|
||||
group.RateMultiplier = *input.RateMultiplier
|
||||
}
|
||||
if input.IsExclusive != nil {
|
||||
group.IsExclusive = *input.IsExclusive
|
||||
}
|
||||
if input.Status != "" {
|
||||
group.Status = input.Status
|
||||
}
|
||||
|
||||
// 订阅相关字段
|
||||
if input.SubscriptionType != "" {
|
||||
group.SubscriptionType = input.SubscriptionType
|
||||
}
|
||||
// 限额字段:nil/负数 表示"无限制",0 表示"不允许用量",正数表示具体限额
|
||||
// 前端始终发送这三个字段,无需 nil 守卫
|
||||
group.DailyLimitUSD = normalizeLimit(input.DailyLimitUSD)
|
||||
group.WeeklyLimitUSD = normalizeLimit(input.WeeklyLimitUSD)
|
||||
group.MonthlyLimitUSD = normalizeLimit(input.MonthlyLimitUSD)
|
||||
// 图片生成计费配置:负数表示清除(使用默认价格)
|
||||
if input.AllowImageGeneration != nil {
|
||||
group.AllowImageGeneration = *input.AllowImageGeneration
|
||||
}
|
||||
if input.AllowBatchImageGeneration != nil {
|
||||
group.AllowBatchImageGeneration = *input.AllowBatchImageGeneration
|
||||
}
|
||||
if !group.AllowImageGeneration || group.Platform != PlatformGemini {
|
||||
group.AllowBatchImageGeneration = false
|
||||
}
|
||||
if input.ImageRateIndependent != nil {
|
||||
group.ImageRateIndependent = *input.ImageRateIndependent
|
||||
}
|
||||
if input.ImageRateMultiplier != nil {
|
||||
if *input.ImageRateMultiplier < 0 {
|
||||
return nil, errors.New("image_rate_multiplier must be >= 0")
|
||||
}
|
||||
group.ImageRateMultiplier = *input.ImageRateMultiplier
|
||||
}
|
||||
if input.BatchImageDiscountMultiplier != nil {
|
||||
if *input.BatchImageDiscountMultiplier < 0 {
|
||||
return nil, errors.New("batch_image_discount_multiplier must be >= 0")
|
||||
}
|
||||
group.BatchImageDiscountMultiplier = *input.BatchImageDiscountMultiplier
|
||||
}
|
||||
if input.BatchImageHoldMultiplier != nil {
|
||||
if *input.BatchImageHoldMultiplier < 0 {
|
||||
return nil, errors.New("batch_image_hold_multiplier must be >= 0")
|
||||
}
|
||||
group.BatchImageHoldMultiplier = *input.BatchImageHoldMultiplier
|
||||
}
|
||||
// 仅在本次更新显式触碰任一比例时校验合并后的不变式(hold >= discount),
|
||||
// 避免存量脏数据阻塞其他字段的正常更新(提交侧另有钳制兜底)。
|
||||
if (input.BatchImageDiscountMultiplier != nil || input.BatchImageHoldMultiplier != nil) &&
|
||||
group.BatchImageHoldMultiplier < group.BatchImageDiscountMultiplier {
|
||||
return nil, errors.New("batch_image_hold_multiplier must be >= batch_image_discount_multiplier")
|
||||
}
|
||||
if input.PeakRateEnabled != nil {
|
||||
group.PeakRateEnabled = *input.PeakRateEnabled
|
||||
}
|
||||
if input.PeakStart != nil {
|
||||
group.PeakStart = *input.PeakStart
|
||||
}
|
||||
if input.PeakEnd != nil {
|
||||
group.PeakEnd = *input.PeakEnd
|
||||
}
|
||||
if input.PeakRateMultiplier != nil {
|
||||
group.PeakRateMultiplier = *input.PeakRateMultiplier
|
||||
}
|
||||
// 先归一化(非订阅分组——含本次更新转为非订阅——静默清空高峰配置,清洗停用状态下的脏字段),
|
||||
// 再收敛校验:Update 可能只传部分 peak 字段,需对合并后的最终配置统一校验,
|
||||
// 防止单独修改 start/end 导致最终 start>=end 等非法配置入库。与 CreateGroup 同一收口。
|
||||
group.PeakRateEnabled, group.PeakStart, group.PeakEnd, group.PeakRateMultiplier = NormalizePeakRateConfig(group.SubscriptionType, group.PeakRateEnabled, group.PeakStart, group.PeakEnd, group.PeakRateMultiplier)
|
||||
if err := ValidatePeakRateConfig(group.SubscriptionType, group.PeakRateEnabled, group.PeakStart, group.PeakEnd, group.PeakRateMultiplier); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if input.ImagePrice1K != nil {
|
||||
group.ImagePrice1K = normalizePrice(input.ImagePrice1K)
|
||||
}
|
||||
if input.ImagePrice2K != nil {
|
||||
group.ImagePrice2K = normalizePrice(input.ImagePrice2K)
|
||||
}
|
||||
if input.ImagePrice4K != nil {
|
||||
group.ImagePrice4K = normalizePrice(input.ImagePrice4K)
|
||||
}
|
||||
|
||||
// Claude Code 客户端限制
|
||||
if input.ClaudeCodeOnly != nil {
|
||||
group.ClaudeCodeOnly = *input.ClaudeCodeOnly
|
||||
}
|
||||
if input.FallbackGroupID != nil {
|
||||
// 校验降级分组
|
||||
if *input.FallbackGroupID > 0 {
|
||||
if err := s.validateFallbackGroup(ctx, id, *input.FallbackGroupID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
group.FallbackGroupID = input.FallbackGroupID
|
||||
} else {
|
||||
// 传入 0 或负数表示清除降级分组
|
||||
group.FallbackGroupID = nil
|
||||
}
|
||||
}
|
||||
fallbackOnInvalidRequest := group.FallbackGroupIDOnInvalidRequest
|
||||
if input.FallbackGroupIDOnInvalidRequest != nil {
|
||||
if *input.FallbackGroupIDOnInvalidRequest > 0 {
|
||||
fallbackOnInvalidRequest = input.FallbackGroupIDOnInvalidRequest
|
||||
} else {
|
||||
fallbackOnInvalidRequest = nil
|
||||
}
|
||||
}
|
||||
if fallbackOnInvalidRequest != nil {
|
||||
if err := s.validateFallbackGroupOnInvalidRequest(ctx, id, group.Platform, group.SubscriptionType, *fallbackOnInvalidRequest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
group.FallbackGroupIDOnInvalidRequest = fallbackOnInvalidRequest
|
||||
|
||||
// 模型路由配置
|
||||
if input.ModelRouting != nil {
|
||||
group.ModelRouting = input.ModelRouting
|
||||
}
|
||||
if input.ModelRoutingEnabled != nil {
|
||||
group.ModelRoutingEnabled = *input.ModelRoutingEnabled
|
||||
}
|
||||
if input.MCPXMLInject != nil {
|
||||
group.MCPXMLInject = *input.MCPXMLInject
|
||||
}
|
||||
|
||||
// 支持的模型系列(仅 antigravity 平台使用)
|
||||
if input.SupportedModelScopes != nil {
|
||||
group.SupportedModelScopes = *input.SupportedModelScopes
|
||||
}
|
||||
|
||||
// OpenAI Messages 调度配置
|
||||
if input.AllowMessagesDispatch != nil {
|
||||
group.AllowMessagesDispatch = *input.AllowMessagesDispatch
|
||||
}
|
||||
if input.RequireOAuthOnly != nil {
|
||||
group.RequireOAuthOnly = *input.RequireOAuthOnly
|
||||
}
|
||||
if input.RequirePrivacySet != nil {
|
||||
group.RequirePrivacySet = *input.RequirePrivacySet
|
||||
}
|
||||
if input.DefaultMappedModel != nil {
|
||||
group.DefaultMappedModel = *input.DefaultMappedModel
|
||||
}
|
||||
if input.MessagesDispatchModelConfig != nil {
|
||||
group.MessagesDispatchModelConfig = normalizeOpenAIMessagesDispatchModelConfig(*input.MessagesDispatchModelConfig)
|
||||
}
|
||||
if input.ModelsListConfig != nil {
|
||||
group.ModelsListConfig = normalizeGroupModelsListConfig(*input.ModelsListConfig)
|
||||
}
|
||||
if input.RPMLimit != nil {
|
||||
group.RPMLimit = *input.RPMLimit
|
||||
}
|
||||
sanitizeGroupMessagesDispatchFields(group)
|
||||
|
||||
if err := s.groupRepo.Update(ctx, group); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByGroupID(ctx, id)
|
||||
}
|
||||
|
||||
// 如果指定了复制账号的源分组,同步绑定(替换当前分组的账号)
|
||||
if len(input.CopyAccountsFromGroupIDs) > 0 {
|
||||
// 去重源分组 IDs
|
||||
seen := make(map[int64]struct{})
|
||||
uniqueSourceGroupIDs := make([]int64, 0, len(input.CopyAccountsFromGroupIDs))
|
||||
for _, srcGroupID := range input.CopyAccountsFromGroupIDs {
|
||||
// 校验:源分组不能是自身
|
||||
if srcGroupID == id {
|
||||
return nil, fmt.Errorf("cannot copy accounts from self")
|
||||
}
|
||||
// 去重
|
||||
if _, exists := seen[srcGroupID]; !exists {
|
||||
seen[srcGroupID] = struct{}{}
|
||||
uniqueSourceGroupIDs = append(uniqueSourceGroupIDs, srcGroupID)
|
||||
}
|
||||
}
|
||||
|
||||
// 校验源分组的平台是否与当前分组一致
|
||||
for _, srcGroupID := range uniqueSourceGroupIDs {
|
||||
srcGroup, err := s.groupRepo.GetByIDLite(ctx, srcGroupID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("source group %d not found: %w", srcGroupID, err)
|
||||
}
|
||||
if srcGroup.Platform != group.Platform {
|
||||
return nil, fmt.Errorf("source group %d platform mismatch: expected %s, got %s", srcGroupID, group.Platform, srcGroup.Platform)
|
||||
}
|
||||
}
|
||||
|
||||
// 获取所有源分组的账号(去重)
|
||||
accountIDsToCopy, err := s.groupRepo.GetAccountIDsByGroupIDs(ctx, uniqueSourceGroupIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get accounts from source groups: %w", err)
|
||||
}
|
||||
|
||||
// 先清空当前分组的所有账号绑定
|
||||
if _, err := s.groupRepo.DeleteAccountGroupsByGroupID(ctx, id); err != nil {
|
||||
return nil, fmt.Errorf("failed to clear existing account bindings: %w", err)
|
||||
}
|
||||
|
||||
// require_oauth_only: 过滤掉 apikey 类型账号
|
||||
if group.RequireOAuthOnly && (group.Platform == PlatformOpenAI || group.Platform == PlatformAntigravity || group.Platform == PlatformAnthropic || group.Platform == PlatformGemini || group.Platform == PlatformGrok) && len(accountIDsToCopy) > 0 {
|
||||
accounts, err := s.accountRepo.GetByIDs(ctx, accountIDsToCopy)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to fetch accounts for oauth filter: %w", err)
|
||||
}
|
||||
oauthIDs := make(map[int64]struct{}, len(accounts))
|
||||
for _, acc := range accounts {
|
||||
if acc.Type != AccountTypeAPIKey {
|
||||
oauthIDs[acc.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
var filtered []int64
|
||||
for _, aid := range accountIDsToCopy {
|
||||
if _, ok := oauthIDs[aid]; ok {
|
||||
filtered = append(filtered, aid)
|
||||
}
|
||||
}
|
||||
accountIDsToCopy = filtered
|
||||
}
|
||||
|
||||
// 再绑定源分组的账号
|
||||
if len(accountIDsToCopy) > 0 {
|
||||
if err := s.groupRepo.BindAccountsToGroup(ctx, id, accountIDsToCopy); err != nil {
|
||||
return nil, fmt.Errorf("failed to bind accounts to group: %w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return group, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) DeleteGroup(ctx context.Context, id int64) error {
|
||||
var groupKeys []string
|
||||
if s.authCacheInvalidator != nil {
|
||||
keys, err := s.apiKeyRepo.ListKeysByGroupID(ctx, id)
|
||||
if err == nil {
|
||||
groupKeys = keys
|
||||
}
|
||||
}
|
||||
|
||||
affectedUserIDs, err := s.groupRepo.DeleteCascade(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 注意:user_group_rate_multipliers 表通过外键 ON DELETE CASCADE 自动清理
|
||||
|
||||
// 事务成功后,异步失效受影响用户的订阅缓存
|
||||
if len(affectedUserIDs) > 0 && s.billingCacheService != nil {
|
||||
groupID := id
|
||||
go func() {
|
||||
cacheCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
for _, userID := range affectedUserIDs {
|
||||
if err := s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID); err != nil {
|
||||
logger.LegacyPrintf("service.admin", "invalidate subscription cache failed: user_id=%d group_id=%d err=%v", userID, groupID, err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
if s.authCacheInvalidator != nil {
|
||||
for _, key := range groupKeys {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByKey(ctx, key)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetGroupAPIKeys(ctx context.Context, groupID int64, page, pageSize int) ([]APIKey, int64, error) {
|
||||
params := pagination.PaginationParams{Page: page, PageSize: pageSize}
|
||||
keys, result, err := s.apiKeyRepo.ListByGroupID(ctx, groupID, params)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return keys, result.Total, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetGroupRateMultipliers(ctx context.Context, groupID int64) ([]UserGroupRateEntry, error) {
|
||||
if s.userGroupRateRepo == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return s.userGroupRateRepo.GetByGroupID(ctx, groupID)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) ClearGroupRateMultipliers(ctx context.Context, groupID int64) error {
|
||||
if s.userGroupRateRepo == nil {
|
||||
return nil
|
||||
}
|
||||
return s.userGroupRateRepo.DeleteByGroupID(ctx, groupID)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) BatchSetGroupRateMultipliers(ctx context.Context, groupID int64, entries []GroupRateMultiplierInput) error {
|
||||
if s.userGroupRateRepo == nil {
|
||||
return nil
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.RateMultiplier <= 0 {
|
||||
return fmt.Errorf("rate_multiplier must be > 0 (user_id=%d)", e.UserID)
|
||||
}
|
||||
}
|
||||
return s.userGroupRateRepo.SyncGroupRateMultipliers(ctx, groupID, entries)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) ClearGroupRPMOverrides(ctx context.Context, groupID int64) error {
|
||||
if s.userGroupRateRepo == nil {
|
||||
return nil
|
||||
}
|
||||
if err := s.userGroupRateRepo.ClearGroupRPMOverrides(ctx, groupID); err != nil {
|
||||
return err
|
||||
}
|
||||
// RPM override 已嵌入 auth cache snapshot (v7),变更后必须失效相关缓存。
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByGroupID(ctx, groupID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) BatchSetGroupRPMOverrides(ctx context.Context, groupID int64, entries []GroupRPMOverrideInput) error {
|
||||
if s.userGroupRateRepo == nil {
|
||||
return nil
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.RPMOverride != nil && *e.RPMOverride < 0 {
|
||||
return infraerrors.BadRequest("INVALID_RPM_OVERRIDE", fmt.Sprintf("rpm_override must be >= 0 (user_id=%d)", e.UserID))
|
||||
}
|
||||
}
|
||||
if err := s.userGroupRateRepo.SyncGroupRPMOverrides(ctx, groupID, entries); err != nil {
|
||||
return err
|
||||
}
|
||||
// RPM override 已嵌入 auth cache snapshot (v7),变更后必须失效相关缓存。
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByGroupID(ctx, groupID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) UpdateGroupSortOrders(ctx context.Context, updates []GroupSortOrderUpdate) error {
|
||||
return s.groupRepo.UpdateSortOrders(ctx, updates)
|
||||
}
|
||||
|
||||
// AdminUpdateAPIKeyGroupID 管理员修改 API Key 分组绑定
|
||||
// groupID: nil=不修改, 指向0=解绑, 指向正整数=绑定到目标分组
|
||||
func (s *adminServiceImpl) AdminUpdateAPIKeyGroupID(ctx context.Context, keyID int64, groupID *int64) (*AdminUpdateAPIKeyGroupIDResult, error) {
|
||||
apiKey, err := s.apiKeyRepo.GetByID(ctx, keyID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if groupID == nil {
|
||||
// nil 表示不修改,直接返回
|
||||
return &AdminUpdateAPIKeyGroupIDResult{APIKey: apiKey}, nil
|
||||
}
|
||||
|
||||
if *groupID < 0 {
|
||||
return nil, infraerrors.BadRequest("INVALID_GROUP_ID", "group_id must be non-negative")
|
||||
}
|
||||
|
||||
result := &AdminUpdateAPIKeyGroupIDResult{}
|
||||
|
||||
if *groupID == 0 {
|
||||
// 0 表示解绑分组(不修改 user_allowed_groups,避免影响用户其他 Key)
|
||||
apiKey.GroupID = nil
|
||||
apiKey.Group = nil
|
||||
} else {
|
||||
// 验证目标分组存在且状态为 active
|
||||
group, err := s.groupRepo.GetByID(ctx, *groupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if group.Status != StatusActive {
|
||||
return nil, infraerrors.BadRequest("GROUP_NOT_ACTIVE", "target group is not active")
|
||||
}
|
||||
// 订阅类型分组:用户须持有该分组的有效订阅才可绑定
|
||||
if group.IsSubscriptionType() {
|
||||
if s.userSubRepo == nil {
|
||||
return nil, infraerrors.InternalServer("SUBSCRIPTION_REPOSITORY_UNAVAILABLE", "subscription repository is not configured")
|
||||
}
|
||||
if _, err := s.userSubRepo.GetActiveByUserIDAndGroupID(ctx, apiKey.UserID, *groupID); err != nil {
|
||||
if errors.Is(err, ErrSubscriptionNotFound) {
|
||||
return nil, infraerrors.BadRequest("SUBSCRIPTION_REQUIRED", "user does not have an active subscription for this group")
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
gid := *groupID
|
||||
apiKey.GroupID = &gid
|
||||
apiKey.Group = group
|
||||
|
||||
// 专属标准分组:使用事务保证「添加分组权限」与「更新 API Key」的原子性
|
||||
if group.IsExclusive && !group.IsSubscriptionType() {
|
||||
opCtx := ctx
|
||||
var tx *dbent.Tx
|
||||
if s.entClient == nil {
|
||||
logger.LegacyPrintf("service.admin", "Warning: entClient is nil, skipping transaction protection for exclusive group binding")
|
||||
} else {
|
||||
var txErr error
|
||||
tx, txErr = s.entClient.Tx(ctx)
|
||||
if txErr != nil {
|
||||
return nil, fmt.Errorf("begin transaction: %w", txErr)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
opCtx = dbent.NewTxContext(ctx, tx)
|
||||
}
|
||||
|
||||
if addErr := s.userRepo.AddGroupToAllowedGroups(opCtx, apiKey.UserID, gid); addErr != nil {
|
||||
return nil, fmt.Errorf("add group to user allowed groups: %w", addErr)
|
||||
}
|
||||
if err := s.apiKeyRepo.Update(opCtx, apiKey); err != nil {
|
||||
return nil, fmt.Errorf("update api key: %w", err)
|
||||
}
|
||||
if tx != nil {
|
||||
if err := tx.Commit(); err != nil {
|
||||
return nil, fmt.Errorf("commit transaction: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
result.AutoGrantedGroupAccess = true
|
||||
result.GrantedGroupID = &gid
|
||||
result.GrantedGroupName = group.Name
|
||||
|
||||
// 失效认证缓存(在事务提交后执行)
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByKey(ctx, apiKey.Key)
|
||||
}
|
||||
|
||||
result.APIKey = apiKey
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
|
||||
// 非专属分组 / 解绑:无需事务,单步更新即可
|
||||
if err := s.apiKeyRepo.Update(ctx, apiKey); err != nil {
|
||||
return nil, fmt.Errorf("update api key: %w", err)
|
||||
}
|
||||
|
||||
// 失效认证缓存
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByKey(ctx, apiKey.Key)
|
||||
}
|
||||
|
||||
result.APIKey = apiKey
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// AdminResetAPIKeyRateLimitUsage resets all API key rate-limit usage windows.
|
||||
func (s *adminServiceImpl) AdminResetAPIKeyRateLimitUsage(ctx context.Context, keyID int64) (*APIKey, error) {
|
||||
apiKey, err := s.apiKeyRepo.GetByID(ctx, keyID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
apiKey.Usage5h = 0
|
||||
apiKey.Usage1d = 0
|
||||
apiKey.Usage7d = 0
|
||||
apiKey.Window5hStart = nil
|
||||
apiKey.Window1dStart = nil
|
||||
apiKey.Window7dStart = nil
|
||||
if err := s.apiKeyRepo.Update(ctx, apiKey); err != nil {
|
||||
return nil, fmt.Errorf("reset api key rate limit usage: %w", err)
|
||||
}
|
||||
if s.authCacheInvalidator != nil {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByKey(ctx, apiKey.Key)
|
||||
}
|
||||
if s.billingCacheService != nil {
|
||||
_ = s.billingCacheService.InvalidateAPIKeyRateLimit(ctx, apiKey.ID)
|
||||
}
|
||||
return apiKey, nil
|
||||
}
|
||||
|
||||
// ReplaceUserGroup 替换用户的专属分组
|
||||
func (s *adminServiceImpl) ReplaceUserGroup(ctx context.Context, userID, oldGroupID, newGroupID int64) (*ReplaceUserGroupResult, error) {
|
||||
if oldGroupID == newGroupID {
|
||||
return nil, infraerrors.BadRequest("SAME_GROUP", "old and new group must be different")
|
||||
}
|
||||
|
||||
// 验证新分组存在且为活跃的专属标准分组
|
||||
newGroup, err := s.groupRepo.GetByID(ctx, newGroupID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if newGroup.Status != StatusActive {
|
||||
return nil, infraerrors.BadRequest("GROUP_NOT_ACTIVE", "target group is not active")
|
||||
}
|
||||
if !newGroup.IsExclusive {
|
||||
return nil, infraerrors.BadRequest("GROUP_NOT_EXCLUSIVE", "target group is not exclusive")
|
||||
}
|
||||
if newGroup.IsSubscriptionType() {
|
||||
return nil, infraerrors.BadRequest("GROUP_IS_SUBSCRIPTION", "subscription groups are not supported for replacement")
|
||||
}
|
||||
|
||||
// 事务保证原子性
|
||||
if s.entClient == nil {
|
||||
return nil, fmt.Errorf("entClient is nil, cannot perform group replacement")
|
||||
}
|
||||
tx, err := s.entClient.Tx(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("begin transaction: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
opCtx := dbent.NewTxContext(ctx, tx)
|
||||
|
||||
// 1. 授予新分组权限
|
||||
if err := s.userRepo.AddGroupToAllowedGroups(opCtx, userID, newGroupID); err != nil {
|
||||
return nil, fmt.Errorf("add new group to allowed groups: %w", err)
|
||||
}
|
||||
|
||||
// 2. 迁移绑定旧分组的 Key 到新分组
|
||||
migrated, err := s.apiKeyRepo.UpdateGroupIDByUserAndGroup(opCtx, userID, oldGroupID, newGroupID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migrate api keys: %w", err)
|
||||
}
|
||||
|
||||
// 3. 移除旧分组权限
|
||||
if err := s.userRepo.RemoveGroupFromUserAllowedGroups(opCtx, userID, oldGroupID); err != nil {
|
||||
return nil, fmt.Errorf("remove old group from allowed groups: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return nil, fmt.Errorf("commit transaction: %w", err)
|
||||
}
|
||||
|
||||
// 失效该用户所有 Key 的认证缓存
|
||||
if s.authCacheInvalidator != nil {
|
||||
keys, keyErr := s.apiKeyRepo.ListKeysByUserID(ctx, userID)
|
||||
if keyErr == nil {
|
||||
for _, k := range keys {
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByKey(ctx, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return &ReplaceUserGroupResult{MigratedKeys: migrated}, nil
|
||||
}
|
||||
@@ -0,0 +1,608 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||||
"github.com/Wei-Shaw/sub2api/internal/util/httputil"
|
||||
)
|
||||
|
||||
// Proxy management implementations
|
||||
func (s *adminServiceImpl) ListProxies(ctx context.Context, page, pageSize int, protocol, status, search string, sortBy, sortOrder string) ([]Proxy, int64, error) {
|
||||
params := pagination.PaginationParams{Page: page, PageSize: pageSize, SortBy: sortBy, SortOrder: sortOrder}
|
||||
proxies, result, err := s.proxyRepo.ListWithFilters(ctx, params, protocol, status, search)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return proxies, result.Total, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) ListProxiesWithAccountCount(ctx context.Context, page, pageSize int, protocol, status, search string, sortBy, sortOrder string) ([]ProxyWithAccountCount, int64, error) {
|
||||
params := pagination.PaginationParams{Page: page, PageSize: pageSize, SortBy: sortBy, SortOrder: sortOrder}
|
||||
proxies, result, err := s.proxyRepo.ListWithFiltersAndAccountCount(ctx, params, protocol, status, search)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
s.attachProxyLatency(ctx, proxies)
|
||||
return proxies, result.Total, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetAllProxies(ctx context.Context) ([]Proxy, error) {
|
||||
return s.proxyRepo.ListActive(ctx)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetAllProxiesWithAccountCount(ctx context.Context) ([]ProxyWithAccountCount, error) {
|
||||
proxies, err := s.proxyRepo.ListActiveWithAccountCount(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.attachProxyLatency(ctx, proxies)
|
||||
return proxies, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetProxy(ctx context.Context, id int64) (*Proxy, error) {
|
||||
return s.proxyRepo.GetByID(ctx, id)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetProxiesByIDs(ctx context.Context, ids []int64) ([]Proxy, error) {
|
||||
return s.proxyRepo.ListByIDs(ctx, ids)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) CreateProxy(ctx context.Context, input *CreateProxyInput) (*Proxy, error) {
|
||||
// 规范化 fallback_mode
|
||||
mode := input.FallbackMode
|
||||
if mode == "" {
|
||||
mode = FallbackModeNone
|
||||
}
|
||||
// 校验:mode=proxy 必须有 backup
|
||||
if mode == FallbackModeProxy && input.BackupProxyID == nil {
|
||||
return nil, infraerrors.BadRequest("PROXY_BACKUP_REQUIRED", "backup proxy required when fallback_mode=proxy")
|
||||
}
|
||||
if input.ExpiryWarnDays < 0 {
|
||||
return nil, infraerrors.BadRequest("PROXY_WARN_DAYS_INVALID", "expiry_warn_days must be >= 0")
|
||||
}
|
||||
|
||||
proxy := &Proxy{
|
||||
Name: input.Name,
|
||||
Protocol: input.Protocol,
|
||||
Host: input.Host,
|
||||
Port: input.Port,
|
||||
Username: input.Username,
|
||||
Password: input.Password,
|
||||
Status: StatusActive,
|
||||
ExpiresAt: input.ExpiresAt,
|
||||
FallbackMode: mode,
|
||||
BackupProxyID: input.BackupProxyID,
|
||||
ExpiryWarnDays: input.ExpiryWarnDays,
|
||||
}
|
||||
if err := s.proxyRepo.Create(ctx, proxy); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Probe latency asynchronously so creation isn't blocked by network timeout.
|
||||
go s.probeProxyLatency(context.Background(), proxy)
|
||||
return proxy, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) UpdateProxy(ctx context.Context, id int64, input *UpdateProxyInput) (*Proxy, error) {
|
||||
// 校验:backup_proxy_id 不能是自身
|
||||
if input.BackupProxyID != nil && *input.BackupProxyID == id {
|
||||
return nil, infraerrors.BadRequest("PROXY_BACKUP_SELF", "backup proxy cannot be itself")
|
||||
}
|
||||
// 规范化 fallback_mode
|
||||
mode := input.FallbackMode
|
||||
if mode == "" {
|
||||
mode = FallbackModeNone
|
||||
}
|
||||
// 校验:mode=proxy 必须有 backup
|
||||
if mode == FallbackModeProxy && input.BackupProxyID == nil {
|
||||
return nil, infraerrors.BadRequest("PROXY_BACKUP_REQUIRED", "backup proxy required when fallback_mode=proxy")
|
||||
}
|
||||
if input.ExpiryWarnDays < 0 {
|
||||
return nil, infraerrors.BadRequest("PROXY_WARN_DAYS_INVALID", "expiry_warn_days must be >= 0")
|
||||
}
|
||||
|
||||
proxy, err := s.proxyRepo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if input.Name != "" {
|
||||
proxy.Name = input.Name
|
||||
}
|
||||
if input.Protocol != "" {
|
||||
proxy.Protocol = input.Protocol
|
||||
}
|
||||
if input.Host != "" {
|
||||
proxy.Host = input.Host
|
||||
}
|
||||
if input.Port != 0 {
|
||||
proxy.Port = input.Port
|
||||
}
|
||||
if input.Username != "" {
|
||||
proxy.Username = input.Username
|
||||
}
|
||||
if input.Password != "" {
|
||||
proxy.Password = input.Password
|
||||
}
|
||||
if input.Status != "" {
|
||||
proxy.Status = input.Status
|
||||
}
|
||||
// 透传有效期与回退字段
|
||||
proxy.ExpiresAt = input.ExpiresAt
|
||||
proxy.FallbackMode = mode
|
||||
proxy.BackupProxyID = input.BackupProxyID
|
||||
proxy.ExpiryWarnDays = input.ExpiryWarnDays
|
||||
|
||||
if err := s.proxyRepo.Update(ctx, proxy); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return proxy, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) DeleteProxy(ctx context.Context, id int64) error {
|
||||
count, err := s.proxyRepo.CountAccountsByProxyID(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if count > 0 {
|
||||
return ErrProxyInUse
|
||||
}
|
||||
return s.proxyRepo.Delete(ctx, id)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) BatchDeleteProxies(ctx context.Context, ids []int64) (*ProxyBatchDeleteResult, error) {
|
||||
result := &ProxyBatchDeleteResult{}
|
||||
if len(ids) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
for _, id := range ids {
|
||||
count, err := s.proxyRepo.CountAccountsByProxyID(ctx, id)
|
||||
if err != nil {
|
||||
result.Skipped = append(result.Skipped, ProxyBatchDeleteSkipped{
|
||||
ID: id,
|
||||
Reason: err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
if count > 0 {
|
||||
result.Skipped = append(result.Skipped, ProxyBatchDeleteSkipped{
|
||||
ID: id,
|
||||
Reason: ErrProxyInUse.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
if err := s.proxyRepo.Delete(ctx, id); err != nil {
|
||||
result.Skipped = append(result.Skipped, ProxyBatchDeleteSkipped{
|
||||
ID: id,
|
||||
Reason: err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
result.DeletedIDs = append(result.DeletedIDs, id)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) GetProxyAccounts(ctx context.Context, proxyID int64) ([]ProxyAccountSummary, error) {
|
||||
return s.proxyRepo.ListAccountSummariesByProxyID(ctx, proxyID)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) CheckProxyExists(ctx context.Context, host string, port int, username, password string) (bool, error) {
|
||||
return s.proxyRepo.ExistsByHostPortAuth(ctx, host, port, username, password)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) TestProxy(ctx context.Context, id int64) (*ProxyTestResult, error) {
|
||||
proxy, err := s.proxyRepo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
proxyURL := proxy.URL()
|
||||
exitInfo, latencyMs, err := s.proxyProber.ProbeProxy(ctx, proxyURL)
|
||||
if err != nil {
|
||||
s.saveProxyLatency(ctx, id, &ProxyLatencyInfo{
|
||||
Success: false,
|
||||
Message: err.Error(),
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
return &ProxyTestResult{
|
||||
Success: false,
|
||||
Message: err.Error(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
latency := latencyMs
|
||||
s.saveProxyLatency(ctx, id, &ProxyLatencyInfo{
|
||||
Success: true,
|
||||
LatencyMs: &latency,
|
||||
Message: "Proxy is accessible",
|
||||
IPAddress: exitInfo.IP,
|
||||
Country: exitInfo.Country,
|
||||
CountryCode: exitInfo.CountryCode,
|
||||
Region: exitInfo.Region,
|
||||
City: exitInfo.City,
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
return &ProxyTestResult{
|
||||
Success: true,
|
||||
Message: "Proxy is accessible",
|
||||
LatencyMs: latencyMs,
|
||||
IPAddress: exitInfo.IP,
|
||||
City: exitInfo.City,
|
||||
Region: exitInfo.Region,
|
||||
Country: exitInfo.Country,
|
||||
CountryCode: exitInfo.CountryCode,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) CheckProxyQuality(ctx context.Context, id int64) (*ProxyQualityCheckResult, error) {
|
||||
proxy, err := s.proxyRepo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result := &ProxyQualityCheckResult{
|
||||
ProxyID: id,
|
||||
Score: 100,
|
||||
Grade: "A",
|
||||
CheckedAt: time.Now().Unix(),
|
||||
Items: make([]ProxyQualityCheckItem, 0, len(proxyQualityTargets)+1),
|
||||
}
|
||||
|
||||
proxyURL := proxy.URL()
|
||||
if s.proxyProber == nil {
|
||||
result.Items = append(result.Items, ProxyQualityCheckItem{
|
||||
Target: "base_connectivity",
|
||||
Status: "fail",
|
||||
Message: "代理探测服务未配置",
|
||||
})
|
||||
result.FailedCount++
|
||||
finalizeProxyQualityResult(result)
|
||||
s.saveProxyQualitySnapshot(ctx, id, result, nil)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
exitInfo, latencyMs, err := s.proxyProber.ProbeProxy(ctx, proxyURL)
|
||||
if err != nil {
|
||||
result.Items = append(result.Items, ProxyQualityCheckItem{
|
||||
Target: "base_connectivity",
|
||||
Status: "fail",
|
||||
LatencyMs: latencyMs,
|
||||
Message: err.Error(),
|
||||
})
|
||||
result.FailedCount++
|
||||
finalizeProxyQualityResult(result)
|
||||
s.saveProxyQualitySnapshot(ctx, id, result, nil)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
result.ExitIP = exitInfo.IP
|
||||
result.Country = exitInfo.Country
|
||||
result.CountryCode = exitInfo.CountryCode
|
||||
result.BaseLatencyMs = latencyMs
|
||||
result.Items = append(result.Items, ProxyQualityCheckItem{
|
||||
Target: "base_connectivity",
|
||||
Status: "pass",
|
||||
LatencyMs: latencyMs,
|
||||
Message: "代理出口连通正常",
|
||||
})
|
||||
result.PassedCount++
|
||||
|
||||
client, err := httpclient.GetClient(httpclient.Options{
|
||||
ProxyURL: proxyURL,
|
||||
Timeout: proxyQualityRequestTimeout,
|
||||
ResponseHeaderTimeout: proxyQualityResponseHeaderTimeout,
|
||||
})
|
||||
if err != nil {
|
||||
result.Items = append(result.Items, ProxyQualityCheckItem{
|
||||
Target: "http_client",
|
||||
Status: "fail",
|
||||
Message: fmt.Sprintf("创建检测客户端失败: %v", err),
|
||||
})
|
||||
result.FailedCount++
|
||||
finalizeProxyQualityResult(result)
|
||||
s.saveProxyQualitySnapshot(ctx, id, result, exitInfo)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
for _, target := range proxyQualityTargets {
|
||||
item := runProxyQualityTarget(ctx, client, target)
|
||||
result.Items = append(result.Items, item)
|
||||
switch item.Status {
|
||||
case "pass":
|
||||
result.PassedCount++
|
||||
case "warn":
|
||||
result.WarnCount++
|
||||
case "challenge":
|
||||
result.ChallengeCount++
|
||||
default:
|
||||
result.FailedCount++
|
||||
}
|
||||
}
|
||||
|
||||
finalizeProxyQualityResult(result)
|
||||
s.saveProxyQualitySnapshot(ctx, id, result, exitInfo)
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func runProxyQualityTarget(ctx context.Context, client *http.Client, target proxyQualityTarget) ProxyQualityCheckItem {
|
||||
item := ProxyQualityCheckItem{
|
||||
Target: target.Target,
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, target.Method, target.URL, nil)
|
||||
if err != nil {
|
||||
item.Status = "fail"
|
||||
item.Message = fmt.Sprintf("构建请求失败: %v", err)
|
||||
return item
|
||||
}
|
||||
req.Header.Set("Accept", "application/json,text/html,*/*")
|
||||
req.Header.Set("User-Agent", proxyQualityClientUserAgent)
|
||||
|
||||
start := time.Now()
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
item.Status = "fail"
|
||||
item.LatencyMs = time.Since(start).Milliseconds()
|
||||
item.Message = fmt.Sprintf("请求失败: %v", err)
|
||||
return item
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
item.LatencyMs = time.Since(start).Milliseconds()
|
||||
item.HTTPStatus = resp.StatusCode
|
||||
|
||||
body, readErr := io.ReadAll(io.LimitReader(resp.Body, proxyQualityMaxBodyBytes+1))
|
||||
if readErr != nil {
|
||||
item.Status = "fail"
|
||||
item.Message = fmt.Sprintf("读取响应失败: %v", readErr)
|
||||
return item
|
||||
}
|
||||
if int64(len(body)) > proxyQualityMaxBodyBytes {
|
||||
body = body[:proxyQualityMaxBodyBytes]
|
||||
}
|
||||
|
||||
// Cloudflare challenge 检测
|
||||
if httputil.IsCloudflareChallengeResponse(resp.StatusCode, resp.Header, body) {
|
||||
item.Status = "challenge"
|
||||
item.CFRay = httputil.ExtractCloudflareRayID(resp.Header, body)
|
||||
item.Message = "命中 Cloudflare challenge"
|
||||
return item
|
||||
}
|
||||
|
||||
if _, ok := target.AllowedStatuses[resp.StatusCode]; ok {
|
||||
// 白名单内的状态码均代表目标可达:2xx 表示接口直接可用,
|
||||
// 401/405 等是无鉴权探测的预期结果,同样视为连通正常,不再扣分。
|
||||
item.Status = "pass"
|
||||
if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
|
||||
item.Message = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||||
} else {
|
||||
item.Message = fmt.Sprintf("HTTP %d(目标可达)", resp.StatusCode)
|
||||
}
|
||||
return item
|
||||
}
|
||||
|
||||
if resp.StatusCode == http.StatusTooManyRequests {
|
||||
item.Status = "warn"
|
||||
item.Message = "目标返回 429,可能存在频控"
|
||||
return item
|
||||
}
|
||||
|
||||
item.Status = "fail"
|
||||
item.Message = fmt.Sprintf("非预期状态码: %d", resp.StatusCode)
|
||||
return item
|
||||
}
|
||||
|
||||
func finalizeProxyQualityResult(result *ProxyQualityCheckResult) {
|
||||
if result == nil {
|
||||
return
|
||||
}
|
||||
score := 100 - result.WarnCount*10 - result.FailedCount*22 - result.ChallengeCount*30
|
||||
if score < 0 {
|
||||
score = 0
|
||||
}
|
||||
result.Score = score
|
||||
result.Grade = proxyQualityGrade(score)
|
||||
result.Summary = fmt.Sprintf(
|
||||
"通过 %d 项,告警 %d 项,失败 %d 项,挑战 %d 项",
|
||||
result.PassedCount,
|
||||
result.WarnCount,
|
||||
result.FailedCount,
|
||||
result.ChallengeCount,
|
||||
)
|
||||
}
|
||||
|
||||
func proxyQualityGrade(score int) string {
|
||||
switch {
|
||||
case score >= 90:
|
||||
return "A"
|
||||
case score >= 75:
|
||||
return "B"
|
||||
case score >= 60:
|
||||
return "C"
|
||||
case score >= 40:
|
||||
return "D"
|
||||
default:
|
||||
return "F"
|
||||
}
|
||||
}
|
||||
|
||||
func proxyQualityOverallStatus(result *ProxyQualityCheckResult) string {
|
||||
if result == nil {
|
||||
return ""
|
||||
}
|
||||
if result.ChallengeCount > 0 {
|
||||
return "challenge"
|
||||
}
|
||||
if result.FailedCount > 0 {
|
||||
return "failed"
|
||||
}
|
||||
if result.WarnCount > 0 {
|
||||
return "warn"
|
||||
}
|
||||
if result.PassedCount > 0 {
|
||||
return "healthy"
|
||||
}
|
||||
return "failed"
|
||||
}
|
||||
|
||||
func proxyQualityFirstCFRay(result *ProxyQualityCheckResult) string {
|
||||
if result == nil {
|
||||
return ""
|
||||
}
|
||||
for _, item := range result.Items {
|
||||
if item.CFRay != "" {
|
||||
return item.CFRay
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func proxyQualityBaseConnectivityPass(result *ProxyQualityCheckResult) bool {
|
||||
if result == nil {
|
||||
return false
|
||||
}
|
||||
for _, item := range result.Items {
|
||||
if item.Target == "base_connectivity" {
|
||||
return item.Status == "pass"
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) saveProxyQualitySnapshot(ctx context.Context, proxyID int64, result *ProxyQualityCheckResult, exitInfo *ProxyExitInfo) {
|
||||
if result == nil {
|
||||
return
|
||||
}
|
||||
score := result.Score
|
||||
checkedAt := result.CheckedAt
|
||||
info := &ProxyLatencyInfo{
|
||||
Success: proxyQualityBaseConnectivityPass(result),
|
||||
Message: result.Summary,
|
||||
QualityStatus: proxyQualityOverallStatus(result),
|
||||
QualityScore: &score,
|
||||
QualityGrade: result.Grade,
|
||||
QualitySummary: result.Summary,
|
||||
QualityCheckedAt: &checkedAt,
|
||||
QualityCFRay: proxyQualityFirstCFRay(result),
|
||||
UpdatedAt: time.Now(),
|
||||
}
|
||||
if result.BaseLatencyMs > 0 {
|
||||
latency := result.BaseLatencyMs
|
||||
info.LatencyMs = &latency
|
||||
}
|
||||
if exitInfo != nil {
|
||||
info.IPAddress = exitInfo.IP
|
||||
info.Country = exitInfo.Country
|
||||
info.CountryCode = exitInfo.CountryCode
|
||||
info.Region = exitInfo.Region
|
||||
info.City = exitInfo.City
|
||||
}
|
||||
s.saveProxyLatency(ctx, proxyID, info)
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) probeProxyLatency(ctx context.Context, proxy *Proxy) {
|
||||
if s.proxyProber == nil || proxy == nil {
|
||||
return
|
||||
}
|
||||
exitInfo, latencyMs, err := s.proxyProber.ProbeProxy(ctx, proxy.URL())
|
||||
if err != nil {
|
||||
s.saveProxyLatency(ctx, proxy.ID, &ProxyLatencyInfo{
|
||||
Success: false,
|
||||
Message: err.Error(),
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
latency := latencyMs
|
||||
s.saveProxyLatency(ctx, proxy.ID, &ProxyLatencyInfo{
|
||||
Success: true,
|
||||
LatencyMs: &latency,
|
||||
Message: "Proxy is accessible",
|
||||
IPAddress: exitInfo.IP,
|
||||
Country: exitInfo.Country,
|
||||
CountryCode: exitInfo.CountryCode,
|
||||
Region: exitInfo.Region,
|
||||
City: exitInfo.City,
|
||||
UpdatedAt: time.Now(),
|
||||
})
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) attachProxyLatency(ctx context.Context, proxies []ProxyWithAccountCount) {
|
||||
if s.proxyLatencyCache == nil || len(proxies) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
ids := make([]int64, 0, len(proxies))
|
||||
for i := range proxies {
|
||||
ids = append(ids, proxies[i].ID)
|
||||
}
|
||||
|
||||
latencies, err := s.proxyLatencyCache.GetProxyLatencies(ctx, ids)
|
||||
if err != nil {
|
||||
logger.LegacyPrintf("service.admin", "Warning: load proxy latency cache failed: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
for i := range proxies {
|
||||
info := latencies[proxies[i].ID]
|
||||
if info == nil {
|
||||
continue
|
||||
}
|
||||
if info.Success {
|
||||
proxies[i].LatencyStatus = "success"
|
||||
proxies[i].LatencyMs = info.LatencyMs
|
||||
} else {
|
||||
proxies[i].LatencyStatus = "failed"
|
||||
}
|
||||
proxies[i].LatencyMessage = info.Message
|
||||
proxies[i].IPAddress = info.IPAddress
|
||||
proxies[i].Country = info.Country
|
||||
proxies[i].CountryCode = info.CountryCode
|
||||
proxies[i].Region = info.Region
|
||||
proxies[i].City = info.City
|
||||
proxies[i].QualityStatus = info.QualityStatus
|
||||
proxies[i].QualityScore = info.QualityScore
|
||||
proxies[i].QualityGrade = info.QualityGrade
|
||||
proxies[i].QualitySummary = info.QualitySummary
|
||||
proxies[i].QualityChecked = info.QualityCheckedAt
|
||||
}
|
||||
}
|
||||
|
||||
func (s *adminServiceImpl) saveProxyLatency(ctx context.Context, proxyID int64, info *ProxyLatencyInfo) {
|
||||
if s.proxyLatencyCache == nil || info == nil {
|
||||
return
|
||||
}
|
||||
|
||||
merged := *info
|
||||
if latencies, err := s.proxyLatencyCache.GetProxyLatencies(ctx, []int64{proxyID}); err == nil {
|
||||
if existing := latencies[proxyID]; existing != nil {
|
||||
if merged.QualityCheckedAt == nil &&
|
||||
merged.QualityScore == nil &&
|
||||
merged.QualityGrade == "" &&
|
||||
merged.QualityStatus == "" &&
|
||||
merged.QualitySummary == "" &&
|
||||
merged.QualityCFRay == "" {
|
||||
merged.QualityStatus = existing.QualityStatus
|
||||
merged.QualityScore = existing.QualityScore
|
||||
merged.QualityGrade = existing.QualityGrade
|
||||
merged.QualitySummary = existing.QualitySummary
|
||||
merged.QualityCheckedAt = existing.QualityCheckedAt
|
||||
merged.QualityCFRay = existing.QualityCFRay
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.proxyLatencyCache.SetProxyLatency(ctx, proxyID, &merged); err != nil {
|
||||
logger.LegacyPrintf("service.admin", "Warning: store proxy latency cache failed: %v", err)
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user