// Package similarity owns the inbound ListenBrainz similarity ingest // pipeline. A periodic worker queries the LB Labs API // (labs.api.listenbrainz.org similar-recordings / similar-artists) for // tracks the user has played, filters returned MBIDs to the local // library, and stores the top-K edges in track_similarity / // artist_similarity for M4c's radio candidate-pool builder. package similarity import ( "context" "errors" "log/slog" "sort" "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/scrobble/listenbrainz" ) // Worker drains played-tracks-and-artists needing similarity and POSTs // the results into track_similarity / artist_similarity. Failures are // passively retried via the timer (no durable queue table — losing one // tick's worth of refresh attempts is "1 hour of staleness," fine). type Worker struct { pool *pgxpool.Pool client *listenbrainz.Client logger *slog.Logger tick time.Duration batch int32 topK int } // NewWorker constructs a worker with production defaults: 1h tick, // batch=25, topK=20. batch is tracks AND artists processed per tick; // at 25/h a freshly-played library converges in hours, not days, while // staying well under ListenBrainz rate limits (429s abort the tick). func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker { return &Worker{ pool: pool, client: client, logger: logger, tick: 1 * time.Hour, batch: 25, topK: 20, } } // Run blocks until ctx is cancelled, ticking every w.tick. func (w *Worker) Run(ctx context.Context) { t := time.NewTicker(w.tick) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: if err := w.tickOnce(ctx); err != nil { w.logger.Error("similarity: tick failed", "err", err) } } } } // tickOnce drains one batch of tracks and one batch of artists. Per-row // errors are logged and skipped (passive retry via timer). 429 aborts the // entire tick. func (w *Worker) tickOnce(ctx context.Context) error { q := dbq.New(w.pool) if err := w.tickTracks(ctx, q); err != nil { return err } return w.tickArtists(ctx, q) } func (w *Worker) tickTracks(ctx context.Context, q *dbq.Queries) error { rows, err := q.ListPlayedTracksNeedingSimilarity(ctx, w.batch) if err != nil { return err } for _, r := range rows { if r.Mbid == nil { continue // defensive — query already filters NULL } results, err := w.client.SimilarRecordings(ctx, *r.Mbid, 100) if err != nil { var ra *listenbrainz.RetryAfterError if errors.As(err, &ra) { w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) return nil } w.logger.Warn("similarity: similar-recordings failed", "track_id", r.ID, "err", err) continue } w.upsertTrackSimilar(ctx, q, r.ID, results) } return nil } func (w *Worker) tickArtists(ctx context.Context, q *dbq.Queries) error { rows, err := q.ListPlayedArtistsNeedingSimilarity(ctx, w.batch) if err != nil { return err } for _, r := range rows { if r.Mbid == nil { continue } results, err := w.client.SimilarArtists(ctx, *r.Mbid, 100) if err != nil { var ra *listenbrainz.RetryAfterError if errors.As(err, &ra) { w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) return nil } w.logger.Warn("similarity: similar-artists failed", "artist_id", r.ID, "err", err) continue } w.upsertArtistSimilar(ctx, q, r.ID, results) } return nil } // upsertTrackSimilar filters returned MBIDs to those in our library, takes // top-K by score, and upserts rows. func (w *Worker) upsertTrackSimilar(ctx context.Context, q *dbq.Queries, trackAID pgtype.UUID, results []listenbrainz.SimilarRecording) { if len(results) == 0 { return } sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) mbids := make([]string, 0, len(results)) for _, r := range results { mbids = append(mbids, r.MBID) } rows, err := q.GetTracksByMBIDs(ctx, mbids) if err != nil { w.logger.Warn("similarity: GetTracksByMBIDs", "err", err) return } idByMBID := make(map[string]pgtype.UUID, len(rows)) for _, r := range rows { if r.Mbid != nil { idByMBID[*r.Mbid] = r.ID } } taken := 0 for _, r := range results { if taken >= w.topK { break } localID, ok := idByMBID[r.MBID] if !ok { continue } if localID == trackAID { continue // defensive — DB CHECK constraint also catches self-edges } if uerr := q.UpsertTrackSimilarity(ctx, dbq.UpsertTrackSimilarityParams{ TrackAID: trackAID, TrackBID: localID, Score: r.Score, }); uerr != nil { w.logger.Warn("similarity: UpsertTrackSimilarity", "err", uerr) continue } taken++ } } func (w *Worker) upsertArtistSimilar(ctx context.Context, q *dbq.Queries, artistAID pgtype.UUID, results []listenbrainz.SimilarArtist) { if len(results) == 0 { return } sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) mbids := make([]string, 0, len(results)) for _, r := range results { mbids = append(mbids, r.MBID) } rows, err := q.GetArtistsByMBIDs(ctx, mbids) if err != nil { w.logger.Warn("similarity: GetArtistsByMBIDs", "err", err) return } idByMBID := make(map[string]pgtype.UUID, len(rows)) for _, r := range rows { if r.Mbid != nil { idByMBID[*r.Mbid] = r.ID } } // Matched: in-library similars → artist_similarity (existing path). takenMatched := 0 for _, r := range results { if takenMatched >= w.topK { break } localID, ok := idByMBID[r.MBID] if !ok { continue } if localID == artistAID { continue } if uerr := q.UpsertArtistSimilarity(ctx, dbq.UpsertArtistSimilarityParams{ ArtistAID: artistAID, ArtistBID: localID, Score: r.Score, }); uerr != nil { w.logger.Warn("similarity: UpsertArtistSimilarity", "err", uerr) continue } takenMatched++ } // Unmatched: out-of-library similars → artist_similarity_unmatched (M5c). // Same top-K cap as the matched path. Skip rows missing a name — we can't // render a suggestion card without one. takenUnmatched := 0 for _, r := range results { if takenUnmatched >= w.topK { break } if _, inLib := idByMBID[r.MBID]; inLib { continue } if r.Name == "" { w.logger.Debug("similarity: skipping unmatched similar with empty name", "mbid", r.MBID) continue } if uerr := q.UpsertArtistSimilarityUnmatched(ctx, dbq.UpsertArtistSimilarityUnmatchedParams{ SeedArtistID: artistAID, CandidateMbid: r.MBID, CandidateName: r.Name, Score: r.Score, Source: "listenbrainz", }); uerr != nil { w.logger.Warn("similarity: UpsertArtistSimilarityUnmatched", "err", uerr) continue } takenUnmatched++ } }