memby/server/internal/credits/service.go

package credits

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"
)

// The service: one worker, one queue, and the rules about when it is allowed to run.
//
// Everything expensive in this package funnels through Process, and the shape of that
// function is the performance claim in miniature — resolve, check the marker, and in the
// common case stop there having touched no media at all. Only a candidate that survives the
// cache check reaches a decoder.

const (
	// livePlaybackDelay is how long a playback has to survive before it is worth scanning
	// for. Somebody browsing the launcher starts and abandons episodes constantly, and
	// scanning on the Play press would turn every one of those into disk activity. Waiting
	// costs nothing: the credits are forty minutes away.
	livePlaybackDelay = 45 * time.Second

	// idlePoll is how often the worker looks for work when the queue is empty. Coarse on
	// purpose — nothing here is urgent, and a tight loop on an idle NAS is exactly the sort
	// of background cost this package is supposed not to have.
	idlePoll = 30 * time.Second

	// busyBackoff is how long the worker stands down when the server is under load.
	busyBackoff = 2 * time.Minute

	// scanBudget bounds one candidate end to end. Past it the answer is not worth the
	// resources, and a scan that overruns is far more likely to be a pathological file than
	// a slow one.
	scanBudget = 90 * time.Second
)

// ResolvedMedia is everything the service needs to turn a candidate into a scan.
type ResolvedMedia struct {
	Version MediaVersion
	// URL is where the bytes are — Emby's direct stream route, since the gateway has no
	// filesystem access to the media.
	URL       string
	SeriesID  string
	Season    int
	Episode   int
	RuntimeMs int64
	// EmbyCreditsMs is a chapter marker Emby found itself, if any. When Emby already knows,
	// this subsystem must do nothing at all.
	EmbyCreditsMs int64
}

// MediaResolver turns an item id into something scannable. Backed by Emby; kept an interface
// so the scheduler, the queue and every rule above them can be tested with no server.
type MediaResolver interface {
	Resolve(ctx context.Context, itemID string) (ResolvedMedia, error)
}

// BehaviourSource reads the stops a household already made. Backed by the tracearr_sessions
// table, which is written by the For You import and by nothing here.
type BehaviourSource interface {
	Stops(ctx context.Context, itemID string) ([]StopEvent, error)
}

// LoadGauge answers whether the server is too busy for speculative work. Deliberately a
// single boolean: building a telemetry subsystem to answer it would cost more than the
// scanning it is meant to defer.
type LoadGauge interface {
	Busy(ctx context.Context) bool
}

// Deps is the service's wiring. Only Repository and MediaResolver are required; a service
// with no detector runs on behavioural evidence alone, which is what happens when ffmpeg is
// missing from the image.
type Deps struct {
	Repository Repository
	History    HistoryRepository
	Resolver   MediaResolver
	Source     CandidateSource
	Detector   Detector
	Behaviour  BehaviourSource
	Load       LoadGauge
	Log        *slog.Logger
	Config     Config
}

type Service struct {
	repo      Repository
	history   HistoryRepository
	resolver  MediaResolver
	source    CandidateSource
	detector  Detector
	behaviour BehaviourSource
	load      LoadGauge
	log       *slog.Logger
	cfg       Config
	cfgMu     sync.RWMutex

	queue  *Queue
	flight *flightGroup

	mu      sync.Mutex
	pending map[string]*pending

	// liveDelay is a field rather than the constant so a test does not have to wait
	// three quarters of a minute to prove that abandonment cancels a scan.
	liveDelay time.Duration
}

func New(deps Deps) *Service {
	cfg := NormaliseConfig(deps.Config)
	detector := deps.Detector
	if detector == nil {
		detector = noopDetector{}
	}
	log := deps.Log
	if log == nil {
		log = slog.Default()
	}
	return &Service{
		repo:      deps.Repository,
		history:   deps.History,
		resolver:  deps.Resolver,
		source:    deps.Source,
		detector:  detector,
		behaviour: deps.Behaviour,
		load:      deps.Load,
		log:       log.With("component", "credits"),
		cfg:       cfg,
		queue:     NewQueue(cfg.QueueLimit),
		flight:    newFlightGroup(),
		pending:   map[string]*pending{},
		liveDelay: livePlaybackDelay,
	}
}

