memby/server/internal/foryou/service.go

// Package foryou keeps Tracearr-derived recommendation data warm in PostgreSQL.
//
// Imports and ranking happen away from television requests. PostgreSQL is also the
// queue: dirty_since records work that still needs doing, so a restart cannot lose it.
package foryou

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

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

const (
	importPageSize        = 100
	incrementalMaxPages   = 10
	incrementalMinPages   = 2
	unchangedPagesToStop  = 2
	preparedPoolReadLimit = 240
	preparedRowSize       = 20
	maxPreparedCandidates = 750

	// Increment only when stored eligibility, scoring, or explanation behavior changes.
	preparedAlgorithmVersion = "2026-08-23.1"
)

type ImportResult struct {
	Kind       string        `json:"kind"`
	Pages      int           `json:"pages"`
	Seen       int           `json:"seen"`
	Changed    int           `json:"changed"`
	Dirtied    int64         `json:"dirtied"`
	Removed    int64         `json:"removed"`
	Duration   time.Duration `json:"-"`
	DurationMs int64         `json:"durationMs"`
}

type Service struct {
	store         *store.Store
	tracearr      *tracearr.Client
	engine        *recommend.Engine
	emby          *emby.Client
	serviceCred   emby.Credentials
	log           *slog.Logger
	minRebuildAge time.Duration
	refreshAge    time.Duration
	location      *time.Location
	now           func() time.Time

	mu            sync.Mutex
	importRunning bool
	building      map[string]bool
}

// ConfigureHouseholdUsers enables background preparation for every enabled Emby user.
// The service token is kept server-side and is only used for read-only profile building.
func (s *Service) ConfigureHouseholdUsers(client *emby.Client, cred emby.Credentials) {
	s.emby = client
	s.serviceCred = cred
}

func New(
	st *store.Store,
	tracearrClient *tracearr.Client,
	engine *recommend.Engine,
	log *slog.Logger,
	minRebuildAge, refreshAge time.Duration,
) *Service {
	return &Service{
		store: st, tracearr: tracearrClient, engine: engine, log: log,
		minRebuildAge: minRebuildAge, refreshAge: refreshAge,
		location: time.Local, now: time.Now, building: map[string]bool{},
	}
}

// ConfigureTimeContext sets the household timezone used for day/time viewing habits.
func (s *Service) ConfigureTimeContext(location *time.Location) {
	if location != nil {
		s.location = location
		s.engine.Location = location
	}
}

func (s *Service) Running() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.importRunning || len(s.building) > 0
}

// Ping asks Tracearr for the smallest thing it will answer with.
//
// It exists so the integrations console can report whether Tracearr is reachable without
// reaching past this service for the client: the service owns the connection, and a
// console holding its own copy of the client would be a second place the address and key
// could be wrong. One user, one page — the answer is discarded and only the error matters.
func (s *Service) Ping(ctx context.Context) error {
	if s == nil || s.tracearr == nil {
		return errors.New("tracearr is not configured")
	}
	_, err := s.tracearr.Users(ctx, 1, 1)
	return err
}

func (s *Service) Stats(ctx context.Context) (store.ForYouStats, error) {
	return s.store.ForYouStats(ctx)
}

