memby/server/internal/store/for_you.go

package store

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

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

const tracearrImportStateKey = "tracearr_import_state"

type TracearrSessionKey struct {
	ServerID  string
	SessionID string
}

type TracearrSessionSignal struct {
	Fingerprint []byte
	Terminal    bool
}

type RecommendationIdentity struct {
	TracearrUserID string
	Username       string
}

type TracearrSession struct {
	ServerID          string
	SessionID         string
	UserID            string
	Username          string
	State             string
	MediaType         string
	MediaTitle        string
	ShowTitle         string
	SeasonNumber      *int
	EpisodeNumber     *int
	ProductionYear    *int
	StartedAt         *time.Time
	StoppedAt         *time.Time
	DurationMs        int64
	ProgressMs        int64
	TotalDurationMs   int64
	Watched           bool
	Device            string
	Player            string
	Product           string
	Platform          string
	IsTranscode       bool
	VideoDecision     string
	AudioDecision     string
	SourceVideoCodec  string
	SourceAudioCodec  string
	EmbyItemID        string
	EmbySeriesID      string
	SourceFingerprint []byte
}

type TracearrImportState struct {
	LastIncrementalAt *time.Time `json:"lastIncrementalAt,omitempty"`
	LastFullAt        *time.Time `json:"lastFullAt,omitempty"`
	LastError         string     `json:"lastError,omitempty"`
	// LastRebuildAt is when the household's For You rows were last rebuilt in full.
	//
	// It lives in this document rather than in a table of its own because it is the same
	// kind of fact as the two stamps above it — where the For You pipeline has got to —
	// and because the daily rebuild is now a scheduled task, which means the alternative
	// was inferring "did today's rebuild happen" from run history that also records the
	// ticks on which it correctly declined to run.
	LastRebuildAt *time.Time `json:"lastRebuildAt,omitempty"`
}

type RecommendationProfile struct {
	EmbyUserID            string
	TracearrUserID        string
	TracearrUsername      string
	SourceSessionCount    int
	MeanCompletionRatio   float64
	TypicalSessionMinutes int
	GenreAffinity         json.RawMessage
	TitleAffinity         json.RawMessage
	StudioAffinity        json.RawMessage
	ContextAffinity       json.RawMessage
	CodecOutcomes         json.RawMessage
	WeightedProfile       json.RawMessage
	AlgorithmVersion      string
	SignalsThrough        *time.Time
	BuiltAt               time.Time
}

type ForYouCandidate struct {
	ItemID                string
	BaseRank              int
	BaseScore             float64
	RuntimeMinutes        int
	AffinityScore         float64
	CompatibilityScore    float64
	CompatibilityLabel    string
	ReasonKind            string
	ReasonGenre           string
	ReasonSourceSessionID string
	ReasonSourceItemID    string
	ReasonSourceTitle     string
	RecommendationReason  string
}

type PreparedForYouItem struct {
	ItemID               string
	BaseRank             int
	BaseScore            float64
	AffinityScore        float64
	Payload              json.RawMessage
	RuntimeMinutes       int
	CompatibilityScore   float64
	CompatibilityLabel   string
	RecommendationReason string
	ReasonKind           string
	ReasonGenre          string
	ReasonSourceItemID   string
	ReasonSourceTitle    string
	ContextAffinity      json.RawMessage
}

type ForYouStats struct {
	TracearrSessions int64      `json:"tracearrSessions"`
	Profiles         int64      `json:"profiles"`
	Candidates       int64      `json:"candidates"`
	LastFullImport   *time.Time `json:"lastFullImport,omitempty"`
}