// Marker is the read path, and the one the API calls. It is a single indexed lookup and it
// is what makes the steady state free: once a household's viewing has settled, almost every
// call to this subsystem is this function returning a row.
func (s *Service) Marker(ctx context.Context, itemID string) (Marker, bool, error) {
	if s == nil || s.repo == nil || itemID == "" {
		return Marker{}, false, nil
	}
	resolved, err := s.resolver.Resolve(ctx, itemID)
	if err != nil {
		return Marker{}, false, err
	}
	return s.repo.GetMarker(ctx, itemID, Fingerprint(resolved.Version))
}

// Refresh rebuilds the speculative queue from demand. Registered as a scheduled task, so its
// interval is the operator's to change and its last run is visible in the console.
//
// The returned detail is the scheduler's one-line summary and is empty when nothing changed,
// which is what stops a task running every ten minutes announcing itself every ten minutes.
func (s *Service) Refresh(ctx context.Context) (string, error) {
	if s.source == nil {
		return "", nil
	}
	candidates, err := s.source.Candidates(ctx)
	if err != nil {
		return "", err
	}
	// The cache check happens here as well as in the worker, and that is the point: an
	// episode whose marker already exists must never occupy a queue slot, or a household
	// that has been watching one show for a week ends up with a permanently full queue of
	// work that will all turn out to be unnecessary.
	wanted := make([]Candidate, 0, len(candidates))
	skipped := 0
	for _, candidate := range candidates {
		known, err := s.known(ctx, candidate.ItemID)
		if err != nil {
			// Trouble reading a marker is not a reason to drop the candidate; the worker
			// will check again and is the one that can afford to be wrong.
			s.log.Debug("marker lookup failed", "item_id", candidate.ItemID, "error", err)
		}
		if known {
			skipped++
			continue
		}
		wanted = append(wanted, candidate)
	}
	deferred := 0
	cfg := s.Configuration()
	if s.history != nil && cfg.RetryCooldown > 0 && len(wanted) > 0 {
		ids := make([]string, 0, len(wanted))
		for _, candidate := range wanted {
			ids = append(ids, candidate.ItemID)
		}
		recent, historyErr := s.history.RecentScanTimes(
			ctx, ids, time.Now().UTC().Add(-cfg.RetryCooldown),
		)
		if historyErr != nil {
			s.log.Debug("credits scan history unavailable", "error", historyErr)
		} else if len(recent) > 0 {
			eligible := wanted[:0]
			for _, candidate := range wanted {
				if _, coolingDown := recent[candidate.ItemID]; coolingDown {
					deferred++
					continue
				}
				eligible = append(eligible, candidate)
			}
			wanted = eligible
		}
	}
	s.queue.Replace(wanted)

	if len(wanted) == 0 {
		if deferred > 0 {
			return fmt.Sprintf("%d candidate%s cooling down, %d already known",
				deferred, plural(deferred), skipped), nil
		}
		return "", nil
	}
	for _, candidate := range wanted {
		s.log.Debug("credits_candidate",
			"item", candidate.ItemID, "reason", candidate.Reason,
			"priority", candidate.Priority, "users", candidate.UserCount)
	}
	return fmt.Sprintf("%d candidate%s queued, %d cooling down, %d already known",
		len(wanted), plural(len(wanted)), deferred, skipped), nil
}

