From b2e500ff51c0d38020895269c277d4d211120300 Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Mon, 9 Nov 2026 00:41:48 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=91=E9=A2=9D=20decimal=20?= =?UTF-8?q?=E7=B2=BE=E7=A1=AE=E8=AE=A1=E8=B4=B9=E3=80=81failover=20?= =?UTF-8?q?=E6=8C=89=E5=A4=B1=E8=B4=A5=E7=B1=BB=E5=9E=8B=E5=B0=81=E9=A1=B6?= =?UTF-8?q?=E3=80=81=E6=B8=A0=E9=81=93=E5=B0=8F=E6=97=B6=E6=A1=B6=E9=A2=84?= =?UTF-8?q?=E8=81=9A=E5=90=88=E3=80=81=E5=88=86=E5=B1=82=E5=AE=88=E6=8A=A4?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 计费：19 个金额字段 float64→decimal.Decimal，DB 列保持 numeric(20,8) 零迁移； API 金额仍为 JSON number（MarshalJSONWithoutQuotes），前端零改动 - failover：保留时间预算，新增 OT_PROXY_FAILOVER_CAPS_NET/SERVER/RATE/QUOTA 按失败类型封顶换渠道次数（429/配额类瞬时失败时间预算形同虚设） - 预聚合：新增 usage_channel_hourlies（schema v10），记账事务内同步 upsert， 长窗口渠道统计走桶不再扫明细，spent_last_30d 改扫 usage_daily，v10 一次性回填幂等 - 守护：internal/arch_test.go 用 go/parser 强制分层依赖（fail-closed） --- .env.example | 7 + server/cmd/server/main.go | 2 +- server/go.mod | 1 + server/go.sum | 2 + server/internal/api/admin.go | 9 +- server/internal/api/admin_models.go | 45 ++--- server/internal/api/admin_stats.go | 119 +++++++++----- server/internal/api/auth.go | 3 +- server/internal/api/keys_test.go | 5 +- server/internal/api/usage.go | 7 +- server/internal/api/user.go | 12 +- server/internal/app/app.go | 7 +- server/internal/arch_test.go | 122 ++++++++++++++ server/internal/cli/serve.go | 4 + server/internal/config/config.go | 41 +++-- server/internal/proxy/gateway.go | 44 ++++- server/internal/proxy/passthrough.go | 83 ++++++---- server/internal/proxy/proxy_test.go | 97 ++++++++++- server/internal/proxy/usage_refine_test.go | 16 +- server/internal/store/backfill_test.go | 64 +++++++ server/internal/store/db.go | 45 ++++- server/internal/store/models.go | 183 ++++++++++++--------- server/internal/usage/retention_test.go | 11 +- server/internal/usage/usage.go | 79 ++++++++- server/internal/usage/usage_test.go | 86 ++++++++-- 25 files changed, 841 insertions(+), 253 deletions(-) create mode 100644 server/internal/arch_test.go create mode 100644 server/internal/store/backfill_test.go diff --git a/.env.example b/.env.example index a06ad34..7a5d584 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,13 @@ OT_PROXY_MAX_BODY_MB=32 # 故障转移给"第二个及以后的渠道"留的时间预算（0=不限）。 # 首个渠道不受它约束——长回答本就要跑几分钟，但不能让坏渠道把客户端拖到 N×timeout。 OT_PROXY_FAILOVER_BUDGET=30s +# 按失败类型限制"换渠道"次数（0=不限）：net=网络错误/超时、server=5xx、 +# rate=429 限流、quota=余额/配额。429/配额类失败瞬时返回，上面的时间预算对它们 +# 形同虚设，按次数封顶才有意义（如全渠道欠费时不至于把候选逐个打一遍）。 +OT_PROXY_FAILOVER_CAPS_NET=4 +OT_PROXY_FAILOVER_CAPS_SERVER=3 +OT_PROXY_FAILOVER_CAPS_RATE=2 +OT_PROXY_FAILOVER_CAPS_QUOTA=3 # 流式空闲多久补一行 SSE 注释心跳（0=关闭）。挂在 nginx/Cloudflare 后面时， # 上游长时间不出字（长思考）会被对端的空闲超时掐掉，心跳用来续住这条流。 OT_PROXY_STREAM_HEARTBEAT=15s diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index a6072c3..5c54537 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -6,4 +6,4 @@ import "github.com/openteam/server/internal/cli" func main() { cli.Execute() -} \ No newline at end of file +} diff --git a/server/go.mod b/server/go.mod index dfb01ca..3ddfe38 100644 --- a/server/go.mod +++ b/server/go.mod @@ -7,6 +7,7 @@ require ( github.com/glebarez/sqlite v1.11.0 github.com/go-webauthn/webauthn v0.17.4 github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/shopspring/decimal v1.6.0 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 github.com/tiktoken-go/tokenizer v0.8.1 diff --git a/server/go.sum b/server/go.sum index 5d29637..2606b87 100644 --- a/server/go.sum +++ b/server/go.sum @@ -112,6 +112,8 @@ github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncj github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= +github.com/shopspring/decimal v1.6.0 h1:DcTwjCj4tJPLuGZYsR9W0hyX9pQwMIkdo26Ig1T9xX4= +github.com/shopspring/decimal v1.6.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= diff --git a/server/internal/api/admin.go b/server/internal/api/admin.go index a7c5080..db056bc 100644 --- a/server/internal/api/admin.go +++ b/server/internal/api/admin.go @@ -11,6 +11,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -177,14 +178,14 @@ func (h *Handler) AdminAdjustBalance(c *gin.Context) { return } var req struct { - Amount float64 `json:"amount" binding:"required"` - Remark string `json:"remark"` + Amount decimal.Decimal `json:"amount" binding:"required"` + Remark string `json:"remark"` } if err := c.ShouldBindJSON(&req); err != nil { resp.Fail(c, http.StatusBadRequest, "invalid input: amount required") return } - if req.Amount == 0 { + if req.Amount.IsZero() { resp.Fail(c, http.StatusBadRequest, "amount must not be zero") return } @@ -204,7 +205,7 @@ func (h *Handler) AdminAdjustBalance(c *gin.Context) { return tx.Create(&store.BalanceLog{ UserID: u.ID, Change: req.Amount, - BalanceAfter: u.Balance + req.Amount, + BalanceAfter: u.Balance.Add(req.Amount), Type: store.BalanceTypeAdminAdjust, RefID: ref, Remark: req.Remark, diff --git a/server/internal/api/admin_models.go b/server/internal/api/admin_models.go index e2b949a..023f766 100644 --- a/server/internal/api/admin_models.go +++ b/server/internal/api/admin_models.go @@ -9,6 +9,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -38,7 +39,7 @@ func (h *Handler) AdminModels(c *gin.Context) { }) } used := len(chs) > 0 - needsPricing := used && m.InputPrice == 0 && m.OutputPrice == 0 && m.CacheReadPrice == 0 + needsPricing := used && m.InputPrice.IsZero() && m.OutputPrice.IsZero() && m.CacheReadPrice.IsZero() denied := containsStr(deny, m.Name) || (len(allow) > 0 && !containsStr(allow, m.Name)) out = append(out, gin.H{ "id": m.ID, "name": m.Name, @@ -81,7 +82,7 @@ func (h *Handler) AdminModels(c *gin.Context) { usedIDs[u.ModelID] = true } for _, m := range ms { - if usedIDs[m.ID] && m.InputPrice == 0 && m.OutputPrice == 0 && m.CacheReadPrice == 0 { + if usedIDs[m.ID] && m.InputPrice.IsZero() && m.OutputPrice.IsZero() && m.CacheReadPrice.IsZero() { unpriced++ } } @@ -121,16 +122,16 @@ func containsStr(list []string, s string) bool { // AdminCreateModel POST /api/v1/admin/models func (h *Handler) AdminCreateModel(c *gin.Context) { var req struct { - Name string `json:"name" binding:"required,min=1,max=128"` - InputPrice float64 `json:"input_price"` - OutputPrice float64 `json:"output_price"` - CacheReadPrice float64 `json:"cache_read_price"` - CacheWritePrice float64 `json:"cache_write_price"` - ContextLength int `json:"context_length"` - SupportsTools bool `json:"supports_tools"` - SupportsVision bool `json:"supports_vision"` - SupportsReasoning bool `json:"supports_reasoning"` - Enabled *bool `json:"enabled"` + Name string `json:"name" binding:"required,min=1,max=128"` + InputPrice decimal.Decimal `json:"input_price"` + OutputPrice decimal.Decimal `json:"output_price"` + CacheReadPrice decimal.Decimal `json:"cache_read_price"` + CacheWritePrice decimal.Decimal `json:"cache_write_price"` + ContextLength int `json:"context_length"` + SupportsTools bool `json:"supports_tools"` + SupportsVision bool `json:"supports_vision"` + SupportsReasoning bool `json:"supports_reasoning"` + Enabled *bool `json:"enabled"` } if err := c.ShouldBindJSON(&req); err != nil { resp.Fail(c, http.StatusBadRequest, "invalid input: "+err.Error()) @@ -159,16 +160,16 @@ func (h *Handler) AdminUpdateModel(c *gin.Context) { return } var req struct { - InputPrice *float64 `json:"input_price"` - OutputPrice *float64 `json:"output_price"` - CacheReadPrice *float64 `json:"cache_read_price"` - CacheWritePrice *float64 `json:"cache_write_price"` - ContextLength *int `json:"context_length"` - SupportsTools *bool `json:"supports_tools"` - SupportsVision *bool `json:"supports_vision"` - SupportsReasoning *bool `json:"supports_reasoning"` - Enabled *bool `json:"enabled"` - Sort *int `json:"sort"` + InputPrice *decimal.Decimal `json:"input_price"` + OutputPrice *decimal.Decimal `json:"output_price"` + CacheReadPrice *decimal.Decimal `json:"cache_read_price"` + CacheWritePrice *decimal.Decimal `json:"cache_write_price"` + ContextLength *int `json:"context_length"` + SupportsTools *bool `json:"supports_tools"` + SupportsVision *bool `json:"supports_vision"` + SupportsReasoning *bool `json:"supports_reasoning"` + Enabled *bool `json:"enabled"` + Sort *int `json:"sort"` } if err := c.ShouldBindJSON(&req); err != nil { resp.Fail(c, http.StatusBadRequest, "invalid input") diff --git a/server/internal/api/admin_stats.go b/server/internal/api/admin_stats.go index 8a7e6e3..b26e0c6 100644 --- a/server/internal/api/admin_stats.go +++ b/server/internal/api/admin_stats.go @@ -10,6 +10,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // AdminStatsOverview GET /api/v1/admin/stats/overview — 运营总览。 @@ -29,7 +30,7 @@ func (h *Handler) AdminStatsOverview(c *gin.Context) { // 全局今日/本月汇总（跨用户） var todayReq, monthReq int64 - var todayCost, monthCost float64 + var todayCost, monthCost decimal.Decimal var todayTokens, monthTokens int64 h.a.DB.Model(&store.UsageDaily{}).Where("date = ?", today). Select("COALESCE(SUM(requests),0), COALESCE(SUM(cost),0), COALESCE(SUM(input_tokens+output_tokens+cache_read_tokens),0)"). @@ -42,7 +43,7 @@ func (h *Handler) AdminStatsOverview(c *gin.Context) { var days []struct { Date string Req int64 - Cost float64 + Cost decimal.Decimal } h.a.DB.Model(&store.UsageDaily{}). Select("date, SUM(requests) req, SUM(cost) cost"). @@ -64,6 +65,10 @@ func (h *Handler) AdminStatsOverview(c *gin.Context) { // AdminChannelStats GET /api/v1/admin/channels/stats?window=1h — 按渠道聚合近一段的 // 请求量、成功率、错误码分布与延迟分位。定位"哪条渠道在掉"不该靠翻日志。 +// +// 短窗口（5m/1h）扫 usage_logs 保留精确延迟分位与错误码分布；长窗口（6h/24h/7d） +// 走 usage_channel_hourly 小时桶——明细扫 5 万行封顶在 7d 下只剩个尾巴，桶是全量， +// 且不受明细保留期清理影响。长窗口延迟给 avg/max（p50/p95 以 avg 填充，桶里没有分布）。 func (h *Handler) AdminChannelStats(c *gin.Context) { win := c.DefaultQuery("window", "1h") d, ok := statWindows[win] @@ -72,40 +77,11 @@ func (h *Handler) AdminChannelStats(c *gin.Context) { } since := time.Now().UTC().Add(-d) - // 只取窗口内最近这么多行，避免大窗口把内存吃光（超出部分不参与分位计算） - const maxRows = 50000 - var rows []channelStatRow - h.a.DB.Model(&store.UsageLog{}). - Select("channel_id, status, latency_ms, cost, "+ - "(input_tokens + output_tokens + cache_read_tokens) AS tokens, error_code, created_at"). - Where("created_at >= ? AND channel_id > 0", since). - Order("id DESC").Limit(maxRows).Scan(&rows) - agg := map[uint64]*chanAgg{} - for _, r := range rows { - a := agg[r.ChannelID] - if a == nil { - a = &chanAgg{lat: make([]int, 0, 64), errs: map[string]int{}} - agg[r.ChannelID] = a - } - a.requests++ - a.cost += r.Cost - a.tokens += r.Tokens - switch r.Status { - case store.UsageStatusSuccess: - a.success++ - a.lat = append(a.lat, r.LatencyMS) // 分位只按正常完成的请求算，中断/失败的延迟不代表体感 - case store.UsageStatusCanceled: - a.canceled++ - default: - a.err++ - if r.ErrorCode != nil && *r.ErrorCode != "" { - a.errs[*r.ErrorCode]++ - } - } - if r.CreatedAt.After(a.lastAt) { - a.lastAt = r.CreatedAt - } + if d > time.Hour { + h.aggChannelStatsHourly(since, agg) + } else { + h.aggChannelStatsRaw(since, agg) } var chans []store.Channel @@ -137,6 +113,75 @@ func (h *Handler) AdminChannelStats(c *gin.Context) { resp.OK(c, gin.H{"window": win, "since": since, "items": out}) } +// aggChannelStatsRaw 短窗口：扫明细（限最近 5 万行防内存失控），保留精确分位与错误码。 +func (h *Handler) aggChannelStatsRaw(since time.Time, agg map[uint64]*chanAgg) { + const maxRows = 50000 + var rows []channelStatRow + h.a.DB.Model(&store.UsageLog{}). + Select("channel_id, status, latency_ms, cost, "+ + "(input_tokens + output_tokens + cache_read_tokens) AS tokens, error_code, created_at"). + Where("created_at >= ? AND channel_id > 0", since). + Order("id DESC").Limit(maxRows).Scan(&rows) + + for _, r := range rows { + a := agg[r.ChannelID] + if a == nil { + a = &chanAgg{lat: make([]int, 0, 64), errs: map[string]int{}} + agg[r.ChannelID] = a + } + a.requests++ + a.cost = a.cost.Add(r.Cost) + a.tokens += r.Tokens + switch r.Status { + case store.UsageStatusSuccess: + a.success++ + a.lat = append(a.lat, r.LatencyMS) // 分位只按正常完成的请求算，中断/失败的延迟不代表体感 + case store.UsageStatusCanceled: + a.canceled++ + default: + a.err++ + if r.ErrorCode != nil && *r.ErrorCode != "" { + a.errs[*r.ErrorCode]++ + } + } + if r.CreatedAt.After(a.lastAt) { + a.lastAt = r.CreatedAt + } + } +} + +// aggChannelStatsHourly 长窗口：读渠道×小时桶。p50/p95 用 avg 近似（桶里没有延迟分布）， +// 错误码分布桶里没有、置空；last_at 取最新桶的起始小时（精度 1h）。 +func (h *Handler) aggChannelStatsHourly(since time.Time, agg map[uint64]*chanAgg) { + var rows []store.UsageChannelHourly + // 桶粒度 1h：包含 since 落在的那个小时（边界桶会带上窗口前的一小段，可接受） + h.a.DB.Where("hour >= ?", since.UTC().Format("2006-01-02T15")).Find(&rows) + + for _, r := range rows { + a := agg[r.ChannelID] + if a == nil { + a = &chanAgg{lat: make([]int, 0, 64), errs: map[string]int{}} + agg[r.ChannelID] = a + } + a.requests += r.Requests + a.success += r.Success + a.canceled += r.Canceled + a.err += r.Errors + a.cost = a.cost.Add(r.Cost) + a.tokens += r.InputTokens + r.OutputTokens + r.CacheReadTokens + if r.Success > 0 { + avg := int(r.LatencySum / r.Success) + a.lat = append(a.lat, avg) + } + if r.LatencyMax > 0 { + a.lat = append(a.lat, r.LatencyMax) // 参与分位计算，max 取两者较大值 + } + if bucketAt, err := time.Parse("2006-01-02T15", r.Hour); err == nil && bucketAt.After(a.lastAt) { + a.lastAt = bucketAt + } + } +} + // statWindows 支持的统计窗口。 var statWindows = map[string]time.Duration{ "5m": 5 * time.Minute, @@ -150,7 +195,7 @@ type channelStatRow struct { ChannelID uint64 Status string LatencyMS int - Cost float64 + Cost decimal.Decimal Tokens int64 ErrorCode *string CreatedAt time.Time @@ -158,7 +203,7 @@ type channelStatRow struct { type chanAgg struct { requests, success, err, canceled int64 - cost float64 + cost decimal.Decimal tokens int64 lat []int errs map[string]int diff --git a/server/internal/api/auth.go b/server/internal/api/auth.go index 7b2f114..b7e501d 100644 --- a/server/internal/api/auth.go +++ b/server/internal/api/auth.go @@ -17,6 +17,7 @@ import ( "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/proxy" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -133,7 +134,7 @@ func (h *Handler) Register(c *gin.Context) { Email: email, PasswordHash: hash, Role: store.RoleUser, - Balance: 5.0, // 新用户赠送体验余额（可通过 admin 调整） + Balance: decimal.NewFromFloat(5.0), // 新用户赠送体验余额（可通过 admin 调整） Status: store.UserStatusActive, } if req.InviteCode != "" { diff --git a/server/internal/api/keys_test.go b/server/internal/api/keys_test.go index 41eda12..a6cb862 100644 --- a/server/internal/api/keys_test.go +++ b/server/internal/api/keys_test.go @@ -13,6 +13,7 @@ import ( "github.com/openteam/server/internal/api/middleware" "github.com/openteam/server/internal/app" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -37,7 +38,7 @@ func newKeysTestEnv(t *testing.T) *keysTestEnv { sqlDB.Close() }) u := store.User{Username: "u1", Email: "u1@x.com", PasswordHash: "x", - Role: store.RoleUser, Balance: 100, Status: store.UserStatusActive} + Role: store.RoleUser, Balance: decimal.NewFromInt(100), Status: store.UserStatusActive} if err := db.Create(&u).Error; err != nil { t.Fatal(err) } @@ -126,7 +127,7 @@ func TestKeysLifecycle(t *testing.T) { func TestKeysUserIsolation(t *testing.T) { e := newKeysTestEnv(t) other := store.User{Username: "u2", Email: "u2@x.com", PasswordHash: "x", - Role: store.RoleUser, Balance: 0, Status: store.UserStatusActive} + Role: store.RoleUser, Balance: decimal.Zero, Status: store.UserStatusActive} e.db.Create(&other) var k store.APIKey k = store.APIKey{UserID: other.ID, Name: "theirs", KeyHash: "h2", KeyPrefix: "pre", diff --git a/server/internal/api/usage.go b/server/internal/api/usage.go index 5de908f..4a84e71 100644 --- a/server/internal/api/usage.go +++ b/server/internal/api/usage.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // UsageSummary GET /api/v1/usage/summary — 今日/本月汇总。 @@ -18,7 +19,7 @@ func (h *Handler) UsageSummary(c *gin.Context) { var todayReq, monthReq int64 var todayTok, monthTok int64 - var todayCost, monthCost float64 + var todayCost, monthCost decimal.Decimal h.a.DB.Model(&store.UsageDaily{}).Where("user_id = ? AND date = ?", u.ID, today). Select("COALESCE(SUM(requests),0), COALESCE(SUM(input_tokens+output_tokens+cache_read_tokens),0), COALESCE(SUM(cost),0)"). Row().Scan(&todayReq, &todayTok, &todayCost) @@ -47,7 +48,7 @@ func (h *Handler) UsageStats(c *gin.Context) { ModelID uint64 Requests int64 Tokens int64 - Cost float64 + Cost decimal.Decimal } q.Select("model_id, SUM(requests) requests, SUM(input_tokens+output_tokens+cache_read_tokens) tokens, SUM(cost) cost"). Group("model_id").Scan(&rows) @@ -64,7 +65,7 @@ func (h *Handler) UsageStats(c *gin.Context) { Date string Requests int64 Tokens int64 - Cost float64 + Cost decimal.Decimal } q.Select("date, SUM(requests) requests, SUM(input_tokens+output_tokens+cache_read_tokens) tokens, SUM(cost) cost"). Group("date").Order("date").Scan(&rows) diff --git a/server/internal/api/user.go b/server/internal/api/user.go index 01b3202..2387c03 100644 --- a/server/internal/api/user.go +++ b/server/internal/api/user.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/pkg/resp" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // UserProfile GET /api/v1/user/profile @@ -16,11 +17,14 @@ func (h *Handler) UserProfile(c *gin.Context) { } // UserBalance GET /api/v1/user/balance — 余额 + 近 30 日消耗。 +// 消耗扫 usage_daily（预聚合）而非 usage_logs：语义等价（usage_daily 只含 cost>0 的 +// 计费行，含 canceled——canceled 请求 token 已实际计费），且不受明细保留期清理影响。 func (h *Handler) UserBalance(c *gin.Context) { u, _ := userFromContext(c) - var spent float64 - h.a.DB.Model(&store.UsageLog{}). - Where("user_id = ? AND status = ? AND created_at >= ?", u.ID, store.UsageStatusSuccess, time.Now().Add(-30*24*time.Hour)). + cutoff := time.Now().UTC().AddDate(0, 0, -30).Format("2006-01-02") + var spent decimal.Decimal + h.a.DB.Model(&store.UsageDaily{}). + Where("user_id = ? AND date >= ?", u.ID, cutoff). Select("COALESCE(SUM(cost),0)").Scan(&spent) resp.OK(c, gin.H{ "balance": u.Balance, @@ -39,7 +43,7 @@ func (h *Handler) availableModelCount() int64 { func (h *Handler) todayUsage(c *gin.Context, userID uint64) gin.H { var requests int64 var tokens int64 - var cost float64 + var cost decimal.Decimal today := time.Now().UTC().Format("2006-01-02") h.a.DB.Model(&store.UsageDaily{}). Where("user_id = ? AND date = ?", userID, today). diff --git a/server/internal/app/app.go b/server/internal/app/app.go index 56fcb8b..5791a66 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -15,6 +15,7 @@ import ( "github.com/openteam/server/internal/pkg/ratelimit" "github.com/openteam/server/internal/store" "github.com/openteam/server/internal/usage" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -115,7 +116,7 @@ func (a *App) Seed() error { Email: envOr("OT_ADMIN_EMAIL", "admin@localhost"), PasswordHash: hash, Role: store.RoleAdmin, - Balance: 1000, // 初始余额，便于联调；生产由充值/调整决定 + Balance: decimal.NewFromInt(1000), // 初始余额，便于联调；生产由充值/调整决定 Status: store.UserStatusActive, } if err := a.DB.Create(&admin).Error; err != nil { @@ -152,8 +153,8 @@ func (a *App) Seed() error { // 默认模型 + 绑定 m := store.Model{ Name: a.Cfg.Proxy.DefaultModel, - InputPrice: 0.15, // 每百万 token，示例价 - OutputPrice: 0.60, + InputPrice: decimal.NewFromFloat(0.15), // 每百万 token，示例价 + OutputPrice: decimal.NewFromFloat(0.60), Enabled: true, } if err := a.DB.Create(&m).Error; err == nil { diff --git a/server/internal/arch_test.go b/server/internal/arch_test.go new file mode 100644 index 0000000..fef3479 --- /dev/null +++ b/server/internal/arch_test.go @@ -0,0 +1,122 @@ +// 架构守护测试：用 AST 扫描 internal 下所有非测试代码的 import，强制分层依赖规则。 +// +// 目标分层（与实际依赖图一致，违规即测试失败）： +// +// pkg/*、config、store ← 叶子层，不得依赖任何 internal 包 +// channel / usage / passkey ← 只可依赖 store + pkg/* +// proxy ← 只可依赖 store、channel、usage、proxy/*、pkg/* +// api ← 不得直接 import channel/usage（必须经 app 持有） +// app ← 组合根：channel、config、passkey、pkg/*、store、usage +// cli ← 顶层装配，不限制 +// +// 本项目没有 arch-go 这类重型工具，一个 go/parser 扫描（毫秒级）足够守住红线； +// 新增包时在这里登记规则，未登记的包默认禁止一切 internal 依赖（fail-closed）。 +package internal_test + +import ( + "go/parser" + "go/token" + "io/fs" + "path/filepath" + "sort" + "strings" + "testing" +) + +const modulePrefix = "github.com/openteam/server/internal/" + +// layerRules 相对 internal 的包目录 → 允许依赖的 internal 包前缀（空切片 = 禁止一切）。 +// 子包继承最长前缀匹配的规则（如 proxy/convert 按 proxy 的规则校验）。 +var layerRules = map[string][]string{ + "config": {}, + "store": {}, + "pkg": {}, + "channel": {"store", "pkg"}, + "usage": {"store", "pkg"}, + "passkey": {"store", "pkg"}, + "proxy": {"store", "channel", "usage", "proxy", "pkg"}, + "api": {"app", "proxy", "passkey", "store", "api", "pkg"}, + "app": {"channel", "config", "passkey", "store", "usage", "pkg"}, + "cli": nil, // nil = 顶层装配，不限制 +} + +// ruleFor 取目录的生效规则：精确匹配优先，否则最长前缀；未登记的包 fail-closed。 +func ruleFor(dir string) (allowed []string, unrestricted bool) { + if r, ok := layerRules[dir]; ok { + return r, r == nil + } + best := "" + for k := range layerRules { + if strings.HasPrefix(dir, k+"/") && len(k) > len(best) { + best = k + } + } + if best != "" { + r := layerRules[best] + return r, r == nil + } + return nil, false +} + +func dependsOn(imported, allowed string) bool { + return imported == allowed || strings.HasPrefix(imported, allowed+"/") +} + +func TestInternalDependencyLayers(t *testing.T) { + root := "." // 测试二进制的 cwd 即 server/internal + var violations []string + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + name := d.Name() + if name != "." && (strings.HasPrefix(name, ".") || name == "testdata" || name == "vendor") { + return filepath.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + dir := filepath.ToSlash(filepath.Dir(path)) + if dir == "." { + dir = "" + } + allowed, unrestricted := ruleFor(dir) + if unrestricted { + return nil + } + f, perr := parser.ParseFile(token.NewFileSet(), path, nil, parser.ImportsOnly) + if perr != nil { + t.Errorf("parse %s: %v", path, perr) + return nil + } + for _, imp := range f.Imports { + p := strings.Trim(imp.Path.Value, `"`) + if !strings.HasPrefix(p, modulePrefix) { + continue + } + target := strings.TrimPrefix(p, modulePrefix) + ok := false + for _, a := range allowed { + if dependsOn(target, a) { + ok = true + break + } + } + if !ok { + violations = append(violations, path+" → "+p) + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if len(violations) > 0 { + sort.Strings(violations) + t.Errorf("internal 分层依赖被破坏（%d 处）：\n%s\n新增合法依赖请更新 arch_test.go 的 layerRules", + len(violations), strings.Join(violations, "\n")) + } +} diff --git a/server/internal/cli/serve.go b/server/internal/cli/serve.go index eb26ae9..7f37ed5 100644 --- a/server/internal/cli/serve.go +++ b/server/internal/cli/serve.go @@ -16,10 +16,13 @@ import ( "github.com/openteam/server/internal/app" "github.com/openteam/server/internal/config" "github.com/openteam/server/internal/proxy" + "github.com/shopspring/decimal" ) // serve 启动 HTTP 服务（默认子命令）。 func serve() error { + // 金额序列化为 JSON number 而非字符串：前端 types（balance/cost 等均为 number）零改动 + decimal.MarshalJSONWithoutQuotes = true cfg, err := config.Load() if err != nil { return err @@ -37,6 +40,7 @@ func serve() error { GWTimeout: cfg.Proxy.Timeout, MaxBodyMB: cfg.Proxy.MaxBodyMB, FailoverBudget: cfg.Proxy.FailoverBudget, + FailoverSwitchCaps: cfg.Proxy.FailoverSwitchCaps, StreamHeartbeat: cfg.Proxy.StreamHeartbeat, FirstTokenWait: cfg.Proxy.FirstTokenWait, AffinityTTL: cfg.Proxy.AffinityTTL, diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 16b5b80..d34ffe5 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -81,13 +81,16 @@ type AuthConfig struct { } type ProxyConfig struct { - DefaultChannelName string // 首次启动自动创建的渠道名（如 openai） - UpstreamBaseURL string // 渠道 base_url 默认值 - UpstreamKey string // 渠道上游 key 默认值 - DefaultModel string // 渠道模型导入时使用的模型名 - Timeout time.Duration - MaxBodyMB int // /v1 请求体上限（MB），超限按客户端协议返回 413 - FailoverBudget time.Duration // 故障转移给"第二个及以后的渠道"留的时间预算，0=不限 + DefaultChannelName string // 首次启动自动创建的渠道名（如 openai） + UpstreamBaseURL string // 渠道 base_url 默认值 + UpstreamKey string // 渠道上游 key 默认值 + DefaultModel string // 渠道模型导入时使用的模型名 + Timeout time.Duration + MaxBodyMB int // /v1 请求体上限（MB），超限按客户端协议返回 413 + FailoverBudget time.Duration // 故障转移给"第二个及以后的渠道"留的时间预算，0=不限 + // FailoverSwitchCaps 按失败类型（net|server|rate|quota）限制换渠道次数，0=不限。 + // 429/配额类失败瞬时返回，时间预算对它们形同虚设，按次数封顶才有意义。 + FailoverSwitchCaps map[string]int StreamHeartbeat time.Duration // 流式空闲多久发一次 SSE 注释心跳，0=关闭 FirstTokenWait time.Duration // 流式放行前等内容的最长时间，0=不限；超时未出内容换下一个渠道 AffinityTTL time.Duration // 会话亲和 TTL，0=关闭；同一会话粘住上次应答的渠道以命中上游缓存 @@ -159,6 +162,10 @@ func Load() (*Config, error) { v.SetDefault("proxy.timeout", "300s") v.SetDefault("proxy.max_body_mb", 32) v.SetDefault("proxy.failover_budget", "30s") + v.SetDefault("proxy.failover_caps.net", 4) + v.SetDefault("proxy.failover_caps.server", 3) + v.SetDefault("proxy.failover_caps.rate", 2) + v.SetDefault("proxy.failover_caps.quota", 3) v.SetDefault("proxy.stream_heartbeat", "15s") v.SetDefault("proxy.first_token_wait", "0s") v.SetDefault("proxy.affinity_ttl", "5m") @@ -206,13 +213,19 @@ func Load() (*Config, error) { SaltLen: v.GetInt("auth.salt_len"), }, Proxy: ProxyConfig{ - DefaultChannelName: v.GetString("proxy.default_channel_name"), - UpstreamBaseURL: v.GetString("proxy.upstream_base_url"), - UpstreamKey: v.GetString("proxy.upstream_key"), - DefaultModel: v.GetString("proxy.default_model"), - Timeout: v.GetDuration("proxy.timeout"), - MaxBodyMB: v.GetInt("proxy.max_body_mb"), - FailoverBudget: v.GetDuration("proxy.failover_budget"), + DefaultChannelName: v.GetString("proxy.default_channel_name"), + UpstreamBaseURL: v.GetString("proxy.upstream_base_url"), + UpstreamKey: v.GetString("proxy.upstream_key"), + DefaultModel: v.GetString("proxy.default_model"), + Timeout: v.GetDuration("proxy.timeout"), + MaxBodyMB: v.GetInt("proxy.max_body_mb"), + FailoverBudget: v.GetDuration("proxy.failover_budget"), + FailoverSwitchCaps: map[string]int{ + "net": v.GetInt("proxy.failover_caps.net"), + "server": v.GetInt("proxy.failover_caps.server"), + "rate": v.GetInt("proxy.failover_caps.rate"), + "quota": v.GetInt("proxy.failover_caps.quota"), + }, StreamHeartbeat: v.GetDuration("proxy.stream_heartbeat"), FirstTokenWait: v.GetDuration("proxy.first_token_wait"), AffinityTTL: v.GetDuration("proxy.affinity_ttl"), diff --git a/server/internal/proxy/gateway.go b/server/internal/proxy/gateway.go index 821f5ec..b8832d7 100644 --- a/server/internal/proxy/gateway.go +++ b/server/internal/proxy/gateway.go @@ -46,6 +46,9 @@ type Gateway struct { // 故障转移给"第二个及以后的渠道"留的时间预算；超了就直接返回最后一个错误， // 不限制首个渠道（长回答本来就可能跑很久）。 failoverBudget time.Duration + // 按失败类型限制换渠道次数（0=不限）：429/配额类失败是瞬时返回， + // 时间预算对它们形同虚设，按次数封顶才有意义。 + failoverCaps map[channel.FailKind]int // 流式空闲多久补一行 SSE 注释心跳（0=关闭）。反向代理的空闲超时会掐掉长思考的流。 streamHeartbeat time.Duration // 流式放行前等多久内容（0=不限制）。超时未出内容取消本次尝试换下一个渠道。 @@ -219,14 +222,36 @@ func contains(list []string, s string) bool { type Options struct { UserRPS int LogRaw bool - GWTimeout time.Duration // 上游超时兜底（渠道 timeout_ms 优先） - MaxBodyMB int // /v1 请求体上限（MB） - FailoverBudget time.Duration // 故障转移重试的时间预算 - StreamHeartbeat time.Duration // 流式空闲心跳间隔 - FirstTokenWait time.Duration // 流式放行前等内容的最长时间（0=不限） - AffinityTTL time.Duration // 会话亲和 TTL（0=关闭） - ChannelAcquireWait time.Duration // 渠道满载排队时长（0=立刻跳过） - AuthCacheTTL time.Duration // 鉴权缓存 TTL（0=关闭） + GWTimeout time.Duration // 上游超时兜底（渠道 timeout_ms 优先） + MaxBodyMB int // /v1 请求体上限（MB） + FailoverBudget time.Duration // 故障转移重试的时间预算 + FailoverSwitchCaps map[string]int // 按失败类型（net|server|rate|quota）限制换渠道次数，0=不限 + StreamHeartbeat time.Duration // 流式空闲心跳间隔 + FirstTokenWait time.Duration // 流式放行前等内容的最长时间（0=不限） + AffinityTTL time.Duration // 会话亲和 TTL（0=关闭） + ChannelAcquireWait time.Duration // 渠道满载排队时长（0=立刻跳过） + AuthCacheTTL time.Duration // 鉴权缓存 TTL（0=关闭） +} + +// failoverCapsFrom 把字符串键的失败类型上限配置转成 FailKind 键；未知键忽略（配置容错）。 +func failoverCapsFrom(m map[string]int) map[channel.FailKind]int { + if len(m) == 0 { + return nil + } + out := make(map[channel.FailKind]int, len(m)) + for k, v := range m { + switch k { + case "net": + out[channel.FailNet] = v + case "server": + out[channel.FailServer] = v + case "rate": + out[channel.FailRate] = v + case "quota": + out[channel.FailQuota] = v + } + } + return out } func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ratelimit.Limiter, o Options) *Gateway { @@ -241,6 +266,7 @@ func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ra gwTimeout: o.GWTimeout, maxBodyMB: o.MaxBodyMB, failoverBudget: o.FailoverBudget, + failoverCaps: failoverCapsFrom(o.FailoverSwitchCaps), streamHeartbeat: o.StreamHeartbeat, firstTokenWait: o.FirstTokenWait, affinityTTL: o.AffinityTTL, @@ -493,7 +519,7 @@ func (g *Gateway) resolveUser(c *gin.Context) (*store.User, bool) { // checkBalance 余额不足返回 402（PLANNING §4.4.3）。 func (g *Gateway) checkBalance(c *gin.Context, u *store.User) bool { - if u.Balance <= 0 { + if !u.Balance.IsPositive() { apiError(c, http.StatusPaymentRequired, "insufficient_balance", "Insufficient balance. Please recharge or contact admin.") return false } diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index c85a843..333c930 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -20,6 +20,7 @@ import ( "github.com/openteam/server/internal/pkg/tokenizer" "github.com/openteam/server/internal/proxy/convert" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // newTraceID 生成请求 trace（用于日志与记账幂等 ref）。 @@ -282,13 +283,25 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto // 那不是「上游坏了」（502），而是「我方满了」（503 + Retry-After）。 saturated := false startAll := time.Now() + // 换渠道次数按失败类型计数：cap 到了就不再为同类失败换渠道（配置见 failover_caps）。 + switches := map[channel.FailKind]int{} + var lastKind channel.FailKind // 从选中的渠道开始遍历，到末尾后再从头遍历到选中渠道之前 for offset := 0; offset < len(cands); offset++ { // 重试预算只约束第二个及以后的渠道：首个渠道不受限（长回答本就要跑几分钟）， // 但坏渠道一个接一个地等下去会让客户端挂到 N×timeout。 - if offset > 0 && g.failoverBudget > 0 && time.Since(startAll) > g.failoverBudget { - lastBody = []byte("failover budget exhausted: " + string(lastBody)) - break + if offset > 0 { + if g.failoverBudget > 0 && time.Since(startAll) > g.failoverBudget { + lastBody = []byte("failover budget exhausted: " + string(lastBody)) + break + } + // 429/配额类失败瞬时返回，时间预算对它们形同虚设；按同类失败次数封顶。 + // switches 记该类型已发生的失败次数（= 已换渠道次数），本次是第 + // switches+1 次换渠道，超过上限即停。 + if cap := g.failoverCaps[lastKind]; cap > 0 && switches[lastKind] > cap { + lastBody = []byte("failover switch cap reached (" + lastKind.String() + "): " + string(lastBody)) + break + } } idx := (startIdx + offset) % len(cands) cand := cands[idx] @@ -314,7 +327,7 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto } // 失败/成功计数由 proxyOne 就地报：只有它读得到上游响应头（429 的 Retry-After） lastCh = ch - done, served, retry, st, b := g.proxyOne(c, ch, plan, stream, sink, sh, clientProto) + done, served, retry, st, b, fkind := g.proxyOne(c, ch, plan, stream, sink, sh, clientProto) release() if done { if served && hasAff { @@ -336,6 +349,8 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto return } lastStatus, lastBody = st, b + switches[fkind]++ + lastKind = fkind } // 全部候选都没打出去、且都是被并发满载挡住的：503 让客户端稍后重试， // 而不是 502（那意味着上游故障，客户端通常会直接放弃）。 @@ -404,11 +419,13 @@ func streamErrorEvent(proto string, status int, msg string) []byte { // served=该渠道成功应答（会话亲和据此绑定）；retry=是否可安全换渠道重试； // status+respBody=失败信息。 // sh 为流式容错缓冲（非流式为 nil）；sink 为请求级记账 sink。 -func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink, sh *streamHold, clientProto string) (done bool, served bool, retry bool, status int, respBody []byte) { +// 最后一个返回值 fkind 是本次尝试的失败类型（仅 retry=true 时被 doProxy 消费， +// 用于按失败类型统计换渠道次数）。 +func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink, sh *streamHold, clientProto string) (done bool, served bool, retry bool, status int, respBody []byte, fkind channel.FailKind) { upKey, err := g.ch.UpstreamKey(ch) if err != nil { g.ch.ReportFailure(ch, channel.FailNet, 0) - return false, false, true, http.StatusInternalServerError, []byte("failed to decrypt channel key") + return false, false, true, http.StatusInternalServerError, []byte("failed to decrypt channel key"), channel.FailNet } upBody := plan.body @@ -449,7 +466,7 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan req, err := http.NewRequestWithContext(tryCtx, http.MethodPost, upstreamURL(ch, plan.proto, plan.path), bytes.NewReader(upBody)) if err != nil { - return false, false, false, http.StatusInternalServerError, []byte("failed to build upstream request") + return false, false, false, http.StatusInternalServerError, []byte("failed to build upstream request"), channel.FailNet } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+upKey) @@ -478,10 +495,10 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan // 客户端断开导致的取消：就地记账并终结本次请求（不计入渠道失败，也不换渠道） if c.Request.Context().Err() != nil { g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") - return true, false, false, 0, nil + return true, false, false, 0, nil, channel.FailNet } g.ch.ReportFailure(ch, channel.FailNet, 0) // 网络错误/超时 - return false, false, true, http.StatusBadGateway, []byte("upstream request failed: " + err.Error()) + return false, false, true, http.StatusBadGateway, []byte("upstream request failed: " + err.Error()), channel.FailNet } defer resp.Body.Close() @@ -492,9 +509,9 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan kind, retryable := classifyUpstreamError(resp.StatusCode, errBody) if retryable { g.ch.ReportFailure(ch, kind, retryAfter(resp.Header)) - return false, false, true, resp.StatusCode, errBody + return false, false, true, resp.StatusCode, errBody, kind } - return false, false, false, resp.StatusCode, errBody + return false, false, false, resp.StatusCode, errBody, kind } g.ch.ReportSuccess(ch) @@ -510,7 +527,7 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan asked, got = "流式", "整包响应" } return false, false, true, http.StatusBadGateway, - []byte(ch.Name + " answered a " + asked + " request with " + got + " (didNotStream)") + []byte(ch.Name + " answered a " + asked + " request with " + got + " (didNotStream)"), channel.FailServer } if stream { @@ -523,20 +540,20 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan // copyAndCapture 非流式：先读完整上游响应再写客户端——响应头与正文都在"确定成功"后 // 才写出，读失败（网络错误/超时）发生在写出之前，可换渠道重试。 // 此前实现先写 200 头再读，超时只能给客户端留一个 200 空响应。 -func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, resp *http.Response, start time.Time, bodyConv func([]byte) ([]byte, error), sink *usageSink) (done bool, served bool, retry bool, status int, respBody []byte) { +func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, resp *http.Response, start time.Time, bodyConv func([]byte) ([]byte, error), sink *usageSink) (done bool, served bool, retry bool, status int, respBody []byte, fkind channel.FailKind) { data, err := io.ReadAll(resp.Body) if err != nil { switch { case c.Request.Context().Err() != nil: // 客户端断开：就地记账终结 g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") - return true, false, false, 0, nil + return true, false, false, 0, nil, channel.FailNet case errors.Is(err, context.DeadlineExceeded): g.ch.ReportFailure(ch, channel.FailNet, 0) - return false, false, true, http.StatusBadGateway, []byte(ch.Name + " upstream timeout") + return false, false, true, http.StatusBadGateway, []byte(ch.Name + " upstream timeout"), channel.FailNet default: g.ch.ReportFailure(ch, channel.FailNet, 0) - return false, false, true, http.StatusBadGateway, []byte(ch.Name + " read error: " + err.Error()) + return false, false, true, http.StatusBadGateway, []byte(ch.Name + " read error: " + err.Error()), channel.FailNet } } if usageRaw := extractUsage(data); usageRaw != nil && sink != nil { @@ -558,7 +575,7 @@ func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, resp *http.R c.Set("raw_response", capRaw(data, rawResponseCap)) // 上游原始响应（未转换） } g.finishUsage(c, ch, start, store.UsageStatusSuccess, "") - return true, true, false, 0, nil + return true, true, false, 0, nil, channel.FailNet } // streamHoldCopy 流式转发：边读上游 SSE 边经 hold 写客户端（放行前缓冲）。 @@ -568,7 +585,7 @@ func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, resp *http.R // 失败渠道的 message_start 用量不能计入最终账单。 func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Channel, r io.Reader, start time.Time, lineConv func([]byte) []byte, clientProto string, trySink, reqSink *usageSink, onRelease func(), tryCtx context.Context, -) (done bool, served bool, retry bool, status int, respBody []byte) { +) (done bool, served bool, retry bool, status int, respBody []byte, fkind channel.FailKind) { // 原始响应捕获：仅管理员且开关开启（raw_request 已 set）时累积上游原始行 _, capture := c.Get("raw_request") var rawResp strings.Builder @@ -580,9 +597,9 @@ func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Chann scanner := newSSEScanner(r) releaseSeen := false - fail := func(msg string, kind channel.FailKind) (bool, bool, bool, int, []byte) { + fail := func(msg string, kind channel.FailKind) (bool, bool, bool, int, []byte, channel.FailKind) { g.ch.ReportFailure(ch, kind, 0) - return false, false, true, http.StatusBadGateway, []byte(ch.Name + ": " + msg) + return false, false, true, http.StatusBadGateway, []byte(ch.Name + ": " + msg), kind } for { line, err := scanner.Next() @@ -605,7 +622,7 @@ func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Chann // 客户端意外断开：按已生成部分收费（canceled） commitRaw() g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") - return true, false, false, 0, nil + return true, false, false, 0, nil, channel.FailNet } if trySink != nil { trySink.addContent(out) @@ -639,12 +656,12 @@ func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Chann } else { g.recordError(c, ch, nil, start, "stream_read_error") } - return true, err == io.EOF, false, 0, nil + return true, err == io.EOF, false, 0, nil, channel.FailNet } // 放行前失败：换渠道重试 if c.Request.Context().Err() != nil { g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") - return true, false, false, 0, nil + return true, false, false, 0, nil, channel.FailNet } if tryCtx != nil && tryCtx.Err() != nil { return fail(fmt.Sprintf("first token timeout (no content in %s)", g.firstTokenWait), channel.FailNet) @@ -1018,7 +1035,7 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time } var model store.Model - var cost float64 + var cost decimal.Decimal var modelID uint64 model, _ = g.modelByName(mn) if model.ID > 0 { @@ -1026,12 +1043,18 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time // 计价口径：in=非缓存输入、cacheRead=缓存读、cacheCreate=缓存写（Anthropic 语义， // messages 流式 message_delta 的 input_tokens 即非缓存部分）。 // 缓存写价未单独配置时按输入价 1.25×（Anthropic 5m 口径）。 + // 金额用 decimal 精确运算：token×价/1e6，结果对齐 numeric(20,8) 列精度。 cwp := model.CacheWritePriceOr() - cost = float64(in)/1e6*model.InputPrice + - float64(out)/1e6*model.OutputPrice + - float64(cacheRead)/1e6*model.CacheReadPrice + - float64(cache5m)/1e6*cwp + - float64(cache1h)/1e6*cwp*1.6 + million := decimal.NewFromInt(1e6) + perMillion := func(tokens int64, price decimal.Decimal) decimal.Decimal { + return decimal.NewFromInt(tokens).Mul(price).Div(million) + } + cost = perMillion(in, model.InputPrice). + Add(perMillion(out, model.OutputPrice)). + Add(perMillion(cacheRead, model.CacheReadPrice)). + Add(perMillion(cache5m, cwp)). + Add(perMillion(cache1h, cwp.Mul(decimal.NewFromFloat(1.6)))). + Round(8) } else { // 模型在记账时已被删除（请求进行中被下架）：无法定价，按 0 记账留痕。 // 虚构单价扣用户的钱既不可审计也无法对账。 @@ -1111,7 +1134,7 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time }) // 请求级一行日志：usage_logs 落库是异步的且可能丢，排障时日志是唯一即时依据 - accessf("proxy: trace=%s user=%d key=%d model=%q served=%q channel=%d proto=%s in=%d out=%d r=%d cache_r=%d cache_w=%d(1h=%d) cost=%.6f status=%s http=%d latency=%dms err=%s", + accessf("proxy: trace=%s user=%d key=%d model=%q served=%q channel=%d proto=%s in=%d out=%d r=%d cache_r=%d cache_w=%d(1h=%d) cost=%s status=%s http=%d latency=%dms err=%s", traceStr, uidVal, kidVal, mn, servedModel, chID, p, in, out, reasoning, cacheRead, cacheCreate, cache1h, cost, status, c.Writer.Status(), latency, errCode) } diff --git a/server/internal/proxy/proxy_test.go b/server/internal/proxy/proxy_test.go index 475831d..9693f5f 100644 --- a/server/internal/proxy/proxy_test.go +++ b/server/internal/proxy/proxy_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -14,11 +15,13 @@ import ( "time" "github.com/gin-gonic/gin" + "github.com/openteam/server/internal/channel" "github.com/openteam/server/internal/pkg/apikey" "github.com/openteam/server/internal/pkg/crypto" "github.com/openteam/server/internal/pkg/ratelimit" "github.com/openteam/server/internal/store" "github.com/openteam/server/internal/usage" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -46,7 +49,7 @@ func newHarness(t *testing.T, baseURL string, timeoutMS int, bind bool, keyModel enc := crypto.NewEncryptor("test-master-key-1234567890") encKey, _ := enc.Encrypt("upstream-key") u := store.User{Username: "u1", Email: "u1@x.com", PasswordHash: "x", Role: store.RoleUser, - Balance: 100, Status: store.UserStatusActive} + Balance: decimal.NewFromInt(100), Status: store.UserStatusActive} if err := db.Create(&u).Error; err != nil { t.Fatal(err) } @@ -56,7 +59,7 @@ func newHarness(t *testing.T, baseURL string, timeoutMS int, bind bool, keyModel if err := db.Create(&k).Error; err != nil { t.Fatal(err) } - m := store.Model{Name: "m1", InputPrice: 0.15, OutputPrice: 0.6, Enabled: true} + m := store.Model{Name: "m1", InputPrice: decimal.NewFromFloat(0.15), OutputPrice: decimal.NewFromFloat(0.6), Enabled: true} if err := db.Create(&m).Error; err != nil { t.Fatal(err) } @@ -426,6 +429,86 @@ func TestFailoverBudgetStopsRetrying(t *testing.T) { } } +// 429/配额类失败瞬时返回，时间预算对它们形同虚设；按失败类型封顶换渠道次数。 +// rate 上限 2 → 首渠道 + 至多 2 次换渠道 = 3 次尝试即止；上限 0 = 不限（4 条全试）。 +func TestFailoverSwitchCaps(t *testing.T) { + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = io.WriteString(w, `{"error":{"message":"rate limited, try again"}}`) + })) + defer srv.Close() + + newMultiChannel := func(t *testing.T, n int) *harness { + t.Helper() + h := newHarness(t, srv.URL, 0, true, nil) + var m store.Model + h.db.Where("name = ?", "m1").First(&m) + encKey, _ := crypto.NewEncryptor("test-master-key-1234567890").Encrypt("upstream-key") + for i := 2; i <= n; i++ { + ch := store.Channel{Name: fmt.Sprintf("c%d", i), Provider: store.ChannelProviderOpenAI, + Formats: []string{store.FormatChat}, BaseURL: srv.URL, APIKeyEnc: encKey, + Weight: 1, TimeoutMS: 0, MaxConcurrency: 4, + HealthStatus: store.ChannelHealthHealthy, Enabled: true} + if err := h.db.Create(&ch).Error; err != nil { + t.Fatal(err) + } + if err := h.db.Create(&store.ChannelModelBinding{ChannelID: ch.ID, ModelID: m.ID, UpstreamModel: "m1"}).Error; err != nil { + t.Fatal(err) + } + } + return h + } + body := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}` + + // rate 上限 2：打满即停，第 4 条渠道不应被碰 + h := newMultiChannel(t, 4) + h.gw.failoverCaps = map[channel.FailKind]int{channel.FailRate: 2} + w := h.run(t, nil, body) + if hits.Load() != 3 { + t.Fatalf("rate 上限 2 应恰好尝试 3 条渠道（1+2），实际 %d", hits.Load()) + } + if w.Code != http.StatusTooManyRequests { + t.Fatalf("应透传最后的 429，实际 %d: %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "failover switch cap reached (rate_limited)") { + t.Fatalf("错误体应说明按类型封顶：%s", w.Body.String()) + } + + // 上限 0 = 不限：4 条渠道全部试完 + hits.Store(0) + h.gw.failoverCaps = map[channel.FailKind]int{channel.FailRate: 0} + if w := h.run(t, nil, body); w.Code != http.StatusTooManyRequests { + t.Fatalf("应 429，实际 %d", w.Code) + } + if hits.Load() != 4 { + t.Fatalf("不限次时应尝试全部 4 条渠道，实际 %d", hits.Load()) + } + + // quota 上限 1：402 打满即停（402 归为 FailQuota，第一次失败即长冷却，但换渠道仍受上限约束） + hits.Store(0) + h2 := newMultiChannel(t, 4) + h2.gw.failoverCaps = map[channel.FailKind]int{channel.FailQuota: 1} + quotaSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusPaymentRequired) + _, _ = io.WriteString(w, `{"error":{"message":"insufficient balance"}}`) + })) + defer quotaSrv.Close() + if err := h2.db.Model(&store.Channel{}).Where("1=1").Update("base_url", quotaSrv.URL).Error; err != nil { + t.Fatal(err) + } + w2 := h2.run(t, nil, body) + if hits.Load() != 2 { + t.Fatalf("quota 上限 1 应恰好尝试 2 条渠道，实际 %d", hits.Load()) + } + if w2.Code != http.StatusPaymentRequired { + t.Fatalf("应透传 402，实际 %d", w2.Code) + } +} + // 流式空闲时补 SSE 注释心跳：上游长时间不出字（长思考）不能被反代空闲超时掐掉。 func TestStreamHeartbeatWhileIdle(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -542,7 +625,7 @@ func TestCacheWritePriceBilledSeparately(t *testing.T) { defer srv.Close() reqBody := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}` - costOf := func(h *harness) float64 { + costOf := func(h *harness) decimal.Decimal { if w := h.run(t, nil, reqBody); w.Code != http.StatusOK { t.Fatalf("code=%d body=%s", w.Code, w.Body.String()) } @@ -560,7 +643,7 @@ func TestCacheWritePriceBilledSeparately(t *testing.T) { // 默认未配置：回退输入价 0.15 × 1.25 = 0.1875，加 in/out 的零头 h1 := newHarness(t, srv.URL, 0, true, nil) fallback := costOf(h1) - if fallback < 0.18 || fallback > 0.20 { + if fallback.LessThan(decimal.NewFromFloat(0.18)) || fallback.GreaterThan(decimal.NewFromFloat(0.20)) { t.Fatalf("未配置缓存写价应按输入价×1.25，实际 cost=%v", fallback) } @@ -571,10 +654,10 @@ func TestCacheWritePriceBilledSeparately(t *testing.T) { t.Fatal(err) } explicit := costOf(h2) - if explicit < 1.99 || explicit > 2.02 { + if explicit.LessThan(decimal.NewFromFloat(1.99)) || explicit.GreaterThan(decimal.NewFromFloat(2.02)) { t.Fatalf("按 cache_write_price=2.0 计费应为 ~2.0，实际 %v", explicit) } - if !(explicit > fallback) { + if !explicit.GreaterThan(fallback) { t.Fatalf("显式价应高于回退价：%v vs %v", explicit, fallback) } } @@ -601,7 +684,7 @@ func TestAllChannelsFailedAttributesLastChannel(t *testing.T) { if l.ChannelID == 0 { t.Fatalf("失败记录没归到渠道上：%+v", l) } - if l.Status != store.UsageStatusError || l.Cost != 0 { + if l.Status != store.UsageStatusError || !l.Cost.IsZero() { t.Fatalf("失败不应产生扣费：%+v", l) } } diff --git a/server/internal/proxy/usage_refine_test.go b/server/internal/proxy/usage_refine_test.go index 48e28c5..81e6eb3 100644 --- a/server/internal/proxy/usage_refine_test.go +++ b/server/internal/proxy/usage_refine_test.go @@ -1,7 +1,6 @@ package proxy import ( - "math" "net/http" "net/http/httptest" "strings" @@ -9,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // runPath 走完整链路（Auth → Handle），路径与请求体由调用方给。 @@ -54,8 +54,8 @@ func TestUsageReasoningAndServedModelRecorded(t *testing.T) { t.Fatalf("reasoning_tokens 应为 3，实际 %d", l.ReasoningTokens) } // 计费不含 reasoning 加成：10×0.15/1e6 + 5×0.6/1e6 = 4.5e-6 - if math.Abs(l.Cost-4.5e-6) > 1e-12 { - t.Fatalf("cost 应为 4.5e-6（reasoning 不重复计费），实际 %g", l.Cost) + if !l.Cost.Equal(decimal.NewFromFloat(4.5e-6)) { + t.Fatalf("cost 应为 4.5e-6（reasoning 不重复计费），实际 %s", l.Cost) } } @@ -100,9 +100,13 @@ func TestAnthropicCache1hBillingSplit(t *testing.T) { t.Fatalf("served_model 应为 claude-x，实际 %q", l.ServedModel) } // 100/1e6*1 + 10/1e6*2 + 200/1e6*0.1 + 100/1e6*1.25 + 200/1e6*(1.25*1.6) = 6.65e-4 - want := 100.0/1e6*1.0 + 10.0/1e6*2.0 + 200.0/1e6*0.1 + 100.0/1e6*1.25 + 200.0/1e6*2.0 - if math.Abs(l.Cost-want) > 1e-9 { - t.Fatalf("cost 应为 %g（1h/5m 分开计价），实际 %g", want, l.Cost) + million := decimal.NewFromInt(1e6) + want := decimal.NewFromInt(100).Div(million).Add(decimal.NewFromInt(10).Mul(decimal.NewFromFloat(2)).Div(million)). + Add(decimal.NewFromInt(200).Mul(decimal.NewFromFloat(0.1)).Div(million)). + Add(decimal.NewFromInt(100).Mul(decimal.NewFromFloat(1.25)).Div(million)). + Add(decimal.NewFromInt(200).Mul(decimal.NewFromFloat(2.0)).Div(million)) + if !l.Cost.Equal(want) { + t.Fatalf("cost 应为 %s（1h/5m 分开计价），实际 %s", want, l.Cost) } // 落库 input 为含缓存的总量口径 if l.InputTokens != 600 || l.OutputTokens != 10 { diff --git a/server/internal/store/backfill_test.go b/server/internal/store/backfill_test.go new file mode 100644 index 0000000..504c632 --- /dev/null +++ b/server/internal/store/backfill_test.go @@ -0,0 +1,64 @@ +package store + +import ( + "testing" + "time" + + "github.com/shopspring/decimal" +) + +// v10 回填：存量 usage_logs 按 (channel_id, 小时) 聚合进桶；标记幂等（二次执行不重复累计）。 +func TestBackfillChannelHourly(t *testing.T) { + db, err := Open("sqlite", "/tmp/store_backfill_test_"+time.Now().Format("150405.000000000")+".db") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + sqlDB, _ := db.DB() + sqlDB.Close() + }) + // Open 已跑过一次空回填并打标；清掉标记与空桶，注入历史明细后重放 + db.Where("key = ?", "channel_hourly_backfill_v10").Delete(&SystemConfig{}) + db.Exec("DELETE FROM usage_channel_hourlies") + + at := time.Date(2026, 11, 9, 14, 30, 0, 0, time.UTC) + atPrev := at.Add(-2 * time.Hour) + rows := []UsageLog{ + {TraceID: "b1", UserID: 1, ChannelID: 5, Status: UsageStatusSuccess, LatencyMS: 300, + InputTokens: 10, OutputTokens: 5, Cost: decimal.NewFromFloat(0.01), CreatedAt: at}, + {TraceID: "b2", UserID: 1, ChannelID: 5, Status: UsageStatusError, ErrorCode: strp("upstream_http_502"), + CreatedAt: at}, + {TraceID: "b3", UserID: 1, ChannelID: 5, Status: UsageStatusSuccess, LatencyMS: 100, + InputTokens: 20, OutputTokens: 10, Cost: decimal.NewFromFloat(0.02), CreatedAt: atPrev}, + {TraceID: "b4", UserID: 1, ChannelID: 0, Status: UsageStatusSuccess, CreatedAt: at}, // channel 0 不入桶 + } + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + + backfillChannelHourly(db) + + var buckets []UsageChannelHourly + db.Where("channel_id = ?", 5).Order("hour DESC").Find(&buckets) + if len(buckets) != 2 { + t.Fatalf("应有两个小时桶，实际 %d", len(buckets)) + } + cur, prev := buckets[0], buckets[1] + if cur.Requests != 2 || cur.Success != 1 || cur.Errors != 1 || cur.LatencyMax != 300 { + t.Fatalf("当前桶 = %+v", cur) + } + if prev.Requests != 1 || prev.Cost.String() != "0.02" { + t.Fatalf("上一桶 = %+v", prev) + } + + // 幂等：清标记重放，计数不应翻倍 + db.Where("key = ?", "channel_hourly_backfill_v10").Delete(&SystemConfig{}) + backfillChannelHourly(db) + var total int64 + db.Model(&UsageChannelHourly{}).Where("channel_id = ?", 5).Select("COALESCE(SUM(requests),0)").Scan(&total) + if total != 3 { + t.Fatalf("回填应幂等，requests 总计 = %d，期望 3", total) + } +} + +func strp(s string) *string { return &s } diff --git a/server/internal/store/db.go b/server/internal/store/db.go index 0a5bc81..5eae5a5 100644 --- a/server/internal/store/db.go +++ b/server/internal/store/db.go @@ -28,7 +28,8 @@ import ( // v7：usage_logs 加 reasoning_tokens / cache_write_1h_tokens / served_model 三列。 // v8：去掉 usage_logs 上没人查询的 key_id / trace_id 索引（每行两次多余的 B-tree 插入）。 // v9：同样去掉没人查询的 idx_channel_created。 -const schemaVersion = 9 +// v10：新增 usage_channel_hourly（渠道×小时预聚合），并从 usage_logs 一次性回填历史。 +const schemaVersion = 10 // Open 打开数据库连接并自动迁移。 // 开发默认 SQLite（自动追加 WAL 与 busy_timeout，见 withSQLitePragmas），生产可切 postgres。 @@ -67,6 +68,7 @@ func Open(driver, dsn string) (*gorm.DB, error) { } reshapeUsageLogIndexes(db) dropUnusedUsageLogIndexes(db) + backfillChannelHourly(db) if driver != "postgres" { setSQLiteVersion(db, schemaVersion) } @@ -102,6 +104,47 @@ func reshapeUsageLogIndexes(db *gorm.DB) { log.Printf("store: usage_logs indexes reshaped (user+created / channel+created)") } +// backfillChannelHourly v10 一次性回填：把存量 usage_logs 按 (channel_id, 小时) 聚合进 +// usage_channel_hourlies（GORM 复数表名）。system_configs 打标（PG 每次启动都走迁移路径， +// 幂等必须靠标记），SQLite 靠 user_version 门控只跑一次，双保险。 +// 保留期清理会删老明细，明细没了历史桶仍在——这正是预聚合的意义。 +func backfillChannelHourly(db *gorm.DB) { + var done string + db.Model(&SystemConfig{}).Where("key = ?", "channel_hourly_backfill_v10").Pluck("value", &done) + if done != "" { + return + } + // 一条 INSERT...SELECT 完成聚合；小时表达式按方言取（SQLite 无 to_char、PG 无 strftime）。 + // ON CONFLICT 兜底并发/重入：理论上标记保证只跑一次，撞上时增量语义仍正确。 + hourExpr := `strftime('%Y-%m-%dT%H', created_at)` + if db.Dialector != nil && db.Dialector.Name() == "postgres" { + hourExpr = `to_char(created_at, 'YYYY-MM-DD"T"HH24')` + } + res := db.Exec(`INSERT INTO usage_channel_hourlies + (channel_id, hour, requests, success, canceled, errors, + input_tokens, output_tokens, cache_read_tokens, cost, latency_sum, latency_max) + SELECT channel_id, + ` + hourExpr + ` AS hour, + COUNT(*), + SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), + SUM(CASE WHEN status = 'canceled' THEN 1 ELSE 0 END), + SUM(CASE WHEN status NOT IN ('success','canceled') THEN 1 ELSE 0 END), + SUM(input_tokens), SUM(output_tokens), SUM(cache_read_tokens), + COALESCE(SUM(cost), 0), + SUM(CASE WHEN status = 'success' THEN latency_ms ELSE 0 END), + COALESCE(MAX(CASE WHEN status = 'success' THEN latency_ms ELSE 0 END), 0) + FROM usage_logs + WHERE channel_id > 0 + GROUP BY channel_id, hour + ON CONFLICT (channel_id, hour) DO NOTHING`) + if res.Error != nil { + log.Printf("store: backfill usage_channel_hourly: %v", res.Error) + return + } + db.Create(&SystemConfig{Key: "channel_hourly_backfill_v10", Value: `"done"`}) + log.Printf("store: channel hourly buckets backfilled (%d rows)", res.RowsAffected) +} + // withSQLitePragmas 给 DSN 补上 busy_timeout 与 WAL。 // glebarez（modernc）驱动只解析 _pragma/_txlock/_time_format/vfs 四种参数—— // 旧注释宣称的 `?_journal_mode=WAL` 实际不生效，缺 WAL 时并发写会频繁 database is locked。 diff --git a/server/internal/store/models.go b/server/internal/store/models.go index ac7b190..a0e834a 100644 --- a/server/internal/store/models.go +++ b/server/internal/store/models.go @@ -1,11 +1,14 @@ // Package store 数据模型与仓储层（GORM）。 -// 字段设计对应 PLANNING.md §6：金额/价格 numeric(20,8)，token bigint，时间 UTC。 +// 字段设计对应 PLANNING.md §6：金额/价格 numeric(20,8)（Go 侧 decimal.Decimal 精确运算， +// 不允许 float64 参与金额计算），token bigint，时间 UTC。 package store import ( "regexp" "strings" "time" + + "github.com/shopspring/decimal" ) // 角色 / 状态枚举（字符串存库，便于阅读与迁移） @@ -49,20 +52,20 @@ const ( // User 用户（PLANNING §6.1） type User struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - Username string `gorm:"uniqueIndex;size:64;not null" json:"username"` - Email string `gorm:"uniqueIndex;size:255;not null" json:"email"` - PasswordHash string `gorm:"size:255;not null" json:"-"` - Role string `gorm:"size:16;not null;default:user" json:"role"` - Balance float64 `gorm:"type:numeric(20,8);not null;default:0" json:"balance"` - Status string `gorm:"size:16;not null;default:active" json:"status"` - TokenVersion int `gorm:"not null;default:0" json:"-"` // 会话版本：改密/登出递增，低于当前版本的 token 全部失效 - AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` // 用户级模型白名单(空=不限制) - DeniedModels []string `gorm:"type:jsonb;serializer:json" json:"denied_models,omitempty"` // 用户级模型黑名单 - InviteCode *string `json:"invite_code,omitempty"` - LastLoginAt *time.Time `json:"last_login_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + Username string `gorm:"uniqueIndex;size:64;not null" json:"username"` + Email string `gorm:"uniqueIndex;size:255;not null" json:"email"` + PasswordHash string `gorm:"size:255;not null" json:"-"` + Role string `gorm:"size:16;not null;default:user" json:"role"` + Balance decimal.Decimal `gorm:"type:numeric(20,8);not null;default:0" json:"balance"` + Status string `gorm:"size:16;not null;default:active" json:"status"` + TokenVersion int `gorm:"not null;default:0" json:"-"` // 会话版本：改密/登出递增，低于当前版本的 token 全部失效 + AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` // 用户级模型白名单(空=不限制) + DeniedModels []string `gorm:"type:jsonb;serializer:json" json:"denied_models,omitempty"` // 用户级模型黑名单 + InviteCode *string `json:"invite_code,omitempty"` + LastLoginAt *time.Time `json:"last_login_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // APIKey 密钥（PLANNING §6.2）：库中只存 SHA-256 哈希 + 展示前缀 @@ -144,30 +147,30 @@ func (c *Channel) UpstreamURL(proto, path string) string { // Model 全局模型 + 定价（PLANNING §6.4，价格按每百万 token，USD） type Model struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - Name string `gorm:"uniqueIndex;size:128;not null" json:"name"` - InputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"input_price"` - OutputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"output_price"` - CacheReadPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"cache_read_price"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + Name string `gorm:"uniqueIndex;size:128;not null" json:"name"` + InputPrice decimal.Decimal `gorm:"type:numeric(20,8);not null;default:0" json:"input_price"` + OutputPrice decimal.Decimal `gorm:"type:numeric(20,8);not null;default:0" json:"output_price"` + CacheReadPrice decimal.Decimal `gorm:"type:numeric(20,8);not null;default:0" json:"cache_read_price"` // 缓存写价（每百万 token）。为 0 时按输入价 ×1.25 计（Anthropic 5m 口径）， // 显式配置它才能支持 1h 缓存（×2）这类不同倍率。 - CacheWritePrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"cache_write_price"` - ContextLength int `gorm:"not null;default:0" json:"context_length"` // 上下文窗口（token），0=未知 - SupportsTools bool `gorm:"not null;default:false" json:"supports_tools"` - SupportsVision bool `gorm:"not null;default:false" json:"supports_vision"` - SupportsReasoning bool `gorm:"not null;default:false" json:"supports_reasoning"` - Enabled bool `gorm:"not null;default:true" json:"enabled"` - Sort int `gorm:"not null;default:0" json:"sort"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + CacheWritePrice decimal.Decimal `gorm:"type:numeric(20,8);not null;default:0" json:"cache_write_price"` + ContextLength int `gorm:"not null;default:0" json:"context_length"` // 上下文窗口（token），0=未知 + SupportsTools bool `gorm:"not null;default:false" json:"supports_tools"` + SupportsVision bool `gorm:"not null;default:false" json:"supports_vision"` + SupportsReasoning bool `gorm:"not null;default:false" json:"supports_reasoning"` + Enabled bool `gorm:"not null;default:true" json:"enabled"` + Sort int `gorm:"not null;default:0" json:"sort"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // CacheWritePriceOr 返回缓存写价；未配置时回退到输入价×1.25（Anthropic 5m 口径）。 -func (m *Model) CacheWritePriceOr() float64 { - if m.CacheWritePrice > 0 { +func (m *Model) CacheWritePriceOr() decimal.Decimal { + if m.CacheWritePrice.IsPositive() { return m.CacheWritePrice } - return m.InputPrice * 1.25 + return m.InputPrice.Mul(decimal.NewFromFloat(1.25)) } // ChannelModelBinding 渠道↔模型绑定（多对多，PLANNING §6.4） @@ -192,26 +195,26 @@ type UsageLog struct { KeyID uint64 `json:"key_id"` // 与 key_id/trace_id 同理：没有任何查询按 channel_id 过滤 usage_logs // （渠道统计走 created_at 时间窗后在内存里分组），复合索引纯属写放大。 - ChannelID uint64 `json:"channel_id"` - ModelID uint64 `json:"model_id"` - ModelName string `gorm:"size:128" json:"model_name"` - Protocol string `gorm:"size:32" json:"protocol"` // responses|chat|messages - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - CacheReadTokens int64 `json:"cache_read_tokens"` - CacheCreationTokens int64 `json:"cache_creation_tokens"` - ReasoningTokens int64 `json:"reasoning_tokens"` // 思考 token（上游已计入 output，展示口径不重复计费） - CacheWrite1hTokens int64 `json:"cache_write_1h_tokens"` // Anthropic 1h 缓存写 token（计价 2× 输入） - ServedModel string `gorm:"size:128" json:"served_model"` // 上游实际应答的模型名（映射/聚合商场景与请求名不同） - InputPrice float64 `gorm:"type:numeric(20,8)" json:"input_price"` // 快照 - OutputPrice float64 `gorm:"type:numeric(20,8)" json:"output_price"` // 快照 - CacheReadPrice float64 `gorm:"type:numeric(20,8)" json:"cache_read_price"` // 快照 - Cost float64 `gorm:"type:numeric(20,8)" json:"cost"` - LatencyMS int `json:"latency_ms"` - Status string `gorm:"size:16;not null" json:"status"` - ErrorCode *string `json:"error_code,omitempty"` - RawRequest string `gorm:"type:text" json:"raw_request"` // 客户端原始请求体（未转换） - RawResponse string `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件） + ChannelID uint64 `json:"channel_id"` + ModelID uint64 `json:"model_id"` + ModelName string `gorm:"size:128" json:"model_name"` + Protocol string `gorm:"size:32" json:"protocol"` // responses|chat|messages + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CacheReadTokens int64 `json:"cache_read_tokens"` + CacheCreationTokens int64 `json:"cache_creation_tokens"` + ReasoningTokens int64 `json:"reasoning_tokens"` // 思考 token（上游已计入 output，展示口径不重复计费） + CacheWrite1hTokens int64 `json:"cache_write_1h_tokens"` // Anthropic 1h 缓存写 token（计价 2× 输入） + ServedModel string `gorm:"size:128" json:"served_model"` // 上游实际应答的模型名（映射/聚合商场景与请求名不同） + InputPrice decimal.Decimal `gorm:"type:numeric(20,8)" json:"input_price"` // 快照 + OutputPrice decimal.Decimal `gorm:"type:numeric(20,8)" json:"output_price"` // 快照 + CacheReadPrice decimal.Decimal `gorm:"type:numeric(20,8)" json:"cache_read_price"` // 快照 + Cost decimal.Decimal `gorm:"type:numeric(20,8)" json:"cost"` + LatencyMS int `json:"latency_ms"` + Status string `gorm:"size:16;not null" json:"status"` + ErrorCode *string `json:"error_code,omitempty"` + RawRequest string `gorm:"type:text" json:"raw_request"` // 客户端原始请求体（未转换） + RawResponse string `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件） // CreatedAt 参与 (user, created) 复合索引（用户侧时间窗过滤），单列索引服务 // 全表时间窗扫描（渠道统计、运营总览）。 CreatedAt time.Time `gorm:"index;index:idx_user_created,priority:2" json:"created_at"` @@ -219,42 +222,61 @@ type UsageLog struct { // UsageDaily 日粒度预聚合（PLANNING §6.6） type UsageDaily struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - UserID uint64 `gorm:"index:idx_daily_user_model,unique" json:"user_id"` - ModelID uint64 `gorm:"index:idx_daily_user_model,unique" json:"model_id"` - Date string `gorm:"size:10;index:idx_daily_user_model,unique" json:"date"` // YYYY-MM-DD (UTC) - Requests int64 `json:"requests"` - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` - CacheReadTokens int64 `json:"cache_read_tokens"` - Cost float64 `gorm:"type:numeric(20,8)" json:"cost"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + UserID uint64 `gorm:"index:idx_daily_user_model,unique" json:"user_id"` + ModelID uint64 `gorm:"index:idx_daily_user_model,unique" json:"model_id"` + Date string `gorm:"size:10;index:idx_daily_user_model,unique" json:"date"` // YYYY-MM-DD (UTC) + Requests int64 `json:"requests"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CacheReadTokens int64 `json:"cache_read_tokens"` + Cost decimal.Decimal `gorm:"type:numeric(20,8)" json:"cost"` +} + +// UsageChannelHourly 渠道×小时预聚合。长窗口的渠道统计（6h/24h/7d）不再扫 usage_logs +// 内存分组（窗口大时 5 万行封顶且只反映最近一段），改从桶里读；短窗口（5m/1h） +// 仍走明细以保留精确分位延迟。只含计费行（success/canceled 且 cost>0）与错误行。 +type UsageChannelHourly struct { + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + ChannelID uint64 `gorm:"index:idx_ch_hour,unique" json:"channel_id"` + Hour string `gorm:"size:13;index:idx_ch_hour,unique" json:"hour"` // "2006-01-02T15" (UTC) + Requests int64 `json:"requests"` + Success int64 `json:"success"` + Canceled int64 `json:"canceled"` + Errors int64 `json:"errors"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CacheReadTokens int64 `json:"cache_read_tokens"` + Cost decimal.Decimal `gorm:"type:numeric(20,8)" json:"cost"` + LatencySum int64 `json:"latency_sum"` // 仅 success 行累计，avg = sum/success + LatencyMax int `json:"latency_max"` } // RechargeOrder 充值订单（PLANNING §6.7，预留：首版不做充值） type RechargeOrder struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - UserID uint64 `gorm:"index;not null" json:"user_id"` - Amount float64 `gorm:"type:numeric(20,8);not null" json:"amount"` - Status string `gorm:"size:16;not null;default:pending" json:"status"` - Method string `gorm:"size:16;not null;default:manual" json:"method"` - TransactionID string `gorm:"size:128" json:"transaction_id,omitempty"` - ReviewedBy *uint64 `json:"reviewed_by,omitempty"` - ReviewedAt *time.Time `json:"reviewed_at,omitempty"` - Remark string `gorm:"size:512" json:"remark,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + UserID uint64 `gorm:"index;not null" json:"user_id"` + Amount decimal.Decimal `gorm:"type:numeric(20,8);not null" json:"amount"` + Status string `gorm:"size:16;not null;default:pending" json:"status"` + Method string `gorm:"size:16;not null;default:manual" json:"method"` + TransactionID string `gorm:"size:128" json:"transaction_id,omitempty"` + ReviewedBy *uint64 `json:"reviewed_by,omitempty"` + ReviewedAt *time.Time `json:"reviewed_at,omitempty"` + Remark string `gorm:"size:512" json:"remark,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // BalanceLog 余额流水（PLANNING §6.8，幂等：ref_id + type 唯一） type BalanceLog struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - UserID uint64 `gorm:"index:idx_balance_user;not null" json:"user_id"` - Change float64 `gorm:"type:numeric(20,8);not null" json:"change"` - BalanceAfter float64 `gorm:"type:numeric(20,8);not null" json:"balance_after"` - Type string `gorm:"size:16;not null" json:"type"` - RefID string `gorm:"size:128;index:idx_balance_ref,unique" json:"ref_id"` - Remark string `gorm:"size:512" json:"remark,omitempty"` - CreatedAt time.Time `json:"created_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + UserID uint64 `gorm:"index:idx_balance_user;not null" json:"user_id"` + Change decimal.Decimal `gorm:"type:numeric(20,8);not null" json:"change"` + BalanceAfter decimal.Decimal `gorm:"type:numeric(20,8);not null" json:"balance_after"` + Type string `gorm:"size:16;not null" json:"type"` + RefID string `gorm:"size:128;index:idx_balance_ref,unique" json:"ref_id"` + Remark string `gorm:"size:512" json:"remark,omitempty"` + CreatedAt time.Time `json:"created_at"` } // Passkey WebAuthn 凭据（passkey 绑定/登录） @@ -283,6 +305,7 @@ func AllModels() []any { &ChannelModelBinding{}, &UsageLog{}, &UsageDaily{}, + &UsageChannelHourly{}, &RechargeOrder{}, &BalanceLog{}, &Passkey{}, diff --git a/server/internal/usage/retention_test.go b/server/internal/usage/retention_test.go index c151938..8a49b94 100644 --- a/server/internal/usage/retention_test.go +++ b/server/internal/usage/retention_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" ) // 保留期只删过期明细，余额流水一条不动；未开启（0）时什么都不做。 @@ -18,13 +19,13 @@ func TestRetentionPurgesOldLogsOnly(t *testing.T) { fresh := time.Now().UTC() for i := 0; i < 5; i++ { db.Create(&store.UsageLog{TraceID: fmt.Sprintf("old-%d", i), UserID: u, ModelName: "m1", - Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: old}) + Cost: decimal.NewFromFloat(0.01), Status: store.UsageStatusSuccess, CreatedAt: old}) } for i := 0; i < 3; i++ { db.Create(&store.UsageLog{TraceID: fmt.Sprintf("new-%d", i), UserID: u, ModelName: "m1", - Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: fresh}) + Cost: decimal.NewFromFloat(0.01), Status: store.UsageStatusSuccess, CreatedAt: fresh}) } - db.Create(&store.BalanceLog{UserID: u, Change: -0.05, BalanceAfter: 99.95, + db.Create(&store.BalanceLog{UserID: u, Change: decimal.NewFromFloat(-0.05), BalanceAfter: decimal.NewFromFloat(99.95), Type: store.BalanceTypeUsage, RefID: "usage:old-0"}) n := purgeOlderThan(context.Background(), db, 30) @@ -47,7 +48,7 @@ func TestRetentionDisabled(t *testing.T) { db := openTestDB(t) u := seedUser(t, db, 100) db.Create(&store.UsageLog{TraceID: "x", UserID: u, ModelName: "m1", - Cost: 0.01, Status: store.UsageStatusSuccess, + Cost: decimal.NewFromFloat(0.01), Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC().AddDate(0, 0, -400)}) stop := StartRetention(context.Background(), db, 0) @@ -67,7 +68,7 @@ func TestRetentionBatches(t *testing.T) { logs := make([]store.UsageLog, 0, retentionBatch+50) for i := 0; i < retentionBatch+50; i++ { logs = append(logs, store.UsageLog{TraceID: fmt.Sprintf("b-%d", i), UserID: u, - ModelName: "m1", Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: old}) + ModelName: "m1", Cost: decimal.NewFromFloat(0.001), Status: store.UsageStatusSuccess, CreatedAt: old}) } if err := db.CreateInBatches(logs, 500).Error; err != nil { t.Fatal(err) diff --git a/server/internal/usage/usage.go b/server/internal/usage/usage.go index 8ad23cd..5880779 100644 --- a/server/internal/usage/usage.go +++ b/server/internal/usage/usage.go @@ -8,6 +8,7 @@ import ( "time" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -207,7 +208,7 @@ func (r *Recorder) bill(tx *gorm.DB, logs []*store.UsageLog) error { billable := make([]*store.UsageLog, 0, len(logs)) refs := make([]string, 0, len(logs)) for _, l := range logs { - if (l.Status != store.UsageStatusSuccess && l.Status != store.UsageStatusCanceled) || l.Cost <= 0 { + if (l.Status != store.UsageStatusSuccess && l.Status != store.UsageStatusCanceled) || !l.Cost.IsPositive() { continue } billable = append(billable, l) @@ -250,9 +251,9 @@ func (r *Recorder) bill(tx *gorm.DB, logs []*store.UsageLog) error { ledger := make([]store.BalanceLog, 0, len(billable)) for _, uid := range order { rows := byUser[uid] - total := 0.0 + total := decimal.Zero for _, l := range rows { - total += l.Cost + total = total.Add(l.Cost) } var user store.User if err := tx.First(&user, uid).Error; err != nil { @@ -270,10 +271,10 @@ func (r *Recorder) bill(tx *gorm.DB, logs []*store.UsageLog) error { // 流水里的 balance_after 是批内累计值，与逐行扣款时的语义一致 running := user.Balance for _, l := range rows { - running -= l.Cost + running = running.Sub(l.Cost) ledger = append(ledger, store.BalanceLog{ UserID: uid, - Change: -l.Cost, + Change: l.Cost.Neg(), BalanceAfter: running, Type: store.BalanceTypeUsage, RefID: usageRefID(l.TraceID), @@ -312,7 +313,7 @@ func (r *Recorder) bill(tx *gorm.DB, logs []*store.UsageLog) error { d.InputTokens += l.InputTokens d.OutputTokens += l.OutputTokens d.CacheReadTokens += l.CacheReadTokens - d.Cost += l.Cost + d.Cost = d.Cost.Add(l.Cost) } for _, k := range dailyOrder { d := agg[k] @@ -329,6 +330,72 @@ func (r *Recorder) bill(tx *gorm.DB, logs []*store.UsageLog) error { return err } } + + // 6. 渠道×小时桶：与日聚合同事务 upsert。长窗口（6h/24h/7d）的渠道统计从桶里 + // 读取，不再扫 usage_logs（明细有保留期，桶是完整历史）。计费口径与日聚合一致： + // token/成本只累计可计费且未重复的行；状态计数（含 error）计全部行。 + type chKey struct { + ch uint64 + hour string + } + chAgg := map[chKey]*store.UsageChannelHourly{} + chOrder := make([]chKey, 0, len(logs)) + for _, l := range logs { + if l.ChannelID == 0 { + continue + } + k := chKey{ch: l.ChannelID, hour: l.CreatedAt.UTC().Format("2006-01-02T15")} + b, ok := chAgg[k] + if !ok { + b = &store.UsageChannelHourly{ChannelID: k.ch, Hour: k.hour} + chAgg[k] = b + chOrder = append(chOrder, k) + } + b.Requests++ + switch l.Status { + case store.UsageStatusSuccess: + b.Success++ + b.LatencySum += int64(l.LatencyMS) + if l.LatencyMS > b.LatencyMax { + b.LatencyMax = l.LatencyMS + } + case store.UsageStatusCanceled: + b.Canceled++ + default: + b.Errors++ + } + if l.Cost.IsPositive() && seen[usageRefID(l.TraceID)] { + b.InputTokens += l.InputTokens + b.OutputTokens += l.OutputTokens + b.CacheReadTokens += l.CacheReadTokens + b.Cost = b.Cost.Add(l.Cost) + } + } + for _, k := range chOrder { + b := chAgg[k] + // SQLite 的双参 MAX 是标量取大，PG 用 GREATEST + latMax := gorm.Expr("MAX(latency_max, ?)", b.LatencyMax) + if tx.Dialector != nil && tx.Dialector.Name() == "postgres" { + latMax = gorm.Expr("GREATEST(latency_max, ?)", b.LatencyMax) + } + if err := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "channel_id"}, {Name: "hour"}}, + DoUpdates: clause.Assignments(map[string]any{ + "requests": gorm.Expr("requests + ?", b.Requests), + "success": gorm.Expr("success + ?", b.Success), + "canceled": gorm.Expr("canceled + ?", b.Canceled), + "errors": gorm.Expr("errors + ?", b.Errors), + "input_tokens": gorm.Expr("input_tokens + ?", b.InputTokens), + "output_tokens": gorm.Expr("output_tokens + ?", b.OutputTokens), + "cache_read_tokens": gorm.Expr("cache_read_tokens + ?", b.CacheReadTokens), + "cost": gorm.Expr("cost + ?", b.Cost), + "latency_sum": gorm.Expr("latency_sum + ?", b.LatencySum), + "latency_max": latMax, + }), + }).Create(b).Error; err != nil { + return err + } + } return nil } diff --git a/server/internal/usage/usage_test.go b/server/internal/usage/usage_test.go index 1732d12..ee940e9 100644 --- a/server/internal/usage/usage_test.go +++ b/server/internal/usage/usage_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/openteam/server/internal/store" + "github.com/shopspring/decimal" "gorm.io/gorm" ) @@ -75,11 +76,13 @@ func TestBillingDeductsBalanceAtomically(t *testing.T) { m := seedModel(t, db) rec := NewRecorder(db) + cost := decimal.NewFromFloat(0.15).Add(decimal.NewFromFloat(0.60)). + Mul(decimal.NewFromInt(1000)).Div(decimal.NewFromInt(1e6)).Round(8) rec.Record(&store.UsageLog{ TraceID: "bill", UserID: u, ModelID: m, ModelName: "m1", InputTokens: 1000, OutputTokens: 1000, - InputPrice: 0.15, OutputPrice: 0.60, - Cost: 0.15*1000/1e6 + 0.60*1000/1e6, + InputPrice: decimal.NewFromFloat(0.15), OutputPrice: decimal.NewFromFloat(0.60), + Cost: cost, Status: store.UsageStatusSuccess, LatencyMS: 5, CreatedAt: time.Now().UTC(), @@ -90,15 +93,15 @@ func TestBillingDeductsBalanceAtomically(t *testing.T) { if err := db.First(&user, u).Error; err != nil { t.Fatal(err) } - want := 1 - (0.15*1000/1e6 + 0.60*1000/1e6) - if diff := user.Balance - want; diff > 1e-9 || diff < -1e-9 { - t.Fatalf("余额应扣到 %.10f，实际 %.10f", want, user.Balance) + want := decimal.NewFromInt(1).Sub(cost) + if !user.Balance.Equal(want) { + t.Fatalf("余额应精确扣到 %s，实际 %s", want, user.Balance) } var bl store.BalanceLog if err := db.Where("user_id = ?", u).First(&bl).Error; err != nil { t.Fatalf("余额流水应落库: %v", err) } - if bl.Change >= 0 { + if !bl.Change.IsNegative() { t.Fatalf("流水应为负数扣费，实际 %v", bl.Change) } } @@ -124,7 +127,7 @@ func openTestDB(t *testing.T) *gorm.DB { func seedUser(t *testing.T, db *gorm.DB, balance float64) uint64 { t.Helper() u := store.User{Username: "u" + fmt.Sprint(time.Now().UnixNano()), Email: "u@x.com", - PasswordHash: "x", Role: store.RoleUser, Balance: balance, Status: store.UserStatusActive} + PasswordHash: "x", Role: store.RoleUser, Balance: decimal.NewFromFloat(balance), Status: store.UserStatusActive} if err := db.Create(&u).Error; err != nil { t.Fatal(err) } @@ -133,7 +136,7 @@ func seedUser(t *testing.T, db *gorm.DB, balance float64) uint64 { func seedModel(t *testing.T, db *gorm.DB) uint64 { t.Helper() - m := store.Model{Name: "m1", InputPrice: 0.15, OutputPrice: 0.60, Enabled: true} + m := store.Model{Name: "m1", InputPrice: decimal.NewFromFloat(0.15), OutputPrice: decimal.NewFromFloat(0.60), Enabled: true} if err := db.Create(&m).Error; err != nil { t.Fatal(err) } @@ -151,7 +154,7 @@ func TestBillingRefIDImpotency(t *testing.T) { rec := NewRecorder(db) mk := func(trace string) *store.UsageLog { return &store.UsageLog{TraceID: trace, UserID: u, ModelName: "m1", - Status: store.UsageStatusSuccess, Cost: 1, CreatedAt: time.Now().UTC()} + Status: store.UsageStatusSuccess, Cost: decimal.NewFromInt(1), CreatedAt: time.Now().UTC()} } rec.Record(mk("dup")) rec.Record(mk("dup")) @@ -161,7 +164,7 @@ func TestBillingRefIDImpotency(t *testing.T) { if err := db.First(&user, u).Error; err != nil { t.Fatal(err) } - if user.Balance != 4 { + if !user.Balance.Equal(decimal.NewFromInt(4)) { t.Errorf("balance = %v, want 4 (重复 trace 不重复扣款)", user.Balance) } var n int64 @@ -182,7 +185,7 @@ func TestBillingAggregatesBatch(t *testing.T) { for i := 0; i < n; i++ { rec.Record(&store.UsageLog{ TraceID: fmt.Sprintf("agg-%d", i), UserID: u, ModelID: m, ModelName: "m1", - InputTokens: 10, OutputTokens: 5, Cost: 0.01, + InputTokens: 10, OutputTokens: 5, Cost: decimal.NewFromFloat(0.01), Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC(), }) } @@ -192,7 +195,7 @@ func TestBillingAggregatesBatch(t *testing.T) { if err := db.First(&user, u).Error; err != nil { t.Fatal(err) } - if want := 100 - 0.01*n; user.Balance-want > 1e-9 || user.Balance-want < -1e-9 { + if want := decimal.NewFromInt(100).Sub(decimal.NewFromFloat(0.01).Mul(decimal.NewFromInt(n))); !user.Balance.Equal(want) { t.Fatalf("余额 = %v，期望 %v", user.Balance, want) } var ledgers int64 @@ -208,13 +211,13 @@ func TestBillingAggregatesBatch(t *testing.T) { if daily[0].Requests != n { t.Fatalf("日聚合 requests = %d，期望 %d", daily[0].Requests, n) } - if d := daily[0].Cost - 0.01*n; d > 1e-9 || d < -1e-9 { - t.Fatalf("日聚合 cost = %v，期望 %v", daily[0].Cost, 0.01*n) + if want := decimal.NewFromFloat(0.01).Mul(decimal.NewFromInt(n)); !daily[0].Cost.Equal(want) { + t.Fatalf("日聚合 cost = %v，期望 %v", daily[0].Cost, want) } // 流水里的 balance_after 应与扣款后余额一致（批内累计） var last store.BalanceLog db.Where("user_id = ?", u).Order("id DESC").First(&last) - if last.BalanceAfter-user.Balance > 1e-9 || user.Balance-last.BalanceAfter > 1e-9 { + if !last.BalanceAfter.Equal(user.Balance) { t.Fatalf("最后一条流水 balance_after = %v，实际余额 %v", last.BalanceAfter, user.Balance) } } @@ -226,7 +229,7 @@ func TestRecorderShardedByUser(t *testing.T) { var users []uint64 for i := 0; i < 3; i++ { u := store.User{Username: fmt.Sprintf("shard-u%d", i), Email: fmt.Sprintf("shard%d@x.com", i), - PasswordHash: "x", Role: store.RoleUser, Balance: 100, Status: store.UserStatusActive} + PasswordHash: "x", Role: store.RoleUser, Balance: decimal.NewFromInt(100), Status: store.UserStatusActive} if err := db.Create(&u).Error; err != nil { t.Fatal(err) } @@ -239,7 +242,7 @@ func TestRecorderShardedByUser(t *testing.T) { for _, u := range users { rec.Record(&store.UsageLog{ TraceID: fmt.Sprintf("u%d-%d", u, i), UserID: u, ModelName: "m1", - Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC(), + Cost: decimal.NewFromFloat(0.001), Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC(), }) } } @@ -258,7 +261,7 @@ func TestRecorderShardedByUser(t *testing.T) { t.Fatalf("用户 %d 流水 %d 条，期望 %d", u, len(logs), per) } for i := 1; i < len(logs); i++ { - if logs[i].BalanceAfter > logs[i-1].BalanceAfter { + if logs[i].BalanceAfter.GreaterThan(logs[i-1].BalanceAfter) { t.Fatalf("用户 %d 的流水余额应单调递减（顺序错乱）：%v → %v", u, logs[i-1].BalanceAfter, logs[i].BalanceAfter) } @@ -266,6 +269,53 @@ func TestRecorderShardedByUser(t *testing.T) { } } +// 渠道×小时桶：状态计数含 error 行，token/成本只累计可计费行，跨小时分桶。 +func TestChannelHourlyBucketAggregated(t *testing.T) { + db := openTestDB(t) + u := seedUser(t, db, 100) + m := seedModel(t, db) + + rec := NewRecorder(db) + now := time.Now().UTC() + hour := now.Format("2006-01-02T15") + prevHour := now.Add(-time.Hour).Format("2006-01-02T15") + mk := func(trace, status string, at time.Time, ch uint64) *store.UsageLog { + return &store.UsageLog{TraceID: trace, UserID: u, ModelID: m, ModelName: "m1", + ChannelID: ch, InputTokens: 100, OutputTokens: 50, Cost: decimal.NewFromFloat(0.02), + Status: status, LatencyMS: 200, CreatedAt: at} + } + rec.Record(mk("h1", store.UsageStatusSuccess, now, 7)) + rec.Record(mk("h2", store.UsageStatusSuccess, now, 7)) + rec.Record(mk("h3", store.UsageStatusError, now, 7)) + rec.Record(mk("h4", store.UsageStatusSuccess, now.Add(-time.Hour), 7)) + rec.Record(mk("h5", store.UsageStatusSuccess, now, 0)) // channel_id=0 不入桶 + rec.Close() + + var buckets []store.UsageChannelHourly + db.Where("channel_id = ?", 7).Order("hour DESC").Find(&buckets) + if len(buckets) != 2 { + t.Fatalf("应有两个小时桶，实际 %d", len(buckets)) + } + cur, prev := buckets[0], buckets[1] + if cur.Hour != hour || prev.Hour != prevHour { + t.Fatalf("桶小时错误：%q / %q，期望 %q / %q", cur.Hour, prev.Hour, hour, prevHour) + } + if cur.Requests != 3 || cur.Success != 2 || cur.Errors != 1 { + t.Fatalf("当前桶计数 = %d/%d/%d，期望 3/2/1（requests/success/errors）", cur.Requests, cur.Success, cur.Errors) + } + // 只有 success 行计延迟 + if cur.LatencySum != 400 || cur.LatencyMax != 200 { + t.Fatalf("延迟聚合 = sum %d max %d，期望 400/200", cur.LatencySum, cur.LatencyMax) + } + // token/成本只累计可计费行（2×success） + if cur.InputTokens != 200 || cur.Cost.String() != "0.04" { + t.Fatalf("token/成本 = %d/%s，期望 200/0.04", cur.InputTokens, cur.Cost) + } + if prev.Requests != 1 || prev.Success != 1 { + t.Fatalf("上一小时桶计数 = %d/%d，期望 1/1", prev.Requests, prev.Success) + } +} + // 分片函数：同一用户恒定同片，且覆盖所有分片。 func TestShardOfStable(t *testing.T) { rec := NewRecorder(openTestDB(t))