fix(quota): 平台限额表只保存已配置限额的记录,仪表盘按用量与限额展示平台

user_platform_quotas 只保存至少配置了一档限额的记录;不存在的行等价于不限额。

- 注册播种只为配置了限额的平台建行,一个都没有时不写入
- IncrementUsageWithReset 无行时不再 fail-open 建行
- BatchSnapshotUsage 改为只更新既有活跃行,不再插入无限额的新行
- 仓储 BulkInsertInitial/UpsertForUser 把三档全空的记录视为未配置;
  管理员全量替换接口同样剔除,清空限额即软删并放弃累计用量
- 迁移 238 删除三档全 NULL 的历史行
- 用户仪表盘平台卡片 = 有用量的平台 ∪ 配置了限额的平台,计数改为卡片数
- 管理员限额弹窗对未保存限额的平台禁用"重置窗口"
This commit is contained in:
feeeei
2026-09-10 16:41:58 +08:00
parent 98d86915be
commit 4a4b50d807
24 changed files with 628 additions and 149 deletions
@@ -790,15 +790,21 @@ func (h *UserHandler) UpdateUserPlatformQuotas(c *gin.Context) {
}
}
// 三档全空的输入不落库:user_platform_quotas 只保存至少配置了一档限额的记录,
// 不存在的行等价于不限额。未进入 records 的平台由 UpsertForUser 软删,审计里记为 removed。
records := make([]service.UserPlatformQuotaRecord, 0, len(req.Quotas))
for _, q := range req.Quotas {
records = append(records, service.UserPlatformQuotaRecord{
rec := service.UserPlatformQuotaRecord{
UserID: userID,
Platform: q.Platform,
DailyLimitUSD: q.DailyLimitUSD,
WeeklyLimitUSD: q.WeeklyLimitUSD,
MonthlyLimitUSD: q.MonthlyLimitUSD,
})
}
if !rec.HasAnyLimit() {
continue
}
records = append(records, rec)
}
ctx := c.Request.Context()
@@ -860,6 +866,7 @@ func (h *UserHandler) UpdateUserPlatformQuotas(c *gin.Context) {
slog.Info("admin.quota_updated",
"actor_admin_id", getAdminIDFromContext(c),
"target_user_id", userID,
"submitted_count", len(req.Quotas),
"platform_count", len(records),
"before_snapshot_available", beforeErr == nil,
"changes", changes)
@@ -112,16 +112,61 @@ func TestUpdateUserPlatformQuotas_Success(t *testing.T) {
if len(repo.upsertCalls) != 1 {
t.Fatalf("UpsertForUser should be called once, got %d", len(repo.upsertCalls))
}
// upsert 记录数 = 请求体中给出的平台数(未给出的平台不落库)。
if repo.upsertCalls[0].userID != 42 || len(repo.upsertCalls[0].records) != 5 {
// upsert 记录数 = 请求体中至少配置了一档限额的平台数;三档全空的平台不落库,
// 由 UpsertForUser 的"不在列表里就软删"分支处理。
if repo.upsertCalls[0].userID != 42 || len(repo.upsertCalls[0].records) != 2 {
t.Errorf("unexpected upsert call: %+v", repo.upsertCalls[0])
}
for _, r := range repo.upsertCalls[0].records {
if r.Platform != "anthropic" && r.Platform != "openai" {
t.Errorf("platform %q has no configured limit and must not be upserted", r.Platform)
}
}
// 缓存失效:按全部允许平台统一失效(含 kimi/zhipu/deepseek)。
if len(cache.deleteCalls) != len(service.AllowedQuotaPlatforms) {
t.Errorf("expected %d cache delete calls, got %d: %+v", len(service.AllowedQuotaPlatforms), len(cache.deleteCalls), cache.deleteCalls)
}
}
// TestUpdateUserPlatformQuotas_AllUnlimitedClearsRows 锁定:全部平台三档全空等价于清空,
// UpsertForUser 收到空列表(软删该用户所有活跃行),且 0 = 显式禁用仍算已配置。
func TestUpdateUserPlatformQuotas_AllUnlimitedClearsRows(t *testing.T) {
repo := &upsertCapturingQuotaRepo{}
cache := &billingCacheStub{}
h := buildTestHandler(repo, cache)
body := `{"quotas":[
{"platform":"anthropic","daily_limit_usd":null,"weekly_limit_usd":null,"monthly_limit_usd":null},
{"platform":"openai"}
]}`
c, w := putReq(t, body)
h.UpdateUserPlatformQuotas(c)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
if len(repo.upsertCalls) != 1 {
t.Fatalf("UpsertForUser should be called once, got %d", len(repo.upsertCalls))
}
if len(repo.upsertCalls[0].records) != 0 {
t.Errorf("all-unlimited input must upsert zero records, got %+v", repo.upsertCalls[0].records)
}
repo = &upsertCapturingQuotaRepo{}
h = buildTestHandler(repo, &billingCacheStub{})
c, w = putReq(t, `{"quotas":[{"platform":"gemini","daily_limit_usd":0}]}`)
h.UpdateUserPlatformQuotas(c)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
if len(repo.upsertCalls) != 1 || len(repo.upsertCalls[0].records) != 1 {
t.Fatalf("zero limit is a configured limit and must be upserted: %+v", repo.upsertCalls)
}
if r := repo.upsertCalls[0].records[0]; r.Platform != "gemini" || r.DailyLimitUSD == nil || *r.DailyLimitUSD != 0 {
t.Errorf("unexpected record: %+v", r)
}
}
func TestUpdateUserPlatformQuotas_RejectsDuplicatePlatform(t *testing.T) {
h := buildTestHandler(&upsertCapturingQuotaRepo{}, &billingCacheStub{})
body := `{"quotas":[
@@ -10,7 +10,6 @@ import (
dbent "github.com/Wei-Shaw/sub2api/ent"
"github.com/Wei-Shaw/sub2api/ent/userplatformquota"
"github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
"github.com/lib/pq"
)
// UserPlatformQuotaRecord 是 repository 层的传输结构体,
@@ -29,10 +28,16 @@ type UserPlatformQuotaRecord struct {
MonthlyWindowStart *time.Time
}
// HasAnyLimit 报告该记录是否至少配置了一档限额(0 也算配置)。
// 三档全空的记录视为未配置:不写入 user_platform_quotas,等价于不限额。
func (r UserPlatformQuotaRecord) HasAnyLimit() bool {
return r.DailyLimitUSD != nil || r.WeeklyLimitUSD != nil || r.MonthlyLimitUSD != nil
}
// ErrUserPlatformQuotaNotFound 用于 ResetExpiredWindow 等需要"必须命中已有记录"的方法。
var ErrUserPlatformQuotaNotFound = fmt.Errorf("user platform quota record not found")
// ErrUserPlatformQuotaFKViolation 当批量 UPSERT 中存在 user_id 不在 users 表的记录时返回。
// ErrUserPlatformQuotaFKViolation 表示批量快照写入命中了 user_id 不在 users 表的记录。
var ErrUserPlatformQuotaFKViolation = errors.New("user platform quota snapshot FK violation")
// UserPlatformQuotaSnapshot 是 BatchSnapshotUsage 的输入结构体,
@@ -50,21 +55,20 @@ type UserPlatformQuotaSnapshot struct {
// UserPlatformQuotaRepository 定义用户平台配额的数据访问接口。
type UserPlatformQuotaRepository interface {
// BulkInsertInitial 幂等批量插入初始配额记录(ON CONFLICT DO NOTHING)。
// BulkInsertInitial 幂等批量插入初始配额记录:冲突时只补既有行为 NULL 的档位;三档全空的记录跳过。
BulkInsertInitial(ctx context.Context, records []UserPlatformQuotaRecord) error
// GetByUserPlatform 查询单条配额记录,未找到时返回 (nil, nil)。
GetByUserPlatform(ctx context.Context, userID int64, platform string) (*UserPlatformQuotaRecord, error)
// ListByUser 查询用户的所有平台配额记录(排除软删除)。
ListByUser(ctx context.Context, userID int64) ([]UserPlatformQuotaRecord, error)
// IncrementUsageWithReset 原子地累加用量,若窗口已过期则先重置再累加。
// IncrementUsageWithReset 原子地累加用量,若窗口已过期则先重置再累加;记录不存在时不建行、直接返回 nil。
IncrementUsageWithReset(ctx context.Context, userID int64, platform string, cost float64, now time.Time) error
// ResetExpiredWindow 重置指定窗口(daily/weekly/monthly)的用量与起始时间。
ResetExpiredWindow(ctx context.Context, userID int64, platform string, window string, newStart time.Time) error
// UpsertForUser 全量替换该用户所有平台限额配置(详见 service.UserPlatformQuotaRepository.UpsertForUser)。
// UpsertForUser 全量替换该用户所有平台限额配置;三档全空的记录视为未配置(详见实现注释)。
UpsertForUser(ctx context.Context, userID int64, records []UserPlatformQuotaRecord) error
// BatchSnapshotUsage 用一条多行 UPSERT 把整批 usage 以绝对值覆盖写入(非累加)。
// usage/window_start 直接取 EXCLUDED(Redis 当前窗口快照),无 CASE。整批共用 now 作 created/updated_at。
// 要求 snapshots 内 (user,platform) 不重复。FK 违反返回 ErrUserPlatformQuotaFKViolation。
// BatchSnapshotUsage 把整批 usage 以绝对值覆盖写入既有活跃行(非累加);行不存在或已软删的 (user,platform) 跳过。
// usage/window_start 直接取 Redis 当前窗口快照。整批共用 now 作 updated_at。要求 snapshots 内 (user,platform) 不重复。
BatchSnapshotUsage(ctx context.Context, snapshots []UserPlatformQuotaSnapshot, now time.Time) error
}
@@ -77,18 +81,17 @@ func NewUserPlatformQuotaRepository(client *dbent.Client) UserPlatformQuotaRepos
return &userPlatformQuotaRepository{client: client}
}
// BulkInsertInitial 用原生 SQL ON CONFLICT 实现幂等批量插入(带条件 limit 覆盖)。
// BulkInsertInitial 用原生 SQL ON CONFLICT 实现幂等批量插入。
// 仅插入 limit_usd 与元数据,usage_usd 用 DB 默认 0,window_start 留 NULL。
// 三档限额全空的记录视为未配置,跳过不插入。
// FK 约束要求 user_id 在 users 表中存在,调用方负责保证。
//
// 冲突策略:CASE WHEN existing.*_limit_usd IS NULL THEN EXCLUDED.*_limit_usd ELSE existing ...
// - 若 IncrementUsageWithReset 因时序问题已先建行(limit 全 NULL),
// 此处会把注册时的默认 limit 写入,避免该用户在该平台永久无限额。
// - 若管理员已通过 UpsertForUser 设置了非 NULL 个性化 limit,**保留不动**
// —— 旧实现无条件 EXCLUDED 覆盖会丢失个性化配置。
// 冲突策略:COALESCE(existing.*_limit_usd, EXCLUDED.*_limit_usd)
// - 既有行某档 limit 为 NULL 时写入本次的值,非 NULL 的既有 limit 保持不动。
// - 不会改 usage_usd / window_start,保留累计的用量。
// - 仅命中 deleted_at IS NULL 的活跃记录(partial unique index 作用域)。
func (r *userPlatformQuotaRepository) BulkInsertInitial(ctx context.Context, records []UserPlatformQuotaRecord) error {
records = configuredRecords(records)
if len(records) == 0 {
return nil
}
@@ -116,8 +119,6 @@ func (r *userPlatformQuotaRepository) BulkInsertInitial(ctx context.Context, rec
}
// 精确命中 partial unique index(deleted_at IS NULL),避免对软删记录的歧义冲突。
// 条件覆盖:仅在现有 limit 为 NULL 时才写入 EXCLUDED,否则保留现有非 NULL 值。
// - 修复 IncrementUsageWithReset 已用 NULL limit 建行的场景(NULL → 注册默认)
// - 保护管理员通过 UpsertForUser 设置的个性化 limit 不被静默覆盖
_, _ = sb.WriteString(` ON CONFLICT (user_id, platform) WHERE deleted_at IS NULL
DO UPDATE SET
daily_limit_usd = COALESCE(user_platform_quotas.daily_limit_usd, EXCLUDED.daily_limit_usd),
@@ -169,13 +170,10 @@ func (r *userPlatformQuotaRepository) ListByUser(ctx context.Context, userID int
// IncrementUsageWithReset 原子累加 cost 到 (user, platform) 三个窗口的 *_usage_usd。
// 行为:
// - 若记录存在:在事务内 SELECT FOR UPDATE,按 (prev_window_start vs current_window_start)
// - 记录存在:在事务内 SELECT FOR UPDATE,按 (prev_window_start vs current_window_start)
// 判断是否需要重置(不同 = 重置为 cost;相同 = 累加 cost)
// - 若记录不存在(fail-open create 分支):插入新记录,**limit 字段保留 nil(无限制)**
// —— 这是预期行为:billing 链路不能因 quota 表缺失而阻断请求,未注册路径
// 的用户 quota 默认放行,由调度层指标观测 + 后台对账补建 limit
//
// 上层正常路径(注册时 BulkInsertInitial)保证 limit 在记录创建时就被写入。
// - 记录不存在:直接返回 nil,不建行。限额只存在于 user_platform_quotas 的行中,
// 不存在的行等价于不限额,没有可累加的对象。
func (r *userPlatformQuotaRepository) IncrementUsageWithReset(ctx context.Context, userID int64, platform string, cost float64, now time.Time) error {
return r.withTx(ctx, func(txCtx context.Context, txClient *dbent.Client) error {
existing, err := txClient.UserPlatformQuota.Query().
@@ -187,25 +185,7 @@ func (r *userPlatformQuotaRepository) IncrementUsageWithReset(ctx context.Contex
ForUpdate().
Only(txCtx)
if dbent.IsNotFound(err) {
// fail-open 建行:limit_* 保留 NULL(无限额)。
// 用 ON CONFLICT DO UPDATE 累加,而非裸 INSERT:并发下另一请求可能在本事务
// SELECT FOR UPDATE 之后、INSERT 之前刚建行,裸 INSERT 会撞 partial unique index
// 致事务回滚、本次 cost 丢失;DO UPDATE 把 cost 累加到既有 usage 上。
// 写法与本文件 insertLimitsRow / BulkInsertInitial 的 ON CONFLICT 一致。
const insertSQL = `INSERT INTO user_platform_quotas
(user_id, platform, daily_usage_usd, weekly_usage_usd, monthly_usage_usd,
daily_window_start, weekly_window_start, monthly_window_start, created_at, updated_at)
VALUES ($1, $2, $3, $3, $3, $4, $5, $6, $7, $7)
ON CONFLICT (user_id, platform) WHERE deleted_at IS NULL DO UPDATE SET
daily_usage_usd = user_platform_quotas.daily_usage_usd + EXCLUDED.daily_usage_usd,
weekly_usage_usd = user_platform_quotas.weekly_usage_usd + EXCLUDED.weekly_usage_usd,
monthly_usage_usd = user_platform_quotas.monthly_usage_usd + EXCLUDED.monthly_usage_usd,
updated_at = EXCLUDED.updated_at`
// $6 = now:30 天滚动月度窗口以当前时刻为起始
_, e := txClient.ExecContext(txCtx, insertSQL,
userID, platform, cost,
timezone.StartOfDay(now), timezone.StartOfWeek(now), now, now)
return e
return nil
}
if err != nil {
return err
@@ -331,12 +311,13 @@ func monthlyMaybeReset(prevUsage float64, prevStart *time.Time, cost float64, no
}
// UpsertForUser 全量替换该用户的所有平台限额(事务内):
// 1. 软删除未在 records 中出现的所有 active 行
// 2. 对每条 record 尝试 UPDATE(含 deleted_at = NULL 兼容重激活);
// UPDATE 行数为 0 时 INSERT 新行
// 1. 软删除未在 records 中出现、或在 records 中但三档全空的所有 active 行
// 2. 对每条至少配置了一档限额的 record 尝试 UPDATE active 行;UPDATE 行数为 0 时 INSERT 新行
//
// 仅改 *_limit_usd + deleted_at + updated_at,保留 *_usage_usd / *_window_start。
// 仅改 *_limit_usd + updated_at,保留 *_usage_usd / *_window_start。
// 清空某平台的限额即软删该行并放弃其累计用量;之后再次配置限额会新建行、从新窗口起算。
func (r *userPlatformQuotaRepository) UpsertForUser(ctx context.Context, userID int64, records []UserPlatformQuotaRecord) error {
records = configuredRecords(records)
return r.withTx(ctx, func(txCtx context.Context, txClient *dbent.Client) error {
platforms := make([]string, 0, len(records))
for _, rec := range records {
@@ -391,13 +372,12 @@ func softDeleteMissingPlatforms(ctx context.Context, client *dbent.Client, userI
}
// updateLimitsRow 尝试 UPDATE active 行(deleted_at IS NULL),返回受影响行数。
// 仅更新 active 行:若存在多条历史软删记录,加 deleted_at IS NULL 守卫可避免
// 批量重激活导致的 partial unique index(userplatformquota_user_id_platform_uq)冲突。
// affected=0 时由调用方 UpsertForUser 走 insertLimitsRow 路径创建新行。
// 软删行不会被命中:历史软删记录可能有多条,重激活会撞 partial unique index
// (userplatformquota_user_id_platform_uq)。affected=0 时由调用方 UpsertForUser 走
// insertLimitsRow 路径创建新行。
func updateLimitsRow(ctx context.Context, client *dbent.Client, userID int64, rec UserPlatformQuotaRecord, now time.Time) (int64, error) {
const query = `UPDATE user_platform_quotas
SET daily_limit_usd = $1, weekly_limit_usd = $2, monthly_limit_usd = $3,
deleted_at = NULL, updated_at = $4
SET daily_limit_usd = $1, weekly_limit_usd = $2, monthly_limit_usd = $3, updated_at = $4
WHERE user_id = $5 AND platform = $6 AND deleted_at IS NULL`
res, err := client.ExecContext(ctx, query,
rec.DailyLimitUSD, rec.WeeklyLimitUSD, rec.MonthlyLimitUSD, now,
@@ -437,14 +417,14 @@ func insertLimitsRow(ctx context.Context, client *dbent.Client, userID int64, re
return nil
}
// batchRows 是 BatchSnapshotUsage 每批最大行数(9 参/行 × 6000 ≈ 54000 参,低于 Postgres 65535 上限)。
// batchRows 是 BatchSnapshotUsage 每批最大行数(8 参/行 × 6000 + 1 ≈ 48001 参,低于 Postgres 65535 上限)。
const batchRows = 6000
// BatchSnapshotUsage 用一条多行 UPSERT 把整批 usage 以绝对值覆盖写入(非累加)。
// BatchSnapshotUsage 用一条多行 UPDATE ... FROM (VALUES ...) 把整批 usage 以绝对值覆盖写入既有活跃行(非累加)。
// 每批最多 batchRows 行;$1=now 共用;每行 8 个 per-row 参(user_id, platform, 3×usage, 3×window_start)。
// FK 违反(user_id 不存在)返回 ErrUserPlatformQuotaFKViolation。
// 只命中 deleted_at IS NULL 的既有行:行不存在或已软删的 (user, platform) 直接跳过,不建行。
//
// 注意:snapshots 超过 batchRows 会分多条 SQL 执行且【非单事务】——若某子批 FK 失败,
// 注意:snapshots 超过 batchRows 会分多条 SQL 执行且【非单事务】——若某子批失败,
// 先前子批已写入无法回滚。调用方(flusher)应保证单次 batchSize ≤ batchRows
// (默认 flush_batch_size=1000 < 6000,安全)。
// 另注:启用 flusher 后,本绝对值覆盖与 admin 直写 DB(ResetExpiredWindow/UpsertForUser)存在覆盖竞态,
@@ -465,19 +445,25 @@ func (r *userPlatformQuotaRepository) BatchSnapshotUsage(ctx context.Context, sn
var sb strings.Builder
_, _ = sb.WriteString(
"INSERT INTO user_platform_quotas" +
" (user_id, platform, daily_usage_usd, weekly_usage_usd, monthly_usage_usd," +
" daily_window_start, weekly_window_start, monthly_window_start, created_at, updated_at)" +
" VALUES ")
"UPDATE user_platform_quotas AS q SET" +
" daily_usage_usd = v.daily_usage_usd," +
" weekly_usage_usd = v.weekly_usage_usd," +
" monthly_usage_usd = v.monthly_usage_usd," +
" daily_window_start = v.daily_window_start," +
" weekly_window_start = v.weekly_window_start," +
" monthly_window_start = v.monthly_window_start," +
" updated_at = $1" +
" FROM (VALUES ")
// $1 = now(共用);每行 8 个 per-row 参,从 $2 起连续编号。
// VALUES 里的占位符显式转型,避免 Postgres 对多行 VALUES 推断不出参数类型。
args := []any{now}
for i, s := range batch {
if i > 0 {
_, _ = sb.WriteString(",")
}
b := len(args) // 当前 per-row 第一个参数的 0-based 索引,实际占位符 = b+1
fmt.Fprintf(&sb, "($%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$1,$1)",
fmt.Fprintf(&sb, "($%d::bigint,$%d::varchar,$%d::numeric,$%d::numeric,$%d::numeric,$%d::timestamptz,$%d::timestamptz,$%d::timestamptz)",
b+1, b+2, b+3, b+4, b+5, b+6, b+7, b+8)
args = append(args,
s.UserID, s.Platform,
@@ -487,22 +473,24 @@ func (r *userPlatformQuotaRepository) BatchSnapshotUsage(ctx context.Context, sn
}
_, _ = sb.WriteString(
" ON CONFLICT (user_id, platform) WHERE deleted_at IS NULL DO UPDATE SET" +
" daily_usage_usd = EXCLUDED.daily_usage_usd," +
" weekly_usage_usd = EXCLUDED.weekly_usage_usd," +
" monthly_usage_usd = EXCLUDED.monthly_usage_usd," +
" daily_window_start = EXCLUDED.daily_window_start," +
" weekly_window_start = EXCLUDED.weekly_window_start," +
" monthly_window_start = EXCLUDED.monthly_window_start," +
" updated_at = EXCLUDED.updated_at")
") AS v(user_id, platform, daily_usage_usd, weekly_usage_usd, monthly_usage_usd," +
" daily_window_start, weekly_window_start, monthly_window_start)" +
" WHERE q.user_id = v.user_id AND q.platform = v.platform AND q.deleted_at IS NULL")
if _, err := client.ExecContext(ctx, sb.String(), args...); err != nil {
var pqErr *pq.Error
if errors.As(err, &pqErr) && pqErr.Code == "23503" {
return ErrUserPlatformQuotaFKViolation
}
return err
}
}
return nil
}
// configuredRecords 过滤出至少配置了一档限额的记录;三档全空的记录视为未配置。
func configuredRecords(records []UserPlatformQuotaRecord) []UserPlatformQuotaRecord {
out := make([]UserPlatformQuotaRecord, 0, len(records))
for _, rec := range records {
if rec.HasAnyLimit() {
out = append(out, rec)
}
}
return out
}
@@ -10,6 +10,8 @@ import (
"time"
dbent "github.com/Wei-Shaw/sub2api/ent"
"github.com/Wei-Shaw/sub2api/ent/schema/mixins"
"github.com/Wei-Shaw/sub2api/ent/userplatformquota"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
@@ -23,6 +25,16 @@ func mustCreateUserForQuota(t *testing.T, client *dbent.Client) int64 {
return u.ID
}
// mustSeedQuotaRow 为 (user, platform) 建一条带 daily limit 的活跃记录。
// IncrementUsageWithReset 只累加已存在的记录,累加类测试先用它建行。
func mustSeedQuotaRow(t *testing.T, ctx context.Context, repo UserPlatformQuotaRepository, userID int64, platform string) {
t.Helper()
limit := 1000.0
require.NoError(t, repo.BulkInsertInitial(ctx, []UserPlatformQuotaRecord{
{UserID: userID, Platform: platform, DailyLimitUSD: &limit},
}))
}
func TestUserPlatformQuotaRepository_BulkInsertInitial_Idempotent(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
@@ -36,17 +48,17 @@ func TestUserPlatformQuotaRepository_BulkInsertInitial_Idempotent(t *testing.T)
daily := 5.0
records := []UserPlatformQuotaRecord{
{UserID: userID, Platform: "anthropic", DailyLimitUSD: &daily},
{UserID: userID, Platform: "openai"},
{UserID: userID, Platform: "openai"}, // 三档全空 = 未配置,不建行
}
// 第一次插入
require.NoError(t, repo.BulkInsertInitial(txCtx, records), "first insert")
// 第二次插入应为 no-op(ON CONFLICT DO NOTHING)
// 第二次插入应为 no-op(ON CONFLICT 只补 NULL 档位)
require.NoError(t, repo.BulkInsertInitial(txCtx, records), "second insert (idempotent)")
list, err := repo.ListByUser(txCtx, userID)
require.NoError(t, err, "list")
require.Len(t, list, 2, "expected 2 records after idempotent insert")
require.Len(t, list, 1, "only the configured platform gets a row, and only once")
// 校验 daily_limit_usd 保留
var anthropicRec *UserPlatformQuotaRecord
@@ -114,9 +126,9 @@ func TestUserPlatformQuotaRepository_BulkInsertInitial_CNProvidersAllowed(t *tes
daily := 12.0
records := []UserPlatformQuotaRecord{
{UserID: userID, Platform: "kimi", DailyLimitUSD: &daily},
{UserID: userID, Platform: "zhipu"},
{UserID: userID, Platform: "deepseek"},
{UserID: userID, Platform: "minimax"},
{UserID: userID, Platform: "zhipu", DailyLimitUSD: &daily},
{UserID: userID, Platform: "deepseek", DailyLimitUSD: &daily},
{UserID: userID, Platform: "minimax", DailyLimitUSD: &daily},
}
require.NoError(t, repo.BulkInsertInitial(txCtx, records),
"kimi/zhipu/deepseek/minimax 平台应可写入(CHECK 约束已含国产供应商)")
@@ -168,8 +180,9 @@ func TestUserPlatformQuotaRepository_IncrementUsageWithReset_SameWindow(t *testi
repo := NewUserPlatformQuotaRepository(client)
now := time.Date(2026, 5, 22, 10, 0, 0, 0, time.UTC) // 周五
mustSeedQuotaRow(t, ctx, repo, userID, "anthropic")
// 首次调用:应新建记录
// 首次累加:window_start 为 NULL 视为窗口过期,usage 重置为 cost
require.NoError(t, repo.IncrementUsageWithReset(ctx, userID, "anthropic", 1.5, now))
rec, err := repo.GetByUserPlatform(ctx, userID, "anthropic")
@@ -195,6 +208,7 @@ func TestUserPlatformQuotaRepository_IncrementUsageWithReset_DailyReset(t *testi
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
mustSeedQuotaRow(t, ctx, repo, userID, "anthropic")
day1 := time.Date(2026, 5, 22, 10, 0, 0, 0, time.UTC) // 周五(同一周、同一月)
day2 := time.Date(2026, 5, 23, 10, 0, 0, 0, time.UTC) // 周六(同一周、同一月)
@@ -215,6 +229,7 @@ func TestUserPlatformQuotaRepository_IncrementUsageWithReset_WeeklyReset(t *test
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
mustSeedQuotaRow(t, ctx, repo, userID, "openai")
// 5月22日(周五)和 5月25日(下周一),不同周
fri := time.Date(2026, 5, 22, 10, 0, 0, 0, time.UTC)
@@ -230,6 +245,26 @@ func TestUserPlatformQuotaRepository_IncrementUsageWithReset_WeeklyReset(t *test
require.InDelta(t, 7.0, rec.MonthlyUsageUSD, 1e-9, "monthly accumulates (same month)")
}
// TestUserPlatformQuotaRepository_IncrementUsageWithReset_NoRowIsNoop 锁定不变式:
// 没有配额记录的 (user, platform) 累加时不建行、不报错。
func TestUserPlatformQuotaRepository_IncrementUsageWithReset_NoRowIsNoop(t *testing.T) {
ctx := context.Background()
client := testEntClient(t)
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
now := time.Date(2026, 5, 22, 10, 0, 0, 0, time.UTC)
require.NoError(t, repo.IncrementUsageWithReset(ctx, userID, "anthropic", 1.5, now))
rec, err := repo.GetByUserPlatform(ctx, userID, "anthropic")
require.NoError(t, err)
require.Nil(t, rec, "increment on a missing row must not create one")
list, err := repo.ListByUser(ctx, userID)
require.NoError(t, err)
require.Empty(t, list)
}
func TestUserPlatformQuotaRepository_ResetExpiredWindow(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
@@ -324,10 +359,10 @@ func TestUserPlatformQuotaRepository_ResetExpiredWindow_NotFoundReturnsSentinel(
"expected ErrUserPlatformQuotaNotFound, got %v", err)
}
// TestBatchSnapshotUsage_InsertOverwriteMultiKey 验证 BatchSnapshotUsage 的绝对值覆盖语义:
// 1. 首批插入 2 条(不同 user),验证 daily 等于首批值;
// TestBatchSnapshotUsage_OverwriteMultiKey 验证 BatchSnapshotUsage 的绝对值覆盖语义:
// 1. 首批写入 2 条既有行(不同 user),验证 daily 等于首批值;
// 2. 对同一 key 传不同值,验证 daily 等于新值(绝对覆盖,非累加)。
func TestBatchSnapshotUsage_InsertOverwriteMultiKey(t *testing.T) {
func TestBatchSnapshotUsage_OverwriteMultiKey(t *testing.T) {
ctx := context.Background()
// BatchSnapshotUsage 不开事务(直接写),使用独立 client 保证跨调用可见性。
client := testEntClient(t)
@@ -336,13 +371,15 @@ func TestBatchSnapshotUsage_InsertOverwriteMultiKey(t *testing.T) {
userID2 := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
mustSeedQuotaRow(t, ctx, repo, userID1, "anthropic")
mustSeedQuotaRow(t, ctx, repo, userID2, "openai")
now := time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)
dailyStart := time.Date(2026, 5, 29, 0, 0, 0, 0, time.UTC)
weeklyStart := time.Date(2026, 5, 25, 0, 0, 0, 0, time.UTC) // 当周一
monthlyStart := time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC)
// ── 第一批:插入 2 行 ──────────────────────────────────────────────────────
// ── 第一批:覆盖 2 行 ──────────────────────────────────────────────────────
firstBatch := []UserPlatformQuotaSnapshot{
{
UserID: userID1,
@@ -421,3 +458,48 @@ func TestBatchSnapshotUsage_InsertOverwriteMultiKey(t *testing.T) {
require.InDelta(t, 18.8, rec2After.WeeklyUsageUSD, 1e-9, "user2 weekly must be overwritten to 18.8")
require.InDelta(t, 28.8, rec2After.MonthlyUsageUSD, 1e-9, "user2 monthly must be overwritten to 28.8")
}
// TestBatchSnapshotUsage_NoRowIsNoop 锁定不变式:快照只落到既有活跃行;
// 行不存在或已软删的 (user, platform) 既不建行也不复活。
func TestBatchSnapshotUsage_NoRowIsNoop(t *testing.T) {
ctx := context.Background()
client := testEntClient(t)
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
now := time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)
snap := func(platform string) UserPlatformQuotaSnapshot {
return UserPlatformQuotaSnapshot{
UserID: userID,
Platform: platform,
DailyUsageUSD: 1.0,
WeeklyUsageUSD: 2.0,
MonthlyUsageUSD: 3.0,
DailyWindowStart: time.Date(2026, 5, 29, 0, 0, 0, 0, time.UTC),
WeeklyWindowStart: time.Date(2026, 5, 25, 0, 0, 0, 0, time.UTC),
MonthlyWindowStart: time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC),
}
}
// 从未有行
require.NoError(t, repo.BatchSnapshotUsage(ctx, []UserPlatformQuotaSnapshot{snap("anthropic")}, now))
rec, err := repo.GetByUserPlatform(ctx, userID, "anthropic")
require.NoError(t, err)
require.Nil(t, rec, "snapshot must not create a row")
// 有过行但已被软删(管理员清空限额)
mustSeedQuotaRow(t, ctx, repo, userID, "openai")
require.NoError(t, repo.UpsertForUser(ctx, userID, nil))
require.NoError(t, repo.BatchSnapshotUsage(ctx, []UserPlatformQuotaSnapshot{snap("openai")}, now))
rec, err = repo.GetByUserPlatform(ctx, userID, "openai")
require.NoError(t, err)
require.Nil(t, rec, "snapshot must not revive or recreate a soft-deleted row")
// 软删 mixin 默认隐藏已删行,计数全部行需显式跳过
all, err := client.UserPlatformQuota.Query().
Where(userplatformquota.UserIDEQ(userID)).
All(mixins.SkipSoftDelete(ctx))
require.NoError(t, err)
require.Len(t, all, 1, "only the soft-deleted openai row may exist")
require.NotNil(t, all[0].DeletedAt)
}
@@ -7,6 +7,7 @@ import (
"testing"
"time"
"github.com/Wei-Shaw/sub2api/ent/schema/mixins"
"github.com/Wei-Shaw/sub2api/ent/userplatformquota"
"github.com/stretchr/testify/require"
)
@@ -146,3 +147,44 @@ func TestUpsertForUser_EmptyClearsAll(t *testing.T) {
require.NoError(t, err)
require.Empty(t, got)
}
// TestUpsertForUser_UnlimitedRecordSoftDeletesAndDiscardsUsage 锁定语义:三档全空的记录视为未配置,
// 等价于该平台不在列表里 → 软删既有行并放弃其累计用量;之后再配置限额会新建行、从新窗口起算。
func TestUpsertForUser_UnlimitedRecordSoftDeletesAndDiscardsUsage(t *testing.T) {
ctx := context.Background()
client := testEntClient(t)
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
d := 10.0
require.NoError(t, repo.UpsertForUser(ctx, userID, []UserPlatformQuotaRecord{
{UserID: userID, Platform: "anthropic", DailyLimitUSD: &d},
}))
now := time.Date(2026, 5, 22, 10, 0, 0, 0, time.UTC)
require.NoError(t, repo.IncrementUsageWithReset(ctx, userID, "anthropic", 7.5, now))
// 清空限额:提交三档全空的记录
require.NoError(t, repo.UpsertForUser(ctx, userID, []UserPlatformQuotaRecord{
{UserID: userID, Platform: "anthropic"},
}))
gone, err := repo.GetByUserPlatform(ctx, userID, "anthropic")
require.NoError(t, err)
require.Nil(t, gone, "an unlimited record must soft-delete the row instead of keeping it with NULL limits")
// 重新配置限额:新行从零起算
require.NoError(t, repo.UpsertForUser(ctx, userID, []UserPlatformQuotaRecord{
{UserID: userID, Platform: "anthropic", DailyLimitUSD: &d},
}))
back, err := repo.GetByUserPlatform(ctx, userID, "anthropic")
require.NoError(t, err)
require.NotNil(t, back)
require.InDelta(t, 0, back.DailyUsageUSD, 1e-9, "usage restarts from zero")
require.Nil(t, back.DailyWindowStart, "window restarts")
// 软删 mixin 默认隐藏已删行,计数全部行需显式跳过
all, err := client.UserPlatformQuota.Query().
Where(userplatformquota.UserIDEQ(userID), userplatformquota.PlatformEQ("anthropic")).
All(mixins.SkipSoftDelete(ctx))
require.NoError(t, err)
require.Len(t, all, 2, "one soft-deleted history row plus one active row")
}
@@ -75,6 +75,7 @@ func TestEmailOAuthAuto_SnapshotsPlatformQuotaDefaults(t *testing.T) {
require.Len(t, quotaRepo.bulkInsertCalls, 1, "createEmailOAuthUser must snapshot platform quotas via BulkInsertInitial")
records := quotaRepo.bulkInsertCalls[0]
require.Len(t, records, 1, "only platforms with a configured limit get a row")
var geminiRecord *UserPlatformQuotaRecord
for i := range records {
if records[i].Platform == "gemini" {
@@ -539,6 +539,7 @@ func TestFinalizeOAuthEmailAccount_SnapshotsPlatformQuotaDefaults(t *testing.T)
require.Len(t, quotaRepo.bulkInsertCalls, 1, "snapshotPlatformQuotaDefaults must call BulkInsertInitial once on successful OAuth signup")
records := quotaRepo.bulkInsertCalls[0]
require.Len(t, records, 1, "only platforms with a configured limit get a row")
var anthropicRecord *UserPlatformQuotaRecord
for i := range records {
if records[i].Platform == "anthropic" {
+15 -10
View File
@@ -1930,7 +1930,7 @@ func resolvedTokenVersion(user *User) int64 {
return user.TokenVersion ^ fingerprint
}
// snapshotPlatformQuotaDefaults 把 plan.PlatformQuotas(platform × 3 window)以
// snapshotPlatformQuotaDefaults 把 plan.PlatformQuotas 中至少配置了一档限额的平台以
// BulkInsertInitial 形式写入 user_platform_quotas 表。失败 fail-open(仅 warn log)。
func (s *AuthService) snapshotPlatformQuotaDefaults(ctx context.Context, userID int64, plan *signupGrantPlan) error {
if s.userPlatformQuotaRepo == nil || plan == nil || len(plan.PlatformQuotas) == 0 {
@@ -1941,18 +1941,23 @@ func (s *AuthService) snapshotPlatformQuotaDefaults(ctx context.Context, userID
// 整个调用方事务被 Postgres 标记 aborted,把"无关紧要的默认配额快照"放大成
// "整笔注册失败"(OAuth pending 路径曾因此 500 → 清 cookie → 404)。
ctx = dbent.WithoutTx(ctx)
// 仅为至少配置了一档限额的平台建行:user_platform_quotas 中不存在的行等价于不限额,
// 三档全空的记录不携带任何可执行的限额。
records := make([]UserPlatformQuotaRecord, 0, len(plan.PlatformQuotas))
for platform, q := range plan.PlatformQuotas {
rec := UserPlatformQuotaRecord{
UserID: userID,
Platform: platform,
if !q.HasAnyLimit() {
continue
}
if q != nil {
rec.DailyLimitUSD = q.DailyLimitUSD
rec.WeeklyLimitUSD = q.WeeklyLimitUSD
rec.MonthlyLimitUSD = q.MonthlyLimitUSD
}
records = append(records, rec)
records = append(records, UserPlatformQuotaRecord{
UserID: userID,
Platform: platform,
DailyLimitUSD: q.DailyLimitUSD,
WeeklyLimitUSD: q.WeeklyLimitUSD,
MonthlyLimitUSD: q.MonthlyLimitUSD,
})
}
if len(records) == 0 {
return nil
}
if err := s.userPlatformQuotaRepo.BulkInsertInitial(ctx, records); err != nil {
logger.LegacyPrintf("service.auth", "[Auth] Warning: snapshot platform quota failed user=%d: %v (fail-open)", userID, err)
@@ -15,6 +15,7 @@ import (
type fakeInsertRecorder struct {
records []UserPlatformQuotaRecord
err error
calls int // BulkInsertInitial 被调用的次数
lastCtx context.Context // 捕获最后一次 BulkInsertInitial 收到的 ctx(用于断言事务隔离)
}
@@ -23,6 +24,7 @@ func (f *fakeInsertRecorder) GetByUserPlatform(_ context.Context, _ int64, _ str
}
func (f *fakeInsertRecorder) BulkInsertInitial(ctx context.Context, recs []UserPlatformQuotaRecord) error {
f.calls++
f.lastCtx = ctx
if f.err != nil {
return f.err
@@ -51,33 +53,70 @@ func (f *fakeInsertRecorder) BatchSnapshotUsage(_ context.Context, _ []UserPlatf
return nil
}
// TestSnapshotPlatformQuotaDefaults_PassesToRepoBulkInsert 锁定不变式:只有至少配置了一档限额的
// 平台才建行;三档全空(含 nil 条目)的平台不进入 BulkInsertInitial。
func TestSnapshotPlatformQuotaDefaults_PassesToRepoBulkInsert(t *testing.T) {
fakeRepo := &fakeInsertRecorder{}
s := &AuthService{userPlatformQuotaRepo: fakeRepo}
five := 5.0
zero := 0.0
plan := &signupGrantPlan{
PlatformQuotas: map[string]*DefaultPlatformQuotaSetting{
"anthropic": {DailyLimitUSD: &five},
"grok": {MonthlyLimitUSD: &zero}, // 0 = 显式禁用,属于已配置
"openai": {},
"gemini": {},
"antigravity": {},
"antigravity": nil,
},
}
if err := s.snapshotPlatformQuotaDefaults(context.Background(), 999, plan); err != nil {
t.Fatal(err)
}
if len(fakeRepo.records) != 4 {
t.Errorf("expected 4 records, got %d", len(fakeRepo.records))
if len(fakeRepo.records) != 2 {
t.Fatalf("expected 2 records (anthropic, grok), got %d: %+v", len(fakeRepo.records), fakeRepo.records)
}
found := false
byPlatform := map[string]UserPlatformQuotaRecord{}
for _, r := range fakeRepo.records {
if r.UserID == 999 && r.Platform == "anthropic" && r.DailyLimitUSD != nil && *r.DailyLimitUSD == 5 {
found = true
if r.UserID != 999 {
t.Errorf("unexpected user id %d", r.UserID)
}
byPlatform[r.Platform] = r
}
if r, ok := byPlatform["anthropic"]; !ok || r.DailyLimitUSD == nil || *r.DailyLimitUSD != 5 {
t.Error("anthropic daily = 5 not snapshotted")
}
if r, ok := byPlatform["grok"]; !ok || r.MonthlyLimitUSD == nil || *r.MonthlyLimitUSD != 0 {
t.Error("grok monthly = 0 (explicit disable) must be snapshotted")
}
for _, p := range []string{"openai", "gemini", "antigravity"} {
if _, ok := byPlatform[p]; ok {
t.Errorf("platform %q has no configured limit and must not get a row", p)
}
}
if !found {
t.Error("anthropic daily = 5 not snapshotted")
}
// TestSnapshotPlatformQuotaDefaults_AllUnlimitedSkipsInsert 锁定:没有任何平台配置限额时,
// 不调用 BulkInsertInitial(不产生全空的占位行)。
func TestSnapshotPlatformQuotaDefaults_AllUnlimitedSkipsInsert(t *testing.T) {
fakeRepo := &fakeInsertRecorder{}
s := &AuthService{userPlatformQuotaRepo: fakeRepo}
plan := &signupGrantPlan{
PlatformQuotas: map[string]*DefaultPlatformQuotaSetting{
"anthropic": {},
"openai": {},
"gemini": nil,
},
}
if err := s.snapshotPlatformQuotaDefaults(context.Background(), 999, plan); err != nil {
t.Fatal(err)
}
if fakeRepo.calls != 0 {
t.Fatalf("BulkInsertInitial must not be called when no platform has a configured limit, got %d calls", fakeRepo.calls)
}
if len(fakeRepo.records) != 0 {
t.Errorf("expected no records, got %d", len(fakeRepo.records))
}
}
@@ -298,19 +298,29 @@ func TestAuthService_Register_SnapshotsPlatformQuotaDefaults(t *testing.T) {
require.Len(t, quotaRepo.bulkInsertCalls, 1)
records := quotaRepo.bulkInsertCalls[0]
var openaiRecord *UserPlatformQuotaRecord
for i := range records {
if records[i].Platform == "openai" {
openaiRecord = &records[i]
break
}
}
require.NotNil(t, openaiRecord, "expected openai platform record")
require.Len(t, records, 1, "only platforms with a configured limit get a row")
openaiRecord := records[0]
require.Equal(t, "openai", openaiRecord.Platform)
require.Equal(t, int64(77), openaiRecord.UserID)
require.NotNil(t, openaiRecord.WeeklyLimitUSD)
require.InDelta(t, 12.34, *openaiRecord.WeeklyLimitUSD, 0.0001)
}
func TestAuthService_Register_NoDefaultQuotasSkipsSnapshot(t *testing.T) {
repo := &userRepoStub{nextID: 78}
quotaRepo := &userPlatformQuotaRepoStub{}
service := newAuthService(repo, map[string]string{
SettingKeyRegistrationEnabled: "true",
}, nil, quotaRepo)
_, user, err := service.Register(context.Background(), "newuser2@test.com", "password")
require.NoError(t, err)
require.NotNil(t, user)
require.Empty(t, quotaRepo.bulkInsertCalls, "no configured default limit must not create quota rows")
}
func TestAuthService_Register_DoesNotSnapshotOnDisabled(t *testing.T) {
repo := &userRepoStub{}
quotaRepo := &userPlatformQuotaRepoStub{}
@@ -1139,10 +1139,9 @@ func (s *BillingCacheService) checkUserPlatformQuotaEligibility(
// 超时 50ms:覆盖正常路径与可接受抖动;Redis 异常时 hot path 不阻塞超过此值。
// 用 context.Background()+短超时,避免请求 ctx 取消导致刷新丢失。
// 显式 setCancel()(而非 defer):缩短 context 生命周期,避免 defer 延迟到函数返回。
// isSentinel 判定「该 entry 无任何 limit」,涵盖两类,跨窗口命中时都跳过 refresh:
// 1) A3 回填的 sentinel(DB 无行,短 TTL):refresh 会把短 TTL 误升级为 86400s,有害;
// 2) DB 有行但三 limit 全未配置的用户(TTL 86400s):refresh 纯属无意义(TTL 升级本身无害)。
// 两类的 enforcement(下方 limit!=nil 比较)都因 limit 全 nil 永远放行,跳过 refresh 均正确。
// isSentinel 判定「该 entry 无任何 limit」(DB 无行时回填的短 TTL sentinel),跨窗口命中时跳过 refresh:
// refresh 会把短 TTL 误升级为 86400s;其 enforcement(下方 limit!=nil 比较)因 limit 全 nil 永远放行,
// 跳过 refresh 不改变结果。
isSentinel := entry.DailyLimitUSD == nil && entry.WeeklyLimitUSD == nil && entry.MonthlyLimitUSD == nil
if windowExpired && s.cache != nil && !isSentinel {
refreshed := &UserPlatformQuotaCacheEntry{
@@ -162,6 +162,11 @@ type DefaultPlatformQuotaSetting struct {
MonthlyLimitUSD *float64 `json:"monthly"`
}
// HasAnyLimit 报告是否至少配置了一档限额(0 也算配置)。nil receiver 视为未配置。
func (q *DefaultPlatformQuotaSetting) HasAnyLimit() bool {
return q != nil && (q.DailyLimitUSD != nil || q.WeeklyLimitUSD != nil || q.MonthlyLimitUSD != nil)
}
type ProviderDefaultGrantSettings struct {
Balance float64
Concurrency int
@@ -180,7 +180,8 @@ func (s *UserPlatformQuotaUsageFlusher) flushOneBatch(parentCtx context.Context)
// 读到旧 usage 快照(此刻 member 已离开脏集),而 admin 随后写 DB、本行 UPSERT 又在 admin 写之后落库,
// 则旧快照会覆盖 admin 刚写入的值;DeleteCache 后 Redis MISS,下次 preflight 从 DB 重载被覆盖的旧值。
// 因 member 已被 SPOP,admin 侧 SREM/清脏标记无法拦截本批(故未做)。影响有限,暂列为已知取舍:
// - UpsertForUser 改 limit,而本 UPSERT 不写 limit 列 → limit 配置不受影响;
// - UpsertForUser 改 limit,而本 UPDATE 只写既有活跃行的 usage/window 列 → limit 配置不受影响,
// 被软删的行不会被本 UPDATE 复活或重建;
// - ResetExpiredWindow 改 usage,但 preflight windowExpired 会在窗口真正过期时自愈重置,
// 仅"强制重置未过期窗口"且与本批精确交错时短暂失效;
// - 低频 admin 操作 + 默认 flusher_enabled=false。彻底消除需 version OCC(DB 加 version 列条件 UPSERT),
@@ -193,11 +194,9 @@ func (s *UserPlatformQuotaUsageFlusher) flushOneBatch(parentCtx context.Context)
if writeErr != nil {
if errors.Is(writeErr, ErrUserPlatformQuotaFKViolation) {
// 注意:PG FK violation 是整条 INSERT 回滚 → 整批(含同批正常用户)均未写入 DB,
// 且这些 key 已被 SPOP 出脏集、此处不 Readd。活跃 key 会在下次请求重新 SADD,
// flusher 读 Redis 当前累计绝对值刷库即自愈;低活跃 key 这轮 DB usage 偏低
// (Redis 仍是 enforcement 权威,不受影响;DB 仅展示)。已删用户边角的接受取舍,不做逐行重试。
// FK 违反:用户已被删除,直接丢弃不 Readd
// FK 违反表示对应用户已不存在,快照无处可落:整批丢弃、不 Readd。
// 同批其它 key 会在下次请求重新 SADD,flusher 读 Redis 当前累计绝对值刷库即自愈
// (Redis 仍是 enforcement 权威;DB 仅展示)。
s.metrics.FlushFKViolationTotal.Add(1)
s.metrics.FlushErrorTotal.Add(1)
logger.LegacyPrintf("quota_flusher", "[QuotaFlusher] FK violation (dropped %d snaps): %v", len(snaps), writeErr)
@@ -11,8 +11,8 @@ import (
// handler 只需引用 service 包,无需直接依赖 repository 包。
var ErrUserPlatformQuotaNotFound = errors.New("user platform quota not found")
// ErrUserPlatformQuotaFKViolation service 层 sentinel:批量 snapshot UPSERT 时存在
// user_id 不在 users 表的记录(外键违反)。adapter 负责将 repository 层同名 sentinel 包装为此错误。
// ErrUserPlatformQuotaFKViolation service 层 sentinel:批量快照写入命中了 user_id 不在 users 表的记录。
// adapter 负责将 repository 层同名 sentinel 包装为此错误。
var ErrUserPlatformQuotaFKViolation = errors.New("user platform quota snapshot FK violation")
// UserPlatformQuotaSnapshot 是 service 层 flusher 向 DB 写入快照时使用的传输结构。
@@ -44,26 +44,33 @@ type UserPlatformQuotaRecord struct {
MonthlyWindowStart *time.Time
}
// HasAnyLimit 报告该记录是否至少配置了一档限额(0 也算配置)。
// 三档全空的记录视为未配置:不写入 user_platform_quotas,等价于不限额。
func (r UserPlatformQuotaRecord) HasAnyLimit() bool {
return r.DailyLimitUSD != nil || r.WeeklyLimitUSD != nil || r.MonthlyLimitUSD != nil
}
// UserPlatformQuotaRepository 定义 service 层所需的 user × platform quota 数据访问端口。
// repository 包的 userPlatformQuotaRepository 实现此接口。
type UserPlatformQuotaRepository interface {
// GetByUserPlatform 查询单条配额记录,未找到时返回 (nil, nil)。
GetByUserPlatform(ctx context.Context, userID int64, platform string) (*UserPlatformQuotaRecord, error)
// BulkInsertInitial 幂等批量插入初始配额记录(ON CONFLICT DO NOTHING)。
// BulkInsertInitial 幂等批量插入初始配额记录:冲突时只补既有行为 NULL 的档位;三档全空的记录跳过。
BulkInsertInitial(ctx context.Context, records []UserPlatformQuotaRecord) error
// IncrementUsageWithReset 原子地累加用量,若窗口已过期则先重置再累加。
// IncrementUsageWithReset 原子地累加用量,若窗口已过期则先重置再累加;记录不存在时不建行、直接返回 nil。
IncrementUsageWithReset(ctx context.Context, userID int64, platform string, cost float64, now time.Time) error
// ListByUser 查询用户的所有平台配额记录。
ListByUser(ctx context.Context, userID int64) ([]UserPlatformQuotaRecord, error)
// UpsertForUser 全量替换该用户所有平台限额配置(事务内):
// 1. 软删除未在 records 中出现的所有 active 行
// 2. 对 records 中每条:UPDATE 已存在的(含重新激活软删行);UPDATE 未命中时 INSERT
// 仅改 *_limit_usd + deleted_at + updated_at,保留 *_usage_usd / *_window_start。
// 1. 软删除未在 records 中出现、或在 records 中但三档全空的所有 active 行
// 2. 对每条至少配置了一档限额的 record:UPDATE 已存在的 active 行;未命中时 INSERT
// 仅改 *_limit_usd + updated_at,保留 *_usage_usd / *_window_start。
// 清空某平台的限额即软删该行并放弃其累计用量;之后再次配置限额会新建行、从新窗口起算。
// records 为空时仅执行步骤 1。
UpsertForUser(ctx context.Context, userID int64, records []UserPlatformQuotaRecord) error
// ResetExpiredWindow 重置指定窗口("daily"|"weekly"|"monthly")的用量与起始时间。
// 未命中活跃记录时返回(service-side wrapper of repository.ErrUserPlatformQuotaNotFound)。
ResetExpiredWindow(ctx context.Context, userID int64, platform string, window string, newStart time.Time) error
// BatchSnapshotUsage 绝对值覆盖写入整批 usage 快照。FK 违反返回 ErrUserPlatformQuotaFKViolation。
// BatchSnapshotUsage 绝对值覆盖写入整批 usage 快照到既有活跃行;行不存在或已软删的 (user,platform) 跳过。
BatchSnapshotUsage(ctx context.Context, snapshots []UserPlatformQuotaSnapshot, now time.Time) error
}
@@ -0,0 +1,10 @@
-- user_platform_quotas 仅保存至少配置了一档限额的记录;不存在的行等价于不限额。
-- 三档限额全为 NULL 的行不携带任何可执行的限额,也不参与任何读取,直接删除。
-- 幂等:重复执行影响 0 行。
-- 行数约为历史用户数 × 平台数,整条语句在一个事务内执行;超大表需按
-- SETUP_MIGRATION_TIMEOUT_SECONDS 调高迁移超时,或在升级前预先分批清理。
DELETE FROM user_platform_quotas
WHERE daily_limit_usd IS NULL
AND weekly_limit_usd IS NULL
AND monthly_limit_usd IS NULL;
@@ -0,0 +1,27 @@
package migrations
import (
"strings"
"testing"
"github.com/stretchr/testify/require"
)
// TestUserPlatformQuotasPurgeUnlimitedMigration 校验 238 号迁移只删除三档限额全为 NULL 的行:
// 这类行等价于"不存在",任何一档非 NULL 的记录(含软删历史)都必须保留。
func TestUserPlatformQuotasPurgeUnlimitedMigration(t *testing.T) {
content, err := FS.ReadFile("238_purge_unlimited_user_platform_quotas.sql")
require.NoError(t, err)
// 去掉注释行后只允许这一条 DELETE。
var stmts []string
for _, line := range strings.Split(string(content), "\n") {
trimmed := strings.TrimSpace(line)
if trimmed == "" || strings.HasPrefix(trimmed, "--") {
continue
}
stmts = append(stmts, trimmed)
}
sql := strings.Join(stmts, " ")
require.Equal(t, "DELETE FROM user_platform_quotas WHERE daily_limit_usd IS NULL AND weekly_limit_usd IS NULL AND monthly_limit_usd IS NULL;", sql)
}
@@ -43,8 +43,8 @@
<button
type="button"
class="text-xs text-gray-400 hover:text-amber-500 disabled:opacity-50"
:disabled="!!resetting[`${row.platform}.daily`]"
:title="t('admin.users.platformQuota.reset.button')"
:disabled="!!resetting[`${row.platform}.daily`] || !savedConfigured.has(row.platform)"
:title="t(savedConfigured.has(row.platform) ? 'admin.users.platformQuota.reset.button' : 'admin.users.platformQuota.reset.unavailable')"
@click="onReset(row.platform, 'daily')"
>↻</button>
</div>
@@ -62,8 +62,8 @@
<button
type="button"
class="text-xs text-gray-400 hover:text-amber-500 disabled:opacity-50"
:disabled="!!resetting[`${row.platform}.weekly`]"
:title="t('admin.users.platformQuota.reset.button')"
:disabled="!!resetting[`${row.platform}.weekly`] || !savedConfigured.has(row.platform)"
:title="t(savedConfigured.has(row.platform) ? 'admin.users.platformQuota.reset.button' : 'admin.users.platformQuota.reset.unavailable')"
@click="onReset(row.platform, 'weekly')"
>↻</button>
</div>
@@ -81,8 +81,8 @@
<button
type="button"
class="text-xs text-gray-400 hover:text-amber-500 disabled:opacity-50"
:disabled="!!resetting[`${row.platform}.monthly`]"
:title="t('admin.users.platformQuota.reset.button')"
:disabled="!!resetting[`${row.platform}.monthly`] || !savedConfigured.has(row.platform)"
:title="t(savedConfigured.has(row.platform) ? 'admin.users.platformQuota.reset.button' : 'admin.users.platformQuota.reset.unavailable')"
@click="onReset(row.platform, 'monthly')"
>↻</button>
</div>
@@ -148,6 +148,18 @@ const loading = ref(false)
const submitting = ref(false)
const resetting = reactive<Record<string, boolean>>({})
const quotas = ref<QuotaRow[]>([])
// 已保存且至少配置了一档限额的平台。只有这些平台在后端有配额记录,重置用量窗口才有对象。
const savedConfigured = ref<Set<PlatformQuotaPlatform>>(new Set())
function configuredPlatforms(items: PlatformQuotaItem[]): Set<PlatformQuotaPlatform> {
const out = new Set<PlatformQuotaPlatform>()
for (const it of items) {
if (it.daily_limit_usd != null || it.weekly_limit_usd != null || it.monthly_limit_usd != null) {
out.add(it.platform)
}
}
return out
}
function emptyRow(p: PlatformQuotaPlatform): QuotaRow {
return {
@@ -190,9 +202,11 @@ async function load() {
try {
const data = await adminAPI.users.getPlatformQuotas(props.user.id)
quotas.value = normalize(data.platform_quotas || [])
savedConfigured.value = configuredPlatforms(data.platform_quotas || [])
} catch {
appStore.showError(t('admin.users.platformQuota.loadFailed'))
quotas.value = PLATFORMS.map(emptyRow)
savedConfigured.value = new Set()
} finally {
loading.value = false
}
@@ -273,6 +287,7 @@ async function onReset(platform: PlatformQuotaPlatform, quotaWindow: PlatformQuo
try {
const data = await adminAPI.users.resetPlatformQuotaWindow(props.user.id, platform, quotaWindow)
quotas.value = normalize(data.platform_quotas || [])
savedConfigured.value = configuredPlatforms(data.platform_quotas || [])
appStore.showSuccess(t('admin.users.platformQuota.reset.success', { platform, window: windowLabel }))
} catch (e: any) {
appStore.showError(e?.response?.data?.message || t('admin.users.platformQuota.reset.failed'))
@@ -168,7 +168,42 @@ describe('UserPlatformQuotaModal', () => {
confirmSpy.mockRestore()
})
// 只有已保存限额的平台在后端有配额记录,重置按钮才可用
const anthropicConfigured = {
platform_quotas: [
{
platform: 'anthropic',
daily_limit_usd: 10,
weekly_limit_usd: null,
monthly_limit_usd: null,
daily_usage_usd: 0,
weekly_usage_usd: 0,
monthly_usage_usd: 0,
},
],
}
it('未配置限额的平台重置按钮禁用并提示不可用', async () => {
const w = await mountAndOpen()
const resetBtns = w.findAll('button').filter((b) => b.text() === '↻')
expect(resetBtns.length).toBe(15) // 5 平台 × 3 窗口
for (const b of resetBtns) {
expect((b.element as HTMLButtonElement).disabled).toBe(true)
expect(b.attributes('title')).toBe('admin.users.platformQuota.reset.unavailable')
}
})
it('已保存限额的平台重置按钮可用,其余仍禁用', async () => {
apiMocks.getPlatformQuotas.mockResolvedValue(anthropicConfigured)
const w = await mountAndOpen()
const resetBtns = w.findAll('button').filter((b) => b.text() === '↻')
const enabled = resetBtns.filter((b) => !(b.element as HTMLButtonElement).disabled)
expect(enabled.length).toBe(3) // anthropic 的 daily/weekly/monthly
expect(enabled[0].attributes('title')).toBe('admin.users.platformQuota.reset.button')
})
it('重置按钮 confirm 取消则不调用 API', async () => {
apiMocks.getPlatformQuotas.mockResolvedValue(anthropicConfigured)
const confirmSpy = vi.spyOn(window, 'confirm').mockReturnValue(false)
const w = await mountAndOpen()
const resetBtns = w.findAll('button').filter((b) => b.text() === '↻')
@@ -180,6 +215,7 @@ describe('UserPlatformQuotaModal', () => {
})
it('重置按钮 confirm 确认则调用 API', async () => {
apiMocks.getPlatformQuotas.mockResolvedValue(anthropicConfigured)
const confirmSpy = vi.spyOn(window, 'confirm').mockReturnValue(true)
const w = await mountAndOpen()
const resetBtns = w.findAll('button').filter((b) => b.text() === '↻')
@@ -137,13 +137,15 @@
<div class="mb-3 flex items-center justify-between">
<h3 class="text-sm font-semibold text-gray-900 dark:text-white">{{ t('dashboard.platformBreakdown') }}</h3>
<span class="text-xs text-gray-500 dark:text-gray-400">
{{ t('dashboard.platformCount', { count: sortedPlatforms.length }) }}
{{ t('dashboard.platformCount', { count: platformCount }) }}
</span>
</div>
<div class="grid grid-cols-1 gap-3 sm:grid-cols-2 lg:grid-cols-4">
<div
v-for="item in platformCards"
:key="item.platform"
data-testid="platform-card"
:data-platform="item.platform"
:class="[
'rounded-lg border p-3',
item.isOther
@@ -226,7 +228,7 @@
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
import Icon from '@/components/icons/Icon.vue'
import type { UserDashboardStats as UserStatsType } from '@/api/usage'
import type { PlatformDashboardStats, UserDashboardStats as UserStatsType } from '@/api/usage'
import type { PlatformQuotaItem } from '@/types'
interface FusedPlatformCard {
@@ -261,27 +263,27 @@ const PLATFORM_LABELS: Record<string, string> = {
const platformLabel = (p: string) => PLATFORM_LABELS[p] ?? p
const sortedPlatforms = computed(() => {
const list = props.stats?.by_platform ?? []
return [...list].sort((a, b) => b.total_actual_cost - a.total_actual_cost)
})
// 处理"各平台之和 < 总值"的差值:后端按平台聚合时过滤了无法归属平台的行
// (group 与 account 都缺 platform)。这里把差值作为"其他"卡片显式展示,
// 避免 Row 1 总值与 Row 3 平台拆分加总对不上、用户困惑。
const OTHER_THRESHOLD = 0.0001
const platformCards = computed<FusedPlatformCard[]>(() => {
// 建立 by_platform Map
const byPlat = new Map<string, (typeof sortedPlatforms.value)[number]>()
const byPlat = new Map<string, PlatformDashboardStats>()
for (const item of props.stats?.by_platform ?? []) byPlat.set(item.platform, item)
// 建立 quota Map
// 建立 quota Map。三档全空的记录不产生卡片,挂到卡片上也不渲染配额区。
const byQuota = new Map<string, PlatformQuotaItem>()
for (const q of props.platformQuotas ?? []) byQuota.set(q.platform, q)
// union 平台集合。后端 by_platform / quota 接口均不会返回 platform='__other__',
// 卡片集合 = 有用量的平台 ∪ 至少配置了一档限额的平台。
// 三档全空的限额记录等价于不限额,不单独产生卡片。
// 后端 by_platform / quota 接口均不会返回 platform='__other__',
// 无需显式排除;__other__ 由下方差值补差逻辑单独追加。
const platforms = new Set<string>([...byPlat.keys(), ...byQuota.keys()])
const platforms = new Set<string>(byPlat.keys())
for (const [platform, q] of byQuota) {
if (hasAnyLimit(q)) platforms.add(platform)
}
const PLATFORM_ORDER = ['anthropic', 'openai', 'gemini', 'antigravity', 'grok']
const cards: FusedPlatformCard[] = []
@@ -330,6 +332,9 @@ const platformCards = computed<FusedPlatformCard[]>(() => {
return cards
})
// 标题右侧的平台计数 = 实际渲染的平台卡片数,不含"其他"差额卡。
const platformCount = computed(() => platformCards.value.filter((c) => !c.isOther).length)
// Quota helpers
type QuotaWindow = 'daily' | 'weekly' | 'monthly'
@@ -0,0 +1,154 @@
import { describe, it, expect, vi } from 'vitest'
import { mount, type VueWrapper } from '@vue/test-utils'
// t() 回显 key;带参数时附上 JSON,便于断言计数
vi.mock('vue-i18n', async () => {
const actual = await vi.importActual<typeof import('vue-i18n')>('vue-i18n')
return {
...actual,
useI18n: () => ({
t: (key: string, params?: Record<string, unknown>) =>
params ? `${key}:${JSON.stringify(params)}` : key,
}),
}
})
import UserDashboardStats from '../UserDashboardStats.vue'
import type { UserDashboardStats as UserStatsType, PlatformDashboardStats } from '@/api/usage'
import type { PlatformQuotaItem } from '@/types'
function makeStats(over: Partial<UserStatsType> = {}): UserStatsType {
return {
total_api_keys: 1,
active_api_keys: 1,
total_requests: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_cache_creation_tokens: 0,
total_cache_read_tokens: 0,
total_tokens: 0,
total_cost: 0,
total_actual_cost: 0,
today_requests: 0,
today_input_tokens: 0,
today_output_tokens: 0,
today_cache_creation_tokens: 0,
today_cache_read_tokens: 0,
today_tokens: 0,
today_cost: 0,
today_actual_cost: 0,
average_duration_ms: 0,
rpm: 0,
tpm: 0,
by_platform: [],
...over,
}
}
function usage(platform: string, cost: number): PlatformDashboardStats {
return {
platform,
total_requests: 1,
total_tokens: 10,
total_actual_cost: cost,
today_requests: 1,
today_tokens: 10,
today_actual_cost: cost,
}
}
function quota(over: Partial<PlatformQuotaItem> & { platform: string }): PlatformQuotaItem {
return {
daily_limit_usd: null,
weekly_limit_usd: null,
monthly_limit_usd: null,
daily_usage_usd: 0,
weekly_usage_usd: 0,
monthly_usage_usd: 0,
...over,
} as PlatformQuotaItem
}
function mountStats(stats: UserStatsType, platformQuotas: PlatformQuotaItem[] | null = null, isSimple = false) {
return mount(UserDashboardStats, {
props: { stats, balance: 0, isSimple, platformQuotas },
global: { stubs: { Icon: true } },
})
}
/** 渲染出的平台卡片,按 DOM 顺序返回 data-platform */
function cardPlatforms(w: VueWrapper): string[] {
return w.findAll('[data-testid="platform-card"]').map((c) => c.attributes('data-platform') ?? '')
}
describe('UserDashboardStats 按平台拆分', () => {
it('只有用量的平台才产生卡片;三档全空的限额记录不产生卡片', () => {
const w = mountStats(
makeStats({ total_actual_cost: 0.03, today_actual_cost: 0.03, by_platform: [usage('grok', 0.03)] }),
[
quota({ platform: 'anthropic' }),
quota({ platform: 'openai' }),
quota({ platform: 'gemini' }),
quota({ platform: 'grok' }),
]
)
expect(cardPlatforms(w)).toEqual(['grok'])
expect(w.text()).toContain('dashboard.platformCount:{"count":1}')
expect(w.html()).not.toContain('dashboard.platformQuota.title')
})
it('配置了限额但没有用量的平台也产生卡片,并渲染配额区', () => {
const w = mountStats(
makeStats({ total_actual_cost: 0.03, today_actual_cost: 0.03, by_platform: [usage('grok', 0.03)] }),
[quota({ platform: 'openai', daily_limit_usd: 10, daily_usage_usd: 2.5 }), quota({ platform: 'anthropic' })]
)
// 固定顺序:openai 排在 grok 前
expect(cardPlatforms(w)).toEqual(['openai', 'grok'])
expect(w.text()).toContain('dashboard.platformQuota.title')
expect(w.text()).toContain('dashboard.platformCount:{"count":2}')
})
it('同一平台既有用量又有限额只产生一张卡片', () => {
const w = mountStats(
makeStats({ total_actual_cost: 1, today_actual_cost: 1, by_platform: [usage('openai', 1)] }),
[quota({ platform: 'openai', daily_limit_usd: 10, daily_usage_usd: 1 })]
)
expect(cardPlatforms(w)).toEqual(['openai'])
expect(w.text()).toContain('dashboard.platformQuota.title')
})
it('限额为 0 的平台视为已配置,渲染禁用态', () => {
const w = mountStats(makeStats(), [quota({ platform: 'gemini', weekly_limit_usd: 0 })])
expect(cardPlatforms(w)).toEqual(['gemini'])
expect(w.text()).toContain('dashboard.platformQuota.disabled')
expect(w.text()).toContain('dashboard.platformCount:{"count":1}')
})
it('固定顺序之外的平台也产生卡片,并排在固定顺序之后', () => {
const w = mountStats(
makeStats({ total_actual_cost: 0.5, today_actual_cost: 0, by_platform: [usage('kimi', 0.3), usage('anthropic', 0.2)] })
)
expect(cardPlatforms(w)).toEqual(['anthropic', 'kimi'])
expect(w.text()).toContain('Kimi')
})
it('总值大于各平台之和时追加"其他"卡片,且不计入平台计数', () => {
const w = mountStats(
makeStats({ total_actual_cost: 1.0, today_actual_cost: 0, by_platform: [usage('anthropic', 0.4)] })
)
expect(cardPlatforms(w)).toEqual(['anthropic', '__other__'])
expect(w.text()).toContain('dashboard.platformOther')
expect(w.text()).toContain('dashboard.platformCount:{"count":1}')
})
it('没有任何用量也没有配置限额时不渲染整块', () => {
const w = mountStats(makeStats(), [quota({ platform: 'anthropic' }), quota({ platform: 'openai' })])
expect(w.html()).not.toContain('dashboard.platformBreakdown')
expect(cardPlatforms(w)).toEqual([])
})
it('简易模式不渲染整块', () => {
const w = mountStats(makeStats({ by_platform: [usage('openai', 1)] }), null, true)
expect(w.html()).not.toContain('dashboard.platformBreakdown')
})
})
@@ -758,6 +758,7 @@ export default {
clearAllConfirm: 'Clear daily / weekly / monthly limits for ALL platforms? All platforms will become "unlimited" with no local undo — you must manually re-enter values before saving.',
reset: {
button: 'Reset window',
unavailable: 'No limit configured for this platform, so there is no usage window to reset',
confirm: 'Reset the {window} usage for {platform} for this user? This is effective immediately.',
success: 'Reset {platform} {window} usage',
failed: 'Reset failed',
@@ -407,7 +407,7 @@ export default {
subscriptionGroup: 'Subscription Group',
subscriptionValidityDays: 'Validity (days)',
defaultPlatformQuotas: 'Default Platform Quotas (on signup)',
defaultPlatformQuotasHint: 'Automatically assigned to new users on signup; existing users are not affected. Leave blank = unlimited.',
defaultPlatformQuotasHint: 'Applied to new users on signup; existing users are not affected. Leave blank = no limit for that platform and window.',
platformQuotaNotice: 'Monthly quota uses a 30-day rolling window, not a calendar month.',
},
platformQuota: {
@@ -755,6 +755,7 @@ export default {
clearAllConfirm: '确认清空全部平台的日 / 周 / 月限额?所有平台将变为"无限额",本地无法撤销,需要在保存前手动重填。',
reset: {
button: '重置该窗口',
unavailable: '该平台未配置限额,没有可重置的用量窗口',
confirm: '确认重置该用户 {platform} 平台的 {window} 用量?此操作立即生效。',
success: '已重置 {platform} {window} 用量',
failed: '重置失败',
@@ -402,7 +402,7 @@ export default {
subscriptionGroup: '订阅分组',
subscriptionValidityDays: '有效期(天)',
defaultPlatformQuotas: '默认平台限额(注册时分配)',
defaultPlatformQuotasHint: '新用户注册时自动写入平台限额记录;已有用户不受影响。留空 = 该平台该窗口不限制。',
defaultPlatformQuotasHint: '新用户注册时自动获得这里配置的限额;已有用户不受影响。留空 = 该平台该窗口不限制。',
platformQuotaNotice: '月限额为 30 天滚动窗口,非自然月',
},
platformQuota: {