From 9490bad36054be0de62d0c820fe110e4a988de91 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 09:20:23 -0400 Subject: [PATCH] feat(scrobble): add Worker with tickOnce, batched submit, retry/fail/auth handling Co-Authored-By: Claude Sonnet 4.6 --- internal/scrobble/worker.go | 203 ++++++++++++++++- internal/scrobble/worker_integration_test.go | 224 +++++++++++++++++++ 2 files changed, 424 insertions(+), 3 deletions(-) create mode 100644 internal/scrobble/worker_integration_test.go diff --git a/internal/scrobble/worker.go b/internal/scrobble/worker.go index 89298602..3a436158 100644 --- a/internal/scrobble/worker.go +++ b/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 -// just happened) to the delay before the next attempt. Per spec §M4a: -// 1m → 5m → 30m → 2h → 6h, then give up. +// backoffSchedule maps the failure count (1-indexed) to the delay before +// the next attempt. Per spec §M4a: 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) + } + } + } +} diff --git a/internal/scrobble/worker_integration_test.go b/internal/scrobble/worker_integration_test.go new file mode 100644 index 00000000..b842863f --- /dev/null +++ b/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()) + } +}