memby/server/internal/credits/candidates.go

package credits

import (
	"sort"
	"strings"
	"time"
)

// Candidate generation, and all of it pure.
//
// The rule this file exists to enforce is the one in the brief that matters most: a viewer
// watching Blue Bloods is *not* a reason to scan 293 episodes of Blue Bloods. It is a reason
// to scan the three they are about to reach. Everything here is arithmetic on where somebody
// has got to and how fast they are moving, and the output is a handful of episodes.

// Config is the tuning. Defaults come from DefaultConfig; an operator changes them through
// the environment rather than by editing constants, because the right look-ahead depends on
// how a particular household watches television.
type Config struct {
	// PrefetchEpisodes is the look-ahead for an ordinary viewer — the "credits.prefetchEpisodes
	// = 3" of the brief. Velocity moves the actual depth either side of it.
	PrefetchEpisodes int
	// MaxPrefetchEpisodes is the ceiling nothing may exceed, however fast somebody watches.
	// The objective is useful precomputation, not speculative scanning, and this is the line
	// between the two.
	MaxPrefetchEpisodes int

	// The decay thresholds. Demand is evidence with a shelf life: three nights of Blue Bloods
	// followed by a week of Slow Horses must stop producing Blue Bloods candidates on its own,
	// without anything having to notice that the household changed its mind.
	StrongWindow time.Duration
	UsefulWindow time.Duration
	WeakWindow   time.Duration

	// QueueLimit bounds the queue. Past it, low-priority speculation is discarded rather than
	// queued — a backlog of candidates for episodes nobody reached is worse than no backlog.
	QueueLimit int
}

func DefaultConfig() Config {
	return Config{
		PrefetchEpisodes:    3,
		MaxPrefetchEpisodes: 5,
		StrongWindow:        24 * time.Hour,
		UsefulWindow:        3 * 24 * time.Hour,
		WeakWindow:          7 * 24 * time.Hour,
		QueueLimit:          20,
	}
}

// Watch is one episode one viewer played, reduced to the four things candidate generation
// needs. It is what a Tracearr session becomes on the way in, and keeping it this narrow is
// what lets every rule below be tested with a literal.
type Watch struct {
	UserKey  string
	SeriesID string
	Season   int
	Episode  int
	// WatchedAt is when the session ended, or started where it never ended. Recency is
	// measured from it and so is velocity.
	WatchedAt time.Time
	// Completed distinguishes "finished this episode, will start the next" from "is
	// part-way through it". The two produce different look-ahead windows, and getting it
	// wrong costs a scan of an episode somebody is already watching.
	Completed bool
}

// EpisodeRef is a position in a series resolved to something scannable.
type EpisodeRef struct {
	ItemID  string
	Season  int
	Episode int
}

// EpisodeIndex resolves "the episodes after this one" — across a season boundary, since a
// season finale is exactly when somebody is most likely to keep going. Backed by the imported
// library, which already holds every episode with its numbering.
type EpisodeIndex interface {
	Following(seriesID string, season, episode, count int) []EpisodeRef
}

// SeriesActivity is one viewer's relationship with one series right now: where they have got
// to, when they were last there, and how fast they are moving.
type SeriesActivity struct {
	UserKey  string
	SeriesID string

	Season  int
	Episode int
	// InProgress means the furthest episode was not finished, so it is itself a candidate
	// rather than something to look past.
	InProgress bool

	LastViewed time.Time
	// EpisodesPerDay over the recent run. See viewingVelocity.
	EpisodesPerDay float64
}

