feat(mixes): tiered rebuilds for New for you + First listens (rule #131)

Both mixes move from a single hard eligibility rule to the tiered
ladder, with their tier stamped onto playlist_tracks.pick_kind via the
#1270 provenance pipeline.

New for you (#1267) — consume on play, degrade by stepping back:
- "Consumed" = any track attempted >=30s; played albums leave the mix
  at the next build instead of crowding it until the calendar window
  expires.
- Tier 1: unconsumed albums added <30d by direct-affinity artists.
  Tier 2: unconsumed affinity albums from the wider 30-90d window —
  added while you weren't looking. Tier 3: any unconsumed album added
  <90d, newest first.

First listens (#1268) — track-level "attempted" threshold:
- A 2-second accidental brush no longer disqualifies a whole album;
  "attempted" is duration_played_ms >= 30000 per track.
- Tier 1: albums with zero attempted tracks. Tier 2: barely-attempted
  albums (<=25% of tracks reached 30s), minus the attempted tracks
  themselves. The artist-affinity ordering signal also moves to the
  >=30s definition so skip-only contact doesn't read as trust.

Producer plumbing: fetch adapters map the tier column onto pick kinds,
finishMix propagates PickKind into the persisted candidates, and
rotateForDay now rotates within contiguous same-pick-kind blocks so
daily rotation can't hoist tier-3 filler above tier-1's exact fits
(untiered pools are one block — original behavior).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TsF3cNoKrqCYsU78cXC8U6

internal/playlists/system_mixes_db_test.go

@@ -57,6 +57,54 @@ func idSet[T any](rows []T, id func(T) pgtype.UUID) map[pgtype.UUID]bool {
 	return out
 }
 
+// seedPlayWithDuration inserts a completed play with an explicit
+// duration_played_ms — the "attempted >= 30s" queries key off it.
+func seedPlayWithDuration(t *testing.T, pool *pgxpool.Pool, userID, trackID pgtype.UUID, startedAt time.Time, durationMs int32) {
+	t.Helper()
+	_, err := pool.Exec(context.Background(), `
+		WITH s AS (
+			INSERT INTO play_sessions (user_id, started_at, last_event_at)
+			VALUES ($1, $3, $3)
+			RETURNING id
+		)
+		INSERT INTO play_events (user_id, track_id, session_id, started_at, duration_played_ms)
+		SELECT $1, $2, s.id, $3, $4 FROM s
+	`, userID, trackID, startedAt, durationMs)
+	if err != nil {
+		t.Fatalf("seed play with duration: %v", err)
+	}
+}
+
+// seedAlbumForArtist creates a fresh album under an existing artist —
+// the tiered album queries need several albums sharing one artist_id.
+func seedAlbumForArtist(t *testing.T, pool *pgxpool.Pool, title string, artistID pgtype.UUID) pgtype.UUID {
+	t.Helper()
+	al, err := dbq.New(pool).UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{
+		Title: title, SortTitle: title, ArtistID: artistID,
+	})
+	if err != nil {
+		t.Fatalf("seed album for artist: %v", err)
+	}
+	return al.ID
+}
+
+func backdateAlbum(t *testing.T, pool *pgxpool.Pool, albumID pgtype.UUID, to time.Time) {
+	t.Helper()
+	if _, err := pool.Exec(context.Background(),
+		`UPDATE albums SET created_at = $2 WHERE id = $1`, albumID, to); err != nil {
+		t.Fatalf("backdate album: %v", err)
+	}
+}
+
+func likeArtist(t *testing.T, pool *pgxpool.Pool, userID, artistID pgtype.UUID) {
+	t.Helper()
+	if _, err := pool.Exec(context.Background(),
+		`INSERT INTO general_likes_artists (user_id, artist_id) VALUES ($1, $2)`,
+		userID, artistID); err != nil {
+		t.Fatalf("like artist: %v", err)
+	}
+}
+
 func TestListDeepCutsTracks_SkippedTracksExcluded(t *testing.T) {
 	// #1257: eligibility used to count only unskipped plays, so a track
 	// skipped repeatedly (and never finished) read as "barely heard".
@@ -133,6 +181,121 @@ func TestListOnThisDayTracks_WrapsYearBoundary(t *testing.T) {
 	}
 }
 
+func TestListNewForYouTracks_TiersAndConsumption(t *testing.T) {
+	// #1267: tiered eligibility with consume-on-play. Tier 1 = fresh
+	// affinity albums; tier 2 = affinity albums from the wider 90-day
+	// window; tier 3 = fresh additions without affinity; any attempted
+	// (>=30s) track consumes the album out of the mix entirely.
+	pool := newPool(t)
+	q := dbq.New(pool)
+	u := seedUser(t, pool, "nfy1")
+	now := time.Now().UTC()
+
+	// Affinity artist via a like; anchor track establishes the artist.
+	anchor := seedTrack(t, pool, "nfy1-anchor", "nfy1-artistX")
+	likeArtist(t, pool, u.ID, anchor.ArtistID)
+	backdateAlbum(t, pool, anchor.AlbumID, now.Add(-200*24*time.Hour)) // out of every window
+
+	// Fresh affinity album → tier 1.
+	freshAlbum := seedAlbumForArtist(t, pool, "nfy1-fresh", anchor.ArtistID)
+	freshTrack := seedTrackForArtist(t, pool, "nfy1-fresh-1", freshAlbum, anchor.ArtistID)
+	// Affinity album added 60 days ago → tier 2.
+	olderAlbum := seedAlbumForArtist(t, pool, "nfy1-older", anchor.ArtistID)
+	olderTrack := seedTrackForArtist(t, pool, "nfy1-older-1", olderAlbum, anchor.ArtistID)
+	backdateAlbum(t, pool, olderAlbum, now.Add(-60*24*time.Hour))
+	// Fresh album by an unrelated artist → tier 3.
+	otherTrack := seedTrack(t, pool, "nfy1-other-1", "nfy1-artistY")
+	// Fresh affinity album with one attempted track → consumed, absent.
+	consumedAlbum := seedAlbumForArtist(t, pool, "nfy1-consumed", anchor.ArtistID)
+	consumedTrack := seedTrackForArtist(t, pool, "nfy1-consumed-1", consumedAlbum, anchor.ArtistID)
+	seedPlayWithDuration(t, pool, u.ID, consumedTrack.ID, now.Add(-time.Hour), 35_000)
+	// Affinity album added 120 days ago → outside every window, absent.
+	ancientAlbum := seedAlbumForArtist(t, pool, "nfy1-ancient", anchor.ArtistID)
+	ancientTrack := seedTrackForArtist(t, pool, "nfy1-ancient-1", ancientAlbum, anchor.ArtistID)
+	backdateAlbum(t, pool, ancientAlbum, now.Add(-120*24*time.Hour))
+
+	rows, err := q.ListNewForYouTracks(context.Background(), u.ID)
+	if err != nil {
+		t.Fatalf("ListNewForYouTracks: %v", err)
+	}
+	tiers := map[pgtype.UUID]int32{}
+	for _, r := range rows {
+		tiers[r.ID] = r.Tier
+	}
+	if got := tiers[freshTrack.ID]; got != 1 {
+		t.Errorf("fresh affinity album tier = %d, want 1", got)
+	}
+	if got := tiers[olderTrack.ID]; got != 2 {
+		t.Errorf("60-day affinity album tier = %d, want 2", got)
+	}
+	if got := tiers[otherTrack.ID]; got != 3 {
+		t.Errorf("fresh non-affinity album tier = %d, want 3", got)
+	}
+	if _, in := tiers[consumedTrack.ID]; in {
+		t.Error("album with an attempted track must be consumed out of the mix")
+	}
+	if _, in := tiers[ancientTrack.ID]; in {
+		t.Error("album added 120 days ago must be outside every window")
+	}
+}
+
+func TestListFirstListensTracks_AttemptThreshold(t *testing.T) {
+	// #1268: "attempted" is track-level at >=30s. A 2-second brush no
+	// longer disqualifies an album; barely-attempted albums step back
+	// to tier 2 with the attempted tracks themselves excluded.
+	pool := newPool(t)
+	q := dbq.New(pool)
+	u := seedUser(t, pool, "fl1")
+	now := time.Now().UTC()
+
+	// Never touched → tier 1.
+	untouched := seedTrack(t, pool, "fl1-untouched-1", "fl1-artistA")
+	// One 2-second accidental brush → still tier 1 (the headline bug).
+	brushed := seedTrack(t, pool, "fl1-brushed-1", "fl1-artistB")
+	seedPlayWithDuration(t, pool, u.ID, brushed.ID, now.Add(-time.Hour), 2_000)
+	// 4-track album with one attempted (25%) → tier 2, attempted track
+	// excluded, siblings present.
+	barely1 := seedTrack(t, pool, "fl1-barely-1", "fl1-artistC")
+	barely2 := seedTrackForArtist(t, pool, "fl1-barely-2", barely1.AlbumID, barely1.ArtistID)
+	barely3 := seedTrackForArtist(t, pool, "fl1-barely-3", barely1.AlbumID, barely1.ArtistID)
+	barely4 := seedTrackForArtist(t, pool, "fl1-barely-4", barely1.AlbumID, barely1.ArtistID)
+	seedPlayWithDuration(t, pool, u.ID, barely1.ID, now.Add(-time.Hour), 40_000)
+	// 2-track album with one attempted (50%) → past the brush
+	// threshold, absent entirely.
+	explored1 := seedTrack(t, pool, "fl1-explored-1", "fl1-artistD")
+	explored2 := seedTrackForArtist(t, pool, "fl1-explored-2", explored1.AlbumID, explored1.ArtistID)
+	seedPlayWithDuration(t, pool, u.ID, explored1.ID, now.Add(-time.Hour), 40_000)
+
+	rows, err := q.ListFirstListensTracks(context.Background(), u.ID)
+	if err != nil {
+		t.Fatalf("ListFirstListensTracks: %v", err)
+	}
+	tiers := map[pgtype.UUID]int32{}
+	for _, r := range rows {
+		tiers[r.ID] = r.Tier
+	}
+	if got := tiers[untouched.ID]; got != 1 {
+		t.Errorf("untouched album tier = %d, want 1", got)
+	}
+	if got := tiers[brushed.ID]; got != 1 {
+		t.Errorf("2-second brush must not disqualify: tier = %d, want 1", got)
+	}
+	if _, in := tiers[barely1.ID]; in {
+		t.Error("the attempted track itself is not a first listen")
+	}
+	for _, tr := range []dbq.Track{barely2, barely3, barely4} {
+		if got := tiers[tr.ID]; got != 2 {
+			t.Errorf("sibling of attempted track tier = %d, want 2", got)
+		}
+	}
+	if _, in := tiers[explored1.ID]; in {
+		t.Error("half-attempted album must be excluded (attempted track)")
+	}
+	if _, in := tiers[explored2.ID]; in {
+		t.Error("half-attempted album must be excluded (sibling)")
+	}
+}
+
 func TestListRediscoverTracks_SkippedResurfacingCoolsDown(t *testing.T) {
 	// #1258: a skip on a rediscover-sourced play is a declined
 	// invitation — the track sits out ~90 days instead of re-qualifying