minstrel/internal/library/duplicate_match.go

package library

import (
	"math"
	"math/bits"
	"sort"
)

// Duplicate matching (M400 #3909).
//
// Pure functions over fingerprints: no database, no files. This is the part that
// decides whether two tracks in the operator's library are proposed as one
// recording, so every rule in it has to be falsifiable in a unit test.
//
// Two tiers, answering different questions:
//
//	exact     equal audio_stream_sha256 — the same encoded audio bytes. No score,
//	          no threshold, no false positives (the #3885 pair).
//	acoustic  chromaprint fingerprints that agree closely once aligned — the
//	          same recording at another bitrate or in another codec.
//
// The acoustic comparison follows the approach of AcoustID's pg_acoustid
// (acoustid_compare.c): vote on the relative offset between two fingerprints
// using items that agree in their high bits, then measure disagreement at the
// winning offset. Reimplemented from that description; no code was copied.
// The alignment window and match-bit width below are taken from it.

// maxAlignOffsetItems bounds how far apart two fingerprints may be shifted and
// still be compared: ±120 items, about 15 seconds at chromaprint's ~8 items per
// second. Covers a leading silence trimmed differently or a short intro; the
// same bound pg_acoustid uses (ACOUSTID_MAX_ALIGN_OFFSET).
const maxAlignOffsetItems = 120

// alignMatchBits is how many high bits two items must share to vote for an
// offset. Matching whole 32-bit items would miss the same recording at another
// bitrate, whose low bits are noisier; 14 is pg_acoustid's MATCH_BITS.
const alignMatchBits = 14

// minOverlapItems is the least overlap worth a verdict: about 10 seconds. A few
// items agreeing perfectly is not evidence that two recordings are one.
const minOverlapItems = 80

// minDistinctFraction rejects low-information fingerprints before they can
// match. Near-silence, a sustained tone or a click track produces the same few
// items over and over, and two such tracks agree closely without being the
// same recording. Real music is overwhelmingly distinct item to item, so this
// floor only catches the pathological case. A judgment value, not a measured
// one — revisit if the sweep reports real tracks refused for it.
const minDistinctFraction = 0.3

// defaultAcousticMaxBitErrorRate is the most disagreement two aligned
// fingerprints may show and still be proposed as one recording. Unrelated audio
// sits near 0.5; the same recording re-encoded lands well under 0.1.
//
// Deliberately conservative. The operator's stated worry is the opposite of a
// missed duplicate: "the same song can appear in different albums, usually it's
// a different recording", and an instrumental shares its vocal version's
// harmony, which chroma features capture. A false merge is the failure that
// matters, and the report is reviewed anyway. This is an unmeasured default:
// calibrate it against real pairs once the backfill (#3908) has populated the
// library, then expose it in Settings (#3913).
const defaultAcousticMaxBitErrorRate = 0.15

// durationToleranceMs is how far apart two tracks' durations may be and still be
// compared. Encoders pad and trim a little; different edits differ by more.
const durationToleranceMs = 3000

// maxAcousticGroupSize caps an acoustic group. A cluster bigger than this is far
// more likely a shared jingle, a skit or a low-information pattern than eight
// copies of one recording, and proposing it would bury the real duplicates.
// Exact-tier groups are not capped: identical bytes are identical however many.
const maxAcousticGroupSize = 8

// acousticScore is the result of comparing two fingerprints.
type acousticScore struct {
	// Offset is how many items b is shifted against a: b[i+Offset] aligns with
	// a[i].
	Offset int
	// Overlap is how many aligned items were compared.
	Overlap int
	// BitErrorRate is the fraction of differing bits over the overlap, 0..1.
	BitErrorRate float64
}

// preparedPrint is a fingerprint with the parts every comparison needs worked
// out once. The sweep compares each track with every other track within a few
// seconds of its duration, so rebuilding the alignment index for each pair would
// dominate its cost.
type preparedPrint struct {
	items       []int32
	index       map[uint32][]int
	informative bool
}

// preparePrint indexes a fingerprint's items by their high bits and records
// whether it varies enough to be compared at all.
func preparePrint(fp []int32) *preparedPrint {
	// Each bucket keeps only a few positions: a value repeating many times is
	// uninformative, and letting it vote once per repeat would make every
	// pairing O(n²).
	const keepPerBucket = 4
	p := &preparedPrint{items: fp, index: make(map[uint32][]int, len(fp))}
	seen := make(map[int32]struct{}, len(fp))
	for i, v := range fp {
		seen[v] = struct{}{}
		key := alignKey(v)
		if pos := p.index[key]; len(pos) < keepPerBucket {
			p.index[key] = append(pos, i)
		}
	}
	p.informative = len(fp) > 0 && float64(len(seen)) >= minDistinctFraction*float64(len(fp))
	return p
}

func alignKey(v int32) uint32 { return uint32(v) >> (32 - alignMatchBits) }