func (s *Service) Import(ctx context.Context, full bool) (result ImportResult, resultErr error) {
	s.mu.Lock()
	if s.importRunning {
		s.mu.Unlock()
		return result, errors.New("for you: a Tracearr import is already running")
	}
	s.importRunning = true
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		s.importRunning = false
		s.mu.Unlock()
	}()

	started := time.Now().UTC()
	result.Kind = "incremental"
	count, err := s.store.TracearrSessionCount(ctx)
	if err != nil {
		return result, err
	}
	if count == 0 {
		full = true
	}
	if full {
		result.Kind = "full"
	}
	state, err := s.store.TracearrImportState(ctx)
	if err != nil {
		return result, err
	}
	defer func() {
		now := time.Now().UTC()
		if resultErr != nil {
			state.LastError = resultErr.Error()
		} else {
			state.LastError = ""
			state.LastIncrementalAt = &now
			if full {
				state.LastFullAt = &now
			}
		}
		if err := s.store.SetTracearrImportState(context.WithoutCancel(ctx), state); err != nil {
			s.log.Error("could not record Tracearr import state", "error", err)
		}
		result.Duration = time.Since(started)
		result.DurationMs = result.Duration.Milliseconds()
	}()

	unchangedPages := 0
	terminalIdentities := map[string]store.RecommendationIdentity{}
	for pageNumber := 1; ; pageNumber++ {
		page, err := s.tracearr.Page(ctx, pageNumber, importPageSize)
		if err != nil {
			return result, err
		}
		result.Pages++
		result.Seen += len(page.Data)

		imported := make([]store.TracearrSession, 0, len(page.Data))
		keys := make([]store.TracearrSessionKey, 0, len(page.Data))
		for _, session := range page.Data {
			value, ok := importedSession(session, s.tracearr.ConfiguredServerID())
			if !ok {
				continue
			}
			imported = append(imported, value)
			keys = append(keys, store.TracearrSessionKey{
				ServerID: value.ServerID, SessionID: value.SessionID,
			})
		}
		current, err := s.store.TracearrSessionSignals(ctx, keys)
		if err != nil {
			return result, err
		}
		pageChanged := false
		for _, session := range imported {
			key := store.TracearrSessionKey{
				ServerID: session.ServerID, SessionID: session.SessionID,
			}
			previous, exists := current[key]
			if !bytes.Equal(previous.Fingerprint, session.SourceFingerprint) {
				pageChanged = true
				result.Changed++
			}
			if tracearrSessionTerminal(session) && (!exists || !previous.Terminal) {
				identity := store.RecommendationIdentity{
					TracearrUserID: session.UserID,
					Username:       session.Username,
				}
				terminalIdentities[identity.TracearrUserID+"|"+
					strings.ToLower(identity.Username)] = identity
			}
		}
		if err := s.store.UpsertTracearrSessions(ctx, imported, started); err != nil {
			return result, err
		}
		if pageChanged {
			unchangedPages = 0
		} else {
			unchangedPages++
		}

		reachedEnd := len(page.Data) == 0 ||
			(page.Meta.Total > 0 && pageNumber*importPageSize >= page.Meta.Total)
		if full && reachedEnd {
			break
		}
		if !full && (reachedEnd ||
			(pageNumber >= incrementalMinPages && unchangedPages >= unchangedPagesToStop) ||
			pageNumber >= incrementalMaxPages) {
			break
		}
	}

	if full {
		removed, err := s.store.DeleteTracearrSessionsNotSeenSince(
			ctx, s.tracearr.ConfiguredServerID(), started,
		)
		if err != nil {
			return result, err
		}
		result.Removed = removed
	}
	affected := make([]store.RecommendationIdentity, 0, len(terminalIdentities))
	for _, identity := range terminalIdentities {
		affected = append(affected, identity)
	}
	dirtied, err := s.store.MarkForYouProfilesDirtyByTracearrIdentity(ctx, affected)
	if err != nil {
		return result, err
	}
	result.Dirtied = dirtied
	s.log.Info("Tracearr import finished",
		"kind", result.Kind, "pages", result.Pages, "seen", result.Seen,
		"changed", result.Changed, "dirtied", result.Dirtied, "removed", result.Removed)
	return result, nil
}

func tracearrSessionTerminal(session store.TracearrSession) bool {
	state := strings.ToLower(strings.TrimSpace(session.State))
	return session.Watched || session.StoppedAt != nil ||
		state == "stopped" || state == "completed" || state == "complete" || state == "ended"
}

