fix(subscription): 修正账号卡片统计口径
This commit is contained in:
33 files changed
+1695
-126
No files matched your search
@@ -465,16 +465,23 @@ func observationWindowRemainingRatio(window ObservationQuotaWindow) (float64, bo
|
||||
}
|
||||
|
||||
// credentialDailyUsageWindow 是账号卡上「近 24 小时成功/失败」的窗口长度。
|
||||
// 取 24 小时是因为 QueryCredentialWindowUsage 在不超过 24 小时时走精确请求日志,
|
||||
// 再长就退化成小时聚合,边界会变成近似值。
|
||||
const credentialDailyUsageWindow = 24 * time.Hour
|
||||
|
||||
func (s *Service) enrichCredentialDailyUsage(
|
||||
func (s *Service) enrichCredentialActivities(
|
||||
ctx context.Context,
|
||||
credentialID uint,
|
||||
item *CredentialItemResponse,
|
||||
items []CredentialItemResponse,
|
||||
) {
|
||||
if s == nil || s.credentialWindowUsage == nil || credentialID == 0 || item == nil {
|
||||
if s == nil || s.credentialActivity == nil || len(items) == 0 {
|
||||
return
|
||||
}
|
||||
credentialIDs := make([]uint, 0, len(items))
|
||||
for index := range items {
|
||||
if items[index].ConnectionType == string(models.ConnectionTypeSubscription) &&
|
||||
items[index].CredentialID != 0 {
|
||||
credentialIDs = append(credentialIDs, items[index].CredentialID)
|
||||
}
|
||||
}
|
||||
if len(credentialIDs) == 0 {
|
||||
return
|
||||
}
|
||||
now := s.now().UTC()
|
||||
@@ -483,33 +490,60 @@ func (s *Service) enrichCredentialDailyUsage(
|
||||
if fromMS < 0 || toMS <= fromMS {
|
||||
return
|
||||
}
|
||||
observed, err := s.credentialWindowUsage.QueryCredentialWindowUsage(ctx, requestlog.CredentialWindowUsageQuery{
|
||||
CredentialID: credentialID,
|
||||
FromMS: fromMS,
|
||||
ToMS: toMS,
|
||||
Source: requestlog.CredentialWindowUsageSourceRequestLogs,
|
||||
})
|
||||
activities, err := s.credentialActivity.QueryCredentialActivity(
|
||||
ctx,
|
||||
requestlog.CredentialActivityQuery{
|
||||
CredentialIDs: credentialIDs,
|
||||
FromMS: fromMS,
|
||||
ToMS: toMS,
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
item.DailyUsage = &CredentialDailyUsageResponse{
|
||||
WindowSeconds: int64(credentialDailyUsageWindow / time.Second),
|
||||
SuccessCount: observed.SuccessCount,
|
||||
FailureCount: observed.FailureCount,
|
||||
DataComplete: observed.DataComplete,
|
||||
for index := range items {
|
||||
activity, exists := activities[items[index].CredentialID]
|
||||
if !exists {
|
||||
continue
|
||||
}
|
||||
items[index].LastUsedAtMS = activity.LastUsedAtMS
|
||||
items[index].DailyUsage = &CredentialDailyUsageResponse{
|
||||
WindowSeconds: int64(credentialDailyUsageWindow / time.Second),
|
||||
SuccessCount: activity.SuccessCount,
|
||||
FailureCount: activity.FailureCount,
|
||||
DataComplete: activity.DataComplete,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) enrichCredentialDailyUsage(
|
||||
ctx context.Context,
|
||||
credentialID uint,
|
||||
item *CredentialItemResponse,
|
||||
) {
|
||||
if item == nil || credentialID == 0 {
|
||||
return
|
||||
}
|
||||
item.CredentialID = credentialID
|
||||
items := []CredentialItemResponse{*item}
|
||||
s.enrichCredentialActivities(ctx, items)
|
||||
*item = items[0]
|
||||
}
|
||||
|
||||
func (s *Service) enrichCredentialObservationUsage(
|
||||
ctx context.Context,
|
||||
credentialID uint,
|
||||
response *CredentialObservationResponse,
|
||||
) {
|
||||
if s == nil || s.credentialWindowUsage == nil || credentialID == 0 ||
|
||||
response == nil || response.Snapshot == nil {
|
||||
response == nil || response.State != string(models.CredentialObservationFresh) ||
|
||||
response.Snapshot == nil || response.FreshUntilMS == nil {
|
||||
return
|
||||
}
|
||||
nowMS := s.now().UTC().UnixMilli()
|
||||
if *response.FreshUntilMS <= nowMS {
|
||||
return
|
||||
}
|
||||
for index := range response.Snapshot.QuotaWindows {
|
||||
window := &response.Snapshot.QuotaWindows[index]
|
||||
if window.Scope != "account" || window.ResetAtMS == nil ||
|
||||
@@ -528,15 +562,11 @@ func (s *Service) enrichCredentialObservationUsage(
|
||||
if fromMS >= nowMS {
|
||||
continue
|
||||
}
|
||||
source := requestlog.CredentialWindowUsageSourceRequestLogs
|
||||
if *window.WindowSeconds > int64((24*time.Hour)/time.Second) {
|
||||
source = requestlog.CredentialWindowUsageSourceHourlyStats
|
||||
}
|
||||
observed, err := s.credentialWindowUsage.QueryCredentialWindowUsage(ctx, requestlog.CredentialWindowUsageQuery{
|
||||
CredentialID: credentialID,
|
||||
FromMS: fromMS,
|
||||
ToMS: nowMS,
|
||||
Source: source,
|
||||
Source: requestlog.CredentialWindowUsageSourceHourlyStats,
|
||||
})
|
||||
if err != nil {
|
||||
continue
|
||||
|
||||
@@ -204,18 +204,20 @@ func TestRefreshCredentialObservationKeepsUsageWhenResetCreditDetailsFail(t *tes
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing.T) {
|
||||
func TestEnrichCredentialObservationUsageUsesHourlyStatsForEveryWindow(t *testing.T) {
|
||||
fixture := newServiceFixture(t)
|
||||
now := time.Date(2026, time.August, 14, 12, 0, 0, 0, time.UTC)
|
||||
fixture.service.now = func() time.Time { return now }
|
||||
shortReset := now.Add(2 * time.Hour).UnixMilli()
|
||||
longReset := now.Add(3 * 24 * time.Hour).UnixMilli()
|
||||
modelReset := now.Add(time.Hour).UnixMilli()
|
||||
freshUntil := now.Add(time.Hour).UnixMilli()
|
||||
shortSeconds := int64((5 * time.Hour) / time.Second)
|
||||
longSeconds := int64((7 * 24 * time.Hour) / time.Second)
|
||||
modelSeconds := int64(time.Hour / time.Second)
|
||||
response := CredentialObservationResponse{
|
||||
State: string(models.CredentialObservationFresh),
|
||||
State: string(models.CredentialObservationFresh),
|
||||
FreshUntilMS: &freshUntil,
|
||||
Snapshot: &CredentialObservationSnapshot{QuotaWindows: []ObservationQuotaWindow{
|
||||
{ID: "short", Scope: "account", ResetAtMS: &shortReset, WindowSeconds: &shortSeconds},
|
||||
{ID: "long", Scope: "account", ResetAtMS: &longReset, WindowSeconds: &longSeconds},
|
||||
@@ -237,7 +239,7 @@ func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing
|
||||
if len(reader.queries) != 2 {
|
||||
t.Fatalf("usage queries = %#v, want two account windows", reader.queries)
|
||||
}
|
||||
if reader.queries[0].Source != requestlog.CredentialWindowUsageSourceRequestLogs ||
|
||||
if reader.queries[0].Source != requestlog.CredentialWindowUsageSourceHourlyStats ||
|
||||
reader.queries[0].FromMS != shortReset-shortSeconds*1000 ||
|
||||
reader.queries[0].ToMS != now.UnixMilli() {
|
||||
t.Fatalf("short window query = %#v", reader.queries[0])
|
||||
@@ -257,6 +259,46 @@ func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnrichCredentialObservationUsageSkipsNonCurrentObservation(t *testing.T) {
|
||||
now := time.Date(2026, time.August, 14, 12, 0, 0, 0, time.UTC)
|
||||
future := now.Add(time.Hour).UnixMilli()
|
||||
expired := now.UnixMilli()
|
||||
resetAt := now.Add(2 * time.Hour).UnixMilli()
|
||||
windowSeconds := int64((5 * time.Hour) / time.Second)
|
||||
tests := []struct {
|
||||
name string
|
||||
state string
|
||||
freshUntilMS *int64
|
||||
}{
|
||||
{name: "stale", state: string(models.CredentialObservationStale), freshUntilMS: &future},
|
||||
{name: "error", state: string(models.CredentialObservationError), freshUntilMS: &future},
|
||||
{name: "missing fresh until", state: string(models.CredentialObservationFresh)},
|
||||
{name: "expired", state: string(models.CredentialObservationFresh), freshUntilMS: &expired},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
fixture := newServiceFixture(t)
|
||||
fixture.service.now = func() time.Time { return now }
|
||||
reader := &recordingCredentialWindowUsageReader{}
|
||||
fixture.service.credentialWindowUsage = reader
|
||||
response := CredentialObservationResponse{
|
||||
State: test.state,
|
||||
FreshUntilMS: test.freshUntilMS,
|
||||
Snapshot: &CredentialObservationSnapshot{QuotaWindows: []ObservationQuotaWindow{{
|
||||
ID: "account", Scope: "account", ResetAtMS: &resetAt, WindowSeconds: &windowSeconds,
|
||||
}}},
|
||||
}
|
||||
|
||||
fixture.service.enrichCredentialObservationUsage(t.Context(), 37, &response)
|
||||
|
||||
if len(reader.queries) != 0 || response.Snapshot.QuotaWindows[0].ObservedUsage != nil {
|
||||
t.Fatalf("non-current observation was enriched: queries=%#v response=%#v", reader.queries, response)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefreshCredentialObservationPublishesAndClearsQuotaRoutingState(t *testing.T) {
|
||||
fixture, groupID, credentialID := newSubscriptionCredentialFixture(t)
|
||||
now := time.Date(2026, time.August, 14, 15, 0, 0, 0, time.UTC)
|
||||
@@ -429,19 +471,35 @@ func (reader *recordingCredentialWindowUsageReader) QueryCredentialWindowUsage(
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func TestEnrichCredentialDailyUsageQueriesExactDayWindow(t *testing.T) {
|
||||
type recordingCredentialActivityReader struct {
|
||||
queries []requestlog.CredentialActivityQuery
|
||||
result map[uint]requestlog.CredentialActivity
|
||||
err error
|
||||
}
|
||||
|
||||
func (reader *recordingCredentialActivityReader) QueryCredentialActivity(
|
||||
_ context.Context,
|
||||
query requestlog.CredentialActivityQuery,
|
||||
) (map[uint]requestlog.CredentialActivity, error) {
|
||||
reader.queries = append(reader.queries, query)
|
||||
return reader.result, reader.err
|
||||
}
|
||||
|
||||
func TestEnrichCredentialDailyUsageQueriesAttemptActivity(t *testing.T) {
|
||||
fixture, _, _ := newSubscriptionCredentialFixture(t)
|
||||
now := time.UnixMilli(1_800_000_000_000)
|
||||
fixture.service.now = func() time.Time { return now }
|
||||
reader := &recordingCredentialWindowUsageReader{}
|
||||
reader.result = requestlog.CredentialWindowUsage{
|
||||
UsageAggregate: requestlog.UsageAggregate{
|
||||
RequestCount: 18, SuccessCount: 15, FailureCount: 3,
|
||||
lastUsedAtMS := now.Add(-time.Minute).UnixMilli()
|
||||
reader := &recordingCredentialActivityReader{result: map[uint]requestlog.CredentialActivity{
|
||||
37: {
|
||||
CredentialID: 37, LastUsedAtMS: &lastUsedAtMS,
|
||||
SuccessCount: 15, FailureCount: 3, DataComplete: true,
|
||||
},
|
||||
DataComplete: true,
|
||||
}}
|
||||
fixture.service.credentialActivity = reader
|
||||
item := CredentialItemResponse{
|
||||
CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription),
|
||||
}
|
||||
fixture.service.credentialWindowUsage = reader
|
||||
item := CredentialItemResponse{CredentialID: 37}
|
||||
|
||||
fixture.service.enrichCredentialDailyUsage(t.Context(), 37, &item)
|
||||
|
||||
@@ -449,25 +507,26 @@ func TestEnrichCredentialDailyUsageQueriesExactDayWindow(t *testing.T) {
|
||||
t.Fatalf("usage queries = %#v, want exactly one day window", reader.queries)
|
||||
}
|
||||
query := reader.queries[0]
|
||||
// 24 小时窗口必须走精确请求日志,走小时聚合会把窗口边界变成近似值。
|
||||
if query.Source != requestlog.CredentialWindowUsageSourceRequestLogs ||
|
||||
query.CredentialID != 37 ||
|
||||
if !reflect.DeepEqual(query.CredentialIDs, []uint{37}) ||
|
||||
query.ToMS != now.UnixMilli() ||
|
||||
query.FromMS != now.Add(-24*time.Hour).UnixMilli() {
|
||||
t.Fatalf("day window query = %#v", query)
|
||||
}
|
||||
if item.DailyUsage == nil || item.DailyUsage.WindowSeconds != 86_400 ||
|
||||
item.DailyUsage.SuccessCount != 15 || item.DailyUsage.FailureCount != 3 ||
|
||||
!item.DailyUsage.DataComplete {
|
||||
t.Fatalf("daily usage = %#v", item.DailyUsage)
|
||||
!item.DailyUsage.DataComplete || item.LastUsedAtMS == nil ||
|
||||
*item.LastUsedAtMS != lastUsedAtMS {
|
||||
t.Fatalf("credential activity = %#v / %#v", item.LastUsedAtMS, item.DailyUsage)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnrichCredentialDailyUsageLeavesItemUntouchedOnQueryFailure(t *testing.T) {
|
||||
fixture, _, _ := newSubscriptionCredentialFixture(t)
|
||||
fixture.service.now = func() time.Time { return time.UnixMilli(1_800_000_000_000) }
|
||||
fixture.service.credentialWindowUsage = failingCredentialWindowUsageReader{}
|
||||
item := CredentialItemResponse{CredentialID: 37}
|
||||
fixture.service.credentialActivity = &recordingCredentialActivityReader{err: errors.New("activity unavailable")}
|
||||
item := CredentialItemResponse{
|
||||
CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription),
|
||||
}
|
||||
|
||||
fixture.service.enrichCredentialDailyUsage(t.Context(), 37, &item)
|
||||
|
||||
@@ -476,13 +535,32 @@ func TestEnrichCredentialDailyUsageLeavesItemUntouchedOnQueryFailure(t *testing.
|
||||
}
|
||||
}
|
||||
|
||||
type failingCredentialWindowUsageReader struct{}
|
||||
func TestEnrichCredentialActivitiesBatchesSubscriptionItems(t *testing.T) {
|
||||
fixture := newServiceFixture(t)
|
||||
now := time.UnixMilli(1_800_000_000_000)
|
||||
fixture.service.now = func() time.Time { return now }
|
||||
reader := &recordingCredentialActivityReader{result: map[uint]requestlog.CredentialActivity{
|
||||
37: {CredentialID: 37, SuccessCount: 4, DataComplete: true},
|
||||
39: {CredentialID: 39, FailureCount: 2, DataComplete: true},
|
||||
}}
|
||||
fixture.service.credentialActivity = reader
|
||||
items := []CredentialItemResponse{
|
||||
{CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription)},
|
||||
{CredentialID: 38, ConnectionType: string(models.ConnectionTypeAPIKey)},
|
||||
{CredentialID: 39, ConnectionType: string(models.ConnectionTypeSubscription)},
|
||||
}
|
||||
|
||||
func (failingCredentialWindowUsageReader) QueryCredentialWindowUsage(
|
||||
_ context.Context,
|
||||
_ requestlog.CredentialWindowUsageQuery,
|
||||
) (requestlog.CredentialWindowUsage, error) {
|
||||
return requestlog.CredentialWindowUsage{}, errors.New("request logs unavailable")
|
||||
fixture.service.enrichCredentialActivities(t.Context(), items)
|
||||
|
||||
if len(reader.queries) != 1 ||
|
||||
!reflect.DeepEqual(reader.queries[0].CredentialIDs, []uint{37, 39}) {
|
||||
t.Fatalf("activity queries = %#v, want one subscription batch", reader.queries)
|
||||
}
|
||||
if items[0].DailyUsage == nil || items[0].DailyUsage.SuccessCount != 4 ||
|
||||
items[1].DailyUsage != nil || items[2].DailyUsage == nil ||
|
||||
items[2].DailyUsage.FailureCount != 2 {
|
||||
t.Fatalf("enriched items = %#v", items)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefreshCredentialObservationPersistsNormalizationFailure(t *testing.T) {
|
||||
|
||||
@@ -83,18 +83,19 @@ type CredentialItemResponse struct {
|
||||
LastFailureCategory string `json:"last_failure_category"`
|
||||
LastStatusCode *int `json:"last_status_code"`
|
||||
CooldownUntilMS *int64 `json:"cooldown_until_ms"`
|
||||
LastUsedAtMS *int64 `json:"last_used_at_ms,omitempty"`
|
||||
DailyUsage *CredentialDailyUsageResponse `json:"daily_usage,omitempty"`
|
||||
Recovery CredentialRecoveryResponse `json:"recovery"`
|
||||
}
|
||||
|
||||
// CredentialDailyUsageResponse 汇报固定 24 小时窗口内的请求结果分布。
|
||||
// CredentialDailyUsageResponse 汇报固定 24 小时窗口内的上游尝试结果分布。
|
||||
// recent_success_count / recent_failure_count 来自 health 的 5 分钟内存窗口,
|
||||
// 是调度判定用的,重启即清零;这里的计数来自请求日志,用于给人看。
|
||||
// 是调度判定用的,重启即清零;这里的计数来自账号小时聚合,用于给人看。
|
||||
type CredentialDailyUsageResponse struct {
|
||||
WindowSeconds int64 `json:"window_seconds"`
|
||||
SuccessCount int64 `json:"success_count"`
|
||||
FailureCount int64 `json:"failure_count"`
|
||||
// DataComplete 为 false 表示请求日志留存期短于该窗口,计数偏低。
|
||||
// DataComplete 为 false 表示统计数据未覆盖完整窗口,计数偏低。
|
||||
DataComplete bool `json:"data_complete"`
|
||||
}
|
||||
|
||||
@@ -426,12 +427,8 @@ func (s *Service) mapCredentialCollection(
|
||||
})
|
||||
total := len(filtered)
|
||||
items := credentialCollectionPage(filtered, query.Page, query.PageSize)
|
||||
for index := range items {
|
||||
s.enrichCredentialObservationUsage(ctx, items[index].CredentialID, items[index].Observation)
|
||||
if items[index].ConnectionType == string(models.ConnectionTypeSubscription) {
|
||||
s.enrichCredentialDailyUsage(ctx, items[index].CredentialID, &items[index])
|
||||
}
|
||||
}
|
||||
// 配额窗口的本地用量只在账号详情被展开时读取,避免列表按卡片聚合。
|
||||
s.enrichCredentialActivities(ctx, items)
|
||||
return CredentialCollectionResponse{
|
||||
ObservedAtMS: observedAtMS, StatsWindowSeconds: credentialCollectionStatsWindow,
|
||||
Summary: summary, Items: items,
|
||||
|
||||
@@ -50,6 +50,7 @@ type Service struct {
|
||||
requestLogs RequestLogReader
|
||||
usageStats UsageStatReader
|
||||
credentialWindowUsage credentialWindowUsageReader
|
||||
credentialActivity credentialActivityReader
|
||||
homeStatistics HomeStatisticsReader
|
||||
stats *health.StatsStore
|
||||
mutations credentialMutationCoordinator
|
||||
@@ -93,6 +94,13 @@ type credentialWindowUsageReader interface {
|
||||
) (requestlog.CredentialWindowUsage, error)
|
||||
}
|
||||
|
||||
type credentialActivityReader interface {
|
||||
QueryCredentialActivity(
|
||||
context.Context,
|
||||
requestlog.CredentialActivityQuery,
|
||||
) (map[uint]requestlog.CredentialActivity, error)
|
||||
}
|
||||
|
||||
type credentialMultiMutationCoordinator interface {
|
||||
DoMany([]uint, func())
|
||||
}
|
||||
@@ -193,6 +201,11 @@ func NewService(
|
||||
} else if reader, ok := requestLogs.(credentialWindowUsageReader); ok {
|
||||
service.credentialWindowUsage = reader
|
||||
}
|
||||
if reader, ok := usageStats.(credentialActivityReader); ok {
|
||||
service.credentialActivity = reader
|
||||
} else if reader, ok := requestLogs.(credentialActivityReader); ok {
|
||||
service.credentialActivity = reader
|
||||
}
|
||||
if provider, ok := executor.(channelDefaultBaseURLProvider); ok {
|
||||
service.channelDefaultBaseURLs = provider
|
||||
}
|
||||
|
||||
@@ -313,6 +313,7 @@ func (recorder *requestRecorder) appendAttempt(
|
||||
}
|
||||
attempt := telemetry.Attempt{
|
||||
Sequence: len(recorder.attempts) + 1,
|
||||
CompletedAt: completedAt,
|
||||
GroupID: selection.GroupID,
|
||||
GroupName: selection.Group.Name,
|
||||
ChannelID: selection.ChannelID,
|
||||
|
||||
@@ -45,6 +45,33 @@ func requestLogSelection(groupID, credentialID uint, name string) scheduler.Sele
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestRecorderCapturesAttemptCompletionTime(t *testing.T) {
|
||||
completedAt := time.Date(2026, time.August, 15, 14, 35, 27, 123_000_000, time.UTC)
|
||||
recorder := newRequestRecorder(
|
||||
&recordingRequestLogSink{},
|
||||
"00000000-0000-4000-8000-000000000001",
|
||||
completedAt.Add(-time.Second),
|
||||
1,
|
||||
protocol.OpenAICompletions,
|
||||
func() time.Time { return completedAt },
|
||||
)
|
||||
|
||||
recorder.appendAttempt(
|
||||
requestLogSelection(11, 21, "first"),
|
||||
UpstreamResult{StatusCode: http.StatusOK},
|
||||
telemetry.FailureCategoryOK,
|
||||
telemetry.ActionTerminate,
|
||||
"",
|
||||
"",
|
||||
completedAt.Add(-250*time.Millisecond),
|
||||
completedAt,
|
||||
)
|
||||
|
||||
if len(recorder.attempts) != 1 || !recorder.attempts[0].CompletedAt.Equal(completedAt) {
|
||||
t.Fatalf("attempt completion = %#v, want %s", recorder.attempts, completedAt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestRecorderCapturesFirstResponseAfterDownstreamCommit(t *testing.T) {
|
||||
startedAt := time.Date(2026, time.August, 5, 8, 0, 0, 0, time.UTC)
|
||||
sink := &recordingRequestLogSink{}
|
||||
|
||||
@@ -0,0 +1,249 @@
|
||||
package requestlog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"gpt-load/internal/platform/epochms"
|
||||
"gpt-load/internal/storage/dbtx"
|
||||
"gpt-load/internal/storage/models"
|
||||
"gpt-load/internal/telemetry"
|
||||
"gpt-load/internal/usage"
|
||||
)
|
||||
|
||||
const (
|
||||
maxCredentialActivityIDs = 100
|
||||
credentialActivityMaxWindowMS = 24 * epochms.MillisecondsPerHour
|
||||
)
|
||||
|
||||
type CredentialActivityQuery struct {
|
||||
CredentialIDs []uint
|
||||
FromMS int64
|
||||
ToMS int64
|
||||
}
|
||||
|
||||
type CredentialActivity struct {
|
||||
CredentialID uint
|
||||
LastUsedAtMS *int64
|
||||
SuccessCount int64
|
||||
FailureCount int64
|
||||
DataComplete bool
|
||||
}
|
||||
|
||||
// QueryCredentialActivity returns the account-card activity for the requested
|
||||
// credentials without issuing one query per account. Full hours come from the
|
||||
// durable hourly aggregate; only the oldest partial hour reads attempt rows.
|
||||
func (service *Service) QueryCredentialActivity(
|
||||
ctx context.Context,
|
||||
input CredentialActivityQuery,
|
||||
) (map[uint]CredentialActivity, error) {
|
||||
if service == nil || service.db == nil {
|
||||
return nil, fmt.Errorf("query credential activity: database is nil")
|
||||
}
|
||||
credentialIDs, err := validateCredentialActivityQuery(input)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
|
||||
hourlyFromMS, err := alignHourUp(input.FromMS)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query credential activity: align window start: %w", err)
|
||||
}
|
||||
hourlyToMS, err := alignHourUp(input.ToMS)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query credential activity: align window end: %w", err)
|
||||
}
|
||||
boundaryToMS := hourlyFromMS
|
||||
if boundaryToMS > input.ToMS {
|
||||
boundaryToMS = input.ToMS
|
||||
}
|
||||
hasHourlyWindow := hourlyFromMS < hourlyToMS
|
||||
hasBoundaryWindow := input.FromMS < boundaryToMS
|
||||
dataComplete := (!hasHourlyWindow || service.hourlyWindowRetained(hourlyFromMS)) &&
|
||||
(!hasBoundaryWindow || service.requestLogWindowRetained(input.FromMS))
|
||||
|
||||
result := make(map[uint]CredentialActivity, len(credentialIDs))
|
||||
for _, credentialID := range credentialIDs {
|
||||
result[credentialID] = CredentialActivity{
|
||||
CredentialID: credentialID,
|
||||
DataComplete: dataComplete,
|
||||
}
|
||||
}
|
||||
|
||||
err = dbtx.Run(ctx, service.db, dbtx.Options{
|
||||
Mode: dbtx.ReadSnapshot,
|
||||
CleanupTimeout: usageRollbackTimeout,
|
||||
Operation: "credential activity read transaction",
|
||||
}, func(connection *gorm.DB) error {
|
||||
if err := queryCredentialLatestActivity(
|
||||
connection, credentialIDs, input.ToMS, result,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
if hasHourlyWindow {
|
||||
if err := queryCredentialHourlyActivity(
|
||||
connection, credentialIDs, hourlyFromMS, hourlyToMS, result,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if hasBoundaryWindow {
|
||||
if err := queryCredentialBoundaryActivity(
|
||||
connection, credentialIDs, input.FromMS, boundaryToMS, result,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query credential activity: %w", err)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func validateCredentialActivityQuery(input CredentialActivityQuery) ([]uint, error) {
|
||||
if len(input.CredentialIDs) == 0 || len(input.CredentialIDs) > maxCredentialActivityIDs ||
|
||||
input.FromMS < 0 || input.ToMS <= input.FromMS ||
|
||||
input.ToMS-input.FromMS > credentialActivityMaxWindowMS {
|
||||
return nil, fmt.Errorf("query credential activity: invalid scope")
|
||||
}
|
||||
credentialIDs := make([]uint, 0, len(input.CredentialIDs))
|
||||
seen := make(map[uint]struct{}, len(input.CredentialIDs))
|
||||
for _, credentialID := range input.CredentialIDs {
|
||||
if credentialID == 0 {
|
||||
return nil, fmt.Errorf("query credential activity: invalid scope")
|
||||
}
|
||||
if _, exists := seen[credentialID]; exists {
|
||||
continue
|
||||
}
|
||||
seen[credentialID] = struct{}{}
|
||||
credentialIDs = append(credentialIDs, credentialID)
|
||||
}
|
||||
return credentialIDs, nil
|
||||
}
|
||||
|
||||
type credentialLatestActivityRow struct {
|
||||
CredentialID uint `gorm:"column:credential_id"`
|
||||
LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"`
|
||||
}
|
||||
|
||||
func queryCredentialLatestActivity(
|
||||
db *gorm.DB,
|
||||
credentialIDs []uint,
|
||||
toMS int64,
|
||||
result map[uint]CredentialActivity,
|
||||
) error {
|
||||
var query strings.Builder
|
||||
args := make([]any, 0, len(credentialIDs)*3)
|
||||
for index, credentialID := range credentialIDs {
|
||||
if index > 0 {
|
||||
query.WriteString(" UNION ALL ")
|
||||
}
|
||||
query.WriteString(`SELECT ? AS credential_id, (
|
||||
SELECT completed_at_ms
|
||||
FROM request_log_attempts
|
||||
WHERE credential_id = ? AND completed_at_ms < ?
|
||||
ORDER BY completed_at_ms DESC
|
||||
LIMIT 1
|
||||
) AS last_used_at_ms`)
|
||||
args = append(args, credentialID, credentialID, toMS)
|
||||
}
|
||||
var rows []credentialLatestActivityRow
|
||||
if err := db.Raw(query.String(), args...).Scan(&rows).Error; err != nil {
|
||||
return fmt.Errorf("query latest credential attempts: %w", err)
|
||||
}
|
||||
for _, row := range rows {
|
||||
activity, exists := result[row.CredentialID]
|
||||
if !exists {
|
||||
continue
|
||||
}
|
||||
activity.LastUsedAtMS = row.LastUsedAtMS
|
||||
result[row.CredentialID] = activity
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type credentialActivityCountRow struct {
|
||||
CredentialID uint `gorm:"column:credential_id"`
|
||||
SuccessCount int64 `gorm:"column:success_count"`
|
||||
FailureCount int64 `gorm:"column:failure_count"`
|
||||
}
|
||||
|
||||
func queryCredentialHourlyActivity(
|
||||
db *gorm.DB,
|
||||
credentialIDs []uint,
|
||||
fromMS int64,
|
||||
toMS int64,
|
||||
result map[uint]CredentialActivity,
|
||||
) error {
|
||||
var rows []credentialActivityCountRow
|
||||
if err := db.Model(&models.CredentialAttemptStat{}).
|
||||
Select("credential_id, COALESCE(SUM(success_count), 0) AS success_count, "+
|
||||
"COALESCE(SUM(failure_count), 0) AS failure_count").
|
||||
Where("credential_id IN ?", credentialIDs).
|
||||
Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fromMS, toMS).
|
||||
Group("credential_id").
|
||||
Find(&rows).Error; err != nil {
|
||||
return fmt.Errorf("query hourly credential attempts: %w", err)
|
||||
}
|
||||
return mergeCredentialActivityCounts(result, rows)
|
||||
}
|
||||
|
||||
func queryCredentialBoundaryActivity(
|
||||
db *gorm.DB,
|
||||
credentialIDs []uint,
|
||||
fromMS int64,
|
||||
toMS int64,
|
||||
result map[uint]CredentialActivity,
|
||||
) error {
|
||||
var rows []credentialActivityCountRow
|
||||
if err := db.Model(&models.RequestLogAttempt{}).
|
||||
Select("credential_id, "+
|
||||
"COALESCE(SUM(CASE WHEN failure_category = ? THEN 1 ELSE 0 END), 0) AS success_count, "+
|
||||
"COALESCE(SUM(CASE WHEN failure_category <> ? AND failure_category <> ? THEN 1 ELSE 0 END), 0) AS failure_count",
|
||||
string(telemetry.FailureCategoryOK),
|
||||
string(telemetry.FailureCategoryOK),
|
||||
string(telemetry.FailureCategoryDownstreamCancel),
|
||||
).
|
||||
Where("credential_id IN ?", credentialIDs).
|
||||
Where("completed_at_ms >= ? AND completed_at_ms < ?", fromMS, toMS).
|
||||
Group("credential_id").
|
||||
Find(&rows).Error; err != nil {
|
||||
return fmt.Errorf("query credential attempt boundary: %w", err)
|
||||
}
|
||||
return mergeCredentialActivityCounts(result, rows)
|
||||
}
|
||||
|
||||
func mergeCredentialActivityCounts(
|
||||
result map[uint]CredentialActivity,
|
||||
rows []credentialActivityCountRow,
|
||||
) error {
|
||||
for _, row := range rows {
|
||||
activity, exists := result[row.CredentialID]
|
||||
if !exists {
|
||||
continue
|
||||
}
|
||||
if row.SuccessCount < 0 || row.FailureCount < 0 {
|
||||
return fmt.Errorf("validate credential activity: negative count")
|
||||
}
|
||||
successCount, ok := usage.CheckedAdd(activity.SuccessCount, row.SuccessCount)
|
||||
if !ok {
|
||||
return fmt.Errorf("aggregate credential activity success count: checked addition failed")
|
||||
}
|
||||
failureCount, ok := usage.CheckedAdd(activity.FailureCount, row.FailureCount)
|
||||
if !ok {
|
||||
return fmt.Errorf("aggregate credential activity failure count: checked addition failed")
|
||||
}
|
||||
activity.SuccessCount = successCount
|
||||
activity.FailureCount = failureCount
|
||||
result[row.CredentialID] = activity
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
package requestlog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"gpt-load/internal/storage/models"
|
||||
"gpt-load/internal/telemetry"
|
||||
)
|
||||
|
||||
func TestQueryCredentialActivityCombinesHourlyStatsAndOldestPartialHour(t *testing.T) {
|
||||
db := openRequestLogQueryDB(t)
|
||||
service := newRequestLogTestService(db)
|
||||
now := time.Date(2026, time.August, 15, 13, 45, 0, 0, time.UTC)
|
||||
service.now = func() time.Time { return now }
|
||||
from := now.Add(-24 * time.Hour)
|
||||
|
||||
createCredentialAttemptStats(t, db,
|
||||
models.CredentialAttemptStat{
|
||||
CredentialID: 11, BucketStartMS: from.Truncate(time.Hour).UnixMilli(),
|
||||
SuccessCount: 100, FailureCount: 100,
|
||||
},
|
||||
models.CredentialAttemptStat{
|
||||
CredentialID: 11, BucketStartMS: from.Truncate(time.Hour).Add(time.Hour).UnixMilli(),
|
||||
SuccessCount: 2, FailureCount: 1,
|
||||
},
|
||||
models.CredentialAttemptStat{
|
||||
CredentialID: 11, BucketStartMS: now.Truncate(time.Hour).UnixMilli(),
|
||||
SuccessCount: 3, FailureCount: 4,
|
||||
},
|
||||
models.CredentialAttemptStat{
|
||||
CredentialID: 12, BucketStartMS: now.Truncate(time.Hour).UnixMilli(),
|
||||
SuccessCount: 7, FailureCount: 8,
|
||||
},
|
||||
models.CredentialAttemptStat{
|
||||
CredentialID: 13, BucketStartMS: now.Truncate(time.Hour).UnixMilli(),
|
||||
SuccessCount: 200, FailureCount: 300,
|
||||
},
|
||||
)
|
||||
|
||||
createCredentialActivityAttempt(t, db, 1, 11, from.Add(5*time.Minute), telemetry.FailureCategoryOK)
|
||||
createCredentialActivityAttempt(t, db, 2, 11, from.Add(6*time.Minute), telemetry.FailureCategoryRateLimited)
|
||||
createCredentialActivityAttempt(t, db, 3, 11, from.Add(7*time.Minute), telemetry.FailureCategoryDownstreamCancel)
|
||||
createCredentialActivityAttempt(t, db, 4, 11, from.Add(-time.Minute), telemetry.FailureCategoryAmbiguous)
|
||||
latest11 := now.Add(-time.Minute)
|
||||
createCredentialActivityAttempt(t, db, 5, 11, latest11, telemetry.FailureCategoryOK)
|
||||
latest12 := now.Add(-2 * time.Minute)
|
||||
createCredentialActivityAttempt(t, db, 6, 12, latest12, telemetry.FailureCategoryOK)
|
||||
createCredentialActivityAttempt(t, db, 7, 13, now.Add(-time.Second), telemetry.FailureCategoryOK)
|
||||
|
||||
queryCount := 0
|
||||
const callbackName = "test:credential_activity_query_count"
|
||||
if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(*gorm.DB) {
|
||||
queryCount++
|
||||
}); err != nil {
|
||||
t.Fatalf("register query counter: %v", err)
|
||||
}
|
||||
if err := db.Callback().Row().After("gorm:row").Register(callbackName, func(*gorm.DB) {
|
||||
queryCount++
|
||||
}); err != nil {
|
||||
t.Fatalf("register row query counter: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = db.Callback().Query().Remove(callbackName)
|
||||
_ = db.Callback().Row().Remove(callbackName)
|
||||
})
|
||||
|
||||
result, err := service.QueryCredentialActivity(context.Background(), CredentialActivityQuery{
|
||||
CredentialIDs: []uint{11, 12},
|
||||
FromMS: from.UnixMilli(),
|
||||
ToMS: now.UnixMilli(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("QueryCredentialActivity() error = %v", err)
|
||||
}
|
||||
if queryCount != 3 {
|
||||
t.Fatalf("query count = %d, want one latest, one hourly, and one boundary query", queryCount)
|
||||
}
|
||||
if len(result) != 2 {
|
||||
t.Fatalf("result = %#v, want only requested credentials", result)
|
||||
}
|
||||
first := result[11]
|
||||
if first.CredentialID != 11 || first.SuccessCount != 6 || first.FailureCount != 6 ||
|
||||
!first.DataComplete || first.LastUsedAtMS == nil || *first.LastUsedAtMS != latest11.UnixMilli() {
|
||||
t.Fatalf("credential 11 activity = %#v", first)
|
||||
}
|
||||
second := result[12]
|
||||
if second.CredentialID != 12 || second.SuccessCount != 7 || second.FailureCount != 8 ||
|
||||
!second.DataComplete || second.LastUsedAtMS == nil || *second.LastUsedAtMS != latest12.UnixMilli() {
|
||||
t.Fatalf("credential 12 activity = %#v", second)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryCredentialActivityReturnsZeroActivityAndIncompleteRetention(t *testing.T) {
|
||||
db := openRequestLogQueryDB(t)
|
||||
service := NewService(db, nil, staticRetentionPolicy{days: 0})
|
||||
now := time.Date(2026, time.August, 15, 13, 45, 0, 0, time.UTC)
|
||||
service.now = func() time.Time { return now }
|
||||
|
||||
result, err := service.QueryCredentialActivity(t.Context(), CredentialActivityQuery{
|
||||
CredentialIDs: []uint{21},
|
||||
FromMS: now.Add(-24 * time.Hour).UnixMilli(),
|
||||
ToMS: now.UnixMilli(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("QueryCredentialActivity() error = %v", err)
|
||||
}
|
||||
activity, exists := result[21]
|
||||
if !exists || activity.CredentialID != 21 || activity.SuccessCount != 0 ||
|
||||
activity.FailureCount != 0 || activity.LastUsedAtMS != nil || activity.DataComplete {
|
||||
t.Fatalf("empty activity = %#v, exists = %v", activity, exists)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryCredentialActivityValidatesScope(t *testing.T) {
|
||||
service := newRequestLogTestService(openRequestLogQueryDB(t))
|
||||
valid := CredentialActivityQuery{CredentialIDs: []uint{1}, FromMS: 1, ToMS: 2}
|
||||
tooMany := make([]uint, 101)
|
||||
for index := range tooMany {
|
||||
tooMany[index] = uint(index + 1)
|
||||
}
|
||||
|
||||
tests := map[string]CredentialActivityQuery{
|
||||
"empty IDs": {FromMS: 1, ToMS: 2},
|
||||
"zero ID": {CredentialIDs: []uint{0}, FromMS: 1, ToMS: 2},
|
||||
"too many IDs": {CredentialIDs: tooMany, FromMS: 1, ToMS: 2},
|
||||
"negative start": {CredentialIDs: []uint{1}, FromMS: -1, ToMS: 2},
|
||||
"empty time range": {CredentialIDs: []uint{1}, FromMS: 2, ToMS: 2},
|
||||
"excessive range": {CredentialIDs: []uint{1}, FromMS: 1, ToMS: 1 + credentialActivityMaxWindowMS + 1},
|
||||
}
|
||||
for name, input := range tests {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
if _, err := service.QueryCredentialActivity(t.Context(), input); err == nil {
|
||||
t.Fatalf("QueryCredentialActivity(%#v) error = nil", input)
|
||||
}
|
||||
})
|
||||
}
|
||||
if _, err := (*Service)(nil).QueryCredentialActivity(t.Context(), valid); err == nil {
|
||||
t.Fatal("nil service error = nil")
|
||||
}
|
||||
}
|
||||
|
||||
func createCredentialAttemptStats(t *testing.T, db *gorm.DB, rows ...models.CredentialAttemptStat) {
|
||||
t.Helper()
|
||||
if err := db.Create(&rows).Error; err != nil {
|
||||
t.Fatalf("create CredentialAttemptStats: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func createCredentialActivityAttempt(
|
||||
t *testing.T,
|
||||
db *gorm.DB,
|
||||
sequence int,
|
||||
credentialID uint,
|
||||
completedAt time.Time,
|
||||
category telemetry.FailureCategory,
|
||||
) {
|
||||
t.Helper()
|
||||
id := fmt.Sprintf("00000000-0000-4000-8000-%012d", sequence)
|
||||
row := requestLogQueryRow(id, completedAt, 1, "activity", []Attempt{{
|
||||
CredentialID: credentialID, FailureCategory: category,
|
||||
}})
|
||||
row.AttemptRows[0].CompletedAtMS = completedAt.UnixMilli()
|
||||
createRequestLogQueryRow(t, db, row)
|
||||
}
|
||||
@@ -5,7 +5,10 @@ import (
|
||||
"fmt"
|
||||
"math"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"gpt-load/internal/platform/epochms"
|
||||
"gpt-load/internal/storage/dbtx"
|
||||
"gpt-load/internal/storage/models"
|
||||
)
|
||||
|
||||
@@ -43,20 +46,30 @@ func (service *Service) QueryCredentialWindowUsage(
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
|
||||
var result CredentialWindowUsage
|
||||
var err error
|
||||
switch input.Source {
|
||||
case CredentialWindowUsageSourceRequestLogs:
|
||||
result, err = service.queryCredentialRequestLogUsage(ctx, input)
|
||||
case CredentialWindowUsageSourceHourlyStats:
|
||||
result, err = service.queryCredentialHourlyUsage(ctx, input)
|
||||
case CredentialWindowUsageSourceRequestLogs, CredentialWindowUsageSourceHourlyStats:
|
||||
default:
|
||||
return CredentialWindowUsage{}, fmt.Errorf(
|
||||
"query credential window usage: unsupported source %q",
|
||||
input.Source,
|
||||
)
|
||||
}
|
||||
|
||||
var result CredentialWindowUsage
|
||||
err := dbtx.Run(ctx, service.db, dbtx.Options{
|
||||
Mode: dbtx.ReadSnapshot,
|
||||
CleanupTimeout: usageRollbackTimeout,
|
||||
Operation: "credential window usage read transaction",
|
||||
}, func(connection *gorm.DB) error {
|
||||
var queryErr error
|
||||
switch input.Source {
|
||||
case CredentialWindowUsageSourceRequestLogs:
|
||||
result, queryErr = service.queryCredentialRequestLogUsage(connection, input)
|
||||
case CredentialWindowUsageSourceHourlyStats:
|
||||
result, queryErr = service.queryCredentialHourlyUsage(connection, input)
|
||||
}
|
||||
return queryErr
|
||||
})
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("query credential window usage: %w", err)
|
||||
}
|
||||
@@ -64,11 +77,11 @@ func (service *Service) QueryCredentialWindowUsage(
|
||||
}
|
||||
|
||||
func (service *Service) queryCredentialRequestLogUsage(
|
||||
ctx context.Context,
|
||||
db *gorm.DB,
|
||||
input CredentialWindowUsageQuery,
|
||||
) (CredentialWindowUsage, error) {
|
||||
var row credentialRequestLogUsageRow
|
||||
query := service.db.WithContext(ctx).Model(&models.RequestLog{}).
|
||||
query := db.Model(&models.RequestLog{}).
|
||||
Where("credential_id = ?", input.CredentialID).
|
||||
Where("completed_at_ms >= ? AND completed_at_ms < ?", input.FromMS, input.ToMS)
|
||||
if err := query.Select(credentialRequestLogUsageSelect).Find(&row).Error; err != nil {
|
||||
@@ -86,40 +99,71 @@ func (service *Service) queryCredentialRequestLogUsage(
|
||||
}
|
||||
|
||||
func (service *Service) queryCredentialHourlyUsage(
|
||||
ctx context.Context,
|
||||
db *gorm.DB,
|
||||
input CredentialWindowUsageQuery,
|
||||
) (CredentialWindowUsage, error) {
|
||||
fromMS, err := epochms.AlignDown(input.FromMS, epochms.MillisecondsPerHour)
|
||||
fullHoursFromMS, err := alignHourUp(input.FromMS)
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("align window start: %w", err)
|
||||
}
|
||||
toMS, err := alignHourUp(input.ToMS)
|
||||
fullHoursToMS, err := alignHourUp(input.ToMS)
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("align window end: %w", err)
|
||||
}
|
||||
scope := service.db.WithContext(ctx).Model(&models.UsageStat{}).
|
||||
Where("credential_id = ?", input.CredentialID).
|
||||
Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fromMS, toMS)
|
||||
if err := validateUsageStatIntegrity(scope); err != nil {
|
||||
return CredentialWindowUsage{}, err
|
||||
result := CredentialWindowUsage{
|
||||
Source: CredentialWindowUsageSourceHourlyStats,
|
||||
DataComplete: true,
|
||||
}
|
||||
aggregate, err := queryUsageSummary(scope)
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, err
|
||||
if fullHoursFromMS < fullHoursToMS {
|
||||
scope := db.Model(&models.UsageStat{}).
|
||||
Where("credential_id = ?", input.CredentialID).
|
||||
Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fullHoursFromMS, fullHoursToMS)
|
||||
if err := validateUsageStatIntegrity(scope); err != nil {
|
||||
return CredentialWindowUsage{}, err
|
||||
}
|
||||
aggregate, err := queryUsageSummary(scope)
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, err
|
||||
}
|
||||
result.UsageAggregate = aggregate
|
||||
var latest struct {
|
||||
LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"`
|
||||
}
|
||||
if err := scope.Select("MAX(bucket_start_ms) AS last_used_at_ms").Find(&latest).Error; err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("query latest usage bucket: %w", err)
|
||||
}
|
||||
result.LastUsedAtMS = latest.LastUsedAtMS
|
||||
result.DataComplete = service.hourlyWindowRetained(fullHoursFromMS)
|
||||
}
|
||||
var latest struct {
|
||||
LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"`
|
||||
|
||||
boundaryToMS := fullHoursFromMS
|
||||
if boundaryToMS > input.ToMS {
|
||||
boundaryToMS = input.ToMS
|
||||
}
|
||||
if err := scope.Select("MAX(bucket_start_ms) AS last_used_at_ms").Find(&latest).Error; err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("query latest usage bucket: %w", err)
|
||||
if input.FromMS < boundaryToMS {
|
||||
boundary, err := service.queryCredentialRequestLogUsage(db, CredentialWindowUsageQuery{
|
||||
CredentialID: input.CredentialID,
|
||||
FromMS: input.FromMS,
|
||||
ToMS: boundaryToMS,
|
||||
Source: CredentialWindowUsageSourceRequestLogs,
|
||||
})
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, err
|
||||
}
|
||||
result.UsageAggregate, err = addUsageAggregates(
|
||||
result.UsageAggregate,
|
||||
boundary.UsageAggregate,
|
||||
)
|
||||
if err != nil {
|
||||
return CredentialWindowUsage{}, fmt.Errorf("merge boundary usage: %w", err)
|
||||
}
|
||||
result.DataComplete = result.DataComplete && boundary.DataComplete
|
||||
if boundary.LastUsedAtMS != nil &&
|
||||
(result.LastUsedAtMS == nil || *boundary.LastUsedAtMS > *result.LastUsedAtMS) {
|
||||
result.LastUsedAtMS = boundary.LastUsedAtMS
|
||||
}
|
||||
}
|
||||
return CredentialWindowUsage{
|
||||
UsageAggregate: aggregate,
|
||||
Source: CredentialWindowUsageSourceHourlyStats,
|
||||
DataComplete: input.FromMS == fromMS && input.ToMS == toMS &&
|
||||
service.hourlyWindowRetained(input.FromMS),
|
||||
LastUsedAtMS: latest.LastUsedAtMS,
|
||||
}, nil
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (service *Service) requestLogWindowRetained(fromMS int64) bool {
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"gpt-load/internal/pricing"
|
||||
"gpt-load/internal/storage/models"
|
||||
"gpt-load/internal/telemetry"
|
||||
@@ -64,6 +66,13 @@ func TestQueryCredentialWindowUsageUsesHourlyStatsForLongWindow(t *testing.T) {
|
||||
createUsageStats(t, db, first, last, other)
|
||||
|
||||
from := now.Add(-7 * 24 * time.Hour).Add(10 * time.Minute)
|
||||
boundary := requestLogWindowRow(
|
||||
"00000000-0000-4000-8000-000000000735",
|
||||
credentialID,
|
||||
from.Add(5*time.Minute),
|
||||
40,
|
||||
)
|
||||
createRequestLogQueryRow(t, db, boundary)
|
||||
result, err := service.QueryCredentialWindowUsage(t.Context(), CredentialWindowUsageQuery{
|
||||
CredentialID: credentialID,
|
||||
FromMS: from.UnixMilli(),
|
||||
@@ -73,14 +82,66 @@ func TestQueryCredentialWindowUsageUsesHourlyStatsForLongWindow(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("QueryCredentialWindowUsage() error = %v", err)
|
||||
}
|
||||
if result.Source != CredentialWindowUsageSourceHourlyStats || result.DataComplete ||
|
||||
result.RequestCount != 5 || result.UncachedInputTokens != 50 ||
|
||||
result.EstimatedCostNanoUSD != 500_000_000 || result.LastUsedAtMS == nil ||
|
||||
if result.Source != CredentialWindowUsageSourceHourlyStats || !result.DataComplete ||
|
||||
result.RequestCount != 4 || result.UncachedInputTokens != 70 ||
|
||||
result.EstimatedCostNanoUSD != 300_000_000 || result.LastUsedAtMS == nil ||
|
||||
*result.LastUsedAtMS != last.BucketStartMS {
|
||||
t.Fatalf("hourly window usage = %#v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryCredentialWindowUsageUsesOneReadSnapshot(t *testing.T) {
|
||||
db, dsn := openRequestLogFileDB(t)
|
||||
service := newRequestLogTestService(db)
|
||||
now := time.Date(2026, time.August, 14, 12, 25, 0, 0, time.UTC)
|
||||
service.now = func() time.Time { return now }
|
||||
credentialID := uint(42)
|
||||
from := now.Add(-5 * time.Hour)
|
||||
hourly := usageStat(from.Add(time.Hour).Truncate(time.Hour), 17, "codex-a", 1)
|
||||
hourly.CredentialID = credentialID
|
||||
createUsageStats(t, db, hourly)
|
||||
|
||||
writerDB, closeWriter := openUsageQueryWriterDB(t, dsn)
|
||||
defer closeWriter()
|
||||
inserted := false
|
||||
const callbackName = "test:credential_window_usage_snapshot_insert"
|
||||
if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) {
|
||||
if inserted || tx.Statement.Table != "usage_stats" {
|
||||
return
|
||||
}
|
||||
inserted = true
|
||||
lateBoundary := requestLogWindowRow(
|
||||
"00000000-0000-4000-8000-000000000736",
|
||||
credentialID,
|
||||
from.Add(5*time.Minute),
|
||||
40,
|
||||
)
|
||||
if err := writerDB.Create(&lateBoundary).Error; err != nil {
|
||||
t.Errorf("insert concurrent RequestLog: %v", err)
|
||||
}
|
||||
}); err != nil {
|
||||
t.Fatalf("register query callback: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := db.Callback().Query().Remove(callbackName); err != nil {
|
||||
t.Errorf("remove query callback: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
result, err := service.QueryCredentialWindowUsage(t.Context(), CredentialWindowUsageQuery{
|
||||
CredentialID: credentialID,
|
||||
FromMS: from.UnixMilli(),
|
||||
ToMS: now.UnixMilli(),
|
||||
Source: CredentialWindowUsageSourceHourlyStats,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("QueryCredentialWindowUsage() error = %v", err)
|
||||
}
|
||||
if !inserted || result.RequestCount != 1 {
|
||||
t.Fatalf("window usage did not retain one snapshot: inserted=%t result=%#v", inserted, result)
|
||||
}
|
||||
}
|
||||
|
||||
func requestLogWindowRow(id string, credentialID uint, completedAt time.Time, input int64) models.RequestLog {
|
||||
return models.RequestLog{
|
||||
ID: id, CompletedAtMS: completedAt.UnixMilli(), AccessKeyID: 1, GroupID: 17,
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"gpt-load/internal/platform/redact"
|
||||
"gpt-load/internal/storage"
|
||||
"gpt-load/internal/storage/models"
|
||||
"gpt-load/internal/telemetry"
|
||||
)
|
||||
|
||||
// TestExternalDatabaseRequestLogLifecycle covers the request-log write,
|
||||
@@ -48,6 +49,13 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) {
|
||||
row := aggregationRow(requestID, now.Add(-40*24*time.Hour), groupID, modelID)
|
||||
row.ChannelID = "openai"
|
||||
row.CredentialID = credentialID
|
||||
row.AttemptRows = []models.RequestLogAttempt{{
|
||||
RequestID: row.ID, Sequence: 1, CompletedAtMS: row.CompletedAtMS,
|
||||
GroupID: groupID, GroupName: "external-request-log", ChannelID: row.ChannelID,
|
||||
CredentialID: credentialID, StatusCode: 200,
|
||||
FailureCategory: string(telemetry.FailureCategoryOK),
|
||||
Action: string(telemetry.ActionTerminate),
|
||||
}}
|
||||
writer := &gormBatchWriter{db: db}
|
||||
if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil {
|
||||
t.Fatalf("first WriteBatch() error = %v", err)
|
||||
@@ -69,8 +77,24 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) {
|
||||
modelID,
|
||||
1,
|
||||
)
|
||||
assertExternalCredentialAttemptStat(t, db, row.CompletedAtMS, credentialID, 1)
|
||||
|
||||
service := NewService(db, redact.New(), staticRetentionPolicy{days: 1})
|
||||
service.now = func() time.Time { return now }
|
||||
activity, err := service.QueryCredentialActivity(t.Context(), CredentialActivityQuery{
|
||||
CredentialIDs: []uint{credentialID},
|
||||
FromMS: row.CompletedAtMS,
|
||||
ToMS: row.CompletedAtMS + int64(time.Hour/time.Millisecond),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("QueryCredentialActivity() error = %v", err)
|
||||
}
|
||||
credentialActivity := activity[credentialID]
|
||||
if credentialActivity.SuccessCount != 1 || credentialActivity.FailureCount != 0 ||
|
||||
credentialActivity.LastUsedAtMS == nil ||
|
||||
*credentialActivity.LastUsedAtMS != row.CompletedAtMS {
|
||||
t.Fatalf("credential activity = %+v, want one exact successful attempt", credentialActivity)
|
||||
}
|
||||
service.Sweep(context.Background(), now)
|
||||
assertExternalRequestLogCount(t, db, "id = ?", []any{requestID}, 0)
|
||||
assertExternalUsageStatCount(
|
||||
@@ -84,6 +108,7 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) {
|
||||
modelID,
|
||||
0,
|
||||
)
|
||||
assertExternalCredentialAttemptStat(t, db, row.CompletedAtMS, credentialID, 0)
|
||||
var journals int64
|
||||
if err := db.Model(&models.UsageAggregationJournal{}).
|
||||
Where("request_id = ?", requestID).
|
||||
@@ -95,6 +120,30 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func assertExternalCredentialAttemptStat(
|
||||
t *testing.T,
|
||||
db *gorm.DB,
|
||||
bucketStartMS int64,
|
||||
credentialID uint,
|
||||
wantRows int64,
|
||||
) {
|
||||
t.Helper()
|
||||
var rows []models.CredentialAttemptStat
|
||||
if err := db.Where(
|
||||
"bucket_start_ms = ? AND credential_id = ?",
|
||||
bucketStartMS,
|
||||
credentialID,
|
||||
).Find(&rows).Error; err != nil {
|
||||
t.Fatalf("query credential attempt stats: %v", err)
|
||||
}
|
||||
if int64(len(rows)) != wantRows {
|
||||
t.Fatalf("credential attempt stat rows = %+v, want %d", rows, wantRows)
|
||||
}
|
||||
if wantRows == 1 && (rows[0].SuccessCount != 1 || rows[0].FailureCount != 0) {
|
||||
t.Fatalf("credential attempt stat = %+v, want one success", rows[0])
|
||||
}
|
||||
}
|
||||
|
||||
func assertExternalRequestLogCount(t *testing.T, db *gorm.DB, query string, args []any, want int64) {
|
||||
t.Helper()
|
||||
var count int64
|
||||
|
||||
@@ -43,6 +43,17 @@ func mapEvent(
|
||||
}
|
||||
attempts := make([]models.RequestLogAttempt, 0, len(event.Attempts))
|
||||
for _, attempt := range event.Attempts {
|
||||
attemptCompletedAtMS := completedAtMS
|
||||
if !attempt.CompletedAt.IsZero() {
|
||||
attemptCompletedAtMS, err = epochms.FromTime(attempt.CompletedAt)
|
||||
if err != nil {
|
||||
return models.RequestLog{}, fmt.Errorf(
|
||||
"map request event attempt %d completion time: %w",
|
||||
attempt.Sequence,
|
||||
err,
|
||||
)
|
||||
}
|
||||
}
|
||||
attemptReceipt := models.JSON(nil)
|
||||
if attempt.Sequence == event.Usage.AttemptSequence {
|
||||
attemptReceipt = receipt
|
||||
@@ -50,7 +61,7 @@ func mapEvent(
|
||||
attempts = append(attempts, models.RequestLogAttempt{
|
||||
RequestID: event.RequestID,
|
||||
Sequence: attempt.Sequence,
|
||||
CompletedAtMS: completedAtMS,
|
||||
CompletedAtMS: attemptCompletedAtMS,
|
||||
GroupID: attempt.GroupID,
|
||||
GroupName: redactIdentityValue(redactor, attempt.GroupName),
|
||||
ChannelID: string(attempt.ChannelID),
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
func TestMapEventPersistsFrozenUsagePricingAndAttribution(t *testing.T) {
|
||||
event := testEvent("00000000-0000-4000-8000-000000000101")
|
||||
event.CompletedAt = time.Date(2026, time.July, 24, 20, 30, 0, 123, time.FixedZone("test", 8*60*60))
|
||||
event.Attempts[0].CompletedAt = event.CompletedAt.Add(-90 * time.Second)
|
||||
event.Protocol = protocol.OpenAICompletions
|
||||
event.Operation = execution.OperationChatCompletion
|
||||
event.ClientModel = "client-alias"
|
||||
@@ -83,7 +84,8 @@ func TestMapEventPersistsFrozenUsagePricingAndAttribution(t *testing.T) {
|
||||
}
|
||||
|
||||
if len(row.AttemptRows) != 1 || row.AttemptRows[0].GroupID != 7 ||
|
||||
row.AttemptRows[0].CredentialID != 8 {
|
||||
row.AttemptRows[0].CredentialID != 8 ||
|
||||
row.AttemptRows[0].CompletedAtMS != event.Attempts[0].CompletedAt.UnixMilli() {
|
||||
t.Fatalf("attempts = %+v", row.AttemptRows)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ const (
|
||||
usageStatRetentionDays = 35
|
||||
)
|
||||
|
||||
// Sweep removes request logs, hourly usage aggregates, and aggregation journals
|
||||
// Sweep removes request logs, hourly aggregates, and aggregation journals
|
||||
// strictly older than their respective retention boundaries. Failures are
|
||||
// isolated from the data plane.
|
||||
func (service *Service) Sweep(ctx context.Context, now time.Time) {
|
||||
@@ -62,9 +62,56 @@ func (service *Service) Sweep(ctx context.Context, now time.Time) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
service.deleteExpiredCredentialAttemptStats(ctx, usageStatCutoffMS, now)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
service.deleteExpiredUsageJournals(ctx, usageStatCutoffMS, now)
|
||||
}
|
||||
|
||||
func (service *Service) deleteExpiredCredentialAttemptStats(
|
||||
ctx context.Context,
|
||||
cutoffMS int64,
|
||||
now time.Time,
|
||||
) bool {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
var ids []uint
|
||||
result := service.db.WithContext(ctx).
|
||||
Model(&models.CredentialAttemptStat{}).
|
||||
Where("bucket_start_ms < ?", cutoffMS).
|
||||
Order("bucket_start_ms ASC").
|
||||
Order("id ASC").
|
||||
Limit(retentionBatchSize).
|
||||
Pluck("id", &ids)
|
||||
if result.Error != nil {
|
||||
if ctx.Err() == nil {
|
||||
service.recordRetentionDeleteFailure(now)
|
||||
}
|
||||
return false
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
result = service.db.WithContext(ctx).
|
||||
Where("id IN ?", ids).
|
||||
Delete(&models.CredentialAttemptStat{})
|
||||
if result.Error != nil {
|
||||
if ctx.Err() == nil {
|
||||
service.recordRetentionDeleteFailure(now)
|
||||
}
|
||||
return false
|
||||
}
|
||||
if len(ids) < retentionBatchSize {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func retentionCutoffMS(nowMS int64, days int) (int64, error) {
|
||||
if days < 0 {
|
||||
return 0, fmt.Errorf("retention days must be non-negative")
|
||||
|
||||
@@ -122,6 +122,11 @@ func TestRetentionSweepUsesFixedThirtyFiveDayIntegerUsageBoundary(t *testing.T)
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create UsageStat at %d: %v", bucketStartMS, err)
|
||||
}
|
||||
if err := db.Create(&models.CredentialAttemptStat{
|
||||
CredentialID: uint(index + 1), BucketStartMS: bucketStartMS, SuccessCount: 1,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create CredentialAttemptStat at %d: %v", bucketStartMS, err)
|
||||
}
|
||||
}
|
||||
for index, item := range []struct {
|
||||
bucketStartMS int64
|
||||
@@ -156,6 +161,18 @@ func TestRetentionSweepUsesFixedThirtyFiveDayIntegerUsageBoundary(t *testing.T)
|
||||
remaining[1].BucketStartMS != usageCutoffMS+3_600_000 {
|
||||
t.Fatalf("remaining UsageStats = %+v, want cutoff and newer integer buckets", remaining)
|
||||
}
|
||||
var remainingAttempts []models.CredentialAttemptStat
|
||||
if err := db.Order("bucket_start_ms ASC").Find(&remainingAttempts).Error; err != nil {
|
||||
t.Fatalf("query remaining CredentialAttemptStats: %v", err)
|
||||
}
|
||||
if len(remainingAttempts) != 2 ||
|
||||
remainingAttempts[0].BucketStartMS != usageCutoffMS ||
|
||||
remainingAttempts[1].BucketStartMS != usageCutoffMS+3_600_000 {
|
||||
t.Fatalf(
|
||||
"remaining CredentialAttemptStats = %+v, want cutoff and newer buckets",
|
||||
remainingAttempts,
|
||||
)
|
||||
}
|
||||
var journals []models.UsageAggregationJournal
|
||||
if err := db.Order("request_id ASC").Find(&journals).Error; err != nil {
|
||||
t.Fatalf("query remaining UsageAggregationJournals: %v", err)
|
||||
@@ -402,6 +419,13 @@ func TestRetentionReplayBoundaryKeepsAggregationIdempotentWithoutRequestLog(t *t
|
||||
row := aggregationRow(aggregationRequestID(70), now.Add(-8*24*time.Hour), 15, "retention-aggregate")
|
||||
row.Status = string(telemetry.RequestStatusSuccess)
|
||||
row.UsageState = string(usage.StateComplete)
|
||||
row.AttemptRows = []models.RequestLogAttempt{credentialAttemptRow(
|
||||
row.ID,
|
||||
1,
|
||||
now.Add(-8*24*time.Hour),
|
||||
23,
|
||||
telemetry.FailureCategoryOK,
|
||||
)}
|
||||
|
||||
if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil {
|
||||
t.Fatalf("initial WriteBatch() error = %v", err)
|
||||
@@ -437,10 +461,21 @@ func assertRetentionReplayState(
|
||||
if err := db.Find(&stats).Error; err != nil {
|
||||
t.Fatalf("query UsageStats: %v", err)
|
||||
}
|
||||
var attemptStats []models.CredentialAttemptStat
|
||||
if err := db.Find(&attemptStats).Error; err != nil {
|
||||
t.Fatalf("query CredentialAttemptStats: %v", err)
|
||||
}
|
||||
if requestLogs != wantRequestLogs || len(stats) != 1 ||
|
||||
stats[0].RequestCount != wantAggregatedRequests {
|
||||
t.Fatalf("retention replay state = logs:%d stats:%+v, want logs:%d requests:%d",
|
||||
requestLogs, stats, wantRequestLogs, wantAggregatedRequests)
|
||||
stats[0].RequestCount != wantAggregatedRequests || len(attemptStats) != 1 ||
|
||||
attemptStats[0].SuccessCount != 1 || attemptStats[0].FailureCount != 0 {
|
||||
t.Fatalf(
|
||||
"retention replay state = logs:%d stats:%+v attempt stats:%+v, want logs:%d requests:%d and one attempt",
|
||||
requestLogs,
|
||||
stats,
|
||||
attemptStats,
|
||||
wantRequestLogs,
|
||||
wantAggregatedRequests,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -168,9 +168,151 @@ func writeRequestLogBatch(tx *gorm.DB, rows []models.RequestLog) error {
|
||||
Order("request_id ASC").Find(&journals).Error; err != nil {
|
||||
return fmt.Errorf("query current usage journals: %w", err)
|
||||
}
|
||||
if len(attemptRows) > 0 && len(journals) > 0 {
|
||||
pendingRequestIDs := make(map[string]struct{}, len(journals))
|
||||
for _, journal := range journals {
|
||||
pendingRequestIDs[journal.RequestID] = struct{}{}
|
||||
}
|
||||
pendingAttempts := make([]models.RequestLogAttempt, 0, len(attemptRows))
|
||||
for _, attempt := range attemptRows {
|
||||
if _, pending := pendingRequestIDs[attempt.RequestID]; pending {
|
||||
pendingAttempts = append(pendingAttempts, attempt)
|
||||
}
|
||||
}
|
||||
if err := applyCredentialAttemptStats(tx, pendingAttempts); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return applyUsageJournalBatch(tx, journals)
|
||||
}
|
||||
|
||||
type credentialAttemptStatKey struct {
|
||||
CredentialID uint
|
||||
BucketStartMS int64
|
||||
}
|
||||
|
||||
type credentialAttemptStatDelta struct {
|
||||
SuccessCount int64
|
||||
FailureCount int64
|
||||
}
|
||||
|
||||
func applyCredentialAttemptStats(tx *gorm.DB, attempts []models.RequestLogAttempt) error {
|
||||
deltas, err := buildCredentialAttemptStatDeltas(attempts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(deltas) == 0 {
|
||||
return nil
|
||||
}
|
||||
keys := make([]credentialAttemptStatKey, 0, len(deltas))
|
||||
credentialSet := make(map[uint]struct{})
|
||||
bucketSet := make(map[int64]struct{})
|
||||
for key := range deltas {
|
||||
keys = append(keys, key)
|
||||
credentialSet[key.CredentialID] = struct{}{}
|
||||
bucketSet[key.BucketStartMS] = struct{}{}
|
||||
}
|
||||
sort.Slice(keys, func(left, right int) bool {
|
||||
if keys[left].CredentialID != keys[right].CredentialID {
|
||||
return keys[left].CredentialID < keys[right].CredentialID
|
||||
}
|
||||
return keys[left].BucketStartMS < keys[right].BucketStartMS
|
||||
})
|
||||
credentialIDs := make([]uint, 0, len(credentialSet))
|
||||
for credentialID := range credentialSet {
|
||||
credentialIDs = append(credentialIDs, credentialID)
|
||||
}
|
||||
bucketStarts := make([]int64, 0, len(bucketSet))
|
||||
for bucketStartMS := range bucketSet {
|
||||
bucketStarts = append(bucketStarts, bucketStartMS)
|
||||
}
|
||||
|
||||
var existingRows []models.CredentialAttemptStat
|
||||
if err := tx.Where("credential_id IN ? AND bucket_start_ms IN ?", credentialIDs, bucketStarts).
|
||||
Find(&existingRows).Error; err != nil {
|
||||
return fmt.Errorf("query credential attempt stats: %w", err)
|
||||
}
|
||||
existing := make(map[credentialAttemptStatKey]models.CredentialAttemptStat, len(existingRows))
|
||||
for _, row := range existingRows {
|
||||
if row.SuccessCount < 0 || row.FailureCount < 0 {
|
||||
return fmt.Errorf("query credential attempt stats: corrupt row")
|
||||
}
|
||||
existing[credentialAttemptStatKey{
|
||||
CredentialID: row.CredentialID, BucketStartMS: row.BucketStartMS,
|
||||
}] = row
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
row := existing[key]
|
||||
row.CredentialID = key.CredentialID
|
||||
row.BucketStartMS = key.BucketStartMS
|
||||
delta := deltas[key]
|
||||
if err := checkedInt64Add(&row.SuccessCount, delta.SuccessCount, "credential attempt success_count"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := checkedInt64Add(&row.FailureCount, delta.FailureCount, "credential attempt failure_count"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "credential_id"}, {Name: "bucket_start_ms"}},
|
||||
DoUpdates: clause.AssignmentColumns([]string{
|
||||
"success_count", "failure_count",
|
||||
}),
|
||||
}).Create(&row).Error; err != nil {
|
||||
return fmt.Errorf("upsert credential attempt stat: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func buildCredentialAttemptStatDeltas(
|
||||
attempts []models.RequestLogAttempt,
|
||||
) (map[credentialAttemptStatKey]credentialAttemptStatDelta, error) {
|
||||
deltas := make(map[credentialAttemptStatKey]credentialAttemptStatDelta)
|
||||
for _, attempt := range attempts {
|
||||
if attempt.CredentialID == 0 {
|
||||
return nil, fmt.Errorf("aggregate credential attempt: credential ID is zero")
|
||||
}
|
||||
if attempt.FailureCategory == string(telemetry.FailureCategoryDownstreamCancel) {
|
||||
continue
|
||||
}
|
||||
bucketStartMS, err := epochms.AlignDown(
|
||||
attempt.CompletedAtMS,
|
||||
epochms.MillisecondsPerHour,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aggregate credential attempt completion time: %w", err)
|
||||
}
|
||||
key := credentialAttemptStatKey{
|
||||
CredentialID: attempt.CredentialID, BucketStartMS: bucketStartMS,
|
||||
}
|
||||
delta := deltas[key]
|
||||
switch telemetry.FailureCategory(attempt.FailureCategory) {
|
||||
case telemetry.FailureCategoryOK:
|
||||
if err := checkedInt64Add(&delta.SuccessCount, 1, "credential attempt success_count"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
case telemetry.FailureCategoryRateLimited,
|
||||
telemetry.FailureCategoryModelUnavailable,
|
||||
telemetry.FailureCategoryInvalidKey,
|
||||
telemetry.FailureCategoryUpstreamHost,
|
||||
telemetry.FailureCategoryClientError,
|
||||
telemetry.FailureCategoryConversionUnsupported,
|
||||
telemetry.FailureCategoryAmbiguous:
|
||||
if err := checkedInt64Add(&delta.FailureCount, 1, "credential attempt failure_count"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf(
|
||||
"aggregate credential attempt: invalid failure category %q",
|
||||
attempt.FailureCategory,
|
||||
)
|
||||
}
|
||||
deltas[key] = delta
|
||||
}
|
||||
return deltas, nil
|
||||
}
|
||||
|
||||
func applyUsageJournalBatch(
|
||||
tx *gorm.DB,
|
||||
journals []models.UsageAggregationJournal,
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -542,6 +543,77 @@ func TestWriteBatchInsertsOnlyNewRequestLogsAndAggregatesUsage(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteBatchAggregatesCredentialAttemptsByOwnCompletionHour(t *testing.T) {
|
||||
db := openRequestLogQueryDB(t)
|
||||
firstHour := time.Date(2026, time.August, 15, 14, 0, 0, 0, time.UTC)
|
||||
row := aggregationRow(aggregationRequestID(90), firstHour.Add(2*time.Hour), 7, "attempt-model")
|
||||
row.AttemptCount = 4
|
||||
row.AttemptRows = []models.RequestLogAttempt{
|
||||
credentialAttemptRow(row.ID, 1, firstHour.Add(time.Minute), 11, telemetry.FailureCategoryOK),
|
||||
credentialAttemptRow(row.ID, 2, firstHour.Add(2*time.Minute), 11, telemetry.FailureCategoryRateLimited),
|
||||
credentialAttemptRow(row.ID, 3, firstHour.Add(time.Hour+time.Minute), 22, telemetry.FailureCategoryInvalidKey),
|
||||
credentialAttemptRow(row.ID, 4, firstHour.Add(time.Hour+2*time.Minute), 22, telemetry.FailureCategoryDownstreamCancel),
|
||||
}
|
||||
|
||||
writer := &gormBatchWriter{db: db}
|
||||
if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil {
|
||||
t.Fatalf("WriteBatch() error = %v", err)
|
||||
}
|
||||
|
||||
var stats []models.CredentialAttemptStat
|
||||
if err := db.Order("credential_id ASC").Order("bucket_start_ms ASC").Find(&stats).Error; err != nil {
|
||||
t.Fatalf("query CredentialAttemptStats: %v", err)
|
||||
}
|
||||
want := []models.CredentialAttemptStat{
|
||||
{CredentialID: 11, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 1, FailureCount: 1},
|
||||
{CredentialID: 22, BucketStartMS: firstHour.Add(time.Hour).UnixMilli(), FailureCount: 1},
|
||||
}
|
||||
if len(stats) != len(want) {
|
||||
t.Fatalf("CredentialAttemptStat rows = %+v, want %+v", stats, want)
|
||||
}
|
||||
persisted := append([]models.CredentialAttemptStat(nil), stats...)
|
||||
for index := range want {
|
||||
stats[index].ID = 0
|
||||
if stats[index] != want[index] {
|
||||
t.Fatalf("CredentialAttemptStat[%d] = %+v, want %+v", index, stats[index], want[index])
|
||||
}
|
||||
}
|
||||
|
||||
if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil {
|
||||
t.Fatalf("replay WriteBatch() error = %v", err)
|
||||
}
|
||||
var replayed []models.CredentialAttemptStat
|
||||
if err := db.Order("credential_id ASC").Order("bucket_start_ms ASC").Find(&replayed).Error; err != nil {
|
||||
t.Fatalf("query replayed CredentialAttemptStats: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(replayed, persisted) {
|
||||
t.Fatalf("replay changed CredentialAttemptStats: before=%+v after=%+v", persisted, replayed)
|
||||
}
|
||||
}
|
||||
|
||||
func credentialAttemptRow(
|
||||
requestID string,
|
||||
sequence int,
|
||||
completedAt time.Time,
|
||||
credentialID uint,
|
||||
category telemetry.FailureCategory,
|
||||
) models.RequestLogAttempt {
|
||||
action := telemetry.ActionRetry
|
||||
statusCode := http.StatusTooManyRequests
|
||||
if category == telemetry.FailureCategoryOK {
|
||||
action = telemetry.ActionTerminate
|
||||
statusCode = http.StatusOK
|
||||
} else if category == telemetry.FailureCategoryDownstreamCancel {
|
||||
action = telemetry.ActionTerminate
|
||||
statusCode = 499
|
||||
}
|
||||
return models.RequestLogAttempt{
|
||||
RequestID: requestID, Sequence: sequence, CompletedAtMS: completedAt.UnixMilli(),
|
||||
GroupID: 7, GroupName: "attempt-group", ChannelID: "openai", CredentialID: credentialID,
|
||||
StatusCode: statusCode, FailureCategory: string(category), Action: string(action),
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteBatchAggregatesStatusQualityAndCompletePricedTotals(t *testing.T) {
|
||||
db := openRequestLogQueryDB(t)
|
||||
hour := time.Date(2026, time.July, 24, 13, 0, 0, 0, time.UTC)
|
||||
|
||||
@@ -89,7 +89,7 @@ func TestExternalDatabaseLifecycle(t *testing.T) {
|
||||
"groups", "credentials", "access_keys", "request_logs", "request_log_attempts",
|
||||
"usage_aggregation_journal", "usage_stats", "model_prices", "system_settings",
|
||||
"jobs", "control_operations", "credential_stages", "credential_observations",
|
||||
"credential_reset_operations", "schema_migrations",
|
||||
"credential_reset_operations", "credential_attempt_stats", "schema_migrations",
|
||||
} {
|
||||
if !db.Migrator().HasTable(table) {
|
||||
t.Fatalf("table %q is missing", table)
|
||||
|
||||
@@ -602,6 +602,7 @@ func TestAutoMigrateCreatesUsageJournalAndMigrationLedger(t *testing.T) {
|
||||
"credential_stages",
|
||||
"credential_observations",
|
||||
"credential_reset_operations",
|
||||
"credential_attempt_stats",
|
||||
"schema_migrations",
|
||||
}
|
||||
for _, table := range wantTables {
|
||||
@@ -619,6 +620,7 @@ func TestAutoMigrateCreatesUsageJournalAndMigrationLedger(t *testing.T) {
|
||||
"0002_subscription_connections",
|
||||
"0003_codex_channel",
|
||||
"0004_subscription_runtime",
|
||||
"0005_credential_attempt_stats",
|
||||
}
|
||||
if !reflect.DeepEqual(migrationIDs, wantMigrationIDs) {
|
||||
t.Fatalf("schema_migrations IDs = %v, want %v", migrationIDs, wantMigrationIDs)
|
||||
|
||||
@@ -58,6 +58,12 @@ var migrations = []migration{
|
||||
Validate: migrationfiles.Validate0004,
|
||||
ValidateRecoverable: migrationfiles.ValidateRecoverable0004,
|
||||
},
|
||||
{
|
||||
ID: migrationfiles.ID0005,
|
||||
Up: migrationfiles.Up0005,
|
||||
Validate: migrationfiles.Validate0005,
|
||||
ValidateRecoverable: migrationfiles.ValidateRecoverable0005,
|
||||
},
|
||||
}
|
||||
|
||||
func applyMigrations(db *gorm.DB) error {
|
||||
|
||||
@@ -11,10 +11,16 @@ import (
|
||||
)
|
||||
|
||||
func TestMigrationRegistryContainsOrderedMigrations(t *testing.T) {
|
||||
if len(migrations) != 4 {
|
||||
t.Fatalf("migration registry length = %d, want 4", len(migrations))
|
||||
if len(migrations) != 5 {
|
||||
t.Fatalf("migration registry length = %d, want 5", len(migrations))
|
||||
}
|
||||
wantIDs := []string{
|
||||
migrationfiles.ID0001,
|
||||
migrationfiles.ID0002,
|
||||
migrationfiles.ID0003,
|
||||
migrationfiles.ID0004,
|
||||
migrationfiles.ID0005,
|
||||
}
|
||||
wantIDs := []string{migrationfiles.ID0001, migrationfiles.ID0002, migrationfiles.ID0003, migrationfiles.ID0004}
|
||||
for index, entry := range migrations {
|
||||
if entry.ID != wantIDs[index] || entry.Up == nil ||
|
||||
entry.Validate == nil || entry.ValidateRecoverable == nil {
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package migrations
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
ID0005 = "0005_credential_attempt_stats"
|
||||
credentialAttemptBackfillHours = 25
|
||||
)
|
||||
|
||||
type credentialAttemptStat struct {
|
||||
ID uint `gorm:"primaryKey;autoIncrement;index:idx_credential_attempt_stats_bucket_id,priority:2"`
|
||||
CredentialID uint `gorm:"not null;check:chk_credential_attempt_stat_credential,credential_id > 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:1"`
|
||||
BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_credential_attempt_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:2;index:idx_credential_attempt_stats_bucket_id,priority:1"`
|
||||
SuccessCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_success_count,success_count >= 0"`
|
||||
FailureCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_failure_count,failure_count >= 0"`
|
||||
}
|
||||
|
||||
func (credentialAttemptStat) TableName() string { return "credential_attempt_stats" }
|
||||
|
||||
type credentialUsageStatIndex struct {
|
||||
CredentialID uint `gorm:"column:credential_id;index:idx_usage_stats_credential_bucket,priority:1"`
|
||||
BucketStartMS int64 `gorm:"column:bucket_start_ms;index:idx_usage_stats_credential_bucket,priority:2"`
|
||||
}
|
||||
|
||||
func (credentialUsageStatIndex) TableName() string { return "usage_stats" }
|
||||
|
||||
// Up0005 creates the credential/hour attempt aggregate and seeds the window
|
||||
// shown by subscription account cards from retained attempt records.
|
||||
func Up0005(db *gorm.DB) error {
|
||||
if err := ValidateRecoverable0005(db); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.AutoMigrate(&credentialAttemptStat{}); err != nil {
|
||||
return fmt.Errorf("create credential attempt stats: %w", err)
|
||||
}
|
||||
if !db.Migrator().HasIndex("usage_stats", "idx_usage_stats_credential_bucket") {
|
||||
if err := db.Migrator().CreateIndex(
|
||||
&credentialUsageStatIndex{},
|
||||
"idx_usage_stats_credential_bucket",
|
||||
); err != nil {
|
||||
return fmt.Errorf("create usage stat credential/time index: %w", err)
|
||||
}
|
||||
}
|
||||
cutoffMS := time.Now().UTC().Add(-credentialAttemptBackfillHours * time.Hour).UnixMilli()
|
||||
if err := backfillCredentialAttemptStats(db, cutoffMS); err != nil {
|
||||
return fmt.Errorf("backfill credential attempt stats: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func backfillCredentialAttemptStats(db *gorm.DB, cutoffMS int64) error {
|
||||
const aggregate = `
|
||||
SELECT
|
||||
attempt.credential_id,
|
||||
attempt.completed_at_ms - (attempt.completed_at_ms % 3600000) AS bucket_start_ms,
|
||||
SUM(CASE WHEN attempt.failure_category = 'ok' THEN 1 ELSE 0 END) AS success_count,
|
||||
SUM(CASE WHEN attempt.failure_category <> 'ok' THEN 1 ELSE 0 END) AS failure_count
|
||||
FROM request_log_attempts AS attempt
|
||||
WHERE attempt.request_id IN (
|
||||
SELECT request_log.id
|
||||
FROM request_logs AS request_log
|
||||
WHERE request_log.completed_at_ms >= ?
|
||||
)
|
||||
AND attempt.completed_at_ms >= ?
|
||||
AND attempt.failure_category <> 'downstream_cancel'
|
||||
GROUP BY
|
||||
attempt.credential_id,
|
||||
attempt.completed_at_ms - (attempt.completed_at_ms % 3600000)`
|
||||
|
||||
var statement string
|
||||
switch strings.ToLower(db.Dialector.Name()) {
|
||||
case "sqlite", "postgres", "postgresql":
|
||||
statement = `
|
||||
INSERT INTO credential_attempt_stats (
|
||||
credential_id, bucket_start_ms, success_count, failure_count
|
||||
)` + aggregate + `
|
||||
ON CONFLICT (credential_id, bucket_start_ms) DO NOTHING`
|
||||
case "mysql":
|
||||
statement = `
|
||||
INSERT INTO credential_attempt_stats (
|
||||
credential_id, bucket_start_ms, success_count, failure_count
|
||||
)` + aggregate + `
|
||||
ON DUPLICATE KEY UPDATE
|
||||
credential_id = credential_attempt_stats.credential_id`
|
||||
default:
|
||||
return fmt.Errorf("unsupported database driver %q", db.Dialector.Name())
|
||||
}
|
||||
return db.Exec(statement, cutoffMS, cutoffMS).Error
|
||||
}
|
||||
|
||||
func ValidateRecoverable0005(db *gorm.DB) error {
|
||||
for _, table := range []string{"request_logs", "request_log_attempts", "usage_stats"} {
|
||||
if !db.Migrator().HasTable(table) {
|
||||
return fmt.Errorf("required credential activity source table %q is missing", table)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Validate0005(db *gorm.DB) error {
|
||||
if err := ValidateRecoverable0005(db); err != nil {
|
||||
return err
|
||||
}
|
||||
if !db.Migrator().HasTable(&credentialAttemptStat{}) {
|
||||
return fmt.Errorf("validate credential attempt stats schema: table is missing")
|
||||
}
|
||||
for _, column := range []string{
|
||||
"id", "credential_id", "bucket_start_ms", "success_count", "failure_count",
|
||||
} {
|
||||
if !db.Migrator().HasColumn(&credentialAttemptStat{}, column) {
|
||||
return fmt.Errorf("validate credential attempt stats schema: column %q is missing", column)
|
||||
}
|
||||
}
|
||||
for _, index := range []string{
|
||||
"idx_credential_attempt_stats_identity",
|
||||
"idx_credential_attempt_stats_bucket_id",
|
||||
} {
|
||||
if !db.Migrator().HasIndex(&credentialAttemptStat{}, index) {
|
||||
return fmt.Errorf("validate credential attempt stats schema: index %q is missing", index)
|
||||
}
|
||||
}
|
||||
if !db.Migrator().HasIndex("usage_stats", "idx_usage_stats_credential_bucket") {
|
||||
return fmt.Errorf("validate credential attempt stats schema: usage stat credential/time index is missing")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,191 @@
|
||||
package migrations_test
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"gpt-load/internal/pricing"
|
||||
"gpt-load/internal/storage/migrations"
|
||||
"gpt-load/internal/storage/models"
|
||||
"gpt-load/internal/telemetry"
|
||||
"gpt-load/internal/usage"
|
||||
)
|
||||
|
||||
func TestCredentialAttemptStatsMigrationCreatesSchemaAndBackfillsRecentAttempts(t *testing.T) {
|
||||
db := openInitialTestDatabase(t)
|
||||
applyCredentialAttemptStatsPrerequisites(t, db)
|
||||
|
||||
now := time.Now().UTC().Truncate(time.Hour)
|
||||
firstHour := now.Add(-2 * time.Hour)
|
||||
secondHour := now.Add(-time.Hour)
|
||||
createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000501", firstHour, []attemptMigrationInput{
|
||||
{credentialID: 11, failureCategory: telemetry.FailureCategoryOK},
|
||||
{credentialID: 11, failureCategory: telemetry.FailureCategoryRateLimited},
|
||||
{credentialID: 12, failureCategory: telemetry.FailureCategoryInvalidKey},
|
||||
{credentialID: 13, failureCategory: telemetry.FailureCategoryDownstreamCancel},
|
||||
})
|
||||
createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000502", secondHour, []attemptMigrationInput{
|
||||
{credentialID: 11, failureCategory: telemetry.FailureCategoryOK},
|
||||
})
|
||||
createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000503", now.Add(-26*time.Hour), []attemptMigrationInput{
|
||||
{credentialID: 14, failureCategory: telemetry.FailureCategoryOK},
|
||||
})
|
||||
|
||||
if err := migrations.Up0005(db); err != nil {
|
||||
t.Fatalf("Up0005() error = %v", err)
|
||||
}
|
||||
if err := migrations.Validate0005(db); err != nil {
|
||||
t.Fatalf("Validate0005() error = %v", err)
|
||||
}
|
||||
|
||||
assertColumns(t, db, "credential_attempt_stats", []string{
|
||||
"id", "credential_id", "bucket_start_ms", "success_count", "failure_count",
|
||||
})
|
||||
assertSQLiteIndex(t, db, "credential_attempt_stats", "idx_credential_attempt_stats_identity", true, []string{
|
||||
"credential_id", "bucket_start_ms",
|
||||
})
|
||||
assertSQLiteIndex(t, db, "credential_attempt_stats", "idx_credential_attempt_stats_bucket_id", false, []string{
|
||||
"bucket_start_ms", "id",
|
||||
})
|
||||
assertSQLiteIndex(t, db, "usage_stats", "idx_usage_stats_credential_bucket", false, []string{
|
||||
"credential_id", "bucket_start_ms",
|
||||
})
|
||||
|
||||
type aggregate struct {
|
||||
CredentialID uint
|
||||
BucketStartMS int64
|
||||
SuccessCount int64
|
||||
FailureCount int64
|
||||
}
|
||||
var got []aggregate
|
||||
if err := db.Table("credential_attempt_stats").
|
||||
Order("credential_id ASC").
|
||||
Order("bucket_start_ms ASC").
|
||||
Find(&got).Error; err != nil {
|
||||
t.Fatalf("query credential attempt stats: %v", err)
|
||||
}
|
||||
want := []aggregate{
|
||||
{CredentialID: 11, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 1, FailureCount: 1},
|
||||
{CredentialID: 11, BucketStartMS: secondHour.UnixMilli(), SuccessCount: 1, FailureCount: 0},
|
||||
{CredentialID: 12, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 0, FailureCount: 1},
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("credential attempt stats = %#v, want %#v", got, want)
|
||||
}
|
||||
|
||||
if err := migrations.Up0005(db); err != nil {
|
||||
t.Fatalf("repeat Up0005() error = %v", err)
|
||||
}
|
||||
var repeated []aggregate
|
||||
if err := db.Table("credential_attempt_stats").
|
||||
Order("credential_id ASC").
|
||||
Order("bucket_start_ms ASC").
|
||||
Find(&repeated).Error; err != nil {
|
||||
t.Fatalf("query repeated credential attempt stats: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(repeated, want) {
|
||||
t.Fatalf("repeated credential attempt stats = %#v, want %#v", repeated, want)
|
||||
}
|
||||
}
|
||||
|
||||
type attemptMigrationInput struct {
|
||||
credentialID uint
|
||||
failureCategory telemetry.FailureCategory
|
||||
}
|
||||
|
||||
func applyCredentialAttemptStatsPrerequisites(t *testing.T, db *gorm.DB) {
|
||||
t.Helper()
|
||||
for id, up := range []struct {
|
||||
id string
|
||||
up func(*gorm.DB) error
|
||||
}{
|
||||
{id: migrations.ID0001, up: migrations.Up0001},
|
||||
{id: migrations.ID0002, up: migrations.Up0002},
|
||||
{id: migrations.ID0003, up: migrations.Up0003},
|
||||
{id: migrations.ID0004, up: migrations.Up0004},
|
||||
} {
|
||||
if err := up.up(db); err != nil {
|
||||
t.Fatalf("apply prerequisite %d (%s): %v", id, up.id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func createAttemptMigrationRequest(
|
||||
t *testing.T,
|
||||
db *gorm.DB,
|
||||
requestID string,
|
||||
completedAt time.Time,
|
||||
inputs []attemptMigrationInput,
|
||||
) {
|
||||
t.Helper()
|
||||
request := models.RequestLog{
|
||||
ID: requestID, CompletedAtMS: completedAt.UnixMilli(), AccessKeyID: 1,
|
||||
GroupID: 1, ChannelID: "openai", CredentialID: inputs[len(inputs)-1].credentialID,
|
||||
Protocol: "openai-completions", ClientModel: "migration-model", UpstreamModel: "migration-model",
|
||||
ModelConsistency: string(telemetry.ModelConsistencyMatch), Status: string(telemetry.RequestStatusSuccess),
|
||||
StatusCode: 200, DurationMs: 1, AttemptCount: len(inputs),
|
||||
UsageState: string(usage.StateNotApplicable), CostState: string(pricing.CostStateNotApplicable),
|
||||
PricingCompleteness: string(pricing.CompletenessNotApplicable),
|
||||
}
|
||||
if err := db.Create(&request).Error; err != nil {
|
||||
t.Fatalf("create request log %s: %v", requestID, err)
|
||||
}
|
||||
for index, input := range inputs {
|
||||
attempt := models.RequestLogAttempt{
|
||||
RequestID: requestID, Sequence: index + 1, CompletedAtMS: completedAt.UnixMilli(),
|
||||
GroupID: 1, GroupName: "migration-group", ChannelID: "openai", CredentialID: input.credentialID,
|
||||
StatusCode: 200, DurationMs: 1, FailureCategory: string(input.failureCategory),
|
||||
Action: string(telemetry.ActionTerminate), ErrorSummary: "",
|
||||
}
|
||||
if err := db.Create(&attempt).Error; err != nil {
|
||||
t.Fatalf("create request attempt %s/%d: %v", requestID, index+1, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func assertSQLiteIndex(
|
||||
t *testing.T,
|
||||
db *gorm.DB,
|
||||
table string,
|
||||
indexName string,
|
||||
wantUnique bool,
|
||||
wantColumns []string,
|
||||
) {
|
||||
t.Helper()
|
||||
var indexes []struct {
|
||||
Name string
|
||||
Unique int
|
||||
}
|
||||
if err := db.Raw("PRAGMA index_list('" + table + "')").Scan(&indexes).Error; err != nil {
|
||||
t.Fatalf("inspect %s indexes: %v", table, err)
|
||||
}
|
||||
for _, index := range indexes {
|
||||
if index.Name != indexName {
|
||||
continue
|
||||
}
|
||||
if (index.Unique == 1) != wantUnique {
|
||||
t.Fatalf("%s unique = %d, want %t", indexName, index.Unique, wantUnique)
|
||||
}
|
||||
var columns []struct {
|
||||
Name string
|
||||
Key int
|
||||
}
|
||||
if err := db.Raw("PRAGMA index_xinfo('" + indexName + "')").Scan(&columns).Error; err != nil {
|
||||
t.Fatalf("inspect %s columns: %v", indexName, err)
|
||||
}
|
||||
gotColumns := make([]string, 0, len(wantColumns))
|
||||
for _, column := range columns {
|
||||
if column.Key == 1 {
|
||||
gotColumns = append(gotColumns, column.Name)
|
||||
}
|
||||
}
|
||||
if !reflect.DeepEqual(gotColumns, wantColumns) {
|
||||
t.Fatalf("%s columns = %v, want %v", indexName, gotColumns, wantColumns)
|
||||
}
|
||||
return
|
||||
}
|
||||
t.Fatalf("%s is missing", indexName)
|
||||
}
|
||||
@@ -72,6 +72,17 @@ type RequestLogAttempt struct {
|
||||
RequestLog *RequestLog `gorm:"foreignKey:RequestID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE"`
|
||||
}
|
||||
|
||||
// CredentialAttemptStat is the hourly account-level outcome aggregate used by
|
||||
// subscription account cards. Downstream cancellations are intentionally not
|
||||
// included in either outcome count.
|
||||
type CredentialAttemptStat struct {
|
||||
ID uint `gorm:"primaryKey;autoIncrement;index:idx_credential_attempt_stats_bucket_id,priority:2"`
|
||||
CredentialID uint `gorm:"not null;check:chk_credential_attempt_stat_credential,credential_id > 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:1"`
|
||||
BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_credential_attempt_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:2;index:idx_credential_attempt_stats_bucket_id,priority:1"`
|
||||
SuccessCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_success_count,success_count >= 0"`
|
||||
FailureCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_failure_count,failure_count >= 0"`
|
||||
}
|
||||
|
||||
// UsageAggregationJournal is the request-idempotent input for hourly usage
|
||||
// aggregation. It is staged, applied, and committed in the same transaction as
|
||||
// its RequestLog, and intentionally excludes request and error payloads.
|
||||
@@ -108,11 +119,11 @@ func (UsageAggregationJournal) TableName() string {
|
||||
// credential, and upstream model.
|
||||
type UsageStat struct {
|
||||
ID uint `gorm:"primaryKey;autoIncrement"`
|
||||
BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_usage_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_usage_stats_identity,priority:1"`
|
||||
BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_usage_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_usage_stats_identity,priority:1;index:idx_usage_stats_credential_bucket,priority:2"`
|
||||
AccessKeyID uint `gorm:"not null;uniqueIndex:idx_usage_stats_identity,priority:2"`
|
||||
ChannelID string `gorm:"type:varchar(64);not null;default:'';uniqueIndex:idx_usage_stats_identity,priority:3"`
|
||||
GroupID uint `gorm:"not null;uniqueIndex:idx_usage_stats_identity,priority:4"`
|
||||
CredentialID uint `gorm:"not null;default:0;uniqueIndex:idx_usage_stats_identity,priority:5"`
|
||||
CredentialID uint `gorm:"not null;default:0;uniqueIndex:idx_usage_stats_identity,priority:5;index:idx_usage_stats_credential_bucket,priority:1"`
|
||||
Model string `gorm:"type:varchar(255);not null;uniqueIndex:idx_usage_stats_identity,priority:6"`
|
||||
RequestCount int64 `gorm:"not null;default:0;check:chk_usage_stat_request_count,request_count >= 0;check:chk_usage_stat_request_outcome,request_count = success_count + failure_count"`
|
||||
SuccessCount int64 `gorm:"not null;default:0;check:chk_usage_stat_success_count,success_count >= 0"`
|
||||
|
||||
@@ -54,6 +54,7 @@ const (
|
||||
|
||||
type Attempt struct {
|
||||
Sequence int
|
||||
CompletedAt time.Time
|
||||
GroupID uint
|
||||
GroupName string
|
||||
ChannelID channel.ID
|
||||
|
||||
@@ -40,6 +40,7 @@ func TestRequestTelemetryContractUsesExactFieldAllowlist(t *testing.T) {
|
||||
},
|
||||
reflect.TypeOf(Attempt{}): {
|
||||
"Sequence",
|
||||
"CompletedAt",
|
||||
"GroupID",
|
||||
"GroupName",
|
||||
"ChannelID",
|
||||
|
||||
@@ -219,19 +219,25 @@ export interface CredentialItemDto {
|
||||
last_failure_category: FailureCategory
|
||||
last_status_code: number | null
|
||||
cooldown_until_ms: number | null
|
||||
last_used_at_ms?: number
|
||||
daily_usage?: CredentialDailyUsageDto
|
||||
recovery: CredentialRecoveryDto
|
||||
}
|
||||
|
||||
/** 固定 24 小时窗口的请求结果分布,来源是请求日志而非 health 的 5 分钟内存窗口。 */
|
||||
/** 固定 24 小时窗口的上游尝试结果分布,来源是小时聚合而非 health 的 5 分钟内存窗口。 */
|
||||
export interface CredentialDailyUsageDto {
|
||||
window_seconds: number
|
||||
success_count: number
|
||||
failure_count: number
|
||||
/** false 表示请求日志留存期短于该窗口,计数偏低。 */
|
||||
/** false 表示统计数据未覆盖完整窗口,计数可能偏低。 */
|
||||
data_complete: boolean
|
||||
}
|
||||
|
||||
export interface CredentialDetailDto {
|
||||
credential: CredentialItemDto
|
||||
observation: CredentialObservationDto
|
||||
}
|
||||
|
||||
export interface CredentialRevealDto {
|
||||
credential_id: number
|
||||
credential: Record<string, string>
|
||||
|
||||
@@ -13,6 +13,7 @@ import type {
|
||||
CredentialCollectionFilters,
|
||||
CredentialConfiguredStatus,
|
||||
CredentialDailyUsageDto,
|
||||
CredentialDetailDto,
|
||||
CredentialItemDto,
|
||||
CredentialObservationDto,
|
||||
CredentialObservationSnapshotDto,
|
||||
@@ -47,6 +48,7 @@ export type {
|
||||
CredentialCollectionFilters,
|
||||
CredentialConfiguredStatus,
|
||||
CredentialDailyUsageDto,
|
||||
CredentialDetailDto,
|
||||
CredentialItemDto,
|
||||
CredentialRecoveryDto,
|
||||
CredentialRevealDto,
|
||||
@@ -98,9 +100,11 @@ const credentialItemFields = [
|
||||
'last_failure_category',
|
||||
'last_status_code',
|
||||
'cooldown_until_ms',
|
||||
'last_used_at_ms',
|
||||
'daily_usage',
|
||||
'recovery',
|
||||
] as const
|
||||
const credentialDetailFields = ['credential', 'observation'] as const
|
||||
const credentialDailyUsageFields = [
|
||||
'window_seconds',
|
||||
'success_count',
|
||||
@@ -450,6 +454,9 @@ export function projectCredentialItem(value: unknown): CredentialItemDto {
|
||||
? null
|
||||
: projectSafeInteger(record.last_status_code, { minimum: 100, maximum: 999 }),
|
||||
cooldown_until_ms: cooldownUntil,
|
||||
...(record.last_used_at_ms === undefined
|
||||
? {}
|
||||
: { last_used_at_ms: projectEpochMilliseconds(record.last_used_at_ms) }),
|
||||
...(record.daily_usage === undefined
|
||||
? {}
|
||||
: { daily_usage: projectDailyUsage(record.daily_usage) }),
|
||||
@@ -468,6 +475,15 @@ function projectDailyUsage(value: unknown): CredentialDailyUsageDto {
|
||||
}
|
||||
}
|
||||
|
||||
export function projectCredentialDetail(value: unknown): CredentialDetailDto {
|
||||
const record = projectRecord(value)
|
||||
assertNoSecretLikeFields(record, credentialDetailFields)
|
||||
return {
|
||||
credential: projectCredentialItem(record.credential),
|
||||
observation: projectObservation(record.observation),
|
||||
}
|
||||
}
|
||||
|
||||
function projectPagination(value: unknown): CredentialCollectionDto['pagination'] {
|
||||
const record = projectRecord(value)
|
||||
assertNoSecretLikeFields(record, credentialPaginationFields)
|
||||
@@ -570,6 +586,22 @@ export async function getCredentialCollection(
|
||||
return result
|
||||
}
|
||||
|
||||
export async function getCredentialDetail(
|
||||
client: ApiClient,
|
||||
groupId: number,
|
||||
credentialId: number,
|
||||
signal?: AbortSignal,
|
||||
): Promise<CredentialDetailDto> {
|
||||
const result = projectCredentialDetail(
|
||||
await client.request(`/api/groups/${groupId}/credentials/${credentialId}`, {
|
||||
method: 'GET',
|
||||
signal,
|
||||
}),
|
||||
)
|
||||
if (result.credential.credential_id !== credentialId) invalidResponse()
|
||||
return result
|
||||
}
|
||||
|
||||
const manualGroupQueryOptions = {
|
||||
refetchOnWindowFocus: false,
|
||||
refetchOnReconnect: false,
|
||||
|
||||
@@ -22,6 +22,7 @@ import {
|
||||
cacheCredentialItem,
|
||||
consumeCredentialResetCredit,
|
||||
credentialCollectionQueryOptions,
|
||||
getCredentialDetail,
|
||||
revealCredential,
|
||||
restoreCredential,
|
||||
refreshCredentialObservation,
|
||||
@@ -347,6 +348,52 @@ function cachedCurrentSummary(): CredentialSummaryDto | undefined {
|
||||
)?.summary
|
||||
}
|
||||
|
||||
function cachedCurrentCredential(id: number): CredentialItemDto | undefined {
|
||||
return queryClient
|
||||
.getQueryData<CredentialCollectionDto>(
|
||||
controlQueryKeys.groups.credentials(props.groupId, filters.value),
|
||||
)
|
||||
?.items.find(({ credential_id }) => credential_id === id)
|
||||
}
|
||||
|
||||
function withPreservedObservedUsage(
|
||||
item: CredentialItemDto,
|
||||
sourceObservation: CredentialItemDto['observation'],
|
||||
): CredentialItemDto {
|
||||
const targetObservation = item.observation
|
||||
if (
|
||||
!targetObservation?.snapshot ||
|
||||
!sourceObservation?.snapshot ||
|
||||
targetObservation.observation_version !== sourceObservation.observation_version ||
|
||||
targetObservation.observed_at_ms !== sourceObservation.observed_at_ms
|
||||
) {
|
||||
return item
|
||||
}
|
||||
|
||||
const observedUsageByWindow = new Map(
|
||||
sourceObservation.snapshot.quota_windows.flatMap((window) =>
|
||||
window.observed_usage === undefined ? [] : [[window.id, window.observed_usage] as const],
|
||||
),
|
||||
)
|
||||
if (observedUsageByWindow.size === 0) return item
|
||||
|
||||
return {
|
||||
...item,
|
||||
observation: {
|
||||
...targetObservation,
|
||||
snapshot: {
|
||||
...targetObservation.snapshot,
|
||||
quota_windows: targetObservation.snapshot.quota_windows.map((window) => {
|
||||
const observedUsage = observedUsageByWindow.get(window.id)
|
||||
return window.observed_usage !== undefined || observedUsage === undefined
|
||||
? window
|
||||
: { ...window, observed_usage: observedUsage }
|
||||
}),
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function synchronizeGroupSummary(summary: CredentialSummaryDto | undefined): void {
|
||||
const queryKey = controlQueryKeys.groups.summary(props.groupId)
|
||||
if (summary === undefined) {
|
||||
@@ -400,8 +447,21 @@ async function refetchActiveCredentialPage(): Promise<void> {
|
||||
|
||||
async function reconcileItem(result: CredentialItemDto, refetchActive: boolean): Promise<void> {
|
||||
try {
|
||||
await cacheCredentialItem(queryClient, props.groupId, result)
|
||||
if (refetchActive) await refetchActiveCredentialPage()
|
||||
const current = cachedCurrentCredential(result.credential_id)
|
||||
const reconciled =
|
||||
current === undefined ? result : withPreservedObservedUsage(result, current.observation)
|
||||
await cacheCredentialItem(queryClient, props.groupId, reconciled)
|
||||
if (refetchActive) {
|
||||
await refetchActiveCredentialPage()
|
||||
const refreshed = cachedCurrentCredential(result.credential_id)
|
||||
if (refreshed !== undefined) {
|
||||
await cacheCredentialItem(
|
||||
queryClient,
|
||||
props.groupId,
|
||||
withPreservedObservedUsage(refreshed, reconciled.observation),
|
||||
)
|
||||
}
|
||||
}
|
||||
const summary = cachedCurrentSummary()
|
||||
if (summary === undefined) {
|
||||
await invalidateReconciliationQueries()
|
||||
@@ -424,8 +484,7 @@ async function refreshObservation(item: CredentialItemDto): Promise<void> {
|
||||
props.groupId,
|
||||
item.credential_id,
|
||||
)
|
||||
const next = { ...item, observation }
|
||||
await reconcileItem(next, false)
|
||||
await reconcileItem({ ...item, observation }, true)
|
||||
} catch (cause) {
|
||||
feedback.value = t(
|
||||
presentSubscriptionErrorKey(cause, 'group.credentials.subscription.syncFailed'),
|
||||
@@ -435,6 +494,19 @@ async function refreshObservation(item: CredentialItemDto): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
async function loadCredentialUsage(item: CredentialItemDto): Promise<void> {
|
||||
if (pending(item.credential_id)) return
|
||||
setPending(item.credential_id, 'usage', true)
|
||||
try {
|
||||
const detail = await getCredentialDetail(client, props.groupId, item.credential_id)
|
||||
await cacheCredentialItem(queryClient, props.groupId, detail.credential)
|
||||
} catch {
|
||||
feedback.value = t('group.credentials.loadFailed')
|
||||
} finally {
|
||||
setPending(item.credential_id, 'usage', false)
|
||||
}
|
||||
}
|
||||
|
||||
function openResetCreditDialog(item: CredentialItemDto): void {
|
||||
if (pending(item.credential_id)) return
|
||||
const idempotencyKey = resetOperationKeys.get(item.credential_id) ?? crypto.randomUUID()
|
||||
@@ -944,6 +1016,7 @@ async function runBatch(
|
||||
@toggle="mutateItem($event, 'toggle')"
|
||||
@restore="mutateItem($event, 'restore')"
|
||||
@refresh="refreshObservation"
|
||||
@load-usage="loadCredentialUsage"
|
||||
@reset="openResetCreditDialog"
|
||||
@reauthorize="openConnectionWorkspace"
|
||||
@remove="deleteTarget = { ids: [$event.credential_id], mask: $event.mask }"
|
||||
|
||||
@@ -22,6 +22,7 @@ const emit = defineEmits<{
|
||||
toggle: [item: CredentialItemDto]
|
||||
restore: [item: CredentialItemDto]
|
||||
refresh: [item: CredentialItemDto]
|
||||
'load-usage': [item: CredentialItemDto]
|
||||
reset: [item: CredentialItemDto]
|
||||
reauthorize: [item: CredentialItemDto]
|
||||
remove: [item: CredentialItemDto]
|
||||
@@ -54,12 +55,7 @@ const quotaWindows = computed(() =>
|
||||
const accountQuotaWindows = computed(() =>
|
||||
quotaWindows.value.filter((window) => window.scope === 'account'),
|
||||
)
|
||||
const lastUsedAtMS = computed(() => {
|
||||
const values = accountQuotaWindows.value
|
||||
.map((window) => window.observed_usage?.last_used_at_ms)
|
||||
.filter((value): value is number => value !== undefined)
|
||||
return values.length ? Math.max(...values) : undefined
|
||||
})
|
||||
const lastUsedAtMS = computed(() => props.item.last_used_at_ms)
|
||||
const constrainedModels = computed(() =>
|
||||
Array.from(new Set(quotaWindows.value.flatMap((window) => window.model_ids ?? []))),
|
||||
)
|
||||
@@ -152,11 +148,11 @@ const authIssue = computed(() => {
|
||||
const key = props.item.auth_error_code ? authErrorKeys[props.item.auth_error_code] : undefined
|
||||
return key ? t(key) : t(`group.credentials.subscription.auth.${props.item.auth_state}`)
|
||||
})
|
||||
// 卡片给的是请求日志里的 24 小时结果分布,不是 health 那个 5 分钟内存窗口
|
||||
// 卡片给的是按账号上游尝试聚合的 24 小时结果分布,不是 health 那个 5 分钟内存窗口
|
||||
//(后者重启即清零,是给调度判定用的)。始终并排给出成功与失败:失败 0 本身是
|
||||
// 信息,也让这一行宽度稳定。
|
||||
const dailyUsage = computed(() => props.item.daily_usage)
|
||||
// 请求日志留存短于 24 小时时计数会偏低,只用 title 提示,不占版面。
|
||||
// 小时统计未覆盖完整 24 小时时计数会偏低,只用 title 提示,不占版面。
|
||||
const dailyIncompleteHint = computed(() =>
|
||||
dailyUsage.value && !dailyUsage.value.data_complete
|
||||
? t('group.credentials.subscription.dailyIncomplete')
|
||||
@@ -179,6 +175,65 @@ function observationErrorLabel(code: string | undefined): string {
|
||||
return key ? t(key) : code
|
||||
}
|
||||
|
||||
function windowUsageCanLoad(window: CredentialQuotaWindowDto): boolean {
|
||||
if (!observationIsCurrent.value) return false
|
||||
const resetAtMS = window.reset_at_ms
|
||||
const windowSeconds = window.window_seconds
|
||||
if (
|
||||
resetAtMS === undefined ||
|
||||
windowSeconds === undefined ||
|
||||
windowSeconds <= 0 ||
|
||||
resetAtMS <= nowMs.value
|
||||
) {
|
||||
return false
|
||||
}
|
||||
const windowMS = windowSeconds * 1_000
|
||||
return (
|
||||
Number.isSafeInteger(windowMS) && windowMS <= resetAtMS && resetAtMS - windowMS < nowMs.value
|
||||
)
|
||||
}
|
||||
|
||||
function loadUsageOnOpen(event: Event): void {
|
||||
const details = event.currentTarget as HTMLDetailsElement
|
||||
if (
|
||||
details.open &&
|
||||
accountQuotaWindows.value.some(
|
||||
(window) => windowUsageCanLoad(window) && window.observed_usage === undefined,
|
||||
)
|
||||
) {
|
||||
emit('load-usage', props.item)
|
||||
}
|
||||
}
|
||||
|
||||
function estimateTitles(...keys: string[]): string | undefined {
|
||||
const titles = keys.filter(Boolean).map((key) => t(key))
|
||||
return titles.length === 0 ? undefined : titles.join(' · ')
|
||||
}
|
||||
|
||||
function requestCountTitle(window: CredentialQuotaWindowDto): string | undefined {
|
||||
return window.observed_usage?.data_complete === false
|
||||
? t('group.credentials.subscription.estimate.dataIncomplete')
|
||||
: undefined
|
||||
}
|
||||
|
||||
function tokenCountTitle(window: CredentialQuotaWindowDto): string | undefined {
|
||||
const observed = window.observed_usage
|
||||
if (!observed) return undefined
|
||||
return estimateTitles(
|
||||
observed.data_complete ? '' : 'group.credentials.subscription.estimate.dataIncomplete',
|
||||
observed.usage_complete ? '' : 'group.credentials.subscription.estimate.usageIncomplete',
|
||||
)
|
||||
}
|
||||
|
||||
function referenceCostTitle(window: CredentialQuotaWindowDto): string | undefined {
|
||||
const observed = window.observed_usage
|
||||
if (!observed) return undefined
|
||||
return estimateTitles(
|
||||
observed.data_complete ? '' : 'group.credentials.subscription.estimate.dataIncomplete',
|
||||
observed.pricing_complete ? '' : 'group.credentials.subscription.estimate.pricingIncomplete',
|
||||
)
|
||||
}
|
||||
|
||||
function remainingPercent(window: CredentialQuotaWindowDto): number | undefined {
|
||||
if (!quotaWindowIsCurrent(window)) return undefined
|
||||
if (window.utilization !== undefined) return Math.round((1 - window.utilization) * 100)
|
||||
@@ -390,10 +445,10 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'):
|
||||
</AppButton>
|
||||
</div>
|
||||
|
||||
<!-- 一行读完「最近跑得怎么样」:计数窗口与 API Key 列表的「近 5 分钟」列同源。
|
||||
<!-- 一行读完「最近跑得怎么样」:最近使用取精确的账号活动时间,结果计数取小时聚合。
|
||||
访问凭据到期这类低频信息下沉到诊断信息,不占这一行。 -->
|
||||
<div class="subscription-account__meta">
|
||||
<span v-if="lastUsedAtMS" class="subscription-account__meta-last-used">
|
||||
<span v-if="lastUsedAtMS !== undefined" class="subscription-account__meta-last-used">
|
||||
{{ t('group.credentials.subscription.lastUsed') }}
|
||||
<AppRelativeTime
|
||||
:instant="lastUsedAtMS"
|
||||
@@ -433,7 +488,11 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'):
|
||||
</span>
|
||||
</div>
|
||||
|
||||
<details v-if="accountQuotaWindows.length" class="subscription-account__estimate">
|
||||
<details
|
||||
v-if="accountQuotaWindows.length"
|
||||
class="subscription-account__estimate"
|
||||
@toggle="loadUsageOnOpen"
|
||||
>
|
||||
<summary>
|
||||
{{ t('group.credentials.subscription.estimate.title') }} ·
|
||||
{{
|
||||
@@ -449,15 +508,15 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'):
|
||||
<span v-if="usedPercentLabel(window)">{{ usedPercentLabel(window) }}</span>
|
||||
</div>
|
||||
<dl v-if="window.observed_usage" class="subscription-account__estimate-rows">
|
||||
<div>
|
||||
<div :title="requestCountTitle(window)">
|
||||
<dt>{{ t('group.credentials.subscription.estimate.requests') }}</dt>
|
||||
<dd>{{ n(window.observed_usage.request_count) }}</dd>
|
||||
</div>
|
||||
<div>
|
||||
<div :title="tokenCountTitle(window)">
|
||||
<dt>{{ t('group.credentials.subscription.estimate.tokens') }}</dt>
|
||||
<dd>{{ formatTokens(window.observed_usage.total_tokens, locale) }}</dd>
|
||||
</div>
|
||||
<div>
|
||||
<div :title="referenceCostTitle(window)">
|
||||
<dt>{{ t('group.credentials.subscription.estimate.referenceCost') }}</dt>
|
||||
<dd class="subscription-account__estimate-money">
|
||||
{{
|
||||
@@ -469,6 +528,9 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'):
|
||||
</dd>
|
||||
</div>
|
||||
</dl>
|
||||
<p v-else class="subscription-account__estimate-unavailable">
|
||||
{{ t('group.credentials.subscription.estimate.unavailable') }}
|
||||
</p>
|
||||
</article>
|
||||
</div>
|
||||
</details>
|
||||
@@ -732,6 +794,11 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'):
|
||||
color: var(--color-text-muted);
|
||||
font-weight: 500;
|
||||
}
|
||||
.subscription-account__estimate-unavailable {
|
||||
margin: 0;
|
||||
color: var(--color-text-faint);
|
||||
font-size: var(--text-label-xs);
|
||||
}
|
||||
.subscription-account__credits {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
|
||||
@@ -381,8 +381,7 @@ export default {
|
||||
dailyWindow: 'Last 24 hours',
|
||||
dailySuccess: 'Success',
|
||||
dailyFailure: 'Failure',
|
||||
dailyIncomplete:
|
||||
'Request logs are retained for less than 24 hours, so the counts may be low',
|
||||
dailyIncomplete: 'Statistics do not cover the full 24 hours, so the counts may be low',
|
||||
syncFailed: 'Unable to sync account information',
|
||||
moreActions: 'More actions',
|
||||
autoRenews: 'renews when used',
|
||||
@@ -417,6 +416,11 @@ export default {
|
||||
requests: 'Requests',
|
||||
tokens: 'Tokens',
|
||||
referenceCost: 'API reference cost',
|
||||
unavailable: 'Local statistics are not currently available',
|
||||
dataIncomplete: 'Statistics do not cover the full window, so the value may be low',
|
||||
usageIncomplete: 'Some requests lack complete token usage, so the value may be low',
|
||||
pricingIncomplete:
|
||||
'Some requests lack complete pricing, so the reference cost may be low',
|
||||
windowSeparator: ', ',
|
||||
},
|
||||
auth: {
|
||||
|
||||
@@ -384,7 +384,7 @@ export default {
|
||||
dailySuccess: '成功',
|
||||
dailyFailure: '失敗',
|
||||
dailyIncomplete:
|
||||
'リクエストログの保持期間が 24 時間未満のため、件数が少なく出る場合があります',
|
||||
'統計データが 24 時間全体をカバーしていないため、件数が少なく出る場合があります',
|
||||
syncFailed: 'アカウント情報を同期できません',
|
||||
moreActions: 'その他の操作',
|
||||
autoRenews: '使用時に自動更新',
|
||||
@@ -419,6 +419,12 @@ export default {
|
||||
requests: 'リクエスト',
|
||||
tokens: 'トークン',
|
||||
referenceCost: 'API 参考コスト',
|
||||
unavailable: '利用可能なローカル統計は現在ありません',
|
||||
dataIncomplete: '統計データが期間全体をカバーしていないため、値が少ない場合があります',
|
||||
usageIncomplete:
|
||||
'一部のリクエストでトークン使用量が不完全なため、値が少ない場合があります',
|
||||
pricingIncomplete:
|
||||
'一部のリクエストで価格情報が不完全なため、参考コストが少ない場合があります',
|
||||
windowSeparator: '、',
|
||||
},
|
||||
auth: {
|
||||
|
||||
@@ -370,7 +370,7 @@ export default {
|
||||
dailyWindow: '近 24 小时',
|
||||
dailySuccess: '成功',
|
||||
dailyFailure: '失败',
|
||||
dailyIncomplete: '请求日志留存期短于 24 小时,计数可能偏低',
|
||||
dailyIncomplete: '统计数据未覆盖完整 24 小时,计数可能偏低',
|
||||
syncFailed: '无法同步账号信息',
|
||||
moreActions: '更多操作',
|
||||
autoRenews: '使用时自动续期',
|
||||
@@ -404,6 +404,10 @@ export default {
|
||||
requests: '请求',
|
||||
tokens: 'Token',
|
||||
referenceCost: 'API 参考成本',
|
||||
unavailable: '暂时没有可用的本地统计',
|
||||
dataIncomplete: '统计数据未覆盖完整窗口,数值可能偏低',
|
||||
usageIncomplete: '部分请求没有完整 Token 用量,数值可能偏低',
|
||||
pricingIncomplete: '部分请求没有完整价格,参考成本可能偏低',
|
||||
windowSeparator: '、',
|
||||
},
|
||||
auth: {
|
||||
|
||||
Reference in new issue
Block a user