minstrel/internal/similarity/worker_integration_test.go

package similarity

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"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", ApiToken: "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)
	}
}

func TestTickOnce_TopKEnforced(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 != 20 {
		t.Errorf("got %d rows, want exactly 20 (all 25 in-library; top 20 by score)", got)
	}
}

func TestTickOnce_RespectsSevenDayCap(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())`, trackA.ID, other.ID); err != nil {
		t.Fatalf("seed sim: %v", err)
	}
	beforeCount := countTrackSim(t, f, trackA.ID)
	srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.99}]`, `[]`, http.StatusOK)
	defer srv.Close()
	w := newTestWorker(f, srv.URL)
	_ = w.tickOnce(context.Background())
	if got := countTrackSim(t, f, trackA.ID); got != beforeCount {
		t.Errorf("worker re-queried fresh track: before=%d after=%d", beforeCount, got)
	}
}

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 '8 days')`, trackA.ID, other.ID); err != nil {
		t.Fatalf("seed sim: %v", err)
	}
	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
}