navidrome/core/artwork/worker.go

package artwork

import (
	"bytes"
	"cmp"
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"math/rand/v2"
	"sync"
	"time"

	"github.com/navidrome/navidrome/conf"
	"github.com/navidrome/navidrome/core/agents"
	"github.com/navidrome/navidrome/core/auth"
	"github.com/navidrome/navidrome/core/ffmpeg"
	"github.com/navidrome/navidrome/log"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/server/events"
	"github.com/navidrome/navidrome/utils/cache"
)

const (
	workerPollInterval = 5 * time.Second
	backoffBase        = 5 * time.Second
	// giveUpAfter bounds the retry budget from enqueue; past it the item settles and only an
	// explicit reprocess retries it.
	giveUpAfter = 12 * time.Hour
	// itemTimeout must outlast a slow but healthy item: each image agent's call plus its download, in turn.
	itemTimeout = 3 * time.Minute
)

// drainPool drains one class of work with its own slot budget, so a blocking kind cannot
// occupy slots another kind needs.
type drainPool struct {
	name        string
	kinds       []string
	concurrency int
}

// Worker drains the artwork queue: each external agent is rate-limited and circuit-broken
// independently, and pruneMu serializes prune against the store-write window.
type Worker struct {
	proc    *processor
	cache   cache.FileCache
	ffmpeg  ffmpeg.FFmpeg
	broker  events.Broker
	pruneMu sync.RWMutex
	pools   []*drainPool
	runCtx  context.Context //nolint:containedctx // worker lifecycle ctx, set at Run
	paused  func() bool

	gatesMu sync.Mutex
	gates   map[string]*extGate

	// busy holds the itemKeys still being acquired, including ones a drain stopped waiting for.
	busy sync.Map
}

type itemKey struct{ kind, id, imageType string }

func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker, imgCache cache.FileCache) *Worker {
	w := &Worker{
		proc:   &processor{ds: ds, store: store},
		cache:  imgCache,
		ffmpeg: ffmpeg,
		broker: broker,
		pools:  newDrainPools(),
		runCtx: context.Background(),
		paused: func() bool { return false },
		gates:  map[string]*extGate{},
	}
	w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate)
	w.proc.pruneLock = w.pruneMu.RLocker()
	return w
}

// newDrainPools splits the drain by what bounds it: gate() holds a slot while waiting for its
// rate-limit permit, so a sleeping lookup would crowd out a cover sitting on disk.
func newDrainPools() []*drainPool {
	budget := conf.MaxOpenConns() // floored at 4, so both remainders below stay positive
	local := min(max(1, conf.Server.DevArtworkWorkerConcurrency), budget-1)
	// More external slots than the rate allows would only sleep in the limiter.
	external := min(max(2, 2*conf.Server.DevArtworkExternalMaxRPS), budget-local)
	return []*drainPool{
		{name: "local", kinds: localDrainKinds, concurrency: local},
		{name: "external", kinds: externalDrainKinds, concurrency: external},
	}
}

// Kind is a proxy for cost: an album that reaches an external agent still costs a local slot.
var (
	externalDrainKinds = []string{model.KindArtistArtwork.Prefix()}
	localDrainKinds    = []string{
		model.KindAlbumArtwork.Prefix(),
		model.KindPlaylistArtwork.Prefix(),
		model.KindRadioArtwork.Prefix(),
		model.KindMediaFileArtwork.Prefix(),
	}
)

// PauseWhile holds off queue draining whenever paused reports true. Call it before Run.
func (w *Worker) PauseWhile(paused func() bool) {
	w.paused = paused
}

// Run blocks draining the queue until ctx is cancelled.
func (w *Worker) Run(ctx context.Context) error {
	w.runCtx = ctx
	var wg sync.WaitGroup
	for _, p := range w.pools {
		wg.Go(func() { w.runPool(ctx, p) })
	}
	wg.Wait()
	return nil
}

