yellowjacket/backend/explore/dumpcatalog.go

//go:build indexbuild

package explore

import (
	"archive/tar"
	"bufio"
	"context"
	"encoding/csv"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"

	"github.com/klauspost/compress/zstd"
)

// Stages 2+3 of the dump import.  Stage 2 picks per-entity popularity
// floors from the aggregated listen counts (top-N by listen count,
// with an absolute minimum).  Stage 3 streams the MusicBrainz
// canonical dump (~2GB tar.zst of CSVs) and keeps only rows above the
// floors, assembling them directly into explore_index.  Nothing except
// the final index rows touches disk; the canonical stream is cheap
// enough to simply restart after an interruption.

const (
	// Entity budgets: the floors are chosen as "the listen count of
	// the Nth most-listened entity", clamped to minListenFloor.
	keepRecordings   = 1_500_000
	keepReleases     = 2_000_000
	keepArtists      = 250_000
	keepReleaseGroup = 400_000

	// minListenFloor cuts long-tail noise even when a budget isn't
	// reached (small dumps, sparse entities).
	minListenFloor = 10

	// releaseMinListenFloor is the (lower) floor for releases, which
	// are not indexed themselves but roll up into release groups.
	releaseMinListenFloor = 5

	// dumpAssembleBatch is the number of index rows per upsert
	// transaction during assembly.
	dumpAssembleBatch = 2000

	// canonicalProgressRows controls progress reporting during the
	// canonical CSV scan (~30M rows total).
	canonicalProgressRows = 2_000_000
)

// Per-artist discography coverage (S2).  The global budgets above keep
// the most-listened entities overall; on their own they leave a
// moderately popular artist with little or nothing on their detail page,
// forcing a slow live ListenBrainz fetch on first view.  S2 guarantees
// each browse-likely artist a graded slice of their own top tracks and
// release groups, selected offline from the same dump — no API calls.
const (
	// perArtistArtistBudget is how many top artists (by listen count)
	// receive graded per-artist coverage.  Library artists are always
	// covered in full regardless of rank (see markLibraryArtists).
	perArtistArtistBudget = 10_000

	// Rank boundaries (1-based) within the top perArtistArtistBudget
	// artists.  Tier A is the most-listened; everything from tier B's
	// boundary down to perArtistArtistBudget is tier C.
	perArtistTierA = 1_000
	perArtistTierB = 4_000

	// Per-tier track/release-group budgets.  These are caps that count
	// globally-kept entities toward the total, so a superstar whose
	// catalogue is already in the global set adds ~nothing.
	perArtistTierATrack = 50
	perArtistTierBTrack = 25
	perArtistTierCTrack = 12
	perArtistTierARG    = 15
	perArtistTierBRG    = 10
	perArtistTierCRG    = 5

	// Library artists get effectively their full discography, capped to
	// bound memory for pathologically prolific credits.
	perArtistLibraryTrack = 500
	perArtistLibraryRG    = 100

	// keepRecordingsSecondary bounds the recording listen counts retained
	// through the canonical scan for per-artist selection (a superset of
	// the global kept set).  Set well above keepRecordings so a top-10K
	// artist's deep cuts still qualify as candidates.
	keepRecordingsSecondary = 5_000_000
)

// uuid16 is a parsed MBID.
type uuid16 [16]byte

// keptSets holds the entities that survived the popularity threshold,
// with their listen counts.
type keptSets struct {
	recordings map[uuid16]uint32
	releases   map[uuid16]uint32
	artists    map[uuid16]uint32

	recFloor uint32
	relFloor uint32
	artFloor uint32

	// Per-artist coverage (S2).  recSecondary holds recording listen
	// counts down to a lower floor than recordings (a superset of it),
	// so a target artist's sub-global-floor tracks stay selectable.
	// trackBudget/rgBudget give the per-artist top-N cap for each browse
	// target (top perArtistArtistBudget artists by listens, plus every
	// library artist).
	recSecondary map[uuid16]uint32
	trackBudget  map[uuid16]int
	rgBudget     map[uuid16]int
}

