memby/server/internal/api/ratings_attach.go

package api

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/mdblist"
	"github.com/ponzischeme89/memby/server/internal/recommend"
	"github.com/ponzischeme89/memby/server/internal/store"
)

// MembyRatings rides on the item payload so a card can draw its ratings the moment the
// row does. The dedicated /v1/items/{id}/ratings endpoint remains for anything the
// database has never seen — this field is only ever what was already stored, so
// attaching it costs one indexed read for a whole launcher and never an external
// request on the request path.
const ratingsItemField = "MembyRatings"

// A launcher is a few hundred cards at most. The cap is a guard against a future row
// type asking for a thousand, not a limit anything reaches today.
const ratingsAttachItemLimit = 600

const (
	// The warmer's pace. MDBList sells a daily allowance rather than a rate, so the
	// interval only stops a burst of navigation becoming a burst of requests, and the
	// daily budget is what actually protects the quota.
	ratingsWarmInterval    = 2 * time.Second
	ratingsWarmDailyBudget = 400
	// A refused request means the allowance is gone; asking again inside the same hour
	// only spends the operator's rate limit on nothing.
	ratingsWarmBackoff = time.Hour
	ratingsWarmQueue   = 512
)

// How long the operator's ratings settings are trusted in memory. Every row, search and
// detail response now consults them, and instant search asks once per keystroke — that
// is a Postgres round trip per key press for a document changed a few times a year. The
// admin write clears this, so a change still takes effect at once.
const mdblistSettingsTTL = 30 * time.Second

type mdblistSettingsCache struct {
	mu       sync.Mutex
	settings store.MDBListSettings
	loadedAt time.Time
}

// mdblistSettings reports the ratings configuration, and whether ratings are on at all.
func (s *Server) mdblistSettings(ctx context.Context) (store.MDBListSettings, bool) {
	if s.store == nil {
		return store.MDBListSettings{}, false
	}
	c := &s.mdblistSettingsCache
	now := time.Now()
	c.mu.Lock()
	if !c.loadedAt.IsZero() && now.Sub(c.loadedAt) < mdblistSettingsTTL {
		settings := c.settings
		c.mu.Unlock()
		return settings, settings.Enabled && settings.APIKey != "" && len(settings.Sources) > 0
	}
	c.mu.Unlock()

	settings, err := s.store.MDBListSettings(ctx)
	if err != nil {
		s.loggerFor(ctx).Warn("MDBList settings unavailable", "error", err)
		return store.MDBListSettings{}, false
	}
	c.mu.Lock()
	c.settings, c.loadedAt = settings, now
	c.mu.Unlock()
	return settings, settings.Enabled && settings.APIKey != "" && len(settings.Sources) > 0
}

// forgetMDBListSettings is called by the admin write so an operator's change is live on
// the next request rather than at the end of the cache window.
func (s *Server) forgetMDBListSettings() {
	c := &s.mdblistSettingsCache
	c.mu.Lock()
	c.loadedAt = time.Time{}
	c.mu.Unlock()
}

// ratingsWarmer renews and fills the durable cache behind the viewer.
//
// Navigation is what feeds it: every row that reaches a television reports the titles it
// could not decorate, and those are fetched once and kept. The queue is bounded and
// deliberately lossy — a title dropped now is offered again the next time somebody
// scrolls past it, which is a far better failure than a growing backlog of requests
// against a daily allowance.
type ratingsWarmer struct {
	once   sync.Once
	queue  chan store.RatingKey
	mu     sync.Mutex
	queued map[store.RatingKey]bool

	spent      int
	windowFrom time.Time
	blockedTil time.Time
}

// warmRatings offers titles to the background warmer. It never blocks the caller.
func (s *Server) warmRatings(keys ...store.RatingKey) {
	if s.mdblist == nil || s.store == nil || len(keys) == 0 {
		return
	}
	w := &s.ratingsWarm
	w.once.Do(func() {
		w.queue = make(chan store.RatingKey, ratingsWarmQueue)
		w.queued = map[store.RatingKey]bool{}
		go s.ratingsWarmLoop()
	})
	for _, key := range keys {
		if key.MediaType == "" || key.Provider == "" || key.ProviderID == "" {
			continue
		}
		w.mu.Lock()
		already := w.queued[key]
		if !already {
			w.queued[key] = true
		}
		w.mu.Unlock()
		if already {
			continue
		}
		select {
		case w.queue <- key:
		default:
			// Full. Forget it rather than wait: the next row that shows this title will
			// offer it again, and a blocked handler would be paying for a cache fill.
			w.mu.Lock()
			delete(w.queued, key)
			w.mu.Unlock()
		}
	}
}

