zurg/pkg/realdebrid/api.go

package realdebrid

import (
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"

	"github.com/debridmediamanager/zurg/internal/config"
	zurghttp "github.com/debridmediamanager/zurg/pkg/http"
	"github.com/debridmediamanager/zurg/pkg/logutil"
	"github.com/panjf2000/ants/v2"
)

type RealDebrid struct {
	apiClient        *zurghttp.HTTPClient
	unrestrictClient *zurghttp.HTTPClient
	downloadClient   *zurghttp.HTTPClient
	workerPool       *ants.Pool
	cfg              config.ConfigInterface
	log              *logutil.Logger
}

func NewRealDebrid(apiClient, unrestrictClient, downloadClient *zurghttp.HTTPClient, workerPool *ants.Pool, cfg config.ConfigInterface, log *logutil.Logger) *RealDebrid {
	return &RealDebrid{
		apiClient:        apiClient,
		unrestrictClient: unrestrictClient,
		downloadClient:   downloadClient,
		workerPool:       workerPool,
		cfg:              cfg,
		log:              log,
	}
}

// currently unused
func (rd *RealDebrid) UnrestrictCheck(link string) (*Download, error) {
	data := url.Values{}
	data.Set("link", link)
	requestBody := strings.NewReader(data.Encode())

	req, err := http.NewRequest("POST", "https://api.real-debrid.com/rest/1.0/unrestrict/check", requestBody)
	if err != nil {
		rd.log.Errorf("Error when creating a unrestrict check request: %v", err)
		return nil, err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

	resp, err := rd.unrestrictClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the unrestrict check request: %v", err)
		return nil, err
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		rd.log.Errorf("Error when reading the body of unrestrict check response: %v", err)
		return nil, err
	}

	var response Download
	err = json.Unmarshal(body, &response)
	if err != nil {
		rd.log.Errorf("Error when decoding unrestrict check JSON: %v", err)
		return nil, err
	}

	rd.log.Debugf("Link %s is streamable? %v", response.Streamable)
	return &response, nil
}

func (rd *RealDebrid) UnrestrictLink(link string, checkFirstByte bool) (*Download, error) {
	data := url.Values{}
	if strings.HasPrefix(link, "https://real-debrid.com/d/") {
		// set link to max 39 chars
		link = link[0:39]
	}
	data.Set("link", link)
	requestBody := strings.NewReader(data.Encode())

	req, err := http.NewRequest("POST", "https://api.real-debrid.com/rest/1.0/unrestrict/link", requestBody)
	if err != nil {
		rd.log.Errorf("Error when creating a unrestrict link request: %v", err)
		return nil, err
	}

	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

	// at this point, any errors mean that the link has expired and we need to repair it
	resp, err := rd.unrestrictClient.Do(req)
	if err != nil {
		// rd.log.Errorf("Error when executing the unrestrict link request: %v", err)
		return nil, fmt.Errorf("unrestrict link request failed: %v", err)
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		// rd.log.Errorf("Error when reading the body of unrestrict link response: %v", err)
		return nil, fmt.Errorf("unreadable body: %v", err)
	}

	var response Download
	err = json.Unmarshal(body, &response)
	if err != nil {
		// rd.log.Errorf("Error when decoding unrestrict link JSON: %v", err)
		return nil, fmt.Errorf("undecodable response: %v", err)
	}

	// will only check for first byte if serving from rclone
	if checkFirstByte && !rd.downloadClient.CanFetchFirstByte(response.Download) {
		return nil, fmt.Errorf("can't fetch first byte")
	}

	// rd.log.Debugf("Unrestricted link %s into %s", link, response.Download)
	return &response, nil
}

type getTorrentsResult struct {
	torrents   []Torrent
	err        error
	totalCount int
}

