memby/server/internal/credits/tracearr.go

package credits

import (
	"context"
	"sort"
	"sync"
	"time"
)

// Tracearr as a candidate source, kept deliberately apart from the detector.
//
// Everything about *why* an episode is worth scanning lives on this side of the boundary,
// and nothing about it crosses. The detector is handed a file and a window; it never learns
// that Paul watched four episodes of Blue Bloods this week, and it must not, or it would
// become possible to tune the detector to agree with the predictor rather than with the
// media.

// Database is the narrow slice of the store this package reads. Narrow so the whole candidate
// pipeline can be exercised against a map in a test, and so it is obvious at a glance that
// nothing here writes anything but a marker.
type Database interface {
	RecentWatches(ctx context.Context, since time.Time, limit int) ([]Watch, error)
	SeriesEpisodes(ctx context.Context, seriesIDs []string) ([]SeriesEpisode, error)
}

// SeriesEpisode is an episode with its series and numbering, which is what an index is built from.
type SeriesEpisode struct {
	ItemID   string
	SeriesID string
	Season   int
	Episode  int
}

// watchLimit bounds the demand query. A household producing more than this many episode
// sessions inside the decay window is one whose oldest sessions cannot possibly still be
// predictive, so the newest-first ordering makes the cap harmless.
const watchLimit = 500

// TracearrSource builds candidates from what the household has actually been watching.
type TracearrSource struct {
	DB  Database
	Cfg Config
	mu  sync.RWMutex
}

func (s *TracearrSource) Configure(cfg Config) {
	s.mu.Lock()
	s.Cfg = NormaliseConfig(cfg)
	s.mu.Unlock()
}

func (s *TracearrSource) config() Config {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return NormaliseConfig(s.Cfg)
}

// Candidates is the whole predictive pipeline: read recent demand, group it into per-viewer
// activities, load the numbering for the few series involved, and project a small window
// ahead of each viewer.
//
// Two database reads for the entire household, whatever it is watching. Nothing here is
// per-candidate and nothing is per-episode.
func (s *TracearrSource) Candidates(ctx context.Context) ([]Candidate, error) {
	if s == nil || s.DB == nil {
		return nil, nil
	}
	cfg := s.config()
	now := time.Now().UTC()

	// The decay window is the query's window too. Anything older cannot survive
	// decayWeight, so fetching it would be reading rows in order to discard them.
	watches, err := s.DB.RecentWatches(ctx, now.Add(-cfg.WeakWindow), watchLimit)
	if err != nil {
		return nil, err
	}
	activities := Activities(watches, now, cfg)
	if len(activities) == 0 {
		return nil, nil
	}

	seriesIDs := make([]string, 0, len(activities))
	seen := map[string]bool{}
	for _, activity := range activities {
		if !seen[activity.SeriesID] {
			seen[activity.SeriesID] = true
			seriesIDs = append(seriesIDs, activity.SeriesID)
		}
	}
	episodes, err := s.DB.SeriesEpisodes(ctx, seriesIDs)
	if err != nil {
		return nil, err
	}
	return BuildCandidates(activities, NewEpisodeIndex(episodes), now, cfg), nil
}

// episodeIndex is an in-memory ordering of each series' episodes.
//
// Built once per refresh cycle from one query, so stepping forward — including across a
// season boundary, which is exactly where somebody is most likely to keep watching — is
// arithmetic rather than a lookup per candidate.
type episodeIndex struct {
	bySeries map[string][]SeriesEpisode
}

// NewEpisodeIndex orders the episodes of each series by season and number.
func NewEpisodeIndex(episodes []SeriesEpisode) EpisodeIndex {
	index := &episodeIndex{bySeries: map[string][]SeriesEpisode{}}
	for _, episode := range episodes {
		if episode.ItemID == "" || episode.SeriesID == "" || episode.Episode <= 0 {
			continue
		}
		index.bySeries[episode.SeriesID] = append(index.bySeries[episode.SeriesID], episode)
	}
	for _, list := range index.bySeries {
		sort.Slice(list, func(a, b int) bool {
			if list[a].Season != list[b].Season {
				return list[a].Season < list[b].Season
			}
			return list[a].Episode < list[b].Episode
		})
	}
	return index
}

// Following returns the next count episodes after a position.
//
// Specials are skipped. Season 0 is a real season and its episodes are real files, but
// nobody finishing S06E07 goes on to a behind-the-scenes featurette, and queueing one would
// spend a scan on an episode that will not be watched.
func (i *episodeIndex) Following(seriesID string, season, episode, count int) []EpisodeRef {
	if count <= 0 {
		return nil
	}
	list := i.bySeries[seriesID]
	out := make([]EpisodeRef, 0, count)
	for _, candidate := range list {
		if candidate.Season <= 0 {
			continue
		}
		after := candidate.Season > season ||
			(candidate.Season == season && candidate.Episode > episode)
		if !after {
			continue
		}
		out = append(out, EpisodeRef{
			ItemID:  candidate.ItemID,
			Season:  candidate.Season,
			Episode: candidate.Episode,
		})
		if len(out) == count {
			break
		}
	}
	return out
}