225 lines
7.8 KiB
Go
225 lines
7.8 KiB
Go
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())
|
|
}
|
|
}
|