minstrel/internal/playevents/writer_test.go

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")
	}
}

// seedForYouSnapshot creates a system For You playlist containing the
// fixture track with the given pick_kind, mimicking what the builder
// persists (#1249).
func seedForYouSnapshot(t *testing.T, f fixture, pickKind string) {
	t.Helper()
	variant := "for_you"
	pl, err := f.q.CreateSystemPlaylist(context.Background(), dbq.CreateSystemPlaylistParams{
		UserID: f.user, Name: "For You", 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)
	seedForYouSnapshot(t, f, "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_NonForYouSource_PickKindNull(t *testing.T) {
	// Even with a For You snapshot containing the track, a play from a
	// different surface must not pick up For You attribution.
	f := newFixture(t, 200_000)
	seedForYouSnapshot(t, f, "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 non-for_you source", *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)
	seedForYouSnapshot(t, f, "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)
	}
}