func (rd *RealDebrid) getPageOfTorrents(page, limit int) getTorrentsResult {
	baseURL := "https://api.real-debrid.com/rest/1.0/torrents"

	params := url.Values{}
	params.Set("page", fmt.Sprintf("%d", page))
	params.Set("limit", fmt.Sprintf("%d", limit))

	reqURL := baseURL + "?" + params.Encode()
	req, err := http.NewRequest("GET", reqURL, nil)
	if err != nil {
		return getTorrentsResult{nil, err, 0}
	}

	resp, err := rd.apiClient.Do(req)
	if err != nil {
		return getTorrentsResult{nil, err, 0}
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusNoContent {
		return getTorrentsResult{nil, nil, 0}
	}

	var torrents []Torrent
	decoder := json.NewDecoder(resp.Body)
	err = decoder.Decode(&torrents)
	if err != nil {
		return getTorrentsResult{nil, err, 0}
	}

	countHeader := resp.Header.Get("x-total-count")
	count, _ := strconv.Atoi(countHeader) // In real use, handle this error

	return getTorrentsResult{torrents, nil, count}
}

func (rd *RealDebrid) GetTorrents(onlyOne bool) ([]Torrent, int, error) {
	var allTorrents []Torrent

	// fetch 1 to get total count
	result := rd.getPageOfTorrents(1, 1)
	allTorrents = append(allTorrents, result.torrents...)
	totalCount := result.totalCount

	if onlyOne {
		return allTorrents, totalCount, nil
	}

	// reset allTorrents
	allTorrents = []Torrent{}
	page := 1
	// compute ceiling of totalCount / limit
	maxPages := (totalCount + 250 - 1) / 250
	rd.log.Debugf("Torrents total count is %d", totalCount)
	maxParallelThreads := 4
	if maxPages < maxParallelThreads {
		maxParallelThreads = maxPages
	}
	for {
		allResults := make(chan getTorrentsResult, maxParallelThreads) // Channel to collect results from goroutines
		for i := 0; i < maxParallelThreads; i++ {                      // Launch GET_PARALLEL concurrent fetches
			idx := i
			rd.workerPool.Submit(func() {
				if page > maxPages {
					allResults <- getTorrentsResult{nil, nil, 0}
					return
				}
				allResults <- rd.getPageOfTorrents(page+idx, 250)
			})
		}
		// Collect results from all goroutines
		for i := 0; i < maxParallelThreads; i++ {
			res := <-allResults
			if res.err != nil {
				rd.log.Warnf("Ignoring error when fetching torrents: %v", res.err)
				continue
			}
			allTorrents = append(allTorrents, res.torrents...)
		}

		rd.log.Debugf("Got %d/%d torrents", len(allTorrents), totalCount)

		if len(allTorrents) >= totalCount || page >= maxPages {
			break
		}

		page += maxParallelThreads
	}

	return allTorrents, totalCount, nil
}

func (rd *RealDebrid) GetTorrentInfo(id string) (*TorrentInfo, error) {
	url := "https://api.real-debrid.com/rest/1.0/torrents/info/" + id

	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		rd.log.Errorf("Error when creating a get info request: %v", err)
		return nil, err
	}

	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the get info request: %v", err)
		return nil, err
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		rd.log.Errorf("Error when reading the body of get info response: %v", err)
		return nil, err
	}

	var response TorrentInfo
	err = json.Unmarshal(body, &response)
	if err != nil {
		rd.log.Errorf("Error when : %v", err)
		return nil, err
	}

	// rd.log.Debugf("Got info for torrent %s (progress=%d%%)", id, response.Progress)
	return &response, nil
}

// SelectTorrentFiles selects files of a torrent to start it.
func (rd *RealDebrid) SelectTorrentFiles(id string, files string) error {
	data := url.Values{}
	data.Set("files", files)
	requestBody := strings.NewReader(data.Encode())

	reqURL := fmt.Sprintf("https://api.real-debrid.com/rest/1.0/torrents/selectFiles/%s", id)
	req, err := http.NewRequest("POST", reqURL, requestBody)
	if err != nil {
		rd.log.Errorf("Error when creating a select files request: %v", err)
		return err
	}

	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the select files request: %v", err)
		return err
	}
	defer resp.Body.Close()

	rd.log.Debugf("Selected %d files and started the download for torrent id=%s (status code: %d)", len(strings.Split(files, ",")), id, resp.StatusCode)
	return nil
}