func (s *Store) TracearrSessionCount(ctx context.Context) (int64, error) {
	var count int64
	if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM tracearr_sessions`).Scan(&count); err != nil {
		return 0, fmt.Errorf("store: count tracearr sessions: %w", err)
	}
	return count, nil
}

func (s *Store) TracearrFingerprints(
	ctx context.Context,
	keys []TracearrSessionKey,
) (map[TracearrSessionKey][]byte, error) {
	out := make(map[TracearrSessionKey][]byte, len(keys))
	if len(keys) == 0 {
		return out, nil
	}
	servers := make([]string, 0, len(keys))
	ids := make([]string, 0, len(keys))
	for _, key := range keys {
		servers = append(servers, key.ServerID)
		ids = append(ids, key.SessionID)
	}
	rows, err := s.pool.Query(ctx, `
		SELECT current.server_id, current.tracearr_session_id, current.source_fingerprint
		FROM tracearr_sessions current
		JOIN unnest($1::text[], $2::text[]) wanted(server_id, session_id)
		  ON current.server_id = wanted.server_id
		 AND current.tracearr_session_id = wanted.session_id`,
		servers, ids)
	if err != nil {
		return nil, fmt.Errorf("store: tracearr fingerprints: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var key TracearrSessionKey
		var fingerprint []byte
		if err := rows.Scan(&key.ServerID, &key.SessionID, &fingerprint); err != nil {
			return nil, err
		}
		out[key] = fingerprint
	}
	return out, rows.Err()
}

func (s *Store) TracearrSessionSignals(
	ctx context.Context,
	keys []TracearrSessionKey,
) (map[TracearrSessionKey]TracearrSessionSignal, error) {
	out := make(map[TracearrSessionKey]TracearrSessionSignal, len(keys))
	if len(keys) == 0 {
		return out, nil
	}
	servers := make([]string, 0, len(keys))
	ids := make([]string, 0, len(keys))
	for _, key := range keys {
		servers = append(servers, key.ServerID)
		ids = append(ids, key.SessionID)
	}
	rows, err := s.pool.Query(ctx, `
		SELECT current.server_id, current.tracearr_session_id, current.source_fingerprint,
			(current.watched OR current.stopped_at IS NOT NULL OR
			 lower(current.state) IN ('stopped', 'completed', 'complete', 'ended'))
		FROM tracearr_sessions current
		JOIN unnest($1::text[], $2::text[]) wanted(server_id, session_id)
		  ON current.server_id = wanted.server_id
		 AND current.tracearr_session_id = wanted.session_id`,
		servers, ids)
	if err != nil {
		return nil, fmt.Errorf("store: tracearr session signals: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var key TracearrSessionKey
		var signal TracearrSessionSignal
		if err := rows.Scan(
			&key.ServerID, &key.SessionID, &signal.Fingerprint, &signal.Terminal,
		); err != nil {
			return nil, err
		}
		out[key] = signal
	}
	return out, rows.Err()
}

func (s *Store) UpsertTracearrSessions(
	ctx context.Context,
	sessions []TracearrSession,
	seenAt time.Time,
) error {
	if len(sessions) == 0 {
		return nil
	}
	batch := &pgx.Batch{}
	for _, session := range sessions {
		batch.Queue(`
			INSERT INTO tracearr_sessions (
				server_id, tracearr_session_id, tracearr_user_id, username, state,
				media_type, media_title, show_title, season_number, episode_number,
				production_year, started_at, stopped_at, duration_ms, progress_ms,
				total_duration_ms, watched, device, player, product, platform,
				is_transcode, video_decision, audio_decision, source_video_codec,
				source_audio_codec, emby_item_id, emby_series_id, source_fingerprint,
				source_seen_at, imported_at, updated_at
			) VALUES (
				$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,
				$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,now(),now()
			)
			ON CONFLICT (server_id, tracearr_session_id) DO UPDATE SET
				tracearr_user_id = EXCLUDED.tracearr_user_id,
				username = EXCLUDED.username,
				state = EXCLUDED.state,
				media_type = EXCLUDED.media_type,
				media_title = EXCLUDED.media_title,
				show_title = EXCLUDED.show_title,
				season_number = EXCLUDED.season_number,
				episode_number = EXCLUDED.episode_number,
				production_year = EXCLUDED.production_year,
				started_at = EXCLUDED.started_at,
				stopped_at = EXCLUDED.stopped_at,
				duration_ms = EXCLUDED.duration_ms,
				progress_ms = EXCLUDED.progress_ms,
				total_duration_ms = EXCLUDED.total_duration_ms,
				watched = EXCLUDED.watched,
				device = EXCLUDED.device,
				player = EXCLUDED.player,
				product = EXCLUDED.product,
				platform = EXCLUDED.platform,
				is_transcode = EXCLUDED.is_transcode,
				video_decision = EXCLUDED.video_decision,
				audio_decision = EXCLUDED.audio_decision,
				source_video_codec = EXCLUDED.source_video_codec,
				source_audio_codec = EXCLUDED.source_audio_codec,
				source_fingerprint = EXCLUDED.source_fingerprint,
				source_seen_at = EXCLUDED.source_seen_at,
				updated_at = CASE
					WHEN tracearr_sessions.source_fingerprint IS DISTINCT FROM EXCLUDED.source_fingerprint
					THEN now() ELSE tracearr_sessions.updated_at END`,
			session.ServerID, session.SessionID, session.UserID, session.Username,
			session.State, session.MediaType, session.MediaTitle, session.ShowTitle,
			session.SeasonNumber, session.EpisodeNumber, session.ProductionYear,
			session.StartedAt, session.StoppedAt, session.DurationMs, session.ProgressMs,
			session.TotalDurationMs, session.Watched, session.Device, session.Player,
			session.Product, session.Platform, session.IsTranscode, session.VideoDecision,
			session.AudioDecision, session.SourceVideoCodec, session.SourceAudioCodec,
			session.EmbyItemID, session.EmbySeriesID, session.SourceFingerprint, seenAt)
	}
	results := s.pool.SendBatch(ctx, batch)
	defer results.Close()
	for range sessions {
		if _, err := results.Exec(); err != nil {
			return fmt.Errorf("store: upsert tracearr sessions: %w", err)
		}
	}
	return nil
}

func (s *Store) DeleteTracearrSessionsNotSeenSince(
	ctx context.Context,
	serverID string,
	cutoff time.Time,
) (int64, error) {
	if serverID == "" {
		tag, err := s.pool.Exec(ctx, `DELETE FROM tracearr_sessions WHERE source_seen_at < $1`, cutoff)
		if err != nil {
			return 0, fmt.Errorf("store: reconcile tracearr sessions: %w", err)
		}
		return tag.RowsAffected(), nil
	}
	tag, err := s.pool.Exec(ctx,
		`DELETE FROM tracearr_sessions WHERE server_id = $1 AND source_seen_at < $2`,
		serverID, cutoff)
	if err != nil {
		return 0, fmt.Errorf("store: reconcile tracearr sessions: %w", err)
	}
	return tag.RowsAffected(), nil
}

func (s *Store) TracearrSessionsForUser(
	ctx context.Context,
	tracearrUserID, username string,
) ([]TracearrSession, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT server_id, tracearr_session_id, tracearr_user_id, username, state,
			media_type, media_title, show_title, season_number, episode_number,
			production_year, started_at, stopped_at, duration_ms, progress_ms,
			total_duration_ms, watched, device, player, product, platform,
			is_transcode, video_decision, audio_decision, source_video_codec,
			source_audio_codec, emby_item_id, emby_series_id, source_fingerprint
		FROM tracearr_sessions
		WHERE ($1 <> '' AND tracearr_user_id = $1)
		   OR ($1 = '' AND lower(username) = lower($2))
		ORDER BY started_at DESC NULLS LAST, tracearr_session_id DESC`,
		tracearrUserID, username)
	if err != nil {
		return nil, fmt.Errorf("store: tracearr user sessions: %w", err)
	}
	defer rows.Close()
	out := []TracearrSession{}
	for rows.Next() {
		var session TracearrSession
		if err := rows.Scan(
			&session.ServerID, &session.SessionID, &session.UserID, &session.Username,
			&session.State, &session.MediaType, &session.MediaTitle, &session.ShowTitle,
			&session.SeasonNumber, &session.EpisodeNumber, &session.ProductionYear,
			&session.StartedAt, &session.StoppedAt, &session.DurationMs, &session.ProgressMs,
			&session.TotalDurationMs, &session.Watched, &session.Device, &session.Player,
			&session.Product, &session.Platform, &session.IsTranscode,
			&session.VideoDecision, &session.AudioDecision, &session.SourceVideoCodec,
			&session.SourceAudioCodec, &session.EmbyItemID, &session.EmbySeriesID,
			&session.SourceFingerprint,
		); err != nil {
			return nil, err
		}
		out = append(out, session)
	}
	return out, rows.Err()
}

