feat: add source cache with stale-while-revalidate

internal/domain/playback.go

@@ -5,6 +5,17 @@ import (
 	"mal/internal/db"
 )
 
+type playbackSourceRefreshKey struct{}
+
+func WithPlaybackSourceRefresh(ctx context.Context) context.Context {
+	return context.WithValue(ctx, playbackSourceRefreshKey{}, true)
+}
+
+func PlaybackSourceRefreshRequested(ctx context.Context) bool {
+	refresh, _ := ctx.Value(playbackSourceRefreshKey{}).(bool)
+	return refresh
+}
+
 type PlaybackService interface {
 	BuildWatchData(ctx context.Context, animeID int, titleCandidates []string, episode string, mode string, userID string) (WatchPageData, error)
 	SaveProgress(ctx context.Context, userID string, animeID int64, episode int, timeSeconds float64) error

internal/playback/source_cache.go

@@ -0,0 +1,152 @@
+package playback
+
+import (
+	"container/list"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"mal/internal/domain"
+)
+
+const (
+	defaultSourceCacheTTL        = 5 * time.Minute
+	defaultSourceCacheStaleTTL   = 10 * time.Minute
+	defaultSourceCacheMaxEntries = 512
+)
+
+type sourceCacheKey struct {
+	animeID int
+	episode string
+	mode    string
+}
+
+func newSourceCacheKey(animeID int, episode string, mode string) sourceCacheKey {
+	return sourceCacheKey{
+		animeID: animeID,
+		episode: strings.TrimSpace(episode),
+		mode:    normalizeSourceMode(mode),
+	}
+}
+
+func (k sourceCacheKey) flightKey() string {
+	return fmt.Sprintf("%d|%q|%q", k.animeID, k.episode, k.mode)
+}
+
+func normalizeSourceMode(mode string) string {
+	mode = strings.ToLower(strings.TrimSpace(mode))
+	if mode == "" {
+		return "sub"
+	}
+	return mode
+}
+
+type sourceCacheEntry struct {
+	key        sourceCacheKey
+	result     *domain.StreamResult
+	freshUntil time.Time
+	staleUntil time.Time
+}
+
+type sourceCacheState uint8
+
+const (
+	sourceCacheMiss sourceCacheState = iota
+	sourceCacheFresh
+	sourceCacheStale
+)
+
+type sourceCache struct {
+	mu         sync.Mutex
+	freshTTL   time.Duration
+	staleTTL   time.Duration
+	maxEntries int
+	entries    map[sourceCacheKey]*list.Element
+	lru        *list.List
+}
+
+func newSourceCache(freshTTL, staleTTL time.Duration, maxEntries int) *sourceCache {
+	if freshTTL <= 0 {
+		freshTTL = defaultSourceCacheTTL
+	}
+	if staleTTL < 0 {
+		staleTTL = 0
+	}
+	if maxEntries <= 0 {
+		maxEntries = defaultSourceCacheMaxEntries
+	}
+	return &sourceCache{
+		freshTTL:   freshTTL,
+		staleTTL:   staleTTL,
+		maxEntries: maxEntries,
+		entries:    make(map[sourceCacheKey]*list.Element, maxEntries),
+		lru:        list.New(),
+	}
+}
+
+func (c *sourceCache) get(key sourceCacheKey, now time.Time) (*domain.StreamResult, sourceCacheState) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	el := c.entries[key]
+	if el == nil {
+		return nil, sourceCacheMiss
+	}
+	entry, ok := el.Value.(sourceCacheEntry)
+	if !ok || !now.Before(entry.staleUntil) {
+		c.remove(el)
+		return nil, sourceCacheMiss
+	}
+	c.lru.MoveToFront(el)
+	if now.Before(entry.freshUntil) {
+		return cloneStreamResult(entry.result), sourceCacheFresh
+	}
+	return cloneStreamResult(entry.result), sourceCacheStale
+}
+
+func (c *sourceCache) set(key sourceCacheKey, result *domain.StreamResult, now time.Time) (evicted bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	entry := sourceCacheEntry{
+		key:        key,
+		result:     cloneStreamResult(result),
+		freshUntil: now.Add(c.freshTTL),
+		staleUntil: now.Add(c.freshTTL + c.staleTTL),
+	}
+	if el := c.entries[key]; el != nil {
+		el.Value = entry
+		c.lru.MoveToFront(el)
+		return false
+	}
+
+	c.entries[key] = c.lru.PushFront(entry)
+	for len(c.entries) > c.maxEntries {
+		back := c.lru.Back()
+		if back == nil {
+			break
+		}
+		c.remove(back)
+		evicted = true
+	}
+	return evicted
+}
+
+func (c *sourceCache) remove(el *list.Element) {
+	entry, ok := el.Value.(sourceCacheEntry)
+	if ok {
+		delete(c.entries, entry.key)
+	}
+	c.lru.Remove(el)
+}
+
+func cloneStreamResult(result *domain.StreamResult) *domain.StreamResult {
+	if result == nil {
+		return nil
+	}
+	cloned := *result
+	cloned.Subtitles = append([]domain.Subtitle(nil), result.Subtitles...)
+	cloned.Qualities = append([]domain.StreamSource(nil), result.Qualities...)
+	return &cloned
+}

internal/playback/source_cache_test.go

