feat(scrobble): add Worker with tickOnce, batched submit, retry/fail/auth handling

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

internal/scrobble/worker.go

@@ -1,12 +1,20 @@
 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"
 )
 
-// backoffSchedule maps the failure count (1-indexed: 1 = first failure
+// backoffSchedule maps the failure count (1-indexed) to the delay before
-// just happened) to the delay before the next attempt. Per spec §M4a:
+// the next attempt. Per spec §M4a: 1m → 5m → 30m → 2h → 6h, then give up.
-// 1m → 5m → 30m → 2h → 6h, then give up.
 var backoffSchedule = []time.Duration{
 	1 * time.Minute,
 	5 * time.Minute,
@@ -28,3 +36,192 @@ func backoffDelay(attempts int) (time.Duration, bool) {
 	}
 	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 {
+			ub = &userBatch{token: *r.LbToken, 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)
+			}
+		}
+	}
+}

internal/scrobble/worker_integration_test.go

@@ -0,0 +1,224 @@
+package scrobble
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz"
+)
+
+// helperEnqueue inserts a queue row directly so we don't go through the
+// threshold path. Reuses the `setup` struct from queue_test.go.
+func helperEnqueue(t *testing.T, s setup) {
+	t.Helper()
+	if _, err := s.pool.Exec(context.Background(),
+		`INSERT INTO scrobble_queue (user_id, play_event_id, status) VALUES ($1, $2, 'pending')`,
+		s.user, s.playEvent); err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+}
+
+func helperPendingCount(t *testing.T, s setup) int {
+	t.Helper()
+	var n int
+	if err := s.pool.QueryRow(context.Background(),
+		`SELECT count(*) FROM scrobble_queue WHERE status = 'pending' AND play_event_id = $1`, s.playEvent).Scan(&n); err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	return n
+}
+
+func helperFailedCount(t *testing.T, s setup) int {
+	t.Helper()
+	var n int
+	if err := s.pool.QueryRow(context.Background(),
+		`SELECT count(*) FROM scrobble_queue WHERE status = 'failed' AND play_event_id = $1`, s.playEvent).Scan(&n); err != nil {
+		t.Fatalf("failed: %v", err)
+	}
+	return n
+}
+
+func helperPlayEventScrobbledAt(t *testing.T, s setup) bool {
+	t.Helper()
+	var v *time.Time
+	if err := s.pool.QueryRow(context.Background(),
+		`SELECT scrobbled_at FROM play_events WHERE id = $1`, s.playEvent).Scan(&v); err != nil {
+		t.Fatalf("scrobbled_at: %v", err)
+	}
+	return v != nil
+}
+
+func helperLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
+
+func newTestWorker(s setup, lb *listenbrainz.Client) *Worker {
+	return NewWorker(s.pool, lb, helperLogger())
+}
+
+func TestWorker_Success_DeletesQueueRowAndStampsScrobbledAt(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	helperEnqueue(t, s)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+
+	if err := w.tickOnce(context.Background()); err != nil {
+		t.Fatalf("tickOnce: %v", err)
+	}
+	if helperPendingCount(t, s) != 0 {
+		t.Error("queue row should be deleted on success")
+	}
+	if helperFailedCount(t, s) != 0 {
+		t.Error("no failed row expected on success")
+	}
+	if !helperPlayEventScrobbledAt(t, s) {
+		t.Error("play_events.scrobbled_at should be populated on success")
+	}
+}
+
+func TestWorker_503Once_RowRemainsAttempts1(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	helperEnqueue(t, s)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+	_ = w.tickOnce(context.Background())
+
+	var attempts int
+	if err := s.pool.QueryRow(context.Background(),
+		`SELECT attempts FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).Scan(&attempts); err != nil {
+		t.Fatalf("attempts: %v", err)
+	}
+	if attempts != 1 {
+		t.Errorf("attempts = %d, want 1 after 503", attempts)
+	}
+	if helperPendingCount(t, s) != 1 {
+		t.Error("row should remain pending after transient failure")
+	}
+}
+
+// TestWorker_503SixTimes_MarksFailed: 6 ticks of 503. Tick 1-5 reschedule
+// (attempts goes 1,2,3,4,5; backoffDelay returns OK for those). Tick 6
+// hits backoffDelay(6) = false and marks failed (attempts stays at 5).
+func TestWorker_503SixTimes_MarksFailed(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	helperEnqueue(t, s)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+	for i := 0; i < 6; i++ {
+		if _, err := s.pool.Exec(context.Background(),
+			`UPDATE scrobble_queue SET next_attempt_at = now() WHERE play_event_id = $1`, s.playEvent); err != nil {
+			t.Fatalf("reset: %v", err)
+		}
+		_ = w.tickOnce(context.Background())
+	}
+	if helperFailedCount(t, s) != 1 {
+		t.Errorf("expected row marked failed after 6 attempts; pending=%d failed=%d",
+			helperPendingCount(t, s), helperFailedCount(t, s))
+	}
+	var attempts int
+	var lastErr *string
+	_ = s.pool.QueryRow(context.Background(),
+		`SELECT attempts, last_error FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).Scan(&attempts, &lastErr)
+	if attempts != 5 {
+		t.Errorf("attempts = %d, want 5 (max retries reached)", attempts)
+	}
+	if lastErr == nil || *lastErr == "" {
+		t.Error("last_error should be populated")
+	}
+}
+
+func TestWorker_401_MarksFailedAndDisablesUser(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	helperEnqueue(t, s)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+	_ = w.tickOnce(context.Background())
+
+	if helperFailedCount(t, s) != 1 {
+		t.Error("401 should mark failed immediately")
+	}
+	cfg, _ := s.q.GetListenBrainzConfig(context.Background(), s.user)
+	if cfg.ListenbrainzEnabled {
+		t.Error("user listenbrainz_enabled should be set to false on auth failure")
+	}
+}
+
+func TestWorker_429_RespectsRetryAfterWithoutIncrementingAttempts(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	helperEnqueue(t, s)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Retry-After", "300")
+		w.WriteHeader(http.StatusTooManyRequests)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+	_ = w.tickOnce(context.Background())
+
+	var attempts int
+	var nextAttemptAt time.Time
+	_ = s.pool.QueryRow(context.Background(),
+		`SELECT attempts, next_attempt_at FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).
+		Scan(&attempts, &nextAttemptAt)
+	if attempts != 0 {
+		t.Errorf("attempts = %d, want 0 (429 should not increment)", attempts)
+	}
+	if d := time.Until(nextAttemptAt); d < 250*time.Second || d > 350*time.Second {
+		t.Errorf("next_attempt_at = +%v, want ~+5m (Retry-After 300)", d)
+	}
+}
+
+func TestWorker_BatchOfTen_OnePost(t *testing.T) {
+	s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
+	// Enqueue 10 distinct play_events to test batching. We synthesize new
+	// play_events by copying the seeded one; each gets its own queue row.
+	for i := 0; i < 10; i++ {
+		var newPE pgtype.UUID
+		if err := s.pool.QueryRow(context.Background(),
+			`INSERT INTO play_events (user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped)
+			 SELECT user_id, track_id, session_id, now(), now(), duration_played_ms, completion_ratio, was_skipped FROM play_events WHERE id = $1
+			 RETURNING id`, s.playEvent).Scan(&newPE); err != nil {
+			t.Fatalf("dup play_event: %v", err)
+		}
+		if _, err := s.pool.Exec(context.Background(),
+			`INSERT INTO scrobble_queue (user_id, play_event_id, status) VALUES ($1, $2, 'pending')`,
+			s.user, newPE); err != nil {
+			t.Fatalf("enqueue: %v", err)
+		}
+	}
+	var posts atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		posts.Add(1)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
+	w := newTestWorker(s, lb)
+	_ = w.tickOnce(context.Background())
+
+	if posts.Load() != 1 {
+		t.Errorf("LB POSTs = %d, want 1 (batch)", posts.Load())
+	}
+}