func (s *Store) UpdateTracearrSessionMapping(
	ctx context.Context,
	key TracearrSessionKey,
	itemID, seriesID string,
) error {
	_, err := s.pool.Exec(ctx, `
		UPDATE tracearr_sessions
		SET emby_item_id = $3, emby_series_id = $4
		WHERE server_id = $1 AND tracearr_session_id = $2
		  AND (emby_item_id, emby_series_id) IS DISTINCT FROM ($3, $4)`,
		key.ServerID, key.SessionID, itemID, seriesID)
	if err != nil {
		return fmt.Errorf("store: map tracearr session: %w", err)
	}
	return nil
}

func (s *Store) ForYouTracearrIdentity(ctx context.Context, userID string) (string, string, error) {
	var tracearrUserID, username string
	err := s.pool.QueryRow(ctx, `
		SELECT tracearr_user_id, tracearr_username
		FROM recommendation_user_profiles WHERE emby_user_id = $1`, userID).
		Scan(&tracearrUserID, &username)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", "", nil
	}
	if err != nil {
		return "", "", fmt.Errorf("store: For You Tracearr identity: %w", err)
	}
	return tracearrUserID, username, nil
}

func (s *Store) TracearrImportState(ctx context.Context) (TracearrImportState, error) {
	var raw []byte
	err := s.pool.QueryRow(ctx,
		`SELECT value FROM app_settings WHERE key = $1`, tracearrImportStateKey).Scan(&raw)
	if errors.Is(err, pgx.ErrNoRows) {
		return TracearrImportState{}, nil
	}
	if err != nil {
		return TracearrImportState{}, fmt.Errorf("store: read tracearr import state: %w", err)
	}
	var state TracearrImportState
	if err := json.Unmarshal(raw, &state); err != nil {
		return state, fmt.Errorf("store: decode tracearr import state: %w", err)
	}
	return state, nil
}

