feat(server/m7-352): BuildSystemPlaylists with atomic replace + concurrency guard
Implements T5 of #352 slice 2: adds CreateSystemPlaylist sqlc query, BuildSystemPlaylists function (For-You + Songs-like mixes, tx atomic replace, in_flight guard, tieBreakHash determinism), and 7 integration tests covering activity, quarantine exclusion, atomic replace, concurrency, daily nonce stability, ListActiveUsers, and stale-in-flight recovery. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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 "<nil>"
|
||||
}
|
||||
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])
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user