feat: cache top-picks with stale-while-refresh

internal/anime/recommendations.go

@@ -4,12 +4,133 @@ import (
 	"context"
 	"mal/internal/anime/recommendations"
 	"mal/internal/domain"
+	"mal/internal/observability"
+	"strings"
+	"sync"
+	"time"
 )
 
-func (s *animeService) GetTopPickForYou(ctx context.Context, userID string) (domain.CatalogSectionData, error) {
+type recommendationComputeFunc func(context.Context, string, int) (domain.CatalogSectionData, error)
-	return recommendations.GetTopPicksForYou(ctx, s.jikan, s.repo, userID, recommendations.TopPickLimit)
+
+type topPicksCacheKey struct {
+	userID string
+	limit  int
 }
 
-func (s *animeService) GetTopPicksForYou(ctx context.Context, userID string) (domain.CatalogSectionData, error) {
+type topPicksCacheEntry struct {
-	return recommendations.GetTopPicksForYou(ctx, s.jikan, s.repo, userID, recommendations.TopPicksLimit)
+	data       domain.CatalogSectionData
+	updatedAt  time.Time
+	hasData    bool
+	refreshing bool
+}
+
+type topPicksCache struct {
+	mu      sync.Mutex
+	entries map[topPicksCacheKey]*topPicksCacheEntry
+}
+
+const topPicksRefreshTimeout = 30 * time.Second
+
+func (s *animeService) GetTopPickForYou(_ context.Context, userID string) (domain.CatalogSectionData, error) {
+	return s.getCachedTopPicksForYou(userID, recommendations.TopPickLimit), nil
+}
+
+func (s *animeService) GetTopPicksForYou(_ context.Context, userID string) (domain.CatalogSectionData, error) {
+	return s.getCachedTopPicksForYou(userID, recommendations.TopPicksLimit), nil
+}
+
+func (s *animeService) fetchTopPicksForYou(ctx context.Context, userID string, limit int) (domain.CatalogSectionData, error) {
+	return recommendations.GetTopPicksForYou(ctx, s.jikan, s.repo, userID, limit)
+}
+
+func (s *animeService) getCachedTopPicksForYou(userID string, limit int) domain.CatalogSectionData {
+	userID = strings.TrimSpace(userID)
+	if userID == "" {
+		return domain.CatalogSectionData{Animes: []domain.Anime{}}
+	}
+
+	key := topPicksCacheKey{userID: userID, limit: limit}
+	now := time.Now()
+
+	s.topPicksCache.mu.Lock()
+	entry := s.topPicksCache.entries[key]
+	if entry != nil && entry.hasData {
+		data := cloneCatalogSectionData(entry.data)
+		if now.Sub(entry.updatedAt) >= s.topPicksCacheTTL && !entry.refreshing {
+			entry.refreshing = true
+			go s.refreshTopPicksForYou(key)
+		}
+		s.topPicksCache.mu.Unlock()
+		return data
+	}
+
+	if entry == nil {
+		entry = &topPicksCacheEntry{}
+		s.topPicksCache.entries[key] = entry
+	}
+	if !entry.refreshing {
+		entry.refreshing = true
+		go s.refreshTopPicksForYou(key)
+	}
+	s.topPicksCache.mu.Unlock()
+
+	return domain.CatalogSectionData{Animes: []domain.Anime{}}
+}
+
+func (s *animeService) refreshTopPicksForYou(key topPicksCacheKey) {
+	ctx, cancel := context.WithTimeout(context.Background(), topPicksRefreshTimeout)
+	defer cancel()
+
+	data, err := s.computeTopPicks(ctx, key.userID, key.limit)
+	if err != nil {
+		observability.WarnContext(ctx,
+			"top_picks_refresh_failed",
+			"anime",
+			"",
+			map[string]any{"user_id": key.userID, "limit": key.limit},
+			err,
+		)
+		s.topPicksCache.mu.Lock()
+		if entry := s.topPicksCache.entries[key]; entry != nil {
+			entry.refreshing = false
+		}
+		s.topPicksCache.mu.Unlock()
+		return
+	}
+
+	s.topPicksCache.mu.Lock()
+	s.topPicksCache.entries[key] = &topPicksCacheEntry{
+		data:      cloneCatalogSectionData(data),
+		updatedAt: time.Now(),
+		hasData:   true,
+	}
+	s.topPicksCache.mu.Unlock()
+}
+
+func (s *animeService) InvalidateTopPicksForUser(userID string) {
+	userID = strings.TrimSpace(userID)
+	if userID == "" {
+		return
+	}
+
+	s.topPicksCache.mu.Lock()
+	defer s.topPicksCache.mu.Unlock()
+	for key := range s.topPicksCache.entries {
+		if key.userID == userID {
+			delete(s.topPicksCache.entries, key)
+		}
+	}
+}
+
+func cloneCatalogSectionData(data domain.CatalogSectionData) domain.CatalogSectionData {
+	data.Animes = append([]domain.Anime(nil), data.Animes...)
+	data.ContinueWatching = append(data.ContinueWatching[:0:0], data.ContinueWatching...)
+	if data.WatchlistMap != nil {
+		watchlistMap := make(map[int64]bool, len(data.WatchlistMap))
+		for id, ok := range data.WatchlistMap {
+			watchlistMap[id] = ok
+		}
+		data.WatchlistMap = watchlistMap
+	}
+	return data
 }

internal/anime/recommendations_cache_test.go

@@ -0,0 +1,123 @@
+package anime
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"mal/integrations/jikan"
+	"mal/internal/domain"
+)
+
+func TestGetTopPicksForYouReturnsEmptyOnCacheMissAndRefreshesInBackground(t *testing.T) {
+	refreshed := make(chan struct{}, 1)
+	svc := NewAnimeService(nil, nil)
+	svc.computeTopPicks = func(context.Context, string, int) (domain.CatalogSectionData, error) {
+		refreshed <- struct{}{}
+		return domain.CatalogSectionData{Animes: []domain.Anime{{Anime: jikan.Anime{MalID: 7}}}}, nil
+	}
+
+	got, err := svc.GetTopPicksForYou(context.Background(), "user-1")
+	if err != nil {
+		t.Fatalf("GetTopPicksForYou cache miss: %v", err)
+	}
+	if len(got.Animes) != 0 {
+		t.Fatalf("cache miss animes = %+v, want empty while refresh runs", got.Animes)
+	}
+
+	waitForRefresh(t, refreshed)
+
+	got, err = svc.GetTopPicksForYou(context.Background(), "user-1")
+	if err != nil {
+		t.Fatalf("GetTopPicksForYou cache hit: %v", err)
+	}
+	if len(got.Animes) != 1 || got.Animes[0].MalID != 7 {
+		t.Fatalf("cache hit animes = %+v, want anime 7", got.Animes)
+	}
+}
+
+func TestGetTopPicksForYouReturnsStaleDataWhenRefreshFails(t *testing.T) {
+	svc := NewAnimeService(nil, nil)
+	svc.topPicksCacheTTL = time.Nanosecond
+	refreshed := make(chan struct{}, 2)
+	svc.computeTopPicks = func(context.Context, string, int) (domain.CatalogSectionData, error) {
+		refreshed <- struct{}{}
+		return domain.CatalogSectionData{Animes: []domain.Anime{{Anime: jikan.Anime{MalID: 11}}}}, nil
+	}
+
+	if _, err := svc.GetTopPicksForYou(context.Background(), "user-1"); err != nil {
+		t.Fatalf("prime cache: %v", err)
+	}
+	waitForRefresh(t, refreshed)
+
+	svc.computeTopPicks = func(context.Context, string, int) (domain.CatalogSectionData, error) {
+		refreshed <- struct{}{}
+		return domain.CatalogSectionData{}, errors.New("provider unavailable")
+	}
+	time.Sleep(time.Nanosecond)
+
+	got, err := svc.GetTopPicksForYou(context.Background(), "user-1")
+	if err != nil {
+		t.Fatalf("stale GetTopPicksForYou: %v", err)
+	}
+	if len(got.Animes) != 1 || got.Animes[0].MalID != 11 {
+		t.Fatalf("stale animes = %+v, want anime 11", got.Animes)
+	}
+}
+
+func TestInvalidateTopPicksForUserRefreshesNextRequest(t *testing.T) {
+	refreshed := make(chan struct{}, 2)
+	results := []int{3, 4}
+
+	svc := newPickSvc(refreshed, &results)
+
+	primePickCache(t, svc, refreshed)
+
+	svc.InvalidateTopPicksForUser("user-1")
+
+	got, err := svc.GetTopPicksForYou(context.Background(), "user-1")
+	if err != nil {
+		t.Fatalf("GetTopPicksForYou after invalidation: %v", err)
+	}
+	if len(got.Animes) != 0 {
+		t.Fatalf("invalidated cache animes = %+v, want empty while refresh runs", got.Animes)
+	}
+	waitForRefresh(t, refreshed)
+
+	got, err = svc.GetTopPicksForYou(context.Background(), "user-1")
+	if err != nil {
+		t.Fatalf("GetTopPicksForYou refreshed cache: %v", err)
+	}
+	if len(got.Animes) != 1 || got.Animes[0].MalID != 4 {
+		t.Fatalf("refreshed animes = %+v, want anime 4", got.Animes)
+	}
+}
+
+func newPickSvc(refreshed chan struct{}, results *[]int) *animeService {
+	svc := NewAnimeService(nil, nil)
+	svc.computeTopPicks = func(context.Context, string, int) (domain.CatalogSectionData, error) {
+		id := (*results)[0]
+		*results = (*results)[1:]
+		refreshed <- struct{}{}
+		return domain.CatalogSectionData{Animes: []domain.Anime{{Anime: jikan.Anime{MalID: id}}}}, nil
+	}
+	return svc
+}
+
+func primePickCache(t *testing.T, svc *animeService, refreshed chan struct{}) {
+	t.Helper()
+	if _, err := svc.GetTopPicksForYou(context.Background(), "user-1"); err != nil {
+		t.Fatalf("prime cache: %v", err)
+	}
+	waitForRefresh(t, refreshed)
+}
+
+func waitForRefresh(t *testing.T, refreshed chan struct{}) {
+	t.Helper()
+	select {
+	case <-refreshed:
+	case <-time.After(time.Second):
+		t.Fatal("background refresh did not run")
+	}
+}

internal/anime/service.go

@@ -17,6 +17,9 @@ import (
 type animeService struct {
 	jikan            *jikan.Client
 	repo             domain.AnimeRepository
+	topPicksCache    *topPicksCache
+	topPicksCacheTTL time.Duration
+	computeTopPicks  recommendationComputeFunc
 }
 
 const continueWatchingCarouselLimit int64 = 24
@@ -30,7 +33,14 @@ func wrapAnimes(in []jikan.Anime) []domain.Anime {
 }
 
 func NewAnimeService(jikan *jikan.Client, repo domain.AnimeRepository) *animeService {
-	return &animeService{jikan: jikan, repo: repo}
+	svc := &animeService{
+		jikan:            jikan,
+		repo:             repo,
+		topPicksCache:    &topPicksCache{entries: map[topPicksCacheKey]*topPicksCacheEntry{}},
+		topPicksCacheTTL: 15 * time.Minute,
+	}
+	svc.computeTopPicks = svc.fetchTopPicksForYou
+	return svc
 }
 
 func (s *animeService) GetCatalogSection(ctx context.Context, userID string, section string) (domain.CatalogSectionData, error) {