// NotePlayback is the live signal, and the strongest one there is: somebody is watching this
// episode now, and if there is no marker they will reach the credits without a button.
//
// It does not scan immediately. A Play press is not yet a viewing — the launcher makes it
// trivially easy to start something and change your mind — so the candidate is held for
// livePlaybackDelay and raised only if the playback is still going. That is the difference
// between a subsystem that reads a file per curious button press and one that reads a file
// per episode actually watched.
func (s *Service) NotePlayback(ctx context.Context, itemID string) {
	if s == nil || itemID == "" {
		return
	}
	s.mu.Lock()
	if _, waiting := s.pending[itemID]; waiting {
		s.mu.Unlock()
		return
	}
	// Detached from the request context deliberately: the HTTP request that reported the
	// playback is over in milliseconds, and hanging the timer off it would cancel every
	// delayed scan the instant it returned.
	timerCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
	s.pending[itemID] = &pending{
		candidate: Candidate{
			ItemID:     itemID,
			Priority:   PriorityLive,
			Reason:     ReasonLivePlayback,
			UserCount:  1,
			LastViewed: time.Now().UTC(),
		},
		due:    time.Now().Add(s.liveDelay),
		cancel: cancel,
	}
	delay := s.liveDelay
	s.mu.Unlock()

	go func() {
		defer cancel()
		select {
		case <-timerCtx.Done():
			return
		case <-time.After(delay):
		}
		s.mu.Lock()
		entry, waiting := s.pending[itemID]
		delete(s.pending, itemID)
		s.mu.Unlock()
		if !waiting {
			return
		}
		// One last cache check before queueing. In the settled case the marker was already
		// there and this costs one indexed read instead of a queue slot.
		if known, err := s.known(timerCtx, itemID); err == nil && known {
			return
		}
		if s.queue.Push(entry.candidate) {
			s.log.Debug("credits_candidate",
				"item", itemID, "reason", ReasonLivePlayback, "priority", PriorityLive)
		}
	}()
}

// AbandonPlayback withdraws a playback that stopped before its delay elapsed. This is what
// makes the delay worth having: without it the timer would fire regardless and the episode
// somebody looked at for ten seconds would be scanned anyway.
func (s *Service) AbandonPlayback(itemID string) {
	if s == nil || itemID == "" {
		return
	}
	s.mu.Lock()
	entry, waiting := s.pending[itemID]
	delete(s.pending, itemID)
	s.mu.Unlock()
	if waiting && entry.cancel != nil {
		entry.cancel()
	}
}

// Run is the worker. One goroutine, for the whole gateway, for ever.
//
// A single worker is not a placeholder for a pool. Concurrent scans would multiply exactly
// the two costs this package is built to minimise — disk reads and decoder CPU — on a machine
// whose primary job is streaming video to televisions. Depth here buys nothing: the queue is
// a handful of episodes, and there are hours before anybody reaches them.
func (s *Service) Run(ctx context.Context) {
	for {
		if ctx.Err() != nil {
			return
		}
		candidate, found := s.queue.Claim()
		if !found {
			if !sleep(ctx, idlePoll) {
				return
			}
			continue
		}
		// Live playback goes ahead regardless of load: somebody is waiting, and the scan is
		// a couple of minutes of ranged reads against a file the server is already serving.
		// Everything else stands down.
		if candidate.Priority < PriorityLive && s.busy(ctx) {
			s.queue.Release(candidate.ItemID)
			s.queue.Push(candidate)
			s.log.Debug("credits scan deferred; server busy", "item", candidate.ItemID)
			if !sleep(ctx, busyBackoff) {
				return
			}
			continue
		}

		started := time.Now().UTC()
		scanCtx, cancel := context.WithTimeout(ctx, scanBudget)
		detection, stored, err := s.Process(scanCtx, candidate.ItemID)
		cancel()
		s.queue.Release(candidate.ItemID)
		s.recordAttempt(ctx, candidate, detection, stored, err, started)

		switch {
		case err != nil && errors.Is(err, context.Canceled):
			return
		case err != nil:
			s.log.Debug("credits scan failed",
				"item", candidate.ItemID, "reason", candidate.Reason, "error", err)
		case stored:
			s.log.Info("credits_detected",
				"item", candidate.ItemID,
				"marker_ms", detection.StartMs,
				"confidence", round2(detection.Confidence),
				"method", detection.Method,
				"frames", detection.FramesSampled,
				"duration_ms", detection.Elapsed.Milliseconds())
		default:
			s.log.Debug("credits not detected",
				"item", candidate.ItemID, "frames", detection.FramesSampled)
		}
	}
}

