refactor: split playback proxy logic into separate handler files
This commit is contained in:
@@ -2,18 +2,12 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"mal/internal/domain"
|
||||
"mal/internal/observability"
|
||||
"mal/internal/server"
|
||||
netutil "mal/pkg/net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -282,254 +276,3 @@ func (h *PlaybackHandler) HandleEpisodeThumbnails(c *gin.Context) {
|
||||
|
||||
c.JSON(http.StatusOK, results)
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) HandleProxyStream(c *gin.Context) {
|
||||
targetURL, referer, ok := h.resolveProxyRequestTarget(c, "stream")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
req, err := newProxyRequest(c.Request.Context(), targetURL, referer)
|
||||
if err != nil {
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
if rangeHeader := c.GetHeader("Range"); rangeHeader != "" {
|
||||
req.Header.Set("Range", rangeHeader)
|
||||
}
|
||||
if ifRangeHeader := c.GetHeader("If-Range"); ifRangeHeader != "" {
|
||||
req.Header.Set("If-Range", ifRangeHeader)
|
||||
}
|
||||
|
||||
resp, err := h.streamingClient.Do(req)
|
||||
if err != nil {
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
observability.ErrorContext(c.Request.Context(), "proxy_stream_upstream_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
|
||||
_ = c.Error(err).SetType(gin.ErrorTypePrivate)
|
||||
}
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if isHLSPlaylistResponse(targetURL, resp.Header) {
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, netutil.MiB2))
|
||||
if err != nil {
|
||||
observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_read_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
|
||||
_ = c.Error(err).SetType(gin.ErrorTypePrivate)
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
rewritten, err := h.rewriteHLSPlaylist(string(body), targetURL, referer)
|
||||
if err != nil {
|
||||
observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_rewrite_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
|
||||
_ = c.Error(err).SetType(gin.ErrorTypePrivate)
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
copyProxyHeaders(c.Writer.Header(), resp.Header)
|
||||
c.Writer.Header().Del("Content-Length")
|
||||
c.Data(resp.StatusCode, "application/vnd.apple.mpegurl", []byte(rewritten))
|
||||
return
|
||||
}
|
||||
|
||||
copyProxyHeaders(c.Writer.Header(), resp.Header)
|
||||
c.Status(resp.StatusCode)
|
||||
_, _ = io.Copy(c.Writer, resp.Body)
|
||||
}
|
||||
|
||||
func isHLSPlaylistResponse(targetURL string, headers http.Header) bool {
|
||||
contentType := strings.ToLower(headers.Get("Content-Type"))
|
||||
if strings.Contains(contentType, "mpegurl") || strings.Contains(contentType, "x-mpegurl") {
|
||||
return true
|
||||
}
|
||||
|
||||
parsed, err := url.Parse(targetURL)
|
||||
if err != nil {
|
||||
return strings.Contains(strings.ToLower(targetURL), ".m3u8")
|
||||
}
|
||||
return strings.Contains(strings.ToLower(parsed.Path), ".m3u8")
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) rewriteHLSPlaylist(body string, playlistURL string, referer string) (string, error) {
|
||||
baseURL, err := url.Parse(playlistURL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
lines := strings.SplitAfter(body, "\n")
|
||||
var out strings.Builder
|
||||
for _, line := range lines {
|
||||
lineBody := strings.TrimSuffix(line, "\n")
|
||||
newline := ""
|
||||
if strings.HasSuffix(line, "\n") {
|
||||
newline = "\n"
|
||||
lineBody = strings.TrimSuffix(lineBody, "\r")
|
||||
if strings.HasSuffix(line, "\r\n") {
|
||||
newline = "\r\n"
|
||||
}
|
||||
}
|
||||
|
||||
trimmed := strings.TrimSpace(lineBody)
|
||||
rewritten := lineBody
|
||||
if trimmed != "" {
|
||||
if strings.HasPrefix(trimmed, "#") {
|
||||
rewritten, err = h.rewriteHLSQuotedURIs(lineBody, baseURL, referer)
|
||||
} else {
|
||||
rewritten, err = h.proxyPlaylistURI(trimmed, baseURL, referer)
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
out.WriteString(rewritten)
|
||||
out.WriteString(newline)
|
||||
}
|
||||
|
||||
return out.String(), nil
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) rewriteHLSQuotedURIs(line string, baseURL *url.URL, referer string) (string, error) {
|
||||
const marker = `URI="`
|
||||
var out strings.Builder
|
||||
remaining := line
|
||||
for {
|
||||
idx := strings.Index(remaining, marker)
|
||||
if idx < 0 {
|
||||
out.WriteString(remaining)
|
||||
return out.String(), nil
|
||||
}
|
||||
out.WriteString(remaining[:idx+len(marker)])
|
||||
remaining = remaining[idx+len(marker):]
|
||||
end := strings.Index(remaining, `"`)
|
||||
if end < 0 {
|
||||
out.WriteString(remaining)
|
||||
return out.String(), nil
|
||||
}
|
||||
proxied, err := h.proxyPlaylistURI(remaining[:end], baseURL, referer)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
out.WriteString(proxied)
|
||||
remaining = remaining[end:]
|
||||
}
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) proxyPlaylistURI(rawURI string, baseURL *url.URL, referer string) (string, error) {
|
||||
target, err := baseURL.Parse(rawURI)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
token, err := h.svc.SignProxyToken(target.String(), referer, "stream")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
params := url.Values{}
|
||||
params.Set("token", token)
|
||||
return "/watch/proxy/stream?" + params.Encode(), nil
|
||||
}
|
||||
|
||||
func copyProxyHeaders(dst http.Header, src http.Header) {
|
||||
// Skip hop-by-hop headers; see RFC 7230 section 6.1.
|
||||
// We intentionally preserve multi-value headers by copying the full slice.
|
||||
for k, v := range src {
|
||||
switch http.CanonicalHeaderKey(k) {
|
||||
case "Connection", "Keep-Alive", "Proxy-Authenticate", "Proxy-Authorization", "Te", "Trailer", "Transfer-Encoding", "Upgrade":
|
||||
continue
|
||||
}
|
||||
// Copy the slice to avoid sharing memory with src.
|
||||
copied := make([]string, len(v))
|
||||
copy(copied, v)
|
||||
dst[k] = copied
|
||||
}
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) resolveProxyRequestTarget(c *gin.Context, scope string) (string, string, bool) {
|
||||
token := c.Query("token")
|
||||
if token == "" {
|
||||
c.Status(http.StatusBadRequest)
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
targetURL, referer, err := h.svc.ResolveProxyToken(token, scope)
|
||||
if err != nil {
|
||||
c.Status(http.StatusForbidden)
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
return targetURL, referer, true
|
||||
}
|
||||
|
||||
func newProxyRequest(ctx context.Context, targetURL string, referer string) (*http.Request, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, targetURL, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if referer != "" {
|
||||
req.Header.Set("Referer", referer)
|
||||
}
|
||||
req.Header.Set("User-Agent", netutil.Firefox121)
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func (h *PlaybackHandler) HandleProxySubtitle(c *gin.Context) {
|
||||
targetURL, referer, ok := h.resolveProxyRequestTarget(c, "subtitle")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
if data, contentType, ok := h.subtitleCache.Get(targetURL, time.Now()); ok {
|
||||
c.Data(http.StatusOK, contentType, data)
|
||||
return
|
||||
}
|
||||
|
||||
req, err := newProxyRequest(c.Request.Context(), targetURL, referer)
|
||||
if err != nil {
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
resp, err := h.proxyClient.Do(req)
|
||||
if err != nil {
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
observability.ErrorContext(c.Request.Context(), "proxy_subtitle_upstream_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
|
||||
_ = c.Error(err).SetType(gin.ErrorTypePrivate)
|
||||
}
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, netutil.MiB2))
|
||||
if err != nil {
|
||||
observability.ErrorContext(c.Request.Context(), "proxy_subtitle_read_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
|
||||
_ = c.Error(err).SetType(gin.ErrorTypePrivate)
|
||||
c.Status(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
contentType := resp.Header.Get("Content-Type")
|
||||
if contentType == "" {
|
||||
contentType = detectSubtitleType(targetURL)
|
||||
}
|
||||
|
||||
h.subtitleCache.Set(targetURL, body, contentType, time.Now())
|
||||
|
||||
c.Data(http.StatusOK, contentType, body)
|
||||
}
|
||||
|
||||
func detectSubtitleType(url string) string {
|
||||
lower := strings.ToLower(url)
|
||||
switch {
|
||||
case strings.Contains(lower, ".vtt"):
|
||||
return "text/vtt"
|
||||
case strings.Contains(lower, ".srt"):
|
||||
return "text/plain; charset=utf-8"
|
||||
case strings.Contains(lower, ".ass") || strings.Contains(lower, ".ssa"):
|
||||
return "text/plain; charset=utf-8"
|
||||
default:
|
||||
return "text/plain; charset=utf-8"
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user