navidrome/core/scrobbler/play_tracker.go

package scrobbler

import (
	"context"
	"encoding/json"
	"maps"
	"slices"
	"sync"
	"time"

	"github.com/navidrome/navidrome/conf"
	"github.com/navidrome/navidrome/consts"
	"github.com/navidrome/navidrome/log"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/model/criteria"
	"github.com/navidrome/navidrome/model/request"
	"github.com/navidrome/navidrome/server/events"
	"github.com/navidrome/navidrome/utils/cache"
	"github.com/navidrome/navidrome/utils/singleton"
	"github.com/navidrome/navidrome/utils/slice"
)

const (
	StateStarting = "starting"
	StatePlaying  = "playing"
	StatePaused   = "paused"
	StateStopped  = "stopped"
	StateExpired  = "expired"
)

var ValidStates = map[string]bool{
	StateStarting: true,
	StatePlaying:  true,
	StatePaused:   true,
	StateStopped:  true,
}

type PlaybackSession struct {
	MediaFile    model.MediaFile
	Start        time.Time
	UserId       string
	Username     string
	PlayerId     string
	PlayerName   string
	State        string
	PositionMs   int64
	PlaybackRate float64
	LastReport   time.Time

	// Verdict from the last report, for the expiry callback: its context carries only
	// a stub user, so it cannot evaluate the filter itself.
	filtered bool
}

type Submission struct {
	TrackID   string
	Timestamp time.Time
}

type ReportPlaybackParams struct {
	MediaId        string
	PositionMs     int64
	State          string
	PlaybackRate   float64
	IgnoreScrobble bool
	ClientId       string
	ClientName     string
}

type nowPlayingEntry struct {
	ctx      context.Context //nolint:containedctx // queued work item carries the request ctx to the worker
	userId   string
	track    *model.MediaFile
	position int
}

type playbackReportEntry struct {
	ctx      context.Context //nolint:containedctx // queued work item carries the request ctx to the worker
	info     PlaybackSession
	filtered bool
}

type PlayTracker interface {
	GetNowPlaying(ctx context.Context) ([]PlaybackSession, error)
	Submit(ctx context.Context, submissions []Submission) error
	ReportPlayback(ctx context.Context, params ReportPlaybackParams) error
}

// PluginLoader is a minimal interface for plugin manager usage in PlayTracker
// (avoids import cycles)
type PluginLoader interface {
	PluginNames(capability string) []string
	LoadScrobbler(name string) (Scrobbler, bool)
}

type playTracker struct {
	ds                model.DataStore
	broker            events.Broker
	playMap           cache.SimpleCache[string, PlaybackSession]
	sessionsMu        sync.Mutex // serializes playMap check-then-write across concurrent reports
	builtinScrobblers map[string]Scrobbler
	pluginScrobblers  map[string]Scrobbler
	pluginLoader      PluginLoader
	mu                sync.RWMutex
	npQueue           map[string]nowPlayingEntry
	npMu              sync.Mutex
	npSignal          chan struct{}
	shutdown          chan struct{}
	workerDone        chan struct{}
	prQueue           []playbackReportEntry
	prMu              sync.Mutex
	prSignal          chan struct{}
	prWorkerDone      chan struct{}
}

func GetPlayTracker(ds model.DataStore, broker events.Broker, pluginManager PluginLoader) PlayTracker {
	return singleton.GetInstance(func() *playTracker {
		return newPlayTracker(ds, broker, pluginManager)
	})
}

// NewPlayTracker creates a new PlayTracker instance. For normal usage, the PlayTracker has to be a singleton,
// returned by the GetPlayTracker function above. This constructor is exported for testing.
func NewPlayTracker(ds model.DataStore, broker events.Broker, pluginManager PluginLoader) PlayTracker {
	return newPlayTracker(ds, broker, pluginManager)
}