func (s *Service) Rebuild(ctx context.Context, sess store.Session, force bool) error {
	if !s.beginBuild(sess.EmbyUserID) {
		return nil
	}
	defer s.endBuild(sess.EmbyUserID)

	_, poolBuiltAt, dirtySince, algorithmVersion, err :=
		s.store.ForYouProfileTimes(ctx, sess.EmbyUserID)
	if err != nil {
		return err
	}
	algorithmChanged := algorithmVersion != preparedAlgorithmVersion
	if !force && !algorithmChanged && poolBuiltAt != nil &&
		time.Since(*poolBuiltAt) < s.minRebuildAge {
		return nil
	}
	if !force && !algorithmChanged && dirtySince == nil && poolBuiltAt != nil &&
		time.Since(*poolBuiltAt) < s.refreshAge {
		return nil
	}

	tracearrUserID, storedUsername, err := s.store.ForYouTracearrIdentity(ctx, sess.EmbyUserID)
	if err != nil {
		return err
	}
	username := strings.TrimSpace(sess.Username)
	if username == "" {
		username = storedUsername
	}
	imported, err := s.store.TracearrSessionsForUser(ctx, tracearrUserID, username)
	if err != nil {
		return err
	}
	sessions := make([]tracearr.Session, 0, len(imported))
	for _, value := range imported {
		sessions = append(sessions, tracearrSession(value))
	}
	result, err := s.engine.PrepareForYou(ctx, credentials(sess), username, sessions)
	if err != nil {
		_ = s.store.SetForYouError(context.WithoutCancel(ctx), sess.EmbyUserID, err)
		return err
	}

	profile, candidates, err := storedResult(sess.EmbyUserID, time.Now().UTC(), result)
	if err != nil {
		return err
	}
	for _, mapping := range result.Mappings {
		if err := s.store.UpdateTracearrSessionMapping(ctx, store.TracearrSessionKey{
			ServerID: mapping.ServerID, SessionID: mapping.SessionID,
		}, mapping.ItemID, mapping.SeriesID); err != nil {
			return err
		}
	}
	if err := s.store.ReplaceForYouPool(ctx, profile, candidates); err != nil {
		_ = s.store.SetForYouError(context.WithoutCancel(ctx), sess.EmbyUserID, err)
		return err
	}
	s.log.Info("For You pool rebuilt",
		"user", sess.EmbyUserID, "sessions", len(sessions), "candidates", len(candidates))
	return nil
}

func (s *Service) PreparedRows(
	ctx context.Context,
	sess store.Session,
	minutes int,
) ([]recommend.Row, bool, bool, error) {
	items, builtAt, err := s.store.PreparedForYou(
		ctx, sess.EmbyUserID, minutes, preparedPoolReadLimit,
	)
	if err != nil {
		return nil, false, false, err
	}
	if len(items) == 0 {
		return nil, false, builtAt != nil && time.Since(*builtAt) >= s.refreshAge, nil
	}
	items = rankPreparedForTime(items, s.now(), s.location)
	rows := buildPreparedRows(items, minutes, s.engine.MinRowItems)
	stale := builtAt == nil || time.Since(*builtAt) >= s.refreshAge
	return rows, len(rows) > 0, stale, nil
}

func rankPreparedForTime(
	items []store.PreparedForYouItem,
	now time.Time,
	location *time.Location,
) []store.PreparedForYouItem {
	if len(items) < 2 || len(items[0].ContextAffinity) == 0 {
		return items
	}
	var profile recommend.ContextAffinityProfile
	if err := json.Unmarshal(items[0].ContextAffinity, &profile); err != nil ||
		len(profile.Slots) == 0 {
		return items
	}
	type contextualItem struct {
		item       store.PreparedForYouItem
		score      float64
		contextRaw float64
		confidence float64
	}
	ranked := make([]contextualItem, 0, len(items))
	var maxContext float64
	for _, value := range items {
		decoded := recommend.Decode([]json.RawMessage{value.Payload})
		raw, confidence := 0.0, 0.0
		if len(decoded) == 1 && value.ReasonKind != "pick-up" {
			raw, confidence = profile.Score(decoded[0], now, location)
			if raw > maxContext {
				maxContext = raw
			}
		}
		ranked = append(ranked, contextualItem{
			item: value, score: value.BaseScore,
			contextRaw: raw, confidence: confidence,
		})
	}
	if maxContext == 0 {
		return items
	}
	for i := range ranked {
		ranked[i].score += 1.5 * ranked[i].confidence * ranked[i].contextRaw / maxContext
		if ranked[i].contextRaw > 0 && ranked[i].confidence >= 0.4 {
			ranked[i].item.RecommendationReason += " · fits what you watch around this time"
		}
	}
	sort.SliceStable(ranked, func(i, j int) bool {
		if ranked[i].score != ranked[j].score {
			return ranked[i].score > ranked[j].score
		}
		return ranked[i].item.BaseRank < ranked[j].item.BaseRank
	})
	out := make([]store.PreparedForYouItem, 0, len(ranked))
	for _, entry := range ranked {
		out = append(out, entry.item)
	}
	return out
}

