mal/internal/playback/watch_data.go

package playback

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"mal/integrations/jikan"
	"mal/internal/domain"
	"mal/internal/observability"
)

const sourceResolutionTimeout = 15 * time.Second

type sourceResolutionResult struct {
	result    *domain.StreamResult
	err       error
	duration  time.Duration
	shared    bool
	completed bool
}

const episodeAvailabilityUncertainWarning = "Episode availability may be incomplete or out of date. Continue only if you understand that the episode list and audio availability may be uncertain."

func (s *playbackService) BuildWatchData(ctx context.Context, animeID int, titleCandidates []string, episode string, mode string, userID string) (domain.WatchPageData, error) {
	animeData, err := s.watchAnime(ctx, animeID)
	if err != nil {
		return domain.WatchPageData{}, fmt.Errorf("failed to fetch anime: %w", err)
	}

	if err := s.ensureAnimeRow(ctx, animeData); err != nil {
		observability.Warn("upsert_anime_failed", "playback", "",
			map[string]any{"anime_id": animeID},
			err,
		)
	}
	anime := animeData.Anime
	searchTitles := buildSearchTitles(animeData, titleCandidates)
	eps, err := s.episodes.GetCanonicalEpisodes(ctx, animeData, false)
	if err != nil {
		return domain.WatchPageData{}, fmt.Errorf("failed to fetch episodes: %w", err)
	}

	// mode fallback
	mode, from := resolveMode(episode, mode, eps.Episodes)
	deferred := domain.PlaybackDataDeferred(ctx)
	modeSources, result, mode, from := s.watchModeSources(
		ctx, animeID, searchTitles, episode, mode, from, !anime.Airing, deferred,
	)
	startTime, watchlistStatus, watchlistIDs := s.loadWatchProgress(ctx, userID, animeID, anime.Episodes, episode)
	segments := s.watchSegments(ctx, userID, animeID, episode, deferred)
	watchData := buildWatchDataPayload(animeData, animeID, episode, startTime, eps.Episodes, modeSources, mode, from, segments)
	pageData := buildWatchPageData(animeData, eps.Episodes, episode, watchlistStatus, watchlistIDs, nil, watchData)
	pageData.EpisodeAvailabilityWarning = episodeAvailabilityWarning(eps, time.Now())
	if deferred {
		return pageData, nil
	}
	if len(modeSources) == 0 {
		return pageData, fmt.Errorf("no streams found")
	}
	if result == nil {
		return pageData, fmt.Errorf("no streams found for mode %s", mode)
	}

	return pageData, nil
}

func (s *playbackService) watchAnime(ctx context.Context, animeID int) (domain.Anime, error) {
	row, err := s.repo.GetAnime(ctx, int64(animeID))
	if err == nil && row.ID > 0 && strings.TrimSpace(row.TitleOriginal) != "" {
		anime := jikan.Anime{
			MalID:         int(row.ID),
			Title:         row.TitleOriginal,
			TitleEnglish:  row.TitleEnglish.String,
			TitleJapanese: row.TitleJapanese.String,
			Airing:        row.Airing.Valid && row.Airing.Bool,
			Status:        row.Status.String,
		}
		return domain.Anime{Anime: anime}, nil
	}

	anime, err := s.jikan.GetAnimeByID(ctx, animeID)
	if err != nil {
		return domain.Anime{}, err
	}
	return domain.Anime{Anime: anime}, nil
}

func (s *playbackService) watchModeSources(ctx context.Context, animeID int, searchTitles []string, episode, mode, from string, allowStale, deferred bool) (map[string]domain.ModeSource, *domain.StreamResult, string, string) {
	if deferred {
		return map[string]domain.ModeSource{}, nil, mode, from
	}

	modeSources, result, resolvedMode, switchedFrom := s.resolveModeSources(
		ctx,
		animeID,
		searchTitles,
		episode,
		mode,
		allowStale,
		domain.PlaybackSourceRefreshRequested(ctx),
	)
	if resolvedMode != "" {
		mode = resolvedMode
	}
	if switchedFrom != "" {
		from = switchedFrom
	}
	return modeSources, result, mode, from
}

func (s *playbackService) watchSegments(ctx context.Context, userID string, animeID int, episode string, deferred bool) []domain.SkipSegment {
	if deferred {
		return []domain.SkipSegment{}
	}
	segments, err := s.fetchSkipSegments(ctx, userID, animeID, episode)
	if err != nil {
		observability.Warn("fetch_skip_segments_failed", "playback", "",
			map[string]any{"anime_id": animeID, "episode": episode},
			err,
		)
	}
	return segments
}