func (w *Worker) runPool(ctx context.Context, p *drainPool) {
	ticker := time.NewTicker(workerPollInterval)
	defer ticker.Stop()
	for {
		n, err := w.drain(ctx, p.concurrency, p.kinds...)
		if err != nil && ctx.Err() == nil {
			log.Warn(ctx, "Artwork: Worker drain failed", "pool", p.name, err)
		}
		if ctx.Err() != nil {
			return
		}
		if n > 0 {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// RunPrune runs prune under the worker's write lock, so no acquisition can place
// a file while orphans are being reclaimed. This is the only sanctioned prune path.
func (w *Worker) RunPrune(ctx context.Context) error {
	w.pruneMu.Lock()
	defer w.pruneMu.Unlock()
	return prune(ctx, w.proc.ds, w.proc.store)
}

// ReconcileConfig records the artwork config fingerprint, or warns when it changed.
func (w *Worker) ReconcileConfig(ctx context.Context) error {
	return ReconcileConfigFingerprint(ctx, w.proc.ds)
}

// EnqueueMissingAll requeues entities with no artwork state row: the safety net for anything
// a scan never enqueued.
func (w *Worker) EnqueueMissingAll(ctx context.Context) error {
	return enqueueMissingAll(ctx, w.proc.ds)
}

func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (int, error) {
	if w.paused() {
		return 0, nil
	}
	// Dequeue well past the pool size so a slow external lookup never idles the other slots.
	// DequeueBatch does not mark rows taken, so this is one query per pass, not per slot.
	items, err := w.proc.ds.ArtworkQueue().DequeueBatch(ctx, max(16, 4*concurrency), kinds...)
	if err != nil {
		return 0, err
	}
	if len(items) == 0 {
		return 0, nil
	}
	drainStart := time.Now()
	// Private playlists need an admin identity; resolved per drain because the worker can
	// start before any admin exists.
	ctx = auth.WithAdminUser(ctx, w.proc.ds)
	sem := make(chan struct{}, concurrency)
	var wg sync.WaitGroup
	var refreshMu sync.Mutex
	var refresh []model.ArtworkQueueItem
	for _, item := range items {
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
		}
		// select picks randomly when both cases are ready, so re-check to never dispatch after cancellation.
		if ctx.Err() != nil {
			wg.Wait()
			return len(items), nil //nolint:nilerr // a cancelled drain is a clean stop, not an error
		}
		if w.paused() {
			break
		}
		wg.Go(func() {
			defer func() { <-sem }()
			out, got := w.process(ctx, item)
			// Absent counts as a visible change too: clients must drop a previously-served
			// immutable cover.
			if out == outcomeFound || out == outcomeFoundStale || out == outcomeAbsent {
				refreshMu.Lock()
				refresh = append(refresh, item)
				refreshMu.Unlock()
			}
			// Post-outcome: the queue row is already settled, so warming the cache can't
			// block or alter queue ops.
			if got != nil {
				w.precache(ctx, got)
			}
		})
	}
	wg.Wait()
	w.broadcastRefresh(ctx, refresh)
	log.Debug(ctx, "Artwork: Drained a batch", "kinds", kinds, "items", len(items),
		"refreshed", len(refresh), "concurrency", concurrency, "elapsed", time.Since(drainStart))
	return len(items), nil
}

// artworkKindToResource maps a kind to its UI resource name; media_file maps to "song", so
// this can't derive from Kind.String().
var artworkKindToResource = map[model.Kind]string{
	model.KindAlbumArtwork:     "album",
	model.KindArtistArtwork:    "artist",
	model.KindPlaylistArtwork:  "playlist",
	model.KindRadioArtwork:     "radio",
	model.KindMediaFileArtwork: "song",
}

// broadcastRefresh emits one coalesced RefreshResource for the batch, so UIs re-fetch the
// affected records and pick up the new coverArt id.
func (w *Worker) broadcastRefresh(ctx context.Context, found []model.ArtworkQueueItem) {
	if len(found) == 0 {
		return
	}
	event := &events.RefreshResource{}
	byResource := map[string][]string{}
	for _, it := range found {
		kind, _ := model.ParseKind(it.ItemKind)
		if res, ok := artworkKindToResource[kind]; ok {
			byResource[res] = append(byResource[res], it.ItemID)
		}
	}
	if len(byResource) == 0 {
		return
	}
	for res, ids := range byResource {
		event = event.With(res, ids...)
	}
	w.broker.SendBroadcastMessage(ctx, event)
}

func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired) {
	item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary)
	ctx = withTrace(ctx, &ChainTrace{})
	out, got, retryIn := w.acquireWithTimeout(ctx, item)
	if ctx.Err() != nil {
		return outcomeFailed, nil // shutting down: leave the row for the next run
	}
	w.settle(ctx, item, out, retryIn)
	return out, got
}