@@ -0,0 +1,149 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"mal/internal/domain"
+)
+
+type sourceCacheProvider struct {
+	calls atomic.Int32
+	get   func(context.Context, int, []string, string, string) (*domain.StreamResult, error)
+}
+
+func (p *sourceCacheProvider) Name() string { return "test" }
+
+func (p *sourceCacheProvider) GetStreams(ctx context.Context, animeID int, titles []string, episode string, mode string) (*domain.StreamResult, error) {
+	p.calls.Add(1)
+	return p.get(ctx, animeID, titles, episode, mode)
+}
+
+func newSourceCacheService(provider domain.Provider) *playbackService {
+	return &playbackService{
+		providers:   []domain.Provider{provider},
+		sourceCache: newSourceCache(time.Minute, time.Minute, 8),
+	}
+}
+
+func TestResolveStreamResultCachesClonedProviderResult(t *testing.T) {
+	provider := &sourceCacheProvider{get: func(context.Context, int, []string, string, string) (*domain.StreamResult, error) {
+		return &domain.StreamResult{
+			URL:       "https://cdn.example.test/episode.m3u8",
+			Subtitles: []domain.Subtitle{{URL: "https://cdn.example.test/sub.vtt", Label: "English"}},
+		}, nil
+	}}
+	svc := newSourceCacheService(provider)
+
+	first := svc.resolveStreamResult(context.Background(), 42, []string{"Title"}, "3", "SUB", false, false)
+	first.Subtitles[0].Label = "changed"
+	second := svc.resolveStreamResult(context.Background(), 42, []string{"Different title"}, "3", "sub", false, false)
+
+	if provider.calls.Load() != 1 {
+		t.Fatalf("provider calls = %d, want 1", provider.calls.Load())
+	}
+	if second == nil || second.Subtitles[0].Label != "English" {
+		t.Fatalf("cached result was not cloned: %#v", second)
+	}
+}
+
+func TestResolveStreamResultCollapsesConcurrentMisses(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	provider := &sourceCacheProvider{get: func(context.Context, int, []string, string, string) (*domain.StreamResult, error) {
+		select {
+		case <-started:
+		default:
+			close(started)
+		}
+		<-release
+		return &domain.StreamResult{URL: "https://cdn.example.test/episode.m3u8"}, nil
+	}}
+	svc := newSourceCacheService(provider)
+
+	const requests = 8
+	results := make(chan *domain.StreamResult, requests)
+	var wg sync.WaitGroup
+	for range requests {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			results <- svc.resolveStreamResult(context.Background(), 42, nil, "3", "sub", false, false)
+		}()
+	}
+	<-started
+	close(release)
+	wg.Wait()
+	close(results)
+
+	for result := range results {
+		if result == nil {
+			t.Fatal("concurrent lookup returned nil")
+		}
+	}
+	if provider.calls.Load() != 1 {
+		t.Fatalf("provider calls = %d, want 1", provider.calls.Load())
+	}
+}
+
+func TestResolveStreamResultSeparatesKeysAndForcesRefresh(t *testing.T) {
+	provider := &sourceCacheProvider{get: func(_ context.Context, _ int, _ []string, episode string, mode string) (*domain.StreamResult, error) {
+		return &domain.StreamResult{URL: "https://cdn.example.test/" + episode + "/" + mode}, nil
+	}}
+	svc := newSourceCacheService(provider)
+
+	svc.resolveStreamResult(context.Background(), 42, nil, "1", "sub", false, false)
+	svc.resolveStreamResult(context.Background(), 42, nil, "2", "sub", false, false)
+	svc.resolveStreamResult(context.Background(), 42, nil, "2", "dub", false, false)
+	svc.resolveStreamResult(context.Background(), 42, nil, "2", "dub", false, true)
+
+	if provider.calls.Load() != 4 {
+		t.Fatalf("provider calls = %d, want 4", provider.calls.Load())
+	}
+}
+
+func TestResolveStreamResultUsesStaleCompletedMediaOnProviderError(t *testing.T) {
+	provider := &sourceCacheProvider{get: func(context.Context, int, []string, string, string) (*domain.StreamResult, error) {
+		return nil, errors.New("provider unavailable")
+	}}
+	svc := newSourceCacheService(provider)
+	svc.sourceCache.set(
+		newSourceCacheKey(42, "3", "sub"),
+		&domain.StreamResult{URL: "https://cdn.example.test/stale.m3u8"},
+		time.Now().Add(-90*time.Second),
+	)
+
+	result := svc.resolveStreamResult(context.Background(), 42, nil, "3", "sub", true, false)
+	if result == nil || result.URL != "https://cdn.example.test/stale.m3u8" {
+		t.Fatalf("result = %#v, want stale source", result)
+	}
+	if airing := svc.resolveStreamResult(context.Background(), 42, nil, "3", "sub", false, false); airing != nil {
+		t.Fatalf("airing result = %#v, want nil after provider failure", airing)
+	}
+}
+
+func TestSourceCacheExpiresAndEvictsLeastRecentlyUsed(t *testing.T) {
+	now := time.Unix(100, 0)
+	cache := newSourceCache(time.Minute, time.Minute, 2)
+	keyA := newSourceCacheKey(1, "1", "sub")
+	keyB := newSourceCacheKey(1, "2", "sub")
+	keyC := newSourceCacheKey(1, "3", "sub")
+	cache.set(keyA, &domain.StreamResult{URL: "a"}, now)
+	cache.set(keyB, &domain.StreamResult{URL: "b"}, now)
+	cache.get(keyA, now)
+	cache.set(keyC, &domain.StreamResult{URL: "c"}, now)
+
+	if _, state := cache.get(keyB, now); state != sourceCacheMiss {
+		t.Fatalf("evicted cache state = %d, want miss", state)
+	}
+	if _, state := cache.get(keyA, now.Add(90*time.Second)); state != sourceCacheStale {
+		t.Fatalf("stale cache state = %d, want stale", state)
+	}
+	if _, state := cache.get(keyA, now.Add(2*time.Minute)); state != sourceCacheMiss {
+		t.Fatalf("expired cache state = %d, want miss", state)
+	}
+}