navidrome/core/scrobbler/buffered_scrobbler_test.go

package scrobbler

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"testing/synctest"
	"time"

	"github.com/navidrome/navidrome/core/agents"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/tests"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
)

var _ = Describe("BufferedScrobbler", func() {
	var ds model.DataStore
	var scr *fakeScrobbler
	var bs *bufferedScrobbler
	var ctx context.Context
	var buffer *tests.MockedScrobbleBufferRepo

	BeforeEach(func() {
		ctx = context.Background()
		buffer = tests.CreateMockedScrobbleBufferRepo()
		userRepo := tests.CreateMockUserRepo()
		Expect(userRepo.Put(ctx, &model.User{ID: "user1", UserName: "alice"})).To(Succeed())
		ds = &tests.MockDataStore{
			MockedScrobbleBuffer: buffer,
			MockedUser:           userRepo,
		}
		scr = &fakeScrobbler{Authorized: true}
		bs = newBufferedScrobbler(ds, scr, "test")
	})

	It("forwards IsAuthorized calls", func() {
		scr.Authorized = true
		Expect(bs.IsAuthorized(ctx, "user1")).To(BeTrue())

		scr.Authorized = false
		Expect(bs.IsAuthorized(ctx, "user1")).To(BeFalse())
	})

	It("forwards NowPlaying calls", func() {
		track := &model.MediaFile{ID: "123", Title: "Test Track"}
		Expect(bs.NowPlaying(ctx, "user1", track, 0)).To(Succeed())
		Expect(scr.GetNowPlayingCalled()).To(BeTrue())
		Expect(scr.GetUserID()).To(Equal("user1"))
		Expect(scr.GetTrack()).To(Equal(track))
	})

	It("enqueues scrobbles to buffer", func() {
		track := model.MediaFile{ID: "123", Title: "Test Track"}
		now := time.Now()
		scrobble := Scrobble{MediaFile: track, TimeStamp: now}
		Expect(buffer.Length(ctx)).To(Equal(int64(0)))
		Expect(scr.ScrobbleCalled.Load()).To(BeFalse())

		Expect(bs.Scrobble(ctx, "user1", scrobble)).To(Succeed())

		// Wait for the background goroutine to process the scrobble.
		// We don't check buffer.Length() here because the background goroutine
		// may dequeue the entry before we can observe it.
		Eventually(scr.ScrobbleCalled.Load).Should(BeTrue())

		lastScrobble := scr.LastScrobble.Load()
		Expect(lastScrobble.MediaFile.ID).To(Equal("123"))
		Expect(lastScrobble.TimeStamp).To(BeTemporally("==", now))
	})

	It("restores the user in the context when draining buffered scrobbles", func() {
		track := model.MediaFile{ID: "123", Title: "Test Track", Artist: "Test Artist"}
		scrobble := Scrobble{MediaFile: track, TimeStamp: time.Now()}

		Expect(bs.Scrobble(ctx, "user1", scrobble)).To(Succeed())

		Eventually(scr.ScrobbleCalled.Load).Should(BeTrue())
		Expect(scr.GetUsername()).To(Equal("alice"))
	})

	It("stops the background goroutine when Stop is called", func() {
		// Replace the real run method with one that signals when it exits
		done := make(chan struct{})

		// Start our instrumented run function that will signal when it exits
		go func() {
			defer close(done)
			bs.run(bs.ctx)
		}()

		// Wait a bit to ensure the goroutine is running
		time.Sleep(10 * time.Millisecond)

		// Call the real Stop method
		bs.Stop()

		// Wait for the goroutine to exit or timeout
		select {
		case <-done:
			// Success, goroutine exited
		case <-time.After(100 * time.Millisecond):
			Fail("Goroutine did not exit in time after Stop was called")
		}
	})
})

var _ = Describe("backoffDelay", func() {
	DescribeTable("computes the exponential backoff curve clamped to the ceiling",
		func(failures int, expected time.Duration) {
			Expect(backoffDelay(failures)).To(Equal(expected))
		},
		Entry("first failure", 0, 5*time.Second),
		Entry("second failure", 1, 10*time.Second),
		Entry("third failure", 2, 20*time.Second),
		Entry("fourth failure", 3, 40*time.Second),
		Entry("fifth failure", 4, 80*time.Second),
		Entry("sixth failure", 5, 160*time.Second),
		Entry("reaches the ceiling", 6, 4*time.Minute),
		Entry("stays clamped past the ceiling", 7, 4*time.Minute),
		Entry("stays clamped for large values", 1000, 4*time.Minute),
		Entry("negative is treated as zero", -1, 5*time.Second),
	)
})