// computeThreshold picks per-entity floors and builds the kept sets.
// This is the "statistical analysis" step: listen counts follow a
// power law, so budget-based cutoffs (keep the top N) adapt to the
// actual distribution instead of hardcoding a magic listen count.
func (imp *dumpImporter) computeThreshold(counts map[mbidKey]uint32) *keptSets {
	var recVals, relVals, artVals []uint32

	var recTotal, relTotal, artTotal uint64

	for k, v := range counts {
		switch k[0] {
		case countKindRecording:
			recVals = append(recVals, v)
			recTotal += uint64(v)
		case countKindRelease:
			relVals = append(relVals, v)
			relTotal += uint64(v)
		case countKindArtist:
			artVals = append(artVals, v)
			artTotal += uint64(v)
		}
	}

	ks := &keptSets{
		recFloor: floorForBudget(recVals, keepRecordings, minListenFloor),
		relFloor: floorForBudget(relVals, keepReleases, releaseMinListenFloor),
		artFloor: floorForBudget(artVals, keepArtists, minListenFloor),

		recordings: make(map[uuid16]uint32, keepRecordings),
		releases:   make(map[uuid16]uint32, keepReleases),
		artists:    make(map[uuid16]uint32, keepArtists),

		recSecondary: make(map[uuid16]uint32, keepRecordingsSecondary),
		trackBudget:  make(map[uuid16]int),
		rgBudget:     make(map[uuid16]int),
	}

	// Per-artist coverage (S2): retain recording counts down to a lower
	// secondary floor, and derive rank-based artist tier floors so each
	// of the top perArtistArtistBudget artists gets a graded track/RG
	// budget.  artVals is sorted descending so rank N is at index N-1.
	recSecFloor := floorForBudget(recVals, keepRecordingsSecondary, minListenFloor)

	sort.Slice(artVals, func(i, j int) bool { return artVals[i] > artVals[j] })
	artTierAFloor := rankFloor(artVals, perArtistTierA)
	artTierBFloor := rankFloor(artVals, perArtistTierB)
	artTierCFloor := rankFloor(artVals, perArtistArtistBudget)

	var recKept, relKept, artKept uint64

	for k, v := range counts {
		var id uuid16

		copy(id[:], k[1:])

		switch {
		case k[0] == countKindRecording && v >= ks.recFloor:
			ks.recordings[id] = v
			ks.recSecondary[id] = v
			recKept += uint64(v)
		case k[0] == countKindRecording && v >= recSecFloor:
			ks.recSecondary[id] = v
		case k[0] == countKindRelease && v >= ks.relFloor:
			ks.releases[id] = v
			relKept += uint64(v)
		case k[0] == countKindArtist && v >= ks.artFloor:
			ks.artists[id] = v
			artKept += uint64(v)

			if v >= artTierCFloor {
				ks.trackBudget[id], ks.rgBudget[id] = tierBudget(
					v, artTierAFloor, artTierBFloor,
				)
			}
		}
	}

	// Library artists are always covered in full — override any rank tier
	// and include those below the global artist floor.
	imp.markLibraryArtists(ks)

	imp.logger.Info("dump import: popularity thresholds chosen",
		"recordingFloor", ks.recFloor,
		"recordings", len(ks.recordings),
		"recordingCoverage", coveragePct(recKept, recTotal),
		"releaseFloor", ks.relFloor,
		"releases", len(ks.releases),
		"artistFloor", ks.artFloor,
		"artists", len(ks.artists),
		"artistCoverage", coveragePct(artKept, artTotal),
		"secondaryFloor", recSecFloor,
		"secondaryRecordings", len(ks.recSecondary),
		"targetArtists", len(ks.trackBudget),
	)

	return ks
}

// rankFloor returns the listen count at the given 1-based rank in a slice
// already sorted in descending order, clamped to the slice length.  Used
// to turn artist-rank tier boundaries into concrete listen-count floors
// that adapt to each dump's distribution.
func rankFloor(sortedDesc []uint32, rank int) uint32 {
	if len(sortedDesc) == 0 {
		return 0
	}

	if rank > len(sortedDesc) {
		rank = len(sortedDesc)
	}

	return sortedDesc[rank-1]
}

// tierBudget maps an artist's listen count to its per-artist track and
// release-group budgets via the rank-derived tier floors.
func tierBudget(listens, tierAFloor, tierBFloor uint32) (int, int) {
	switch {
	case listens >= tierAFloor:
		return perArtistTierATrack, perArtistTierARG
	case listens >= tierBFloor:
		return perArtistTierBTrack, perArtistTierBRG
	default:
		return perArtistTierCTrack, perArtistTierCRG
	}
}

