minstrel/internal/acquire/fanout.go

package acquire

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"

	"git.fabledsword.com/bvandeusen/minstrel/internal/indexer"
)

// Searching every source at once. A source that is down narrows a search rather
// than failing it: the results come back with the list of who did not answer.

// SourceFailure is one source that could not be reached.
type SourceFailure struct {
	Code  string
	Label string
	Err   error
}

// PartialError is a search that reached some sources and not others. It is
// returned alongside the results, not instead of them.
type PartialError struct {
	Failures []SourceFailure
	// Answered is how many sources did answer. Zero means the search failed.
	Answered int
}

func (e *PartialError) Error() string {
	if e == nil || len(e.Failures) == 0 {
		return "some sources did not answer"
	}
	parts := make([]string, 0, len(e.Failures))
	for _, f := range e.Failures {
		parts = append(parts, fmt.Sprintf("%s: %s", f.Label, indexer.FirstLine(f.Err.Error())))
	}
	return "some sources did not answer — " + strings.Join(parts, "; ")
}

// Unwrap lets errors.Is find indexer.ErrUnreachable through a partial failure.
func (e *PartialError) Unwrap() []error {
	if e == nil {
		return nil
	}
	out := make([]error, 0, len(e.Failures))
	for _, f := range e.Failures {
		out = append(out, f.Err)
	}
	return out
}

// Fanout searches every source and merges what comes back. It is both the
// Searcher and the Fetcher, routing a .torrent fetch to the source that
// offered the result.
type Fanout struct {
	// Sources in code order: the fold keeps a duplicate's first arrival, so
	// the order decides which source a grab is fetched through.
	Sources []Source
	// Pace spaces each source's searches. Nil paces nothing.
	Pace *Pacer
}

// Search asks every source concurrently. Each client bounds its own call by
// its configured timeout, so there is no shared deadline here.
func (f *Fanout) Search(ctx context.Context, query indexer.Query) ([]indexer.Result, error) {
	if f == nil || len(f.Sources) == 0 {
		return nil, indexer.ErrUnconfigured
	}
	found := make([][]indexer.Result, len(f.Sources))
	errs := make([]error, len(f.Sources))
	var wg sync.WaitGroup
	for i, src := range f.Sources {
		if src.Search == nil {
			continue
		}
		f.Pace.spend(src)
		wg.Add(1)
		go func(i int, src Source) {
			defer wg.Done()
			found[i], errs[i] = src.Search.Search(ctx, query)
		}(i, src)
	}
	wg.Wait()

	out := make([]indexer.Result, 0, 32)
	var failures []SourceFailure
	answered, unconfigured := 0, 0
	for i, src := range f.Sources {
		if src.Search == nil {
			continue
		}
		if errors.Is(errs[i], indexer.ErrUnconfigured) {
			unconfigured++
			continue
		}
		if rl, ok := indexer.AsRateLimited(errs[i]); ok {
			f.Pace.hold(src.Code, rl)
		}
		if errs[i] != nil {
			failures = append(failures, SourceFailure{Code: src.Code, Label: src.Label, Err: errs[i]})
			continue
		}
		answered++
		for j := range found[i] {
			if found[i][j].Source == "" {
				found[i][j].Source = src.Code
			}
		}
		out = append(out, found[i]...)
	}
	switch {
	case answered == 0 && len(failures) > 0:
		return nil, &PartialError{Failures: failures}
	case len(failures) > 0:
		return out, &PartialError{Failures: failures, Answered: answered}
	case answered == 0 && unconfigured > 0:
		return nil, indexer.ErrUnconfigured
	}
	return out, nil
}

// Download fetches a result's .torrent from the source that offered it: a
// Prowlarr link carries Prowlarr's key and only Prowlarr can redeem it.
func (f *Fanout) Download(ctx context.Context, r indexer.Result) ([]byte, error) {
	if f == nil {
		return nil, indexer.ErrUnconfigured
	}
	for _, src := range f.Sources {
		if src.Code != r.Source {
			continue
		}
		if src.Fetch == nil {
			return nil, fmt.Errorf("%w: %s cannot fetch torrent files", indexer.ErrUnconfigured, src.Label)
		}
		return src.Fetch.Download(ctx, r)
	}
	if r.Source == "" {
		return nil, fmt.Errorf("%w: this result does not say which source it came from", indexer.ErrUnconfigured)
	}
	return nil, fmt.Errorf("%w: %q is no longer a configured source", indexer.ErrUnconfigured, r.Source)
}

// Narrowing is one source that did not answer, as a page shows it.
type Narrowing struct {
	Source string `json:"source"`
	Label  string `json:"label"`
	Reason string `json:"reason"`
}

// narrowings reads a PartialError out of err, and whether any source
// answered. Nil when err is not a partial failure.
func narrowings(err error) ([]Narrowing, bool) {
	var partial *PartialError
	if !errors.As(err, &partial) || len(partial.Failures) == 0 {
		return nil, false
	}
	out := make([]Narrowing, 0, len(partial.Failures))
	for _, f := range partial.Failures {
		out = append(out, Narrowing{Source: f.Code, Label: f.Label, Reason: indexer.FirstLine(f.Err.Error())})
	}
	return out, partial.Answered > 0
}