minstrel/internal/library/album_loudness.go

package library

import (
	"context"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync"
)

// Album loudness (M464 #4996).
//
// Album-mode normalization plays a whole album at one gain, so the quiet
// interlude stays quieter than the single it sits between. That gain comes
// from the album's loudness: BS.1770's gated loudness over every block on the
// album, which is what summing the tracks' block histograms and gating the sum
// computes. An average of the tracks' values is not the same thing: gating
// over the whole album drops a near-silent hidden track, where averaging
// would let it drag the album quieter.
//
// The values are derived, so they are recomputed rather than maintained: each
// worker tick lists the albums whose inputs digest has moved (see
// ListAlbumsNeedingLoudness) and recomputes those from the stored histograms.

// albumLoudnessBatch is how many albums one query hands the album pass.
// Recomputing an album is a few small reads and arithmetic, no decode.
const albumLoudnessBatch = 200

// albumLoudness is one album's computed values.
type albumLoudness struct {
	// integratedLUFS is nil until every track is settled, or when no block on
	// the album passed the gate.
	integratedLUFS *float32
	truePeakDBTP   *float32
	total, settled int32
}

// computeAlbumLoudness sums the present tracks' histograms and gates the sum.
func computeAlbumLoudness(inputs []dbq.ListAlbumLoudnessInputsRow) albumLoudness {
	a := albumLoudness{total: int32(len(inputs))}
	hists := make([]blockHistogram, 0, len(inputs))
	for _, in := range inputs {
		if !in.Settled {
			continue
		}
		a.settled++
		if in.TruePeakDbtp != nil && (a.truePeakDBTP == nil || *in.TruePeakDbtp > *a.truePeakDBTP) {
			peak := *in.TruePeakDbtp
			a.truePeakDBTP = &peak
		}
		if in.BlockHistStart != nil && len(in.BlockHist) > 0 {
			hists = append(hists, blockHistogram{start: *in.BlockHistStart, counts: in.BlockHist})
		}
	}
	// Leveling from part of an album would change its gain as the rest is
	// measured, audibly, mid-listen. Wait for all of it.
	if a.total == 0 || a.settled < a.total {
		return a
	}
	if lufs, ok := mergeHistograms(hists).gatedLoudness(); ok {
		v := float32(lufs)
		a.integratedLUFS = &v
	}
	return a
}

// mergeHistograms sums histograms that may cover different bin ranges into
// one, trimmed to the occupied range. A histogram reaching past the bin range
// (only a corrupt row could) is clipped rather than trusted.
func mergeHistograms(hs []blockHistogram) blockHistogram {
	var bins [loudnessHistBins]int32
	for _, h := range hs {
		for i, c := range h.counts {
			if b := int(h.start) + i; b >= 0 && b < loudnessHistBins {
				bins[b] += c
			}
		}
	}
	return trimBins(&bins)
}

// AlbumLoudnessResult tallies one album pass.
type AlbumLoudnessResult struct {
	Recomputed int
	Leveled    int // stored with an album loudness
	Waiting    int // stored without one: a track is not yet measured
	Failed     int
	Orphans    int64 // rows dropped because the album has no present track
}

