package playevents import ( "context" "encoding/json" "io" "log/slog" "os" "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" ) func testPool(t *testing.T) *pgxpool.Pool { t.Helper() if testing.Short() { t.Skip("skipping integration test 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 } type fixture struct { pool *pgxpool.Pool q *dbq.Queries w *Writer user pgtype.UUID track pgtype.UUID } func newFixture(t *testing.T, durationMs int32) fixture { t.Helper() pool := testPool(t) q := dbq.New(pool) u, err := q.CreateUser(context.Background(), dbq.CreateUserParams{ Username: dbtest.TestUserPrefix + "tester", PasswordHash: "x", ApiToken: "x", IsAdmin: false, }) if err != nil { t.Fatalf("user: %v", err) } a, err := q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{Name: "Artist", SortName: "Artist"}) if err != nil { t.Fatalf("artist: %v", err) } al, err := q.UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{Title: "Album", SortTitle: "Album", ArtistID: a.ID}) if err != nil { t.Fatalf("album: %v", err) } tr, err := q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ Title: "Track", AlbumID: al.ID, ArtistID: a.ID, FilePath: "/tmp/track-fixture.flac", DurationMs: durationMs, }) if err != nil { t.Fatalf("track: %v", err) } return fixture{ pool: pool, q: q, w: NewWriter(pool, slog.New(slog.NewTextHandler(io.Discard, nil)), 30*time.Minute, 0.5, 30000), user: u.ID, track: tr.ID, } } func TestRecordPlayStarted_InsertsRow(t *testing.T) { f := newFixture(t, 200_000) now := time.Now().UTC() res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now) if err != nil { t.Fatalf("RecordPlayStarted: %v", err) } if !res.PlayEventID.Valid { t.Fatalf("play_event_id invalid") } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.EndedAt.Valid { t.Errorf("ended_at should be NULL right after start") } if got.WasSkipped { t.Errorf("was_skipped should default false") } } func TestRecordPlayStarted_AutoClosesPriorOpenRow(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC() first, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) if err != nil { t.Fatalf("first: %v", err) } t2 := t1.Add(45 * time.Second) _, err = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) if err != nil { t.Fatalf("second: %v", err) } prior, err := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if !prior.EndedAt.Valid { t.Errorf("prior should be closed") } // 45s elapsed on a 200s track clears the 30s duration threshold, so the // skip rule's AND fails -> NOT a skip. An orphan that played long enough // to count must still land in history rather than being force-hidden. if prior.WasSkipped { t.Errorf("auto-close at 45s should apply the skip rule -> was_skipped=false") } if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 45_000 { t.Errorf("duration_played_ms = %v, want 45000", prior.DurationPlayedMs) } } func TestRecordPlayStarted_AutoCloseShortPlayIsSkip(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC() first, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // 5s elapsed: ratio 0.025 < 0.5 AND 5000 < 30000 -> both fail -> a skip. t2 := t1.Add(5 * time.Second) if _, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2); err != nil { t.Fatalf("second: %v", err) } prior, _ := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) if !prior.WasSkipped { t.Errorf("auto-close at 5s should be was_skipped=true") } } func TestRecordPlayStarted_AutoCloseCapsAtTrackDuration(t *testing.T) { f := newFixture(t, 60_000) // 60s track t1 := time.Now().UTC() first, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // Start a second play 5 minutes later — elapsed > track duration. t2 := t1.Add(5 * time.Minute) _, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) prior, _ := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 60_000 { t.Errorf("duration_played_ms = %v, want capped at 60000", prior.DurationPlayedMs) } } func TestRecordPlayEnded_AppliesSkipRule_BothFail_NotSkip(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC() res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // 15% completion (30_000ms), 30s duration -> 30s satisfies "duration >= 30s" // so the rule's AND fails -> NOT a skip. t2 := t1.Add(30 * time.Second) if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 30_000, t2); err != nil { t.Fatalf("RecordPlayEnded: %v", err) } got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if got.WasSkipped { t.Errorf("was_skipped = true, want false (duration_played_ms == 30000 fails the AND)") } } func TestRecordPlayEnded_AppliesSkipRule_BothPass_IsSkip(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC() res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // 5% completion (10_000ms), 10s duration -> both conditions hold -> SKIP. t2 := t1.Add(10 * time.Second) if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 10_000, t2); err != nil { t.Fatalf("RecordPlayEnded: %v", err) } got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if !got.WasSkipped { t.Errorf("expected was_skipped=true") } rows, err := f.pool.Query(context.Background(), "SELECT id FROM skip_events WHERE user_id=$1 AND track_id=$2", f.user, f.track) if err != nil { t.Fatalf("query skips: %v", err) } defer rows.Close() count := 0 for rows.Next() { count++ } if count != 1 { t.Errorf("skip_events count = %d, want 1", count) } } func TestRecordPlayEnded_DurationOver50Percent_NotSkip(t *testing.T) { f := newFixture(t, 200_000) // 200s track t1 := time.Now().UTC() res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // 50% completion (100_000ms), 5s duration. Completion fails the "< 0.5" // check — so AND fails, NOT a skip. t2 := t1.Add(5 * time.Second) _ = f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 100_000, t2) got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if got.WasSkipped { t.Errorf("expected was_skipped=false at completion >= 0.5") } } // A lost-response retry of an offline play (same user/track/started_at) // must not create a second history row. Guards the at-least-once replay // path added in the 2026-06-11 audit. func TestRecordOfflinePlay_DedupsByUserTrackStartedAt(t *testing.T) { f := newFixture(t, 200_000) ctx := context.Background() at := time.Now().UTC() if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil { t.Fatalf("first offline: %v", err) } if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil { t.Fatalf("replay offline: %v", err) } var count int if err := f.pool.QueryRow(ctx, "SELECT COUNT(*) FROM play_events WHERE user_id=$1 AND track_id=$2 AND started_at=$3", f.user, f.track, pgtype.Timestamptz{Time: at, Valid: true}).Scan(&count); err != nil { t.Fatalf("count: %v", err) } if count != 1 { t.Errorf("play_events count = %d, want 1 (deduped)", count) } } // Re-closing an already-ended row (the client's durable close-by-id replay) // updates the row but must NOT insert a second skip_events row. func TestRecordPlayEnded_IdempotentClose_NoDuplicateSkipEvent(t *testing.T) { f := newFixture(t, 200_000) ctx := context.Background() t1 := time.Now().UTC() res, _ := f.w.RecordPlayStarted(ctx, f.user, f.track, "c", t1) t2 := t1.Add(10 * time.Second) // 5% / 10s -> skip if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil { t.Fatalf("first end: %v", err) } if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil { t.Fatalf("replay end: %v", err) } var count int if err := f.pool.QueryRow(ctx, "SELECT COUNT(*) FROM skip_events WHERE user_id=$1 AND track_id=$2", f.user, f.track).Scan(&count); err != nil { t.Fatalf("count: %v", err) } if count != 1 { t.Errorf("skip_events count = %d, want 1 (idempotent close)", count) } } func TestRecordPlaySkipped_AlwaysSkipFlagAndSkipEventRow(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC() res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) // User-initiated skip at 90% — overrides the rule, always treated as skip. t2 := t1.Add(180 * time.Second) if err := f.w.RecordPlaySkipped(context.Background(), res.PlayEventID, 180_000, t2); err != nil { t.Fatalf("RecordPlaySkipped: %v", err) } got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if !got.WasSkipped { t.Errorf("RecordPlaySkipped must always set was_skipped=true") } } func TestRecordSyntheticCompletedPlay_WritesBothRows(t *testing.T) { f := newFixture(t, 200_000) now := time.Now().UTC() if err := f.w.RecordSyntheticCompletedPlay(context.Background(), f.user, f.track, "feishin", now); err != nil { t.Fatalf("synthetic: %v", err) } rows, _ := f.pool.Query(context.Background(), "SELECT id FROM play_events WHERE user_id=$1 AND ended_at IS NOT NULL", f.user) defer rows.Close() count := 0 for rows.Next() { count++ } if count != 1 { t.Errorf("closed play_events count = %d, want 1", count) } } func TestRecordPlayStarted_PersistsSessionVector_Seed(t *testing.T) { f := newFixture(t, 200_000) now := time.Now().UTC() res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now) if err != nil { t.Fatalf("RecordPlayStarted: %v", err) } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.SessionVectorAtPlay == nil { t.Fatalf("session_vector_at_play is NULL") } var vec map[string]any if err := json.Unmarshal(got.SessionVectorAtPlay, &vec); err != nil { t.Fatalf("unmarshal: %v", err) } if seed, _ := vec["seed"].(bool); !seed { t.Errorf("seed = %v, want true (first play in session)", vec["seed"]) } if artists, _ := vec["artists"].([]any); len(artists) != 0 { t.Errorf("artists not empty: %v", artists) } } func TestRecordPlayStarted_PersistsSessionVector_Populated(t *testing.T) { f := newFixture(t, 200_000) // Seed 4 prior plays with the same track (reusing the fixture's single // track is fine — vector dedup means we'll see 1 artist regardless). t0 := time.Now().UTC().Add(-1 * time.Hour) for i := 0; i < 4; i++ { at := t0.Add(time.Duration(i) * time.Minute) _, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", at) } // 5th play — should see vector with 4 prior tracks. at := t0.Add(10 * time.Minute) res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", at) if err != nil { t.Fatalf("RecordPlayStarted: %v", err) } got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if got.SessionVectorAtPlay == nil { t.Fatal("session_vector_at_play is NULL") } var vec map[string]any _ = json.Unmarshal(got.SessionVectorAtPlay, &vec) if seed, _ := vec["seed"].(bool); seed { t.Errorf("seed = true after 4 prior tracks, want false") } recentIDs, _ := vec["recent_track_ids"].([]any) if len(recentIDs) != 4 { t.Errorf("recent_track_ids len = %d, want 4", len(recentIDs)) } } func TestRecordPlayStarted_VectorScopedToSession(t *testing.T) { f := newFixture(t, 200_000) // Play 1: at t=0. t0 := time.Now().UTC().Add(-2 * time.Hour) _, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t0) // Play 2: t = t0 + 31 minutes (exceeds 30-min session timeout). t1 := t0.Add(31 * time.Minute) res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) if err != nil { t.Fatalf("RecordPlayStarted: %v", err) } got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if got.SessionVectorAtPlay == nil { t.Fatal("vector NULL") } var vec map[string]any _ = json.Unmarshal(got.SessionVectorAtPlay, &vec) if seed, _ := vec["seed"].(bool); !seed { t.Errorf("seed = false in fresh session, want true") } recentIDs, _ := vec["recent_track_ids"].([]any) if len(recentIDs) != 0 { t.Errorf("recent_track_ids has %d entries from prior session, want 0", len(recentIDs)) } } func TestRecordPlayEnded_QualifyingPlay_EnqueuesScrobble(t *testing.T) { f := newFixture(t, 300_000) // 300s track ctx := context.Background() // Enable LB for the user. tk := "tk" if err := f.q.SetListenBrainzToken(ctx, dbq.SetListenBrainzTokenParams{ID: f.user, ListenbrainzToken: &tk}); err != nil { t.Fatalf("token: %v", err) } if err := f.q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{ID: f.user, ListenbrainzEnabled: true}); err != nil { t.Fatalf("enable: %v", err) } res, err := f.w.RecordPlayStarted(ctx, f.user, f.track, "test", time.Now().Add(-1*time.Minute)) if err != nil { t.Fatalf("start: %v", err) } // 250s played out of 300s = ~83% — well above LB threshold. if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 250_000, time.Now()); err != nil { t.Fatalf("end: %v", err) } var n int if err := f.pool.QueryRow(ctx, `SELECT count(*) FROM scrobble_queue WHERE play_event_id = $1`, res.PlayEventID).Scan(&n); err != nil { t.Fatalf("count: %v", err) } if n != 1 { t.Errorf("scrobble_queue rows = %d, want 1", n) } } func TestRecordPlayEnded_SubThreshold_NoEnqueue(t *testing.T) { f := newFixture(t, 300_000) ctx := context.Background() tk := "tk" _ = f.q.SetListenBrainzToken(ctx, dbq.SetListenBrainzTokenParams{ID: f.user, ListenbrainzToken: &tk}) _ = f.q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{ID: f.user, ListenbrainzEnabled: true}) res, err := f.w.RecordPlayStarted(ctx, f.user, f.track, "test", time.Now().Add(-1*time.Minute)) if err != nil { t.Fatalf("start: %v", err) } // 60s played, 20% — well below threshold. if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 60_000, time.Now()); err != nil { t.Fatalf("end: %v", err) } var n int _ = f.pool.QueryRow(ctx, `SELECT count(*) FROM scrobble_queue WHERE play_event_id = $1`, res.PlayEventID).Scan(&n) if n != 0 { t.Errorf("expected 0 queue rows for sub-threshold play, got %d", n) } } func TestRecordPlayStartedWithSource_AppendsRotation(t *testing.T) { f := newFixture(t, 200_000) now := time.Now().UTC() if _, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "for_you", now, ); err != nil { t.Fatalf("RecordPlayStartedWithSource: %v", err) } st, err := f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ UserID: f.user, PlaylistKind: "for_you", }) if err != nil { t.Fatalf("GetRotationState: %v", err) } if len(st.PlayedTrackIds) != 1 || st.PlayedTrackIds[0] != f.track { t.Errorf("played_track_ids = %v, want [%v]", st.PlayedTrackIds, f.track) } // Re-playing the same track stays a set (no duplicate append). if _, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "for_you", now.Add(time.Minute), ); err != nil { t.Fatalf("RecordPlayStartedWithSource 2: %v", err) } st, err = f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ UserID: f.user, PlaylistKind: "for_you", }) if err != nil { t.Fatalf("GetRotationState 2: %v", err) } if len(st.PlayedTrackIds) != 1 { t.Errorf("played_track_ids should dedupe, got %v", st.PlayedTrackIds) } } func TestRecordPlayStarted_NoRotationWithoutSource(t *testing.T) { f := newFixture(t, 200_000) now := time.Now().UTC() if _, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now); err != nil { t.Fatalf("RecordPlayStarted: %v", err) } _, err := f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ UserID: f.user, PlaylistKind: "for_you", }) if err == nil { t.Errorf("expected no rotation row for a source-less play") } } // seedSystemSnapshot creates a system playlist of the given variant // containing the fixture track with the given pick_kind, mimicking // what the builder persists (#1249, generalized #1270). func seedSystemSnapshot(t *testing.T, f fixture, variant, pickKind string) { t.Helper() pl, err := f.q.CreateSystemPlaylist(context.Background(), dbq.CreateSystemPlaylistParams{ UserID: f.user, Name: variant, SystemVariant: &variant, }) if err != nil { t.Fatalf("CreateSystemPlaylist: %v", err) } if _, err := f.q.AppendPlaylistTrack(context.Background(), dbq.AppendPlaylistTrackParams{ PlaylistID: pl.ID, TrackID: f.track, PickKind: &pickKind, }); err != nil { t.Fatalf("AppendPlaylistTrack: %v", err) } } func TestRecordPlayStartedWithSource_StampsForYouPickKind(t *testing.T) { f := newFixture(t, 200_000) seedSystemSnapshot(t, f, "for_you", "fresh") res, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "for_you", time.Now().UTC(), ) if err != nil { t.Fatalf("RecordPlayStartedWithSource: %v", err) } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.PickKind == nil || *got.PickKind != "fresh" { t.Errorf("pick_kind = %v, want fresh (frozen from live snapshot)", got.PickKind) } } func TestRecordPlayStartedWithSource_NoSnapshotMatch_PickKindNull(t *testing.T) { // A for_you play whose track isn't in the current snapshot (rotated // out, or no snapshot exists) stays unattributed rather than guessing. f := newFixture(t, 200_000) res, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "for_you", time.Now().UTC(), ) if err != nil { t.Fatalf("RecordPlayStartedWithSource: %v", err) } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.PickKind != nil { t.Errorf("pick_kind = %q, want NULL (track not in snapshot)", *got.PickKind) } } func TestRecordPlayStartedWithSource_CrossVariant_PickKindNull(t *testing.T) { // Attribution is variant-scoped: even with a For You snapshot // containing the track, a play from a different surface must not // pick up For You's stamp — it reads its OWN variant's snapshot, // which here doesn't exist. f := newFixture(t, 200_000) seedSystemSnapshot(t, f, "for_you", "taste") res, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "discover", time.Now().UTC(), ) if err != nil { t.Fatalf("RecordPlayStartedWithSource: %v", err) } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.PickKind != nil { t.Errorf("pick_kind = %q, want NULL for a cross-variant play", *got.PickKind) } } func TestRecordPlayStartedWithSource_StampsDiscoverBucket(t *testing.T) { // Discover stamps its candidate bucket (#1270); a discover play whose // track is in the live snapshot freezes that bucket onto the play. f := newFixture(t, 200_000) seedSystemSnapshot(t, f, "discover", "dormant") res, err := f.w.RecordPlayStartedWithSource( context.Background(), f.user, f.track, "c", "discover", time.Now().UTC(), ) if err != nil { t.Fatalf("RecordPlayStartedWithSource: %v", err) } got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) if err != nil { t.Fatalf("get: %v", err) } if got.PickKind == nil || *got.PickKind != "dormant" { t.Errorf("pick_kind = %v, want dormant (frozen from live snapshot)", got.PickKind) } } func TestRecordOfflinePlay_StampsForYouPickKind(t *testing.T) { // The offline replay path threads source the same way; a replayed // for_you play whose track is still in the snapshot gets attributed. f := newFixture(t, 200_000) seedSystemSnapshot(t, f, "for_you", "taste") at := time.Now().UTC().Add(-time.Hour) if err := f.w.RecordOfflinePlay( context.Background(), f.user, f.track, "c", "for_you", at, 180_000, ); err != nil { t.Fatalf("RecordOfflinePlay: %v", err) } var pickKind *string if err := f.pool.QueryRow(context.Background(), `SELECT pick_kind FROM play_events WHERE user_id = $1 AND track_id = $2`, f.user, f.track, ).Scan(&pickKind); err != nil { t.Fatalf("select: %v", err) } if pickKind == nil || *pickKind != "taste" { t.Errorf("pick_kind = %v, want taste", pickKind) } }