diff --git a/internal/app/app.go b/internal/app/app.go index 152f1e2..a4a9162 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -196,6 +196,84 @@ func (m *Main) markDownloaded(ctx context.Context, source, id string) error { return m.Store.MarkDownloaded(writeCtx, source, id) } +// defaultMaxWorkers is used when max_connections is unset or nonsensical. +const defaultMaxWorkers = 6 + +// runTrackWorkers rips ids, either serially or across max_connections workers +// depending on config, and returns how many failed. +// +// This replaces three copy-pasted dispatch blocks that had drifted apart: two +// counted failures under a mutex and logged inline, while the third collected +// errors in a channel and logged them all after wg.Wait() without the track id. +// The inline form is kept, since it reports failures as they happen and names +// the track. pos is 1-based to match the progress output's numbering. +func (m *Main) runTrackWorkers(ctx context.Context, ids []string, rip func(pos int, id string) error) int { + failures := 0 + + maxWorkers := m.Config.Session.Downloads.MaxConnections + if !m.Config.Session.Downloads.Concurrency || maxWorkers == 1 { + for i, id := range ids { + if ctx.Err() != nil { + break + } + if err := rip(i+1, id); err != nil { + failures++ + m.logf("track failed: id=%s reason=%v\n", id, err) + } + } + return failures + } + if maxWorkers <= 0 { + maxWorkers = defaultMaxWorkers + } + + sem := make(chan struct{}, maxWorkers) + var wg sync.WaitGroup + var mu sync.Mutex + for i, id := range ids { + if ctx.Err() != nil { + break + } + wg.Add(1) + sem <- struct{}{} + go func(pos int, tid string) { + defer wg.Done() + defer func() { <-sem }() + if err := rip(pos, tid); err != nil { + // Holding mu across the log keeps failure lines from + // interleaving with each other. + mu.Lock() + failures++ + m.logf("track failed: id=%s reason=%v\n", tid, err) + mu.Unlock() + } + }(i+1, id) + } + wg.Wait() + return failures +} + +// anySlice normalizes a JSON array that may have been decoded as either +// []any or []map[string]any, which differs per provider because some clients +// hand-build their metadata maps instead of round-tripping through encoding/json. +// +// It reports false when the value is neither, so callers can distinguish +// "missing" from "empty". +func anySlice(v any) ([]any, bool) { + switch items := v.(type) { + case []any: + return items, true + case []map[string]any: + out := make([]any, 0, len(items)) + for _, item := range items { + out = append(out, item) + } + return out, true + default: + return nil, false + } +} + // safeJoin joins base with parts and asserts the result stays inside the // configured downloads folder. // @@ -387,15 +465,8 @@ func (m *Main) ripTrackCollection(ctx context.Context, p provider.Client, source if !ok { return fmt.Errorf("%s missing tracks data", strings.ToLower(kind)) } - rawItems := make([]any, 0) - switch items := tracksMap["items"].(type) { - case []any: - rawItems = items - case []map[string]any: - for _, item := range items { - rawItems = append(rawItems, item) - } - default: + rawItems, ok := anySlice(tracksMap["items"]) + if !ok { return fmt.Errorf("%s tracks missing items", strings.ToLower(kind)) } ids := make([]string, 0, len(rawItems)) @@ -409,59 +480,15 @@ func (m *Main) ripTrackCollection(ctx context.Context, p provider.Client, source } } m.logf("%s: %s (%d tracks)\n", kind, name, len(ids)) - failures := 0 - runOne := func(i int, trackID string) { - opts := ripTrackOptions{albumFolder: folder, index: i, total: len(ids)} + failures := m.runTrackWorkers(ctx, ids, func(pos int, trackID string) error { + opts := ripTrackOptions{albumFolder: folder, index: pos, total: len(ids)} if playlistLike { opts.forPlaylist = true opts.playlistName = name - opts.playlistPos = i + opts.playlistPos = pos } - if err := m.ripTrack(ctx, p, source, trackID, "", opts); err != nil { - failures++ - m.logf("track failed: id=%s reason=%v\n", trackID, err) - } - } - if !m.Config.Session.Downloads.Concurrency || m.Config.Session.Downloads.MaxConnections == 1 { - for i, trackID := range ids { - if ctx.Err() != nil { - break - } - runOne(i+1, trackID) - } - } else { - maxWorkers := m.Config.Session.Downloads.MaxConnections - if maxWorkers <= 0 { - maxWorkers = 6 - } - sem := make(chan struct{}, maxWorkers) - var wg sync.WaitGroup - var mu sync.Mutex - for i, trackID := range ids { - if ctx.Err() != nil { - break - } - wg.Add(1) - sem <- struct{}{} - go func(pos int, tid string) { - defer wg.Done() - defer func() { <-sem }() - opts := ripTrackOptions{albumFolder: folder, index: pos, total: len(ids)} - if playlistLike { - opts.forPlaylist = true - opts.playlistName = name - opts.playlistPos = pos - } - if err := m.ripTrack(ctx, p, source, tid, "", opts); err != nil { - mu.Lock() - failures++ - m.logf("track failed: id=%s reason=%v\n", tid, err) - mu.Unlock() - } - }(i+1, trackID) - } - wg.Wait() - } + return m.ripTrack(ctx, p, source, trackID, "", opts) + }) if failures > 0 { m.logf("%s done with %d failed track(s)\n", kind, failures) } @@ -606,23 +633,11 @@ func extractAlbumIDs(meta map[string]any) []string { return nil } - items := make([]any, 0) - switch a := albumsObj.(type) { - case map[string]any: - switch v := a["items"].(type) { - case []any: - items = v - case []map[string]any: - for _, it := range v { - items = append(items, it) - } - } - case []any: - items = a - case []map[string]any: - for _, it := range a { - items = append(items, it) - } + var items []any + if nested, isMap := albumsObj.(map[string]any); isMap { + items, _ = anySlice(nested["items"]) + } else { + items, _ = anySlice(albumsObj) } out := make([]string, 0, len(items)) @@ -715,14 +730,8 @@ func (m *Main) ripAlbum(ctx context.Context, p provider.Client, source, albumID if !ok { return fmt.Errorf("album missing tracks data") } - rawItems := make([]any, 0) - if itemsAny, ok := tracksMap["items"].([]any); ok { - rawItems = itemsAny - } else if itemsMap, ok := tracksMap["items"].([]map[string]any); ok { - for _, item := range itemsMap { - rawItems = append(rawItems, item) - } - } else { + rawItems, ok := anySlice(tracksMap["items"]) + if !ok { return fmt.Errorf("album tracks missing items") } trackIDs := make([]string, 0, len(rawItems)) @@ -752,59 +761,22 @@ func (m *Main) ripAlbum(ctx context.Context, p provider.Client, source, albumID discTotal = jsonutil.IntFromAny(albumMeta["numberOfVolumes"]) } m.logf("Album: %s (%d tracks)\n", albumTitle, total) - failures := 0 - - if !m.Config.Session.Downloads.Concurrency || m.Config.Session.Downloads.MaxConnections == 1 { - for i, trackID := range trackIDs { - if ctx.Err() != nil { - break - } - opts := ripTrackOptions{albumFolder: folder, albumEmbedCover: artRes.EmbedPath, albumArtist: albumArtist, index: i + 1, total: total, albumDiscTotal: discTotal} - if i == 0 { - opts.prefetched = prefetched - } - if err := m.ripTrack(ctx, p, source, trackID, "", opts); err != nil { - failures++ - m.logf("track failed: id=%s reason=%v\n", trackID, err) - } + failures := m.runTrackWorkers(ctx, trackIDs, func(pos int, trackID string) error { + opts := ripTrackOptions{ + albumFolder: folder, + albumEmbedCover: artRes.EmbedPath, + albumArtist: albumArtist, + index: pos, + total: total, + albumDiscTotal: discTotal, } - if failures > 0 { - m.logf("Album done with %d failed track(s)\n", failures) + // The first track's downloadable was already fetched to decide the + // album folder's audio values, so reuse it instead of refetching. + if pos == 1 { + opts.prefetched = prefetched } - return nil - } - - maxWorkers := m.Config.Session.Downloads.MaxConnections - if maxWorkers <= 0 { - maxWorkers = 6 - } - sem := make(chan struct{}, maxWorkers) - var wg sync.WaitGroup - errCh := make(chan error, len(trackIDs)) - for i, trackID := range trackIDs { - if ctx.Err() != nil { - break - } - wg.Add(1) - sem <- struct{}{} - go func(idx int, tid string) { - defer wg.Done() - defer func() { <-sem }() - opts := ripTrackOptions{albumFolder: folder, albumEmbedCover: artRes.EmbedPath, albumArtist: albumArtist, index: idx, total: total, albumDiscTotal: discTotal} - if idx == 1 { - opts.prefetched = prefetched - } - if err := m.ripTrack(ctx, p, source, tid, "", opts); err != nil { - errCh <- err - } - }(i+1, trackID) - } - wg.Wait() - close(errCh) - for err := range errCh { - failures++ - m.logf("track failed: reason=%v\n", err) - } + return m.ripTrack(ctx, p, source, trackID, "", opts) + }) if failures > 0 { m.logf("Album done with %d failed track(s)\n", failures) } @@ -833,16 +805,9 @@ func (m *Main) ripPlaylist(ctx context.Context, p provider.Client, source, playl if !ok { return fmt.Errorf("playlist missing tracks data") } - rawItems, ok := tracksMap["items"].([]any) + rawItems, ok := anySlice(tracksMap["items"]) if !ok { - if itemsMap, ok2 := tracksMap["items"].([]map[string]any); ok2 { - rawItems = make([]any, 0, len(itemsMap)) - for _, it := range itemsMap { - rawItems = append(rawItems, it) - } - } else { - return fmt.Errorf("playlist tracks missing items") - } + return fmt.Errorf("playlist tracks missing items") } ids := make([]string, 0, len(rawItems)) @@ -862,58 +827,17 @@ func (m *Main) ripPlaylist(ctx context.Context, p provider.Client, source, playl total := len(ids) m.logf("Playlist: %s (%d tracks)\n", name, total) - failures := 0 - - runOne := func(i int, id string) { + failures := m.runTrackWorkers(ctx, ids, func(pos int, id string) error { opts := ripTrackOptions{ albumFolder: folder, - index: i, + index: pos, total: total, forPlaylist: true, playlistName: name, - playlistPos: i, + playlistPos: pos, } - if err := m.ripTrack(ctx, p, source, id, "", opts); err != nil { - failures++ - m.logf("track failed: id=%s reason=%v\n", id, err) - } - } - - if !m.Config.Session.Downloads.Concurrency || m.Config.Session.Downloads.MaxConnections == 1 { - for i, id := range ids { - if ctx.Err() != nil { - break - } - runOne(i+1, id) - } - } else { - maxWorkers := m.Config.Session.Downloads.MaxConnections - if maxWorkers <= 0 { - maxWorkers = 6 - } - sem := make(chan struct{}, maxWorkers) - var wg sync.WaitGroup - var mu sync.Mutex - for i, id := range ids { - if ctx.Err() != nil { - break - } - wg.Add(1) - sem <- struct{}{} - go func(pos int, tid string) { - defer wg.Done() - defer func() { <-sem }() - opts := ripTrackOptions{albumFolder: folder, index: pos, total: total, forPlaylist: true, playlistName: name, playlistPos: pos} - if err := m.ripTrack(ctx, p, source, tid, "", opts); err != nil { - mu.Lock() - failures++ - m.logf("track failed: id=%s reason=%v\n", tid, err) - mu.Unlock() - } - }(i+1, id) - } - wg.Wait() - } + return m.ripTrack(ctx, p, source, id, "", opts) + }) if failures > 0 { m.logf("Playlist done with %d failed track(s)\n", failures) diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 7868664..99115cc 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "fmt" "net/http" "net/http/httptest" @@ -9,6 +10,7 @@ import ( "os/exec" "path/filepath" "strings" + "sync" "testing" "streamrip-go/internal/audio/tag" @@ -957,3 +959,182 @@ func TestBuildTagMetadataReplayGainFallsBackToDeezerGain(t *testing.T) { t.Fatalf("track replaygain gain=%q", tags.ReplaygainTrackGain) } } + +// countingAlbumProvider serves a large album and records how many track +// downloads were attempted, so a cancellation test can assert the rip actually +// stopped instead of grinding through the rest of the queue. +type countingAlbumProvider struct { + url string + trackCount int + + mu sync.Mutex + attempts int + cancel context.CancelFunc + cancelAt int +} + +func (f *countingAlbumProvider) Source() string { return "qobuz" } +func (f *countingAlbumProvider) Login(context.Context) error { return nil } +func (f *countingAlbumProvider) LoggedIn() bool { return true } +func (f *countingAlbumProvider) Close() error { return nil } +func (f *countingAlbumProvider) Search(context.Context, string, string, int) ([]map[string]any, error) { + return nil, nil +} + +func (f *countingAlbumProvider) GetMetadata(_ context.Context, id string, mediaType string) (map[string]any, error) { + if mediaType == "album" { + items := make([]any, 0, f.trackCount) + for i := 0; i < f.trackCount; i++ { + items = append(items, map[string]any{"id": fmt.Sprintf("t%d", i)}) + } + return map[string]any{ + "title": "Big Album", + "release_date_original": "1977-02-04", + "artist": map[string]any{"name": "Fleetwood Mac"}, + "tracks": map[string]any{"items": items}, + }, nil + } + + // Count metadata fetches as download attempts and trip the cancel once + // enough tracks have been seen. + f.mu.Lock() + f.attempts++ + n := f.attempts + f.mu.Unlock() + if f.cancel != nil && n == f.cancelAt { + f.cancel() + } + + return map[string]any{ + "title": "Track " + id, + "track_number": float64(1), + "performer": map[string]any{"name": "Fleetwood Mac"}, + "album": map[string]any{ + "title": "Big Album", + "artist": map[string]any{"name": "Fleetwood Mac"}, + }, + }, nil +} + +func (f *countingAlbumProvider) GetDownloadable(_ context.Context, _ string, _ int) (*provider.Downloadable, error) { + return &provider.Downloadable{URL: f.url, Extension: "flac", Source: "qobuz"}, nil +} + +func (f *countingAlbumProvider) attemptCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.attempts +} + +// TestRipStopsOnCancellation is the regression test for Ctrl-C doing nothing. +// ctx was threaded through the rip but never checked, so cancelling left the +// loops running through every remaining track (and the retry ladder retrying +// on a dead context). +func TestRipStopsOnCancellation(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("fLaC-test-bytes")) + })) + defer ts.Close() + + tmp := t.TempDir() + d := config.DefaultConfigData() + d.Downloads.Folder = tmp + d.Downloads.SourceSubdirectories = false + // Serial, so the cancellation point is deterministic. + d.Downloads.Concurrency = false + d.Artwork.Embed = false + d.Artwork.SaveArtwork = false + cfg := &config.Config{File: d, Session: d} + + const total = 40 + const cancelAfter = 5 + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + prov := &countingAlbumProvider{url: ts.URL, trackCount: total, cancel: cancel, cancelAt: cancelAfter} + + m := &Main{ + Config: cfg, + Providers: map[string]provider.Client{"qobuz": prov}, + Store: store.NewDummy(), + DL: download.New(), + Tagger: noopTagger{}, + } + + if err := m.AddByID(context.Background(), "qobuz", "album", "a1"); err != nil { + t.Fatalf("AddByID() error = %v", err) + } + if err := m.Resolve(context.Background()); err != nil { + t.Fatalf("Resolve() error = %v", err) + } + + err := m.Rip(ctx) + if err == nil { + t.Fatal("Rip() returned nil after cancellation, want an error") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("Rip() error = %v, want it to wrap context.Canceled", err) + } + + // The rip must stop promptly. Without the cancellation checks it walked + // all 40 tracks; allow a small margin for the in-flight track. + if got := prov.attemptCount(); got > cancelAfter+2 { + t.Errorf("attempted %d track(s) after cancelling at %d; the rip did not stop", got, cancelAfter) + } +} + +// TestRipTrackReturnsEarlyOnCancellation pins the funnel check: every worker +// path goes through ripTrack, so one guard there stops all of them. +func TestRipTrackReturnsEarlyOnCancellation(t *testing.T) { + tmp := t.TempDir() + d := config.DefaultConfigData() + d.Downloads.Folder = tmp + cfg := &config.Config{File: d, Session: d} + + m := &Main{ + Config: cfg, + Providers: map[string]provider.Client{"qobuz": &fakeProvider{}}, + Store: store.NewDummy(), + DL: download.New(), + Tagger: noopTagger{}, + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := m.ripTrack(ctx, &fakeProvider{}, "qobuz", "t1", "", ripTrackOptions{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("ripTrack() error = %v, want context.Canceled", err) + } +} + +// TestBookkeepingSurvivesCancellation checks that the failure record still +// lands when the rip was cancelled, which is when it matters most. The store +// writes deliberately use a detached context. +func TestBookkeepingSurvivesCancellation(t *testing.T) { + tmp := t.TempDir() + sqlite, err := store.NewSQLite(filepath.Join(tmp, "db.sqlite")) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sqlite.Close() }() + + d := config.DefaultConfigData() + d.Downloads.Folder = tmp + m := &Main{Config: &config.Config{File: d, Session: d}, Store: sqlite, DL: download.New()} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + m.markFailed(ctx, "qobuz", "track", "t1") + + // Read back with a live context. + failed, err := sqlite.IsFailed(context.Background(), "qobuz", "track", "t1") + if err != nil { + t.Fatal(err) + } + if !failed { + t.Fatal("failure record not written under a cancelled context") + } +} diff --git a/internal/download/downloader.go b/internal/download/downloader.go index ae078f1..64d58b8 100644 --- a/internal/download/downloader.go +++ b/internal/download/downloader.go @@ -7,6 +7,7 @@ import ( "crypto/cipher" "crypto/md5" "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -70,11 +71,24 @@ func (d *Downloader) FileVideo(ctx context.Context, sourceURL, outputPath string return d.file(ctx, sourceURL, outputPath, true, true) } -func (d *Downloader) FileDeezerEncrypted(ctx context.Context, sourceURL, outputPath, trackID string) error { +// streamEncrypted performs the shared scaffolding for an encrypted download: +// create the output directory, GET the URL, check the status, create the output +// file, attach a progress bar, clean up a partial file on failure, verify the +// byte count against Content-Length and fsync. +// +// decode does the format-specific work of turning the response body into plain +// audio on dst and returns the number of bytes written. FileDeezerEncrypted and +// FileYandexEncrypted were ~90% identical before this. +func (d *Downloader) streamEncrypted( + ctx context.Context, + sourceURL, outputPath string, + decode func(dst io.Writer, src io.Reader, bar *mpb.Bar) (int64, error), +) error { logDownloadStart(sourceURL, outputPath) if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil { return err } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, sourceURL, nil) if err != nil { return err @@ -87,6 +101,7 @@ func (d *Downloader) FileDeezerEncrypted(ctx context.Context, sourceURL, outputP if resp.StatusCode != http.StatusOK { return fmt.Errorf("download failed: status=%d", resp.StatusCode) } + out, err := os.Create(outputPath) if err != nil { return err @@ -99,97 +114,72 @@ func (d *Downloader) FileDeezerEncrypted(ctx context.Context, sourceURL, outputP } }() - var bar *mpb.Bar - if d.ProgressEnabled() { - d.barStarted.Store(1) - desc := shortenName(filepath.Base(outputPath), 54) - if resp.ContentLength > 0 { - bar = d.progress.AddBar( - resp.ContentLength, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - decor.Percentage(decor.WCSyncWidthR), - ), - mpb.AppendDecorators( - decor.CountersKibiByte("% .1f / % .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.AverageSpeed(decor.SizeB1024(0), "% .1f", decor.WCSyncWidthR), - decor.Name(" | ETA ", decor.WCSyncWidth), - decor.AverageETA(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - } else { - bar = d.progress.AddSpinner( - 0, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - ), - mpb.AppendDecorators( - decor.CurrentKibiByte("% .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.Elapsed(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - defer bar.SetTotal(-1, true) - } - defer func() { - if !success && bar != nil { - bar.Abort(true) - } - }() - } + bar, finishBar := d.newDownloadBar(outputPath, resp.ContentLength) + defer func() { finishBar(success) }() - block, err := blowfish.NewCipher(deriveDeezerBlowfishKey(d.blowfishSecret(), trackID)) + written, err := decode(out, resp.Body, bar) if err != nil { return err } - buf := make([]byte, deezerBFChunkSize) - dec := make([]byte, deezerBFChunkSize) - chunkIndex := 0 - totalRead := int64(0) - for { - n, readErr := io.ReadFull(resp.Body, buf) - if readErr == io.EOF { - break - } - if readErr != nil && readErr != io.ErrUnexpectedEOF { - return readErr - } - chunk := buf[:n] - if chunkIndex%3 == 0 && n == deezerBFChunkSize { - mode := cipher.NewCBCDecrypter(block, deezerBFIV) - mode.CryptBlocks(dec[:n], chunk) - chunk = dec[:n] - } - if _, err = out.Write(chunk); err != nil { - return err - } - totalRead += int64(n) - if bar != nil { - bar.IncrBy(n) - } - chunkIndex++ - if readErr == io.ErrUnexpectedEOF { - break - } - } - if resp.ContentLength > 0 && totalRead != resp.ContentLength { + if resp.ContentLength > 0 && written != resp.ContentLength { return io.ErrUnexpectedEOF } if err = out.Sync(); err != nil { return err } + success = true return nil } -func (d *Downloader) FileYandexEncrypted(ctx context.Context, sourceURL, outputPath, key string) error { - logDownloadStart(sourceURL, outputPath) - if err := os.MkdirAll(filepath.Dir(outputPath), 0o755); err != nil { +// FileDeezerEncrypted downloads a Deezer stream and decrypts it. Deezer +// encrypts every third 2048-byte chunk with Blowfish-CBC; a short final chunk +// is left as-is. +func (d *Downloader) FileDeezerEncrypted(ctx context.Context, sourceURL, outputPath, trackID string) error { + block, err := blowfish.NewCipher(deriveDeezerBlowfishKey(d.blowfishSecret(), trackID)) + if err != nil { return err } + + return d.streamEncrypted(ctx, sourceURL, outputPath, func(dst io.Writer, src io.Reader, bar *mpb.Bar) (int64, error) { + buf := make([]byte, deezerBFChunkSize) + dec := make([]byte, deezerBFChunkSize) + chunkIndex := 0 + total := int64(0) + for { + n, readErr := io.ReadFull(src, buf) + if errors.Is(readErr, io.EOF) { + return total, nil + } + if readErr != nil && !errors.Is(readErr, io.ErrUnexpectedEOF) { + return total, readErr + } + + chunk := buf[:n] + if chunkIndex%3 == 0 && n == deezerBFChunkSize { + mode := cipher.NewCBCDecrypter(block, deezerBFIV) + mode.CryptBlocks(dec[:n], chunk) + chunk = dec[:n] + } + if _, writeErr := dst.Write(chunk); writeErr != nil { + return total, writeErr + } + total += int64(n) + if bar != nil { + bar.IncrBy(n) + } + chunkIndex++ + + if errors.Is(readErr, io.ErrUnexpectedEOF) { + return total, nil + } + } + }) +} + +// FileYandexEncrypted downloads a Yandex stream and decrypts it. Yandex uses +// AES-CTR over the whole body with a zero IV. +func (d *Downloader) FileYandexEncrypted(ctx context.Context, sourceURL, outputPath, key string) error { keyBytes, err := hex.DecodeString(strings.TrimSpace(key)) if err != nil { return fmt.Errorf("invalid yandex key: %w", err) @@ -201,102 +191,11 @@ func (d *Downloader) FileYandexEncrypted(ctx context.Context, sourceURL, outputP if err != nil { return err } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, sourceURL, nil) - if err != nil { - return err - } - resp, err := d.http.Do(req) - if err != nil { - return err - } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("download failed: status=%d", resp.StatusCode) - } - out, err := os.Create(outputPath) - if err != nil { - return err - } - success := false - defer func() { - _ = out.Close() - if !success { - _ = os.Remove(outputPath) - } - }() - var bar *mpb.Bar - if d.ProgressEnabled() { - d.barStarted.Store(1) - desc := shortenName(filepath.Base(outputPath), 54) - if resp.ContentLength > 0 { - bar = d.progress.AddBar( - resp.ContentLength, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - decor.Percentage(decor.WCSyncWidthR), - ), - mpb.AppendDecorators( - decor.CountersKibiByte("% .1f / % .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.AverageSpeed(decor.SizeB1024(0), "% .1f", decor.WCSyncWidthR), - decor.Name(" | ETA ", decor.WCSyncWidth), - decor.AverageETA(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - } else { - bar = d.progress.AddSpinner( - 0, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - ), - mpb.AppendDecorators( - decor.CurrentKibiByte("% .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.Elapsed(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - defer bar.SetTotal(-1, true) - } - defer func() { - if !success && bar != nil { - bar.Abort(true) - } - }() - } - - stream := cipher.NewCTR(block, make([]byte, aes.BlockSize)) - reader := &cipher.StreamReader{S: stream, R: resp.Body} - buf := make([]byte, downloadBufferSize) - totalWritten := int64(0) - for { - n, readErr := reader.Read(buf) - if n > 0 { - if _, writeErr := out.Write(buf[:n]); writeErr != nil { - return writeErr - } - totalWritten += int64(n) - if bar != nil { - bar.IncrBy(n) - } - } - if readErr != nil { - if readErr == io.EOF { - break - } - return readErr - } - } - if resp.ContentLength > 0 && totalWritten != resp.ContentLength { - return io.ErrUnexpectedEOF - } - if err = out.Sync(); err != nil { - return err - } - success = true - return nil + return d.streamEncrypted(ctx, sourceURL, outputPath, func(dst io.Writer, src io.Reader, bar *mpb.Bar) (int64, error) { + stream := cipher.NewCTR(block, make([]byte, aes.BlockSize)) + return copyWithProgress(dst, &cipher.StreamReader{S: stream, R: src}, bar) + }) } func (d *Downloader) file(ctx context.Context, sourceURL, outputPath string, allowProgress bool, includeVideo bool) error { @@ -339,81 +238,22 @@ func (d *Downloader) file(ctx context.Context, sourceURL, outputPath string, all } }() - if d.ProgressEnabled() && allowProgress { - d.barStarted.Store(1) - desc := shortenName(filepath.Base(outputPath), 54) - var bar *mpb.Bar - if resp.ContentLength > 0 { - bar = d.progress.AddBar( - resp.ContentLength, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - decor.Percentage(decor.WCSyncWidthR), - ), - mpb.AppendDecorators( - decor.CountersKibiByte("% .1f / % .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.AverageSpeed(decor.SizeB1024(0), "% .1f", decor.WCSyncWidthR), - decor.Name(" | ETA ", decor.WCSyncWidth), - decor.AverageETA(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - } else { - bar = d.progress.AddSpinner( - 0, - mpb.PrependDecorators( - decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), - ), - mpb.AppendDecorators( - decor.CurrentKibiByte("% .1f", decor.WCSyncWidthR), - decor.Name(" | ", decor.WCSyncWidth), - decor.Elapsed(decor.ET_STYLE_GO, decor.WCSyncWidthR), - ), - mpb.BarRemoveOnComplete(), - ) - defer bar.SetTotal(-1, true) - } - defer func() { - if !success && bar != nil { - bar.Abort(true) - } - }() - buf := make([]byte, downloadBufferSize) - totalWritten := int64(0) - for { - n, readErr := reader.Read(buf) - if n > 0 { - if _, writeErr := out.Write(buf[:n]); writeErr != nil { - return writeErr - } - totalWritten += int64(n) - bar.IncrBy(n) - } - if readErr != nil { - if readErr == io.EOF { - break - } - return readErr - } - } - if resp.ContentLength > 0 && totalWritten != resp.ContentLength { - return io.ErrUnexpectedEOF - } - if err = out.Sync(); err != nil { - return err - } - } else { - written, copyErr := io.CopyBuffer(out, reader, make([]byte, downloadBufferSize)) - if copyErr != nil { - return copyErr - } - if resp.ContentLength > 0 && written != resp.ContentLength { - return io.ErrUnexpectedEOF - } - if err = out.Sync(); err != nil { - return err - } + var bar *mpb.Bar + finishBar := func(bool) {} + if allowProgress { + bar, finishBar = d.newDownloadBar(outputPath, resp.ContentLength) + } + defer func() { finishBar(success) }() + + written, err := copyWithProgress(out, reader, bar) + if err != nil { + return err + } + if resp.ContentLength > 0 && written != resp.ContentLength { + return io.ErrUnexpectedEOF + } + if err = out.Sync(); err != nil { + return err } success = true @@ -439,6 +279,92 @@ func (d *Downloader) Logf(format string, args ...any) { fmt.Print(msg) } +// copyWithProgress copies src to dst, reporting progress to bar when it is +// non-nil. It replaces three hand-rolled read/write loops that differed only +// in whether they touched a progress bar. +func copyWithProgress(dst io.Writer, src io.Reader, bar *mpb.Bar) (int64, error) { + if bar == nil { + return io.CopyBuffer(dst, src, make([]byte, downloadBufferSize)) + } + buf := make([]byte, downloadBufferSize) + total := int64(0) + for { + n, readErr := src.Read(buf) + if n > 0 { + if _, writeErr := dst.Write(buf[:n]); writeErr != nil { + return total, writeErr + } + total += int64(n) + bar.IncrBy(n) + } + if readErr != nil { + if errors.Is(readErr, io.EOF) { + return total, nil + } + return total, readErr + } + } +} + +// newDownloadBar creates the progress bar for a download: a determinate bar +// when the content length is known, a spinner when it is not. It returns a nil +// bar when progress display is off, so callers can use the result +// unconditionally, plus a finish func to defer. +// +// This block was copy-pasted into all three download paths; the decorator +// stacks had already started to drift apart. +func (d *Downloader) newDownloadBar(outputPath string, contentLength int64) (*mpb.Bar, func(success bool)) { + if !d.ProgressEnabled() { + return nil, func(bool) {} + } + d.barStarted.Store(1) + desc := shortenName(filepath.Base(outputPath), 54) + + if contentLength > 0 { + bar := d.progress.AddBar( + contentLength, + mpb.PrependDecorators( + decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), + decor.Percentage(decor.WCSyncWidthR), + ), + mpb.AppendDecorators( + decor.CountersKibiByte("% .1f / % .1f", decor.WCSyncWidthR), + decor.Name(" | ", decor.WCSyncWidth), + decor.AverageSpeed(decor.SizeB1024(0), "% .1f", decor.WCSyncWidthR), + decor.Name(" | ETA ", decor.WCSyncWidth), + decor.AverageETA(decor.ET_STYLE_GO, decor.WCSyncWidthR), + ), + mpb.BarRemoveOnComplete(), + ) + return bar, func(success bool) { + if !success { + bar.Abort(true) + } + } + } + + bar := d.progress.AddSpinner( + 0, + mpb.PrependDecorators( + decor.Name(desc+" ", decor.WC{W: 56, C: decor.DSyncWidth | decor.DindentRight}), + ), + mpb.AppendDecorators( + decor.CurrentKibiByte("% .1f", decor.WCSyncWidthR), + decor.Name(" | ", decor.WCSyncWidth), + decor.Elapsed(decor.ET_STYLE_GO, decor.WCSyncWidthR), + ), + mpb.BarRemoveOnComplete(), + ) + // Order matches the original defers, which ran LIFO: abort first, then + // finalise the spinner total. + return bar, func(success bool) { + if !success { + bar.Abort(true) + } + bar.SetTotal(-1, true) + } +} + // logDownloadStart emits the source URL and destination filename when the // user passed -v or higher. The transport-level logger covers the same // requests at -vv, but this line gives a friendlier per-track summary. diff --git a/internal/download/downloader_test.go b/internal/download/downloader_test.go index c51f052..67bb7bd 100644 --- a/internal/download/downloader_test.go +++ b/internal/download/downloader_test.go @@ -1,6 +1,7 @@ package download import ( + "bytes" "context" "crypto/aes" "crypto/cipher" @@ -12,6 +13,7 @@ import ( "os" "os/exec" "path/filepath" + "strconv" "strings" "testing" "time" @@ -367,3 +369,144 @@ func containsArgPair(args []string, key, value string) bool { } return false } + +// newProgressDownloader builds a Downloader with the progress bars actually +// enabled. The bar code paths are otherwise never exercised, because tests do +// not run on a terminal; STREAMRIP_GO_FORCE_PROGRESS is the documented escape +// hatch. Output goes to stderr, which `go test` captures. +func newProgressDownloader(t *testing.T) *Downloader { + t.Helper() + t.Setenv("STREAMRIP_GO_FORCE_PROGRESS", "1") + d := NewWithOptions(true, true, 0) + if !d.ProgressEnabled() { + t.Fatal("expected progress to be enabled") + } + t.Cleanup(d.Close) + return d +} + +// TestFileWithProgressBar covers the determinate-bar path (Content-Length +// known) through the shared newDownloadBar/copyWithProgress helpers. +func TestFileWithProgressBar(t *testing.T) { + payload := bytes.Repeat([]byte("abcd"), 4096) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(payload))) + _, _ = w.Write(payload) + })) + defer ts.Close() + + d := newProgressDownloader(t) + out := filepath.Join(t.TempDir(), "sub", "a.flac") + if err := d.File(context.Background(), ts.URL, out); err != nil { + t.Fatalf("File() error = %v", err) + } + got, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatalf("content mismatch: got %d bytes, want %d", len(got), len(payload)) + } +} + +// TestFileWithSpinnerUnknownLength covers the spinner path, taken when the +// server does not send a Content-Length. +func TestFileWithSpinnerUnknownLength(t *testing.T) { + payload := bytes.Repeat([]byte("xyz"), 3000) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Transfer-Encoding", "chunked") + _, _ = w.Write(payload) + })) + defer ts.Close() + + d := newProgressDownloader(t) + out := filepath.Join(t.TempDir(), "a.flac") + if err := d.File(context.Background(), ts.URL, out); err != nil { + t.Fatalf("File() error = %v", err) + } + got, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatalf("content mismatch: got %d bytes, want %d", len(got), len(payload)) + } +} + +// TestFileDeezerEncryptedWithProgressBar runs the Deezer decryption through +// the progress-enabled path, so the shared streamEncrypted scaffolding is +// covered with a bar attached as well as without. +func TestFileDeezerEncryptedWithProgressBar(t *testing.T) { + plain := make([]byte, deezerBFChunkSize*4+100) + for i := range plain { + plain[i] = byte((i * 7) % 253) + } + trackID := "123456" + enc := encryptDeezerPayload(t, plain, trackID) + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(enc))) + _, _ = w.Write(enc) + })) + defer ts.Close() + + d := newProgressDownloader(t) + out := filepath.Join(t.TempDir(), "a.mp3") + if err := d.FileDeezerEncrypted(context.Background(), ts.URL, out, trackID); err != nil { + t.Fatalf("FileDeezerEncrypted() error = %v", err) + } + got, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, plain) { + t.Fatal("decrypted output does not match the original plaintext") + } +} + +// encryptDeezerPayload applies Deezer's scheme: Blowfish-CBC on every third +// full 2048-byte chunk, everything else left in the clear. +func encryptDeezerPayload(t *testing.T, plain []byte, trackID string) []byte { + t.Helper() + block, err := blowfish.NewCipher(deriveDeezerBlowfishKey(defaultDeezerBlowfishSecret, trackID)) + if err != nil { + t.Fatal(err) + } + enc := make([]byte, len(plain)) + copy(enc, plain) + for i, off := 0, 0; off < len(plain); i, off = i+1, off+deezerBFChunkSize { + end := off + deezerBFChunkSize + if end > len(plain) { + break + } + if i%3 != 0 { + continue + } + mode := cipher.NewCBCEncrypter(block, deezerBFIV) + mode.CryptBlocks(enc[off:end], plain[off:end]) + } + return enc +} + +// TestProgressBarFailurePathRemovesPartialFile checks that a mid-download +// failure still runs the bar cleanup and removes the partial output, which is +// the behaviour the finishBar(success) defer replaced. +func TestProgressBarFailurePathRemovesPartialFile(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", "100000") + _, _ = w.Write(bytes.Repeat([]byte("a"), 10)) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + })) + defer ts.Close() + + d := newProgressDownloader(t) + out := filepath.Join(t.TempDir(), "a.flac") + if err := d.File(context.Background(), ts.URL, out); err == nil { + t.Fatal("expected a truncated-response error") + } + if _, err := os.Stat(out); !os.IsNotExist(err) { + t.Fatalf("partial file not removed, stat err = %v", err) + } +} diff --git a/internal/store/sqlite.go b/internal/store/sqlite.go index 5fb29bf..5666869 100644 --- a/internal/store/sqlite.go +++ b/internal/store/sqlite.go @@ -300,6 +300,25 @@ func (s *SQLite) MarkFailed(ctx context.Context, source, mediaType, id string) e return err } +// IsFailed reports whether a previous attempt at this item was recorded as +// failed. It complements MarkFailed and is what the retry/repair workflows and +// `rip database browse failed` are built on. +func (s *SQLite) IsFailed(ctx context.Context, source, mediaType, id string) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + + var count int + err := s.db.QueryRowContext( + ctx, + `SELECT COUNT(1) FROM failed_downloads WHERE source = ? AND media_type = ? AND id = ?`, + source, mediaType, id, + ).Scan(&count) + if err != nil { + return false, err + } + return count > 0, nil +} + func (s *SQLite) Close() error { if s.db == nil { return nil