// markLibraryArtists grants full per-artist coverage to every library
// artist, overriding any rank-based tier and including artists below the
// global artist floor so their discography is covered even when globally
// obscure.
func (imp *dumpImporter) markLibraryArtists(ks *keptSets) {
	n := 0

	for _, s := range imp.si.getLibraryArtistMBIDs() {
		var id uuid16

		if !parseUUID(s, id[:]) {
			continue
		}

		ks.trackBudget[id] = perArtistLibraryTrack
		ks.rgBudget[id] = perArtistLibraryRG
		n++
	}

	imp.logger.Info("dump import: library artists granted full coverage", "artists", n)
}

// floorForBudget returns the listen count of the budget-th largest
// value (so keeping everything >= floor yields ~budget entries),
// clamped below by minFloor.
func floorForBudget(vals []uint32, budget int, minFloor uint32) uint32 {
	if len(vals) == 0 {
		return minFloor
	}

	if len(vals) <= budget {
		return minFloor
	}

	sort.Slice(vals, func(i, j int) bool { return vals[i] > vals[j] })

	floor := vals[budget-1]
	if floor < minFloor {
		floor = minFloor
	}

	return floor
}

func coveragePct(kept, total uint64) string {
	if total == 0 {
		return "0%"
	}

	return fmt.Sprintf("%.1f%%", float64(kept)/float64(total)*100)
}

// ---------------------------------------------------------------------------
// Canonical dump scan
// ---------------------------------------------------------------------------

// keptRecordingRow is a canonical-dump row that survived the filter.
type keptRecordingRow struct {
	mbid        uuid16
	name        string
	artistName  string
	artistMBID  string
	releaseMBID uuid16
	releaseName string
	listens     uint32
}

// releaseInfo carries the display fields for a kept release, used to
// title its release group.
type releaseInfo struct {
	name       string
	artistName string
	artistMBID string
}

// rgTarget is a release's redirect target.
type rgTarget struct {
	rg        uuid16
	canonical uuid16
}

// canonicalScan is the in-RAM result of streaming the canonical dump.
type canonicalScan struct {
	recordings   []keptRecordingRow
	releaseInfos map[uuid16]releaseInfo
	releaseToRG  map[uuid16]rgTarget
	artistNames  map[uuid16]string

	// artistTracks accumulates each target artist's top recordings for
	// S2 coverage; merged into recordings before assembly.
	artistTracks *perArtistTracks
}

// perArtistTracks holds, per target artist, that artist's top recordings
// by listen count.  Populated during the canonical scan.
type perArtistTracks struct {
	byArtist map[uuid16]*artistTopN
}

func newPerArtistTracks() *perArtistTracks {
	return &perArtistTracks{byArtist: make(map[uuid16]*artistTopN)}
}

func (p *perArtistTracks) add(artist uuid16, budget int, row keptRecordingRow) {
	a := p.byArtist[artist]
	if a == nil {
		a = &artistTopN{n: budget, inSet: make(map[uuid16]struct{}, budget)}
		p.byArtist[artist] = a
	}

	a.add(row)
}

// artistTopN keeps the top-n recordings for a single artist by listen
// count, deduplicated by recording MBID.  n is small (tens to a few
// hundred), so a linear min-scan on eviction is cheaper than the overhead
// of container/heap.
type artistTopN struct {
	n     int
	rows  []keptRecordingRow
	inSet map[uuid16]struct{}
}

func (a *artistTopN) add(row keptRecordingRow) {
	if _, dup := a.inSet[row.mbid]; dup {
		return
	}

	if len(a.rows) < a.n {
		a.rows = append(a.rows, row)
		a.inSet[row.mbid] = struct{}{}

		return
	}

	minIdx := 0

	for i := 1; i < len(a.rows); i++ {
		if a.rows[i].listens < a.rows[minIdx].listens {
			minIdx = i
		}
	}

	if row.listens <= a.rows[minIdx].listens {
		return
	}

	delete(a.inSet, a.rows[minIdx].mbid)
	a.rows[minIdx] = row
	a.inSet[row.mbid] = struct{}{}
}

// rgCandidate is a release group offered to an artist's top-K selection.
type rgCandidate struct {
	rg      uuid16
	listens uint32
}

// artistTopRG keeps the top-n release groups for a single artist by
// rolled-up listen count, deduplicated by release-group MBID.
type artistTopRG struct {
	n     int
	rgs   []rgCandidate
	inSet map[uuid16]struct{}
}

