minstrel/internal/library/duplicate_resolve.go

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; without Lidarr since 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.
//   - 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
	// 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, ClassMismatch:
		if i := g.lone(); i >= 0 {
			return true, i
		}
	}
	return false, -1
}

// ResolveDuplicates runs one pass. act is the operator's auto-resolve setting;
// with it off, the pass only classifies.
func ResolveDuplicates(
	ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, act bool,
) (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 {
		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{}
	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++
	}

	linked, err := linkCrossRelease(ctx, pool, logger, groups, merged)
	res.SongsLinked = linked
	if err != nil {
		return res, err
	}

	if res.Merged > 0 {
		notifyAdmins(ctx, notifications.KindDuplicatesResolved, notifications.Payload{
			Count:  int64(res.Merged),
			Detail: fmt.Sprintf("Merged %d duplicate %s.", res.Merged, pluralWord(res.Merged, "copy", "copies")),
		})
	}
	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 "Identical audio filed under different titles: one file carries another song's tags."
	}
	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
}

func pluralWord(n int, one, many string) string {
	if n == 1 {
		return one
	}
	return many
}

// DuplicateResolveWorker runs a resolver pass every hour.
type DuplicateResolveWorker struct {
	pool     *pgxpool.Pool
	logger   *slog.Logger
	dataDir  string
	settings *FingerprintSettingsService
	tick     time.Duration
}

// NewDuplicateResolveWorker builds a worker with the production cadence.
func NewDuplicateResolveWorker(
	pool *pgxpool.Pool, logger *slog.Logger, dataDir string, settings *FingerprintSettingsService,
) *DuplicateResolveWorker {
	return &DuplicateResolveWorker{pool: pool, logger: logger, dataDir: dataDir, settings: settings, 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 := ResolveDuplicates(ctx, w.pool, w.logger, w.dataDir, w.settings.Get().AutoResolve)
	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, "held_back", res.HeldBack, "released", res.Released)
}