memby/server/internal/store/ratings.go

package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// ErrMediaRatingsNotFound means no durable external-rating response has been stored yet.
var ErrMediaRatingsNotFound = errors.New("store: media ratings not found")

// MediaRatings returns the raw provider response and the time it was fetched. Keeping the
// raw response lets API settings select different sources without invalidating this cache.
func (s *Store) MediaRatings(
	ctx context.Context, mediaType, provider, providerID string,
) (json.RawMessage, time.Time, error) {
	var ratings []byte
	var fetchedAt time.Time
	err := s.pool.QueryRow(ctx, `
		SELECT ratings, fetched_at
		FROM external_media_ratings
		WHERE media_type = $1 AND provider = $2 AND provider_id = $3`,
		mediaType, provider, providerID).Scan(&ratings, &fetchedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, time.Time{}, ErrMediaRatingsNotFound
	}
	if err != nil {
		return nil, time.Time{}, fmt.Errorf("store: load media ratings: %w", err)
	}
	return json.RawMessage(ratings), fetchedAt, nil
}

// RatingKey identifies one title at the external provider. The same key serves every
// television and every viewer, which is why ratings are stored per title rather than
// per Emby item — a series and each of its episodes share one row.
type RatingKey struct {
	MediaType  string // movie | show
	Provider   string // tmdb | imdb
	ProviderID string
}

func (k RatingKey) valid() bool {
	return k.MediaType != "" && k.Provider != "" && k.ProviderID != ""
}

// MediaRatingsEntry is one stored provider response and its age.
type MediaRatingsEntry struct {
	Ratings   json.RawMessage
	FetchedAt time.Time
}