// acquireWithTimeout stops waiting on a stuck item, which Go can't stop, and never runs one item twice at once.
func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired, time.Duration) {
	key := itemKey{item.ItemKind, item.ItemID, item.ImageType}
	if _, running := w.busy.LoadOrStore(key, struct{}{}); running {
		log.Debug(ctx, "Artwork: Item still running from an earlier drain", "kind", item.ItemKind, "id", item.ItemID)
		traceStage(ctx, "worker", errors.New("previous attempt still running"))
		return outcomeFailed, nil, 0
	}
	type result struct {
		out     outcome
		got     *acquired
		retryIn time.Duration
	}
	done := make(chan result, 1)
	go func() {
		// The deadline stops work that honors ctx; the select below covers work that doesn't.
		actx, cancel := context.WithTimeout(ctx, itemTimeout)
		defer cancel()
		out, got, retryIn := w.proc.acquire(actx, item)
		// Cleared before the send, so the next drain never sees a finished item as busy.
		w.busy.Delete(key)
		done <- result{out, got, retryIn}
	}()
	select {
	case r := <-done:
		return r.out, r.got, r.retryIn
	case <-ctx.Done():
		return outcomeFailed, nil, 0
	case <-time.After(itemTimeout):
		log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout)
		traceStage(ctx, "worker", fmt.Errorf("timed out after %s", itemTimeout))
		return outcomeFailed, nil, 0
	}
}

// settle deletes or reschedules the queue row for an acquire outcome.
func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration) {
	queue := w.proc.ds.ArtworkQueue()
	switch out {
	case outcomeFound, outcomeAbsent:
		// A scan that re-enqueued this row mid-flight reset its retry_at, so the row survives
		// here and the next drain re-resolves it.
		if err := queue.DeleteIfUnchanged(ctx, item.ItemKind, item.ItemID, item.ImageType, item.RetryAt); err != nil {
			log.Warn(ctx, "Artwork: Could not delete processed queue item", "kind", item.ItemKind, "id", item.ItemID, err)
		}
	case outcomeFoundStale, outcomeFailed:
		retryAt := time.Now().Add(retryDelay(item.Attempts, retryIn))
		encoded := traceFrom(ctx).encode("")
		if retryAt.Before(item.EnqueuedAt.Add(giveUpAfter)) {
			// A mid-flight re-enqueue reset retry_at; stale backoff must not stomp its
			// fresh, immediate eligibility.
			if err := queue.MarkFailedIfUnchanged(ctx, item.ItemKind, item.ItemID, item.ImageType, item.RetryAt, retryAt, encoded); err != nil {
				log.Warn(ctx, "Artwork: Could not reschedule failed queue item", "kind", item.ItemKind, "id", item.ItemID, err)
			}
			log.Debug(ctx, "Artwork: Rescheduled item", "kind", item.ItemKind, "id", item.ItemID,
				"outcome", out, "attempts", item.Attempts+1, "retryIn", time.Until(retryAt),
				"budgetLeft", time.Until(item.EnqueuedAt.Add(giveUpAfter)))
			break
		}
		// Art already being served is kept: exhaustion means unreachable, not removed.
		settled := "kept previous state"
		if out == outcomeFailed && settlesAbsentOnGiveUp(item.ItemKind) && !w.hasResolvedArtwork(ctx, item) {
			writeAbsent(ctx, w.proc.ds.Artwork(), item)
			settled = "recorded absent"
		}
		// The queue row is about to go, taking the only record of the failure with it. This write is
		// unconditional (not CAS-guarded) — safe only because the drain resolves each item serially.
		w.recordGiveUp(ctx, item, encoded)
		log.Info(ctx, "Artwork: Retry budget exhausted, giving up", "kind", item.ItemKind, "id", item.ItemID,
			"outcome", out, "attempts", item.Attempts+1, "budget", giveUpAfter, "settled", settled)
		if err := queue.DeleteIfUnchanged(ctx, item.ItemKind, item.ItemID, item.ImageType, item.RetryAt); err != nil {
			log.Warn(ctx, "Artwork: Could not remove exhausted queue item", "kind", item.ItemKind, "id", item.ItemID, err)
		}
	}
}