func buildWatchDataPayload(anime domain.Anime, animeID int, episode string, startTime float64, episodes []domain.CanonicalEpisode, modeSources map[string]domain.ModeSource, mode string, modeSwitchedFrom string, segments []domain.SkipSegment) domain.WatchData {
	return domain.WatchData{
		MalID:            animeID,
		Title:            anime.DisplayTitle(),
		CurrentEpisode:   episode,
		StartTimeSeconds: startTime,
		Episodes:         episodes,
		Providers: []domain.ProviderData{{Streams: []domain.ProviderStream{{
			Name:      "Primary",
			Quality:   "Auto",
			MalID:     animeID,
			IsCurrent: true,
		}}}},
		ModeSources:      modeSources,
		InitialMode:      mode,
		ModeSwitchedFrom: modeSwitchedFrom,
		AvailableModes:   availableModes(modeSources),
		Segments:         segments,
		Airing:           anime.Airing,
	}
}

func buildWatchPageData(anime domain.Anime, episodes []domain.CanonicalEpisode, episode string, watchlistStatus string, watchlistIDs []int64, seasons []domain.SeasonEntry, watchData domain.WatchData) domain.WatchPageData {
	return domain.WatchPageData{
		WatchData:       watchData,
		Anime:           anime,
		Episodes:        episodes,
		CurrentEpID:     episode,
		WatchlistStatus: watchlistStatus,
		WatchlistIDs:    watchlistIDs,
		Seasons:         seasons,
	}
}

func episodeAvailabilityWarning(episodeList domain.CanonicalEpisodeList, now time.Time) string {
	if episodeList.FailureCount > 0 {
		return episodeAvailabilityUncertainWarning
	}
	if episodeList.NextRefreshAt == "" {
		return ""
	}
	nextRefresh, err := time.Parse(time.RFC3339, episodeList.NextRefreshAt)
	if err != nil {
		return ""
	}
	if nextRefresh.After(now) {
		return ""
	}
	return episodeAvailabilityUncertainWarning
}

func buildSearchTitles(anime domain.Anime, titleCandidates []string) []string {
	seen := map[string]struct{}{}
	out := make([]string, 0, 3+len(anime.TitleSynonyms)+len(titleCandidates))

	appendTitle := func(title string) {
		title = strings.TrimSpace(title)
		if title == "" {
			return
		}
		if _, ok := seen[title]; ok {
			return
		}
		seen[title] = struct{}{}
		out = append(out, title)
	}

	appendTitle(anime.Title)
	appendTitle(anime.TitleEnglish)
	appendTitle(anime.TitleJapanese)
	for _, syn := range anime.TitleSynonyms {
		appendTitle(syn)
	}
	for _, candidate := range titleCandidates {
		appendTitle(candidate)
	}

	return out
}

func resolveMode(episode string, requestedMode string, episodes []domain.CanonicalEpisode) (string, string) {
	if requestedMode != "dub" {
		return requestedMode, ""
	}

	epNum, err := strconv.Atoi(episode)
	if err != nil {
		return requestedMode, ""
	}

	for _, ep := range episodes {
		if ep.Number == epNum && !ep.HasDub && ep.HasSub {
			return "sub", requestedMode
		}
	}

	return requestedMode, ""
}

func (s *playbackService) resolveModeSources(ctx context.Context, animeID int, searchTitles []string, episode string, requestedMode string, allowStale bool, forceRefresh bool) (map[string]domain.ModeSource, *domain.StreamResult, string, string) {
	requestedMode = normalizeSourceMode(requestedMode)
	if res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, requestedMode, allowStale, forceRefresh); res != nil {
		return map[string]domain.ModeSource{
			requestedMode: s.buildModeSource(res),
		}, res, requestedMode, ""
	}

	for _, fallbackMode := range fallbackModes(requestedMode) {
		res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, fallbackMode, allowStale, forceRefresh)
		if res == nil {
			continue
		}
		return map[string]domain.ModeSource{
			fallbackMode: s.buildModeSource(res),
		}, res, fallbackMode, requestedMode
	}

	return map[string]domain.ModeSource{}, nil, requestedMode, ""
}

