melosh/mal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden watch playback flow

Browse Source
This commit is contained in:
mkelvers
2026-04-19 02:10:20 +02:00
parent 7c333da8e5
commit afba550da3
3 changed files with 213 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
internal/features/playback/handler.go
View File

@@ -88,9 +88,9 @@ func (h *Handler) HandleWatchPage(w http.ResponseWriter, r *http.Request) {
}
}
title := anime.DisplayTitle()
titleCandidates := playbackTitleCandidates(anime)
userID := watchlistUserIDFromRequest(r)
data, err := h.svc.BuildWatchPageData(ctx, malID, title, episode, mode, userID)
data, err := h.svc.BuildWatchPageData(ctx, malID, titleCandidates, episode, mode, userID)
if err != nil {
log.Printf("watch page error for mal_id=%d: %v", malID, err)
http.Error(w, "Failed to load playback", http.StatusBadGateway)
@@ -127,6 +127,35 @@ func watchlistUserIDFromRequest(r *http.Request) string {
return user.ID
}
func playbackTitleCandidates(anime jikan.Anime) []string {
out := make([]string, 0, 3+len(anime.TitleSynonyms))
seen := make(map[string]struct{})
add := func(value string) {
normalized := strings.TrimSpace(value)
if normalized == "" {
return
}
key := strings.ToLower(normalized)
if _, exists := seen[key]; exists {
return
}
seen[key] = struct{}{}
out = append(out, normalized)
}
add(anime.Title)
add(anime.TitleEnglish)
add(anime.TitleJapanese)
for _, synonym := range anime.TitleSynonyms {
add(synonym)
}
return out
}
func convertModeSources(sources map[string]ModeSource) map[string]templates.ModeSource {
result := make(map[string]templates.ModeSource, len(sources))
for k, v := range sources {
@@ -354,42 +383,55 @@ func (h *Handler) HandleCompleteAnime(w http.ResponseWriter, r *http.Request) {
}
animeID := int64(payload.MalID)
if _, err := h.svc.db.GetAnime(r.Context(), animeID); err != nil {
anime, fetchErr := h.jikanClient.GetAnimeByID(r.Context(), payload.MalID)
if fetchErr != nil {
log.Printf("complete anime failed to fetch anime user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, fetchErr)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
if _, upsertErr := h.svc.db.UpsertAnime(r.Context(), database.UpsertAnimeParams{
ID: animeID,
TitleOriginal: anime.Title,
TitleEnglish: sql.NullString{String: anime.TitleEnglish, Valid: anime.TitleEnglish != ""},
TitleJapanese: sql.NullString{String: anime.TitleJapanese, Valid: anime.TitleJapanese != ""},
ImageUrl: anime.ImageURL(),
Airing: sql.NullBool{Bool: anime.Airing, Valid: true},
}); upsertErr != nil {
log.Printf("complete anime failed to upsert anime user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, upsertErr)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
}
if _, err := h.svc.db.UpsertWatchListEntry(r.Context(), database.UpsertWatchListEntryParams{
ID: uuid.New().String(),
UserID: user.ID,
AnimeID: animeID,
Status: "completed",
CurrentEpisode: sql.NullInt64{Int64: int64(payload.Episode), Valid: true},
CurrentTimeSeconds: 0,
}); err != nil {
log.Printf("complete anime failed to upsert watchlist user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, err)
watchListEntry, watchListErr := h.svc.db.GetWatchListEntry(r.Context(), database.GetWatchListEntryParams{
UserID: user.ID,
AnimeID: animeID,
})
if watchListErr != nil && !errors.Is(watchListErr, sql.ErrNoRows) {
log.Printf("complete anime failed to load watchlist user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, watchListErr)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
alreadyCompleted := watchListErr == nil && watchListEntry.Status == "completed"
if !alreadyCompleted {
if _, err := h.svc.db.GetAnime(r.Context(), animeID); err != nil {
anime, fetchErr := h.jikanClient.GetAnimeByID(r.Context(), payload.MalID)
if fetchErr != nil {
log.Printf("complete anime failed to fetch anime user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, fetchErr)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
if _, upsertErr := h.svc.db.UpsertAnime(r.Context(), database.UpsertAnimeParams{
ID: animeID,
TitleOriginal: anime.Title,
TitleEnglish: sql.NullString{String: anime.TitleEnglish, Valid: anime.TitleEnglish != ""},
TitleJapanese: sql.NullString{String: anime.TitleJapanese, Valid: anime.TitleJapanese != ""},
ImageUrl: anime.ImageURL(),
Airing: sql.NullBool{Bool: anime.Airing, Valid: true},
}); upsertErr != nil {
log.Printf("complete anime failed to upsert anime user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, upsertErr)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
}
if _, err := h.svc.db.UpsertWatchListEntry(r.Context(), database.UpsertWatchListEntryParams{
ID: uuid.New().String(),
UserID: user.ID,
AnimeID: animeID,
Status: "completed",
CurrentEpisode: sql.NullInt64{Int64: int64(payload.Episode), Valid: true},
CurrentTimeSeconds: 0,
}); err != nil {
log.Printf("complete anime failed to upsert watchlist user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, err)
http.Error(w, "failed to mark anime completed", http.StatusInternalServerError)
return
}
}
if err := h.svc.db.DeleteContinueWatchingEntry(r.Context(), database.DeleteContinueWatchingEntryParams{
UserID: user.ID,
AnimeID: animeID,
@@ -418,7 +460,7 @@ func (h *Handler) HandleCompleteAnime(w http.ResponseWriter, r *http.Request) {
UserID: user.ID,
AnimeID: animeID,
}); err != nil {
if err.Error() != "sql: no rows in result set" {
if !errors.Is(err, sql.ErrNoRows) {
log.Printf("complete anime failed to reset watchlist progress user_id=%s mal_id=%d err=%v", user.ID, payload.MalID, err)
}
}
@@ -438,6 +480,10 @@ func (h *Handler) proxyUpstream(w http.ResponseWriter, r *http.Request, targetUR
statusCode, headers, rewrittenBody, streamBody, err := h.svc.ProxyStream(ctx, targetURL, referer, r.Header.Get("Range"))
if err != nil {
if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) || errors.Is(r.Context().Err(), context.Canceled) {
return
}
log.Printf("proxy error for url=%s: %v", targetURL, err)
http.Error(w, "upstream request failed", http.StatusBadGateway)
return

129
internal/features/playback/service.go
View File

@@ -93,7 +93,7 @@ func NewService(db database.Querier) *Service {
}
}
func (s *Service) BuildWatchPageData(ctx context.Context, malID int, title string, episode string, mode string, userID string) (WatchPageData, error) {
func (s *Service) BuildWatchPageData(ctx context.Context, malID int, titleCandidates []string, episode string, mode string, userID string) (WatchPageData, error) {
if malID <= 0 {
return WatchPageData{}, errors.New("invalid mal id")
}
@@ -113,7 +113,7 @@ func (s *Service) BuildWatchPageData(ctx context.Context, malID int, title strin
cacheKey := playbackDataCacheKey(malID, normalizedEpisode)
baseData, cacheHit := s.getPlaybackBaseDataCache(cacheKey)
if !cacheHit {
showID, resolvedTitle, err := s.resolveShowCached(ctx, malID, title)
showID, resolvedTitle, err := s.resolveShowCached(ctx, malID, titleCandidates)
if err != nil {
return WatchPageData{}, err
}
@@ -125,7 +125,7 @@ func (s *Service) BuildWatchPageData(ctx context.Context, malID int, title strin
watchTitle := strings.TrimSpace(resolvedTitle)
if watchTitle == "" {
watchTitle = strings.TrimSpace(title)
watchTitle = firstNonEmptyTitle(titleCandidates)
}
if watchTitle == "" {
watchTitle = fmt.Sprintf("MAL #%d", malID)
@@ -233,7 +233,7 @@ func (s *Service) setPlaybackBaseDataCache(key string, data playbackBaseData) {
s.cacheMu.Unlock()
}
func (s *Service) resolveShowCached(ctx context.Context, malID int, title string) (string, string, error) {
func (s *Service) resolveShowCached(ctx context.Context, malID int, titleCandidates []string) (string, string, error) {
now := time.Now()
s.cacheMu.RLock()
@@ -244,7 +244,7 @@ func (s *Service) resolveShowCached(ctx context.Context, malID int, title string
return item.ShowID, item.Title, nil
}
showID, resolvedTitle, err := s.resolveShow(ctx, malID, title)
showID, resolvedTitle, err := s.resolveShow(ctx, malID, titleCandidates)
if err != nil {
return "", "", err
}
@@ -365,43 +365,22 @@ func cloneSegments(segments []SkipSegment) []SkipSegment {
return cloned
}
func (s *Service) resolveShow(ctx context.Context, malID int, title string) (string, string, error) {
func (s *Service) resolveShow(ctx context.Context, malID int, titleCandidates []string) (string, string, error) {
malText := strconv.Itoa(malID)
modeCandidates := []string{"sub", "dub"}
queries := buildTitleSearchQueries(titleCandidates)
resultsByMode := make(map[string][]searchResult, len(modeCandidates))
searchCh := make(chan searchModeResult, len(modeCandidates))
for _, query := range queries {
resultsByMode := s.searchShowResultsByMode(ctx, query, modeCandidates)
var wg sync.WaitGroup
for _, mode := range modeCandidates {
modeValue := mode
wg.Add(1)
go func() {
defer wg.Done()
results, err := s.allAnimeClient.Search(ctx, title, modeValue)
searchCh <- searchModeResult{Mode: modeValue, Results: results, Err: err}
}()
}
wg.Wait()
close(searchCh)
for result := range searchCh {
if result.Err != nil {
continue
}
resultsByMode[result.Mode] = result.Results
}
for _, mode := range modeCandidates {
for _, result := range resultsByMode[mode] {
if strings.TrimSpace(result.MalID) == malText && strings.TrimSpace(result.ID) != "" {
return result.ID, result.Name, nil
for _, mode := range modeCandidates {
for _, result := range resultsByMode[mode] {
if strings.TrimSpace(result.MalID) == malText && strings.TrimSpace(result.ID) != "" {
return result.ID, result.Name, nil
}
}
}
}
if strings.TrimSpace(title) != "" {
for _, mode := range modeCandidates {
results := resultsByMode[mode]
if len(results) == 0 {
@@ -418,6 +397,86 @@ func (s *Service) resolveShow(ctx context.Context, malID int, title string) (str
return "", "", errors.New("unable to resolve allanime show")
}
func (s *Service) searchShowResultsByMode(ctx context.Context, query string, modeCandidates []string) map[string][]searchResult {
resultsByMode := make(map[string][]searchResult, len(modeCandidates))
searchCh := make(chan searchModeResult, len(modeCandidates))
var wg sync.WaitGroup
for _, mode := range modeCandidates {
modeValue := mode
wg.Add(1)
go func() {
defer wg.Done()
results, err := s.allAnimeClient.Search(ctx, query, modeValue)
searchCh <- searchModeResult{Mode: modeValue, Results: results, Err: err}
}()
}
wg.Wait()
close(searchCh)
for result := range searchCh {
if result.Err != nil {
continue
}
resultsByMode[result.Mode] = result.Results
}
return resultsByMode
}
func buildTitleSearchQueries(titleCandidates []string) []string {
queries := make([]string, 0, len(titleCandidates)*4)
seen := make(map[string]struct{})
add := func(raw string) {
normalized := normalizeSearchQuery(raw)
if normalized == "" {
return
}
key := strings.ToLower(normalized)
if _, exists := seen[key]; exists {
return
}
seen[key] = struct{}{}
queries = append(queries, normalized)
}
for _, candidate := range titleCandidates {
normalized := normalizeSearchQuery(candidate)
if normalized == "" {
continue
}
add(normalized)
add(strings.ReplaceAll(normalized, "+", " "))
withoutApostrophes := strings.NewReplacer("'", "", "’", "", "`", "").Replace(normalized)
add(withoutApostrophes)
add(strings.ReplaceAll(withoutApostrophes, "+", " "))
}
return queries
}
func normalizeSearchQuery(raw string) string {
return strings.Join(strings.Fields(strings.TrimSpace(raw)), " ")
}
func firstNonEmptyTitle(values []string) string {
for _, value := range values {
normalized := strings.TrimSpace(value)
if normalized != "" {
return normalized
}
}
return ""
}
func (s *Service) resolveModeSource(ctx context.Context, showID string, episode string, mode string, quality string) (StreamSource, error) {
sources, err := s.allAnimeClient.GetEpisodeSources(ctx, showID, episode, mode)
if err != nil {