package library import ( "context" "errors" "fmt" "log/slog" "sort" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) // Duplicate merge (M400 #3911). // ErrDuplicateGroupNotPending means the group was already merged or dismissed, // no longer exists, or no longer has two members to merge. var ErrDuplicateGroupNotPending = errors.New("library: duplicate group is not pending") // ErrSurvivorNotInGroup means the copy chosen to keep is not a member of the group. var ErrSurvivorNotInGroup = errors.New("library: survivor is not a member of the group") // MergedCopy is one copy a merge kept or removed. type MergedCopy struct { TrackID pgtype.UUID FilePath string TrackMbid *string AlbumMbid *string } // MergeResult says what a merge did. type MergeResult struct { Tier string Survivor MergedCopy Removed []MergedCopy // What moved onto the survivor — reported so the operator, and the audit // log, can see that the history was kept rather than take it on trust. PlayEvents int64 SkipEvents int64 Likes int // users whose like now sits on the survivor PlaylistEntries int DeletedAlbumIDs []pgtype.UUID DeletedArtistIDs []pgtype.UUID } // MergeDuplicateGroup keeps one copy of a duplicate group and removes the rest, // carrying everything the removed copies held onto the one kept. // // survivorID chooses the copy to keep; an invalid (zero) id takes the proposal // from ProposeSurvivor. // // The danger this is built around: every table referencing tracks does so ON // DELETE CASCADE, so deleting a duplicate's row outright silently destroys its // likes, plays, playlist entries and tags. The merge moves all of that onto the // survivor first, and only then deletes the now-empty row. // // It deletes the removed copies' FILES too, and first, before any row changes // (#3918, note #3926). A merge that left the file behind would be undone by the // next scan, which re-imports it as a new track with no history. If a file cannot // be removed, the *FileRemoveError comes back and nothing in the database changes. // With several copies to remove, one file may already be gone when a later one // fails; that copy's row keeps all its history and is marked missing by the next // scan, and retrying the merge picks up where it stopped. // // Everything else happens in one transaction, which holds a lock on the group so // two merges of it cannot run at once. Sync changes for clients' caches are logged // inside it, the way the playlists service logs its own. func MergeDuplicateGroup( ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, dataDir string, groupID, survivorID pgtype.UUID, ) (MergeResult, error) { if logger == nil { logger = slog.Default() } tx, err := pool.Begin(ctx) if err != nil { return MergeResult{}, fmt.Errorf("begin merge: %w", err) } defer func() { _ = tx.Rollback(ctx) }() tq := dbq.New(tx) group, err := tq.LockDuplicateGroupForMerge(ctx, groupID) if errors.Is(err, pgx.ErrNoRows) { return MergeResult{}, ErrDuplicateGroupNotPending } if err != nil { return MergeResult{}, fmt.Errorf("lock duplicate group: %w", err) } if group.Status != "pending" { return MergeResult{}, ErrDuplicateGroupNotPending } members, err := tq.ListDuplicateGroupMergeMembers(ctx, groupID) if err != nil { return MergeResult{}, fmt.Errorf("list group members: %w", err) } if len(members) < 2 { return MergeResult{}, ErrDuplicateGroupNotPending } survivor, losers, err := splitSurvivor(members, survivorID) if err != nil { return MergeResult{}, err } for _, l := range losers { if err := removeTrackFileOnDisk(l.FilePath); err != nil { return MergeResult{}, err } } res := MergeResult{Tier: group.Tier, Survivor: mergedCopyOf(survivor)} likers := map[string]struct{}{} changes := mergeChanges{} survivorKey := syncpkg.FormatUUID(survivor.ID) for _, l := range losers { ids := struct{ survivor, loser pgtype.UUID }{survivor.ID, l.ID} loserKey := syncpkg.FormatUUID(l.ID) n, err := tq.MergeRepointPlayEvents(ctx, dbq.MergeRepointPlayEventsParams{SurvivorID: ids.survivor, LoserID: ids.loser}) if err != nil { return MergeResult{}, fmt.Errorf("move play events: %w", err) } res.PlayEvents += n n, err = tq.MergeRepointSkipEvents(ctx, dbq.MergeRepointSkipEventsParams{SurvivorID: ids.survivor, LoserID: ids.loser}) if err != nil { return MergeResult{}, fmt.Errorf("move skip events: %w", err) } res.SkipEvents += n if _, err := tq.MergeRepointContextualLikes(ctx, dbq.MergeRepointContextualLikesParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("move contextual likes: %w", err) } if _, err := tq.MergeRepointPlaybackErrors(ctx, dbq.MergeRepointPlaybackErrorsParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("move playback errors: %w", err) } if _, err := tq.MergeRepointLidarrRequests(ctx, dbq.MergeRepointLidarrRequestsParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("move lidarr request matches: %w", err) } playlists, err := tq.MergeRepointPlaylistTracks(ctx, dbq.MergeRepointPlaylistTracksParams{SurvivorID: ids.survivor, LoserID: ids.loser}) if err != nil { return MergeResult{}, fmt.Errorf("move playlist entries: %w", err) } res.PlaylistEntries += len(playlists) for _, pl := range playlists { plKey := syncpkg.FormatUUID(pl) changes.playlistDelete(syncpkg.EncodePlaylistTrackID(plKey, loserKey)) changes.playlistUpsert(syncpkg.EncodePlaylistTrackID(plKey, survivorKey)) } users, err := tq.MergeCopyGeneralLikes(ctx, dbq.MergeCopyGeneralLikesParams{SurvivorID: ids.survivor, LoserID: ids.loser}) if err != nil { return MergeResult{}, fmt.Errorf("move likes: %w", err) } for _, u := range users { userKey := syncpkg.FormatUUID(u) likers[userKey] = struct{}{} changes.likeDelete(syncpkg.EncodeLikeID(userKey, loserKey)) changes.likeUpsert(syncpkg.EncodeLikeID(userKey, survivorKey)) } if _, err := tq.MergeCopyTrackTags(ctx, dbq.MergeCopyTrackTagsParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("merge tags: %w", err) } if _, err := tq.MergeCopyTrackSimilarity(ctx, dbq.MergeCopyTrackSimilarityParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("merge similarity: %w", err) } if err := tq.MergeInheritTrackMbid(ctx, dbq.MergeInheritTrackMbidParams{SurvivorID: ids.survivor, LoserID: ids.loser}); err != nil { return MergeResult{}, fmt.Errorf("inherit recording mbid: %w", err) } // Everything the loser carried now sits on the survivor, so the CASCADE // this delete sets off has nothing left to destroy. deleted, err := tq.DeleteTrack(ctx, l.ID) if err != nil { return MergeResult{}, fmt.Errorf("delete merged copy: %w", err) } tidied, err := tidyEmptiedAlbum(ctx, tq, deleted.AlbumID) if err != nil { return MergeResult{}, err } if tidied.AlbumID != nil { res.DeletedAlbumIDs = append(res.DeletedAlbumIDs, *tidied.AlbumID) } if tidied.ArtistID != nil { res.DeletedArtistIDs = append(res.DeletedArtistIDs, *tidied.ArtistID) } res.Removed = append(res.Removed, mergedCopyOf(l)) changes.trackDelete(loserKey) } res.Likes = len(likers) marked, err := tq.MarkDuplicateGroupMerged(ctx, groupID) if err != nil { return MergeResult{}, fmt.Errorf("mark group merged: %w", err) } if marked != 1 { return MergeResult{}, ErrDuplicateGroupNotPending } if err := changes.log(ctx, tx); err != nil { return MergeResult{}, err } if err := tx.Commit(ctx); err != nil { return MergeResult{}, fmt.Errorf("commit merge: %w", err) } // After commit, like DeleteTrackFile: a leftover art directory is only disk. if dataDir != "" { for _, artistID := range res.DeletedArtistIDs { if err := coverart.CleanupArtistArt(dataDir, artistID); err != nil { logger.Warn("duplicate merge: artist-art cleanup failed", "artist_id", syncpkg.FormatUUID(artistID), "err", err) } } } return res, nil } // splitSurvivor separates the copy to keep from the copies to remove. An // invalid survivorID takes ProposeSurvivor's choice. func splitSurvivor( members []dbq.ListDuplicateGroupMergeMembersRow, survivorID pgtype.UUID, ) (dbq.ListDuplicateGroupMergeMembersRow, []dbq.ListDuplicateGroupMergeMembersRow, error) { want := "" if survivorID.Valid { want = syncpkg.FormatUUID(survivorID) } else { cands := make([]SurvivorCandidate, len(members)) for i, m := range members { cands[i] = SurvivorCandidate{ TrackID: syncpkg.FormatUUID(m.ID), FileFormat: m.FileFormat, FileSize: m.FileSize, AddedAt: m.AddedAt.Time, } } want, _ = ProposeSurvivor(cands) } var survivor dbq.ListDuplicateGroupMergeMembersRow found := false var losers []dbq.ListDuplicateGroupMergeMembersRow for _, m := range members { if syncpkg.FormatUUID(m.ID) == want { survivor, found = m, true continue } losers = append(losers, m) } if !found { return dbq.ListDuplicateGroupMergeMembersRow{}, nil, ErrSurvivorNotInGroup } return survivor, losers, nil } func mergedCopyOf(m dbq.ListDuplicateGroupMergeMembersRow) MergedCopy { return MergedCopy{TrackID: m.ID, FilePath: m.FilePath, TrackMbid: m.Mbid, AlbumMbid: m.AlbumMbid} } // mergeChanges collects the sync-log entries a merge owes clients' caches, each // once: a user who liked two removed copies still gets one upsert for the // survivor. type mergeChanges struct { tracks, likeDeletes, likeUpserts, playlistDeletes, playlistUpserts map[string]struct{} } func addTo(set *map[string]struct{}, id string) { if *set == nil { *set = map[string]struct{}{} } (*set)[id] = struct{}{} } func (c *mergeChanges) trackDelete(id string) { addTo(&c.tracks, id) } func (c *mergeChanges) likeDelete(id string) { addTo(&c.likeDeletes, id) } func (c *mergeChanges) likeUpsert(id string) { addTo(&c.likeUpserts, id) } func (c *mergeChanges) playlistDelete(id string) { addTo(&c.playlistDeletes, id) } func (c *mergeChanges) playlistUpsert(id string) { addTo(&c.playlistUpserts, id) } func (c *mergeChanges) log(ctx context.Context, tx pgx.Tx) error { for _, entry := range []struct { kind syncpkg.EntityType ids map[string]struct{} op syncpkg.Op }{ {syncpkg.EntityTrack, c.tracks, syncpkg.OpDelete}, {syncpkg.EntityLikeTrack, c.likeDeletes, syncpkg.OpDelete}, {syncpkg.EntityLikeTrack, c.likeUpserts, syncpkg.OpUpsert}, {syncpkg.EntityPlaylistTrack, c.playlistDeletes, syncpkg.OpDelete}, {syncpkg.EntityPlaylistTrack, c.playlistUpserts, syncpkg.OpUpsert}, } { if len(entry.ids) == 0 { continue } ids := make([]string, 0, len(entry.ids)) for id := range entry.ids { ids = append(ids, id) } sort.Strings(ids) if err := syncpkg.LogChanges(ctx, tx, entry.kind, ids, entry.op); err != nil { return fmt.Errorf("log merge changes: %w", err) } } return nil }