func (s *Server) ratingsWarmLoop() {
	w := &s.ratingsWarm
	for key := range w.queue {
		w.mu.Lock()
		delete(w.queued, key)
		w.mu.Unlock()
		s.warmOne(key)
		time.Sleep(ratingsWarmInterval)
	}
}

func (s *Server) warmOne(key store.RatingKey) {
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	settings, enabled := s.mdblistSettings(ctx)
	if !enabled {
		return
	}
	if !s.claimRatingsBudget(time.Now()) {
		return
	}
	if _, err := s.fetchAndStoreRatings(ctx, settings.APIKey, key); err != nil {
		var apiErr *mdblist.APIError
		if errors.As(err, &apiErr) && (apiErr.StatusCode == 429 || apiErr.StatusCode == 402) {
			s.blockRatingsWarming(time.Now().Add(ratingsWarmBackoff))
		}
		if s.log != nil {
			s.log.Debug("MDBList warm failed",
				"component", "ratings", "provider", key.Provider, "id", key.ProviderID, "error", err)
		}
		return
	}
	if s.log != nil {
		s.log.Debug("MDBList rating stored",
			"component", "ratings", "provider", key.Provider, "id", key.ProviderID)
	}
}

// claimRatingsBudget spends one of the day's allowed external requests. The window is
// in-process: a restart forgives what was already spent, which is the right way round
// for a counter whose only job is to stop a runaway fill.
func (s *Server) claimRatingsBudget(now time.Time) bool {
	w := &s.ratingsWarm
	w.mu.Lock()
	defer w.mu.Unlock()
	if now.Before(w.blockedTil) {
		return false
	}
	if w.windowFrom.IsZero() || now.Sub(w.windowFrom) >= 24*time.Hour {
		w.windowFrom = now
		w.spent = 0
	}
	if w.spent >= ratingsWarmDailyBudget {
		return false
	}
	w.spent++
	return true
}

func (s *Server) blockRatingsWarming(until time.Time) {
	w := &s.ratingsWarm
	w.mu.Lock()
	defer w.mu.Unlock()
	w.blockedTil = until
}

// rememberRatingRef records an item's external identity without delaying the response
// that discovered it.
func (s *Server) rememberRatingRef(ctx context.Context, itemID string, key store.RatingKey) {
	if s.store == nil || itemID == "" {
		return
	}
	ctx = context.WithoutCancel(ctx)
	go func() {
		writeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		if err := s.store.SaveItemRatingRef(writeCtx, itemID, key); err != nil && s.log != nil {
			s.loggerFor(ctx).Debug("rating reference write failed", "item", itemID, "error", err)
		}
	}()
}

// decorateHomeRatings attaches stored ratings to every collection of items in one pass.
//
// One pass rather than one per row: the fixed rows and the composed rows share their
// backing arrays, so decorating each separately would resolve the same ids repeatedly,
// and writing back in place is what keeps both views agreeing.
func (s *Server) decorateHomeRatings(ctx context.Context, out *homeResponse) {
	collections := make([][]json.RawMessage, 0, len(out.Rows)+4)
	for _, row := range out.Rows {
		collections = append(collections, row.Items)
	}
	collections = append(collections,
		out.ContinueWatching, out.NextUp, out.Favorites, out.LatestMovies)
	s.decorateItems(ctx, collections...)
}

// decorateItems is the one door items leave the gateway through.
//
// It attaches the facts Memby knows about a title beyond Emby's payload: the latest stored
// Sonarr series status, stored review scores, and — for a shadow viewer — whose progress
// this is. The concerns stay in separate functions, but every call site needs one door;
// a decoration added at seven sites is a decoration missing from the eighth.
func (s *Server) decorateItems(ctx context.Context, collections ...[]json.RawMessage) {
	s.decorateSeriesStatuses(ctx, collections...)
	s.decorateItemRatings(ctx, collections...)
	s.decorateViewerState(ctx, collections...)
}

