minstrel/internal/downloader/exchange.go

package downloader

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"path"
	"strings"
	"time"
)

// Exchange drives FabledExchange's native /api/downloads: not its qBittorrent
// facade, so its own states, failure reasons and engine status come through.
type Exchange struct {
	Config Config
	// HTTP is injectable for tests. Nil takes a plain client.
	HTTP *http.Client
}

type exchangeItem struct {
	InfoHash   string    `json:"infohash"`
	Name       string    `json:"name"`
	Bytes      int64     `json:"bytes"`
	Length     int64     `json:"length"`
	Progress   float64   `json:"progress"`
	State      string    `json:"state"`
	Peers      int       `json:"peers"`
	Category   string    `json:"category"`
	Tags       []string  `json:"tags"`
	SavePath   string    `json:"save_path"`
	AddedAt    time.Time `json:"added_at"`
	Failed     bool      `json:"failed"`
	FailReason string    `json:"fail_reason"`
	Ratio      float64   `json:"ratio"`
}

type exchangeSnapshot struct {
	Enabled        bool           `json:"enabled"`
	Running        bool           `json:"running"`
	HeldReason     string         `json:"held_reason"`
	InboundConns   int64          `json:"inbound_conns"`
	RestorePending int            `json:"restore_pending"`
	Downloads      []exchangeItem `json:"downloads"`
}

func (c *Exchange) AddTorrent(ctx context.Context, a Add) error {
	if a.Magnet == "" && len(a.Torrent) == 0 {
		return fmt.Errorf("add: neither a magnet nor torrent bytes")
	}
	if a.SavePath == "" {
		return fmt.Errorf("add: no save path, which would land the download outside the folder Minstrel imports from")
	}
	if len(a.Torrent) > 0 {
		q := url.Values{"save_path": {a.SavePath}}
		if c.Config.Category != "" {
			q.Set("category", c.Config.Category)
		}
		for _, t := range a.Tags {
			q.Add("tag", t)
		}
		if a.Paused {
			q.Set("paused", "true")
		}
		_, err := c.call(ctx, http.MethodPost, "/api/downloads/torrent?"+q.Encode(), "application/x-bittorrent", a.Torrent)
		return err
	}
	body, err := json.Marshal(map[string]any{
		"magnet": a.Magnet, "category": c.Config.Category, "save_path": a.SavePath,
		"tags": a.Tags, "paused": a.Paused,
	})
	if err != nil {
		return fmt.Errorf("add: %w", err)
	}
	_, err = c.call(ctx, http.MethodPost, "/api/downloads", "application/json", body)
	return err
}

// Torrents maps Exchange's downloads onto qBittorrent's shape. While Exchange
// is still restoring after a restart its list is incomplete, so it answers
// ErrRestoring rather than a list that looks like downloads vanished.
func (c *Exchange) Torrents(ctx context.Context, hashes ...string) ([]Torrent, error) {
	snap, err := c.snapshot(ctx)
	if err != nil {
		return nil, err
	}
	if snap.RestorePending > 0 {
		return nil, fmt.Errorf("%w (%d still to come back)", ErrRestoring, snap.RestorePending)
	}
	want := map[string]bool{}
	for _, h := range hashes {
		want[strings.ToLower(strings.TrimSpace(h))] = true
	}
	down := ""
	switch {
	case snap.Running:
	case !snap.Enabled:
		down = "the download engine is switched off in FabledExchange"
	case snap.HeldReason != "":
		down = "the download engine is held (" + snap.HeldReason + ") and resumes by itself"
	default:
		down = "the download engine is enabled but not running; FabledExchange is retrying it"
	}
	out := make([]Torrent, 0, len(snap.Downloads))
	for _, it := range snap.Downloads {
		if c.Config.Category != "" && it.Category != c.Config.Category {
			continue
		}
		t := asTorrent(it)
		t.EngineDown = down
		if len(want) > 0 && !want[t.Hash] {
			continue
		}
		out = append(out, t)
	}
	return out, nil
}

func asTorrent(it exchangeItem) Torrent {
	content := it.SavePath
	if it.Name != "" {
		content = path.Join(it.SavePath, it.Name)
	}
	left := max(it.Length-it.Bytes, 0)
	state := exchangeState(it.State, it.Progress)
	if it.Failed {
		state = "error"
	}
	var added int64
	if !it.AddedAt.IsZero() {
		added = it.AddedAt.Unix()
	}
	return Torrent{
		Hash: strings.ToLower(it.InfoHash), Name: it.Name, Size: it.Length,
		Progress: it.Progress, State: state, Category: it.Category,
		Tags: strings.Join(it.Tags, ","), SavePath: it.SavePath, ContentPath: content,
		AddedOn: added, NumLeechs: it.Peers, AmountLeft: left,
		FailReason: it.FailReason, Ratio: it.Ratio,
	}
}

// exchangeState names Exchange's state in qBittorrent's vocabulary.
func exchangeState(tier string, progress float64) string {
	switch tier {
	case "fast", "slow":
		return "downloading"
	case "metadata":
		return "metaDL"
	case "stalled":
		return "stalledDL"
	case "queued":
		return "queuedDL"
	case "seeding":
		return "uploading"
	case "relocating":
		return "moving"
	case "paused":
		if progress >= 1 {
			return "pausedUP"
		}
		return "pausedDL"
	}
	return "unknown"
}

// Connected is a running engine that has had an inbound connection: the
// nearest Exchange comes to qBittorrent's "connected".
func (c *Exchange) Connected(ctx context.Context) (bool, error) {
	snap, err := c.snapshot(ctx)
	if err != nil {
		return false, err
	}
	return snap.Running && snap.InboundConns > 0, nil
}

