package library import ( "context" "errors" "fmt" "log/slog" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/audit" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/notifications" syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) // Duplicate resolver (M498 #5435, #5437; M502 #5517). // // The sweep proposes duplicate groups; the resolver decides what each one is // and settles the ones that are safe to settle without asking. Minstrel is the // only program writing the library, so no copy is kept because something else // depends on it. What decides is what the copies are: // // - On one album, a song is held once: the extra copies are merged into the // best by ProposeSurvivor. // - Across releases, each copy belongs to its own release (a single and its // album): nothing is removed, and the copies are linked as one song. An // album holding the song twice among them still holds it once: its extra // copies are merged into its best (Humanz held Saturnz Barz twice beside // the single, #5401). // - Except that a video rip is no release's copy. When every copy but one is // a rip, across releases or under another title, the rips are merged into // the one that is not. // - Two versions of a song on one album that sound alike are not duplicates: // the group is dismissed (duplicate_settle.go). // - One file filling several of an album's track slots: MusicBrainz's track // lengths say which slot it belongs to, or AcoustID which song it is when // each copy was cut to its own slot's length, and the copies in the other // slots are removed so the album is fetched again (duplicate_settle.go). // - Anything else waits for the operator. const ( // resolveMergeCap bounds the merges one pass makes. The first pass after // deploy meets the whole backlog; a cap spreads it over a few hours, each // pass short. resolveMergeCap = 200 // duplicateResolveTick is how often the worker runs a pass. duplicateResolveTick = time.Hour // scanWaitRetry is how soon a pass put off for a running scan is tried // again: a boot scan takes under a minute on the operator's library. scanWaitRetry = time.Minute ) // DuplicateResolveResult tallies one pass. type DuplicateResolveResult struct { Groups int Classes map[DuplicateClass]int Merged int MergeFailed int // SongsLinked counts cross-release groups newly linked as one song. SongsLinked int // Dismissed counts groups of distinct versions dismissed; WrongFiles the // copies removed from track slots that were not theirs. Dismissed, WrongFiles int // HeldBack counts tracks newly held back from radio and the mixes as video // rips; Released counts held-back tracks whose name no longer says so. HeldBack, Released int64 } // resolveGroup is one pending group and its present copies. type resolveGroup struct { id pgtype.UUID tier string // worstBER is how far apart the least alike copies' fingerprints are, for // an acoustic group; nil for an exact one. worstBER *float32 members []dbq.ListDuplicateGroupsForResolveRow class DuplicateClass note string } // isRip says whether a copy is a video rip: held back as one, or named like // one and not kept by the operator. The name is read as well as the verdict // because a pass reads the groups before it flags new rips. func isRip(m dbq.ListDuplicateGroupsForResolveRow) bool { if m.SourceVerdict != nil { return *m.SourceVerdict == "suspect" } return len(SourceMarkersFor(m.FilePath)) > 0 } // lone returns the index of the group's only copy that is not a rip, or -1 // when there is none or more than one. func (g *resolveGroup) lone() int { at := -1 for i, m := range g.members { if isRip(m) { continue } if at >= 0 { return -1 } at = i } return at } // mergePlan says whether the resolver may merge the group on its own, and into // which copy (an index into members; -1 means ProposeSurvivor decides). func mergePlan(g *resolveGroup) (ok bool, survivor int) { switch g.class { case ClassSameRelease: return true, -1 case ClassCrossRelease: if i := g.lone(); i >= 0 { return true, i } } return false, -1 } // TrackListFetcher reads a release group's releases and track lists from // MusicBrainz into the catalogue (catalogue.Service.RefreshReleaseGroup). type TrackListFetcher func(ctx context.Context, releaseGroupMBID string) error // RecordingIdentifier asks AcoustID which recordings a file may be // (AcoustIDLookupWorker.IdentifyRecordings). type RecordingIdentifier func(ctx context.Context, path string) ([]string, error) // ResolveSources are what a pass may ask beyond the library, a few times per // pass each. Either may be nil, and the pass decides from what is held. type ResolveSources struct { // TrackLists reads the track list of an album whose wrong slot the pass // must decide and the catalogue has never read (trackListFetchesPerPass). TrackLists TrackListFetcher // Identify names the recording cut to several slots' lengths, when no // copy's own lookup has (identifiesPerPass). Identify RecordingIdentifier } // ResolveDuplicates runs one pass. act is the operator's auto-resolve setting; // with it off, the pass only classifies. It decides from what the library and // catalogue hold; ResolveDuplicatesWith may also ask MusicBrainz and AcoustID. func ResolveDuplicates( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, act bool, ) (DuplicateResolveResult, error) { return ResolveDuplicatesWith(ctx, pool, logger, dataDir, act, ResolveSources{}) } // ResolveDuplicatesWith is ResolveDuplicates that may ask src. func ResolveDuplicatesWith( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, act bool, src ResolveSources, ) (DuplicateResolveResult, error) { if logger == nil { logger = slog.Default() } q := dbq.New(pool) res := DuplicateResolveResult{Classes: map[DuplicateClass]int{}} rows, err := q.ListDuplicateGroupsForResolve(ctx, pgtype.UUID{}) if err != nil { return res, fmt.Errorf("list duplicate groups: %w", err) } groups := foldResolveGroups(rows) res.Groups = len(groups) for _, g := range groups { g.classify() res.Classes[g.class]++ } if err := writeResolveVerdicts(ctx, q, groups); err != nil { return res, err } if !act { notifyNewForReview(ctx, groups, nil) return res, nil } if res.HeldBack, err = q.FlagSuspectSourceTracks(ctx, SuspectSourcePattern); err != nil { return res, fmt.Errorf("hold back video rips: %w", err) } if res.Released, err = q.ClearStaleSuspectSourceTracks(ctx, SuspectSourcePattern); err != nil { return res, fmt.Errorf("release renamed tracks: %w", err) } // Merges before links, so a rip merged away is not linked first. merged := map[pgtype.UUID]bool{} var mergedNames []string for _, g := range groups { if res.Merged >= resolveMergeCap { break } if ctx.Err() != nil { return res, ctx.Err() } ok, survivor := mergePlan(g) if !ok { continue } if err := autoMerge(ctx, pool, logger, dataDir, g, survivor); err != nil { res.MergeFailed++ logger.Warn("duplicate resolve: merge failed", "group_id", syncpkg.FormatUUID(g.id), "err", err) continue } merged[g.id] = true res.Merged++ if len(mergedNames) < notifyItems { mergedNames = append(mergedNames, TrackLabel(g.members[0].ArtistName, g.members[0].Title)) } } folded, foldedNames := foldAlbumCopies(ctx, pool, logger, dataDir, groups, merged, resolveMergeCap-res.Merged) res.Merged += folded mergedNames = append(mergedNames, foldedNames...) res.Dismissed = dismissDistinct(ctx, pool, logger, groups, merged) wrong, names := settleWrongSlots(ctx, pool, logger, groups, merged, src) res.WrongFiles = wrong mergedNames = append(mergedNames, names...) linked, err := linkCrossRelease(ctx, pool, logger, groups, merged) res.SongsLinked = linked if err != nil { return res, err } if settledCopies := res.Merged + res.WrongFiles; settledCopies > 0 { notifyAdmins(ctx, notifications.KindDuplicatesResolved, notifications.Payload{ Count: int64(settledCopies), Items: mergedNames[:min(len(mergedNames), notifyItems)], }) } notifyNewForReview(ctx, groups, merged) return res, nil } // foldAlbumCopies merges, in each cross-release group, the extra copies one // album holds of the song into the best of that album's copies: the one that // is not a video rip when only one is not, else ProposeSurvivor's choice. // Copies whose title or file name name different versions (versionWords) are // not folded: "Suite-Pee" and "Suite-Pee (Clean Version)" carried one title tag // on 2026-10-20, and the explicit copy went. The // copies on other releases stay and are linked as one song. At most budget // copies go. It returns how many went and the songs they were, and drops them // from their groups. func foldAlbumCopies( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, groups []*resolveGroup, merged map[pgtype.UUID]bool, budget int, ) (int, []string) { n := 0 var names []string for _, g := range groups { if g.class != ClassCrossRelease || merged[g.id] || ctx.Err() != nil || n >= budget { continue } onAlbum := map[string][]int{} for i, m := range g.members { k := syncpkg.FormatUUID(m.AlbumID) + "\x00" + MatchTitleKey(m.Title, m.ArtistName) + "\x00" + versionWords(m.Title, m.FilePath) onAlbum[k] = append(onAlbum[k], i) } gone := map[int]bool{} for _, copies := range onAlbum { if len(copies) < 2 { continue } keep, reason := albumCopySurvivor(g, copies) for _, i := range copies { if i == keep || n >= budget { continue } if err := foldAlbumCopy(ctx, pool, logger, dataDir, g, i, keep, reason); err != nil { logger.Warn("duplicate resolve: merging an album's extra copy failed", "group_id", syncpkg.FormatUUID(g.id), "path", g.members[i].FilePath, "err", err) continue } gone[i] = true n++ if len(names) < notifyItems { names = append(names, TrackLabel(g.members[i].ArtistName, g.members[i].Title)) } } } if len(gone) > 0 { kept := g.members[:0:0] for i, m := range g.members { if !gone[i] { kept = append(kept, m) } } g.members = kept } } return n, names } // albumCopySurvivor picks which of one album's copies (indexes into the // group's members) to keep, and why. func albumCopySurvivor(g *resolveGroup, copies []int) (int, string) { notRip := -1 for _, i := range copies { if isRip(g.members[i]) { continue } if notRip >= 0 { notRip = -1 break } notRip = i } if notRip >= 0 { return notRip, "the copy that is not a video rip" } all := survivorCandidates(g) cands := make([]SurvivorCandidate, len(copies)) for j, i := range copies { cands[j] = all[i] } id, reason := ProposeSurvivor(cands) for _, i := range copies { if all[i].TrackID == id { return i, reason } } return copies[0], reason } // foldAlbumCopy replaces one of an album's copies with the copy kept, history // and all, and records it as the resolver's merge. func foldAlbumCopy( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, g *resolveGroup, gone, keep int, reason string, ) error { from, into := g.members[gone], g.members[keep] moved, err := ReplaceTrack(ctx, pool, logger, dataDir, from.TrackID, into.TrackID) if err != nil { return err } audit.WriteOrLog(ctx, pool, logger, pgtype.UUID{}, pgtype.UUID{}, audit.ActionDuplicateMerge, map[string]any{ "automatic": true, "group_id": syncpkg.FormatUUID(g.id), "tier": g.tier, "class": string(g.class), "survivor_track_id": syncpkg.FormatUUID(into.TrackID), "survivor_path": into.FilePath, "survivor_reason": reason + "; the album held the song twice", "removed": []map[string]string{ {"track_id": syncpkg.FormatUUID(from.TrackID), "file_path": from.FilePath}, }, "moved": map[string]any{ "play_events": moved.PlayEvents, "skip_events": moved.SkipEvents, "likes": moved.Likes, "playlist_entries": moved.PlaylistEntries, }, }) return nil } // linkCrossRelease links each cross-release group's copies as one song // (#5438): a like on one is a like on all, and the Liked list, shuffle and the // mixes count the song once. A group the pass merged has nothing left to link. Linking an already-linked group changes nothing, // so this runs every pass; likes are shared only when a link is new, since a // like made after it reaches every copy as it is made. func linkCrossRelease( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, groups []*resolveGroup, merged map[pgtype.UUID]bool, ) (int, error) { linked := 0 for _, g := range groups { if g.class != ClassCrossRelease || merged[g.id] { continue } if ctx.Err() != nil { return linked, ctx.Err() } ids := make([]pgtype.UUID, len(g.members)) for i, m := range g.members { ids[i] = m.TrackID } moved, err := linkSong(ctx, pool, ids) if err != nil { logger.Warn("duplicate resolve: song link failed", "group_id", syncpkg.FormatUUID(g.id), "err", err) continue } if moved { linked++ } } return linked, nil } // linkSong links the tracks as one song and, when that changed anything, // shares their likes across the song and logs each added like for sync. func linkSong(ctx context.Context, pool *pgxpool.Pool, ids []pgtype.UUID) (bool, error) { tx, err := pool.Begin(ctx) if err != nil { return false, err } defer func() { _ = tx.Rollback(ctx) }() tq := dbq.New(tx) link, err := tq.LinkTracksAsSong(ctx, ids) if err != nil { return false, fmt.Errorf("link: %w", err) } if link.Moved == 0 { return false, nil } added, err := tq.ShareSongLikes(ctx, link.SongKey) if err != nil { return false, fmt.Errorf("share likes: %w", err) } likeIDs := make([]string, len(added)) for i, a := range added { likeIDs[i] = syncpkg.EncodeLikeID(syncpkg.FormatUUID(a.UserID), syncpkg.FormatUUID(a.TrackID)) } if err := syncpkg.LogChanges(ctx, tx, syncpkg.EntityLikeTrack, likeIDs, syncpkg.OpUpsert); err != nil { return false, fmt.Errorf("log shared likes: %w", err) } return true, tx.Commit(ctx) } func foldResolveGroups(rows []dbq.ListDuplicateGroupsForResolveRow) []*resolveGroup { var groups []*resolveGroup for _, r := range rows { if n := len(groups); n == 0 || groups[n-1].id != r.GroupID { groups = append(groups, &resolveGroup{id: r.GroupID, tier: r.Tier, worstBER: r.WorstBitErrorRate}) } g := groups[len(groups)-1] g.members = append(g.members, r) } // A group whose other copies went missing has nothing left to resolve. out := groups[:0] for _, g := range groups { if len(g.members) >= 2 { out = append(out, g) } } return out } // classify sets the group's class and the note the report shows beside it. func (g *resolveGroup) classify() { cm := make([]ClassifyMember, len(g.members)) for i, m := range g.members { cm[i] = ClassifyMember{AlbumID: syncpkg.FormatUUID(m.AlbumID), Title: m.Title, ArtistName: m.ArtistName} } g.class = ClassifyDuplicateGroup(g.tier, cm) g.note = resolveNote(g) } // resolveNote is what the report says beside a group, when there is something // to say. func resolveNote(g *resolveGroup) string { switch g.class { case ClassCrossRelease: if g.lone() < 0 { return "Each copy belongs to its own release, so none is removed; they count as one song." } case ClassMismatch: return "One recording fills several of this album's track slots, and MusicBrainz's track lengths could not say which is its own." } return "" } func writeResolveVerdicts(ctx context.Context, q *dbq.Queries, groups []*resolveGroup) error { classes := dbq.SetDuplicateGroupClassesParams{} for _, g := range groups { classes.GroupIds = append(classes.GroupIds, g.id) classes.Classes = append(classes.Classes, string(g.class)) classes.Notes = append(classes.Notes, g.note) } if len(classes.GroupIds) == 0 { return nil } if err := q.SetDuplicateGroupClasses(ctx, classes); err != nil { return fmt.Errorf("write duplicate classes: %w", err) } return nil } // survivorCandidates builds what ProposeSurvivor weighs for a group's copies. func survivorCandidates(g *resolveGroup) []SurvivorCandidate { cands := make([]SurvivorCandidate, len(g.members)) for i, m := range g.members { cands[i] = SurvivorCandidate{ TrackID: syncpkg.FormatUUID(m.TrackID), FileFormat: m.FileFormat, FileSize: m.FileSize, AddedAt: m.AddedAt.Time, TagFit: TagFitScore(m.TrackNumber != nil, m.PositionClash, m.FilePath, m.HasMbid), } } return cands } // autoMerge merges one group into the copy at index survivor, or the copy // ProposeSurvivor picks when survivor is -1, and records the merge as the // resolver's. func autoMerge( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, g *resolveGroup, survivor int, ) error { survivorKey, reason := ProposeSurvivor(survivorCandidates(g)) if survivor >= 0 { survivorKey, reason = syncpkg.FormatUUID(g.members[survivor].TrackID), "the copy that is not a video rip" } var survivorID pgtype.UUID if err := survivorID.Scan(survivorKey); err != nil { return fmt.Errorf("parse survivor id: %w", err) } res, err := MergeDuplicateGroup(ctx, pool, logger, dataDir, g.id, survivorID) if err != nil { return err } if err := dbq.New(pool).MarkDuplicateGroupResolvedAutomatically(ctx, g.id); err != nil { logger.Warn("duplicate resolve: marking the merge automatic failed", "group_id", syncpkg.FormatUUID(g.id), "err", err) } removed := make([]map[string]string, 0, len(res.Removed)) for _, c := range res.Removed { removed = append(removed, map[string]string{"track_id": syncpkg.FormatUUID(c.TrackID), "file_path": c.FilePath}) } audit.WriteOrLog(ctx, pool, logger, pgtype.UUID{}, pgtype.UUID{}, audit.ActionDuplicateMerge, map[string]any{ "automatic": true, "group_id": syncpkg.FormatUUID(g.id), "tier": res.Tier, "class": string(g.class), "survivor_track_id": syncpkg.FormatUUID(res.Survivor.TrackID), "survivor_path": res.Survivor.FilePath, "survivor_reason": reason, "removed": removed, "moved": map[string]any{ "play_events": res.PlayEvents, "skip_events": res.SkipEvents, "likes": res.Likes, "playlist_entries": res.PlaylistEntries, }, }) return nil } // DuplicateResolveWorker runs a resolver pass every hour. type DuplicateResolveWorker struct { pool *pgxpool.Pool logger *slog.Logger dataDir string settings *FingerprintSettingsService src ResolveSources tick time.Duration } // NewDuplicateResolveWorker builds a worker with the production cadence. src // is what a pass may ask beyond the library. func NewDuplicateResolveWorker( pool *pgxpool.Pool, logger *slog.Logger, dataDir string, settings *FingerprintSettingsService, src ResolveSources, ) *DuplicateResolveWorker { return &DuplicateResolveWorker{ pool: pool, logger: logger, dataDir: dataDir, settings: settings, src: src, tick: duplicateResolveTick, } } // Run blocks until ctx is cancelled, running a pass at start and then each // tick. A pass put off for a running scan is tried again after scanWaitRetry. func (w *DuplicateResolveWorker) Run(ctx context.Context) { t := time.NewTicker(w.tick) defer t.Stop() soon := time.NewTimer(0) defer soon.Stop() for { select { case <-ctx.Done(): return case <-t.C: case <-soon.C: } if w.tickOnce(ctx) { soon.Reset(scanWaitRetry) } } } // tickOnce contains one pass so nothing it does can stop the next tick (rule // 157). It reports whether the pass was put off for a running scan. func (w *DuplicateResolveWorker) tickOnce(ctx context.Context) (waitForScan bool) { defer func() { if r := recover(); r != nil { w.logger.Error("duplicate resolve: tick panicked", "panic", r) } }() q := dbq.New(w.pool) // A sweep in flight is rewriting the groups; the next tick sees its result. _, err := q.GetInFlightDuplicateSweep(ctx) switch { case err == nil: return false case !errors.Is(err, pgx.ErrNoRows): if ctx.Err() == nil { w.logger.Warn("duplicate resolve: sweep check failed", "err", err) } return false } // A scan in flight has listed the files it will reconcile: a copy removed // now is marked present again when it does (7 of the 13 removed at boot // on 2026-10-20, #6169). A run past StuckScanThreshold is a dead one. run, err := q.GetInFlightScanRun(ctx) switch { case err == nil && time.Since(run.StartedAt.Time) <= StuckScanThreshold: return true case err != nil && !errors.Is(err, pgx.ErrNoRows): if ctx.Err() == nil { w.logger.Warn("duplicate resolve: scan check failed", "err", err) } return false } res, err := ResolveDuplicatesWith(ctx, w.pool, w.logger, w.dataDir, w.settings.Get().AutoResolve, w.src) if err != nil && ctx.Err() == nil { w.logger.Warn("duplicate resolve: pass failed", "err", err) } w.logger.Info("duplicate resolve complete", "groups", res.Groups, "merged", res.Merged, "merge_failed", res.MergeFailed, "songs_linked", res.SongsLinked, "dismissed", res.Dismissed, "wrong_files", res.WrongFiles, "held_back", res.HeldBack, "released", res.Released) return false }