diff --git a/internal/episodes/service/cache_store.go b/internal/episodes/service/cache_store.go index 91cecb54..00140534 100644 --- a/internal/episodes/service/cache_store.go +++ b/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, - Episodes: episodes, - Source: source, - LastAttemptAt: now.Format(time.RFC3339), - FailureCount: 0, + AnimeID: anime.MalID, + Episodes: episodes, + Source: source, + AvailabilityVersion: episodeAvailabilityPayloadVersion, + 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") +} diff --git a/internal/episodes/service/merge.go b/internal/episodes/service/merge.go index 99f7bb2b..95fb44df 100644 --- a/internal/episodes/service/merge.go +++ b/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 } diff --git a/internal/episodes/service/service.go b/internal/episodes/service/service.go index 1b43db7c..2c1912b1 100644 --- a/internal/episodes/service/service.go +++ b/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, - Episodes: mergeEpisodes(episodes, domain.EpisodeAvailability{}, anime.Episodes), - Source: source, + AnimeID: anime.MalID, + Episodes: mergeEpisodesForAnime(anime, episodes, domain.EpisodeAvailability{}, s.clock.Now(), false), + Source: source, + AvailabilityVersion: episodeAvailabilityPayloadVersion, + ReleaseChecked: true, }, nil } diff --git a/internal/episodes/service/service_test.go b/internal/episodes/service/service_test.go index 77394a7f..896c2c35 100644 --- a/internal/episodes/service/service_test.go +++ b/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 {