func (s *Store) SetTracearrImportState(ctx context.Context, state TracearrImportState) error {
	raw, err := json.Marshal(state)
	if err != nil {
		return err
	}
	_, err = s.pool.Exec(ctx, `
		INSERT INTO app_settings (key, value, updated_at)
		VALUES ($1, $2::jsonb, now())
		ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`,
		tracearrImportStateKey, string(raw))
	if err != nil {
		return fmt.Errorf("store: write tracearr import state: %w", err)
	}
	return nil
}

// MarkForYouRebuild records that the household's rows have just been rebuilt.
//
// Read-modify-write rather than a whole-document put, because the importer owns the other
// two stamps in this document and an import running beside a rebuild must not lose its own.
func (s *Store) MarkForYouRebuild(ctx context.Context, at time.Time) error {
	state, err := s.TracearrImportState(ctx)
	if err != nil {
		return err
	}
	stamp := at.UTC()
	state.LastRebuildAt = &stamp
	return s.SetTracearrImportState(ctx, state)
}

func (s *Store) ActiveRecommendationUsers(ctx context.Context) ([]Session, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT DISTINCT ON (emby_user_id)
			token_hash, emby_user_id, emby_token, username, server_id, device_id,
			device_name, client_version, client_protocol, client_capabilities, last_seen_at
		FROM sessions
		ORDER BY emby_user_id, last_seen_at DESC`)
	if err != nil {
		return nil, fmt.Errorf("store: active recommendation users: %w", err)
	}
	defer rows.Close()
	out := []Session{}
	for rows.Next() {
		var session Session
		if err := rows.Scan(
			&session.TokenHash, &session.EmbyUserID, &session.EmbyToken, &session.Username,
			&session.ServerID, &session.DeviceID, &session.DeviceName, &session.ClientVersion,
			&session.ClientProtocol, &session.ClientCapabilities, &session.LastSeenAt,
		); err != nil {
			return nil, err
		}
		out = append(out, session)
	}
	return out, rows.Err()
}

func (s *Store) MarkForYouDirty(ctx context.Context, userID, username string) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO recommendation_user_profiles (emby_user_id, tracearr_username, dirty_since)
		VALUES ($1, $2, now())
		ON CONFLICT (emby_user_id) DO UPDATE SET
			tracearr_username = CASE WHEN $2 <> '' THEN $2 ELSE recommendation_user_profiles.tracearr_username END,
			dirty_since = coalesce(recommendation_user_profiles.dirty_since, now())`,
		userID, username)
	if err != nil {
		return fmt.Errorf("store: mark For You dirty: %w", err)
	}
	return nil
}

