diff --git a/internal/api/admin_duplicates.go b/internal/api/admin_duplicates.go index 05fc1be7..db319b3a 100644 --- a/internal/api/admin_duplicates.go +++ b/internal/api/admin_duplicates.go @@ -2,11 +2,14 @@ package api import ( "context" + "encoding/json" "errors" + "io" "net/http" "github.com/go-chi/chi/v5" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/library" @@ -228,3 +231,86 @@ func (h *handlers) handleDismissDuplicateGroup(w http.ResponseWriter, r *http.Re } writeJSON(w, http.StatusOK, map[string]string{"status": "dismissed"}) } + +// mergeDuplicateRequest chooses the copy to keep. An empty survivor_track_id +// keeps the report's proposal. +type mergeDuplicateRequest struct { + SurvivorTrackID string `json:"survivor_track_id"` + Unmonitor bool `json:"unmonitor"` +} + +// mergeDuplicateResponse reports what the merge removed. RemovedPaths are files +// deleted from disk; the operator reads them to know exactly what went. +type mergeDuplicateResponse struct { + SurvivorTrackID string `json:"survivor_track_id"` + RemovedPaths []string `json:"removed_paths"` + LidarrUnmonitorFailed *bool `json:"lidarr_unmonitor_failed,omitempty"` +} + +// mergeRequestBodyLimit bounds the request body. It holds one id and a flag. +const mergeRequestBodyLimit = 1 << 16 + +// handleMergeDuplicateGroup implements POST /api/admin/library/duplicates/{id}/merge +// (#3911): keep one copy, move the others' likes, plays and playlist entries onto +// it, and delete the others' files and rows. +// +// Errors: +// - 409 library_not_writable / 500 file_delete_failed when a file could not be +// removed — nothing was changed (fileRemoveAPIError) +// - 404 duplicate_group_not_pending when the group was already resolved +// - 400 survivor_not_in_group, invalid_id, invalid_body +func (h *handlers) handleMergeDuplicateGroup(w http.ResponseWriter, r *http.Request) { + admin, ok := requireUser(w, r) + if !ok { + return + } + groupID, ok := parseUUID(chi.URLParam(r, "id")) + if !ok { + writeAdminJSONErr(w, http.StatusBadRequest, "invalid_id") + return + } + var body mergeDuplicateRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, mergeRequestBodyLimit)).Decode(&body); err != nil && !errors.Is(err, io.EOF) { + writeAdminJSONErr(w, http.StatusBadRequest, "invalid_body") + return + } + var survivorID pgtype.UUID // invalid: keep the proposal + if body.SurvivorTrackID != "" { + if survivorID, ok = parseUUID(body.SurvivorTrackID); !ok { + writeAdminJSONErr(w, http.StatusBadRequest, "invalid_id") + return + } + } + + res, unmonitorFailed, err := h.tracks.MergeDuplicates(r.Context(), groupID, survivorID, admin.ID, body.Unmonitor) + if err != nil { + if apiErr, ok := fileRemoveAPIError(err); ok { + logFileRemoveFailure(h.logger, apiErr, "group_id", uuidToString(groupID)) + writeErr(w, apiErr) + return + } + switch { + case errors.Is(err, library.ErrDuplicateGroupNotPending): + writeAdminJSONErr(w, http.StatusNotFound, "duplicate_group_not_pending") + case errors.Is(err, library.ErrSurvivorNotInGroup): + writeAdminJSONErr(w, http.StatusBadRequest, "survivor_not_in_group") + default: + h.logger.Error("admin: merge duplicate group", "group_id", uuidToString(groupID), "err", err) + writeAdminJSONErr(w, http.StatusInternalServerError, "server_error") + } + return + } + + resp := mergeDuplicateResponse{ + SurvivorTrackID: uuidToString(res.Survivor.TrackID), + RemovedPaths: make([]string, 0, len(res.Removed)), + } + for _, c := range res.Removed { + resp.RemovedPaths = append(resp.RemovedPaths, c.FilePath) + } + if body.Unmonitor && unmonitorFailed { + failed := true + resp.LidarrUnmonitorFailed = &failed + } + writeJSON(w, http.StatusOK, resp) +} diff --git a/internal/api/api.go b/internal/api/api.go index d3018b9d..85647a5c 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -217,10 +217,12 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev admin.Get("/library/coverage", h.handleGetLibraryCoverage) admin.Get("/library/fingerprints", h.handleGetFingerprintCoverage) // Duplicates report (#3912): proposals from the duplicate sweep, a - // trigger to sweep now, and dismissal. Nothing here merges or deletes. + // trigger to sweep now, dismissal, and the merge (#3911), which deletes + // the removed copies' files after moving their history onto the kept one. admin.Get("/library/duplicates", h.handleListDuplicates) admin.Post("/library/duplicates/sweep", h.handleRunDuplicateSweep) admin.Post("/library/duplicates/{id}/dismiss", h.handleDismissDuplicateGroup) + admin.Post("/library/duplicates/{id}/merge", h.handleMergeDuplicateGroup) admin.Get("/invites", h.handleListInvites) admin.Post("/invites", h.handleCreateInvite) diff --git a/internal/audit/audit.go b/internal/audit/audit.go index eb2e8a5b..ef2c5f60 100644 --- a/internal/audit/audit.go +++ b/internal/audit/audit.go @@ -55,6 +55,11 @@ const ( // exercised. ActionSessionRevoke Action = "session_revoke" ActionSessionRevokeOthers Action = "session_revoke_others" + + // Duplicate merge (#3911). Irreversible: a copy's file and row are removed + // and its history moved onto the copy kept. The metadata names both, so the + // log can answer "where did that file go" long after the report is gone. + ActionDuplicateMerge Action = "duplicate_merge" ) // Write inserts one audit_log row. metadata is marshaled as JSON; diff --git a/internal/audit/audit_test.go b/internal/audit/audit_test.go index ee413893..84faacd7 100644 --- a/internal/audit/audit_test.go +++ b/internal/audit/audit_test.go @@ -168,6 +168,7 @@ func TestWrite_AllActionConstantsArePersisted(t *testing.T) { audit.ActionTokenRegenerate, audit.ActionForgotPasswordInit, audit.ActionPasswordResetByEmail, + audit.ActionDuplicateMerge, } for _, a := range actions { if err := audit.Write(context.Background(), pool, nilUUID, nilUUID, a, nil); err != nil { diff --git a/internal/db/dbq/merge.sql.go b/internal/db/dbq/merge.sql.go new file mode 100644 index 00000000..95240836 --- /dev/null +++ b/internal/db/dbq/merge.sql.go @@ -0,0 +1,339 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: merge.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const listDuplicateGroupMergeMembers = `-- name: ListDuplicateGroupMergeMembers :many +SELECT t.id, t.file_path, t.file_format, t.file_size, t.added_at, t.album_id, + t.mbid, albums.mbid AS album_mbid + FROM duplicate_group_members m + JOIN tracks t ON t.id = m.track_id + JOIN albums ON albums.id = t.album_id + WHERE m.group_id = $1 + ORDER BY t.id +` + +type ListDuplicateGroupMergeMembersRow struct { + ID pgtype.UUID + FilePath string + FileFormat string + FileSize int64 + AddedAt pgtype.Timestamptz + AlbumID pgtype.UUID + Mbid *string + AlbumMbid *string +} + +func (q *Queries) ListDuplicateGroupMergeMembers(ctx context.Context, groupID pgtype.UUID) ([]ListDuplicateGroupMergeMembersRow, error) { + rows, err := q.db.Query(ctx, listDuplicateGroupMergeMembers, groupID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListDuplicateGroupMergeMembersRow + for rows.Next() { + var i ListDuplicateGroupMergeMembersRow + if err := rows.Scan( + &i.ID, + &i.FilePath, + &i.FileFormat, + &i.FileSize, + &i.AddedAt, + &i.AlbumID, + &i.Mbid, + &i.AlbumMbid, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const lockDuplicateGroupForMerge = `-- name: LockDuplicateGroupForMerge :one + +SELECT id, tier, status + FROM duplicate_groups + WHERE id = $1 + FOR UPDATE +` + +type LockDuplicateGroupForMergeRow struct { + ID pgtype.UUID + Tier string + Status string +} + +// Duplicate merge (Scribe #3911). Every statement here runs inside the one +// transaction library.MergeDuplicateGroup opens, after the removed copy's file +// is already gone. The loser's own track row is deleted last with DeleteTrack; +// what these do is move everything it carries onto the survivor first, so that +// delete's CASCADE finds nothing left to destroy. +// Locks the group for the rest of the transaction, so two merges of one group +// cannot run at once. +func (q *Queries) LockDuplicateGroupForMerge(ctx context.Context, id pgtype.UUID) (LockDuplicateGroupForMergeRow, error) { + row := q.db.QueryRow(ctx, lockDuplicateGroupForMerge, id) + var i LockDuplicateGroupForMergeRow + err := row.Scan(&i.ID, &i.Tier, &i.Status) + return i, err +} + +const markDuplicateGroupMerged = `-- name: MarkDuplicateGroupMerged :execrows +UPDATE duplicate_groups + SET status = 'merged', resolved_at = now() + WHERE id = $1 AND status = 'pending' +` + +func (q *Queries) MarkDuplicateGroupMerged(ctx context.Context, id pgtype.UUID) (int64, error) { + result, err := q.db.Exec(ctx, markDuplicateGroupMerged, id) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeCopyGeneralLikes = `-- name: MergeCopyGeneralLikes :many + +INSERT INTO general_likes (user_id, track_id, liked_at) +SELECT user_id, $1::uuid, liked_at + FROM general_likes + WHERE track_id = $2::uuid +ON CONFLICT (user_id, track_id) DO UPDATE + SET liked_at = LEAST(general_likes.liked_at, EXCLUDED.liked_at) +RETURNING user_id +` + +type MergeCopyGeneralLikesParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +// Collision-safe merges: a unique key includes track_id, so the survivor may +// already hold a matching row. Copy what it lacks; DeleteTrack's CASCADE then +// removes the loser's originals. +// One like per user. A user who liked both copies keeps a single like, dated to +// the earlier of the two. +func (q *Queries) MergeCopyGeneralLikes(ctx context.Context, arg MergeCopyGeneralLikesParams) ([]pgtype.UUID, error) { + rows, err := q.db.Query(ctx, mergeCopyGeneralLikes, arg.SurvivorID, arg.LoserID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []pgtype.UUID + for rows.Next() { + var user_id pgtype.UUID + if err := rows.Scan(&user_id); err != nil { + return nil, err + } + items = append(items, user_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const mergeCopyTrackSimilarity = `-- name: MergeCopyTrackSimilarity :execrows +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +SELECT CASE WHEN track_a_id = $1::uuid THEN $2::uuid ELSE track_a_id END, + CASE WHEN track_b_id = $1::uuid THEN $2::uuid ELSE track_b_id END, + score, source, fetched_at + FROM track_similarity + WHERE (track_a_id = $1::uuid OR track_b_id = $1::uuid) + AND (CASE WHEN track_a_id = $1::uuid THEN $2::uuid ELSE track_a_id END) + <> (CASE WHEN track_b_id = $1::uuid THEN $2::uuid ELSE track_b_id END) +ON CONFLICT (track_a_id, track_b_id, source) DO NOTHING +` + +type MergeCopyTrackSimilarityParams struct { + LoserID pgtype.UUID + SurvivorID pgtype.UUID +} + +// Rewrites the loser to the survivor on either side of an edge. An edge between +// the two copies would become a track similar to itself — the table forbids +// that, and it means nothing — so it is dropped. An edge the survivor already +// has from the same source is kept as it is. +func (q *Queries) MergeCopyTrackSimilarity(ctx context.Context, arg MergeCopyTrackSimilarityParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeCopyTrackSimilarity, arg.LoserID, arg.SurvivorID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeCopyTrackTags = `-- name: MergeCopyTrackTags :execrows +INSERT INTO track_tags (track_id, tag, weight) +SELECT $1::uuid, tag, weight + FROM track_tags + WHERE track_id = $2::uuid +ON CONFLICT (track_id, tag) DO NOTHING +` + +type MergeCopyTrackTagsParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +func (q *Queries) MergeCopyTrackTags(ctx context.Context, arg MergeCopyTrackTagsParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeCopyTrackTags, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeInheritTrackMbid = `-- name: MergeInheritTrackMbid :exec +UPDATE tracks AS survivor + SET mbid = loser.mbid + FROM tracks AS loser + WHERE survivor.id = $1::uuid + AND loser.id = $2::uuid + AND survivor.mbid IS NULL + AND loser.mbid IS NOT NULL +` + +type MergeInheritTrackMbidParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +// A recording MBID is what the similarity pipeline keys on. If only the removed +// copy carried one, the survivor takes it rather than going dark to similarity. +func (q *Queries) MergeInheritTrackMbid(ctx context.Context, arg MergeInheritTrackMbidParams) error { + _, err := q.db.Exec(ctx, mergeInheritTrackMbid, arg.SurvivorID, arg.LoserID) + return err +} + +const mergeRepointContextualLikes = `-- name: MergeRepointContextualLikes :execrows +UPDATE contextual_likes SET track_id = $1::uuid WHERE track_id = $2::uuid +` + +type MergeRepointContextualLikesParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +func (q *Queries) MergeRepointContextualLikes(ctx context.Context, arg MergeRepointContextualLikesParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeRepointContextualLikes, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeRepointLidarrRequests = `-- name: MergeRepointLidarrRequests :execrows +UPDATE lidarr_requests SET matched_track_id = $1::uuid + WHERE matched_track_id = $2::uuid +` + +type MergeRepointLidarrRequestsParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +func (q *Queries) MergeRepointLidarrRequests(ctx context.Context, arg MergeRepointLidarrRequestsParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeRepointLidarrRequests, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeRepointPlayEvents = `-- name: MergeRepointPlayEvents :execrows + +UPDATE play_events SET track_id = $1::uuid WHERE track_id = $2::uuid +` + +type MergeRepointPlayEventsParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +// Plain repoints: no unique key involves track_id, so moving rows cannot collide. +func (q *Queries) MergeRepointPlayEvents(ctx context.Context, arg MergeRepointPlayEventsParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeRepointPlayEvents, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeRepointPlaybackErrors = `-- name: MergeRepointPlaybackErrors :execrows +UPDATE playback_errors SET track_id = $1::uuid WHERE track_id = $2::uuid +` + +type MergeRepointPlaybackErrorsParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +func (q *Queries) MergeRepointPlaybackErrors(ctx context.Context, arg MergeRepointPlaybackErrorsParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeRepointPlaybackErrors, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const mergeRepointPlaylistTracks = `-- name: MergeRepointPlaylistTracks :many +UPDATE playlist_tracks SET track_id = $1::uuid + WHERE track_id = $2::uuid +RETURNING playlist_id +` + +type MergeRepointPlaylistTracksParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +// playlist_tracks is keyed by (playlist_id, position), so repointing keeps every +// entry exactly where it was. A playlist that held both copies simply holds the +// survivor twice — the user put two entries there, and both stay. +func (q *Queries) MergeRepointPlaylistTracks(ctx context.Context, arg MergeRepointPlaylistTracksParams) ([]pgtype.UUID, error) { + rows, err := q.db.Query(ctx, mergeRepointPlaylistTracks, arg.SurvivorID, arg.LoserID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []pgtype.UUID + for rows.Next() { + var playlist_id pgtype.UUID + if err := rows.Scan(&playlist_id); err != nil { + return nil, err + } + items = append(items, playlist_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const mergeRepointSkipEvents = `-- name: MergeRepointSkipEvents :execrows +UPDATE skip_events SET track_id = $1::uuid WHERE track_id = $2::uuid +` + +type MergeRepointSkipEventsParams struct { + SurvivorID pgtype.UUID + LoserID pgtype.UUID +} + +func (q *Queries) MergeRepointSkipEvents(ctx context.Context, arg MergeRepointSkipEventsParams) (int64, error) { + result, err := q.db.Exec(ctx, mergeRepointSkipEvents, arg.SurvivorID, arg.LoserID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} diff --git a/internal/db/queries/merge.sql b/internal/db/queries/merge.sql new file mode 100644 index 00000000..6c71dfb1 --- /dev/null +++ b/internal/db/queries/merge.sql @@ -0,0 +1,101 @@ +-- Duplicate merge (Scribe #3911). Every statement here runs inside the one +-- transaction library.MergeDuplicateGroup opens, after the removed copy's file +-- is already gone. The loser's own track row is deleted last with DeleteTrack; +-- what these do is move everything it carries onto the survivor first, so that +-- delete's CASCADE finds nothing left to destroy. + +-- name: LockDuplicateGroupForMerge :one +-- Locks the group for the rest of the transaction, so two merges of one group +-- cannot run at once. +SELECT id, tier, status + FROM duplicate_groups + WHERE id = sqlc.arg(id) + FOR UPDATE; + +-- name: ListDuplicateGroupMergeMembers :many +SELECT t.id, t.file_path, t.file_format, t.file_size, t.added_at, t.album_id, + t.mbid, albums.mbid AS album_mbid + FROM duplicate_group_members m + JOIN tracks t ON t.id = m.track_id + JOIN albums ON albums.id = t.album_id + WHERE m.group_id = sqlc.arg(group_id) + ORDER BY t.id; + +-- Plain repoints: no unique key involves track_id, so moving rows cannot collide. + +-- name: MergeRepointPlayEvents :execrows +UPDATE play_events SET track_id = sqlc.arg(survivor_id)::uuid WHERE track_id = sqlc.arg(loser_id)::uuid; + +-- name: MergeRepointSkipEvents :execrows +UPDATE skip_events SET track_id = sqlc.arg(survivor_id)::uuid WHERE track_id = sqlc.arg(loser_id)::uuid; + +-- name: MergeRepointContextualLikes :execrows +UPDATE contextual_likes SET track_id = sqlc.arg(survivor_id)::uuid WHERE track_id = sqlc.arg(loser_id)::uuid; + +-- name: MergeRepointPlaybackErrors :execrows +UPDATE playback_errors SET track_id = sqlc.arg(survivor_id)::uuid WHERE track_id = sqlc.arg(loser_id)::uuid; + +-- name: MergeRepointLidarrRequests :execrows +UPDATE lidarr_requests SET matched_track_id = sqlc.arg(survivor_id)::uuid + WHERE matched_track_id = sqlc.arg(loser_id)::uuid; + +-- name: MergeRepointPlaylistTracks :many +-- playlist_tracks is keyed by (playlist_id, position), so repointing keeps every +-- entry exactly where it was. A playlist that held both copies simply holds the +-- survivor twice — the user put two entries there, and both stay. +UPDATE playlist_tracks SET track_id = sqlc.arg(survivor_id)::uuid + WHERE track_id = sqlc.arg(loser_id)::uuid +RETURNING playlist_id; + +-- Collision-safe merges: a unique key includes track_id, so the survivor may +-- already hold a matching row. Copy what it lacks; DeleteTrack's CASCADE then +-- removes the loser's originals. + +-- name: MergeCopyGeneralLikes :many +-- One like per user. A user who liked both copies keeps a single like, dated to +-- the earlier of the two. +INSERT INTO general_likes (user_id, track_id, liked_at) +SELECT user_id, sqlc.arg(survivor_id)::uuid, liked_at + FROM general_likes + WHERE track_id = sqlc.arg(loser_id)::uuid +ON CONFLICT (user_id, track_id) DO UPDATE + SET liked_at = LEAST(general_likes.liked_at, EXCLUDED.liked_at) +RETURNING user_id; + +-- name: MergeCopyTrackTags :execrows +INSERT INTO track_tags (track_id, tag, weight) +SELECT sqlc.arg(survivor_id)::uuid, tag, weight + FROM track_tags + WHERE track_id = sqlc.arg(loser_id)::uuid +ON CONFLICT (track_id, tag) DO NOTHING; + +-- name: MergeCopyTrackSimilarity :execrows +-- Rewrites the loser to the survivor on either side of an edge. An edge between +-- the two copies would become a track similar to itself — the table forbids +-- that, and it means nothing — so it is dropped. An edge the survivor already +-- has from the same source is kept as it is. +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +SELECT CASE WHEN track_a_id = sqlc.arg(loser_id)::uuid THEN sqlc.arg(survivor_id)::uuid ELSE track_a_id END, + CASE WHEN track_b_id = sqlc.arg(loser_id)::uuid THEN sqlc.arg(survivor_id)::uuid ELSE track_b_id END, + score, source, fetched_at + FROM track_similarity + WHERE (track_a_id = sqlc.arg(loser_id)::uuid OR track_b_id = sqlc.arg(loser_id)::uuid) + AND (CASE WHEN track_a_id = sqlc.arg(loser_id)::uuid THEN sqlc.arg(survivor_id)::uuid ELSE track_a_id END) + <> (CASE WHEN track_b_id = sqlc.arg(loser_id)::uuid THEN sqlc.arg(survivor_id)::uuid ELSE track_b_id END) +ON CONFLICT (track_a_id, track_b_id, source) DO NOTHING; + +-- name: MergeInheritTrackMbid :exec +-- A recording MBID is what the similarity pipeline keys on. If only the removed +-- copy carried one, the survivor takes it rather than going dark to similarity. +UPDATE tracks AS survivor + SET mbid = loser.mbid + FROM tracks AS loser + WHERE survivor.id = sqlc.arg(survivor_id)::uuid + AND loser.id = sqlc.arg(loser_id)::uuid + AND survivor.mbid IS NULL + AND loser.mbid IS NOT NULL; + +-- name: MarkDuplicateGroupMerged :execrows +UPDATE duplicate_groups + SET status = 'merged', resolved_at = now() + WHERE id = sqlc.arg(id) AND status = 'pending'; diff --git a/internal/library/delete.go b/internal/library/delete.go index 70d3ece5..a12aec3c 100644 --- a/internal/library/delete.go +++ b/internal/library/delete.go @@ -114,10 +114,8 @@ func DeleteTrackFile( return DeletedTrack{}, fmt.Errorf("get track: %w", err) } - if err := removeFile(track.FilePath); err != nil && !errors.Is(err, fs.ErrNotExist) { - return DeletedTrack{}, &FileRemoveError{ - Path: track.FilePath, UID: os.Getuid(), GID: os.Getgid(), Err: err, - } + if err := removeTrackFileOnDisk(track.FilePath); err != nil { + return DeletedTrack{}, err } // The row and any album or artist it empties go together, so a failure @@ -138,25 +136,9 @@ func DeleteTrackFile( return DeletedTrack{}, fmt.Errorf("delete track: %w", err) } - var out DeletedTrack - album, err := tq.DeleteAlbumIfEmpty(ctx, deleted.AlbumID) - switch { - case err == nil: - albumID := album.ID - out.AlbumID = &albumID - artistID, aerr := tq.DeleteArtistIfEmpty(ctx, album.ArtistID) - switch { - case aerr == nil: - out.ArtistID = &artistID - case errors.Is(aerr, pgx.ErrNoRows): - // The artist still has other albums or stray tracks. - default: - return DeletedTrack{}, fmt.Errorf("delete artist if empty: %w", aerr) - } - case errors.Is(err, pgx.ErrNoRows): - // The album still has other tracks. - default: - return DeletedTrack{}, fmt.Errorf("delete album if empty: %w", err) + out, err := tidyEmptiedAlbum(ctx, tq, deleted.AlbumID) + if err != nil { + return DeletedTrack{}, err } if err := tx.Commit(ctx); err != nil { @@ -179,3 +161,41 @@ func DeleteTrackFile( } return out, nil } + +// removeTrackFileOnDisk is the one rule for removing a track's file, shared by +// DeleteTrackFile and the duplicate merge. A file already gone is fine; anything +// else comes back as a *FileRemoveError, and the caller must then change nothing +// in the database (#3918). +func removeTrackFileOnDisk(path string) error { + if err := removeFile(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + return &FileRemoveError{Path: path, UID: os.Getuid(), GID: os.Getgid(), Err: err} + } + return nil +} + +// tidyEmptiedAlbum removes an album a track delete left with no tracks, and its +// artist if that album was the artist's last. It runs on the caller's +// transaction, so the tidy-up commits or rolls back with the delete itself. +func tidyEmptiedAlbum(ctx context.Context, tq *dbq.Queries, albumID pgtype.UUID) (DeletedTrack, error) { + var out DeletedTrack + album, err := tq.DeleteAlbumIfEmpty(ctx, albumID) + switch { + case err == nil: + id := album.ID + out.AlbumID = &id + artistID, aerr := tq.DeleteArtistIfEmpty(ctx, album.ArtistID) + switch { + case aerr == nil: + out.ArtistID = &artistID + case errors.Is(aerr, pgx.ErrNoRows): + // The artist still has other albums or stray tracks. + default: + return DeletedTrack{}, fmt.Errorf("delete artist if empty: %w", aerr) + } + case errors.Is(err, pgx.ErrNoRows): + // The album still has other tracks. + default: + return DeletedTrack{}, fmt.Errorf("delete album if empty: %w", err) + } + return out, nil +} diff --git a/internal/library/duplicate_merge.go b/internal/library/duplicate_merge.go new file mode 100644 index 00000000..1182c510 --- /dev/null +++ b/internal/library/duplicate_merge.go @@ -0,0 +1,309 @@ +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 +} diff --git a/internal/library/duplicate_merge_test.go b/internal/library/duplicate_merge_test.go new file mode 100644 index 00000000..729cc942 --- /dev/null +++ b/internal/library/duplicate_merge_test.go @@ -0,0 +1,281 @@ +package library + +import ( + "context" + "errors" + "io/fs" + "os" + "path/filepath" + "syscall" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/dbtest" +) + +// mergeFixture is a library with one duplicate pair carrying history on both +// copies, and a neighbour track for similarity edges. +type mergeFixture struct { + pool *pgxpool.Pool + keep, remove, other dbq.Track + keepPath, removePath string + groupID pgtype.UUID + alice, bob dbq.User + playlistID pgtype.UUID + removePlaylistPos int32 + aliceEarlierLikeOnRem time.Time +} + +func newMergeFixture(t *testing.T) mergeFixture { + t.Helper() + pool := newPool(t) + ctx := context.Background() + q := dbq.New(pool) + dir := t.TempDir() + f := mergeFixture{pool: pool} + + f.keepPath = filepath.Join(dir, "keep.flac") + f.removePath = filepath.Join(dir, "remove.mp3") + for _, p := range []string{f.keepPath, f.removePath} { + if err := os.WriteFile(p, []byte("audio"), 0o644); err != nil { + t.Fatalf("write %s: %v", p, err) + } + } + var album dbq.Album + var artist dbq.Artist + f.keep, album, artist = seedTrack(t, pool, f.keepPath) + upsert := func(title, path string) dbq.Track { + t.Helper() + tr, err := q.UpsertTrack(ctx, dbq.UpsertTrackParams{ + Title: title, AlbumID: album.ID, ArtistID: artist.ID, + DurationMs: 215000, FilePath: path, FileSize: 100, FileFormat: "mp3", + }) + if err != nil { + t.Fatalf("track %s: %v", title, err) + } + return tr + } + f.remove = upsert("WWW (copy)", f.removePath) + f.other = upsert("Neighbour", filepath.Join(dir, "other.mp3")) + + mustExec := func(sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("exec %q: %v", sql, err) + } + } + // Only the copy being removed carries a recording MBID. + mustExec(`UPDATE tracks SET mbid = 'rec-www' WHERE id = $1`, f.remove.ID) + + user := func(name string) dbq.User { + t.Helper() + u, err := q.CreateUser(ctx, dbq.CreateUserParams{ + Username: dbtest.TestUserPrefix + name, PasswordHash: "x", ApiToken: name + "-merge-token", + }) + if err != nil { + t.Fatalf("user %s: %v", name, err) + } + return u + } + f.alice, f.bob = user("merge-alice"), user("merge-bob") + + // Alice liked both copies, the removed one first; Bob liked only the removed one. + f.aliceEarlierLikeOnRem = time.Now().Add(-72 * time.Hour).UTC().Truncate(time.Microsecond) + mustExec(`INSERT INTO general_likes (user_id, track_id, liked_at) VALUES ($1, $2, $3), ($1, $4, now()), ($5, $2, now())`, + f.alice.ID, f.remove.ID, f.aliceEarlierLikeOnRem, f.keep.ID, f.bob.ID) + + now := pgtype.Timestamptz{Time: time.Now(), Valid: true} + session, err := q.InsertPlaySession(ctx, dbq.InsertPlaySessionParams{UserID: f.alice.ID, StartedAt: now}) + if err != nil { + t.Fatalf("session: %v", err) + } + for _, track := range []dbq.Track{f.remove, f.remove, f.keep} { + if _, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ + UserID: f.alice.ID, TrackID: track.ID, SessionID: session.ID, StartedAt: now, + }); err != nil { + t.Fatalf("play event: %v", err) + } + } + if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ + UserID: f.alice.ID, TrackID: f.remove.ID, SessionID: session.ID, SkippedAt: now, PositionMs: 1000, + }); err != nil { + t.Fatalf("skip event: %v", err) + } + + pl, err := q.CreatePlaylist(ctx, dbq.CreatePlaylistParams{UserID: f.alice.ID, Name: "merge-mix"}) + if err != nil { + t.Fatalf("playlist: %v", err) + } + f.playlistID = pl.ID + entry, err := q.AppendPlaylistTrack(ctx, dbq.AppendPlaylistTrackParams{PlaylistID: pl.ID, TrackID: f.remove.ID}) + if err != nil { + t.Fatalf("playlist entry: %v", err) + } + f.removePlaylistPos = entry.Position + + mustExec(`INSERT INTO track_tags (track_id, tag, weight) VALUES ($1, 'j-pop', 1), ($1, 'house', 0.5), ($2, 'house', 0.9)`, + f.remove.ID, f.keep.ID) + mustExec(`INSERT INTO track_similarity (track_a_id, track_b_id, score, source) VALUES + ($1, $3, 0.8, 'listenbrainz'), + ($2, $3, 0.7, 'listenbrainz'), + ($1, $2, 0.99, 'listenbrainz'), + ($3, $1, 0.6, 'musicbrainz_tag')`, f.remove.ID, f.keep.ID, f.other.ID) + + if err := pool.QueryRow(ctx, + `INSERT INTO duplicate_groups (member_key, tier) VALUES ('merge-fixture', 'exact') RETURNING id`, + ).Scan(&f.groupID); err != nil { + t.Fatalf("group: %v", err) + } + mustExec(`INSERT INTO duplicate_group_members (group_id, track_id) VALUES ($1, $2), ($1, $3)`, + f.groupID, f.keep.ID, f.remove.ID) + return f +} + +func (f mergeFixture) count(t *testing.T, sql string, args ...any) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), sql, args...).Scan(&n); err != nil { + t.Fatalf("count %q: %v", sql, err) + } + return n +} + +// The #3911 proof: after a merge, every piece of history the removed copy held +// is on the copy kept, nothing is doubled, and the removed copy — row and file — +// is gone. +func TestMergeDuplicateGroup_Integration(t *testing.T) { + f := newMergeFixture(t) + ctx := context.Background() + + res, err := MergeDuplicateGroup(ctx, f.pool, nil, "", f.groupID, f.keep.ID) + if err != nil { + t.Fatalf("merge: %v", err) + } + if len(res.Removed) != 1 || res.Removed[0].FilePath != f.removePath || res.Survivor.TrackID != f.keep.ID { + t.Fatalf("result = %+v, want the removed copy reported and the kept one as survivor", res) + } + if res.PlayEvents != 2 || res.SkipEvents != 1 || res.Likes != 2 || res.PlaylistEntries != 1 { + t.Errorf("moved = plays %d skips %d likes %d playlist %d, want 2, 1, 2, 1", + res.PlayEvents, res.SkipEvents, res.Likes, res.PlaylistEntries) + } + + if _, err := os.Stat(f.removePath); !errors.Is(err, os.ErrNotExist) { + t.Errorf("removed copy's file still on disk: %v", err) + } + if _, err := os.Stat(f.keepPath); err != nil { + t.Errorf("kept copy's file is gone: %v", err) + } + if n := f.count(t, `SELECT count(*) FROM tracks WHERE id = $1`, f.remove.ID); n != 0 { + t.Errorf("removed copy's row still exists") + } + + // Likes: one per user, Alice's dated to her earlier like. + if n := f.count(t, `SELECT count(*) FROM general_likes WHERE track_id = $1`, f.keep.ID); n != 2 { + t.Errorf("likes on the kept copy = %d, want 2 (Alice once, Bob)", n) + } + var aliceLiked time.Time + if err := f.pool.QueryRow(ctx, `SELECT liked_at FROM general_likes WHERE user_id = $1 AND track_id = $2`, + f.alice.ID, f.keep.ID).Scan(&aliceLiked); err != nil { + t.Fatalf("alice's like: %v", err) + } + if !aliceLiked.Equal(f.aliceEarlierLikeOnRem) { + t.Errorf("alice's like dated %v, want her earlier like %v", aliceLiked, f.aliceEarlierLikeOnRem) + } + + // Plays and skips move exactly: none lost, none invented. + if n := f.count(t, `SELECT count(*) FROM play_events WHERE track_id = $1`, f.keep.ID); n != 3 { + t.Errorf("plays on the kept copy = %d, want 3", n) + } + if n := f.count(t, `SELECT count(*) FROM skip_events WHERE track_id = $1`, f.keep.ID); n != 1 { + t.Errorf("skips on the kept copy = %d, want 1", n) + } + + // The playlist entry stays where it was and now plays the kept copy. + if n := f.count(t, `SELECT count(*) FROM playlist_tracks WHERE playlist_id = $1 AND position = $2 AND track_id = $3`, + f.playlistID, f.removePlaylistPos, f.keep.ID); n != 1 { + t.Errorf("playlist entry at position %d does not point at the kept copy", f.removePlaylistPos) + } + + // Tags are a union; the kept copy's own weight wins where both had the tag. + if n := f.count(t, `SELECT count(*) FROM track_tags WHERE track_id = $1`, f.keep.ID); n != 2 { + t.Errorf("tags on the kept copy = %d, want 2 (house, j-pop)", n) + } + if n := f.count(t, `SELECT count(*) FROM track_tags WHERE track_id = $1 AND tag = 'house' AND weight = 0.9`, f.keep.ID); n != 1 { + t.Errorf("the kept copy's own house weight was overwritten") + } + + // Similarity: rewritten onto the kept copy, no duplicate edge, no self-edge, + // nothing left pointing at the removed copy. + if n := f.count(t, `SELECT count(*) FROM track_similarity WHERE track_a_id = $1 AND track_b_id = $2 AND source = 'listenbrainz'`, + f.keep.ID, f.other.ID); n != 1 { + t.Errorf("listenbrainz edge keep→other = %d rows, want exactly 1", n) + } + if n := f.count(t, `SELECT count(*) FROM track_similarity WHERE track_a_id = $1 AND track_b_id = $2 AND source = 'musicbrainz_tag'`, + f.other.ID, f.keep.ID); n != 1 { + t.Errorf("musicbrainz_tag edge other→keep was not carried over") + } + if n := f.count(t, `SELECT count(*) FROM track_similarity WHERE track_a_id = track_b_id`); n != 0 { + t.Errorf("a self-edge was written") + } + + // The removed copy's recording MBID is inherited; the group is closed. + if n := f.count(t, `SELECT count(*) FROM tracks WHERE id = $1 AND mbid = 'rec-www'`, f.keep.ID); n != 1 { + t.Errorf("the kept copy did not inherit the recording MBID") + } + if n := f.count(t, `SELECT count(*) FROM duplicate_groups WHERE id = $1 AND status = 'merged' AND resolved_at IS NOT NULL`, f.groupID); n != 1 { + t.Errorf("group was not marked merged") + } + + // A second merge of the same group is refused rather than repeated. + if _, err := MergeDuplicateGroup(ctx, f.pool, nil, "", f.groupID, f.keep.ID); !errors.Is(err, ErrDuplicateGroupNotPending) { + t.Errorf("second merge err = %v, want ErrDuplicateGroupNotPending", err) + } +} + +// When the removed copy's file cannot go, nothing may change: its likes, plays +// and row stay exactly where they were, and the group stays pending. +func TestMergeDuplicateGroup_UnremovableFileChangesNothing(t *testing.T) { + f := newMergeFixture(t) + stubRemoveFile(t, func(name string) error { + return &fs.PathError{Op: "remove", Path: name, Err: syscall.EROFS} + }) + + _, err := MergeDuplicateGroup(context.Background(), f.pool, nil, "", f.groupID, f.keep.ID) + var fre *FileRemoveError + if !errors.As(err, &fre) || !fre.NotWritable() { + t.Fatalf("err = %v, want a not-writable *FileRemoveError", err) + } + if n := f.count(t, `SELECT count(*) FROM tracks WHERE id = $1`, f.remove.ID); n != 1 { + t.Errorf("the copy's row was deleted although its file was not") + } + if n := f.count(t, `SELECT count(*) FROM general_likes WHERE track_id = $1`, f.remove.ID); n != 2 { + t.Errorf("likes on the copy = %d, want both still there", n) + } + if n := f.count(t, `SELECT count(*) FROM play_events WHERE track_id = $1`, f.remove.ID); n != 2 { + t.Errorf("plays on the copy = %d, want both still there", n) + } + if n := f.count(t, `SELECT count(*) FROM duplicate_groups WHERE id = $1 AND status = 'pending'`, f.groupID); n != 1 { + t.Errorf("group left pending = false, want it still pending") + } +} + +func TestMergeDuplicateGroup_SurvivorMustBeAMember(t *testing.T) { + f := newMergeFixture(t) + var stranger pgtype.UUID + stranger.Bytes[15], stranger.Valid = 0xEE, true + + _, err := MergeDuplicateGroup(context.Background(), f.pool, nil, "", f.groupID, stranger) + if !errors.Is(err, ErrSurvivorNotInGroup) { + t.Fatalf("err = %v, want ErrSurvivorNotInGroup", err) + } + if _, err := os.Stat(f.removePath); err != nil { + t.Errorf("a refused merge removed a file: %v", err) + } + if n := f.count(t, `SELECT count(*) FROM duplicate_groups WHERE id = $1 AND status = 'pending'`, f.groupID); n != 1 { + t.Errorf("a refused merge changed the group") + } +} diff --git a/internal/tracks/merge_test.go b/internal/tracks/merge_test.go new file mode 100644 index 00000000..c83d5ba7 --- /dev/null +++ b/internal/tracks/merge_test.go @@ -0,0 +1,39 @@ +package tracks + +import ( + "testing" + + "git.fabledsword.com/bvandeusen/minstrel/internal/library" +) + +func mbids(track, album string) library.MergedCopy { + c := library.MergedCopy{} + if track != "" { + c.TrackMbid = &track + } + if album != "" { + c.AlbumMbid = &album + } + return c +} + +// Unmonitoring a second file of the kept copy's own album track would stop +// Lidarr managing the kept file too; that case must be recognised. +func TestSameLidarrTrack(t *testing.T) { + for _, tc := range []struct { + name string + kept library.MergedCopy + removed library.MergedCopy + wantSame bool + }{ + {"same recording on the same album", mbids("rec", "alb"), mbids("rec", "alb"), true}, + {"same recording on a compilation", mbids("rec", "alb"), mbids("rec", "comp"), false}, + {"different recordings on one album", mbids("rec", "alb"), mbids("rec-2", "alb"), false}, + {"kept copy has no mbids", mbids("", ""), mbids("rec", "alb"), false}, + {"removed copy has no album mbid", mbids("rec", "alb"), mbids("rec", ""), false}, + } { + if got := sameLidarrTrack(tc.kept, tc.removed); got != tc.wantSame { + t.Errorf("%s: sameLidarrTrack = %v, want %v", tc.name, got, tc.wantSame) + } + } +} diff --git a/internal/tracks/service.go b/internal/tracks/service.go index 7abb98ec..1d576bd1 100644 --- a/internal/tracks/service.go +++ b/internal/tracks/service.go @@ -28,8 +28,10 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/apierror" + "git.fabledsword.com/bvandeusen/minstrel/internal/audit" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/library" + syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) // ErrNotFound is returned when the track id doesn't resolve. Aliased @@ -147,3 +149,70 @@ func (s *Service) RemoveTrack( return deleted.AlbumID, deleted.ArtistID, lidarrUnmonitorFailed, nil } + +// MergeDuplicates merges a duplicate group into the copy to keep +// (library.MergeDuplicateGroup), then — when asked — tells Lidarr to stop +// monitoring the removed copies so it does not download them again, and records +// the merge in the audit log. +// +// lidarrUnmonitorFailed reports that unmonitoring was asked for and at least one +// removed copy could not be unmonitored. Like RemoveTrack, that never fails the +// merge: the files and rows are already gone. +func (s *Service) MergeDuplicates( + ctx context.Context, groupID, survivorID, actorID pgtype.UUID, unmonitor bool, +) (res library.MergeResult, lidarrUnmonitorFailed bool, err error) { + res, err = library.MergeDuplicateGroup(ctx, s.pool, s.logger, s.dataDir, groupID, survivorID) + if err != nil { + return res, false, err + } + + if unmonitor && s.lidarr != nil { + for _, removed := range res.Removed { + if sameLidarrTrack(res.Survivor, removed) { + continue + } + if !hasLidarrIdentity(removed) { + s.logger.Warn("duplicate merge: lidarr unmonitor skipped — removed copy has no mbids", + "track_id", syncpkg.FormatUUID(removed.TrackID)) + lidarrUnmonitorFailed = true + continue + } + if uerr := s.lidarr.UnmonitorTrack(ctx, *removed.TrackMbid, *removed.AlbumMbid); uerr != nil { + s.logger.Warn("duplicate merge: lidarr unmonitor failed", + "track_id", syncpkg.FormatUUID(removed.TrackID), "err", uerr) + lidarrUnmonitorFailed = true + } + } + } + + 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, s.pool, s.logger, actorID, pgtype.UUID{}, audit.ActionDuplicateMerge, map[string]any{ + "group_id": syncpkg.FormatUUID(groupID), + "tier": res.Tier, + "survivor_track_id": syncpkg.FormatUUID(res.Survivor.TrackID), + "survivor_path": res.Survivor.FilePath, + "removed": removed, + "moved": map[string]any{ + "play_events": res.PlayEvents, "skip_events": res.SkipEvents, + "likes": res.Likes, "playlist_entries": res.PlaylistEntries, + }, + }) + return res, lidarrUnmonitorFailed, nil +} + +// sameLidarrTrack reports whether two copies are the same Lidarr track: one +// recording on one album. Lidarr monitors per album track, so when the removed +// copy is a second file of the kept copy's own album track — the #3885 case — +// unmonitoring it would also stop Lidarr managing the file the operator chose to +// keep. Those are skipped, and are not a failure. +func sameLidarrTrack(a, b library.MergedCopy) bool { + return hasLidarrIdentity(a) && hasLidarrIdentity(b) && + *a.TrackMbid == *b.TrackMbid && *a.AlbumMbid == *b.AlbumMbid +} + +func hasLidarrIdentity(c library.MergedCopy) bool { + return c.TrackMbid != nil && *c.TrackMbid != "" && c.AlbumMbid != nil && *c.AlbumMbid != "" +} diff --git a/web/src/lib/api/admin.duplicates.test.ts b/web/src/lib/api/admin.duplicates.test.ts index f3a2da1c..5be0cac9 100644 --- a/web/src/lib/api/admin.duplicates.test.ts +++ b/web/src/lib/api/admin.duplicates.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { dismissDuplicateGroup, listDuplicates, runDuplicateSweep } from './admin'; +import { dismissDuplicateGroup, listDuplicates, mergeDuplicateGroup, runDuplicateSweep } from './admin'; vi.mock('./client', () => ({ api: { get: vi.fn(), post: vi.fn() } @@ -27,4 +27,13 @@ describe('admin duplicates API', () => { await dismissDuplicateGroup('g/1'); expect(api.post).toHaveBeenCalledWith('/api/admin/library/duplicates/g%2F1/dismiss', {}); }); + + it('mergeDuplicateGroup POSTs the chosen survivor', async () => { + (api.post as unknown as ReturnType).mockResolvedValueOnce({ removed_paths: [] }); + await mergeDuplicateGroup('g-1', { survivor_track_id: 't-1', unmonitor: true }); + expect(api.post).toHaveBeenCalledWith('/api/admin/library/duplicates/g-1/merge', { + survivor_track_id: 't-1', + unmonitor: true + }); + }); }); diff --git a/web/src/lib/api/admin.ts b/web/src/lib/api/admin.ts index a560b7df..07e3940e 100644 --- a/web/src/lib/api/admin.ts +++ b/web/src/lib/api/admin.ts @@ -5,6 +5,7 @@ import type { ActionResult, AdminMissingResponse, AdminDuplicatesResponse, + MergeDuplicateResult, AdminPlaybackError, AdminQuarantineRow, LidarrConfig, @@ -723,6 +724,19 @@ export async function runDuplicateSweep(): Promise<{ started: boolean }> { return api.post<{ started: boolean }>('/api/admin/library/duplicates/sweep', {}); } +// Merges a group into the copy chosen to keep, removing the other copies' files. +// survivor_track_id is the operator's choice; the server checks it belongs to +// the group. +export async function mergeDuplicateGroup( + id: string, + body: { survivor_track_id: string; unmonitor: boolean } +): Promise { + return api.post( + `/api/admin/library/duplicates/${encodeURIComponent(id)}/merge`, + body + ); +} + export async function dismissDuplicateGroup(id: string): Promise { await api.post(`/api/admin/library/duplicates/${encodeURIComponent(id)}/dismiss`, {}); } diff --git a/web/src/lib/api/types.ts b/web/src/lib/api/types.ts index 2d5ecef0..76c5a645 100644 --- a/web/src/lib/api/types.ts +++ b/web/src/lib/api/types.ts @@ -460,6 +460,15 @@ export type AdminDuplicateSweep = { error_message: string | null; }; +// What a merge did (#3911). removed_paths are the files that were deleted from +// disk; lidarr_unmonitor_failed appears only when unmonitoring was asked for and +// failed. +export type MergeDuplicateResult = { + survivor_track_id: string; + removed_paths: string[]; + lidarr_unmonitor_failed?: boolean; +}; + export type AdminDuplicatesResponse = { sweep: AdminDuplicateSweep; fingerprints: { total: number; fingerprinted: number; rejected: number; pending: number }; diff --git a/web/src/lib/styles/error-copy.json b/web/src/lib/styles/error-copy.json index 260b3060..c18028a3 100644 --- a/web/src/lib/styles/error-copy.json +++ b/web/src/lib/styles/error-copy.json @@ -44,6 +44,7 @@ "file_delete_failed": "The file couldn't be deleted.", "sweep_in_progress": "A duplicate sweep is already running.", "duplicate_group_not_pending": "That group has already been resolved.", + "survivor_not_in_group": "That copy isn't part of this group any more.", "album_not_found": "That album no longer exists.", "artist_not_found": "That artist no longer exists.", "playlist_not_found": "That playlist no longer exists.", diff --git a/web/src/routes/admin/duplicates/+page.svelte b/web/src/routes/admin/duplicates/+page.svelte index 2e0881cb..ce80f41d 100644 --- a/web/src/routes/admin/duplicates/+page.svelte +++ b/web/src/routes/admin/duplicates/+page.svelte @@ -4,22 +4,30 @@ import { createDuplicatesQuery, runDuplicateSweep, - dismissDuplicateGroup + dismissDuplicateGroup, + mergeDuplicateGroup } from '$lib/api/admin'; import { errMessage } from '$lib/api/errors'; import { pushToast } from '$lib/stores/toast.svelte'; import { relativeTime } from '$lib/utils/relativeTime'; import type { AdminDuplicateGroup, AdminDuplicateMember } from '$lib/api/types'; - // Tracks the duplicate sweep believes hold one recording (#3912). A group is a - // proposal: nothing here deletes or merges. Dismissing one says "these are not - // duplicates", and the sweep will not propose that set again. + // Tracks the duplicate sweep believes hold one recording (#3912). Dismissing a + // group says "these are not duplicates", and the sweep will not propose that + // set again. Merging (#3911) keeps one copy, moves the others' likes, plays and + // playlist entries onto it, and deletes their files — so it asks twice. const PAGE_SIZE = 25; let offset = $state(0); let sweeping = $state(false); let dismissing = $state(null); + // Per group: which copy to keep (defaults to the proposed survivor), whether to + // unmonitor the removed copies in Lidarr, and the two-click confirm. + let keepChoice = $state>({}); + let unmonitorChoice = $state>({}); + let confirmingMerge = $state(null); + let merging = $state(null); const queryStore = $derived(createDuplicatesQuery(offset, PAGE_SIZE)); const query = $derived($queryStore); @@ -56,6 +64,40 @@ } } + function keeperOf(group: AdminDuplicateGroup): string { + return keepChoice[group.id] ?? group.survivor_track_id; + } + + function fileCountLabel(n: number): string { + return n === 1 ? '1 file' : `${n} files`; + } + + async function onMerge(group: AdminDuplicateGroup) { + // First click arms; the second, on the button that now names how many files + // go, does it. A merge deletes files, and nothing brings them back. + if (confirmingMerge !== group.id) { + confirmingMerge = group.id; + return; + } + confirmingMerge = null; + merging = group.id; + try { + const result = await mergeDuplicateGroup(group.id, { + survivor_track_id: keeperOf(group), + unmonitor: unmonitorChoice[group.id] ?? false + }); + pushToast(`Merged. Removed ${fileCountLabel(result.removed_paths.length)}.`); + if (result.lidarr_unmonitor_failed) { + pushToast("Merged, but Lidarr couldn't be told to stop monitoring the removed copies.", 'error'); + } + query.refetch(); + } catch (e: unknown) { + pushToast(errMessage(e), 'error'); + } finally { + merging = null; + } + } + // "Identical audio" and "same recording" are different claims, and an // operator deciding whether to merge needs to know which one they are // looking at before anything else. @@ -177,20 +219,71 @@

