feat: reject stale caches and filter future episodes

internal/episodes/service/cache_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"strings"
 	"time"
 
 	"mal/integrations/jikan"
@@ -12,15 +13,19 @@ import (
 	"mal/internal/observability"
 )
 
+const episodeAvailabilityPayloadVersion = 2
+
 func (s *EpisodeService) store(ctx context.Context, anime domain.Anime, jikanEpisodes []jikan.Episode, availability domain.EpisodeAvailability, source string, now time.Time, providerSuccess bool) (domain.CanonicalEpisodeList, error) {
 	nextRefreshSQL := nextRefreshAt(anime, now)
-	episodes := mergeEpisodes(jikanEpisodes, availability, anime.Episodes)
+	episodes := mergeEpisodesForAnime(anime, jikanEpisodes, availability, now, providerSuccess)
 	payload := domain.CanonicalEpisodeList{
-		AnimeID:       anime.MalID,
+		AnimeID:             anime.MalID,
-		Episodes:      episodes,
+		Episodes:            episodes,
-		Source:        source,
+		Source:              source,
-		LastAttemptAt: now.Format(time.RFC3339),
+		AvailabilityVersion: episodeAvailabilityPayloadVersion,
-		FailureCount:  0,
+		ReleaseChecked:      !providerSuccess,
+		LastAttemptAt:       now.Format(time.RFC3339),
+		FailureCount:        0,
 	}
 	if providerSuccess {
 		payload.LastSuccessAt = now.Format(time.RFC3339)
@@ -238,6 +243,35 @@ func (s *EpisodeService) decodeCachedPayload(anime domain.Anime, raw string) (do
 		return domain.CanonicalEpisodeList{}, false
 	}
 
+	if isStaleAiringEpisodePayload(anime, payload) {
+		observability.Info(
+			"episodes_cached_payload_rejected_stale_version",
+			"episodes",
+			"",
+			map[string]any{
+				"anime_id":             anime.MalID,
+				"cached_episodes":      len(payload.Episodes),
+				"source":               payload.Source,
+				"availability_version": payload.AvailabilityVersion,
+			},
+		)
+		return domain.CanonicalEpisodeList{}, false
+	}
+
+	if isUncheckedAiringJikanFallback(anime, payload) {
+		observability.Info(
+			"episodes_cached_payload_rejected_unchecked_fallback",
+			"episodes",
+			"",
+			map[string]any{
+				"anime_id":        anime.MalID,
+				"cached_episodes": len(payload.Episodes),
+				"source":          payload.Source,
+			},
+		)
+		return domain.CanonicalEpisodeList{}, false
+	}
+
 	if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) {
 		observability.Info(
 			"episodes_cached_payload_rejected",
@@ -254,3 +288,18 @@ func (s *EpisodeService) decodeCachedPayload(anime domain.Anime, raw string) (do
 
 	return payload, true
 }
+
+func isStaleAiringEpisodePayload(anime domain.Anime, payload domain.CanonicalEpisodeList) bool {
+	return isAiring(anime) && len(payload.Episodes) > 0 && payload.AvailabilityVersion < episodeAvailabilityPayloadVersion
+}
+
+func isUncheckedAiringJikanFallback(anime domain.Anime, payload domain.CanonicalEpisodeList) bool {
+	if !isAiring(anime) || payload.ReleaseChecked || len(payload.Episodes) == 0 {
+		return false
+	}
+	return payload.Source == "jikan_fallback" || payload.Source == "legacy_disabled"
+}
+
+func isAiring(anime domain.Anime) bool {
+	return anime.Airing || strings.EqualFold(strings.TrimSpace(anime.Status), "Currently Airing")
+}

internal/episodes/service/merge.go

@@ -5,6 +5,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"mal/integrations/jikan"
 	"mal/internal/domain"
@@ -62,15 +63,23 @@ func providerBackedPayloadHasAvailability(payload domain.CanonicalEpisodeList) b
 }
 
 func mergeEpisodes(jikanEpisodes []jikan.Episode, availability domain.EpisodeAvailability, expectedCount int) []domain.CanonicalEpisode {
+	return mergeEpisodeData(jikanEpisodes, availability, expectedCount, time.Now(), false, "", false)
+}
+
+func mergeEpisodesForAnime(anime domain.Anime, jikanEpisodes []jikan.Episode, availability domain.EpisodeAvailability, now time.Time, providerVerified bool) []domain.CanonicalEpisode {
+	return mergeEpisodeData(jikanEpisodes, availability, anime.Episodes, now, providerVerified, anime.Aired.From, anime.Airing)
+}
+
+func mergeEpisodeData(jikanEpisodes []jikan.Episode, availability domain.EpisodeAvailability, expectedCount int, now time.Time, providerVerified bool, firstAired string, requireJikanAiredDates bool) []domain.CanonicalEpisode {
 	byNumber := map[int]episodePartial{}
 	providerNumbers := availableEpisodeNumbers(availability, expectedCount)
-	providerBacked := len(providerNumbers) > 0
+	providerBacked := providerVerified || len(providerNumbers) > 0
 
 	for number := range providerNumbers {
 		mergeEpisode(&byNumber, number, func(item *episodePartial) {})
 	}
 
-	mergeJikanEpisodes(&byNumber, jikanEpisodes, providerNumbers, providerBacked, expectedCount)
+	mergeJikanEpisodes(&byNumber, jikanEpisodes, providerNumbers, providerBacked, expectedCount, now, firstAired, requireJikanAiredDates)
 	mergeAvailability(&byNumber, availability.Sub, expectedCount, func(item *episodePartial) { item.sub = true })
 	mergeAvailability(&byNumber, availability.Dub, expectedCount, func(item *episodePartial) { item.dub = true })
 
@@ -100,13 +109,26 @@ func mergeEpisodes(jikanEpisodes []jikan.Episode, availability domain.EpisodeAva
 	return episodes
 }
 
-func mergeJikanEpisodes(byNumber *map[int]episodePartial, episodes []jikan.Episode, providerNumbers map[int]bool, providerBacked bool, expectedCount int) {
+func mergeJikanEpisodes(byNumber *map[int]episodePartial, episodes []jikan.Episode, providerNumbers map[int]bool, providerBacked bool, expectedCount int, now time.Time, firstAired string, requireAiredDates bool) {
+	if shouldSkipJikanMerge(providerBacked, firstAired, now) {
+		return
+	}
+
 	for i, ep := range episodes {
 		if exceedsExpectedCount(i+1, expectedCount) {
 			break
 		}
 		number, ok := jikanEpisodeNumber(ep, i)
-		if !ok || exceedsExpectedCount(number, expectedCount) || (providerBacked && !providerNumbers[number]) {
+		if !ok {
+			continue
+		}
+		if exceedsExpectedCount(number, expectedCount) {
+			continue
+		}
+		if providerBacked && !providerNumbers[number] {
+			continue
+		}
+		if !providerBacked && !hasEpisodeAired(ep, now, requireAiredDates) {
 			continue
 		}
 		mergeEpisode(byNumber, number, func(item *episodePartial) {
@@ -117,6 +139,10 @@ func mergeJikanEpisodes(byNumber *map[int]episodePartial, episodes []jikan.Episo
 	}
 }
 
+func shouldSkipJikanMerge(providerBacked bool, firstAired string, now time.Time) bool {
+	return !providerBacked && !hasStartedAiring(firstAired, now)
+}
+
 func availableEpisodeNumbers(availability domain.EpisodeAvailability, expectedCount int) map[int]bool {
 	numbers := map[int]bool{}
 	for _, number := range availability.Sub {
@@ -158,6 +184,28 @@ func jikanEpisodeNumber(ep jikan.Episode, index int) (int, bool) {
 	return index + 1, true
 }
 
+func hasStartedAiring(firstAired string, now time.Time) bool {
+	if strings.TrimSpace(firstAired) == "" {
+		return true
+	}
+	startedAt, err := time.Parse(time.RFC3339, firstAired)
+	if err != nil {
+		return true
+	}
+	return !now.Before(startedAt)
+}
+
+func hasEpisodeAired(ep jikan.Episode, now time.Time, requireAiredDate bool) bool {
+	if strings.TrimSpace(ep.Aired) == "" {
+		return !requireAiredDate
+	}
+	airedAt, err := time.Parse(time.RFC3339, ep.Aired)
+	if err != nil {
+		return !requireAiredDate
+	}
+	return !now.Before(airedAt)
+}
+
 func exceedsExpectedCount(number int, expectedCount int) bool {
 	return expectedCount > 0 && number > expectedCount
 }

internal/episodes/service/service.go

@@ -222,8 +222,10 @@ func (s *EpisodeService) jikanOnly(ctx context.Context, anime domain.Anime, sour
 		return domain.CanonicalEpisodeList{}, err
 	}
 	return domain.CanonicalEpisodeList{
-		AnimeID:  anime.MalID,
+		AnimeID:             anime.MalID,
-		Episodes: mergeEpisodes(episodes, domain.EpisodeAvailability{}, anime.Episodes),
+		Episodes:            mergeEpisodesForAnime(anime, episodes, domain.EpisodeAvailability{}, s.clock.Now(), false),
-		Source:   source,
+		Source:              source,
+		AvailabilityVersion: episodeAvailabilityPayloadVersion,
+		ReleaseChecked:      true,
 	}, nil
 }

internal/episodes/service/service_test.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"database/sql"
+	"encoding/json"
 	"mal/integrations/jikan"
 	"mal/internal/db"
 	"mal/internal/domain"
@@ -43,6 +44,74 @@ func TestMergeEpisodesUsesJikanWhenProviderAvailabilityMissing(t *testing.T) {
 	assertEpisode(t, episodes[1], 2, "Second", false, false, false, false)
 }
 
+func TestMergeEpisodesSkipsFutureJikanEpisodesWithoutProviderAvailability(t *testing.T) {
+	now := time.Date(2026, time.July, 1, 12, 0, 0, 0, time.UTC)
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+		Aired:  jikan.Aired{From: "2026-06-03T00:00:00+00:00"},
+	}}
+	episodes := mergeEpisodesForAnime(anime, decodeJikanEpisodes(t, `[
+		{"mal_id":1,"title":"Episode 1","episode":null,"aired":"2026-07-09T00:00:00+00:00"},
+		{"mal_id":2,"title":"Episode 2","episode":null,"aired":"2026-07-16T00:00:00+00:00"}
+	]`), domain.EpisodeAvailability{}, now, false)
+
+	if len(episodes) != 0 {
+		t.Fatalf("len(episodes) = %d, want 0", len(episodes))
+	}
+}
+
+func TestMergeEpisodesSkipsUndatedJikanEpisodesForAiringAnimeWithoutProviderAvailability(t *testing.T) {
+	now := time.Date(2026, time.July, 1, 12, 0, 0, 0, time.UTC)
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+		Aired:  jikan.Aired{From: "2026-06-03T00:00:00+00:00"},
+	}}
+	episodes := mergeEpisodesForAnime(anime, []jikan.Episode{
+		{MalID: 1, Title: "Episode 1"},
+		{MalID: 2, Title: "Episode 2"},
+	}, domain.EpisodeAvailability{}, now, false)
+
+	if len(episodes) != 0 {
+		t.Fatalf("len(episodes) = %d, want 0", len(episodes))
+	}
+}
+
+func TestMergeEpisodesKeepsReleasedJikanEpisodesWithoutProviderAvailability(t *testing.T) {
+	now := time.Date(2026, time.July, 10, 12, 0, 0, 0, time.UTC)
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+		Aired:  jikan.Aired{From: "2026-06-03T00:00:00+00:00"},
+	}}
+	episodes := mergeEpisodesForAnime(anime, decodeJikanEpisodes(t, `[
+		{"mal_id":1,"title":"Episode 1","episode":null,"aired":"2026-07-09T00:00:00+00:00"},
+		{"mal_id":2,"title":"Episode 2","episode":null,"aired":"2026-07-16T00:00:00+00:00"}
+	]`), domain.EpisodeAvailability{}, now, false)
+
+	if len(episodes) != 1 {
+		t.Fatalf("len(episodes) = %d, want 1", len(episodes))
+	}
+	assertEpisode(t, episodes[0], 1, "Episode 1", false, false, false, false)
+}
+
+func TestMergeEpisodesTreatsEmptyProviderAvailabilityAsAuthoritative(t *testing.T) {
+	now := time.Date(2026, time.July, 10, 12, 0, 0, 0, time.UTC)
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+		Aired:  jikan.Aired{From: "2026-06-03T00:00:00+00:00"},
+	}}
+	episodes := mergeEpisodesForAnime(anime, decodeJikanEpisodes(t, `[
+		{"mal_id":1,"title":"Episode 1","episode":null,"aired":"2026-07-09T00:00:00+00:00"}
+	]`), domain.EpisodeAvailability{}, now, true)
+
+	if len(episodes) != 0 {
+		t.Fatalf("len(episodes) = %d, want 0", len(episodes))
+	}
+}
+
 func TestMergeEpisodesIgnoresInvalidJikanEpisodeNumbers(t *testing.T) {
 	episodes := mergeEpisodes([]jikan.Episode{
 		{MalID: 201, Episode: "", Title: "Missing"},
@@ -129,6 +198,58 @@ func TestIsCanonicalEpisodePayloadValidAllowsJikanFallbackWithoutAvailability(t
 	}
 }
 
+func TestDecodeCachedPayloadRejectsUncheckedAiringJikanFallback(t *testing.T) {
+	svc := &EpisodeService{}
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+	}}
+	raw := `{"anime_id":62076,"source":"jikan_fallback","episodes":[{"number":1,"title":"Episode 1"}]}`
+
+	if _, ok := svc.decodeCachedPayload(anime, raw); ok {
+		t.Fatal("expected unchecked airing jikan fallback cache to be rejected")
+	}
+}
+
+func TestDecodeCachedPayloadRejectsOldReleaseCheckedAiringFallback(t *testing.T) {
+	svc := &EpisodeService{}
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+	}}
+	raw := `{"anime_id":62076,"source":"jikan_fallback","release_checked":true,"episodes":[{"number":1,"title":"Episode 1"}]}`
+
+	if _, ok := svc.decodeCachedPayload(anime, raw); ok {
+		t.Fatal("expected old release-checked jikan fallback cache to be rejected")
+	}
+}
+
+func TestDecodeCachedPayloadRejectsOldAiringProviderPayload(t *testing.T) {
+	svc := &EpisodeService{}
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+	}}
+	raw := `{"anime_id":62076,"source":"AllAnime","episodes":[{"number":1,"title":"Episode 1","has_sub":true}]}`
+
+	if _, ok := svc.decodeCachedPayload(anime, raw); ok {
+		t.Fatal("expected old airing provider cache to be rejected")
+	}
+}
+
+func TestDecodeCachedPayloadAllowsCurrentReleaseCheckedJikanFallback(t *testing.T) {
+	svc := &EpisodeService{}
+	anime := domain.Anime{Anime: jikan.Anime{
+		MalID:  62076,
+		Airing: true,
+	}}
+	raw := `{"anime_id":62076,"source":"jikan_fallback","availability_version":2,"release_checked":true,"episodes":[{"number":1,"title":"Episode 1"}]}`
+
+	if _, ok := svc.decodeCachedPayload(anime, raw); !ok {
+		t.Fatal("expected release-checked jikan fallback cache to be valid")
+	}
+}
+
 func TestEnrichCachedPayloadAddsRefreshMetadata(t *testing.T) {
 	now := time.Date(2026, time.June, 27, 11, 0, 0, 0, time.UTC)
 	payload := enrichCachedPayload(domain.CanonicalEpisodeList{
@@ -193,6 +314,16 @@ func TestNextRetryTimeWithinAndAfterRetryWindow(t *testing.T) {
 	}
 }
 
+func decodeJikanEpisodes(t *testing.T, raw string) []jikan.Episode {
+	t.Helper()
+
+	var episodes []jikan.Episode
+	if err := json.Unmarshal([]byte(raw), &episodes); err != nil {
+		t.Fatalf("json.Unmarshal episodes: %v", err)
+	}
+	return episodes
+}
+
 func assertEpisode(t *testing.T, got domain.CanonicalEpisode, number int, title string, sub bool, dub bool, subOnly bool, filler bool) {
 	t.Helper()
 	if got.Number != number || got.Title != title || got.HasSub != sub || got.HasDub != dub || got.SubOnly != subOnly || got.Filler != filler || got.Recap {