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 { if current.UpdatedAt.After(time.Now().Add(-10 * time.Minute)) { 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 }