minstrel/internal/library/leveled.go

package library

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"math"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"
)

// Leveled streams.
//
// Sonos and other UPnP renderers fetch a track's URL themselves, so the phone
// that sent them there cannot change what they play. For them the server
// renders the track with its gain already applied: a FLAC, written to a cache
// directory and served as a plain file, so Content-Length, Range and seeking
// behave as for the original. Tags are stripped; the renderer is told the
// metadata in the DIDL-Lite it is handed, and an embedded ReplayGain tag would
// invite it to adjust a second time.

// The same limits as the web and Android players (web/src/lib/player/gain.ts,
// android .../player/gain/GainMath.kt): headroom mode stops a boost 1 dB under
// the true peak, and no boost exceeds 12 dB.
const (
	leveledPeakCeilingDBTP = -1.0
	leveledMaxBoostDB      = 12.0
	// The cut is bounded too: a gain is a request parameter, and a value no
	// measurement could produce is refused rather than rendered.
	leveledMaxCutDB = -60.0
)

// LeveledGainDB is the gain, in dB, a track gets under prefs: 0 when leveling
// is off or the track has not been measured. asAlbum says whether the track is
// being played as part of its album in order, which only the client that
// holds the queue can know.
func LeveledGainDB(prefs NormalizationPrefs, g ReplayGain, asAlbum bool) float64 {
	if prefs.Mode == "off" {
		return 0
	}
	wantAlbum := prefs.Mode == "album" || (prefs.Mode == "auto" && asAlbum)
	// Album gain falls back to track gain while the album is still being
	// measured; track gain never falls back to album gain.
	gain, peak := g.TrackGain, g.TrackPeak
	if wantAlbum && g.AlbumGain != nil {
		gain, peak = g.AlbumGain, g.AlbumPeak
	}
	if gain == nil {
		return 0
	}
	db := float64(*gain) + float64(prefs.TargetLUFS) - ReplayGainReferenceLUFS
	if prefs.Boost == "headroom" && peak != nil && *peak > 0 {
		db = math.Min(db, leveledPeakCeilingDBTP-20*math.Log10(float64(*peak)))
	}
	return math.Min(db, leveledMaxBoostDB)
}

// LeveledGain is a render request: a gain in hundredths of a dB, and whether
// a limiter holds the peaks of a boost. Hundredths, so it travels in a URL and
// a signature exactly.
type LeveledGain struct {
	CentiDB int
	Limiter bool
}

// NewLeveledGain rounds db to a render request. The limiter is only ever
// engaged for a boost: a cut cannot raise a peak.
func NewLeveledGain(db float64, boost string) LeveledGain {
	c := int(math.Round(db * 100))
	return LeveledGain{CentiDB: c, Limiter: boost == "limiter" && c > 0}
}

// Unity reports whether the request would reproduce the original.
func (g LeveledGain) Unity() bool { return g.CentiDB == 0 && !g.Limiter }

// Valid reports whether the gain is one a measurement could have produced.
func (g LeveledGain) Valid() bool {
	return g.CentiDB >= int(leveledMaxCutDB*100) && g.CentiDB <= int(leveledMaxBoostDB*100)
}

// leveledFilter is the ffmpeg audio filter for g. The limiter's ceiling is
// -1 dBFS (0.891 linear); level=disabled stops alimiter from raising the
// output back to full scale afterwards, which would undo the leveling.
func leveledFilter(g LeveledGain) string {
	f := "volume=" + strconv.FormatFloat(float64(g.CentiDB)/100, 'f', 2, 64) + "dB"
	if g.Limiter {
		f += ",alimiter=limit=0.891:level=disabled"
	}
	return f
}

// leveledSource is what the renderer needs to know about the original file.
type leveledSource struct {
	SampleRate int
	// Bits is the source's sample depth; 0 for a lossy source, which has none.
	Bits int
}

// leveledRenderArgs is the ffmpeg command line rendering src to dst. Output is
// FLAC at the source's depth (24-bit for a hi-res source, 16 otherwise) and at
// most 48 kHz: renderers that take FLAC take those, and few take more.
func leveledRenderArgs(src, dst string, g LeveledGain, s leveledSource) []string {
	args := []string{
		"-hide_banner", "-nostdin", "-nostats", "-loglevel", "error",
		"-i", src,
		"-map", "0:a:0",
		"-map_metadata", "-1",
		"-af", leveledFilter(g),
		"-c:a", "flac",
	}
	if s.Bits > 16 {
		args = append(args, "-sample_fmt", "s32", "-bits_per_raw_sample", "24")
	} else {
		args = append(args, "-sample_fmt", "s16")
	}
	if s.SampleRate > 48000 {
		args = append(args, "-ar", "48000")
	}
	return append(args, "-f", "flac", "-y", dst)
}