// Drives the real run loop and asserts the exact retry schedule + recovery. Plain
// test: testing/synctest's fake clock needs a *testing.T, which Ginkgo doesn't give.
func TestBufferedScrobblerBackoffSchedule(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		g := NewWithT(t)
		buffer := tests.CreateMockedScrobbleBufferRepo()
		userRepo := tests.CreateMockUserRepo()
		g.Expect(userRepo.Put(t.Context(), &model.User{ID: "user1", UserName: "alice"})).To(Succeed())
		ds := &tests.MockDataStore{MockedScrobbleBuffer: buffer, MockedUser: userRepo}

		flaky := &recoveringScrobbler{}
		flaky.fail(ErrRetryLater)
		bs := newBufferedScrobbler(ds, flaky, "flaky")
		defer func() { bs.Stop(); synctest.Wait() }()

		// Let the loop settle on the empty buffer, then enqueue a scrobble.
		synctest.Wait()
		track := model.MediaFile{ID: "123", Title: "Test Track", Artist: "Test Artist"}
		g.Expect(bs.Scrobble(context.Background(), "user1", Scrobble{MediaFile: track, TimeStamp: time.Now()})).To(Succeed())

		// First attempt fires immediately on the enqueue wake and is left buffered.
		synctest.Wait()
		g.Expect(flaky.count.Load()).To(Equal(int32(1)))
		g.Expect(buffer.Length(t.Context())).To(Equal(int64(1)))

		// Each subsequent retry waits exactly double the previous: 5s, 10s, 20s, 40s.
		for i, gap := range []time.Duration{5 * time.Second, 10 * time.Second, 20 * time.Second, 40 * time.Second} {
			want := int32(i + 2)
			time.Sleep(gap - time.Nanosecond)
			synctest.Wait()
			g.Expect(flaky.count.Load()).To(Equal(want-1), "retry fired before the %s backoff", gap)
			time.Sleep(time.Nanosecond)
			synctest.Wait()
			g.Expect(flaky.count.Load()).To(Equal(want), "retry did not fire after the %s backoff", gap)
		}

		// Once the service recovers, the buffered entry drains when the open
		// backoff window closes (a wake alone must not drain it early).
		flaky.succeed()
		bs.sendWakeSignal()
		synctest.Wait()
		g.Expect(buffer.Length(t.Context())).To(Equal(int64(1)), "wake during backoff drained early")
		time.Sleep(80 * time.Second)
		synctest.Wait()
		g.Expect(buffer.Length(t.Context())).To(Equal(int64(0)))
	})
}

func TestBufferedScrobblerBackoffWindow(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		buffer := tests.CreateMockedScrobbleBufferRepo()
		userRepo := tests.CreateMockUserRepo()
		_ = userRepo.Put(t.Context(), &model.User{ID: "user1", UserName: "alice"})
		ds := &tests.MockDataStore{MockedScrobbleBuffer: buffer, MockedUser: userRepo}
		scr := &fakeScrobbler{Authorized: true}
		scr.SetError(errors.Join(errors.New("boom"), ErrRetryLater))
		bs := newBufferedScrobbler(ds, scr, "test")
		defer bs.Stop()

		// First enqueue: one immediate attempt, then a 5s window opens.
		_ = bs.Scrobble(context.Background(), "user1", Scrobble{MediaFile: model.MediaFile{ID: "1"}, TimeStamp: time.Now()})
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 1 {
			t.Fatalf("expected 1 attempt after first enqueue, got %d", got)
		}

		// A wake inside the window must NOT trigger an early attempt.
		time.Sleep(1 * time.Second)
		_ = bs.Scrobble(context.Background(), "user1", Scrobble{MediaFile: model.MediaFile{ID: "2"}, TimeStamp: time.Now()})
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 1 {
			t.Fatalf("wake during backoff drained early: %d attempts", got)
		}

		// When the 5s window closes, the retry happens.
		time.Sleep(4100 * time.Millisecond)
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 2 {
			t.Fatalf("expected retry after window, got %d attempts", got)
		}
	})
}

