feat(sections): add trending refresher with persisted snapshots
This commit is contained in:
Quick
@@ -0,0 +1,252 @@
|
||||
package sections
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/Silo-Server/silo-server/internal/catalog"
|
||||
"github.com/Silo-Server/silo-server/internal/sections/recipes"
|
||||
)
|
||||
|
||||
// trendingFetchCap is the over-fetch size for each refresh. Library-only
|
||||
// matching drops globally-trending titles the server does not own, so we fetch
|
||||
// well beyond any section's display limit and store the matched, ordered list.
|
||||
const trendingFetchCap = 200
|
||||
|
||||
// trendingSectionConfigLister enumerates enabled trending_discover section
|
||||
// configs. Satisfied by *Repository.
|
||||
type trendingSectionConfigLister interface {
|
||||
ListTrendingDiscoverConfigs(ctx context.Context) ([]json.RawMessage, error)
|
||||
}
|
||||
|
||||
// trendingSnapshotStore is the write side of the snapshot table. Satisfied by
|
||||
// *TrendingSnapshotRepository.
|
||||
type trendingSnapshotStore interface {
|
||||
SaveSuccess(ctx context.Context, source, window string, contentIDs []string, entryCount int, status string, at time.Time) error
|
||||
RecordAttempt(ctx context.Context, source, window, status, message string, at time.Time) error
|
||||
}
|
||||
|
||||
// trendingExternalIDResolver resolves external IDs to library content IDs.
|
||||
// Satisfied by *catalog.ItemRepository.
|
||||
type trendingExternalIDResolver interface {
|
||||
GetByExternalIDs(ctx context.Context, batch catalog.ExternalIDBatch, itemType string) (*catalog.ExternalIDLookup, error)
|
||||
}
|
||||
|
||||
// TrendingRefresher fetches external global trending (TMDB/Trakt), resolves it
|
||||
// to library content IDs, and persists one snapshot per canonical
|
||||
// (source, window). It is driven by a TaskManager task on an interval.
|
||||
type TrendingRefresher struct {
|
||||
Sections trendingSectionConfigLister
|
||||
Snapshots trendingSnapshotStore
|
||||
Resolver trendingExternalIDResolver
|
||||
TMDBTrending catalog.TMDBCollectionFetcher
|
||||
TraktTrending catalog.TraktCollectionFetcher
|
||||
|
||||
// Clock defaults to recipes.RealClock{}. Tests inject recipes.FixedClock.
|
||||
Clock recipes.Clock
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewTrendingRefresher creates a refresher with real-clock and default logger.
|
||||
func NewTrendingRefresher(
|
||||
sectionsRepo trendingSectionConfigLister,
|
||||
snapshots trendingSnapshotStore,
|
||||
resolver trendingExternalIDResolver,
|
||||
tmdb catalog.TMDBCollectionFetcher,
|
||||
trakt catalog.TraktCollectionFetcher,
|
||||
) *TrendingRefresher {
|
||||
return &TrendingRefresher{
|
||||
Sections: sectionsRepo,
|
||||
Snapshots: snapshots,
|
||||
Resolver: resolver,
|
||||
TMDBTrending: tmdb,
|
||||
TraktTrending: trakt,
|
||||
Clock: recipes.RealClock{},
|
||||
logger: slog.Default(),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *TrendingRefresher) now() time.Time {
|
||||
if r.Clock != nil {
|
||||
return r.Clock.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
func (r *TrendingRefresher) log() *slog.Logger {
|
||||
if r.logger != nil {
|
||||
return r.logger
|
||||
}
|
||||
return slog.Default()
|
||||
}
|
||||
|
||||
// TrendingRefreshResult is the JSON summary attached to the task execution.
|
||||
type TrendingRefreshResult struct {
|
||||
Combos int `json:"combos"`
|
||||
Refreshed int `json:"refreshed"`
|
||||
Empty int `json:"empty"`
|
||||
Failed int `json:"failed"`
|
||||
}
|
||||
|
||||
type trendingCombo struct {
|
||||
source string
|
||||
window string
|
||||
}
|
||||
|
||||
// distinctTrendingCombos parses section configs and returns the deduplicated set
|
||||
// of canonical (source, window) pairs that need a snapshot.
|
||||
func distinctTrendingCombos(configs []json.RawMessage) []trendingCombo {
|
||||
seen := make(map[trendingCombo]struct{}, len(configs))
|
||||
out := make([]trendingCombo, 0, len(configs))
|
||||
for _, raw := range configs {
|
||||
var p recipes.TrendingDiscoverParams
|
||||
if len(raw) > 0 {
|
||||
_ = json.Unmarshal(raw, &p)
|
||||
}
|
||||
source, window := canonicalTrendingKey(p.Source, p.Window)
|
||||
c := trendingCombo{source: source, window: window}
|
||||
if _, ok := seen[c]; ok {
|
||||
continue
|
||||
}
|
||||
seen[c] = struct{}{}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// RunOnce refreshes every (source, window) used by an enabled trending_discover
|
||||
// section. Per-combo failures are recorded and never abort the others. The JSON
|
||||
// summary is suitable for task result data.
|
||||
func (r *TrendingRefresher) RunOnce(ctx context.Context) (json.RawMessage, error) {
|
||||
configs, err := r.Sections.ListTrendingDiscoverConfigs(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing trending_discover sections: %w", err)
|
||||
}
|
||||
|
||||
combos := distinctTrendingCombos(configs)
|
||||
result := TrendingRefreshResult{Combos: len(combos)}
|
||||
for _, c := range combos {
|
||||
switch r.refreshCombo(ctx, c.source, c.window) {
|
||||
case "ok":
|
||||
result.Refreshed++
|
||||
case "empty":
|
||||
result.Empty++
|
||||
default:
|
||||
result.Failed++
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(result)
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// refreshCombo refreshes a single canonical (source, window) and returns its
|
||||
// outcome: "ok", "empty", or "error". A fetch failure or an unconfigured/empty
|
||||
// provider preserves the last-good content list (RecordAttempt). When the
|
||||
// provider returns entries, the list is replaced even if nothing matched the
|
||||
// catalog ("empty" status with an empty list) — that genuinely reflects current
|
||||
// trending having no library matches.
|
||||
func (r *TrendingRefresher) refreshCombo(ctx context.Context, source, window string) string {
|
||||
now := r.now()
|
||||
|
||||
entries, err := r.fetchEntries(ctx, source, window, trendingFetchCap)
|
||||
if err != nil {
|
||||
r.log().Error("trending refresh: fetch failed", "source", source, "window", window, "error", err)
|
||||
_ = r.Snapshots.RecordAttempt(ctx, source, window, "error", err.Error(), now)
|
||||
return "error"
|
||||
}
|
||||
if len(entries) == 0 {
|
||||
// Provider unconfigured or returned nothing: keep last-good, mark empty.
|
||||
_ = r.Snapshots.RecordAttempt(ctx, source, window, "empty", "", now)
|
||||
return "empty"
|
||||
}
|
||||
|
||||
contentIDs, err := r.resolveIDs(ctx, entries)
|
||||
if err != nil {
|
||||
r.log().Error("trending refresh: resolve failed", "source", source, "window", window, "error", err)
|
||||
_ = r.Snapshots.RecordAttempt(ctx, source, window, "error", err.Error(), now)
|
||||
return "error"
|
||||
}
|
||||
|
||||
status := "ok"
|
||||
if len(contentIDs) == 0 {
|
||||
status = "empty"
|
||||
}
|
||||
if err := r.Snapshots.SaveSuccess(ctx, source, window, contentIDs, len(entries), status, now); err != nil {
|
||||
r.log().Error("trending refresh: save failed", "source", source, "window", window, "error", err)
|
||||
return "error"
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
// fetchEntries pulls the raw trending list from the configured provider. A
|
||||
// nil/unconfigured provider yields an empty list (no error).
|
||||
func (r *TrendingRefresher) fetchEntries(ctx context.Context, source, window string, fetchLimit int) ([]trendingDiscoverEntry, error) {
|
||||
if source == "trakt" {
|
||||
if r.TraktTrending == nil {
|
||||
return nil, nil
|
||||
}
|
||||
movies, movieErr := r.TraktTrending.GetCollectionPreset(ctx, "trending", "movie", fetchLimit, "")
|
||||
shows, showErr := r.TraktTrending.GetCollectionPreset(ctx, "trending", "tv", fetchLimit, "")
|
||||
if movieErr != nil && showErr != nil {
|
||||
return nil, fmt.Errorf("trakt trending: %v / %v", movieErr, showErr)
|
||||
}
|
||||
out := make([]trendingDiscoverEntry, 0, len(movies)+len(shows))
|
||||
for _, e := range movies {
|
||||
out = append(out, newTrendingEntry(e.TMDBID, e.TVDBID, e.IMDbID, e.MediaType))
|
||||
}
|
||||
for _, e := range shows {
|
||||
out = append(out, newTrendingEntry(e.TMDBID, e.TVDBID, e.IMDbID, e.MediaType))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
if r.TMDBTrending == nil {
|
||||
return nil, nil
|
||||
}
|
||||
entries, err := r.TMDBTrending.GetCollectionPreset(ctx, "trending", "all", window, fetchLimit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]trendingDiscoverEntry, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
out = append(out, newTrendingEntry(e.ID, e.TVDBID, e.IMDbID, e.MediaType))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// resolveIDs matches trending entries to library content IDs via two batched
|
||||
// external-ID lookups (movies, series), preserving trending order.
|
||||
func (r *TrendingRefresher) resolveIDs(ctx context.Context, entries []trendingDiscoverEntry) ([]string, error) {
|
||||
if r.Resolver == nil {
|
||||
return nil, fmt.Errorf("trending_discover: external ID resolver not configured")
|
||||
}
|
||||
var movieBatch, seriesBatch catalog.ExternalIDBatch
|
||||
for _, e := range entries {
|
||||
batch := &movieBatch
|
||||
if e.mediaType == "tv" {
|
||||
batch = &seriesBatch
|
||||
}
|
||||
if e.tmdbID != "" {
|
||||
batch.TMDBIDs = append(batch.TMDBIDs, e.tmdbID)
|
||||
}
|
||||
if e.imdbID != "" {
|
||||
batch.IMDbIDs = append(batch.IMDbIDs, e.imdbID)
|
||||
}
|
||||
if e.tvdbID != "" {
|
||||
batch.TVDBIDs = append(batch.TVDBIDs, e.tvdbID)
|
||||
}
|
||||
}
|
||||
movieLookup, err := r.Resolver.GetByExternalIDs(ctx, movieBatch, "movie")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
seriesLookup, err := r.Resolver.GetByExternalIDs(ctx, seriesBatch, "series")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return orderedTrendingContentIDs(entries, movieLookup, seriesLookup), nil
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
package sections
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Silo-Server/silo-server/internal/catalog"
|
||||
"github.com/Silo-Server/silo-server/internal/sections/recipes"
|
||||
)
|
||||
|
||||
type fakeSectionLister struct {
|
||||
configs []json.RawMessage
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeSectionLister) ListTrendingDiscoverConfigs(context.Context) ([]json.RawMessage, error) {
|
||||
return f.configs, f.err
|
||||
}
|
||||
|
||||
type savedSnap struct {
|
||||
contentIDs []string
|
||||
entryCount int
|
||||
status string
|
||||
}
|
||||
|
||||
type attemptRec struct {
|
||||
status string
|
||||
message string
|
||||
}
|
||||
|
||||
type fakeSnapshotStore struct {
|
||||
saved map[string]savedSnap
|
||||
attempts map[string]attemptRec
|
||||
}
|
||||
|
||||
func newFakeSnapshotStore() *fakeSnapshotStore {
|
||||
return &fakeSnapshotStore{saved: map[string]savedSnap{}, attempts: map[string]attemptRec{}}
|
||||
}
|
||||
|
||||
func (f *fakeSnapshotStore) SaveSuccess(_ context.Context, source, window string, contentIDs []string, entryCount int, status string, _ time.Time) error {
|
||||
f.saved[source+"|"+window] = savedSnap{contentIDs: contentIDs, entryCount: entryCount, status: status}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeSnapshotStore) RecordAttempt(_ context.Context, source, window, status, message string, _ time.Time) error {
|
||||
f.attempts[source+"|"+window] = attemptRec{status: status, message: message}
|
||||
return nil
|
||||
}
|
||||
|
||||
type fakeTMDB struct {
|
||||
entries []catalog.TMDBCollectionEntry
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeTMDB) GetCollectionPreset(context.Context, string, string, string, int) ([]catalog.TMDBCollectionEntry, error) {
|
||||
return f.entries, f.err
|
||||
}
|
||||
|
||||
type fakeResolver struct {
|
||||
byType map[string]*catalog.ExternalIDLookup
|
||||
}
|
||||
|
||||
func (f fakeResolver) GetByExternalIDs(_ context.Context, _ catalog.ExternalIDBatch, itemType string) (*catalog.ExternalIDLookup, error) {
|
||||
if lk, ok := f.byType[itemType]; ok {
|
||||
return lk, nil
|
||||
}
|
||||
return &catalog.ExternalIDLookup{ByTMDB: map[string]string{}, ByIMDb: map[string]string{}, ByTVDB: map[string]string{}}, nil
|
||||
}
|
||||
|
||||
func tmdbConfig(t *testing.T, source, window string) json.RawMessage {
|
||||
t.Helper()
|
||||
raw, err := json.Marshal(recipes.TrendingDiscoverParams{Source: source, Window: window})
|
||||
if err != nil {
|
||||
t.Fatalf("marshal config: %v", err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func TestRefresherSavesOrderedContentIDs(t *testing.T) {
|
||||
store := newFakeSnapshotStore()
|
||||
r := &TrendingRefresher{
|
||||
Sections: fakeSectionLister{configs: []json.RawMessage{tmdbConfig(t, "tmdb", "week")}},
|
||||
Snapshots: store,
|
||||
Resolver: fakeResolver{byType: map[string]*catalog.ExternalIDLookup{
|
||||
"movie": {ByTMDB: map[string]string{"10": "c-movie"}, ByIMDb: map[string]string{}, ByTVDB: map[string]string{}},
|
||||
"series": {ByTMDB: map[string]string{"20": "c-series"}, ByIMDb: map[string]string{}, ByTVDB: map[string]string{}},
|
||||
}},
|
||||
TMDBTrending: fakeTMDB{entries: []catalog.TMDBCollectionEntry{
|
||||
{ID: 10, MediaType: "movie"},
|
||||
{ID: 20, MediaType: "tv"},
|
||||
}},
|
||||
Clock: recipes.FixedClock(time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)),
|
||||
}
|
||||
|
||||
data, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
|
||||
var result TrendingRefreshResult
|
||||
if err := json.Unmarshal(data, &result); err != nil {
|
||||
t.Fatalf("unmarshal result: %v", err)
|
||||
}
|
||||
if result.Combos != 1 || result.Refreshed != 1 || result.Failed != 0 || result.Empty != 0 {
|
||||
t.Fatalf("result = %+v; want {Combos:1 Refreshed:1 Empty:0 Failed:0}", result)
|
||||
}
|
||||
|
||||
got := store.saved["tmdb|week"]
|
||||
want := []string{"c-movie", "c-series"}
|
||||
if len(got.contentIDs) != len(want) || got.contentIDs[0] != want[0] || got.contentIDs[1] != want[1] {
|
||||
t.Fatalf("saved content IDs = %v; want %v", got.contentIDs, want)
|
||||
}
|
||||
if got.status != "ok" || got.entryCount != 2 {
|
||||
t.Fatalf("saved snap = %+v; want status ok, entryCount 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefresherFailurePreservesLastGood(t *testing.T) {
|
||||
store := newFakeSnapshotStore()
|
||||
r := &TrendingRefresher{
|
||||
Sections: fakeSectionLister{configs: []json.RawMessage{tmdbConfig(t, "tmdb", "week")}},
|
||||
Snapshots: store,
|
||||
Resolver: fakeResolver{},
|
||||
TMDBTrending: fakeTMDB{err: errors.New("tmdb 503")},
|
||||
Clock: recipes.FixedClock(time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)),
|
||||
}
|
||||
|
||||
data, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
|
||||
if _, ok := store.saved["tmdb|week"]; ok {
|
||||
t.Fatal("SaveSuccess must not be called on fetch failure (would clear last-good)")
|
||||
}
|
||||
att, ok := store.attempts["tmdb|week"]
|
||||
if !ok || att.status != "error" {
|
||||
t.Fatalf("attempt = %+v, ok=%v; want status error", att, ok)
|
||||
}
|
||||
|
||||
var result TrendingRefreshResult
|
||||
_ = json.Unmarshal(data, &result)
|
||||
if result.Failed != 1 {
|
||||
t.Fatalf("result.Failed = %d; want 1", result.Failed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefresherEmptyProviderPreservesLastGood(t *testing.T) {
|
||||
store := newFakeSnapshotStore()
|
||||
r := &TrendingRefresher{
|
||||
Sections: fakeSectionLister{configs: []json.RawMessage{tmdbConfig(t, "tmdb", "week")}},
|
||||
Snapshots: store,
|
||||
Resolver: fakeResolver{},
|
||||
// TMDBTrending nil => provider unconfigured => empty entries, no error.
|
||||
Clock: recipes.FixedClock(time.Date(2026, 5, 29, 12, 0, 0, 0, time.UTC)),
|
||||
}
|
||||
|
||||
data, err := r.RunOnce(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("RunOnce: %v", err)
|
||||
}
|
||||
if _, ok := store.saved["tmdb|week"]; ok {
|
||||
t.Fatal("SaveSuccess must not be called when provider returns no entries")
|
||||
}
|
||||
att := store.attempts["tmdb|week"]
|
||||
if att.status != "empty" {
|
||||
t.Fatalf("attempt status = %q; want empty", att.status)
|
||||
}
|
||||
var result TrendingRefreshResult
|
||||
_ = json.Unmarshal(data, &result)
|
||||
if result.Empty != 1 {
|
||||
t.Fatalf("result.Empty = %d; want 1", result.Empty)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDistinctTrendingCombosCollapsesTrakt(t *testing.T) {
|
||||
configs := []json.RawMessage{
|
||||
tmdbConfig(t, "trakt", "day"),
|
||||
tmdbConfig(t, "trakt", "week"),
|
||||
tmdbConfig(t, "tmdb", "day"),
|
||||
tmdbConfig(t, "tmdb", "day"),
|
||||
}
|
||||
got := distinctTrendingCombos(configs)
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("distinctTrendingCombos len = %d (%+v); want 2", len(got), got)
|
||||
}
|
||||
seen := map[trendingCombo]bool{}
|
||||
for _, c := range got {
|
||||
seen[c] = true
|
||||
}
|
||||
if !seen[trendingCombo{"trakt", "week"}] || !seen[trendingCombo{"tmdb", "day"}] {
|
||||
t.Fatalf("combos = %+v; want {trakt week} and {tmdb day}", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user