mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
refactor(repository): 纯移动拆分 usage_log_repo.go（4701→212行）
This commit is contained in:
shaw
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,628 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
)
|
||||
|
||||
// getPerformanceStats 获取 RPM 和 TPM（近5分钟平均值，可选按用户过滤）
|
||||
func (r *usageLogRepository) getPerformanceStats(ctx context.Context, userID int64) (rpm, tpm int64, err error) {
|
||||
fiveMinutesAgo := time.Now().Add(-5 * time.Minute)
|
||||
query := `
|
||||
SELECT
|
||||
COUNT(*) as request_count,
|
||||
COALESCE(SUM(input_tokens + output_tokens), 0) as token_count
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1`
|
||||
args := []any{fiveMinutesAgo}
|
||||
if userID > 0 {
|
||||
query += " AND user_id = $2"
|
||||
args = append(args, userID)
|
||||
}
|
||||
|
||||
var requestCount int64
|
||||
var tokenCount int64
|
||||
if err := scanSingleRow(ctx, r.sql, query, args, &requestCount, &tokenCount); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
return requestCount / 5, tokenCount / 5, nil
|
||||
}
|
||||
|
||||
// UserStats 用户使用统计
|
||||
type UserStats struct {
|
||||
TotalRequests int64 `json:"total_requests"`
|
||||
TotalTokens int64 `json:"total_tokens"`
|
||||
TotalCost float64 `json:"total_cost"`
|
||||
InputTokens int64 `json:"input_tokens"`
|
||||
OutputTokens int64 `json:"output_tokens"`
|
||||
CacheReadTokens int64 `json:"cache_read_tokens"`
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) GetUserStats(ctx context.Context, userID int64, startTime, endTime time.Time) (*UserStats, error) {
|
||||
query := `
|
||||
SELECT
|
||||
COUNT(*) as total_requests,
|
||||
COALESCE(SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(actual_cost), 0) as total_cost,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as cache_read_tokens
|
||||
FROM usage_logs
|
||||
WHERE user_id = $1 AND created_at >= $2 AND created_at < $3
|
||||
`
|
||||
|
||||
stats := &UserStats{}
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
query,
|
||||
[]any{userID, startTime, endTime},
|
||||
&stats.TotalRequests,
|
||||
&stats.TotalTokens,
|
||||
&stats.TotalCost,
|
||||
&stats.InputTokens,
|
||||
&stats.OutputTokens,
|
||||
&stats.CacheReadTokens,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// DashboardStats 仪表盘统计
|
||||
type DashboardStats = usagestats.DashboardStats
|
||||
|
||||
func (r *usageLogRepository) GetDashboardStats(ctx context.Context) (*DashboardStats, error) {
|
||||
stats := &DashboardStats{}
|
||||
now := timezone.Now()
|
||||
todayStart := timezone.Today()
|
||||
|
||||
if err := r.fillDashboardEntityStats(ctx, stats, todayStart, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := r.fillDashboardUsageStatsAggregated(ctx, stats, todayStart, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rpm, tpm, err := r.getPerformanceStats(ctx, 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.Rpm = rpm
|
||||
stats.Tpm = tpm
|
||||
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) GetDashboardStatsWithRange(ctx context.Context, start, end time.Time) (*DashboardStats, error) {
|
||||
startUTC := start.UTC()
|
||||
endUTC := end.UTC()
|
||||
if !endUTC.After(startUTC) {
|
||||
return nil, errors.New("统计时间范围无效")
|
||||
}
|
||||
|
||||
stats := &DashboardStats{}
|
||||
now := timezone.Now()
|
||||
todayStart := timezone.Today()
|
||||
|
||||
if err := r.fillDashboardEntityStats(ctx, stats, todayStart, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := r.fillDashboardUsageStatsFromUsageLogs(ctx, stats, startUTC, endUTC, todayStart, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rpm, tpm, err := r.getPerformanceStats(ctx, 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.Rpm = rpm
|
||||
stats.Tpm = tpm
|
||||
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) fillDashboardEntityStats(ctx context.Context, stats *DashboardStats, todayUTC, now time.Time) error {
|
||||
userStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as total_users,
|
||||
COUNT(CASE WHEN created_at >= $1 THEN 1 END) as today_new_users
|
||||
FROM users
|
||||
WHERE deleted_at IS NULL
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
userStatsQuery,
|
||||
[]any{todayUTC},
|
||||
&stats.TotalUsers,
|
||||
&stats.TodayNewUsers,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
apiKeyStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as total_api_keys,
|
||||
COUNT(CASE WHEN status = $1 THEN 1 END) as active_api_keys
|
||||
FROM api_keys
|
||||
WHERE deleted_at IS NULL
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
apiKeyStatsQuery,
|
||||
[]any{service.StatusActive},
|
||||
&stats.TotalAPIKeys,
|
||||
&stats.ActiveAPIKeys,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
accountStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as total_accounts,
|
||||
COUNT(CASE WHEN status = $1 AND schedulable = true THEN 1 END) as normal_accounts,
|
||||
COUNT(CASE WHEN status = $2 THEN 1 END) as error_accounts,
|
||||
COUNT(CASE WHEN rate_limited_at IS NOT NULL AND rate_limit_reset_at > $3 THEN 1 END) as ratelimit_accounts,
|
||||
COUNT(CASE WHEN overload_until IS NOT NULL AND overload_until > $4 THEN 1 END) as overload_accounts
|
||||
FROM accounts
|
||||
WHERE deleted_at IS NULL
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
accountStatsQuery,
|
||||
[]any{service.StatusActive, service.StatusError, now, now},
|
||||
&stats.TotalAccounts,
|
||||
&stats.NormalAccounts,
|
||||
&stats.ErrorAccounts,
|
||||
&stats.RateLimitAccounts,
|
||||
&stats.OverloadAccounts,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) fillDashboardUsageStatsAggregated(ctx context.Context, stats *DashboardStats, todayUTC, now time.Time) error {
|
||||
totalStatsQuery := `
|
||||
SELECT
|
||||
COALESCE(SUM(total_requests), 0) as total_requests,
|
||||
COALESCE(SUM(input_tokens), 0) as total_input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as total_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as total_cost,
|
||||
COALESCE(SUM(actual_cost), 0) as total_actual_cost,
|
||||
COALESCE(SUM(account_cost), 0) as total_account_cost,
|
||||
COALESCE(SUM(total_duration_ms), 0) as total_duration_ms
|
||||
FROM usage_dashboard_daily
|
||||
`
|
||||
var totalDurationMs int64
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
totalStatsQuery,
|
||||
nil,
|
||||
&stats.TotalRequests,
|
||||
&stats.TotalInputTokens,
|
||||
&stats.TotalOutputTokens,
|
||||
&stats.TotalCacheCreationTokens,
|
||||
&stats.TotalCacheReadTokens,
|
||||
&stats.TotalCost,
|
||||
&stats.TotalActualCost,
|
||||
&stats.TotalAccountCost,
|
||||
&totalDurationMs,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheCreationTokens + stats.TotalCacheReadTokens
|
||||
if stats.TotalRequests > 0 {
|
||||
stats.AverageDurationMs = float64(totalDurationMs) / float64(stats.TotalRequests)
|
||||
}
|
||||
|
||||
todayStatsQuery := `
|
||||
SELECT
|
||||
total_requests as today_requests,
|
||||
input_tokens as today_input_tokens,
|
||||
output_tokens as today_output_tokens,
|
||||
cache_creation_tokens as today_cache_creation_tokens,
|
||||
cache_read_tokens as today_cache_read_tokens,
|
||||
total_cost as today_cost,
|
||||
actual_cost as today_actual_cost,
|
||||
account_cost as today_account_cost,
|
||||
active_users as active_users
|
||||
FROM usage_dashboard_daily
|
||||
WHERE bucket_date = $1::date
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
todayStatsQuery,
|
||||
[]any{todayUTC},
|
||||
&stats.TodayRequests,
|
||||
&stats.TodayInputTokens,
|
||||
&stats.TodayOutputTokens,
|
||||
&stats.TodayCacheCreationTokens,
|
||||
&stats.TodayCacheReadTokens,
|
||||
&stats.TodayCost,
|
||||
&stats.TodayActualCost,
|
||||
&stats.TodayAccountCost,
|
||||
&stats.ActiveUsers,
|
||||
); err != nil {
|
||||
if err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
}
|
||||
stats.TodayTokens = stats.TodayInputTokens + stats.TodayOutputTokens + stats.TodayCacheCreationTokens + stats.TodayCacheReadTokens
|
||||
|
||||
hourlyActiveQuery := `
|
||||
SELECT active_users
|
||||
FROM usage_dashboard_hourly
|
||||
WHERE bucket_start = $1
|
||||
`
|
||||
hourStart := now.In(timezone.Location()).Truncate(time.Hour)
|
||||
if err := scanSingleRow(ctx, r.sql, hourlyActiveQuery, []any{hourStart}, &stats.HourlyActiveUsers); err != nil {
|
||||
if err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) fillDashboardUsageStatsFromUsageLogs(ctx context.Context, stats *DashboardStats, startUTC, endUTC, todayUTC, now time.Time) error {
|
||||
todayEnd := todayUTC.Add(24 * time.Hour)
|
||||
combinedStatsQuery := `
|
||||
WITH scoped AS (
|
||||
SELECT
|
||||
created_at,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
cache_creation_tokens,
|
||||
cache_read_tokens,
|
||||
total_cost,
|
||||
actual_cost,
|
||||
COALESCE(account_stats_cost, total_cost) * COALESCE(account_rate_multiplier, 1) AS account_cost,
|
||||
COALESCE(duration_ms, 0) AS duration_ms
|
||||
FROM usage_logs
|
||||
WHERE created_at >= LEAST($1::timestamptz, $3::timestamptz)
|
||||
AND created_at < GREATEST($2::timestamptz, $4::timestamptz)
|
||||
)
|
||||
SELECT
|
||||
COUNT(*) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz) AS total_requests,
|
||||
COALESCE(SUM(input_tokens) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_input_tokens,
|
||||
COALESCE(SUM(output_tokens) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_cost,
|
||||
COALESCE(SUM(actual_cost) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_actual_cost,
|
||||
COALESCE(SUM(account_cost) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_account_cost,
|
||||
COALESCE(SUM(duration_ms) FILTER (WHERE created_at >= $1::timestamptz AND created_at < $2::timestamptz), 0) AS total_duration_ms,
|
||||
COUNT(*) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz) AS today_requests,
|
||||
COALESCE(SUM(input_tokens) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_input_tokens,
|
||||
COALESCE(SUM(output_tokens) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_cost,
|
||||
COALESCE(SUM(actual_cost) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_actual_cost,
|
||||
COALESCE(SUM(account_cost) FILTER (WHERE created_at >= $3::timestamptz AND created_at < $4::timestamptz), 0) AS today_account_cost
|
||||
FROM scoped
|
||||
`
|
||||
var totalDurationMs int64
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
combinedStatsQuery,
|
||||
[]any{startUTC, endUTC, todayUTC, todayEnd},
|
||||
&stats.TotalRequests,
|
||||
&stats.TotalInputTokens,
|
||||
&stats.TotalOutputTokens,
|
||||
&stats.TotalCacheCreationTokens,
|
||||
&stats.TotalCacheReadTokens,
|
||||
&stats.TotalCost,
|
||||
&stats.TotalActualCost,
|
||||
&stats.TotalAccountCost,
|
||||
&totalDurationMs,
|
||||
&stats.TodayRequests,
|
||||
&stats.TodayInputTokens,
|
||||
&stats.TodayOutputTokens,
|
||||
&stats.TodayCacheCreationTokens,
|
||||
&stats.TodayCacheReadTokens,
|
||||
&stats.TodayCost,
|
||||
&stats.TodayActualCost,
|
||||
&stats.TodayAccountCost,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheCreationTokens + stats.TotalCacheReadTokens
|
||||
if stats.TotalRequests > 0 {
|
||||
stats.AverageDurationMs = float64(totalDurationMs) / float64(stats.TotalRequests)
|
||||
}
|
||||
|
||||
stats.TodayTokens = stats.TodayInputTokens + stats.TodayOutputTokens + stats.TodayCacheCreationTokens + stats.TodayCacheReadTokens
|
||||
|
||||
hourStart := now.UTC().Truncate(time.Hour)
|
||||
hourEnd := hourStart.Add(time.Hour)
|
||||
activeUsersQuery := `
|
||||
WITH scoped AS (
|
||||
SELECT user_id, created_at
|
||||
FROM usage_logs
|
||||
WHERE created_at >= LEAST($1::timestamptz, $3::timestamptz)
|
||||
AND created_at < GREATEST($2::timestamptz, $4::timestamptz)
|
||||
)
|
||||
SELECT
|
||||
COUNT(DISTINCT CASE WHEN created_at >= $1::timestamptz AND created_at < $2::timestamptz THEN user_id END) AS active_users,
|
||||
COUNT(DISTINCT CASE WHEN created_at >= $3::timestamptz AND created_at < $4::timestamptz THEN user_id END) AS hourly_active_users
|
||||
FROM scoped
|
||||
`
|
||||
if err := scanSingleRow(ctx, r.sql, activeUsersQuery, []any{todayUTC, todayEnd, hourStart, hourEnd}, &stats.ActiveUsers, &stats.HourlyActiveUsers); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// UserDashboardStats 用户仪表盘统计
|
||||
type UserDashboardStats = usagestats.UserDashboardStats
|
||||
|
||||
// PlatformDashboardStats 单平台用量明细
|
||||
type PlatformDashboardStats = usagestats.PlatformDashboardStats
|
||||
|
||||
// GetUserDashboardStats 获取用户专属的仪表盘统计
|
||||
func (r *usageLogRepository) GetUserDashboardStats(ctx context.Context, userID int64) (*UserDashboardStats, error) {
|
||||
stats := &UserDashboardStats{}
|
||||
today := timezone.Today()
|
||||
|
||||
// API Key 统计
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
"SELECT COUNT(*) FROM api_keys WHERE user_id = $1 AND deleted_at IS NULL",
|
||||
[]any{userID},
|
||||
&stats.TotalAPIKeys,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
"SELECT COUNT(*) FROM api_keys WHERE user_id = $1 AND status = $2 AND deleted_at IS NULL",
|
||||
[]any{userID, service.StatusActive},
|
||||
&stats.ActiveAPIKeys,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 累计 Token 统计
|
||||
totalStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as total_requests,
|
||||
COALESCE(SUM(input_tokens), 0) as total_input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as total_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as total_cost,
|
||||
COALESCE(SUM(actual_cost), 0) as total_actual_cost,
|
||||
COALESCE(AVG(duration_ms), 0) as avg_duration_ms
|
||||
FROM usage_logs
|
||||
WHERE user_id = $1
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
totalStatsQuery,
|
||||
[]any{userID},
|
||||
&stats.TotalRequests,
|
||||
&stats.TotalInputTokens,
|
||||
&stats.TotalOutputTokens,
|
||||
&stats.TotalCacheCreationTokens,
|
||||
&stats.TotalCacheReadTokens,
|
||||
&stats.TotalCost,
|
||||
&stats.TotalActualCost,
|
||||
&stats.AverageDurationMs,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheCreationTokens + stats.TotalCacheReadTokens
|
||||
|
||||
// 今日 Token 统计
|
||||
todayStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as today_requests,
|
||||
COALESCE(SUM(input_tokens), 0) as today_input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as today_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as today_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as today_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as today_cost,
|
||||
COALESCE(SUM(actual_cost), 0) as today_actual_cost
|
||||
FROM usage_logs
|
||||
WHERE user_id = $1 AND created_at >= $2
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
todayStatsQuery,
|
||||
[]any{userID, today},
|
||||
&stats.TodayRequests,
|
||||
&stats.TodayInputTokens,
|
||||
&stats.TodayOutputTokens,
|
||||
&stats.TodayCacheCreationTokens,
|
||||
&stats.TodayCacheReadTokens,
|
||||
&stats.TodayCost,
|
||||
&stats.TodayActualCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.TodayTokens = stats.TodayInputTokens + stats.TodayOutputTokens + stats.TodayCacheCreationTokens + stats.TodayCacheReadTokens
|
||||
|
||||
// 性能指标：RPM 和 TPM（最近1分钟，仅统计该用户的请求）
|
||||
rpm, tpm, err := r.getPerformanceStats(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.Rpm = rpm
|
||||
stats.Tpm = tpm
|
||||
|
||||
// 按"有效平台"维度拆分（group.platform 优先，否则 account.platform）。
|
||||
// 与 ops 路径口径一致；HAVING 过滤掉无法确定平台的行（避免出现空字符串平台）。
|
||||
// 与上面 totalStatsQuery/todayStatsQuery 的总值可能略微差异，原因有二：
|
||||
// 1) 无平台归属的极少数行（group/account 都没 platform）会被 HAVING 排除；
|
||||
// 2) usageLogSuccessFilterUL 会把 actual_cost = 0 的失败 placeholder 行排除，
|
||||
// 而 totalStatsQuery/todayStatsQuery 没有这层过滤、会把这些行的 request 计数算进去。
|
||||
platformQuery := `
|
||||
SELECT
|
||||
` + usageLogEffectivePlatformExpr + ` as platform,
|
||||
COUNT(*) as total_requests,
|
||||
COALESCE(SUM(ul.input_tokens + ul.output_tokens + ul.cache_creation_tokens + ul.cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(ul.actual_cost), 0) as total_actual_cost,
|
||||
COUNT(*) FILTER (WHERE ul.created_at >= $2) as today_requests,
|
||||
COALESCE(SUM(ul.input_tokens + ul.output_tokens + ul.cache_creation_tokens + ul.cache_read_tokens) FILTER (WHERE ul.created_at >= $2), 0) as today_tokens,
|
||||
COALESCE(SUM(ul.actual_cost) FILTER (WHERE ul.created_at >= $2), 0) as today_actual_cost
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN groups g ON g.id = ul.group_id
|
||||
LEFT JOIN accounts a ON a.id = ul.account_id
|
||||
WHERE ul.user_id = $1
|
||||
AND ` + usageLogSuccessFilterUL + `
|
||||
GROUP BY ` + usageLogEffectivePlatformExpr + `
|
||||
HAVING ` + usageLogEffectivePlatformExpr + ` IS NOT NULL AND ` + usageLogEffectivePlatformExpr + ` <> ''
|
||||
ORDER BY total_actual_cost DESC
|
||||
`
|
||||
rows, err := r.sql.QueryContext(ctx, platformQuery, userID, today)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var p PlatformDashboardStats
|
||||
if err := rows.Scan(
|
||||
&p.Platform,
|
||||
&p.TotalRequests,
|
||||
&p.TotalTokens,
|
||||
&p.TotalActualCost,
|
||||
&p.TodayRequests,
|
||||
&p.TodayTokens,
|
||||
&p.TodayActualCost,
|
||||
); err != nil {
|
||||
_ = rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
stats.ByPlatform = append(stats.ByPlatform, p)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// getPerformanceStatsByAPIKey 获取指定 API Key 的 RPM 和 TPM（近5分钟平均值）
|
||||
func (r *usageLogRepository) getPerformanceStatsByAPIKey(ctx context.Context, apiKeyID int64) (rpm, tpm int64, err error) {
|
||||
fiveMinutesAgo := time.Now().Add(-5 * time.Minute)
|
||||
query := `
|
||||
SELECT
|
||||
COUNT(*) as request_count,
|
||||
COALESCE(SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0) as token_count
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1 AND api_key_id = $2`
|
||||
args := []any{fiveMinutesAgo, apiKeyID}
|
||||
|
||||
var requestCount int64
|
||||
var tokenCount int64
|
||||
if err := scanSingleRow(ctx, r.sql, query, args, &requestCount, &tokenCount); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
return requestCount / 5, tokenCount / 5, nil
|
||||
}
|
||||
|
||||
// GetAPIKeyDashboardStats 获取指定 API Key 的仪表盘统计（按 api_key_id 过滤）
|
||||
func (r *usageLogRepository) GetAPIKeyDashboardStats(ctx context.Context, apiKeyID int64) (*UserDashboardStats, error) {
|
||||
stats := &UserDashboardStats{}
|
||||
today := timezone.Today()
|
||||
|
||||
// API Key 维度不需要统计 key 数量，设为 1
|
||||
stats.TotalAPIKeys = 1
|
||||
stats.ActiveAPIKeys = 1
|
||||
|
||||
// 累计 Token 统计
|
||||
totalStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as total_requests,
|
||||
COALESCE(SUM(input_tokens), 0) as total_input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as total_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as total_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as total_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as total_cost,
|
||||
COALESCE(SUM(actual_cost), 0) as total_actual_cost,
|
||||
COALESCE(AVG(duration_ms), 0) as avg_duration_ms
|
||||
FROM usage_logs
|
||||
WHERE api_key_id = $1
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
totalStatsQuery,
|
||||
[]any{apiKeyID},
|
||||
&stats.TotalRequests,
|
||||
&stats.TotalInputTokens,
|
||||
&stats.TotalOutputTokens,
|
||||
&stats.TotalCacheCreationTokens,
|
||||
&stats.TotalCacheReadTokens,
|
||||
&stats.TotalCost,
|
||||
&stats.TotalActualCost,
|
||||
&stats.AverageDurationMs,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheCreationTokens + stats.TotalCacheReadTokens
|
||||
|
||||
// 今日 Token 统计
|
||||
todayStatsQuery := `
|
||||
SELECT
|
||||
COUNT(*) as today_requests,
|
||||
COALESCE(SUM(input_tokens), 0) as today_input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as today_output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as today_cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as today_cache_read_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as today_cost,
|
||||
COALESCE(SUM(actual_cost), 0) as today_actual_cost
|
||||
FROM usage_logs
|
||||
WHERE api_key_id = $1 AND created_at >= $2
|
||||
`
|
||||
if err := scanSingleRow(
|
||||
ctx,
|
||||
r.sql,
|
||||
todayStatsQuery,
|
||||
[]any{apiKeyID, today},
|
||||
&stats.TodayRequests,
|
||||
&stats.TodayInputTokens,
|
||||
&stats.TodayOutputTokens,
|
||||
&stats.TodayCacheCreationTokens,
|
||||
&stats.TodayCacheReadTokens,
|
||||
&stats.TodayCost,
|
||||
&stats.TodayActualCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.TodayTokens = stats.TodayInputTokens + stats.TodayOutputTokens + stats.TodayCacheCreationTokens + stats.TodayCacheReadTokens
|
||||
|
||||
// 性能指标：RPM 和 TPM（最近5分钟，按 API Key 过滤）
|
||||
rpm, tpm, err := r.getPerformanceStatsByAPIKey(ctx, apiKeyID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats.Rpm = rpm
|
||||
stats.Tpm = tpm
|
||||
|
||||
return stats, nil
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,712 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dbaccount "github.com/Wei-Shaw/sub2api/ent/account"
|
||||
dbapikey "github.com/Wei-Shaw/sub2api/ent/apikey"
|
||||
dbgroup "github.com/Wei-Shaw/sub2api/ent/group"
|
||||
"github.com/Wei-Shaw/sub2api/ent/schema/mixins"
|
||||
dbuser "github.com/Wei-Shaw/sub2api/ent/user"
|
||||
dbusersub "github.com/Wei-Shaw/sub2api/ent/usersubscription"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
)
|
||||
|
||||
const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, service_tier, reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, created_at"
|
||||
|
||||
func (r *usageLogRepository) GetByID(ctx context.Context, id int64) (log *service.UsageLog, err error) {
|
||||
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE id = $1"
|
||||
rows, err := r.sql.QueryContext(ctx, query, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
log = nil
|
||||
}
|
||||
}()
|
||||
if !rows.Next() {
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, service.ErrUsageLogNotFound
|
||||
}
|
||||
log, err = scanUsageLog(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return log, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByUser(ctx context.Context, userID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
return r.listUsageLogsWithPagination(ctx, "WHERE user_id = $1", []any{userID}, params)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByAPIKey(ctx context.Context, apiKeyID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
return r.listUsageLogsWithPagination(ctx, "WHERE api_key_id = $1", []any{apiKeyID}, params)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByAccount(ctx context.Context, accountID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
return r.listUsageLogsWithPagination(ctx, "WHERE account_id = $1", []any{accountID}, params)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByUserAndTimeRange(ctx context.Context, userID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE user_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
|
||||
logs, err := r.queryUsageLogs(ctx, query, userID, startTime, endTime)
|
||||
return logs, nil, err
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByAPIKeyAndTimeRange(ctx context.Context, apiKeyID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE api_key_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
|
||||
logs, err := r.queryUsageLogs(ctx, query, apiKeyID, startTime, endTime)
|
||||
return logs, nil, err
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByAccountAndTimeRange(ctx context.Context, accountID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE account_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
|
||||
logs, err := r.queryUsageLogs(ctx, query, accountID, startTime, endTime)
|
||||
return logs, nil, err
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) ListByModelAndTimeRange(ctx context.Context, modelName string, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
query := fmt.Sprintf("SELECT %s FROM usage_logs WHERE %s = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000", usageLogSelectColumns, rawUsageLogModelColumn)
|
||||
logs, err := r.queryUsageLogs(ctx, query, modelName, startTime, endTime)
|
||||
return logs, nil, err
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) Delete(ctx context.Context, id int64) error {
|
||||
_, err := r.sql.ExecContext(ctx, "DELETE FROM usage_logs WHERE id = $1", id)
|
||||
return err
|
||||
}
|
||||
|
||||
// UsageLogFilters represents filters for usage log queries
|
||||
type UsageLogFilters = usagestats.UsageLogFilters
|
||||
|
||||
// ListWithFilters lists usage logs with optional filters (for admin)
|
||||
func (r *usageLogRepository) ListWithFilters(ctx context.Context, params pagination.PaginationParams, filters UsageLogFilters) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
conditions := make([]string, 0, 9)
|
||||
args := make([]any, 0, 9)
|
||||
|
||||
if filters.UserID > 0 {
|
||||
conditions = append(conditions, fmt.Sprintf("user_id = $%d", len(args)+1))
|
||||
args = append(args, filters.UserID)
|
||||
}
|
||||
if filters.APIKeyID > 0 {
|
||||
conditions = append(conditions, fmt.Sprintf("api_key_id = $%d", len(args)+1))
|
||||
args = append(args, filters.APIKeyID)
|
||||
}
|
||||
if filters.AccountID > 0 {
|
||||
conditions = append(conditions, fmt.Sprintf("account_id = $%d", len(args)+1))
|
||||
args = append(args, filters.AccountID)
|
||||
}
|
||||
if filters.GroupID > 0 {
|
||||
conditions = append(conditions, fmt.Sprintf("group_id = $%d", len(args)+1))
|
||||
args = append(args, filters.GroupID)
|
||||
}
|
||||
conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource)
|
||||
conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream)
|
||||
if filters.BillingType != nil {
|
||||
conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1))
|
||||
args = append(args, int16(*filters.BillingType))
|
||||
}
|
||||
conditions, args = appendUsageLogBillingModeWhereCondition(conditions, args, filters.BillingMode)
|
||||
if filters.StartTime != nil {
|
||||
conditions = append(conditions, fmt.Sprintf("created_at >= $%d", len(args)+1))
|
||||
args = append(args, *filters.StartTime)
|
||||
}
|
||||
if filters.EndTime != nil {
|
||||
conditions = append(conditions, fmt.Sprintf("created_at < $%d", len(args)+1))
|
||||
args = append(args, *filters.EndTime)
|
||||
}
|
||||
|
||||
whereClause := buildWhere(conditions)
|
||||
var (
|
||||
logs []service.UsageLog
|
||||
page *pagination.PaginationResult
|
||||
err error
|
||||
)
|
||||
if shouldUseFastUsageLogTotal(filters) {
|
||||
logs, page, err = r.listUsageLogsWithFastPagination(ctx, whereClause, args, params)
|
||||
} else {
|
||||
logs, page, err = r.listUsageLogsWithPagination(ctx, whereClause, args, params)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if err := r.hydrateUsageLogAssociations(ctx, logs); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return logs, page, nil
|
||||
}
|
||||
|
||||
func shouldUseFastUsageLogTotal(filters UsageLogFilters) bool {
|
||||
if filters.ExactTotal {
|
||||
return false
|
||||
}
|
||||
// 强选择过滤下记录集通常较小，保留精确总数。
|
||||
return filters.UserID == 0 && filters.APIKeyID == 0 && filters.AccountID == 0
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) listUsageLogsWithPagination(ctx context.Context, whereClause string, args []any, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
countQuery := "SELECT COUNT(*) FROM usage_logs " + whereClause
|
||||
var total int64
|
||||
if err := scanSingleRow(ctx, r.sql, countQuery, args, &total); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
limitPos := len(args) + 1
|
||||
offsetPos := len(args) + 2
|
||||
listArgs := append(append([]any{}, args...), params.Limit(), params.Offset())
|
||||
query := fmt.Sprintf("SELECT %s FROM usage_logs %s ORDER BY %s LIMIT $%d OFFSET $%d", usageLogSelectColumns, whereClause, usageLogOrderBy(params), limitPos, offsetPos)
|
||||
logs, err := r.queryUsageLogs(ctx, query, listArgs...)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return logs, paginationResultFromTotal(total, params), nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) listUsageLogsWithFastPagination(ctx context.Context, whereClause string, args []any, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
|
||||
limit := params.Limit()
|
||||
offset := params.Offset()
|
||||
|
||||
limitPos := len(args) + 1
|
||||
offsetPos := len(args) + 2
|
||||
listArgs := append(append([]any{}, args...), limit+1, offset)
|
||||
query := fmt.Sprintf("SELECT %s FROM usage_logs %s ORDER BY %s LIMIT $%d OFFSET $%d", usageLogSelectColumns, whereClause, usageLogOrderBy(params), limitPos, offsetPos)
|
||||
|
||||
logs, err := r.queryUsageLogs(ctx, query, listArgs...)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
hasMore := false
|
||||
if len(logs) > limit {
|
||||
hasMore = true
|
||||
logs = logs[:limit]
|
||||
}
|
||||
|
||||
total := int64(offset) + int64(len(logs))
|
||||
if hasMore {
|
||||
// 只保证“还有下一页”，避免对超大表做全量 COUNT(*)。
|
||||
total = int64(offset) + int64(limit) + 1
|
||||
}
|
||||
|
||||
return logs, paginationResultFromTotal(total, params), nil
|
||||
}
|
||||
|
||||
func usageLogOrderBy(params pagination.PaginationParams) string {
|
||||
sortBy := strings.ToLower(strings.TrimSpace(params.SortBy))
|
||||
sortOrder := strings.ToUpper(params.NormalizedSortOrder(pagination.SortOrderDesc))
|
||||
|
||||
var column string
|
||||
switch sortBy {
|
||||
case "model":
|
||||
column = "COALESCE(NULLIF(TRIM(requested_model), ''), model)"
|
||||
case "created_at":
|
||||
column = "created_at"
|
||||
default:
|
||||
column = "id"
|
||||
}
|
||||
|
||||
if column == "id" {
|
||||
return fmt.Sprintf("id %s", sortOrder)
|
||||
}
|
||||
return fmt.Sprintf("%s %s, id %s", column, sortOrder, sortOrder)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) queryUsageLogs(ctx context.Context, query string, args ...any) (logs []service.UsageLog, err error) {
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
logs = nil
|
||||
}
|
||||
}()
|
||||
|
||||
logs = make([]service.UsageLog, 0)
|
||||
for rows.Next() {
|
||||
var log *service.UsageLog
|
||||
log, err = scanUsageLog(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
logs = append(logs, *log)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return logs, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) hydrateUsageLogAssociations(ctx context.Context, logs []service.UsageLog) error {
|
||||
// 关联数据使用 Ent 批量加载，避免把复杂 SQL 继续膨胀。
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
ids := collectUsageLogIDs(logs)
|
||||
users, err := r.loadUsers(ctx, ids.userIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
apiKeys, err := r.loadAPIKeys(ctx, ids.apiKeyIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
accounts, err := r.loadAccounts(ctx, ids.accountIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
groups, err := r.loadGroups(ctx, ids.groupIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
subs, err := r.loadSubscriptions(ctx, ids.subscriptionIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i := range logs {
|
||||
if user, ok := users[logs[i].UserID]; ok {
|
||||
logs[i].User = user
|
||||
}
|
||||
if key, ok := apiKeys[logs[i].APIKeyID]; ok {
|
||||
logs[i].APIKey = key
|
||||
}
|
||||
if acc, ok := accounts[logs[i].AccountID]; ok {
|
||||
logs[i].Account = acc
|
||||
}
|
||||
if logs[i].GroupID != nil {
|
||||
if group, ok := groups[*logs[i].GroupID]; ok {
|
||||
logs[i].Group = group
|
||||
}
|
||||
}
|
||||
if logs[i].SubscriptionID != nil {
|
||||
if sub, ok := subs[*logs[i].SubscriptionID]; ok {
|
||||
logs[i].Subscription = sub
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type usageLogIDs struct {
|
||||
userIDs []int64
|
||||
apiKeyIDs []int64
|
||||
accountIDs []int64
|
||||
groupIDs []int64
|
||||
subscriptionIDs []int64
|
||||
}
|
||||
|
||||
func collectUsageLogIDs(logs []service.UsageLog) usageLogIDs {
|
||||
idSet := func() map[int64]struct{} { return make(map[int64]struct{}) }
|
||||
|
||||
userIDs := idSet()
|
||||
apiKeyIDs := idSet()
|
||||
accountIDs := idSet()
|
||||
groupIDs := idSet()
|
||||
subscriptionIDs := idSet()
|
||||
|
||||
for i := range logs {
|
||||
userIDs[logs[i].UserID] = struct{}{}
|
||||
apiKeyIDs[logs[i].APIKeyID] = struct{}{}
|
||||
accountIDs[logs[i].AccountID] = struct{}{}
|
||||
if logs[i].GroupID != nil {
|
||||
groupIDs[*logs[i].GroupID] = struct{}{}
|
||||
}
|
||||
if logs[i].SubscriptionID != nil {
|
||||
subscriptionIDs[*logs[i].SubscriptionID] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
return usageLogIDs{
|
||||
userIDs: setToSlice(userIDs),
|
||||
apiKeyIDs: setToSlice(apiKeyIDs),
|
||||
accountIDs: setToSlice(accountIDs),
|
||||
groupIDs: setToSlice(groupIDs),
|
||||
subscriptionIDs: setToSlice(subscriptionIDs),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) loadUsers(ctx context.Context, ids []int64) (map[int64]*service.User, error) {
|
||||
out := make(map[int64]*service.User)
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
// 无条件穿透软删除：ids 来自调用方已按 user_id 筛选的日志行；普通用户路径强制 UserID=本人（本人必为活跃用户），不会借此解析他人已删身份；仅 admin 路径可借此显示已删用户。
|
||||
models, err := r.client.User.Query().Where(dbuser.IDIn(ids...)).All(mixins.SkipSoftDelete(ctx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range models {
|
||||
out[m.ID] = userEntityToService(m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) loadAPIKeys(ctx context.Context, ids []int64) (map[int64]*service.APIKey, error) {
|
||||
out := make(map[int64]*service.APIKey)
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
models, err := r.client.APIKey.Query().Where(dbapikey.IDIn(ids...)).All(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range models {
|
||||
out[m.ID] = apiKeyEntityToService(m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) loadAccounts(ctx context.Context, ids []int64) (map[int64]*service.Account, error) {
|
||||
out := make(map[int64]*service.Account)
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
models, err := r.client.Account.Query().Where(dbaccount.IDIn(ids...)).All(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range models {
|
||||
out[m.ID] = accountEntityToService(m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) loadGroups(ctx context.Context, ids []int64) (map[int64]*service.Group, error) {
|
||||
out := make(map[int64]*service.Group)
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
models, err := r.client.Group.Query().Where(dbgroup.IDIn(ids...)).All(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range models {
|
||||
out[m.ID] = groupEntityToService(m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) loadSubscriptions(ctx context.Context, ids []int64) (map[int64]*service.UserSubscription, error) {
|
||||
out := make(map[int64]*service.UserSubscription)
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
models, err := r.client.UserSubscription.Query().Where(dbusersub.IDIn(ids...)).All(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, m := range models {
|
||||
out[m.ID] = userSubscriptionEntityToService(m)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, error) {
|
||||
var (
|
||||
id int64
|
||||
userID int64
|
||||
apiKeyID int64
|
||||
accountID int64
|
||||
requestID sql.NullString
|
||||
model string
|
||||
requestedModel sql.NullString
|
||||
upstreamModel sql.NullString
|
||||
groupID sql.NullInt64
|
||||
subscriptionID sql.NullInt64
|
||||
inputTokens int
|
||||
outputTokens int
|
||||
cacheCreationTokens int
|
||||
cacheReadTokens int
|
||||
cacheCreation5m int
|
||||
cacheCreation1h int
|
||||
imageOutputTokens int
|
||||
imageOutputCost float64
|
||||
inputCost float64
|
||||
outputCost float64
|
||||
cacheCreationCost float64
|
||||
cacheReadCost float64
|
||||
totalCost float64
|
||||
actualCost float64
|
||||
rateMultiplier float64
|
||||
accountRateMultiplier sql.NullFloat64
|
||||
billingType int16
|
||||
requestTypeRaw int16
|
||||
stream bool
|
||||
openaiWSMode bool
|
||||
durationMs sql.NullInt64
|
||||
firstTokenMs sql.NullInt64
|
||||
userAgent sql.NullString
|
||||
ipAddress sql.NullString
|
||||
imageCount int
|
||||
imageSize sql.NullString
|
||||
imageInputSize sql.NullString
|
||||
imageOutputSize sql.NullString
|
||||
imageSizeSource sql.NullString
|
||||
imageSizeBreakdown sql.NullString
|
||||
serviceTier sql.NullString
|
||||
reasoningEffort sql.NullString
|
||||
inboundEndpoint sql.NullString
|
||||
upstreamEndpoint sql.NullString
|
||||
cacheTTLOverridden bool
|
||||
channelID sql.NullInt64
|
||||
modelMappingChain sql.NullString
|
||||
billingTier sql.NullString
|
||||
billingMode sql.NullString
|
||||
accountStatsCost sql.NullFloat64
|
||||
createdAt time.Time
|
||||
)
|
||||
|
||||
if err := scanner.Scan(
|
||||
&id,
|
||||
&userID,
|
||||
&apiKeyID,
|
||||
&accountID,
|
||||
&requestID,
|
||||
&model,
|
||||
&requestedModel,
|
||||
&upstreamModel,
|
||||
&groupID,
|
||||
&subscriptionID,
|
||||
&inputTokens,
|
||||
&outputTokens,
|
||||
&cacheCreationTokens,
|
||||
&cacheReadTokens,
|
||||
&cacheCreation5m,
|
||||
&cacheCreation1h,
|
||||
&imageOutputTokens,
|
||||
&imageOutputCost,
|
||||
&inputCost,
|
||||
&outputCost,
|
||||
&cacheCreationCost,
|
||||
&cacheReadCost,
|
||||
&totalCost,
|
||||
&actualCost,
|
||||
&rateMultiplier,
|
||||
&accountRateMultiplier,
|
||||
&billingType,
|
||||
&requestTypeRaw,
|
||||
&stream,
|
||||
&openaiWSMode,
|
||||
&durationMs,
|
||||
&firstTokenMs,
|
||||
&userAgent,
|
||||
&ipAddress,
|
||||
&imageCount,
|
||||
&imageSize,
|
||||
&imageInputSize,
|
||||
&imageOutputSize,
|
||||
&imageSizeSource,
|
||||
&imageSizeBreakdown,
|
||||
&serviceTier,
|
||||
&reasoningEffort,
|
||||
&inboundEndpoint,
|
||||
&upstreamEndpoint,
|
||||
&cacheTTLOverridden,
|
||||
&channelID,
|
||||
&modelMappingChain,
|
||||
&billingTier,
|
||||
&billingMode,
|
||||
&accountStatsCost,
|
||||
&createdAt,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log := &service.UsageLog{
|
||||
ID: id,
|
||||
UserID: userID,
|
||||
APIKeyID: apiKeyID,
|
||||
AccountID: accountID,
|
||||
Model: model,
|
||||
RequestedModel: coalesceTrimmedString(requestedModel, model),
|
||||
InputTokens: inputTokens,
|
||||
OutputTokens: outputTokens,
|
||||
CacheCreationTokens: cacheCreationTokens,
|
||||
CacheReadTokens: cacheReadTokens,
|
||||
CacheCreation5mTokens: cacheCreation5m,
|
||||
CacheCreation1hTokens: cacheCreation1h,
|
||||
ImageOutputTokens: imageOutputTokens,
|
||||
ImageOutputCost: imageOutputCost,
|
||||
InputCost: inputCost,
|
||||
OutputCost: outputCost,
|
||||
CacheCreationCost: cacheCreationCost,
|
||||
CacheReadCost: cacheReadCost,
|
||||
TotalCost: totalCost,
|
||||
ActualCost: actualCost,
|
||||
RateMultiplier: rateMultiplier,
|
||||
AccountRateMultiplier: nullFloat64Ptr(accountRateMultiplier),
|
||||
BillingType: int8(billingType),
|
||||
RequestType: service.RequestTypeFromInt16(requestTypeRaw),
|
||||
ImageCount: imageCount,
|
||||
CacheTTLOverridden: cacheTTLOverridden,
|
||||
CreatedAt: createdAt,
|
||||
}
|
||||
// 先回填 legacy 字段，再基于 legacy + request_type 计算最终请求类型，保证历史数据兼容。
|
||||
log.Stream = stream
|
||||
log.OpenAIWSMode = openaiWSMode
|
||||
log.RequestType = log.EffectiveRequestType()
|
||||
log.Stream, log.OpenAIWSMode = service.ApplyLegacyRequestFields(log.RequestType, stream, openaiWSMode)
|
||||
|
||||
if requestID.Valid {
|
||||
log.RequestID = requestID.String
|
||||
}
|
||||
if groupID.Valid {
|
||||
value := groupID.Int64
|
||||
log.GroupID = &value
|
||||
}
|
||||
if subscriptionID.Valid {
|
||||
value := subscriptionID.Int64
|
||||
log.SubscriptionID = &value
|
||||
}
|
||||
if durationMs.Valid {
|
||||
value := int(durationMs.Int64)
|
||||
log.DurationMs = &value
|
||||
}
|
||||
if firstTokenMs.Valid {
|
||||
value := int(firstTokenMs.Int64)
|
||||
log.FirstTokenMs = &value
|
||||
}
|
||||
if userAgent.Valid {
|
||||
log.UserAgent = &userAgent.String
|
||||
}
|
||||
if ipAddress.Valid {
|
||||
log.IPAddress = &ipAddress.String
|
||||
}
|
||||
if imageSize.Valid {
|
||||
log.ImageSize = &imageSize.String
|
||||
}
|
||||
if imageInputSize.Valid {
|
||||
log.ImageInputSize = &imageInputSize.String
|
||||
}
|
||||
if imageOutputSize.Valid {
|
||||
log.ImageOutputSize = &imageOutputSize.String
|
||||
}
|
||||
if imageSizeSource.Valid {
|
||||
log.ImageSizeSource = &imageSizeSource.String
|
||||
}
|
||||
log.ImageSizeBreakdown = stringIntMapFromNullJSON(imageSizeBreakdown)
|
||||
if serviceTier.Valid {
|
||||
log.ServiceTier = &serviceTier.String
|
||||
}
|
||||
if reasoningEffort.Valid {
|
||||
log.ReasoningEffort = &reasoningEffort.String
|
||||
}
|
||||
if inboundEndpoint.Valid {
|
||||
log.InboundEndpoint = &inboundEndpoint.String
|
||||
}
|
||||
if upstreamEndpoint.Valid {
|
||||
log.UpstreamEndpoint = &upstreamEndpoint.String
|
||||
}
|
||||
if upstreamModel.Valid {
|
||||
log.UpstreamModel = &upstreamModel.String
|
||||
}
|
||||
if channelID.Valid {
|
||||
value := channelID.Int64
|
||||
log.ChannelID = &value
|
||||
}
|
||||
if modelMappingChain.Valid {
|
||||
log.ModelMappingChain = &modelMappingChain.String
|
||||
}
|
||||
if billingTier.Valid {
|
||||
log.BillingTier = &billingTier.String
|
||||
}
|
||||
if billingMode.Valid {
|
||||
log.BillingMode = &billingMode.String
|
||||
}
|
||||
if accountStatsCost.Valid {
|
||||
log.AccountStatsCost = &accountStatsCost.Float64
|
||||
}
|
||||
|
||||
return log, nil
|
||||
}
|
||||
|
||||
func nullInt64(v *int64) sql.NullInt64 {
|
||||
if v == nil {
|
||||
return sql.NullInt64{}
|
||||
}
|
||||
return sql.NullInt64{Int64: *v, Valid: true}
|
||||
}
|
||||
|
||||
func nullInt(v *int) sql.NullInt64 {
|
||||
if v == nil {
|
||||
return sql.NullInt64{}
|
||||
}
|
||||
return sql.NullInt64{Int64: int64(*v), Valid: true}
|
||||
}
|
||||
|
||||
func nullFloat64Ptr(v sql.NullFloat64) *float64 {
|
||||
if !v.Valid {
|
||||
return nil
|
||||
}
|
||||
out := v.Float64
|
||||
return &out
|
||||
}
|
||||
|
||||
func nullString(v *string) sql.NullString {
|
||||
if v == nil || *v == "" {
|
||||
return sql.NullString{}
|
||||
}
|
||||
return sql.NullString{String: *v, Valid: true}
|
||||
}
|
||||
|
||||
func nullStringIntMapJSON(v map[string]int) any {
|
||||
if len(v) == 0 {
|
||||
return nil
|
||||
}
|
||||
payload, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return string(payload)
|
||||
}
|
||||
|
||||
func stringIntMapFromNullJSON(v sql.NullString) map[string]int {
|
||||
if !v.Valid || strings.TrimSpace(v.String) == "" {
|
||||
return nil
|
||||
}
|
||||
var out map[string]int
|
||||
if err := json.Unmarshal([]byte(v.String), &out); err != nil {
|
||||
return nil
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func coalesceTrimmedString(v sql.NullString, fallback string) string {
|
||||
if v.Valid && strings.TrimSpace(v.String) != "" {
|
||||
return v.String
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func setToSlice(set map[int64]struct{}) []int64 {
|
||||
out := make([]int64, 0, len(set))
|
||||
for id := range set {
|
||||
out = append(out, id)
|
||||
}
|
||||
return out
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,799 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
|
||||
)
|
||||
|
||||
// TrendDataPoint represents a single point in trend data
|
||||
type TrendDataPoint = usagestats.TrendDataPoint
|
||||
|
||||
// ModelStat represents usage statistics for a single model
|
||||
type ModelStat = usagestats.ModelStat
|
||||
|
||||
// UserUsageTrendPoint represents user usage trend data point
|
||||
type UserUsageTrendPoint = usagestats.UserUsageTrendPoint
|
||||
|
||||
// UserSpendingRankingItem represents a user spending ranking row.
|
||||
type UserSpendingRankingItem = usagestats.UserSpendingRankingItem
|
||||
type UserSpendingRankingResponse = usagestats.UserSpendingRankingResponse
|
||||
|
||||
// APIKeyUsageTrendPoint represents API key usage trend data point
|
||||
type APIKeyUsageTrendPoint = usagestats.APIKeyUsageTrendPoint
|
||||
|
||||
// GetAPIKeyUsageTrend returns usage trend data grouped by API key and date
|
||||
func (r *usageLogRepository) GetAPIKeyUsageTrend(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) (results []APIKeyUsageTrendPoint, err error) {
|
||||
dateFormat := safeDateFormat(granularity)
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
WITH top_keys AS (
|
||||
SELECT api_key_id
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1 AND created_at < $2
|
||||
GROUP BY api_key_id
|
||||
ORDER BY SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens) DESC
|
||||
LIMIT $3
|
||||
)
|
||||
SELECT
|
||||
TO_CHAR(u.created_at, '%s') as date,
|
||||
u.api_key_id,
|
||||
COALESCE(k.name, '') as key_name,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(u.input_tokens + u.output_tokens + u.cache_creation_tokens + u.cache_read_tokens), 0) as tokens
|
||||
FROM usage_logs u
|
||||
LEFT JOIN api_keys k ON u.api_key_id = k.id
|
||||
WHERE u.api_key_id IN (SELECT api_key_id FROM top_keys)
|
||||
AND u.created_at >= $4 AND u.created_at < $5
|
||||
GROUP BY date, u.api_key_id, k.name
|
||||
ORDER BY date ASC, tokens DESC
|
||||
`, dateFormat)
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, startTime, endTime, limit, startTime, endTime)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results = make([]APIKeyUsageTrendPoint, 0)
|
||||
for rows.Next() {
|
||||
var row APIKeyUsageTrendPoint
|
||||
if err = rows.Scan(&row.Date, &row.APIKeyID, &row.KeyName, &row.Requests, &row.Tokens); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetUserUsageTrend returns usage trend data grouped by user and date
|
||||
func (r *usageLogRepository) GetUserUsageTrend(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) (results []UserUsageTrendPoint, err error) {
|
||||
dateFormat := safeDateFormat(granularity)
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
WITH top_users AS (
|
||||
SELECT user_id
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1 AND created_at < $2
|
||||
GROUP BY user_id
|
||||
ORDER BY SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens) DESC
|
||||
LIMIT $3
|
||||
)
|
||||
SELECT
|
||||
TO_CHAR(u.created_at, '%s') as date,
|
||||
u.user_id,
|
||||
COALESCE(us.email, '') as email,
|
||||
COALESCE(us.username, '') as username,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(u.input_tokens + u.output_tokens + u.cache_creation_tokens + u.cache_read_tokens), 0) as tokens,
|
||||
COALESCE(SUM(u.total_cost), 0) as cost,
|
||||
COALESCE(SUM(u.actual_cost), 0) as actual_cost
|
||||
FROM usage_logs u
|
||||
LEFT JOIN users us ON u.user_id = us.id
|
||||
WHERE u.user_id IN (SELECT user_id FROM top_users)
|
||||
AND u.created_at >= $4 AND u.created_at < $5
|
||||
GROUP BY date, u.user_id, us.email, us.username
|
||||
ORDER BY date ASC, tokens DESC
|
||||
`, dateFormat)
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, startTime, endTime, limit, startTime, endTime)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results = make([]UserUsageTrendPoint, 0)
|
||||
for rows.Next() {
|
||||
var row UserUsageTrendPoint
|
||||
if err = rows.Scan(&row.Date, &row.UserID, &row.Email, &row.Username, &row.Requests, &row.Tokens, &row.Cost, &row.ActualCost); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetUserSpendingRanking returns user spending ranking aggregated within the time range.
|
||||
func (r *usageLogRepository) GetUserSpendingRanking(ctx context.Context, startTime, endTime time.Time, limit int) (result *UserSpendingRankingResponse, err error) {
|
||||
if limit <= 0 {
|
||||
limit = 12
|
||||
}
|
||||
|
||||
query := `
|
||||
WITH user_spend AS (
|
||||
SELECT
|
||||
u.user_id,
|
||||
COALESCE(us.email, '') as email,
|
||||
COALESCE(SUM(u.actual_cost), 0) as actual_cost,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(u.input_tokens + u.output_tokens + u.cache_creation_tokens + u.cache_read_tokens), 0) as tokens
|
||||
FROM usage_logs u
|
||||
LEFT JOIN users us ON u.user_id = us.id
|
||||
WHERE u.created_at >= $1 AND u.created_at < $2
|
||||
GROUP BY u.user_id, us.email
|
||||
),
|
||||
ranked AS (
|
||||
SELECT
|
||||
user_id,
|
||||
email,
|
||||
actual_cost,
|
||||
requests,
|
||||
tokens,
|
||||
COALESCE(SUM(actual_cost) OVER (), 0) as total_actual_cost,
|
||||
COALESCE(SUM(requests) OVER (), 0) as total_requests,
|
||||
COALESCE(SUM(tokens) OVER (), 0) as total_tokens
|
||||
FROM user_spend
|
||||
ORDER BY actual_cost DESC, tokens DESC, user_id ASC
|
||||
LIMIT $3
|
||||
)
|
||||
SELECT
|
||||
user_id,
|
||||
email,
|
||||
actual_cost,
|
||||
requests,
|
||||
tokens,
|
||||
total_actual_cost,
|
||||
total_requests,
|
||||
total_tokens
|
||||
FROM ranked
|
||||
ORDER BY actual_cost DESC, tokens DESC, user_id ASC
|
||||
`
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, startTime, endTime, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
result = nil
|
||||
}
|
||||
}()
|
||||
|
||||
ranking := make([]UserSpendingRankingItem, 0)
|
||||
totalActualCost := 0.0
|
||||
totalRequests := int64(0)
|
||||
totalTokens := int64(0)
|
||||
for rows.Next() {
|
||||
var row UserSpendingRankingItem
|
||||
if err = rows.Scan(&row.UserID, &row.Email, &row.ActualCost, &row.Requests, &row.Tokens, &totalActualCost, &totalRequests, &totalTokens); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ranking = append(ranking, row)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &UserSpendingRankingResponse{
|
||||
Ranking: ranking,
|
||||
TotalActualCost: totalActualCost,
|
||||
TotalRequests: totalRequests,
|
||||
TotalTokens: totalTokens,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetUserUsageTrendByUserID 获取指定用户的使用趋势
|
||||
func (r *usageLogRepository) GetUserUsageTrendByUserID(ctx context.Context, userID int64, startTime, endTime time.Time, granularity string) (results []TrendDataPoint, err error) {
|
||||
dateFormat := safeDateFormat(granularity)
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
SELECT
|
||||
TO_CHAR(created_at, '%s') as date,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as cache_read_tokens,
|
||||
COALESCE(SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as cost,
|
||||
COALESCE(SUM(actual_cost), 0) as actual_cost
|
||||
FROM usage_logs
|
||||
WHERE user_id = $1 AND created_at >= $2 AND created_at < $3
|
||||
GROUP BY date
|
||||
ORDER BY date ASC
|
||||
`, dateFormat)
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, userID, startTime, endTime)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results, err = scanTrendRows(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetUserModelStats 获取指定用户的模型统计
|
||||
func (r *usageLogRepository) GetUserModelStats(ctx context.Context, userID int64, startTime, endTime time.Time) (results []ModelStat, err error) {
|
||||
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, 0, 0, 0, "", nil, nil, nil, usagestats.ModelSourceRequested, "")
|
||||
}
|
||||
|
||||
// GetUsageTrendWithFilters returns usage trend data with optional filters
|
||||
func (r *usageLogRepository) GetUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) (results []TrendDataPoint, err error) {
|
||||
return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, userID, apiKeyID, accountID, groupID, model, "", requestType, stream, billingType, "")
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) GetUsageTrendWithUsageFilters(ctx context.Context, startTime, endTime time.Time, granularity string, filters UsageLogFilters) (results []TrendDataPoint, err error) {
|
||||
return r.getUsageTrendWithFilters(ctx, startTime, endTime, granularity, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.ModelFilterSource, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) getUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, modelSource string, requestType *int16, stream *bool, billingType *int8, billingMode string) (results []TrendDataPoint, err error) {
|
||||
if shouldUsePreaggregatedTrend(granularity, userID, apiKeyID, accountID, groupID, model, requestType, stream, billingType, billingMode) {
|
||||
aggregated, aggregatedErr := r.getUsageTrendFromAggregates(ctx, startTime, endTime, granularity)
|
||||
if aggregatedErr == nil && len(aggregated) > 0 {
|
||||
return aggregated, nil
|
||||
}
|
||||
}
|
||||
|
||||
dateFormat := safeDateFormat(granularity)
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
SELECT
|
||||
TO_CHAR(created_at, '%s') as date,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as cache_read_tokens,
|
||||
COALESCE(SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as cost,
|
||||
COALESCE(SUM(actual_cost), 0) as actual_cost
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1 AND created_at < $2
|
||||
`, dateFormat)
|
||||
|
||||
args := []any{startTime, endTime}
|
||||
if userID > 0 {
|
||||
query += fmt.Sprintf(" AND user_id = $%d", len(args)+1)
|
||||
args = append(args, userID)
|
||||
}
|
||||
if apiKeyID > 0 {
|
||||
query += fmt.Sprintf(" AND api_key_id = $%d", len(args)+1)
|
||||
args = append(args, apiKeyID)
|
||||
}
|
||||
if accountID > 0 {
|
||||
query += fmt.Sprintf(" AND account_id = $%d", len(args)+1)
|
||||
args = append(args, accountID)
|
||||
}
|
||||
if groupID > 0 {
|
||||
query += fmt.Sprintf(" AND group_id = $%d", len(args)+1)
|
||||
args = append(args, groupID)
|
||||
}
|
||||
query, args = appendUsageLogModelQueryFilter(query, args, model, modelSource)
|
||||
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
|
||||
if billingType != nil {
|
||||
query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1)
|
||||
args = append(args, int16(*billingType))
|
||||
}
|
||||
query, args = appendUsageLogBillingModeQueryFilter(query, args, billingMode, "")
|
||||
query += " GROUP BY date ORDER BY date ASC"
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results, err = scanTrendRows(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func shouldUsePreaggregatedTrend(granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string) bool {
|
||||
if granularity != "day" && granularity != "hour" {
|
||||
return false
|
||||
}
|
||||
return userID == 0 &&
|
||||
apiKeyID == 0 &&
|
||||
accountID == 0 &&
|
||||
groupID == 0 &&
|
||||
model == "" &&
|
||||
requestType == nil &&
|
||||
stream == nil &&
|
||||
billingType == nil &&
|
||||
billingMode == ""
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) getUsageTrendFromAggregates(ctx context.Context, startTime, endTime time.Time, granularity string) (results []TrendDataPoint, err error) {
|
||||
dateFormat := safeDateFormat(granularity)
|
||||
query := ""
|
||||
args := []any{startTime, endTime}
|
||||
|
||||
switch granularity {
|
||||
case "hour":
|
||||
query = fmt.Sprintf(`
|
||||
SELECT
|
||||
TO_CHAR(bucket_start, '%s') as date,
|
||||
total_requests as requests,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
cache_creation_tokens,
|
||||
cache_read_tokens,
|
||||
(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens) as total_tokens,
|
||||
total_cost as cost,
|
||||
actual_cost
|
||||
FROM usage_dashboard_hourly
|
||||
WHERE bucket_start >= $1 AND bucket_start < $2
|
||||
ORDER BY bucket_start ASC
|
||||
`, dateFormat)
|
||||
case "day":
|
||||
query = fmt.Sprintf(`
|
||||
SELECT
|
||||
TO_CHAR(bucket_date::timestamp, '%s') as date,
|
||||
total_requests as requests,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
cache_creation_tokens,
|
||||
cache_read_tokens,
|
||||
(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens) as total_tokens,
|
||||
total_cost as cost,
|
||||
actual_cost
|
||||
FROM usage_dashboard_daily
|
||||
WHERE bucket_date >= $1::date AND bucket_date < $2::date
|
||||
ORDER BY bucket_date ASC
|
||||
`, dateFormat)
|
||||
default:
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results, err = scanTrendRows(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetModelStatsWithFilters returns model statistics with optional filters
|
||||
func (r *usageLogRepository) GetModelStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []ModelStat, err error) {
|
||||
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, usagestats.ModelSourceRequested, "")
|
||||
}
|
||||
|
||||
// GetModelStatsWithFiltersBySource returns model statistics with optional filters and model source dimension.
|
||||
// source: requested | upstream | mapping.
|
||||
func (r *usageLogRepository) GetModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8, source string) (results []ModelStat, err error) {
|
||||
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, source, "")
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) GetModelStatsWithUsageFiltersBySource(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters, source string) (results []ModelStat, err error) {
|
||||
return r.getModelStatsWithFiltersBySource(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, source, filters.BillingMode)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) getModelStatsWithFiltersBySource(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, source string, billingMode string) (results []ModelStat, err error) {
|
||||
actualCostExpr := "COALESCE(SUM(actual_cost), 0) as actual_cost"
|
||||
// 当仅按 account_id 聚合时，实际费用使用账号倍率（total_cost * account_rate_multiplier）。
|
||||
if accountID > 0 && userID == 0 && apiKeyID == 0 {
|
||||
actualCostExpr = "COALESCE(SUM(COALESCE(account_stats_cost, total_cost) * COALESCE(account_rate_multiplier, 1)), 0) as actual_cost"
|
||||
}
|
||||
accountCostExpr := "COALESCE(SUM(COALESCE(account_stats_cost, total_cost) * COALESCE(account_rate_multiplier, 1)), 0) as account_cost"
|
||||
modelExpr := resolveModelDimensionExpression(source)
|
||||
|
||||
query := fmt.Sprintf(`
|
||||
SELECT
|
||||
%s as model,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(cache_creation_tokens), 0) as cache_creation_tokens,
|
||||
COALESCE(SUM(cache_read_tokens), 0) as cache_read_tokens,
|
||||
COALESCE(SUM(input_tokens + output_tokens + cache_creation_tokens + cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as cost,
|
||||
%s,
|
||||
%s
|
||||
FROM usage_logs
|
||||
WHERE created_at >= $1 AND created_at < $2
|
||||
`, modelExpr, actualCostExpr, accountCostExpr)
|
||||
|
||||
args := []any{startTime, endTime}
|
||||
if userID > 0 {
|
||||
query += fmt.Sprintf(" AND user_id = $%d", len(args)+1)
|
||||
args = append(args, userID)
|
||||
}
|
||||
if apiKeyID > 0 {
|
||||
query += fmt.Sprintf(" AND api_key_id = $%d", len(args)+1)
|
||||
args = append(args, apiKeyID)
|
||||
}
|
||||
if accountID > 0 {
|
||||
query += fmt.Sprintf(" AND account_id = $%d", len(args)+1)
|
||||
args = append(args, accountID)
|
||||
}
|
||||
if groupID > 0 {
|
||||
query += fmt.Sprintf(" AND group_id = $%d", len(args)+1)
|
||||
args = append(args, groupID)
|
||||
}
|
||||
if strings.TrimSpace(model) != "" {
|
||||
query += fmt.Sprintf(" AND %s = $%d", modelExpr, len(args)+1)
|
||||
args = append(args, model)
|
||||
}
|
||||
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
|
||||
if billingType != nil {
|
||||
query += fmt.Sprintf(" AND billing_type = $%d", len(args)+1)
|
||||
args = append(args, int16(*billingType))
|
||||
}
|
||||
query, args = appendUsageLogBillingModeQueryFilter(query, args, billingMode, "")
|
||||
query += fmt.Sprintf(" GROUP BY %s ORDER BY total_tokens DESC", modelExpr)
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// 保持主错误优先；仅在无错误时回传 Close 失败。
|
||||
// 同时清空返回值，避免误用不完整结果。
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results, err = scanModelStatsRows(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetGroupStatsWithFilters returns group usage statistics with optional filters
|
||||
func (r *usageLogRepository) GetGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) (results []usagestats.GroupStat, err error) {
|
||||
return r.getGroupStatsWithFilters(ctx, startTime, endTime, userID, apiKeyID, accountID, groupID, "", requestType, stream, billingType, "")
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) GetGroupStatsWithUsageFilters(ctx context.Context, startTime, endTime time.Time, filters UsageLogFilters) (results []usagestats.GroupStat, err error) {
|
||||
return r.getGroupStatsWithFilters(ctx, startTime, endTime, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType, filters.BillingMode)
|
||||
}
|
||||
|
||||
func (r *usageLogRepository) getGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8, billingMode string) (results []usagestats.GroupStat, err error) {
|
||||
query := `
|
||||
SELECT
|
||||
COALESCE(ul.group_id, 0) as group_id,
|
||||
COALESCE(g.name, '') as group_name,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(ul.input_tokens + ul.output_tokens + ul.cache_creation_tokens + ul.cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(ul.total_cost), 0) as cost,
|
||||
COALESCE(SUM(ul.actual_cost), 0) as actual_cost,
|
||||
COALESCE(SUM(COALESCE(ul.account_stats_cost, ul.total_cost) * COALESCE(ul.account_rate_multiplier, 1)), 0) as account_cost
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN groups g ON g.id = ul.group_id
|
||||
WHERE ul.created_at >= $1 AND ul.created_at < $2
|
||||
`
|
||||
|
||||
args := []any{startTime, endTime}
|
||||
if userID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.user_id = $%d", len(args)+1)
|
||||
args = append(args, userID)
|
||||
}
|
||||
if apiKeyID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.api_key_id = $%d", len(args)+1)
|
||||
args = append(args, apiKeyID)
|
||||
}
|
||||
if accountID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.account_id = $%d", len(args)+1)
|
||||
args = append(args, accountID)
|
||||
}
|
||||
if groupID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.group_id = $%d", len(args)+1)
|
||||
args = append(args, groupID)
|
||||
}
|
||||
if strings.TrimSpace(model) != "" {
|
||||
modelExpr := resolveModelDimensionExpressionWithAlias(usagestats.ModelSourceRequested, "ul")
|
||||
query += fmt.Sprintf(" AND %s = $%d", modelExpr, len(args)+1)
|
||||
args = append(args, model)
|
||||
}
|
||||
query, args = appendRequestTypeOrStreamQueryFilter(query, args, requestType, stream)
|
||||
if billingType != nil {
|
||||
query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1)
|
||||
args = append(args, int16(*billingType))
|
||||
}
|
||||
query, args = appendUsageLogBillingModeQueryFilter(query, args, billingMode, "ul")
|
||||
query += " GROUP BY ul.group_id, g.name ORDER BY total_tokens DESC"
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results = make([]usagestats.GroupStat, 0)
|
||||
for rows.Next() {
|
||||
var row usagestats.GroupStat
|
||||
if err := rows.Scan(
|
||||
&row.GroupID,
|
||||
&row.GroupName,
|
||||
&row.Requests,
|
||||
&row.TotalTokens,
|
||||
&row.Cost,
|
||||
&row.ActualCost,
|
||||
&row.AccountCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetUserBreakdownStats returns per-user usage breakdown within a specific dimension.
|
||||
func (r *usageLogRepository) GetUserBreakdownStats(ctx context.Context, startTime, endTime time.Time, dim usagestats.UserBreakdownDimension, limit int) (results []usagestats.UserBreakdownItem, err error) {
|
||||
query := `
|
||||
SELECT
|
||||
COALESCE(ul.user_id, 0) as user_id,
|
||||
COALESCE(u.email, '') as email,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(ul.input_tokens + ul.output_tokens + ul.cache_creation_tokens + ul.cache_read_tokens), 0) as total_tokens,
|
||||
COALESCE(SUM(ul.total_cost), 0) as cost,
|
||||
COALESCE(SUM(ul.actual_cost), 0) as actual_cost,
|
||||
COALESCE(SUM(COALESCE(ul.account_stats_cost, ul.total_cost) * COALESCE(ul.account_rate_multiplier, 1)), 0) as account_cost
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN users u ON u.id = ul.user_id
|
||||
WHERE ul.created_at >= $1 AND ul.created_at < $2
|
||||
`
|
||||
args := []any{startTime, endTime}
|
||||
|
||||
if dim.GroupID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.group_id = $%d", len(args)+1)
|
||||
args = append(args, dim.GroupID)
|
||||
}
|
||||
if dim.Model != "" {
|
||||
query += fmt.Sprintf(" AND %s = $%d", resolveModelDimensionExpression(dim.ModelType), len(args)+1)
|
||||
args = append(args, dim.Model)
|
||||
}
|
||||
if dim.Endpoint != "" {
|
||||
col := resolveEndpointColumn(dim.EndpointType)
|
||||
query += fmt.Sprintf(" AND %s = $%d", col, len(args)+1)
|
||||
args = append(args, dim.Endpoint)
|
||||
}
|
||||
if dim.UserID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.user_id = $%d", len(args)+1)
|
||||
args = append(args, dim.UserID)
|
||||
}
|
||||
if dim.APIKeyID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.api_key_id = $%d", len(args)+1)
|
||||
args = append(args, dim.APIKeyID)
|
||||
}
|
||||
if dim.AccountID > 0 {
|
||||
query += fmt.Sprintf(" AND ul.account_id = $%d", len(args)+1)
|
||||
args = append(args, dim.AccountID)
|
||||
}
|
||||
if dim.RequestType != nil {
|
||||
query += fmt.Sprintf(" AND ul.request_type = $%d", len(args)+1)
|
||||
args = append(args, *dim.RequestType)
|
||||
}
|
||||
if dim.Stream != nil {
|
||||
query += fmt.Sprintf(" AND ul.stream = $%d", len(args)+1)
|
||||
args = append(args, *dim.Stream)
|
||||
}
|
||||
if dim.BillingType != nil {
|
||||
query += fmt.Sprintf(" AND ul.billing_type = $%d", len(args)+1)
|
||||
args = append(args, *dim.BillingType)
|
||||
}
|
||||
|
||||
query += " GROUP BY ul.user_id, u.email ORDER BY actual_cost DESC"
|
||||
if limit > 0 {
|
||||
query += fmt.Sprintf(" LIMIT %d", limit)
|
||||
}
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
if closeErr := rows.Close(); closeErr != nil && err == nil {
|
||||
err = closeErr
|
||||
results = nil
|
||||
}
|
||||
}()
|
||||
|
||||
results = make([]usagestats.UserBreakdownItem, 0)
|
||||
for rows.Next() {
|
||||
var row usagestats.UserBreakdownItem
|
||||
if err := rows.Scan(
|
||||
&row.UserID,
|
||||
&row.Email,
|
||||
&row.Requests,
|
||||
&row.TotalTokens,
|
||||
&row.Cost,
|
||||
&row.ActualCost,
|
||||
&row.AccountCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetAllGroupUsageSummary returns today's and cumulative actual_cost for every group.
|
||||
// todayStart is the start-of-day in the caller's timezone (UTC-based).
|
||||
// TODO(perf): This query scans ALL usage_logs rows for total_cost aggregation.
|
||||
// When usage_logs exceeds ~1M rows, consider adding a short-lived cache (30s)
|
||||
// or a materialized view / pre-aggregation table for cumulative costs.
|
||||
func (r *usageLogRepository) GetAllGroupUsageSummary(ctx context.Context, todayStart time.Time) ([]usagestats.GroupUsageSummary, error) {
|
||||
query := `
|
||||
SELECT
|
||||
g.id AS group_id,
|
||||
COALESCE(SUM(ul.actual_cost), 0) AS total_cost,
|
||||
COALESCE(SUM(CASE WHEN ul.created_at >= $1 THEN ul.actual_cost ELSE 0 END), 0) AS today_cost
|
||||
FROM groups g
|
||||
LEFT JOIN usage_logs ul ON ul.group_id = g.id
|
||||
GROUP BY g.id
|
||||
`
|
||||
|
||||
rows, err := r.sql.QueryContext(ctx, query, todayStart)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
var results []usagestats.GroupUsageSummary
|
||||
for rows.Next() {
|
||||
var row usagestats.GroupUsageSummary
|
||||
if err := rows.Scan(&row.GroupID, &row.TotalCost, &row.TodayCost); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// resolveModelDimensionExpression maps model source type to a safe SQL expression.
|
||||
func resolveModelDimensionExpression(modelType string) string {
|
||||
return resolveModelDimensionExpressionWithAlias(modelType, "")
|
||||
}
|
||||
|
||||
func resolveModelDimensionExpressionWithAlias(modelType, alias string) string {
|
||||
column := func(name string) string {
|
||||
if alias == "" {
|
||||
return name
|
||||
}
|
||||
return alias + "." + name
|
||||
}
|
||||
requestedExpr := fmt.Sprintf("COALESCE(NULLIF(TRIM(%s), ''), %s)", column("requested_model"), column("model"))
|
||||
switch usagestats.NormalizeModelSource(modelType) {
|
||||
case usagestats.ModelSourceUpstream:
|
||||
return fmt.Sprintf("COALESCE(NULLIF(TRIM(%s), ''), %s)", column("upstream_model"), requestedExpr)
|
||||
case usagestats.ModelSourceMapping:
|
||||
return fmt.Sprintf("(%s || ' -> ' || COALESCE(NULLIF(TRIM(%s), ''), %s))", requestedExpr, column("upstream_model"), requestedExpr)
|
||||
default:
|
||||
return requestedExpr
|
||||
}
|
||||
}
|
||||
|
||||
func scanTrendRows(rows *sql.Rows) ([]TrendDataPoint, error) {
|
||||
results := make([]TrendDataPoint, 0)
|
||||
for rows.Next() {
|
||||
var row TrendDataPoint
|
||||
if err := rows.Scan(
|
||||
&row.Date,
|
||||
&row.Requests,
|
||||
&row.InputTokens,
|
||||
&row.OutputTokens,
|
||||
&row.CacheCreationTokens,
|
||||
&row.CacheReadTokens,
|
||||
&row.TotalTokens,
|
||||
&row.Cost,
|
||||
&row.ActualCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func scanModelStatsRows(rows *sql.Rows) ([]ModelStat, error) {
|
||||
results := make([]ModelStat, 0)
|
||||
for rows.Next() {
|
||||
var row ModelStat
|
||||
if err := rows.Scan(
|
||||
&row.Model,
|
||||
&row.Requests,
|
||||
&row.InputTokens,
|
||||
&row.OutputTokens,
|
||||
&row.CacheCreationTokens,
|
||||
&row.CacheReadTokens,
|
||||
&row.TotalTokens,
|
||||
&row.Cost,
|
||||
&row.ActualCost,
|
||||
&row.AccountCost,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results = append(results, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
Reference in New Issue
Block a user