diff --git a/internal/playlists/discover.go b/internal/playlists/discover.go new file mode 100644 index 00000000..e3b7aad8 --- /dev/null +++ b/internal/playlists/discover.go @@ -0,0 +1,259 @@ +package playlists + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgtype" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +const ( + discoverTotalSlots = 100 + discoverDormantSlots = 40 + discoverCrossUserSlots = 30 + discoverRandomSlots = 30 + discoverMaxTracksPerAlbum = 2 + discoverMaxTracksPerArtist = 3 +) + +// discoverTrack is the common shape used by the bucket allocator. The +// three sqlc-generated row types collapse into this internal struct so +// downstream functions don't need to be generic over them. +type discoverTrack struct { + ID pgtype.UUID + AlbumID pgtype.UUID + ArtistID pgtype.UUID +} + +// buildDiscoverCandidates assembles the Discover playlist track list. +// Pulls from three buckets, applies per-album/per-artist caps, then +// redistributes any deficit equally across the remaining buckets. +// +// Returns up to discoverTotalSlots track IDs in the order they should +// appear in the playlist (round-robin interleaved across buckets). +// Empty slice when the library has no eligible tracks at all. +func buildDiscoverCandidates(ctx context.Context, q *dbq.Queries, userID pgtype.UUID, dateStr string) ([]rankedCandidate, error) { + dormantRows, err := q.ListDormantArtistTracksForDiscover(ctx, dbq.ListDormantArtistTracksForDiscoverParams{ + UserID: userID, Column2: dateStr, + }) + if err != nil { + return nil, fmt.Errorf("dormant bucket: %w", err) + } + crossUserRows, err := q.ListCrossUserLikedTracksForDiscover(ctx, dbq.ListCrossUserLikedTracksForDiscoverParams{ + UserID: userID, Column2: dateStr, + }) + if err != nil { + return nil, fmt.Errorf("cross-user bucket: %w", err) + } + randomRows, err := q.ListRandomUnheardTracksForDiscover(ctx, dbq.ListRandomUnheardTracksForDiscoverParams{ + UserID: userID, Column2: dateStr, + }) + if err != nil { + return nil, fmt.Errorf("random bucket: %w", err) + } + + // Adapt sqlc-generated rows into the internal struct, then apply + // per-album / per-artist caps. + dormantPool := capByAlbumAndArtist(dormantRowsToTracks(dormantRows)) + crossUserPool := capByAlbumAndArtist(crossUserRowsToTracks(crossUserRows)) + randomPool := capByAlbumAndArtist(randomRowsToTracks(randomRows)) + + // Allocate slots with redistribution. + allocations := redistributeSlots([]bucketRequest{ + {want: discoverDormantSlots, available: len(dormantPool)}, + {want: discoverCrossUserSlots, available: len(crossUserPool)}, + {want: discoverRandomSlots, available: len(randomPool)}, + }) + + // Take the head of each bucket's capped pool. + dormantTake := dormantPool[:allocations[0]] + crossUserTake := crossUserPool[:allocations[1]] + randomTake := randomPool[:allocations[2]] + + // Round-robin interleave so the playlist doesn't front-load one + // flavour. + out := interleaveBuckets(dormantTake, crossUserTake, randomTake) + + // Convert to rankedCandidate (the type insertSystemPlaylist accepts). + // Score is unused for Discover — the daily-deterministic md5 ordering + // already gave us the ranking. + ranked := make([]rankedCandidate, 0, len(out)) + for _, t := range out { + ranked = append(ranked, rankedCandidate{TrackID: t.ID}) + } + return ranked, nil +} + +func dormantRowsToTracks(rows []dbq.ListDormantArtistTracksForDiscoverRow) []discoverTrack { + out := make([]discoverTrack, len(rows)) + for i, r := range rows { + out[i] = discoverTrack{ID: r.ID, AlbumID: r.AlbumID, ArtistID: r.ArtistID} + } + return out +} + +func crossUserRowsToTracks(rows []dbq.ListCrossUserLikedTracksForDiscoverRow) []discoverTrack { + out := make([]discoverTrack, len(rows)) + for i, r := range rows { + out[i] = discoverTrack{ID: r.ID, AlbumID: r.AlbumID, ArtistID: r.ArtistID} + } + return out +} + +func randomRowsToTracks(rows []dbq.ListRandomUnheardTracksForDiscoverRow) []discoverTrack { + out := make([]discoverTrack, len(rows)) + for i, r := range rows { + out[i] = discoverTrack{ID: r.ID, AlbumID: r.AlbumID, ArtistID: r.ArtistID} + } + return out +} + +// capByAlbumAndArtist trims a bucket's pool so no single album appears +// more than discoverMaxTracksPerAlbum times and no single artist +// appears more than discoverMaxTracksPerArtist times. Preserves input +// order (which is the daily-deterministic md5 sort). +// +// pgtype.UUID's Bytes field is [16]byte, which IS comparable as a map +// key. +func capByAlbumAndArtist(rows []discoverTrack) []discoverTrack { + albumCount := map[pgtype.UUID]int{} + artistCount := map[pgtype.UUID]int{} + out := make([]discoverTrack, 0, len(rows)) + for _, r := range rows { + if albumCount[r.AlbumID] >= discoverMaxTracksPerAlbum { + continue + } + if artistCount[r.ArtistID] >= discoverMaxTracksPerArtist { + continue + } + albumCount[r.AlbumID]++ + artistCount[r.ArtistID]++ + out = append(out, r) + } + return out +} + +// bucketRequest captures one bucket's desired and available count. +type bucketRequest struct { + want int + available int +} + +// redistributeSlots returns the final per-bucket allocation given each +// bucket's want and available count. When a bucket can't fill its want, +// the deficit splits equally between the OTHER buckets that still have +// supply. Sum of returned allocations <= sum of `want`. +// +// Algorithm: +// 1. Initial pass: allocate min(want, available) to each bucket; +// compute supply = available - allocated. +// 2. For each deficit bucket (allocated < want), split the deficit +// equally across peers that still have supply > 0. +// 3. Repeat until no further movement (deficit redistribution +// can create new deficits if the recipient was already at its +// `want`; we cap each bucket at its `available` after every pass). +// +// For 3 buckets, two redistribution passes are sufficient in practice; +// we cap at 4 passes defensively. +func redistributeSlots(buckets []bucketRequest) []int { + n := len(buckets) + final := make([]int, n) + supply := make([]int, n) + for i, b := range buckets { + final[i] = b.want + if final[i] > b.available { + final[i] = b.available + } + supply[i] = b.available - final[i] + } + + for pass := 0; pass < 4; pass++ { + anyMoved := false + for i, b := range buckets { + deficit := b.want - final[i] + if deficit <= 0 { + continue + } + // Find peers with supply. + peers := make([]int, 0, n-1) + for j := 0; j < n; j++ { + if j == i { + continue + } + if supply[j] > 0 { + peers = append(peers, j) + } + } + if len(peers) == 0 { + continue + } + share := deficit / len(peers) + rem := deficit % len(peers) + for k, j := range peers { + take := share + if k < rem { + take++ + } + if take > supply[j] { + take = supply[j] + } + if take > 0 { + final[j] += take + supply[j] -= take + anyMoved = true + } + } + } + if !anyMoved { + break + } + } + + // Final clamp: total <= discoverTotalSlots. + sum := 0 + for _, f := range final { + sum += f + } + for sum > discoverTotalSlots { + // Trim the largest bucket. Should be rare. + maxVal, idx := -1, -1 + for i, f := range final { + if f > maxVal { + maxVal, idx = f, i + } + } + if idx < 0 { + break + } + final[idx]-- + sum-- + } + return final +} + +// interleaveBuckets round-robins items from the buckets in order. The +// first item of each bucket appears before the second of any bucket. +func interleaveBuckets(buckets ...[]discoverTrack) []discoverTrack { + total := 0 + for _, b := range buckets { + total += len(b) + } + out := make([]discoverTrack, 0, total) + indices := make([]int, len(buckets)) + for { + anyAppended := false + for bi, b := range buckets { + if indices[bi] < len(b) { + out = append(out, b[indices[bi]]) + indices[bi]++ + anyAppended = true + } + } + if !anyAppended { + break + } + } + return out +} diff --git a/internal/playlists/discover_test.go b/internal/playlists/discover_test.go new file mode 100644 index 00000000..e58d93a7 --- /dev/null +++ b/internal/playlists/discover_test.go @@ -0,0 +1,139 @@ +package playlists + +import ( + "testing" + + "github.com/jackc/pgx/v5/pgtype" +) + +// uuidN generates a deterministic pgtype.UUID from an integer for cap tests. +func uuidN(n int) pgtype.UUID { + var u pgtype.UUID + u.Valid = true + u.Bytes[15] = byte(n) + u.Bytes[14] = byte(n >> 8) + return u +} + +func TestRedistributeSlots_AllAvailable(t *testing.T) { + got := redistributeSlots([]bucketRequest{ + {want: 40, available: 100}, + {want: 30, available: 100}, + {want: 30, available: 100}, + }) + if got[0] != 40 || got[1] != 30 || got[2] != 30 { + t.Errorf("got %v, want [40, 30, 30]", got) + } +} + +func TestRedistributeSlots_ColdStart(t *testing.T) { + // Both signal buckets empty; all 100 slots roll into random. + got := redistributeSlots([]bucketRequest{ + {want: 40, available: 0}, + {want: 30, available: 0}, + {want: 30, available: 200}, + }) + if got[0] != 0 || got[1] != 0 { + t.Errorf("dormant/cross-user should be 0; got %v", got) + } + if got[2] != 100 { + t.Errorf("random = %d, want 100 (cold start fills from random)", got[2]) + } +} + +func TestRedistributeSlots_SingleUser(t *testing.T) { + // Cross-user empty; 30 slots split 15/15 between dormant + random. + got := redistributeSlots([]bucketRequest{ + {want: 40, available: 100}, + {want: 30, available: 0}, + {want: 30, available: 100}, + }) + if got[1] != 0 { + t.Errorf("cross-user = %d, want 0", got[1]) + } + // dormant + random = 40+15 + 30+15 = 100; order of distribution + // can put the +1 on either side if the deficit was odd, but 30 + // is even so 15/15. + if got[0] != 55 || got[2] != 45 { + t.Errorf("got %v, want [55, 0, 45]", got) + } +} + +func TestRedistributeSlots_PartialDormant(t *testing.T) { + // Dormant returns 25 of 40; deficit = 15, splits 7+8 across peers. + got := redistributeSlots([]bucketRequest{ + {want: 40, available: 25}, + {want: 30, available: 100}, + {want: 30, available: 100}, + }) + if got[0] != 25 { + t.Errorf("dormant = %d, want 25 (capped at available)", got[0]) + } + sum := got[0] + got[1] + got[2] + if sum != 100 { + t.Errorf("total = %d, want 100", sum) + } + // got[1] and got[2] should each be 30 + (7 or 8). + if got[1] < 37 || got[1] > 38 { + t.Errorf("cross-user = %d, want 37 or 38", got[1]) + } + if got[2] < 37 || got[2] > 38 { + t.Errorf("random = %d, want 37 or 38", got[2]) + } +} + +func TestRedistributeSlots_TotalLibraryEmpty(t *testing.T) { + // All buckets empty (degenerate library). + got := redistributeSlots([]bucketRequest{ + {want: 40, available: 0}, + {want: 30, available: 0}, + {want: 30, available: 0}, + }) + if got[0] != 0 || got[1] != 0 || got[2] != 0 { + t.Errorf("got %v, want [0, 0, 0]", got) + } +} + +func TestCapByAlbumAndArtist_AlbumCap(t *testing.T) { + rows := []discoverTrack{ + {ID: uuidN(1), AlbumID: uuidN(10), ArtistID: uuidN(100)}, + {ID: uuidN(2), AlbumID: uuidN(10), ArtistID: uuidN(101)}, + {ID: uuidN(3), AlbumID: uuidN(10), ArtistID: uuidN(102)}, // 3rd from album 10 → drop + {ID: uuidN(4), AlbumID: uuidN(11), ArtistID: uuidN(103)}, + } + got := capByAlbumAndArtist(rows) + if len(got) != 3 { + t.Errorf("len = %d, want 3 (album cap drops the 3rd from album 10)", len(got)) + } +} + +func TestCapByAlbumAndArtist_ArtistCap(t *testing.T) { + rows := []discoverTrack{ + {ID: uuidN(1), AlbumID: uuidN(10), ArtistID: uuidN(100)}, + {ID: uuidN(2), AlbumID: uuidN(11), ArtistID: uuidN(100)}, + {ID: uuidN(3), AlbumID: uuidN(12), ArtistID: uuidN(100)}, + {ID: uuidN(4), AlbumID: uuidN(13), ArtistID: uuidN(100)}, // 4th by artist → drop + {ID: uuidN(5), AlbumID: uuidN(14), ArtistID: uuidN(101)}, + } + got := capByAlbumAndArtist(rows) + if len(got) != 4 { + t.Errorf("len = %d, want 4 (artist cap drops the 4th by artist 100)", len(got)) + } +} + +func TestInterleaveBuckets_RoundRobin(t *testing.T) { + a := []discoverTrack{{ID: uuidN(1)}, {ID: uuidN(2)}} + b := []discoverTrack{{ID: uuidN(10)}, {ID: uuidN(20)}} + c := []discoverTrack{{ID: uuidN(100)}} + got := interleaveBuckets(a, b, c) + if len(got) != 5 { + t.Fatalf("len = %d, want 5", len(got)) + } + // Expected order: a[0], b[0], c[0], a[1], b[1]. + wantOrder := []byte{1, 10, 100, 2, 20} + for i, w := range wantOrder { + if got[i].ID.Bytes[15] != w { + t.Errorf("got[%d].ID = %d, want %d", i, got[i].ID.Bytes[15], w) + } + } +} diff --git a/internal/playlists/system.go b/internal/playlists/system.go index 5d131586..268beb30 100644 --- a/internal/playlists/system.go +++ b/internal/playlists/system.go @@ -226,6 +226,14 @@ func BuildSystemPlaylists(ctx context.Context, pool *pgxpool.Pool, logger *slog. }) } + // 4. Discover: surface unheard tracks, biased toward dormant artists. + discoverTracks, derr := buildDiscoverCandidates(ctx, q, userID, dateStr) + if derr != nil { + logger.Warn("system playlist: discover candidates load failed; skipping", + "user_id", uuidStringPL(userID), "err", derr) + discoverTracks = nil + } + // 3. Atomic replace inside a transaction. tx, err := pool.Begin(ctx) if err != nil { @@ -270,6 +278,16 @@ func BuildSystemPlaylists(ctx context.Context, pool *pgxpool.Pool, logger *slog. createdIDs = append(createdIDs, id) } + if len(discoverTracks) > 0 { + id, err := insertSystemPlaylist(ctx, qtx, userID, "Discover", "discover", + pgtype.UUID{}, discoverTracks) + if err != nil { + buildErr = fmt.Errorf("insert discover: %w", err) + return buildErr + } + createdIDs = append(createdIDs, id) + } + if err := tx.Commit(ctx); err != nil { buildErr = fmt.Errorf("commit tx: %w", err) return buildErr diff --git a/internal/playlists/system_test.go b/internal/playlists/system_test.go index ee67c746..64d7c456 100644 --- a/internal/playlists/system_test.go +++ b/internal/playlists/system_test.go @@ -105,6 +105,8 @@ func TestBuildSystemPlaylists_SufficientActivity(t *testing.T) { if !r.SeedArtistID.Valid { t.Errorf("songs_like_artist row should have non-NULL seed_artist_id") } + case "discover": + // Discover playlist is valid — no seed_artist_id required. default: t.Errorf("unknown system_variant=%q", *r.SystemVariant) } @@ -320,6 +322,57 @@ func TestListActiveUsersForSystemPlaylists(t *testing.T) { } } +func TestBuildSystemPlaylists_DiscoverColdStart(t *testing.T) { + // User has no plays / likes → Discover's dormant + cross-user + // buckets are empty (or minimal), all slots come from random unheard. + pool := newPool(t) + logger := discardLogger() + u, _ := seedActiveLibrary(t, pool, "discover_cold", 4, 5) + // seedActiveLibrary seeds 3 plays per track; delete them so dormant + // bucket (< 10 plays per artist) and random bucket both operate on + // a true cold-start user with no prior play history. + if _, err := pool.Exec(context.Background(), + "DELETE FROM play_events WHERE user_id = $1", u.ID); err != nil { + t.Fatalf("clear plays: %v", err) + } + + if err := playlists.BuildSystemPlaylists(context.Background(), pool, logger, u.ID, time.Now().UTC(), t.TempDir()); 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) + } + var foundDiscover bool + for _, r := range rows { + if r.SystemVariant != nil && *r.SystemVariant == "discover" { + foundDiscover = true + if r.TrackCount == 0 { + t.Errorf("Discover playlist has zero tracks on cold start (random bucket should fill)") + } + if r.TrackCount > 100 { + t.Errorf("Discover track_count = %d, want <= 100", r.TrackCount) + } + } + } + if !foundDiscover { + t.Errorf("expected a 'discover' system playlist; got variants: %v", systemVariantsOf(rows)) + } +} + +func systemVariantsOf(rows []dbq.ListPlaylistsByUserAndKindRow) []string { + out := make([]string, 0, len(rows)) + for _, r := range rows { + if r.SystemVariant != nil { + out = append(out, *r.SystemVariant) + } + } + return out +} + func TestStartupRecoveryClearsStaleInFlight(t *testing.T) { pool := newPool(t) ctx := context.Background()