package redisqueue import ( "context" "encoding/json" "net/http" "strings" "time" internallogging "github.com/router-for-me/CLIProxyAPI/v7/internal/logging" coresession "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/session" coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage") func init() { coreusage.RegisterPlugin(&usageQueuePlugin{})} type usageQueuePlugin struct{} func (p *usageQueuePlugin) HandleUsage(ctx context.Context, record coreusage.Record) { if p == nil { return } if !Enabled() || !UsageStatisticsEnabled() { return } timestamp := record.RequestedAt if timestamp.IsZero() { timestamp = time.Now() } modelName := strings.TrimSpace(record.Model) if modelName == "" { modelName = "unknown" } aliasName := strings.TrimSpace(record.Alias) if aliasName == "" { aliasName = modelName } provider := strings.TrimSpace(record.Provider) if provider == "" { provider = "unknown" } executorType := strings.TrimSpace(record.ExecutorType) if executorType == "" { executorType = "unknown" } authType := strings.TrimSpace(record.AuthType) if authType == "" { authType = "unknown" } apiKey := strings.TrimSpace(record.APIKey) requestID := strings.TrimSpace(internallogging.GetRequestID(ctx)) reasoningEffort := strings.TrimSpace(record.ReasoningEffort) if reasoningEffort == "" { reasoningEffort = coreusage.ReasoningEffortFromContext(ctx) } serviceTier := strings.TrimSpace(record.ServiceTier) if serviceTier == "" { serviceTier = strings.TrimSpace(record.RequestServiceTier) } if serviceTier == "" { serviceTier = coreusage.ServiceTierFromContext(ctx) } responseServiceTier := strings.TrimSpace(record.ResponseServiceTier) clientRequestMetadata := internallogging.GetClientRequestMetadata(ctx) sessionID := strings.TrimSpace(record.SessionID) parentSessionID := strings.TrimSpace(record.ParentSessionID) if sessionID == "" { sessionID = strings.TrimSpace(clientRequestMetadata.SessionID) parentSessionID = strings.TrimSpace(clientRequestMetadata.ParentSessionID) } else if parentSessionID == "" && sessionID == strings.TrimSpace(clientRequestMetadata.SessionID) { parentSessionID = strings.TrimSpace(clientRequestMetadata.ParentSessionID) } sessionID = coresession.NormalizeToCanonicalUUID(sessionID) parentSessionID = coresession.NormalizeToCanonicalUUID(parentSessionID) if sessionID == "" || sessionID == parentSessionID { parentSessionID = "" } usageDetail := coreusage.EnsureTokenBreakdownForProvider(record.Detail, record.Provider, record.ExecutorType) tokens := tokenStats{ InputTokens: usageDetail.InputTokens, OutputTokens: usageDetail.OutputTokens, ReasoningTokens: usageDetail.ReasoningTokens, CachedTokens: usageDetail.CachedTokens, CacheReadTokens: usageDetail.CacheReadTokens, CacheReadTokensPresent: true, CacheCreationTokens: usageDetail.CacheCreationTokens, TotalTokens: usageDetail.TotalTokens, } failed := record.Failed if !failed { failed = !resolveSuccess(ctx) } fail := resolveFail(ctx, record, failed) stream := record.Stream if !stream { stream = coreusage.StreamFromContext(ctx) } detail := requestDetail{ Timestamp: timestamp, LatencyMs: record.Latency.Milliseconds(), TTFTMs: record.TTFT.Milliseconds(), Source: record.Source, AuthIndex: record.AuthIndex, AccessTokenHash: record.AccessTokenSHA256, ClientIP: clientRequestMetadata.ClientIP, XForwardedFor: clientRequestMetadata.XForwardedFor, UserAgent: clientRequestMetadata.UserAgent, Tokens: tokens, Failed: failed, Generate: coreusage.GenerateEnabled(record.Generate), Stream: stream, Fail: fail, ResponseHeaders: record.ResponseHeaders, } payload, err := json.Marshal(queuedUsageDetail{ requestDetail: detail, AccountingVersion: coreusage.TokenAccountingSchemaVersion, TokenBreakdown: usageDetail.TokenBreakdown, Provider: provider, ExecutorType: executorType, Model: modelName, Alias: aliasName, Endpoint: resolveEndpoint(ctx), AuthType: authType, APIKey: apiKey, RequestID: requestID, SessionID: sessionID, ParentSessionID: parentSessionID, ReasoningEffort: reasoningEffort, ServiceTier: serviceTier, ResponseServiceTier: responseServiceTier, }) if err != nil { return } Enqueue(payload)} type queuedUsageDetail struct { requestDetail AccountingVersion int `json:"accounting_version"` TokenBreakdown coreusage.TokenBreakdown `json:"token_breakdown"` Provider string `json:"provider"` ExecutorType string `json:"executor_type"` Model string `json:"model"` Alias string `json:"alias"` Endpoint string `json:"endpoint"` AuthType string `json:"auth_type"` APIKey string `json:"api_key"` RequestID string `json:"request_id"` SessionID string `json:"session_id,omitempty"` ParentSessionID string `json:"parent_session_id,omitempty"` ReasoningEffort string `json:"reasoning_effort"` ServiceTier string `json:"service_tier"` ResponseServiceTier string `json:"response_service_tier,omitempty"`} type requestDetail struct { Timestamp time.Time `json:"timestamp"` LatencyMs int64 `json:"latency_ms"` TTFTMs int64 `json:"ttft_ms"` Source string `json:"source"` AuthIndex string `json:"auth_index"` AccessTokenHash string `json:"access_token_sha256,omitempty"` ClientIP string `json:"client_ip"` XForwardedFor string `json:"x_forwarded_for"` UserAgent string `json:"user_agent"` Tokens tokenStats `json:"tokens"` Failed bool `json:"failed"` Generate bool `json:"generate"` Stream bool `json:"stream"` Fail failDetail `json:"fail"` ResponseHeaders http.Header `json:"response_headers,omitempty"`} type tokenStats struct { InputTokens int64 `json:"input_tokens"` OutputTokens int64 `json:"output_tokens"` ReasoningTokens int64 `json:"reasoning_tokens"` CachedTokens int64 `json:"cached_tokens"` CacheReadTokens int64 `json:"cache_read_tokens"` CacheReadTokensPresent bool `json:"cache_read_tokens_present"` CacheCreationTokens int64 `json:"cache_creation_tokens"` TotalTokens int64 `json:"total_tokens"`} type failDetail struct { StatusCode int `json:"status_code"` Body string `json:"body"`} func resolveFail(ctx context.Context, record coreusage.Record, failed bool) failDetail { fail := failDetail{ StatusCode: record.Fail.StatusCode, Body: strings.TrimSpace(record.Fail.Body), } if !failed { return failDetail{StatusCode: 200} } if fail.StatusCode <= 0 { fail.StatusCode = internallogging.GetResponseStatus(ctx) } if fail.StatusCode <= 0 { fail.StatusCode = 500 } return fail} func resolveSuccess(ctx context.Context) bool { status := internallogging.GetResponseStatus(ctx) if status == 0 { return true } return status < httpStatusBadRequest} func resolveEndpoint(ctx context.Context) string { return strings.TrimSpace(internallogging.GetEndpoint(ctx))} const httpStatusBadRequest = 400
package redisqueue
import (
"context"
"encoding/json"
"net/http"
"strings"
"time"
internallogging "github.com/router-for-me/CLIProxyAPI/v7/internal/logging"
coresession "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/session"
coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage"
)
func init() {
coreusage.RegisterPlugin(&usageQueuePlugin{})
}
type usageQueuePlugin struct{}
func (p *usageQueuePlugin) HandleUsage(ctx context.Context, record coreusage.Record) {
if p == nil {
return
}
if !Enabled() || !UsageStatisticsEnabled() {
return
}
timestamp := record.RequestedAt
if timestamp.IsZero() {
timestamp = time.Now()
}
modelName := strings.TrimSpace(record.Model)
if modelName == "" {
modelName = "unknown"
}
aliasName := strings.TrimSpace(record.Alias)
if aliasName == "" {
aliasName = modelName
}
provider := strings.TrimSpace(record.Provider)
if provider == "" {
provider = "unknown"
}
executorType := strings.TrimSpace(record.ExecutorType)
if executorType == "" {
executorType = "unknown"
}
authType := strings.TrimSpace(record.AuthType)
if authType == "" {
authType = "unknown"
}
apiKey := strings.TrimSpace(record.APIKey)
requestID := strings.TrimSpace(internallogging.GetRequestID(ctx))
reasoningEffort := strings.TrimSpace(record.ReasoningEffort)
if reasoningEffort == "" {
reasoningEffort = coreusage.ReasoningEffortFromContext(ctx)
}
serviceTier := strings.TrimSpace(record.ServiceTier)
if serviceTier == "" {
serviceTier = strings.TrimSpace(record.RequestServiceTier)
}
if serviceTier == "" {
serviceTier = coreusage.ServiceTierFromContext(ctx)
}
responseServiceTier := strings.TrimSpace(record.ResponseServiceTier)
clientRequestMetadata := internallogging.GetClientRequestMetadata(ctx)
sessionID := strings.TrimSpace(record.SessionID)
parentSessionID := strings.TrimSpace(record.ParentSessionID)
if sessionID == "" {
sessionID = strings.TrimSpace(clientRequestMetadata.SessionID)
parentSessionID = strings.TrimSpace(clientRequestMetadata.ParentSessionID)
} else if parentSessionID == "" && sessionID == strings.TrimSpace(clientRequestMetadata.SessionID) {
parentSessionID = strings.TrimSpace(clientRequestMetadata.ParentSessionID)
}
sessionID = coresession.NormalizeToCanonicalUUID(sessionID)
parentSessionID = coresession.NormalizeToCanonicalUUID(parentSessionID)
if sessionID == "" || sessionID == parentSessionID {
parentSessionID = ""
}
usageDetail := coreusage.EnsureTokenBreakdownForProvider(record.Detail, record.Provider, record.ExecutorType)
tokens := tokenStats{
InputTokens: usageDetail.InputTokens,
OutputTokens: usageDetail.OutputTokens,
ReasoningTokens: usageDetail.ReasoningTokens,
CachedTokens: usageDetail.CachedTokens,
CacheReadTokens: usageDetail.CacheReadTokens,
CacheReadTokensPresent: true,
CacheCreationTokens: usageDetail.CacheCreationTokens,
TotalTokens: usageDetail.TotalTokens,
}
failed := record.Failed
if !failed {
failed = !resolveSuccess(ctx)
}
fail := resolveFail(ctx, record, failed)
stream := record.Stream
if !stream {
stream = coreusage.StreamFromContext(ctx)
}
detail := requestDetail{
Timestamp: timestamp,
LatencyMs: record.Latency.Milliseconds(),
TTFTMs: record.TTFT.Milliseconds(),
Source: record.Source,
AuthIndex: record.AuthIndex,
AccessTokenHash: record.AccessTokenSHA256,
ClientIP: clientRequestMetadata.ClientIP,
XForwardedFor: clientRequestMetadata.XForwardedFor,
UserAgent: clientRequestMetadata.UserAgent,
Tokens: tokens,
Failed: failed,
Generate: coreusage.GenerateEnabled(record.Generate),
Stream: stream,
Fail: fail,
ResponseHeaders: record.ResponseHeaders,
}
payload, err := json.Marshal(queuedUsageDetail{
requestDetail: detail,
AccountingVersion: coreusage.TokenAccountingSchemaVersion,
TokenBreakdown: usageDetail.TokenBreakdown,
Provider: provider,
ExecutorType: executorType,
Model: modelName,
Alias: aliasName,
Endpoint: resolveEndpoint(ctx),
AuthType: authType,
APIKey: apiKey,
RequestID: requestID,
SessionID: sessionID,
ParentSessionID: parentSessionID,
ReasoningEffort: reasoningEffort,
ServiceTier: serviceTier,
ResponseServiceTier: responseServiceTier,
})
if err != nil {
return
}
Enqueue(payload)
}
type queuedUsageDetail struct {
requestDetail
AccountingVersion int `json:"accounting_version"`
TokenBreakdown coreusage.TokenBreakdown `json:"token_breakdown"`
Provider string `json:"provider"`
ExecutorType string `json:"executor_type"`
Model string `json:"model"`
Alias string `json:"alias"`
Endpoint string `json:"endpoint"`
AuthType string `json:"auth_type"`
APIKey string `json:"api_key"`
RequestID string `json:"request_id"`
SessionID string `json:"session_id,omitempty"`
ParentSessionID string `json:"parent_session_id,omitempty"`
ReasoningEffort string `json:"reasoning_effort"`
ServiceTier string `json:"service_tier"`
ResponseServiceTier string `json:"response_service_tier,omitempty"`
}
type requestDetail struct {
Timestamp time.Time `json:"timestamp"`
LatencyMs int64 `json:"latency_ms"`
TTFTMs int64 `json:"ttft_ms"`
Source string `json:"source"`
AuthIndex string `json:"auth_index"`
AccessTokenHash string `json:"access_token_sha256,omitempty"`
ClientIP string `json:"client_ip"`
XForwardedFor string `json:"x_forwarded_for"`
UserAgent string `json:"user_agent"`
Tokens tokenStats `json:"tokens"`
Failed bool `json:"failed"`
Generate bool `json:"generate"`
Stream bool `json:"stream"`
Fail failDetail `json:"fail"`
ResponseHeaders http.Header `json:"response_headers,omitempty"`
}
type tokenStats struct {
InputTokens int64 `json:"input_tokens"`
OutputTokens int64 `json:"output_tokens"`
ReasoningTokens int64 `json:"reasoning_tokens"`
CachedTokens int64 `json:"cached_tokens"`
CacheReadTokens int64 `json:"cache_read_tokens"`
CacheReadTokensPresent bool `json:"cache_read_tokens_present"`
CacheCreationTokens int64 `json:"cache_creation_tokens"`
TotalTokens int64 `json:"total_tokens"`
}
type failDetail struct {
StatusCode int `json:"status_code"`
Body string `json:"body"`
}
func resolveFail(ctx context.Context, record coreusage.Record, failed bool) failDetail {
fail := failDetail{
StatusCode: record.Fail.StatusCode,
Body: strings.TrimSpace(record.Fail.Body),
}
if !failed {
return failDetail{StatusCode: 200}
}
if fail.StatusCode <= 0 {
fail.StatusCode = internallogging.GetResponseStatus(ctx)
}
if fail.StatusCode <= 0 {
fail.StatusCode = 500
}
return fail
}
func resolveSuccess(ctx context.Context) bool {
status := internallogging.GetResponseStatus(ctx)
if status == 0 {
return true
}
return status < httpStatusBadRequest
}
func resolveEndpoint(ctx context.Context) string {
return strings.TrimSpace(internallogging.GetEndpoint(ctx))
}
const httpStatusBadRequest = 400