// albumPass recomputes every album whose inputs changed, keyset-paged on album
// id so a pass ends even when one album keeps failing to store.
//
// The digest stored is the one the list query computed. If a track changes
// between that query and the read of its inputs, the stored digest is already
// stale and the next pass recomputes the album again, so the values always
// converge on the inputs.
func (w *LoudnessBackfillWorker) albumPass(ctx context.Context) (AlbumLoudnessResult, error) {
	q := dbq.New(w.pool)
	var res AlbumLoudnessResult
	after := pgtype.UUID{Valid: true}
	for {
		if err := ctx.Err(); err != nil {
			return res, err
		}
		rows, err := q.ListAlbumsNeedingLoudness(ctx, dbq.ListAlbumsNeedingLoudnessParams{
			CurrentVersion: loudnessVersion,
			AfterID:        after,
			BatchLimit:     w.albumBatch,
		})
		if err != nil {
			return res, fmt.Errorf("list albums needing loudness: %w", err)
		}
		if len(rows) == 0 {
			break
		}
		for _, row := range rows {
			res.Recomputed++
			if err := w.storeAlbumLoudness(ctx, q, row.AlbumID, row.Digest, &res); err != nil {
				res.Failed++
				w.logger.Warn("album loudness: recompute failed", "album_id", row.AlbumID, "err", err)
			}
		}
		after = rows[len(rows)-1].AlbumID
	}
	// Listed, logged, then deleted: the same log-first order as above.
	orphans, err := q.ListOrphanAlbumLoudness(ctx)
	if err != nil {
		return res, fmt.Errorf("list orphan album loudness: %w", err)
	}
	if len(orphans) == 0 {
		return res, nil
	}
	ids := make([]string, len(orphans))
	for i, id := range orphans {
		ids[i] = syncpkg.FormatUUID(id)
	}
	if err := syncpkg.LogChanges(ctx, w.pool, syncpkg.EntityAlbum, ids, syncpkg.OpUpsert); err != nil {
		return res, fmt.Errorf("log orphan album changes: %w", err)
	}
	if err := q.DeleteAlbumLoudness(ctx, orphans); err != nil {
		return res, fmt.Errorf("drop orphan album loudness: %w", err)
	}
	res.Orphans = int64(len(orphans))
	return res, nil
}

// storeAlbumLoudness recomputes one album. The sync feed hears about it only
// when the values clients see move: during the backfill an album's digest
// changes with every track measured, and most of those recomputes still end
// with no album value.
func (w *LoudnessBackfillWorker) storeAlbumLoudness(
	ctx context.Context, q *dbq.Queries, albumID pgtype.UUID, digest string, res *AlbumLoudnessResult,
) error {
	before, err := q.GetAlbumLoudness(ctx, albumID)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("read stored values: %w", err)
	}
	inputs, err := q.ListAlbumLoudnessInputs(ctx, dbq.ListAlbumLoudnessInputsParams{
		CurrentVersion: loudnessVersion,
		AlbumID:        albumID,
	})
	if err != nil {
		return fmt.Errorf("read inputs: %w", err)
	}
	a := computeAlbumLoudness(inputs)
	// Logged before the write, for the reason storeLoudness gives (#2704). A
	// failed log stores nothing, so the digest still differs and the next
	// pass tries again.
	if albumVisibleChange(before, a) {
		if err := syncpkg.LogChange(ctx, w.pool, syncpkg.EntityAlbum, syncpkg.FormatUUID(albumID), syncpkg.OpUpsert); err != nil {
			return fmt.Errorf("log album change: %w", err)
		}
	}
	if err := q.UpsertAlbumLoudness(ctx, dbq.UpsertAlbumLoudnessParams{
		AlbumID:        albumID,
		IntegratedLufs: a.integratedLUFS,
		TruePeakDbtp:   a.truePeakDBTP,
		TracksTotal:    a.total,
		TracksSettled:  a.settled,
		InputsDigest:   digest,
	}); err != nil {
		return fmt.Errorf("store: %w", err)
	}
	if a.integratedLUFS != nil {
		res.Leveled++
	} else {
		res.Waiting++
	}
	return nil
}

// albumVisibleChange reports whether clients would see different album gains.
// The peak is only delivered beside a loudness, so a waiting album whose peak
// moves as tracks are measured has nothing new to tell anyone.
func albumVisibleChange(before dbq.GetAlbumLoudnessRow, after albumLoudness) bool {
	if !sameFloat(before.IntegratedLufs, after.integratedLUFS) {
		return true
	}
	return after.integratedLUFS != nil && !sameFloat(before.TruePeakDbtp, after.truePeakDBTP)
}

func sameFloat(a, b *float32) bool {
	if a == nil || b == nil {
		return a == b
	}
	return *a == *b
}