mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 03:52:26 +08:00
Merge pull request #5543 from luckydududu/fix/invalidate-channel-cache-on-group-platform-change
fix(group): 改分组 platform 后失效渠道缓存(否则最长 10 分钟按旧平台计价)
This commit is contained in:
@@ -189,7 +189,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
|
||||
adminAccountRepository := repository.NewAdminAccountRepository(client, db, schedulerCache)
|
||||
proxyExitInfoProber := repository.NewProxyExitInfoProber(configConfig)
|
||||
proxyLatencyCache := repository.NewProxyLatencyCache(redisClient)
|
||||
adminService := service.NewAdminService(userRepository, adminGroupRepository, adminAccountRepository, proxyRepository, apiKeyRepository, redeemCodeRepository, userGroupRateRepository, userRPMCache, billingCacheService, proxyExitInfoProber, proxyLatencyCache, apiKeyAuthCacheInvalidator, client, settingService, subscriptionService, userSubscriptionRepository, privacyClientFactory, openAIGatewayService, affiliateService, compositeModelRouteRepository, compositeRouteResolver)
|
||||
adminService := service.NewAdminService(userRepository, adminGroupRepository, adminAccountRepository, proxyRepository, apiKeyRepository, redeemCodeRepository, userGroupRateRepository, userRPMCache, billingCacheService, proxyExitInfoProber, proxyLatencyCache, apiKeyAuthCacheInvalidator, client, settingService, subscriptionService, userSubscriptionRepository, privacyClientFactory, openAIGatewayService, affiliateService, compositeModelRouteRepository, compositeRouteResolver, channelService)
|
||||
adminUserHandler := admin.NewUserHandler(adminService, concurrencyService, serviceUserPlatformQuotaRepository, billingCache, totpService, userService, settingService)
|
||||
groupCapacityService := service.NewGroupCapacityService(accountRepository, groupRepository, concurrencyService, sessionLimitCache, rpmCache)
|
||||
groupHandler := admin.NewGroupHandler(adminService, dashboardService, groupCapacityService)
|
||||
|
||||
@@ -1468,7 +1468,7 @@ func newContractDeps(t *testing.T) *contractDeps {
|
||||
settingRepo := newStubSettingRepo()
|
||||
settingService := service.NewSettingService(settingRepo, cfg)
|
||||
|
||||
adminService := service.NewAdminService(userRepo, groupRepo, &accountRepo, proxyRepo, apiKeyRepo, redeemRepo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
|
||||
adminService := service.NewAdminService(userRepo, groupRepo, &accountRepo, proxyRepo, apiKeyRepo, redeemRepo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
|
||||
authHandler := handler.NewAuthHandler(cfg, nil, userService, settingService, nil, redeemService, nil, nil)
|
||||
apiKeyHandler := handler.NewAPIKeyHandler(apiKeyService)
|
||||
usageHandler := handler.NewUsageHandler(usageService, apiKeyService, nil, nil)
|
||||
|
||||
@@ -641,6 +641,9 @@ func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *Upd
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 渠道缓存里存了 groupID → platform 的映射,改了平台要让它失效(见函数末尾)
|
||||
previousPlatform := group.Platform
|
||||
|
||||
if input.Name != "" {
|
||||
group.Name = input.Name
|
||||
}
|
||||
@@ -900,6 +903,13 @@ func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *Upd
|
||||
s.authCacheInvalidator.InvalidateAuthCacheByGroupID(ctx, id)
|
||||
}
|
||||
|
||||
// 平台变了就失效渠道缓存:该缓存持有 groupID → platform,而渠道定价 / 模型映射 /
|
||||
// 模型白名单都按平台严格隔离。不失效的话,缓存最长 10 分钟仍按旧平台匹配,
|
||||
// 期间定价查不到会静默回落到 LiteLLM 价格表、映射与白名单也不生效。
|
||||
if group.Platform != previousPlatform && s.channelCacheInvalidator != nil {
|
||||
s.channelCacheInvalidator.InvalidateCache()
|
||||
}
|
||||
|
||||
// 如果指定了复制账号的源分组,同步绑定(替换当前分组的账号)
|
||||
if len(input.CopyAccountsFromGroupIDs) > 0 {
|
||||
// 去重源分组 IDs
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// groupPlatformRepoStub 只实现 UpdateGroup 走到的两个方法,其余靠内嵌接口占位。
|
||||
type groupPlatformRepoStub struct {
|
||||
GroupRepository
|
||||
group *Group
|
||||
updated *Group
|
||||
}
|
||||
|
||||
func (r *groupPlatformRepoStub) GetByID(_ context.Context, _ int64) (*Group, error) {
|
||||
cloned := *r.group
|
||||
return &cloned, nil
|
||||
}
|
||||
|
||||
func (r *groupPlatformRepoStub) Update(_ context.Context, group *Group) error {
|
||||
r.updated = group
|
||||
return nil
|
||||
}
|
||||
|
||||
type channelCacheInvalidatorSpy struct {
|
||||
calls int
|
||||
}
|
||||
|
||||
func (s *channelCacheInvalidatorSpy) InvalidateCache() { s.calls++ }
|
||||
|
||||
// 渠道缓存持有 groupID → platform,而渠道定价/模型映射/模型白名单都按平台严格隔离。
|
||||
// 改了分组平台却不失效缓存,最长 10 分钟内这些查找仍按旧平台匹配(静默走错价)。
|
||||
func TestUpdateGroupInvalidatesChannelCacheOnPlatformChange(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
fromPlatform string
|
||||
inputPlatform string
|
||||
wantCalls int
|
||||
}{
|
||||
{
|
||||
name: "platform changed invalidates",
|
||||
fromPlatform: PlatformAnthropic,
|
||||
inputPlatform: PlatformOpenAI,
|
||||
wantCalls: 1,
|
||||
},
|
||||
{
|
||||
name: "same platform does not invalidate",
|
||||
fromPlatform: PlatformAnthropic,
|
||||
inputPlatform: PlatformAnthropic,
|
||||
wantCalls: 0,
|
||||
},
|
||||
{
|
||||
// 请求里不带 platform 字段时不应该动缓存
|
||||
name: "platform omitted does not invalidate",
|
||||
fromPlatform: PlatformAnthropic,
|
||||
inputPlatform: "",
|
||||
wantCalls: 0,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
repo := &groupPlatformRepoStub{group: &Group{ID: 7, Name: "g", Platform: tt.fromPlatform}}
|
||||
spy := &channelCacheInvalidatorSpy{}
|
||||
svc := &adminServiceImpl{groupRepo: repo, channelCacheInvalidator: spy}
|
||||
|
||||
got, err := svc.UpdateGroup(context.Background(), 7, &UpdateGroupInput{Platform: tt.inputPlatform})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, got)
|
||||
require.Equal(t, tt.wantCalls, spy.calls)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 依赖可以不注入(例如测试或裁剪构建),此时不应 panic——缓存靠 TTL 自然重建。
|
||||
func TestUpdateGroupWithoutChannelCacheInvalidator(t *testing.T) {
|
||||
repo := &groupPlatformRepoStub{group: &Group{ID: 7, Name: "g", Platform: PlatformAnthropic}}
|
||||
svc := &adminServiceImpl{groupRepo: repo}
|
||||
|
||||
got, err := svc.UpdateGroup(context.Background(), 7, &UpdateGroupInput{Platform: PlatformOpenAI})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, PlatformOpenAI, got.Platform)
|
||||
}
|
||||
@@ -666,6 +666,14 @@ type adminServiceImpl struct {
|
||||
affiliateService adminRechargeAffiliateAccruer
|
||||
compositeRouteRepo CompositeModelRouteRepository
|
||||
compositeResolver *CompositeRouteResolver
|
||||
// 分组平台变更后用来失效渠道缓存;可为 nil(缓存会在 TTL 到期后自然重建)
|
||||
channelCacheInvalidator ChannelCacheInvalidator
|
||||
}
|
||||
|
||||
// ChannelCacheInvalidator 失效渠道缓存。
|
||||
// 窄接口,避免 admin 服务依赖整个 ChannelService——与 APIKeyAuthCacheInvalidator 同一思路。
|
||||
type ChannelCacheInvalidator interface {
|
||||
InvalidateCache()
|
||||
}
|
||||
|
||||
type adminRechargeAffiliateAccruer interface {
|
||||
@@ -699,6 +707,7 @@ func NewAdminService(
|
||||
affiliateService *AffiliateService,
|
||||
compositeRouteRepo CompositeModelRouteRepository,
|
||||
compositeResolver *CompositeRouteResolver,
|
||||
channelCacheInvalidator ChannelCacheInvalidator,
|
||||
) AdminService {
|
||||
return &adminServiceImpl{
|
||||
userRepo: userRepo,
|
||||
@@ -725,5 +734,7 @@ func NewAdminService(
|
||||
affiliateService: affiliateService,
|
||||
compositeRouteRepo: compositeRouteRepo,
|
||||
compositeResolver: compositeResolver,
|
||||
|
||||
channelCacheInvalidator: channelCacheInvalidator,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -375,6 +375,13 @@ func channelLookupPlatform(ctx context.Context, groupPlatform string) string {
|
||||
}
|
||||
return groupPlatform
|
||||
}
|
||||
|
||||
// InvalidateCache 失效并重建渠道缓存。
|
||||
// 供渠道以外、但会影响渠道缓存内容的变更调用(如分组平台变更)。
|
||||
func (s *ChannelService) InvalidateCache() {
|
||||
s.invalidateCache()
|
||||
}
|
||||
|
||||
func (s *ChannelService) invalidateCache() {
|
||||
s.cache.Store((*channelCache)(nil))
|
||||
s.cacheSF.Forget("channel_cache")
|
||||
|
||||
@@ -852,6 +852,7 @@ var ProviderSet = wire.NewSet(
|
||||
ProvideScheduledTestRunnerService,
|
||||
NewGroupCapacityService,
|
||||
NewChannelService,
|
||||
wire.Bind(new(ChannelCacheInvalidator), new(*ChannelService)),
|
||||
NewModelPricingResolver,
|
||||
NewContentModerationService,
|
||||
NewAffiliateService,
|
||||
|
||||
Reference in New Issue
Block a user