// leveledCacheKey names the rendered file. It changes with the source file's
// size and modification time, so a replaced file is never served from an old
// render, and with leveledRenderVersion, so a change to how renders are made
// retires the old ones.
func leveledCacheKey(trackID string, info fs.FileInfo, g LeveledGain) string {
	h := sha256.New()
	_, _ = fmt.Fprintf(h, "%d|%s|%d|%d|%d|%t",
		leveledRenderVersion, trackID, info.Size(), info.ModTime().UnixNano(), g.CentiDB, g.Limiter)
	return hex.EncodeToString(h.Sum(nil))[:32] + ".flac"
}

const leveledRenderVersion = 1

// LeveledSource identifies the original a render is made from.
type LeveledSource struct {
	TrackID    string
	Path       string
	DurationMs int32
}

// LeveledRenderer renders leveled FLACs into a cache directory and keeps that
// directory under the size the loudness settings allow. Safe for concurrent
// use: renders of the same file and gain are coalesced into one.
type LeveledRenderer struct {
	dir      string
	settings *LoudnessSettingsService
	logger   *slog.Logger
	group    singleflight.Group
	evictMu  sync.Mutex

	// prerenders bounds the background renders running at once. A fetch
	// renders on demand regardless, so a prerender that finds no slot is
	// dropped, not queued.
	prerenders chan struct{}

	// render and probe are ffmpeg and ffprobe; tests replace them.
	render func(ctx context.Context, src, dst string, g LeveledGain, s leveledSource) error
	probe  func(ctx context.Context, src string) (leveledSource, error)
}

// NewLeveledRenderer renders into dir, creating it if needed.
func NewLeveledRenderer(dir string, settings *LoudnessSettingsService, logger *slog.Logger) (*LeveledRenderer, error) {
	if err := os.MkdirAll(dir, 0o750); err != nil {
		return nil, fmt.Errorf("leveled cache: %w", err)
	}
	return &LeveledRenderer{
		dir:        dir,
		settings:   settings,
		logger:     logger,
		prerenders: make(chan struct{}, maxLeveledPrerenders),
		render:     ffmpegRenderLeveled,
		probe:      ffprobeLeveledSource,
	}, nil
}

// ErrLeveledSourceMissing is returned when the original file cannot be read.
var ErrLeveledSourceMissing = errors.New("leveled: source file missing")