// Activities groups raw watches into one activity per viewer and series.
//
// "Where they have got to" is the *furthest* episode watched recently, not the most recent
// session: somebody who dips back to rewatch an earlier episode has not un-watched the ones
// after it, and predicting from the rewatch would queue episodes they finished a fortnight
// ago. Recency still comes from the latest session, because that is what decay measures.
func Activities(watches []Watch, now time.Time, cfg Config) []SeriesActivity {
	type key struct{ user, series string }
	grouped := map[key][]Watch{}
	for _, watch := range watches {
		if strings.TrimSpace(watch.SeriesID) == "" || watch.Season < 0 || watch.Episode <= 0 {
			continue
		}
		// Anything past the discard threshold is not evidence about tonight. Dropping it
		// here rather than at the end keeps it out of the velocity calculation too, where
		// a six-month-old session would otherwise drag an active binge down to a crawl.
		if now.Sub(watch.WatchedAt) > cfg.WeakWindow {
			continue
		}
		group := key{user: watch.UserKey, series: watch.SeriesID}
		grouped[group] = append(grouped[group], watch)
	}

	out := make([]SeriesActivity, 0, len(grouped))
	for group, items := range grouped {
		activity := SeriesActivity{UserKey: group.user, SeriesID: group.series}
		for _, watch := range items {
			if watch.WatchedAt.After(activity.LastViewed) {
				activity.LastViewed = watch.WatchedAt
			}
			if watch.Season > activity.Season ||
				(watch.Season == activity.Season && watch.Episode > activity.Episode) {
				activity.Season, activity.Episode = watch.Season, watch.Episode
				activity.InProgress = !watch.Completed
			}
		}
		activity.EpisodesPerDay = viewingVelocity(items, now)
		out = append(out, activity)
	}
	// Sorted so a queue built from identical input is identical, which is what makes the
	// scheduler's behaviour reproducible in a test and its log readable in production.
	sort.Slice(out, func(a, b int) bool {
		if !out[a].LastViewed.Equal(out[b].LastViewed) {
			return out[a].LastViewed.After(out[b].LastViewed)
		}
		if out[a].SeriesID != out[b].SeriesID {
			return out[a].SeriesID < out[b].SeriesID
		}
		return out[a].UserKey < out[b].UserKey
	})
	return out
}

// viewingVelocity is episodes per day over the run somebody is actually on.
//
// Measured across the span from the first watch in the window to now, rather than to the
// last watch: a viewer who watched four episodes on Monday and nothing since is not still
// watching four a day, and rating them as a binge would queue five episodes for somebody
// who has moved on. The span floors at a day, so one evening's four episodes reads as four
// a day rather than as an infinite rate.
func viewingVelocity(watches []Watch, now time.Time) float64 {
	if len(watches) == 0 {
		return 0
	}
	// Distinct episodes: a session resumed three times is one episode watched, and counting
	// the resumes would read an unreliable connection as enthusiasm.
	type slot struct{ season, episode int }
	seen := map[slot]bool{}
	earliest := now
	for _, watch := range watches {
		seen[slot{watch.Season, watch.Episode}] = true
		if watch.WatchedAt.Before(earliest) {
			earliest = watch.WatchedAt
		}
	}
	days := now.Sub(earliest).Hours() / 24
	if days < 1 {
		days = 1
	}
	return float64(len(seen)) / days
}

// prefetchDepth turns a rate into a number of episodes to prepare.
//
// Deliberately a step function rather than anything cleverer. The brief's own guidance is
// slow → 1, normal → 2-3, binge → 4-5, and prediction beyond that is not worth the risk of
// being confidently wrong about somebody's evening: every episode of depth is a real scan.
func prefetchDepth(episodesPerDay float64, cfg Config) int {
	base := cfg.PrefetchEpisodes
	if base <= 0 {
		base = 3
	}
	ceiling := cfg.MaxPrefetchEpisodes
	if ceiling <= 0 {
		ceiling = 5
	}

	depth := base
	switch {
	case episodesPerDay < 0.5:
		// One episode every few days. The next one is all that is worth preparing, and
		// there will be plenty of time to prepare the one after it.
		depth = 1
	case episodesPerDay < 1.5:
		depth = base - 1
	case episodesPerDay < 3:
		depth = base
	case episodesPerDay < 5:
		depth = base + 1
	default:
		depth = base + 2
	}
	if depth < 1 {
		depth = 1
	}
	if depth > ceiling {
		depth = ceiling
	}
	return depth
}

// decayWeight is how much a piece of demand still counts for.
//
// A multiplier rather than a filter because the queue is ordered, not gated: a three-day-old
// binge is still worth preparing when the machine is idle, it simply must not outrank
// somebody who watched an episode an hour ago.
func decayWeight(lastViewed, now time.Time, cfg Config) float64 {
	age := now.Sub(lastViewed)
	switch {
	case age < 0:
		// A clock disagreement, not a prediction about the future.
		return 1
	case age <= cfg.StrongWindow:
		return 1
	case age <= cfg.UsefulWindow:
		return 0.75
	case age <= cfg.WeakWindow:
		return 0.4
	default:
		return 0
	}
}

// priorityForOffset is the look-ahead hierarchy: how much less an episode is worth for each
// step further from where somebody actually is.
func priorityForOffset(offset int) (int, string) {
	switch offset {
	case 0, 1:
		return PriorityNext, ReasonNext
	case 2:
		return PriorityAhead2, ReasonBinge
	case 3:
		return PriorityAhead3, ReasonBinge
	default:
		return PrioritySpeculative, ReasonBinge
	}
}