func (c *Exchange) Stop(ctx context.Context, hashes ...string) error {
	return c.each(ctx, hashes, func(h string) error {
		_, err := c.call(ctx, http.MethodPost, "/api/downloads/"+url.PathEscape(h)+"/pause", "", nil)
		return err
	})
}

func (c *Exchange) Start(ctx context.Context, hashes ...string) error {
	return c.each(ctx, hashes, func(h string) error {
		_, err := c.call(ctx, http.MethodPost, "/api/downloads/"+url.PathEscape(h)+"/resume", "", nil)
		return err
	})
}

func (c *Exchange) SkipFiles(ctx context.Context, hash string, indices []int) error {
	return c.filePriority(ctx, hash, indices, "skip")
}

func (c *Exchange) KeepFiles(ctx context.Context, hash string, indices []int) error {
	return c.filePriority(ctx, hash, indices, "normal")
}

func (c *Exchange) filePriority(ctx context.Context, hash string, indices []int, priority string) error {
	if len(indices) == 0 {
		return nil
	}
	body, err := json.Marshal(map[string]any{"indices": indices, "priority": priority})
	if err != nil {
		return err
	}
	_, err = c.call(ctx, http.MethodPatch, "/api/downloads/"+url.PathEscape(hash)+"/files", "application/json", body)
	return err
}

func (c *Exchange) Files(ctx context.Context, hash string) ([]File, error) {
	raw, err := c.files(ctx, hash)
	if err != nil {
		return nil, err
	}
	out := make([]File, len(raw))
	for i, f := range raw {
		out[i] = File{Index: f.Index, Path: f.Path, Size: f.Length}
	}
	return out, nil
}

func (c *Exchange) FileStates(ctx context.Context, hash string) ([]FileState, error) {
	raw, err := c.files(ctx, hash)
	if err != nil {
		return nil, err
	}
	out := make([]FileState, len(raw))
	for i, f := range raw {
		if f.Bytes == nil {
			return nil, fmt.Errorf("%w: FabledExchange does not say how much of each file it has", ErrUnsupported)
		}
		out[i] = FileState{Path: f.Path, Size: f.Length, Done: *f.Bytes, Skipped: f.Priority == "skip"}
	}
	return out, nil
}

type exchangeFile struct {
	Index    int    `json:"index"`
	Path     string `json:"path"`
	Length   int64  `json:"length"`
	Bytes    *int64 `json:"bytes"`
	Priority string `json:"priority"`
}

func (c *Exchange) files(ctx context.Context, hash string) ([]exchangeFile, error) {
	body, err := c.call(ctx, http.MethodGet, "/api/downloads/"+url.PathEscape(hash), "", nil)
	if err != nil {
		return nil, err
	}
	var detail struct {
		Files []exchangeFile `json:"files"`
	}
	if err := json.Unmarshal(body, &detail); err != nil {
		return nil, fmt.Errorf("%w: parse download detail: %v", ErrUnreachable, err)
	}
	return detail.Files, nil
}

func (c *Exchange) Reannounce(context.Context, ...string) error {
	return fmt.Errorf("%w: FabledExchange schedules its own announces", ErrUnsupported)
}

func (c *Exchange) Recheck(context.Context, ...string) error {
	return fmt.Errorf("%w: FabledExchange has no recheck", ErrUnsupported)
}

func (c *Exchange) Delete(ctx context.Context, deleteFiles bool, hashes ...string) error {
	return c.each(ctx, hashes, func(h string) error {
		p := "/api/downloads/" + url.PathEscape(h)
		if deleteFiles {
			p += "?delete_files=true"
		}
		_, err := c.call(ctx, http.MethodDelete, p, "", nil)
		return err
	})
}

func (c *Exchange) each(ctx context.Context, hashes []string, do func(string) error) error {
	for _, h := range hashes {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if err := do(strings.ToLower(strings.TrimSpace(h))); err != nil {
			return err
		}
	}
	return nil
}

func (c *Exchange) snapshot(ctx context.Context) (exchangeSnapshot, error) {
	var snap exchangeSnapshot
	body, err := c.call(ctx, http.MethodGet, "/api/downloads", "", nil)
	if err != nil {
		return snap, err
	}
	if err := json.Unmarshal(body, &snap); err != nil {
		return snap, fmt.Errorf("%w: parse download list: %v", ErrUnreachable, err)
	}
	return snap, nil
}

func (c *Exchange) call(ctx context.Context, method, p, contentType string, body []byte) ([]byte, error) {
	if c.Config.BaseURL == "" {
		return nil, ErrUnconfigured
	}
	httpc := c.HTTP
	if httpc == nil {
		httpc = &http.Client{}
	}
	status, out, err := send(ctx, httpc, c.Config.Timeout, method,
		strings.TrimRight(c.Config.BaseURL, "/")+p, contentType, body, func(h http.Header) {
			h.Set("Accept", "application/json")
			if c.Config.APIKey != "" {
				h.Set("Authorization", "Bearer "+c.Config.APIKey)
			}
		})
	if err != nil {
		return nil, err
	}
	if status >= 200 && status < 300 {
		return out, nil
	}
	var e struct {
		Error  string `json:"error"`
		Reason string `json:"reason"`
	}
	_ = json.Unmarshal(out, &e)
	said := strings.TrimSpace(e.Error)
	if said == "" {
		said = fmt.Sprintf("HTTP %d", status)
	}
	switch {
	case status == http.StatusUnauthorized || status == http.StatusForbidden:
		return nil, fmt.Errorf("%w: FabledExchange refused the API key (%s)", ErrAuth, said)
	case status == http.StatusConflict && e.Reason == "duplicate":
		return nil, fmt.Errorf("%w: %s", ErrDuplicate, said)
	}
	return nil, fmt.Errorf("%w: %s", ErrUnreachable, said)
}