From b853a239ff0a396ed6932799fb6e8b747bb2e798 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Thu, 17 Sep 2026 02:30:31 +0800 Subject: [PATCH] feat: add shared media assets and fix credit lifecycle --- admin_users.go | 42 ++++++++++++++- assets.go | 125 ++++++++++++++++++++++++++++++++++++++++++++ credits.go | 6 ++- credits_test.go | 125 ++++++++++++++++++++++++++++++++++++++++++++ downloader.go | 27 ++++++---- handlers.go | 30 +++++++++-- main.go | 7 ++- models.go | 1 + multiuser_models.go | 2 +- redeem.go | 4 ++ util.go | 1 + 11 files changed, 351 insertions(+), 19 deletions(-) create mode 100644 assets.go create mode 100644 credits_test.go diff --git a/admin_users.go b/admin_users.go index 1b641ec..6879012 100644 --- a/admin_users.go +++ b/admin_users.go @@ -57,6 +57,7 @@ func (a *App) adminCreateUser(c *gin.Context) { c.JSON(500, gin.H{"error": err.Error()}) return } + admin := c.MustGet("user").(*User) var u User err = a.db.Transaction(func(tx *gorm.DB) error { u = User{Username: strings.TrimSpace(in.Username), PasswordHash: hash, Role: in.Role, Status: UserActive} @@ -83,6 +84,7 @@ func (a *App) adminCreateUser(c *gin.Context) { return } u.PasswordHash = "" + a.audit(admin.ID, "user.create", "user", strconv.FormatUint(uint64(u.ID), 10), "role="+u.Role) c.JSON(201, u) } @@ -145,10 +147,48 @@ func (a *App) adminUpdateUser(c *gin.Context) { } u.Status = *in.Status } - if err := a.db.Save(&u).Error; err != nil { + operator := c.MustGet("user").(*User) + if u.ID == operator.ID && (u.Role != RoleAdmin || u.Status == UserDisabled) { + var admins int64 + if err := a.db.Model(&User{}).Where("role = ? AND status = ?", RoleAdmin, UserActive).Count(&admins).Error; err != nil { + c.JSON(500, gin.H{"error": err.Error()}) + return + } + if admins <= 1 { + c.JSON(400, gin.H{"error": "cannot remove the last active admin account"}) + return + } + } + if err := a.db.Transaction(func(tx *gorm.DB) error { + if err := tx.Save(&u).Error; err != nil { + return err + } + if in.Role != nil { + if err := tx.Where("user_id = ?", u.ID).Delete(&UserPermission{}).Error; err != nil { + return err + } + for _, p := range defaultPermissions(u.Role) { + if err := tx.Create(&UserPermission{UserID: u.ID, Permission: p, Enabled: true}).Error; err != nil { + return err + } + } + } + return nil + }); err != nil { c.JSON(500, gin.H{"error": err.Error()}) return } u.PasswordHash = "" + changes := []string{} + if in.Password != nil { + changes = append(changes, "password") + } + if in.Role != nil { + changes = append(changes, "role="+u.Role) + } + if in.Status != nil { + changes = append(changes, "status="+u.Status) + } + a.audit(operator.ID, "user.update", "user", strconv.FormatUint(uint64(u.ID), 10), strings.Join(changes, ",")) c.JSON(200, u) } diff --git a/assets.go b/assets.go new file mode 100644 index 0000000..ebb1cc2 --- /dev/null +++ b/assets.go @@ -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(¤t, 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(¤t).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 +} diff --git a/credits.go b/credits.go index 717245a..66429aa 100644 --- a/credits.go +++ b/credits.go @@ -33,7 +33,7 @@ func (a *App) reserveCredit(userID, taskID uint, amount int64) (*CreditReservati } } var existing CreditReservation - if err := tx.Where("task_id = ?", taskID).First(&existing).Error; err == nil { + if err := tx.Where("task_id = ? AND status = ?", taskID, "reserved").Order("id desc").First(&existing).Error; err == nil { reservation = existing return nil } else if !errors.Is(err, gorm.ErrRecordNotFound) { @@ -106,7 +106,9 @@ func (a *App) refundCredit(task *Task) error { return nil } if r.Status == "committed" { - return errReservationState + // A committed reservation belongs to a completed download. A later + // storage/retry error must never debit or refund it a second time. + return nil } if r.Status != "reserved" { return errReservationState diff --git a/credits_test.go b/credits_test.go new file mode 100644 index 0000000..0ab161c --- /dev/null +++ b/credits_test.go @@ -0,0 +1,125 @@ +package main + +import ( + "testing" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" +) + +func newCreditTestApp(t *testing.T) *App { + t.Helper() + db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&Task{}, &MediaAsset{}, &User{}, &CreditAccount{}, &CreditTransaction{}, &CreditReservation{}); err != nil { + t.Fatal(err) + } + return &App{db: db} +} + +func TestCreditReservationRetryAfterRefund(t *testing.T) { + a := newCreditTestApp(t) + u := User{Username: "u", Role: RoleUser, Status: UserActive} + if err := a.db.Create(&u).Error; err != nil { + t.Fatal(err) + } + if err := a.db.Create(&CreditAccount{UserID: u.ID, Balance: 1}).Error; err != nil { + t.Fatal(err) + } + task := Task{UserID: u.ID, Source: "kw", RID: "1", Quality: "standard", Status: "error"} + if err := a.db.Create(&task).Error; err != nil { + t.Fatal(err) + } + r1, err := a.reserveCredit(u.ID, task.ID, 1) + if err != nil { + t.Fatal(err) + } + task.CreditReservationID = r1.ID + if err := a.refundCredit(&task); err != nil { + t.Fatal(err) + } + r2, err := a.reserveCredit(u.ID, task.ID, 1) + if err != nil { + t.Fatal(err) + } + if r2.ID == r1.ID { + t.Fatal("retry reused refunded reservation") + } + var account CreditAccount + if err := a.db.Where("user_id = ?", u.ID).First(&account).Error; err != nil { + t.Fatal(err) + } + if account.Balance != 1 || account.Reserved != 1 { + t.Fatalf("unexpected account: %+v", account) + } +} + +func TestSharedAssetIsReusedAcrossUsers(t *testing.T) { + a := newCreditTestApp(t) + u1 := User{Username: "a1", Role: RoleUser, Status: UserActive} + u2 := User{Username: "a2", Role: RoleUser, Status: UserActive} + if err := a.db.Create(&u1).Error; err != nil { + t.Fatal(err) + } + if err := a.db.Create(&u2).Error; err != nil { + t.Fatal(err) + } + t1 := Task{UserID: u1.ID, Source: "kw", RID: "same", Quality: "standard", Title: "Song", Artist: "Artist"} + t2 := Task{UserID: u2.ID, Source: "kw", RID: "same", Quality: "standard", Title: "Song", Artist: "Artist"} + if err := a.db.Create(&t1).Error; err != nil { + t.Fatal(err) + } + if err := a.db.Create(&t2).Error; err != nil { + t.Fatal(err) + } + asset, reused, err := a.acquireAsset(&t1) + if err != nil || reused { + t.Fatalf("first acquire: asset=%+v reused=%v err=%v", asset, reused, err) + } + if asset.ID == 0 { + t.Fatal("missing asset id") + } + if err := a.db.Model(&MediaAsset{}).Where("id = ?", asset.ID).Updates(map[string]any{"status": "done", "owner_task_id": 0, "local_path": "/etc/hosts", "format": "mp3", "size": 123}).Error; err != nil { + t.Fatal(err) + } + t2.AssetID = asset.ID + asset2, reused, err := a.acquireAsset(&t2) + if err != nil || !reused || asset2.ID != asset.ID { + t.Fatalf("second acquire: asset=%+v reused=%v err=%v", asset2, reused, err) + } +} + +func TestCommittedCreditIsNotRefunded(t *testing.T) { + a := newCreditTestApp(t) + u := User{Username: "u2", Role: RoleUser, Status: UserActive} + if err := a.db.Create(&u).Error; err != nil { + t.Fatal(err) + } + if err := a.db.Create(&CreditAccount{UserID: u.ID, Balance: 1}).Error; err != nil { + t.Fatal(err) + } + task := Task{UserID: u.ID, Source: "kw", RID: "2", Quality: "standard", Status: "pending"} + if err := a.db.Create(&task).Error; err != nil { + t.Fatal(err) + } + r, err := a.reserveCredit(u.ID, task.ID, 1) + if err != nil { + t.Fatal(err) + } + task.CreditReservationID = r.ID + if err := a.commitCredit(&task); err != nil { + t.Fatal(err) + } + if err := a.refundCredit(&task); err != nil { + t.Fatal(err) + } + var account CreditAccount + if err := a.db.Where("user_id = ?", u.ID).First(&account).Error; err != nil { + t.Fatal(err) + } + if account.Balance != 0 || account.Reserved != 0 { + t.Fatalf("unexpected account after committed refund: %+v", account) + } +} diff --git a/downloader.go b/downloader.go index 74f6a69..9250450 100644 --- a/downloader.go +++ b/downloader.go @@ -40,17 +40,17 @@ func (a *App) download(id uint) { if ext == "" { ext = "mp3" } - filename := safe(t.Artist) + " - " + safe(t.Title) + "." + ext - local := filepath.Join(a.cfg.DownloadDir, filename) - t.Format, t.ResolvedQuality, t.ActualQuality, t.LocalPath = ext, info.Level, "", local - - // Reuse only the path recorded on this task. Never borrow another task's quality. - if t.LocalPath != "" { - if fi, statErr := os.Stat(t.LocalPath); statErr == nil && fi.Size() > 0 { - a.finishExisting(&t, fi.Size()) - return - } + t.Format, t.ResolvedQuality, t.ActualQuality = ext, info.Level, "" + asset, reused, err := a.acquireAsset(&t) + if err != nil { + a.fail(&t, err) + return } + if reused { + a.finishExisting(&t, asset.Size) + return + } + a.db.Model(&t).Updates(map[string]any{"format": t.Format, "local_path": t.LocalPath}) a.downloadFile(&t, info) } @@ -141,6 +141,13 @@ func (a *App) downloadFile(t *Task, info playInfo) { t.Progress = 100 } a.db.Save(t) + a.markAssetDone(t) + // The credit pays for the successful media download, not the later WebDAV sync. + // Commit before upload so a storage outage cannot turn a successful download into a free retry. + if err := a.commitCredit(t); err != nil { + a.fail(t, err) + return + } a.upload(t) } diff --git a/handlers.go b/handlers.go index 265931e..325a420 100644 --- a/handlers.go +++ b/handlers.go @@ -125,8 +125,21 @@ func (a *App) retryTask(c *gin.Context) { c.Status(404) return } - a.db.Model(&t).Updates(map[string]any{"status": "pending", "error": "", "progress": 0, "downloaded_bytes": 0}) - t.Status, t.Error, t.Progress, t.DownloadedBytes = "pending", "", 0, 0 + if t.Status != "error" { + c.JSON(409, gin.H{"error": "only failed tasks can be retried"}) + return + } + r, reserveErr := a.reserveCredit(user.ID, t.ID, 1) + if reserveErr != nil { + c.JSON(402, gin.H{"error": reserveErr.Error()}) + return + } + if err := a.db.Model(&t).Updates(map[string]any{"status": "pending", "error": "", "progress": 0, "downloaded_bytes": 0, "credit_reservation_id": r.ID}).Error; err != nil { + _ = a.refundCredit(&Task{ID: t.ID, UserID: user.ID, CreditReservationID: r.ID}) + c.JSON(500, gin.H{"error": err.Error()}) + return + } + t.Status, t.Error, t.Progress, t.DownloadedBytes, t.CreditReservationID = "pending", "", 0, 0, r.ID a.queue <- t.ID c.JSON(202, t) } @@ -138,7 +151,18 @@ func (a *App) deleteTask(c *gin.Context) { return } user := c.MustGet("user").(*User) - if err := a.db.Where("id = ? AND user_id = ?", id, user.ID).Delete(&Task{}).Error; err != nil { + var t Task + if err := a.db.Where("id = ? AND user_id = ?", id, user.ID).First(&t).Error; err != nil { + c.Status(404) + return + } + if t.CreditReservationID != 0 { + if err := a.refundCredit(&t); err != nil { + c.JSON(409, gin.H{"error": "cannot release task credit: " + err.Error()}) + return + } + } + if err := a.db.Delete(&t).Error; err != nil { c.JSON(500, gin.H{"error": err.Error()}) return } diff --git a/main.go b/main.go index 31c6845..c848182 100644 --- a/main.go +++ b/main.go @@ -15,9 +15,12 @@ func main() { _ = os.MkdirAll(filepath.Dir(dbPath), 0755) db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) must(err) - must(db.AutoMigrate(&Task{}, &Setting{}, &User{}, &UserPermission{}, &CreditAccount{}, &CreditTransaction{}, &CreditReservation{}, &RedeemBatch{}, &RedeemCode{}, &RedeemRecord{}, &WebDAVConfig{}, &Session{}, &AuditLog{})) - // Replace the legacy global task uniqueness constraint with per-user uniqueness. + must(db.AutoMigrate(&Task{}, &MediaAsset{}, &Setting{}, &User{}, &UserPermission{}, &CreditAccount{}, &CreditTransaction{}, &CreditReservation{}, &RedeemBatch{}, &RedeemCode{}, &RedeemRecord{}, &WebDAVConfig{}, &Session{}, &AuditLog{})) + // Replace legacy global task/reservation uniqueness constraints with scoped indexes. must(db.Exec("DROP INDEX IF EXISTS idx_tasks_source_rid_quality").Error) + must(db.Exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_user_source_rid_quality ON tasks(user_id, source, r_id, quality)").Error) + must(db.Exec("DROP INDEX IF EXISTS idx_credit_reservations_task_id").Error) + must(db.Exec("CREATE INDEX IF NOT EXISTS idx_credit_reservations_task_id ON credit_reservations(task_id)").Error) client := &http.Client{Timeout: 2 * 60 * 1e9} cfg := loadConfig(db) diff --git a/models.go b/models.go index 9022de2..04beb1f 100644 --- a/models.go +++ b/models.go @@ -5,6 +5,7 @@ import "time" type Task struct { ID uint `gorm:"primaryKey" json:"id"` UserID uint `gorm:"uniqueIndex:idx_tasks_user_source_rid_quality" json:"userId"` + AssetID uint `gorm:"index" json:"assetId,omitempty"` CreditReservationID uint `gorm:"index" json:"creditReservationId,omitempty"` Source string `gorm:"uniqueIndex:idx_tasks_user_source_rid_quality" json:"source"` RID string `gorm:"uniqueIndex:idx_tasks_user_source_rid_quality" json:"rid"` diff --git a/multiuser_models.go b/multiuser_models.go index b0daab5..fc9a66f 100644 --- a/multiuser_models.go +++ b/multiuser_models.go @@ -42,7 +42,7 @@ type CreditTransaction struct { type CreditReservation struct { ID uint `gorm:"primaryKey"` UserID uint `gorm:"index" json:"userId"` - TaskID uint `gorm:"uniqueIndex" json:"taskId"` + TaskID uint `gorm:"index" json:"taskId"` Amount int64 `json:"amount"` Status string `gorm:"index;size:20" json:"status"` CreatedAt time.Time `json:"createdAt"` diff --git a/redeem.go b/redeem.go index 912f997..1fa2a22 100644 --- a/redeem.go +++ b/redeem.go @@ -66,6 +66,7 @@ func (a *App) adminCreateRedeemBatch(c *gin.Context) { c.JSON(500, gin.H{"error": err.Error()}) return } + a.audit(admin.ID, "redeem.batch.create", "redeem_batch", fmt.Sprint(batch.ID), fmt.Sprintf("credits=%d,count=%d", batch.Credits, batch.Count)) c.JSON(201, gin.H{"batch": batch, "codes": codes}) } @@ -126,6 +127,7 @@ func (a *App) redeem(c *gin.Context) { c.JSON(status, gin.H{"error": err.Error()}) return } + a.audit(user.ID, "redeem.use", "redeem_code", "hash:"+hashRedeemCode(in.Code)[:12], fmt.Sprintf("credits=%d", amount)) c.JSON(200, gin.H{"credits": amount}) } @@ -135,6 +137,8 @@ func (a *App) adminRevokeRedeemBatch(c *gin.Context) { c.JSON(500, gin.H{"error": err.Error()}) return } + admin := c.MustGet("user").(*User) + a.audit(admin.ID, "redeem.batch.revoke", "redeem_batch", id, "unused codes revoked") c.JSON(200, gin.H{"ok": true}) } diff --git a/util.go b/util.go index 49b4119..643d52c 100644 --- a/util.go +++ b/util.go @@ -13,6 +13,7 @@ func (a *App) fail(t *Task, err error) { if t.LocalPath != "" { _ = os.Remove(t.LocalPath + ".part") } + a.releaseAsset(t) if err == nil { err = errors.New("unknown download error") }