func (a *artistTopRG) add(rg uuid16, listens uint32) {
	if _, dup := a.inSet[rg]; dup {
		return
	}

	if len(a.rgs) < a.n {
		a.rgs = append(a.rgs, rgCandidate{rg: rg, listens: listens})
		a.inSet[rg] = struct{}{}

		return
	}

	minIdx := 0

	for i := 1; i < len(a.rgs); i++ {
		if a.rgs[i].listens < a.rgs[minIdx].listens {
			minIdx = i
		}
	}

	if listens <= a.rgs[minIdx].listens {
		return
	}

	delete(a.inSet, a.rgs[minIdx].rg)
	a.rgs[minIdx] = rgCandidate{rg: rg, listens: listens}
	a.inSet[rg] = struct{}{}
}

// scanCanonicalDump streams the canonical dump and filters it against
// the kept sets.  Member order inside the tar doesn't matter — the two
// CSVs populate independent maps that are joined during assembly.
func (imp *dumpImporter) scanCanonicalDump(
	ctx context.Context, url string, ks *keptSets,
) (*canonicalScan, error) {
	stream := imp.openDumpStream(ctx, url, 0)

	defer func() { _ = stream.Close() }()

	zr, err := zstd.NewReader(bufio.NewReaderSize(stream, 1<<20))
	if err != nil {
		return nil, fmt.Errorf("canonical zstd: %w", err)
	}

	defer zr.Close()

	scan := &canonicalScan{
		releaseInfos: make(map[uuid16]releaseInfo, len(ks.releases)),
		releaseToRG:  make(map[uuid16]rgTarget, len(ks.releases)),
		artistNames:  make(map[uuid16]string, len(ks.artists)),
		artistTracks: newPerArtistTracks(),
	}

	sawData, sawRedirect := false, false
	tr := tar.NewReader(zr)

	for {
		if err := ctx.Err(); err != nil {
			return nil, err
		}

		hdr, err := tr.Next()
		if errors.Is(err, io.EOF) {
			break
		}

		if err != nil {
			return nil, fmt.Errorf("canonical tar: %w", err)
		}

		if hdr.Typeflag != tar.TypeReg {
			continue
		}

		var member io.Reader = tr

		// Individual members may themselves be zstd-compressed.
		if strings.HasSuffix(hdr.Name, ".zst") {
			mzr, err := zstd.NewReader(member)
			if err != nil {
				return nil, fmt.Errorf("canonical member zstd: %w", err)
			}

			member = mzr.IOReadCloser()
		}

		switch {
		case strings.Contains(hdr.Name, "canonical_musicbrainz_data.csv"):
			if err := imp.scanCanonicalData(ctx, member, ks, scan); err != nil {
				return nil, err
			}

			sawData = true
		case strings.Contains(hdr.Name, "canonical_release_redirect.csv"):
			if err := imp.scanReleaseRedirect(ctx, member, ks, scan); err != nil {
				return nil, err
			}

			sawRedirect = true
		}
	}

	if !sawData || !sawRedirect {
		return nil, fmt.Errorf("%w: canonical dump missing expected CSVs (data=%t redirect=%t)",
			ErrDumpFormat, sawData, sawRedirect)
	}

	return scan, nil
}

// canonicalDataColumns maps the columns of canonical_musicbrainz_data.csv.
type canonicalDataColumns struct {
	artistMBIDs      int
	artistCreditName int
	releaseMBID      int
	releaseName      int
	recordingMBID    int
	recordingName    int
}

// defaultCanonicalDataColumns matches the documented column order.
func defaultCanonicalDataColumns() canonicalDataColumns {
	return canonicalDataColumns{
		artistMBIDs:      2,
		artistCreditName: 3,
		releaseMBID:      4,
		releaseName:      5,
		recordingMBID:    6,
		recordingName:    7,
	}
}

