diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index aa57a6a7..6ad3e0ec 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -220,6 +220,11 @@ func run() error { // internal/library/fingerprint_backfill.go for why. go library.NewFingerprintBackfillWorker(pool, logger.With("component", "fingerprint_backfill")).Run(ctx) + // Duplicate sweep (M400 #3910): proposes groups of tracks holding one + // recording, from the fingerprints above. Sweeps only when fingerprints have + // changed since the last sweep. + go library.NewDuplicateSweepWorker(pool, logger.With("component", "duplicate_sweep")).Run(ctx) + // Start the tag-enrichment worker (#1490). Reconciles the compiled-in // tag providers with tag_provider_settings, bumps the sources version if // the provider set changed (re-opening settled rows), then drains tracks diff --git a/internal/db/dbq/duplicates.sql.go b/internal/db/dbq/duplicates.sql.go new file mode 100644 index 00000000..94a8cb0e --- /dev/null +++ b/internal/db/dbq/duplicates.sql.go @@ -0,0 +1,320 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: duplicates.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const addDuplicateGroupMember = `-- name: AddDuplicateGroupMember :exec +INSERT INTO duplicate_group_members (group_id, track_id) +VALUES ($1, $2) +ON CONFLICT DO NOTHING +` + +type AddDuplicateGroupMemberParams struct { + GroupID pgtype.UUID + TrackID pgtype.UUID +} + +func (q *Queries) AddDuplicateGroupMember(ctx context.Context, arg AddDuplicateGroupMemberParams) error { + _, err := q.db.Exec(ctx, addDuplicateGroupMember, arg.GroupID, arg.TrackID) + return err +} + +const deleteStalePendingDuplicateGroups = `-- name: DeleteStalePendingDuplicateGroups :execrows +DELETE FROM duplicate_groups g + WHERE g.status = 'pending' + AND g.last_seen_sweep_id IS DISTINCT FROM $1 + AND (g.last_seen_sweep_id IS NULL + OR (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = g.last_seen_sweep_id) + < (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = $1)) +` + +// A pending proposal this sweep did not find again no longer describes the +// library: a member was re-fingerprinted, merged away or went missing. Dismissed +// groups are kept regardless — they are the memory of a decision. +// +// Only proposals last confirmed by an EARLIER sweep go. Should two sweeps ever +// overlap (a manual trigger racing the worker), neither may delete what the other +// has just found. +func (q *Queries) DeleteStalePendingDuplicateGroups(ctx context.Context, sweepID pgtype.UUID) (int64, error) { + result, err := q.db.Exec(ctx, deleteStalePendingDuplicateGroups, sweepID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const finishDuplicateSweep = `-- name: FinishDuplicateSweep :exec +UPDATE duplicate_sweeps + SET finished_at = now(), + candidates = $1, + groups_found = $2, + oversize_clusters = $3, + error_message = NULLIF($4::text, '') + WHERE id = $5 +` + +type FinishDuplicateSweepParams struct { + Candidates *int32 + GroupsFound *int32 + OversizeClusters *int32 + ErrorMessage string + ID pgtype.UUID +} + +func (q *Queries) FinishDuplicateSweep(ctx context.Context, arg FinishDuplicateSweepParams) error { + _, err := q.db.Exec(ctx, finishDuplicateSweep, + arg.Candidates, + arg.GroupsFound, + arg.OversizeClusters, + arg.ErrorMessage, + arg.ID, + ) + return err +} + +const getInFlightDuplicateSweep = `-- name: GetInFlightDuplicateSweep :one +SELECT id, started_at + FROM duplicate_sweeps + WHERE finished_at IS NULL + ORDER BY started_at DESC + LIMIT 1 +` + +type GetInFlightDuplicateSweepRow struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz +} + +// The guard against two sweeps at once: "in flight" is finished_at IS NULL. +func (q *Queries) GetInFlightDuplicateSweep(ctx context.Context) (GetInFlightDuplicateSweepRow, error) { + row := q.db.QueryRow(ctx, getInFlightDuplicateSweep) + var i GetInFlightDuplicateSweepRow + err := row.Scan(&i.ID, &i.StartedAt) + return i, err +} + +const getLatestDuplicateSweep = `-- name: GetLatestDuplicateSweep :one +SELECT id, started_at, finished_at, candidates, groups_found, oversize_clusters, error_message + FROM duplicate_sweeps + ORDER BY started_at DESC + LIMIT 1 +` + +func (q *Queries) GetLatestDuplicateSweep(ctx context.Context) (DuplicateSweep, error) { + row := q.db.QueryRow(ctx, getLatestDuplicateSweep) + var i DuplicateSweep + err := row.Scan( + &i.ID, + &i.StartedAt, + &i.FinishedAt, + &i.Candidates, + &i.GroupsFound, + &i.OversizeClusters, + &i.ErrorMessage, + ) + return i, err +} + +const getLatestFingerprintComputedAt = `-- name: GetLatestFingerprintComputedAt :one +SELECT max(computed_at)::timestamptz AS latest FROM track_fingerprints +` + +// Whether a sweep has anything new to look at: fingerprints written since the +// last sweep started. +func (q *Queries) GetLatestFingerprintComputedAt(ctx context.Context) (pgtype.Timestamptz, error) { + row := q.db.QueryRow(ctx, getLatestFingerprintComputedAt) + var latest pgtype.Timestamptz + err := row.Scan(&latest) + return latest, err +} + +const listDismissedDuplicateMemberSets = `-- name: ListDismissedDuplicateMemberSets :many +SELECT g.id, array_agg(m.track_id ORDER BY m.track_id)::uuid[] AS track_ids + FROM duplicate_groups g + JOIN duplicate_group_members m ON m.group_id = g.id + WHERE g.status = 'dismissed' + GROUP BY g.id +` + +type ListDismissedDuplicateMemberSetsRow struct { + ID pgtype.UUID + TrackIds []pgtype.UUID +} + +// What the operator has already said are not duplicates. A new proposal whose +// every member sat together in one of these is not proposed again. +func (q *Queries) ListDismissedDuplicateMemberSets(ctx context.Context) ([]ListDismissedDuplicateMemberSetsRow, error) { + rows, err := q.db.Query(ctx, listDismissedDuplicateMemberSets) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListDismissedDuplicateMemberSetsRow + for rows.Next() { + var i ListDismissedDuplicateMemberSetsRow + if err := rows.Scan(&i.ID, &i.TrackIds); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDuplicateCandidates = `-- name: ListDuplicateCandidates :many +SELECT t.id, t.duration_ms, f.audio_stream_sha256, f.chromaprint + FROM tracks t + JOIN track_fingerprints f ON f.track_id = t.id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= $1 + AND f.chromaprint IS NOT NULL + AND (t.duration_ms, t.id) > ($2::integer, $3::uuid) + ORDER BY t.duration_ms, t.id + LIMIT $4 +` + +type ListDuplicateCandidatesParams struct { + CurrentVersion int16 + AfterDurationMs int32 + AfterID pgtype.UUID + PageLimit int32 +} + +type ListDuplicateCandidatesRow struct { + ID pgtype.UUID + DurationMs int32 + AudioStreamSha256 []byte + Chromaprint []int32 +} + +// The acoustic tier's input, one page at a time in (duration_ms, id) order so the +// sweep holds only a sliding window of durations. Tracks without a chromaprint +// cannot be compared acoustically and are left out; any exact duplicates among +// them come from ListExactDuplicateHashes. +func (q *Queries) ListDuplicateCandidates(ctx context.Context, arg ListDuplicateCandidatesParams) ([]ListDuplicateCandidatesRow, error) { + rows, err := q.db.Query(ctx, listDuplicateCandidates, + arg.CurrentVersion, + arg.AfterDurationMs, + arg.AfterID, + arg.PageLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListDuplicateCandidatesRow + for rows.Next() { + var i ListDuplicateCandidatesRow + if err := rows.Scan( + &i.ID, + &i.DurationMs, + &i.AudioStreamSha256, + &i.Chromaprint, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listExactDuplicateHashes = `-- name: ListExactDuplicateHashes :many +SELECT f.audio_stream_sha256, + array_agg(t.id ORDER BY t.id)::uuid[] AS track_ids + FROM track_fingerprints f + JOIN tracks t ON t.id = f.track_id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= $1 + AND f.audio_stream_sha256 IS NOT NULL + GROUP BY f.audio_stream_sha256 +HAVING count(*) > 1 +` + +type ListExactDuplicateHashesRow struct { + AudioStreamSha256 []byte + TrackIds []pgtype.UUID +} + +// The exact tier, library-wide in one pass: identical encoded audio shared by +// more than one present track. +func (q *Queries) ListExactDuplicateHashes(ctx context.Context, currentVersion int16) ([]ListExactDuplicateHashesRow, error) { + rows, err := q.db.Query(ctx, listExactDuplicateHashes, currentVersion) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListExactDuplicateHashesRow + for rows.Next() { + var i ListExactDuplicateHashesRow + if err := rows.Scan(&i.AudioStreamSha256, &i.TrackIds); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const startDuplicateSweep = `-- name: StartDuplicateSweep :one +INSERT INTO duplicate_sweeps DEFAULT VALUES RETURNING id, started_at +` + +type StartDuplicateSweepRow struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz +} + +func (q *Queries) StartDuplicateSweep(ctx context.Context) (StartDuplicateSweepRow, error) { + row := q.db.QueryRow(ctx, startDuplicateSweep) + var i StartDuplicateSweepRow + err := row.Scan(&i.ID, &i.StartedAt) + return i, err +} + +const upsertDuplicateGroup = `-- name: UpsertDuplicateGroup :one +INSERT INTO duplicate_groups (member_key, tier, worst_bit_error_rate, last_seen_sweep_id) +VALUES ($1, $2, $3, $4) +ON CONFLICT (member_key) DO UPDATE + SET tier = EXCLUDED.tier, + worst_bit_error_rate = EXCLUDED.worst_bit_error_rate, + last_seen_sweep_id = EXCLUDED.last_seen_sweep_id + WHERE duplicate_groups.status = 'pending' +RETURNING id +` + +type UpsertDuplicateGroupParams struct { + MemberKey string + Tier string + WorstBitErrorRate *float32 + SweepID pgtype.UUID +} + +// Proposes a group, or refreshes one already pending. A group already dismissed +// or merged is left exactly as it is: the WHERE on the update makes the conflict +// a no-op, and the caller sees no row. +func (q *Queries) UpsertDuplicateGroup(ctx context.Context, arg UpsertDuplicateGroupParams) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, upsertDuplicateGroup, + arg.MemberKey, + arg.Tier, + arg.WorstBitErrorRate, + arg.SweepID, + ) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index 07ca6f75..cddadd0c 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -297,6 +297,32 @@ type DiscoverTuning struct { UpdatedAt pgtype.Timestamptz } +type DuplicateGroup struct { + ID pgtype.UUID + MemberKey string + Tier string + WorstBitErrorRate *float32 + Status string + DetectedAt pgtype.Timestamptz + LastSeenSweepID pgtype.UUID + ResolvedAt pgtype.Timestamptz +} + +type DuplicateGroupMember struct { + GroupID pgtype.UUID + TrackID pgtype.UUID +} + +type DuplicateSweep struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz + FinishedAt pgtype.Timestamptz + Candidates *int32 + GroupsFound *int32 + OversizeClusters *int32 + ErrorMessage *string +} + type GeneralLike struct { UserID pgtype.UUID TrackID pgtype.UUID diff --git a/internal/db/migrations/0059_duplicate_groups.down.sql b/internal/db/migrations/0059_duplicate_groups.down.sql new file mode 100644 index 00000000..1d37bf4c --- /dev/null +++ b/internal/db/migrations/0059_duplicate_groups.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS tracks_duration_id_idx; +DROP TABLE duplicate_group_members; +DROP TABLE duplicate_groups; +DROP TABLE duplicate_sweeps; diff --git a/internal/db/migrations/0059_duplicate_groups.up.sql b/internal/db/migrations/0059_duplicate_groups.up.sql new file mode 100644 index 00000000..17a112b5 --- /dev/null +++ b/internal/db/migrations/0059_duplicate_groups.up.sql @@ -0,0 +1,54 @@ +-- 0059_duplicate_groups.up.sql — proposed duplicates and the sweeps that find +-- them (Scribe milestone #400: #3910). +-- +-- The sweep compares fingerprints (track_fingerprints, 0058) and proposes groups +-- of tracks that hold one recording. Nothing here merges anything: a group is a +-- proposal the operator reviews, and the merge (#3911) is a separate act. + +-- One row per sweep. Lets the report tell "the sweep has never run" apart from +-- "it ran and found nothing", and gives the in-flight guard something to check, +-- the same way scan_runs does for the library scan. +CREATE TABLE duplicate_sweeps ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + started_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz, + candidates integer, + groups_found integer, + oversize_clusters integer, + error_message text +); +CREATE INDEX duplicate_sweeps_started_at_idx ON duplicate_sweeps (started_at DESC); + +CREATE TABLE duplicate_groups ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + -- The group's identity: its member track ids, sorted and joined. A sweep + -- that finds the same tracks again updates this row rather than proposing + -- them twice, and a dismissal stays attached to the set it was made about. + member_key text NOT NULL UNIQUE, + -- Rule 36: a new value for either CHECK swaps the constraint in the same + -- migration. + tier text NOT NULL CHECK (tier IN ('exact', 'acoustic')), + -- Largest disagreement between any two members; NULL for exact groups, + -- which have no score. + worst_bit_error_rate real, + status text NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'dismissed', 'merged')), + detected_at timestamptz NOT NULL DEFAULT now(), + last_seen_sweep_id uuid REFERENCES duplicate_sweeps (id) ON DELETE SET NULL, + resolved_at timestamptz +); +CREATE INDEX duplicate_groups_status_idx ON duplicate_groups (status); + +CREATE TABLE duplicate_group_members ( + group_id uuid NOT NULL REFERENCES duplicate_groups (id) ON DELETE CASCADE, + -- CASCADE is right here: a track that genuinely leaves the library has no + -- place in a proposal about its duplicates. + track_id uuid NOT NULL REFERENCES tracks (id) ON DELETE CASCADE, + PRIMARY KEY (group_id, track_id) +); +CREATE INDEX duplicate_group_members_track_idx ON duplicate_group_members (track_id); + +-- The sweep streams candidates in (duration_ms, id) order, keyset-paged, so it +-- only ever holds a few seconds' worth of durations in memory. Without this each +-- page would sort the whole library again. +CREATE INDEX tracks_duration_id_idx ON tracks (duration_ms, id); diff --git a/internal/db/queries/duplicates.sql b/internal/db/queries/duplicates.sql new file mode 100644 index 00000000..def52620 --- /dev/null +++ b/internal/db/queries/duplicates.sql @@ -0,0 +1,100 @@ +-- name: StartDuplicateSweep :one +INSERT INTO duplicate_sweeps DEFAULT VALUES RETURNING id, started_at; + +-- name: FinishDuplicateSweep :exec +UPDATE duplicate_sweeps + SET finished_at = now(), + candidates = sqlc.arg(candidates), + groups_found = sqlc.arg(groups_found), + oversize_clusters = sqlc.arg(oversize_clusters), + error_message = NULLIF(sqlc.arg(error_message)::text, '') + WHERE id = sqlc.arg(id); + +-- name: GetInFlightDuplicateSweep :one +-- The guard against two sweeps at once: "in flight" is finished_at IS NULL. +SELECT id, started_at + FROM duplicate_sweeps + WHERE finished_at IS NULL + ORDER BY started_at DESC + LIMIT 1; + +-- name: GetLatestDuplicateSweep :one +SELECT id, started_at, finished_at, candidates, groups_found, oversize_clusters, error_message + FROM duplicate_sweeps + ORDER BY started_at DESC + LIMIT 1; + +-- name: GetLatestFingerprintComputedAt :one +-- Whether a sweep has anything new to look at: fingerprints written since the +-- last sweep started. +SELECT max(computed_at)::timestamptz AS latest FROM track_fingerprints; + +-- name: ListExactDuplicateHashes :many +-- The exact tier, library-wide in one pass: identical encoded audio shared by +-- more than one present track. +SELECT f.audio_stream_sha256, + array_agg(t.id ORDER BY t.id)::uuid[] AS track_ids + FROM track_fingerprints f + JOIN tracks t ON t.id = f.track_id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= sqlc.arg(current_version) + AND f.audio_stream_sha256 IS NOT NULL + GROUP BY f.audio_stream_sha256 +HAVING count(*) > 1; + +-- name: ListDuplicateCandidates :many +-- The acoustic tier's input, one page at a time in (duration_ms, id) order so the +-- sweep holds only a sliding window of durations. Tracks without a chromaprint +-- cannot be compared acoustically and are left out; any exact duplicates among +-- them come from ListExactDuplicateHashes. +SELECT t.id, t.duration_ms, f.audio_stream_sha256, f.chromaprint + FROM tracks t + JOIN track_fingerprints f ON f.track_id = t.id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= sqlc.arg(current_version) + AND f.chromaprint IS NOT NULL + AND (t.duration_ms, t.id) > (sqlc.arg(after_duration_ms)::integer, sqlc.arg(after_id)::uuid) + ORDER BY t.duration_ms, t.id + LIMIT sqlc.arg(page_limit); + +-- name: ListDismissedDuplicateMemberSets :many +-- What the operator has already said are not duplicates. A new proposal whose +-- every member sat together in one of these is not proposed again. +SELECT g.id, array_agg(m.track_id ORDER BY m.track_id)::uuid[] AS track_ids + FROM duplicate_groups g + JOIN duplicate_group_members m ON m.group_id = g.id + WHERE g.status = 'dismissed' + GROUP BY g.id; + +-- name: UpsertDuplicateGroup :one +-- Proposes a group, or refreshes one already pending. A group already dismissed +-- or merged is left exactly as it is: the WHERE on the update makes the conflict +-- a no-op, and the caller sees no row. +INSERT INTO duplicate_groups (member_key, tier, worst_bit_error_rate, last_seen_sweep_id) +VALUES (sqlc.arg(member_key), sqlc.arg(tier), sqlc.narg(worst_bit_error_rate), sqlc.arg(sweep_id)) +ON CONFLICT (member_key) DO UPDATE + SET tier = EXCLUDED.tier, + worst_bit_error_rate = EXCLUDED.worst_bit_error_rate, + last_seen_sweep_id = EXCLUDED.last_seen_sweep_id + WHERE duplicate_groups.status = 'pending' +RETURNING id; + +-- name: AddDuplicateGroupMember :exec +INSERT INTO duplicate_group_members (group_id, track_id) +VALUES (sqlc.arg(group_id), sqlc.arg(track_id)) +ON CONFLICT DO NOTHING; + +-- name: DeleteStalePendingDuplicateGroups :execrows +-- A pending proposal this sweep did not find again no longer describes the +-- library: a member was re-fingerprinted, merged away or went missing. Dismissed +-- groups are kept regardless — they are the memory of a decision. +-- +-- Only proposals last confirmed by an EARLIER sweep go. Should two sweeps ever +-- overlap (a manual trigger racing the worker), neither may delete what the other +-- has just found. +DELETE FROM duplicate_groups g + WHERE g.status = 'pending' + AND g.last_seen_sweep_id IS DISTINCT FROM sqlc.arg(sweep_id) + AND (g.last_seen_sweep_id IS NULL + OR (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = g.last_seen_sweep_id) + < (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = sqlc.arg(sweep_id))); diff --git a/internal/dbtest/reset.go b/internal/dbtest/reset.go index a99db739..25e0b076 100644 --- a/internal/dbtest/reset.go +++ b/internal/dbtest/reset.go @@ -87,6 +87,9 @@ var dataTables = []string{ // pristine Discover knobs rather than whatever a previous test tuned. "discover_tuning", "recommendation_tuning_audit", + "duplicate_group_members", // M400 + "duplicate_groups", + "duplicate_sweeps", "track_fingerprints", // M400 "tracks", "albums", diff --git a/internal/library/duplicate_match.go b/internal/library/duplicate_match.go index 75539e88..d45b83f4 100644 --- a/internal/library/duplicate_match.go +++ b/internal/library/duplicate_match.go @@ -82,15 +82,52 @@ type acousticScore struct { BitErrorRate float64 } +// preparedPrint is a fingerprint with the parts every comparison needs worked +// out once. The sweep compares each track with every other track within a few +// seconds of its duration, so rebuilding the alignment index for each pair would +// dominate its cost. +type preparedPrint struct { + items []int32 + index map[uint32][]int + informative bool +} + +// preparePrint indexes a fingerprint's items by their high bits and records +// whether it varies enough to be compared at all. +func preparePrint(fp []int32) *preparedPrint { + // Each bucket keeps only a few positions: a value repeating many times is + // uninformative, and letting it vote once per repeat would make every + // pairing O(n²). + const keepPerBucket = 4 + p := &preparedPrint{items: fp, index: make(map[uint32][]int, len(fp))} + seen := make(map[int32]struct{}, len(fp)) + for i, v := range fp { + seen[v] = struct{}{} + key := alignKey(v) + if pos := p.index[key]; len(pos) < keepPerBucket { + p.index[key] = append(pos, i) + } + } + p.informative = len(fp) > 0 && float64(len(seen)) >= minDistinctFraction*float64(len(fp)) + return p +} + +func alignKey(v int32) uint32 { return uint32(v) >> (32 - alignMatchBits) } + // compareChromaprint aligns two raw fingerprints and measures how much they // disagree. ok is false when no verdict is possible: no offset gathered any // votes, the overlap at the best offset is too short, or either side carries // too little information to mean anything. func compareChromaprint(a, b []int32) (acousticScore, bool) { - if len(a) < minOverlapItems || len(b) < minOverlapItems { + return comparePrepared(preparePrint(a), preparePrint(b)) +} + +// comparePrepared is compareChromaprint over fingerprints already prepared. +func comparePrepared(a, b *preparedPrint) (acousticScore, bool) { + if len(a.items) < minOverlapItems || len(b.items) < minOverlapItems { return acousticScore{}, false } - if !informative(a) || !informative(b) { + if !a.informative || !b.informative { return acousticScore{}, false } @@ -101,14 +138,14 @@ func compareChromaprint(a, b []int32) (acousticScore, bool) { // a[i] aligns with b[i+offset]; walk the indices valid on both sides. start := max(0, -offset) - end := min(len(a), len(b)-offset) + end := min(len(a.items), len(b.items)-offset) overlap := end - start if overlap < minOverlapItems { return acousticScore{}, false } errBits := 0 for i := start; i < end; i++ { - errBits += bits.OnesCount32(uint32(a[i]) ^ uint32(b[i+offset])) + errBits += bits.OnesCount32(uint32(a.items[i]) ^ uint32(b.items[i+offset])) } return acousticScore{ Offset: offset, @@ -118,22 +155,10 @@ func compareChromaprint(a, b []int32) (acousticScore, bool) { } // bestOffset returns the relative shift most items agree on. -func bestOffset(a, b []int32) (int, bool) { - // Index a's items by their high bits. Each bucket keeps only a few - // positions: a value repeating many times is uninformative, and letting it - // vote once per repeat would make every pairing O(n²). - const keepPerBucket = 4 - positions := make(map[uint32][]int, len(a)) - for i, v := range a { - key := uint32(v) >> (32 - alignMatchBits) - if p := positions[key]; len(p) < keepPerBucket { - positions[key] = append(p, i) - } - } - +func bestOffset(a, b *preparedPrint) (int, bool) { votes := make([]int, 2*maxAlignOffsetItems+1) - for j, v := range b { - for _, i := range positions[uint32(v)>>(32-alignMatchBits)] { + for j, v := range b.items { + for _, i := range a.index[alignKey(v)] { off := j - i if off >= -maxAlignOffsetItems && off <= maxAlignOffsetItems { votes[off+maxAlignOffsetItems]++ @@ -152,15 +177,6 @@ func bestOffset(a, b []int32) (int, bool) { return best, bestVotes > 0 } -// informative reports whether a fingerprint varies enough to be compared. -func informative(fp []int32) bool { - seen := make(map[int32]struct{}, len(fp)) - for _, v := range fp { - seen[v] = struct{}{} - } - return float64(len(seen)) >= minDistinctFraction*float64(len(fp)) -} - // fingerprintCandidate is one track as the grouping sees it. type fingerprintCandidate struct { ID string @@ -197,149 +213,206 @@ type groupingResult struct { OversizeClusters int } -// groupDuplicates proposes duplicate groups among candidates. +// groupUnit is one thing the acoustic pass compares: a single track, or an exact +// group standing in for all its byte-identical copies. +type groupUnit struct { + ids []string // every member, sorted + durationMs int32 + sortKey string // the representative's id: ties on duration break on it + print *preparedPrint + exact bool // more than one member with identical audio + assigned bool +} + +// streamGrouper is the acoustic pass over units arriving in (durationMs, sortKey) +// order. It holds only the units within durationToleranceMs of the oldest one +// not yet settled, so memory is bounded by the densest few seconds of the +// library rather than by its size — the whole library's fingerprints would be +// hundreds of megabytes. // -// Exact groups come first: tracks sharing an audio stream hash. Each exact group -// is then treated as a single unit for the acoustic pass, so its members are -// never compared with each other again. +// A seed can be settled as soon as a unit arrives beyond its window: everything +// it could group with has already arrived, and no later seed can reach back to +// it because seeds only look forward. That is what makes the streamed result +// identical to running the same pass over the whole sorted list. // -// Acoustic grouping is COMPLETE-LINKAGE: a unit joins a group only if it matches -// every unit already in it, within the duration tolerance and the bit-error -// limit. Single-linkage would let a chain of near-misses — A close to B, B close -// to C — drag A and C, which are not close, into one proposed merge. Complete -// linkage also means any member can be chosen as the survivor (#3911). +// Grouping is COMPLETE-LINKAGE: a unit joins a group only if it matches every +// unit already in it, within the duration tolerance and the bit-error limit. +// Single-linkage would let a chain of near-misses — A close to B, B close to C — +// drag A and C, which are not close, into one proposed merge. Complete linkage +// also means any member can be chosen as the survivor (#3911). // -// When an acoustic group absorbs an exact group, the result is tier acoustic: -// a group is only as certain as its weakest link. +// When an acoustic group absorbs an exact group, the result is tier acoustic: a +// group is only as certain as its weakest link. +type streamGrouper struct { + maxBitErrorRate float64 + window []*groupUnit + res groupingResult +} + +func newStreamGrouper(maxBitErrorRate float64) *streamGrouper { + return &streamGrouper{maxBitErrorRate: maxBitErrorRate} +} + +// push adds the next unit. Units must arrive in non-decreasing +// (durationMs, sortKey) order. +func (g *streamGrouper) push(u *groupUnit) { + g.window = append(g.window, u) + for len(g.window) > 1 && u.durationMs-g.window[0].durationMs > durationToleranceMs { + g.settleOldest() + } +} + +// finish settles every unit still waiting and returns what was found. Groups +// are in no particular order; callers sort with sortGroups. +func (g *streamGrouper) finish() groupingResult { + for len(g.window) > 0 { + g.settleOldest() + } + return g.res +} + +func (g *streamGrouper) settleOldest() { + seed := g.window[0] + g.window[0] = nil // release it: the window's backing array outlives the slide + g.window = g.window[1:] + if seed.assigned { + return + } + + group := []*groupUnit{seed} + worst := 0.0 + for _, cand := range g.window { + if cand.durationMs-seed.durationMs > durationToleranceMs { + break + } + if cand.assigned { + continue + } + joined, worstWithCand := true, worst + for _, member := range group { + if abs32(cand.durationMs-member.durationMs) > durationToleranceMs { + joined = false + break + } + score, ok := comparePrepared(member.print, cand.print) + if !ok || score.BitErrorRate > g.maxBitErrorRate { + joined = false + break + } + worstWithCand = math.Max(worstWithCand, score.BitErrorRate) + } + if joined { + group = append(group, cand) + worst = worstWithCand + } + } + + if len(group) == 1 { + if seed.exact { + g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: seed.ids}) + } + return + } + // Count units, not tracks: an absorbed exact group is one piece of acoustic + // evidence however many identical files it holds. + if len(group) > maxAcousticGroupSize { + g.res.OversizeClusters++ + for _, member := range group { + member.assigned = true + // The acoustic evidence is untrustworthy; identical bytes are not. + // An exact group caught inside an oversize cluster is still proposed. + if member.exact { + g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: member.ids}) + } + } + return + } + var members []string + for _, member := range group { + member.assigned = true + members = append(members, member.ids...) + } + sort.Strings(members) + g.res.Groups = append(g.res.Groups, duplicateGroup{ + Tier: tierAcoustic, Members: members, WorstBitErrorRate: worst, + }) +} + +// groupDuplicates proposes duplicate groups among candidates held in memory. It +// runs the same streamGrouper the sweep uses, so there is one grouping rule. +// +// Exact groups come first: tracks sharing an audio stream hash. Each becomes a +// single unit for the acoustic pass, represented by its member with the lowest +// (duration, id) that has a chromaprint. That is the member the sweep's +// duration-ordered stream meets first, which keeps the two identical. An exact +// group with no chromaprint at all cannot be compared acoustically and stands +// on its own. // // The output does not depend on input order. func groupDuplicates(cands []fingerprintCandidate, maxBitErrorRate float64) groupingResult { - var res groupingResult - - // Exact tier. byHash := map[string][]fingerprintCandidate{} - var noHash []fingerprintCandidate + var units []*groupUnit + var printless []duplicateGroup for _, c := range cands { - if len(c.StreamSHA256) == 0 { - noHash = append(noHash, c) + if len(c.StreamSHA256) > 0 { + byHash[string(c.StreamSHA256)] = append(byHash[string(c.StreamSHA256)], c) continue } - k := string(c.StreamSHA256) - byHash[k] = append(byHash[k], c) + if len(c.Chromaprint) > 0 { + units = append(units, &groupUnit{ + ids: []string{c.ID}, durationMs: c.DurationMs, sortKey: c.ID, print: preparePrint(c.Chromaprint), + }) + } } - - // A unit is one exact group, or one track with no exact duplicate. - type unit struct { - members []fingerprintCandidate - durationMs int32 - print []int32 - exact bool - } - var units []unit for _, group := range byHash { - sortCandidates(group) - u := unit{members: group, durationMs: group[0].DurationMs, exact: len(group) > 1} - for _, m := range group { - if len(m.Chromaprint) > 0 { - u.print = m.Chromaprint - break + ids := make([]string, len(group)) + for i, m := range group { + ids[i] = m.ID + } + sort.Strings(ids) + + var rep *fingerprintCandidate + for i := range group { + m := &group[i] + if len(m.Chromaprint) == 0 { + continue + } + if rep == nil || m.DurationMs < rep.DurationMs || (m.DurationMs == rep.DurationMs && m.ID < rep.ID) { + rep = m } } - units = append(units, u) - } - for _, c := range noHash { - units = append(units, unit{members: []fingerprintCandidate{c}, durationMs: c.DurationMs, print: c.Chromaprint}) + if rep == nil { + if len(group) > 1 { + printless = append(printless, duplicateGroup{Tier: tierExact, Members: ids}) + } + continue + } + units = append(units, &groupUnit{ + ids: ids, durationMs: rep.DurationMs, sortKey: rep.ID, + print: preparePrint(rep.Chromaprint), exact: len(group) > 1, + }) } - // Deterministic order: duration, then the first member's ID. Sorting by - // duration also lets the scan below stop as soon as durations are too far - // apart, which is the blocking #3910 relies on. sort.Slice(units, func(i, j int) bool { if units[i].durationMs != units[j].durationMs { return units[i].durationMs < units[j].durationMs } - return units[i].members[0].ID < units[j].members[0].ID + return units[i].sortKey < units[j].sortKey }) - - assigned := make([]bool, len(units)) - for i := range units { - if assigned[i] || len(units[i].print) == 0 { - continue - } - group := []int{i} - worst := 0.0 - for j := i + 1; j < len(units); j++ { - if units[j].durationMs-units[i].durationMs > durationToleranceMs { - break - } - if assigned[j] || len(units[j].print) == 0 { - continue - } - // Complete linkage: j must match every member so far. - joined, worstWithJ := true, worst - for _, g := range group { - if abs32(units[j].durationMs-units[g].durationMs) > durationToleranceMs { - joined = false - break - } - score, ok := compareChromaprint(units[g].print, units[j].print) - if !ok || score.BitErrorRate > maxBitErrorRate { - joined = false - break - } - worstWithJ = math.Max(worstWithJ, score.BitErrorRate) - } - if joined { - group = append(group, j) - worst = worstWithJ - } - } - - if len(group) == 1 { - continue - } - // Count units, not tracks: an absorbed exact group is one piece of - // acoustic evidence however many identical files it holds. - if len(group) > maxAcousticGroupSize { - res.OversizeClusters++ - for _, g := range group { - assigned[g] = true - } - continue - } - var members []string - for _, g := range group { - assigned[g] = true - for _, m := range units[g].members { - members = append(members, m.ID) - } - } - sort.Strings(members) - res.Groups = append(res.Groups, duplicateGroup{ - Tier: tierAcoustic, Members: members, WorstBitErrorRate: worst, - }) + g := newStreamGrouper(maxBitErrorRate) + for _, u := range units { + g.push(u) } - - // Exact groups that no acoustic group absorbed stand on their own. - for i, u := range units { - if assigned[i] || !u.exact { - continue - } - members := make([]string, len(u.members)) - for k, m := range u.members { - members[k] = m.ID - } - res.Groups = append(res.Groups, duplicateGroup{Tier: tierExact, Members: members}) - } - - sort.Slice(res.Groups, func(i, j int) bool { - return res.Groups[i].Members[0] < res.Groups[j].Members[0] - }) + res := g.finish() + res.Groups = append(res.Groups, printless...) + sortGroups(res.Groups) return res } -func sortCandidates(cs []fingerprintCandidate) { - sort.Slice(cs, func(i, j int) bool { return cs[i].ID < cs[j].ID }) +// sortGroups orders groups by their first member. Groups are disjoint, so that +// is a total order. +func sortGroups(groups []duplicateGroup) { + sort.Slice(groups, func(i, j int) bool { return groups[i].Members[0] < groups[j].Members[0] }) } func abs(n int) int { diff --git a/internal/library/duplicate_sweep.go b/internal/library/duplicate_sweep.go new file mode 100644 index 00000000..f1c3cf0f --- /dev/null +++ b/internal/library/duplicate_sweep.go @@ -0,0 +1,360 @@ +package library + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sort" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" +) + +// Duplicate sweep (M400 #3910). +// +// Reads fingerprints, runs them through the matcher and records what it proposes +// in duplicate_groups. It never merges or deletes anything the operator has not +// asked for: a group is a proposal, reviewed in the admin report (#3912). +// +// Blocked on duration, deliberately not on title: the #3885 pair are titled +// "WWW" and "WWW (instrumental)", so a title block would have missed the case +// that started the milestone. Candidates stream in (duration_ms, id) order and +// the grouper holds only a few seconds of durations at a time. + +// duplicateCandidatePage is how many candidates one query returns. Each row +// carries a ~4 KB fingerprint, so a page is about 2 MB. +const duplicateCandidatePage = 500 + +// duplicateSweepTick is how often the worker checks for anything new to sweep. +// With nothing new, a tick is two cheap aggregate queries. +const duplicateSweepTick = time.Hour + +// staleDuplicateSweepThreshold is the age past which an in-flight sweep is +// assumed dead — a crash mid-sweep leaves finished_at NULL for ever — and another +// may start. Twice the library scan's threshold, because a sweep compares +// fingerprints across the whole library and can legitimately run long on a big +// one. +const staleDuplicateSweepThreshold = 2 * time.Hour + +// duplicateSweepFinishTimeout bounds recording that a sweep ended. It runs on a +// context detached from the sweep's own, so a sweep cancelled at shutdown still +// closes its row rather than leaving it in flight until the reaper. +const duplicateSweepFinishTimeout = 10 * time.Second + +// DuplicateSweepResult tallies one sweep. +type DuplicateSweepResult struct { + Candidates int // tracks with a chromaprint that were streamed + Groups int // groups the matcher found + Proposed int // written as pending, new or refreshed + Suppressed int // not proposed: already dismissed or resolved by the operator + Retired int // pending proposals this sweep did not find again, removed + Oversize int // acoustic clusters too large to propose +} + +// RunDuplicateSweep runs one sweep and records it in duplicate_sweeps. +func RunDuplicateSweep(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) (DuplicateSweepResult, error) { + return runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage) +} + +func runDuplicateSweep( + ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, pageSize int32, +) (DuplicateSweepResult, error) { + q := dbq.New(pool) + sweep, err := q.StartDuplicateSweep(ctx) + if err != nil { + return DuplicateSweepResult{}, fmt.Errorf("start duplicate sweep: %w", err) + } + + res, runErr := sweepDuplicates(ctx, q, sweep.ID, pageSize) + + finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), duplicateSweepFinishTimeout) + defer cancel() + errMsg := "" + if runErr != nil { + errMsg = runErr.Error() + } + candidates, groups, oversize := int32(res.Candidates), int32(res.Groups), int32(res.Oversize) + if ferr := q.FinishDuplicateSweep(finishCtx, dbq.FinishDuplicateSweepParams{ + ID: sweep.ID, Candidates: &candidates, GroupsFound: &groups, OversizeClusters: &oversize, + ErrorMessage: errMsg, + }); ferr != nil { + logger.Error("duplicate sweep: recording the end of the sweep failed", "err", ferr) + if runErr == nil { + runErr = fmt.Errorf("finish duplicate sweep: %w", ferr) + } + } + + logger.Info("duplicate sweep complete", + "candidates", res.Candidates, "groups", res.Groups, "proposed", res.Proposed, + "suppressed", res.Suppressed, "retired", res.Retired, "oversize", res.Oversize, "err", runErr) + return res, runErr +} + +func sweepDuplicates( + ctx context.Context, q *dbq.Queries, sweepID pgtype.UUID, pageSize int32, +) (DuplicateSweepResult, error) { + var res DuplicateSweepResult + + // Exact tier, library-wide, in one query. + exactRows, err := q.ListExactDuplicateHashes(ctx, fingerprintVersion) + if err != nil { + return res, fmt.Errorf("list exact duplicates: %w", err) + } + exactMembers := make([][]string, len(exactRows)) + exactOf := map[string]int{} + for i, row := range exactRows { + exactMembers[i] = formatUUIDs(row.TrackIds) + for _, id := range exactMembers[i] { + exactOf[id] = i + } + } + exactSeen := make([]bool, len(exactRows)) + + // Acoustic tier, streamed in duration order. The first member of an exact + // group the stream meets stands in for the whole group; the rest are skipped. + grouper := newStreamGrouper(defaultAcousticMaxBitErrorRate) + params := dbq.ListDuplicateCandidatesParams{ + CurrentVersion: fingerprintVersion, + // Durations are never negative, and the all-zero uuid sorts first: every + // row is after this cursor. Valid must be true, or "> NULL" matches nothing. + AfterDurationMs: -1, + AfterID: pgtype.UUID{Valid: true}, + PageLimit: pageSize, + } + for { + if err := ctx.Err(); err != nil { + return res, err + } + rows, err := q.ListDuplicateCandidates(ctx, params) + if err != nil { + return res, fmt.Errorf("list duplicate candidates: %w", err) + } + for _, row := range rows { + res.Candidates++ + id := syncpkg.FormatUUID(row.ID) + unit := &groupUnit{ids: []string{id}, durationMs: row.DurationMs, sortKey: id} + if gi, ok := exactOf[id]; ok { + if exactSeen[gi] { + continue + } + exactSeen[gi] = true + unit.ids, unit.exact = exactMembers[gi], true + } + unit.print = preparePrint(row.Chromaprint) + grouper.push(unit) + } + if int32(len(rows)) < pageSize { + break + } + last := rows[len(rows)-1] + params.AfterDurationMs, params.AfterID = last.DurationMs, last.ID + } + + found := grouper.finish() + // Exact groups none of whose members has a chromaprint never reached the + // stream. Identical bytes need no acoustic evidence. + for gi, seen := range exactSeen { + if !seen { + found.Groups = append(found.Groups, duplicateGroup{Tier: tierExact, Members: exactMembers[gi]}) + } + } + sortGroups(found.Groups) + res.Groups, res.Oversize = len(found.Groups), found.OversizeClusters + + dismissed, err := q.ListDismissedDuplicateMemberSets(ctx) + if err != nil { + return res, fmt.Errorf("list dismissed duplicate groups: %w", err) + } + dismissedSets := make([]map[string]struct{}, 0, len(dismissed)) + for _, d := range dismissed { + set := map[string]struct{}{} + for _, id := range formatUUIDs(d.TrackIds) { + set[id] = struct{}{} + } + dismissedSets = append(dismissedSets, set) + } + + for _, group := range found.Groups { + if coveredByDismissal(group.Members, dismissedSets) { + res.Suppressed++ + continue + } + up := dbq.UpsertDuplicateGroupParams{ + MemberKey: strings.Join(group.Members, ","), + Tier: string(group.Tier), + SweepID: sweepID, + } + if group.Tier == tierAcoustic { + worst := float32(group.WorstBitErrorRate) + up.WorstBitErrorRate = &worst + } + groupID, err := q.UpsertDuplicateGroup(ctx, up) + if errors.Is(err, pgx.ErrNoRows) { + // This exact member set was already dismissed or merged. + res.Suppressed++ + continue + } + if err != nil { + return res, fmt.Errorf("upsert duplicate group: %w", err) + } + for _, id := range group.Members { + var trackID pgtype.UUID + if err := trackID.Scan(id); err != nil { + return res, fmt.Errorf("parse track id %q: %w", id, err) + } + if err := q.AddDuplicateGroupMember(ctx, dbq.AddDuplicateGroupMemberParams{ + GroupID: groupID, TrackID: trackID, + }); err != nil { + return res, fmt.Errorf("add duplicate group member: %w", err) + } + } + res.Proposed++ + } + + // Only after a complete sweep: a sweep that failed partway has no basis for + // concluding that anything it did not reach has gone. + retired, err := q.DeleteStalePendingDuplicateGroups(ctx, sweepID) + if err != nil { + return res, fmt.Errorf("retire stale duplicate groups: %w", err) + } + res.Retired = int(retired) + return res, nil +} + +// coveredByDismissal reports whether every member of a proposal sat together in +// one group the operator dismissed. A subset counts: dismissing {A, B, C} said +// none of them are copies of each other, so proposing {A, B} again would be +// asking the same question twice. A superset does not count: a new copy joining +// is new evidence, and worth asking about. +func coveredByDismissal(members []string, dismissed []map[string]struct{}) bool { + for _, set := range dismissed { + covered := true + for _, id := range members { + if _, ok := set[id]; !ok { + covered = false + break + } + } + if covered { + return true + } + } + return false +} + +func formatUUIDs(ids []pgtype.UUID) []string { + out := make([]string, len(ids)) + for i, id := range ids { + out[i] = syncpkg.FormatUUID(id) + } + sort.Strings(out) + return out +} + +// TryStartDuplicateSweep starts a sweep in the background unless one is already +// running, reaping a sweep that has been in flight past +// staleDuplicateSweepThreshold. Mirrors TryStartScan. The sweep runs on ctx, so +// a caller answering an HTTP request must pass a context that outlives it. +func TryStartDuplicateSweep(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) (bool, error) { + q := dbq.New(pool) + row, err := q.GetInFlightDuplicateSweep(ctx) + switch { + case err == nil: + age := time.Since(row.StartedAt.Time) + if age <= staleDuplicateSweepThreshold { + return false, nil + } + logger.Warn("reaping stale duplicate sweep", "id", syncpkg.FormatUUID(row.ID), "age", age) + if ferr := q.FinishDuplicateSweep(ctx, dbq.FinishDuplicateSweepParams{ + ID: row.ID, ErrorMessage: "reaped (stale)", + }); ferr != nil { + return false, fmt.Errorf("reap stale duplicate sweep: %w", ferr) + } + case !errors.Is(err, pgx.ErrNoRows): + return false, fmt.Errorf("duplicate sweep in-flight check: %w", err) + } + + go func() { + if _, err := RunDuplicateSweep(ctx, pool, logger); err != nil { + logger.Warn("duplicate sweep failed", "err", err) + } + }() + return true, nil +} + +// DuplicateSweepWorker sweeps whenever fingerprints have changed. +type DuplicateSweepWorker struct { + pool *pgxpool.Pool + logger *slog.Logger + tick time.Duration +} + +// NewDuplicateSweepWorker builds a worker with the production cadence. +func NewDuplicateSweepWorker(pool *pgxpool.Pool, logger *slog.Logger) *DuplicateSweepWorker { + return &DuplicateSweepWorker{pool: pool, logger: logger, tick: duplicateSweepTick} +} + +// Run blocks until ctx is cancelled, checking once at start and then each tick. +func (w *DuplicateSweepWorker) 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 check so nothing it does can stop the next tick (rule 157). +func (w *DuplicateSweepWorker) tickOnce(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + w.logger.Error("duplicate sweep: tick panicked", "panic", r) + } + }() + due, err := duplicateSweepDue(ctx, dbq.New(w.pool)) + if err != nil { + if ctx.Err() == nil { + w.logger.Warn("duplicate sweep: due check failed", "err", err) + } + return + } + if !due { + return + } + if _, err := TryStartDuplicateSweep(ctx, w.pool, w.logger); err != nil { + w.logger.Warn("duplicate sweep: start failed", "err", err) + } +} + +// duplicateSweepDue reports whether any fingerprint was written after the latest +// sweep started. Fingerprints are the sweep's only input, so nothing else can +// change its answer; while the backfill is running this is true every tick. +func duplicateSweepDue(ctx context.Context, q *dbq.Queries) (bool, error) { + latest, err := q.GetLatestFingerprintComputedAt(ctx) + if err != nil { + return false, fmt.Errorf("latest fingerprint: %w", err) + } + if !latest.Valid { + return false, nil // nothing fingerprinted yet + } + last, err := q.GetLatestDuplicateSweep(ctx) + if errors.Is(err, pgx.ErrNoRows) { + return true, nil + } + if err != nil { + return false, fmt.Errorf("latest duplicate sweep: %w", err) + } + return latest.Time.After(last.StartedAt.Time), nil +} diff --git a/internal/library/duplicate_sweep_test.go b/internal/library/duplicate_sweep_test.go new file mode 100644 index 00000000..9f30c778 --- /dev/null +++ b/internal/library/duplicate_sweep_test.go @@ -0,0 +1,173 @@ +package library + +import ( + "bytes" + "context" + "io" + "log/slog" + "path/filepath" + "sort" + "strings" + "testing" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" +) + +func TestCoveredByDismissal(t *testing.T) { + dismissed := []map[string]struct{}{{"a": {}, "b": {}, "c": {}}} + for _, tc := range []struct { + name string + members []string + want bool + }{ + {"the same set", []string{"a", "b", "c"}, true}, + {"a subset of it", []string{"a", "b"}, true}, + // A new copy joining is new evidence: ask again. + {"a superset of it", []string{"a", "b", "c", "d"}, false}, + {"overlapping only in part", []string{"a", "d"}, false}, + {"unrelated", []string{"x", "y"}, false}, + } { + if got := coveredByDismissal(tc.members, dismissed); got != tc.want { + t.Errorf("%s: coveredByDismissal = %v, want %v", tc.name, got, tc.want) + } + } +} + +// TestDuplicateSweep_Integration pins what the sweep proposes, what it leaves out, +// and how re-sweeping treats a dismissal and a proposal that no longer holds. +func TestDuplicateSweep_Integration(t *testing.T) { + pool := newPool(t) + ctx := context.Background() + q := dbq.New(pool) + dir := t.TempDir() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + // seedTrack's own track has no fingerprint row: it must be absent from the + // report, not grouped with every other track lacking one. + _, album, artist := seedTrack(t, pool, filepath.Join(dir, "unfingerprinted.mp3")) + hash := func(b byte) []byte { return bytes.Repeat([]byte{b}, 32) } + add := func(name string, durationMs int32, sum []byte, print []int32) string { + t.Helper() + tr, err := q.UpsertTrack(ctx, dbq.UpsertTrackParams{ + Title: name, AlbumID: album.ID, ArtistID: artist.ID, DurationMs: durationMs, + FilePath: filepath.Join(dir, name+".mp3"), FileSize: 100, FileFormat: "mp3", + }) + if err != nil { + t.Fatalf("track %s: %v", name, err) + } + if err := q.UpsertTrackFingerprint(ctx, dbq.UpsertTrackFingerprintParams{ + TrackID: tr.ID, AudioStreamSha256: sum, Chromaprint: print, FingerprintVersion: fingerprintVersion, + }); err != nil { + t.Fatalf("fingerprint %s: %v", name, err) + } + return syncpkg.FormatUUID(tr.ID) + } + key := func(ids ...string) string { + sorted := append([]string(nil), ids...) + sort.Strings(sorted) + return strings.Join(sorted, ",") + } + + recording := randomPrint(200, printLen) + onAlbum := add("recording-album", 240000, hash(1), recording) + onCompilation := add("recording-compilation", 241000, hash(2), withBitNoise(recording, 0.05, 201)) + www1 := add("www-01", 215000, hash(9), randomPrint(210, printLen)) + www2 := add("www-02", 215000, hash(9), randomPrint(210, printLen)) + // Near-identical duration to the recording, different audio. + add("different-song", 240500, hash(3), randomPrint(220, printLen)) + // Identical to the album copy, but its file is gone: nothing to compare. + missing := add("missing-copy", 240000, hash(4), recording) + if _, err := pool.Exec(ctx, "UPDATE tracks SET missing_since = now() WHERE file_path LIKE '%missing-copy.mp3'"); err != nil { + t.Fatalf("mark missing: %v", err) + } + + type stored struct { + tier, status string + } + groups := func() map[string]stored { + t.Helper() + rows, err := pool.Query(ctx, `SELECT member_key, tier, status FROM duplicate_groups`) + if err != nil { + t.Fatalf("read groups: %v", err) + } + defer rows.Close() + out := map[string]stored{} + for rows.Next() { + var k string + var s stored + if err := rows.Scan(&k, &s.tier, &s.status); err != nil { + t.Fatalf("scan group: %v", err) + } + out[k] = s + } + return out + } + + // 1. A page size of one forces the keyset cursor across every candidate. + res, err := runDuplicateSweep(ctx, pool, logger, 1) + if err != nil { + t.Fatalf("first sweep: %v", err) + } + // Five tracks carry a chromaprint and a present file. + if res.Candidates != 5 || res.Groups != 2 || res.Proposed != 2 { + t.Fatalf("first sweep = %+v, want 5 candidates, 2 groups, 2 proposed", res) + } + acousticKey, exactKey := key(onAlbum, onCompilation), key(www1, www2) + got := groups() + want := map[string]stored{ + acousticKey: {"acoustic", "pending"}, + exactKey: {"exact", "pending"}, + } + if len(got) != len(want) || got[acousticKey] != want[acousticKey] || got[exactKey] != want[exactKey] { + t.Fatalf("groups = %+v, want %+v", got, want) + } + for k := range got { + if strings.Contains(k, missing) { + t.Fatalf("a missing track was proposed: %s", k) + } + } + + // 2. A dismissed group is not proposed again, and the pending one is + // refreshed in place rather than duplicated. + if _, err := pool.Exec(ctx, "UPDATE duplicate_groups SET status = 'dismissed' WHERE member_key = $1", acousticKey); err != nil { + t.Fatalf("dismiss: %v", err) + } + res, err = runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage) + if err != nil { + t.Fatalf("second sweep: %v", err) + } + if res.Proposed != 1 || res.Suppressed != 1 { + t.Fatalf("second sweep = %+v, want 1 proposed, 1 suppressed", res) + } + got = groups() + if len(got) != 2 || got[acousticKey].status != "dismissed" || got[exactKey].status != "pending" { + t.Fatalf("after dismissal groups = %+v, want the dismissal kept and one pending group", got) + } + + // 3. A proposal that no longer holds is retired; the dismissal survives it. + if _, err := pool.Exec(ctx, + "DELETE FROM track_fingerprints f USING tracks t WHERE f.track_id = t.id AND t.file_path LIKE '%www-02.mp3'"); err != nil { + t.Fatalf("drop fingerprint: %v", err) + } + res, err = runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage) + if err != nil { + t.Fatalf("third sweep: %v", err) + } + if res.Retired != 1 { + t.Fatalf("third sweep = %+v, want 1 retired", res) + } + got = groups() + if len(got) != 1 || got[acousticKey].status != "dismissed" { + t.Fatalf("after retiring groups = %+v, want only the dismissal", got) + } + + // 4. The sweep record reflects the last run. + last, err := q.GetLatestDuplicateSweep(ctx) + if err != nil { + t.Fatalf("latest sweep: %v", err) + } + if !last.FinishedAt.Valid || last.ErrorMessage != nil { + t.Fatalf("latest sweep = %+v, want finished without error", last) + } +}