feat(usage): track native compaction requests

This commit is contained in:
Jarl
2026-08-28 23:32:11 +08:00
parent 7b693ae429
commit 1cc6999ada
42 changed files with 797 additions and 69 deletions
@@ -262,13 +262,18 @@ func (h *DashboardHandler) GetUsageTrend(c *gin.Context) {
return return
} }
} }
upstreamModelMismatch, err := parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch") nativeCompactionV2, err := parseOptionalBoolDashboardFilter(c, "native_compaction_v2")
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
upstreamModelMismatch, err = parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch")
if err != nil { if err != nil {
response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false") response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false")
return return
} }
trend, hit, err := h.getUsageTrendCached(c.Request.Context(), startTime, endTime, granularity, userID, apiKeyID, accountID, groupID, model, requestType, stream, billingType, upstreamModelMismatch) trend, hit, err := h.getUsageTrendCached(c.Request.Context(), startTime, endTime, granularity, userID, apiKeyID, accountID, groupID, model, requestType, stream, nativeCompactionV2, billingType, upstreamModelMismatch)
if err != nil { if err != nil {
response.Error(c, 500, "Failed to get usage trend") response.Error(c, 500, "Failed to get usage trend")
return return
@@ -349,13 +354,18 @@ func (h *DashboardHandler) GetModelStats(c *gin.Context) {
return return
} }
} }
upstreamModelMismatch, err := parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch") nativeCompactionV2, err := parseOptionalBoolDashboardFilter(c, "native_compaction_v2")
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
upstreamModelMismatch, err = parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch")
if err != nil { if err != nil {
response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false") response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false")
return return
} }
stats, hit, err := h.getModelStatsCached(c.Request.Context(), startTime, endTime, userID, apiKeyID, accountID, groupID, modelSource, requestType, stream, billingType, upstreamModelMismatch) stats, hit, err := h.getModelStatsCached(c.Request.Context(), startTime, endTime, userID, apiKeyID, accountID, groupID, modelSource, requestType, stream, nativeCompactionV2, billingType, upstreamModelMismatch)
if err != nil { if err != nil {
response.Error(c, 500, "Failed to get model statistics") response.Error(c, 500, "Failed to get model statistics")
return return
@@ -426,13 +436,18 @@ func (h *DashboardHandler) GetGroupStats(c *gin.Context) {
return return
} }
} }
upstreamModelMismatch, err := parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch") nativeCompactionV2, err := parseOptionalBoolDashboardFilter(c, "native_compaction_v2")
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
upstreamModelMismatch, err = parseOptionalBoolDashboardFilter(c, "upstream_model_mismatch")
if err != nil { if err != nil {
response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false") response.BadRequest(c, "Invalid upstream_model_mismatch value, use true or false")
return return
} }
stats, hit, err := h.getGroupStatsCached(c.Request.Context(), startTime, endTime, userID, apiKeyID, accountID, groupID, requestType, stream, billingType, upstreamModelMismatch) stats, hit, err := h.getGroupStatsCached(c.Request.Context(), startTime, endTime, userID, apiKeyID, accountID, groupID, requestType, stream, nativeCompactionV2, billingType, upstreamModelMismatch)
if err != nil { if err != nil {
response.Error(c, 500, "Failed to get group statistics") response.Error(c, 500, "Failed to get group statistics")
return return
@@ -701,6 +716,14 @@ func (h *DashboardHandler) GetUserBreakdown(c *gin.Context) {
dim.Stream = &s dim.Stream = &s
} }
} }
if v := strings.TrimSpace(c.Query("native_compaction_v2")); v != "" {
value, err := strconv.ParseBool(v)
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
dim.NativeCompactionV2 = &value
}
if v := c.Query("billing_type"); v != "" { if v := c.Query("billing_type"); v != "" {
if bt, err := strconv.ParseInt(v, 10, 8); err == nil { if bt, err := strconv.ParseInt(v, 10, 8); err == nil {
btVal := int8(bt) btVal := int8(bt)
@@ -15,16 +15,19 @@ import (
type dashboardUsageRepoCapture struct { type dashboardUsageRepoCapture struct {
service.UsageLogRepository service.UsageLogRepository
trendRequestType *int16 trendRequestType *int16
trendStream *bool trendStream *bool
modelRequestType *int16 trendNativeCompaction *bool
modelStream *bool modelRequestType *int16
trendMismatch *bool modelStream *bool
modelMismatch *bool modelNativeCompaction *bool
groupMismatch *bool groupNativeCompaction *bool
rankingLimit int trendMismatch *bool
ranking []usagestats.UserSpendingRankingItem modelMismatch *bool
rankingTotal float64 groupMismatch *bool
rankingLimit int
ranking []usagestats.UserSpendingRankingItem
rankingTotal float64
} }
func (s *dashboardUsageRepoCapture) GetUsageTrendWithUsageFilters( func (s *dashboardUsageRepoCapture) GetUsageTrendWithUsageFilters(
@@ -35,6 +38,7 @@ func (s *dashboardUsageRepoCapture) GetUsageTrendWithUsageFilters(
) ([]usagestats.TrendDataPoint, error) { ) ([]usagestats.TrendDataPoint, error) {
s.trendRequestType = filters.RequestType s.trendRequestType = filters.RequestType
s.trendStream = filters.Stream s.trendStream = filters.Stream
s.trendNativeCompaction = filters.NativeCompactionV2
s.trendMismatch = filters.UpstreamModelMismatch s.trendMismatch = filters.UpstreamModelMismatch
return []usagestats.TrendDataPoint{}, nil return []usagestats.TrendDataPoint{}, nil
} }
@@ -62,6 +66,7 @@ func (s *dashboardUsageRepoCapture) GetModelStatsWithUsageFiltersBySource(
) ([]usagestats.ModelStat, error) { ) ([]usagestats.ModelStat, error) {
s.modelRequestType = filters.RequestType s.modelRequestType = filters.RequestType
s.modelStream = filters.Stream s.modelStream = filters.Stream
s.modelNativeCompaction = filters.NativeCompactionV2
s.modelMismatch = filters.UpstreamModelMismatch s.modelMismatch = filters.UpstreamModelMismatch
return []usagestats.ModelStat{}, nil return []usagestats.ModelStat{}, nil
} }
@@ -71,6 +76,7 @@ func (s *dashboardUsageRepoCapture) GetGroupStatsWithUsageFilters(
startTime, endTime time.Time, startTime, endTime time.Time,
filters usagestats.UsageLogFilters, filters usagestats.UsageLogFilters,
) ([]usagestats.GroupStat, error) { ) ([]usagestats.GroupStat, error) {
s.groupNativeCompaction = filters.NativeCompactionV2
s.groupMismatch = filters.UpstreamModelMismatch s.groupMismatch = filters.UpstreamModelMismatch
return []usagestats.GroupStat{}, nil return []usagestats.GroupStat{}, nil
} }
@@ -208,6 +214,48 @@ func TestDashboardModelStatsValidModelSource(t *testing.T) {
require.Equal(t, http.StatusOK, rec.Code) require.Equal(t, http.StatusOK, rec.Code)
} }
func TestDashboardNativeCompactionFilterPropagatesAlongsideTransport(t *testing.T) {
resetDashboardReadCachesForTest()
repo := &dashboardUsageRepoCapture{}
router := newDashboardRequestTypeTestRouter(repo)
for _, path := range []string{
"/admin/dashboard/trend?request_type=stream&native_compaction_v2=true",
"/admin/dashboard/models?request_type=stream&native_compaction_v2=true",
"/admin/dashboard/groups?request_type=stream&native_compaction_v2=true",
} {
req := httptest.NewRequest(http.MethodGet, path, nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code, path)
}
require.NotNil(t, repo.trendNativeCompaction)
require.True(t, *repo.trendNativeCompaction)
require.NotNil(t, repo.modelNativeCompaction)
require.True(t, *repo.modelNativeCompaction)
require.NotNil(t, repo.groupNativeCompaction)
require.True(t, *repo.groupNativeCompaction)
require.NotNil(t, repo.trendRequestType)
require.Equal(t, int16(service.RequestTypeStream), *repo.trendRequestType)
}
func TestDashboardNativeCompactionFilterRejectsInvalidBoolean(t *testing.T) {
repo := &dashboardUsageRepoCapture{}
router := newDashboardRequestTypeTestRouter(repo)
for _, path := range []string{
"/admin/dashboard/trend?native_compaction_v2=invalid",
"/admin/dashboard/models?native_compaction_v2=invalid",
"/admin/dashboard/groups?native_compaction_v2=invalid",
} {
req := httptest.NewRequest(http.MethodGet, path, nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusBadRequest, rec.Code, path)
}
}
func TestDashboardModelAuditFilterPropagatesToTrendModelAndGroupQueries(t *testing.T) { func TestDashboardModelAuditFilterPropagatesToTrendModelAndGroupQueries(t *testing.T) {
resetDashboardReadCachesForTest() resetDashboardReadCachesForTest()
repo := &dashboardUsageRepoCapture{} repo := &dashboardUsageRepoCapture{}
@@ -28,6 +28,7 @@ type dashboardTrendCacheKey struct {
Model string `json:"model"` Model string `json:"model"`
RequestType *int16 `json:"request_type"` RequestType *int16 `json:"request_type"`
Stream *bool `json:"stream"` Stream *bool `json:"stream"`
NativeCompactionV2 *bool `json:"native_compaction_v2"`
BillingType *int8 `json:"billing_type"` BillingType *int8 `json:"billing_type"`
UpstreamModelMismatch *bool `json:"upstream_model_mismatch"` UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
} }
@@ -42,6 +43,7 @@ type dashboardModelGroupCacheKey struct {
ModelSource string `json:"model_source,omitempty"` ModelSource string `json:"model_source,omitempty"`
RequestType *int16 `json:"request_type"` RequestType *int16 `json:"request_type"`
Stream *bool `json:"stream"` Stream *bool `json:"stream"`
NativeCompactionV2 *bool `json:"native_compaction_v2"`
BillingType *int8 `json:"billing_type"` BillingType *int8 `json:"billing_type"`
UpstreamModelMismatch *bool `json:"upstream_model_mismatch"` UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
} }
@@ -85,6 +87,7 @@ func (h *DashboardHandler) getUsageTrendCached(
model string, model string,
requestType *int16, requestType *int16,
stream *bool, stream *bool,
nativeCompactionV2 *bool,
billingType *int8, billingType *int8,
upstreamModelMismatch *bool, upstreamModelMismatch *bool,
) ([]usagestats.TrendDataPoint, bool, error) { ) ([]usagestats.TrendDataPoint, bool, error) {
@@ -99,13 +102,14 @@ func (h *DashboardHandler) getUsageTrendCached(
Model: model, Model: model,
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
NativeCompactionV2: nativeCompactionV2,
BillingType: billingType, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}) })
entry, hit, err := dashboardTrendCache.GetOrLoad(key, func() (any, error) { entry, hit, err := dashboardTrendCache.GetOrLoad(key, func() (any, error) {
return h.dashboardService.GetUsageTrendWithUsageFilters(ctx, startTime, endTime, granularity, usagestats.UsageLogFilters{ return h.dashboardService.GetUsageTrendWithUsageFilters(ctx, startTime, endTime, granularity, usagestats.UsageLogFilters{
UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID, UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
Model: model, RequestType: requestType, Stream: stream, BillingType: billingType, Model: model, RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}) })
}) })
@@ -123,6 +127,7 @@ func (h *DashboardHandler) getModelStatsCached(
modelSource string, modelSource string,
requestType *int16, requestType *int16,
stream *bool, stream *bool,
nativeCompactionV2 *bool,
billingType *int8, billingType *int8,
upstreamModelMismatch *bool, upstreamModelMismatch *bool,
) ([]usagestats.ModelStat, bool, error) { ) ([]usagestats.ModelStat, bool, error) {
@@ -136,13 +141,14 @@ func (h *DashboardHandler) getModelStatsCached(
ModelSource: usagestats.NormalizeModelSource(modelSource), ModelSource: usagestats.NormalizeModelSource(modelSource),
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
NativeCompactionV2: nativeCompactionV2,
BillingType: billingType, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}) })
entry, hit, err := dashboardModelStatsCache.GetOrLoad(key, func() (any, error) { entry, hit, err := dashboardModelStatsCache.GetOrLoad(key, func() (any, error) {
return h.dashboardService.GetModelStatsWithUsageFiltersBySource(ctx, startTime, endTime, usagestats.UsageLogFilters{ return h.dashboardService.GetModelStatsWithUsageFiltersBySource(ctx, startTime, endTime, usagestats.UsageLogFilters{
UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID, UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
RequestType: requestType, Stream: stream, BillingType: billingType, RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}, modelSource) }, modelSource)
}) })
@@ -159,6 +165,7 @@ func (h *DashboardHandler) getGroupStatsCached(
userID, apiKeyID, accountID, groupID int64, userID, apiKeyID, accountID, groupID int64,
requestType *int16, requestType *int16,
stream *bool, stream *bool,
nativeCompactionV2 *bool,
billingType *int8, billingType *int8,
upstreamModelMismatch *bool, upstreamModelMismatch *bool,
) ([]usagestats.GroupStat, bool, error) { ) ([]usagestats.GroupStat, bool, error) {
@@ -171,13 +178,14 @@ func (h *DashboardHandler) getGroupStatsCached(
GroupID: groupID, GroupID: groupID,
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
NativeCompactionV2: nativeCompactionV2,
BillingType: billingType, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}) })
entry, hit, err := dashboardGroupStatsCache.GetOrLoad(key, func() (any, error) { entry, hit, err := dashboardGroupStatsCache.GetOrLoad(key, func() (any, error) {
return h.dashboardService.GetGroupStatsWithUsageFilters(ctx, startTime, endTime, usagestats.UsageLogFilters{ return h.dashboardService.GetGroupStatsWithUsageFilters(ctx, startTime, endTime, usagestats.UsageLogFilters{
UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID, UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
RequestType: requestType, Stream: stream, BillingType: billingType, RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
}) })
}) })
@@ -44,6 +44,7 @@ type dashboardSnapshotV2Filters struct {
Model string Model string
RequestType *int16 RequestType *int16
Stream *bool Stream *bool
NativeCompactionV2 *bool
BillingType *int8 BillingType *int8
UpstreamModelMismatch *bool UpstreamModelMismatch *bool
} }
@@ -59,6 +60,7 @@ type dashboardSnapshotV2CacheKey struct {
Model string `json:"model"` Model string `json:"model"`
RequestType *int16 `json:"request_type"` RequestType *int16 `json:"request_type"`
Stream *bool `json:"stream"` Stream *bool `json:"stream"`
NativeCompactionV2 *bool `json:"native_compaction_v2"`
BillingType *int8 `json:"billing_type"` BillingType *int8 `json:"billing_type"`
UpstreamModelMismatch *bool `json:"upstream_model_mismatch"` UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
IncludeStats bool `json:"include_stats"` IncludeStats bool `json:"include_stats"`
@@ -105,6 +107,7 @@ func (h *DashboardHandler) GetSnapshotV2(c *gin.Context) {
Model: filters.Model, Model: filters.Model,
RequestType: filters.RequestType, RequestType: filters.RequestType,
Stream: filters.Stream, Stream: filters.Stream,
NativeCompactionV2: filters.NativeCompactionV2,
BillingType: filters.BillingType, BillingType: filters.BillingType,
UpstreamModelMismatch: filters.UpstreamModelMismatch, UpstreamModelMismatch: filters.UpstreamModelMismatch,
IncludeStats: includeStats, IncludeStats: includeStats,
@@ -186,6 +189,7 @@ func (h *DashboardHandler) buildSnapshotV2Response(
filters.Model, filters.Model,
filters.RequestType, filters.RequestType,
filters.Stream, filters.Stream,
filters.NativeCompactionV2,
filters.BillingType, filters.BillingType,
filters.UpstreamModelMismatch, filters.UpstreamModelMismatch,
) )
@@ -207,6 +211,7 @@ func (h *DashboardHandler) buildSnapshotV2Response(
usagestats.ModelSourceRequested, usagestats.ModelSourceRequested,
filters.RequestType, filters.RequestType,
filters.Stream, filters.Stream,
filters.NativeCompactionV2,
filters.BillingType, filters.BillingType,
filters.UpstreamModelMismatch, filters.UpstreamModelMismatch,
) )
@@ -227,6 +232,7 @@ func (h *DashboardHandler) buildSnapshotV2Response(
filters.GroupID, filters.GroupID,
filters.RequestType, filters.RequestType,
filters.Stream, filters.Stream,
filters.NativeCompactionV2,
filters.BillingType, filters.BillingType,
filters.UpstreamModelMismatch, filters.UpstreamModelMismatch,
) )
@@ -296,6 +302,14 @@ func parseDashboardSnapshotV2Filters(c *gin.Context) (*dashboardSnapshotV2Filter
filters.Stream = &streamVal filters.Stream = &streamVal
} }
if nativeCompactionV2Str := strings.TrimSpace(c.Query("native_compaction_v2")); nativeCompactionV2Str != "" {
value, err := strconv.ParseBool(nativeCompactionV2Str)
if err != nil {
return nil, err
}
filters.NativeCompactionV2 = &value
}
if billingTypeStr := strings.TrimSpace(c.Query("billing_type")); billingTypeStr != "" { if billingTypeStr := strings.TrimSpace(c.Query("billing_type")); billingTypeStr != "" {
v, err := strconv.ParseInt(billingTypeStr, 10, 8) v, err := strconv.ParseInt(billingTypeStr, 10, 8)
if err != nil { if err != nil {
@@ -132,6 +132,12 @@ func (h *UsageHandler) List(c *gin.Context) {
stream = &val stream = &val
} }
nativeCompactionV2, err := parseOptionalBoolDashboardFilter(c, "native_compaction_v2")
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
var billingType *int8 var billingType *int8
if billingTypeStr := c.Query("billing_type"); billingTypeStr != "" { if billingTypeStr := c.Query("billing_type"); billingTypeStr != "" {
val, err := strconv.ParseInt(billingTypeStr, 10, 8) val, err := strconv.ParseInt(billingTypeStr, 10, 8)
@@ -192,6 +198,7 @@ func (h *UsageHandler) List(c *gin.Context) {
ModelFilterSource: usagestats.ModelSourceRequested, ModelFilterSource: usagestats.ModelSourceRequested,
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
NativeCompactionV2: nativeCompactionV2,
BillingType: billingType, BillingType: billingType,
BillingMode: billingMode, BillingMode: billingMode,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
@@ -276,6 +283,12 @@ func (h *UsageHandler) Stats(c *gin.Context) {
stream = &val stream = &val
} }
nativeCompactionV2, err := parseOptionalBoolDashboardFilter(c, "native_compaction_v2")
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return
}
var billingType *int8 var billingType *int8
if billingTypeStr := c.Query("billing_type"); billingTypeStr != "" { if billingTypeStr := c.Query("billing_type"); billingTypeStr != "" {
val, err := strconv.ParseInt(billingTypeStr, 10, 8) val, err := strconv.ParseInt(billingTypeStr, 10, 8)
@@ -344,6 +357,7 @@ func (h *UsageHandler) Stats(c *gin.Context) {
ModelFilterSource: usagestats.ModelSourceRequested, ModelFilterSource: usagestats.ModelSourceRequested,
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
NativeCompactionV2: nativeCompactionV2,
BillingType: billingType, BillingType: billingType,
BillingMode: billingMode, BillingMode: billingMode,
UpstreamModelMismatch: upstreamModelMismatch, UpstreamModelMismatch: upstreamModelMismatch,
@@ -95,6 +95,32 @@ func TestAdminUsageListInvalidStream(t *testing.T) {
require.Equal(t, http.StatusBadRequest, rec.Code) require.Equal(t, http.StatusBadRequest, rec.Code)
} }
func TestAdminUsageListNativeCompactionFilter(t *testing.T) {
repo := &adminUsageRepoCapture{}
router := newAdminUsageRequestTypeTestRouter(repo)
req := httptest.NewRequest(http.MethodGet, "/admin/usage?request_type=stream&native_compaction_v2=true", nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code)
require.NotNil(t, repo.listFilters.RequestType)
require.Equal(t, int16(service.RequestTypeStream), *repo.listFilters.RequestType)
require.NotNil(t, repo.listFilters.NativeCompactionV2)
require.True(t, *repo.listFilters.NativeCompactionV2)
}
func TestAdminUsageListInvalidNativeCompactionFilter(t *testing.T) {
repo := &adminUsageRepoCapture{}
router := newAdminUsageRequestTypeTestRouter(repo)
req := httptest.NewRequest(http.MethodGet, "/admin/usage?native_compaction_v2=oops", nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusBadRequest, rec.Code)
}
func TestAdminUsageListExactTotalTrue(t *testing.T) { func TestAdminUsageListExactTotalTrue(t *testing.T) {
repo := &adminUsageRepoCapture{} repo := &adminUsageRepoCapture{}
router := newAdminUsageRequestTypeTestRouter(repo) router := newAdminUsageRequestTypeTestRouter(repo)
@@ -144,6 +170,19 @@ func TestAdminUsageStatsRequestTypePriority(t *testing.T) {
require.Nil(t, repo.statsFilters.Stream) require.Nil(t, repo.statsFilters.Stream)
} }
func TestAdminUsageStatsNativeCompactionFilter(t *testing.T) {
repo := &adminUsageRepoCapture{}
router := newAdminUsageRequestTypeTestRouter(repo)
req := httptest.NewRequest(http.MethodGet, "/admin/usage/stats?native_compaction_v2=true", nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code)
require.NotNil(t, repo.statsFilters.NativeCompactionV2)
require.True(t, *repo.statsFilters.NativeCompactionV2)
}
func TestAdminUsageStatsUsesRequestedModelForDisplayModelFilter(t *testing.T) { func TestAdminUsageStatsUsesRequestedModelForDisplayModelFilter(t *testing.T) {
repo := &adminUsageRepoCapture{} repo := &adminUsageRepoCapture{}
router := newAdminUsageRequestTypeTestRouter(repo) router := newAdminUsageRequestTypeTestRouter(repo)
@@ -21,6 +21,7 @@ type usageStatsCacheKeyData struct {
BillingMode string `json:"billing_mode"` BillingMode string `json:"billing_mode"`
RequestType *int16 `json:"request_type"` RequestType *int16 `json:"request_type"`
Stream *bool `json:"stream"` Stream *bool `json:"stream"`
NativeCompactionV2 *bool `json:"native_compaction_v2"`
BillingType *int8 `json:"billing_type"` BillingType *int8 `json:"billing_type"`
UpstreamModelMismatch *bool `json:"upstream_model_mismatch"` UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
} }
@@ -45,6 +46,7 @@ func usageStatsCacheKey(filters usagestats.UsageLogFilters) string {
BillingMode: filters.BillingMode, BillingMode: filters.BillingMode,
RequestType: filters.RequestType, RequestType: filters.RequestType,
Stream: filters.Stream, Stream: filters.Stream,
NativeCompactionV2: filters.NativeCompactionV2,
BillingType: filters.BillingType, BillingType: filters.BillingType,
UpstreamModelMismatch: filters.UpstreamModelMismatch, UpstreamModelMismatch: filters.UpstreamModelMismatch,
}) })
+1
View File
@@ -667,6 +667,7 @@ func usageLogFromServiceUser(l *service.UsageLog) UsageLog {
RequestType: requestType.String(), RequestType: requestType.String(),
Stream: stream, Stream: stream,
OpenAIWSMode: openAIWSMode, OpenAIWSMode: openAIWSMode,
NativeCompactionV2: l.NativeCompactionV2,
DurationMs: l.DurationMs, DurationMs: l.DurationMs,
FirstTokenMs: l.FirstTokenMs, FirstTokenMs: l.FirstTokenMs,
ImageCount: l.ImageCount, ImageCount: l.ImageCount,
@@ -28,6 +28,27 @@ func TestUsageLogFromService_IncludesOpenAIWSMode(t *testing.T) {
require.False(t, UsageLogFromServiceAdmin(httpLog).OpenAIWSMode) require.False(t, UsageLogFromServiceAdmin(httpLog).OpenAIWSMode)
} }
func TestUsageLogFromService_PreservesNativeCompactionAndStream(t *testing.T) {
t.Parallel()
log := &service.UsageLog{
RequestID: "resp_compaction",
Model: "gpt-5.6-sol",
RequestType: service.RequestTypeStream,
Stream: true,
NativeCompactionV2: true,
}
userDTO := UsageLogFromService(log)
adminDTO := UsageLogFromServiceAdmin(log)
require.Equal(t, "stream", userDTO.RequestType)
require.True(t, userDTO.Stream)
require.True(t, userDTO.NativeCompactionV2)
require.Equal(t, "stream", adminDTO.RequestType)
require.True(t, adminDTO.Stream)
require.True(t, adminDTO.NativeCompactionV2)
}
func TestUsageLogFromService_PrefersRequestTypeForLegacyFields(t *testing.T) { func TestUsageLogFromService_PrefersRequestTypeForLegacyFields(t *testing.T) {
t.Parallel() t.Parallel()
+5 -2
View File
@@ -522,8 +522,11 @@ type UsageLog struct {
RequestType string `json:"request_type"` RequestType string `json:"request_type"`
Stream bool `json:"stream"` Stream bool `json:"stream"`
OpenAIWSMode bool `json:"openai_ws_mode"` OpenAIWSMode bool `json:"openai_ws_mode"`
DurationMs *int `json:"duration_ms"` // NativeCompactionV2 is true only for requests positively identified at
FirstTokenMs *int `json:"first_token_ms"` // runtime as the native OpenAI remote compaction v2 wire.
NativeCompactionV2 bool `json:"native_compaction_v2"`
DurationMs *int `json:"duration_ms"`
FirstTokenMs *int `json:"first_token_ms"`
// 图片生成字段 // 图片生成字段
ImageCount int `json:"image_count"` ImageCount int `json:"image_count"`
@@ -778,6 +778,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel), ChannelUsageFields: clientRequestedUsageFields(c, channelMapping, reqModel, res.UpstreamModel),
PricingAt: pricingAt, PricingAt: pricingAt,
CyberBlocked: cyberBlocked, CyberBlocked: cyberBlocked,
NativeCompactionV2: nativeV2,
}); err != nil { }); err != nil {
logger.L().With( logger.L().With(
zap.String("component", "handler.openai_gateway.responses"), zap.String("component", "handler.openai_gateway.responses"),
@@ -3675,6 +3676,7 @@ func (h *OpenAIGatewayHandler) recordCyberPolicyIfMarked(c *gin.Context, apiKey
} }
// 提前拍成标量,避免在下方 goroutine 内访问 gin.Context。 // 提前拍成标量,避免在下方 goroutine 内访问 gin.Context。
sessionID := service.ExtractClientSessionID(c) sessionID := service.ExtractClientSessionID(c)
nativeCompactionV2 := service.IsOpenAINativeCompactionV2(c)
apiKeyPrefix := "" apiKeyPrefix := ""
if apiKey != nil { if apiKey != nil {
apiKeyPrefix = keyPrefix(apiKey.Key, 8) apiKeyPrefix = keyPrefix(apiKey.Key, 8)
@@ -3742,6 +3744,7 @@ func (h *OpenAIGatewayHandler) recordCyberPolicyIfMarked(c *gin.Context, apiKey
SessionID: sessionID, SessionID: sessionID,
RequestPayloadHash: requestPayloadHash, RequestPayloadHash: requestPayloadHash,
APIKeyService: apiKeySvc, APIKeyService: apiKeySvc,
NativeCompactionV2: nativeCompactionV2,
ChannelUsageFields: channelFields, ChannelUsageFields: channelFields,
}) })
} }
+22 -11
View File
@@ -126,6 +126,16 @@ func (h *UsageHandler) parseUserUsageFilters(c *gin.Context, requireRange bool)
stream = &val stream = &val
} }
var nativeCompactionV2 *bool
if raw := strings.TrimSpace(c.Query("native_compaction_v2")); raw != "" {
value, err := strconv.ParseBool(raw)
if err != nil {
response.BadRequest(c, "Invalid native_compaction_v2 value, use true or false")
return nil, false
}
nativeCompactionV2 = &value
}
var billingType *int8 var billingType *int8
if billingTypeStr := strings.TrimSpace(c.Query("billing_type")); billingTypeStr != "" { if billingTypeStr := strings.TrimSpace(c.Query("billing_type")); billingTypeStr != "" {
val, err := strconv.ParseInt(billingTypeStr, 10, 8) val, err := strconv.ParseInt(billingTypeStr, 10, 8)
@@ -195,17 +205,18 @@ func (h *UsageHandler) parseUserUsageFilters(c *gin.Context, requireRange bool)
return &userUsageFilters{ return &userUsageFilters{
Filters: usagestats.UsageLogFilters{ Filters: usagestats.UsageLogFilters{
UserID: subject.UserID, UserID: subject.UserID,
APIKeyID: apiKeyID, APIKeyID: apiKeyID,
GroupID: groupID, GroupID: groupID,
Model: strings.TrimSpace(c.Query("model")), Model: strings.TrimSpace(c.Query("model")),
ModelFilterSource: usagestats.ModelSourceRequested, ModelFilterSource: usagestats.ModelSourceRequested,
RequestType: requestType, RequestType: requestType,
Stream: stream, Stream: stream,
BillingType: billingType, NativeCompactionV2: nativeCompactionV2,
BillingMode: billingMode, BillingType: billingType,
StartTime: startPtr, BillingMode: billingMode,
EndTime: endPtr, StartTime: startPtr,
EndTime: endPtr,
}, },
StartTime: derefTime(startPtr), StartTime: derefTime(startPtr),
EndTime: derefTime(endPtr), EndTime: derefTime(endPtr),
@@ -131,6 +131,32 @@ func TestUserUsageListInvalidStream(t *testing.T) {
require.Equal(t, http.StatusBadRequest, rec.Code) require.Equal(t, http.StatusBadRequest, rec.Code)
} }
func TestUserUsageListNativeCompactionFilter(t *testing.T) {
repo := &userUsageRepoCapture{}
router := newUserUsageRequestTypeTestRouter(repo)
req := httptest.NewRequest(http.MethodGet, "/usage?request_type=stream&native_compaction_v2=true", nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code)
require.NotNil(t, repo.listFilters.RequestType)
require.Equal(t, int16(service.RequestTypeStream), *repo.listFilters.RequestType)
require.NotNil(t, repo.listFilters.NativeCompactionV2)
require.True(t, *repo.listFilters.NativeCompactionV2)
}
func TestUserUsageListInvalidNativeCompactionFilter(t *testing.T) {
repo := &userUsageRepoCapture{}
router := newUserUsageRequestTypeTestRouter(repo)
req := httptest.NewRequest(http.MethodGet, "/usage?native_compaction_v2=unknown", nil)
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusBadRequest, rec.Code)
}
func TestUserUsageListAdvancedFilters(t *testing.T) { func TestUserUsageListAdvancedFilters(t *testing.T) {
repo := &userUsageRepoCapture{} repo := &userUsageRepoCapture{}
router := newUserUsageRequestTypeTestRouter(repo) router := newUserUsageRequestTypeTestRouter(repo)
@@ -189,6 +215,7 @@ func TestUserUsageListKeepsUserBillingAndIPWithoutAdminCostFields(t *testing.T)
AccountID: 5, AccountID: 5,
RequestID: "req_user_billing", RequestID: "req_user_billing",
Model: "gpt-5", Model: "gpt-5",
NativeCompactionV2: true,
InputCost: 0.01, InputCost: 0.01,
OutputCost: 0.02, OutputCost: 0.02,
CacheCreationCost: 0.03, CacheCreationCost: 0.03,
@@ -219,6 +246,7 @@ func TestUserUsageListKeepsUserBillingAndIPWithoutAdminCostFields(t *testing.T)
require.Contains(t, body, `"total_cost":0.1`) require.Contains(t, body, `"total_cost":0.1`)
require.Contains(t, body, `"actual_cost":0.08`) require.Contains(t, body, `"actual_cost":0.08`)
require.Contains(t, body, `"rate_multiplier":0.8`) require.Contains(t, body, `"rate_multiplier":0.8`)
require.Contains(t, body, `"native_compaction_v2":true`)
require.Contains(t, body, `"ip_address":"203.0.113.10"`) require.Contains(t, body, `"ip_address":"203.0.113.10"`)
require.NotContains(t, body, "upstream_endpoint") require.NotContains(t, body, "upstream_endpoint")
require.NotContains(t, body, "account_rate_multiplier") require.NotContains(t, body, "account_rate_multiplier")
@@ -186,12 +186,13 @@ type UserBreakdownDimension struct {
Endpoint string // filter by endpoint value (non-empty to enable) Endpoint string // filter by endpoint value (non-empty to enable)
EndpointType string // "inbound", "upstream", or "path" EndpointType string // "inbound", "upstream", or "path"
// Additional filter conditions // Additional filter conditions
UserID int64 // filter by user_id (>0 to enable) UserID int64 // filter by user_id (>0 to enable)
APIKeyID int64 // filter by api_key_id (>0 to enable) APIKeyID int64 // filter by api_key_id (>0 to enable)
AccountID int64 // filter by account_id (>0 to enable) AccountID int64 // filter by account_id (>0 to enable)
RequestType *int16 // filter by request_type (non-nil to enable) RequestType *int16 // filter by request_type (non-nil to enable)
Stream *bool // filter by stream flag (non-nil to enable) Stream *bool // filter by stream flag (non-nil to enable)
BillingType *int8 // filter by billing_type (non-nil to enable) NativeCompactionV2 *bool // filter by native compaction v2 flag (non-nil to enable)
BillingType *int8 // filter by billing_type (non-nil to enable)
// SortBy 指定排序列(空 = 默认按 actual_cost)。合法值由 repo 层 allowlist 校验。 // SortBy 指定排序列(空 = 默认按 actual_cost)。合法值由 repo 层 allowlist 校验。
SortBy string SortBy string
} }
@@ -278,6 +279,7 @@ type UsageLogFilters struct {
ModelFilterSource string ModelFilterSource string
RequestType *int16 RequestType *int16
Stream *bool Stream *bool
NativeCompactionV2 *bool
BillingType *int8 BillingType *int8
BillingMode string BillingMode string
UpstreamModelMismatch *bool UpstreamModelMismatch *bool
@@ -69,6 +69,8 @@ func TestMigrationsRunner_IsIdempotent_AndSchemaIsUpToDate(t *testing.T) {
requireColumn(t, tx, "usage_logs", "billing_type", "smallint", 0, false) requireColumn(t, tx, "usage_logs", "billing_type", "smallint", 0, false)
requireColumn(t, tx, "usage_logs", "request_type", "smallint", 0, false) requireColumn(t, tx, "usage_logs", "request_type", "smallint", 0, false)
requireColumn(t, tx, "usage_logs", "openai_ws_mode", "boolean", 0, false) requireColumn(t, tx, "usage_logs", "openai_ws_mode", "boolean", 0, false)
requireColumn(t, tx, "usage_logs", "native_compaction_v2", "boolean", 0, false)
requireColumnDefaultContains(t, tx, "usage_logs", "native_compaction_v2", "false")
requireColumn(t, tx, "usage_logs", "image_input_size", "character varying", 32, true) requireColumn(t, tx, "usage_logs", "image_input_size", "character varying", 32, true)
requireColumn(t, tx, "usage_logs", "image_output_size", "character varying", 32, true) requireColumn(t, tx, "usage_logs", "image_output_size", "character varying", 32, true)
requireColumn(t, tx, "usage_logs", "image_size_source", "character varying", 16, true) requireColumn(t, tx, "usage_logs", "image_size_source", "character varying", 16, true)
@@ -199,6 +199,27 @@ func appendRequestTypeOrStreamQueryFilter(query string, args []any, requestType
return query, args return query, args
} }
func appendNativeCompactionV2WhereCondition(conditions []string, args []any, nativeCompactionV2 *bool, alias string) ([]string, []any) {
if nativeCompactionV2 == nil {
return conditions, args
}
column := "native_compaction_v2"
if alias != "" {
column = alias + "." + column
}
conditions = append(conditions, fmt.Sprintf("%s = $%d", column, len(args)+1))
args = append(args, *nativeCompactionV2)
return conditions, args
}
func appendNativeCompactionV2QueryFilter(query string, args []any, nativeCompactionV2 *bool, alias string) (string, []any) {
conditions, args := appendNativeCompactionV2WhereCondition(nil, args, nativeCompactionV2, alias)
if len(conditions) == 0 {
return query, args
}
return query + " AND " + conditions[0], args
}
// buildRequestTypeFilterCondition 在 request_type 过滤时兼容 legacy 字段,避免历史数据漏查。 // buildRequestTypeFilterCondition 在 request_type 过滤时兼容 legacy 字段,避免历史数据漏查。
func buildRequestTypeFilterCondition(startArgIndex int, requestType int16) (string, []any) { func buildRequestTypeFilterCondition(startArgIndex int, requestType int16) (string, []any) {
return buildRequestTypeFilterConditionWithAlias(startArgIndex, requestType, "") return buildRequestTypeFilterConditionWithAlias(startArgIndex, requestType, "")
@@ -77,3 +77,26 @@ func TestGetUserBreakdownStatsRequestTypeIncludesLegacyFallback(t *testing.T) {
require.Empty(t, rows) require.Empty(t, rows)
require.NoError(t, mock.ExpectationsWereMet()) require.NoError(t, mock.ExpectationsWereMet())
} }
func TestGetUserBreakdownStatsFiltersNativeCompactionV2(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
start := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
end := start.Add(24 * time.Hour)
nativeCompactionV2 := true
mock.ExpectQuery(regexp.QuoteMeta("AND ul.native_compaction_v2 = $3")).
WithArgs(start, end, true).
WillReturnRows(sqlmock.NewRows([]string{
"user_id", "email", "requests", "input_tokens", "output_tokens",
"cache_tokens", "total_tokens", "cost", "actual_cost", "account_cost",
}))
rows, err := repo.GetUserBreakdownStats(context.Background(), start, end, usagestats.UserBreakdownDimension{
NativeCompactionV2: &nativeCompactionV2,
}, 0)
require.NoError(t, err)
require.Empty(t, rows)
require.NoError(t, mock.ExpectationsWereMet())
}
@@ -83,6 +83,7 @@ var usageLogInsertArgTypes = [...]string{
"text", // billing_mode "text", // billing_mode
"numeric", // account_stats_cost "numeric", // account_stats_cost
"text", // session_id "text", // session_id
"boolean", // native_compaction_v2
"timestamptz", // created_at "timestamptz", // created_at
} }
@@ -282,6 +283,7 @@ func (r *usageLogRepository) createSingle(ctx context.Context, sqlq sqlExecutor,
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) VALUES ( ) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $1, $2, $3, $4, $5, $6, $7, $8, $9,
@@ -289,7 +291,7 @@ func (r *usageLogRepository) createSingle(ctx context.Context, sqlq sqlExecutor,
$12, $13, $14, $15, $12, $13, $14, $15,
$16, $17, $18, $19, $16, $17, $18, $19,
$20, $21, $22, $23, $24, $25, $20, $21, $22, $23, $24, $25,
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60 $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61
) )
ON CONFLICT (request_id, api_key_id) DO NOTHING ON CONFLICT (request_id, api_key_id) DO NOTHING
RETURNING id, created_at RETURNING id, created_at
@@ -740,6 +742,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) AS (VALUES `) ) AS (VALUES `)
@@ -833,6 +836,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) )
SELECT SELECT
@@ -895,6 +899,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
FROM input FROM input
ON CONFLICT (request_id, api_key_id) DO NOTHING ON CONFLICT (request_id, api_key_id) DO NOTHING
@@ -997,6 +1002,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) AS (VALUES `) ) AS (VALUES `)
@@ -1085,6 +1091,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) )
SELECT SELECT
@@ -1147,6 +1154,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
FROM input FROM input
ON CONFLICT (request_id, api_key_id) DO NOTHING ON CONFLICT (request_id, api_key_id) DO NOTHING
@@ -1217,6 +1225,7 @@ func execUsageLogInsertNoResult(ctx context.Context, sqlq sqlExecutor, prepared
billing_mode, billing_mode,
account_stats_cost, account_stats_cost,
session_id, session_id,
native_compaction_v2,
created_at created_at
) VALUES ( ) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $1, $2, $3, $4, $5, $6, $7, $8, $9,
@@ -1224,7 +1233,7 @@ func execUsageLogInsertNoResult(ctx context.Context, sqlq sqlExecutor, prepared
$12, $13, $14, $15, $12, $13, $14, $15,
$16, $17, $18, $19, $16, $17, $18, $19,
$20, $21, $22, $23, $24, $25, $20, $21, $22, $23, $24, $25,
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60 $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61
) )
ON CONFLICT (request_id, api_key_id) DO NOTHING ON CONFLICT (request_id, api_key_id) DO NOTHING
`, prepared.args...) `, prepared.args...)
@@ -1345,6 +1354,7 @@ func prepareUsageLogInsert(log *service.UsageLog) usageLogInsertPrepared {
billingMode, billingMode,
log.AccountStatsCost, // account_stats_cost log.AccountStatsCost, // account_stats_cost
sessionID, // session_id sessionID, // session_id
log.NativeCompactionV2,
createdAt, createdAt,
}, },
} }
@@ -19,7 +19,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/service" "github.com/Wei-Shaw/sub2api/internal/service"
) )
const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, upstream_response_model, upstream_model_mismatch, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, image_input_tokens, image_input_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, video_count, video_resolution, video_duration_seconds, service_tier, reasoning_effort, requested_reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, long_context_billing_applied, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, session_id, created_at" const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, upstream_response_model, upstream_model_mismatch, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, image_input_tokens, image_input_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, video_count, video_resolution, video_duration_seconds, service_tier, reasoning_effort, requested_reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, long_context_billing_applied, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, session_id, native_compaction_v2, created_at"
func (r *usageLogRepository) GetByID(ctx context.Context, id int64) (log *service.UsageLog, err error) { func (r *usageLogRepository) GetByID(ctx context.Context, id int64) (log *service.UsageLog, err error) {
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE id = $1" query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE id = $1"
@@ -122,6 +122,7 @@ func (r *usageLogRepository) ListWithFilters(ctx context.Context, params paginat
} }
conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource) conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource)
conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream) conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream)
conditions, args = appendNativeCompactionV2WhereCondition(conditions, args, filters.NativeCompactionV2, "")
if filters.BillingType != nil { if filters.BillingType != nil {
conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1)) conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1))
args = append(args, int16(*filters.BillingType)) args = append(args, int16(*filters.BillingType))
@@ -499,6 +500,7 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
billingMode sql.NullString billingMode sql.NullString
accountStatsCost sql.NullFloat64 accountStatsCost sql.NullFloat64
sessionID sql.NullString sessionID sql.NullString
nativeCompactionV2 bool
createdAt time.Time createdAt time.Time
) )
@@ -563,6 +565,7 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
&billingMode, &billingMode,
&accountStatsCost, &accountStatsCost,
&sessionID, &sessionID,
&nativeCompactionV2,
&createdAt, &createdAt,
); err != nil { ); err != nil {
return nil, err return nil, err
@@ -595,6 +598,7 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
AccountRateMultiplier: nullFloat64Ptr(accountRateMultiplier), AccountRateMultiplier: nullFloat64Ptr(accountRateMultiplier),
BillingType: int8(billingType), BillingType: int8(billingType),
RequestType: service.RequestTypeFromInt16(requestTypeRaw), RequestType: service.RequestTypeFromInt16(requestTypeRaw),
NativeCompactionV2: nativeCompactionV2,
ImageCount: imageCount, ImageCount: imageCount,
VideoCount: videoCount, VideoCount: videoCount,
CacheTTLOverridden: cacheTTLOverridden, CacheTTLOverridden: cacheTTLOverridden,
@@ -100,6 +100,7 @@ func TestUsageLogRepositoryCreateSyncRequestTypeAndLegacyFields(t *testing.T) {
sqlmock.AnyArg(), // billing_mode sqlmock.AnyArg(), // billing_mode
sqlmock.AnyArg(), // account_stats_cost sqlmock.AnyArg(), // account_stats_cost
sqlmock.AnyArg(), // session_id sqlmock.AnyArg(), // session_id
log.NativeCompactionV2,
createdAt, createdAt,
). ).
WillReturnRows(sqlmock.NewRows([]string{"id", "created_at"}).AddRow(int64(99), createdAt)) WillReturnRows(sqlmock.NewRows([]string{"id", "created_at"}).AddRow(int64(99), createdAt))
@@ -193,6 +194,7 @@ func TestUsageLogRepositoryCreate_PersistsServiceTier(t *testing.T) {
sqlmock.AnyArg(), // billing_mode sqlmock.AnyArg(), // billing_mode
sqlmock.AnyArg(), // account_stats_cost sqlmock.AnyArg(), // account_stats_cost
sqlmock.AnyArg(), // session_id sqlmock.AnyArg(), // session_id
log.NativeCompactionV2,
createdAt, createdAt,
). ).
WillReturnRows(sqlmock.NewRows([]string{"id", "created_at"}).AddRow(int64(100), createdAt)) WillReturnRows(sqlmock.NewRows([]string{"id", "created_at"}).AddRow(int64(100), createdAt))
@@ -258,6 +260,30 @@ func TestPrepareUsageLogInsert_ArgCountMatchesTypes(t *testing.T) {
require.Len(t, prepared.args, len(usageLogInsertArgTypes)) require.Len(t, prepared.args, len(usageLogInsertArgTypes))
} }
func TestPrepareUsageLogInsert_PersistsNativeCompactionV2WithoutChangingRequestType(t *testing.T) {
log := &service.UsageLog{
UserID: 1,
APIKeyID: 2,
AccountID: 3,
RequestID: "req-native-compaction-v2",
Model: "gpt-5",
RequestedModel: "gpt-5",
RequestType: service.RequestTypeStream,
NativeCompactionV2: true,
CreatedAt: time.Date(2025, 1, 5, 13, 0, 0, 0, time.UTC),
}
prepared := prepareUsageLogInsert(log)
require.Len(t, prepared.args, len(usageLogInsertArgTypes))
require.Equal(t, "boolean", usageLogInsertArgTypes[len(usageLogInsertArgTypes)-2])
require.Equal(t, true, prepared.args[len(prepared.args)-2])
require.Equal(t, int16(service.RequestTypeStream), prepared.args[30])
require.Equal(t, service.RequestTypeStream, log.RequestType)
require.True(t, log.Stream)
require.False(t, log.OpenAIWSMode)
}
func TestPrepareUsageLogInsert_PersistsImageSizeMetadata(t *testing.T) { func TestPrepareUsageLogInsert_PersistsImageSizeMetadata(t *testing.T) {
imageSize := "4K" imageSize := "4K"
inputSize := "1024x1024" inputSize := "1024x1024"
@@ -380,6 +406,26 @@ func TestUsageLogRepositoryListWithFiltersRequestTypePriority(t *testing.T) {
require.NoError(t, mock.ExpectationsWereMet()) require.NoError(t, mock.ExpectationsWereMet())
} }
func TestUsageLogRepositoryListWithFiltersNativeCompactionV2(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
nativeCompactionV2 := true
filters := usagestats.UsageLogFilters{NativeCompactionV2: &nativeCompactionV2, ExactTotal: true}
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM usage_logs WHERE native_compaction_v2 = \\$1").
WithArgs(true).
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(int64(0)))
mock.ExpectQuery("SELECT .* FROM usage_logs WHERE native_compaction_v2 = \\$1 ORDER BY id DESC LIMIT \\$2 OFFSET \\$3").
WithArgs(true, 20, 0).
WillReturnRows(sqlmock.NewRows([]string{"id"}))
logs, page, err := repo.ListWithFilters(context.Background(), pagination.PaginationParams{Page: 1, PageSize: 20}, filters)
require.NoError(t, err)
require.Empty(t, logs)
require.NotNil(t, page)
require.NoError(t, mock.ExpectationsWereMet())
}
func TestUsageLogRepositoryListWithFiltersRequestID(t *testing.T) { func TestUsageLogRepositoryListWithFiltersRequestID(t *testing.T) {
db, mock := newSQLMock(t) db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db} repo := &usageLogRepository{sql: db}
@@ -457,6 +503,74 @@ func TestUsageLogRepositoryGetUsageTrendWithUsageFiltersRequestedModelSource(t *
require.NoError(t, mock.ExpectationsWereMet()) require.NoError(t, mock.ExpectationsWereMet())
} }
func TestUsageLogRepositoryUsageAggregatesFilterNativeCompactionV2(t *testing.T) {
start := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
end := start.Add(24 * time.Hour)
nativeCompactionV2 := true
filters := usagestats.UsageLogFilters{NativeCompactionV2: &nativeCompactionV2}
t.Run("stats", func(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
mock.ExpectQuery("(?s)FROM usage_logs\\s+WHERE native_compaction_v2 = \\$1.*GROUP BY GROUPING SETS").
WithArgs(true).
WillReturnRows(sqlmock.NewRows([]string{
"inbound_grouped", "upstream_grouped", "inbound_endpoint", "upstream_endpoint",
"requests", "input_tokens", "output_tokens", "cache_creation_tokens", "cache_read_tokens",
"cost", "actual_cost", "account_cost", "avg_duration_ms",
}))
_, err := repo.GetStatsWithFilters(context.Background(), filters)
require.NoError(t, err)
require.NoError(t, mock.ExpectationsWereMet())
})
t.Run("trend bypasses preaggregate", func(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
mock.ExpectQuery("(?s)FROM usage_logs.*AND native_compaction_v2 = \\$3").
WithArgs(start, end, true).
WillReturnRows(sqlmock.NewRows([]string{"date", "requests", "input_tokens", "output_tokens", "cache_creation_tokens", "cache_read_tokens", "total_tokens", "cost", "actual_cost"}))
_, err := repo.GetUsageTrendWithUsageFilters(context.Background(), start, end, "day", filters)
require.NoError(t, err)
require.NoError(t, mock.ExpectationsWereMet())
})
t.Run("model", func(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
mock.ExpectQuery("(?s)FROM usage_logs.*AND native_compaction_v2 = \\$3").
WithArgs(start, end, true).
WillReturnRows(sqlmock.NewRows([]string{
"model", "requests", "input_tokens", "output_tokens", "cache_creation_tokens",
"cache_read_tokens", "total_tokens", "cost", "actual_cost", "account_cost",
}))
_, err := repo.GetModelStatsWithUsageFiltersBySource(context.Background(), start, end, filters, usagestats.ModelSourceRequested)
require.NoError(t, err)
require.NoError(t, mock.ExpectationsWereMet())
})
t.Run("group", func(t *testing.T) {
db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db}
mock.ExpectQuery("(?s)FROM usage_logs ul.*AND ul.native_compaction_v2 = \\$3").
WithArgs(start, end, true).
WillReturnRows(sqlmock.NewRows([]string{"group_id", "group_name", "requests", "total_tokens", "cost", "actual_cost", "account_cost"}))
_, err := repo.GetGroupStatsWithUsageFilters(context.Background(), start, end, filters)
require.NoError(t, err)
require.NoError(t, mock.ExpectationsWereMet())
})
}
func TestShouldUsePreaggregatedTrendRejectsNativeCompactionV2Filter(t *testing.T) {
nativeCompactionV2 := true
require.True(t, shouldUsePreaggregatedTrend("day", 0, 0, 0, 0, "", nil, nil, nil, "", nil, nil))
require.False(t, shouldUsePreaggregatedTrend("day", 0, 0, 0, 0, "", nil, nil, nil, "", nil, &nativeCompactionV2))
}
func TestUsageLogRepositoryGetModelStatsWithFiltersRequestTypePriority(t *testing.T) { func TestUsageLogRepositoryGetModelStatsWithFiltersRequestTypePriority(t *testing.T) {
db, mock := newSQLMock(t) db, mock := newSQLMock(t)
repo := &usageLogRepository{sql: db} repo := &usageLogRepository{sql: db}
@@ -842,6 +956,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
sql.NullString{}, sql.NullString{},
sql.NullFloat64{}, sql.NullFloat64{},
sql.NullString{}, sql.NullString{},
false, // native_compaction_v2
now, now,
}}) }})
require.NoError(t, err) require.NoError(t, err)
@@ -920,6 +1035,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
sql.NullString{}, // billing_mode sql.NullString{}, // billing_mode
sql.NullFloat64{}, // account_stats_cost sql.NullFloat64{}, // account_stats_cost
sql.NullString{}, // session_id sql.NullString{}, // session_id
false, // native_compaction_v2
now, now,
}}) }})
require.NoError(t, err) require.NoError(t, err)
@@ -981,6 +1097,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
sql.NullString{}, // billing_mode sql.NullString{}, // billing_mode
sql.NullFloat64{}, // account_stats_cost sql.NullFloat64{}, // account_stats_cost
sql.NullString{}, // session_id sql.NullString{}, // session_id
true, // native_compaction_v2
now, now,
}}) }})
require.NoError(t, err) require.NoError(t, err)
@@ -989,6 +1106,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
require.Equal(t, service.RequestTypeStream, log.RequestType) require.Equal(t, service.RequestTypeStream, log.RequestType)
require.True(t, log.Stream) require.True(t, log.Stream)
require.False(t, log.OpenAIWSMode) require.False(t, log.OpenAIWSMode)
require.True(t, log.NativeCompactionV2)
}) })
t.Run("service_tier_is_scanned", func(t *testing.T) { t.Run("service_tier_is_scanned", func(t *testing.T) {
@@ -1042,6 +1160,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
sql.NullString{}, // billing_mode sql.NullString{}, // billing_mode
sql.NullFloat64{}, // account_stats_cost sql.NullFloat64{}, // account_stats_cost
sql.NullString{}, // session_id sql.NullString{}, // session_id
false, // native_compaction_v2
now, now,
}}) }})
require.NoError(t, err) require.NoError(t, err)
@@ -677,6 +677,7 @@ func (r *usageLogRepository) GetStatsWithFilters(ctx context.Context, filters Us
} }
conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource) conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource)
conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream) conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream)
conditions, args = appendNativeCompactionV2WhereCondition(conditions, args, filters.NativeCompactionV2, "")
if filters.BillingType != nil { if filters.BillingType != nil {
conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1)) conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1))
args = append(args, int16(*filters.BillingType)) args = append(args, int16(*filters.BillingType))
@@ -265,20 +265,20 @@ func (r *usageLogRepository) GetUserUsageTrendByUserID(ctx context.Context, user
// GetUserModelStats 获取指定用户的模型统计 // GetUserModelStats 获取指定用户的模型统计
func (r *usageLogRepository) GetUserModelStats(ctx context.Context, userID int64, startTime, endTime time.Time) (results []ModelStat, err error) { func (r *usageLogRepository) GetUserModelStats(ctx context.Context, userID int64, startTime, endTime time.Time) (results []ModelStat, err error) {
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, 0, 0, 0, "", nil, nil, nil, usagestats.ModelSourceRequested, "", nil) return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, 0, 0, 0, "", nil, nil, nil, usagestats.ModelSourceRequested, "", nil, nil)
} }
// GetUsageTrendWithFilters returns usage trend data with optional filters // GetUsageTrendWithFilters returns usage trend data with optional filters
func (r *usageLogRepository) GetUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) (results []TrendDataPoint, err error) { func (r *usageLogRepository) GetUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) (results []TrendDataPoint, err error) {
return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, userID, apiKeyID, accountID, groupID, model, "", requestType, stream, billingType, "", nil) return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, userID, apiKeyID, accountID, groupID, model, "", requestType, stream, billingType, "", nil, nil)
} }
func (r *usageLogRepository) GetUsageTrendWithUsageFilters(ctx context.Context, startTime, endTime time.Time, granularity string, filters UsageLogFilters) (results []TrendDataPoint, err error) { func (r *usageLogRepository) GetUsageTrendWithUsageFilters(ctx context.Context, startTime, endTime time.Time, granularity string, filters UsageLogFilters) (results []TrendDataPoint, err error) {
return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.ModelFilterSource, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode, filters.UpstreamModelMismatch) return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.ModelFilterSource, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode, filters.UpstreamModelMismatch, filters.NativeCompactionV2)
} }
func (r *usageLogRepository) getUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, modelSource string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool) (results []TrendDataPoint, err error) { func (r *usageLogRepository) getUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, modelSource string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool, nativeCompactionV2 *bool) (results []TrendDataPoint, err error) {
if shouldUsePreaggregatedTrend(granularity, userID, apiKeyID, accountID, groupID, model, requestType, stream, billingType, billingMode, upstreamModelMismatch) { if shouldUsePreaggregatedTrend(granularity, userID, apiKeyID, accountID, groupID, model, requestType, stream, billingType, billingMode, upstreamModelMismatch, nativeCompactionV2) {
aggregated, aggregatedErr := r.getUsageTrendFromAggregates(ctx, startTime, endTime, granularity) aggregated, aggregatedErr := r.getUsageTrendFromAggregates(ctx, startTime, endTime, granularity)
if aggregatedErr == nil && len(aggregated) > 0 { if aggregatedErr == nil && len(aggregated) > 0 {
return aggregated, nil return aggregated, nil
@@ -321,6 +321,7 @@ func (r *usageLogRepository) getUsageTrendWithFilters(ctx context.Context, start
} }
query, args = appendUsageLogModelQueryFilter(query, args, model, modelSource) query, args = appendUsageLogModelQueryFilter(query, args, model, modelSource)
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream) query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
query, args = appendNativeCompactionV2QueryFilter(query, args, nativeCompactionV2, "")
if billingType != nil { if billingType != nil {
query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1) query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1)
args = append(args, int16(*billingType)) args = append(args, int16(*billingType))
@@ -351,7 +352,7 @@ func (r *usageLogRepository) getUsageTrendWithFilters(ctx context.Context, start
return results, nil return results, nil
} }
func shouldUsePreaggregatedTrend(granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool) bool { func shouldUsePreaggregatedTrend(granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool, nativeCompactionV2 *bool) bool {
if granularity != "day" && granularity != "hour" { if granularity != "day" && granularity != "hour" {
return false return false
} }
@@ -364,7 +365,8 @@ func shouldUsePreaggregatedTrend(granularity string, userID, apiKeyID, accountID
stream == nil && stream == nil &&
billingType == nil && billingType == nil &&
billingMode == "" && billingMode == "" &&
upstreamModelMismatch == nil upstreamModelMismatch == nil &&
nativeCompactionV2 == nil
} }
func (r *usageLogRepository) getUsageTrendFromAggregates(ctx context.Context, startTime, endTime time.Time, granularity string) (results []TrendDataPoint, err error) { func (r *usageLogRepository) getUsageTrendFromAggregates(ctx context.Context, startTime, endTime time.Time, granularity string) (results []TrendDataPoint, err error) {
@@ -429,20 +431,20 @@ func (r *usageLogRepository) getUsageTrendFromAggregates(ctx context.Context, st
// GetModelStatsWithFilters returns model statistics with optional filters // GetModelStatsWithFilters returns model statistics with optional filters
func (r *usageLogRepository) GetModelStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []ModelStat, err error) { func (r *usageLogRepository) GetModelStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []ModelStat, err error) {
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, usagestats.ModelSourceRequested, "", nil) return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, usagestats.ModelSourceRequested, "", nil, nil)
} }
// GetModelStatsWithFiltersBySource returns model statistics with optional filters and model source dimension. // GetModelStatsWithFiltersBySource returns model statistics with optional filters and model source dimension.
// source: requested | upstream | mapping. // source: requested | upstream | mapping.
func (r *usageLogRepository) GetModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8, source string) (results []ModelStat, err error) { func (r *usageLogRepository) GetModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8, source string) (results []ModelStat, err error) {
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, source, "", nil) return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, source, "", nil, nil)
} }
func (r *usageLogRepository) GetModelStatsWithUsageFiltersBySource(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters, source string) (results []ModelStat, err error) { func (r *usageLogRepository) GetModelStatsWithUsageFiltersBySource(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters, source string) (results []ModelStat, err error) {
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, source, filters.BillingMode, filters.UpstreamModelMismatch) return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, source, filters.BillingMode, filters.UpstreamModelMismatch, filters.NativeCompactionV2)
} }
func (r *usageLogRepository) getModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, source string, billingMode string, upstreamModelMismatch *bool) (results []ModelStat, err error) { func (r *usageLogRepository) getModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, source string, billingMode string, upstreamModelMismatch *bool, nativeCompactionV2 *bool) (results []ModelStat, err error) {
actualCostExpr := "COALESCE(SUM(actual_cost), 0) as actual_cost" actualCostExpr := "COALESCE(SUM(actual_cost), 0) as actual_cost"
// 当仅按 account_id 聚合时,实际费用使用账号倍率(total_cost * account_rate_multiplier)。 // 当仅按 account_id 聚合时,实际费用使用账号倍率(total_cost * account_rate_multiplier)。
if accountID > 0 && userID == 0 && apiKeyID == 0 { if accountID > 0 && userID == 0 && apiKeyID == 0 {
@@ -489,6 +491,7 @@ func (r *usageLogRepository) getModelStatsWithFiltersBySource(ctx context.Contex
args = append(args, model) args = append(args, model)
} }
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream) query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
query, args = appendNativeCompactionV2QueryFilter(query, args, nativeCompactionV2, "")
if billingType != nil { if billingType != nil {
query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1) query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1)
args = append(args, int16(*billingType)) args = append(args, int16(*billingType))
@@ -521,14 +524,14 @@ func (r *usageLogRepository) getModelStatsWithFiltersBySource(ctx context.Contex
// GetGroupStatsWithFilters returns group usage statistics with optional filters // GetGroupStatsWithFilters returns group usage statistics with optional filters
func (r *usageLogRepository) GetGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []usagestats.GroupStat, err error) { func (r *usageLogRepository) GetGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []usagestats.GroupStat, err error) {
return r.getGroupStatsWithFilters(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, "", nil) return r.getGroupStatsWithFilters(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, "", nil, nil)
} }
func (r *usageLogRepository) GetGroupStatsWithUsageFilters(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters) (results []usagestats.GroupStat, err error) { func (r *usageLogRepository) GetGroupStatsWithUsageFilters(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters) (results []usagestats.GroupStat, err error) {
return r.getGroupStatsWithFilters(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode, filters.UpstreamModelMismatch) return r.getGroupStatsWithFilters(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode, filters.UpstreamModelMismatch, filters.NativeCompactionV2)
} }
func (r *usageLogRepository) getGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool) (results []usagestats.GroupStat, err error) { func (r *usageLogRepository) getGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string, upstreamModelMismatch *bool, nativeCompactionV2 *bool) (results []usagestats.GroupStat, err error) {
query := ` query := `
SELECT SELECT
COALESCE(ul.group_id, 0) as group_id, COALESCE(ul.group_id, 0) as group_id,
@@ -566,6 +569,7 @@ func (r *usageLogRepository) getGroupStatsWithFilters(ctx context.Context, start
args = append(args, model) args = append(args, model)
} }
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream) query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
query, args = appendNativeCompactionV2QueryFilter(query, args, nativeCompactionV2, "ul")
if billingType != nil { if billingType != nil {
query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1) query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1)
args = append(args, int16(*billingType)) args = append(args, int16(*billingType))
@@ -663,6 +667,7 @@ func (r *usageLogRepository) GetUserBreakdownStats(ctx context.Context, startTim
query += fmt.Sprintf(" AND ul.stream = $%d", len(args)+1) query += fmt.Sprintf(" AND ul.stream = $%d", len(args)+1)
args = append(args, *dim.Stream) args = append(args, *dim.Stream)
} }
query, args = appendNativeCompactionV2QueryFilter(query, args, dim.NativeCompactionV2, "ul")
if dim.BillingType != nil { if dim.BillingType != nil {
query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1) query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1)
args = append(args, *dim.BillingType) args = append(args, *dim.BillingType)
@@ -70,6 +70,13 @@ func isOpenAINativeCompactionV2(c *gin.Context) bool {
return c.GetBool(openAINativeCompactionV2Key) return c.GetBool(openAINativeCompactionV2Key)
} }
// IsOpenAINativeCompactionV2 reports whether the handler identified this
// request as the native remote compaction v2 wire. It exposes only the
// request-scoped boolean marker; no request payload is retained.
func IsOpenAINativeCompactionV2(c *gin.Context) bool {
return isOpenAINativeCompactionV2(c)
}
// ensureOpenAIRemoteCompactionV2BetaFeature 确保出站 x-codex-beta-features // ensureOpenAIRemoteCompactionV2BetaFeature 确保出站 x-codex-beta-features
// 头包含 remote_compaction_v2。真实 Codex 发送 compaction_trigger 时总会同时 // 头包含 remote_compaction_v2。真实 Codex 发送 compaction_trigger 时总会同时
// 携带该协商头(codex-rs build_model_client_beta_features_header 对该 feature // 携带该协商头(codex-rs build_model_client_beta_features_header 对该 feature
@@ -5,10 +5,20 @@ package service
import ( import (
"testing" "testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/tidwall/gjson" "github.com/tidwall/gjson"
) )
func TestIsOpenAINativeCompactionV2ReadsOnlyRequestMarker(t *testing.T) {
c, _ := gin.CreateTestContext(nil)
require.False(t, IsOpenAINativeCompactionV2(c))
MarkOpenAINativeCompactionV2(c)
require.True(t, IsOpenAINativeCompactionV2(c))
require.False(t, IsOpenAINativeCompactionV2(nil))
}
func TestHasCompactionTriggerInInput_DetectsCompactSignal(t *testing.T) { func TestHasCompactionTriggerInInput_DetectsCompactSignal(t *testing.T) {
body := []byte(`{ body := []byte(`{
"model":"gpt-5.5", "model":"gpt-5.5",
@@ -2866,6 +2866,30 @@ func TestGatewayServiceCalculateRecordUsageCost_GroupImagePriceOverridesChannelI
require.InDelta(t, 0.042, cost.ActualCost, 1e-12) require.InDelta(t, 0.042, cost.ActualCost, 1e-12)
} }
func TestRecordUsageKeepsCompactionSemanticFlagOrthogonalToTransport(t *testing.T) {
logStub := &openAIRecordUsageLogRepoStub{inserted: true}
svc := newOpenAIRecordUsageServiceForTest(logStub, &openAIRecordUsageUserRepoStub{}, &openAIRecordUsageSubRepoStub{}, nil)
apiKey := &APIKey{ID: 20, UserID: 21, Group: &Group{RateMultiplier: 1}}
require.NoError(t, svc.RecordUsage(context.Background(), &OpenAIRecordUsageInput{
NativeCompactionV2: true,
Result: &OpenAIForwardResult{
Model: "gpt-5",
Stream: true,
Duration: time.Second,
Usage: OpenAIUsage{InputTokens: 100, OutputTokens: 10},
},
APIKey: apiKey,
User: &User{ID: 21},
Account: &Account{ID: 22},
}))
require.NotNil(t, logStub.lastLog)
require.True(t, logStub.lastLog.Stream)
require.True(t, logStub.lastLog.NativeCompactionV2)
require.Equal(t, RequestTypeStream, RequestTypeFromLegacy(logStub.lastLog.Stream, logStub.lastLog.OpenAIWSMode))
}
func TestRecordUsageMarksCyberRequestType(t *testing.T) { func TestRecordUsageMarksCyberRequestType(t *testing.T) {
logStub := &openAIRecordUsageLogRepoStub{inserted: true} logStub := &openAIRecordUsageLogRepoStub{inserted: true}
userStub := &openAIRecordUsageUserRepoStub{} userStub := &openAIRecordUsageUserRepoStub{}
@@ -2874,7 +2898,8 @@ func TestRecordUsageMarksCyberRequestType(t *testing.T) {
svc := newOpenAIRecordUsageServiceForTest(logStub, userStub, subStub, rateStub) svc := newOpenAIRecordUsageServiceForTest(logStub, userStub, subStub, rateStub)
in := &OpenAIRecordUsageInput{ in := &OpenAIRecordUsageInput{
CyberBlocked: true, CyberBlocked: true,
NativeCompactionV2: true,
Result: &OpenAIForwardResult{ Result: &OpenAIForwardResult{
Model: "gpt-5", Model: "gpt-5",
Duration: time.Second, Duration: time.Second,
@@ -2887,6 +2912,7 @@ func TestRecordUsageMarksCyberRequestType(t *testing.T) {
require.NoError(t, svc.RecordUsage(context.Background(), in)) require.NoError(t, svc.RecordUsage(context.Background(), in))
require.NotNil(t, logStub.lastLog) require.NotNil(t, logStub.lastLog)
require.Equal(t, RequestTypeCyberBlocked, logStub.lastLog.RequestType) require.Equal(t, RequestTypeCyberBlocked, logStub.lastLog.RequestType)
require.True(t, logStub.lastLog.NativeCompactionV2)
require.Equal(t, 100, logStub.lastLog.InputTokens, "计费 token 不变(正常计费)") require.Equal(t, 100, logStub.lastLog.InputTokens, "计费 token 不变(正常计费)")
} }
@@ -39,6 +39,10 @@ type OpenAIRecordUsageInput struct {
PricingAt time.Time PricingAt time.Time
// CyberBlocked 为 true 时把该用量行标记为 cyber(request_type=cyber),计费逻辑不变。 // CyberBlocked 为 true 时把该用量行标记为 cyber(request_type=cyber),计费逻辑不变。
CyberBlocked bool CyberBlocked bool
// NativeCompactionV2 is an orthogonal semantic flag captured by the
// Responses handler from stream=true + compaction_trigger. It never stores
// the request payload and does not replace the transport request type.
NativeCompactionV2 bool
ChannelUsageFields ChannelUsageFields
} }
@@ -63,6 +67,7 @@ type CyberPolicyUsageInput struct {
SessionID string SessionID string
RequestPayloadHash string RequestPayloadHash string
APIKeyService APIKeyQuotaUpdater APIKeyService APIKeyQuotaUpdater
NativeCompactionV2 bool
ChannelUsageFields ChannelUsageFields
} }
@@ -100,6 +105,7 @@ func (s *OpenAIGatewayService) RecordCyberPolicyUsageLog(ctx context.Context, in
APIKeyService: in.APIKeyService, APIKeyService: in.APIKeyService,
ChannelUsageFields: in.ChannelUsageFields, ChannelUsageFields: in.ChannelUsageFields,
CyberBlocked: true, CyberBlocked: true,
NativeCompactionV2: in.NativeCompactionV2,
}); err != nil { }); err != nil {
logger.LegacyPrintf("service.openai_gateway", "cyber usage record failed: request_id=%s err=%v", in.RequestID, err) logger.LegacyPrintf("service.openai_gateway", "cyber usage record failed: request_id=%s err=%v", in.RequestID, err)
} }
@@ -347,6 +353,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec
ImageOutputSize: optionalTrimmedStringPtr(result.ImageOutputSize), ImageOutputSize: optionalTrimmedStringPtr(result.ImageOutputSize),
ImageSizeSource: optionalTrimmedStringPtr(result.ImageSizeSource), ImageSizeSource: optionalTrimmedStringPtr(result.ImageSizeSource),
ImageSizeBreakdown: result.ImageSizeBreakdown, ImageSizeBreakdown: result.ImageSizeBreakdown,
NativeCompactionV2: input.NativeCompactionV2,
} }
isVideoUsage := isGrokVideoUsageResult(result, billingModels) isVideoUsage := isGrokVideoUsageResult(result, billingModels)
if isVideoUsage { if isVideoUsage {
+9 -8
View File
@@ -173,14 +173,15 @@ type UsageLog struct {
// AccountStatsCost 账号统计定价预计算费用(nil = 使用默认公式 total_cost × account_rate_multiplier) // AccountStatsCost 账号统计定价预计算费用(nil = 使用默认公式 total_cost × account_rate_multiplier)
AccountStatsCost *float64 AccountStatsCost *float64
BillingType int8 BillingType int8
RequestType RequestType RequestType RequestType
Stream bool Stream bool
OpenAIWSMode bool OpenAIWSMode bool
DurationMs *int NativeCompactionV2 bool
FirstTokenMs *int DurationMs *int
UserAgent *string FirstTokenMs *int
IPAddress *string UserAgent *string
IPAddress *string
// SessionID is the explicit client-provided request correlation identifier // SessionID is the explicit client-provided request correlation identifier
// (e.g. the session_id / X-Session-Id headers). Nil when the client sent no // (e.g. the session_id / X-Session-Id headers). Nil when the client sent no
// valid session header. It is never derived from prompt_cache_key or content. // valid session header. It is never derived from prompt_cache_key or content.
@@ -0,0 +1,5 @@
ALTER TABLE usage_logs
ADD COLUMN IF NOT EXISTS native_compaction_v2 BOOLEAN NOT NULL DEFAULT FALSE;
COMMENT ON COLUMN usage_logs.native_compaction_v2 IS
'True only when the request was identified at runtime as native OpenAI remote compaction v2';
+4
View File
@@ -55,6 +55,7 @@ export interface TrendParams {
group_id?: number group_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
upstream_model_mismatch?: boolean upstream_model_mismatch?: boolean
} }
@@ -87,6 +88,7 @@ export interface ModelStatsParams {
group_id?: number group_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
upstream_model_mismatch?: boolean upstream_model_mismatch?: boolean
} }
@@ -116,6 +118,7 @@ export interface GroupStatsParams {
group_id?: number group_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
upstream_model_mismatch?: boolean upstream_model_mismatch?: boolean
} }
@@ -178,6 +181,7 @@ export interface UserBreakdownParams {
account_id?: number account_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
} }
+1
View File
@@ -124,6 +124,7 @@ export async function getStats(params: {
model?: string model?: string
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
upstream_model_mismatch?: boolean upstream_model_mismatch?: boolean
period?: string period?: string
start_date?: string start_date?: string
+2
View File
@@ -64,6 +64,7 @@ export interface TrendParams {
group_id?: number group_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
billing_mode?: string | null billing_mode?: string | null
timezone?: string timezone?: string
@@ -285,6 +286,7 @@ export async function getDashboardModels(params?: {
group_id?: number group_id?: number
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
billing_mode?: string | null billing_mode?: string | null
timezone?: string timezone?: string
@@ -127,6 +127,12 @@
<Select v-model="filters.request_type" :options="requestTypeOptions" @change="emitChange" /> <Select v-model="filters.request_type" :options="requestTypeOptions" @change="emitChange" />
</div> </div>
<!-- Native compaction is independent of the transport request type. -->
<div v-if="mode !== 'errors'" class="w-full sm:w-auto sm:min-w-[180px]">
<label class="input-label">{{ t('usage.compactionFilter') }}</label>
<Select v-model="filters.native_compaction_v2" :options="compactionOptions" @change="emitChange" />
</div>
<!-- Billing Type Filter (usage only) --> <!-- Billing Type Filter (usage only) -->
<div v-if="mode !== 'errors'" class="w-full sm:w-auto sm:min-w-[200px]"> <div v-if="mode !== 'errors'" class="w-full sm:w-auto sm:min-w-[200px]">
<label class="input-label">{{ t('admin.usage.billingType') }}</label> <label class="input-label">{{ t('admin.usage.billingType') }}</label>
@@ -274,6 +280,11 @@ const requestTypeOptions = ref<SelectOption[]>([
{ value: 'cyber', label: t('usage.cyber') } { value: 'cyber', label: t('usage.cyber') }
]) ])
const compactionOptions = ref<SelectOption[]>([
{ value: null, label: t('usage.allCompactionTypes') },
{ value: true, label: t('usage.compactionOnly') }
])
const billingTypeOptions = ref<SelectOption[]>([ const billingTypeOptions = ref<SelectOption[]>([
{ value: null, label: t('admin.usage.allBillingTypes') }, { value: null, label: t('admin.usage.allBillingTypes') },
{ value: 0, label: t('admin.usage.billingTypeBalance') }, { value: 0, label: t('admin.usage.billingTypeBalance') },
@@ -125,9 +125,18 @@
</template> </template>
<template #cell-stream="{ row }"> <template #cell-stream="{ row }">
<span class="inline-flex items-center rounded px-2 py-0.5 text-xs font-medium" :class="getRequestTypeBadgeClass(row)"> <div class="flex flex-wrap items-center gap-1">
{{ getRequestTypeLabel(row) }} <span data-testid="request-type-badge" class="inline-flex items-center rounded px-2 py-0.5 text-xs font-medium" :class="getRequestTypeBadgeClass(row)">
</span> {{ getRequestTypeLabel(row) }}
</span>
<span
v-if="row.native_compaction_v2"
data-testid="native-compaction-badge"
class="inline-flex items-center rounded bg-teal-100 px-2 py-0.5 text-xs font-medium text-teal-800 dark:bg-teal-900 dark:text-teal-200"
>
{{ t('usage.nativeCompactionV2') }}
</span>
</div>
</template> </template>
<template #cell-billing_mode="{ row }"> <template #cell-billing_mode="{ row }">
@@ -19,6 +19,9 @@ const messages: Record<string, string> = {
'usage.ws': 'WS', 'usage.ws': 'WS',
'usage.stream': 'Stream', 'usage.stream': 'Stream',
'usage.sync': 'Sync', 'usage.sync': 'Sync',
'usage.compactionFilter': 'Request Kind',
'usage.allCompactionTypes': 'All Requests',
'usage.compactionOnly': 'Compaction Only',
'admin.usage.billingType': 'Billing Type', 'admin.usage.billingType': 'Billing Type',
'admin.usage.allBillingTypes': 'All Billing Types', 'admin.usage.allBillingTypes': 'All Billing Types',
'admin.usage.billingTypeBalance': 'Balance', 'admin.usage.billingTypeBalance': 'Balance',
@@ -77,6 +80,7 @@ const defaultFilters = () => ({
account_id: undefined, account_id: undefined,
model: null, model: null,
request_type: null, request_type: null,
native_compaction_v2: null,
billing_type: null, billing_type: null,
billing_mode: null, billing_mode: null,
upstream_model_mismatch: null, upstream_model_mismatch: null,
@@ -259,3 +263,45 @@ describe('UsageFilters — model options come from prop (no dup request)', () =>
expect(opts.map((o) => o.value)).toEqual([null, 'claude-3', 'gpt-4o']) expect(opts.map((o) => o.value)).toEqual([null, 'claude-3', 'gpt-4o'])
}) })
}) })
describe('UsageFilters — native compaction filter', () => {
it('offers only All/Compaction and emits the independent boolean filter', async () => {
const SelectStub = {
name: 'Select',
props: ['modelValue', 'options'],
emits: ['update:modelValue', 'change'],
template: '<div />',
}
const filters = defaultFilters()
const wrapper = mount(UsageFilters, {
props: {
modelValue: filters,
exporting: false,
startDate: '2026-05-01',
endDate: '2026-05-28',
showActions: false,
modelOptions: [],
},
global: { stubs: { Select: SelectStub, Teleport: true } },
})
const compactionSelect = wrapper.findAllComponents(SelectStub).find((select: any) =>
(select.props('options') as Array<{ value: unknown }>).some((option) => option.value === true)
)
expect(compactionSelect).toBeDefined()
expect(compactionSelect!.props('options')).toEqual([
{ value: null, label: 'All Requests' },
{ value: true, label: 'Compaction Only' },
])
expect(compactionSelect!.props('options')).not.toEqual(
expect.arrayContaining([expect.objectContaining({ value: false })])
)
compactionSelect!.vm.$emit('update:modelValue', true)
compactionSelect!.vm.$emit('change')
await wrapper.vm.$nextTick()
expect(filters.native_compaction_v2).toBe(true)
expect(wrapper.emitted('change')).toBeTruthy()
})
})
@@ -54,6 +54,9 @@ const messages: Record<string, string> = {
'usage.imageSizeUnknown': 'unknown', 'usage.imageSizeUnknown': 'unknown',
'usage.imageUnitPrice': 'Per-image price', 'usage.imageUnitPrice': 'Per-image price',
'usage.imageTotalPrice': 'Image total price', 'usage.imageTotalPrice': 'Image total price',
'usage.stream': 'Stream',
'usage.sync': 'Sync',
'usage.nativeCompactionV2': 'Compaction',
'admin.usage.billingModeToken': 'Token', 'admin.usage.billingModeToken': 'Token',
'admin.usage.billingModePerRequest': 'Per request', 'admin.usage.billingModePerRequest': 'Per request',
'admin.usage.billingModeImage': 'Image', 'admin.usage.billingModeImage': 'Image',
@@ -169,6 +172,56 @@ describe('admin UsageTable tooltip', () => {
expect(wrapper.get('[data-testid="long-context-billing-marker"]').text()).toBe('x2') expect(wrapper.get('[data-testid="long-context-billing-marker"]').text()).toBe('x2')
}) })
it('keeps the request type badge and adds a separate badge only for native compaction rows', () => {
const DataTableStreamStub = {
props: ['data'],
template: `
<div>
<div v-for="row in data" :key="row.request_id">
<slot name="cell-stream" :row="row" />
</div>
</div>
`,
}
const wrapper = mount(UsageTable, {
props: {
data: [
{
...baseImageRow,
request_id: 'req-compaction-stream',
request_type: 'stream',
stream: true,
native_compaction_v2: true,
},
{
...baseImageRow,
request_id: 'req-historical-sync',
request_type: 'sync',
stream: false,
native_compaction_v2: false,
},
],
loading: false,
columns: [],
},
global: {
stubs: {
DataTable: DataTableStreamStub,
EmptyState: true,
Icon: true,
Teleport: true,
},
},
})
const requestBadges = wrapper.findAll('[data-testid="request-type-badge"]')
expect(requestBadges).toHaveLength(2)
expect(requestBadges[0].text()).toBe('Stream')
expect(requestBadges[1].text()).toBe('Sync')
expect(wrapper.findAll('[data-testid="native-compaction-badge"]')).toHaveLength(1)
expect(wrapper.get('[data-testid="native-compaction-badge"]').text()).toBe('Compaction')
})
it('shows service tier and billing breakdown in cost tooltip', async () => { it('shows service tier and billing breakdown in cost tooltip', async () => {
const row = { const row = {
request_id: 'req-admin-1', request_id: 'req-admin-1',
@@ -356,6 +356,10 @@ export default {
ws: 'WS', ws: 'WS',
stream: 'Stream', stream: 'Stream',
sync: 'Sync', sync: 'Sync',
nativeCompactionV2: 'Compaction',
compactionFilter: 'Request Kind',
allCompactionTypes: 'All Requests',
compactionOnly: 'Compaction Only',
cyber: 'Cyber', cyber: 'Cyber',
live: 'Live', live: 'Live',
unknown: 'Unknown', unknown: 'Unknown',
@@ -361,6 +361,10 @@ export default {
ws: 'WS', ws: 'WS',
stream: '流式', stream: '流式',
sync: '同步', sync: '同步',
nativeCompactionV2: '压缩',
compactionFilter: '请求类别',
allCompactionTypes: '全部请求',
compactionOnly: '仅原生压缩',
cyber: '安全策略', cyber: '安全策略',
live: 'Live', live: 'Live',
unknown: '未知', unknown: '未知',
+2
View File
@@ -1668,6 +1668,7 @@ export interface UsageLog {
request_type?: UsageRequestType request_type?: UsageRequestType
stream: boolean stream: boolean
openai_ws_mode?: boolean openai_ws_mode?: boolean
native_compaction_v2: boolean
duration_ms: number | null duration_ms: number | null
first_token_ms: number | null first_token_ms: number | null
@@ -2100,6 +2101,7 @@ export interface UsageQueryParams {
model?: string model?: string
request_type?: UsageRequestType request_type?: UsageRequestType
stream?: boolean stream?: boolean
native_compaction_v2?: boolean | null
billing_type?: number | null billing_type?: number | null
billing_mode?: string | null billing_mode?: string | null
start_date?: string start_date?: string
+5 -2
View File
@@ -246,6 +246,7 @@ const breakdownFilters = computed(() => {
if (filters.value.account_id) f.account_id = filters.value.account_id if (filters.value.account_id) f.account_id = filters.value.account_id
if (filters.value.group_id) f.group_id = filters.value.group_id if (filters.value.group_id) f.group_id = filters.value.group_id
if (filters.value.request_type != null) f.request_type = filters.value.request_type if (filters.value.request_type != null) f.request_type = filters.value.request_type
if (filters.value.native_compaction_v2 != null) f.native_compaction_v2 = filters.value.native_compaction_v2
if (filters.value.billing_type != null) f.billing_type = filters.value.billing_type if (filters.value.billing_type != null) f.billing_type = filters.value.billing_type
return f return f
}) })
@@ -297,7 +298,7 @@ const getGranularityForRange = (start: string, end: string): 'day' | 'hour' => {
} }
const defaultRange = getLast24HoursRangeDates() const defaultRange = getLast24HoursRangeDates()
const startDate = ref(defaultRange.start); const endDate = ref(defaultRange.end) const startDate = ref(defaultRange.start); const endDate = ref(defaultRange.end)
const filters = ref<AdminUsageQueryParams>({ user_id: undefined, model: undefined, group_id: undefined, request_type: undefined, billing_type: null, start_date: startDate.value, end_date: endDate.value }) const filters = ref<AdminUsageQueryParams>({ user_id: undefined, model: undefined, group_id: undefined, request_type: undefined, native_compaction_v2: null, billing_type: null, start_date: startDate.value, end_date: endDate.value })
const pagination = reactive({ page: 1, page_size: getPersistedPageSize(), total: 0 }) const pagination = reactive({ page: 1, page_size: getPersistedPageSize(), total: 0 })
const sortState = reactive({ const sortState = reactive({
sort_by: 'created_at', sort_by: 'created_at',
@@ -451,6 +452,7 @@ const loadModelStats = async (source: ModelDistributionSource, force = false) =>
group_id: filters.value.group_id, group_id: filters.value.group_id,
request_type: requestType, request_type: requestType,
stream: legacyStream === null ? undefined : legacyStream, stream: legacyStream === null ? undefined : legacyStream,
native_compaction_v2: filters.value.native_compaction_v2,
billing_type: filters.value.billing_type, billing_type: filters.value.billing_type,
upstream_model_mismatch: filters.value.upstream_model_mismatch, upstream_model_mismatch: filters.value.upstream_model_mismatch,
} }
@@ -501,6 +503,7 @@ const loadChartData = async () => {
group_id: filters.value.group_id, group_id: filters.value.group_id,
request_type: requestType, request_type: requestType,
stream: legacyStream === null ? undefined : legacyStream, stream: legacyStream === null ? undefined : legacyStream,
native_compaction_v2: filters.value.native_compaction_v2,
billing_type: filters.value.billing_type, billing_type: filters.value.billing_type,
upstream_model_mismatch: filters.value.upstream_model_mismatch, upstream_model_mismatch: filters.value.upstream_model_mismatch,
include_stats: false, include_stats: false,
@@ -541,7 +544,7 @@ const resetFilters = () => {
const range = getLast24HoursRangeDates() const range = getLast24HoursRangeDates()
startDate.value = range.start startDate.value = range.start
endDate.value = range.end endDate.value = range.end
filters.value = { start_date: startDate.value, end_date: endDate.value, request_type: undefined, billing_type: null, billing_mode: undefined } filters.value = { start_date: startDate.value, end_date: endDate.value, request_type: undefined, native_compaction_v2: null, billing_type: null, billing_mode: undefined }
granularity.value = getGranularityForRange(startDate.value, endDate.value) granularity.value = getGranularityForRange(startDate.value, endDate.value)
applyFilters() applyFilters()
} }
@@ -271,6 +271,71 @@ describe('admin UsageView route filters', () => {
}) })
}) })
describe('admin UsageView native compaction filter', () => {
beforeEach(() => {
vi.useFakeTimers()
list.mockReset().mockResolvedValue({ items: [], total: 0, pages: 0 })
getStats.mockReset().mockResolvedValue({
total_requests: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_tokens: 0,
total_tokens: 0,
total_cost: 0,
total_actual_cost: 0,
average_duration_ms: 0,
})
getSnapshotV2.mockReset().mockResolvedValue({ trend: [], models: [], groups: [] })
getModelStats.mockReset().mockResolvedValue({ models: [] })
})
afterEach(() => {
vi.useRealTimers()
})
it('propagates the filter to list/stats/model/snapshot requests and clears it on reset', async () => {
const wrapper = mountRouteFilteredUsageView()
vi.advanceTimersByTime(120)
await flushPromises()
list.mockClear()
getStats.mockClear()
getModelStats.mockClear()
getSnapshotV2.mockClear()
;(wrapper.vm as any).filters.native_compaction_v2 = true
;(wrapper.vm as any).applyFilters()
await flushPromises()
expect((wrapper.vm as any).breakdownFilters.native_compaction_v2).toBe(true)
expect(list).toHaveBeenCalledWith(
expect.objectContaining({ native_compaction_v2: true }),
expect.anything()
)
expect(getStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
expect(getModelStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
expect(getSnapshotV2).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
list.mockClear()
getStats.mockClear()
getModelStats.mockClear()
getSnapshotV2.mockClear()
;(wrapper.vm as any).resetFilters()
await flushPromises()
expect((wrapper.vm as any).filters.native_compaction_v2).toBeNull()
expect((wrapper.vm as any).breakdownFilters).not.toHaveProperty('native_compaction_v2')
expect(list).toHaveBeenCalledWith(
expect.objectContaining({ native_compaction_v2: null }),
expect.anything()
)
expect(getStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
expect(getModelStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
expect(getSnapshotV2).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
})
})
describe('admin UsageView distribution metric toggles', () => { describe('admin UsageView distribution metric toggles', () => {
beforeEach(() => { beforeEach(() => {
vi.useFakeTimers() vi.useFakeTimers()
@@ -671,10 +736,16 @@ describe('admin UsageView model audit export', () => {
const wrapper = mountRouteFilteredUsageView() const wrapper = mountRouteFilteredUsageView()
vi.advanceTimersByTime(120) vi.advanceTimersByTime(120)
await flushPromises() await flushPromises()
;(wrapper.vm as any).filters.native_compaction_v2 = true
await (wrapper.vm as any).exportToExcel() await (wrapper.vm as any).exportToExcel()
await flushPromises() await flushPromises()
expect(exportList).toHaveBeenCalledWith(
expect.objectContaining({ native_compaction_v2: true }),
expect.anything()
)
const headers = aoaToSheet.mock.calls[0][0][0] const headers = aoaToSheet.mock.calls[0][0][0]
expect(headers.slice(4, 8)).toEqual([ expect(headers.slice(4, 8)).toEqual([
'Requested model', 'Requested model',
+10
View File
@@ -111,6 +111,10 @@
<label class="input-label">{{ t('usage.type') }}</label> <label class="input-label">{{ t('usage.type') }}</label>
<Select v-model="filters.request_type" :options="requestTypeOptions" @change="applyFilters" /> <Select v-model="filters.request_type" :options="requestTypeOptions" @change="applyFilters" />
</div> </div>
<div class="w-full sm:w-auto sm:min-w-[180px]">
<label class="input-label">{{ t('usage.compactionFilter') }}</label>
<Select v-model="filters.native_compaction_v2" :options="compactionOptions" @change="applyFilters" />
</div>
<div class="w-full sm:w-auto sm:min-w-[200px]"> <div class="w-full sm:w-auto sm:min-w-[200px]">
<label class="input-label">{{ t('admin.usage.billingType') }}</label> <label class="input-label">{{ t('admin.usage.billingType') }}</label>
<Select v-model="filters.billing_type" :options="billingTypeOptions" @change="applyFilters" /> <Select v-model="filters.billing_type" :options="billingTypeOptions" @change="applyFilters" />
@@ -357,6 +361,7 @@ const filters = ref<UsageQueryParams>({
start_date: startDate.value, start_date: startDate.value,
end_date: endDate.value, end_date: endDate.value,
request_type: undefined, request_type: undefined,
native_compaction_v2: null,
billing_type: null, billing_type: null,
billing_mode: null, billing_mode: null,
}) })
@@ -382,6 +387,10 @@ const requestTypeOptions = computed<SelectOption[]>(() => [
{ value: 'stream', label: t('usage.stream') }, { value: 'stream', label: t('usage.stream') },
{ value: 'sync', label: t('usage.sync') }, { value: 'sync', label: t('usage.sync') },
]) ])
const compactionOptions = computed<SelectOption[]>(() => [
{ value: null, label: t('usage.allCompactionTypes') },
{ value: true, label: t('usage.compactionOnly') },
])
const billingTypeOptions = computed<SelectOption[]>(() => [ const billingTypeOptions = computed<SelectOption[]>(() => [
{ value: null, label: t('admin.usage.allBillingTypes') }, { value: null, label: t('admin.usage.allBillingTypes') },
{ value: 0, label: t('admin.usage.billingTypeBalance') }, { value: 0, label: t('admin.usage.billingTypeBalance') },
@@ -553,6 +562,7 @@ const resetFilters = () => {
start_date: range.start, start_date: range.start,
end_date: range.end, end_date: range.end,
request_type: undefined, request_type: undefined,
native_compaction_v2: null,
billing_type: null, billing_type: null,
billing_mode: null, billing_mode: null,
} }
@@ -53,6 +53,9 @@ const messages: Record<string, string> = {
'usage.ws': 'WS', 'usage.ws': 'WS',
'usage.stream': 'Stream', 'usage.stream': 'Stream',
'usage.sync': 'Sync', 'usage.sync': 'Sync',
'usage.compactionFilter': 'Request Kind',
'usage.allCompactionTypes': 'All Requests',
'usage.compactionOnly': 'Compaction Only',
'usage.exporting': 'Exporting', 'usage.exporting': 'Exporting',
'usage.exportCsv': 'Export CSV', 'usage.exportCsv': 'Export CSV',
'usage.failedToLoad': 'Failed to load', 'usage.failedToLoad': 'Failed to load',
@@ -125,6 +128,7 @@ const usageLog = {
billing_mode: 'token', billing_mode: 'token',
request_type: 'sync', request_type: 'sync',
stream: false, stream: false,
native_compaction_v2: false,
} }
function mountUsageView() { function mountUsageView() {
@@ -207,9 +211,54 @@ describe('user UsageView', () => {
expect(getAvailable).toHaveBeenCalled() expect(getAvailable).toHaveBeenCalled()
}) })
it('propagates and resets the native compaction filter across page requests', async () => {
const wrapper = mountUsageView()
await flushPromises()
expect((wrapper.vm as any).compactionOptions).toEqual([
{ value: null, label: 'All Requests' },
{ value: true, label: 'Compaction Only' },
])
query.mockClear()
getStats.mockClear()
getDashboardModels.mockClear()
getDashboardSnapshotV2.mockClear()
;(wrapper.vm as any).filters.native_compaction_v2 = true
;(wrapper.vm as any).applyFilters()
await flushPromises()
expect(query).toHaveBeenCalledWith(
expect.objectContaining({ native_compaction_v2: true }),
expect.anything()
)
expect(getStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
expect(getDashboardModels).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
expect(getDashboardSnapshotV2).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: true }))
query.mockClear()
getStats.mockClear()
getDashboardModels.mockClear()
getDashboardSnapshotV2.mockClear()
;(wrapper.vm as any).resetFilters()
await flushPromises()
expect((wrapper.vm as any).filters.native_compaction_v2).toBeNull()
expect(query).toHaveBeenCalledWith(
expect.objectContaining({ native_compaction_v2: null }),
expect.anything()
)
expect(getStats).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
expect(getDashboardModels).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
expect(getDashboardSnapshotV2).toHaveBeenCalledWith(expect.objectContaining({ native_compaction_v2: null }))
})
it('exports csv with current filters and without admin-only fields', async () => { it('exports csv with current filters and without admin-only fields', async () => {
const wrapper = mountUsageView() const wrapper = mountUsageView()
await flushPromises() await flushPromises()
;(wrapper.vm as any).filters.native_compaction_v2 = true
let exportedBlob: Blob | null = null let exportedBlob: Blob | null = null
let csvContent = '' let csvContent = ''
@@ -234,6 +283,7 @@ describe('user UsageView', () => {
page_size: 100, page_size: 100,
sort_by: 'created_at', sort_by: 'created_at',
sort_order: 'desc', sort_order: 'desc',
native_compaction_v2: true,
})) }))
expect(clickSpy).toHaveBeenCalled() expect(clickSpy).toHaveBeenCalled()
expect(showSuccess).toHaveBeenCalled() expect(showSuccess).toHaveBeenCalled()