func (imp *dumpImporter) scanCanonicalData(
	ctx context.Context, r io.Reader, ks *keptSets, scan *canonicalScan,
) error {
	cr := csv.NewReader(bufio.NewReaderSize(r, 1<<20))
	cr.ReuseRecord = true
	cr.FieldsPerRecord = -1

	cols := defaultCanonicalDataColumns()
	rows := 0

	for {
		rec, err := cr.Read()
		if errors.Is(err, io.EOF) {
			break
		}

		if err != nil {
			return fmt.Errorf("canonical data csv: %w", err)
		}

		rows++

		if rows == 1 && looksLikeHeader(rec) {
			cols = headerColumns(rec, cols)

			continue
		}

		if rows%canonicalProgressRows == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}

			imp.logger.Info("dump import: canonical scan progress",
				"rows", rows,
				"kept", len(scan.recordings),
			)
			imp.setStageProgress(dumpStageCatalog, rows, 0)
		}

		maxCol := cols.recordingName
		if cols.recordingMBID > maxCol {
			maxCol = cols.recordingMBID
		}

		if len(rec) <= maxCol {
			continue
		}

		var recMBID uuid16

		if !parseUUID(rec[cols.recordingMBID], recMBID[:]) {
			continue
		}

		var relMBID uuid16

		relOK := parseUUID(rec[cols.releaseMBID], relMBID[:])

		artistMBIDs := parsePGStringArray(rec[cols.artistMBIDs])
		creditName := rec[cols.artistCreditName]

		// Artist names: a single-artist credit names that artist.
		if len(artistMBIDs) == 1 {
			var artMBID uuid16

			if parseUUID(artistMBIDs[0], artMBID[:]) {
				if _, kept := ks.artists[artMBID]; kept {
					if _, seen := scan.artistNames[artMBID]; !seen {
						scan.artistNames[artMBID] = creditName
					}
				}
			}
		}

		// Release display info for release-group titling.
		if relOK {
			if _, kept := ks.releases[relMBID]; kept {
				if _, seen := scan.releaseInfos[relMBID]; !seen {
					firstArtist := ""
					if len(artistMBIDs) > 0 {
						firstArtist = artistMBIDs[0]
					}

					scan.releaseInfos[relMBID] = releaseInfo{
						name:       rec[cols.releaseName],
						artistName: creditName,
						artistMBID: firstArtist,
					}
				}
			}
		}

		globalListens, keptGlobal := ks.recordings[recMBID]
		secListens, keptSecondary := ks.recSecondary[recMBID]

		// Below even the secondary floor: neither globally kept nor a
		// per-artist candidate, so there's nothing to do.
		if !keptGlobal && !keptSecondary {
			continue
		}

		firstArtist := ""
		if len(artistMBIDs) > 0 {
			firstArtist = artistMBIDs[0]
		}

		// Build the row once (globalListens == secListens for global-kept
		// recordings, since the global set is a subset of the secondary).
		row := keptRecordingRow{
			mbid:        recMBID,
			name:        strings.Clone(rec[cols.recordingName]),
			artistName:  strings.Clone(creditName),
			artistMBID:  firstArtist,
			releaseMBID: relMBID,
			releaseName: strings.Clone(rec[cols.releaseName]),
			listens:     secListens,
		}

		// S2: offer this recording to its primary artist's top-N when
		// that artist is a browse target.
		if keptSecondary && firstArtist != "" {
			var faID uuid16

			if parseUUID(firstArtist, faID[:]) {
				if budget, target := ks.trackBudget[faID]; target {
					scan.artistTracks.add(faID, budget, row)
				}
			}
		}

		// Global-kept recordings are written unconditionally.
		if keptGlobal {
			row.listens = globalListens
			scan.recordings = append(scan.recordings, row)
		}
	}

	imp.logger.Info("dump import: canonical data scanned",
		"rows", rows,
		"keptRecordings", len(scan.recordings),
		"artistNames", len(scan.artistNames),
	)

	return nil
}

func (imp *dumpImporter) scanReleaseRedirect(
	ctx context.Context, r io.Reader, ks *keptSets, scan *canonicalScan,
) error {
	cr := csv.NewReader(bufio.NewReaderSize(r, 1<<20))
	cr.ReuseRecord = true
	cr.FieldsPerRecord = -1

	// Documented order: release_mbid, canonical_release_mbid,
	// release_group_mbid.
	relCol, canonCol, rgCol := 0, 1, 2
	rows := 0

	for {
		rec, err := cr.Read()
		if errors.Is(err, io.EOF) {
			break
		}

		if err != nil {
			return fmt.Errorf("release redirect csv: %w", err)
		}

		rows++

		if rows == 1 && looksLikeHeader(rec) {
			for i, name := range rec {
				switch strings.TrimSpace(name) {
				case "release_mbid":
					relCol = i
				case "canonical_release_mbid":
					canonCol = i
				case "release_group_mbid":
					rgCol = i
				}
			}

			continue
		}

		if rows%canonicalProgressRows == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}
		}

		if len(rec) <= rgCol || len(rec) <= relCol || len(rec) <= canonCol {
			continue
		}

		var rel uuid16

		if !parseUUID(rec[relCol], rel[:]) {
			continue
		}

		if _, kept := ks.releases[rel]; !kept {
			continue
		}

		var target rgTarget

		if !parseUUID(rec[rgCol], target.rg[:]) {
			continue
		}

		if !parseUUID(rec[canonCol], target.canonical[:]) {
			target.canonical = rel
		}

		scan.releaseToRG[rel] = target
	}

	imp.logger.Info("dump import: release redirects scanned",
		"rows", rows,
		"keptReleases", len(scan.releaseToRG),
	)

	return nil
}

