repo-sync/fetch/source.go

package fetch

import (
	"bufio"
	"bytes"
	"context"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	cfg "github.com/grmrgecko/repo-sync/config"
	log "github.com/sirupsen/logrus"
)

// Source is the set of base URLs a repository can be fetched from. Direct
// repositories have a single base; mirrorlists provide several with
// failover between them.
type Source struct {
	// mu guards bases and active.
	mu     sync.Mutex
	bases  []string
	active int
}

// NewSource builds a Source from base URLs, normalizing trailing slashes.
func NewSource(bases []string) *Source {
	s := &Source{}
	for _, base := range bases {
		s.bases = append(s.bases, strings.TrimRight(base, "/"))
	}
	return s
}

// Rebase trims a common relative suffix off every base URL so later fetches
// resolve against the parent directory. Bases without the suffix are kept
// unchanged.
func (s *Source) Rebase(suffix string) {
	suffix = "/" + strings.Trim(suffix, "/")
	s.mu.Lock()
	defer s.mu.Unlock()
	for i, base := range s.bases {
		if strings.HasSuffix(base, suffix) {
			s.bases[i] = strings.TrimSuffix(base, suffix)
		} else {
			log.WithField("mirror", base).Warn("Mirror URL does not share the expected path suffix; using it unchanged.")
		}
	}
}

// GetOptions adjusts a single Source request. A non-zero ModifiedSince
// adds a conditional header, surfacing a 304 as ErrNotModified. A positive
// RangeFrom requests the remainder of a partially downloaded file.
type GetOptions struct {
	ModifiedSince time.Time
	RangeFrom     int64
}

// Get requests a repository-relative path, failing over to the next mirror
// on transport or server errors. A 404 from the responding mirror is
// treated as authoritative so optional files do not trigger failover.
func (s *Source) Get(ctx context.Context, reqPath string, o GetOptions) (*http.Response, error) {
	// Snapshot the bases so a concurrent Rebase cannot race the loop.
	s.mu.Lock()
	start := s.active
	bases := append([]string(nil), s.bases...)
	s.mu.Unlock()
	if len(bases) == 0 {
		return nil, errors.New("source has no mirror URLs")
	}

	var lastErr error
	for i := 0; i < len(bases); i++ {
		idx := (start + i) % len(bases)
		target := bases[idx] + "/" + strings.TrimPrefix(reqPath, "/")
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
		if err != nil {
			return nil, err
		}
		req.Header.Set("User-Agent", agent())
		if !o.ModifiedSince.IsZero() {
			req.Header.Set("If-Modified-Since", o.ModifiedSince.UTC().Format(http.TimeFormat))
		}
		if o.RangeFrom > 0 {
			req.Header.Set("Range", fmt.Sprintf("bytes=%d-", o.RangeFrom))
		}
		resp, err := client.Load().Do(req)
		if err != nil {
			lastErr = fmt.Errorf("fetch %s: %w", target, err)
			log.WithError(err).WithField("url", target).Debug("Mirror request failed.")
			continue
		}
		if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
			resp.Body.Close()
			return nil, fmt.Errorf("fetch %s: %w", target, ErrNotFound)
		}
		if resp.StatusCode == http.StatusNotModified && !o.ModifiedSince.IsZero() {
			resp.Body.Close()
			return nil, fmt.Errorf("fetch %s: %w", target, ErrNotModified)
		}
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			resp.Body.Close()
			lastErr = fmt.Errorf("fetch %s: status %d", target, resp.StatusCode)
			log.WithField("url", target).WithField("status", resp.StatusCode).Debug("Mirror request failed.")
			continue
		}
		if idx != start {
			s.mu.Lock()
			s.active = idx
			s.mu.Unlock()
		}
		return resp, nil
	}
	return nil, lastErr
}

// ReadURL retrieves a raw URL fully into memory, limited to 4 MiB. It is
// used for probing mirrorlists, whose URLs may carry query strings that a
// Source cannot join paths onto.
func ReadURL(ctx context.Context, rawURL string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", agent())
	resp, err := client.Load().Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
		return nil, fmt.Errorf("fetch %s: %w", rawURL, ErrNotFound)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return nil, fmt.Errorf("fetch %s: status %d", rawURL, resp.StatusCode)
	}
	// Read one byte past the cap so an oversized document can be detected
	// and reported rather than silently truncated into a bad parse.
	const maxSize = 4 << 20
	data, err := io.ReadAll(io.LimitReader(resp.Body, maxSize+1))
	if err != nil {
		return nil, err
	}
	if len(data) > maxSize {
		log.WithField("url", rawURL).Warn("Response exceeds the read limit and was truncated.")
		data = data[:maxSize]
	}
	return data, nil
}

// ParseMirrorlist extracts mirror base URLs from a mirrorlist body,
// skipping comments and lines that are not absolute HTTP URLs.
func ParseMirrorlist(body []byte) []string {
	seen := map[string]bool{}
	var bases []string
	scanner := bufio.NewScanner(bytes.NewReader(body))
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		u, err := url.Parse(line)
		if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
			continue
		}
		base := strings.TrimRight(line, "/")
		if seen[base] {
			continue
		}
		seen[base] = true
		bases = append(bases, base)
	}
	return bases
}

// ParseMetalink extracts mirror base URLs from a metalink document's
// repomd.xml entries, keeping the document's preference order.
func ParseMetalink(body []byte) []string {
	const repomdSuffix = "/repodata/repomd.xml"
	var ml struct {
		Files []struct {
			Name string `xml:"name,attr"`
			URLs []struct {
				Value string `xml:",chardata"`
			} `xml:"resources>url"`
		} `xml:"files>file"`
	}
	if xml.Unmarshal(body, &ml) != nil {
		return nil
	}
	seen := map[string]bool{}
	var bases []string
	for _, file := range ml.Files {
		if file.Name != "repomd.xml" {
			continue
		}
		for _, u := range file.URLs {
			v := strings.TrimSpace(u.Value)
			if !strings.HasPrefix(v, "http://") && !strings.HasPrefix(v, "https://") {
				continue
			}
			base := strings.TrimSuffix(v, repomdSuffix)
			if base == v {
				continue
			}
			base = strings.TrimRight(base, "/")
			if seen[base] {
				continue
			}
			seen[base] = true
			bases = append(bases, base)
		}
	}
	return bases
}

// agent returns the user agent sent with every upstream request, falling
// back to the build identity when the configuration clears it.
func agent() string {
	if ua := cfg.C.Crawler.UserAgent; ua != "" {
		return ua
	}
	return cfg.Name + "/" + cfg.Version
}

// Get performs a plain GET of a raw URL with the mirror user agent, used
// for directory listings and other non-repository requests.
func Get(ctx context.Context, rawURL string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", agent())
	return client.Load().Do(req)
}

// Bases returns a copy of the mirror base URLs, primarily for tests and
// diagnostics.
func (s *Source) Bases() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]string(nil), s.bases...)
}

// Active returns the base URL of the mirror currently being fetched from,
// which for a mirrorlist or metalink is the one that last answered rather
// than the list's own URL. It is empty when the source has no mirrors.
func (s *Source) Active() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.active < 0 || s.active >= len(s.bases) {
		return ""
	}
	return s.bases[s.active]
}