// Path returns the rendered file for src at gain g, rendering it first if it
// is not cached. A caller arriving while the same render runs waits for it
// rather than starting another.
func (r *LeveledRenderer) Path(ctx context.Context, src LeveledSource, g LeveledGain) (string, error) {
	info, err := os.Stat(src.Path)
	if err != nil {
		return "", fmt.Errorf("%w: %w", ErrLeveledSourceMissing, err)
	}
	dst := filepath.Join(r.dir, leveledCacheKey(src.TrackID, info, g))
	if _, err := os.Stat(dst); err == nil {
		// A hit counts as a use, for eviction.
		now := time.Now()
		_ = os.Chtimes(dst, now, now)
		return dst, nil
	}
	// The render runs detached from the caller: a renderer that gives up on
	// one request (Sonos retries quickly) must not cancel the render a second
	// request is about to wait on.
	ch := r.group.DoChan(dst, func() (any, error) {
		rctx, cancel := context.WithTimeout(context.Background(), loudnessTimeout(src.DurationMs))
		defer cancel()
		return dst, r.renderTo(rctx, src.Path, dst, g)
	})
	select {
	case res := <-ch:
		if res.Err != nil {
			return "", res.Err
		}
		return dst, nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

// maxLeveledPrerenders is how many prerenders may run at once. A speaker
// needs the track it is about to play and the one after; more than that is a
// client asking for too much, and the excess is dropped.
const maxLeveledPrerenders = 2

// Prerender starts rendering src at g in the background, so the speaker's
// fetch finds it ready. It reports false, and does nothing, when the
// prerender slots are full.
func (r *LeveledRenderer) Prerender(src LeveledSource, g LeveledGain) bool {
	select {
	case r.prerenders <- struct{}{}:
	default:
		r.logger.Debug("leveled: prerender dropped, slots full", "track", src.TrackID)
		return false
	}
	go func() {
		defer func() { <-r.prerenders }()
		ctx, cancel := context.WithTimeout(context.Background(), loudnessTimeout(src.DurationMs))
		defer cancel()
		if _, err := r.Path(ctx, src, g); err != nil {
			r.logger.Warn("leveled: prerender failed", "track", src.TrackID, "err", err)
		}
	}()
	return true
}

func (r *LeveledRenderer) renderTo(ctx context.Context, src, dst string, g LeveledGain) error {
	s, err := r.probe(ctx, src)
	if err != nil {
		return fmt.Errorf("leveled: probe: %w", err)
	}
	// Rendered beside the destination and renamed into place, so a reader
	// never sees a half-written file and a failed render leaves nothing.
	tmp := dst + ".part"
	if err := r.render(ctx, src, tmp, g, s); err != nil {
		_ = os.Remove(tmp)
		return fmt.Errorf("leveled: render: %w", err)
	}
	if err := os.Rename(tmp, dst); err != nil {
		_ = os.Remove(tmp)
		return fmt.Errorf("leveled: %w", err)
	}
	r.evict(dst)
	return nil
}

// evict removes the least recently used renders until the cache fits the
// configured size. keep is never removed: it is the render about to be served.
func (r *LeveledRenderer) evict(keep string) {
	r.evictMu.Lock()
	defer r.evictMu.Unlock()
	limit := int64(r.settings.Get().LeveledCacheMB) << 20
	entries, err := os.ReadDir(r.dir)
	if err != nil {
		r.logger.Warn("leveled: read cache", "err", err)
		return
	}
	type cached struct {
		path string
		size int64
		used time.Time
	}
	var files []cached
	var total int64
	for _, e := range entries {
		if e.IsDir() || !strings.HasSuffix(e.Name(), ".flac") {
			continue
		}
		info, err := e.Info()
		if err != nil {
			continue
		}
		files = append(files, cached{filepath.Join(r.dir, e.Name()), info.Size(), info.ModTime()})
		total += info.Size()
	}
	slices.SortFunc(files, func(a, b cached) int { return a.used.Compare(b.used) })
	for _, f := range files {
		if total <= limit {
			return
		}
		if f.path == keep {
			continue
		}
		// A file still being sent stays readable: the open descriptor holds it.
		if err := os.Remove(f.path); err == nil {
			total -= f.size
		}
	}
}

func ffmpegRenderLeveled(ctx context.Context, src, dst string, g LeveledGain, s leveledSource) error {
	out, err := exec.CommandContext(ctx, "ffmpeg", leveledRenderArgs(src, dst, g, s)...).CombinedOutput()
	if err != nil {
		return fmt.Errorf("ffmpeg: %w: %s", err, strings.TrimSpace(string(out)))
	}
	return nil
}

// ffprobeLeveledSource reads the first audio stream's rate and depth.
// bits_per_raw_sample is set for lossless codecs and 0 or N/A for lossy ones.
func ffprobeLeveledSource(ctx context.Context, src string) (leveledSource, error) {
	out, err := exec.CommandContext(ctx, "ffprobe",
		"-v", "error", "-select_streams", "a:0",
		"-show_entries", "stream=sample_rate,bits_per_raw_sample,bits_per_sample",
		"-of", "default=noprint_wrappers=1", src).Output()
	if err != nil {
		return leveledSource{}, fmt.Errorf("ffprobe: %w", err)
	}
	return parseLeveledProbe(string(out)), nil
}

// parseLeveledProbe reads ffprobe's key=value lines. Unknown or N/A values
// read as 0: a 16-bit, at-most-48 kHz render, which suits every source.
func parseLeveledProbe(out string) leveledSource {
	var s leveledSource
	for _, line := range strings.Split(out, "\n") {
		k, v, ok := strings.Cut(strings.TrimSpace(line), "=")
		if !ok {
			continue
		}
		n, err := strconv.Atoi(v)
		if err != nil {
			continue
		}
		switch k {
		case "sample_rate":
			s.SampleRate = n
		case "bits_per_raw_sample", "bits_per_sample":
			s.Bits = max(s.Bits, n)
		}
	}
	return s
}