// looksLikeHeader reports whether a CSV record is a header row (no
// parseable UUIDs, contains a known column name).
func looksLikeHeader(rec []string) bool {
	for _, f := range rec {
		switch strings.TrimSpace(f) {
		case "recording_mbid", "release_mbid", "artist_mbids", "release_group_mbid":
			return true
		}
	}

	return false
}

// headerColumns resolves column indexes from a header row, falling
// back to the documented defaults for any missing name.
func headerColumns(rec []string, fallback canonicalDataColumns) canonicalDataColumns {
	cols := fallback

	for i, name := range rec {
		switch strings.TrimSpace(name) {
		case "artist_mbids":
			cols.artistMBIDs = i
		case "artist_credit_name":
			cols.artistCreditName = i
		case "release_mbid":
			cols.releaseMBID = i
		case "release_name":
			cols.releaseName = i
		case "recording_mbid":
			cols.recordingMBID = i
		case "recording_name":
			cols.recordingName = i
		}
	}

	return cols
}

// parsePGStringArray parses the artist_mbids CSV field, tolerating
// Postgres array syntax ({a,b}), JSON-ish lists (['a', 'b']), and bare
// single values.
func parsePGStringArray(s string) []string {
	s = strings.TrimSpace(s)
	if s == "" {
		return nil
	}

	if len(s) >= 2 {
		first, last := s[0], s[len(s)-1]
		if (first == '{' && last == '}') || (first == '[' && last == ']') {
			s = s[1 : len(s)-1]
		}
	}

	if s == "" {
		return nil
	}

	parts := strings.Split(s, ",")
	out := make([]string, 0, len(parts))

	for _, p := range parts {
		p = strings.Trim(strings.TrimSpace(p), `"'`)
		if p != "" {
			out = append(out, p)
		}
	}

	return out
}

// ---------------------------------------------------------------------------
// Assembly into explore_index
// ---------------------------------------------------------------------------

