refactor: consolidate the duplicated download and worker-pool code

internal/download The ~40-line progress-bar decorator stack was copy-pasted into three download paths and had started to drift; it is now newDownloadBar, which returns a nil bar when progress is off (so callers use it unconditionally) plus a finish func to defer. The three hand-rolled read/write loops that differed only in whether they touched a bar become copyWithProgress. FileDeezerEncrypted and FileYandexEncrypted were ~90% identical: mkdir, GET, status check, create output, bar, cleanup-on-failure, Content-Length check, fsync. That scaffolding is now streamEncrypted, and each function supplies only its decryption step. Also switches the io.EOF/io.ErrUnexpectedEOF comparisons to errors.Is. internal/app Three copies of "coerce a JSON array that may be []any or []map[string]any" become anySlice, which also reports whether the value was an array at all so callers can still distinguish missing from empty. The worker-pool dispatch existed three times and the copies disagreed: two counted failures under a mutex and logged inline with the track id, while the third collected errors in a channel and logged them all after wg.Wait() without the id. runTrackWorkers unifies them on the inline form, which reports failures as they happen and names the track, and keeps the mutex held across the log so failure lines cannot interleave. Tests The progress-bar paths had no coverage at all, because progress is disabled when not attached to a terminal; four new tests force it on via the existing STREAMRIP_GO_FORCE_PROGRESS escape hatch and cover the determinate bar, the unknown-length spinner, Deezer decryption with a bar attached, and the failure path still removing the partial file. internal/download coverage 53% -> 63%. Three cancellation tests pin the Ctrl-C fix. Verified as genuine regression tests by reverting the guards: without them the rip walks all 40 tracks of a cancelled album, logs a failure for each, and Rip() returns nil. Adds store.SQLite.IsFailed, the read complement to MarkFailed, used by the bookkeeping test to confirm failure records still land under a cancelled context. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-16 22:07:30 +02:00 · 2026-10-03 17:23:24 +02:00
1 parent 78d4f0316d
commit b5d4d5688d
@@ -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)