feat(model): 同步 vivid 最新变更 — 去模型开关,配价即上架
对齐 vivid a338edb / 7fb5c57 / 4f454b6: - model_configs.enabled 列删除:上架唯一门槛 = 绑定了启用渠道 + 全部档位 有价(视频另需时长有价)。/v1/models、/api/models、video-presets、 managed-models 全部改按 sellable 透出 - 保存渠道即全量同步上游模型(消失解绑/孤儿删除/失败告警不删库) - v1 job 持久化 upstream jobId,重启后维护循环按 jobId 续等不丢单 Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)"),
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+115
-11
@@ -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).
|
||||
|
||||
Reference in New Issue
Block a user