Add episode streaming functionality and caching

8 files changed, 319 insertions, 14 deletions

diff --git a/controllers/anime.go b/controllers/anime.go
index 11da647..83e0039 100644
--- a/controllers/anime.go
+++ b/controllers/anime.go
@@ -66,6 +66,56 @@ func GetAnimeEpisodes(c *fiber.Ctx) error {
 	return c.JSON(anime.Episodes)
 }
 
+// GetAnimeEpisode fetches a single episode by anime ID and episode ID
+func GetAnimeEpisode(c *fiber.Ctx) error {
+	mapping, err := getAnimeMapping(c)
+	if err != nil {
+		return err
+	}
+
+	episodeID := c.Params("episodeId")
+	if episodeID == "" {
+		return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{
+			"error": "Episode ID is required",
+		})
+	}
+
+	service := getAnimeService()
+	anime, err := service.GetAnimeDetails(mapping)
+	if err != nil {
+		logger.Log("Failed to fetch anime details: "+err.Error(), logger.LogOptions{
+			Level:  logger.Error,
+			Prefix: "AnimeAPI",
+		})
+		return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
+			"error": "Failed to fetch anime details",
+		})
+	}
+
+	// Find the episode with matching ID
+	for i, episode := range anime.Episodes.Episodes {
+		if episode.ID == episodeID {
+			// Fetch streaming sources for this specific episode
+			episodeNumber := i + 1
+			streaming, err := service.GetEpisodeStreaming(anime.Titles.Romaji, episodeNumber, episode.ID, uint(anime.MALID))
+			if err != nil {
+				logger.Log("Failed to fetch streaming sources: "+err.Error(), logger.LogOptions{
+					Level:  logger.Warn,
+					Prefix: "AnimeAPI",
+				})
+				// Continue without streaming data
+			} else {
+				episode.Streaming = streaming
+			}
+			return c.JSON(episode)
+		}
+	}
+
+	return c.Status(fiber.StatusNotFound).JSON(fiber.Map{
+		"error": "Episode not found",
+	})
+}
+
 func GetAnimeCharacters(c *fiber.Ctx) error {
 	mapping, err := getAnimeMapping(c)
 	if err != nil {
diff --git a/database/anime.go b/database/anime.go
index 02bac41..e210b62 100644
--- a/database/anime.go
+++ b/database/anime.go
@@ -314,3 +314,91 @@ func GetAnimeMappingsByTVDBID(tvdbID int) ([]entities.AnimeMapping, error) {
 	}
 	return mappings, nil
 }
+
+// GetEpisodeStreaming retrieves cached streaming data for an episode
+func GetEpisodeStreaming(episodeID string, animeID uint) (*entities.EpisodeStreaming, error) {
+	var streaming entities.EpisodeStreaming
+	result := DB.Preload("SubSources").
+		Preload("DubSources").
+		Where("episode_id = ? AND anime_id = ?", episodeID, animeID).
+		First(&streaming)
+
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	// Check if data is stale (older than 7 days)
+	if time.Since(streaming.LastFetch) > 7*24*time.Hour {
+		return nil, fmt.Errorf("streaming data is stale")
+	}
+
+	return &streaming, nil
+}
+
+// SaveEpisodeStreaming saves streaming data to the database
+func SaveEpisodeStreaming(episodeID string, animeID uint, subSources, dubSources []types.AnimeStreamingSource) error {
+	tx := DB.Begin()
+	if tx.Error != nil {
+		return tx.Error
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			tx.Rollback()
+		}
+	}()
+
+	// Delete existing streaming data for this episode
+	var existing entities.EpisodeStreaming
+	if err := tx.Where("episode_id = ? AND anime_id = ?", episodeID, animeID).First(&existing).Error; err == nil {
+		if err := tx.Delete(&existing).Error; err != nil {
+			tx.Rollback()
+			return err
+		}
+	}
+
+	// Create new streaming record
+	streaming := &entities.EpisodeStreaming{
+		EpisodeID: episodeID,
+		AnimeID:   animeID,
+		LastFetch: time.Now(),
+	}
+
+	// Save the main record first
+	if err := tx.Create(streaming).Error; err != nil {
+		tx.Rollback()
+		return err
+	}
+
+	// Save sub sources
+	for _, source := range subSources {
+		subSource := entities.EpisodeStreamingSource{
+			EpisodeStreamingID: streaming.ID,
+			URL:                source.URL,
+			Server:             source.Server,
+			Type:               source.Type,
+		}
+		if err := tx.Create(&subSource).Error; err != nil {
+			tx.Rollback()
+			return err
+		}
+		streaming.SubSources = append(streaming.SubSources, subSource)
+	}
+
+	// Save dub sources
+	for _, source := range dubSources {
+		dubSource := entities.EpisodeStreamingSource{
+			EpisodeStreamingID: streaming.ID,
+			URL:                source.URL,
+			Server:             source.Server,
+			Type:               source.Type,
+		}
+		if err := tx.Create(&dubSource).Error; err != nil {
+			tx.Rollback()
+			return err
+		}
+		streaming.DubSources = append(streaming.DubSources, dubSource)
+	}
+
+	return tx.Commit().Error
+}
diff --git a/database/migrate.go b/database/migrate.go
index 89cdb7d..9e7c1b9 100644
--- a/database/migrate.go
+++ b/database/migrate.go
@@ -32,6 +32,10 @@ func AutoMigrate() {
 		&entities.AnimeCharacter{},
 		&entities.AnimeVoiceActor{},
 		&entities.AnimeSeason{},
+
+		// Streaming entities
+		&entities.EpisodeStreaming{},
+		&entities.EpisodeStreamingSource{},
 	)
 	if err != nil {
 		logger.Log(fmt.Sprintf("Failed to migrate database: %v", err), logger.LogOptions{
diff --git a/entities/anime.go b/entities/anime.go
index a2a9988..737cae4 100644
--- a/entities/anime.go
+++ b/entities/anime.go
@@ -515,3 +515,22 @@ type CachedScheduleEpisode struct {
 	AiringAt int
 	Episode  int
 }
+
+// EpisodeStreamingSource stores individual streaming sources for episodes
+type EpisodeStreamingSource struct {
+	gorm.Model
+	EpisodeStreamingID uint
+	URL                string
+	Server             string
+	Type               string // M3U8, MP4, or embed
+}
+
+// EpisodeStreaming stores streaming data for a specific episode
+type EpisodeStreaming struct {
+	gorm.Model
+	EpisodeID  string                   `gorm:"uniqueIndex:idx_episode_streaming;size:32"`
+	AnimeID    uint                     `gorm:"uniqueIndex:idx_episode_streaming"`
+	SubSources []EpisodeStreamingSource `gorm:"foreignKey:EpisodeStreamingID;constraint:OnDelete:CASCADE"`
+	DubSources []EpisodeStreamingSource `gorm:"foreignKey:EpisodeStreamingID;constraint:OnDelete:CASCADE"`
+	LastFetch  time.Time
+}
diff --git a/router/router.go b/router/router.go
index 4591bc5..4358477 100644
--- a/router/router.go
+++ b/router/router.go
@@ -14,6 +14,7 @@ func Initialize(router *fiber.App) {
 	animeRouter := router.Group("/a")
 	animeRouter.Get("/:id", controllers.GetAnime)
 	animeRouter.Get("/:id/episodes", controllers.GetAnimeEpisodes)
+	animeRouter.Get("/:id/episodes/:episodeId", controllers.GetAnimeEpisode)
 	animeRouter.Get("/:id/characters", controllers.GetAnimeCharacters)
 
 	// 404 Default
diff --git a/services/anime/service.go b/services/anime/service.go
index f4a48e5..34a32df 100644
--- a/services/anime/service.go
+++ b/services/anime/service.go
@@ -463,6 +463,89 @@ func (s *Service) GetAnimeDetails(mapping *entities.AnimeMapping) (*types.Anime,
 	return s.GetAnimeDetailsWithSource(mapping, "api")
 }
 
+// GetEpisodeStreaming fetches streaming sources for a specific episode
+func (s *Service) GetEpisodeStreaming(title string, episodeNumber int, episodeID string, animeID uint) (*types.AnimeStreaming, error) {
+	// Try to get from database first
+	cached, err := database.GetEpisodeStreaming(episodeID, animeID)
+	if err == nil && cached != nil {
+		logger.Log(fmt.Sprintf("Using cached streaming data for episode %d", episodeNumber), logger.LogOptions{
+			Level:  logger.Debug,
+			Prefix: "AnimeService",
+		})
+
+		result := &types.AnimeStreaming{
+			Sub: make([]types.AnimeStreamingSource, len(cached.SubSources)),
+			Dub: make([]types.AnimeStreamingSource, len(cached.DubSources)),
+		}
+
+		for i, source := range cached.SubSources {
+			result.Sub[i] = types.AnimeStreamingSource{
+				URL:    source.URL,
+				Server: source.Server,
+				Type:   source.Type,
+			}
+		}
+
+		for i, source := range cached.DubSources {
+			result.Dub[i] = types.AnimeStreamingSource{
+				URL:    source.URL,
+				Server: source.Server,
+				Type:   source.Type,
+			}
+		}
+
+		return result, nil
+	}
+
+	// If not in cache or stale, fetch from API
+	logger.Log(fmt.Sprintf("Fetching fresh streaming data for episode %d from API", episodeNumber), logger.LogOptions{
+		Level:  logger.Debug,
+		Prefix: "AnimeService",
+	})
+
+	streaming, err := s.streamingClient.GetStreamingSources(title, episodeNumber)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get streaming sources: %w", err)
+	}
+
+	// Convert streaming API type to types package type
+	result := &types.AnimeStreaming{
+		Sub: make([]types.AnimeStreamingSource, len(streaming.Sub)),
+		Dub: make([]types.AnimeStreamingSource, len(streaming.Dub)),
+	}
+
+	for i, source := range streaming.Sub {
+		result.Sub[i] = types.AnimeStreamingSource{
+			URL:    source.URL,
+			Server: source.Server,
+			Type:   source.Type,
+		}
+	}
+
+	for i, source := range streaming.Dub {
+		result.Dub[i] = types.AnimeStreamingSource{
+			URL:    source.URL,
+			Server: source.Server,
+			Type:   source.Type,
+		}
+	}
+
+	// Save to database for future requests
+	if err := database.SaveEpisodeStreaming(episodeID, animeID, result.Sub, result.Dub); err != nil {
+		logger.Log(fmt.Sprintf("Failed to cache streaming data: %v", err), logger.LogOptions{
+			Level:  logger.Warn,
+			Prefix: "AnimeService",
+		})
+	} else {
+		logger.Log(fmt.Sprintf("Cached streaming data for episode %d", episodeNumber), logger.LogOptions{
+			Level:  logger.Debug,
+			Prefix: "AnimeService",
+		})
+	}
+
+	return result, nil
+}
+
 // getSeasonDetails fetches details for anime seasons
 func (s *Service) getSeasonDetails(mappings *[]entities.AnimeMapping, currentMALID int) []types.AnimeSeason {
 	// Helper function to fetch anime details for a single mapping
diff --git a/types/anime.go b/types/anime.go
index b5bda0b..0f79ef9 100644
--- a/types/anime.go
+++ b/types/anime.go
@@ -35,18 +35,32 @@ type EpisodeTitles struct {
 	Romaji   string `json:"romaji"`
 }
 
+// AnimeStreamingSource represents a single streaming source
+type AnimeStreamingSource struct {
+	URL    string `json:"url"`
+	Server string `json:"server"`
+	Type   string `json:"type"` // direct or embed
+}
+
+// AnimeStreaming represents all available streaming sources
+type AnimeStreaming struct {
+	Sub []AnimeStreamingSource `json:"sub"`
+	Dub []AnimeStreamingSource `json:"dub"`
+}
+
 // AnimeSingleEpisode contains information about a single anime episode
 type AnimeSingleEpisode struct {
-	ID           string        `json:"id"`
-	Titles       EpisodeTitles `json:"titles"`
-	Description  string        `json:"description"`
-	Aired        string        `json:"aired"`
-	Score        float64       `json:"score"`
-	Filler       bool          `json:"filler"`
-	Recap        bool          `json:"recap"`
-	ForumURL     string        `json:"forum_url"`
-	URL          string        `json:"url"`
-	ThumbnailURL string        `json:"thumbnail_url"`
+	ID           string          `json:"id"`
+	Titles       EpisodeTitles   `json:"titles"`
+	Description  string          `json:"description"`
+	Aired        string          `json:"aired"`
+	Score        float64         `json:"score"`
+	Filler       bool            `json:"filler"`
+	Recap        bool            `json:"recap"`
+	ForumURL     string          `json:"forum_url"`
+	URL          string          `json:"url"`
+	ThumbnailURL string          `json:"thumbnail_url"`
+	Streaming    *AnimeStreaming `json:"streaming,omitempty"`
 }
 
 // AnimeEpisodes contains information about all episodes of an anime
diff --git a/utils/api/streaming/streaming.go b/utils/api/streaming/streaming.go
index 4d0f625..cf55a70 100644
--- a/utils/api/streaming/streaming.go
+++ b/utils/api/streaming/streaming.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"maps"
+	"metachan/utils/logger"
 	"metachan/utils/mappers"
 	"net/http"
 	"net/url"
@@ -169,7 +170,7 @@ func (c *AllAnimeClient) processSourceURL(sourceURL, sourceType string) *AnimeSt
 	}
 
 	// Check if it's a direct stream link
-	directPatterns := []string{"fast4speed.rsvp", "sharepoint.com", ".m3u8", ".mp4"}
+	directPatterns := []string{"sharepoint.com", ".m3u8", ".mp4"}
 	for _, pattern := range directPatterns {
 		if strings.Contains(processedURL, pattern) {
 			return &AnimeStreamingSource{
@@ -191,12 +192,12 @@ func (c *AllAnimeClient) processSourceURL(sourceURL, sourceType string) *AnimeSt
 // getServerName maps AllAnime source types to readable server names
 func getServerName(sourceType string) string {
 	switch strings.ToLower(sourceType) {
-	case "default":
+	case "s-mp4":
 		return "Maria"
 	case "luf-mp4":
-		return "Rose"
-	case "s-mp4":
 		return "Sina"
+	case "default":
+		return "Rose"
 	default:
 		return sourceType
 	}
@@ -417,6 +418,12 @@ func (c *AllAnimeClient) GetEpisodeLinks(showID, episode, mode string) ([]AnimeS
 
 			// Only add direct sources
 			if sourceInfo.Type == "direct" {
+				// Transform type to M3U8 or MP4 based on URL
+				if strings.HasSuffix(sourceInfo.URL, ".m3u8") {
+					sourceInfo.Type = "M3U8"
+				} else {
+					sourceInfo.Type = "MP4"
+				}
 				links = append(links, *sourceInfo)
 			}
 		}
@@ -427,18 +434,35 @@ func (c *AllAnimeClient) GetEpisodeLinks(showID, episode, mode string) ([]AnimeS
 
 // GetStreamingSources fetches both sub and dub streaming sources for an anime episode
 func (c *AllAnimeClient) GetStreamingSources(title string, episodeNumber int) (*AnimeStreaming, error) {
+	logger.Log(fmt.Sprintf("Fetching streaming sources for '%s' episode %d", title, episodeNumber), logger.LogOptions{
+		Level:  logger.Debug,
+		Prefix: "Streaming",
+	})
+
 	// Search for the anime
 	searchResults, err := c.SearchAnime(title)
 	if err != nil {
+		logger.Log(fmt.Sprintf("Failed to search anime '%s': %v", title, err), logger.LogOptions{
+			Level:  logger.Error,
+			Prefix: "Streaming",
+		})
 		return nil, fmt.Errorf("failed to search for anime: %w", err)
 	}
 
 	if len(searchResults) == 0 {
+		logger.Log(fmt.Sprintf("No streaming sources found for '%s'", title), logger.LogOptions{
+			Level:  logger.Warn,
+			Prefix: "Streaming",
+		})
 		return nil, fmt.Errorf("no streaming sources found for '%s'", title)
 	}
 
 	// Use the best match (first result)
 	bestMatch := searchResults[0]
+	logger.Log(fmt.Sprintf("Best match: '%s' (ID: %s, Sub: %d, Dub: %d)", bestMatch.Name, bestMatch.ID, bestMatch.SubEpisodes, bestMatch.DubEpisodes), logger.LogOptions{
+		Level:  logger.Debug,
+		Prefix: "Streaming",
+	})
 
 	streaming := &AnimeStreaming{
 		Sub: []AnimeStreamingSource{},
@@ -464,6 +488,15 @@ func (c *AllAnimeClient) GetStreamingSources(title string, episodeNumber int) (*
 				subSources, err := c.GetEpisodeLinks(bestMatch.ID, closestEpisode, "sub")
 				if err == nil {
 					streaming.Sub = subSources
+					logger.Log(fmt.Sprintf("Found %d sub sources for episode %d", len(subSources), episodeNumber), logger.LogOptions{
+						Level:  logger.Debug,
+						Prefix: "Streaming",
+					})
+				} else {
+					logger.Log(fmt.Sprintf("Failed to get sub sources: %v", err), logger.LogOptions{
+						Level:  logger.Warn,
+						Prefix: "Streaming",
+					})
 				}
 			}
 		}
@@ -488,11 +521,24 @@ func (c *AllAnimeClient) GetStreamingSources(title string, episodeNumber int) (*
 				dubSources, err := c.GetEpisodeLinks(bestMatch.ID, closestEpisode, "dub")
 				if err == nil {
 					streaming.Dub = dubSources
+					logger.Log(fmt.Sprintf("Found %d dub sources for episode %d", len(dubSources), episodeNumber), logger.LogOptions{
+						Level:  logger.Debug,
+						Prefix: "Streaming",
+					})
+				} else {
+					logger.Log(fmt.Sprintf("Failed to get dub sources: %v", err), logger.LogOptions{
+						Level:  logger.Warn,
+						Prefix: "Streaming",
+					})
 				}
 			}
 		}
 	}
 
+	logger.Log(fmt.Sprintf("Successfully fetched streaming sources for episode %d (Sub: %d, Dub: %d)", episodeNumber, len(streaming.Sub), len(streaming.Dub)), logger.LogOptions{
+		Level:  logger.Info,
+		Prefix: "Streaming",
+	})
 	return streaming, nil
 }