From 5cf6206ee9b03f935318fa89cf8c8495340375c6 Mon Sep 17 00:00:00 2001 From: chiyi Date: Tue, 30 Jun 2026 13:52:31 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B9=B6=E5=8F=91=E5=88=86=E7=BB=84?= =?UTF-8?q?=E7=B3=BB=E7=BB=9F=20+=20=E5=93=81=E7=89=8C/Logo=20=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0=20+=20=E5=85=91=E6=8D=A2=E7=A0=81=E5=BC=80=E5=85=B3?= =?UTF-8?q?=20+=20=E6=96=87=E6=A1=A3=E5=88=86=E8=BE=A8=E7=8E=87=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 并发分组(新功能): - 新表 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 --- backend/internal/bootstrap/app.go | 17 ++- backend/internal/bootstrap/seed.go | 1 + backend/internal/http/handler/admin_read.go | 1 + backend/internal/http/handler/app_settings.go | 83 ++++++++++ .../http/handler/concurrency_group.go | 80 ++++++++++ backend/internal/http/handler/site.go | 9 +- .../internal/http/handler/site_settings.go | 5 +- .../internal/http/handler/user_generation.go | 4 +- backend/internal/http/handler/v1.go | 4 +- backend/internal/http/router/router.go | 9 ++ backend/internal/model/models.go | 14 ++ .../internal/repo/concurrency_group_repo.go | 128 ++++++++++++++++ backend/internal/repo/event_repo.go | 9 ++ backend/internal/service/admin_write.go | 52 ++++++- backend/internal/service/app_settings.go | 97 ++++++++++-- backend/internal/service/auth.go | 25 +++ backend/internal/service/cdk.go | 18 ++- backend/internal/service/concurrency.go | 103 +++++++++++++ backend/internal/service/concurrency_group.go | 107 +++++++++++++ backend/internal/service/image_access.go | 5 + backend/internal/service/maintenance.go | 8 + backend/internal/service/site.go | 21 ++- backend/internal/service/user_generation.go | 9 +- backend/internal/service/v1.go | 142 ++++++++++------- frontend/src/components/CustomModelModal.vue | 4 +- frontend/src/layouts/AdminLayout.vue | 1 + frontend/src/layouts/PublicLayout.vue | 13 +- frontend/src/main.js | 2 + frontend/src/site.js | 16 ++ frontend/src/views/AboutView.vue | 2 +- frontend/src/views/ConcurrencyView.vue | 143 ++++++++++++++++++ frontend/src/views/ConfigView.vue | 79 +++++++--- frontend/src/views/DocsView.vue | 67 ++++++-- frontend/src/views/HomeView.vue | 2 +- frontend/src/views/SettingsView.vue | 41 ++++- frontend/src/views/ShowcaseView.vue | 56 ++++--- frontend/src/views/UsersView.vue | 41 ++++- 37 files changed, 1257 insertions(+), 161 deletions(-) create mode 100644 backend/internal/http/handler/concurrency_group.go create mode 100644 backend/internal/repo/concurrency_group_repo.go create mode 100644 backend/internal/service/concurrency.go create mode 100644 backend/internal/service/concurrency_group.go create mode 100644 frontend/src/views/ConcurrencyView.vue diff --git a/backend/internal/bootstrap/app.go b/backend/internal/bootstrap/app.go index b4c191b..942abb2 100644 --- a/backend/internal/bootstrap/app.go +++ b/backend/internal/bootstrap/app.go @@ -3,6 +3,7 @@ package bootstrap import ( "context" "fmt" + "log" "os" "time" @@ -96,12 +97,19 @@ func NewApp(ctx context.Context) (*App, error) { apiKeyRepo := repo.NewAPIKeyRepository(db) tokenRepo := repo.NewTokenRepository(db) refreshRepo := repo.NewRefreshProfileRepository(db) + cgroupRepo := repo.NewConcurrencyGroupRepository(db) + // Seed the "默认并发" group (cap 10) and bind any ungrouped users to it. + if err := cgroupRepo.EnsureDefault(ctx); err != nil { + log.Printf("ensure default concurrency group: %v", err) + } + concSvc := service.NewConcurrencyService(rdb) + cgroupSvc := service.NewConcurrencyGroupService(cgroupRepo, concSvc) sessionSvc := service.NewSessionService(rdb, cfg.SessionTTL, cfg.SessionSlideAfter) emailCodeSvc := service.NewEmailCodeService(rdb) smtpSvc := service.NewSMTPService() rateLimitSvc := service.NewRateLimitService(rdb) rustfsClient := storage.New(cfg.RustFSEndpoint, cfg.RustFSBucket, cfg.RustFSAccessKey, cfg.RustFSSecretKey) - authSvc := service.NewAuthService(userRepo, siteRepo, sessionSvc, emailCodeSvc, smtpSvc) + authSvc := service.NewAuthService(userRepo, siteRepo, sessionSvc, emailCodeSvc, smtpSvc, cgroupRepo) appSettingsSvc := service.NewAppSettingsService(siteRepo, eventRepo, smtpSvc, rustfsClient) imageAccessSvc := service.NewImageAccessService(cfg.GeneratedRoot, showcaseRepo, authSvc) adobeClient := adobe.NewClient("clio-playground-web", "") @@ -112,12 +120,12 @@ func NewApp(ctx context.Context) (*App, error) { imagineClient := imagine.NewClient("") grokClient := grok.NewClient("") customClient := custom.NewClient() - v1Svc := service.NewV1Service(cfg, modelRepo, userRepo, eventRepo, tokenRepo, siteRepo, adobeClient, chatGPTClient, runwayClient, leonardoClient, kreaClient, imagineClient, grokClient, customClient, rustfsClient) + v1Svc := service.NewV1Service(cfg, modelRepo, userRepo, eventRepo, tokenRepo, siteRepo, cgroupRepo, concSvc, adobeClient, chatGPTClient, runwayClient, leonardoClient, kreaClient, imagineClient, grokClient, customClient, rustfsClient) siteSvc := service.NewSiteService(siteRepo, cfg.AppTitle) showcaseSvc := service.NewShowcaseService(showcaseRepo) adminReadSvc := service.NewAdminReadService(cfg, userRepo, modelRepo, eventRepo, siteRepo, tokenRepo, cdkRepo, rustfsClient) - adminWriteSvc := service.NewAdminWriteService(userRepo, showcaseRepo, modelRepo, eventRepo, apiKeyRepo) - cdkSvc := service.NewCDKService(cdkRepo, userRepo) + adminWriteSvc := service.NewAdminWriteService(userRepo, showcaseRepo, modelRepo, eventRepo, apiKeyRepo, tokenRepo) + cdkSvc := service.NewCDKService(cdkRepo, userRepo, siteRepo) apiKeySvc := service.NewAPIKeyService(apiKeyRepo) tokenSvc := service.NewTokenService(tokenRepo, refreshRepo, eventRepo, siteRepo, adobeClient, chatGPTClient, runwayClient, leonardoClient, kreaClient, imagineClient, grokClient) refreshSvc := service.NewRefreshProfileService(refreshRepo, tokenRepo, adobeClient) @@ -141,6 +149,7 @@ func NewApp(ctx context.Context) (*App, error) { UserTools: handler.NewUserToolsHandler(apiKeySvc, cdkSvc), UserGen: handler.NewUserGenerationHandler(userGenSvc, adminReadSvc), ProviderAdmin: handler.NewProviderAdminHandler(tokenSvc, refreshSvc), + ConcGroups: handler.NewConcurrencyGroupHandler(cgroupSvc), }) // Background self-healing sweep (quota recovery, cookie refresh, stale-pending diff --git a/backend/internal/bootstrap/seed.go b/backend/internal/bootstrap/seed.go index 0fc6bfb..839eca3 100644 --- a/backend/internal/bootstrap/seed.go +++ b/backend/internal/bootstrap/seed.go @@ -34,6 +34,7 @@ func seedDefaults(ctx context.Context, db *gorm.DB) error { {Key: "credits.checkin_reward", Value: "3"}, {Key: "credits.invite_enabled", Value: "true"}, {Key: "credits.invite_reward", Value: "3"}, + {Key: "credits.cdk_redeem_enabled", Value: "true"}, {Key: "logs.retention_days", Value: "30"}, {Key: "media.retention_days", Value: "30"}, } diff --git a/backend/internal/http/handler/admin_read.go b/backend/internal/http/handler/admin_read.go index 2d7b3ad..33d6b15 100644 --- a/backend/internal/http/handler/admin_read.go +++ b/backend/internal/http/handler/admin_read.go @@ -182,6 +182,7 @@ func userPublic(user model.User) gin.H { "status": user.Status, "credits": user.Credits, "notes": user.Notes, + "concurrency_group_id": user.ConcurrencyGroupID, "created_at": unixSec(user.CreatedAt), "last_login_at": unixSecPtr(user.LastLoginAt), "last_login_ip": user.LastLoginIP, diff --git a/backend/internal/http/handler/app_settings.go b/backend/internal/http/handler/app_settings.go index 5bcb36e..ab25afb 100644 --- a/backend/internal/http/handler/app_settings.go +++ b/backend/internal/http/handler/app_settings.go @@ -1,7 +1,9 @@ package handler import ( + "encoding/base64" "net/http" + "strings" "backend/internal/service" "github.com/gin-gonic/gin" @@ -15,6 +17,87 @@ func NewAppSettingsHandler(settings *service.AppSettingsService) *AppSettingsHan return &AppSettingsHandler{settings: settings} } +// LogoUpload stores a base64 image as the site logo in RustFS (deleting the old +// one) and persists site.logo. Called on 保存 — not on file pick. +func (h *AppSettingsHandler) LogoUpload(c *gin.Context) { + var body struct { + Data string `json:"data"` // base64, optionally a "data:...;base64," URL + ContentType string `json:"content_type"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "invalid request body"}) + return + } + raw := strings.TrimSpace(body.Data) + if strings.HasPrefix(raw, "data:") { + if i := strings.Index(raw, ","); i >= 0 { + if body.ContentType == "" { + meta := raw[5:i] // e.g. image/png;base64 + if j := strings.Index(meta, ";"); j >= 0 { + body.ContentType = meta[:j] + } + } + raw = raw[i+1:] + } + } + data, err := base64.StdEncoding.DecodeString(raw) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "图片解码失败"}) + return + } + url, err := h.settings.UploadLogo(c.Request.Context(), data, body.ContentType) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "logo": url}) +} + +// AssetUpload stores a public image (homepage 底图 etc.) in RustFS and returns +// its storage path for the caller to save (e.g. as a showcase card's image). +func (h *AppSettingsHandler) AssetUpload(c *gin.Context) { + var body struct { + Data string `json:"data"` + ContentType string `json:"content_type"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "invalid request body"}) + return + } + raw := strings.TrimSpace(body.Data) + if strings.HasPrefix(raw, "data:") { + if i := strings.Index(raw, ","); i >= 0 { + if body.ContentType == "" { + meta := raw[5:i] + if j := strings.Index(meta, ";"); j >= 0 { + body.ContentType = meta[:j] + } + } + raw = raw[i+1:] + } + } + data, err := base64.StdEncoding.DecodeString(raw) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "图片解码失败"}) + return + } + path, err := h.settings.UploadAsset(c.Request.Context(), data, body.ContentType) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "path": path}) +} + +// LogoDelete removes the uploaded logo and falls back to the built-in default. +func (h *AppSettingsHandler) LogoDelete(c *gin.Context) { + if err := h.settings.RemoveLogo(c.Request.Context()); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"detail": "failed to remove logo"}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "logo": ""}) +} + func (h *AppSettingsHandler) RegistrationGet(c *gin.Context) { data, err := h.settings.Registration(c.Request.Context()) if err != nil { diff --git a/backend/internal/http/handler/concurrency_group.go b/backend/internal/http/handler/concurrency_group.go new file mode 100644 index 0000000..17aa6d1 --- /dev/null +++ b/backend/internal/http/handler/concurrency_group.go @@ -0,0 +1,80 @@ +package handler + +import ( + "errors" + "net/http" + + "backend/internal/service" + "github.com/gin-gonic/gin" +) + +type ConcurrencyGroupHandler struct { + svc *service.ConcurrencyGroupService +} + +func NewConcurrencyGroupHandler(svc *service.ConcurrencyGroupService) *ConcurrencyGroupHandler { + return &ConcurrencyGroupHandler{svc: svc} +} + +func (h *ConcurrencyGroupHandler) List(c *gin.Context) { + items, err := h.svc.List(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"detail": "failed to load concurrency groups"}) + return + } + c.JSON(http.StatusOK, gin.H{"data": items}) +} + +func (h *ConcurrencyGroupHandler) Create(c *gin.Context) { + var body struct { + Name string `json:"name"` + MaxConcurrency int `json:"max_concurrency"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "invalid request body"}) + return + } + g, err := h.svc.Create(c.Request.Context(), body.Name, body.MaxConcurrency) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "id": g.ID}) +} + +func (h *ConcurrencyGroupHandler) Update(c *gin.Context) { + var body struct { + Name *string `json:"name"` + MaxConcurrency *int `json:"max_concurrency"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": "invalid request body"}) + return + } + g, err := h.svc.Update(c.Request.Context(), c.Param("id"), body.Name, body.MaxConcurrency) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "id": g.ID}) +} + +func (h *ConcurrencyGroupHandler) SetDefault(c *gin.Context) { + if err := h.svc.SetDefault(c.Request.Context(), c.Param("id")); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} + +func (h *ConcurrencyGroupHandler) Delete(c *gin.Context) { + if err := h.svc.Delete(c.Request.Context(), c.Param("id")); err != nil { + if errors.Is(err, service.ErrNotFound) { + c.JSON(http.StatusNotFound, gin.H{"detail": "分组不存在"}) + return + } + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true}) +} diff --git a/backend/internal/http/handler/site.go b/backend/internal/http/handler/site.go index d9182a1..77ce1a8 100644 --- a/backend/internal/http/handler/site.go +++ b/backend/internal/http/handler/site.go @@ -23,9 +23,10 @@ func (h *SiteHandler) Public(c *gin.Context) { } ctx := c.Request.Context() c.JSON(http.StatusOK, gin.H{ - "title": title, - "logo": h.site.Logo(ctx), - "subtitle": h.site.Subtitle(ctx), - "contact": h.site.Contact(ctx), + "title": title, + "logo": h.site.Logo(ctx), + "subtitle": h.site.Subtitle(ctx), + "cdk_redeem_enabled": h.site.CDKRedeemEnabled(ctx), + "contact": h.site.Contact(ctx), }) } diff --git a/backend/internal/http/handler/site_settings.go b/backend/internal/http/handler/site_settings.go index 0e579ec..cfc0a40 100644 --- a/backend/internal/http/handler/site_settings.go +++ b/backend/internal/http/handler/site_settings.go @@ -35,7 +35,6 @@ func (h *SiteSettingsHandler) Put(c *gin.Context) { ctx := c.Request.Context() var body struct { Title string `json:"title"` - Logo string `json:"logo"` Subtitle string `json:"subtitle"` Contact service.Contact `json:"contact"` } @@ -53,8 +52,8 @@ func (h *SiteSettingsHandler) Put(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"detail": "failed to save site settings"}) return } - if err := h.site.SetBranding(ctx, body.Logo, body.Subtitle); err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"detail": "failed to save branding"}) + if err := h.site.SetSubtitle(ctx, body.Subtitle); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"detail": "failed to save subtitle"}) return } if err := h.site.SetContact(ctx, body.Contact); err != nil { diff --git a/backend/internal/http/handler/user_generation.go b/backend/internal/http/handler/user_generation.go index 390664b..a38af9b 100644 --- a/backend/internal/http/handler/user_generation.go +++ b/backend/internal/http/handler/user_generation.go @@ -80,7 +80,7 @@ func (h *UserGenerationHandler) Generate(c *gin.Context) { c.JSON(http.StatusServiceUnavailable, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrProviderQuota): c.JSON(http.StatusTooManyRequests, gin.H{"detail": err.Error()}) - case errors.Is(err, service.ErrConcurrencyFull): + case errors.Is(err, service.ErrConcurrencyFull), errors.Is(err, service.ErrUserConcurrencyFull): c.JSON(http.StatusTooManyRequests, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrProviderExecution): c.JSON(http.StatusBadGateway, gin.H{"detail": err.Error()}) @@ -136,7 +136,7 @@ func (h *UserGenerationHandler) Test(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrProviderQuota): c.JSON(http.StatusTooManyRequests, gin.H{"detail": err.Error()}) - case errors.Is(err, service.ErrConcurrencyFull): + case errors.Is(err, service.ErrConcurrencyFull), errors.Is(err, service.ErrUserConcurrencyFull): c.JSON(http.StatusTooManyRequests, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrNoProviderAccount): c.JSON(http.StatusServiceUnavailable, gin.H{"detail": err.Error()}) diff --git a/backend/internal/http/handler/v1.go b/backend/internal/http/handler/v1.go index bd7e039..f86480e 100644 --- a/backend/internal/http/handler/v1.go +++ b/backend/internal/http/handler/v1.go @@ -72,7 +72,6 @@ func (h *V1Handler) ImageGenerations(c *gin.Context) { Prompt: body.Prompt, N: body.N, Size: body.Size, - Quality: body.Quality, BaseURL: requestBaseURL(c), }) if err != nil { @@ -110,7 +109,6 @@ func (h *V1Handler) ImageEdits(c *gin.Context) { Prompt: c.PostForm("prompt"), N: n, Size: c.PostForm("size"), - Quality: c.PostForm("quality"), ReferenceImages: refs, BaseURL: requestBaseURL(c), }) @@ -336,7 +334,7 @@ func (h *V1Handler) writeV1Error(c *gin.Context, err error, payload map[string]a c.JSON(http.StatusUnauthorized, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrProviderTemporary): c.JSON(http.StatusServiceUnavailable, gin.H{"detail": err.Error()}) - case errors.Is(err, service.ErrConcurrencyFull): + case errors.Is(err, service.ErrConcurrencyFull), errors.Is(err, service.ErrUserConcurrencyFull): c.JSON(http.StatusTooManyRequests, gin.H{"detail": err.Error()}) case errors.Is(err, service.ErrVideoJobNotFound): c.JSON(http.StatusNotFound, gin.H{"detail": err.Error()}) diff --git a/backend/internal/http/router/router.go b/backend/internal/http/router/router.go index 997cb6f..e05a13e 100644 --- a/backend/internal/http/router/router.go +++ b/backend/internal/http/router/router.go @@ -25,6 +25,7 @@ type Handlers struct { UserTools *handler.UserToolsHandler UserGen *handler.UserGenerationHandler ProviderAdmin *handler.ProviderAdminHandler + ConcGroups *handler.ConcurrencyGroupHandler } func New(cfg *config.Config, auth *service.AuthService, handlers Handlers) *gin.Engine { @@ -96,6 +97,11 @@ func New(cfg *config.Config, auth *service.AuthService, handlers Handlers) *gin. authed.POST("/users/:user_id/credits", handlers.AdminWrite.AdjustUserCredits) authed.POST("/users/:user_id/api-keys", handlers.AdminWrite.CreateUserAPIKey) authed.DELETE("/users/:user_id/api-keys/:key_id", handlers.AdminWrite.DeleteUserAPIKey) + authed.GET("/concurrency-groups", handlers.ConcGroups.List) + authed.POST("/concurrency-groups", handlers.ConcGroups.Create) + authed.PATCH("/concurrency-groups/:id", handlers.ConcGroups.Update) + authed.POST("/concurrency-groups/:id/default", handlers.ConcGroups.SetDefault) + authed.DELETE("/concurrency-groups/:id", handlers.ConcGroups.Delete) authed.GET("/cdks", handlers.CDK.List) authed.POST("/cdks", handlers.CDK.Create) authed.POST("/cdks/delete-bulk", handlers.CDK.DeleteBulk) @@ -135,6 +141,9 @@ func New(cfg *config.Config, auth *service.AuthService, handlers Handlers) *gin. { settings.GET("/site", handlers.SiteSettings.Get) settings.PUT("/site", handlers.SiteSettings.Put) + settings.POST("/logo", handlers.AppSettings.LogoUpload) + settings.DELETE("/logo", handlers.AppSettings.LogoDelete) + settings.POST("/asset", handlers.AppSettings.AssetUpload) settings.GET("/registration", handlers.AppSettings.RegistrationGet) settings.PUT("/registration", handlers.AppSettings.RegistrationPut) settings.GET("/smtp", handlers.AppSettings.SMTPGet) diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index c15fc6e..5bc37e1 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -15,6 +15,7 @@ type User struct { Status string `gorm:"size:32;index;not null"` Credits float64 `gorm:"not null;default:0"` Notes string `gorm:"type:text"` + ConcurrencyGroupID string `gorm:"size:32;index"` InviteCode string `gorm:"size:32;uniqueIndex"` InvitedBy *string `gorm:"size:32;index"` InviteRewardDone bool `gorm:"not null;default:false"` @@ -203,9 +204,22 @@ func AutoMigrateModels() []any { &RefreshProfile{}, &SiteSetting{}, &StatCounter{}, + &ConcurrencyGroup{}, } } +// ConcurrencyGroup caps how many generations a member user may run AT ONCE +// (across their API key + 画图台). MaxConcurrency 0 = unlimited. Exactly one +// group is IsDefault — new users are bound to it and it can't be deleted. +type ConcurrencyGroup struct { + ID string `gorm:"primaryKey;size:32"` + Name string `gorm:"size:100;not null"` + MaxConcurrency int `gorm:"not null;default:10"` // 0 = 不限制 + IsDefault bool `gorm:"not null;default:false;index"` + CreatedAt time.Time + UpdatedAt time.Time +} + // StatCounter is a persistent monotonic counter (key → value), independent of the // event_log (which is retention-pruned / clearable). Used for the dashboard // cumulative cards (total/success/failed/image/video/api) so they never reset. diff --git a/backend/internal/repo/concurrency_group_repo.go b/backend/internal/repo/concurrency_group_repo.go new file mode 100644 index 0000000..7196746 --- /dev/null +++ b/backend/internal/repo/concurrency_group_repo.go @@ -0,0 +1,128 @@ +package repo + +import ( + "context" + "time" + + "backend/internal/model" + "gorm.io/gorm" +) + +type ConcurrencyGroupRepository struct { + db *gorm.DB +} + +func NewConcurrencyGroupRepository(db *gorm.DB) *ConcurrencyGroupRepository { + return &ConcurrencyGroupRepository{db: db} +} + +func (r *ConcurrencyGroupRepository) List(ctx context.Context) ([]model.ConcurrencyGroup, error) { + var items []model.ConcurrencyGroup + // Default first, then by name — stable ordering for the admin table. + err := r.db.WithContext(ctx).Order("is_default desc, created_at asc").Find(&items).Error + return items, err +} + +func (r *ConcurrencyGroupRepository) Get(ctx context.Context, id string) (*model.ConcurrencyGroup, error) { + var g model.ConcurrencyGroup + if err := r.db.WithContext(ctx).First(&g, "id = ?", id).Error; err != nil { + return nil, err + } + return &g, nil +} + +func (r *ConcurrencyGroupRepository) GetDefault(ctx context.Context) (*model.ConcurrencyGroup, error) { + var g model.ConcurrencyGroup + if err := r.db.WithContext(ctx).First(&g, "is_default = ?", true).Error; err != nil { + return nil, err + } + return &g, nil +} + +func (r *ConcurrencyGroupRepository) Create(ctx context.Context, g *model.ConcurrencyGroup) error { + now := time.Now() + g.CreatedAt = now + g.UpdatedAt = now + return r.db.WithContext(ctx).Create(g).Error +} + +func (r *ConcurrencyGroupRepository) Update(ctx context.Context, id string, patch map[string]any) (*model.ConcurrencyGroup, error) { + patch["updated_at"] = time.Now() + if err := r.db.WithContext(ctx).Model(&model.ConcurrencyGroup{}).Where("id = ?", id).Updates(patch).Error; err != nil { + return nil, err + } + return r.Get(ctx, id) +} + +// SetDefault makes id the sole default (used for new registrations). +func (r *ConcurrencyGroupRepository) SetDefault(ctx context.Context, id string) error { + return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Model(&model.ConcurrencyGroup{}).Where("is_default = ?", true).Update("is_default", false).Error; err != nil { + return err + } + return tx.Model(&model.ConcurrencyGroup{}).Where("id = ?", id).Update("is_default", true).Error + }) +} + +// Delete removes a group and reassigns its members to the default group, so no +// user is left without a concurrency limit. The default group itself is never +// deletable (guarded in the service). +func (r *ConcurrencyGroupRepository) Delete(ctx context.Context, id, defaultID string) (int64, error) { + var rows int64 + err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Model(&model.User{}).Where("concurrency_group_id = ?", id). + Update("concurrency_group_id", defaultID).Error; err != nil { + return err + } + res := tx.Delete(&model.ConcurrencyGroup{}, "id = ?", id) + rows = res.RowsAffected + return res.Error + }) + return rows, err +} + +// UserCounts returns group_id → number of bound users. +func (r *ConcurrencyGroupRepository) UserCounts(ctx context.Context) (map[string]int64, error) { + type row struct { + GroupID string + N int64 + } + var rows []row + if err := r.db.WithContext(ctx).Model(&model.User{}). + Select("concurrency_group_id as group_id, count(*) as n"). + Where("concurrency_group_id <> ''"). + Group("concurrency_group_id").Scan(&rows).Error; err != nil { + return nil, err + } + out := make(map[string]int64, len(rows)) + for _, x := range rows { + out[x.GroupID] = x.N + } + return out, nil +} + +// EnsureDefault creates the seed "默认并发" group (MaxConcurrency 10) when none +// exists, and binds any ungrouped users to the default. Idempotent — safe at boot. +func (r *ConcurrencyGroupRepository) EnsureDefault(ctx context.Context) error { + var count int64 + if err := r.db.WithContext(ctx).Model(&model.ConcurrencyGroup{}).Count(&count).Error; err != nil { + return err + } + if count == 0 { + now := time.Now() + if err := r.db.WithContext(ctx).Create(&model.ConcurrencyGroup{ + ID: "cg-default", Name: "默认并发", MaxConcurrency: 10, IsDefault: true, + CreatedAt: now, UpdatedAt: now, + }).Error; err != nil { + return err + } + } + def, err := r.GetDefault(ctx) + if err != nil { + return err + } + // Bind ungrouped users to the default group. + return r.db.WithContext(ctx).Model(&model.User{}). + Where("concurrency_group_id = '' OR concurrency_group_id IS NULL"). + Update("concurrency_group_id", def.ID).Error +} diff --git a/backend/internal/repo/event_repo.go b/backend/internal/repo/event_repo.go index 036460c..2c01f89 100644 --- a/backend/internal/repo/event_repo.go +++ b/backend/internal/repo/event_repo.go @@ -238,6 +238,15 @@ func (r *EventRepository) CountBetween(ctx context.Context, start, end time.Time return n, err } +// CountPendingByUser returns how many generations the user currently has +// in-flight (status=pending) — used to enforce the per-user concurrency limit. +func (r *EventRepository) CountPendingByUser(ctx context.Context, userID string) (int64, error) { + var n int64 + err := r.db.WithContext(ctx).Model(&model.EventLog{}). + Where("user_id = ? AND status = ?", userID, "pending").Count(&n).Error + return n, err +} + // DistinctUsersSince counts distinct (non-empty) user_ids active since `since`. func (r *EventRepository) DistinctUsersSince(ctx context.Context, since time.Time) (int64, error) { var n int64 diff --git a/backend/internal/service/admin_write.go b/backend/internal/service/admin_write.go index c38b8ca..b81f9c5 100644 --- a/backend/internal/service/admin_write.go +++ b/backend/internal/service/admin_write.go @@ -27,15 +27,17 @@ type AdminWriteService struct { models *repo.ModelRepository events *repo.EventRepository apiKeys *repo.APIKeyRepository + tokens *repo.TokenRepository } -func NewAdminWriteService(users *repo.UserRepository, showcase *repo.ShowcaseRepository, models *repo.ModelRepository, events *repo.EventRepository, apiKeys *repo.APIKeyRepository) *AdminWriteService { +func NewAdminWriteService(users *repo.UserRepository, showcase *repo.ShowcaseRepository, models *repo.ModelRepository, events *repo.EventRepository, apiKeys *repo.APIKeyRepository, tokens *repo.TokenRepository) *AdminWriteService { return &AdminWriteService{ users: users, showcase: showcase, models: models, events: events, apiKeys: apiKeys, + tokens: tokens, } } @@ -60,6 +62,7 @@ func (s *AdminWriteService) CreateUser(ctx context.Context, body map[string]any) status := normalizedStatus(stringValue(body["status"])) credits := maxFloat(0, floatValue(body["credits"])) notes := strings.TrimSpace(stringValue(body["notes"])) + cgroupID := strings.TrimSpace(stringValue(body["concurrency_group_id"])) exists, err := s.users.ExistsEmail(ctx, email, "") if err != nil { @@ -99,6 +102,7 @@ func (s *AdminWriteService) CreateUser(ctx context.Context, body map[string]any) Status: status, Credits: credits, Notes: notes, + ConcurrencyGroupID: cgroupID, InviteCode: randomInviteCode(), CreatedAt: time.Now(), UpdatedAt: time.Now(), @@ -161,6 +165,9 @@ func (s *AdminWriteService) UpdateUser(ctx context.Context, userID string, body if _, ok := body["notes"]; ok { patch["notes"] = strings.TrimSpace(stringValue(body["notes"])) } + if _, ok := body["concurrency_group_id"]; ok { + patch["concurrency_group_id"] = strings.TrimSpace(stringValue(body["concurrency_group_id"])) + } if _, ok := body["password"]; ok && strings.TrimSpace(stringValue(body["password"])) != "" { if err := ValidatePassword(stringValue(body["password"])); err != nil { return nil, err @@ -458,9 +465,52 @@ func (s *AdminWriteService) DeleteModel(ctx context.Context, modelID string) err if rows == 0 { return ErrNotFound } + // Clean up: strip this model id from every custom upstream account's + // supported-models list so no orphan reference is left behind. + s.removeModelFromUpstreams(ctx, modelID) return nil } +// removeModelFromUpstreams drops modelID from the CSV in each custom account's +// Meta["models"]. Best-effort — a failure here doesn't undo the model delete. +func (s *AdminWriteService) removeModelFromUpstreams(ctx context.Context, modelID string) { + if s.tokens == nil { + return + } + items, err := s.tokens.ListByPool(ctx, "custom") + if err != nil { + return + } + for _, it := range items { + raw, _ := it.Meta["models"].(string) + if strings.TrimSpace(raw) == "" { + continue + } + kept := make([]string, 0, len(strings.Split(raw, ","))) + changed := false + for _, p := range strings.Split(raw, ",") { + p = strings.TrimSpace(p) + if p == "" { + continue + } + if p == modelID { + changed = true + continue + } + kept = append(kept, p) + } + if !changed { + continue + } + meta := datatypes.JSONMap{} + for k, v := range it.Meta { + meta[k] = v + } + meta["models"] = strings.Join(kept, ",") + _, _ = s.tokens.Update(ctx, "custom", it.ID, map[string]any{"meta": meta}) + } +} + func (s *AdminWriteService) ClearLogs(ctx context.Context) (int64, error) { return s.events.DeleteAll(ctx) } diff --git a/backend/internal/service/app_settings.go b/backend/internal/service/app_settings.go index 2d5a999..eeb3d15 100644 --- a/backend/internal/service/app_settings.go +++ b/backend/internal/service/app_settings.go @@ -41,10 +41,11 @@ type SMTPSettings struct { } type CreditSettings struct { - CheckinEnabled bool `json:"checkin_enabled"` - CheckinReward int `json:"checkin_reward"` - InviteEnabled bool `json:"invite_enabled"` - InviteReward int `json:"invite_reward"` + CheckinEnabled bool `json:"checkin_enabled"` + CheckinReward int `json:"checkin_reward"` + InviteEnabled bool `json:"invite_enabled"` + InviteReward int `json:"invite_reward"` + CDKRedeemEnabled bool `json:"cdk_redeem_enabled"` } type ProxySettings struct { @@ -70,6 +71,75 @@ func NewAppSettingsService(settings *repo.SiteSettingRepository, events *repo.Ev } } +// UploadLogo stores a new site logo in object storage under branding/, deletes +// the previously-uploaded one (if any), persists site.logo, and returns its URL. +func (s *AppSettingsService) UploadLogo(ctx context.Context, data []byte, contentType string) (string, error) { + if s.store == nil || !s.store.Configured() { + return "", errors.New("对象存储未配置") + } + if len(data) == 0 { + return "", errors.New("空文件") + } + if len(data) > 4*1024*1024 { + return "", errors.New("logo 不能超过 4MB") + } + key := "branding/logo-" + randomUpper(10) + "." + logoExt(contentType) + if err := s.store.Put(ctx, key, data, contentType); err != nil { + return "", err + } + url := "/images/" + key + // Delete the previous uploaded logo (best-effort), then point site.logo at the new one. + if old, _ := s.settings.GetValue(ctx, "site.logo"); strings.HasPrefix(old, "/images/branding/") { + _ = s.store.Delete(ctx, strings.TrimPrefix(old, "/images/")) + } + if err := s.settings.UpsertValue(ctx, "site.logo", url); err != nil { + return "", err + } + return url, nil +} + +// UploadAsset stores a public image (e.g. a 首页内容 底图) under branding/ and +// returns its storage path (for form.image). Does NOT touch site settings. +func (s *AppSettingsService) UploadAsset(ctx context.Context, data []byte, contentType string) (string, error) { + if s.store == nil || !s.store.Configured() { + return "", errors.New("对象存储未配置") + } + if len(data) == 0 { + return "", errors.New("空文件") + } + if len(data) > 8*1024*1024 { + return "", errors.New("图片不能超过 8MB") + } + key := "branding/sc-" + randomUpper(10) + "." + logoExt(contentType) + if err := s.store.Put(ctx, key, data, contentType); err != nil { + return "", err + } + return key, nil +} + +// RemoveLogo deletes the uploaded logo and resets site.logo to the built-in default (empty). +func (s *AppSettingsService) RemoveLogo(ctx context.Context) error { + if old, _ := s.settings.GetValue(ctx, "site.logo"); strings.HasPrefix(old, "/images/branding/") && s.store != nil { + _ = s.store.Delete(ctx, strings.TrimPrefix(old, "/images/")) + } + return s.settings.UpsertValue(ctx, "site.logo", "") +} + +func logoExt(contentType string) string { + switch strings.ToLower(strings.TrimSpace(contentType)) { + case "image/jpeg", "image/jpg": + return "jpg" + case "image/webp": + return "webp" + case "image/svg+xml": + return "svg" + case "image/gif": + return "gif" + default: + return "png" + } +} + func (s *AppSettingsService) Registration(ctx context.Context) (*RegistrationSettings, error) { openRaw, err := s.settings.GetValue(ctx, "auth.open") if err != nil { @@ -281,11 +351,13 @@ func (s *AppSettingsService) Credits(ctx context.Context) (*CreditSettings, erro if err != nil { return nil, err } + cdkRaw, _ := s.settings.GetValue(ctx, "credits.cdk_redeem_enabled") return &CreditSettings{ - CheckinEnabled: parseBoolSetting(checkinEnabledRaw, true), - CheckinReward: parseIntSetting(checkinRewardRaw, 3), - InviteEnabled: parseBoolSetting(inviteEnabledRaw, true), - InviteReward: parseIntSetting(inviteRewardRaw, 3), + CheckinEnabled: parseBoolSetting(checkinEnabledRaw, true), + CheckinReward: parseIntSetting(checkinRewardRaw, 3), + InviteEnabled: parseBoolSetting(inviteEnabledRaw, true), + InviteReward: parseIntSetting(inviteRewardRaw, 3), + CDKRedeemEnabled: parseBoolSetting(cdkRaw, true), }, nil } @@ -297,10 +369,11 @@ func (s *AppSettingsService) SaveCredits(ctx context.Context, in CreditSettings) in.InviteReward = 0 } if err := s.settings.UpsertValues(ctx, map[string]string{ - "credits.checkin_enabled": strconv.FormatBool(in.CheckinEnabled), - "credits.checkin_reward": strconv.Itoa(in.CheckinReward), - "credits.invite_enabled": strconv.FormatBool(in.InviteEnabled), - "credits.invite_reward": strconv.Itoa(in.InviteReward), + "credits.checkin_enabled": strconv.FormatBool(in.CheckinEnabled), + "credits.checkin_reward": strconv.Itoa(in.CheckinReward), + "credits.invite_enabled": strconv.FormatBool(in.InviteEnabled), + "credits.invite_reward": strconv.Itoa(in.InviteReward), + "credits.cdk_redeem_enabled": strconv.FormatBool(in.CDKRedeemEnabled), }); err != nil { return nil, err } diff --git a/backend/internal/service/auth.go b/backend/internal/service/auth.go index bb2aaf3..4d92f8b 100644 --- a/backend/internal/service/auth.go +++ b/backend/internal/service/auth.go @@ -24,6 +24,7 @@ type AuthService struct { codes *EmailCodeService smtp *SMTPService loginGuard *LoginGuard + cgroups *repo.ConcurrencyGroupRepository } type AuthSettings struct { @@ -39,6 +40,7 @@ func NewAuthService( sessions *SessionService, codes *EmailCodeService, smtp *SMTPService, + cgroups *repo.ConcurrencyGroupRepository, ) *AuthService { return &AuthService{ users: users, @@ -47,6 +49,7 @@ func NewAuthService( codes: codes, smtp: smtp, loginGuard: NewLoginGuard(codes.Redis()), + cgroups: cgroups, } } @@ -300,6 +303,12 @@ func (s *AuthService) Register(ctx context.Context, email, username, password, i CreatedAt: now, UpdatedAt: now, } + // Bind new users to the default concurrency group. + if s.cgroups != nil { + if def, derr := s.cgroups.GetDefault(ctx); derr == nil && def != nil { + user.ConcurrencyGroupID = def.ID + } + } if err := s.users.Create(ctx, user); err != nil { return nil, "", nil, err } @@ -426,6 +435,20 @@ func (s *AuthService) PublicUser(ctx context.Context, user *model.User) (map[str if err != nil { return nil, err } + // Concurrency group + its cap (0 = unlimited) for the profile page. + concName, concMax := "", 0 + if s.cgroups != nil { + 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 { + concName, concMax = g.Name, g.MaxConcurrency + } + } return map[string]any{ "id": user.ID, "email": user.Email, @@ -433,6 +456,8 @@ func (s *AuthService) PublicUser(ctx context.Context, user *model.User) (map[str "role": user.Role, "status": user.Status, "credits": user.Credits, + "concurrency_group": concName, + "concurrency_limit": concMax, "checkin_last": user.CheckinLast, "checkin_streak": user.CheckinStreak, "checkin_today": user.CheckinLast == time.Now().Format("2006-01-02"), diff --git a/backend/internal/service/cdk.go b/backend/internal/service/cdk.go index adb2531..ba7d18c 100644 --- a/backend/internal/service/cdk.go +++ b/backend/internal/service/cdk.go @@ -12,14 +12,16 @@ import ( ) type CDKService struct { - cdks *repo.CDKRepository - users *repo.UserRepository + cdks *repo.CDKRepository + users *repo.UserRepository + settings *repo.SiteSettingRepository } -func NewCDKService(cdks *repo.CDKRepository, users *repo.UserRepository) *CDKService { +func NewCDKService(cdks *repo.CDKRepository, users *repo.UserRepository, settings *repo.SiteSettingRepository) *CDKService { return &CDKService{ - cdks: cdks, - users: users, + cdks: cdks, + users: users, + settings: settings, } } @@ -127,6 +129,12 @@ func (s *CDKService) DeleteBulk(ctx context.Context, codes []string) (int, error } func (s *CDKService) Redeem(ctx context.Context, userID, code string) (map[string]any, error) { + // Honor the admin "兑换码" switch — when off, no code can be redeemed. + if s.settings != nil { + if v, _ := s.settings.GetValue(ctx, "credits.cdk_redeem_enabled"); v == "false" { + return nil, errors.New("兑换功能已关闭") + } + } code = strings.TrimSpace(strings.ToUpper(code)) if code == "" { return nil, errors.New("请输入兑换码") diff --git a/backend/internal/service/concurrency.go b/backend/internal/service/concurrency.go new file mode 100644 index 0000000..6a9958d --- /dev/null +++ b/backend/internal/service/concurrency.go @@ -0,0 +1,103 @@ +package service + +import ( + "context" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// ConcurrencyService is a Redis-backed, self-healing concurrency limiter shared +// by the per-user gate (画图台 + API key) and the per-account upstream gate. +// +// Each slot is a member of a sorted set keyed by the subject (user/account), +// scored with its expiry time. Acquire prunes expired members first, so a slot +// whose Release was lost (crash / missed defer) auto-frees after the TTL — the +// count can never leak forever. It's intentionally lossy-tolerant: if Redis is +// unavailable it FAILS OPEN (allows the work) rather than blocking generation. +type ConcurrencyService struct { + redis *redis.Client + // ttl is the max lifetime of a slot — the longest a generation can run + // (video ~3min) plus head-room, after which a stuck slot self-heals. + ttl int +} + +func NewConcurrencyService(rdb *redis.Client) *ConcurrencyService { + return &ConcurrencyService{redis: rdb, ttl: 900} // 15 min +} + +// acquireScript: KEYS[1]=set, ARGV[1]=max (0=unlimited), ARGV[2]=ttl secs, +// ARGV[3]=token. Prunes expired members, then admits the token iff under max. +// Returns 1 on success, 0 when full. +var acquireScript = redis.NewScript(` +local t = redis.call('TIME') +local now = tonumber(t[1]) +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now) +local n = redis.call('ZCARD', KEYS[1]) +local max = tonumber(ARGV[1]) +if max > 0 and n >= max then return 0 end +redis.call('ZADD', KEYS[1], now + tonumber(ARGV[2]), ARGV[3]) +redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2])) +return 1 +`) + +// Acquire takes one slot under `key` (capped at max; 0 = unlimited), tagged with +// `token`. Returns true if admitted. Fail-open when Redis is down/unset. +func (c *ConcurrencyService) Acquire(ctx context.Context, key string, max int, token string) bool { + if c == nil || c.redis == nil { + return true + } + res, err := acquireScript.Run(ctx, c.redis, []string{key}, max, c.ttl, token).Int() + if err != nil { + return true // fail open — never block a generation on Redis trouble + } + return res == 1 +} + +// Release frees the slot held by `token` under `key`. Safe to call even if the +// slot already expired. +func (c *ConcurrencyService) Release(ctx context.Context, key, token string) { + if c == nil || c.redis == nil { + return + } + _ = c.redis.ZRem(ctx, key, token).Err() +} + +// Count returns the live (non-expired) slot count under `key` — for display. +func (c *ConcurrencyService) Count(ctx context.Context, key string) int { + if c == nil || c.redis == nil { + return 0 + } + now := time.Now().Unix() + _ = c.redis.ZRemRangeByScore(ctx, key, "-inf", strconv.FormatInt(now, 10)).Err() + n, err := c.redis.ZCard(ctx, key).Result() + if err != nil { + return 0 + } + return int(n) +} + +// CountUsers returns live concurrency for many users in one round-trip +// (group_id display etc. don't need this, but the user list does). Keyed by the +// raw subject id passed in. +func (c *ConcurrencyService) CountMany(ctx context.Context, prefix string, ids []string) map[string]int { + out := make(map[string]int, len(ids)) + if c == nil || c.redis == nil || len(ids) == 0 { + return out + } + pipe := c.redis.Pipeline() + cmds := make(map[string]*redis.IntCmd, len(ids)) + for _, id := range ids { + cmds[id] = pipe.ZCard(ctx, prefix+id) + } + if _, err := pipe.Exec(ctx); err != nil { + return out + } + for id, cmd := range cmds { + if n, err := cmd.Result(); err == nil && n > 0 { + out[id] = int(n) + } + } + return out +} diff --git a/backend/internal/service/concurrency_group.go b/backend/internal/service/concurrency_group.go new file mode 100644 index 0000000..fd3d65b --- /dev/null +++ b/backend/internal/service/concurrency_group.go @@ -0,0 +1,107 @@ +package service + +import ( + "context" + "errors" + "strings" + + "backend/internal/model" + "backend/internal/repo" +) + +// ConcurrencyGroupService manages the admin-facing concurrency groups: the +// definitions live in the DB (repo), the live in-flight counts come from Redis. +type ConcurrencyGroupService struct { + repo *repo.ConcurrencyGroupRepository + conc *ConcurrencyService +} + +func NewConcurrencyGroupService(r *repo.ConcurrencyGroupRepository, conc *ConcurrencyService) *ConcurrencyGroupService { + return &ConcurrencyGroupService{repo: r, conc: conc} +} + +// List returns every group with its bound-user count. +func (s *ConcurrencyGroupService) List(ctx context.Context) ([]map[string]any, error) { + groups, err := s.repo.List(ctx) + if err != nil { + return nil, err + } + counts, _ := s.repo.UserCounts(ctx) + out := make([]map[string]any, 0, len(groups)) + for _, g := range groups { + out = append(out, map[string]any{ + "id": g.ID, + "name": g.Name, + "max_concurrency": g.MaxConcurrency, + "is_default": g.IsDefault, + "user_count": counts[g.ID], + }) + } + return out, nil +} + +func (s *ConcurrencyGroupService) Create(ctx context.Context, name string, max int) (*model.ConcurrencyGroup, error) { + name = strings.TrimSpace(name) + if name == "" { + return nil, errors.New("名称不能为空") + } + if max < 0 { + max = 0 + } + g := &model.ConcurrencyGroup{ + ID: "cg-" + randomUpper(10), Name: name, MaxConcurrency: max, IsDefault: false, + } + if err := s.repo.Create(ctx, g); err != nil { + return nil, err + } + return g, nil +} + +func (s *ConcurrencyGroupService) Update(ctx context.Context, id string, name *string, max *int) (*model.ConcurrencyGroup, error) { + patch := map[string]any{} + if name != nil { + n := strings.TrimSpace(*name) + if n == "" { + return nil, errors.New("名称不能为空") + } + patch["name"] = n + } + if max != nil { + m := *max + if m < 0 { + m = 0 + } + patch["max_concurrency"] = m + } + if len(patch) == 0 { + return s.repo.Get(ctx, id) + } + return s.repo.Update(ctx, id, patch) +} + +func (s *ConcurrencyGroupService) SetDefault(ctx context.Context, id string) error { + if _, err := s.repo.Get(ctx, id); err != nil { + return ErrNotFound + } + return s.repo.SetDefault(ctx, id) +} + +// Delete removes a group (members fall back to the default group). The default +// group itself can never be deleted. +func (s *ConcurrencyGroupService) Delete(ctx context.Context, id string) error { + def, err := s.repo.GetDefault(ctx) + if err != nil || def == nil { + return errors.New("缺少默认并发分组") + } + if id == def.ID { + return errors.New("默认并发分组不可删除") + } + rows, err := s.repo.Delete(ctx, id, def.ID) + if err != nil { + return err + } + if rows == 0 { + return ErrNotFound + } + return nil +} diff --git a/backend/internal/service/image_access.go b/backend/internal/service/image_access.go index 6b9976b..99a9745 100644 --- a/backend/internal/service/image_access.go +++ b/backend/internal/service/image_access.go @@ -40,6 +40,11 @@ func (s *ImageAccessService) Resolve(user, name string) (string, error) { } func (s *ImageAccessService) IsPublic(ctx context.Context, rel string) (bool, error) { + // Branding assets (the site logo) are public — they render on the homepage / + // header for logged-out visitors. + if strings.HasPrefix(rel, "branding/") { + return true, nil + } return s.showcase.IsPublicFile(ctx, rel) } diff --git a/backend/internal/service/maintenance.go b/backend/internal/service/maintenance.go index 0a4d736..64dca3d 100644 --- a/backend/internal/service/maintenance.go +++ b/backend/internal/service/maintenance.go @@ -238,6 +238,14 @@ func (m *MaintenanceService) pruneMedia(ctx context.Context) { pinned = nil } } + // The site logo is permanent too — pin it like a showcase image so the + // retention sweep never deletes it. site.logo is "/images/". + if logo, _ := m.settings.GetValue(ctx, "site.logo"); strings.TrimSpace(logo) != "" { + if pinned == nil { + pinned = map[string]struct{}{} + } + pinned[strings.TrimPrefix(strings.TrimLeft(logo, "/"), "images/")] = struct{}{} + } removed, skipped := 0, 0 var clearedKeys []string for _, o := range objs { diff --git a/backend/internal/service/site.go b/backend/internal/service/site.go index f854065..26eb56c 100644 --- a/backend/internal/service/site.go +++ b/backend/internal/service/site.go @@ -51,17 +51,16 @@ func (s *SiteService) get(ctx context.Context, key string) string { func (s *SiteService) Logo(ctx context.Context) string { return s.get(ctx, "site.logo") } func (s *SiteService) Subtitle(ctx context.Context) string { return s.get(ctx, "site.subtitle") } -// SetBranding persists logo / subtitle (either may be empty). -func (s *SiteService) SetBranding(ctx context.Context, logo, subtitle string) error { - for k, v := range map[string]string{ - "site.logo": strings.TrimSpace(logo), - "site.subtitle": strings.TrimSpace(subtitle), - } { - if err := s.settings.UpsertValue(ctx, k, v); err != nil { - return err - } - } - return nil +// CDKRedeemEnabled reflects the admin "兑换码" switch (default on). The front-end +// hides the redeem UI when off. +func (s *SiteService) CDKRedeemEnabled(ctx context.Context) bool { + return s.get(ctx, "credits.cdk_redeem_enabled") != "false" +} + +// SetSubtitle persists the homepage 子标题. (The logo is managed separately via +// the upload/delete endpoints so a site-form save never clobbers it.) +func (s *SiteService) SetSubtitle(ctx context.Context, subtitle string) error { + return s.settings.UpsertValue(ctx, "site.subtitle", strings.TrimSpace(subtitle)) } // Contact is the admin-editable "联系我们" info shown in the public 关于 section. diff --git a/backend/internal/service/user_generation.go b/backend/internal/service/user_generation.go index 945aad8..0def13e 100644 --- a/backend/internal/service/user_generation.go +++ b/backend/internal/service/user_generation.go @@ -39,13 +39,8 @@ func (s *UserGenerationService) Generate(ctx context.Context, user *model.User, if user == nil || strings.TrimSpace(user.ID) == "" { return nil, errors.New("未登录或会话已过期") } - pending, err := s.events.PendingByUser(ctx, user.ID, "user") - if err != nil { - return nil, err - } - if pending != nil { - return nil, errors.New("已有正在生成的任务,请稍候") - } + // No single-job lock anymore — concurrent generations are allowed, capped by + // the user's concurrency group (enforced in prepareImageExecution/Video). modelItem, err := s.models.Get(ctx, strings.TrimSpace(in.Model)) if err != nil { diff --git a/backend/internal/service/v1.go b/backend/internal/service/v1.go index 40cfb22..319278a 100644 --- a/backend/internal/service/v1.go +++ b/backend/internal/service/v1.go @@ -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:] { diff --git a/frontend/src/components/CustomModelModal.vue b/frontend/src/components/CustomModelModal.vue index e1e2250..ef53498 100644 --- a/frontend/src/components/CustomModelModal.vue +++ b/frontend/src/components/CustomModelModal.vue @@ -6,7 +6,9 @@ import SelectMenu from './SelectMenu.vue' const emit = defineEmits(['close', 'saved']) -const RATIO_OPTS = ['1:1', '16:9', '9:16', '4:3', '3:4', '21:9', '3:2', '5:4', '4:5', '2:3', '2:1'] +// 14 ratios — the union of what our models actually support; kept in sync with +// the backend guessRatio() and the docs 对照表. +const RATIO_OPTS = ['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'] const IMG_RES = ['1K', '2K', '4K'] const VID_RES = ['720p', '1080p', '2K', '4K'] const ALL_RES = ['1K', '2K', '4K', '720p', '1080p'] diff --git a/frontend/src/layouts/AdminLayout.vue b/frontend/src/layouts/AdminLayout.vue index 1000b19..7519eee 100644 --- a/frontend/src/layouts/AdminLayout.vue +++ b/frontend/src/layouts/AdminLayout.vue @@ -13,6 +13,7 @@ const tabs = [ { label: '模型管理', to: '/admin/models', icon: 'models' }, { label: '账号管理', to: '/admin/accounts', icon: 'plug' }, { label: '用户管理', to: '/admin/users', icon: 'accounts' }, + { label: '并发分组', to: '/admin/concurrency', icon: 'shield' }, { label: '兑换码', to: '/admin/cdks', icon: 'spark' }, { label: '邀请日志', to: '/admin/invites', icon: 'accounts' }, { label: '图片管理', to: '/admin/images', icon: 'files' }, diff --git a/frontend/src/layouts/PublicLayout.vue b/frontend/src/layouts/PublicLayout.vue index e3f21ac..d5b345a 100644 --- a/frontend/src/layouts/PublicLayout.vue +++ b/frontend/src/layouts/PublicLayout.vue @@ -16,8 +16,8 @@ const nav = computed(() => { const items = [{ to: '/', label: '首页', icon: 'overview' }] if (isAuthed()) { items.push({ to: '/user', label: '画图', icon: 'spark' }) - items.push({ to: '/logs', label: '记录', icon: 'log' }) - items.push({ to: '/mylogs', label: '日志', icon: 'files' }) + items.push({ to: '/logs', label: '图片', icon: 'files' }) + items.push({ to: '/mylogs', label: '日志', icon: 'log' }) items.push({ to: '/invite', label: '邀请', icon: 'accounts' }) } // 文档 + 关于 are public — visible to guests too. @@ -53,7 +53,9 @@ const currentLabel = computed(() => {