// DeleteTorrent deletes a torrent from the torrents list.
func (rd *RealDebrid) DeleteTorrent(id string) error {
	// Construct request URL
	reqURL := fmt.Sprintf("https://api.real-debrid.com/rest/1.0/torrents/delete/%s", id)
	req, err := http.NewRequest("DELETE", reqURL, nil)
	if err != nil {
		rd.log.Errorf("Error when creating a delete torrent request: %v", err)
		return err
	}

	// Send the request
	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the delete torrent request: %v", err)
		return err
	}
	defer resp.Body.Close()

	rd.log.Debugf("Deleted torrent with id=%s", id)
	return nil
}

// AddMagnetHash adds a magnet link to download.
func (rd *RealDebrid) AddMagnetHash(magnet string) (*MagnetResponse, error) {
	// Prepare request data
	data := url.Values{}
	data.Set("magnet", fmt.Sprintf("magnet:?xt=urn:btih:%s", magnet))
	requestBody := strings.NewReader(data.Encode())

	// Construct request URL
	reqURL := "https://api.real-debrid.com/rest/1.0/torrents/addMagnet"
	req, err := http.NewRequest("POST", reqURL, requestBody)
	if err != nil {
		rd.log.Errorf("Error when creating an add magnet request: %v", err)
		return nil, err
	}

	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

	// Send the request
	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the add magnet request: %v", err)
		return nil, err
	}
	defer resp.Body.Close()

	var response MagnetResponse
	err = json.NewDecoder(resp.Body).Decode(&response)
	if err != nil {
		rd.log.Errorf("Error when decoding add magnet JSON: %v", err)
		return nil, err
	}

	rd.log.Debugf("Added magnet %s with id=%s", magnet, response.ID)
	return &response, nil
}

// GetActiveTorrentCount gets the number of currently active torrents and the current maximum limit.
func (rd *RealDebrid) GetActiveTorrentCount() (*ActiveTorrentCountResponse, error) {
	// Construct request URL
	reqURL := "https://api.real-debrid.com/rest/1.0/torrents/activeCount"
	req, err := http.NewRequest("GET", reqURL, nil)
	if err != nil {
		rd.log.Errorf("Error when creating a active torrents request: %v", err)
		return nil, err
	}

	// Send the request
	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the active torrents request: %v", err)
		return nil, err
	}
	defer resp.Body.Close()

	var response ActiveTorrentCountResponse
	err = json.NewDecoder(resp.Body).Decode(&response)
	if err != nil {
		rd.log.Errorf("Error when decoding active torrents JSON: %v", err)
		return nil, err
	}
	return &response, nil
}

