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. // - 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, 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 ) // 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 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 // ResolveDuplicates runs one pass. act is the operator's auto-resolve setting; // with it off, the pass only classifies. It reads only track lists already in // the catalogue; ResolveDuplicatesFetching may also fetch missing ones. func ResolveDuplicates( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, act bool, ) (DuplicateResolveResult, error) { return ResolveDuplicatesFetching(ctx, pool, logger, dataDir, act, nil) } // ResolveDuplicatesFetching is ResolveDuplicates that asks MusicBrainz, through // fetch, for the track list of an album whose wrong slot it must decide and // the catalogue has never read: a few per pass (trackListFetchesPerPass). func ResolveDuplicatesFetching( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, act bool, fetch TrackListFetcher, ) (DuplicateResolveResult, error) { if logger == nil { logger = slog.Default() } q := dbq.New(pool) res := DuplicateResolveResult{Classes: map[DuplicateClass]int{}} rows, err := q.ListDuplicateGroupsForResolve(ctx) if err != nil { return res, fmt.Errorf("list duplicate groups: %w", err) } groups := foldResolveGroups(rows) res.Groups = len(groups) for _, g := range groups { 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) 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)) } } res.Dismissed = dismissDistinct(ctx, pool, logger, groups, merged) wrong, names := settleWrongSlots(ctx, pool, logger, groups, merged, fetch) 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 } // 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}) } 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 } // 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 fetch TrackListFetcher tick time.Duration } // NewDuplicateResolveWorker builds a worker with the production cadence. fetch // reads a missing track list from MusicBrainz; nil reads only what is held. func NewDuplicateResolveWorker( pool *pgxpool.Pool, logger *slog.Logger, dataDir string, settings *FingerprintSettingsService, fetch TrackListFetcher, ) *DuplicateResolveWorker { return &DuplicateResolveWorker{ pool: pool, logger: logger, dataDir: dataDir, settings: settings, fetch: fetch, tick: duplicateResolveTick, } } // Run blocks until ctx is cancelled, running a pass at start and then each tick. func (w *DuplicateResolveWorker) Run(ctx context.Context) { w.tickOnce(ctx) t := time.NewTicker(w.tick) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: w.tickOnce(ctx) } } } // tickOnce contains one pass so nothing it does can stop the next tick (rule 157). func (w *DuplicateResolveWorker) tickOnce(ctx context.Context) { defer func() { if r := recover(); r != nil { w.logger.Error("duplicate resolve: tick panicked", "panic", r) } }() // A sweep in flight is rewriting the groups; the next tick sees its result. _, err := dbq.New(w.pool).GetInFlightDuplicateSweep(ctx) switch { case err == nil: return case !errors.Is(err, pgx.ErrNoRows): if ctx.Err() == nil { w.logger.Warn("duplicate resolve: sweep check failed", "err", err) } return } res, err := ResolveDuplicatesFetching(ctx, w.pool, w.logger, w.dataDir, w.settings.Get().AutoResolve, w.fetch) 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) }