func newPlayTracker(ds model.DataStore, broker events.Broker, pluginManager PluginLoader) *playTracker {
	m := cache.NewSimpleCache[string, PlaybackSession]()
	p := &playTracker{
		ds:                ds,
		playMap:           m,
		broker:            broker,
		builtinScrobblers: make(map[string]Scrobbler),
		pluginScrobblers:  make(map[string]Scrobbler),
		pluginLoader:      pluginManager,
		npQueue:           make(map[string]nowPlayingEntry),
		npSignal:          make(chan struct{}, 1),
		shutdown:          make(chan struct{}),
		workerDone:        make(chan struct{}),
		prSignal:          make(chan struct{}, 1),
		prWorkerDone:      make(chan struct{}),
	}
	enableNowPlaying := conf.Server.EnableNowPlaying
	m.OnExpiration(func(_ string, info PlaybackSession) {
		log.Debug("PlaybackSession expired", "clientId", info.PlayerId, "mediaId", info.MediaFile.ID, "state",
			info.State, "username", info.Username, "userId", info.UserId)
		if enableNowPlaying {
			broker.SendBroadcastMessage(context.Background(), &events.NowPlayingCount{Count: m.Len()})
		}
		ctx := request.WithUser(context.Background(), model.User{ID: info.UserId, UserName: info.Username})
		if info.State != StateStopped {
			log.Trace("Enqueueing PlaybackReport for expired session", "session", info)
			info.State = StateExpired
			info.LastReport = time.Now()
			p.enqueuePlaybackReport(ctx, info, info.filtered)
		}
	})

	var enabled []string
	for name, constructor := range constructors {
		s := constructor(ds)
		if s == nil {
			log.Debug("Scrobbler not available. Missing configuration?", "name", name)
			continue
		}
		enabled = append(enabled, name)
		s = newBufferedScrobbler(ds, s, name)
		p.builtinScrobblers[name] = s
	}
	log.Debug("List of builtin scrobblers enabled", "names", enabled)
	go p.nowPlayingWorker()
	go p.playbackReportWorker()
	return p
}

// stopBackgroundWorkers stops the background workers. This is primarily for testing.
func (p *playTracker) stopBackgroundWorkers() {
	close(p.shutdown)
	<-p.workerDone   // Wait for nowPlaying worker to finish
	<-p.prWorkerDone // Wait for playbackReport worker to finish
}

// pluginNamesMatchScrobblers returns true if the set of pluginNames matches the keys in pluginScrobblers.
func pluginNamesMatchScrobblers(pluginNames []string, scrobblers map[string]Scrobbler) bool {
	if len(pluginNames) != len(scrobblers) {
		return false
	}
	for _, name := range pluginNames {
		if _, ok := scrobblers[name]; !ok {
			return false
		}
	}
	return true
}

// refreshPluginScrobblers updates the pluginScrobblers map to match the current set of plugin scrobblers.
// The buffered scrobblers use a loader function to dynamically get the current plugin instance,
// so we only need to add/remove scrobblers when plugins are added/removed (not when reloaded).
func (p *playTracker) refreshPluginScrobblers() {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.pluginLoader == nil {
		return
	}

	// Get the list of available plugin names
	pluginNames := p.pluginLoader.PluginNames("Scrobbler")

	// Early return if plugin names match existing scrobblers (no change)
	if pluginNamesMatchScrobblers(pluginNames, p.pluginScrobblers) {
		return
	}

	// Build a set of current plugins for faster lookups
	current := make(map[string]struct{}, len(pluginNames))

	// Process additions - add new plugins with a loader that dynamically fetches the current instance
	for _, name := range pluginNames {
		current[name] = struct{}{}
		if _, exists := p.pluginScrobblers[name]; !exists {
			// Capture the name for the closure
			pluginName := name
			loader := p.pluginLoader
			p.pluginScrobblers[name] = newBufferedScrobblerWithLoader(p.ds, name, func() (Scrobbler, bool) {
				return loader.LoadScrobbler(pluginName)
			})
		}
	}

	type stoppableScrobbler interface {
		Scrobbler
		Stop()
	}

	// Process removals - remove plugins that no longer exist
	for name, scrobbler := range p.pluginScrobblers {
		if _, exists := current[name]; !exists {
			// If the scrobbler implements stoppableScrobbler, call Stop() before removing it
			if stoppable, ok := scrobbler.(stoppableScrobbler); ok {
				log.Debug("Stopping scrobbler", "name", name)
				stoppable.Stop()
			}
			delete(p.pluginScrobblers, name)
		}
	}
}

// getActiveScrobblers refreshes plugin scrobblers, acquires a read lock,
// combines builtin and plugin scrobblers into a new map, releases the lock,
// and returns the combined map.
func (p *playTracker) getActiveScrobblers() map[string]Scrobbler {
	p.refreshPluginScrobblers()
	p.mu.RLock()
	defer p.mu.RUnlock()
	combined := maps.Clone(p.builtinScrobblers)
	maps.Copy(combined, p.pluginScrobblers)
	return combined
}

