//go:build indexbuild package explore import ( "archive/tar" "bufio" "compress/bzip2" "context" "errors" "fmt" "io" "path" "regexp" "strconv" "strings" ) // Multi-artist credits, from the core MusicBrainz dump. // // A credit is ordered parts and the credit *string* is derived from // them; MusicBrainz's own artist_credit.name is a cached render. What // this pass extracts is the decomposition: for each catalog recording // and release group whose credit names more than one artist, the // credited artists in order, each with the name *as credited* and the // join phrase that follows it. See artist_credit_part.sql for why that // is stored rather than derived, and why nothing may reconstruct a // credit by searching a name inside a credit string. // // It is a separate dump from everything else here, and it has to be. // The canonical dump this importer already streams gives artist_mbids // (an ordered list) and artist_credit_name (the *rendered* string) -- // no join phrases, and no per-artist as-credited names. Splitting the // rendered string using canonical artist names fails on exactly the // credits that matter: measured on a real library, 21% of multi-artist // credits name an artist differently from the artist's own name // ("Snoop Dogg" credited on a track by "Snoop Doggy Dogg"), so the // substring is simply not there. The JSON dumps were checked too and // cover 153,691 recordings of ~35M, with zero overlap against a real // library. This dump is the only source. // // Cost, measured on the 20260815 export: 7.1 GB compressed, decompressed // by pure-Go compress/bzip2 at ~26 MB/s uncompressed (~13.7 min for the // whole file, single-threaded). cmd/indexbuild is built CGO_ENABLED=0, // so the stdlib decompressor is what there is -- and it is fine, because // the 2 MB/s origin throttle dominates, as it does for every other dump // here. const ( // defaultMBDumpBaseURL is the core MusicBrainz export. Only // mbdump.tar.bz2 is fetched; the other tarballs there hold data this // app has no use for. defaultMBDumpBaseURL = "https://data.metabrainz.org/pub/musicbrainz/data/fullexport/" ) var ( mbdumpDirRe = regexp.MustCompile(`^\d{8}-\d+$`) mbdumpFileRe = regexp.MustCompile(`^mbdump\.tar\.bz2$`) // ErrDumpShape is returned when a dump member does not have the // columns this code was written against. It is deliberately fatal: // reading the wrong column silently produces a catalog whose credits // are subtly wrong, which is far worse than a failed build. ErrDumpShape = errors.New("musicbrainz dump member has an unexpected shape") ) // Column positions in the Postgres COPY output, verified against the // 20260815 export. There is no header row to read them from, so they // are asserted instead -- see checkShape. const ( artistColID = 0 artistColGID = 1 artistColMin = 2 creditColID = 0 creditColArtistCount = 2 creditColMin = 3 partColCredit = 0 partColPosition = 1 partColArtist = 2 partColName = 3 partColJoin = 4 partColMin = 5 // recording and release_group share a layout in the columns this // pass reads: id, gid, name, artist_credit, ... entityColGID = 1 entityColCredit = 3 entityColMin = 4 ) // creditPart is one credited artist within a credit. type creditPart struct { position int artistID int32 name string join string } // creditScan is what one pass over the dump collects. type creditScan struct { // artistGIDs maps an artist row id to its MBID. artist_credit_name // references artists by row id, and the tar orders `artist` before // it, so this is complete by the time it is read. artistGIDs map[int32]uuid16 // multiCredits are the credit ids naming more than one artist, from // artist_credit.artist_count. Taking the count from the dump rather // than counting parts means a credit can be rejected before its // parts are stored. multiCredits map[int32]struct{} // parts are the decompositions of multiCredits, keyed by credit id. parts map[int32][]creditPart // refs maps a kept catalog entity to its credit. Only entities in // explore_index and only multi-artist credits: everything else is // already described by explore_index's own artist_name/artist_mbid. refs map[uuid16]int32 // used are the credits some ref actually points at, which is a small // fraction of multiCredits -- the catalog keeps ~1.8M entities of // MusicBrainz's tens of millions. used map[int32]struct{} skippedUnknownArtist int } // creditsImportDoneKey marks in explore_index_meta that the credit pass // has run against the current catalog. // // It is its own marker rather than part of the import's stage state for // a resume reason: the credit pass runs *after* the catalog is // assembled, and a failure in it must not send the next run back // through the ~205 GB it just finished. Marking separately means a // retry retries only this. const creditsImportDoneKey = "credits_import_done" // ensureArtistCredits runs the credit pass unless it has already run // against this catalog, reporting whether it newly populated them. // // Called from both of run's paths -- the full import and the resume // that finds the rows already assembled -- and from the maintenance // entry point below, since a catalog built before credits existed is // otherwise never offered a chance to gain them: the index job picks // its mode from the index's own state, and a complete import means // "refresh", which never enters run() at all. // // The return value is what tells the job there is something new worth // publishing. A refresh otherwise reports "changed" only when the // listens series advanced, so credits would sit in the CI database and // never reach an artifact. func (imp *dumpImporter) ensureArtistCredits(ctx context.Context) bool { if imp.si.hasMeta(creditsImportDoneKey) { return false } url, err := discoverDumpFile( ctx, imp.httpClient, imp.mbdumpBaseURL, mbdumpDirRe, mbdumpFileRe, ) if err != nil { imp.logger.Warn("credit import: could not find the dump", "error", err) return false } if err := imp.importArtistCredits(ctx, url); err != nil { // A catalog without credits is the catalog this app shipped // before them: every credit falls back to its single artist. // That is worth far less than failing an import that otherwise // succeeded. imp.logger.Warn("credit import: failed", "error", err) return false } imp.si.setMeta(creditsImportDoneKey, "1") return true } // EnsureArtistCredits tops up the credit tables outside a full import. // // It exists because the index job's modes are decided from the index's // own state: a cache holding a completed import chooses `refresh`, // which folds in incremental listens and never enters the dump // importer. Without this, a catalog built before the credit pass // existed could only gain credits from a `rebuild` -- and a rebuild // re-downloads ~205 GB to reproduce rows it already has, to add // something that costs 7 GB on its own. // // Reports whether credits were newly populated, so the caller knows // there is a new artifact worth publishing. func (e *Service) EnsureArtistCredits(ctx context.Context) bool { imp, err := newDumpImporter(e.index, e.lb) if err != nil { e.index.logger.Warn("credit import: could not start", "error", err) return false } return imp.ensureArtistCredits(ctx) } // importArtistCredits streams the core MusicBrainz dump and fills // artist_credit_part and artist_credit_ref for the entities the catalog // kept. // // It runs after assembleIndex because it asks explore_index which // entities those are: the popularity filter decides what is worth // carrying credits for, and asking the table rather than the kept sets // means this stays correct if that filter changes. func (imp *dumpImporter) importArtistCredits(ctx context.Context, url string) error { kept, err := imp.keptEntityMBIDs(ctx) if err != nil { return err } if len(kept) == 0 { imp.logger.Warn("credit import: no catalog entities, skipping") return nil } imp.logger.Info("credit import: starting", "url", url, "entities", len(kept)) imp.logJob("Streaming MusicBrainz dump for artist credits") scan, err := imp.scanCreditDump(ctx, url, kept) if err != nil { return err } imp.logger.Info("credit import: scanned", "multiArtistCredits", len(scan.multiCredits), "entitiesWithMultiArtistCredit", len(scan.refs), "creditsUsed", len(scan.used), ) return imp.writeCredits(ctx, scan) } // keptEntityMBIDs is every recording and release group in the catalog. // Artists are excluded: an artist is not credited to a credit. func (imp *dumpImporter) keptEntityMBIDs(ctx context.Context) (map[uuid16]struct{}, error) { rows, err := imp.si.db.QueryContextWith(ctx, `SELECT mbid FROM explore_index WHERE entity_type IN (2 /* release_group */, 3 /* recording */)`, ) if err != nil { return nil, fmt.Errorf("credit import: read catalog entities: %w", err) } defer func() { _ = rows.Close() }() out := make(map[uuid16]struct{}) for rows.Next() { var raw []byte if err := rows.Scan(&raw); err != nil { return nil, fmt.Errorf("credit import: scan mbid: %w", err) } if len(raw) != len(uuid16{}) { continue } var id uuid16 copy(id[:], raw) out[id] = struct{}{} } if err := rows.Err(); err != nil { return nil, fmt.Errorf("credit import: read catalog entities: %w", err) } return out, nil } // scanCreditDump makes one sequential pass over mbdump.tar.bz2. // // The tar's members are alphabetical, which is what makes a single pass // possible without buffering the big ones: `artist` and // `artist_credit_name` both arrive before `recording` and // `release_group`, so by the time an entity names a credit, that // credit's parts and their artists' MBIDs are already known and the // entity can be resolved and dropped. 35M recording rows are never // held. // // The order is not depended on blindly: an entity naming a credit that // has not been seen is counted and reported rather than silently // producing an empty catalog, which is what a reordered export would // otherwise look like. func (imp *dumpImporter) scanCreditDump( ctx context.Context, url string, kept map[uuid16]struct{}, ) (*creditScan, error) { stream := imp.openDumpStream(ctx, url, 0) defer func() { _ = stream.Close() }() return imp.scanCreditTar( ctx, tar.NewReader(bzip2.NewReader(bufio.NewReaderSize(stream, 1<<20))), kept, ) } // scanCreditTar is the parse, separated from the fetch so it can be // driven by a tar built in a test. compress/bzip2 is decompress-only, // so a test cannot produce the real container. func (imp *dumpImporter) scanCreditTar( ctx context.Context, tr *tar.Reader, kept map[uuid16]struct{}, ) (*creditScan, error) { scan := &creditScan{ artistGIDs: make(map[int32]uuid16), multiCredits: make(map[int32]struct{}), parts: make(map[int32][]creditPart), refs: make(map[uuid16]int32), used: make(map[int32]struct{}), } 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("credit import: tar: %w", err) } if hdr.Typeflag != tar.TypeReg { continue } done, err := imp.scanCreditMember(ctx, hdr.Name, tr, kept, scan) if err != nil { return nil, err } if done { // Everything this pass needs has been read; the rest of the // tarball is other entities' data and decompressing it would // cost minutes for nothing. break } } if scan.skippedUnknownArtist > 0 { imp.logger.Warn("credit import: credits dropped for unknown artists", "count", scan.skippedUnknownArtist, ) } return scan, nil } // scanCreditMember dispatches one tar member, reporting whether the // pass has everything it needs. func (imp *dumpImporter) scanCreditMember( ctx context.Context, name string, r io.Reader, kept map[uuid16]struct{}, scan *creditScan, ) (bool, error) { switch path.Base(name) { case "artist": return false, imp.scanArtists(ctx, r, scan) case "artist_credit": return false, imp.scanCredits(ctx, r, scan) case "artist_credit_name": return false, imp.scanCreditParts(ctx, r, scan) case "recording", "release_group": if err := imp.scanCreditedEntities(ctx, r, kept, scan); err != nil { return false, err } // release_group sorts after recording, so the pass is complete // once it has been read. return path.Base(name) == "release_group", nil default: return false, nil } } // scanArtists records every artist's MBID by row id. func (imp *dumpImporter) scanArtists( ctx context.Context, r io.Reader, scan *creditScan, ) error { return scanTSV(ctx, r, artistColMin, "artist", func(fields []string) error { id, ok := parseInt32(fields[artistColID]) if !ok { return nil } var gid uuid16 if !parseUUID(fields[artistColGID], gid[:]) { return fmt.Errorf("%w: artist.gid is not a UUID: %q", ErrDumpShape, truncate(fields[artistColGID])) } scan.artistGIDs[id] = gid return nil }) } // scanCredits records which credits name more than one artist. func (imp *dumpImporter) scanCredits( ctx context.Context, r io.Reader, scan *creditScan, ) error { return scanTSV(ctx, r, creditColMin, "artist_credit", func(fields []string) error { id, ok := parseInt32(fields[creditColID]) if !ok { return nil } count, ok := parseInt32(fields[creditColArtistCount]) if !ok { return fmt.Errorf("%w: artist_credit.artist_count is not a number: %q", ErrDumpShape, truncate(fields[creditColArtistCount])) } if count > 1 { scan.multiCredits[id] = struct{}{} } return nil }) } // scanCreditParts records the decomposition of every multi-artist // credit. func (imp *dumpImporter) scanCreditParts( ctx context.Context, r io.Reader, scan *creditScan, ) error { return scanTSV(ctx, r, partColMin, "artist_credit_name", func(fields []string) error { credit, ok := parseInt32(fields[partColCredit]) if !ok { return nil } if _, multi := scan.multiCredits[credit]; !multi { return nil } position, ok := parseInt32(fields[partColPosition]) if !ok { return nil } artist, ok := parseInt32(fields[partColArtist]) if !ok { return nil } scan.parts[credit] = append(scan.parts[credit], creditPart{ position: int(position), artistID: artist, name: fields[partColName], join: fields[partColJoin], }) return nil }) } // scanCreditedEntities resolves recordings and release groups against // the catalog, keeping only those the catalog holds and whose credit // names more than one artist. func (imp *dumpImporter) scanCreditedEntities( ctx context.Context, r io.Reader, kept map[uuid16]struct{}, scan *creditScan, ) error { return scanTSV(ctx, r, entityColMin, "recording/release_group", func(fields []string) error { var gid uuid16 if !parseUUID(fields[entityColGID], gid[:]) { return fmt.Errorf("%w: entity gid is not a UUID: %q", ErrDumpShape, truncate(fields[entityColGID])) } if _, want := kept[gid]; !want { return nil } credit, ok := parseInt32(fields[entityColCredit]) if !ok { return fmt.Errorf("%w: entity artist_credit is not a number: %q", ErrDumpShape, truncate(fields[entityColCredit])) } if _, multi := scan.multiCredits[credit]; !multi { return nil } scan.refs[gid] = credit scan.used[credit] = struct{}{} return nil }) } // scanTSV reads Postgres COPY output a line at a time, unescaping each // field and handing the row to fn. // // The shape is asserted on the first row rather than trusted: this dump // has no header, so a column that moved would otherwise be read as a // neighbouring one and produce a catalog that is quietly wrong. func scanTSV( ctx context.Context, r io.Reader, minCols int, member string, fn func(fields []string) error, ) error { sc := bufio.NewScanner(r) sc.Buffer(make([]byte, 0, 1<<20), 1<<24) checked := false rows := 0 for sc.Scan() { rows++ if rows%(1<<20) == 0 { if err := ctx.Err(); err != nil { return err } } line := sc.Text() if line == "" { continue } fields := strings.Split(line, "\t") if len(fields) < minCols { if !checked { return fmt.Errorf("%w: %s has %d columns, need at least %d", ErrDumpShape, member, len(fields), minCols) } continue } checked = true for i := range fields { fields[i] = unescapeCopy(fields[i]) } if err := fn(fields); err != nil { return err } } if err := sc.Err(); err != nil { return fmt.Errorf("credit import: read %s: %w", member, err) } return nil } // unescapeCopy undoes Postgres COPY's text escaping. A NULL (\N) is // returned as an empty string: every field this pass reads is either a // number it will reject or a name whose absence means the same as // empty. func unescapeCopy(s string) string { if s == `\N` { return "" } if !strings.ContainsRune(s, '\\') { return s } var b strings.Builder b.Grow(len(s)) for i := 0; i < len(s); i++ { if s[i] != '\\' || i+1 >= len(s) { b.WriteByte(s[i]) continue } i++ switch s[i] { case 'n': b.WriteByte('\n') case 't': b.WriteByte('\t') case 'r': b.WriteByte('\r') case 'b': b.WriteByte('\b') case 'f': b.WriteByte('\f') case 'v': b.WriteByte('\v') case '\\': b.WriteByte('\\') default: b.WriteByte('\\') b.WriteByte(s[i]) } } return b.String() } func parseInt32(s string) (int32, bool) { n, err := strconv.ParseInt(s, 10, 32) if err != nil { return 0, false } return int32(n), true } // truncate bounds an error message built from dump data, which is // attacker-free but can be long. func truncate(s string) string { const limit = 64 if len(s) <= limit { return s } return s[:limit] + "..." }