From a0d11e3a7b18077e4c53ff62a2313e9eb0df98a8 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 5 Jul 2026 23:08:26 +0200 Subject: [PATCH] feat: add source cache with stale-while-revalidate --- internal/domain/playback.go | 11 ++ internal/playback/source_cache.go | 152 +++++++++++++++++++++++++ internal/playback/source_cache_test.go | 149 ++++++++++++++++++++++++ 3 files changed, 312 insertions(+) create mode 100644 internal/playback/source_cache.go create mode 100644 internal/playback/source_cache_test.go diff --git a/internal/domain/playback.go b/internal/domain/playback.go index f819a10f..46e26475 100644 --- a/internal/domain/playback.go +++ b/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 diff --git a/internal/playback/source_cache.go b/internal/playback/source_cache.go new file mode 100644 index 00000000..6c2c59b3 --- /dev/null +++ b/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 +} diff --git a/internal/playback/source_cache_test.go b/internal/playback/source_cache_test.go new file mode 100644 index 00000000..aa4b6760 --- /dev/null +++ b/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) + } +}