func TestBufferedScrobblerHonorsServerDelay(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		buffer := tests.CreateMockedScrobbleBufferRepo()
		userRepo := tests.CreateMockUserRepo()
		_ = userRepo.Put(t.Context(), &model.User{ID: "user1", UserName: "alice"})
		ds := &tests.MockDataStore{MockedScrobbleBuffer: buffer, MockedUser: userRepo}
		scr := &fakeScrobbler{Authorized: true}
		scr.SetError(errors.Join(errors.New("429"), &agents.RetryLaterError{RetryIn: 30 * time.Second}))
		bs := newBufferedScrobbler(ds, scr, "test")
		defer bs.Stop()

		_ = bs.Scrobble(context.Background(), "user1", Scrobble{MediaFile: model.MediaFile{ID: "1"}, TimeStamp: time.Now()})
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 1 {
			t.Fatalf("expected 1 attempt, got %d", got)
		}

		// The 5s exponential floor is overridden by the 30s server delay.
		time.Sleep(20 * time.Second)
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 1 {
			t.Fatalf("retried before server delay elapsed: %d attempts", got)
		}
		time.Sleep(10100 * time.Millisecond)
		synctest.Wait()
		if got := scr.ScrobbleAttempts(); got != 2 {
			t.Fatalf("expected retry after server delay, got %d attempts", got)
		}
	})
}

// The drain visits users in an arbitrary order, so the longest delay must win regardless
// of which user was seen last.
func TestBufferedScrobblerTakesTheLongestServerDelayAcrossUsers(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		buffer := tests.CreateMockedScrobbleBufferRepo()
		userRepo := tests.CreateMockUserRepo()
		_ = userRepo.Put(t.Context(), &model.User{ID: "user1", UserName: "alice"})
		_ = userRepo.Put(t.Context(), &model.User{ID: "user2", UserName: "bob"})
		ds := &tests.MockDataStore{MockedScrobbleBuffer: buffer, MockedUser: userRepo}
		scr := &recoveringScrobbler{delays: map[string]time.Duration{
			"user1": 10 * time.Second,
			"user2": 45 * time.Second,
		}}
		// Both are buffered before the drain goroutine exists: it drains once on startup, and
		// seeing only one user there would park it on that user's delay, ignoring the other.
		_ = buffer.Enqueue(t.Context(), "test", "user1", "1", time.Now())
		_ = buffer.Enqueue(t.Context(), "test", "user2", "2", time.Now())
		bs := newBufferedScrobbler(ds, scr, "test")
		defer bs.Stop()

		synctest.Wait()
		if got := scr.count.Load(); got != 2 {
			t.Fatalf("expected both users drained, got %d attempts", got)
		}

		time.Sleep(30 * time.Second)
		synctest.Wait()
		if got := scr.count.Load(); got != 2 {
			t.Fatalf("retried on the shorter delay: %d attempts", got)
		}
		time.Sleep(15100 * time.Millisecond)
		synctest.Wait()
		if got := scr.count.Load(); got != 4 {
			t.Fatalf("expected a retry after the longest delay, got %d attempts", got)
		}
	})
}

// recoveringScrobbler is a race-safe Scrobbler whose error can be toggled while
// the buffered scrobbler's goroutine is draining, to exercise retry then recovery.
// With delays set, it instead fails every scrobble asking for that user's delay.
type recoveringScrobbler struct {
	err    atomic.Pointer[error]
	count  atomic.Int32
	delays map[string]time.Duration
}

func (f *recoveringScrobbler) fail(err error) { f.err.Store(&err) }
func (f *recoveringScrobbler) succeed()       { f.err.Store(nil) }

func (f *recoveringScrobbler) IsAuthorized(context.Context, string) bool { return true }

func (f *recoveringScrobbler) NowPlaying(context.Context, string, *model.MediaFile, int) error {
	return nil
}

func (f *recoveringScrobbler) Scrobble(_ context.Context, userId string, _ Scrobble) error {
	f.count.Add(1)
	if f.delays != nil {
		return errors.Join(errors.New("429"), &agents.RetryLaterError{RetryIn: f.delays[userId]})
	}
	if e := f.err.Load(); e != nil {
		return *e
	}
	return nil
}

func (f *recoveringScrobbler) PlaybackReport(context.Context, PlaybackSession) error { return nil }