diff --git a/backend/internal/bootstrap/app.go b/backend/internal/bootstrap/app.go index 85ac10b..37a9e4d 100644 --- a/backend/internal/bootstrap/app.go +++ b/backend/internal/bootstrap/app.go @@ -68,6 +68,10 @@ func NewApp(ctx context.Context) (*App, error) { `ON cdk_codes (batch_id, redeemed_by) WHERE type = 'marketing' AND redeemed_by IS NOT NULL`).Error; err != nil { return nil, fmt.Errorf("cdk marketing index: %w", err) } + // 配价即上架（vivid 同款）：models 不再有启停开关，删遗留列。 + if err := db.WithContext(ctx).Exec(`ALTER TABLE model_configs DROP COLUMN IF EXISTS enabled`).Error; err != nil { + return nil, fmt.Errorf("drop model_configs.enabled: %w", err) + } if err := seedDefaults(ctx, db); err != nil { return nil, fmt.Errorf("seed defaults: %w", err) } @@ -159,7 +163,7 @@ func NewApp(ctx context.Context) (*App, error) { // Background self-healing sweep (order expiry, stale-pending cleanup+refund, // log/media retention). - maintenanceSvc := service.NewMaintenanceService(eventRepo, userRepo, siteRepo, rustfsClient, v1Svc.Inflight(), showcaseRepo, orderRepo) + maintenanceSvc := service.NewMaintenanceService(eventRepo, userRepo, siteRepo, rustfsClient, v1Svc, showcaseRepo, orderRepo) loopCtx, loopCancel := context.WithCancel(context.Background()) go maintenanceSvc.Run(loopCtx) diff --git a/backend/internal/http/handler/channel_admin.go b/backend/internal/http/handler/channel_admin.go index 7df5813..deeafc8 100644 --- a/backend/internal/http/handler/channel_admin.go +++ b/backend/internal/http/handler/channel_admin.go @@ -38,7 +38,7 @@ func (h *ChannelAdminHandler) Save(c *gin.Context) { return } body.BaseURL = strings.TrimSpace(body.BaseURL) - ch, err := h.channels.Save(c.Request.Context(), body) + ch, warning, err := h.channels.Save(c.Request.Context(), body) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { c.JSON(http.StatusNotFound, gin.H{"detail": "channel not found"}) @@ -47,7 +47,7 @@ func (h *ChannelAdminHandler) Save(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) return } - c.JSON(http.StatusOK, gin.H{"ok": true, "id": ch.ID}) + c.JSON(http.StatusOK, gin.H{"ok": true, "id": ch.ID, "warning": warning}) } func (h *ChannelAdminHandler) Delete(c *gin.Context) { diff --git a/backend/internal/http/handler/user_generation.go b/backend/internal/http/handler/user_generation.go index 55a2cf5..60bb853 100644 --- a/backend/internal/http/handler/user_generation.go +++ b/backend/internal/http/handler/user_generation.go @@ -461,7 +461,7 @@ func (h *UserGenerationHandler) VideoPresets(c *gin.Context) { } out := make([]gin.H, 0, len(models)) for _, m := range models { - if m.Type != "video" || !m.Enabled { + if m.Type != "video" { continue } out = append(out, modelPreset(m)) @@ -547,7 +547,7 @@ func (h *UserGenerationHandler) catalogEntries(c *gin.Context) ([]gin.H, error) } // publicModels — GET /api/models for the public site (docs/landing pickers). -// Enabled models only, in the same shape as before minus the provider id. +// Fully-priced models only (配价即上架), same shape as before. func (h *UserGenerationHandler) publicModels(c *gin.Context) ([]gin.H, error) { models, err := h.admin.Models(c.Request.Context()) if err != nil { @@ -555,7 +555,8 @@ func (h *UserGenerationHandler) publicModels(c *gin.Context) ([]gin.H, error) { } items := make([]gin.H, 0, len(models)) for _, m := range models { - if !m.Enabled { + // 配价即上架：未配齐价的模型对前台不可见（绑定/配价都满足才进 /v1）。 + if !service.ModelFullyPriced(&m) { continue } items = append(items, gin.H{ diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 172afea..1522f72 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -126,7 +126,6 @@ type ModelConfig struct { Name string `gorm:"size:255;not null"` Alias string `gorm:"column:alias;size:255;not null;default:''"` Provider string `gorm:"size:100;index;not null"` - Enabled bool `gorm:"not null;default:true"` FreeAllowed bool `gorm:"not null;default:false"` // 普号(free)能否调度此模型，默认不可用 Ratios datatypes.JSON `gorm:"type:jsonb"` Prices datatypes.JSONMap `gorm:"type:jsonb"` diff --git a/backend/internal/repo/channel_repo.go b/backend/internal/repo/channel_repo.go index c196bca..5dbc9dc 100644 --- a/backend/internal/repo/channel_repo.go +++ b/backend/internal/repo/channel_repo.go @@ -80,6 +80,15 @@ func (r *ChannelRepository) Unbind(ctx context.Context, channelID, modelID strin return r.db.WithContext(ctx).Delete(&model.ChannelModel{}, "channel_id = ? AND model_id = ?", channelID, modelID).Error } +// UnbindExcept removes every binding of a channel whose upstream model name is +// NOT in the keep list (upstream_model, falling back to model_id for legacy +// rows) — the "true sync" counterpart used after saving a channel. +func (r *ChannelRepository) UnbindExcept(ctx context.Context, channelID string, upstreamIDs []string) error { + return r.db.WithContext(ctx). + Where("channel_id = ? AND NOT (COALESCE(NULLIF(upstream_model, ''), model_id) IN ?)", channelID, upstreamIDs). + Delete(&model.ChannelModel{}).Error +} + func (r *ChannelRepository) DeleteBindingsByChannel(ctx context.Context, channelID string) error { return r.db.WithContext(ctx).Delete(&model.ChannelModel{}, "channel_id = ?", channelID).Error } @@ -158,6 +167,22 @@ func (r *ChannelRepository) UpstreamModelMap(ctx context.Context, channelID stri return out, nil } +// RoutableModelIDs returns the set of local model ids bound to at least one +// ENABLED channel — the "can actually dispatch" set behind the sellable flag. +func (r *ChannelRepository) RoutableModelIDs(ctx context.Context) (map[string]struct{}, error) { + var ids []string + err := r.db.WithContext(ctx). + Table("channel_models"). + Select("DISTINCT channel_models.model_id"). + Joins("JOIN channels ON channels.id = channel_models.channel_id AND channels.enabled = ?", true). + Scan(&ids).Error + out := make(map[string]struct{}, len(ids)) + for _, id := range ids { + out[id] = struct{}{} + } + return out, err +} + // ChannelIDsByModel lists channel ids bound to a model (any enabled state). func (r *ChannelRepository) ChannelIDsByModel(ctx context.Context, modelID string) ([]string, error) { var ids []string diff --git a/backend/internal/repo/event_repo.go b/backend/internal/repo/event_repo.go index 1bdc9b4..5953ddc 100644 --- a/backend/internal/repo/event_repo.go +++ b/backend/internal/repo/event_repo.go @@ -430,12 +430,41 @@ const stalePendingWhere = "status = ? AND ((kind = ? AND ts < ?) OR (kind <> ? A // StaleEvent identifies a purged pending event so the caller can refund the // credits debited up-front AND attribute the failure to the account the -// (now-abandoned) generation was using. +// (now-abandoned) generation was using. UpstreamJob/Kind/Model/File/Source are +// what the channel-resume path needs to pick the job back up. type StaleEvent struct { - ID string `gorm:"column:id"` - UserID string `gorm:"column:user_id"` - AccountID string `gorm:"column:account_id"` - Cost float64 `gorm:"column:cost"` + ID string `gorm:"column:id"` + UserID string `gorm:"column:user_id"` + AccountID string `gorm:"column:account_id"` + UpstreamJob string `gorm:"column:upstream_job"` + Kind string `gorm:"column:kind"` + Model string `gorm:"column:model"` + File string `gorm:"column:file"` + Source string `gorm:"column:source"` + Cost float64 `gorm:"column:cost"` + TS time.Time `gorm:"column:ts"` +} + +// ResumableStale returns pending events past their per-kind deadline that still +// carry an upstream job id — a process restart orphaned the local goroutine but +// the upstream job may still be running, so these get resumed (not purged). +// Events older than maxResumeAge are excluded and fall through to PurgeStale. +func (r *EventRepository) ResumableStale(ctx context.Context, maxAge, videoMaxAge, maxResumeAge time.Duration) ([]StaleEvent, error) { + if maxAge <= 0 { + maxAge = 600 * time.Second + } + if videoMaxAge <= 0 { + videoMaxAge = maxAge + } + cutoff := time.Now().Add(-maxAge) + videoCutoff := time.Now().Add(-videoMaxAge) + resumeFloor := time.Now().Add(-maxResumeAge) + var stale []StaleEvent + err := r.db.WithContext(ctx).Model(&model.EventLog{}). + Where(stalePendingWhere+" AND upstream_job <> '' AND ts >= ?", "pending", "video", videoCutoff, "video", cutoff, resumeFloor). + Select("id", "user_id", "account_id", "upstream_job", "kind", "model", "file", "source", "cost", "ts"). + Scan(&stale).Error + return stale, err } // PurgeStale marks long-pending entries as failed/abandoned and RETURNS them so @@ -446,7 +475,10 @@ type StaleEvent struct { // // videoMaxAge 单独给视频用：出片本来就慢（上游长镜头能跑二十多分钟），按图片的 // 期限扫会在它出片前就判死并退款。 -func (r *EventRepository) PurgeStale(ctx context.Context, maxAge, videoMaxAge time.Duration) ([]StaleEvent, error) { +// +// Events carrying an upstream job id and younger than resumeFloor are skipped — +// ResumableStale resumes them against the channel instead of killing them. +func (r *EventRepository) PurgeStale(ctx context.Context, maxAge, videoMaxAge, maxResumeAge time.Duration) ([]StaleEvent, error) { if maxAge <= 0 { maxAge = 600 * time.Second } @@ -455,13 +487,14 @@ func (r *EventRepository) PurgeStale(ctx context.Context, maxAge, videoMaxAge ti } cutoff := time.Now().Add(-maxAge) videoCutoff := time.Now().Add(-videoMaxAge) + resumeFloor := time.Now().Add(-maxResumeAge) var stale []StaleEvent err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // Snapshot who/what to refund BEFORE flipping status, so a concurrent // sweep can't double-count (the UPDATE in the same tx removes them from // the pending set). if err := tx.Model(&model.EventLog{}). - Where(stalePendingWhere, "pending", "video", videoCutoff, "video", cutoff). + Where(stalePendingWhere+" AND (upstream_job IS NULL OR upstream_job = '' OR ts < ?)", "pending", "video", videoCutoff, "video", cutoff, resumeFloor). Select("id", "user_id", "account_id", "cost"). Scan(&stale).Error; err != nil { return err @@ -470,7 +503,7 @@ func (r *EventRepository) PurgeStale(ctx context.Context, maxAge, videoMaxAge ti return nil } return tx.Model(&model.EventLog{}). - Where(stalePendingWhere, "pending", "video", videoCutoff, "video", cutoff). + Where(stalePendingWhere+" AND (upstream_job IS NULL OR upstream_job = '' OR ts < ?)", "pending", "video", videoCutoff, "video", cutoff, resumeFloor). Updates(map[string]any{ "status": "failed", "error": gorm.Expr("COALESCE(NULLIF(error, ''), ?)", "abandoned (process restarted or request interrupted)"), diff --git a/backend/internal/service/admin_read.go b/backend/internal/service/admin_read.go index 37ed753..d4de6c5 100644 --- a/backend/internal/service/admin_read.go +++ b/backend/internal/service/admin_read.go @@ -84,15 +84,18 @@ func (s *AdminReadService) ModelsView(ctx context.Context) ([]map[string]any, er if err != nil { return nil, err } + routable, _ := s.channels.RoutableModelIDs(ctx) out := make([]map[string]any, 0, len(items)) for _, item := range items { + _, bound := routable[item.ID] out = append(out, map[string]any{ "id": item.ID, "alias": item.Alias, "type": item.Type, "name": item.Name, "provider": item.Provider, - "enabled": item.Enabled, + // 配价即上架（无启停开关）：全档位有价 + 绑定了启用渠道才算可售。 + "sellable": bound && ModelFullyPriced(&item), "ratios": repo.JSONStrings(item.Ratios), "prices": map[string]any(item.Prices), "resolutions": repo.JSONStrings(item.Resolutions), diff --git a/backend/internal/service/admin_write.go b/backend/internal/service/admin_write.go index dbb4335..31699a2 100644 --- a/backend/internal/service/admin_write.go +++ b/backend/internal/service/admin_write.go @@ -391,7 +391,6 @@ func (s *AdminWriteService) CreateModel(ctx context.Context, body map[string]any Name: defaultString(strings.TrimSpace(stringValue(body["name"])), modelID), Alias: alias, Provider: provider, - Enabled: boolValueWithDefault(body["enabled"], true), Ratios: jsonArray(body["ratios"]), Prices: prices, Resolutions: resolutions, @@ -440,13 +439,6 @@ func (s *AdminWriteService) UpdateModel(ctx context.Context, modelID string, bod 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). - // Without this guard a PATCH that omits the field would default it to false - // and silently disable the model. - if raw, ok := body["enabled"]; ok && raw != nil { - patch["enabled"] = boolValueWithDefault(raw, true) - } if _, ok := body["ratios"]; ok { patch["ratios"] = jsonArray(body["ratios"]) } diff --git a/backend/internal/service/channel.go b/backend/internal/service/channel.go index 50fffbe..dbb229d 100644 --- a/backend/internal/service/channel.go +++ b/backend/internal/service/channel.go @@ -120,14 +120,17 @@ type SaveChannelDto struct { Weight int `json:"weight"` } -func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Channel, error) { +// Save upserts a channel then auto-syncs its upstream models (same semantics +// as vivid: upstream /v1/models is the truth). The returned warning is a +// non-fatal sync message — "" means fully synced. +func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Channel, string, error) { name := strings.TrimSpace(d.Name) base := strings.TrimRight(strings.TrimSpace(d.BaseURL), "/") if base == "" { - return nil, errors.New("base_url required") + 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://") + return nil, "", errors.New("base_url must start with http:// or https://") } weight := d.Weight if weight < 1 { @@ -135,11 +138,11 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha } if d.ID == "" { if strings.TrimSpace(d.APIKey) == "" { - return nil, errors.New("api_key required") + return nil, "", errors.New("api_key required") } enc, err := s.encrypt(d.APIKey) if err != nil { - return nil, err + return nil, "", err } ch := &model.Channel{ ID: "ch-" + randomUpper(10), @@ -153,7 +156,10 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha if ch.Name == "" { ch.Name = base } - return ch, s.channels.Create(ctx, ch) + if err := s.channels.Create(ctx, ch); err != nil { + return nil, "", err + } + return ch, s.syncChannelModels(ctx, ch.ID), nil } patch := map[string]any{ "name": name, @@ -165,11 +171,95 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha if key := strings.TrimSpace(d.APIKey); key != "" && !strings.Contains(key, "•") && !strings.Contains(key, "*") { enc, err := s.encrypt(key) if err != nil { - return nil, err + return nil, "", err } patch["api_key_enc"] = enc } - return s.channels.Update(ctx, d.ID, patch) + ch, err := s.channels.Update(ctx, d.ID, patch) + if err != nil { + return nil, "", err + } + return ch, s.syncChannelModels(ctx, d.ID), nil +} + +// syncChannelModels runs after Save: upstream /v1/models is the truth — every +// sellable upstream model is upserted locally (alias = upstream id) and bound; +// bindings on this channel not in the upstream list are removed, and orphan +// local models are deleted. An unreachable upstream or empty/anomalous list +// only produces a warning — the DB is left untouched. +func (s *ChannelService) syncChannelModels(ctx context.Context, channelID string) string { + ch, err := s.channels.Get(ctx, channelID) + if err != nil { + return "模型自动同步失败：" + err.Error() + } + key := s.decrypt(ch.APIKeyEnc) + if key == "" { + return "模型自动同步失败：无法解密上游密钥" + } + fctx, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + models, err := s.up.Models(fctx, ch.BaseURL, key) + if err != nil { + return "模型自动同步失败：" + err.Error() + } + if len(models) == 0 { + return "上游未返回任何模型，已跳过同步" + } + specs := make([]ChannelModelSpec, 0, len(models)) + upstreamIDs := make([]string, 0, len(models)) + var skipped []string + for _, m := range models { + spec, serr := normalizeRemoteModel(m) + if serr != nil { + skipped = append(skipped, m.ID+"："+serr.Error()) + continue + } + specs = append(specs, spec) + upstreamIDs = append(upstreamIDs, spec.ID) + } + if len(upstreamIDs) == 0 { + return fmt.Sprintf("上游模型均规格异常，已跳过同步：%s", strings.Join(skipped, "；")) + } + if err := s.ImportModels(ctx, channelID, specs); err != nil { + return "模型自动同步失败：" + err.Error() + } + if err := s.channels.UnbindExcept(ctx, channelID, upstreamIDs); err != nil { + return "模型自动同步失败：" + err.Error() + } + if orphans, oerr := s.channels.OrphanModelIDs(ctx); oerr == nil { + for _, mid := range orphans { + _, _ = s.models.Delete(ctx, mid) + } + } + if len(skipped) > 0 { + return fmt.Sprintf("已同步上游模型，跳过 %d 个异常项：%s", len(skipped), strings.Join(skipped, "；")) + } + return "" +} + +// normalizeRemoteModel validates an upstream /v1/models item into an import +// spec (alias = upstream id). +func normalizeRemoteModel(m upstream.RemoteModel) (ChannelModelSpec, error) { + spec := ChannelModelSpec{ + ID: strings.TrimSpace(m.ID), + Kind: strings.TrimSpace(m.Kind), + Qualities: m.Qualities, + Duration: m.Duration, + RefImages: m.RefImages, + RefVideos: m.RefVideos, + RefAudios: m.RefAudios, + RefVideoSeconds: m.RefVideoSeconds, + RefAudioSeconds: m.RefAudioSeconds, + } + if spec.ID == "" { + return spec, errors.New("模型 id 为空") + } + switch spec.Kind { + case "image", "video", "text": + default: + return spec, fmt.Errorf("未知类型 %q", spec.Kind) + } + return spec, nil } // Delete removes a channel, its bindings, and any local model that ends up @@ -348,7 +438,6 @@ func (s *ChannelService) ImportModels(ctx context.Context, channelID string, spe Name: localID, Type: spec.Kind, Provider: "channel", - Enabled: true, Ratios: jsonArray([]string{"16:9"}), Prices: datatypes.JSONMap{}, CreatedAt: time.Now(), @@ -459,6 +548,11 @@ func (s *ChannelService) ChannelsForModelAnyState(ctx context.Context, modelID s } // Decrypt exposes the upstream key for dispatch. +// Get loads one channel row by id (resume path needs the owning channel). +func (s *ChannelService) Get(ctx context.Context, id string) (*model.Channel, error) { + return s.channels.Get(ctx, id) +} + func (s *ChannelService) Decrypt(ch model.Channel) string { return s.decrypt(ch.APIKeyEnc) } // Upstream returns the shared protocol client (dispatch needs it for diff --git a/backend/internal/service/maintenance.go b/backend/internal/service/maintenance.go index 22ae379..b1dae30 100644 --- a/backend/internal/service/maintenance.go +++ b/backend/internal/service/maintenance.go @@ -20,23 +20,24 @@ type MaintenanceService struct { users *repo.UserRepository settings *repo.SiteSettingRepository store *storage.Client - inflight *InflightRegistry + v1 *V1Service showcase *repo.ShowcaseRepository orders *repo.OrderRepository interval time.Duration stalePending time.Duration stalePendingVid time.Duration + maxResumeAge time.Duration mediaPruneEvery time.Duration lastMediaPrune time.Time } -func NewMaintenanceService(events *repo.EventRepository, users *repo.UserRepository, 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, v1 *V1Service, showcase *repo.ShowcaseRepository, orders *repo.OrderRepository) *MaintenanceService { return &MaintenanceService{ events: events, users: users, settings: settings, store: store, - inflight: inflight, + v1: v1, showcase: showcase, orders: orders, interval: 60 * time.Second, @@ -44,6 +45,9 @@ func NewMaintenanceService(events *repo.EventRepository, users *repo.UserReposit // 视频比图片慢得多，单条执行预算就是 30 分钟（videoGenBudget），用图片的 // 10 分钟去扫会在它出片前判死并退款。 stalePendingVid: 30 * time.Minute, + // 带上游 jobId 的逾期任务先尝试续等而不是判死；超过 3 小时总时限 + // （vivid 同款）才放弃走判死退款。 + maxResumeAge: 3 * time.Hour, mediaPruneEvery: 60 * time.Second, } } @@ -75,10 +79,26 @@ func (m *MaintenanceService) tick(ctx context.Context) { } } - // 1. Fail long-pending events so they stop blocking the per-user gate, and + // 1a. Pending events with an upstream job id that outlived their local + // goroutine (process restart) get resumed against the owning channel — + // the upstream job keeps running even while we were down. + if m.v1 != nil { + if resumable, err := m.events.ResumableStale(ctx, m.stalePending, m.stalePendingVid, m.maxResumeAge); err != nil { + log.Printf("maintenance: resumable_stale: %v", err) + } else { + for _, e := range resumable { + m.v1.ResumeStaleJob(ctx, e) + } + if len(resumable) > 0 { + log.Printf("maintenance: resuming %d stale job(s) on their channels", len(resumable)) + } + } + } + + // 1b. 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 { + if purged, err := m.events.PurgeStale(ctx, m.stalePending, m.stalePendingVid, m.maxResumeAge); err != nil { log.Printf("maintenance: purge_stale: %v", err) } else if len(purged) > 0 { refunded := 0 @@ -87,7 +107,7 @@ func (m *MaintenanceService) tick(ctx context.Context) { // Stop the generation goroutine if it's still running, so it doesn't // keep grinding for minutes and surface a late "success" on this // just-abandoned event. - if m.inflight != nil && m.inflight.Cancel(e.ID) { + if m.v1 != nil && m.v1.Inflight().Cancel(e.ID) { cancelled++ } if e.UserID == "" || e.Cost <= 0 { diff --git a/backend/internal/service/v1.go b/backend/internal/service/v1.go index 552dedf..19da6ac 100644 --- a/backend/internal/service/v1.go +++ b/backend/internal/service/v1.go @@ -159,6 +159,16 @@ func (r *InflightRegistry) Add(eventID string, cancel context.CancelFunc) { } } +// TryAdd registers an event only when nothing is tracked for it — the resume +// sweep uses it so two ticks can't double-attach to the same upstream job. +func (r *InflightRegistry) TryAdd(eventID string, cancel context.CancelFunc) bool { + if eventID == "" { + return false + } + _, loaded := r.m.LoadOrStore(eventID, cancel) + return !loaded +} + // Done deregisters an event (called on normal completion). func (r *InflightRegistry) Done(eventID string) { r.m.Delete(eventID) } @@ -342,9 +352,9 @@ type V1ModelDur struct { Options []int64 `json:"options,omitempty"` } -// SellableModels — GET /v1/models. Only enabled models bound to at least one -// enabled channel AND fully priced are sellable (a model with an unpriced tier -// would charge-reject at submit time — hide it instead). +// SellableModels — GET /v1/models. 配价即上架（无启停开关）：绑定到至少一个 +// enabled 渠道 AND 全部档位有价才是可售模型（缺价的档位提交时会扣费拒绝， +// 直接不下发）。kind 不限——image/video/text 同构透出，下游才能连导。 func (s *V1Service) SellableModels(ctx context.Context) ([]V1ModelItem, error) { items, err := s.models.List(ctx) if err != nil { @@ -352,9 +362,6 @@ func (s *V1Service) SellableModels(ctx context.Context) ([]V1ModelItem, error) { } out := make([]V1ModelItem, 0, len(items)) for _, item := range items { - if !item.Enabled { - continue - } if !s.modelSellable(ctx, &item) { continue } @@ -393,6 +400,12 @@ func (s *V1Service) modelSellable(ctx context.Context, item *model.ModelConfig) if err != nil || len(channels) == 0 { return false } + return ModelFullyPriced(item) +} + +// ModelFullyPriced — 上架门槛（无启停开关，配价即上架）：每个清晰度档位在 +// Prices 里有价；视频模型还要求时长有价（离散档逐档，或区间档按秒价）。 +func ModelFullyPriced(item *model.ModelConfig) bool { for _, res := range repo.JSONStrings(item.Resolutions) { if _, ok := jsonMapFloat(item.Prices, res); !ok { return false @@ -564,6 +577,10 @@ func (s *V1Service) generateOnChannel(ctx context.Context, eventID string, ch mo if upstreamJob == "" { upstreamJob = job _ = s.events.SetUpstreamJob(ctx, eventID, job) + // Stamp the owning channel the moment the upstream accepts the job — + // a restart between submit and finish would otherwise leave a + // pending row that can't tell which channel to resume on. + _ = s.events.SetAccount(ctx, eventID, ch.ID, ch.Name) } }) if err != nil { @@ -881,9 +898,6 @@ func (s *V1Service) CreateJob(ctx context.Context, principal *APIPrincipal, in V } return nil, err } - if !modelItem.Enabled { - return nil, ErrUnknownModel - } kind := modelItem.Type if kind != "image" && kind != "video" { return nil, ErrUnknownModel @@ -1052,6 +1066,96 @@ func (s *V1Service) runV1Job(ctx context.Context, principal *APIPrincipal, event _ = s.maybeGrantInviteReward(ctx, principal) } +// ResumeStaleJob re-attaches to an upstream job whose local goroutine died with +// a process restart. Called by the maintenance sweep; TryAdd guards against +// double-attach across ticks. Terminal failure (upstream job lost/failed, +// channel gone or key broken) fails the event and refunds; transient failure +// leaves it pending for the next sweep (events older than maxResumeAge get +// purged instead of resumed). +func (s *V1Service) ResumeStaleJob(ctx context.Context, ev repo.StaleEvent) { + if ev.UpstreamJob == "" || ev.AccountID == "" { + return + } + rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), videoGenBudget) + if !s.inflight.TryAdd(ev.ID, cancel) { + cancel() + return + } + go s.resumeUpstreamJob(rctx, ev) +} + +func (s *V1Service) resumeUpstreamJob(ctx context.Context, ev repo.StaleEvent) { + defer s.inflight.Done(ev.ID) + fail := func(msg string) { + s.refundStale(ctx, ev) + elapsed := int(time.Since(ev.TS).Milliseconds()) + _ = s.events.UpdateStatus(ctx, ev.ID, "failed", msg, elapsed) + } + ch, err := s.channels.Get(ctx, ev.AccountID) + if err != nil || ch == nil { + fail("上游渠道已删除，任务无法续等") + return + } + apiKey := s.channels.Decrypt(*ch) + if apiKey == "" { + fail("上游渠道密钥不可用，任务无法续等") + return + } + base := strings.TrimRight(ch.BaseURL, "/") + resultURL, credits, err := s.up.Generate(ctx, base, apiKey, upstream.GenParams{JobID: ev.UpstreamJob}, int64(videoGenBudget/time.Second), nil) + if err != nil { + switch upstream.KindOf(err) { + case upstream.KindJobFailed, upstream.KindInvalid, upstream.KindAuth: + fail(upstreamUserError(err).Error()) + default: + // Retryable/transport/timeout — the upstream job may still run; + // leave the event pending for the next sweep. + log.Printf("[v1] resume upstream job %s (event %s) retry later: %v", ev.UpstreamJob, ev.ID, err) + } + return + } + data, ct, err := s.up.Download(ctx, resultURL) + if err != nil { + log.Printf("[v1] resume download %s (event %s) retry later: %v", ev.UpstreamJob, ev.ID, err) + return + } + rel := strings.TrimSpace(ev.File) + if rel == "" { + rel = filepath.ToSlash(filepath.Join("tmp", ev.ID+"."+resultExt(ct, ev.Kind))) + } + if err := s.store.Put(ctx, rel, data, ct); err != nil { + fail("storage upload failed: " + err.Error()) + return + } + _ = s.events.SetFile(ctx, ev.ID, rel) + if err := s.events.UpdateStatus(ctx, ev.ID, "success", "", int(time.Since(ev.TS).Milliseconds())); err != nil { + return + } + s.channels.BumpSuccess(ctx, ch.ID, float64(credits)) + if ev.Model != "" { + _ = s.models.IncrementGenerationCount(ctx, ev.Model) + } + if ev.UserID != "" { + _ = s.users.IncrementGenerationCount(ctx, ev.UserID) + } + log.Printf("[v1] resumed upstream job %s → event %s succeeded", ev.UpstreamJob, ev.ID) +} + +// refundStale returns the up-front charge for an abandoned/resume-dead event, +// claimed exactly once via the refunded flag. +func (s *V1Service) refundStale(ctx context.Context, ev repo.StaleEvent) { + if ev.UserID == "" || ev.Cost <= 0 { + return + } + claimed, err := s.events.MarkRefunded(ctx, ev.ID) + if err != nil || !claimed { + return + } + if _, err := s.users.RefundCredits(ctx, ev.UserID, ev.Cost); err != nil { + log.Printf("[v1] refund stale event %s (user %s, %.0f): %v", ev.ID, ev.UserID, ev.Cost, err) + } +} + func resultExt(contentType, kind string) string { switch { case strings.Contains(contentType, "webp"): @@ -1300,7 +1404,7 @@ func (s *V1Service) prepareImage(ctx context.Context, principal *APIPrincipal, i } return nil, "", "", 0, err } - if !modelItem.Enabled || modelItem.Type != "image" { + if modelItem.Type != "image" { return nil, "", "", 0, ErrUnknownModel } // Fail fast before charging if no channel serves this model. @@ -1361,7 +1465,7 @@ func (s *V1Service) prepareVideo(ctx context.Context, principal *APIPrincipal, i } return nil, "", "", "", 0, err } - if !modelItem.Enabled || modelItem.Type != "video" { + if modelItem.Type != "video" { return nil, "", "", "", 0, ErrUnknownModel } // Validate duration against the model's spec (options list or min/max range). diff --git a/frontend/src/views/CanvasView.vue b/frontend/src/views/CanvasView.vue index ee0eccc..b3b1ddc 100644 --- a/frontend/src/views/CanvasView.vue +++ b/frontend/src/views/CanvasView.vue @@ -50,7 +50,7 @@ watch(genMsg, (v) => { if (v) genMsgTimer = setTimeout(() => { genMsg.value = '' }, 3500) }) const kindModels = computed(() => - (allModels.value || []).filter((m) => m.enabled !== false && m.type === gen.kind), + (allModels.value || []).filter((m) => m.sellable && m.type === gen.kind), ) const model = computed(() => kindModels.value.find((m) => m.id === gen.modelId) || null) const ratios = computed(() => model.value?.ratios?.length ? model.value.ratios : ['1:1']) diff --git a/frontend/src/views/ChannelsView.vue b/frontend/src/views/ChannelsView.vue index d7bcac8..c2d94a1 100644 --- a/frontend/src/views/ChannelsView.vue +++ b/frontend/src/views/ChannelsView.vue @@ -9,6 +9,7 @@ const KIND_LABEL = { image: '图像', video: '视频', text: '文本' } const channels = ref([]) const loading = ref(false) const err = ref('') +const warn = ref('') // 保存后自动同步的非致命告警 // create/edit form const form = ref(null) @@ -59,6 +60,7 @@ async function save() { })) saving.value = false if (!r.ok) { formErr.value = r.data?.detail || '保存失败'; return } + warn.value = r.data?.warning || '' form.value = null load() } @@ -70,7 +72,8 @@ async function toggle(c) { id: c.id, name: c.name, base_url: c.base_url, api_key: '', enabled: c.enabled, concurrency: c.concurrency || 0, weight: c.weight || 1, })) - if (!r.ok) { c.enabled = prev; err.value = r.data?.detail || '状态更新失败' } + if (!r.ok) { c.enabled = prev; err.value = r.data?.detail || '状态更新失败'; return } + warn.value = r.data?.warning || '' } async function del(c) { @@ -117,6 +120,7 @@ async function toggleBind(s) { enabled: target.enabled !== false, concurrency: target.concurrency || 0, weight: target.weight || 1, })) if (!sr.ok) { fetchErr.value = sr.data?.detail || '渠道保存失败'; return } + if (sr.data?.warning) warn.value = sr.data.warning target.id = sr.data?.id form.value = null } @@ -159,7 +163,7 @@ onMounted(load)
渠道
-

同构 /v1 协议上游实例 · 模型经「拉取上游模型」导入 · 余额实时探测

+

同构 /v1 协议上游实例 · 保存时自动同步上游模型 · 余额实时探测

{{ err }}

+

{{ warn }}

@@ -292,7 +297,7 @@ onMounted(load)
正在从上游拉取模型列表…