// BuildCandidates is the whole predictor: activities in, a small deduplicated queue out.
//
// Three things happen here and they are easy to conflate. Each viewer produces a *window* of
// episodes ahead of where they are, sized by how fast they watch. Those windows are then
// merged across the household, so an episode two people are approaching is one candidate
// rather than two. And the merged candidate is boosted for the agreement, because one scan
// now serves both of them.
func BuildCandidates(
	activities []SeriesActivity, index EpisodeIndex, now time.Time, cfg Config,
) []Candidate {
	if index == nil {
		return nil
	}
	merged := map[string]*Candidate{}
	users := map[string]map[string]bool{}

	for _, activity := range activities {
		weight := decayWeight(activity.LastViewed, now, cfg)
		if weight <= 0 {
			continue
		}
		depth := prefetchDepth(activity.EpisodesPerDay, cfg)

		// An unfinished episode is itself the first candidate; a finished one means the
		// window starts at the episode after it. Asking the index for one extra and dropping
		// the head is wrong here — the unfinished episode has an item id of its own that the
		// index would not return as "following".
		refs := index.Following(activity.SeriesID, activity.Season, activity.Episode, depth)
		window := make([]EpisodeRef, 0, depth+1)
		if activity.InProgress {
			window = append(window, EpisodeRef{
				ItemID:  currentItemID(index, activity),
				Season:  activity.Season,
				Episode: activity.Episode,
			})
		}
		window = append(window, refs...)

		for offset, ref := range window {
			if strings.TrimSpace(ref.ItemID) == "" {
				continue
			}
			if offset >= depth+boolToInt(activity.InProgress) {
				break
			}
			base, reason := priorityForOffset(offset)
			priority := int(float64(base) * weight)

			existing, found := merged[ref.ItemID]
			if !found {
				existing = &Candidate{
					ItemID:     ref.ItemID,
					SeriesID:   activity.SeriesID,
					Season:     ref.Season,
					Episode:    ref.Episode,
					Priority:   priority,
					Reason:     reason,
					LastViewed: activity.LastViewed,
				}
				merged[ref.ItemID] = existing
				users[ref.ItemID] = map[string]bool{}
			}
			// The strongest case for an episode wins; a second, weaker viewer never lowers
			// a candidate somebody else is about to reach.
			if priority > existing.Priority {
				existing.Priority, existing.Reason = priority, reason
			}
			if activity.LastViewed.After(existing.LastViewed) {
				existing.LastViewed = activity.LastViewed
			}
			users[ref.ItemID][activity.UserKey] = true
		}
	}

	out := make([]Candidate, 0, len(merged))
	for itemID, candidate := range merged {
		candidate.UserCount = len(users[itemID])
		if candidate.UserCount > 1 {
			// Agreement is the cheapest signal there is: the same single scan now serves
			// more than one person, so it is worth doing sooner.
			candidate.Priority += MultiUserBonus * (candidate.UserCount - 1)
			candidate.Reason = ReasonMultiUser
		}
		// Nothing predicted may reach live playback's priority. Live playback is a viewer
		// waiting; everything here is a guess about one.
		if candidate.Priority > MaxPredictedPriority {
			candidate.Priority = MaxPredictedPriority
		}
		out = append(out, *candidate)
	}
	sortCandidates(out)

	if cfg.QueueLimit > 0 && len(out) > cfg.QueueLimit {
		out = out[:cfg.QueueLimit]
	}
	return out
}

// currentItemID resolves the episode a viewer is part-way through. The index is asked for
// the episode at that exact position by requesting the one before it and taking the head,
// which keeps EpisodeIndex to a single method.
func currentItemID(index EpisodeIndex, activity SeriesActivity) string {
	refs := index.Following(activity.SeriesID, activity.Season, activity.Episode-1, 1)
	for _, ref := range refs {
		if ref.Season == activity.Season && ref.Episode == activity.Episode {
			return ref.ItemID
		}
	}
	return ""
}

// sortCandidates puts the queue in the order the worker should take it: priority first, then
// the most recent demand, then item id so the order is total and a test can assert on it.
func sortCandidates(candidates []Candidate) {
	sort.Slice(candidates, func(a, b int) bool {
		if candidates[a].Priority != candidates[b].Priority {
			return candidates[a].Priority > candidates[b].Priority
		}
		if !candidates[a].LastViewed.Equal(candidates[b].LastViewed) {
			return candidates[a].LastViewed.After(candidates[b].LastViewed)
		}
		return candidates[a].ItemID < candidates[b].ItemID
	})
}

func boolToInt(value bool) int {
	if value {
		return 1
	}
	return 0
}