func (s *Service) MarkDirty(ctx context.Context, sess store.Session) {
	if err := s.store.MarkForYouDirty(ctx, sess.EmbyUserID, sess.Username); err != nil {
		s.log.Warn("could not mark For You dirty", "user", sess.EmbyUserID, "error", err)
	}
}

func (s *Service) RefreshAsync(sess store.Session, force bool) {
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		if err := s.Rebuild(ctx, sess, force); err != nil {
			s.log.Warn("For You background rebuild failed", "user", sess.EmbyUserID, "error", err)
		}
	}()
}

// RebuildResult is what one pass over the household did.
//
// Counted because a rebuild that quietly failed for every viewer and one that succeeded
// for every viewer are the same "no error" from outside: a per-user failure is logged and
// swallowed on purpose — one viewer's broken profile must not stop the rest being built —
// so the count is the only thing that can say it happened.
type RebuildResult struct {
	Users   int
	Built   int
	Failed  int
	Skipped int
}

func (s *Service) RebuildAll(ctx context.Context, force bool) (RebuildResult, error) {
	result := RebuildResult{}
	users, err := s.recommendationUsers(ctx)
	if err != nil {
		return result, err
	}
	result.Users = len(users)
	for _, user := range users {
		if err := s.Rebuild(ctx, user, force); err != nil {
			result.Failed++
			s.log.Warn("For You user rebuild failed", "user", user.EmbyUserID, "error", err)
			continue
		}
		result.Built++
	}
	return result, nil
}

// RebuildOutdated refreshes only profiles produced by an older algorithm. This makes
// startup migrations deterministic without rebuilding every fresh profile on every boot.
func (s *Service) RebuildOutdated(ctx context.Context) error {
	users, err := s.recommendationUsers(ctx)
	if err != nil {
		return err
	}
	for _, user := range users {
		_, _, _, version, stateErr := s.store.ForYouProfileTimes(ctx, user.EmbyUserID)
		if stateErr != nil {
			return stateErr
		}
		if version == preparedAlgorithmVersion {
			continue
		}
		if err := s.Rebuild(ctx, user, true); err != nil {
			s.log.Warn("outdated For You rebuild failed", "user", user.EmbyUserID, "error", err)
		}
	}
	return nil
}

func (s *Service) MarkAllDirty(ctx context.Context) {
	if err := s.store.MarkAllForYouProfilesDirty(ctx); err != nil {
		s.log.Warn("could not mark For You profiles dirty", "error", err)
	}
	users, err := s.store.ActiveRecommendationUsers(ctx)
	if err != nil {
		s.log.Warn("could not list For You users", "error", err)
		return
	}
	for _, user := range users {
		s.MarkDirty(ctx, user)
	}
}

func (s *Service) recommendationUsers(ctx context.Context) ([]store.Session, error) {
	active, err := s.store.ActiveRecommendationUsers(ctx)
	if err != nil {
		return nil, err
	}
	byID := make(map[string]store.Session, len(active))
	for _, user := range active {
		byID[user.EmbyUserID] = user
	}
	if s.emby == nil || strings.TrimSpace(s.serviceCred.Token) == "" {
		return active, nil
	}

	embyUsers, err := s.emby.Users(ctx, s.serviceCred)
	if err != nil {
		s.log.Warn("Emby user catalogue unavailable; using signed-in users", "error", err)
		return active, nil
	}
	tracearrUsers, traceErr := s.allTracearrUsers(ctx)
	if traceErr != nil {
		s.log.Warn("Tracearr users unavailable; preparing Emby-only profiles", "error", traceErr)
	}
	traceByName := map[string]tracearr.User{}
	for _, user := range tracearrUsers {
		key := strings.ToLower(strings.TrimSpace(user.Username))
		current, exists := traceByName[key]
		if key != "" && (!exists || user.SessionCount > current.SessionCount) {
			traceByName[key] = user
		}
	}

	for _, user := range embyUsers {
		if user.Policy.IsDisabled || strings.TrimSpace(user.ID) == "" {
			continue
		}
		matched := traceByName[strings.ToLower(strings.TrimSpace(user.Name))]
		if traceErr == nil {
			if err := s.store.MatchRecommendationUser(
				ctx, user.ID, user.Name, matched.ID, matched.Username,
			); err != nil {
				return nil, err
			}
		} else if err := s.store.MarkForYouDirty(ctx, user.ID, user.Name); err != nil {
			return nil, err
		}
		byID[user.ID] = store.Session{
			EmbyUserID: user.ID,
			EmbyToken:  s.serviceCred.Token,
			Username:   user.Name,
			DeviceID:   builderDeviceID,
			DeviceName: builderDeviceName,
		}
	}
	out := make([]store.Session, 0, len(byID))
	for _, user := range byID {
		out = append(out, user)
	}
	sort.Slice(out, func(i, j int) bool {
		return strings.ToLower(out[i].Username) < strings.ToLower(out[j].Username)
	})
	return out, nil
}