// decorateItemRatings rewrites each item in place with whatever the database already
// holds for it, and offers the rest to the warmer.
func (s *Server) decorateItemRatings(ctx context.Context, collections ...[]json.RawMessage) {
	settings, enabled := s.mdblistSettings(ctx)
	if !enabled {
		return
	}
	ids := itemIDsIn(collections, ratingsAttachItemLimit)
	if len(ids) == 0 {
		return
	}
	keys := s.ratingKeysForItems(ctx, ids)
	if len(keys) == 0 {
		return
	}
	stored, err := s.store.MediaRatingsBatch(ctx, distinctRatingKeys(keys))
	if err != nil {
		s.loggerFor(ctx).Warn("stored ratings read failed", "error", err)
		return
	}
	now := time.Now()
	byItem := make(map[string][]movieRating, len(keys))
	warm := make([]store.RatingKey, 0)
	warmed := make(map[store.RatingKey]bool, len(keys))
	for itemID, key := range keys {
		entry, ok := stored[key]
		var ratings []mdblist.Rating
		if ok {
			if json.Unmarshal(entry.Ratings, &ratings) != nil {
				ratings = nil
			}
			byItem[itemID] = selectedMovieRatings(settings.Sources, ratings)
		}
		if (!ok || ratingsNeedRefresh(ratings, entry.FetchedAt, now)) && !warmed[key] {
			warmed[key] = true
			warm = append(warm, key)
		}
	}
	for _, items := range collections {
		for index, raw := range items {
			id := itemIDOf(raw)
			ratings, ok := byItem[id]
			if !ok || len(ratings) == 0 {
				continue
			}
			items[index] = injectItemRatings(raw, ratings)
		}
	}
	// Only what the household actually looked at is warmed, which is what keeps a large
	// library from being imported into MDBList's quota all at once.
	s.warmRatings(warm...)
}

// decorateRowRatings is the row-shaped entry point used outside Home.
func (s *Server) decorateRowRatings(ctx context.Context, rows []recommend.Row) {
	collections := make([][]json.RawMessage, 0, len(rows))
	for _, row := range rows {
		collections = append(collections, row.Items)
	}
	s.decorateItems(ctx, collections...)
}

// ratingKeysForItems resolves Emby ids to external titles, preferring the index built by
// navigation and falling back to the imported library.
func (s *Server) ratingKeysForItems(ctx context.Context, ids []string) map[string]store.RatingKey {
	keys, err := s.store.ItemRatingRefs(ctx, ids)
	if err != nil {
		s.loggerFor(ctx).Warn("rating references read failed", "error", err)
		keys = map[string]store.RatingKey{}
	}
	missing := make([]string, 0, len(ids))
	for _, id := range ids {
		if _, ok := keys[id]; !ok {
			missing = append(missing, id)
		}
	}
	if len(missing) == 0 {
		return keys
	}
	refs, err := s.store.LibraryProviderIDs(ctx, missing)
	if err != nil {
		s.loggerFor(ctx).Warn("library provider ids read failed", "error", err)
		return keys
	}
	for id, ref := range refs {
		if key, ok := ratingKeyFor(ref.Type, ref.ProviderIDs); ok {
			keys[id] = key
		}
	}
	return keys
}

// ratingKeyFor applies the same rule the live lookup does: MDBList rates films and
// shows, an episode is rated as its series, and tmdb wins over imdb.
func ratingKeyFor(itemType string, ids map[string]string) (store.RatingKey, bool) {
	var mediaType string
	switch {
	case strings.EqualFold(itemType, "Movie"):
		mediaType = "movie"
	case strings.EqualFold(itemType, "Series"), strings.EqualFold(itemType, "Episode"):
		mediaType = "show"
	default:
		return store.RatingKey{}, false
	}
	provider, providerID := movieProvider(ids)
	if providerID == "" {
		return store.RatingKey{}, false
	}
	return store.RatingKey{MediaType: mediaType, Provider: provider, ProviderID: providerID}, true
}

func distinctRatingKeys(keys map[string]store.RatingKey) []store.RatingKey {
	seen := make(map[store.RatingKey]bool, len(keys))
	out := make([]store.RatingKey, 0, len(keys))
	for _, key := range keys {
		if seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, key)
	}
	return out
}

// injectItemRatings adds the ratings field to one item payload. Emby's JSON is otherwise
// forwarded verbatim, so this decodes into raw members and re-encodes rather than
// through any struct: a field this build does not know about must survive the round trip.
func injectItemRatings(raw json.RawMessage, ratings []movieRating) json.RawMessage {
	if len(ratings) == 0 {
		return raw
	}
	var members map[string]json.RawMessage
	if json.Unmarshal(raw, &members) != nil || members == nil {
		return raw
	}
	encoded, err := json.Marshal(ratings)
	if err != nil {
		return raw
	}
	members[ratingsItemField] = encoded
	out, err := json.Marshal(members)
	if err != nil {
		return raw
	}
	return out
}

func itemIDOf(raw json.RawMessage) string {
	var item struct {
		ID string `json:"Id"`
	}
	if json.Unmarshal(raw, &item) != nil {
		return ""
	}
	return item.ID
}

func itemIDsIn(collections [][]json.RawMessage, limit int) []string {
	seen := make(map[string]bool)
	ids := make([]string, 0, limit)
	for _, items := range collections {
		for _, raw := range items {
			id := itemIDOf(raw)
			if id == "" || seen[id] {
				continue
			}
			seen[id] = true
			ids = append(ids, id)
			if len(ids) >= limit {
				return ids
			}
		}
	}
	return ids
}