diff --git a/internal/db/dbq/system_playlists.sql.go b/internal/db/dbq/system_playlists.sql.go index 6fe365d8..86088ccf 100644 --- a/internal/db/dbq/system_playlists.sql.go +++ b/internal/db/dbq/system_playlists.sql.go @@ -23,6 +23,52 @@ func (q *Queries) ClearStaleSystemPlaylistInFlight(ctx context.Context) error { return err } +const createSystemPlaylist = `-- name: CreateSystemPlaylist :one +INSERT INTO playlists ( + user_id, name, description, is_public, + kind, system_variant, seed_artist_id, cover_path +) VALUES ($1, $2, '', false, 'system', $3, $4, $5) +RETURNING id, user_id, name, description, is_public, cover_path, track_count, duration_sec, created_at, updated_at, kind, system_variant, seed_artist_id +` + +type CreateSystemPlaylistParams struct { + UserID pgtype.UUID + Name string + SystemVariant *string + SeedArtistID pgtype.UUID + CoverPath *string +} + +// Inserts a system-generated playlist. Used by BuildSystemPlaylists. +// For 'for_you' variant, pass seed_artist_id as NULL (zero-value pgtype.UUID +// with Valid=false). +func (q *Queries) CreateSystemPlaylist(ctx context.Context, arg CreateSystemPlaylistParams) (Playlist, error) { + row := q.db.QueryRow(ctx, createSystemPlaylist, + arg.UserID, + arg.Name, + arg.SystemVariant, + arg.SeedArtistID, + arg.CoverPath, + ) + var i Playlist + err := row.Scan( + &i.ID, + &i.UserID, + &i.Name, + &i.Description, + &i.IsPublic, + &i.CoverPath, + &i.TrackCount, + &i.DurationSec, + &i.CreatedAt, + &i.UpdatedAt, + &i.Kind, + &i.SystemVariant, + &i.SeedArtistID, + ) + return i, err +} + const deleteSystemPlaylistsForUser = `-- name: DeleteSystemPlaylistsForUser :exec DELETE FROM playlists WHERE user_id = $1 AND kind = 'system' ` diff --git a/internal/db/queries/system_playlists.sql b/internal/db/queries/system_playlists.sql index 0ae5326a..4f77875b 100644 --- a/internal/db/queries/system_playlists.sql +++ b/internal/db/queries/system_playlists.sql @@ -140,3 +140,13 @@ SELECT id, user_id, name, description, is_public, kind, system_variant, WHERE user_id = $1 AND ($2::text = 'all' OR kind = $2) ORDER BY updated_at DESC; + +-- name: CreateSystemPlaylist :one +-- Inserts a system-generated playlist. Used by BuildSystemPlaylists. +-- For 'for_you' variant, pass seed_artist_id as NULL (zero-value pgtype.UUID +-- with Valid=false). +INSERT INTO playlists ( + user_id, name, description, is_public, + kind, system_variant, seed_artist_id, cover_path +) VALUES ($1, $2, '', false, 'system', $3, $4, $5) +RETURNING *; diff --git a/internal/playlists/system.go b/internal/playlists/system.go index 93549218..30291c1f 100644 --- a/internal/playlists/system.go +++ b/internal/playlists/system.go @@ -5,10 +5,20 @@ package playlists import ( + "context" + "errors" + "fmt" "hash/fnv" + "log/slog" "sort" + "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/recommendation" ) // seedArtistRow mirrors the sqlc-generated PickSeedArtistsRow shape. @@ -65,3 +75,264 @@ func stableSortByScoreThenHash(cands []rankedCandidate, dateStr string) { return tieBreakHash(cands[i].TrackID, dateStr) < tieBreakHash(cands[j].TrackID, dateStr) }) } + +const ( + systemMixLength = 25 + systemMixSeedLimit = 3 +) + +// systemMixWeights are the fixed scoring weights used by the cron worker. +// JitterMagnitude is 0 because daily determinism comes from tieBreakHash; +// any randomness would defeat the within-day stability invariant. +var systemMixWeights = recommendation.ScoringWeights{ + BaseWeight: 1.0, + LikeBoost: 2.0, + RecencyWeight: 1.0, + SkipPenalty: 2.0, + JitterMagnitude: 0.0, + ContextWeight: 0.5, + SimilarityWeight: 1.5, +} + +// noopRNG returns 0 for any call. Combined with JitterMagnitude=0 it makes +// recommendation.Score fully deterministic. +func noopRNG() float64 { return 0 } + +// BuildSystemPlaylists builds the user's daily system mixes (one For-You + +// up to 3 Songs-like-{seed} mixes). Atomic-replace inside one tx; +// concurrency-guarded via system_playlist_runs.in_flight; deterministic +// within a day via tieBreakHash(track_id, now.UTC().Format("2006-01-02")). +// +// Callable from both the cron loop (system_cron.go) and the lazy fallback. +// Returns nil when the user has no library activity to build from. Errors +// are captured in system_playlist_runs.last_error before returning. +func BuildSystemPlaylists(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, userID pgtype.UUID, now time.Time) error { + q := dbq.New(pool) + dateStr := now.UTC().Format("2006-01-02") + dateOnly := pgtype.Date{Time: now.UTC(), Valid: true} + + // Concurrency guard: try to claim the run row. + if _, err := q.TryClaimSystemPlaylistRun(ctx, userID); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + logger.Debug("system playlist build: another run is in flight", + "user_id", uuidStringPL(userID)) + return nil + } + return fmt.Errorf("claim run: %w", err) + } + + // On any return path, mark the run finished. Failure capture is + // best-effort; success is updated in the happy path. + var buildErr error + defer func() { + if buildErr == nil { + _ = q.FinishSystemPlaylistRun(ctx, dbq.FinishSystemPlaylistRunParams{ + UserID: userID, + LastRunAt: pgtype.Timestamptz{Time: now.UTC(), Valid: true}, + LastRunDate: dateOnly, + }) + } else { + errStr := buildErr.Error() + _ = q.FailSystemPlaylistRun(ctx, dbq.FailSystemPlaylistRunParams{ + UserID: userID, + LastError: &errStr, + }) + } + }() + + // 1. For-You: synth seed from top recently-played track, pull candidates. + forYouSeed, err := q.PickTopPlayedTrackForUser(ctx, userID) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + buildErr = fmt.Errorf("pick for-you seed: %w", err) + return buildErr + } + var forYouTracks []rankedCandidate + if err == nil && forYouSeed.Valid { + zeroVec := recommendation.SessionVector{Seed: true} + cands, cerr := recommendation.LoadCandidatesFromSimilarity( + ctx, q, userID, forYouSeed, + 1, // recentlyPlayedHours — small to avoid filtering the seed's recent neighbourhood + zeroVec, + []pgtype.UUID{forYouSeed}, + recommendation.DefaultCandidateSourceLimits(), + ) + if cerr == nil { + forYouTracks = pickTopN(cands, dateStr, now, systemMixLength) + } else { + logger.Warn("system playlist: for-you candidates load failed; skipping", + "user_id", uuidStringPL(userID), "err", cerr) + } + } + + // 2. Seed artists for "Songs like {X}". + seedRows, err := q.PickSeedArtists(ctx, userID) + if err != nil { + buildErr = fmt.Errorf("pick seed artists: %w", err) + return buildErr + } + seedRowsLocal := make([]seedArtistRow, 0, len(seedRows)) + for _, r := range seedRows { + seedRowsLocal = append(seedRowsLocal, seedArtistRow{ + ArtistID: r.ArtistID, Score: r.Score, + }) + } + seeds := pickSeedArtistsFromRows(seedRowsLocal) + + type seedMix struct { + ArtistID pgtype.UUID + ArtistName string + CoverPath *string + Tracks []rankedCandidate + } + seedMixes := make([]seedMix, 0, len(seeds)) + for _, artistID := range seeds { + artistRow, aerr := q.GetArtistByID(ctx, artistID) + if aerr != nil { + logger.Warn("system playlist: seed artist load failed; skipping", + "artist_id", uuidStringPL(artistID), "err", aerr) + continue + } + seedTrack, terr := q.PickTopPlayedTrackForArtistByUser(ctx, dbq.PickTopPlayedTrackForArtistByUserParams{ + UserID: userID, ArtistID: artistID, + }) + if terr != nil || !seedTrack.Valid { + continue + } + zeroVec := recommendation.SessionVector{Seed: true} + cands, cerr := recommendation.LoadCandidatesFromSimilarity( + ctx, q, userID, seedTrack, 1, zeroVec, []pgtype.UUID{seedTrack}, + recommendation.DefaultCandidateSourceLimits(), + ) + if cerr != nil { + logger.Warn("system playlist: seed candidates load failed; skipping", + "artist_id", uuidStringPL(artistID), "err", cerr) + continue + } + // Keep only tracks NOT by the seed artist (we want "songs like X", + // not X's own songs). + filtered := make([]recommendation.Candidate, 0, len(cands)) + for _, c := range cands { + if !pgtypeUUIDEqual(c.Track.ArtistID, artistID) { + filtered = append(filtered, c) + } + } + tracks := pickTopN(filtered, dateStr, now, systemMixLength) + coverPath, _ := q.PickTopAlbumCoverForArtistByUser(ctx, dbq.PickTopAlbumCoverForArtistByUserParams{ + UserID: userID, ArtistID: artistID, + }) + seedMixes = append(seedMixes, seedMix{ + ArtistID: artistID, + ArtistName: artistRow.Name, + CoverPath: coverPath, + Tracks: tracks, + }) + } + + // 3. Atomic replace inside a transaction. + tx, err := pool.Begin(ctx) + if err != nil { + buildErr = fmt.Errorf("begin tx: %w", err) + return buildErr + } + defer func() { _ = tx.Rollback(ctx) }() + qtx := dbq.New(tx) + + if err := qtx.DeleteSystemPlaylistsForUser(ctx, userID); err != nil { + buildErr = fmt.Errorf("delete old system playlists: %w", err) + return buildErr + } + + if len(forYouTracks) > 0 { + if err := insertSystemPlaylist(ctx, qtx, userID, "For You", "for_you", + pgtype.UUID{}, nil, forYouTracks); err != nil { + buildErr = fmt.Errorf("insert for-you: %w", err) + return buildErr + } + } + + for _, m := range seedMixes { + if len(m.Tracks) == 0 { + continue + } + name := fmt.Sprintf("Songs like %s", m.ArtistName) + if err := insertSystemPlaylist(ctx, qtx, userID, name, "songs_like_artist", + m.ArtistID, m.CoverPath, m.Tracks); err != nil { + buildErr = fmt.Errorf("insert songs-like %s: %w", uuidStringPL(m.ArtistID), err) + return buildErr + } + } + + if err := tx.Commit(ctx); err != nil { + buildErr = fmt.Errorf("commit tx: %w", err) + return buildErr + } + return nil +} + +// pickTopN ranks candidates with recommendation.Score, sorts by score-DESC +// breaking ties via tieBreakHash(track_id, dateStr), then truncates to N. +func pickTopN(cands []recommendation.Candidate, dateStr string, now time.Time, n int) []rankedCandidate { + ranked := make([]rankedCandidate, 0, len(cands)) + for _, c := range cands { + s := recommendation.Score(c.Inputs, systemMixWeights, now, noopRNG) + ranked = append(ranked, rankedCandidate{TrackID: c.Track.ID, Score: s}) + } + stableSortByScoreThenHash(ranked, dateStr) + if len(ranked) > n { + ranked = ranked[:n] + } + return ranked +} + +// insertSystemPlaylist inserts a kind='system' playlists row plus its +// playlist_tracks via AppendPlaylistTrack (slice 1's existing query, which +// does the snapshot lookup from tracks/albums/artists). Then recomputes rollups. +// +// Pass seed_artist_id with Valid=false (zero pgtype.UUID) for the For-You +// variant; the CreateSystemPlaylist query persists NULL. +func insertSystemPlaylist(ctx context.Context, qtx *dbq.Queries, userID pgtype.UUID, + name, variant string, seedArtistID pgtype.UUID, coverPath *string, + tracks []rankedCandidate) error { + variantStr := variant + p, err := qtx.CreateSystemPlaylist(ctx, dbq.CreateSystemPlaylistParams{ + UserID: userID, + Name: name, + SystemVariant: &variantStr, + SeedArtistID: seedArtistID, + CoverPath: coverPath, + }) + if err != nil { + return fmt.Errorf("insert playlist row: %w", err) + } + + for _, t := range tracks { + if _, err := qtx.AppendPlaylistTrack(ctx, dbq.AppendPlaylistTrackParams{ + PlaylistID: p.ID, + TrackID: t.TrackID, + }); err != nil { + // Track may have been deleted between candidate-load and insert; + // skip silently rather than failing the whole build. + if errors.Is(err, pgx.ErrNoRows) { + continue + } + return fmt.Errorf("append track %s: %w", uuidStringPL(t.TrackID), err) + } + } + + if err := qtx.UpdatePlaylistRollups(ctx, p.ID); err != nil { + return fmt.Errorf("update rollups: %w", err) + } + return nil +} + +// uuidStringPL renders a pgtype.UUID as the canonical 8-4-4-4-12 form. +// (pgtype.UUID's String method exists on some pgx versions but not others; +// also the playlists package needs this independent of test helpers.) +// Suffix `PL` distinguishes it from any test helper named uuidString. +func uuidStringPL(u pgtype.UUID) string { + if !u.Valid { + return "" + } + return fmt.Sprintf("%x-%x-%x-%x-%x", + u.Bytes[0:4], u.Bytes[4:6], u.Bytes[6:8], u.Bytes[8:10], u.Bytes[10:16]) +} diff --git a/internal/playlists/system_test.go b/internal/playlists/system_test.go new file mode 100644 index 00000000..0f58ba52 --- /dev/null +++ b/internal/playlists/system_test.go @@ -0,0 +1,341 @@ +package playlists_test + +import ( + "context" + "io" + "log/slog" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/playlists" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// seedPlayEvent inserts a play_events row with the given was_skipped flag, +// at startedAt, for the given user/track. Uses raw SQL because the sqlc +// `InsertPlayEvent` doesn't accept was_skipped (that's set by an UPDATE). +func seedPlayEvent(t *testing.T, pool *pgxpool.Pool, userID, trackID pgtype.UUID, startedAt time.Time, wasSkipped bool) { + t.Helper() + _, err := pool.Exec(context.Background(), ` + INSERT INTO play_events (user_id, track_id, session_id, started_at, was_skipped) + VALUES ($1, $2, gen_random_uuid(), $3, $4) + `, userID, trackID, startedAt, wasSkipped) + if err != nil { + t.Fatalf("seed play_event: %v", err) + } +} + +// seedQuarantine inserts a lidarr_quarantine row for (user, track). +func seedQuarantine(t *testing.T, pool *pgxpool.Pool, userID, trackID pgtype.UUID) { + t.Helper() + _, err := pool.Exec(context.Background(), ` + INSERT INTO lidarr_quarantine (user_id, track_id, reason) + VALUES ($1, $2, 'test-hide') + `, userID, trackID) + if err != nil { + t.Fatalf("seed quarantine: %v", err) + } +} + +// seedActiveLibrary creates a user with N played tracks across M artists, +// all within the last 7 days. Returns the user and the slice of tracks. +// Each track gets ≥3 unskipped plays so PickSeedArtists has a real signal. +func seedActiveLibrary(t *testing.T, pool *pgxpool.Pool, name string, numArtists, tracksPerArtist int) (dbq.User, []dbq.Track) { + t.Helper() + u := seedUser(t, pool, name) + now := time.Now().UTC() + var allTracks []dbq.Track + for a := 0; a < numArtists; a++ { + artistName := name + "-artist-" + string(rune('A'+a)) + for k := 0; k < tracksPerArtist; k++ { + tk := seedTrack(t, pool, name+"-track-"+string(rune('A'+a))+string(rune('0'+k)), artistName) + allTracks = append(allTracks, tk) + // 3 unskipped plays per track to populate engagement stats. + for p := 0; p < 3; p++ { + seedPlayEvent(t, pool, u.ID, tk.ID, now.Add(-time.Duration(a*10+k+p)*time.Hour), false) + } + } + } + return u, allTracks +} + +func TestBuildSystemPlaylists_SufficientActivity(t *testing.T) { + pool := newPool(t) + logger := discardLogger() + u, _ := seedActiveLibrary(t, pool, "act1", 4, 3) + if err := playlists.BuildSystemPlaylists(context.Background(), pool, logger, u.ID, time.Now().UTC()); err != nil { + t.Fatalf("build: %v", err) + } + + rows, err := dbq.New(pool).ListPlaylistsByUserAndKind(context.Background(), dbq.ListPlaylistsByUserAndKindParams{ + UserID: u.ID, Column2: "system", + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(rows) == 0 { + t.Fatalf("expected at least one system playlist; got 0") + } + + var hasForYou, hasSongsLike bool + for _, r := range rows { + if r.Kind != "system" { + t.Errorf("row %s: kind=%q want 'system'", uuidString(r.ID), r.Kind) + } + if r.SystemVariant == nil { + t.Errorf("row %s: nil system_variant", uuidString(r.ID)) + continue + } + switch *r.SystemVariant { + case "for_you": + hasForYou = true + if r.SeedArtistID.Valid { + t.Errorf("for_you row should have NULL seed_artist_id; got valid") + } + case "songs_like_artist": + hasSongsLike = true + if !r.SeedArtistID.Valid { + t.Errorf("songs_like_artist row should have non-NULL seed_artist_id") + } + default: + t.Errorf("unknown system_variant=%q", *r.SystemVariant) + } + if r.TrackCount == 0 { + t.Errorf("row %s: empty track_count", uuidString(r.ID)) + } + if r.TrackCount > 25 { + t.Errorf("row %s: track_count=%d exceeds 25", uuidString(r.ID), r.TrackCount) + } + } + if !hasForYou { + t.Error("expected a for_you playlist") + } + if !hasSongsLike { + t.Error("expected at least one songs_like_artist playlist") + } +} + +func TestBuildSystemPlaylists_QuarantineExcluded(t *testing.T) { + pool := newPool(t) + logger := discardLogger() + u, tracks := seedActiveLibrary(t, pool, "qact", 3, 3) + + quarantined := tracks[0] + seedQuarantine(t, pool, u.ID, quarantined.ID) + + if err := playlists.BuildSystemPlaylists(context.Background(), pool, logger, u.ID, time.Now().UTC()); err != nil { + t.Fatalf("build: %v", err) + } + + // Inspect playlist_tracks for the user's system playlists; quarantined + // track must not appear. + rows, err := pool.Query(context.Background(), ` + SELECT pt.track_id + FROM playlist_tracks pt + JOIN playlists p ON p.id = pt.playlist_id + WHERE p.user_id = $1 AND p.kind = 'system' + `, u.ID) + if err != nil { + t.Fatalf("query playlist_tracks: %v", err) + } + defer rows.Close() + for rows.Next() { + var tid pgtype.UUID + if err := rows.Scan(&tid); err != nil { + t.Fatalf("scan: %v", err) + } + if tid.Bytes == quarantined.ID.Bytes { + t.Fatalf("quarantined track %s appeared in a system playlist", uuidString(quarantined.ID)) + } + } +} + +func TestBuildSystemPlaylists_AtomicReplace(t *testing.T) { + pool := newPool(t) + logger := discardLogger() + u, _ := seedActiveLibrary(t, pool, "atomic", 3, 3) + ctx := context.Background() + + // First build. + if err := playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, time.Now().UTC()); err != nil { + t.Fatalf("first build: %v", err) + } + var ids1 []pgtype.UUID + rows1, _ := dbq.New(pool).ListPlaylistsByUserAndKind(ctx, dbq.ListPlaylistsByUserAndKindParams{ + UserID: u.ID, Column2: "system", + }) + for _, r := range rows1 { + ids1 = append(ids1, r.ID) + } + if len(ids1) == 0 { + t.Skip("no system playlists from first build; can't verify replace") + } + + // Second build — old rows must be gone, replaced with fresh ones. + if err := playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, time.Now().UTC()); err != nil { + t.Fatalf("second build: %v", err) + } + rows2, _ := dbq.New(pool).ListPlaylistsByUserAndKind(ctx, dbq.ListPlaylistsByUserAndKindParams{ + UserID: u.ID, Column2: "system", + }) + for _, r := range rows2 { + for _, oldID := range ids1 { + if r.ID.Bytes == oldID.Bytes { + t.Errorf("playlist id %s persisted across rebuilds — atomic replace failed", uuidString(oldID)) + } + } + } +} + +func TestBuildSystemPlaylists_Concurrency(t *testing.T) { + pool := newPool(t) + logger := discardLogger() + u, _ := seedActiveLibrary(t, pool, "conc", 3, 3) + ctx := context.Background() + now := time.Now().UTC() + + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); _ = playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, now) }() + go func() { defer wg.Done(); _ = playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, now) }() + wg.Wait() + + // Exactly one run must have completed; in_flight must be false either way. + run, err := dbq.New(pool).GetSystemPlaylistRun(ctx, u.ID) + if err != nil { + t.Fatalf("get run: %v", err) + } + if run.InFlight { + t.Errorf("in_flight should be false after both calls return") + } + if !run.LastRunAt.Valid { + t.Errorf("last_run_at should be set after at least one successful build") + } +} + +func TestBuildSystemPlaylists_DailyNonceDeterminism(t *testing.T) { + pool := newPool(t) + logger := discardLogger() + u, _ := seedActiveLibrary(t, pool, "nonce", 4, 5) + ctx := context.Background() + day1 := time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC) + + if err := playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, day1); err != nil { + t.Fatalf("build day1 first: %v", err) + } + snap1 := snapshotSystemTracks(t, pool, u.ID) + + if err := playlists.BuildSystemPlaylists(ctx, pool, logger, u.ID, day1); err != nil { + t.Fatalf("build day1 second: %v", err) + } + snap2 := snapshotSystemTracks(t, pool, u.ID) + + if !equalSnapshots(snap1, snap2) { + t.Errorf("same-day rebuild produced different track lists") + } +} + +// snapshotSystemTracks returns a per-playlist-name → []track_id mapping +// for the user's system playlists, in playlist_tracks position order. +func snapshotSystemTracks(t *testing.T, pool *pgxpool.Pool, userID pgtype.UUID) map[string][]string { + t.Helper() + rows, err := pool.Query(context.Background(), ` + SELECT p.name, pt.track_id + FROM playlists p + JOIN playlist_tracks pt ON pt.playlist_id = p.id + WHERE p.user_id = $1 AND p.kind = 'system' + ORDER BY p.name, pt.position + `, userID) + if err != nil { + t.Fatalf("snapshot query: %v", err) + } + defer rows.Close() + out := map[string][]string{} + for rows.Next() { + var name string + var tid pgtype.UUID + if err := rows.Scan(&name, &tid); err != nil { + t.Fatalf("snapshot scan: %v", err) + } + out[name] = append(out[name], uuidString(tid)) + } + return out +} + +func equalSnapshots(a, b map[string][]string) bool { + if len(a) != len(b) { + return false + } + for k, va := range a { + vb, ok := b[k] + if !ok || len(va) != len(vb) { + return false + } + for i := range va { + if va[i] != vb[i] { + return false + } + } + } + return true +} + +func TestListActiveUsersForSystemPlaylists(t *testing.T) { + pool := newPool(t) + ctx := context.Background() + activeU, tracks := seedActiveLibrary(t, pool, "active", 1, 1) + inactiveU := seedUser(t, pool, "inactive") + // inactive user has plays older than 7 days. + seedPlayEvent(t, pool, inactiveU.ID, tracks[0].ID, time.Now().UTC().AddDate(0, 0, -30), false) + + ids, err := dbq.New(pool).ListActiveUsersForSystemPlaylists(ctx) + if err != nil { + t.Fatalf("list: %v", err) + } + + var sawActive, sawInactive bool + for _, id := range ids { + if id.Bytes == activeU.ID.Bytes { + sawActive = true + } + if id.Bytes == inactiveU.ID.Bytes { + sawInactive = true + } + } + if !sawActive { + t.Error("active user not returned") + } + if sawInactive { + t.Error("inactive user (plays > 7d old) was returned") + } +} + +func TestStartupRecoveryClearsStaleInFlight(t *testing.T) { + pool := newPool(t) + ctx := context.Background() + u := seedUser(t, pool, "stale") + // Seed a stale in_flight=true row. + if _, err := pool.Exec(ctx, `INSERT INTO system_playlist_runs (user_id, in_flight) VALUES ($1, true)`, u.ID); err != nil { + t.Fatalf("seed runs: %v", err) + } + + if err := dbq.New(pool).ClearStaleSystemPlaylistInFlight(ctx); err != nil { + t.Fatalf("clear: %v", err) + } + + run, err := dbq.New(pool).GetSystemPlaylistRun(ctx, u.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if run.InFlight { + t.Errorf("in_flight should be false after ClearStale; got true") + } +}