func (s *Store) MarkAllForYouProfilesDirty(ctx context.Context) error {
	_, err := s.pool.Exec(ctx, `
		UPDATE recommendation_user_profiles
		SET dirty_since = coalesce(dirty_since, now())`)
	if err != nil {
		return fmt.Errorf("store: mark all For You profiles dirty: %w", err)
	}
	return nil
}

func (s *Store) MarkForYouProfilesDirtyByTracearrIdentity(
	ctx context.Context,
	identities []RecommendationIdentity,
) (int64, error) {
	if len(identities) == 0 {
		return 0, nil
	}
	userIDs := make([]string, 0, len(identities))
	usernames := make([]string, 0, len(identities))
	for _, identity := range identities {
		if identity.TracearrUserID != "" {
			userIDs = append(userIDs, identity.TracearrUserID)
		}
		if identity.Username != "" {
			usernames = append(usernames, strings.ToLower(identity.Username))
		}
	}
	tag, err := s.pool.Exec(ctx, `
		UPDATE recommendation_user_profiles
		SET dirty_since = coalesce(dirty_since, now())
		WHERE tracearr_user_id = ANY($1)
		   OR lower(tracearr_username) = ANY($2)`,
		userIDs, usernames)
	if err != nil {
		return 0, fmt.Errorf("store: mark affected For You profiles dirty: %w", err)
	}
	return tag.RowsAffected(), nil
}