// MediaRatingsBatch reads many stored responses in one query. Rows attach ratings to
// every card they carry, so the per-title read of MediaRatings would otherwise mean a
// round trip per poster.
func (s *Store) MediaRatingsBatch(
	ctx context.Context, keys []RatingKey,
) (map[RatingKey]MediaRatingsEntry, error) {
	found := make(map[RatingKey]MediaRatingsEntry, len(keys))
	mediaTypes, providers, providerIDs := ratingKeyColumns(keys)
	if len(mediaTypes) == 0 {
		return found, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT media_type, provider, provider_id, ratings, fetched_at
		FROM external_media_ratings
		WHERE (media_type, provider, provider_id) IN (
			SELECT * FROM unnest($1::text[], $2::text[], $3::text[])
		)`, mediaTypes, providers, providerIDs)
	if err != nil {
		return nil, fmt.Errorf("store: load media ratings batch: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var key RatingKey
		var ratings []byte
		var fetchedAt time.Time
		if err := rows.Scan(&key.MediaType, &key.Provider, &key.ProviderID, &ratings, &fetchedAt); err != nil {
			return nil, fmt.Errorf("store: scan media ratings batch: %w", err)
		}
		found[key] = MediaRatingsEntry{Ratings: json.RawMessage(ratings), FetchedAt: fetchedAt}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("store: load media ratings batch: %w", err)
	}
	return found, nil
}

// ratingKeyColumns flattens keys into the three parallel arrays unnest expects, dropping
// incomplete ones so a single unresolved item cannot break the whole read.
func ratingKeyColumns(keys []RatingKey) ([]string, []string, []string) {
	mediaTypes := make([]string, 0, len(keys))
	providers := make([]string, 0, len(keys))
	providerIDs := make([]string, 0, len(keys))
	seen := make(map[RatingKey]bool, len(keys))
	for _, key := range keys {
		if !key.valid() || seen[key] {
			continue
		}
		seen[key] = true
		mediaTypes = append(mediaTypes, key.MediaType)
		providers = append(providers, key.Provider)
		providerIDs = append(providerIDs, key.ProviderID)
	}
	return mediaTypes, providers, providerIDs
}

// SaveItemRatingRef remembers which external title an Emby item is, so later rows do not
// have to ask Emby for its ProviderIds again.
func (s *Store) SaveItemRatingRef(ctx context.Context, itemID string, key RatingKey) error {
	if itemID == "" || !key.valid() {
		return nil
	}
	_, err := s.pool.Exec(ctx, `
		INSERT INTO item_rating_refs (item_id, media_type, provider, provider_id, updated_at)
		VALUES ($1, $2, $3, $4, now())
		ON CONFLICT (item_id) DO UPDATE SET
			media_type = EXCLUDED.media_type,
			provider = EXCLUDED.provider,
			provider_id = EXCLUDED.provider_id,
			updated_at = EXCLUDED.updated_at`,
		itemID, key.MediaType, key.Provider, key.ProviderID)
	if err != nil {
		return fmt.Errorf("store: save item rating ref: %w", err)
	}
	return nil
}

// ItemRatingRefs resolves Emby item ids that have been looked up before.
func (s *Store) ItemRatingRefs(ctx context.Context, itemIDs []string) (map[string]RatingKey, error) {
	found := make(map[string]RatingKey, len(itemIDs))
	if len(itemIDs) == 0 {
		return found, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item_id, media_type, provider, provider_id
		FROM item_rating_refs
		WHERE item_id = ANY($1)`, itemIDs)
	if err != nil {
		return nil, fmt.Errorf("store: load item rating refs: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var itemID string
		var key RatingKey
		if err := rows.Scan(&itemID, &key.MediaType, &key.Provider, &key.ProviderID); err != nil {
			return nil, fmt.Errorf("store: scan item rating refs: %w", err)
		}
		found[itemID] = key
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("store: load item rating refs: %w", err)
	}
	return found, nil
}

// LibraryProviderIDs reads the provider identifiers the library import stored, following
// an episode to its series — MDBList rates shows, never single episodes.
//
// The returned map is keyed by the *requested* item id and carries Emby's own
// ProviderIds object, so the caller applies the same provider precedence it uses for a
// live lookup rather than a second copy of that rule living in SQL.
func (s *Store) LibraryProviderIDs(
	ctx context.Context, itemIDs []string,
) (map[string]LibraryProviderRef, error) {
	found := make(map[string]LibraryProviderRef, len(itemIDs))
	if len(itemIDs) == 0 {
		return found, nil
	}
	rows, err := s.pool.Query(ctx, `
		SELECT item.id,
		       item.type,
		       CASE WHEN item.type = 'Episode' THEN series.payload -> 'ProviderIds'
		            ELSE item.payload -> 'ProviderIds' END
		FROM library_items item
		LEFT JOIN library_items series
		  ON item.type = 'Episode' AND series.id = item.series_id
		WHERE item.id = ANY($1)`, itemIDs)
	if err != nil {
		return nil, fmt.Errorf("store: load library provider ids: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var itemID, itemType string
		var raw []byte
		if err := rows.Scan(&itemID, &itemType, &raw); err != nil {
			return nil, fmt.Errorf("store: scan library provider ids: %w", err)
		}
		ids := map[string]string{}
		if len(raw) > 0 {
			// Emby writes provider ids as strings, but a hand-edited or future payload
			// need not: an undecodable object leaves the item unresolved rather than
			// failing the whole read.
			_ = json.Unmarshal(raw, &ids)
		}
		found[itemID] = LibraryProviderRef{Type: itemType, ProviderIDs: ids}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("store: load library provider ids: %w", err)
	}
	return found, nil
}

// LibraryProviderRef is one imported item's type and external identifiers.
type LibraryProviderRef struct {
	Type        string
	ProviderIDs map[string]string
}

// MediaRatingsStats reports how much of the durable cache exists and how much of it is
// old enough to be refreshed, which is the only way the admin page can say whether the
// integration is still spending external requests.
func (s *Store) MediaRatingsStats(ctx context.Context, staleBefore time.Time) (total, stale int, err error) {
	err = s.pool.QueryRow(ctx, `
		SELECT count(*), count(*) FILTER (WHERE fetched_at < $1)
		FROM external_media_ratings`, staleBefore).Scan(&total, &stale)
	if err != nil {
		return 0, 0, fmt.Errorf("store: media ratings stats: %w", err)
	}
	return total, stale, nil
}

// StaleRatingKeys is the oldest stored titles due to be re-checked, oldest first.
//
// Oldest first rather than by any measure of popularity, because the refresh is bounded
// per run: taking the oldest means every title comes round eventually, where taking the
// most-watched would leave the tail of the library permanently on scores from the year it
// was imported. The limit is what keeps a run's cost — and therefore the day's external
// allowance — a figure the operator can reason about.
func (s *Store) StaleRatingKeys(
	ctx context.Context, staleBefore time.Time, limit int,
) ([]RatingKey, error) {
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	rows, err := s.pool.Query(ctx, `
		SELECT media_type, provider, provider_id
		FROM external_media_ratings
		WHERE fetched_at < $1
		ORDER BY fetched_at ASC
		LIMIT $2`, staleBefore, limit)
	if err != nil {
		return nil, fmt.Errorf("store: stale rating keys: %w", err)
	}
	defer rows.Close()
	keys := []RatingKey{}
	for rows.Next() {
		var key RatingKey
		if err := rows.Scan(&key.MediaType, &key.Provider, &key.ProviderID); err != nil {
			return nil, fmt.Errorf("store: scan stale rating key: %w", err)
		}
		keys = append(keys, key)
	}
	return keys, rows.Err()
}

// SaveMediaRatings upserts a successfully loaded provider response.
func (s *Store) SaveMediaRatings(
	ctx context.Context, mediaType, provider, providerID string, ratings json.RawMessage,
) error {
	if !json.Valid(ratings) {
		return errors.New("store: save media ratings: invalid JSON")
	}
	_, err := s.pool.Exec(ctx, `
		INSERT INTO external_media_ratings (media_type, provider, provider_id, ratings, fetched_at)
		VALUES ($1, $2, $3, $4, now())
		ON CONFLICT (media_type, provider, provider_id) DO UPDATE SET
			ratings = EXCLUDED.ratings,
			fetched_at = EXCLUDED.fetched_at`,
		mediaType, provider, providerID, ratings)
	if err != nil {
		return fmt.Errorf("store: save media ratings: %w", err)
	}
	return nil
}