// recordGiveUp keeps the last failure on the state row after the queue row is deleted. An item
// that never resolved has no row to update, and creating one would settle it absent.
func (w *Worker) recordGiveUp(ctx context.Context, item model.ArtworkQueueItem, trace string) {
	kind, ok := model.ParseKind(item.ItemKind)
	if !ok {
		return
	}
	if err := w.proc.ds.Artwork().PutLastFailure(ctx, kind, item.ItemID, item.ImageType, trace); err != nil {
		log.Warn(ctx, "Artwork: Could not record the last failure", "kind", item.ItemKind, "id", item.ItemID, err)
	}
}

func (w *Worker) hasResolvedArtwork(ctx context.Context, item model.ArtworkQueueItem) bool {
	kind, ok := model.ParseKind(item.ItemKind)
	if !ok {
		return false
	}
	ia, err := w.proc.ds.Artwork().GetItemArtwork(ctx, kind, item.ItemID, item.ImageType)
	return err == nil && ia.Hash != ""
}

// precache warms the resize cache at the UI cover size from the bytes just acquired, so the
// first UI request hits without re-reading the rows or the file.
func (w *Worker) precache(ctx context.Context, got *acquired) {
	if !conf.Server.EnableArtworkPrecache || w.cache == nil || w.cache.Disabled(ctx) {
		return
	}
	precacheStart := time.Now()
	// Same key as the serving path: square must match what the list surfaces request, or this
	// warms a key nothing reads.
	item := &resizedItem{
		hash:   got.ia.Hash,
		size:   conf.Server.UICoverArtSize,
		square: true,
		ffmpeg: w.ffmpeg,
		open:   func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(got.data)), nil },
	}
	stream, err := w.cache.Get(ctx, item)
	if err != nil {
		log.Debug(ctx, "Artwork: Precache failed", "kind", got.ia.ItemKind, "id", got.ia.ItemID, err)
		return
	}
	_, _ = io.Copy(io.Discard, stream)
	_ = stream.Close()
	log.Trace(ctx, "Artwork: Precached UI size", "kind", got.ia.ItemKind, "id", got.ia.ItemID,
		"size", conf.Server.UICoverArtSize, "elapsed", time.Since(precacheStart))
}

// backoffFor returns min(5s×4^n, giveUpAfter) scaled by (1+jitter), with jitter in [-0.4, 0.4].
func backoffFor(attempts int, jitter float64) time.Duration {
	d := min(float64(backoffBase)*math.Pow(4, float64(attempts)), float64(giveUpAfter))
	return time.Duration(d * (1 + jitter))
}

func backoff(attempts int) time.Duration {
	return backoffFor(attempts, rand.Float64()*0.8-0.4) //nolint:gosec // retry jitter, not security-sensitive
}

// retryDelay is how long a failed item waits: our backoff, unless the provider asked for longer.
func retryDelay(attempts int, hint time.Duration) time.Duration {
	return max(backoff(attempts), hint)
}