navidrome/core/scrobbler/buffered_scrobbler.go

package scrobbler

import (
	"context"
	"errors"
	"time"

	"github.com/navidrome/navidrome/core/agents"
	"github.com/navidrome/navidrome/log"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/model/request"
)

const (
	minRetryDelay = 5 * time.Second
	maxRetryDelay = 4 * time.Minute
	// maxRetryShift caps the exponent so the shift never overflows int64.
	// minRetryDelay<<6 = 320s already exceeds maxRetryDelay, so 6 reaches the ceiling.
	maxRetryShift = 6
)

// backoffDelay returns the delay for a zero-based retry index (0 = first retry):
// minRetryDelay doubled per prior failure, clamped to maxRetryDelay.
func backoffDelay(failures int) time.Duration {
	if failures < 0 {
		failures = 0
	}
	if failures >= maxRetryShift {
		return maxRetryDelay
	}
	d := minRetryDelay << failures
	if d > maxRetryDelay {
		return maxRetryDelay
	}
	return d
}

// Loader is a function that loads a scrobbler by name.
// It returns the scrobbler and true if found, or nil and false if not available.
// This allows the buffered scrobbler to always get the current plugin instance.
type Loader func() (Scrobbler, bool)

// newBufferedScrobbler creates a buffered scrobbler that wraps a static scrobbler instance.
// Use this for builtin scrobblers that don't change.
func newBufferedScrobbler(ds model.DataStore, s Scrobbler, service string) *bufferedScrobbler {
	return newBufferedScrobblerWithLoader(ds, service, func() (Scrobbler, bool) {
		return s, true
	})
}

// newBufferedScrobblerWithLoader creates a buffered scrobbler that dynamically loads
// the underlying scrobbler on each call. Use this for plugin scrobblers that may be
// reloaded (e.g., after configuration changes).
func newBufferedScrobblerWithLoader(ds model.DataStore, service string, loader Loader) *bufferedScrobbler {
	ctx, cancel := context.WithCancel(context.Background())
	b := &bufferedScrobbler{
		ds:         ds,
		loader:     loader,
		service:    service,
		wakeSignal: make(chan struct{}, 1),
		ctx:        ctx,
		cancel:     cancel,
	}
	go b.run(ctx)
	return b
}

type bufferedScrobbler struct {
	ds         model.DataStore
	loader     Loader
	service    string
	wakeSignal chan struct{}
	ctx        context.Context //nolint:containedctx // scrobbler lifecycle ctx, cancelled by Stop
	cancel     context.CancelFunc
}

func (b *bufferedScrobbler) Stop() {
	if b.cancel != nil {
		b.cancel()
	}
}

func (b *bufferedScrobbler) IsAuthorized(ctx context.Context, userId string) bool {
	s, ok := b.loader()
	if !ok {
		return false
	}
	return s.IsAuthorized(ctx, userId)
}

func (b *bufferedScrobbler) NowPlaying(ctx context.Context, userId string, track *model.MediaFile, position int) error {
	s, ok := b.loader()
	if !ok {
		return errors.New("scrobbler not available")
	}
	return s.NowPlaying(ctx, userId, track, position)
}

func (b *bufferedScrobbler) Scrobble(ctx context.Context, userId string, s Scrobble) error {
	err := b.ds.ScrobbleBuffer().Enqueue(ctx, b.service, userId, s.ID, s.TimeStamp)
	if err != nil {
		return err
	}

	b.sendWakeSignal()
	return nil
}

func (b *bufferedScrobbler) PlaybackReport(ctx context.Context, info PlaybackSession) error {
	s, ok := b.loader()
	if !ok {
		return errors.New("scrobbler not available")
	}
	return s.PlaybackReport(ctx, info)
}

func (b *bufferedScrobbler) sendWakeSignal() {
	// Don't block if the previous signal was not read yet
	select {
	case b.wakeSignal <- struct{}{}:
	default:
	}
}

func (b *bufferedScrobbler) run(ctx context.Context) {
	timer := time.NewTimer(time.Hour)
	timer.Stop()
	defer timer.Stop()
	failures := 0
	backingOff := false
	for {
		// While a backoff window is open the timer is already armed for the rest of it, so a
		// wake (a new play enqueued) must not drain: that is the hammering this avoids.
		if !backingOff {
			if ok, retryIn := b.processQueue(ctx); ok {
				failures = 0
				timer.Stop()
			} else {
				timer.Reset(max(backoffDelay(failures), retryIn))
				backingOff = true
				if failures < maxRetryShift {
					failures++
				}
			}
		}
		select {
		case <-b.wakeSignal:
		case <-timer.C:
			backingOff = false
		case <-ctx.Done():
			return
		}
	}
}

func (b *bufferedScrobbler) processQueue(ctx context.Context) (bool, time.Duration) {
	buffer := b.ds.ScrobbleBuffer()
	userIds, err := buffer.UserIDs(ctx, b.service)
	if err != nil {
		log.Error(ctx, "Error retrieving userIds from scrobble buffer", "scrobbler", b.service, err)
		return false, 0
	}
	result := true
	var retryIn time.Duration
	for _, userId := range userIds {
		ok, d := b.processUserQueue(ctx, userId)
		if !ok {
			result = false
			retryIn = max(retryIn, d)
		}
	}
	return result, retryIn
}

func (b *bufferedScrobbler) processUserQueue(ctx context.Context, userId string) (bool, time.Duration) {
	// Scrobbles are drained on a background context that no longer carries the
	// request's authenticated user. Restore it from the buffered userId so that
	// scrobblers relying on the user in the context (e.g. plugins) still get it.
	if user, err := b.ds.User().Get(ctx, userId); err != nil {
		log.Warn(ctx, "Could not load user for buffered scrobble", "userId", userId, "scrobbler", b.service, err)
	} else {
		ctx = request.WithUser(ctx, *user)
	}
	buffer := b.ds.ScrobbleBuffer()
	for {
		entry, err := buffer.Next(ctx, b.service, userId)
		if err != nil {
			log.Error(ctx, "Error reading from scrobble buffer", "scrobbler", b.service, err)
			return false, 0
		}
		if entry == nil {
			return true, 0
		}
		s, ok := b.loader()
		if !ok {
			log.Warn(ctx, "Scrobbler not available, will retry later", "scrobbler", b.service)
			return false, 0
		}
		log.Debug(ctx, "Sending scrobble", "scrobbler", b.service, "track", entry.Title, "artist", entry.Artist)
		err = s.Scrobble(ctx, entry.UserID, Scrobble{
			MediaFile: entry.MediaFile,
			TimeStamp: entry.PlayTime,
		})
		if retry, ok := errors.AsType[*agents.RetryLaterError](err); ok {
			log.Warn(ctx, "Could not send scrobble. Will be retried", "userId", entry.UserID,
				"track", entry.Title, "artist", entry.Artist, "scrobbler", b.service, err)
			return false, retry.RetryIn
		}
		if err != nil {
			log.Error(ctx, "Error sending scrobble to service. Discarding", "scrobbler", b.service,
				"userId", entry.UserID, "artist", entry.Artist, "track", entry.Title, err)
		}
		err = buffer.Dequeue(ctx, entry)
		if err != nil {
			log.Error(ctx, "Error removing entry from scrobble buffer", "userId", entry.UserID,
				"track", entry.Title, "artist", entry.Artist, "scrobbler", b.service, err)
			return false, 0
		}
	}
}

var _ Scrobbler = (*bufferedScrobbler)(nil)