func (s *Service) allTracearrUsers(ctx context.Context) ([]tracearr.User, error) {
	const pageSize = 100
	out := []tracearr.User{}
	for pageNumber := 1; ; pageNumber++ {
		page, err := s.tracearr.Users(ctx, pageNumber, pageSize)
		if err != nil {
			return nil, err
		}
		out = append(out, page.Data...)
		if len(page.Data) == 0 || page.Meta.Total <= pageNumber*pageSize {
			return out, nil
		}
	}
}

func buildPreparedRows(
	items []store.PreparedForYouItem,
	minutes, minRowItems int,
) []recommend.Row {
	if len(items) == 0 {
		return nil
	}
	eligibleItems := make([]store.PreparedForYouItem, 0, len(items))
	for _, item := range items {
		if item.ReasonKind == "pick-up" &&
			!strings.Contains(item.RecommendationReason, "season 1") {
			continue
		}
		eligibleItems = append(eligibleItems, item)
	}
	items = eligibleItems
	used := map[string]bool{}
	rows := make([]recommend.Row, 0, 6)
	appendRowWithMinimum := func(
		id, title string,
		candidates []store.PreparedForYouItem,
		minimum int,
	) bool {
		selected := make([]store.PreparedForYouItem, 0, preparedRowSize)
		for _, item := range candidates {
			if used[item.ItemID] {
				continue
			}
			selected = append(selected, item)
			if len(selected) == preparedRowSize {
				break
			}
		}
		if len(selected) < minimum {
			return false
		}
		rowItems := make([]json.RawMessage, 0, len(selected))
		for _, item := range selected {
			used[item.ItemID] = true
			rowItems = append(rowItems, recommend.EnrichPreparedRecommendation(
				item.Payload, item.RecommendationReason, item.CompatibilityLabel, minutes,
			))
		}
		rows = append(rows, recommend.Row{
			ID: id, Title: title, Kind: "for-you", Items: rowItems,
		})
		return true
	}
	appendRow := func(id, title string, candidates []store.PreparedForYouItem) bool {
		return appendRowWithMinimum(id, title, candidates, minRowItems)
	}

	pickups := make([]store.PreparedForYouItem, 0, preparedRowSize)
	for _, item := range items {
		// The reason guard also keeps an already-prepared pool from an older server
		// version from briefly promoting later-season lapses after deployment.
		if item.ReasonKind == "pick-up" &&
			strings.Contains(item.RecommendationReason, "season 1") {
			pickups = append(pickups, item)
		}
	}
	// A pickup is valuable even when only one genuinely abandoned, unfinished series
	// qualifies. Unlike generic recommendations, padding this shelf would make it lie.
	appendRowWithMinimum("for-you:pick-up", "Pick this show up again", pickups, 1)

	topTitle := "Top picks for you"
	if minutes > 0 {
		topTitle = fmt.Sprintf("Top picks that fit in %d minutes", minutes)
	}
	appendRow("for-you:picks", topTitle, items)

	appendGroupedRows := func(
		prefix string,
		key func(store.PreparedForYouItem) string,
		title func(store.PreparedForYouItem) string,
		maxRows int,
		filter func(store.PreparedForYouItem) bool,
	) {
		groups := map[string][]store.PreparedForYouItem{}
		order := []string{}
		first := map[string]store.PreparedForYouItem{}
		for _, item := range items {
			if used[item.ItemID] || !filter(item) {
				continue
			}
			groupKey := key(item)
			if groupKey == "" {
				continue
			}
			if _, exists := groups[groupKey]; !exists {
				order = append(order, groupKey)
				first[groupKey] = item
			}
			groups[groupKey] = append(groups[groupKey], item)
		}
		added := 0
		for _, groupKey := range order {
			if added == maxRows {
				return
			}
			candidates := groups[groupKey]
			if len(candidates) < minRowItems {
				continue
			}
			if appendRow(prefix+rowKey(groupKey), title(first[groupKey]), candidates) {
				added++
			}
		}
	}

	appendGroupedRows(
		"for-you:because:",
		func(item store.PreparedForYouItem) string { return item.ReasonSourceItemID },
		func(item store.PreparedForYouItem) string {
			switch item.ReasonKind {
			case "completed-title":
				return "Because you finished " + item.ReasonSourceTitle
			case "favourite-title":
				return "Because you like " + item.ReasonSourceTitle
			default:
				return "Because you watched " + item.ReasonSourceTitle
			}
		},
		2,
		func(item store.PreparedForYouItem) bool {
			return strings.TrimSpace(item.ReasonSourceTitle) != ""
		},
	)
	appendGroupedRows(
		"for-you:genre:",
		func(item store.PreparedForYouItem) string {
			return strings.ToLower(strings.TrimSpace(item.ReasonGenre))
		},
		func(item store.PreparedForYouItem) string {
			return "More " + item.ReasonGenre + " for you"
		},
		2,
		func(item store.PreparedForYouItem) bool {
			return strings.TrimSpace(item.ReasonGenre) != ""
		},
	)
	beforeBed := make([]store.PreparedForYouItem, 0, len(items))
	for _, item := range items {
		if item.ReasonKind != "pick-up" && item.RuntimeMinutes >= 15 &&
			item.RuntimeMinutes <= 50 {
			decoded := recommend.Decode([]json.RawMessage{item.Payload})
			if len(decoded) == 1 && strings.EqualFold(decoded[0].Type, "Series") {
				beforeBed = append(beforeBed, item)
			}
		}
	}
	appendRow(
		"for-you:one-episode-before-bed",
		"One episode before bed",
		beforeBed,
	)

	// The prepared pool contains only unseen candidates. Taking the remaining strong
	// matches produces a genuine library-discovery shelf; request-time impression
	// fatigue then rotates repeatedly ignored posters out of its leading positions.
	hidden := make([]store.PreparedForYouItem, 0, len(items))
	for _, item := range items {
		if !used[item.ItemID] && item.ReasonKind != "pick-up" {
			hidden = append(hidden, item)
		}
	}
	appendRow("for-you:hidden", "Hidden in your library", hidden)

	compatible := make([]store.PreparedForYouItem, 0, len(items))
	for _, item := range items {
		if !used[item.ItemID] && item.CompatibilityScore > 0.2 {
			compatible = append(compatible, item)
		}
	}
	appendRow("for-you:tv-ready", "Plays well on this TV", compatible)
	return rows
}

