minstrel/internal/scrobble/worker.go

package scrobble

import (
	"context"
	"errors"
	"log/slog"
	"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"
	"git.fabledsword.com/bvandeusen/minstrel/internal/secret"
)

// backoffSchedule maps the failure count (1-indexed) to the delay before
// the next attempt: 1m → 5m → 30m → 2h → 6h, then give up.
var backoffSchedule = []time.Duration{
	1 * time.Minute,
	5 * time.Minute,
	30 * time.Minute,
	2 * time.Hour,
	6 * time.Hour,
}

const maxAttempts = 5

// backoffDelay returns the delay before the next attempt given the value
// of the `attempts` column AFTER the failure has been recorded (1-indexed:
// attempts=1 means the first failure just happened, the next retry should
// wait 1 minute). Returns (_, false) when attempts > maxAttempts (the
// caller should mark the row failed).
func backoffDelay(attempts int) (time.Duration, bool) {
	if attempts < 1 || attempts > maxAttempts {
		return 0, false
	}
	return backoffSchedule[attempts-1], true
}

// Worker drains scrobble_queue rows and POSTs them to ListenBrainz.
type Worker struct {
	pool   *pgxpool.Pool
	client *listenbrainz.Client
	logger *slog.Logger
	tick   time.Duration
	batch  int32
}

// NewWorker constructs a worker with production defaults: 30s tick, 50-row
// batch.
func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker {
	return &Worker{pool: pool, client: client, logger: logger, tick: 30 * time.Second, batch: 50}
}

// 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("scrobble: tick failed", "err", err)
			}
		}
	}
}

// tickOnce drains up to w.batch pending rows. Returns the first error
// encountered loading rows; per-row errors are logged and don't abort
// the batch.
func (w *Worker) tickOnce(ctx context.Context) error {
	q := dbq.New(w.pool)
	rows, err := q.ListPendingScrobbles(ctx, w.batch)
	if err != nil {
		return err
	}
	if len(rows) == 0 {
		return nil
	}

	// Group rows by user (each user has their own token + may produce its
	// own batch). Map keyed by user UUID string.
	type userBatch struct {
		token    string
		queueIDs []pgtype.UUID
		listens  []listenbrainz.Listen
		userID   pgtype.UUID
	}
	batches := map[string]*userBatch{}
	for _, r := range rows {
		if r.LbToken == nil || *r.LbToken == "" || !r.LbEnabled {
			// Defensive: row got into queue but user since disabled. Skip.
			continue
		}
		key := r.UserID.String()
		ub := batches[key]
		if ub == nil {
			token := secret.Reveal(secret.ListenBrainzToken.Purpose(), key, *r.LbToken)
			if token == "" {
				continue
			}
			ub = &userBatch{token: token, userID: r.UserID}
			batches[key] = ub
		}
		ub.queueIDs = append(ub.queueIDs, r.QueueID)
		l := listenbrainz.Listen{
			ListenedAt: r.StartedAt.Time.Unix(),
			Track: listenbrainz.Track{
				ArtistName:  r.ArtistName,
				TrackName:   r.TrackTitle,
				ReleaseName: r.AlbumTitle,
				DurationMs:  int(r.TrackDurationMs),
			},
		}
		if r.TrackMbid != nil {
			l.Track.RecordingMBID = *r.TrackMbid
		}
		if r.AlbumMbid != nil {
			l.Track.ReleaseMBID = *r.AlbumMbid
		}
		if r.ArtistMbid != nil {
			l.Track.ArtistMBIDs = []string{*r.ArtistMbid}
		}
		ub.listens = append(ub.listens, l)
	}

	for _, ub := range batches {
		w.processBatch(ctx, q, ub.userID, ub.token, ub.queueIDs, ub.listens)
	}
	return nil
}

// processBatch submits one user's pending batch and updates queue rows
// according to the response.
func (w *Worker) processBatch(
	ctx context.Context,
	q *dbq.Queries,
	userID pgtype.UUID,
	token string,
	queueIDs []pgtype.UUID,
	listens []listenbrainz.Listen,
) {
	err := w.client.SubmitListens(ctx, token, listens)
	now := time.Now().UTC()

	if err == nil {
		for _, id := range queueIDs {
			if uerr := q.MarkScrobbleSent(ctx, id); uerr != nil {
				w.logger.Error("scrobble: MarkScrobbleSent", "queue_id", id, "err", uerr)
			}
		}
		return
	}

	// 429: schedule retry per Retry-After, do NOT increment attempts.
	var ra *listenbrainz.RetryAfterError
	if errors.As(err, &ra) {
		next := pgtype.Timestamptz{Time: now.Add(ra.Wait), Valid: true}
		for _, id := range queueIDs {
			if uerr := q.MarkScrobbleRetryAfter(ctx, dbq.MarkScrobbleRetryAfterParams{
				ID: id, NextAttemptAt: next,
			}); uerr != nil {
				w.logger.Error("scrobble: MarkScrobbleRetryAfter", "err", uerr)
			}
		}
		return
	}

	// 401: mark failed AND disable user (defensive — bad token shouldn't keep
	// retrying or feed future rows).
	if errors.Is(err, listenbrainz.ErrAuth) {
		w.logger.Warn("scrobble: auth rejected; disabling user listenbrainz", "user_id", userID)
		if uerr := q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{
			ID: userID, ListenbrainzEnabled: false,
		}); uerr != nil {
			w.logger.Error("scrobble: SetListenBrainzEnabled", "err", uerr)
		}
		errStr := err.Error()
		for _, id := range queueIDs {
			if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
				ID: id, LastError: &errStr,
			}); uerr != nil {
				w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
			}
		}
		return
	}

	// 4xx: permanent. Mark failed, no retry.
	if errors.Is(err, listenbrainz.ErrPermanent) {
		errStr := err.Error()
		for _, id := range queueIDs {
			if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
				ID: id, LastError: &errStr,
			}); uerr != nil {
				w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
			}
		}
		return
	}

	// Transient (5xx, network): per-row, look up backoffDelay(currentAttempts+1).
	// If OK → RescheduleScrobble (which increments). If NOT OK → MarkScrobbleFailed
	// (don't increment further; current attempts already represents the cap).
	errStr := err.Error()
	for _, id := range queueIDs {
		var attempts int32
		if rerr := w.pool.QueryRow(ctx,
			`SELECT attempts FROM scrobble_queue WHERE id = $1`, id).Scan(&attempts); rerr != nil {
			w.logger.Error("scrobble: read attempts", "err", rerr)
			continue
		}
		if delay, ok := backoffDelay(int(attempts) + 1); ok {
			nextAt := pgtype.Timestamptz{Time: now.Add(delay), Valid: true}
			if uerr := q.RescheduleScrobble(ctx, dbq.RescheduleScrobbleParams{
				ID: id, NextAttemptAt: nextAt, LastError: &errStr,
			}); uerr != nil {
				w.logger.Error("scrobble: RescheduleScrobble", "err", uerr)
			}
		} else {
			if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
				ID: id, LastError: &errStr,
			}); uerr != nil {
				w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
			}
		}
	}
}