zurg/pkg/http/client.go

package http

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"math/rand"
	"net"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/debridmediamanager/zurg/internal/config"
	"github.com/debridmediamanager/zurg/pkg/hosts"
	"github.com/debridmediamanager/zurg/pkg/logutil"
	http_dialer "github.com/mwitkow/go-http-dialer"
	"golang.org/x/net/proxy"

	cmap "github.com/orcaman/concurrent-map/v2"
)

type HTTPClient struct {
	client      *http.Client
	maxRetries  int
	timeoutSecs int
	backoff     func(attempt int) time.Duration
	bearerToken string
	supportIPv6 bool
	cfg         config.ConfigInterface
	ipv6        cmap.ConcurrentMap[string, string]
	ipv6Hosts   []string
	log         *logutil.Logger
}

// {
// "error": "infringing_file",
// "error_code": 35
// }

type ApiErrorResponse struct {
	Message string `json:"error"`
	Code    int    `json:"error_code"`
}

func (e *ApiErrorResponse) Error() string {
	return fmt.Sprintf("api response error: %s (code: %d)", e.Message, e.Code)
}

func NewHTTPClient(
	token string,
	maxRetries int,
	timeoutSecs int,
	supportIPv6 bool,
	cfg config.ConfigInterface,
	log *logutil.Logger,
) *HTTPClient {
	client := HTTPClient{
		bearerToken: token,
		client:      http.DefaultClient,
		maxRetries:  maxRetries,
		timeoutSecs: timeoutSecs,
		backoff:     backoffFunc,
		supportIPv6: supportIPv6,
		cfg:         cfg,
		ipv6:        cmap.New[string](),
		log:         log,
	}

	var dialer proxy.Dialer = &net.Dialer{
		Timeout: time.Duration(timeoutSecs) * time.Second, // timeout for dns resolution, tcp handshake
	}

	if proxyURLString := cfg.GetProxy(); proxyURLString != "" {
		proxyURL, err := url.Parse(proxyURLString)
		if err != nil {
			log.Errorf("Failed to parse proxy URL: %v", err)
			return nil
		}
		dialer, err = client.proxyDialer(proxyURL)
		if err != nil {
			log.Errorf("Failed to create proxy dialer: %v", err)
			return nil
		}
	}

	maxConnections := cfg.GetNumOfWorkers()
	if maxConnections > 32 {
		maxConnections = 32 // real-debrid has a limit of 32 connections per server/host
	}

	client.client.Transport = &http.Transport{
		ResponseHeaderTimeout: time.Duration(timeoutSecs) * time.Second,
		MaxIdleConns:          0,
		MaxConnsPerHost:       maxConnections,
		DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
			return dialer.Dial(network, address)
		},
	}

	if cfg.ShouldForceIPv6() {
		// fetch IPv6 hosts
		ipv6List, err := hosts.FetchHosts(hosts.IPV6)
		if err != nil {
			log.Warnf("Failed to fetch IPv6 hosts: %v", err)
			// Decide if you want to return nil here or continue without IPv6
		} else {
			client.ipv6Hosts = ipv6List
			log.Debugf("Fetched %d IPv6 hosts", len(ipv6List))
		}

		// replace the default dialer with a custom one that resolves hostnames to IPv6 addresses
		client.client.Transport.(*http.Transport).DialContext = func(ctx context.Context, network, address string) (net.Conn, error) {
			// if address is already cached, use it
			if ipv6Address, ok := client.ipv6.Get(address); ok {
				return dialer.Dial(network, ipv6Address)
			}

			host, port, err := net.SplitHostPort(address)
			if err != nil {
				return nil, err
			}
			ips, err := net.DefaultResolver.LookupIPAddr(ctx, host)
			if err != nil {
				return nil, err
			}

			for _, ip := range ips {
				if ip.IP.To4() == nil { // IPv6 address found
					ipv6Address := net.JoinHostPort(ip.IP.String(), port)
					client.ipv6.Set(address, ipv6Address)
					return dialer.Dial(network, ipv6Address)
				}
			}

			// no IPv6 address found, use the original address
			log.Warnf("No IPv6 address found for host %s", host)
			for _, ip := range ips {
				if ip.IP.To4() != nil { // IPv4 address found
					ipV4Address := net.JoinHostPort(ip.IP.String(), port)
					client.ipv6.Set(address, ipV4Address)
					return dialer.Dial(network, ipV4Address)
				}
			}

			return dialer.Dial(network, address)
		}

	}

	return &client
}

func (r *HTTPClient) Do(req *http.Request) (*http.Response, error) {
	if r.bearerToken != "" {
		req.Header.Set("Authorization", "Bearer "+r.bearerToken)
	}

	var resp *http.Response
	var err error

	attempt := 0
	var origBody []byte
	if req.Method == "POST" || req.Method == "PUT" || req.Method == "PATCH" {
		origBody, _ = io.ReadAll(req.Body)
	}

	for {
		if origBody != nil {
			req.Body = io.NopCloser(bytes.NewReader(origBody))
		}
		if resp != nil && resp.Body != nil {
			resp.Body.Close()
		}

		r.replaceWithIPv6Host(req) // needed for ipv6

		resp, err = r.client.Do(req)

		// http 4xx and 5xx errors
		if resp != nil && resp.StatusCode/100 >= 4 {
			body, _ := io.ReadAll(resp.Body)
			if body != nil {
				var errResp ApiErrorResponse
				jsonErr := json.Unmarshal(body, &errResp)
				if jsonErr == nil {
					errResp.Message += fmt.Sprintf(" (status code: %d)", resp.StatusCode)
					err = &errResp
				}
			}
		}

		incr := r.shouldRetry(req, resp, err, r.cfg.GetRateLimitSleepSecs())
		if incr > 0 {
			attempt += incr
			if attempt > r.maxRetries {
				r.log.Warnf("Request to %s failed after %d attempts", req.URL.String(), r.maxRetries)
				break
			}
			time.Sleep(r.backoff(attempt))
		} else if incr == 0 {
			time.Sleep(10 * time.Millisecond)
		} else {
			// don't retry anymore
			break
		}
	}
	okWithTimeout := strings.HasSuffix(req.URL.Path, "unrestrict/link") && strings.HasSuffix(req.URL.Path, "torrents/addMagnet")
	if err != nil && strings.Contains(err.Error(), "timeout") && req.Host == "api.real-debrid.com" && !okWithTimeout {
		r.log.Warnf("Adjust your API timeout settings, request to %s timed out", req.URL.String())
	}
	return resp, err
}