func rowKey(value string) string {
	value = strings.TrimSpace(strings.ToLower(value))
	var b strings.Builder
	for _, r := range value {
		if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' {
			b.WriteRune(r)
		}
	}
	if b.Len() == 0 {
		sum := sha256.Sum256([]byte(value))
		return fmt.Sprintf("%x", sum[:6])
	}
	return b.String()
}

// ImportIfDue runs an import only when the persisted state says one is owed. Due-ness is
// measured from the last recorded import rather than from process uptime, so a restart —
// a redeploy, a crash loop, a container the healthcheck bounced — cannot buy an extra
// pass, and a gateway that is restarted more often than fullEvery still reconciles.
func (s *Service) ImportIfDue(
	ctx context.Context,
	importEvery, fullEvery time.Duration,
) (ImportResult, bool, error) {
	state, err := s.store.TracearrImportState(ctx)
	if err != nil {
		return ImportResult{}, false, err
	}
	full, due := importDue(state, time.Now().UTC(), importEvery, fullEvery)
	if !due {
		s.log.Debug("Tracearr import not due",
			"lastIncremental", state.LastIncrementalAt, "lastFull", state.LastFullAt)
		return ImportResult{}, false, nil
	}
	result, err := s.Import(ctx, full)
	return result, err == nil, err
}

