package similarity import ( "context" "encoding/json" "fmt" "io" "log/slog" "net/http" "net/http/httptest" "os" "strings" "sync" "sync/atomic" "testing" "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/db" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/dbtest" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" ) func testPool(t *testing.T) (*pgxpool.Pool, *dbq.Queries) { t.Helper() if testing.Short() { t.Skip("skipping in -short mode") } dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") if dsn == "" { t.Skip("MINSTREL_TEST_DATABASE_URL not set") } if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { t.Fatalf("migrate: %v", err) } pool, err := pgxpool.New(context.Background(), dsn) if err != nil { t.Fatalf("pool: %v", err) } t.Cleanup(pool.Close) dbtest.ResetDB(t, pool) return pool, dbq.New(pool) } type fixture struct { pool *pgxpool.Pool q *dbq.Queries user pgtype.UUID artist dbq.Artist album dbq.Album } func newFixture(t *testing.T) fixture { t.Helper() pool, q := testPool(t) ctx := context.Background() u, err := q.CreateUser(ctx, dbq.CreateUserParams{ Username: dbtest.TestUserPrefix + "tester", PasswordHash: "x", ApiTokenHash: "x", IsAdmin: false, }) if err != nil { t.Fatalf("user: %v", err) } artistMbid := "aaaaaaaa-1111-1111-1111-111111111111" a, err := q.UpsertArtist(ctx, dbq.UpsertArtistParams{ Name: "X", SortName: "X", Mbid: &artistMbid, }) if err != nil { t.Fatalf("artist: %v", err) } al, err := q.UpsertAlbum(ctx, dbq.UpsertAlbumParams{ Title: "X", SortTitle: "X", ArtistID: a.ID, }) if err != nil { t.Fatalf("album: %v", err) } return fixture{pool: pool, q: q, user: u.ID, artist: a, album: al} } func seedTrack(t *testing.T, f fixture, title string, mbid *string) dbq.Track { t.Helper() tr, err := f.q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ Title: title, AlbumID: f.album.ID, ArtistID: f.artist.ID, FilePath: "/tmp/" + title + ".flac", DurationMs: 200_000, Mbid: mbid, }) if err != nil { t.Fatalf("track: %v", err) } return tr } func markPlayed(t *testing.T, f fixture, trackID pgtype.UUID) { t.Helper() var sessionID pgtype.UUID if err := f.pool.QueryRow(context.Background(), `INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id) VALUES ($1, now() - interval '5 minutes', now(), 'test') RETURNING id`, f.user).Scan(&sessionID); err != nil { t.Fatalf("session: %v", err) } if _, err := f.pool.Exec(context.Background(), `INSERT INTO play_events (user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped) VALUES ($1, $2, $3, now() - interval '1 minute', now(), 250000, 0.83, false)`, f.user, trackID, sessionID); err != nil { t.Fatalf("play_event: %v", err) } } func newTestWorker(f fixture, lbBaseURL string) *Worker { logger := slog.New(slog.NewTextHandler(io.Discard, nil)) return &Worker{ pool: f.pool, // LabsBaseURL must point at the stub too: similarity calls hit // the Labs API (commit 4fca0e6), so an unset LabsBaseURL would // fall through to the real labs.api and the stub never runs. client: &listenbrainz.Client{BaseURL: lbBaseURL, LabsBaseURL: lbBaseURL, HTTP: http.DefaultClient}, logger: logger, tick: 1 * time.Hour, batch: 5, topK: 20, } } // stubLB returns an httptest server that responds to similar-recordings and // similar-artists with the given JSON payloads. func stubLB(recordingsBody, artistsBody string, status int) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if status != 0 && status != http.StatusOK { w.WriteHeader(status) return } if strings.Contains(r.URL.Path, "/similar-recordings/") { _, _ = w.Write([]byte(recordingsBody)) return } if strings.Contains(r.URL.Path, "/similar-artists/") { _, _ = w.Write([]byte(artistsBody)) return } w.WriteHeader(http.StatusNotFound) })) } func countTrackSim(t *testing.T, f fixture, a pgtype.UUID) int { t.Helper() var n int if err := f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity WHERE track_a_id = $1`, a).Scan(&n); err != nil { t.Fatalf("count: %v", err) } return n } func countArtistSim(t *testing.T, f fixture, a pgtype.UUID) int { t.Helper() var n int if err := f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity WHERE artist_a_id = $1`, a).Scan(&n); err != nil { t.Fatalf("count: %v", err) } return n } func TestTickOnce_NoPlayedTracks_NoOp(t *testing.T) { f := newFixture(t) srv := stubLB(`[]`, `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } var n int _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity`).Scan(&n) if n != 0 { t.Errorf("track_similarity rows = %d, want 0", n) } _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity`).Scan(&n) if n != 0 { t.Errorf("artist_similarity rows = %d, want 0", n) } } func TestTickOnce_MapsLBResponseToLocalLibrary(t *testing.T) { f := newFixture(t) mbidA := "11111111-1111-1111-1111-111111111111" mbidB := "22222222-2222-2222-2222-222222222222" mbidC := "99999999-9999-9999-9999-999999999999" // NOT in library trackA := seedTrack(t, f, "A", &mbidA) _ = seedTrack(t, f, "B", &mbidB) markPlayed(t, f, trackA.ID) body := `[ {"recording_mbid": "` + mbidB + `", "score": 0.9}, {"recording_mbid": "` + mbidC + `", "score": 0.7} ]` srv := stubLB(body, `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if got := countTrackSim(t, f, trackA.ID); got != 1 { t.Errorf("track_similarity rows = %d, want 1 (only mbidB is in-library)", got) } } // The 20-per-seed cap is gone with the cache (#5296): LB's answer is at most // 50, and every recording of it that is in the library becomes an edge. func TestTickOnce_KeepsEveryInLibraryMatch(t *testing.T) { f := newFixture(t) mbidSeed := "11111111-1111-1111-1111-111111111111" seed := seedTrack(t, f, "Seed", &mbidSeed) markPlayed(t, f, seed.ID) type lbRow struct { MBID string `json:"recording_mbid"` Score float64 `json:"score"` } rows := make([]lbRow, 0, 25) for i := 0; i < 25; i++ { mbid := fmt.Sprintf("20000000-0000-0000-0000-%012d", i+1) _ = seedTrack(t, f, fmt.Sprintf("T%02d", i+1), &mbid) rows = append(rows, lbRow{MBID: mbid, Score: 1.0 - 0.01*float64(i)}) } body, _ := json.Marshal(rows) srv := stubLB(string(body), `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if got := countTrackSim(t, f, seed.ID); got != 25 { t.Errorf("got %d rows, want all 25 in-library matches", got) } } func TestTickOnce_RecentFetchIsNotRepeated(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) recordFetch(t, f, trackA.ID, "now() - interval '29 days'") srv, asked := recordingLB(func(string) string { return `[]` }) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if n := len(asked()); n != 0 { t.Errorf("a seed fetched 29 days ago was asked again (%d requests)", n) } } func TestTickOnce_RefreshesStaleRow(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) otherMbid := "55555555-5555-5555-5555-555555555555" other := seedTrack(t, f, "Other", &otherMbid) if _, err := f.pool.Exec(context.Background(), `INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) VALUES ($1, $2, 0.5, 'listenbrainz', now() - interval '31 days')`, trackA.ID, other.ID); err != nil { t.Fatalf("seed sim: %v", err) } recordFetch(t, f, trackA.ID, "now() - interval '31 days'") srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.95}]`, `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } var score float64 var fetchedAt time.Time _ = f.pool.QueryRow(context.Background(), `SELECT score, fetched_at FROM track_similarity WHERE track_a_id = $1 AND track_b_id = $2 AND source = 'listenbrainz'`, trackA.ID, other.ID).Scan(&score, &fetchedAt) if score != 0.95 { t.Errorf("score = %v, want 0.95 (refreshed)", score) } if time.Since(fetchedAt) > time.Minute { t.Errorf("fetched_at not bumped: %v ago", time.Since(fetchedAt)) } } func TestTickOnce_429AbortsTick(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Retry-After", "60") w.WriteHeader(http.StatusTooManyRequests) })) defer srv.Close() w := newTestWorker(f, srv.URL) _ = w.tickOnce(context.Background()) if got := countTrackSim(t, f, trackA.ID); got != 0 { t.Errorf("rows on 429 = %d, want 0 (tick aborted before any inserts)", got) } } func TestTickOnce_TransientErrorSkipsTrack(t *testing.T) { f := newFixture(t) mbidA := "11111111-1111-1111-1111-111111111111" mbidB := "22222222-2222-2222-2222-222222222222" otherMbid := "55555555-5555-5555-5555-555555555555" trackA := seedTrack(t, f, "A", &mbidA) trackB := seedTrack(t, f, "B", &mbidB) _ = seedTrack(t, f, "Other", &otherMbid) markPlayed(t, f, trackA.ID) markPlayed(t, f, trackB.ID) var seen atomic.Int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !strings.Contains(r.URL.Path, "/similar-recordings/") { _, _ = w.Write([]byte(`[]`)) return } n := seen.Add(1) if n == 1 { w.WriteHeader(http.StatusServiceUnavailable) return } _, _ = w.Write([]byte(`[{"recording_mbid":"` + otherMbid + `","score":0.7}]`)) })) defer srv.Close() w := newTestWorker(f, srv.URL) _ = w.tickOnce(context.Background()) a := countTrackSim(t, f, trackA.ID) b := countTrackSim(t, f, trackB.ID) if a+b != 1 { t.Errorf("expected exactly one track to succeed: a=%d b=%d", a, b) } } func TestTickOnce_FiltersInLibrary(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) body := `[ {"recording_mbid":"99999999-9999-9999-9999-999999999991","score":0.9}, {"recording_mbid":"99999999-9999-9999-9999-999999999992","score":0.8}, {"recording_mbid":"99999999-9999-9999-9999-999999999993","score":0.7} ]` srv := stubLB(body, `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if got := countTrackSim(t, f, trackA.ID); got != 0 { t.Errorf("filtered out-of-library: got %d, want 0", got) } } func TestTickOnce_ArtistPassMirrors(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) bMbid := "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" _, err := f.q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{ Name: "Other", SortName: "Other", Mbid: &bMbid, }) if err != nil { t.Fatalf("artist: %v", err) } body := `[{"artist_mbid":"` + bMbid + `","score":0.85}]` srv := stubLB(`[]`, body, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if got := countArtistSim(t, f, f.artist.ID); got != 1 { t.Errorf("artist_similarity rows = %d, want 1", got) } } func TestTickOnce_NoMBIDOnTrack_Skipped(t *testing.T) { f := newFixture(t) trackNoMbid := seedTrack(t, f, "NoMbid", nil) markPlayed(t, f, trackNoMbid.ID) srv := stubLB(`[{"recording_mbid":"11111111-1111-1111-1111-111111111111","score":0.9}]`, `[]`, http.StatusOK) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } if got := countTrackSim(t, f, trackNoMbid.ID); got != 0 { t.Errorf("no-MBID track produced rows: %d", got) } } // TestUpsertArtistSimilar_PersistsUnmatchedToTable: M5c. Calls // upsertArtistSimilar directly with a mix of in-library and out-of-library // similar-artist payloads; verifies the matched path lands in // artist_similarity (1 row), the unmatched path lands in // artist_similarity_unmatched (3 rows), and rows with empty Name are // skipped (we can't render a suggestion without a name). func TestUpsertArtistSimilar_PersistsUnmatchedToTable(t *testing.T) { f := newFixture(t) ctx := context.Background() // Seed one additional in-library artist that will be the in-library match. inLibMBID := "in-lib-mbid-123" inLibArtist, err := f.q.UpsertArtist(ctx, dbq.UpsertArtistParams{ Name: "InLib Artist", SortName: "InLib Artist", Mbid: &inLibMBID, }) if err != nil { t.Fatalf("seed in-lib artist: %v", err) } w := newTestWorker(f, "") similars := []listenbrainz.SimilarArtist{ {MBID: inLibMBID, Name: "InLib Artist", Score: 0.95}, {MBID: "out-mbid-1", Name: "Outsider One", Score: 0.85}, {MBID: "out-mbid-2", Name: "Outsider Two", Score: 0.80}, {MBID: "out-mbid-3", Name: "Outsider Three", Score: 0.70}, {MBID: "out-mbid-4", Name: "", Score: 0.60}, // empty name — skipped } w.upsertArtistSimilar(ctx, f.q, f.artist.ID, similars) // Matched path: 1 row in artist_similarity for the in-library candidate. var matchedCount int if err := f.pool.QueryRow(ctx, "SELECT count(*) FROM artist_similarity WHERE artist_a_id = $1", f.artist.ID, ).Scan(&matchedCount); err != nil { t.Fatalf("count matched: %v", err) } if matchedCount != 1 { t.Errorf("artist_similarity rows = %d, want 1 (only the in-library match)", matchedCount) } // Unmatched path: 3 rows (out-mbid-1/2/3); the empty-name row is skipped. var unmatchedCount int if err := f.pool.QueryRow(ctx, "SELECT count(*) FROM artist_similarity_unmatched WHERE seed_artist_id = $1", f.artist.ID, ).Scan(&unmatchedCount); err != nil { t.Fatalf("count unmatched: %v", err) } if unmatchedCount != 3 { t.Errorf("artist_similarity_unmatched rows = %d, want 3", unmatchedCount) } // Verify a specific row's name + score round-tripped correctly. var name string var score float64 if err := f.pool.QueryRow(ctx, "SELECT candidate_name, score FROM artist_similarity_unmatched WHERE seed_artist_id = $1 AND candidate_mbid = $2", f.artist.ID, "out-mbid-1", ).Scan(&name, &score); err != nil { t.Fatalf("fetch out-mbid-1: %v", err) } if name != "Outsider One" || score != 0.85 { t.Errorf("row = (%q, %v), want (Outsider One, 0.85)", name, score) } _ = inLibArtist } // recordFetch stamps a seed as answered by ListenBrainz at the given SQL time. func recordFetch(t *testing.T, f fixture, trackID pgtype.UUID, at string) { t.Helper() if _, err := f.pool.Exec(context.Background(), `INSERT INTO track_similarity_fetches (track_id, fetched_at, returned) VALUES ($1, `+at+`, 0)`, trackID); err != nil { t.Fatalf("record fetch: %v", err) } } // recordingLB answers similar-recordings with answer(mbid) and similar-artists // with an empty list, and reports which recording MBIDs it was asked about. func recordingLB(answer func(mbid string) string) (*httptest.Server, func() []string) { var mu sync.Mutex var asked []string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !strings.Contains(r.URL.Path, "/similar-recordings/") { _, _ = w.Write([]byte(`[]`)) return } mbid := r.URL.Query().Get("recording_mbids") mu.Lock() asked = append(asked, mbid) mu.Unlock() _, _ = w.Write([]byte(answer(mbid))) })) return srv, func() []string { mu.Lock() defer mu.Unlock() return append([]string(nil), asked...) } } // The starvation #3879 measured: a seed whose answer matched nothing in the // library wrote nothing, stayed due, and was asked again every tick ahead of // everything behind it. Recording the fetch moves the queue on. func TestTickOnce_UnmatchedAnswerDoesNotHoldTheQueue(t *testing.T) { f := newFixture(t) mbids := []string{ "11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222", "33333333-3333-3333-3333-333333333333", } for i := range mbids { tr := seedTrack(t, f, fmt.Sprintf("S%d", i), &mbids[i]) markPlayed(t, f, tr.ID) } srv, asked := recordingLB(func(string) string { return `[{"recording_mbid":"99999999-9999-9999-9999-999999999999","score":0.9}]` }) defer srv.Close() w := newTestWorker(f, srv.URL) w.batch = 1 for i := 0; i < len(mbids); i++ { if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tick %d: %v", i, err) } } seen := map[string]bool{} for _, m := range asked() { seen[m] = true } if len(seen) != len(mbids) { t.Errorf("three ticks of one asked about %d distinct seeds (%v), want all %d", len(seen), asked(), len(mbids)) } var fetched int _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity_fetches`).Scan(&fetched) if fetched != len(mbids) { t.Errorf("track_similarity_fetches = %d, want %d", fetched, len(mbids)) } } // The point of the cache (#5296): a recording LB named while it was not in the // library becomes an edge once it arrives, on the next tick's resolve, without // asking LB again. One edge per recording when the library holds two copies, // and none to a copy that is missing. func TestTickOnce_CachedAnswerResolvesWhenTheRecordingArrives(t *testing.T) { f := newFixture(t) ctx := context.Background() seedMbid := "11111111-1111-1111-1111-111111111111" laterMbid := "77777777-7777-7777-7777-777777777777" goneMbid := "88888888-8888-8888-8888-888888888888" seed := seedTrack(t, f, "Seed", &seedMbid) markPlayed(t, f, seed.ID) srv, asked := recordingLB(func(string) string { return `[{"recording_mbid":"` + laterMbid + `","score":0.9}, {"recording_mbid":"` + goneMbid + `","score":0.8}]` }) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(ctx); err != nil { t.Fatalf("tick 1: %v", err) } if got := countTrackSim(t, f, seed.ID); got != 0 { t.Fatalf("edges before the recording is in the library = %d, want 0", got) } var cached int _ = f.pool.QueryRow(ctx, `SELECT count(*) FROM listenbrainz_similar_recordings WHERE seed_track_id = $1`, seed.ID).Scan(&cached) if cached != 2 { t.Fatalf("cached answer rows = %d, want 2 (both out of library)", cached) } // The recording arrives twice (an mp3 beside its flac); the other one // arrives only as a missing row. _ = seedTrack(t, f, "Later mp3", &laterMbid) _ = seedTrack(t, f, "Later flac", &laterMbid) gone := seedTrack(t, f, "Gone", &goneMbid) if _, err := f.pool.Exec(ctx, `UPDATE tracks SET missing_since = now() WHERE id = $1`, gone.ID); err != nil { t.Fatal(err) } if err := w.tickOnce(ctx); err != nil { t.Fatalf("tick 2: %v", err) } if got := countTrackSim(t, f, seed.ID); got != 1 { t.Errorf("edges after the recording arrived = %d, want 1 (one per recording, none to a missing copy)", got) } if n := len(asked()); n != 1 { t.Errorf("LB asked %d times, want 1: the resolve needs no new fetch", n) } } // A 4xx answer is permanent for that MBID; it is recorded like an empty one // so the seed waits its turn instead of being re-asked every hour. func TestTickOnce_PermanentErrorCountsAsFetched(t *testing.T) { f := newFixture(t) mbid := "11111111-1111-1111-1111-111111111111" trackA := seedTrack(t, f, "A", &mbid) markPlayed(t, f, trackA.ID) srv := stubLB(``, ``, http.StatusBadRequest) defer srv.Close() w := newTestWorker(f, srv.URL) if err := w.tickOnce(context.Background()); err != nil { t.Fatalf("tickOnce: %v", err) } var tracks, artists int _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity_fetches`).Scan(&tracks) _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity_fetches`).Scan(&artists) if tracks != 1 || artists != 1 { t.Errorf("fetches recorded: tracks=%d artists=%d, want 1 and 1", tracks, artists) } }