// GetDownloads returns all torrents, paginated
func (rd *RealDebrid) GetDownloads() []Download {
	_, totalCount, err := rd.fetchPageOfDownloads(1, 1)
	if err != nil {
		return nil
	}

	// reset allDownloads
	allDownloads := []Download{}
	page := 1
	limit := 250

	// compute ceiling of totalCount / limit
	maxPages := (totalCount + limit - 1) / limit
	rd.log.Debugf("Total downloads count is %d", totalCount)
	maxParallelThreads := 4
	if maxPages < maxParallelThreads {
		maxParallelThreads = maxPages
	}
	for {
		allResults := make(chan []Download, maxParallelThreads) // Channel to collect results from goroutines
		errChan := make(chan error, maxParallelThreads)         // Channel to collect errors from goroutines
		for i := 0; i < maxParallelThreads; i++ {               // Launch GET_PARALLEL concurrent fetches
			idx := i
			rd.workerPool.Submit(func() {
				if page+idx > maxPages {
					allResults <- nil
					errChan <- nil
					return
				}
				result, _, err := rd.fetchPageOfDownloads(page+idx, limit)
				if err != nil {
					allResults <- nil
					errChan <- err
					return
				}
				allResults <- result
				errChan <- nil
			})
		}
		// Collect results from all goroutines
		for i := 0; i < maxParallelThreads; i++ {
			res := <-allResults
			err := <-errChan
			if err != nil {
				rd.log.Warnf("Ignoring error when fetching downloads: %v", err)
				continue
			}
			allDownloads = append(allDownloads, res...)
		}

		rd.log.Debugf("Got %d/%d downloads", len(allDownloads), totalCount)

		if len(allDownloads) >= totalCount || page >= maxPages {
			break
		}

		page += maxParallelThreads
	}

	return allDownloads
}

func (rd *RealDebrid) fetchPageOfDownloads(page, limit int) ([]Download, int, error) {
	baseURL := "https://api.real-debrid.com/rest/1.0/downloads"
	var downloads []Download
	totalCount := 0

	params := url.Values{}
	params.Set("page", fmt.Sprintf("%d", page))
	params.Set("limit", fmt.Sprintf("%d", limit))
	// params.Set("filter", "active")

	reqURL := baseURL + "?" + params.Encode()

	req, err := http.NewRequest("GET", reqURL, nil)
	if err != nil {
		rd.log.Errorf("Error when creating a get downloads request: %v", err)
		return nil, 0, err
	}

	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the get downloads request: %v", err)
		return nil, 0, err
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusNoContent {
		return downloads, 0, nil
	}

	if resp.StatusCode != http.StatusOK {
		err := fmt.Errorf("unexpected status code: %d", resp.StatusCode)
		rd.log.Errorf("Error when executing the get downloads request: %v", err)
		return nil, 0, err
	}

	decoder := json.NewDecoder(resp.Body)
	err = decoder.Decode(&downloads)
	if err != nil {
		rd.log.Errorf("Error when decoding get downloads JSON: %v", err)
		return nil, 0, err
	}

	totalCountHeader := resp.Header.Get("x-total-count")
	totalCount, err = strconv.Atoi(totalCountHeader)
	if err != nil {
		totalCount = 0
	}

	return downloads, totalCount, nil
}

// GetUserInformation gets the current user information.
func (rd *RealDebrid) GetUserInformation() (*User, error) {
	// Construct request URL
	reqURL := "https://api.real-debrid.com/rest/1.0/user"
	req, err := http.NewRequest("GET", reqURL, nil)
	if err != nil {
		rd.log.Errorf("Error when creating a user information request: %v", err)
		return nil, err
	}

	// Send the request
	resp, err := rd.apiClient.Do(req)
	if err != nil {
		rd.log.Errorf("Error when executing the user information request: %v", err)
		return nil, err
	}
	defer resp.Body.Close()

	// Decode the JSON response into the User struct
	var user User
	err = json.NewDecoder(resp.Body).Decode(&user)
	if err != nil {
		rd.log.Errorf("Error when decoding user information JSON: %v", err)
		return nil, err
	}

	return &user, nil
}

// AvailabilityCheck checks the instant availability of torrents
func (rd *RealDebrid) AvailabilityCheck(hashes []string) (AvailabilityResponse, error) {
	if len(hashes) == 0 {
		return nil, fmt.Errorf("no hashes provided")
	}

	baseURL := "https://api.real-debrid.com/rest/1.0"
	url := fmt.Sprintf("%s/torrents/instantAvailability/%s", baseURL, strings.Join(hashes, "/"))
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return nil, err
	}

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

	var response AvailabilityResponse
	err = json.NewDecoder(resp.Body).Decode(&response)
	if err != nil {
		return nil, err
	}

	return response, nil
}