package main import ( "fmt" "io" "net/http" "os" "path/filepath" "strings" ) func (a *App) startWorkers() { for i := 0; i < a.cfg.WorkerCount; i++ { a.wg.Add(1) go func() { defer a.wg.Done() for id := range a.queue { a.download(id) } }() } } func (a *App) download(id uint) { var t Task if a.db.First(&t, id).Error != nil || t.Status != "pending" { return } a.db.Model(&t).Updates(map[string]any{"status": "resolving", "error": "", "progress": 0}) info, err := a.am.Resolve(t.Source, t.RID, t.Quality) if err != nil { a.fail(&t, err) return } ext := normalizedFormat(info.Format) if ext == "" { ext = normalizedFormat(filepath.Ext(info.URL)) } 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 } } a.downloadFile(&t, info) } func (a *App) finishExisting(t *Task, size int64) { actual, err := detectAudioFormat(t.LocalPath) if err == nil { actual = normalizedFormat(actual) if actual != t.Format { if renamed, renameErr := renameToActualFormat(t.LocalPath, actual); renameErr == nil { t.LocalPath = renamed } else { a.fail(t, renameErr) return } } t.Format = actual t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual) } t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = size, size, 100, "done" a.db.Save(t) // Existing local media is reused and therefore does not consume a credit. if err := a.refundCredit(t); err != nil { a.fail(t, err) return } a.upload(t) } func (a *App) downloadFile(t *Task, info playInfo) { a.db.Model(t).Updates(map[string]any{"status": "downloading", "format": t.Format, "local_path": t.LocalPath}) req, err := http.NewRequest("GET", info.URL, nil) if err != nil { a.fail(t, err) return } for k, v := range info.Headers { req.Header.Set(k, v) } resp, err := a.httpClient.Do(req) if err != nil { a.fail(t, err) return } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { a.fail(t, fmt.Errorf("media http %d", resp.StatusCode)) return } t.Status, t.TotalBytes, t.DownloadedBytes, t.Progress = "downloading", resp.ContentLength, 0, 0 a.db.Save(t) tmp := t.LocalPath + ".part" f, err := os.Create(tmp) if err != nil { a.fail(t, err) return } n, err := a.copyWithProgress(f, resp.Body, t) closeErr := f.Close() if err != nil { a.fail(t, err) return } if closeErr != nil { a.fail(t, closeErr) return } if t.TotalBytes > 0 && n != t.TotalBytes { a.fail(t, fmt.Errorf("incomplete download: got %d of %d bytes", n, t.TotalBytes)) return } actual, err := detectAudioFormat(tmp) if err != nil { a.fail(t, fmt.Errorf("detect audio format: %w", err)) _ = os.Remove(tmp) return } actual = normalizedFormat(actual) finalPath := strings.TrimSuffix(t.LocalPath, filepath.Ext(t.LocalPath)) + "." + actual if err = os.Rename(tmp, finalPath); err != nil { a.fail(t, err) _ = os.Remove(tmp) return } t.Format, t.LocalPath, t.ActualQuality, t.DownloadedBytes = actual, finalPath, actualQuality(t.Quality, t.ResolvedQuality, actual), n if t.TotalBytes > 0 { t.Progress = 100 } a.db.Save(t) a.upload(t) } func (a *App) copyWithProgress(dst *os.File, src io.Reader, t *Task) (int64, error) { var n, last int64 buf := make([]byte, 256*1024) for { nr, er := src.Read(buf) if nr > 0 { wn, ew := dst.Write(buf[:nr]) n += int64(wn) if ew != nil { return n, ew } if n-last >= 1024*1024 { progress := 0.0 if t.TotalBytes > 0 { progress = float64(n) * 100 / float64(t.TotalBytes) } t.DownloadedBytes, t.Progress = n, progress a.db.Model(t).Updates(map[string]any{"downloaded_bytes": n, "progress": progress}) last = n } } if er == io.EOF { return n, nil } if er != nil { return n, er } } }