feat(taste): household co-play similarity — #1533
Milestone #160 Opt 5. A collaborative candidate arm: tracks by artists co-played across the instance with the seed's artist. Minstrel is a single shared-library, multi-user server (no per-user library ACL — verified: no owner/share/group model), so the "household" is the whole instance's user set; the rule #47 scoping is satisfied by the shared-library boundary. Single-user servers produce no edges. - No migration: source='user_cooccurrence' was pre-whitelisted in the 0009 similarity CHECK from day one. - internal/db/queries/coplay.sql: Delete + Insert artist co-play edges. Score = Jaccard of the two artists' distinct-player sets (controls for globally-popular artists); >= 2 co-players AND Jaccard >= floor kept (the floor also self-limits hub artists). Completed plays, 365d window. - internal/coplay: periodic worker (6h) that atomic-replaces the user_cooccurrence edge set from play_events — pure local SQL, no external calls. Wired in main.go alongside the similarity worker. - LoadRadioCandidatesV2: new coplay_artists arm (source='user_cooccurrence', seed-artist based, 0.5 damp like similar_artists) + $11 limit; CandidateSourceLimits.UserCoplay (default 20, For-You 40). - Integration tests: perfect-overlap Jaccard=1.0 edge + single-user empty-set gate. Device axis and AcousticBrainz (Opt 4) are separately tracked; this closes the milestone-#160 sequential options. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
// Package coplay builds household co-play similarity edges (#1533, milestone
|
||||
// #160 Opt 5). Minstrel is a single shared-library, multi-user server (no
|
||||
// per-user library ACL), so the "household" is the whole instance's user set.
|
||||
// A periodic worker recomputes artist–artist co-occurrence from play_events
|
||||
// entirely in SQL (no external API) and stores the edges in artist_similarity
|
||||
// under source='user_cooccurrence' — the pre-provisioned co-occurrence source
|
||||
// from the 0009 schema — which the radio/mix candidate pool consumes as a
|
||||
// collaborative arm. Single-user servers produce no edges (the >= 2 co-player
|
||||
// gate), so the feature is a graceful no-op there.
|
||||
package coplay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
// minJaccard is the co-play edge floor: prune weak pairs and self-limit hub
|
||||
// artists, whose large combined-audience denominators pull their Jaccard down.
|
||||
// 0.1 keeps only pairs whose co-player overlap is a real fraction of their
|
||||
// combined audience.
|
||||
const minJaccard = 0.1
|
||||
|
||||
// Worker periodically rebuilds the household co-play edge set. Pure local SQL,
|
||||
// so there are no rate limits — but the pair self-join is roughly
|
||||
// O(users × artists_per_user²), so it ticks slowly: co-listening shifts over
|
||||
// days, not minutes, and a frequent rebuild would burn CPU for no fresher
|
||||
// signal.
|
||||
type Worker struct {
|
||||
pool *pgxpool.Pool
|
||||
logger *slog.Logger
|
||||
tick time.Duration
|
||||
}
|
||||
|
||||
// NewWorker constructs a worker with the production default 6h tick.
|
||||
func NewWorker(pool *pgxpool.Pool, logger *slog.Logger) *Worker {
|
||||
return &Worker{pool: pool, logger: logger, tick: 6 * time.Hour}
|
||||
}
|
||||
|
||||
// Run rebuilds once at startup (so edges exist without waiting a full tick),
|
||||
// then every w.tick until ctx is cancelled. Runs in its own goroutine, so a
|
||||
// slow initial rebuild never blocks boot.
|
||||
func (w *Worker) Run(ctx context.Context) {
|
||||
if err := w.rebuild(ctx); err != nil {
|
||||
w.logger.Error("coplay: initial rebuild failed", "err", err)
|
||||
}
|
||||
t := time.NewTicker(w.tick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
if err := w.rebuild(ctx); err != nil {
|
||||
w.logger.Error("coplay: rebuild failed", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rebuild atomic-replaces the user_cooccurrence edge set inside one
|
||||
// transaction — delete the old edges, recompute from current play history — so
|
||||
// the candidate pool never observes a half-built set.
|
||||
func (w *Worker) rebuild(ctx context.Context) error {
|
||||
tx, err := w.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
q := dbq.New(tx)
|
||||
if err := q.DeleteArtistCoplayEdges(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := q.InsertArtistCoplayEdges(ctx, minJaccard); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
w.logger.Debug("coplay: edges rebuilt")
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package coplay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"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 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
|
||||
}
|
||||
|
||||
func discardLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
|
||||
|
||||
func seedUser(t *testing.T, q *dbq.Queries, name string) pgtype.UUID {
|
||||
t.Helper()
|
||||
u, err := q.CreateUser(context.Background(), dbq.CreateUserParams{
|
||||
Username: dbtest.TestUserPrefix + name, PasswordHash: "x", ApiToken: name + "-tok", IsAdmin: false,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("seed user %s: %v", name, err)
|
||||
}
|
||||
return u.ID
|
||||
}
|
||||
|
||||
// seedArtistTrack creates an artist + album + one track, returning both ids.
|
||||
func seedArtistTrack(t *testing.T, q *dbq.Queries, name string) (pgtype.UUID, pgtype.UUID) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
a, err := q.UpsertArtist(ctx, dbq.UpsertArtistParams{Name: name, SortName: name})
|
||||
if err != nil {
|
||||
t.Fatalf("seed artist: %v", err)
|
||||
}
|
||||
al, err := q.UpsertAlbum(ctx, dbq.UpsertAlbumParams{Title: name, SortTitle: name, ArtistID: a.ID})
|
||||
if err != nil {
|
||||
t.Fatalf("seed album: %v", err)
|
||||
}
|
||||
tr, err := q.UpsertTrack(ctx, dbq.UpsertTrackParams{
|
||||
Title: name, AlbumID: al.ID, ArtistID: a.ID,
|
||||
DurationMs: 1000, FilePath: "/tmp/coplay-" + name + ".mp3",
|
||||
FileSize: 1, FileFormat: "mp3",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("seed track: %v", err)
|
||||
}
|
||||
return a.ID, tr.ID
|
||||
}
|
||||
|
||||
func play(t *testing.T, pool *pgxpool.Pool, userID, trackID pgtype.UUID) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
at := time.Now()
|
||||
var sessionID pgtype.UUID
|
||||
if err := pool.QueryRow(ctx,
|
||||
`INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id)
|
||||
VALUES ($1, $2, $2, 'coplay-test') RETURNING id`,
|
||||
userID, at).Scan(&sessionID); err != nil {
|
||||
t.Fatalf("insert play_session: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx,
|
||||
`INSERT INTO play_events (user_id, track_id, session_id, started_at, was_skipped)
|
||||
VALUES ($1, $2, $3, $4, false)`,
|
||||
userID, trackID, sessionID, at); err != nil {
|
||||
t.Fatalf("insert play_event: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func coplayEdgeCount(t *testing.T, pool *pgxpool.Pool) int {
|
||||
t.Helper()
|
||||
var n int
|
||||
if err := pool.QueryRow(context.Background(),
|
||||
`SELECT count(*) FROM artist_similarity WHERE source='user_cooccurrence'`).Scan(&n); err != nil {
|
||||
t.Fatalf("count edges: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// TestCoplayWorker_BuildsEdges: two users who both play two artists produce a
|
||||
// perfect-overlap co-play edge (Jaccard 1.0) in each direction.
|
||||
func TestCoplayWorker_BuildsEdges(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
ctx := context.Background()
|
||||
q := dbq.New(pool)
|
||||
|
||||
u1 := seedUser(t, q, "cop1")
|
||||
u2 := seedUser(t, q, "cop2")
|
||||
aA, tA := seedArtistTrack(t, q, "ArtA")
|
||||
aB, tB := seedArtistTrack(t, q, "ArtB")
|
||||
|
||||
for _, u := range []pgtype.UUID{u1, u2} {
|
||||
play(t, pool, u, tA)
|
||||
play(t, pool, u, tB)
|
||||
}
|
||||
|
||||
if err := NewWorker(pool, discardLogger()).rebuild(ctx); err != nil {
|
||||
t.Fatalf("rebuild: %v", err)
|
||||
}
|
||||
|
||||
var score float64
|
||||
if err := pool.QueryRow(ctx,
|
||||
`SELECT score FROM artist_similarity
|
||||
WHERE artist_a_id=$1 AND artist_b_id=$2 AND source='user_cooccurrence'`,
|
||||
aA, aB).Scan(&score); err != nil {
|
||||
t.Fatalf("expected co-play edge ArtA→ArtB: %v", err)
|
||||
}
|
||||
// Jaccard of perfectly-overlapping player sets (both users play both):
|
||||
// coplayers=2, players each=2 → 2/(2+2-2) = 1.0.
|
||||
if score != 1.0 {
|
||||
t.Errorf("edge score = %v, want 1.0 (perfect overlap)", score)
|
||||
}
|
||||
if got := coplayEdgeCount(t, pool); got != 2 {
|
||||
t.Errorf("edge count = %d, want 2 (A→B and B→A)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCoplayWorker_SingleUserNoEdges: one user's own co-listening yields no
|
||||
// edges — the >= 2 distinct-co-player gate keeps single-user servers empty.
|
||||
func TestCoplayWorker_SingleUserNoEdges(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
ctx := context.Background()
|
||||
q := dbq.New(pool)
|
||||
|
||||
u := seedUser(t, q, "solo")
|
||||
_, tA := seedArtistTrack(t, q, "SoloA")
|
||||
_, tB := seedArtistTrack(t, q, "SoloB")
|
||||
play(t, pool, u, tA)
|
||||
play(t, pool, u, tB)
|
||||
|
||||
if err := NewWorker(pool, discardLogger()).rebuild(ctx); err != nil {
|
||||
t.Fatalf("rebuild: %v", err)
|
||||
}
|
||||
if got := coplayEdgeCount(t, pool); got != 0 {
|
||||
t.Errorf("single-user edge count = %d, want 0 (>= 2 co-player gate)", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user