// hasPlayingSession reports whether clientId's current session is already playing mediaId.
func (p *playTracker) hasPlayingSession(clientId, mediaId string) bool {
	cur, err := p.playMap.Get(clientId)
	return err == nil && cur.MediaFile.ID == mediaId && cur.State == StatePlaying
}

func remainingTTL(durationSec float32, positionMs int64, rate float64) time.Duration {
	if rate <= 0 {
		rate = 1.0
	}
	remainingMs := float64(int64(durationSec*1000)-positionMs) / rate
	remainingSec := max(int(remainingMs/1000), 0)
	return time.Duration(remainingSec+5) * time.Second
}

func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackParams) error {
	player, _ := request.PlayerFrom(ctx)
	user, _ := request.UserFrom(ctx)
	clientId := params.ClientId
	client := params.ClientName

	now := time.Now()

	// One verdict per report, reused by every dispatch below, so a filter reading
	// annotations cannot decide differently on either side of incPlay.
	var filtered bool

	switch params.State {
	case StateStarting:
		// Clients may send starting/playing unordered; a late "starting" must not downgrade
		// a playing session, or position estimation freezes until the next report.
		if p.hasPlayingSession(clientId, params.MediaId) {
			log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
			return nil
		}
		mf, err := p.ds.MediaFile().GetWithParticipants(ctx, params.MediaId)
		if err != nil {
			return err
		}
		filtered = p.isFilteredOut(ctx, mf)
		info := PlaybackSession{
			MediaFile:    *mf,
			filtered:     filtered,
			Start:        now,
			UserId:       user.ID,
			Username:     user.UserName,
			PlayerId:     clientId,
			PlayerName:   client,
			State:        params.State,
			PositionMs:   params.PositionMs,
			PlaybackRate: params.PlaybackRate,
			LastReport:   now,
		}
		p.sessionsMu.Lock()
		// re-check: a concurrent "playing" report may have created the session during the load above
		if p.hasPlayingSession(clientId, params.MediaId) {
			p.sessionsMu.Unlock()
			log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
			return nil
		}
		err = p.playMap.AddWithTTL(clientId, info, remainingTTL(mf.Duration, params.PositionMs, params.PlaybackRate))
		p.sessionsMu.Unlock()
		if err != nil {
			log.Warn(ctx, "Error adding PlaybackSession to cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
		}
		p.enqueuePlaybackReport(ctx, info, filtered)

	case StatePlaying, StatePaused:
		info, getErr := p.playMap.Get(clientId)
		if getErr != nil || info.MediaFile.ID != params.MediaId {
			mf, err := p.ds.MediaFile().GetWithParticipants(ctx, params.MediaId)
			if err != nil {
				return err
			}
			info = PlaybackSession{
				MediaFile:  *mf,
				Start:      now.Add(-time.Duration(params.PositionMs) * time.Millisecond),
				UserId:     user.ID,
				Username:   user.UserName,
				PlayerId:   clientId,
				PlayerName: client,
			}
		}
		info.State = params.State
		info.PositionMs = params.PositionMs
		info.PlaybackRate = params.PlaybackRate
		info.LastReport = now
		filtered = p.isFilteredOut(ctx, &info.MediaFile)
		info.filtered = filtered
		ttl := 30 * time.Minute
		if params.State == StatePlaying {
			ttl = remainingTTL(info.MediaFile.Duration, params.PositionMs, params.PlaybackRate)
		}
		log.Trace(ctx, "Updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, "positionMs", params.PositionMs, "playbackRate", params.PlaybackRate, "ttl", ttl)
		p.sessionsMu.Lock()
		err := p.playMap.AddWithTTL(clientId, info, ttl)
		p.sessionsMu.Unlock()
		if err != nil {
			log.Warn(ctx, "Error updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
		}
		p.enqueuePlaybackReport(ctx, info, filtered)

	case StateStopped:
		var loadedMF *model.MediaFile
		haveVerdict := false
		if !params.IgnoreScrobble && player.ScrobbleEnabled {
			mf, err := p.ds.MediaFile().GetWithParticipants(ctx, params.MediaId)
			if err != nil {
				return err
			}
			loadedMF = mf
			filtered = p.isFilteredOut(ctx, mf)
			haveVerdict = true
			trackDurationMs := int64(mf.Duration * 1000)
			threshold := min(trackDurationMs*50/100, 240_000)
			if params.PositionMs >= threshold {
				err = p.incPlay(ctx, mf, now)
				if err != nil {
					log.Warn(ctx, "Error updating play counts", "id", mf.ID, "track", mf.Title, "user", user.UserName, err)
				}
				p.dispatchScrobble(ctx, mf, now, filtered)
			}
		}
		p.sessionsMu.Lock()
		info, getErr := p.playMap.Get(clientId)
		// A late stop for a previous track must not end the current session nor reach
		// playback reporters, or presence-style plugins would clear the active track.
		if getErr == nil && info.MediaFile.ID != params.MediaId {
			p.sessionsMu.Unlock()
			log.Trace(ctx, "Ignoring out-of-order stopped report for different track", "clientId", clientId, "stoppedMediaId", params.MediaId, "currentMediaId", info.MediaFile.ID)
			return nil
		}
		p.playMap.Remove(clientId)
		p.sessionsMu.Unlock()
		stoppedInfo := PlaybackSession{
			UserId:       user.ID,
			Username:     user.UserName,
			PlayerId:     clientId,
			PlayerName:   client,
			State:        params.State,
			PositionMs:   params.PositionMs,
			PlaybackRate: params.PlaybackRate,
			LastReport:   now,
		}
		if getErr == nil {
			stoppedInfo.MediaFile = info.MediaFile
			stoppedInfo.Start = info.Start
		} else {
			mf := loadedMF
			if mf == nil {
				var mfErr error
				mf, mfErr = p.ds.MediaFile().GetWithParticipants(ctx, params.MediaId)
				if mfErr != nil {
					return mfErr
				}
			}
			stoppedInfo.MediaFile = *mf
		}
		if !haveVerdict {
			filtered = p.isFilteredOut(ctx, &stoppedInfo.MediaFile)
		}
		p.enqueuePlaybackReport(ctx, stoppedInfo, filtered)
	}

	if conf.Server.EnableNowPlaying {
		p.broker.SendBroadcastMessage(ctx, &events.NowPlayingCount{Count: p.playMap.Len()})
	}

	// NowPlaying gating, by design distinct from scrobble submission:
	//   - IgnoreScrobble=true   -> still send NowPlaying (suppresses only the
	//     scrobble submission/play-count above), mirroring the legacy scrobble
	//     endpoint's submission=false behavior.
	//   - player.ScrobbleEnabled=false -> never send NowPlaying.
	// External agents here are the active scrobblers (Last.fm, ListenBrainz, and
	// scrobbler plugins) returned by getActiveScrobblers; see dispatchNowPlaying.
	if player.ScrobbleEnabled &&
		(params.State == StateStarting || params.State == StatePlaying) {
		if filtered {
			log.Debug(ctx, "Ignoring external NowPlaying update for filtered track", "mediaId", params.MediaId)
		} else if info, err := p.playMap.Get(clientId); err == nil {
			p.enqueueNowPlaying(ctx, clientId, user.ID, &info.MediaFile, int(params.PositionMs/1000))
		}
	}

	return nil
}

func (p *playTracker) GetNowPlaying(ctx context.Context) ([]PlaybackSession, error) {
	// The cache is process-global, so it holds every user's playback, across all libraries.
	res := p.playMap.Values()
	if user, ok := request.UserFrom(ctx); ok {
		res = slice.Filter(res, func(s PlaybackSession) bool {
			return user.HasLibraryAccess(s.MediaFile.LibraryID)
		})
	}
	slices.SortFunc(res, func(a, b PlaybackSession) int {
		return b.Start.Compare(a.Start)
	})
	for i := range res {
		if res[i].State == StatePlaying {
			elapsed := time.Since(res[i].LastReport).Milliseconds()
			estimated := res[i].PositionMs + int64(float64(elapsed)*res[i].PlaybackRate)
			trackDurationMs := int64(res[i].MediaFile.Duration * 1000)
			res[i].PositionMs = min(estimated, trackDurationMs)
		}
	}
	return res, nil
}

func (p *playTracker) Submit(ctx context.Context, submissions []Submission) error {
	username, _ := request.UsernameFrom(ctx)
	player, _ := request.PlayerFrom(ctx)
	if !player.ScrobbleEnabled {
		log.Debug(ctx, "External scrobbling disabled for this player", "player", player.Name, "ip", player.IP, "user", username)
	}
	event := &events.RefreshResource{}
	success := 0

	for _, s := range submissions {
		mf, err := p.ds.MediaFile().GetWithParticipants(ctx, s.TrackID)
		if err != nil {
			log.Error(ctx, "Cannot find track for scrobbling", "id", s.TrackID, "user", username, err)
			continue
		}
		filtered := p.isFilteredOut(ctx, mf)
		err = p.incPlay(ctx, mf, s.Timestamp)
		if err != nil {
			log.Error(ctx, "Error updating play counts", "id", mf.ID, "track", mf.Title, "user", username, err)
		} else {
			success++
			event.With("song", mf.ID).With("album", mf.AlbumID).With("artist", mf.AlbumArtistID)
			log.Info(ctx, "Scrobbled", "title", mf.Title, "artist", mf.Artist, "user", username, "timestamp", s.Timestamp)
			if player.ScrobbleEnabled {
				p.dispatchScrobble(ctx, mf, s.Timestamp, filtered)
			}
		}
	}

	if success > 0 {
		p.broker.SendMessage(ctx, event)
	}
	return nil
}

func (p *playTracker) incPlay(ctx context.Context, track *model.MediaFile, timestamp time.Time) error {
	return p.ds.WithTx(func(tx model.DataStore) error {
		err := tx.MediaFile().IncPlayCount(ctx, track.ID, timestamp)
		if err != nil {
			return err
		}
		err = tx.Album().IncPlayCount(ctx, track.AlbumID, timestamp)
		if err != nil {
			return err
		}
		for _, artist := range track.Participants[model.RoleArtist] {
			err = tx.Artist().IncPlayCount(ctx, artist.ID, timestamp)
			if err != nil {
				return err
			}
		}
		if conf.Server.EnableScrobbleHistory {
			return tx.Scrobble().RecordScrobble(ctx, track.ID, timestamp)
		}
		return nil
	})
}

// Take this verdict before incPlay mutates what a filter reads, and independently of
// which scrobblers are active: it can be stored on a session and dispatched much later.
// Any parse or query failure fails open, because filtering must not break scrobbling.
func (p *playTracker) isFilteredOut(ctx context.Context, t *model.MediaFile) bool {
	u, _ := request.UserFrom(ctx)
	if u.ScrobbleFilter == "" {
		return false
	}
	var c criteria.Criteria
	if err := json.Unmarshal([]byte(u.ScrobbleFilter), &c); err != nil {
		log.Warn(ctx, "Invalid scrobble filter, ignoring", "user", u.UserName, err)
		return false
	}
	match, err := p.ds.MediaFile().MatchesCriteria(ctx, t.ID, c)
	if err != nil {
		log.Warn(ctx, "Error evaluating scrobble filter, ignoring", "user", u.UserName, "track", t.Title, err)
		return false
	}
	return match
}

func (p *playTracker) dispatchScrobble(ctx context.Context, t *model.MediaFile, playTime time.Time, filtered bool) {
	if t.Artist == consts.UnknownArtist {
		log.Debug(ctx, "Ignoring external Scrobble for track with unknown artist", "track", t.Title, "artist", t.Artist)
		return
	}
	if filtered {
		log.Debug(ctx, "Ignoring external Scrobble for filtered track", "track", t.Title, "artist", t.Artist)
		return
	}

	allScrobblers := p.getActiveScrobblers()
	u, _ := request.UserFrom(ctx)
	scrobble := Scrobble{MediaFile: *t, TimeStamp: playTime}
	for name, s := range allScrobblers {
		if !s.IsAuthorized(ctx, u.ID) {
			continue
		}
		log.Debug(ctx, "Buffering Scrobble", "scrobbler", name, "track", t.Title, "artist", t.Artist)
		err := s.Scrobble(ctx, u.ID, scrobble)
		if err != nil {
			log.Error(ctx, "Error sending Scrobble", "scrobbler", name, "track", t.Title, "artist", t.Artist, err)
			continue
		}
	}
}

var constructors map[string]Constructor

func Register(name string, init Constructor) {
	if constructors == nil {
		constructors = make(map[string]Constructor)
	}
	constructors[name] = init
}

// IsBuiltinScrobbler reports whether name belongs to a registered builtin
// scrobbler (e.g. "lastfm", "listenbrainz").
func IsBuiltinScrobbler(name string) bool {
	_, ok := constructors[name]
	return ok
}