func (s *Store) MatchRecommendationUser(
	ctx context.Context,
	embyUserID, embyUsername, tracearrUserID, tracearrUsername string,
) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO recommendation_user_profiles (
			emby_user_id, tracearr_user_id, tracearr_username, dirty_since
		) VALUES ($1, $2, CASE WHEN $3 <> '' THEN $3 ELSE $4 END, now())
		ON CONFLICT (emby_user_id) DO UPDATE SET
			tracearr_user_id = EXCLUDED.tracearr_user_id,
			tracearr_username = CASE
				WHEN EXCLUDED.tracearr_username <> '' THEN EXCLUDED.tracearr_username
				ELSE $4
			END,
			dirty_since = CASE
				WHEN recommendation_user_profiles.tracearr_user_id IS DISTINCT FROM EXCLUDED.tracearr_user_id
				  OR recommendation_user_profiles.tracearr_username IS DISTINCT FROM
					CASE WHEN EXCLUDED.tracearr_username <> '' THEN EXCLUDED.tracearr_username ELSE $4 END
				THEN coalesce(recommendation_user_profiles.dirty_since, now())
				ELSE recommendation_user_profiles.dirty_since
			END`,
		embyUserID, tracearrUserID, tracearrUsername, embyUsername)
	if err != nil {
		return fmt.Errorf("store: match recommendation user: %w", err)
	}
	return nil
}

func (s *Store) ForYouProfileTimes(
	ctx context.Context,
	userID string,
) (builtAt, poolBuiltAt, dirtySince *time.Time, algorithmVersion string, err error) {
	err = s.pool.QueryRow(ctx, `
		SELECT built_at, pool_built_at, dirty_since, algorithm_version
		FROM recommendation_user_profiles WHERE emby_user_id = $1`, userID).
		Scan(&builtAt, &poolBuiltAt, &dirtySince, &algorithmVersion)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil, nil, "", nil
	}
	if err != nil {
		return nil, nil, nil, "", fmt.Errorf("store: For You freshness: %w", err)
	}
	return builtAt, poolBuiltAt, dirtySince, algorithmVersion, nil
}

func (s *Store) SetForYouError(ctx context.Context, userID string, buildErr error) error {
	message := ""
	if buildErr != nil {
		message = buildErr.Error()
	}
	_, err := s.pool.Exec(ctx, `
		INSERT INTO recommendation_user_profiles (emby_user_id, last_error, dirty_since)
		VALUES ($1, $2, now())
		ON CONFLICT (emby_user_id) DO UPDATE SET last_error = $2,
			dirty_since = coalesce(recommendation_user_profiles.dirty_since, now())`,
		userID, message)
	return err
}

func (s *Store) ReplaceForYouPool(
	ctx context.Context,
	profile RecommendationProfile,
	candidates []ForYouCandidate,
) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return fmt.Errorf("store: begin For You rebuild: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	_, err = tx.Exec(ctx, `
		INSERT INTO recommendation_user_profiles (
			emby_user_id, tracearr_user_id, tracearr_username, source_session_count,
			mean_completion_ratio, typical_session_minutes, genre_affinity,
			title_affinity, studio_affinity, context_affinity, codec_outcomes,
			weighted_profile, algorithm_version, signals_through, built_at, pool_built_at, dirty_since, last_error
		) VALUES (
			$1,$2,$3,$4,$5,$6,$7::jsonb,$8::jsonb,$9::jsonb,$10::jsonb,$11::jsonb,
			$12::jsonb,$13,$14,$15,$15,NULL,''
		)
		ON CONFLICT (emby_user_id) DO UPDATE SET
			tracearr_user_id = EXCLUDED.tracearr_user_id,
			tracearr_username = EXCLUDED.tracearr_username,
			source_session_count = EXCLUDED.source_session_count,
			mean_completion_ratio = EXCLUDED.mean_completion_ratio,
			typical_session_minutes = EXCLUDED.typical_session_minutes,
			genre_affinity = EXCLUDED.genre_affinity,
			title_affinity = EXCLUDED.title_affinity,
			studio_affinity = EXCLUDED.studio_affinity,
			context_affinity = EXCLUDED.context_affinity,
			codec_outcomes = EXCLUDED.codec_outcomes,
			weighted_profile = EXCLUDED.weighted_profile,
			algorithm_version = EXCLUDED.algorithm_version,
			signals_through = EXCLUDED.signals_through,
			built_at = EXCLUDED.built_at,
			pool_built_at = EXCLUDED.pool_built_at,
			dirty_since = NULL,
			last_error = ''`,
		profile.EmbyUserID, profile.TracearrUserID, profile.TracearrUsername,
		profile.SourceSessionCount, profile.MeanCompletionRatio,
		profile.TypicalSessionMinutes, string(profile.GenreAffinity),
		string(profile.TitleAffinity), string(profile.StudioAffinity),
		string(profile.ContextAffinity), string(profile.CodecOutcomes),
		string(profile.WeightedProfile), profile.AlgorithmVersion,
		profile.SignalsThrough, profile.BuiltAt)
	if err != nil {
		return fmt.Errorf("store: write For You profile: %w", err)
	}
	if _, err := tx.Exec(ctx,
		`DELETE FROM for_you_candidates WHERE emby_user_id = $1`, profile.EmbyUserID); err != nil {
		return fmt.Errorf("store: clear For You pool: %w", err)
	}

	batch := &pgx.Batch{}
	for _, candidate := range candidates {
		batch.Queue(`
			INSERT INTO for_you_candidates (
				emby_user_id, item_id, base_rank, base_score, runtime_minutes,
				affinity_score, compatibility_score, compatibility_label, reason_kind,
				reason_genre, reason_source_session_id, reason_source_item_id,
				reason_source_title, recommendation_reason, built_at
			) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)`,
			profile.EmbyUserID, candidate.ItemID, candidate.BaseRank, candidate.BaseScore,
			candidate.RuntimeMinutes, candidate.AffinityScore, candidate.CompatibilityScore,
			candidate.CompatibilityLabel, candidate.ReasonKind, candidate.ReasonGenre,
			candidate.ReasonSourceSessionID, candidate.ReasonSourceItemID,
			candidate.ReasonSourceTitle, candidate.RecommendationReason, profile.BuiltAt)
	}
	results := tx.SendBatch(ctx, batch)
	for range candidates {
		if _, err := results.Exec(); err != nil {
			_ = results.Close()
			return fmt.Errorf("store: insert For You candidates: %w", err)
		}
	}
	if err := results.Close(); err != nil {
		return fmt.Errorf("store: close For You candidate batch: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("store: commit For You rebuild: %w", err)
	}
	return nil
}

func (s *Store) PreparedForYou(
	ctx context.Context,
	userID string,
	minutes, limit int,
) ([]PreparedForYouItem, *time.Time, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT fc.item_id, fc.base_rank, fc.base_score, fc.affinity_score,
			li.payload, fc.runtime_minutes,
			fc.compatibility_score, fc.compatibility_label,
			fc.recommendation_reason, fc.reason_kind, fc.reason_genre,
			fc.reason_source_item_id, fc.reason_source_title, p.context_affinity,
			p.pool_built_at
		FROM for_you_candidates fc
		JOIN library_items li ON li.id = fc.item_id
		JOIN recommendation_user_profiles p ON p.emby_user_id = fc.emby_user_id
		WHERE fc.emby_user_id = $1
		  AND (
			$2 = 0 OR fc.reason_kind = 'pick-up' OR
			(fc.runtime_minutes > 0 AND fc.runtime_minutes <= $2)
		  )
		ORDER BY fc.base_rank
		LIMIT $3`,
		userID, minutes, limit)
	if err != nil {
		return nil, nil, fmt.Errorf("store: prepared For You: %w", err)
	}
	defer rows.Close()
	out := []PreparedForYouItem{}
	var builtAt *time.Time
	for rows.Next() {
		var payload []byte
		var item PreparedForYouItem
		var rowBuiltAt *time.Time
		if err := rows.Scan(
			&item.ItemID, &item.BaseRank, &item.BaseScore, &item.AffinityScore,
			&payload, &item.RuntimeMinutes,
			&item.CompatibilityScore, &item.CompatibilityLabel,
			&item.RecommendationReason, &item.ReasonKind, &item.ReasonGenre,
			&item.ReasonSourceItemID, &item.ReasonSourceTitle, &item.ContextAffinity,
			&rowBuiltAt,
		); err != nil {
			return nil, nil, err
		}
		item.Payload = json.RawMessage(payload)
		out = append(out, item)
		if builtAt == nil {
			builtAt = rowBuiltAt
		}
	}
	return out, builtAt, rows.Err()
}

func (s *Store) ForYouStats(ctx context.Context) (ForYouStats, error) {
	var stats ForYouStats
	if err := s.pool.QueryRow(ctx, `
		SELECT
			(SELECT count(*) FROM tracearr_sessions),
			(SELECT count(*) FROM recommendation_user_profiles),
			(SELECT count(*) FROM for_you_candidates)`).
		Scan(&stats.TracearrSessions, &stats.Profiles, &stats.Candidates); err != nil {
		return stats, fmt.Errorf("store: For You stats: %w", err)
	}
	state, err := s.TracearrImportState(ctx)
	if err != nil {
		return stats, err
	}
	stats.LastFullImport = state.LastFullAt
	return stats, nil
}