func (s *playbackService) resolveStreamResult(ctx context.Context, animeID int, searchTitles []string, episode string, mode string, allowStale bool, forceRefresh bool) *domain.StreamResult {
	key := newSourceCacheKey(animeID, episode, mode)
	stale, state := s.sourceCache.get(key, time.Now())
	if !forceRefresh && state == sourceCacheFresh {
		observability.Info("playback_source_cache_hit", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode})
		return stale
	}

	observability.Info("playback_source_cache_miss", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode, "forced": forceRefresh})
	resolved := s.waitForSourceResult(ctx, key, animeID, searchTitles, forceRefresh)
	if !resolved.completed {
		return nil
	}
	observability.Info("playback_source_resolution", "playback", "", map[string]any{
		"anime_id": animeID, "episode": episode, "mode": key.mode,
		"duration_ms": resolved.duration.Milliseconds(), "shared": resolved.shared,
	})
	if resolved.err == nil {
		return cloneStreamResult(resolved.result)
	}
	if allowStale && state == sourceCacheStale && stale != nil {
		observability.Warn("playback_source_cache_stale_hit", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode}, errors.New("provider source refresh failed"))
		return stale
	}
	observability.Warn("playback_source_resolution_failed", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode}, errors.New("provider source resolution failed"))
	return nil
}

func (s *playbackService) waitForSourceResult(ctx context.Context, key sourceCacheKey, animeID int, searchTitles []string, forceRefresh bool) sourceResolutionResult {
	startedAt := time.Now()
	resultCh := s.sourceFlight.DoChan(key.flightKey(), func() (any, error) {
		return s.resolveSource(key, animeID, searchTitles, forceRefresh)
	})
	select {
	case <-ctx.Done():
		return sourceResolutionResult{err: ctx.Err(), duration: time.Since(startedAt)}
	case resolved := <-resultCh:
		result, _ := resolved.Val.(*domain.StreamResult)
		return sourceResolutionResult{
			result: result, err: resolved.Err, shared: resolved.Shared,
			duration: time.Since(startedAt), completed: true,
		}
	}
}

func (s *playbackService) resolveSource(key sourceCacheKey, animeID int, searchTitles []string, forceRefresh bool) (*domain.StreamResult, error) {
	if !forceRefresh {
		if cached, state := s.sourceCache.get(key, time.Now()); state == sourceCacheFresh {
			return cached, nil
		}
	}
	resolveCtx, cancel := context.WithTimeout(context.Background(), sourceResolutionTimeout)
	defer cancel()

	var lastErr error
	for _, provider := range s.providers {
		result, err := provider.GetStreams(resolveCtx, animeID, searchTitles, key.episode, key.mode)
		if err == nil && result != nil {
			if s.sourceCache.set(key, result, time.Now()) {
				observability.Info("playback_source_cache_eviction", "playback", "", nil)
			}
			return cloneStreamResult(result), nil
		}
		if err != nil {
			lastErr = err
		}
	}
	if lastErr == nil {
		lastErr = errors.New("no provider returned a stream")
	}
	return nil, lastErr
}

func (s *playbackService) buildModeSource(res *domain.StreamResult) domain.ModeSource {
	subtitles := make([]domain.SubtitleItem, 0, len(res.Subtitles))
	for _, sub := range res.Subtitles {
		token, err := s.SignProxyToken(sub.URL, res.Referer, "subtitle")
		if err != nil {
			observability.Warn("sign_subtitle_token_failed", "playback", "", nil, err)
		}
		subtitles = append(subtitles, domain.SubtitleItem{
			Lang:  sub.Label,
			Token: token,
		})
	}

	streamToken, err := s.SignProxyToken(res.URL, res.Referer, "stream")
	if err != nil {
		observability.Warn("sign_stream_token_failed", "playback", "", nil, err)
	}
	return domain.ModeSource{
		Token:     streamToken,
		Type:      res.Type,
		Subtitles: subtitles,
	}
}

func availableModes(modeSources map[string]domain.ModeSource) []string {
	modes := make([]string, 0, len(modeSources))
	for mode := range modeSources {
		modes = append(modes, mode)
	}
	sort.Strings(modes)
	return modes
}

func fallbackModes(requestedMode string) []string {
	switch requestedMode {
	case "sub":
		return []string{"dub"}
	case "dub":
		return []string{"sub"}
	default:
		return []string{"sub", "dub"}
	}
}