// importDue is the whole scheduling rule, kept pure so it can be tested without a
// database. A stamp in the future is treated as due: a clock correction must not be able
// to strand the importer for an arbitrary length of time.
func importDue(
	state store.TracearrImportState,
	now time.Time,
	importEvery, fullEvery time.Duration,
) (full bool, due bool) {
	elapsed := func(at *time.Time) (time.Duration, bool) {
		if at == nil {
			return 0, false
		}
		return now.Sub(*at), true
	}
	if fullEvery > 0 {
		since, recorded := elapsed(state.LastFullAt)
		if !recorded || since >= fullEvery || since < 0 {
			return true, true
		}
	}
	if importEvery > 0 {
		since, recorded := elapsed(state.LastIncrementalAt)
		if !recorded || since >= importEvery || since < 0 {
			return false, true
		}
	}
	return false, false
}

// Schedule used to own both the Tracearr import ticker and the daily rebuild timer, and
// nextDailyRebuild was when the household's off-peak hour next came round. Both are gone:
// the import and the rebuild are scheduler tasks now (see api.RegisterIntegrationTasks),
// because a ticker in here could report nothing to an operator, could not be started by
// hand, and — since the tasks carry an integration id — could not be counted towards
// Tracearr's run history. The "is the rebuild due" rule moved with the work, to
// api.forYouRebuildDue, and is still pure and still tested.

func (s *Service) beginBuild(userID string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.building[userID] {
		return false
	}
	s.building[userID] = true
	return true
}

func (s *Service) endBuild(userID string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.building, userID)
}

func importedSession(session tracearr.Session, configuredServerID string) (store.TracearrSession, bool) {
	if strings.TrimSpace(session.ID) == "" {
		return store.TracearrSession{}, false
	}
	serverID := strings.TrimSpace(session.ServerID)
	if serverID == "" {
		serverID = strings.TrimSpace(configuredServerID)
	}
	if serverID == "" {
		serverID = "default"
	}
	startedAt := parsedTime(session.StartedAt)
	stoppedAt := parsedTime(session.StoppedAt)
	fingerprintRaw, _ := json.Marshal(session)
	fingerprint := sha256.Sum256(fingerprintRaw)
	return store.TracearrSession{
		ServerID: serverID, SessionID: session.ID, UserID: session.User.ID,
		Username: session.User.Username, State: session.State,
		MediaType: session.MediaType, MediaTitle: session.MediaTitle,
		ShowTitle: session.ShowTitle, SeasonNumber: session.SeasonNumber,
		EpisodeNumber: session.EpisodeNumber, ProductionYear: session.Year,
		StartedAt: startedAt, StoppedAt: stoppedAt,
		DurationMs: int64(session.DurationMs), ProgressMs: int64(session.ProgressMs),
		TotalDurationMs: int64(session.TotalDurationMs), Watched: session.Watched,
		Device: session.Device, Player: session.Player, Product: session.Product,
		Platform: session.Platform, IsTranscode: session.IsTranscode,
		VideoDecision: session.VideoDecision, AudioDecision: session.AudioDecision,
		SourceVideoCodec: session.SourceVideoCodec, SourceAudioCodec: session.SourceAudioCodec,
		SourceFingerprint: fingerprint[:],
	}, true
}