// compareChromaprint aligns two raw fingerprints and measures how much they
// disagree. ok is false when no verdict is possible: no offset gathered any
// votes, the overlap at the best offset is too short, or either side carries
// too little information to mean anything.
func compareChromaprint(a, b []int32) (acousticScore, bool) {
	return comparePrepared(preparePrint(a), preparePrint(b))
}

// comparePrepared is compareChromaprint over fingerprints already prepared.
func comparePrepared(a, b *preparedPrint) (acousticScore, bool) {
	if len(a.items) < minOverlapItems || len(b.items) < minOverlapItems {
		return acousticScore{}, false
	}
	if !a.informative || !b.informative {
		return acousticScore{}, false
	}

	offset, ok := bestOffset(a, b)
	if !ok {
		return acousticScore{}, false
	}

	// a[i] aligns with b[i+offset]; walk the indices valid on both sides.
	start := max(0, -offset)
	end := min(len(a.items), len(b.items)-offset)
	overlap := end - start
	if overlap < minOverlapItems {
		return acousticScore{}, false
	}
	errBits := 0
	for i := start; i < end; i++ {
		errBits += bits.OnesCount32(uint32(a.items[i]) ^ uint32(b.items[i+offset]))
	}
	return acousticScore{
		Offset:       offset,
		Overlap:      overlap,
		BitErrorRate: float64(errBits) / float64(32*overlap),
	}, true
}

// bestOffset returns the relative shift most items agree on.
func bestOffset(a, b *preparedPrint) (int, bool) {
	votes := make([]int, 2*maxAlignOffsetItems+1)
	for j, v := range b.items {
		for _, i := range a.index[alignKey(v)] {
			off := j - i
			if off >= -maxAlignOffsetItems && off <= maxAlignOffsetItems {
				votes[off+maxAlignOffsetItems]++
			}
		}
	}

	best, bestVotes := 0, 0
	for k, n := range votes {
		// Strictly greater keeps the smallest shift on a tie, which is the more
		// likely truth and keeps the result deterministic.
		if n > bestVotes || (n == bestVotes && n > 0 && abs(k-maxAlignOffsetItems) < abs(best)) {
			best, bestVotes = k-maxAlignOffsetItems, n
		}
	}
	return best, bestVotes > 0
}

// fingerprintCandidate is one track as the grouping sees it.
type fingerprintCandidate struct {
	ID           string
	DurationMs   int32
	StreamSHA256 []byte
	Chromaprint  []int32
}

// duplicateTier names what a group's evidence is.
type duplicateTier string

const (
	tierExact    duplicateTier = "exact"
	tierAcoustic duplicateTier = "acoustic"
)

// duplicateGroup is a set of tracks proposed as one recording. Members are
// sorted by ID.
type duplicateGroup struct {
	Tier    duplicateTier
	Members []string
	// WorstBitErrorRate is the largest disagreement between any two members of
	// an acoustic group — the weakest evidence the group rests on. Zero for
	// exact groups.
	WorstBitErrorRate float64
}

// groupingResult is what one grouping pass found.
type groupingResult struct {
	Groups []duplicateGroup
	// OversizeClusters counts acoustic clusters discarded for exceeding
	// maxAcousticGroupSize. Reported rather than silent: a sudden rise means the
	// cap or the information floor needs attention.
	OversizeClusters int
}

// groupUnit is one thing the acoustic pass compares: a single track, or an exact
// group standing in for all its byte-identical copies.
type groupUnit struct {
	ids        []string // every member, sorted
	durationMs int32
	sortKey    string // the representative's id: ties on duration break on it
	print      *preparedPrint
	exact      bool // more than one member with identical audio
	assigned   bool
}

// streamGrouper is the acoustic pass over units arriving in (durationMs, sortKey)
// order. It holds only the units within durationToleranceMs of the oldest one
// not yet settled, so memory is bounded by the densest few seconds of the
// library rather than by its size — the whole library's fingerprints would be
// hundreds of megabytes.
//
// A seed can be settled as soon as a unit arrives beyond its window: everything
// it could group with has already arrived, and no later seed can reach back to
// it because seeds only look forward. That is what makes the streamed result
// identical to running the same pass over the whole sorted list.
//
// Grouping is COMPLETE-LINKAGE: a unit joins a group only if it matches every
// unit already in it, within the duration tolerance and the bit-error limit.
// Single-linkage would let a chain of near-misses — A close to B, B close to C —
// drag A and C, which are not close, into one proposed merge. Complete linkage
// also means any member can be chosen as the survivor (#3911).
//
// When an acoustic group absorbs an exact group, the result is tier acoustic: a
// group is only as certain as its weakest link.
type streamGrouper struct {
	maxBitErrorRate float64
	window          []*groupUnit
	res             groupingResult
}

func newStreamGrouper(maxBitErrorRate float64) *streamGrouper {
	return &streamGrouper{maxBitErrorRate: maxBitErrorRate}
}

// push adds the next unit. Units must arrive in non-decreasing
// (durationMs, sortKey) order.
func (g *streamGrouper) push(u *groupUnit) {
	g.window = append(g.window, u)
	for len(g.window) > 1 && u.durationMs-g.window[0].durationMs > durationToleranceMs {
		g.settleOldest()
	}
}

