Files
memby/server/internal/foryou/service.go
T

750 lines
22 KiB
Go
Raw Normal View History

2026-07-29 15:26:40 +12:00
// 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
)
type ImportResult struct {
Kind string `json:"kind"`
Pages int `json:"pages"`
Seen int `json:"seen"`
Changed int `json:"changed"`
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
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,
building: map[string]bool{},
}
}
func (s *Service) Running() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.importRunning || len(s.building) > 0
}
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
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.TracearrFingerprints(ctx, keys)
if err != nil {
return result, err
}
pageChanged := false
for _, session := range imported {
key := store.TracearrSessionKey{
ServerID: session.ServerID, SessionID: session.SessionID,
}
if !bytes.Equal(current[key], session.SourceFingerprint) {
pageChanged = true
result.Changed++
}
}
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
}
if result.Changed > 0 || result.Removed > 0 {
if err := s.store.MarkAllForYouProfilesDirty(ctx); err != nil {
return result, err
}
users, err := s.store.ActiveRecommendationUsers(ctx)
if err != nil {
return result, err
}
for _, user := range users {
if err := s.store.MarkForYouDirty(ctx, user.EmbyUserID, user.Username); err != nil {
return result, err
}
}
}
s.log.Info("Tracearr import finished",
"kind", result.Kind, "pages", result.Pages, "seen", result.Seen,
"changed", result.Changed, "removed", result.Removed)
return result, nil
}
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, err := s.store.ForYouProfileTimes(ctx, sess.EmbyUserID)
if err != nil {
return err
}
if !force && poolBuiltAt != nil && time.Since(*poolBuiltAt) < s.minRebuildAge {
return nil
}
if !force && 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
}
rows := buildPreparedRows(items, minutes, s.engine.MinRowItems)
stale := builtAt == nil || time.Since(*builtAt) >= s.refreshAge
return rows, len(rows) > 0, stale, nil
}
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)
}
}()
}
func (s *Service) RebuildAll(ctx context.Context, force bool) error {
users, err := s.recommendationUsers(ctx)
if err != nil {
return err
}
for _, user := range users {
if err := s.Rebuild(ctx, user, force); err != nil {
s.log.Warn("For You user 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 household users 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: "memby-for-you-builder",
DeviceName: "Memby For You builder",
}
}
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
}
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 {
if item.ReasonKind == "pick-up" {
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 these 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 {
return "Because you finished " + 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) != ""
},
)
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()
}
func (s *Service) Schedule(
ctx context.Context,
importEvery, fullEvery, refreshEvery time.Duration,
) {
if importEvery <= 0 {
s.log.Info("Tracearr auto-import disabled")
return
}
importTicker := time.NewTicker(importEvery)
defer importTicker.Stop()
var fullC, refreshC <-chan time.Time
var fullTicker, refreshTicker *time.Ticker
if fullEvery > 0 {
fullTicker = time.NewTicker(fullEvery)
fullC = fullTicker.C
defer fullTicker.Stop()
}
if refreshEvery > 0 {
refreshTicker = time.NewTicker(refreshEvery)
refreshC = refreshTicker.C
defer refreshTicker.Stop()
}
for {
select {
case <-ctx.Done():
return
case <-importTicker.C:
if _, err := s.Import(ctx, false); err != nil {
s.log.Warn("scheduled Tracearr import failed", "error", err)
} else {
_ = s.RebuildAll(ctx, false)
}
case <-fullC:
if _, err := s.Import(ctx, true); err != nil {
s.log.Warn("scheduled full Tracearr import failed", "error", err)
} else {
_ = s.RebuildAll(ctx, false)
}
case <-refreshC:
_ = s.RebuildAll(ctx, false)
}
}
}
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
}
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,
CodecOutcomes: codecs, SignalsThrough: result.Profile.SignalsThrough, BuiltAt: builtAt,
}
candidates := make([]store.ForYouCandidate, 0, len(result.Candidates))
for _, value := range result.Candidates {
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
}
func credentials(sess store.Session) emby.Credentials {
return emby.Credentials{
UserID: sess.EmbyUserID, Token: sess.EmbyToken,
DeviceID: sess.DeviceID, DeviceName: sess.DeviceName,
}
}