// assembleIndex writes the filtered catalog into explore_index via the
// standard upsert path (idempotent — safe to re-run after a crash).
func (imp *dumpImporter) assembleIndex(
	ctx context.Context, ks *keptSets, scan *canonicalScan,
) error {
	batch := make([]SearchIndexResult, 0, dumpAssembleBatch)

	flush := func() error {
		if len(batch) == 0 {
			return nil
		}

		if err := ctx.Err(); err != nil {
			return err
		}

		imp.si.upsertBatch(batch)
		batch = batch[:0]

		return nil
	}

	// S2: fold each target artist's top recordings into the kept set.
	imp.mergePerArtistTracks(ks, scan)

	// Recordings.
	written := 0

	for _, row := range scan.recordings {
		caa := row.releaseMBID
		if target, ok := scan.releaseToRG[row.releaseMBID]; ok {
			caa = target.canonical
		}

		batch = append(batch, SearchIndexResult{
			EntityType:     "recording",
			MBID:           formatUUID(row.mbid[:]),
			Title:          row.name,
			ArtistName:     row.artistName,
			ArtistMBID:     row.artistMBID,
			Popularity:     int(row.listens),
			ReleaseName:    row.releaseName,
			CAAReleaseMBID: formatUUID(caa[:]),
		})

		written++

		if len(batch) >= dumpAssembleBatch {
			if err := flush(); err != nil {
				return err
			}

			if written%500_000 == 0 {
				imp.logger.Info("dump import: assembling recordings", "written", written)
				imp.setStageProgress(dumpStageCatalog, written, len(scan.recordings))
			}
		}
	}

	if err := flush(); err != nil {
		return err
	}

	// Release groups: roll release listen counts up to the redirect
	// target, keep the top keepReleaseGroup, and title each group
	// after its most-listened kept release.
	type rgAgg struct {
		listens   uint32
		bestRel   uuid16
		bestCnt   uint32
		canonical uuid16
	}

	rgs := make(map[uuid16]*rgAgg, len(scan.releaseToRG))

	for rel, target := range scan.releaseToRG {
		cnt := ks.releases[rel]

		agg := rgs[target.rg]
		if agg == nil {
			agg = &rgAgg{}
			rgs[target.rg] = agg
		}

		agg.listens += cnt

		if cnt >= agg.bestCnt {
			agg.bestCnt = cnt
			agg.bestRel = rel
			agg.canonical = target.canonical
		}
	}

	rgFloorVals := make([]uint32, 0, len(rgs))
	for _, agg := range rgs {
		rgFloorVals = append(rgFloorVals, agg.listens)
	}

	rgFloor := floorForBudget(rgFloorVals, keepReleaseGroup, releaseMinListenFloor)

	// S2: keep each target artist's top release groups even below the
	// global RG floor.  Attributed via the best release's artist credit.
	keepRG := make(map[uuid16]struct{})

	rgPickers := make(map[uuid16]*artistTopRG)

	for rg, agg := range rgs {
		info, ok := scan.releaseInfos[agg.bestRel]
		if !ok {
			continue
		}

		var aID uuid16

		if !parseUUID(info.artistMBID, aID[:]) {
			continue
		}

		budget, target := ks.rgBudget[aID]
		if !target {
			continue
		}

		p := rgPickers[aID]
		if p == nil {
			p = &artistTopRG{n: budget, inSet: make(map[uuid16]struct{}, budget)}
			rgPickers[aID] = p
		}

		p.add(rg, agg.listens)
	}

	for _, p := range rgPickers {
		for _, c := range p.rgs {
			keepRG[c.rg] = struct{}{}
		}
	}

	rgWritten := 0
	rgFromPerArtist := 0

	for rg, agg := range rgs {
		if agg.listens < rgFloor {
			if _, keep := keepRG[rg]; !keep {
				continue
			}

			rgFromPerArtist++
		}

		info, ok := scan.releaseInfos[agg.bestRel]
		if !ok {
			// No kept release carries display info for this group.
			continue
		}

		batch = append(batch, SearchIndexResult{
			EntityType:     "release_group",
			MBID:           formatUUID(rg[:]),
			Title:          info.name,
			ArtistName:     info.artistName,
			ArtistMBID:     info.artistMBID,
			Popularity:     int(agg.listens),
			CAAReleaseMBID: formatUUID(agg.canonical[:]),
		})

		rgWritten++

		if len(batch) >= dumpAssembleBatch {
			if err := flush(); err != nil {
				return err
			}
		}
	}

	if err := flush(); err != nil {
		return err
	}

	// Artists (only those whose name is derivable from a solo credit;
	// the metadata patch pass fills the rest from ListenBrainz).
	artWritten := 0

	for mbid, name := range scan.artistNames {
		batch = append(batch, SearchIndexResult{
			EntityType: "artist",
			MBID:       formatUUID(mbid[:]),
			Title:      name,
			ArtistName: name,
			ArtistMBID: formatUUID(mbid[:]),
			Popularity: int(ks.artists[mbid]),
		})

		artWritten++

		if len(batch) >= dumpAssembleBatch {
			if err := flush(); err != nil {
				return err
			}
		}
	}

	if err := flush(); err != nil {
		return err
	}

	imp.logger.Info("dump import: index assembled",
		"recordings", written,
		"releaseGroups", rgWritten,
		"releaseGroupsPerArtist", rgFromPerArtist,
		"artists", artWritten,
	)

	return nil
}

// mergePerArtistTracks folds the S2 per-artist top-N recordings into the
// kept recordings, skipping any already covered by the global floor so no
// row is written twice (upsert dedupes by MBID as a backstop anyway).
func (imp *dumpImporter) mergePerArtistTracks(ks *keptSets, scan *canonicalScan) {
	added := 0

	for _, a := range scan.artistTracks.byArtist {
		for _, row := range a.rows {
			if _, global := ks.recordings[row.mbid]; global {
				continue
			}

			scan.recordings = append(scan.recordings, row)
			added++
		}
	}

	imp.logger.Info("dump import: per-artist track coverage",
		"targetArtists", len(scan.artistTracks.byArtist),
		"recordingsAdded", added,
	)
}