// finish settles every unit still waiting and returns what was found. Groups
// are in no particular order; callers sort with sortGroups.
func (g *streamGrouper) finish() groupingResult {
	for len(g.window) > 0 {
		g.settleOldest()
	}
	return g.res
}

func (g *streamGrouper) settleOldest() {
	seed := g.window[0]
	g.window[0] = nil // release it: the window's backing array outlives the slide
	g.window = g.window[1:]
	if seed.assigned {
		return
	}

	group := []*groupUnit{seed}
	worst := 0.0
	for _, cand := range g.window {
		if cand.durationMs-seed.durationMs > durationToleranceMs {
			break
		}
		if cand.assigned {
			continue
		}
		joined, worstWithCand := true, worst
		for _, member := range group {
			if abs32(cand.durationMs-member.durationMs) > durationToleranceMs {
				joined = false
				break
			}
			score, ok := comparePrepared(member.print, cand.print)
			if !ok || score.BitErrorRate > g.maxBitErrorRate {
				joined = false
				break
			}
			worstWithCand = math.Max(worstWithCand, score.BitErrorRate)
		}
		if joined {
			group = append(group, cand)
			worst = worstWithCand
		}
	}

	if len(group) == 1 {
		if seed.exact {
			g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: seed.ids})
		}
		return
	}
	// Count units, not tracks: an absorbed exact group is one piece of acoustic
	// evidence however many identical files it holds.
	if len(group) > maxAcousticGroupSize {
		g.res.OversizeClusters++
		for _, member := range group {
			member.assigned = true
			// The acoustic evidence is untrustworthy; identical bytes are not.
			// An exact group caught inside an oversize cluster is still proposed.
			if member.exact {
				g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: member.ids})
			}
		}
		return
	}
	var members []string
	for _, member := range group {
		member.assigned = true
		members = append(members, member.ids...)
	}
	sort.Strings(members)
	g.res.Groups = append(g.res.Groups, duplicateGroup{
		Tier: tierAcoustic, Members: members, WorstBitErrorRate: worst,
	})
}

// groupDuplicates proposes duplicate groups among candidates held in memory. It
// runs the same streamGrouper the sweep uses, so there is one grouping rule.
//
// Exact groups come first: tracks sharing an audio stream hash. Each becomes a
// single unit for the acoustic pass, represented by its member with the lowest
// (duration, id) that has a chromaprint. That is the member the sweep's
// duration-ordered stream meets first, which keeps the two identical. An exact
// group with no chromaprint at all cannot be compared acoustically and stands
// on its own.
//
// The output does not depend on input order.
func groupDuplicates(cands []fingerprintCandidate, maxBitErrorRate float64) groupingResult {
	byHash := map[string][]fingerprintCandidate{}
	var units []*groupUnit
	var printless []duplicateGroup
	for _, c := range cands {
		if len(c.StreamSHA256) > 0 {
			byHash[string(c.StreamSHA256)] = append(byHash[string(c.StreamSHA256)], c)
			continue
		}
		if len(c.Chromaprint) > 0 {
			units = append(units, &groupUnit{
				ids: []string{c.ID}, durationMs: c.DurationMs, sortKey: c.ID, print: preparePrint(c.Chromaprint),
			})
		}
	}
	for _, group := range byHash {
		ids := make([]string, len(group))
		for i, m := range group {
			ids[i] = m.ID
		}
		sort.Strings(ids)

		var rep *fingerprintCandidate
		for i := range group {
			m := &group[i]
			if len(m.Chromaprint) == 0 {
				continue
			}
			if rep == nil || m.DurationMs < rep.DurationMs || (m.DurationMs == rep.DurationMs && m.ID < rep.ID) {
				rep = m
			}
		}
		if rep == nil {
			if len(group) > 1 {
				printless = append(printless, duplicateGroup{Tier: tierExact, Members: ids})
			}
			continue
		}
		units = append(units, &groupUnit{
			ids: ids, durationMs: rep.DurationMs, sortKey: rep.ID,
			print: preparePrint(rep.Chromaprint), exact: len(group) > 1,
		})
	}

	sort.Slice(units, func(i, j int) bool {
		if units[i].durationMs != units[j].durationMs {
			return units[i].durationMs < units[j].durationMs
		}
		return units[i].sortKey < units[j].sortKey
	})
	g := newStreamGrouper(maxBitErrorRate)
	for _, u := range units {
		g.push(u)
	}
	res := g.finish()
	res.Groups = append(res.Groups, printless...)
	sortGroups(res.Groups)
	return res
}

// sortGroups orders groups by their first member. Groups are disjoint, so that
// is a total order.
func sortGroups(groups []duplicateGroup) {
	sort.Slice(groups, func(i, j int) bool { return groups[i].Members[0] < groups[j].Members[0] })
}

func abs(n int) int {
	if n < 0 {
		return -n
	}
	return n
}

func abs32(n int32) int32 {
	if n < 0 {
		return -n
	}
	return n
}