func tracearrSession(session store.TracearrSession) tracearr.Session {
	value := tracearr.Session{
		ID: session.SessionID, ServerID: session.ServerID, State: session.State,
		MediaType: session.MediaType, MediaTitle: session.MediaTitle,
		ShowTitle: session.ShowTitle, SeasonNumber: session.SeasonNumber,
		EpisodeNumber: session.EpisodeNumber, Year: session.ProductionYear,
		DurationMs:      tracearr.FlexibleInt64(session.DurationMs),
		ProgressMs:      tracearr.FlexibleInt64(session.ProgressMs),
		TotalDurationMs: tracearr.FlexibleInt64(session.TotalDurationMs),
		Watched:         session.Watched, Device: session.Device, Player: session.Player,
		Product: session.Product, Platform: session.Platform,
		IsTranscode: session.IsTranscode, VideoDecision: session.VideoDecision,
		AudioDecision: session.AudioDecision, SourceVideoCodec: session.SourceVideoCodec,
		SourceAudioCodec: session.SourceAudioCodec,
	}
	if session.StartedAt != nil {
		value.StartedAt = session.StartedAt.UTC().Format(time.RFC3339Nano)
	}
	if session.StoppedAt != nil {
		value.StoppedAt = session.StoppedAt.UTC().Format(time.RFC3339Nano)
	}
	value.User.ID = session.UserID
	value.User.Username = session.Username
	return value
}

func storedResult(
	userID string,
	builtAt time.Time,
	result recommend.PreparedResult,
) (store.RecommendationProfile, []store.ForYouCandidate, error) {
	genre, err := json.Marshal(result.Profile.GenreAffinity)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	title, err := json.Marshal(result.Profile.TitleAffinity)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	studio, err := json.Marshal(result.Profile.StudioAffinity)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	codecs, err := json.Marshal(result.Profile.CodecOutcomes)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	contextAffinity, err := json.Marshal(result.Profile.ContextAffinity)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	weightedProfile, err := json.Marshal(result.Profile.Weighted)
	if err != nil {
		return store.RecommendationProfile{}, nil, err
	}
	profile := store.RecommendationProfile{
		EmbyUserID: userID, TracearrUserID: result.Profile.TracearrUserID,
		TracearrUsername:      result.Profile.TracearrUsername,
		SourceSessionCount:    result.Profile.SourceSessionCount,
		MeanCompletionRatio:   result.Profile.MeanCompletionRatio,
		TypicalSessionMinutes: result.Profile.TypicalSessionMinutes,
		GenreAffinity:         genre, TitleAffinity: title, StudioAffinity: studio,
		ContextAffinity: contextAffinity, CodecOutcomes: codecs,
		WeightedProfile:  weightedProfile,
		AlgorithmVersion: preparedAlgorithmVersion,
		SignalsThrough:   result.Profile.SignalsThrough, BuiltAt: builtAt,
	}
	candidateCount := min(len(result.Candidates), maxPreparedCandidates)
	candidates := make([]store.ForYouCandidate, 0, candidateCount)
	for _, value := range result.Candidates[:candidateCount] {
		candidates = append(candidates, store.ForYouCandidate{
			ItemID: value.ItemID, BaseRank: value.BaseRank, BaseScore: value.BaseScore,
			RuntimeMinutes: value.RuntimeMinutes, AffinityScore: value.AffinityScore,
			CompatibilityScore: value.CompatibilityScore,
			CompatibilityLabel: value.CompatibilityLabel, ReasonKind: value.ReasonKind,
			ReasonGenre:           value.ReasonGenre,
			ReasonSourceSessionID: value.ReasonSourceSessionID,
			ReasonSourceItemID:    value.ReasonSourceItemID,
			ReasonSourceTitle:     value.ReasonSourceTitle,
			RecommendationReason:  value.RecommendationReason,
		})
	}
	return profile, candidates, nil
}

func parsedTime(value string) *time.Time {
	parsed, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(value))
	if err != nil {
		return nil
	}
	parsed = parsed.UTC()
	return &parsed
}

// builderDeviceID and builderDeviceName stand in for a television when the gateway
// rebuilds a viewer's rows on its own. It is the server asking, so Emby records it under
// the gateway's client name rather than as a set in the house — see Credentials.Gateway.
const (
	builderDeviceID   = "memby-for-you-builder"
	builderDeviceName = "MbyGateway For You"
)

func credentials(sess store.Session) emby.Credentials {
	return emby.Credentials{
		UserID: sess.EmbyUserID, Token: sess.EmbyToken,
		DeviceID: sess.DeviceID, DeviceName: sess.DeviceName,
		Gateway: sess.DeviceID == builderDeviceID,
	}
}