// Process is one candidate, start to finish. Single-flighted on the media version, so the
// four ways an episode can be asked for — live playback, a predicted next, another viewer's
// prediction, a refresh cycle — collapse into one scan.
//
// Returns whether anything was stored, which is not the same as whether anything was found:
// a detection below the confidence threshold, or one too close to an existing marker to be
// worth a write, is a successful scan that deliberately produces no database activity.
func (s *Service) Process(ctx context.Context, itemID string) (Detection, bool, error) {
	resolved, err := s.resolver.Resolve(ctx, itemID)
	if err != nil {
		return Detection{}, false, err
	}
	fingerprint := Fingerprint(resolved.Version)

	result, err := s.flight.Do(itemID+":"+fingerprint, func() (any, error) {
		detection, stored, err := s.process(ctx, itemID, fingerprint, resolved)
		return processResult{detection: detection, stored: stored}, err
	})
	outcome, _ := result.(processResult)
	return outcome.detection, outcome.stored, err
}

type processResult struct {
	detection Detection
	stored    bool
}

func (s *Service) process(
	ctx context.Context, itemID, fingerprint string, resolved ResolvedMedia,
) (Detection, bool, error) {
	// Emby already knows. Nothing to do, and nothing to store: the live path reads Emby's
	// chapters directly, so duplicating the answer here would be a row that could only ever
	// go stale.
	if resolved.EmbyCreditsMs > 0 {
		return Detection{Found: true, StartMs: resolved.EmbyCreditsMs, Method: MethodEmby}, false, nil
	}
	existing, found, err := s.repo.GetMarker(ctx, itemID, fingerprint)
	if err != nil {
		return Detection{}, false, err
	}
	// The common path, and the one everything is optimised for: already decided, no media
	// access, no decoder, no write.
	if found && existing.Confidence >= ConfidenceThreshold {
		return Detection{
			Found: true, StartMs: existing.CreditsStartMs,
			Confidence: existing.Confidence, Method: existing.DetectionMethod,
		}, false, nil
	}
	if resolved.RuntimeMs <= 0 {
		return Detection{}, false, nil
	}

	// Behaviour first, because it is free. On a well-watched show it can settle the question
	// without any media being opened at all, and where it cannot it still narrows the window
	// the decoder has to read.
	evidence := s.evidence(ctx, itemID, resolved.RuntimeMs)

	var visual Detection
	if !evidence.StandaloneMarker() || !evidence.Usable() {
		history, runtimes := s.seasonHistory(ctx, resolved)
		window := NarrowWindow(resolved.RuntimeMs, history, runtimes, evidence)
		visual, err = s.detector.Detect(ctx, MediaInfo{
			URL: resolved.URL, RuntimeMs: resolved.RuntimeMs, Window: window,
		})
		if err != nil {
			if errors.Is(err, ErrNoFFmpeg) {
				// Run on behaviour alone rather than failing. An image without a decoder is
				// a deliberate deployment, not a fault.
				s.log.Debug("visual credits detection unavailable")
			} else {
				return Detection{}, false, err
			}
		}
		visual.Window = window
	}

	detection, acceptable := Combine(visual, evidence)
	if !acceptable {
		return visual, false, nil
	}
	if found && !ShouldRewrite(existing, detection) {
		// Stability: the same answer, or a marginally different one, is not worth a write.
		return detection, false, nil
	}

	// A fingerprint too weak to detect a file replacement is one whose marker could outlive
	// the file it describes. Better to keep re-deriving it than to store something nothing
	// will ever invalidate.
	if resolved.Version.Weak() {
		s.log.Debug("credits marker withheld; weak fingerprint", "item", itemID)
		return detection, false, nil
	}

	now := time.Now().UTC()
	marker := Marker{
		ItemID:           itemID,
		MediaFingerprint: fingerprint,
		CreditsStartMs:   detection.StartMs,
		Confidence:       detection.Confidence,
		DetectionMethod:  detection.Method,
		SeriesID:         resolved.SeriesID,
		Season:           resolved.Season,
		CreatedAt:        now,
		UpdatedAt:        now,
	}
	if err := s.repo.SaveMarker(ctx, marker); err != nil {
		return detection, false, err
	}
	return detection, true, nil
}

// evidence reads the household's stops. Failures are swallowed: behaviour is an optimisation
// and an enhancement, never a precondition.
func (s *Service) evidence(ctx context.Context, itemID string, runtimeMs int64) BehaviourEvidence {
	if s.behaviour == nil {
		return BehaviourEvidence{}
	}
	stops, err := s.behaviour.Stops(ctx, itemID)
	if err != nil {
		s.log.Debug("behavioural evidence unavailable", "item", itemID, "error", err)
		return BehaviourEvidence{}
	}
	return AnalyseStops(stops, runtimeMs)
}

