fix(grok): 修正 voice 路由推导、视频价归一化与门禁缓存驻留

- custom-voices endpoint 改由匹配到的路由模板推导(c.FullPath())。
  原实现按请求 URL 字面后缀判 /audio,voice_id 恰为 "audio" 时
  GET /custom-voices/audio 会被改写成 custom-voices/audio/audio,
  把档案查询变成音频下载。补 voice_id="audio" 的路由推导测试。

- NormalizeVideoModelPrices 不再把无法识别的分辨率静默折算成 480p:
  新增 LookupVideoBillingResolution 报告未知档位,配置解析路径丢弃并告警,
  运行时计费仍走 OrDefault 兜底。model/tier 两层遍历改为排序遍历,
  多个别名收敛到同一 family 时结果不再随 Go map 顺序漂移,冲突单价告警。

- grok free quota 软门禁缓存新增过期淘汰。条目按 account_id 键控且只写不删,
  账号下线后会驻留至进程结束;淘汰只挂在已受 TTL 约束的查询路径上,
  不影响缓存命中热路径。

- 迁移 220 清空非 Grok 分组视频价前先落快照表 groups_video_price_backup_220,
  原 UPDATE 不可回滚。

- 简化 grok_search_count 两处等价冗余的事件类型分支。
This commit is contained in:
IanShaw027 committed 2026-08-08 11:02:22 +08:00
1 parent b7112d596f
commit 07b46e93e4
9 files changed
+277 -30

No files matched your search

