minstrel/internal/similarity/worker.go

// 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++
	}
}