feat!: 账号池改渠道网关,对外协议换成 vivid /v1

- 新增 channels/channel_models:上游 URL + AES-GCM 加密 key +
  并发/权重/本地统计,模型经渠道 /v1/models 导入绑定
- internal/upstream:vivid 协议客户端(NDJSON 流 + jobId 断流重连 +
  uploads 预签名 + models + balance + 错误分类)
- service/v1 重写:权重分组调度、Redis 并发槽 conc:c:、故障转移,
  站内/测试/对外统一走渠道执行器
- 对外端点换成 vivid 四件套:POST /v1/generate(NDJSON+jobId 重连)、
  POST /v1/uploads(SigV4 预签名 PUT)、GET /v1/models、GET /v1/balance,
  错误包 {"error":{"code","message"}}
- API key 新发改 vk_+48hex;删除全部内置 provider 与 token 池
- 管理端:渠道 CRUD/拉模型/导入/余额端点,前端账号页换渠道页,
  概览/文档/测试弹窗适配

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
2026-10-16 21:14:18 +08:00
co-authored by Devin
parent ecdbc0b281
commit b4efe6b192
75 changed files with 3598 additions and 23848 deletions
+38 -70
View File
@@ -19,20 +19,20 @@ type AdminReadService struct {
models *repo.ModelRepository
events *repo.EventRepository
settings *repo.SiteSettingRepository
tokens *repo.TokenRepository
channels *repo.ChannelRepository
cdks *repo.CDKRepository
store *storage.Client
showcase *repo.ShowcaseRepository
}
func NewAdminReadService(cfg *config.Config, users *repo.UserRepository, models *repo.ModelRepository, events *repo.EventRepository, settings *repo.SiteSettingRepository, tokens *repo.TokenRepository, cdks *repo.CDKRepository, store *storage.Client, showcase *repo.ShowcaseRepository) *AdminReadService {
func NewAdminReadService(cfg *config.Config, users *repo.UserRepository, models *repo.ModelRepository, events *repo.EventRepository, settings *repo.SiteSettingRepository, channels *repo.ChannelRepository, cdks *repo.CDKRepository, store *storage.Client, showcase *repo.ShowcaseRepository) *AdminReadService {
return &AdminReadService{
cfg: cfg,
users: users,
models: models,
events: events,
settings: settings,
tokens: tokens,
channels: channels,
cdks: cdks,
store: store,
showcase: showcase,
@@ -104,6 +104,10 @@ func (s *AdminReadService) ModelsView(ctx context.Context) ([]map[string]any, er
"max_reference_images": item.MaxReferenceImages,
"max_reference_videos": item.MaxReferenceVideos,
"max_reference_audios": item.MaxReferenceAudios,
"duration_min": item.DurationMin,
"duration_max": item.DurationMax,
"max_ref_video_seconds": item.MaxRefVideoSeconds,
"max_ref_audio_seconds": item.MaxRefAudioSeconds,
"reference_mode": item.ReferenceMode,
"weight": item.Weight,
"generation_count": item.GenerationCount,
@@ -196,24 +200,24 @@ func (s *AdminReadService) UserNameMap(ctx context.Context) (map[string]string,
return out, nil
}
// AccountNameMap builds a token-account id -> display label lookup (account
// email, else display name, else id) used to annotate log rows with which
// provider account fulfilled each generation (event_logs.account_id).
// AccountNameMap builds a channel id -> display name lookup used to annotate
// log rows with which channel fulfilled each generation (event_logs.account_id
// now stores the channel id; account_email carries a denormalized name stamp).
func (s *AdminReadService) AccountNameMap(ctx context.Context) (map[string]string, error) {
accounts, err := s.tokens.List(ctx)
if s.channels == nil {
return map[string]string{}, nil
}
channels, err := s.channels.List(ctx)
if err != nil {
return nil, err
}
out := make(map[string]string, len(accounts))
for _, a := range accounts {
label := strings.TrimSpace(a.AccountEmail)
out := make(map[string]string, len(channels))
for _, ch := range channels {
label := strings.TrimSpace(ch.Name)
if label == "" {
label = strings.TrimSpace(a.AccountDisplayName)
label = ch.ID
}
if label == "" {
label = a.ID
}
out[a.ID] = label
out[ch.ID] = label
}
return out, nil
}
@@ -422,66 +426,30 @@ func (s *AdminReadService) Invites(ctx context.Context) ([]repo.InviteRecord, *r
return s.users.AllInvites(ctx, reward)
}
// Providers is now a channel summary (the account pools are gone): one row per
// channel with its bound-model count and local usage stats.
func (s *AdminReadService) Providers(ctx context.Context) ([]map[string]any, error) {
models, err := s.models.List(ctx)
if s.channels == nil {
return []map[string]any{}, nil
}
channels, err := s.channels.List(ctx)
if err != nil {
return nil, err
}
tokens, err := s.tokens.List(ctx)
if err != nil {
return nil, err
}
modelCounts := map[string]int{}
for _, item := range models {
modelCounts[item.Provider]++
}
type aggregate struct {
active int
disabled int
quota int
}
tokenCounts := map[string]*aggregate{}
for _, item := range tokens {
if _, ok := tokenCounts[item.Pool]; !ok {
tokenCounts[item.Pool] = &aggregate{}
}
switch item.Status {
case "active":
tokenCounts[item.Pool].active++
case "quota":
tokenCounts[item.Pool].quota++
default:
tokenCounts[item.Pool].disabled++
}
}
providers := []struct {
Name string
Pool string
Type string
}{
{Name: "chatgpt", Pool: "chatgpt", Type: "openai"},
{Name: "adobe", Pool: "adobe", Type: "adobe"},
{Name: "runway", Pool: "runway", Type: "runway"},
{Name: "leonardo", Pool: "leonardo", Type: "leonardo"},
{Name: "krea", Pool: "krea", Type: "krea"},
{Name: "imagine", Pool: "imagine", Type: "imagine"},
{Name: "grok", Pool: "grok", Type: "grok"},
}
out := make([]map[string]any, 0, len(providers))
for _, item := range providers {
count := tokenCounts[item.Pool]
if count == nil {
count = &aggregate{}
}
out := make([]map[string]any, 0, len(channels))
for _, ch := range channels {
modelIDs, _ := s.channels.ModelIDsByChannel(ctx, ch.ID)
out = append(out, map[string]any{
"name": item.Name,
"token_pool": item.Pool,
"type": item.Type,
"model_count": modelCounts[item.Name],
"tokens_total": count.active + count.disabled + count.quota,
"tokens_active": count.active,
"tokens_disabled": count.disabled,
"tokens_quota": count.quota,
"id": ch.ID,
"name": ch.Name,
"base_url": ch.BaseURL,
"enabled": ch.Enabled,
"concurrency": ch.Concurrency,
"weight": ch.Weight,
"model_count": len(modelIDs),
"success_total": ch.SuccessTotal,
"used_credits": ch.UsedCredits,
"type": "vivid",
})
}
return out, nil
+23 -50
View File
@@ -28,18 +28,18 @@ type AdminWriteService struct {
models *repo.ModelRepository
events *repo.EventRepository
apiKeys *repo.APIKeyRepository
tokens *repo.TokenRepository
channels *repo.ChannelRepository
orders *repo.OrderRepository
}
func NewAdminWriteService(users *repo.UserRepository, showcase *repo.ShowcaseRepository, models *repo.ModelRepository, events *repo.EventRepository, apiKeys *repo.APIKeyRepository, tokens *repo.TokenRepository, orders *repo.OrderRepository) *AdminWriteService {
func NewAdminWriteService(users *repo.UserRepository, showcase *repo.ShowcaseRepository, models *repo.ModelRepository, events *repo.EventRepository, apiKeys *repo.APIKeyRepository, channels *repo.ChannelRepository, orders *repo.OrderRepository) *AdminWriteService {
return &AdminWriteService{
users: users,
showcase: showcase,
models: models,
events: events,
apiKeys: apiKeys,
tokens: tokens,
channels: channels,
orders: orders,
}
}
@@ -371,7 +371,7 @@ func (s *AdminWriteService) CreateModel(ctx context.Context, body map[string]any
return nil, errors.New("type must be image or video")
}
if provider == "" {
return nil, errors.New("provider required")
provider = "channel"
}
if err := s.validateModelNameSpace(ctx, "", modelID, alias); err != nil {
return nil, err
@@ -436,10 +436,9 @@ func (s *AdminWriteService) UpdateModel(ctx context.Context, modelID string, bod
}
if _, ok := body["provider"]; ok {
provider := strings.TrimSpace(stringValue(body["provider"]))
if provider == "" {
return nil, errors.New("provider required")
if provider != "" {
patch["provider"] = provider
}
patch["provider"] = provider
}
// Only touch `enabled` when the caller explicitly sends a non-null value;
// mirrors Python models_store.update ("enabled" in fields and is not None).
@@ -488,6 +487,18 @@ func (s *AdminWriteService) UpdateModel(ctx context.Context, modelID string, bod
if _, ok := body["reference_mode"]; ok {
patch["reference_mode"] = defaultString(strings.TrimSpace(stringValue(body["reference_mode"])), "none")
}
if _, ok := body["duration_min"]; ok {
patch["duration_min"] = intValue(body["duration_min"])
}
if _, ok := body["duration_max"]; ok {
patch["duration_max"] = intValue(body["duration_max"])
}
if _, ok := body["max_ref_video_seconds"]; ok {
patch["max_ref_video_seconds"] = intValue(body["max_ref_video_seconds"])
}
if _, ok := body["max_ref_audio_seconds"]; ok {
patch["max_ref_audio_seconds"] = intValue(body["max_ref_audio_seconds"])
}
if _, ok := body["weight"]; ok {
patch["weight"] = intValue(body["weight"])
}
@@ -532,52 +543,14 @@ func (s *AdminWriteService) DeleteModel(ctx context.Context, modelID string) err
if rows == 0 {
return ErrNotFound
}
// Clean up: strip this model id from every custom upstream account's
// supported-models list so no orphan reference is left behind.
s.removeModelFromUpstreams(ctx, modelID)
// Clean up: drop every channel binding pointing at the deleted model so no
// orphan reference is left behind.
if s.channels != nil {
_ = s.channels.DeleteBindingsByModel(ctx, modelID)
}
return nil
}
// removeModelFromUpstreams drops modelID from the CSV in each custom account's
// Meta["models"]. Best-effort — a failure here doesn't undo the model delete.
func (s *AdminWriteService) removeModelFromUpstreams(ctx context.Context, modelID string) {
if s.tokens == nil {
return
}
items, err := s.tokens.ListByPool(ctx, "custom")
if err != nil {
return
}
for _, it := range items {
raw, _ := it.Meta["models"].(string)
if strings.TrimSpace(raw) == "" {
continue
}
kept := make([]string, 0, len(strings.Split(raw, ",")))
changed := false
for _, p := range strings.Split(raw, ",") {
p = strings.TrimSpace(p)
if p == "" {
continue
}
if p == modelID {
changed = true
continue
}
kept = append(kept, p)
}
if !changed {
continue
}
meta := datatypes.JSONMap{}
for k, v := range it.Meta {
meta[k] = v
}
meta["models"] = strings.Join(kept, ",")
_, _ = s.tokens.Update(ctx, "custom", it.ID, map[string]any{"meta": meta})
}
}
func (s *AdminWriteService) ClearLogs(ctx context.Context) (int64, error) {
return s.events.DeleteAll(ctx)
}
+9 -1
View File
@@ -2,6 +2,7 @@ package service
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"errors"
@@ -109,8 +110,15 @@ func (s *APIKeyService) DeleteOne(ctx context.Context, userID, keyID string) err
return s.keys.DeleteByID(ctx, userID, keyID)
}
// generatePlainAPIKey mints a vivid-shaped key: vk_ + 48 lowercase hex chars
// (upstream channel api_keys use the same shape, so one key format rules
// everything — no per-direction prefixes).
func generatePlainAPIKey() (string, error) {
return "sk-" + randomUpper(38), nil
b := make([]byte, 24)
if _, err := rand.Read(b); err != nil {
return "", err
}
return "vk_" + hex.EncodeToString(b), nil
}
func previewAPIKey(plain string) string {
+450
View File
@@ -0,0 +1,450 @@
package service
import (
"context"
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"strings"
"sync"
"time"
"backend/internal/model"
"backend/internal/repo"
"backend/internal/upstream"
"gorm.io/datatypes"
"gorm.io/gorm"
)
// ChannelService manages upstream vivid-protocol channels: CRUD, upstream key
// encryption at rest, model fetching/import/binding, and live balance probes.
// The actual request dispatch lives on V1Service (it owns concurrency + events).
type ChannelService struct {
channels *repo.ChannelRepository
models *repo.ModelRepository
up *upstream.Client
secret []byte
}
func NewChannelService(channels *repo.ChannelRepository, models *repo.ModelRepository, up *upstream.Client, secret string) *ChannelService {
sum := sha256.Sum256([]byte("image2api-channel-keys|" + secret))
return &ChannelService{channels: channels, models: models, up: up, secret: sum[:]}
}
// ChannelView is the admin list row: local stats + bound model names + a live
// upstream balance probe (nil when the upstream doesn't implement /v1/balance
// or the probe failed).
type ChannelView struct {
ID string `json:"id"`
Name string `json:"name"`
BaseURL string `json:"base_url"`
Enabled bool `json:"enabled"`
Concurrency int `json:"concurrency"`
Weight int `json:"weight"`
Models []string `json:"models"`
SuccessTotal int64 `json:"success_total"`
UsedCredits float64 `json:"used_credits"`
UpstreamBalance *int64 `json:"upstream_balance"`
LastUsedAt *int64 `json:"last_used_at"`
CreatedAt int64 `json:"created_at"`
UpdatedAt int64 `json:"updated_at"`
}
func (s *ChannelService) List(ctx context.Context) ([]ChannelView, error) {
items, err := s.channels.List(ctx)
if err != nil {
return nil, err
}
out := make([]ChannelView, len(items))
var wg sync.WaitGroup
for i, ch := range items {
out[i] = ChannelView{
ID: ch.ID,
Name: ch.Name,
BaseURL: ch.BaseURL,
Enabled: ch.Enabled,
Concurrency: ch.Concurrency,
Weight: ch.Weight,
SuccessTotal: ch.SuccessTotal,
UsedCredits: ch.UsedCredits,
CreatedAt: ch.CreatedAt.Unix(),
UpdatedAt: ch.UpdatedAt.Unix(),
}
if ch.LastUsedAt != nil {
t := ch.LastUsedAt.Unix()
out[i].LastUsedAt = &t
}
if ids, err := s.channels.ModelIDsByChannel(ctx, ch.ID); err == nil {
names := make([]string, 0, len(ids))
for _, id := range ids {
if m, err := s.models.Get(ctx, id); err == nil {
names = append(names, m.EffectiveName())
} else {
names = append(names, id)
}
}
out[i].Models = names
}
if key := s.decrypt(ch.APIKeyEnc); key != "" {
wg.Add(1)
go func(i int, ch model.Channel, key string) {
defer wg.Done()
bctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
defer cancel()
if bal, err := s.up.Balance(bctx, ch.BaseURL, key); err == nil {
out[i].UpstreamBalance = bal
}
}(i, ch, key)
}
}
wg.Wait()
return out, nil
}
// SaveChannelDto is the admin create/update payload. ApiKey empty or the mask
// placeholder keeps the existing key on update; required on create.
type SaveChannelDto struct {
ID string `json:"id"`
Name string `json:"name"`
BaseURL string `json:"base_url"`
APIKey string `json:"api_key"`
Enabled bool `json:"enabled"`
Concurrency int `json:"concurrency"`
Weight int `json:"weight"`
}
func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Channel, error) {
name := strings.TrimSpace(d.Name)
base := strings.TrimRight(strings.TrimSpace(d.BaseURL), "/")
if base == "" {
return nil, errors.New("base_url required")
}
if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
return nil, errors.New("base_url must start with http:// or https://")
}
weight := d.Weight
if weight < 1 {
weight = 1
}
if d.ID == "" {
if strings.TrimSpace(d.APIKey) == "" {
return nil, errors.New("api_key required")
}
enc, err := s.encrypt(d.APIKey)
if err != nil {
return nil, err
}
ch := &model.Channel{
ID: "ch-" + randomUpper(10),
Name: name,
BaseURL: base,
APIKeyEnc: enc,
Enabled: d.Enabled,
Concurrency: d.Concurrency,
Weight: weight,
}
if ch.Name == "" {
ch.Name = base
}
return ch, s.channels.Create(ctx, ch)
}
patch := map[string]any{
"name": name,
"base_url": base,
"enabled": d.Enabled,
"concurrency": d.Concurrency,
"weight": weight,
}
if key := strings.TrimSpace(d.APIKey); key != "" && !strings.Contains(key, "•") && !strings.Contains(key, "*") {
enc, err := s.encrypt(key)
if err != nil {
return nil, err
}
patch["api_key_enc"] = enc
}
return s.channels.Update(ctx, d.ID, patch)
}
// Delete removes a channel, its bindings, and any local model that ends up
// bound to no channel at all (same semantics as vivid).
func (s *ChannelService) Delete(ctx context.Context, id string) error {
modelIDs, _ := s.channels.ModelIDsByChannel(ctx, id)
if err := s.channels.DeleteBindingsByChannel(ctx, id); err != nil {
return err
}
res := 0
if n, err := s.channels.Delete(ctx, id); err != nil {
return err
} else {
res = int(n)
}
if res == 0 {
return gorm.ErrRecordNotFound
}
for _, mid := range modelIDs {
rest, _ := s.channels.ChannelIDsByModel(ctx, mid)
if len(rest) == 0 {
_, _ = s.models.Delete(ctx, mid)
}
}
return nil
}
// FetchModelsDto resolves credentials for a model fetch: Id uses the stored
// channel's key; otherwise BaseURL+ApiKey preview an unsaved channel.
type FetchModelsDto struct {
ID string `json:"id"`
BaseURL string `json:"base_url"`
APIKey string `json:"api_key"`
}
// FetchModels lists the upstream's sellable models and flags which are already
// imported locally.
func (s *ChannelService) FetchModels(ctx context.Context, d FetchModelsDto) ([]upstream.RemoteModel, error) {
base, key, err := s.resolveCreds(ctx, d.ID, d.BaseURL, d.APIKey)
if err != nil {
return nil, err
}
items, err := s.up.Models(ctx, base, key)
if err != nil {
return nil, err
}
return items, nil
}
// ChannelModelSpec is one model the admin ticks for import (same shape as the
// upstream /v1/models item plus an optional local alias).
type ChannelModelSpec struct {
ID string `json:"id"`
Alias string `json:"alias"`
Kind string `json:"kind"`
Qualities []string `json:"qualities"`
Duration *upstream.RemoteDuration `json:"duration,omitempty"`
RefImages int64 `json:"refImages"`
RefVideos int64 `json:"refVideos"`
RefAudios int64 `json:"refAudios"`
RefVideoSeconds int64 `json:"refVideoSeconds,omitempty"`
RefAudioSeconds int64 `json:"refAudioSeconds,omitempty"`
}
func (s *ChannelService) resolveCreds(ctx context.Context, id, base, key string) (string, string, error) {
if strings.TrimSpace(id) != "" {
ch, err := s.channels.Get(ctx, id)
if err != nil {
return "", "", err
}
return strings.TrimRight(ch.BaseURL, "/"), s.decrypt(ch.APIKeyEnc), nil
}
base = strings.TrimRight(strings.TrimSpace(base), "/")
if base == "" || strings.TrimSpace(key) == "" {
return "", "", errors.New("id or base_url+api_key required")
}
return base, strings.TrimSpace(key), nil
}
// ImportModels creates/updates local model rows for the ticked specs and binds
// them to the channel. Pricing stays whatever the local row already has
// (empty for new rows — admin prices them in 模型管理).
func (s *ChannelService) ImportModels(ctx context.Context, channelID string, specs []ChannelModelSpec) error {
ch, err := s.channels.Get(ctx, channelID)
if err != nil {
return err
}
for _, spec := range specs {
upID := strings.TrimSpace(spec.ID)
if upID == "" {
continue
}
localID := strings.TrimSpace(spec.Alias)
if localID == "" {
localID = upID
}
existing, gerr := s.models.Get(ctx, localID)
if gerr != nil && !errors.Is(gerr, gorm.ErrRecordNotFound) {
existing = nil
}
var local *model.ModelConfig
if existing != nil && existing.ID == localID {
local = existing
}
patch := modelPatchFromSpec(spec, local)
if local == nil {
local = &model.ModelConfig{
ID: localID,
Name: localID,
Type: spec.Kind,
Provider: "channel",
Enabled: true,
Ratios: jsonArray([]string{"16:9"}),
Prices: datatypes.JSONMap{},
CreatedAt: time.Now(),
}
applySpec(local, patch)
if err := s.models.Create(ctx, local); err != nil {
return fmt.Errorf("model %s: %w", localID, err)
}
} else {
if _, err := s.models.Update(ctx, localID, patch); err != nil {
return fmt.Errorf("model %s: %w", localID, err)
}
}
if err := s.channels.Bind(ctx, ch.ID, localID, upID); err != nil {
return err
}
}
return nil
}
// modelPatchFromSpec maps an upstream model spec onto model_configs columns.
// Prices/enabled/alias are intentionally untouched on update so admin pricing
// survives re-import.
func modelPatchFromSpec(spec ChannelModelSpec, existing *model.ModelConfig) map[string]any {
patch := map[string]any{
"type": spec.Kind,
"resolutions": jsonArray(spec.Qualities),
"max_reference_images": spec.RefImages,
"max_reference_videos": spec.RefVideos,
"max_reference_audios": spec.RefAudios,
"max_ref_video_seconds": spec.RefVideoSeconds,
"max_ref_audio_seconds": spec.RefAudioSeconds,
}
if spec.RefImages+spec.RefVideos+spec.RefAudios > 0 {
patch["reference_mode"] = "asset"
patch["image_to_image"] = spec.RefImages > 0
}
if d := spec.Duration; d != nil {
switch d.Mode {
case "fixed":
patch["durations"] = jsonArray([]string{fmt.Sprintf("%ds", d.Seconds)})
patch["duration_min"], patch["duration_max"] = int64(0), int64(0)
case "options":
opts := make([]string, 0, len(d.Options))
for _, s := range d.Options {
opts = append(opts, fmt.Sprintf("%ds", s))
}
patch["durations"] = jsonArray(opts)
patch["duration_min"], patch["duration_max"] = int64(0), int64(0)
case "range":
patch["duration_min"], patch["duration_max"] = d.Min, d.Max
}
}
return patch
}
func applySpec(m *model.ModelConfig, patch map[string]any) {
if v, ok := patch["resolutions"]; ok {
m.Resolutions = datatypes.JSON(mustJSON(v))
}
if v, ok := patch["durations"]; ok {
m.Durations = datatypes.JSON(mustJSON(v))
}
if v, ok := patch["duration_min"]; ok {
m.DurationMin = int(v.(int64))
}
if v, ok := patch["duration_max"]; ok {
m.DurationMax = int(v.(int64))
}
m.MaxReferenceImages = int(specInt(patch, "max_reference_images"))
m.MaxReferenceVideos = int(specInt(patch, "max_reference_videos"))
m.MaxReferenceAudios = int(specInt(patch, "max_reference_audios"))
m.MaxRefVideoSeconds = int(specInt(patch, "max_ref_video_seconds"))
m.MaxRefAudioSeconds = int(specInt(patch, "max_ref_audio_seconds"))
if v, ok := patch["reference_mode"].(string); ok {
m.ReferenceMode = v
}
if v, ok := patch["image_to_image"].(bool); ok {
m.ImageToImage = v
}
}
func specInt(patch map[string]any, key string) int64 {
switch v := patch[key].(type) {
case int64:
return v
case int:
return int64(v)
default:
return 0
}
}
func mustJSON(v any) []byte {
b, _ := json.Marshal(v)
return b
}
// ChannelsForModel returns enabled channels bound to the model with their
// upstream model names — dispatch order is decided by the caller.
func (s *ChannelService) ChannelsForModel(ctx context.Context, modelID string) ([]repo.ChannelWithUpstream, error) {
return s.channels.ForModel(ctx, modelID)
}
// ChannelsForModelAnyState includes disabled channels (admin pin tests).
func (s *ChannelService) ChannelsForModelAnyState(ctx context.Context, modelID string) ([]repo.ChannelWithUpstream, error) {
return s.channels.ForModelAnyState(ctx, modelID)
}
// Decrypt exposes the upstream key for dispatch.
func (s *ChannelService) Decrypt(ch model.Channel) string { return s.decrypt(ch.APIKeyEnc) }
// Upstream returns the shared protocol client (dispatch needs it for
// UploadRef/Generate/Download; admin handlers for balance probes).
func (s *ChannelService) Upstream() *upstream.Client { return s.up }
// BumpSuccess forwards local stats updates to the repo.
func (s *ChannelService) BumpSuccess(ctx context.Context, channelID string, credits float64) {
_ = s.channels.BumpSuccess(ctx, channelID, credits)
}
// --- AES-GCM key encryption ---
const encPrefix = "enc:v1:"
func (s *ChannelService) encrypt(plain string) (string, error) {
block, err := aes.NewCipher(s.secret)
if err != nil {
return "", err
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return "", err
}
nonce := make([]byte, gcm.NonceSize())
if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
return "", err
}
return encPrefix + base64.StdEncoding.EncodeToString(gcm.Seal(nonce, nonce, []byte(plain), nil)), nil
}
func (s *ChannelService) decrypt(stored string) string {
if !strings.HasPrefix(stored, encPrefix) {
return stored // plaintext rows (e.g. seeded by hand)
}
raw, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(stored, encPrefix))
if err != nil {
return ""
}
block, err := aes.NewCipher(s.secret)
if err != nil {
return ""
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return ""
}
if len(raw) < gcm.NonceSize() {
return ""
}
plain, err := gcm.Open(nil, raw[:gcm.NonceSize()], raw[gcm.NonceSize():], nil)
if err != nil {
return ""
}
return string(plain)
}
+34 -158
View File
@@ -5,25 +5,19 @@ import (
"log"
"strconv"
"strings"
"sync"
"time"
"backend/internal/model"
"backend/internal/repo"
"backend/internal/storage"
)
// MaintenanceService runs the periodic self-healing sweep that the Python
// original did via a 60s daemon thread plus read-time lazy cleanup. Without it
// the Go token pool only ever loses capacity: tokens never re-activate after a
// quota reset, cookies never auto-renew, stale pending events permanently block
// a user's generation gate, and old media/logs accumulate unbounded.
// MaintenanceService runs the periodic self-healing sweep. Without it stale
// pending events permanently block a user's generation gate, and old media /
// logs / temp objects accumulate unbounded. Token-pool and cookie-refresh
// upkeep is gone — channels are stateless upstream credentials.
type MaintenanceService struct {
tokens *repo.TokenRepository
tokenSvc *TokenService
events *repo.EventRepository
users *repo.UserRepository
refresh *RefreshProfileService
settings *repo.SiteSettingRepository
store *storage.Client
inflight *InflightRegistry
@@ -36,20 +30,17 @@ type MaintenanceService struct {
lastMediaPrune time.Time
}
func NewMaintenanceService(tokens *repo.TokenRepository, tokenSvc *TokenService, events *repo.EventRepository, users *repo.UserRepository, refresh *RefreshProfileService, settings *repo.SiteSettingRepository, store *storage.Client, inflight *InflightRegistry, showcase *repo.ShowcaseRepository, orders *repo.OrderRepository) *MaintenanceService {
func NewMaintenanceService(events *repo.EventRepository, users *repo.UserRepository, settings *repo.SiteSettingRepository, store *storage.Client, inflight *InflightRegistry, showcase *repo.ShowcaseRepository, orders *repo.OrderRepository) *MaintenanceService {
return &MaintenanceService{
tokens: tokens,
tokenSvc: tokenSvc,
events: events,
users: users,
refresh: refresh,
settings: settings,
store: store,
inflight: inflight,
showcase: showcase,
orders: orders,
interval: 60 * time.Second,
stalePending: 600 * time.Second,
events: events,
users: users,
settings: settings,
store: store,
inflight: inflight,
showcase: showcase,
orders: orders,
interval: 60 * time.Second,
stalePending: 600 * time.Second,
// 视频比图片慢得多,单条执行预算就是 30 分钟(videoGenBudget),用图片的
// 10 分钟去扫会在它出片前判死并退款。
stalePendingVid: 30 * time.Minute,
@@ -63,10 +54,6 @@ func NewMaintenanceService(tokens *repo.TokenRepository, tokenSvc *TokenService,
func (m *MaintenanceService) Run(ctx context.Context) {
ticker := time.NewTicker(m.interval)
defer ticker.Stop()
// The leonardo session keep-alive gets its own loop: one tick() can take tens
// of minutes (219 adobe cookie profiles alone), which would stretch a 5-minute
// keep-alive to the tick's real duration.
go m.runLeonardoKeepalive(ctx)
m.tick(ctx)
for {
select {
@@ -78,55 +65,6 @@ func (m *MaintenanceService) Run(ctx context.Context) {
}
}
// runLeonardoKeepalive re-checks every minute which leonardo accounts are due for
// a session renewal (the 5-minute due-ness itself is read per account from the DB,
// so a restart can't skip one).
func (m *MaintenanceService) runLeonardoKeepalive(ctx context.Context) {
if m.tokenSvc == nil {
return
}
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for {
m.tokenSvc.RefreshLeonardoSessions(ctx)
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}
// syncRecoveredQuota re-probes each just-recovered account so its displayed
// balance reflects the post-reset value (these providers only sync quota when
// accessed). krea additionally needs /app (Activate) to actually grant the daily
// free balance before billing-data reports it. Bounded concurrency avoids a
// thundering herd at the daily reset.
func (m *MaintenanceService) syncRecoveredQuota(accs []model.TokenAccount) {
sem := make(chan struct{}, 4)
var wg sync.WaitGroup
for _, acc := range accs {
switch acc.Pool {
case "chatgpt", "leonardo", "krea", "imagine":
default:
continue
}
wg.Add(1)
sem <- struct{}{}
go func(a model.TokenAccount) {
defer wg.Done()
defer func() { <-sem }()
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
defer cancel()
if a.Pool == "krea" && m.tokenSvc.krea != nil {
m.tokenSvc.krea.Activate(ctx, a.Value)
}
_, _ = m.tokenSvc.Quota(ctx, a.Pool, a.ID)
}(acc)
}
wg.Wait()
}
func (m *MaintenanceService) tick(ctx context.Context) {
// 0. Auto-cancel unpaid recharge orders past their 30-min TTL.
if m.orders != nil {
@@ -137,74 +75,7 @@ func (m *MaintenanceService) tick(ctx context.Context) {
}
}
// 1. Re-activate quota-exhausted tokens whose reset time has passed, then
// auto-sync their real balance — these providers only refresh quota when
// accessed, so recovery alone would leave a stale 0/—. For krea the sync
// must first load /app (Activate) to grant the daily free balance.
if recovered, err := m.tokens.RecoverQuota(ctx); err != nil {
log.Printf("maintenance: recover_quota: %v", err)
} else if len(recovered) > 0 {
log.Printf("maintenance: recovered %d quota token(s)", len(recovered))
if m.tokenSvc != nil {
go m.syncRecoveredQuota(recovered)
}
}
// 1a. Roll the 恢复时间 marker of ACTIVE daily-reset accounts forward to the next
// future reset (same time-of-day, +1 day) so the column never shows a stale
// past time. Limited accounts are intentionally skipped (RecoverQuota owns
// their marker). adobe/leonardo/krea/imagine all renew daily.
if _, err := m.tokens.RollResetMarkers(ctx, []string{"adobe", "leonardo", "krea", "imagine"}); err != nil {
log.Printf("maintenance: roll_reset: %v", err)
}
// 1b. Runway tokens have no refresh — once the JWT expiry marker passes the
// token can only 401, so flip it to disabled+dead proactively instead of
// leaving a doomed account "active". Grok is intentionally NOT swept here:
// its reset marker is billingPeriodEnd (a credits-renewal date), not a
// death deadline — a grok sso keeps working past billingPeriodEnd, so
// expiring on it kills live accounts. Grok death is caught for real by the
// import-time FetchSession check and by marking dead on a 401 at use.
for _, pool := range []string{"runway"} {
if n, err := m.tokens.ExpireByReset(ctx, pool); err != nil {
log.Printf("maintenance: expire_%s: %v", pool, err)
} else if n > 0 {
log.Printf("maintenance: expired %d %s token(s)", n, pool)
}
}
// 1c. Proactively renew krea/imagine sessions ~10min before expiry so a
// dormant account's rotating refresh_token never lapses (a dead token
// can't be recovered and, for krea, blocks the daily free-credit meter
// from being re-created). Only near-expiry accounts hit the network.
if m.tokenSvc != nil {
m.tokenSvc.RefreshExpiringTokens(ctx)
// 1d. Once-per-day krea /app activation for accounts not yet synced since the
// daily reset — krea only grants the free balance after /app loads, so an
// always-active account (never went 限额) would otherwise read 0 / 402
// after each reset. Self-guarded + background; no-op once all are done.
m.tokenSvc.ActivateKreaDue(ctx)
// 1e. Re-validate grok accounts: an empty /rest/subscriptions (or 401)
// means the membership lapsed → disable+dead; otherwise re-sync the
// credits balance and 恢复时间 (from the credits' weekly reset).
m.tokenSvc.RefreshGrokLiveness(ctx)
}
// 2. Auto-renew Adobe cookies whose refresh interval has elapsed.
if m.refresh != nil {
if m.settings != nil {
if proxy, err := m.settings.GetValue(ctx, "proxy.url"); err == nil && proxy != "" {
m.refresh.SetProxy(proxy)
}
}
if n, err := m.refresh.RefreshDue(ctx); err != nil {
log.Printf("maintenance: refresh_due: %v", err)
} else if n > 0 {
log.Printf("maintenance: refreshed %d cookie profile(s)", n)
}
}
// 3. Fail long-pending events so they stop blocking the per-user gate, and
// 1. Fail long-pending events so they stop blocking the per-user gate, and
// refund the credits debited up-front for each abandoned generation (the
// normal failure-refund path never ran for a process-restart orphan).
if purged, err := m.events.PurgeStale(ctx, m.stalePending, m.stalePendingVid); err != nil {
@@ -219,13 +90,6 @@ func (m *MaintenanceService) tick(ctx context.Context) {
if m.inflight != nil && m.inflight.Cancel(e.ID) {
cancelled++
}
// Attribute the abandoned failure back to the account it was using
// (the normal markTokenFailure path never ran for an orphaned job).
if e.AccountID != "" {
if err := m.tokens.IncrementFail(ctx, e.AccountID); err != nil {
log.Printf("maintenance: fail-count abandoned event %s (account %s): %v", e.ID, e.AccountID, err)
}
}
if e.UserID == "" || e.Cost <= 0 {
continue
}
@@ -248,10 +112,11 @@ func (m *MaintenanceService) tick(ctx context.Context) {
log.Printf("maintenance: marked %d stale pending event(s) failed, refunded %d, cancelled %d in-flight", len(purged), refunded, cancelled)
}
// 4. Enforce the admin-configured log retention window.
// 2. Enforce the admin-configured log retention window.
m.pruneLogs(ctx)
// 5. Enforce the media retention window. Runs every 60s like the log prune;
// 3. Enforce the media retention window + sweep expired tmp/ (v1 API job
// results) and stale refs/ objects. Runs every 60s like the log prune;
// mediaPruneEvery still gates it in case the interval is ever shortened.
if time.Since(m.lastMediaPrune) >= m.mediaPruneEvery {
m.pruneMedia(ctx)
@@ -274,10 +139,11 @@ func (m *MaintenanceService) pruneMedia(ctx context.Context) {
return
}
days := m.retentionDays(ctx, "media.retention_days")
if days <= 0 {
return
}
cutoff := time.Now().Add(-time.Duration(days) * 24 * time.Hour)
// tmp/ holds /v1 job results: presigned for 30 min, pruned shortly after.
tmpCutoff := time.Now().Add(-40 * time.Minute)
// <uid>/refs/ holds uploaded reference material: live for the presign TTL.
refCutoff := time.Now().Add(-2 * PresignTTL)
objs, err := m.store.List(ctx, "")
if err != nil {
log.Printf("maintenance: list media: %v", err)
@@ -303,10 +169,20 @@ func (m *MaintenanceService) pruneMedia(ctx context.Context) {
removed, skipped := 0, 0
var clearedKeys []string
for _, o := range objs {
if !o.LastModified.Before(cutoff) {
key := strings.TrimLeft(o.Key, "/")
expired := false
switch {
case strings.HasPrefix(key, "tmp/"):
expired = o.LastModified.Before(tmpCutoff)
case strings.Contains(key, "/refs/"):
expired = o.LastModified.Before(refCutoff)
case days > 0:
expired = o.LastModified.Before(cutoff)
}
if !expired {
continue
}
if _, ok := pinned[strings.TrimLeft(o.Key, "/")]; ok {
if _, ok := pinned[key]; ok {
skipped++
continue
}
@@ -1,235 +0,0 @@
package service
import (
"context"
"errors"
"strings"
"time"
"backend/internal/model"
"backend/internal/provider/adobe"
"backend/internal/repo"
"gorm.io/datatypes"
)
type RefreshProfileService struct {
profiles *repo.RefreshProfileRepository
tokens *repo.TokenRepository
adobe *adobe.Client
}
func NewRefreshProfileService(profiles *repo.RefreshProfileRepository, tokens *repo.TokenRepository, adobeClient *adobe.Client) *RefreshProfileService {
return &RefreshProfileService{
profiles: profiles,
tokens: tokens,
adobe: adobeClient,
}
}
func (s *RefreshProfileService) SetProxy(proxy string) {
if s.adobe != nil {
s.adobe.SetProxy(proxy)
}
}
func (s *RefreshProfileService) List(ctx context.Context) ([]model.RefreshProfile, error) {
return s.profiles.List(ctx)
}
func (s *RefreshProfileService) RefreshNow(ctx context.Context, id string) error {
if s.adobe == nil || s.tokens == nil {
return errors.New("refresh client not configured")
}
profile, err := s.profiles.Get(ctx, id)
if err != nil {
return err
}
if profile.Pool != "adobe" || profile.Kind != "adobe_cookie" {
return errors.New("unsupported refresh profile")
}
now := time.Now()
_, _ = s.profiles.Update(ctx, id, map[string]any{
"last_attempt_at": now,
})
result, err := s.adobe.ExchangeCookie(ctx, profile.Cookie)
if err != nil {
failures := profile.ConsecutiveFailures + 1
// Exponential backoff: 60s per consecutive failure, capped at 1h.
secs := 60 * failures
if secs > 3600 {
secs = 3600
}
msg := err.Error()
if len(msg) > 300 {
msg = msg[:300]
}
_, _ = s.profiles.Update(ctx, id, map[string]any{
"last_error": msg,
"consecutive_failures": failures,
"next_retry_at": now.Add(time.Duration(secs) * time.Second),
})
// Mark the token dead when the cookie exchange keeps failing:
// - 5 consecutive failures → cookie is likely expired/revoked
// - ride_AdobeID_acct_actreq (Adobe requires account action) → permanently broken
if failures >= 5 || strings.Contains(msg, "ride_AdobeID_acct_actreq") {
_, _ = s.tokens.Update(ctx, "adobe", id, map[string]any{
"status": "disabled",
"dead": true,
})
}
return err
}
tokenPatch := map[string]any{
"value": result.AccessToken,
"status": "active",
"dead": false,
"fails": 0,
"updated_at": now,
}
email, _ := parseJWTEmailExpiry(result.AccessToken)
if email != "" {
tokenPatch["account_email"] = email
}
if profile.IntervalSeconds > 0 {
tokenPatch["cached_quota_reset_after"] = now.Add(time.Duration(profile.IntervalSeconds) * time.Second).Format(time.RFC3339)
} else {
tokenPatch["cached_quota_reset_after"] = now.Add(54000 * time.Second).Format(time.RFC3339)
}
if profileData, profileErr := s.adobe.FetchAccountProfile(ctx, result.AccessToken); profileErr == nil {
if email := strings.TrimSpace(stringValue(profileData["email"])); email != "" {
tokenPatch["account_email"] = email
}
if displayName := strings.TrimSpace(stringValue(profileData["display_name"])); displayName != "" {
tokenPatch["account_display_name"] = displayName
}
}
planKnown := false
if quotaData, quotaErr := s.adobe.FetchCreditsBalance(ctx, result.AccessToken); quotaErr == nil {
meta := datatypes.JSONMap{
"cached_quota_at": int(time.Now().Unix()),
}
if remaining, ok := quotaData["remaining"].(int); ok {
meta["cached_quota_remaining"] = remaining
}
if used, ok := quotaData["used"].(int); ok {
meta["cached_quota_used"] = used
}
if total, ok := quotaData["total"].(int); ok {
meta["cached_quota_total"] = total
}
tokenPatch["meta"] = meta
planCap := strings.ToLower(strings.TrimSpace(stringValue(quotaData["plan"])))
isVIP := planCap != "" && !strings.EqualFold(planCap, "free")
planKnown = planCap != ""
// 额度打到 0 的 VIP(母号/子号) 视为普号:写死 plan=free,额度恢复(>0)后下次刷新自动还原真实身份。
if isVIP {
if rem, ok := quotaData["remaining"].(int); ok && rem <= 0 {
isVIP = false
planCap = "free"
}
}
meta["plan"] = planCap
// VIP: use Adobe's reset time; set concurrency 5.
// 母号/子号 身份固定,不随积分动态变化——只有降级普号才刷新身份。
if isVIP {
if resetAfter := strings.TrimSpace(stringValue(quotaData["available_until"])); resetAfter != "" {
tokenPatch["cached_quota_reset_after"] = resetAfter
}
tokenPatch["concurrency"] = 5
// 身份固定:这里重建了整个 meta,若不带上 is_sub_account,刷新会把导入时
// 确定的 母号/子号 标记冲掉。身份只在导入时确定,刷新只沿用、绝不靠积分猜:
// 已有标记就带过来;取不到(没 meta / 缺字段)说明身份未知,直接置死号。
if existing, gerr := s.tokens.Get(ctx, "adobe", id); gerr == nil {
if v, ok := existing.Meta["is_sub_account"]; ok {
meta["is_sub_account"] = v
}
}
if _, ok := meta["is_sub_account"]; !ok {
tokenPatch["status"] = "disabled"
tokenPatch["dead"] = true
}
} else {
// Free account: concurrency 1, limit image+video
tokenPatch["concurrency"] = 1
tokenPatch["image_limited"] = true
tokenPatch["video_limited"] = true
meta["is_sub_account"] = false
}
}
// 探测不到会员身份(credits 接口失败或没返回 plan)的号既不算普号也不算会员号,
// 留在池里只会在需要会员的请求上撞 403 —— 直接置死号。
if !planKnown {
tokenPatch["status"] = "disabled"
tokenPatch["dead"] = true
}
if _, err := s.tokens.Update(ctx, "adobe", id, tokenPatch); err != nil {
return err
}
interval := profile.IntervalSeconds
if interval <= 0 {
interval = 54000
}
_, err = s.profiles.Update(ctx, id, map[string]any{
"last_success_at": now,
"next_retry_at": now.Add(time.Duration(interval) * time.Second),
"last_error": "",
"consecutive_failures": 0,
})
return err
}
// RefreshDue refreshes every enabled profile whose next_retry_at has passed.
// Driven by the background maintenance loop so Adobe cookies auto-renew without
// an admin clicking "refresh". Individual failures are recorded on the profile
// (backoff + dead escalation) and don't abort the sweep.
func (s *RefreshProfileService) RefreshDue(ctx context.Context) (int, error) {
if s.adobe == nil || s.tokens == nil {
return 0, nil
}
due, err := s.profiles.ListDue(ctx, time.Now())
if err != nil {
return 0, err
}
refreshed := 0
for _, p := range due {
if p.Pool != "adobe" || p.Kind != "adobe_cookie" {
continue
}
if err := s.RefreshNow(ctx, p.ID); err != nil {
continue
}
refreshed++
}
return refreshed, nil
}
func (s *RefreshProfileService) Update(ctx context.Context, id string, body map[string]any) (*model.RefreshProfile, error) {
patch := map[string]any{}
if raw, ok := body["enabled"]; ok {
patch["enabled"] = boolValueWithDefault(raw, false)
}
if raw, ok := body["name"]; ok {
patch["name"] = stringValue(raw)
}
if raw, ok := body["interval_seconds"]; ok {
n := intValue(raw)
if n <= 0 {
return nil, errors.New("interval_seconds must be positive")
}
patch["interval_seconds"] = n
}
if len(patch) == 0 {
return s.profiles.Get(ctx, id)
}
return s.profiles.Update(ctx, id, patch)
}
func (s *RefreshProfileService) Delete(ctx context.Context, id string) error {
return s.profiles.Delete(ctx, id)
}
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -40,8 +40,8 @@ type UserGenerateRequest struct {
// DeAI applies 去AI特征 post-processing to the generated image (image only)
// and charges the per-tier surcharge on top of the model price.
DeAI bool
// AccountID pins an admin test to one specific provider account (账号生图测试).
AccountID string
// ChannelID pins an admin test to one specific channel (渠道生图测试).
ChannelID string
}
func (s *UserGenerationService) Generate(ctx context.Context, user *model.User, in UserGenerateRequest) (map[string]any, error) {
@@ -138,7 +138,7 @@ func (s *UserGenerationService) AdminTest(ctx context.Context, user *model.User,
AspectRatio: in.Ratio,
Resolution: in.Resolution,
ReferenceImages: in.ReferenceImages,
AccountID: in.AccountID,
ChannelID: in.ChannelID,
})
default:
return s.v1.prepareAdminTestImage(ctx, principal, V1ImageRequest{
@@ -147,7 +147,7 @@ func (s *UserGenerationService) AdminTest(ctx context.Context, user *model.User,
AspectRatio: in.Ratio,
Resolution: in.Resolution,
ReferenceImages: in.ReferenceImages,
AccountID: in.AccountID,
ChannelID: in.ChannelID,
})
}
}
File diff suppressed because it is too large Load Diff