feat: 并发分组系统 + 品牌/Logo 上传 + 兑换码开关 + 文档分辨率表

并发分组(新功能): - 新表 concurrency_groups(名称/上限/默认),用户加 concurrency_group_id - 启动自动建「默认并发」组(上限10、默认),老用户回填、新注册自动绑定 - 并发计数改用 Redis(自愈 sorted-set + TTL + fail-open): · 用户并发(画图台 + API key 合计)受其分组上限限制,0=不限制 → 超返回 429 · 账号级并发也从内存 gate 换成同一套 Redis(6 处调用点) · 移除旧的「已有正在生成的任务」单任务锁 - 后台「并发分组」新菜单:增删改、设默认、用户数;默认组不可删(删别的组成员转默认) - 用户管理:并发列 + 新建/编辑可选分组 - 个人设置页:账户信息卡(用户名/邮箱/角色/余额/并发);/me 暴露 concurrency_group/limit 品牌 / Logo(上传到 RustFS): - Logo 改成拖拽/点击上传,点保存才上传;替换自动删旧;branding/ 设为公开且被清理任务 pin 住(永不删) - 有自定义就用:前台左侧 nav + 后台侧栏 + favicon(浏览器标签);没有则默认 V 图标 - 前台页头还原成文字;首页 Hero 子标题用 site.subtitle(默认那句宣传语,设置页预填) - 邮件验证码标题用站点名;新增 POST/DELETE /settings/logo + POST /settings/asset(首页底图上传) 兑换码开关: - 系统设置→积分新增「开启兑换码」(默认开);关闭后后端拒绝兑换、前台隐藏兑换入口(/site 暴露 cdk_redeem_enabled) 文档 / 分辨率: - 去掉 quality 参数:size(宽x高)同时决定比例 + 分辨率档(长边映射 1K/2K/4K) - 文档加「分辨率对照表」(14 个比例 × 1K/2K/4K → size 该传的值);guessRatio 与自定义模型 RATIO_OPTS 对齐到 14 个其它: - 删模型时同步清掉各上游账号「支持模型」里的该 id - 首页设置/兑换码弹窗去固定高度滚动条;展示位弹窗浅色主题适配 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 13:52:31 +08:00
parent 1137d0bd1d
commit 5cf6206ee9
37 changed files with 1257 additions and 161 deletions
@@ -49,6 +49,9 @@ var (
 	// ErrConcurrencyFull — every eligible account is busy (each account runs at
 	// most ONE generation at a time). English message: surfaced to API / UI.
 	ErrConcurrencyFull = errors.New("all accounts are busy (1 concurrent job each), please try again shortly")
+	// ErrUserConcurrencyFull — the caller already has their concurrency-group's max
+	// generations in flight (画图台 + API key combined). 0 = unlimited.
+	ErrUserConcurrencyFull = errors.New("too many generations in progress, please wait for one to finish")
 	// ErrVideoJobNotFound / ErrVideoNotReady — /v1/videos async job lookups.
 	ErrVideoJobNotFound = errors.New("video job not found")
 	ErrVideoNotReady    = errors.New("video is not ready yet")
@@ -66,6 +69,7 @@ type V1Service struct {
 	events   *repo.EventRepository
 	tokens   *repo.TokenRepository
 	settings *repo.SiteSettingRepository
+	cgroups  *repo.ConcurrencyGroupRepository
 	adobe    *adobe.Client
 	chatgpt  *chatgpt.Client
 	runway   *runway.Client
@@ -93,48 +97,57 @@ type V1Service struct {
 	// for minutes and surface a late "success" on an already-abandoned event).
 	inflight *InflightRegistry

-	// gate enforces 1 concurrent generation PER account: a scheduler skips any
-	// account that's currently busy, and fails with ErrConcurrencyFull when every
-	// eligible account is occupied. In-memory (single process).
-	gate accountGate
+	// conc is the Redis-backed concurrency limiter for BOTH the per-account
+	// upstream gate (1+ jobs per account) and the per-user gate (画图台 + API key,
+	// capped by the user's concurrency group). Self-healing + fail-open.
+	conc *ConcurrencyService
 }

-// accountGate is a 1-slot-per-account in-flight gate. tryAcquire wins only if the
-// account isn't already running a generation; release frees it when done.
-type accountGate struct{ m sync.Map } // accountID -> struct{} held while busy
-
-// tryAcquireN wins if the account has fewer than max in-flight jobs, atomically
-// bumping its counter. max=1 is the default 1-job-per-account policy; some
-// providers (grok) allow more.
-func (g *accountGate) tryAcquireN(id string, max int) bool {
-	if id == "" {
-		return true
-	}
+// acctAcquire takes one per-account upstream slot (capped at max; 0/1 = single),
+// tagged with the generation's eventID (unique per job; a generation only ever
+// holds one slot on a given account at a time, so failover reuses it cleanly).
+func (s *V1Service) acctAcquire(ctx context.Context, accountID, eventID string, max int) bool {
 	if max < 1 {
 		max = 1
 	}
-	v, _ := g.m.LoadOrStore(id, new(int64))
-	cnt := v.(*int64)
-	for {
-		cur := atomic.LoadInt64(cnt)
-		if cur >= int64(max) {
-			return false
-		}
-		if atomic.CompareAndSwapInt64(cnt, cur, cur+1) {
-			return true
-		}
-	}
+	return s.conc.Acquire(ctx, "conc:a:"+accountID, max, eventID)
 }

-func (g *accountGate) tryAcquire(id string) bool { return g.tryAcquireN(id, 1) }
+func (s *V1Service) acctRelease(ctx context.Context, accountID, eventID string) {
+	s.conc.Release(ctx, "conc:a:"+accountID, eventID)
+}

-func (g *accountGate) release(id string) {
-	if id == "" {
-		return
+// userAcquire takes one per-user generation slot, capped by the user's
+// concurrency group (0 = unlimited). Returns false when the user is already at
+// their limit. `token` is a unique per-generation tag passed back to userRelease.
+func (s *V1Service) userAcquire(ctx context.Context, user *model.User, token string) bool {
+	if user == nil {
+		return true
 	}
-	if v, ok := g.m.Load(id); ok {
-		atomic.AddInt64(v.(*int64), -1)
+	return s.conc.Acquire(ctx, "conc:u:"+user.ID, s.userConcurrencyLimit(ctx, user), token)
+}
+
+func (s *V1Service) userRelease(ctx context.Context, userID, token string) {
+	s.conc.Release(ctx, "conc:u:"+userID, token)
+}
+
+// userConcurrencyLimit resolves the user's concurrency-group cap (0 = unlimited),
+// falling back to the default group when unset/missing.
+func (s *V1Service) userConcurrencyLimit(ctx context.Context, user *model.User) int {
+	if s.cgroups == nil || user == nil {
+		return 0
 	}
+	var g *model.ConcurrencyGroup
+	if user.ConcurrencyGroupID != "" {
+		g, _ = s.cgroups.Get(ctx, user.ConcurrencyGroupID)
+	}
+	if g == nil {
+		g, _ = s.cgroups.GetDefault(ctx)
+	}
+	if g == nil {
+		return 0
+	}
+	return g.MaxConcurrency
 }

 // InflightRegistry tracks the cancel func of every in-progress generation by
@@ -199,7 +212,7 @@ type V1VideoRequest struct {
 	BaseURL string
 }

-func NewV1Service(cfg *config.Config, models *repo.ModelRepository, users *repo.UserRepository, events *repo.EventRepository, tokens *repo.TokenRepository, settings *repo.SiteSettingRepository, adobeClient *adobe.Client, chatGPTClient *chatgpt.Client, runwayClient *runway.Client, leonardoClient *leonardo.Client, kreaClient *krea.Client, imagineClient *imagine.Client, grokClient *grok.Client, customClient *custom.Client, store *storage.Client) *V1Service {
+func NewV1Service(cfg *config.Config, models *repo.ModelRepository, users *repo.UserRepository, events *repo.EventRepository, tokens *repo.TokenRepository, settings *repo.SiteSettingRepository, cgroups *repo.ConcurrencyGroupRepository, conc *ConcurrencyService, adobeClient *adobe.Client, chatGPTClient *chatgpt.Client, runwayClient *runway.Client, leonardoClient *leonardo.Client, kreaClient *krea.Client, imagineClient *imagine.Client, grokClient *grok.Client, customClient *custom.Client, store *storage.Client) *V1Service {
 	return &V1Service{
 		cfg:      cfg,
 		models:   models,
@@ -207,6 +220,8 @@ func NewV1Service(cfg *config.Config, models *repo.ModelRepository, users *repo.
 		events:   events,
 		tokens:   tokens,
 		settings: settings,
+		cgroups:  cgroups,
+		conc:     conc,
 		adobe:    adobeClient,
 		chatgpt:  chatGPTClient,
 		runway:   runwayClient,
@@ -327,6 +342,16 @@ func (s *V1Service) prepareImageExecution(ctx context.Context, principal *APIPri
 	genCtx, cancel := context.WithTimeout(ctx, 8*time.Minute)
 	defer cancel()

+	// Per-user concurrency gate (画图台 + API key combined). Admin model-tests are
+	// exempt. Held for the whole generation; released on return.
+	if source != "admin" && principal != nil && principal.User != nil {
+		slot := randomUpper(12)
+		if !s.userAcquire(ctx, principal.User, slot) {
+			return nil, ErrUserConcurrencyFull
+		}
+		defer s.userRelease(ctx, principal.User.ID, slot)
+	}
+
 	modelItem, resolution, aspectRatio, price, err := s.prepareImage(ctx, principal, in, charge)
 	if err != nil {
 		return nil, err
@@ -550,6 +575,15 @@ func (s *V1Service) prepareVideoExecution(ctx context.Context, principal *APIPri
 	genCtx, cancel := context.WithTimeout(ctx, 12*time.Minute)
 	defer cancel()

+	// Per-user concurrency gate (画图台 + API key combined); admin tests exempt.
+	if source != "admin" && principal != nil && principal.User != nil {
+		slot := randomUpper(12)
+		if !s.userAcquire(ctx, principal.User, slot) {
+			return nil, ErrUserConcurrencyFull
+		}
+		defer s.userRelease(ctx, principal.User.ID, slot)
+	}
+
 	modelItem, resolution, aspectRatio, duration, price, err := s.prepareVideo(ctx, principal, in, charge)
 	if err != nil {
 		return nil, err
@@ -922,12 +956,11 @@ func (s *V1Service) prepareImage(ctx context.Context, principal *APIPrincipal, i
 	if err := ensureReferenceSizes(in.ReferenceImages); err != nil {
 		return nil, "", "", 0, err
 	}
+	// `size` (WxH) drives BOTH the aspect ratio AND the resolution tier — its long
+	// edge maps to a tier (<1800→1K, 1800–3499→2K, ≥3500→4K). The web path passes
+	// an explicit resolution; the OpenAI /v1 path derives it from size. There is no
+	// `quality` param — size is the single source of truth for resolution.
 	aspectRatio, resolution := parseImageSize(in.Size, in.AspectRatio, in.Resolution)
-	// Strict OpenAI path (/v1) sends no resolution — pick the tier from `quality`
-	// (low/medium/high/auto → 1K/2K/4K/default), clamped to the model's tiers.
-	if strings.TrimSpace(in.Resolution) == "" {
-		resolution = resolutionForQuality(modelItem, in.Quality)
-	}
 	// parseImageSize defaults a blank resolution to "2K" (OpenAI-size parity).
 	// For a model that doesn't price that tier — e.g. gpt-image-2 is 1K-only —
 	// fall back to its first supported tier so a missing/stale resolution from
@@ -1240,13 +1273,13 @@ func (s *V1Service) runPoolWithFailover(ctx context.Context, eventID, pool strin
 	tempDeadCount := 0
 	for _, token := range active {
 		// 1 concurrent job per account: skip any account already generating.
-		if !s.gate.tryAcquire(token.ID) {
+		if !s.acctAcquire(ctx, token.ID, eventID, 1) {
 			busy++
 			continue
 		}
 		// release via defer so a panic in tryAccount can't leak the 1-job slot.
 		data, err, failover, tempDead := func() ([]byte, error, bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			return s.tryAccount(ctx, eventID, pool, token, kind, attempt, classify, refreshOnAuth, tempAsDead)
 		}()
 		if err == nil {
@@ -1524,13 +1557,13 @@ func (s *V1Service) generateRunwayVideo(ctx context.Context, eventID string, mod
 	busy := 0
 	for _, token := range active {
 		// 1 concurrent job per account: skip any account already generating.
-		if !s.gate.tryAcquire(token.ID) {
+		if !s.acctAcquire(ctx, token.ID, eventID, 1) {
 			busy++
 			continue
 		}
 		var data []byte
 		done, failover := func() (bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			_ = s.events.SetAccount(ctx, eventID, token.ID)
 			_ = s.tokens.TouchLastUsed(ctx, token.ID)
 			teamID := ""
@@ -1663,13 +1696,13 @@ func (s *V1Service) generateCustomImage(ctx context.Context, eventID string, mod
 	var lastErr error
 	busy := 0
 	for _, token := range active {
-		if !s.gate.tryAcquireN(token.ID, accountConcurrency(token)) {
+		if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) {
 			busy++
 			continue
 		}
 		var data []byte
 		done, failover := func() (bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			_ = s.events.SetAccount(ctx, eventID, token.ID)
 			_ = s.tokens.TouchLastUsed(ctx, token.ID)
 			baseURL := stringValue(token.Meta["base_url"])
@@ -1730,13 +1763,13 @@ func (s *V1Service) generateCustomVideo(ctx context.Context, eventID string, mod
 	var videoURL string
 	busy := 0
 	for _, token := range active {
-		if !s.gate.tryAcquireN(token.ID, accountConcurrency(token)) {
+		if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) {
 			busy++
 			continue
 		}
 		var data []byte
 		done, failover := func() (bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			_ = s.events.SetAccount(ctx, eventID, token.ID)
 			_ = s.tokens.TouchLastUsed(ctx, token.ID)
 			baseURL := stringValue(token.Meta["base_url"])
@@ -1869,13 +1902,13 @@ func (s *V1Service) generateGrokVideo(ctx context.Context, eventID string, model
 	for _, token := range active {
 		// grok allows 10 concurrent jobs per account (unlike the 1-per-account
 		// default of the other pools).
-		if !s.gate.tryAcquireN(token.ID, grokConcurrencyPerAccount) {
+		if !s.acctAcquire(ctx, token.ID, eventID, grokConcurrencyPerAccount) {
 			busy++
 			continue
 		}
 		var data []byte
 		done, failover := func() (bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			_ = s.events.SetAccount(ctx, eventID, token.ID)
 			_ = s.tokens.TouchLastUsed(ctx, token.ID)
 			d, meta, genErr := s.grok.GenerateVideo(ctx, token.Value, in.Prompt, aspectRatio, res, durationSeconds, frames, downloadResult)
@@ -1970,13 +2003,13 @@ func (s *V1Service) generateRunwayImage(ctx context.Context, eventID string, mod
 	busy := 0
 	for _, token := range active {
 		// 1 concurrent job per account: skip any account already generating.
-		if !s.gate.tryAcquire(token.ID) {
+		if !s.acctAcquire(ctx, token.ID, eventID, 1) {
 			busy++
 			continue
 		}
 		var data []byte
 		done, failover := func() (bool, bool) {
-			defer s.gate.release(token.ID)
+			defer s.acctRelease(ctx, token.ID, eventID)
 			_ = s.events.SetAccount(ctx, eventID, token.ID)
 			_ = s.tokens.TouchLastUsed(ctx, token.ID)
 			teamID := ""
@@ -2571,7 +2604,14 @@ func guessRatio(w, h int) string {
 		W int
 		H int
 	}
-	candidates := []candidate{{1, 1}, {16, 9}, {9, 16}, {4, 3}, {3, 4}, {4, 1}, {1, 4}, {8, 1}, {1, 8}}
+	// The 14 ratios actually used across our models. Must stay in sync with the
+	// custom-model picker (CustomModelModal RATIO_OPTS) and the docs 对照表, so a
+	// /v1 `size` maps to exactly one of them.
+	candidates := []candidate{
+		{1, 1},
+		{5, 4}, {4, 3}, {3, 2}, {16, 9}, {2, 1}, {21, 9}, {3, 1}, // 横
+		{4, 5}, {3, 4}, {2, 3}, {9, 16}, {9, 21}, {1, 3}, // 竖
+	}
 	best := candidates[0]
 	bestDelta := absFloat(float64(w)/float64(h) - float64(best.W)/float64(best.H))
 	for _, item := range candidates[1:] {