// seasonHistory reads what is already known about this episode's neighbours, and the runtime
// of each so the tail offsets can be compared. Failures degrade to a generic tail window.
func (s *Service) seasonHistory(
	ctx context.Context, resolved ResolvedMedia,
) ([]Marker, func(Marker) int64) {
	if resolved.SeriesID == "" || resolved.Season <= 0 {
		return nil, nil
	}
	const historyLimit = 6
	history, err := s.repo.SeasonMarkers(ctx, resolved.SeriesID, resolved.Season, historyLimit)
	if err != nil || len(history) == 0 {
		return nil, nil
	}
	// The neighbours' runtimes come from resolving them, which would be a request each. The
	// episodes of one season are within a minute or two of each other, so this episode's own
	// runtime is a good enough stand-in — and being slightly wrong here only widens or
	// narrows a window that carries a ninety-second margin either way.
	runtime := resolved.RuntimeMs
	return history, func(Marker) int64 { return runtime }
}

// known is the cache check in its cheapest form: does a marker exist for the current version
// of this item. One resolve, one indexed read.
func (s *Service) known(ctx context.Context, itemID string) (bool, error) {
	resolved, err := s.resolver.Resolve(ctx, itemID)
	if err != nil {
		return false, err
	}
	if resolved.EmbyCreditsMs > 0 {
		return true, nil
	}
	marker, found, err := s.repo.GetMarker(ctx, itemID, Fingerprint(resolved.Version))
	if err != nil {
		return false, err
	}
	return found && marker.Confidence >= ConfidenceThreshold, nil
}

func (s *Service) busy(ctx context.Context) bool {
	return s.load != nil && s.load.Busy(ctx)
}

// Configuration is a snapshot safe to expose through the admin API.
func (s *Service) Configuration() Config {
	s.cfgMu.RLock()
	defer s.cfgMu.RUnlock()
	return s.cfg
}

// Configure applies persisted operator tuning to future candidate refreshes and trims an
// over-full queue immediately. A scan already in flight is deliberately left alone.
func (s *Service) Configure(cfg Config) {
	cfg = NormaliseConfig(cfg)
	s.cfgMu.Lock()
	s.cfg = cfg
	s.cfgMu.Unlock()
	if source, ok := s.source.(ConfigurableCandidateSource); ok {
		source.Configure(cfg)
	}
	s.queue.SetLimit(cfg.QueueLimit)
}

func (s *Service) recordAttempt(
	ctx context.Context, candidate Candidate, detection Detection, stored bool, scanErr error,
	started time.Time,
) {
	if s.history == nil {
		return
	}
	outcome := "no_match"
	switch {
	case scanErr != nil:
		outcome = "failed"
	case stored:
		outcome = "detected"
	case detection.Found:
		outcome = "unchanged"
	}
	errorText := ""
	if scanErr != nil {
		errorText = scanErr.Error()
	}
	finished := time.Now().UTC()
	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer cancel()
	if err := s.history.SaveScanAttempt(writeCtx, ScanAttempt{
		ItemID: candidate.ItemID, SeriesID: candidate.SeriesID,
		Season: candidate.Season, Episode: candidate.Episode,
		Reason: candidate.Reason, Priority: candidate.Priority, Outcome: outcome,
		MarkerMs: detection.StartMs, Confidence: detection.Confidence,
		Method: detection.Method, Frames: detection.FramesSampled, Error: errorText,
		StartedAt: started, FinishedAt: finished,
	}); err != nil {
		s.log.Debug("credits scan history write failed", "item", candidate.ItemID, "error", err)
	}
}

// QueueDepth is what the console reads.
func (s *Service) QueueDepth() int { return s.queue.Len() }

// Pending is the queue in worker order, for the admin console.
func (s *Service) Pending() []Candidate { return s.queue.Snapshot() }

func sleep(ctx context.Context, duration time.Duration) bool {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func plural(count int) string {
	if count == 1 {
		return ""
	}
	return "s"
}

func round2(value float64) float64 {
	return float64(int(value*100+0.5)) / 100
}