+18 -10
View File
@@ -292,11 +292,7 @@ func RegisterGatewayRoutes(
c.JSON(http.StatusNotFound, gin.H{"error": gin.H{"type": "not_found_error", "message": "Voice API is not supported for this platform"}})
return
}
endpoint := "custom-voices/" + c.Param("voice_id")
if strings.HasSuffix(strings.TrimRight(c.Request.URL.Path, "/"), "/audio") {
endpoint += "/audio"
}
h.OpenAIGateway.GrokVoice(c, endpoint)
h.OpenAIGateway.GrokVoice(c, grokCustomVoiceEndpoint(c))
}
gateway.GET("/custom-voices", voiceHandler("custom-voices"))
gateway.GET("/custom-voices/:voice_id/audio", customVoicePathHandler)
@@ -424,11 +420,7 @@ func RegisterGatewayRoutes(
c.JSON(http.StatusNotFound, gin.H{"error": gin.H{"type": "not_found_error", "message": "Voice API is not supported for this platform"}})
return
}
endpoint := "custom-voices/" + c.Param("voice_id")
if strings.HasSuffix(strings.TrimRight(c.Request.URL.Path, "/"), "/audio") {
endpoint += "/audio"
}
h.OpenAIGateway.GrokVoice(c, endpoint)
h.OpenAIGateway.GrokVoice(c, grokCustomVoiceEndpoint(c))
}
r.GET("/custom-voices", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), compositeTarget, requireGroupAnthropic, rootVoiceHandler("custom-voices"))
r.GET("/custom-voices/:voice_id/audio", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), compositeTarget, requireGroupAnthropic, rootCustomVoicePathHandler)
@@ -615,6 +607,22 @@ func compositeGeminiTargetPlatformMiddleware(resolver *service.CompositeRouteRes
}
}
// grokCustomVoiceEndpoint derives the upstream Voice endpoint for the
// /custom-voices/:voice_id[/audio] routes.
//
// The /audio suffix must be decided from the matched route template, not from
// the raw URL path: a voice literally named "audio" makes GET
// /custom-voices/audio match /custom-voices/:voice_id, and a raw-path suffix
// check would rewrite it to custom-voices/audio/audio — turning a profile
// lookup into an audio download.
func grokCustomVoiceEndpoint(c *gin.Context) string {
endpoint := "custom-voices/" + c.Param("voice_id")
if strings.HasSuffix(c.FullPath(), "/:voice_id/audio") {
endpoint += "/audio"
}
return endpoint
}
func compositeGeminiModelFromParams(c *gin.Context) string {
if c == nil {
return ""
@@ -220,6 +220,38 @@ func TestGatewayRoutesGrokCustomVoiceCRUDPathsAreRegistered(t *testing.T) {
}
}
func TestGrokCustomVoiceEndpointUsesRouteTemplateNotRawPath(t *testing.T) {
gin.SetMode(gin.TestMode)
router := gin.New()
var got string
capture := func(c *gin.Context) {
got = grokCustomVoiceEndpoint(c)
c.Status(http.StatusOK)
}
router.GET("/v1/custom-voices/:voice_id/audio", capture)
router.GET("/v1/custom-voices/:voice_id", capture)
for _, tc := range []struct {
path string
want string
}{
{path: "/v1/custom-voices/voice-123", want: "custom-voices/voice-123"},
{path: "/v1/custom-voices/voice-123/audio", want: "custom-voices/voice-123/audio"},
// A voice literally named "audio" matches /:voice_id, not /:voice_id/audio.
// A raw-path suffix check would turn this profile lookup into an audio download.
{path: "/v1/custom-voices/audio", want: "custom-voices/audio"},
{path: "/v1/custom-voices/audio/audio", want: "custom-voices/audio/audio"},
} {
got = ""
req := httptest.NewRequest(http.MethodGet, tc.path, nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
require.Equal(t, http.StatusOK, w.Code, "path=%s", tc.path)
require.Equal(t, tc.want, got, "path=%s", tc.path)
}
}
func TestGatewayRoutesCompositeVideoLookupsUseGrokHandler(t *testing.T) {
router := newGatewayRoutesTestRouter(service.PlatformComposite)
@@ -203,6 +203,9 @@ func filterGrokFreeQuotaAccountsCore(
}
}
}
// Only sweep on the query path: it is already TTL-bounded, so the Range
// cost stays off the cache-hit hot path.
sweepGrokFreeQuotaGateCache(cache, now, settings.cacheTTL)
}
filtered := make([]Account, 0, len(accounts))
@@ -218,6 +221,33 @@ func filterGrokFreeQuotaAccountsCore(
return filtered
}
// grokFreeQuotaGateCacheMinSweepAge floors the eviction age so a tiny cacheTTL
// does not turn the cache into a per-call re-query.
const grokFreeQuotaGateCacheMinSweepAge = 5 * time.Minute
// sweepGrokFreeQuotaGateCache drops entries far past their TTL.
//
// Entries are keyed by account ID and only ever overwritten, so an account that
// stops being scheduled (deleted, or moved off the free tier) would otherwise
// sit in the map for the process lifetime. A still-live account simply
// re-populates its entry on the next miss.
func sweepGrokFreeQuotaGateCache(cache *sync.Map, now time.Time, cacheTTL time.Duration) {
if cache == nil || cacheTTL <= 0 {
return
}
maxAge := cacheTTL * 20
if maxAge < grokFreeQuotaGateCacheMinSweepAge {
maxAge = grokFreeQuotaGateCacheMinSweepAge
}
cache.Range(func(key, value any) bool {
entry, ok := value.(grokFreeQuotaGateCacheEntry)
if !ok || now.Sub(entry.checkedAt) > maxAge {
cache.Delete(key)
}
return true
})
}
func queryGrokFreeQuotaWindowStats(ctx context.Context, usageLogRepo UsageLogRepository, accountIDs []int64, start time.Time) (map[int64]*usagestats.AccountStats, error) {
if usageLogRepo == nil {
return nil, nil
@@ -199,6 +199,56 @@ func TestGrokFreeQuotaGateIsSchedulerOnlyAdminPathUnfiltered(t *testing.T) {
require.Equal(t, int64(9), overGate.ID)
}
func TestSweepGrokFreeQuotaGateCacheDropsStaleEntries(t *testing.T) {
now := time.Now().UTC()
cacheTTL := 5 * time.Second
// maxAge is floored at grokFreeQuotaGateCacheMinSweepAge, not 20*cacheTTL.
var cache sync.Map
cache.Store(int64(1), grokFreeQuotaGateCacheEntry{tokens: 10, checkedAt: now, known: true})
cache.Store(int64(2), grokFreeQuotaGateCacheEntry{tokens: 20, checkedAt: now.Add(-time.Minute), known: true})
cache.Store(int64(3), grokFreeQuotaGateCacheEntry{tokens: 30, checkedAt: now.Add(-time.Hour), known: true})
cache.Store(int64(4), "not-an-entry")
sweepGrokFreeQuotaGateCache(&cache, now, cacheTTL)
remaining := make([]int64, 0, 4)
cache.Range(func(key, _ any) bool {
if id, ok := key.(int64); ok {
remaining = append(remaining, id)
}
return true
})
require.ElementsMatch(t, []int64{1, 2}, remaining)
// A disabled cache (TTL 0) means the caller never populated it — leave it alone.
var untouched sync.Map
untouched.Store(int64(7), grokFreeQuotaGateCacheEntry{checkedAt: now.Add(-time.Hour), known: true})
sweepGrokFreeQuotaGateCache(&untouched, now, 0)
_, stillThere := untouched.Load(int64(7))
require.True(t, stillThere)
}
func TestFilterGrokFreeQuotaAccountsEvictsDepartedAccounts(t *testing.T) {
repo := &grokFreeQuotaUsageRepoStub{stats: map[int64]*usagestats.AccountStats{
1: {Tokens: 1_000},
}}
var cache sync.Map
// Account 99 was scheduled long ago and no longer appears in any batch. Its
// entry must not survive a run that queries for a different account.
cache.Store(int64(99), grokFreeQuotaGateCacheEntry{tokens: 5, checkedAt: time.Now().UTC().Add(-2 * time.Hour), known: true})
accounts := []Account{
{ID: 1, Platform: PlatformGrok, Type: AccountTypeOAuth, Credentials: map[string]any{"subscription_tier": "FREE"}},
}
filtered := filterGrokFreeQuotaAccountsCore(context.Background(), grokFreeQuotaTestConfig(), repo, &cache, accounts)
require.Equal(t, []int64{1}, accountIDs(filtered))
_, departedStillCached := cache.Load(int64(99))
require.False(t, departedStillCached)
_, freshCached := cache.Load(int64(1))
require.True(t, freshCached)
}
func accountIDs(accounts []Account) []int64 {
ids := make([]int64, 0, len(accounts))
for i := range accounts {
+8 -12
View File
@@ -108,13 +108,12 @@ func collectGrokNativeSearchCallKeys(data []byte) []string {
if len(data) == 0 || !gjson.ValidBytes(data) {
return nil
}
eventType := strings.TrimSpace(gjson.GetBytes(data, "type").String())
switch eventType {
// An empty type means a bare item object without an SSE envelope; anything
// else that is not a completion event carries no billable call.
switch strings.TrimSpace(gjson.GetBytes(data, "type").String()) {
case "response.output_item.done", "response.completed", "response.done", "":
default:
if eventType != "" {
return nil
}
return nil
}
var keys []string
syntheticOrdinals := make(map[string]int)
@@ -161,15 +160,12 @@ func countGrokNativeSearchCallsInSSEDataWithKeys(data []byte) (int, []string) {
if len(data) == 0 || !gjson.ValidBytes(data) {
return 0, nil
}
eventType := strings.TrimSpace(gjson.GetBytes(data, "type").String())
// Count once on item completion / completed response, not on every delta.
switch eventType {
case "response.output_item.done", "response.completed", "response.done":
// An empty type is a bare item object without an SSE envelope.
switch strings.TrimSpace(gjson.GetBytes(data, "type").String()) {
case "response.output_item.done", "response.completed", "response.done", "":
default:
// Also allow bare item objects without envelope.
if eventType != "" {
return 0, nil
}
return 0, nil
}
var keys []string
n := 0
+38 -3
View File
@@ -1,6 +1,8 @@
package service
import (
"log/slog"
"sort"
"strings"
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
@@ -52,12 +54,24 @@ func CanonicalGrokImagineVideoPriceFamily(model string) string {
// NormalizeVideoModelPrices cleans and canonicalizes a per-model resolution map.
// Keys become price families; tiers use 480p/720p/1080p. Negative prices dropped.
//
// Model keys are walked in sorted order rather than in Go map order: several
// aliases can canonicalize onto the same family, and an unordered walk would
// make the winning price for a conflicting tier vary between processes.
// Unrecognized tiers are dropped with a warning instead of silently collapsing
// into the 480p bucket.
func NormalizeVideoModelPrices(in map[string]map[string]float64) map[string]map[string]float64 {
if len(in) == 0 {
return nil
}
modelKeys := make([]string, 0, len(in))
for modelKey := range in {
modelKeys = append(modelKeys, modelKey)
}
sort.Strings(modelKeys)
out := make(map[string]map[string]float64)
for modelKey, tierPrices := range in {
for _, modelKey := range modelKeys {
tierPrices := in[modelKey]
if len(tierPrices) == 0 {
continue
}
@@ -78,11 +92,32 @@ func NormalizeVideoModelPrices(in map[string]map[string]float64) map[string]map[
if normalizedTiers == nil {
normalizedTiers = make(map[string]float64)
}
for tierKey, price := range tierPrices {
tierKeys := make([]string, 0, len(tierPrices))
for tierKey := range tierPrices {
tierKeys = append(tierKeys, tierKey)
}
sort.Strings(tierKeys)
for _, tierKey := range tierKeys {
price := tierPrices[tierKey]
if price < 0 {
continue
}
tier := NormalizeVideoBillingResolutionOrDefault(tierKey)
tier, ok := LookupVideoBillingResolution(tierKey)
if !ok {
slog.Warn("video_model_prices_unknown_resolution_dropped",
"model_key", modelKey,
"family", family,
"resolution", tierKey)
continue
}
if existing, exists := normalizedTiers[tier]; exists && existing != price {
slog.Warn("video_model_prices_conflicting_tier_price",
"model_key", modelKey,
"family", family,
"resolution", tier,
"previous_price", existing,
"price", price)
}
normalizedTiers[tier] = price
}
if len(normalizedTiers) > 0 {
@@ -31,15 +31,27 @@ func NormalizeVideoBillingDurationSecondsOrDefault(durationSeconds int) int {
return durationSeconds
}
func NormalizeVideoBillingResolutionOrDefault(resolution string) string {
// LookupVideoBillingResolution 归一化分辨率并报告是否为已知档位。
// 配置解析路径必须用它而不是 OrDefault:把无法识别的档位(如 "4k"、拼错的
// "1080i")静默折算成 480p,会让管理员配的高分辨率单价被挂到低分辨率档上。
func LookupVideoBillingResolution(resolution string) (string, bool) {
switch strings.ToLower(strings.TrimSpace(resolution)) {
case "480", "480p", "sd":
return VideoBillingResolution480P
return VideoBillingResolution480P, true
case "720", "720p", "hd":
return VideoBillingResolution720P
return VideoBillingResolution720P, true
case "1080", "1080p", "full_hd", "full-hd", "fhd":
return VideoBillingResolution1080P
return VideoBillingResolution1080P, true
default:
return VideoBillingResolution480P
return "", false
}
}
// NormalizeVideoBillingResolutionOrDefault 用于运行时计费:上游回传的分辨率
// 缺失或无法识别时按最低档兜底,保证请求仍可计费。
func NormalizeVideoBillingResolutionOrDefault(resolution string) string {
if normalized, ok := LookupVideoBillingResolution(resolution); ok {
return normalized
}
return VideoBillingResolution480P
}
@@ -48,6 +48,66 @@ func TestNormalizeAndLookupVideoModelPrices(t *testing.T) {
require.Nil(t, LookupVideoModelPrice(norm, "unknown-model", "480p"))
}
func TestNormalizeVideoModelPricesDropsUnknownResolutions(t *testing.T) {
t.Parallel()
// "4k" and "1080i" are not billable tiers. Collapsing them into 480p would
// charge a 480p request at the operator's high-resolution price.
norm := NormalizeVideoModelPrices(map[string]map[string]float64{
"grok-imagine-video": {"480p": 0.05, "4k": 0.50, "1080i": 0.30},
})
require.NotNil(t, norm)
require.Equal(t, map[string]float64{VideoBillingResolution480P: 0.05}, norm[VideoPriceFamilyGrokImagineVideo])
// A model whose tiers are all unrecognized contributes no family at all.
require.Nil(t, NormalizeVideoModelPrices(map[string]map[string]float64{
"grok-imagine-video": {"4k": 0.50},
}))
}
func TestNormalizeVideoModelPricesIsDeterministicAcrossAliasConflicts(t *testing.T) {
t.Parallel()
// Both keys canonicalize onto grok-imagine-video-1.5 and disagree on 480p.
// Whichever price wins, it must be the same one on every run — a Go map walk
// would let two processes bill the same request differently.
raw := map[string]map[string]float64{
"grok-imagine-video-1.5": {"480p": 0.08},
"grok-imagine-video-1.5-preview": {"480p": 0.11},
"grok-video-1.5": {"480p": 0.09},
}
first := NormalizeVideoModelPrices(raw)
require.NotNil(t, first)
for i := 0; i < 50; i++ {
require.Equal(t, first, NormalizeVideoModelPrices(raw), "run %d diverged", i)
}
// Aliases within one model key normalize to the same tier deterministically too.
aliased := map[string]map[string]float64{
"grok-imagine-video": {"720": 0.12, "720p": 0.13, "hd": 0.14},
}
firstAliased := NormalizeVideoModelPrices(aliased)
require.NotNil(t, firstAliased)
for i := 0; i < 50; i++ {
require.Equal(t, firstAliased, NormalizeVideoModelPrices(aliased), "run %d diverged", i)
}
}
func TestLookupVideoBillingResolutionReportsUnknownTiers(t *testing.T) {
t.Parallel()
for _, in := range []string{"480", "480p", "SD", "720", "hd", "1080", "full-hd", " fhd "} {
normalized, ok := LookupVideoBillingResolution(in)
require.True(t, ok, "input=%q", in)
require.NotEmpty(t, normalized)
}
for _, in := range []string{"", "4k", "1080i", "2160p", "potato"} {
normalized, ok := LookupVideoBillingResolution(in)
require.False(t, ok, "input=%q", in)
require.Empty(t, normalized)
}
// Runtime billing still needs a tier for unrecognized upstream values.
require.Equal(t, VideoBillingResolution480P, NormalizeVideoBillingResolutionOrDefault("4k"))
require.Equal(t, VideoBillingResolution1080P, NormalizeVideoBillingResolutionOrDefault("full_hd"))
}
func TestVideoModelPriceMissingTierFallsBackToFlatTierPrice(t *testing.T) {
t.Parallel()
flat720P := 0.7
@@ -2,6 +2,30 @@
-- Columns match migrations 170/217 (video_price_* / video_model_prices), not a
-- separate allow_video_generation flag which was never applied on this branch.
-- Snapshot before clearing. The UPDATE below is irreversible, and an operator
-- who deliberately priced video on a non-Grok group would otherwise lose that
-- configuration with no way to recover it. CREATE TABLE IF NOT EXISTS ... AS
-- SELECT is a no-op on re-run, so this stays idempotent.
CREATE TABLE IF NOT EXISTS groups_video_price_backup_220 AS
SELECT id AS group_id,
platform,
video_price_480p,
video_price_720p,
video_price_1080p,
video_model_prices,
now() AS backed_up_at
FROM groups
WHERE platform IS DISTINCT FROM 'grok'
AND (
video_price_480p IS NOT NULL
OR video_price_720p IS NOT NULL
OR video_price_1080p IS NOT NULL
OR video_model_prices IS NOT NULL
);
COMMENT ON TABLE groups_video_price_backup_220 IS
'迁移 220 清空非 Grok 分组视频价前的快照。确认无需回滚后可安全 DROP;回滚方式:UPDATE groups g SET video_price_480p = b.video_price_480p, ... FROM groups_video_price_backup_220 b WHERE g.id = b.group_id';
UPDATE groups
SET video_price_480p = NULL,
video_price_720p = NULL,