{tierLabel(group)}

Found {relativeTime(group.detected_at)}

- +
+ + +
+ {#if confirmingMerge === group.id} + +
+

+ The copy marked Keep stays. The other {fileCountLabel(group.members.length - 1)} will be + deleted from disk, and their likes, plays and playlist entries move to the copy kept. +

+ + +
+ {/if} +
    {#each group.members as m (m.track_id)} {@const keep = m.track_id === group.survivor_track_id}
  • + (keepChoice[group.id] = m.track_id)} + aria-label="Keep {m.title}, {m.file_format.toUpperCase()}, {m.file_path}" + />
    {m.title} diff --git a/web/src/routes/admin/duplicates/duplicates.test.ts b/web/src/routes/admin/duplicates/duplicates.test.ts index c8701d30..98d3c9d3 100644 --- a/web/src/routes/admin/duplicates/duplicates.test.ts +++ b/web/src/routes/admin/duplicates/duplicates.test.ts @@ -6,11 +6,20 @@ import type { AdminDuplicatesResponse } from '$lib/api/types'; vi.mock('$lib/api/admin', () => ({ createDuplicatesQuery: vi.fn(), runDuplicateSweep: vi.fn().mockResolvedValue({ started: true }), - dismissDuplicateGroup: vi.fn().mockResolvedValue(undefined) + dismissDuplicateGroup: vi.fn().mockResolvedValue(undefined), + mergeDuplicateGroup: vi.fn().mockResolvedValue({ + survivor_track_id: 'www-01', + removed_paths: ['/music/Moe Shop/WWW (2020)/www-02.mp3'] + }) })); import AdminDuplicatesPage from './+page.svelte'; -import { createDuplicatesQuery, dismissDuplicateGroup, runDuplicateSweep } from '$lib/api/admin'; +import { + createDuplicatesQuery, + dismissDuplicateGroup, + mergeDuplicateGroup, + runDuplicateSweep +} from '$lib/api/admin'; const HOUR = 3_600_000; const ago = (ms: number) => new Date(Date.now() - ms).toISOString(); @@ -144,4 +153,40 @@ describe('admin duplicates', () => { await fireEvent.click(screen.getByRole('button', { name: 'Sweep now' })); expect(runDuplicateSweep).toHaveBeenCalledTimes(1); }); + + // A merge deletes files. The first click must only arm it. + test('Merge needs a second click, and keeps the proposed copy by default', async () => { + renderWith(response()); + await fireEvent.click(screen.getByRole('button', { name: 'Merge…' })); + expect(mergeDuplicateGroup).not.toHaveBeenCalled(); + expect(text(screen.getByTestId('merge-confirm'))).toContain('The other 1 file will be deleted from disk'); + + await fireEvent.click(screen.getByRole('button', { name: 'Remove 1 file and merge' })); + expect(mergeDuplicateGroup).toHaveBeenCalledWith('g-1', { + survivor_track_id: 'www-01', + unmonitor: false + }); + }); + + test('choosing another copy to keep sends that copy', async () => { + renderWith(response()); + const radios = screen.getAllByRole('radio'); + expect((radios[0] as HTMLInputElement).checked).toBe(true); + await fireEvent.click(radios[1]); + + await fireEvent.click(screen.getByRole('button', { name: 'Merge…' })); + await fireEvent.click(screen.getByRole('button', { name: 'Remove 1 file and merge' })); + expect(mergeDuplicateGroup).toHaveBeenCalledWith('g-1', { + survivor_track_id: 'www-02', + unmonitor: false + }); + }); + + test('Cancel disarms the merge', async () => { + renderWith(response()); + await fireEvent.click(screen.getByRole('button', { name: 'Merge…' })); + await fireEvent.click(screen.getByRole('button', { name: 'Cancel' })); + expect(screen.queryByTestId('merge-confirm')).toBeNull(); + expect(screen.getByRole('button', { name: 'Merge…' })).toBeTruthy(); + }); });