feat: add shared media assets and fix credit lifecycle

This commit is contained in:
OpenClaw Agent
2026-09-17 02:30:31 +08:00
parent 7d79d43b61
commit b853a239ff
11 changed files with 351 additions and 19 deletions
+125
View File
@@ -0,0 +1,125 @@
package main
import (
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"time"
"gorm.io/gorm"
)
type MediaAsset struct {
ID uint `gorm:"primaryKey" json:"id"`
AssetKey string `gorm:"uniqueIndex;size:512" json:"assetKey"`
Source string `gorm:"index;size:32" json:"source"`
RID string `gorm:"index;size:256" json:"rid"`
Quality string `gorm:"index;size:32" json:"quality"`
LocalPath string `json:"localPath,omitempty"`
Format string `json:"format,omitempty"`
ActualQuality string `json:"actualQuality,omitempty"`
Size int64 `json:"size"`
Status string `gorm:"index;size:20" json:"status"`
OwnerTaskID uint `gorm:"index" json:"ownerTaskId,omitempty"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
func assetKey(source, rid, quality string) string {
return strings.Join([]string{source, rid, quality}, "\x00")
}
func (a *App) ensureAsset(t *Task) (*MediaAsset, error) {
if t.AssetID != 0 {
var asset MediaAsset
if err := a.db.First(&asset, t.AssetID).Error; err == nil {
return &asset, nil
}
}
key := assetKey(t.Source, t.RID, t.Quality)
var asset MediaAsset
err := a.db.Where("asset_key = ?", key).First(&asset).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
asset = MediaAsset{AssetKey: key, Source: t.Source, RID: t.RID, Quality: t.Quality, Status: "new"}
if err := a.db.Create(&asset).Error; err != nil {
if a.db.Where("asset_key = ?", key).First(&asset).Error != nil {
return nil, err
}
}
} else if err != nil {
return nil, err
}
t.AssetID = asset.ID
if err := a.db.Model(t).Update("asset_id", asset.ID).Error; err != nil {
return nil, err
}
return &asset, nil
}
// acquireAsset serializes physical downloads for the same source/rid/quality.
// A waiting task yields its worker briefly instead of creating duplicate files.
func (a *App) acquireAsset(t *Task) (*MediaAsset, bool, error) {
for i := 0; i < 120; i++ {
asset, err := a.ensureAsset(t)
if err != nil {
return nil, false, err
}
if asset.Status == "done" && asset.LocalPath != "" {
if fi, statErr := os.Stat(asset.LocalPath); statErr == nil && fi.Size() > 0 {
t.LocalPath, t.Format, t.ActualQuality = asset.LocalPath, asset.Format, asset.ActualQuality
t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = fi.Size(), fi.Size(), 100, "done"
if err := a.db.Save(t).Error; err != nil {
return nil, false, err
}
return asset, true, nil
}
}
claimed := false
err = a.db.Transaction(func(tx *gorm.DB) error {
var current MediaAsset
if err := tx.First(&current, asset.ID).Error; err != nil {
return err
}
if current.Status == "downloading" && current.OwnerTaskID != t.ID {
return nil
}
current.Status, current.OwnerTaskID = "downloading", t.ID
if current.LocalPath == "" {
current.LocalPath = filepath.Join(a.cfg.DownloadDir, safe(t.Artist)+" - "+safe(t.Title)+"."+t.Format)
}
if err := tx.Save(&current).Error; err != nil {
return err
}
*asset = current
claimed = true
return nil
})
if err != nil {
return nil, false, err
}
if claimed {
t.LocalPath = asset.LocalPath
return asset, false, nil
}
time.Sleep(500 * time.Millisecond)
}
return nil, false, fmt.Errorf("timed out waiting for shared media asset")
}
func (a *App) markAssetDone(t *Task) {
if t.AssetID == 0 {
return
}
_ = a.db.Model(&MediaAsset{}).Where("id = ? AND owner_task_id = ?", t.AssetID, t.ID).Updates(map[string]any{
"status": "done", "owner_task_id": 0, "local_path": t.LocalPath, "format": t.Format, "actual_quality": t.ActualQuality, "size": t.DownloadedBytes,
}).Error
}
func (a *App) releaseAsset(t *Task) {
if t.AssetID == 0 {
return
}
_ = a.db.Model(&MediaAsset{}).Where("id = ? AND owner_task_id = ?", t.AssetID, t.ID).Updates(map[string]any{"status": "error", "owner_task_id": 0}).Error
}