package service import ( "context" "encoding/base64" "encoding/json" "errors" "fmt" "io" "log" "net/http" "path/filepath" "sort" "strings" "sync" "sync/atomic" "time" "strconv" "backend/internal/config" "backend/internal/model" "backend/internal/provider/adobe" "backend/internal/provider/chatgpt" "backend/internal/provider/custom" "backend/internal/provider/grok" "backend/internal/provider/imagine" "backend/internal/provider/krea" "backend/internal/provider/leonardo" "backend/internal/provider/runway" "backend/internal/repo" "backend/internal/storage" "gorm.io/gorm" ) var ( ErrMissingAPIKey = errors.New("missing api key") ErrInvalidAPIKey = errors.New("invalid api key") ErrUnknownModel = errors.New("unknown model") ErrUnsupportedParams = errors.New("unsupported or unpriced parameters for this model") ErrBannedPrompt = errors.New("prompt contains banned content") ErrInsufficientFunds = errors.New("insufficient credits") ErrGenerationPending = errors.New("generation executor not implemented yet") ErrProviderAuth = errors.New("provider token invalid or expired") ErrNoProviderAccount = errors.New("no provider account available, please ask an admin to configure one") ErrProviderQuota = errors.New("provider quota exhausted") ErrProviderTemporary = errors.New("provider temporary unavailable") ErrProviderExecution = errors.New("provider request failed") ErrProviderUnsupported = errors.New("provider not implemented") ErrReferenceTooLarge = errors.New("reference image too large") // 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") ) // maxReferenceImageBytes bounds a single decoded reference image. 20 MB // comfortably covers real photos/screenshots; anything larger is almost // certainly abuse or a mistake. Mirrors Python core/refs.py. const maxReferenceImageBytes = 20 * 1024 * 1024 type V1Service struct { cfg *config.Config models *repo.ModelRepository users *repo.UserRepository events *repo.EventRepository tokens *repo.TokenRepository settings *repo.SiteSettingRepository cgroups *repo.ConcurrencyGroupRepository adobe *adobe.Client chatgpt *chatgpt.Client runway *runway.Client leonardo *leonardo.Client krea *krea.Client imagine *imagine.Client grok *grok.Client custom *custom.Client store *storage.Client // refresh re-mints an Adobe access token from its cookie when a request hits a // 401 mid-flight (set via SetRefresh — wired after construction to avoid an // init cycle). nil for deployments without cookie refresh. refresh *RefreshProfileService // banned is the admin-managed prompt blocklist (set via SetBannedWords). // nil disables the check. banned *repo.BannedWordRepository // tokenCursors holds one strict round-robin cursor per pool (key: pool name, // value: *uint64). Each pick advances the pool's cursor by one so accounts // are used in a fixed, even rotation (acct1→acct2→acct3→acct1…) independent // of fails/last_used. The atomic counter also serializes concurrent picks so // two simultaneous requests never start on the same account. tokenCursors sync.Map // inflight maps an in-progress event ID → the cancel func of its generation // work context, so the maintenance sweep can stop a stuck generation the // moment it abandons the row (instead of letting an orphaned goroutine run on // for minutes and surface a late "success" on an already-abandoned event). inflight *InflightRegistry // 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 } // 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 } return s.conc.Acquire(ctx, "conc:a:"+accountID, max, eventID) } func (s *V1Service) acctRelease(ctx context.Context, accountID, eventID string) { s.conc.Release(ctx, "conc:a:"+accountID, eventID) } // 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 } 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 // event ID. The generation registers on start and removes on finish; the // maintenance sweep calls Cancel when it gives up on (abandons) an event. type InflightRegistry struct { m sync.Map // eventID -> context.CancelFunc } func (r *InflightRegistry) Add(eventID string, cancel context.CancelFunc) { if eventID != "" { r.m.Store(eventID, cancel) } } // Done deregisters an event (called on normal completion). func (r *InflightRegistry) Done(eventID string) { r.m.Delete(eventID) } // Cancel stops an in-flight generation by event ID. Returns true if one was // running and got cancelled. No-op (false) if it already finished. func (r *InflightRegistry) Cancel(eventID string) bool { if v, ok := r.m.LoadAndDelete(eventID); ok { v.(context.CancelFunc)() return true } return false } type APIPrincipal struct { User *model.User TokenType string } type V1ImageRequest struct { Model string Prompt string Size string // Quality is OpenAI's image quality (low|medium|high|auto). For our tiered // models it selects the resolution (low→1K, medium→2K, high→4K, auto→default), // clamped to whatever tiers the model actually prices. Only used when // Resolution is left blank (the strict /v1 OpenAI path); the playground passes // Resolution directly and ignores this. Quality string AspectRatio string Resolution string N int ReferenceImages []string // DeAI applies 去AI特征 post-processing (crop / noise / tone jitter + // re-encode) to the output and charges the per-tier surcharge on top of // the model price. Playground-only; the /v1 OpenAI path never sets it. DeAI bool // BaseURL is the scheme+host of the inbound request (e.g. "https://host"), // used to build absolute, directly-downloadable output URLs. Empty falls // back to a relative "/images/..." path. BaseURL string // AccountID pins the generation to one specific provider account (admin // account-test). Empty keeps the normal pool selection with failover. AccountID string } type V1VideoRequest struct { Model string Prompt string Duration string AspectRatio string Resolution string ReferenceImages []string ReferenceMode string // "frame" or "asset", overrides model default // BaseURL — see V1ImageRequest.BaseURL. BaseURL string // AccountID — see V1ImageRequest.AccountID. AccountID string } 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, users: users, events: events, tokens: tokens, settings: settings, cgroups: cgroups, conc: conc, adobe: adobeClient, chatgpt: chatGPTClient, runway: runwayClient, leonardo: leonardoClient, krea: kreaClient, imagine: imagineClient, grok: grokClient, custom: customClient, store: store, inflight: &InflightRegistry{}, } } // Inflight exposes the registry so the maintenance sweep can cancel a stuck // generation when it abandons that event. func (s *V1Service) Inflight() *InflightRegistry { return s.inflight } // SetRefresh wires the Adobe cookie-refresh service in after construction // (RefreshProfileService is built later in bootstrap, so it can't be a ctor arg // without reordering). Enables refresh-then-retry on a mid-request 401. func (s *V1Service) SetRefresh(r *RefreshProfileService) { s.refresh = r } // SetBannedWords wires the prompt blocklist in after construction. func (s *V1Service) SetBannedWords(r *repo.BannedWordRepository) { s.banned = r } // checkBannedPrompt rejects the request when the prompt contains any banned // word (case-insensitive substring). A hit bumps the word's counter and the // user's 违禁词触发次数 before rejecting. func (s *V1Service) checkBannedPrompt(ctx context.Context, principal *APIPrincipal, prompt string) error { if s.banned == nil || strings.TrimSpace(prompt) == "" { return nil } words, err := s.banned.List(ctx) if err != nil || len(words) == 0 { return nil } lower := strings.ToLower(prompt) for _, w := range words { term := strings.ToLower(strings.TrimSpace(w.Word)) if term == "" || !strings.Contains(lower, term) { continue } userID, userName := "", "" if principal != nil && principal.User != nil { userID = principal.User.ID userName = principal.User.Name if userName == "" { userName = principal.User.Email } } s.banned.RecordHit(ctx, w.ID, w.Word, userID, userName, prompt) return fmt.Errorf("%w: banned word \"%s\"", ErrBannedPrompt, w.Word) } return nil } // logRejectedEvent records a request rejected BEFORE the pending event exists // (banned word, concurrency full, unknown model, insufficient credits…) as a // failed event, so every attempt shows up in the logs. func (s *V1Service) logRejectedEvent(ctx context.Context, kind, modelID string, principal *APIPrincipal, prompt, source, reason string) { event := &model.EventLog{ ID: "evt-" + randomUpper(12), TS: time.Now(), Kind: kind, Status: "failed", Model: strings.TrimSpace(modelID), Prompt: prompt, Source: source, Error: reason, CreatedAt: time.Now(), UpdatedAt: time.Now(), } if m, err := s.models.Get(ctx, event.Model); err == nil { event.Model = m.ID event.Provider = m.Provider } if principal != nil && principal.User != nil { event.UserID = principal.User.ID } _ = s.events.Create(ctx, event) } // refreshAdobeToken re-mints an Adobe account's access token from its cookie // (RefreshNow) and returns the updated row. Used to retry a 401 with a fresh // token instead of replaying the stale one. Returns false if refresh is // unavailable or the cookie can no longer mint a token (genuinely dead). func (s *V1Service) refreshAdobeToken(ctx context.Context, tokenID string) (model.TokenAccount, bool) { if s.refresh == nil { return model.TokenAccount{}, false } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil && proxy != "" { s.refresh.SetProxy(proxy) } } if err := s.refresh.RefreshNow(ctx, tokenID); err != nil { return model.TokenAccount{}, false } t, err := s.tokens.Get(ctx, "adobe", tokenID) if err != nil || t == nil { return model.TokenAccount{}, false } return *t, true } func (s *V1Service) Authenticate(ctx context.Context, authHeader string) (*APIPrincipal, error) { token := ParseBearer(authHeader) if token == "" { return nil, ErrMissingAPIKey } // Only per-user API keys (hashed in the DB) authenticate to /v1. The old // global/shared API_KEY backdoor has been removed. user, err := s.users.GetByAPIKeyHash(ctx, HashAPIKey(token)) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, ErrInvalidAPIKey } return nil, err } if user.Status != "active" { return nil, ErrInvalidAPIKey } _ = s.users.TouchAPIKeyUsage(ctx, HashAPIKey(token)) return &APIPrincipal{ User: user, TokenType: "user", }, nil } func (s *V1Service) ListModels(ctx context.Context) ([]map[string]any, error) { items, err := s.models.List(ctx) if err != nil { return nil, err } now := time.Now().Unix() out := make([]map[string]any, 0, len(items)) for _, item := range items { if !item.Enabled { continue } out = append(out, map[string]any{ "id": item.EffectiveName(), "object": "model", "created": now, "owned_by": item.Provider, "kind": item.Type, "supported_ratios": repo.JSONStrings(item.Ratios), "supported_resolutions": repo.JSONStrings(item.Resolutions), }) } return out, nil } // UserBalance — GET /v1/user/balance 的数据。重新读一次用户行保证实时 // (principal 里那份是鉴权时读的,可能已经过期)。 func (s *V1Service) UserBalance(ctx context.Context, principal *APIPrincipal) (map[string]any, error) { user := principal.User if fresh, err := s.users.GetByID(ctx, user.ID); err == nil && fresh != nil { user = fresh } return map[string]any{ "object": "user.balance", "balance": user.Credits, "used": user.CreditsUsed, "total": user.Credits + user.CreditsUsed, }, nil } func (s *V1Service) PrepareImageRequest(ctx context.Context, principal *APIPrincipal, in V1ImageRequest) (map[string]any, error) { return s.prepareImageExecution(ctx, principal, in, "v1", true) } func (s *V1Service) prepareSessionImage(ctx context.Context, principal *APIPrincipal, in V1ImageRequest) (map[string]any, error) { return s.prepareImageExecution(ctx, principal, in, "user", true) } func (s *V1Service) prepareAdminTestImage(ctx context.Context, principal *APIPrincipal, in V1ImageRequest) (map[string]any, error) { return s.prepareImageExecution(ctx, principal, in, "admin", false) } func (s *V1Service) prepareImageExecution(ctx context.Context, principal *APIPrincipal, in V1ImageRequest, source string, charge bool) (map[string]any, error) { // Detach the whole execution from the request lifecycle. The frontend tracks // progress by polling /jobs/mine, so a client disconnect — or an nginx/CDN // gateway timeout on the slow synchronous response — must NOT cancel an // in-flight generation. Binding to the request ctx meant a cancelled request // (a) spun uselessly in the upstream poll until its 180s timeout and // (b) silently dropped the refund + final status write, leaving the row stuck // pending until the maintenance sweep mislabeled it "abandoned". // // `ctx` (WithoutCancel) is durable and used for ALL bookkeeping (status / // refund / cleanup) so those always land. `genCtx` is the cancellable WORK // context: an 8-min backstop, AND registered in s.inflight so the maintenance // sweep can cancel it the instant it abandons the row — stopping a stuck // generation from running on for minutes and surfacing a late "success" on an // already-abandoned event. ctx = context.WithoutCancel(ctx) if source != "admin" { if err := s.checkBannedPrompt(ctx, principal, in.Prompt); err != nil { s.logRejectedEvent(ctx, "image", in.Model, principal, in.Prompt, source, err.Error()) return nil, err } } // 去AI特征 is gated by a system-settings switch (default off) — drop the // flag when disabled so no surcharge is charged and no processing runs. if in.DeAI && !s.deaiEnabled(ctx) { in.DeAI = false } 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) { s.logRejectedEvent(ctx, "image", in.Model, principal, in.Prompt, source, ErrUserConcurrencyFull.Error()) 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 { s.logRejectedEvent(ctx, "image", in.Model, principal, in.Prompt, source, err.Error()) return nil, err } refCount := len(in.ReferenceImages) // API-key (source "v1") requests don't persist the output: we return the image // as base64 inline (OpenAI gpt-image-1 also returns only b64_json) and never // upload to RustFS, so there's no URL. The event is still logged (empty file) // for usage; the customer logs page hides source="v1" rows. noStore := source == "v1" var fileURL, relativePath string if !noStore { fileURL, relativePath = s.allocateOutput(principal, "png", in.BaseURL) } // upstreamURL is the provider's original artifact URL. For API-key (source // "v1") requests we return it instead of base64. When gatedURL is true the URL // is auth-gated (chatgpt files.oaiusercontent.com — a plain GET 403s), so we // store it on the event and hand the caller a proxy URL // ({base}/v1/images/{eventID}/content) that re-fetches with the account token. var upstreamURL string var gatedURL bool eventID, err := s.logPendingEvent(ctx, "image", modelItem, principal, in.Prompt, aspectRatio, resolution, "", refCount, price, relativePath, source, nil, in.DeAI) if err != nil { return nil, err } // Register so the maintenance sweep can cancel this generation if it abandons // the row; deregister on return. s.inflight.Add(eventID, cancel) defer s.inflight.Done(eventID) startedAt := time.Now() var imageBytes []byte switch s.effectiveProvider(genCtx, modelItem) { case "adobe": b, u, execErr := s.generateAdobeImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, adobe.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, adobe.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, adobe.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "chatgpt": b, u, execErr := s.generateChatGPTImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, chatgpt.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, chatgpt.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, chatgpt.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u gatedURL = true // chatgpt URL needs the account token → proxy it case "leonardo": b, u, execErr := s.generateLeonardoImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, leonardo.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, leonardo.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, leonardo.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "krea": b, u, execErr := s.generateKreaImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, krea.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, krea.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, krea.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "imagine": b, u, execErr := s.generateImagineImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, imagine.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, imagine.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, imagine.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "grok": b, u, execErr := s.generateGrokImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, grok.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, grok.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, grok.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "runway": b, u, execErr := s.generateRunwayImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, runway.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, runway.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, runway.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u case "custom": b, u, execErr := s.generateCustomImage(genCtx, eventID, modelItem, in, aspectRatio, resolution, noStore) if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, custom.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, custom.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, custom.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } imageBytes = b upstreamURL = u default: _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "provider not implemented", 0) return nil, fmt.Errorf("%w: %s", ErrProviderUnsupported, modelItem.Provider) } // 去AI特征: post-process before storing/returning. Best-effort — a decode // failure keeps the original bytes rather than failing a paid generation. if in.DeAI { if processed, derr := applyDeAI(imageBytes); derr == nil { imageBytes = processed } } if !noStore { // Upload to RustFS. On failure the generation fails and credits are // refunded — we never fall back to local disk. if err := s.store.Put(genCtx, relativePath, imageBytes, "image/png"); err != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "storage upload failed: "+err.Error(), 0) return nil, fmt.Errorf("%w: %v", ErrProviderExecution, err) } // Best-effort thumbnail for list views; the image serving route falls // back to the original when the thumb object is missing. if thumb, terr := makeThumbnail(imageBytes); terr == nil { _ = s.store.Put(genCtx, ThumbKey(relativePath), thumb, "image/jpeg") } } elapsedMS := int(time.Since(startedAt).Milliseconds()) if err := s.events.UpdateStatus(ctx, eventID, "success", "", elapsedMS); err != nil { return nil, err } _ = s.models.IncrementGenerationCount(ctx, modelItem.ID) if principal != nil && principal.User != nil { _ = s.users.IncrementGenerationCount(ctx, principal.User.ID) } if charge { _ = s.maybeGrantInviteReward(ctx, principal) } if noStore { // Prefer the provider's original URL — return it directly, no base64. // (API-key requests don't support DeAI, so there's no post-processing that // would invalidate the upstream URL.) if strings.TrimSpace(upstreamURL) != "" { outURL := upstreamURL if gatedURL { // Auth-gated URL (chatgpt): store it on the event and return a proxy // URL that re-fetches with the account token (see OpenImageContent). _ = s.events.SetFile(ctx, eventID, upstreamURL) if base := strings.TrimRight(strings.TrimSpace(in.BaseURL), "/"); base != "" { outURL = base + "/v1/images/" + eventID + "/content" } } return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"url": outURL}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "image", "url": outURL, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } // Fallback: providers without an upstream URL still return base64. b64 := base64.StdEncoding.EncodeToString(imageBytes) return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"b64_json": b64}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "image", "b64_json": b64, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"url": fileURL, "b64_json": nil}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "image", "url": fileURL, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } func (s *V1Service) PrepareVideoRequest(ctx context.Context, principal *APIPrincipal, in V1VideoRequest) (map[string]any, error) { return s.prepareVideoExecution(ctx, principal, in, "v1", true) } func (s *V1Service) prepareSessionVideo(ctx context.Context, principal *APIPrincipal, in V1VideoRequest) (map[string]any, error) { return s.prepareVideoExecution(ctx, principal, in, "user", true) } func (s *V1Service) prepareAdminTestVideo(ctx context.Context, principal *APIPrincipal, in V1VideoRequest) (map[string]any, error) { return s.prepareVideoExecution(ctx, principal, in, "admin", false) } func (s *V1Service) prepareVideoExecution(ctx context.Context, principal *APIPrincipal, in V1VideoRequest, source string, charge bool) (map[string]any, error) { // Detach from the request lifecycle — see prepareImageExecution. `ctx` // (WithoutCancel) carries all bookkeeping; `genCtx` is the cancellable work // context (12-min backstop — video polls up to 10 min — and registered so the // maintenance sweep can cancel a stuck render when it abandons the row). ctx = context.WithoutCancel(ctx) if source != "admin" { if err := s.checkBannedPrompt(ctx, principal, in.Prompt); err != nil { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, source, err.Error()) return nil, err } } genCtx, cancel := context.WithTimeout(ctx, videoGenBudget) 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) { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, source, ErrUserConcurrencyFull.Error()) 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 { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, source, err.Error()) return nil, err } refCount := len(in.ReferenceImages) // API-key (source "v1") requests return base64 inline and never persist a // file — see prepareImageExecution for the rationale. noStore := source == "v1" var fileURL, relativePath string if !noStore { fileURL, relativePath = s.allocateOutput(principal, "mp4", in.BaseURL) } eventID, err := s.logPendingEvent(ctx, "video", modelItem, principal, in.Prompt, aspectRatio, resolution, duration, refCount, price, relativePath, source, nil, false) if err != nil { return nil, err } // Register so the maintenance sweep can cancel this render if it abandons the // row; deregister on return. s.inflight.Add(eventID, cancel) defer s.inflight.Done(eventID) startedAt := time.Now() // API-key (noStore) requests return the upstream video URL directly. // downloadResult=false skips the download. grok asset URLs are auth-gated // (a plain GET 403s) → gatedVideoURL routes them through the /content proxy. prov := s.effectiveProvider(genCtx, modelItem) urlOnly := noStore gatedVideoURL := prov == "grok" var videoBytes []byte var videoURL string var execErr error switch prov { case "adobe": videoBytes, videoURL, execErr = s.generateAdobeVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), !urlOnly) case "runway": videoBytes, videoURL, execErr = s.generateRunwayVideo(genCtx, eventID, modelItem, in, aspectRatio, parseDurationSeconds(duration), !urlOnly) case "grok": videoBytes, videoURL, execErr = s.generateGrokVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), !urlOnly) case "leonardo": videoBytes, videoURL, execErr = s.generateLeonardoVideo(genCtx, eventID, modelItem, in, aspectRatio, parseDurationSeconds(duration), !urlOnly) case "custom": videoBytes, videoURL, execErr = s.generateCustomVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), !urlOnly) default: _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "provider not implemented", 0) return nil, fmt.Errorf("%w: %s", ErrProviderUnsupported, modelItem.Provider) } if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) switch { case errors.Is(execErr, ErrNoProviderAccount): return nil, ErrNoProviderAccount case errors.Is(execErr, adobe.ErrAuth), errors.Is(execErr, runway.ErrAuth), errors.Is(execErr, grok.ErrAuth), errors.Is(execErr, leonardo.ErrAuth), errors.Is(execErr, custom.ErrAuth): return nil, ErrProviderAuth case errors.Is(execErr, adobe.ErrQuotaExhausted), errors.Is(execErr, runway.ErrQuotaExhausted), errors.Is(execErr, grok.ErrQuotaExhausted), errors.Is(execErr, leonardo.ErrQuotaExhausted), errors.Is(execErr, custom.ErrQuotaExhausted): return nil, ErrProviderQuota case errors.Is(execErr, adobe.ErrTemporaryUpstream), errors.Is(execErr, runway.ErrTemporaryUpstream), errors.Is(execErr, grok.ErrTemporaryUpstream), errors.Is(execErr, leonardo.ErrTemporaryUpstream), errors.Is(execErr, custom.ErrTemporaryUpstream): return nil, ErrProviderTemporary default: return nil, fmt.Errorf("%w: %v", ErrProviderExecution, execErr) } } if !noStore { if err := s.store.Put(genCtx, relativePath, videoBytes, "video/mp4"); err != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "storage upload failed: "+err.Error(), 0) return nil, fmt.Errorf("%w: %v", ErrProviderExecution, err) } // Best-effort stills: first frame (downscaled) for list thumbnails and // the full-res last frame for 首尾帧 continuation. Missing objects fall // back to the video itself at serve time. if thumb, last, terr := extractVideoFrames(genCtx, videoBytes); terr == nil { if len(thumb) > 0 { _ = s.store.Put(genCtx, ThumbKey(relativePath), thumb, "image/jpeg") } if len(last) > 0 { _ = s.store.Put(genCtx, LastFrameKey(relativePath), last, "image/jpeg") } } } elapsedMS := int(time.Since(startedAt).Milliseconds()) if err := s.events.UpdateStatus(ctx, eventID, "success", "", elapsedMS); err != nil { return nil, err } _ = s.models.IncrementGenerationCount(ctx, modelItem.ID) if principal != nil && principal.User != nil { _ = s.users.IncrementGenerationCount(ctx, principal.User.ID) } if charge { _ = s.maybeGrantInviteReward(ctx, principal) } if noStore && strings.TrimSpace(videoURL) != "" { // Return the upstream video URL. grok URLs are auth-gated → store on the // event and hand back the /content proxy (re-fetches with the account token). outURL := videoURL if gatedVideoURL { _ = s.events.SetFile(ctx, eventID, videoURL) if base := strings.TrimRight(strings.TrimSpace(in.BaseURL), "/"); base != "" { outURL = base + "/v1/videos/" + eventID + "/content" } } return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"url": outURL}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "video", "url": outURL, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } if noStore { b64 := base64.StdEncoding.EncodeToString(videoBytes) return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"b64_json": b64}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "video", "b64_json": b64, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } return map[string]any{ "created": time.Now().Unix(), "data": []map[string]any{{"url": fileURL}}, "model": modelItem.EffectiveName(), "provider": modelItem.Provider, "kind": "video", "url": fileURL, "elapsed_ms": elapsedMS, "charged": price, "credits": principalCredits(principal), }, nil } // ===== /v1/videos — OpenAI Sora-style async jobs ===== // POST /v1/videos charges + creates a pending event and renders in the // background; the render captures only the UPSTREAM video URL (no download, no // RustFS). GET /v1/videos/{id} polls status; /content proxies the upstream URL. // StartVideoJob validates+charges, creates the job event, kicks the render off in // the background, and returns the OpenAI video object (status "queued"). func (s *V1Service) StartVideoJob(ctx context.Context, principal *APIPrincipal, in V1VideoRequest) (map[string]any, error) { ctx = context.WithoutCancel(ctx) if err := s.checkBannedPrompt(ctx, principal, in.Prompt); err != nil { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, "v1", err.Error()) return nil, err } modelItem, resolution, aspectRatio, duration, price, err := s.prepareVideo(ctx, principal, in, true) if err != nil { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, "v1", err.Error()) return nil, err } // Validate reference_mode against model capabilities and reference count. if rm := strings.TrimSpace(in.ReferenceMode); rm != "" { supported := strings.TrimSpace(modelItem.ReferenceMode) if supported == "none" || supported == "" { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, "v1", "reference_mode not supported for this model") return nil, errors.New("reference_mode not supported for this model") } if rm != "frame" && rm != "asset" { s.logRejectedEvent(ctx, "video", in.Model, principal, in.Prompt, "v1", "reference_mode must be 'frame' or 'asset'") return nil, errors.New("reference_mode must be 'frame' or 'asset'") } if rm == "frame" && len(in.ReferenceImages) > 2 { return nil, fmt.Errorf("frame mode supports at most 2 reference images (first+last frame), got %d", len(in.ReferenceImages)) } if strings.TrimSpace(in.ReferenceMode) == modelItem.ReferenceMode { in.ReferenceMode = "" // same as default, don't override } } // Source "v1": no output file is allocated — the result is the upstream URL, // stored on the event when the render completes. eventID, err := s.logPendingEvent(ctx, "video", modelItem, principal, in.Prompt, aspectRatio, resolution, duration, len(in.ReferenceImages), price, "", "v1", nil, false) if err != nil { return nil, err } go s.runVideoJob(ctx, principal, in, modelItem, eventID, aspectRatio, resolution, duration, price) return videoJobObject(eventID, modelItem.EffectiveName(), "queued", 0, duration, sizeFromRatioRes(aspectRatio, resolution), time.Now().Unix(), 0, ""), nil } // runVideoJob renders the clip in the background, capturing the upstream URL // (downloadResult=false → no bytes, no RustFS) and storing it on the event. func (s *V1Service) runVideoJob(ctx context.Context, principal *APIPrincipal, in V1VideoRequest, modelItem *model.ModelConfig, eventID, aspectRatio, resolution, duration string, price float64) { genCtx, cancel := context.WithTimeout(ctx, videoGenBudget) defer cancel() s.inflight.Add(eventID, cancel) defer s.inflight.Done(eventID) startedAt := time.Now() // No-store: capture only the UPSTREAM video URL. /content streams it on demand // (grok URLs are auth-gated → fetched with the generating account's token). var videoURL string var execErr error switch s.effectiveProvider(genCtx, modelItem) { case "adobe": _, videoURL, execErr = s.generateAdobeVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), false) case "runway": _, videoURL, execErr = s.generateRunwayVideo(genCtx, eventID, modelItem, in, aspectRatio, parseDurationSeconds(duration), false) case "grok": _, videoURL, execErr = s.generateGrokVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), false) case "leonardo": _, videoURL, execErr = s.generateLeonardoVideo(genCtx, eventID, modelItem, in, aspectRatio, parseDurationSeconds(duration), false) case "custom": _, videoURL, execErr = s.generateCustomVideo(genCtx, eventID, modelItem, in, aspectRatio, resolution, parseDurationSeconds(duration), false) default: _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "provider not implemented", 0) return } if execErr != nil { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", execErr.Error(), 0) return } if strings.TrimSpace(videoURL) == "" { _ = s.refundIfNeeded(ctx, principal, eventID, price) _ = s.events.UpdateStatus(ctx, eventID, "failed", "upstream returned no video url", 0) return } // Store the upstream URL as the event's "file"; /content fetches it on demand. if err := s.events.MarkVideoReady(ctx, eventID, videoURL, int(time.Since(startedAt).Milliseconds())); err != nil { return } _ = s.models.IncrementGenerationCount(ctx, modelItem.ID) if principal != nil && principal.User != nil { _ = s.users.IncrementGenerationCount(ctx, principal.User.ID) } _ = s.maybeGrantInviteReward(ctx, principal) } // VideoJob returns the OpenAI video object for a job, scoped to the caller. func (s *V1Service) VideoJob(ctx context.Context, principal *APIPrincipal, id string) (map[string]any, error) { ev, err := s.videoEventForUser(ctx, principal, id) if err != nil { return nil, err } status, progress := videoJobStatus(ev) completedAt := int64(0) if ev.Status == "success" || ev.Status == "failed" { completedAt = ev.UpdatedAt.Unix() } errMsg := "" if ev.Status == "failed" { errMsg = ev.Error } modelName := ev.Model if nameByID, nerr := s.models.NameMap(ctx); nerr == nil { if name, ok := nameByID[ev.Model]; ok && strings.TrimSpace(name) != "" { modelName = name } } return videoJobObject(ev.ID, modelName, status, progress, ev.Duration, sizeFromRatioRes(ev.Ratio, ev.Resolution), ev.TS.Unix(), completedAt, errMsg), nil } // OpenVideoContent streams a completed job's video by proxying the stored // upstream URL (downloaded on demand — never persisted). func (s *V1Service) OpenVideoContent(ctx context.Context, principal *APIPrincipal, id string) (io.ReadCloser, string, error) { ev, err := s.videoEventForUser(ctx, principal, id) if err != nil { return nil, "", err } if ev.Status != "success" || strings.TrimSpace(ev.File) == "" { return nil, "", ErrVideoNotReady } // grok asset URLs (assets.grok.com) are auth-gated — a plain GET 403s. Stream // them through the SAME account that generated the clip, using its token. If // that account is gone (grok pools churn often), the clip is unrecoverable. if ev.Provider == "grok" && s.grok != nil { if s.settings != nil { if proxy, perr := s.settings.GetValue(ctx, "proxy.url"); perr == nil { s.grok.SetProxy(proxy) } } acct, _ := s.tokens.Get(ctx, "grok", ev.AccountID) if acct == nil || strings.TrimSpace(acct.Value) == "" { return nil, "", fmt.Errorf("%w: grok account no longer available for this video", ErrProviderTemporary) } return s.grok.OpenAsset(ctx, acct.Value, ev.File) } // Other providers return publicly-fetchable URLs — proxy directly. req, err := http.NewRequestWithContext(ctx, http.MethodGet, ev.File, nil) if err != nil { return nil, "", err } resp, err := (&http.Client{Timeout: 5 * time.Minute}).Do(req) if err != nil { return nil, "", fmt.Errorf("%w: fetch upstream video: %v", ErrProviderTemporary, err) } if resp.StatusCode != http.StatusOK { resp.Body.Close() return nil, "", fmt.Errorf("%w: upstream video status %d", ErrProviderTemporary, resp.StatusCode) } ct := strings.TrimSpace(resp.Header.Get("Content-Type")) if ct == "" { ct = "video/mp4" } return resp.Body, ct, nil } // OpenImageContent streams a no-store image by proxying the stored upstream URL. // chatgpt URLs are auth-gated (files.oaiusercontent.com — a plain GET 403s), so // they're fetched through the generating account's token; other providers' // URLs are public and proxied directly. Never persisted. func (s *V1Service) OpenImageContent(ctx context.Context, principal *APIPrincipal, id string) (io.ReadCloser, string, error) { ev, err := s.events.GetByID(ctx, strings.TrimSpace(id)) if err != nil { return nil, "", err } if ev == nil || ev.Kind != "image" { return nil, "", ErrVideoJobNotFound } if principal != nil && principal.User != nil && ev.UserID != principal.User.ID { return nil, "", ErrVideoJobNotFound } if ev.Status != "success" || strings.TrimSpace(ev.File) == "" { return nil, "", ErrVideoNotReady } if ev.Provider == "chatgpt" && s.chatgpt != nil { if s.settings != nil { if proxy, perr := s.settings.GetValue(ctx, "proxy.url"); perr == nil { s.chatgpt.SetProxy(proxy) } } acct, _ := s.tokens.Get(ctx, "chatgpt", ev.AccountID) if acct == nil || strings.TrimSpace(acct.Value) == "" { return nil, "", fmt.Errorf("%w: chatgpt account no longer available for this image", ErrProviderTemporary) } return s.chatgpt.OpenAsset(ctx, acct.Value, ev.File) } req, err := http.NewRequestWithContext(ctx, http.MethodGet, ev.File, nil) if err != nil { return nil, "", err } resp, err := (&http.Client{Timeout: 5 * time.Minute}).Do(req) if err != nil { return nil, "", fmt.Errorf("%w: fetch upstream image: %v", ErrProviderTemporary, err) } if resp.StatusCode != http.StatusOK { resp.Body.Close() return nil, "", fmt.Errorf("%w: upstream image status %d", ErrProviderTemporary, resp.StatusCode) } ct := strings.TrimSpace(resp.Header.Get("Content-Type")) if ct == "" { ct = "image/png" } return resp.Body, ct, nil } func (s *V1Service) videoEventForUser(ctx context.Context, principal *APIPrincipal, id string) (*model.EventLog, error) { ev, err := s.events.GetByID(ctx, strings.TrimSpace(id)) if err != nil { return nil, err } if ev == nil || ev.Kind != "video" { return nil, ErrVideoJobNotFound } if principal != nil && principal.User != nil && ev.UserID != principal.User.ID { return nil, ErrVideoJobNotFound } return ev, nil } // videoJobStatus maps our event status → OpenAI's (queued|in_progress|completed| // failed) plus a coarse progress. func videoJobStatus(ev *model.EventLog) (string, int) { switch ev.Status { case "success": return "completed", 100 case "failed": return "failed", 0 default: if strings.TrimSpace(ev.AccountID) != "" { return "in_progress", 50 } return "queued", 0 } } func videoJobObject(id, modelID, status string, progress int, seconds, size string, createdAt, completedAt int64, errMsg string) map[string]any { obj := map[string]any{ "id": id, "object": "video", "model": modelID, "status": status, "progress": progress, "created_at": createdAt, "size": size, "seconds": strings.TrimSuffix(strings.TrimSpace(seconds), "s"), } if completedAt > 0 { obj["completed_at"] = completedAt } else { obj["completed_at"] = nil } if errMsg != "" { obj["error"] = map[string]any{"message": errMsg} } else { obj["error"] = nil } return obj } // sizeFromRatioRes reconstructs an OpenAI-style "WxH" label from our stored ratio // + resolution tier (best-effort; only for display in the job object). func sizeFromRatioRes(ratio, resolution string) string { long := 720 res := strings.ToUpper(resolution) switch { case strings.Contains(res, "1080") || strings.Contains(res, "2K"): long = 1080 case strings.Contains(res, "4K") || strings.Contains(res, "2160"): long = 2160 } w, h := long, long switch strings.TrimSpace(ratio) { case "16:9": w, h = long, long*9/16 case "9:16": w, h = long*9/16, long case "4:3": w, h = long, long*3/4 case "3:4": w, h = long*3/4, long case "1:1": w, h = long, long default: w, h = long, long*9/16 } return fmt.Sprintf("%dx%d", w, h) } // hasActiveProviderToken reports whether the provider pool holds at least one // usable token for this kind of generation — mirrors the selection filter in // the generate* paths. Used to fail fast (before charging / creating a job) // with a clear "no account" error instead of dialing upstream with no token. func (s *V1Service) hasActiveProviderToken(ctx context.Context, provider, kind string) (bool, error) { items, err := s.tokens.ListByPool(ctx, provider) if err != nil { return false, err } for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } // Adobe accounts are credit-based (积分号) — no per-kind quota locks. return true, nil } return false, nil } func (s *V1Service) prepareImage(ctx context.Context, principal *APIPrincipal, in V1ImageRequest, charge bool) (*model.ModelConfig, string, string, float64, error) { modelID := strings.TrimSpace(in.Model) prompt := strings.TrimSpace(in.Prompt) if modelID == "" || prompt == "" { return nil, "", "", 0, errors.New("model and prompt required") } modelItem, err := s.models.Get(ctx, modelID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, "", "", 0, ErrUnknownModel } return nil, "", "", 0, err } if !modelItem.Enabled || modelItem.Type != "image" { return nil, "", "", 0, ErrUnknownModel } // Fail fast before charging if the provider has no usable account. Use the // effective provider: a custom upstream serving this model id routes to // "custom" (effectiveProvider only returns it when such an account exists, so // the precheck is satisfied); otherwise check the native provider pool. if eff := s.effectiveProvider(ctx, modelItem); eff != "custom" { if ok, err := s.hasActiveProviderToken(ctx, eff, "image"); err != nil { return nil, "", "", 0, err } else if !ok { return nil, "", "", 0, ErrNoProviderAccount } } refLimit := 0 if modelItem.ImageToImage { refLimit = modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 1 } } if len(in.ReferenceImages) > refLimit { return nil, "", "", 0, errors.New("too many reference images") } // Reject oversized reference images before charging (all providers, all paths). 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) // Snap to the nearest ratio the model actually supports — a `size`-derived // ratio (e.g. 1:3) must never be passed through to an upstream that rejects // it (Runway 400s on ratios outside its list). aspectRatio = snapRatio(aspectRatio, repo.JSONStrings(modelItem.Ratios)) // 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 // the client doesn't get rejected as "unsupported or unpriced". if _, ok := modelPrice(modelItem, "image", resolution, "", false); !ok { if fb := firstPricedResolution(modelItem); fb != "" { resolution = fb } } var surcharge float64 if in.DeAI { surcharge = s.deaiSurcharge(ctx, resolution) } price, err := s.chargeForModel(ctx, principal, modelItem, "image", resolution, "", surcharge, charge) if err != nil { return nil, "", "", 0, err } return modelItem, resolution, aspectRatio, price, nil } func (s *V1Service) prepareVideo(ctx context.Context, principal *APIPrincipal, in V1VideoRequest, charge bool) (*model.ModelConfig, string, string, string, float64, error) { modelID := strings.TrimSpace(in.Model) prompt := strings.TrimSpace(in.Prompt) duration := strings.TrimSpace(in.Duration) if modelID == "" || prompt == "" { return nil, "", "", "", 0, errors.New("model and prompt required") } if duration == "" { return nil, "", "", "", 0, errors.New("duration required") } modelItem, err := s.models.Get(ctx, modelID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, "", "", "", 0, ErrUnknownModel } return nil, "", "", "", 0, err } if !modelItem.Enabled || modelItem.Type != "video" { return nil, "", "", "", 0, ErrUnknownModel } // Validate duration against model's supported range (from Durations JSON array). if secs := parseDurationSeconds(duration); secs > 0 { if durList := repo.JSONStrings(modelItem.Durations); len(durList) > 0 { minSecs, maxSecs := 9999, 0 for _, d := range durList { n := parseDurationSeconds(d) if n > 0 { if n < minSecs { minSecs = n } if n > maxSecs { maxSecs = n } } } if secs < minSecs || secs > maxSecs { return nil, "", "", "", 0, fmt.Errorf("duration %ds out of range [%d-%d] for model %s", secs, minSecs, maxSecs, modelItem.EffectiveName()) } } } // Fail fast before charging — effective provider (custom upstream by id, else native). if eff := s.effectiveProvider(ctx, modelItem); eff == "custom" { // custom serves this id (effectiveProvider guaranteed it) — precheck ok } else if ok, err := s.hasActiveProviderToken(ctx, eff, "video"); err != nil { return nil, "", "", "", 0, err } else if !ok { return nil, "", "", "", 0, ErrNoProviderAccount } refLimit := modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 10 } if len(in.ReferenceImages) > refLimit { return nil, "", "", "", 0, errors.New("too many reference images") } // Reject oversized reference images before charging (all providers, all paths). if err := ensureReferenceSizes(in.ReferenceImages); err != nil { return nil, "", "", "", 0, err } // Runway i2v strictly requires exactly one first-frame image. Enforce it here, // BEFORE charging, so a missing/extra frame fails fast instead of charge → // upstream reject → refund. generateRunwayVideo keeps its own guard too. if modelItem.Provider == "runway" { n := 0 for _, r := range in.ReferenceImages { if strings.TrimSpace(r) != "" { n++ } } if n != 1 { return nil, "", "", "", 0, errors.New("runway 图生视频需要且仅需 1 张首帧图") } } // Leonardo seedance 的参考资产分三类且各有上限/时长限制,在扣费前拦掉, // 不让请求带着非法参考走到上游。 if modelItem.Provider == "leonardo" { if _, err := classifyLeonardoVideoRefs(in.ReferenceImages, leonardoVideoSpecOf(modelItem.ID)); err != nil { return nil, "", "", "", 0, err } } aspectRatio := strings.TrimSpace(strings.ReplaceAll(in.AspectRatio, "x", ":")) if aspectRatio == "" { aspectRatio = "16:9" } resolution := strings.TrimSpace(in.Resolution) if resolution == "" { resolution = "720p" } price, err := s.chargeForModel(ctx, principal, modelItem, "video", resolution, duration, 0, charge) if err != nil { return nil, "", "", "", 0, err } // 规范化 duration 字段:前端 per_second 计费模式可能发来 "per_second" 字符串, // 统一转为 "Xs" 格式(如 "4s")存库,避免日志显示原始键名。 if n := parseDurationSeconds(duration); n > 0 { duration = fmt.Sprintf("%ds", n) } return modelItem, resolution, aspectRatio, duration, price, nil } func (s *V1Service) chargeForModel(ctx context.Context, principal *APIPrincipal, modelItem *model.ModelConfig, kind, resolution, duration string, surcharge float64, charge bool) (float64, error) { // 代理用户走代理价(某档未设代理价则回退普通价)。principal.User 即将被扣费的 // 用户,无论画图台还是 key 调用都从这里取,所以一处即覆盖所有路径。 agent := principal != nil && principal.User != nil && principal.User.Role == "agent" price, ok := modelPrice(modelItem, kind, resolution, duration, agent) if !ok { return 0, ErrUnsupportedParams } price += surcharge if !charge || principal == nil || principal.User == nil { return 0, nil } updated, debited, err := s.users.TryDebitCredits(ctx, principal.User.ID, price) if err != nil { return 0, err } if !debited { if updated != nil { principal.User = updated } return 0, ErrInsufficientFunds } principal.User = updated return price, nil } func (s *V1Service) userDir(principal *APIPrincipal) string { if principal == nil { return "anon" } return OwnerDir(principal.User) } // OwnerDir is the storage directory (= /images// segment) a user's outputs // live under: sanitized name → sanitized email-local → id → "anon". func OwnerDir(user *model.User) string { if user != nil { if d := sanitizeOwnerName(user.Name); d != "" { return d } if d := sanitizeOwnerName(strings.Split(user.Email, "@")[0]); d != "" { return d } if user.ID != "" { return user.ID } } return "anon" } // contentTypeForExt maps a file extension to a MIME type for storage uploads. func contentTypeForExt(ext string) string { switch strings.ToLower(strings.TrimPrefix(ext, ".")) { case "png": return "image/png" case "jpg", "jpeg": return "image/jpeg" case "webp": return "image/webp" case "gif": return "image/gif" case "mp4": return "video/mp4" case "webm": return "video/webm" case "mov": return "video/quicktime" default: return "application/octet-stream" } } // imageExtFromBytes sniffs a sensible file extension from the magic bytes so the // saved reference keeps its real type (the /images handler types by extension). func imageExtFromBytes(b []byte) string { switch { case len(b) >= 3 && b[0] == 0xFF && b[1] == 0xD8 && b[2] == 0xFF: return "jpg" case len(b) >= 6 && string(b[0:6]) == "GIF89a", len(b) >= 6 && string(b[0:6]) == "GIF87a": return "gif" case len(b) >= 12 && string(b[0:4]) == "RIFF" && string(b[8:12]) == "WEBP": return "webp" default: return "png" } } // allocateOutput builds the object key (= relative path, user-scoped) and the // directly-downloadable URL pointing at this site's /images proxy. Nothing is // written here — the bytes are uploaded to RustFS by the caller. func (s *V1Service) allocateOutput(principal *APIPrincipal, ext, baseURL string) (string, string) { userDir := s.userDir(principal) filename := time.Now().Format("20060102-150405") + "-" + randomUpper(8) + "." + strings.TrimPrefix(ext, ".") relativePath := filepath.ToSlash(filepath.Join(userDir, filename)) // OpenAI-style clients need a directly-downloadable absolute URL. When the // inbound request's base URL is known, build "{scheme}://{host}/images/..."; // otherwise fall back to the relative path for backward compatibility. if base := strings.TrimRight(strings.TrimSpace(baseURL), "/"); base != "" { return base + "/images/" + relativePath, relativePath } return "/images/" + relativePath, relativePath } func (s *V1Service) logPendingEvent(ctx context.Context, kind string, modelItem *model.ModelConfig, principal *APIPrincipal, prompt, ratio, resolution, duration string, refs int, cost float64, file, source string, refFiles []string, deai bool) (string, error) { event := &model.EventLog{ ID: "evt-" + randomUpper(12), TS: time.Now(), Kind: kind, Status: "pending", Model: modelItem.ID, Provider: modelItem.Provider, Prompt: prompt, Ratio: ratio, Resolution: resolution, Duration: duration, Refs: refs, DeAI: deai, Source: source, Cost: cost, File: file, CreatedAt: time.Now(), UpdatedAt: time.Now(), } if len(refFiles) > 0 { event.RefFiles = jsonArray(refFiles) } if principal != nil && principal.User != nil { event.UserID = principal.User.ID } if err := s.events.Create(ctx, event); err != nil { return "", err } return event.ID, nil } func (s *V1Service) finishUnimplementedEvent(ctx context.Context, eventID string) error { return s.events.UpdateStatus(ctx, eventID, "failed", "generation executor not implemented yet", 0) } // videoGenBudget caps one video render end-to-end (submit + poll + download). // 上游慢的时候(seedance 长镜头)12 分钟不够,统一给 30 分钟。 const videoGenBudget = 30 * time.Minute // grokConcurrencyPerAccount is how many simultaneous generations one grok account // may run (grok tolerates 10, unlike the 1-per-account default elsewhere). const grokConcurrencyPerAccount = 10 // maxTempDeadAccounts caps how many accounts the "temporary error = fail over" // policy may burn per request before giving up, so an upstream-wide blip // ("system under load") can't fan a single request out across the whole pool. // After this many accounts fail this way, the request fails. const maxTempDeadAccounts = 10 // runPoolWithFailover drives a generation across a round-robin-ordered account // list with per-error-class behavior, so a bad request never burns the whole // pool while genuinely limited accounts still fail over: // - 额度耗尽 quota → mark the account and FAIL OVER to the next account // immediately (same-account retry can't help). Repeats until one succeeds or // the pool is exhausted. // - 认证失效 auth → refresh the token from its cookie and retry ONCE with the // fresh token; if it still auth-fails (or there's nothing to refresh, e.g. // chatgpt's JWT IS the credential), mark the account and fail over. // - 上游临时 temporary → record the failure (no disable/dead) and FAIL OVER to // the next account immediately, capped at maxTempDeadAccounts accounts so a // pool-wide blip can't fan a single request out across everything. // - 参数错 / request-level (anything else) → return immediately, no retry, no // account penalty (the account isn't at fault). // // Returns the actual upstream error (never a synthetic "retry failed"). On // success it stamps success_total/fails=0 on the winning account. classify maps // a provider error to (isAuth, isQuota, isTemporary). refreshOnAuth (nil for // providers whose token IS the credential) re-mints the account's token so an // auth retry uses a FRESH token instead of replaying the stale one. func (s *V1Service) runPoolWithFailover(ctx context.Context, eventID, pool string, active []model.TokenAccount, kind string, attempt func(token model.TokenAccount) ([]byte, error), classify func(error) (isAuth, isQuota, isTemporary, isDead bool), refreshOnAuth func(tokenID string) (model.TokenAccount, bool), tempFailover bool, ) ([]byte, error) { var lastErr error busy := 0 tempDeadCount := 0 for _, token := range active { // Per-account concurrency gate (defaults to 1 for built-in pools). if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { 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.acctRelease(ctx, token.ID, eventID) return s.tryAccount(ctx, eventID, pool, token, kind, attempt, classify, refreshOnAuth, tempFailover) }() if err == nil { return data, nil } lastErr = err if tempDead { // temp-failover policy: this account hit a temporary upstream error. // Cap how many accounts one request may burn before we stop, so an // upstream-wide blip doesn't fan out across the whole pool. tempDeadCount++ if tempDeadCount >= maxTempDeadAccounts { return nil, lastErr } } if failover { continue } // temporary exhausted or request-level error → surface it, no fan-out. return nil, lastErr } // Nothing ran. If accounts were skipped ONLY because they were all busy // (no real failure), tell the caller the pool is at its concurrency cap. if lastErr == nil { if busy > 0 { return nil, ErrConcurrencyFull } return nil, ErrProviderExecution } return nil, lastErr } // tryAccount runs one account's attempt with the pool's retry policy: // 额度耗尽/认证失效 → mark + failover; 上游临时 → record failure + failover (capped // via the tempDead return); 参数错 → fail fast. Returns (data, err, failover, // tempDead) — failover=true means move on to the next account. The per-account // concurrency gate is held by the caller. func (s *V1Service) tryAccount(ctx context.Context, eventID, pool string, token model.TokenAccount, kind string, attempt func(token model.TokenAccount) ([]byte, error), classify func(error) (isAuth, isQuota, isTemporary, isDead bool), refreshOnAuth func(tokenID string) (model.TokenAccount, bool), tempFailover bool, ) ([]byte, error, bool, bool) { _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) authRefreshed := false if strings.TrimSpace(token.Value) == "" && refreshOnAuth != nil { if refreshed, ok := refreshOnAuth(token.ID); ok { token = refreshed authRefreshed = true } else { s.markTokenDead(ctx, pool, token, kind) return nil, ErrProviderExecution, true, true } } for { data, err := attempt(token) if err == nil { _, _ = s.tokens.Update(ctx, pool, token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) return data, nil, false, false } isAuth, isQuota, isTemp, isDead := classify(err) if isQuota { s.markTokenFailure(ctx, pool, token, kind, false, true) return nil, err, true, false } if isAuth { // A 403 user_not_entitled means the account has no Firefly entitlement // — refreshing the access token can't grant one, so kill it now instead // of leaving it in rotation to burn every future request. if errors.Is(err, adobe.ErrNotEntitled) { s.markTokenDead(ctx, pool, token, kind) return nil, err, true, true } // Refresh from cookie and retry ONCE; otherwise the credential is dead. if refreshOnAuth != nil && !authRefreshed { if refreshed, ok := refreshOnAuth(token.ID); ok { token = refreshed authRefreshed = true continue } } s.markTokenFailure(ctx, pool, token, kind, true, false) return nil, err, true, false } // Fatal / temporary-under-failover-policy upstream error. if isDead || (isTemp && tempFailover) { if tempFailover { // Ops policy (adobe): NEVER kill on these upstream errors — a // genuinely bad account and a transient Adobe blip (429/5xx/ // overload) look the same, and killing wipes healthy accounts. // Record the failure and fail over to the next account (no // disable/dead). The 4th return value caps how many accounts one // request may burn this way (maxTempDeadAccounts) so a pool-wide // blip can't fan a single request across the whole pool. s.markTokenFailure(ctx, pool, token, kind, false, false) return nil, err, true, true } s.markTokenDead(ctx, pool, token, kind) return nil, err, true, true } if isTemp { // Temporary upstream error → record the failure (no disable/dead) and // fail over to the NEXT account, capped via the tempDead return so a // pool-wide blip can't fan one request across the whole pool. s.markTokenFailure(ctx, pool, token, kind, false, false) return nil, err, true, true } return nil, err, false, false // 参数错 / request-level } } func adobeErrClass(e error) (bool, bool, bool, bool) { return errors.Is(e, adobe.ErrAuth), errors.Is(e, adobe.ErrQuotaExhausted), errors.Is(e, adobe.ErrTemporaryUpstream) || errors.Is(e, adobe.ErrRateLimited), errors.Is(e, adobe.ErrDeadUpstream) } // noStore url-only mode: adobe returns a presigned image URL (meta["image_url"]); // skip the download and return it directly. func (s *V1Service) generateAdobeImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.adobe == nil { return nil, "", errors.New("adobe client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.adobe.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "adobe") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { // Adobe accounts are credit-based (积分号) — no per-kind quota locks. // Only skip accounts that are dead or disabled. if item.Status != "active" || item.Dead { continue } // plan 未探测到的号既不算普号也不算会员号,置死号、不参与调度 if planUnknown(item.Meta) { s.markPlanUnknownDead(ctx, "adobe", item.ID) continue } // 普号(free)只能调度 free_allowed 的模型(香蕉2 仅 1K) if !freeAccountsAllowed(modelItem, resolution) && isFreeAccount(item.Meta) { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("adobe", active) // 非 seedance 图片生成:普号 → 子号 → 母号 active = prioritizeSubAccounts(active) refs, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } // Round-robin order. Adobe uses tempFailover=true: a temporary upstream error // ("system under load") fails over to the next account without penalizing the // current one, capped at maxTempDeadAccounts; auth/quota also fail over // (see runPoolWithFailover). imageURL is captured from the successful attempt. var imageURL string data, err := s.runPoolWithFailover(ctx, eventID, "adobe", active, "image", func(token model.TokenAccount) ([]byte, error) { var blobIDs []string for _, ref := range refs { id, upErr := s.adobe.UploadImage(ctx, token.Value, ref, "image/png", "") if upErr != nil { if errors.Is(upErr, adobe.ErrRateLimited) { recoverAt := time.Now().Add(4 * time.Hour) s.tokens.Update(ctx, "adobe", token.ID, map[string]any{ "status": "quota", "quota_recover_at": &recoverAt, }) } return nil, upErr } blobIDs = append(blobIDs, id) } d, meta, genErr := s.adobe.GenerateImage(ctx, token.Value, modelItem.ID, in.Prompt, aspectRatio, resolution, blobIDs, !urlOnly) if genErr == nil { imageURL = strings.TrimSpace(stringValue(meta["image_url"])) } return d, genErr }, adobeErrClass, func(id string) (model.TokenAccount, bool) { return s.refreshAdobeToken(ctx, id) }, true) return data, imageURL, err } func (s *V1Service) generateAdobeVideo(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1VideoRequest, aspectRatio, resolution string, durationSeconds int, downloadResult bool) ([]byte, string, error) { if s.adobe == nil { return nil, "", errors.New("adobe client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.adobe.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "adobe") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead { continue } // plan 未探测到的号既不算普号也不算会员号,置死号、不参与调度 if planUnknown(item.Meta) { s.markPlanUnknownDead(ctx, "adobe", item.ID) continue } // Seedance 模型只允许 VIP 母号:必须正向识别(plan 非 free、非子号、 // 积分 >4000),plan/额度未探测的账号一律不参与调度 if isSeedanceModel(modelItem.ID) && !isVipMotherAccount(item.Meta) { continue } // 普号(free)只能调度 free_allowed 的模型 if !freeAccountsAllowed(modelItem, resolution) && isFreeAccount(item.Meta) { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("adobe", active) // 非 seedance 视频生成:普号 → 子号 → 母号 // (seedance 已在上面过滤掉子号,此处无需额外处理) if !isSeedanceModel(modelItem.ID) { active = prioritizeSubAccounts(active) } refLimit := modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 10 } refs, err := decodeReferenceImages(in.ReferenceImages, refLimit) if err != nil { return nil, "", err } // Classify refs for seedance: images (usage:style), videos, audio (usage:source). var imgRefs, vidRefs, audRefs [][]byte for _, r := range refs { switch detectMediaType(r) { case "video": vidRefs = append(vidRefs, r) case "audio": audRefs = append(audRefs, r) default: imgRefs = append(imgRefs, r) } } prompt := in.Prompt engine, upstreamModel := resolveAdobeVideoEngine(modelItem.ID) referenceMode := defaultString(strings.TrimSpace(modelItem.ReferenceMode), "frame") if rm := strings.TrimSpace(in.ReferenceMode); rm != "" { referenceMode = rm } // Round-robin order; fail over to the next account on auth/quota; temporary // upstream errors fail over too without penalizing the account (tempFailover, // capped at maxTempDeadAccounts). videoURL is // captured from the successful attempt's meta (the upstream presigned URL). var videoURL string data, err := s.runPoolWithFailover(ctx, eventID, "adobe", active, "video", func(token model.TokenAccount) ([]byte, error) { var blobIDs []string for _, ref := range imgRefs { id, upErr := s.adobe.UploadImage(ctx, token.Value, ref, "image/png", engine) if upErr != nil { if errors.Is(upErr, adobe.ErrRateLimited) { recoverAt := time.Now().Add(4 * time.Hour) s.tokens.Update(ctx, "adobe", token.ID, map[string]any{ "status": "quota", "quota_recover_at": &recoverAt, }) } return nil, upErr } blobIDs = append(blobIDs, id) } var videoBlobIDs []string for _, ref := range vidRefs { id, upErr := s.adobe.UploadImage(ctx, token.Value, ref, "video/mp4", engine) if upErr != nil { return nil, upErr } videoBlobIDs = append(videoBlobIDs, id) } var audioBlobIDs []string for _, ref := range audRefs { id, upErr := s.adobe.UploadImage(ctx, token.Value, ref, "audio/mp3", engine) if upErr != nil { return nil, upErr } audioBlobIDs = append(audioBlobIDs, id) } bytes, meta, genErr := s.adobe.GenerateVideo(ctx, token.Value, engine, prompt, aspectRatio, durationSeconds, resolution, referenceMode, upstreamModel, blobIDs, videoBlobIDs, audioBlobIDs, downloadResult) if genErr == nil { videoURL = strings.TrimSpace(stringValue(meta["video_url"])) } return bytes, genErr }, adobeErrClass, func(id string) (model.TokenAccount, bool) { return s.refreshAdobeToken(ctx, id) }, true) return data, videoURL, err } // leonardoMinCredits is the per-generation token cost (one Leonardo image = 30 // tokens). An account with fewer is treated as 限额 and skipped — it can't afford // a generation. Daily renewal (tokenRenewalDate) drives auto-recovery. const leonardoMinCredits = 30 func (s *V1Service) generateRunwayVideo(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1VideoRequest, aspectRatio string, durationSeconds int, downloadResult bool) ([]byte, string, error) { if s.runway == nil { return nil, "", errors.New("runway client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.runway.SetProxy(proxy) } } // Runway i2v strictly requires exactly one first-frame image. refs, err := decodeReferenceImages(in.ReferenceImages, 1) if err != nil { return nil, "", err } if len(refs) != 1 { return nil, "", errors.New("runway 图生视频需要且仅需 1 张首帧图") } frame := refs[0] items, err := s.tokens.ListByPool(ctx, "runway") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } // No pre-deduct (same policy as the image flow): skip only accounts we KNOW // are out of credits (cached remaining <= 0) — those are treated as dead. // Unknown balance gets the benefit of the doubt. if rem, ok := jsonMapInt(item.Meta, "cached_quota_remaining"); ok && rem <= 0 { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("runway", active) var lastErr error var videoURL string busy := 0 for _, token := range active { // Per-account concurrency gate (defaults to 1 for built-in pools). if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) teamID := "" if token.Meta != nil { teamID = strings.TrimSpace(stringValue(token.Meta["team_id"])) } d, meta, genErr := s.runway.GenerateVideo(ctx, token.Value, teamID, in.Prompt, aspectRatio, durationSeconds, frame, downloadResult) if genErr == nil { _, _ = s.tokens.Update(ctx, "runway", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) data = d videoURL = strings.TrimSpace(stringValue(meta["video_url"])) return true, false } lastErr = genErr switch { case errors.Is(genErr, runway.ErrAuth), errors.Is(genErr, runway.ErrQuotaExhausted): // 额度没了 / token 失效 → 当 401 判死(status=disabled, dead),换号。 s.markTokenFailure(ctx, "runway", token, "video", true, false) return false, true case errors.Is(genErr, runway.ErrTemporaryUpstream): // 上游临时错误 → 直接换下一个号。 return false, true default: // 参数级错误(如 prompt 未过审)→ 直接失败,不换号。 return false, false } }() if done { return data, videoURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // customAccountServes reports whether a custom (upstream) account is usable for a // given model id: active, not dead, has a base_url, and its meta.models list (csv // of model ids it serves) contains the id. An empty models list serves ALL ids. func customAccountServes(item model.TokenAccount, modelID string) bool { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { return false } if item.Meta == nil || strings.TrimSpace(stringValue(item.Meta["base_url"])) == "" { return false } list := strings.TrimSpace(stringValue(item.Meta["models"])) if list == "" { return true } for _, m := range strings.Split(list, ",") { if strings.EqualFold(strings.TrimSpace(m), modelID) { return true } } return false } // customActive returns the custom accounts that serve modelID, ordered by weight // (higher first; ties by id) so heavier upstreams are preferred. func (s *V1Service) customActive(ctx context.Context, modelID string) ([]model.TokenAccount, error) { items, err := s.tokens.ListByPool(ctx, "custom") if err != nil { return nil, err } var active []model.TokenAccount for _, item := range items { if customAccountServes(item, modelID) { active = append(active, item) } } s.rotateRoundRobin("custom", active) // weight priority + round-robin within ties return active, nil } // accountConcurrency is the per-account simultaneous-job cap. Custom accounts use // their configured Concurrency (default 1); built-in pools use the system value. func accountConcurrency(item model.TokenAccount) int { if item.Pool == "adobe" { if isFreeAccount(item.Meta) { return 1 // FREE 普号 / 降级号限制为 1 并发 } if item.Concurrency > 0 { return item.Concurrency } return 5 // VIP 会员号默认 5 并发 } if item.Concurrency > 0 { return item.Concurrency } if item.Pool == "grok" { return grokConcurrencyPerAccount // 10 } return 1 } // effectiveProvider routes a model to the "custom" upstream whenever a custom // account declares it serves that model id (id-based override of the model's // native provider) — so an upstream can take over any model by matching its id. // Otherwise the model's own provider is used. func (s *V1Service) effectiveProvider(ctx context.Context, modelItem *model.ModelConfig) string { if s.custom != nil { if active, err := s.customActive(ctx, modelItem.ID); err == nil && len(active) > 0 { return "custom" } } return modelItem.Provider } // generateCustomImage forwards an image generation to an OpenAI-compatible // upstream. The upstream (custom account) is matched by model id; calls go direct // (no proxy). Billing uses the local model price. func (s *V1Service) generateCustomImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.custom == nil { return nil, "", errors.New("custom client not configured") } refs, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } active, err := s.customActive(ctx, modelItem.ID) if err != nil { return nil, "", err } active = pinTestAccount(active, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } size := upstreamSize(aspectRatio, resolution) quality := upstreamQuality(resolution) var lastErr error busy := 0 for _, token := range active { if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte var imgURL string done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) baseURL := stringValue(token.Meta["base_url"]) d, u, genErr := s.custom.GenerateImage(ctx, baseURL, token.Value, modelItem.ID, in.Prompt, size, quality, refs, !urlOnly) if genErr == nil { _, _ = s.tokens.Update(ctx, "custom", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) data = d imgURL = u return true, false } lastErr = genErr switch { case errors.Is(genErr, custom.ErrAuth): s.markTokenFailure(ctx, "custom", token, "image", true, false) return false, true case errors.Is(genErr, custom.ErrQuotaExhausted): s.markTokenFailure(ctx, "custom", token, "image", false, true) return false, true case errors.Is(genErr, custom.ErrTemporaryUpstream): return false, true default: return false, false } }() if done { return data, imgURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // generateCustomVideo forwards a video generation to an OpenAI-compatible // (Sora-style) upstream, matched by model id. No proxy; local-price billing. func (s *V1Service) generateCustomVideo(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1VideoRequest, aspectRatio, resolution string, durationSeconds int, downloadResult bool) ([]byte, string, error) { if s.custom == nil { return nil, "", errors.New("custom client not configured") } active, err := s.customActive(ctx, modelItem.ID) if err != nil { return nil, "", err } active = pinTestAccount(active, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } size := upstreamVideoSize(aspectRatio, resolution) // Optional reference frames (image-to-video / first-last frames) — forwarded // to the upstream as multipart input_reference[] files. frames, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } var lastErr error var videoURL string busy := 0 for _, token := range active { if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) baseURL := stringValue(token.Meta["base_url"]) d, url, genErr := s.custom.GenerateVideo(ctx, baseURL, token.Value, modelItem.ID, in.Prompt, size, durationSeconds, frames, downloadResult) if genErr == nil { _, _ = s.tokens.Update(ctx, "custom", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) data = d videoURL = url return true, false } lastErr = genErr switch { case errors.Is(genErr, custom.ErrAuth): s.markTokenFailure(ctx, "custom", token, "video", true, false) return false, true case errors.Is(genErr, custom.ErrQuotaExhausted): s.markTokenFailure(ctx, "custom", token, "video", false, true) return false, true case errors.Is(genErr, custom.ErrTemporaryUpstream): return false, true default: return false, false } }() if done { return data, videoURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // upstreamSize maps our (ratio, resolution) to an OpenAI-style "WxH" size string // for the upstream. The pixel base scales with the tier (1K/2K/4K); the ratio // sets the shape. Upstreams that key off ratio (our own /v1) read it fine. func upstreamSize(aspectRatio, resolution string) string { base := 1024 switch strings.ToUpper(strings.TrimSpace(resolution)) { case "2K": base = 2048 case "4K": base = 4096 } w, h := 1, 1 parts := strings.Split(strings.ReplaceAll(strings.TrimSpace(aspectRatio), "x", ":"), ":") if len(parts) == 2 { if a, e1 := strconv.Atoi(strings.TrimSpace(parts[0])); e1 == nil && a > 0 { if b, e2 := strconv.Atoi(strings.TrimSpace(parts[1])); e2 == nil && b > 0 { w, h = a, b } } } if w >= h { return fmt.Sprintf("%dx%d", base, base*h/w) } return fmt.Sprintf("%dx%d", base*w/h, base) } // upstreamVideoSize maps our (ratio, resolution) to a "WxH" size for video // upstreams. Video "Np" tiers set the SHORT edge in pixels (like grok: // 720p 1:1 → 720x720, 720p 16:9 → 1280x720); 2K/4K fall back to the // long-edge mapping shared with images. func upstreamVideoSize(aspectRatio, resolution string) string { short := 0 switch res := strings.ToLower(strings.TrimSpace(resolution)); res { case "540p": short = 540 case "720p", "": short = 720 case "1080p": short = 1080 } if short == 0 { return upstreamSize(aspectRatio, resolution) } w, h := 1, 1 parts := strings.Split(strings.ReplaceAll(strings.TrimSpace(aspectRatio), "x", ":"), ":") if len(parts) == 2 { if a, e1 := strconv.Atoi(strings.TrimSpace(parts[0])); e1 == nil && a > 0 { if b, e2 := strconv.Atoi(strings.TrimSpace(parts[1])); e2 == nil && b > 0 { w, h = a, b } } } if w >= h { return fmt.Sprintf("%dx%d", short*w/h, short) } return fmt.Sprintf("%dx%d", short, short*h/w) } // upstreamQuality maps a resolution tier to the OpenAI quality enum. func upstreamQuality(resolution string) string { switch strings.ToUpper(strings.TrimSpace(resolution)) { case "2K": return "medium" case "4K": return "high" case "1K": return "low" } return "" } // generateGrokVideo runs grok's imagine video pipeline across the grok pool, // via Grok Console (console.x.ai) — the same sso account, but the clean JSON // media API instead of the anti-bot gated grok.com website flow. // 额度是本地写死的(每号 图 5 / 视频 2):视频计数归零的号不再调度,下单先预扣一个、 // 失败退回(并发不超扣),图/视频都归零直接判死;auth / 额度错误同样判死换号 // (grok sso 不续期,失效就失效)。 func (s *V1Service) generateGrokVideo(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1VideoRequest, aspectRatio, resolution string, durationSeconds int, downloadResult bool) ([]byte, string, error) { if s.grok == nil { return nil, "", errors.New("grok client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.grok.SetProxy(proxy) } } // Optional reference frames (image-to-video), up to the model's max. frames, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } items, err := s.tokens.ListByPool(ctx, "grok") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } if rem, ok := jsonMapInt(item.Meta, repo.GrokVideoQuotaKey); ok && rem <= 0 { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("grok", active) res := strings.TrimSpace(resolution) if res == "" { res = "720p" } var lastErr error var videoURL string busy := 0 for _, token := range active { // Per-account concurrency gate (defaults to 1 for built-in pools). if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) // 下单先预扣本地额度(并发时不会超扣),失败再退回。 allowed, reserveErr := s.tokens.ReserveGrokQuota(ctx, token.ID, "video") if reserveErr != nil || !allowed { return false, true } d, meta, genErr := s.grok.GenerateConsoleVideo(ctx, token.Value, in.Prompt, aspectRatio, res, durationSeconds, frames, downloadResult) if genErr == nil { _, _ = s.tokens.Update(ctx, "grok", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) // 图/视频都归零时账号直接判死。 _ = s.tokens.FinalizeGrokQuota(ctx, token.ID) data = d videoURL = strings.TrimSpace(stringValue(meta["video_url"])) return true, false } _ = s.tokens.RefundGrokQuota(ctx, token.ID, "video") lastErr = genErr switch { case errors.Is(genErr, grok.ErrAuth), errors.Is(genErr, grok.ErrQuotaExhausted): // 失效 / 额度没了 → 当 401 判死(不续期),换号。 s.markTokenFailure(ctx, "grok", token, "video", true, false) return false, true case errors.Is(genErr, grok.ErrTemporaryUpstream): return false, true default: return false, false } }() if done { return data, videoURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // generateGrokImage runs Grok Console's image pipeline (grok-imagine-image) // across the grok pool. 额度策略同视频路径,只是扣的是图片那份计数。带参考图时 // (最多 3 张,内联在请求里)自动走 /images/edits 的 quality 上游 — 图生图。 func (s *V1Service) generateGrokImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { // API-key (noStore) requests skip the download and return the upstream URL. urlOnly := noStore if s.grok == nil { return nil, "", errors.New("grok client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.grok.SetProxy(proxy) } } refs, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } items, err := s.tokens.ListByPool(ctx, "grok") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } if rem, ok := jsonMapInt(item.Meta, repo.GrokImageQuotaKey); ok && rem <= 0 { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("grok", active) var lastErr error busy := 0 for _, token := range active { // Per-account concurrency gate. if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte var artURL string done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) // 下单先预扣本地额度(并发时不会超扣),失败再退回。 allowed, reserveErr := s.tokens.ReserveGrokQuota(ctx, token.ID, "image") if reserveErr != nil || !allowed { return false, true } d, meta, genErr := s.grok.GenerateConsoleImage(ctx, token.Value, in.Prompt, aspectRatio, resolution, refs, urlOnly) if genErr == nil { _, _ = s.tokens.Update(ctx, "grok", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) // 图/视频都归零时账号直接判死。 _ = s.tokens.FinalizeGrokQuota(ctx, token.ID) data = d artURL = strings.TrimSpace(stringValue(meta["image_url"])) return true, false } _ = s.tokens.RefundGrokQuota(ctx, token.ID, "image") lastErr = genErr switch { case errors.Is(genErr, grok.ErrAuth), errors.Is(genErr, grok.ErrQuotaExhausted): // 失效 / 额度没了 → 当 401 判死(不续期),换号。 s.markTokenFailure(ctx, "grok", token, "image", true, false) return false, true case errors.Is(genErr, grok.ErrTemporaryUpstream): return false, true default: return false, false } }() if done { return data, artURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // generateRunwayImage runs the Runway gemini image pipeline (Nano Banana Pro or // Nano Banana 2, selected by the model id) across the runway pool. Unlike the // video path it does NOT pre-deduct credits: it simply round-robins the pool and // generates. Per ops decision an out-of-credits account is treated like a dead // 401 — marked dead (status=disabled) and skipped — because Runway credits don't // refill daily, so a "quota" mark (which the maintenance loop would revive) is // wrong. Reference images (up to the model's max) are uploaded per attempt. // noStore url-only mode (API-key requests without DeAI): skip the artifact // download and return the upstream image URL directly, no bytes. func (s *V1Service) generateRunwayImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { // API-key (noStore) requests don't support DeAI (only the web drawing board // does), so url-only mode == noStore — skip the download, return the URL. urlOnly := noStore if s.runway == nil { return nil, "", errors.New("runway client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.runway.SetProxy(proxy) } } refs, err := decodeReferenceImages(in.ReferenceImages, max(1, modelItem.MaxReferenceImages)) if err != nil { return nil, "", err } items, err := s.tokens.ListByPool(ctx, "runway") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } // No pre-deduct: skip only accounts we KNOW are out of credits // (cached remaining <= 0); they're treated as dead. Unknown balance gets // the benefit of the doubt — upstream rejects if it's truly empty. if rem, ok := jsonMapInt(item.Meta, "cached_quota_remaining"); ok && rem <= 0 { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("runway", active) imageSize := strings.TrimSpace(resolution) if imageSize == "" { imageSize = "1K" } var lastErr error busy := 0 for _, token := range active { // Per-account concurrency gate (defaults to 1 for built-in pools). if !s.acctAcquire(ctx, token.ID, eventID, accountConcurrency(token)) { busy++ continue } var data []byte var artURL string done, failover := func() (bool, bool) { defer s.acctRelease(ctx, token.ID, eventID) _ = s.events.SetAccount(ctx, eventID, token.ID, token.AccountEmail) _ = s.tokens.TouchLastUsed(ctx, token.ID) teamID := "" if token.Meta != nil { teamID = strings.TrimSpace(stringValue(token.Meta["team_id"])) } // downloadResult=false in url-only mode → skip the artifact download and // just return meta["image_url"]. d, meta, genErr := s.runway.GenerateImage(ctx, token.Value, teamID, modelItem.ID, in.Prompt, aspectRatio, imageSize, refs, !urlOnly) if genErr == nil { _, _ = s.tokens.Update(ctx, "runway", token.ID, map[string]any{ "last_used_at": time.Now(), "success_total": gorm.Expr("success_total + 1"), "fails": 0, }) data = d artURL = strings.TrimSpace(stringValue(meta["image_url"])) return true, false } lastErr = genErr switch { case errors.Is(genErr, runway.ErrAuth), errors.Is(genErr, runway.ErrQuotaExhausted): // 额度没了 / token 失效 → 当 401 判死(status=disabled, dead),换号。 s.markTokenFailure(ctx, "runway", token, "image", true, false) return false, true case errors.Is(genErr, runway.ErrTemporaryUpstream): // 上游临时错误 → 直接换下一个号。 return false, true default: // 参数级错误(如 prompt 未过审)→ 直接失败,不换号。 return false, false } }() if done { return data, artURL, nil } if failover { continue } return nil, "", lastErr } if lastErr == nil { if busy > 0 { return nil, "", ErrConcurrencyFull } lastErr = ErrProviderExecution } return nil, "", lastErr } // reconcileChatGPTQuota re-reads OpenAI's image_gen remaining right after a // successful generation and writes it back (negative / unknown clamp to 0), // flipping the account to 限额 when it hits 0 — so accounts limit one-by-one as // they're used, not all at once on a later batch probe. Runs while the // per-account concurrency gate is still held. Best-effort (never fails the render). func (s *V1Service) reconcileChatGPTQuota(ctx context.Context, tokenID, accessToken string) { if s.chatgpt == nil { return } data, err := s.chatgpt.FetchImageQuota(ctx, accessToken) if err != nil || boolValueWithDefault(data["auth_failed"], false) { return } rem, exhausted := chatgptRemaining(data) item, err := s.tokens.Get(ctx, "chatgpt", tokenID) if err != nil { return } meta := cloneJSONMap(item.Meta) meta["cached_quota_remaining"] = rem meta["cached_quota_at"] = int(time.Now().Unix()) patch := map[string]any{"meta": meta} if reset := strings.TrimSpace(stringValue(data["reset_after"])); reset != "" { patch["cached_quota_reset_after"] = reset } else if strings.TrimSpace(item.CachedQuotaResetAfter) == "" { patch["cached_quota_reset_after"] = leonardoResetAfter("") } if exhausted && item.Status == "active" { patch["status"] = "quota" } _, _ = s.tokens.Update(ctx, "chatgpt", tokenID, patch) } // chatgpt image URLs are auth-gated (files.oaiusercontent.com — a plain GET // 403s), so url-only mode returns the URL for the caller to proxy via // OpenImageContent using the generating account's token. func (s *V1Service) generateChatGPTImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.chatgpt == nil { return nil, "", errors.New("chatgpt client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.chatgpt.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "chatgpt") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status == "active" && !item.Dead && strings.TrimSpace(item.Value) != "" { active = append(active, item) } } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("chatgpt", active) refLimit := modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 1 } refs, err := decodeReferenceImages(in.ReferenceImages, refLimit) if err != nil { return nil, "", err } // Round-robin order; on a transient upstream error (e.g. "image generation // did not start (no async marker)") FAIL OVER to the next account // (tempFailover=true, capped at maxTempDeadAccounts) — never mark the // account dead. Auth/quota fail over immediately (see runPoolWithFailover). var imageURL string data, err := s.runPoolWithFailover(ctx, eventID, "chatgpt", active, "image", func(token model.TokenAccount) ([]byte, error) { d, meta, genErr := s.chatgpt.GenerateImage(ctx, token.Value, in.Prompt, modelItem.ID, aspectRatio, resolution, refs, !urlOnly) if genErr == nil { imageURL = strings.TrimSpace(stringValue(meta["image_url"])) // Sync the real OpenAI quota BEFORE the concurrency gate releases, so the // freshly-decremented remaining (and 限额 flip at 0) gates the next pick. s.reconcileChatGPTQuota(ctx, token.ID, token.Value) } return d, genErr }, func(e error) (bool, bool, bool, bool) { return errors.Is(e, chatgpt.ErrAuth), errors.Is(e, chatgpt.ErrQuotaExhausted), errors.Is(e, chatgpt.ErrTemporaryUpstream), false }, nil, true) // chatgpt token IS the credential — no cookie to refresh; switch accounts on transient errors return data, imageURL, err } // leonardoResetAfter returns when a Leonardo account's daily free tokens renew. // Leonardo resets at 08:00 Beijing == 00:00 UTC, so when the upstream gives no // explicit renewal time we deterministically use the next UTC midnight — this is // filled at import so 恢复时间 is always populated, not left blank. func leonardoResetAfter(availableUntil string) string { if v := strings.TrimSpace(availableUntil); v != "" { return v } return time.Unix((time.Now().Unix()/86400+1)*86400, 0).UTC().Format(time.RFC3339) } // leonardoDimensions maps the catalog's resolution+ratio to Leonardo pixel sizes. func leonardoDimensions(resolution, aspectRatio string) (int, int) { res := strings.ToUpper(strings.TrimSpace(resolution)) ar := strings.TrimSpace(aspectRatio) if res == "4K" { switch ar { case "2:3": return 2000, 3000 case "16:9": return 4096, 2304 case "4:3": return 4096, 3072 case "4:5": return 3264, 4080 case "9:16": return 2160, 3840 case "2:1": return 4096, 2048 default: // 1:1 return 4096, 4096 } } switch ar { // 2K (default) case "2:3": return 1664, 2496 case "16:9": return 2560, 1440 case "4:3": return 2304, 1728 case "4:5": return 2432, 3040 case "9:16": return 1440, 2560 case "2:1": return 3232, 1616 default: // 1:1 return 2048, 2048 } } func (s *V1Service) generateLeonardoImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.leonardo == nil { return nil, "", errors.New("leonardo client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.leonardo.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "leonardo") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } // Skip accounts under the per-generation floor (treated as 限额). Unknown // balance gets the benefit of the doubt (upstream rejects if truly empty). if rem, ok := jsonMapInt(item.Meta, "cached_quota_remaining"); ok && rem < leonardoMinCredits { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("leonardo", active) width, height := leonardoDimensions(resolution, aspectRatio) // The catalog model id is the upstream Leonardo model name (e.g. seedream-4.5). upstreamModel := strings.TrimSpace(modelItem.ID) // Optional image-to-image: decode the reference image once up front (Leonardo // seedream takes at most one). refLimit := modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 1 } refs, err := decodeReferenceImages(in.ReferenceImages, refLimit) if err != nil { return nil, "", err } // token.Value is the cookie; GenerateImage mints a fresh JWT each attempt (and // re-mints it internally when the bearer is rejected), so an auth failure means // the cookie itself no longer authenticates — no refresher (nil). var imageURL string data, err := s.runPoolWithFailover(ctx, eventID, "leonardo", active, "image", func(token model.TokenAccount) ([]byte, error) { // Atomically pre-deduct the per-generation cost so concurrent picks of the // same near-empty account can't over-commit it. A known-insufficient // balance surfaces as quota → the driver fails over to the next account. allowed, deducted, rerr := s.tokens.ReserveQuota(ctx, "leonardo", token.ID, leonardoMinCredits) if rerr != nil { return nil, fmt.Errorf("%w: reserve: %v", leonardo.ErrTemporaryUpstream, rerr) } if !allowed { return nil, leonardo.ErrQuotaExhausted } data, meta, genErr := s.leonardo.GenerateImage(ctx, token.Value, upstreamModel, in.Prompt, width, height, nil, refs, !urlOnly) cookie := s.leonardoPersistCookie(ctx, token.ID, token.Value) if genErr != nil { // Release the hold so a failed render doesn't burn credits. if deducted { _ = s.tokens.RefundQuota(ctx, "leonardo", token.ID, leonardoMinCredits) } return nil, genErr } imageURL = strings.TrimSpace(stringValue(meta["image_url"])) // Success → overwrite the held value with the REAL upstream balance and // sink to 限额 if below the floor (best-effort; never fails a done render). s.reconcileLeonardoCredits(ctx, token.ID, cookie) return data, nil }, func(e error) (bool, bool, bool, bool) { return errors.Is(e, leonardo.ErrAuth), errors.Is(e, leonardo.ErrQuotaExhausted), errors.Is(e, leonardo.ErrTemporaryUpstream), false }, nil, true) return data, imageURL, err } // leonardoPrivateSuffix 标记目录里 Leonardo 私有视频模型(public:false 生成)的 // 后缀,上游 slug 就是去掉它之后的 id。 const leonardoPrivateSuffix = "-不卡人脸" // leonardoVideoSpec 描述一个 Leonardo 视频模型的上游 slug、输出尺寸和参考资产 // 限制(各类上限 + 时长约束),校验在扣费前跑,免得非法参考白扣积分。 type leonardoVideoSpec struct { upstream string // long/short 是长边/短边像素,按比例组合成 16:9 或 9:16。 long int short int maxImages int maxAudios int maxVideos int // 视频参考单个时长区间与总时长上限(秒),0 表示不限。 videoMinSeconds float64 videoMaxSeconds float64 videoTotalSeconds float64 // 音频参考总时长上限(秒),0 表示不限。 audioTotalSeconds float64 } var leonardoVideoSpecs = map[string]leonardoVideoSpec{ "seedance-2.0" + leonardoPrivateSuffix: { upstream: "seedance-2.0", long: 1280, short: 720, maxImages: 4, maxAudios: 1, maxVideos: 3, videoMinSeconds: 3, videoMaxSeconds: 10, videoTotalSeconds: 15, }, "seedance-2.0-fast" + leonardoPrivateSuffix: { upstream: "seedance-2.0-fast", long: 1280, short: 720, maxImages: 4, maxAudios: 1, maxVideos: 3, videoMinSeconds: 3, videoMaxSeconds: 10, videoTotalSeconds: 15, }, "minimax-h3": { upstream: "hailuo-03", long: 2560, short: 1440, maxImages: 5, maxAudios: 3, maxVideos: 0, audioTotalSeconds: 15, }, } func leonardoVideoSpecOf(modelID string) leonardoVideoSpec { id := strings.TrimSpace(modelID) if spec, ok := leonardoVideoSpecs[id]; ok { return spec } // 目录里新增的同族模型退化成 seedance 规格,上游 slug 取去掉私有后缀的 id。 return leonardoVideoSpec{ upstream: strings.TrimSuffix(id, leonardoPrivateSuffix), long: 1280, short: 720, maxImages: 4, maxAudios: 1, maxVideos: 3, videoMinSeconds: 3, videoMaxSeconds: 10, videoTotalSeconds: 15, } } // dimensions 按比例返回像素尺寸,只支持 16:9 / 9:16。 func (spec leonardoVideoSpec) dimensions(aspectRatio string) (int, int) { if strings.TrimSpace(aspectRatio) == "9:16" { return spec.short, spec.long } return spec.long, spec.short } // classifyLeonardoVideoRefs decodes the mixed reference payload (画图台把图片/ // 音频/视频一起塞进 reference_images)并按类型分流,同时按模型规格校验各类上限和 // 时长。 func classifyLeonardoVideoRefs(inputs []string, spec leonardoVideoSpec) (leonardo.VideoAssets, error) { var refs leonardo.VideoAssets decoded, err := decodeReferenceImages(inputs, spec.maxImages+spec.maxAudios+spec.maxVideos) if err != nil { return refs, err } var videoTotal, audioTotal float64 for _, ref := range decoded { switch detectMediaType(ref) { case "video": if spec.maxVideos == 0 { return refs, errors.New("该模型不支持视频参考") } secs := leonardo.MediaDurationSeconds(ref) if secs <= 0 { return refs, errors.New("无法解析视频参考的时长,请换一个 mp4") } if secs < spec.videoMinSeconds || secs > spec.videoMaxSeconds { return refs, fmt.Errorf("单个视频参考需 %.0f-%.0f 秒,当前 %.1f 秒", spec.videoMinSeconds, spec.videoMaxSeconds, secs) } videoTotal += secs if spec.videoTotalSeconds > 0 && videoTotal >= spec.videoTotalSeconds { return refs, fmt.Errorf("视频参考总时长 %.1f 秒,需短于 %.0f 秒", videoTotal, spec.videoTotalSeconds) } refs.Videos = append(refs.Videos, ref) case "audio": if spec.maxAudios == 0 { return refs, errors.New("该模型不支持音频参考") } secs := leonardo.MediaDurationSeconds(ref) if spec.audioTotalSeconds > 0 { if secs <= 0 { return refs, errors.New("无法解析音频参考的时长,请换一个 mp3") } audioTotal += secs if audioTotal > spec.audioTotalSeconds { return refs, fmt.Errorf("音频参考总时长 %.1f 秒,最多 %.0f 秒", audioTotal, spec.audioTotalSeconds) } } refs.Audios = append(refs.Audios, ref) default: refs.Images = append(refs.Images, ref) } } switch { case len(refs.Images) > spec.maxImages: return refs, fmt.Errorf("最多 %d 张参考图", spec.maxImages) case len(refs.Audios) > spec.maxAudios: return refs, fmt.Errorf("最多 %d 段音频参考", spec.maxAudios) case len(refs.Videos) > spec.maxVideos: return refs, fmt.Errorf("最多 %d 段视频参考", spec.maxVideos) } return refs, nil } // generateLeonardoVideo renders a Leonardo video across the leonardo pool. Mirrors // generateLeonardoImage (cookie → JWT, quota reserve, cookie 轮换持久化), only the // upstream call differs: 私有生成 + 三类参考资产 + motionMP4URL。 func (s *V1Service) generateLeonardoVideo(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1VideoRequest, aspectRatio string, durationSeconds int, downloadResult bool) ([]byte, string, error) { if s.leonardo == nil { return nil, "", errors.New("leonardo client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.leonardo.SetProxy(proxy) } } spec := leonardoVideoSpecOf(modelItem.ID) refs, err := classifyLeonardoVideoRefs(in.ReferenceImages, spec) if err != nil { return nil, "", err } items, err := s.tokens.ListByPool(ctx, "leonardo") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { if item.Status != "active" || item.Dead || strings.TrimSpace(item.Value) == "" { continue } if rem, ok := jsonMapInt(item.Meta, "cached_quota_remaining"); ok && rem < leonardoMinCredits { continue } active = append(active, item) } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("leonardo", active) width, height := spec.dimensions(aspectRatio) var videoURL string data, err := s.runPoolWithFailover(ctx, eventID, "leonardo", active, "video", func(token model.TokenAccount) ([]byte, error) { allowed, deducted, rerr := s.tokens.ReserveQuota(ctx, "leonardo", token.ID, leonardoMinCredits) if rerr != nil { return nil, fmt.Errorf("%w: reserve: %v", leonardo.ErrTemporaryUpstream, rerr) } if !allowed { return nil, leonardo.ErrQuotaExhausted } data, meta, genErr := s.leonardo.GenerateVideo(ctx, token.Value, spec.upstream, in.Prompt, width, height, durationSeconds, refs, downloadResult) cookie := s.leonardoPersistCookie(ctx, token.ID, token.Value) if genErr != nil { if deducted { _ = s.tokens.RefundQuota(ctx, "leonardo", token.ID, leonardoMinCredits) } return nil, genErr } videoURL = strings.TrimSpace(stringValue(meta["video_url"])) s.reconcileLeonardoCredits(ctx, token.ID, cookie) return data, nil }, func(e error) (bool, bool, bool, bool) { return errors.Is(e, leonardo.ErrAuth), errors.Is(e, leonardo.ErrQuotaExhausted), errors.Is(e, leonardo.ErrTemporaryUpstream), false }, nil, true) return data, videoURL, err } // leonardoPersistCookie writes back the account cookie when Leonardo rotated its // better-auth session_data cache (that cache is what actually authenticates // get-session, so a stale copy would eventually look like a dead account). // Returns the value now stored. func (s *V1Service) leonardoPersistCookie(ctx context.Context, tokenID, cookie string) string { if s.leonardo == nil { return cookie } fresh, ok := s.leonardo.RotatedCookie(cookie) if !ok || strings.TrimSpace(fresh) == "" { return cookie } _, _ = s.tokens.SwapValue(ctx, "leonardo", tokenID, cookie, fresh) return fresh } // reconcileLeonardoCredits re-fetches an account's real token balance after a // render and writes it back, flipping the account to 限额 when below the per-gen // floor. Stores the daily renewal time so RecoverQuota can auto-recover it. func (s *V1Service) reconcileLeonardoCredits(ctx context.Context, tokenID, cookie string) { if s.leonardo == nil { return } data, err := s.leonardo.FetchCreditsBalance(ctx, cookie) s.leonardoPersistCookie(ctx, tokenID, cookie) if err != nil { return } rem, ok := data["remaining"].(int) if !ok { return } item, err := s.tokens.Get(ctx, "leonardo", tokenID) if err != nil { return } meta := cloneJSONMap(item.Meta) meta["cached_quota_remaining"] = rem meta["cached_quota_at"] = int(time.Now().Unix()) patch := map[string]any{"meta": meta} patch["cached_quota_reset_after"] = leonardoResetAfter(stringValue(data["available_until"])) if rem < leonardoMinCredits && item.Status == "active" { patch["status"] = "quota" } _, _ = s.tokens.Update(ctx, "leonardo", tokenID, patch) } // kreaRefreshAndPersist ensures the account's Krea cookie has a valid access token // (refreshing via the rotating refresh_token when expired) and persists the new // cookie — the refresh_token is single-use, so the rotated value MUST be saved. func kreaRefreshAndPersist(ctx context.Context, client *krea.Client, tokens *repo.TokenRepository, tokenID, cookie string) (string, error) { if client == nil { return cookie, nil } fresh, changed, err := client.RefreshIfNeeded(ctx, cookie) if err != nil { return "", err } if changed && tokenID != "" { _, _ = tokens.Update(ctx, "krea", tokenID, map[string]any{"value": fresh}) } return fresh, nil } // kreaDimensions maps the catalog's resolution+ratio to Krea pixel sizes. func kreaDimensions(resolution, aspectRatio string) (int, int) { res := strings.ToUpper(strings.TrimSpace(resolution)) ar := strings.TrimSpace(aspectRatio) if res == "2K" { switch ar { case "4:3": return 2048, 1536 case "3:4": return 1536, 2048 case "16:9": return 2048, 1152 case "9:16": return 1152, 2048 default: // 1:1 return 2048, 2048 } } switch ar { // 1K (default) case "4:3": return 1024, 768 case "3:4": return 768, 1024 case "16:9": return 1024, 576 case "9:16": return 576, 1024 default: // 1:1 return 1024, 1024 } } func (s *V1Service) generateKreaImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.krea == nil { return nil, "", errors.New("krea client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.krea.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "krea") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { // No numeric floor — Krea signals 限额 with a 402 at generation time, which // the failover driver turns into mark-quota + next account. if item.Status == "active" && !item.Dead && strings.TrimSpace(item.Value) != "" { active = append(active, item) } } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("krea", active) width, height := kreaDimensions(resolution, aspectRatio) refLimit := modelItem.MaxReferenceImages if refLimit <= 0 { refLimit = 1 } refs, err := decodeReferenceImages(in.ReferenceImages, refLimit) if err != nil { return nil, "", err } var imageURL string data, err := s.runPoolWithFailover(ctx, eventID, "krea", active, "image", func(token model.TokenAccount) ([]byte, error) { // Refresh the (rotating) Supabase token if expired and persist the new // cookie, then generate with the fresh cookie. cookie, rerr := kreaRefreshAndPersist(ctx, s.krea, s.tokens, token.ID, token.Value) if rerr != nil { return nil, rerr } data, meta, genErr := s.krea.GenerateImage(ctx, cookie, in.Prompt, width, height, refs, !urlOnly) if genErr == nil { imageURL = strings.TrimSpace(stringValue(meta["image_url"])) } return data, genErr }, func(e error) (bool, bool, bool, bool) { return errors.Is(e, krea.ErrAuth), errors.Is(e, krea.ErrQuotaExhausted), errors.Is(e, krea.ErrTemporaryUpstream), false }, nil, true) return data, imageURL, err } // imagineRefreshAndPersist ensures the account's Imagine credential has a valid // access token (refreshing via the rotating refreshToken when expired) and // persists the new credential — both tokens rotate, so the value MUST be saved. func imagineRefreshAndPersist(ctx context.Context, client *imagine.Client, tokens *repo.TokenRepository, tokenID, cred string) (string, error) { if client == nil { return cred, nil } fresh, changed, err := client.RefreshIfNeeded(ctx, cred) if err != nil { return "", err } if changed && tokenID != "" { _, _ = tokens.Update(ctx, "imagine", tokenID, map[string]any{"value": fresh}) } return fresh, nil } // imagineStyle maps the catalog model id to its upstream style_id + resolution. func imagineStyle(modelID string) (int, string) { if strings.TrimSpace(modelID) == "imagine-1.5pro" { return 41004, "4K" } return 41001, "2K" } func (s *V1Service) generateImagineImage(ctx context.Context, eventID string, modelItem *model.ModelConfig, in V1ImageRequest, aspectRatio, resolution string, noStore bool) ([]byte, string, error) { urlOnly := noStore if s.imagine == nil { return nil, "", errors.New("imagine client not configured") } if s.settings != nil { if proxy, err := s.settings.GetValue(ctx, "proxy.url"); err == nil { s.imagine.SetProxy(proxy) } } items, err := s.tokens.ListByPool(ctx, "imagine") if err != nil { return nil, "", err } var active []model.TokenAccount for _, item := range items { // No numeric floor — Imagine signals 限额 with a 402 at generation time, // which the failover driver turns into mark-quota + next account. if item.Status == "active" && !item.Dead && strings.TrimSpace(item.Value) != "" { active = append(active, item) } } active = pinTestAccount(items, active, in.AccountID) if len(active) == 0 { return nil, "", ErrNoProviderAccount } s.rotateRoundRobin("imagine", active) // Each model supports exactly one resolution (2K / 4K) — force it per model. styleID, res := imagineStyle(modelItem.ID) var imageURL string data, err := s.runPoolWithFailover(ctx, eventID, "imagine", active, "image", func(token model.TokenAccount) ([]byte, error) { // Refresh the (rotating) access token if expired and persist the new // credential, then generate with the fresh token. cred, rerr := imagineRefreshAndPersist(ctx, s.imagine, s.tokens, token.ID, token.Value) if rerr != nil { return nil, rerr } data, meta, genErr := s.imagine.GenerateImage(ctx, cred, styleID, res, aspectRatio, in.Prompt, !urlOnly) if genErr != nil { return nil, genErr } imageURL = strings.TrimSpace(stringValue(meta["image_url"])) return data, nil }, func(e error) (bool, bool, bool, bool) { return errors.Is(e, imagine.ErrAuth), errors.Is(e, imagine.ErrQuotaExhausted), errors.Is(e, imagine.ErrTemporaryUpstream), false }, nil, true) return data, imageURL, err } func (s *V1Service) refundIfNeeded(ctx context.Context, principal *APIPrincipal, eventID string, price float64) error { if principal == nil || principal.User == nil || price <= 0 { return nil } // Exactly-once: claim the refund via the event's `refunded` flag. If another // path (e.g. the abandoned-purge sweep) already refunded, MarkRefunded // returns false and we skip — no double refund. claimed, err := s.events.MarkRefunded(ctx, eventID) if err != nil { return err } if !claimed { return nil } updated, err := s.users.RefundCredits(ctx, principal.User.ID, price) if err == nil { principal.User = updated } return err } func (s *V1Service) maybeGrantInviteReward(ctx context.Context, principal *APIPrincipal) error { if principal == nil || principal.User == nil || s.settings == nil { return nil } enabledRaw, err := s.settings.GetValue(ctx, "credits.invite_enabled") if err != nil { return err } if !parseBoolSetting(enabledRaw, true) { return nil } rewardRaw, err := s.settings.GetValue(ctx, "credits.invite_reward") if err != nil { return err } _, err = s.users.GrantInviteReward(ctx, principal.User.ID, parseIntSetting(rewardRaw, 3)) return err } // ensureReferenceSizes rejects any reference image over the byte cap BEFORE // charging, so an oversized image fails fast (no charge, no pending-log churn) // across every entry path — session /generate, API-key /v1, and admin /test. // decodeReferenceImages re-checks at decode time as a backstop; this mirrors its // base64 length pre-check (decoded ≈ len(b64)*3/4). func ensureReferenceSizes(inputs []string) error { for _, raw := range inputs { v := strings.TrimSpace(raw) if v == "" { continue } if (len(v)*3)/4 > maxReferenceImageBytes { return ErrReferenceTooLarge } } return nil } func decodeReferenceImages(inputs []string, limit int) ([][]byte, error) { if limit <= 0 { limit = 1 } if len(inputs) > limit { return nil, errors.New("too many reference images") } out := make([][]byte, 0, len(inputs)) for _, raw := range inputs { v := strings.TrimSpace(raw) if v == "" { continue } // Only raw base64 is accepted (no "data:...;base64," URL prefix). A data // URL now fails to decode rather than being silently stripped. // decoded size ≈ len(b64) * 3 / 4 — reject oversized payloads up front, // before allocating the decoded buffer. if (len(v)*3)/4 > maxReferenceImageBytes { return nil, ErrReferenceTooLarge } data, err := base64.StdEncoding.DecodeString(v) if err != nil { data, err = base64.RawStdEncoding.DecodeString(v) if err != nil { return nil, errors.New("invalid reference image encoding") } } if len(data) == 0 { return nil, errors.New("empty reference image") } if len(data) > maxReferenceImageBytes { return nil, ErrReferenceTooLarge } out = append(out, data) } return out, nil } // detectMediaType inspects the first bytes of a decoded reference to classify it // as "video", "audio", or "image". Used to route refs to the correct upload MIME // and the correct referenceBlobs usage for seedance. func detectMediaType(data []byte) string { n := len(data) if n < 8 { return "image" } // MP4 / ISOBMFF if n >= 12 && string(data[4:8]) == "ftyp" { return "video" } // WebM if n >= 4 && data[0] == 0x1A && data[1] == 0x45 && data[2] == 0xDF && data[3] == 0xA3 { return "video" } // MP3: ID3 header or sync word 0xFFFx if n >= 3 && data[0] == 0x49 && data[1] == 0x44 && data[2] == 0x33 { return "audio" } if n >= 2 && data[0] == 0xFF && (data[1]&0xE0) == 0xE0 { return "audio" } // WAV if n >= 4 && data[0] == 0x52 && data[1] == 0x49 && data[2] == 0x46 && data[3] == 0x46 { return "audio" } // OGG if n >= 4 && data[0] == 0x4F && data[1] == 0x67 && data[2] == 0x67 && data[3] == 0x53 { return "audio" } return "image" } func parseImageSize(size, aspectRatio, resolution string) (string, string) { ar := strings.TrimSpace(strings.ReplaceAll(aspectRatio, "x", ":")) rs := strings.TrimSpace(resolution) if size != "" && strings.Contains(strings.ToLower(size), "x") { var w, h int _, _ = fmt.Sscanf(strings.ToLower(size), "%dx%d", &w, &h) if w > 0 && h > 0 { if ar == "" { ar = guessRatio(w, h) } if rs == "" { maxEdge := w if h > maxEdge { maxEdge = h } switch { case maxEdge >= 3500: rs = "4K" case maxEdge >= 1800: rs = "2K" default: rs = "1K" } } } } if ar == "" { ar = "1:1" } if rs == "" { rs = "2K" } return ar, rs } // snapRatio returns the entry in supported closest in value to ar ("W:H"). // ar is returned as-is when it's already supported, unparsable, or the model // has no ratio list. func snapRatio(ar string, supported []string) string { parse := func(s string) (float64, bool) { var w, h int if _, err := fmt.Sscanf(strings.TrimSpace(s), "%d:%d", &w, &h); err != nil || w <= 0 || h <= 0 { return 0, false } return float64(w) / float64(h), true } v, ok := parse(ar) if !ok || len(supported) == 0 { return ar } best, bestDelta := "", 0.0 for _, s := range supported { if strings.TrimSpace(strings.ReplaceAll(s, "x", ":")) == ar { return ar } sv, sok := parse(strings.ReplaceAll(s, "x", ":")) if !sok { continue } if d := absFloat(v - sv); best == "" || d < bestDelta { best, bestDelta = strings.TrimSpace(strings.ReplaceAll(s, "x", ":")), d } } if best == "" { return ar } return best } func guessRatio(w, h int) string { type candidate struct { W int H int } // The 17 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. 9:21 is intentionally absent — // no image provider accepts it (Runway 400s on it). snapRatio then clamps // the guess to the target model's own supported list. candidates := []candidate{ {1, 1}, {5, 4}, {4, 3}, {3, 2}, {16, 9}, {2, 1}, {21, 9}, {3, 1}, {4, 1}, {8, 1}, // 横 {4, 5}, {3, 4}, {2, 3}, {9, 16}, {1, 3}, {1, 4}, {1, 8}, // 竖 } best := candidates[0] bestDelta := absFloat(float64(w)/float64(h) - float64(best.W)/float64(best.H)) for _, item := range candidates[1:] { delta := absFloat(float64(w)/float64(h) - float64(item.W)/float64(item.H)) if delta < bestDelta { best = item bestDelta = delta } } return fmt.Sprintf("%d:%d", best.W, best.H) } // firstPricedResolution returns the model's lowest priced image tier (1K/2K/4K // order), or "" if none is priced. Used to rescue a request whose resolution // the model doesn't support. // deaiEnabled reports whether the 去AI特征 feature is switched on in system // settings (default off). When off, an incoming deai flag is ignored entirely. func (s *V1Service) deaiEnabled(ctx context.Context) bool { if s.settings == nil { return false } raw, err := s.settings.GetValue(ctx, "deai.enabled") if err != nil { return false } return parseBoolSetting(raw, false) } // deaiSurcharge returns the 去AI特征 surcharge (积分) for an image resolution // tier, from site settings (defaults: 1K=1, 2K=2, 4K=3). func (s *V1Service) deaiSurcharge(ctx context.Context, resolution string) float64 { key, def := "deai.price_1k", 1 switch strings.ToUpper(strings.TrimSpace(resolution)) { case "2K": key, def = "deai.price_2k", 2 case "4K": key, def = "deai.price_4k", 3 } if s.settings == nil { return float64(def) } raw, err := s.settings.GetValue(ctx, key) if err != nil { return float64(def) } n := parseIntSetting(raw, def) if n < 0 { n = 0 } return float64(n) } func firstPricedResolution(item *model.ModelConfig) string { if item == nil { return "" } for _, r := range []string{"1K", "2K", "4K"} { if _, ok := jsonMapFloat(item.Prices, r); ok { return r } } return "" } // resolutionForQuality maps OpenAI's `quality` to one of the model's priced // resolution tiers: low→1K, medium→2K, high→4K, auto/blank→the model's lowest // priced tier. The desired tier is clamped to the nearest tier the model // actually prices (e.g. seedream is 2K/4K only: low→2K, high→4K). func resolutionForQuality(item *model.ModelConfig, quality string) string { order := []string{"1K", "2K", "4K"} var priced []string for _, r := range order { if _, ok := jsonMapFloat(item.Prices, r); ok { priced = append(priced, r) } } if len(priced) == 0 { return firstPricedResolution(item) } rank := map[string]int{"low": 0, "medium": 1, "high": 2} want, ok := rank[strings.ToLower(strings.TrimSpace(quality))] if !ok { return priced[0] // auto / unknown → model default (lowest priced) } idxOf := func(r string) int { for i, v := range order { if v == r { return i } } return 0 } best, bestDist := priced[0], 99 for _, r := range priced { d := idxOf(r) - want if d < 0 { d = -d } if d < bestDist { best, bestDist = r, d } } return best } // modelPrice returns the charge for (kind, resolution, duration). The set of // supported tiers is always driven by the NORMAL prices; `agent` only overrides // the amount with the agent price when one is set for that tier (else it falls // back to the normal price). func modelPrice(item *model.ModelConfig, kind, resolution, duration string, agent bool) (float64, bool) { if item == nil { return 0, false } // tierPrice: normal price gates support; agent price (if present) overrides. tierPrice := func(normal, agentMap map[string]any, key string) (float64, bool) { nv, ok := jsonMapFloat(normal, key) if !ok { return 0, false } if agent { if av, aok := jsonMapFloat(agentMap, key); aok { return av, true } } return nv, true } if kind == "video" { rv, rok := tierPrice(item.Prices, item.PricesAgent, resolution) var dv float64 var dok bool if pps, hasPerSec := jsonMapFloat(item.DurationPrices, "per_second"); hasPerSec { secs := parseDurationSeconds(duration) dv = pps * float64(secs) dok = true if agent { if av, aok := jsonMapFloat(item.DurationPricesAgent, "per_second"); aok { dv = av * float64(secs) } } } else { dv, dok = tierPrice(item.DurationPrices, item.DurationPricesAgent, duration) } if !rok || !dok { return 0, false } return rv + dv, true } return tierPrice(item.Prices, item.PricesAgent, resolution) } func jsonMapFloat(m map[string]any, key string) (float64, bool) { if m == nil { return 0, false } v, ok := m[key] if !ok || v == nil { return 0, false } switch x := v.(type) { case float64: return x, true case float32: return float64(x), true case int: return float64(x), true case int64: return float64(x), true case json.Number: // datatypes.JSONMap.Scan decodes with UseNumber(), so values loaded from // the DB arrive as json.Number — NOT float64. Without this case every // price read back from Postgres looked "unpriced". if f, err := x.Float64(); err == nil { return f, true } case string: var out float64 if _, err := fmt.Sscanf(strings.TrimSpace(x), "%f", &out); err == nil { return out, true } } return 0, false } func sanitizeOwnerName(v string) string { v = strings.TrimSpace(v) if v == "" { return "" } var b strings.Builder for _, r := range v { switch { case r >= 'a' && r <= 'z': b.WriteRune(r) case r >= 'A' && r <= 'Z': b.WriteRune(r) case r >= '0' && r <= '9': b.WriteRune(r) } } return b.String() } func parseDurationSeconds(raw string) int { raw = strings.ToLower(strings.TrimSpace(raw)) raw = strings.TrimSuffix(raw, "s") var n int if _, err := fmt.Sscanf(raw, "%d", &n); err != nil || n <= 0 { return 5 } return n } func resolveAdobeVideoEngine(modelID string) (string, string) { switch strings.ToLower(strings.TrimSpace(modelID)) { case "gemini-veo3.1-fast", "gemini-veo31", "firefly-veo31": return "veo31-fast", "" case "gemini-veo3.1": return "veo31-standard", "" case "seedance-2.0-fast": return "seedance-2.0-fast", "" case "seedance-2.0": return "seedance-2.0", "" case "firefly-ray": return "luma", "" case "firefly-video": return "firefly-video", "" default: return "sora2", "" } } func absFloat(v float64) float64 { if v < 0 { return -v } return v } func max(a, b int) int { if a > b { return a } return b } func principalCredits(principal *APIPrincipal) float64 { if principal == nil || principal.User == nil { return 0 } return principal.User.Credits } // markTokenFailure applies Python mark_bad semantics for a failed generation // attempt against a pool token. It always bumps fail counters; the status side // effects depend on the failure reason and the provider/pool. // // - quota: status="quota"; when no cached_quota_reset_after is present, set // quota_recover_at to next UTC midnight so the maintenance loop can revive it. // - auth on chatgpt: status="disabled" + dead=true (the access token IS the // credential; a 401 means it's dead). // - auth on adobe: NOT disabled/dead — the access token auto-refreshes from the // cookie, so rotate for this request and let the refresh loop mint a new one. // - other (non-auth/non-quota): NEITHER pool is auto-disabled — accounts stay // active/green and fails is tracked only for rotation ordering. func (s *V1Service) markTokenFailure(ctx context.Context, pool string, token model.TokenAccount, kind string, isAuth, isQuota bool) { patch := map[string]any{ "last_used_at": time.Now(), "fail_total": gorm.Expr("fail_total + 1"), "fails": gorm.Expr("fails + 1"), } switch { case isQuota: patch["status"] = "quota" if strings.TrimSpace(token.CachedQuotaResetAfter) == "" { recoverAt := time.Unix((time.Now().Unix()/86400+1)*86400, 0).UTC() patch["quota_recover_at"] = &recoverAt } case isAuth: // Adobe auth failures are NOT disabling: the access token refreshes from // the cookie. chatgpt/runway/leonardo auth means the stored credential is // dead — a raw JWT (chatgpt/runway) or a cookie whose session no longer // authenticates (leonardo) — there's nothing left to refresh from. // grok is intentionally excluded: a grok sso can momentarily 401 while // still valid (upstream blip / proxy / anti-bot), so an auth failure just // fails over for this request without permanently killing the account. disable := pool == "chatgpt" || pool == "runway" || pool == "leonardo" || pool == "krea" || pool == "imagine" if disable && pool == "leonardo" { // 两道保险:先重新 get-session 复核(单次失败常是 bearer 轮换竞态),复核 // 也不过就只记一次连续失败,连续到上限才判死。 if s.leonardoCookieAlive(ctx, token) { log.Printf("leonardo %s: auth failure on %s but cookie still authenticates — kept active", token.ID, kind) disable = false } else { meta, strikes, kill := leonardoAuthStrike(&token, "auth failure on "+kind) patch["meta"] = meta disable = kill if kill { log.Printf("account leonardo/%s disabled after %d consecutive auth failures: %s", token.ID, strikes, kind) } else { log.Printf("leonardo %s: auth failure %d/%d on %s — kept active", token.ID, strikes, leonardoAuthStrikeLimit, kind) } } } if disable { patch["status"] = "disabled" patch["dead"] = true if pool != "leonardo" { log.Printf("account %s/%s disabled: auth failure on %s", pool, token.ID, kind) } } default: // Neither pool is auto-disabled on generic (non-auth / non-quota) failures // — the account usually still works, so it stays active (green). fails is // only tracked for rotation ordering. (A chatgpt *auth* failure still marks // the token dead in the isAuth case above; that is a genuinely dead token.) } _, _ = s.tokens.Update(ctx, pool, token.ID, patch) } // leonardoCookieAlive re-checks a Leonardo cookie after an auth failure by // force-minting a session (bypassing the cached bearer). Only a cookie that // still fails to authenticate counts as dead; a temporary upstream answer // (403/429 人机校验) also keeps the account alive. func (s *V1Service) leonardoCookieAlive(ctx context.Context, token model.TokenAccount) bool { if s.leonardo == nil || strings.TrimSpace(token.Value) == "" { return false } probeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) defer cancel() sess, err := s.leonardo.ProbeSession(probeCtx, token.Value) if err == nil && sess != nil && strings.TrimSpace(sess.AccessToken) != "" { s.leonardoPersistCookie(probeCtx, token.ID, token.Value) return true } return !errors.Is(err, leonardo.ErrAuth) } // markTokenDead disables an account and marks it dead on a fatal upstream error // (a non-overload temporary Adobe failure that ops policy treats as account death). func (s *V1Service) markTokenDead(ctx context.Context, pool string, token model.TokenAccount, kind string) { _, _ = s.tokens.Update(ctx, pool, token.ID, map[string]any{ "last_used_at": time.Now(), "fail_total": gorm.Expr("fail_total + 1"), "fails": gorm.Expr("fails + 1"), "status": "disabled", "dead": true, }) } // nextCursor returns the pool's current round-robin position and atomically // advances it by one. Concurrent callers each get a distinct value, so parallel // picks land on different accounts instead of racing onto the same one. The // counter is in-memory (per process): it resets on restart, which only shifts // the rotation's starting point — distribution stays even. func (s *V1Service) nextCursor(pool string) uint64 { v, _ := s.tokenCursors.LoadOrStore(pool, new(uint64)) return atomic.AddUint64(v.(*uint64), 1) - 1 } // rotateRoundRobin orders the active accounts by a stable key (ID) and rotates // the slice in place so iteration begins at the pool's current cursor position, // then advances the cursor. This is strict round-robin: account selection // cycles in fixed order regardless of fails or last_used. The fall-through // retry chain is preserved — on failure the caller's loop simply continues to // the next account in rotation order. // pinTestAccount narrows account selection to the single account requested by // an admin 账号生图测试. The pinned account is taken from the pool's full list // (bypassing active/dead/limited filters) so a limited or disabled account can // still be probed. Returns nil when the account isn't in this pool. func pinTestAccount(items, active []model.TokenAccount, accountID string) []model.TokenAccount { id := strings.TrimSpace(accountID) if id == "" { return active } for _, item := range items { if item.ID == id && strings.TrimSpace(item.Value) != "" { return []model.TokenAccount{item} } } return nil } func (s *V1Service) rotateRoundRobin(pool string, items []model.TokenAccount) { if len(items) <= 1 { return } // Weight = priority: higher-weight accounts come first, so the scheduler tries // them before lower-weight ones (and only falls through when they're at their // concurrency cap). Within the SAME weight all accounts are equal, so they're // rotated by the pool cursor for even distribution. sort.SliceStable(items, func(i, j int) bool { if items[i].Weight != items[j].Weight { return items[i].Weight > items[j].Weight } return items[i].ID < items[j].ID }) start := int(s.nextCursor(pool)) for i := 0; i < len(items); { j := i + 1 for j < len(items) && items[j].Weight == items[i].Weight { j++ } if g := j - i; g > 1 { off := start % g if off != 0 { grp := items[i:j] rot := make([]model.TokenAccount, 0, g) rot = append(rot, grp[off:]...) rot = append(rot, grp[:off]...) copy(grp, rot) } } i = j } } // freeOnly1KModelID is the one free-allowed model 普号 may only serve at 1K // (香蕉2 的 2K/4K 需要会员号) — see freeAccountsAllowed. const freeOnly1KModelID = "nano-banana-2" func isSeedanceModel(modelID string) bool { return modelID == "seedance-2.0-fast" || modelID == "seedance-2.0" } // freeAccountsAllowed reports whether 普号(free) may serve this request: the model // must be marked free_allowed. 香蕉2 另外只允许 1K 档,它的 2K/4K 只走会员号。 func freeAccountsAllowed(modelItem *model.ModelConfig, resolution string) bool { if modelItem == nil || !modelItem.FreeAllowed { return false } if modelItem.ID == freeOnly1KModelID { return strings.EqualFold(strings.TrimSpace(resolution), "1K") } return true } func isFreeAccount(meta map[string]interface{}) bool { if meta == nil { return false } plan := strings.ToLower(strings.TrimSpace(stringValue(meta["plan"]))) return plan == "free" } // planUnknown 报告账号的会员身份还没探测出来(meta.plan 缺失或为空)。这类号 // 既不能当普号也不能当会员号用:当会员号派出去会在需要会员的模型上撞 403 // user_not_entitled。 func planUnknown(meta map[string]interface{}) bool { if meta == nil { return true } return strings.TrimSpace(stringValue(meta["plan"])) == "" } // markPlanUnknownDead 把选号时遇到的 plan 未探测账号置为死号,等重新探测到 // plan 后再由额度刷新恢复。 func (s *V1Service) markPlanUnknownDead(ctx context.Context, pool, id string) { s.tokens.Update(ctx, pool, id, map[string]any{"status": "disabled", "dead": true}) } // prioritizeSubAccounts 对非 Seedance 模型按 普号 → 子号 → 母号 的顺序排序: // 先消耗普号,普号不可用再用低积分子号,最后才动 vip 母号。 func prioritizeSubAccounts(active []model.TokenAccount) []model.TokenAccount { var frees, subs, mothers []model.TokenAccount for _, a := range active { switch { case isFreeAccount(a.Meta): frees = append(frees, a) case isLowCredits(a.Meta): subs = append(subs, a) default: mothers = append(mothers, a) } } return append(append(frees, subs...), mothers...) } // isVipMotherAccount 正向识别 VIP 母号:plan 已知且非 free,且 is_sub_account // 显式为 false。只看身份不看积分余额(低积分母号也可用);plan 未探测或 // is_sub_account 缺失的账号返回 false,等刷新补齐后才可被 Seedance 调度。 func isVipMotherAccount(meta map[string]interface{}) bool { if meta == nil { return false } plan := strings.ToLower(strings.TrimSpace(stringValue(meta["plan"]))) if plan == "" || plan == "free" { return false } v, ok := meta["is_sub_account"] if !ok { return false } switch val := v.(type) { case bool: return !val case float64: return val == 0 } return false } func isLowCredits(meta map[string]interface{}) bool { if meta == nil { return false } if v, ok := meta["is_sub_account"]; ok { switch val := v.(type) { case bool: return val case float64: return val != 0 } } // 兼容存量账号:is_sub_account 字段不存在时,用积分余额判断(>0 且 ≤4000 视为子号) if rem, ok := jsonMapInt(meta, "cached_quota_remaining"); ok { return rem > 0 && rem <= 4000 } return false }