func (r *HTTPClient) replaceWithIPv6Host(req *http.Request) {
	// don't replace host if IPv6 is not supported or not forced
	if !r.supportIPv6 || !r.cfg.ShouldForceIPv6() {
		return
	}
	// this host should be replaced
	if !strings.HasSuffix(req.Host, ".download.real-debrid.com") {
		return
	}

	req.Host = strings.Replace(req.Host, ".com", ".cloud", 1)
	req.URL.Host = req.Host
	// check if this host is in the list of IPv6 hosts
	for _, h := range r.ipv6Hosts {
		if h == req.Host {
			return
		}
	}

	// if not then just assign a random IPv6 host
	req.Host = r.ipv6Hosts[rand.Intn(len(r.ipv6Hosts))]
	req.URL.Host = req.Host
	r.log.Debugf("Host %s is not a valid IPv6 host, assigning a random IPv6 host: %s", req.URL.Host, req.Host)
	// if !found && !r.CanFetchFirstByte(req.URL.String()) {
}

func (r *HTTPClient) proxyDialer(proxyURL *url.URL) (proxy.Dialer, error) {
	if proxyURL.Scheme == "http" || proxyURL.Scheme == "https" {
		httpProxyDialer := http_dialer.New(proxyURL, http_dialer.WithConnectionTimeout(time.Duration(r.timeoutSecs)*time.Second))
		return httpProxyDialer, nil
	} else if proxyURL.Scheme == "socks5" {
		return proxy.FromURL(proxyURL, proxy.Direct)
	}
	return nil, fmt.Errorf("unsupported proxy scheme: %s", proxyURL.Scheme)
}

func (r *HTTPClient) shouldRetry(req *http.Request, resp *http.Response, err error, rateLimitSleep int) int {
	if strings.HasSuffix(req.URL.Path, "torrents/addMagnet") {
		return -1 // don't retry to prevent duplicate torrents
	}
	if err != nil && strings.HasPrefix(err.Error(), "api response error:") {
		if apiErr, ok := err.(*ApiErrorResponse); ok {
			switch apiErr.Code {
			case -1: // Internal error
				return 1
			case 5: // Slow down (retry infinitely), default: 4 secs
				time.Sleep(time.Duration(rateLimitSleep) * time.Second)
				return 0
			case 6: // Ressource unreachable
				return 1
			case 17: // Hoster in maintenance
				return 1
			case 18: // Hoster limit reached
				return 1
			case 19: // Hoster temporarily unavailable
				return 1
			case 25: // Service unavailable
				return 1
			case 34: // Too many requests (retry infinitely), default: 4 secs
				time.Sleep(time.Duration(rateLimitSleep) * time.Second)
				return 0
			case 36: // Fair Usage Limit
				return 1
			default:
				return -1 // don't retry
			}
		}
	}
	if err != nil && strings.Contains(err.Error(), "timeout") {
		return 1
	}
	if resp != nil {
		if resp.StatusCode == 429 {
			// Too many requests: retry infinitely, default: 4 secs
			time.Sleep(time.Duration(rateLimitSleep) * time.Second)
			return 0
		}
		if resp.StatusCode/100 == 4 {
			// other client errors: retry
			return 1
		}
		if resp.StatusCode/100 == 5 {
			// server errors: don't retry
			return -1
		}
		// if the request has a Range header but the server doesn't respond with a Content-Range header
		hasRangeHeader := req.Header.Get("Range") != "" && !strings.HasPrefix(req.Header.Get("Range"), "bytes=0-")
		if hasRangeHeader && resp.StatusCode/100 == 2 && resp.Header.Get("Content-Range") == "" {
			time.Sleep(10 * time.Millisecond)
			return 0
		}
		return -1 // don't retry
	}
	return 1
}

func backoffFunc(attempt int) time.Duration {
	maxDuration := 60
	backoff := int(math.Pow(2, float64(attempt)))
	if backoff > maxDuration {
		backoff = maxDuration
	}
	return time.Duration(backoff) * time.Second
}

func (r *HTTPClient) CanFetchFirstByte(url string) bool {
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return false
	}
	req.Header.Set("Range", "bytes=0-0")
	timeout := time.Duration(r.timeoutSecs) * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	req = req.WithContext(ctx)

	resp, err := r.Do(req)
	if err != nil {
		return false
	}
	defer resp.Body.Close()

	// If server supports partial content
	if resp.StatusCode/100 == 2 {
		buffer := make([]byte, 1)
		_, err = resp.Body.Read(buffer)
		if err == nil {
			return true
		}
	}
	return false
}