pkg-proxy/internal/handler/handler.go

// Package handler provides HTTP protocol handlers for package manager proxying.
package handler

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/git-pkgs/artifacts"
	"github.com/git-pkgs/cooldown"
	"github.com/git-pkgs/proxy/internal/database"
	"github.com/git-pkgs/proxy/internal/denylist"
	"github.com/git-pkgs/proxy/internal/httpclient"
	"github.com/git-pkgs/proxy/internal/metrics"
	"github.com/git-pkgs/proxy/internal/packageurl"
	"github.com/git-pkgs/proxy/internal/scanner"
	"github.com/git-pkgs/proxy/internal/storage"
	"github.com/git-pkgs/purl"
	"github.com/git-pkgs/registries/fetch"
	"github.com/opencontainers/go-digest"
)

// containsPathTraversal returns true if the path contains ".." segments
// that could be used to escape the intended directory. It checks the path
// as given and after URL-decoding, and treats backslashes as separators.
func containsPathTraversal(path string) bool {
	if hasDotDotSegment(path) {
		return true
	}
	if decoded, err := url.PathUnescape(path); err == nil && decoded != path {
		return hasDotDotSegment(decoded)
	}
	return false
}

func hasDotDotSegment(path string) bool {
	path = strings.ReplaceAll(path, "\\", "/")
	for segment := range strings.SplitSeq(path, "/") {
		if segment == ".." {
			return true
		}
	}
	return false
}

func configuredUpstreamURL(value, defaultValue string) string {
	if value == "" {
		value = defaultValue
	}
	return strings.TrimRight(value, "/")
}

const defaultHTTPTimeout = 30 * time.Second

const artifactCopyBufferSize = 32 << 10

var artifactCopyBufferPool = sync.Pool{ //nolint:gochecknoglobals // shared across artifact responses
	New: func() any {
		buffer := make([]byte, artifactCopyBufferSize)
		return &buffer
	},
}

// canonicalPackagePURL returns a versionless PURL in canonical form so cooldown
// lookups match keys produced by config.CooldownConfig.NormalizedPackages.
func canonicalPackagePURL(ecosystem, name string) string {
	return packageurl.MakeString(ecosystem, name, "")
}

// canonicalVersionPURL returns a versioned PURL in canonical form, matching
// the keys the artifact cache writes to the versions table.
func canonicalVersionPURL(ecosystem, name, version string) string {
	return packageurl.MakeString(ecosystem, name, version)
}

var errUnsupportedPackageIdentity = errors.New("package identity cannot be represented as a PURL")

func packagePURLStrings(ecosystem, name, version string) (string, string, error) {
	packagePURL := packageurl.MakeString(ecosystem, name, "")
	versionPURL := packageurl.MakeString(ecosystem, name, version)
	if packagePURL == "" || versionPURL == "" {
		return "", "", fmt.Errorf("%w: %s %q", errUnsupportedPackageIdentity, ecosystem, name)
	}
	return packagePURL, versionPURL, nil
}

const contentTypeJSON = "application/json"

const (
	headerAccept          = "Accept"
	headerAcceptEncoding  = "Accept-Encoding"
	headerContentType     = "Content-Type"
	headerContentLength   = "Content-Length"
	headerContentEncoding = "Content-Encoding"
	headerETag            = "ETag"
	headerLastModified    = "Last-Modified"
)

// ifNoneMatchHits reports whether the given If-None-Match header value
// matches the current entity tag using weak comparison, so "*" matches any
// tag, W/ prefixes are ignored on both sides, and a comma-separated list is
// scanned. An empty header or an empty stored tag never match.
func ifNoneMatchHits(header, etag string) bool {
	if etag == "" || header == "" {
		return false
	}
	if header == "*" {
		return true
	}
	etag = strings.TrimPrefix(etag, "W/")
	for value := range strings.SplitSeq(header, ",") {
		if strings.TrimPrefix(strings.TrimSpace(value), "W/") == etag {
			return true
		}
	}
	return false
}

// defaultMetadataMaxSize is used when Proxy.MetadataMaxSize is unset.
const defaultMetadataMaxSize = 100 << 20

// ErrMetadataTooLarge is returned when upstream metadata exceeds the configured limit.
var ErrMetadataTooLarge = errors.New("metadata response exceeds size limit")

// ReadMetadata reads an upstream response body with a size limit to prevent OOM
// from unexpectedly large responses. Returns ErrMetadataTooLarge if the response
// is truncated by the limit.
func (p *Proxy) ReadMetadata(r io.Reader) ([]byte, error) {
	limit := p.MetadataMaxSize
	if limit <= 0 {
		limit = defaultMetadataMaxSize
	}
	data, err := io.ReadAll(io.LimitReader(r, limit+1))
	if err != nil {
		return nil, err
	}
	if int64(len(data)) > limit {
		return nil, ErrMetadataTooLarge
	}
	return data, nil
}

// Proxy provides shared functionality for protocol handlers.
type Proxy struct {
	DB                  *database.DB
	Storage             storage.Storage
	Fetcher             fetch.FetcherInterface
	Resolver            *fetch.Resolver
	Logger              *slog.Logger
	Cooldown            CooldownPolicy
	Denylist            *denylist.Policy
	CacheMetadata       bool
	MetadataTTL         time.Duration
	MetadataMaxSize     int64
	GradleReadOnly      bool
	HostedEnabled       bool
	GradleMaxUploadSize int64
	// NPMFullMetadata requests full npm packuments from upstream even when
	// cooldown is disabled, so served metadata carries publish times.
	NPMFullMetadata bool
	DirectServe     bool
	DirectServeTTL  time.Duration
	// DirectServeBaseURL, if set, replaces the scheme and host of presigned
	// URLs so clients receive a public address even when the proxy reaches
	// storage at an internal one.
	DirectServeBaseURL string
	HTTPClient         *http.Client

	// RateLimits is the gate under the upstream clients. Artifact fetches
	// check it before calling Fetcher, so held hosts never reach its breaker.
	RateLimits *httpclient.RateLimitGate

	// HostedNPMScopes and HostedComposerVendors name namespaces served from
	// pushed packages alone, without asking the upstream.
	HostedNPMScopes       []string
	HostedComposerVendors []string

	// StreamArtifacts streams artifacts from upstream without storing them.
	// Each request fetches its own copy: there is no cache to check and
	// nothing for concurrent misses to share.
	StreamArtifacts bool

	// Scanners runs pre-cache artifact scanning (e.g. trivy, ClamAV, Wiz).
	// Nil or disabled means artifacts are cached without scanning.
	Scanners *scanner.Group

	// ScanSigningKey authenticates pull requests to the internal
	// /_internal/scan-fetch route used by scanners to retrieve staged
	// artifacts, for every storage backend.
	ScanSigningKey []byte

	// ScanFetchBaseURL is the address scanners use to reach this proxy to
	// pull staged artifacts.
	ScanFetchBaseURL string

	// inFlight coalesces concurrent cache misses for one artifact, so a single
	// upstream fetch serves every waiting caller. Keyed by artifactCoalesceKey.
	fetchMu  sync.Mutex
	inFlight map[string]*inflightFetch

	// detachSlots caps the artifact fetches that keep running after their
	// client went away (see DetachUpstreamFetch). Guarded by fetchMu while it
	// is created.
	detachSlots chan struct{}

	// inFlightMeta does the same for metadata misses. Keyed by
	// metadataCoalesceKey.
	metaMu       sync.Mutex
	inFlightMeta map[string]*inflightMetadata

	// rewrites caches metadata documents after their handler rewrites them.
	// Nil leaves every request to rewrite its own copy.
	rewrites *rewriteCache

	// now is the clock rewritten metadata is dated by, and started the time
	// the process started; see metadataDate.
	now     func() time.Time
	started time.Time
}

// CooldownPolicy supplies version-age filtering and package-specific durations.
type CooldownPolicy interface {
	IsAllowed(ecosystem, packagePURL string, publishedAt time.Time) bool
	Evaluate(ecosystem, packagePURL string, publishedAt, evaluatedAt time.Time) cooldown.Decision
	For(ecosystem, packagePURL string) time.Duration
	Enabled() bool
}

// NewProxy creates a new Proxy with the given dependencies.
func NewProxy(db *database.DB, store storage.Storage, fetcher fetch.FetcherInterface, resolver *fetch.Resolver, logger *slog.Logger) *Proxy {
	if logger == nil {
		logger = slog.Default()
	}
	return &Proxy{
		DB:       db,
		Storage:  store,
		Fetcher:  fetcher,
		Resolver: resolver,
		Logger:   logger,
		HTTPClient: &http.Client{
			Timeout: defaultHTTPTimeout,
		},
		now:     time.Now,
		started: time.Now(),
	}
}

// CacheResult contains information about a cached or fetched artifact.
type CacheResult struct {
	Reader      io.ReadCloser
	RedirectURL string
	Artifact    artifacts.Artifact
	Cached      bool
	storagePath string
}

// GetOrFetchArtifact retrieves an artifact from cache or fetches from upstream.
func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) {
	// NuGet ids are case-insensitive and the NuGet handler stores them
	// lowercased, so a mirrored Newtonsoft.Json must land on the same record
	// clients are served.
	if ecosystem == "nuget" {
		name = strings.ToLower(name)
	}
	pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version)
	if err != nil {
		return nil, err
	}
	// Deny before resolution as well as before reading the cache. Mirror callers
	// do not supply a filename, and resolving it can contact an upstream registry.
	if p.Denylist.Denied(versionPURL) {
		return nil, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL)
	}
	var info *fetch.ArtifactInfo
	if filename == "" {
		info, err = p.resolveArtifact(ctx, ecosystem, name, version)
		if err != nil {
			return nil, err
		}
		filename = info.Filename
		if filename == "" {
			return nil, errors.New("resolved artifact has no filename")
		}
	}
	if p.StreamArtifacts {
		if info == nil {
			if info, err = p.resolveArtifact(ctx, ecosystem, name, version); err != nil {
				return nil, err
			}
		}
		return p.streamFromUpstream(ctx, ecosystem, name, version, filename, versionPURL, info.URL, "", "",
			func(fetchCtx context.Context) (*fetch.Artifact, error) {
				return p.Fetcher.Fetch(fetchCtx, info.URL)
			})
	}
	if cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename); err != nil {
		return nil, err
	} else if cached != nil {
		return cached, nil
	}
	metrics.RecordCacheMiss(ecosystem)

	key := artifactCoalesceKey(versionPURL, filename, "", "", "")
	recheck := func() (artifacts.Artifact, string, bool) {
		return p.cachedArtifactRecord(pkgPURL, versionPURL, filename, "")
	}
	return p.coalesceFetch(ctx, key, recheck, func(fetchCtx context.Context) (artifacts.Artifact, string, error) {
		return p.fetchAndCache(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL, info)
	})
}

// GetCachedArtifact retrieves an artifact from cache without contacting an upstream.
// It returns nil when no usable cache entry exists.
func (p *Proxy) GetCachedArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) {
	pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version)
	if err != nil {
		return nil, err
	}
	return p.checkCache(ctx, pkgPURL, versionPURL, filename)
}

// ClearCachedArtifact clears an artifact cache record after an external
// integrity check fails, and queues its stored bytes for deletion.
func (p *Proxy) ClearCachedArtifact(ecosystem, name, version, filename string) error {
	if p.DB == nil || p.Storage == nil {
		return nil
	}
	pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version)
	if err != nil {
		return err
	}
	cached, err := p.DB.GetCachedArtifact(pkgPURL, versionPURL, filename)
	if err != nil {
		return fmt.Errorf("looking up cached artifact: %w", err)
	}
	if cached == nil {
		return nil
	}
	return p.DB.DiscardArtifact(versionPURL, filename, cached.StoragePath)
}

// checkCache looks up an artifact in the cache. Returns nil if not cached.
// With StreamArtifacts set it always reports a miss, so entries stored before the
// mode was enabled are never served.
//
// A miss on a pushed artifact is an error rather than a miss: the caller would
// otherwise fetch the upstream copy and replace the pushed one with it.
func (p *Proxy) checkCache(ctx context.Context, pkgPURL, versionPURL, filename string) (*CacheResult, error) {
	result, err := p.lookupCache(ctx, pkgPURL, versionPURL, filename)
	if err != nil || result != nil || p.StreamArtifacts {
		return result, err
	}
	return nil, p.localArtifactMiss(versionPURL, filename)
}

// localArtifactMiss returns ErrHostedArtifactUnavailable when the artifact
// was pushed and is stored but couldn't be served, ErrHostedArtifactMissing
// when it was pushed and its stored object is gone, and nil otherwise.
func (p *Proxy) localArtifactMiss(versionPURL, filename string) error {
	record, err := p.DB.GetArtifact(versionPURL, filename)
	if err != nil {
		return fmt.Errorf("checking artifact source: %w", err)
	}
	if record == nil || record.Source != database.SourceLocal {
		return nil
	}
	if record.StoragePath.Valid {
		return fmt.Errorf("%w: %s %s", ErrHostedArtifactUnavailable, versionPURL, filename)
	}
	return fmt.Errorf("%w: %s %s", ErrHostedArtifactMissing, versionPURL, filename)
}

// lookupCache looks up an artifact in the cache. Returns nil if not cached.
func (p *Proxy) lookupCache(ctx context.Context, pkgPURL, versionPURL, filename string) (*CacheResult, error) {
	if p.Denylist.Denied(versionPURL) {
		return nil, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL)
	}
	if p.StreamArtifacts {
		return nil, nil
	}
	return p.readStored(ctx, pkgPURL, versionPURL, filename)
}

// readStored opens the stored object of an artifact, whether or not
// StreamArtifacts is set. Returns nil if no usable object is stored.
func (p *Proxy) readStored(ctx context.Context, pkgPURL, versionPURL, filename string) (*CacheResult, error) {
	artifact, err := p.DB.GetCachedArtifact(pkgPURL, versionPURL, filename)
	if err != nil {
		return nil, fmt.Errorf("checking artifact cache: %w", err)
	}
	if artifact == nil {
		return nil, nil
	}
	checks, err := newIntegrityChecks(artifact.Artifact.Digest.Encoded(), artifact.Integrity.String)
	if err != nil {
		p.rejectUnusableCacheRecord(artifact, versionPURL, filename, err)
		return nil, nil
	}

	result := &CacheResult{
		Artifact:    artifact.Artifact,
		Cached:      true,
		storagePath: artifact.StoragePath,
	}

	if p.DirectServe {
		signed, err := p.Storage.SignedURL(ctx, artifact.StoragePath, p.DirectServeTTL)
		if err == nil {
			result.RedirectURL = rewriteSignedURLHost(signed, p.DirectServeBaseURL)
			p.recordCacheHit(artifact.Ecosystem, versionPURL, filename)
			return result, nil
		}
		if !errors.Is(err, storage.ErrSignedURLUnsupported) {
			p.Logger.Warn("failed to sign storage URL, falling back to streaming",
				"path", artifact.StoragePath, "error", err)
		}
	}

	start := time.Now()
	reader, err := p.Storage.Open(ctx, artifact.StoragePath)
	metrics.RecordStorageOperation("read", time.Since(start))
	if err != nil {
		metrics.RecordStorageError("read", metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, Path: artifact.StoragePath, Err: err})
		p.Logger.Warn("cached artifact missing from storage, will refetch",
			"path", artifact.StoragePath, "error", err)
		return nil, nil
	}

	result.Reader, err = checks.wrap(reader,
		func(reason string) {
			p.Logger.Error("cached artifact failed integrity check",
				"purl", versionPURL, "filename", filename,
				"path", artifact.StoragePath, "reason", reason)
			metrics.RecordIntegrityFailure(purl.NormalizeEcosystem(artifact.Ecosystem), metrics.FailureDetail{
				PURL: versionPURL, Filename: filename, Path: artifact.StoragePath, Err: errors.New(reason)})
			if err := p.DB.DiscardArtifact(versionPURL, filename, artifact.StoragePath); err != nil {
				p.Logger.Warn("failed to clear corrupt artifact from cache", "error", err)
			}
		})
	if err != nil {
		_ = reader.Close()
		p.rejectUnusableCacheRecord(artifact, versionPURL, filename, err)
		return nil, nil
	}
	p.recordCacheHit(artifact.Ecosystem, versionPURL, filename)
	return result, nil
}

// rewriteSignedURLHost replaces the scheme and host of a signed URL with those
// from baseURL, preserving the path and query (which carry the signature).
// Returns signed unchanged if baseURL is empty or either URL fails to parse.
func rewriteSignedURLHost(signed, baseURL string) string {
	if baseURL == "" {
		return signed
	}
	s, err := url.Parse(signed)
	if err != nil {
		return signed
	}
	b, err := url.Parse(baseURL)
	if err != nil || b.Scheme == "" || b.Host == "" {
		return signed
	}
	s.Scheme = b.Scheme
	s.Host = b.Host
	return s.String()
}

func (p *Proxy) recordCacheHit(ecosystem, versionPURL, filename string) {
	_ = p.DB.RecordArtifactHit(versionPURL, filename)
	metrics.RecordCacheHit(ecosystem)
}

func (p *Proxy) rejectUnusableCacheRecord(artifact *database.CachedArtifact, versionPURL, filename string, cause error) {
	p.Logger.Warn("cached artifact has unusable integrity metadata",
		"purl", versionPURL, "filename", filename,
		"path", artifact.StoragePath, "error", cause)
	metrics.RecordIntegrityFailure(purl.NormalizeEcosystem(artifact.Ecosystem), metrics.FailureDetail{
		PURL: versionPURL, Filename: filename, Path: artifact.StoragePath, Err: cause})
	if err := p.DB.DiscardArtifact(versionPURL, filename, artifact.StoragePath); err != nil {
		p.Logger.Warn("failed to clear unusable artifact from cache", "error", err)
	}
}

func (p *Proxy) resolveArtifact(ctx context.Context, ecosystem, name, version string) (*fetch.ArtifactInfo, error) {
	info, err := p.Resolver.Resolve(ctx, ecosystem, name, version)
	if err != nil {
		if errors.Is(err, fetch.ErrNotFound) {
			return nil, ErrUpstreamNotFound
		}
		return nil, fmt.Errorf("resolving download URL: %w", err)
	}
	return info, nil
}

func (p *Proxy) fetchAndCache(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL string, info *fetch.ArtifactInfo) (artifacts.Artifact, string, error) {
	// Reuse the resolution used for an unnamed cache lookup. Named requests
	// still resolve only on a cache miss, inside the coalesced fetch.
	if info == nil {
		var err error
		info, err = p.resolveArtifact(ctx, ecosystem, name, version)
		if err != nil {
			return artifacts.Artifact{}, "", err
		}
	}

	if err := p.RateLimits.Held(info.URL); err != nil {
		return artifacts.Artifact{}, "", err
	}
	p.Logger.Info("fetching from upstream",
		"ecosystem", ecosystem, "name", name, "version", version, "url", info.URL)

	// Fetch from upstream with timing
	fetchStart := time.Now()
	artifact, err := p.Fetcher.Fetch(ctx, info.URL)
	fetchDuration := time.Since(fetchStart)

	if err != nil {
		metrics.RecordUpstreamFetch(ecosystem, fetchDuration)
		metrics.RecordUpstreamError(ecosystem, "fetch_failed", metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: info.URL, Err: err})
		if errors.Is(err, fetch.ErrNotFound) {
			return artifacts.Artifact{}, "", ErrUpstreamNotFound
		}
		return artifacts.Artifact{}, "", fmt.Errorf("fetching from upstream: %w", err)
	}
	metrics.RecordUpstreamFetch(ecosystem, fetchDuration)

	return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, info.URL, "", "", artifact)
}

// storeArtifact writes a fetched artifact to storage, verifies it against
// upstreamHash and declared, the integrity the registry's metadata declares,
// if they are non-empty, runs it through the scan gate if scanning is enabled,
// and commits it to the cache database, recording declared with the version
// so cache hits are verified against it too.
//
// The scan gate sits between Storage.Store and updateCacheDB: updateCacheDB
// is the only thing that makes an artifact visible to clients (checkCache
// looks up its row before touching Storage), so deferring it until after a
// verdict means a blocked artifact was never reachable by any client. On
// block, the just-stored bytes are deleted and ErrArtifactBlocked is
// returned; updateCacheDB is never called.
//
// It returns the artifact and its storage path, not a reader; callers get one
// from openStoredArtifact.
func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, declared string, artifact *fetch.Artifact) (artifacts.Artifact, string, error) {
	storagePath := storage.FetchPath(ecosystem, name, version, storage.NewFetchID(), filename)

	body, verified, err := verifyIntegrity(artifact.Body, declared)
	if err != nil {
		_ = artifact.Body.Close()
		return artifacts.Artifact{}, "", err
	}

	storeStart := time.Now()
	size, hash, err := p.Storage.Store(ctx, storagePath, body)
	_ = artifact.Body.Close()
	metrics.RecordStorageOperation("write", time.Since(storeStart))
	if err != nil {
		metrics.RecordStorageError("write", metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: upstreamURL, Path: storagePath, Err: err})
		return artifacts.Artifact{}, "", fmt.Errorf("storing artifact: %w", err)
	}

	if !artifactHashMatches(hash, upstreamHash) {
		if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil {
			p.Logger.Warn("failed to discard artifact with mismatched checksum", "path", storagePath, "error", delErr)
		}
		return artifacts.Artifact{}, "", fmt.Errorf("%w: upstream declared %s, got %s", ErrArtifactDigestMismatch, upstreamHash, hash)
	}
	if err := verified(); err != nil {
		p.Logger.Error("fetched artifact failed integrity check",
			"purl", versionPURL, "filename", filename, "url", upstreamURL, "error", err)
		metrics.RecordIntegrityFailure(purl.NormalizeEcosystem(ecosystem), metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: upstreamURL, Err: err})
		if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil {
			p.Logger.Warn("failed to discard artifact with mismatched integrity", "path", storagePath, "error", delErr)
		}
		return artifacts.Artifact{}, "", err
	}

	if p.Scanners != nil && p.Scanners.Enabled() {
		if err := p.runScan(ctx, ecosystem, name, version, filename, versionPURL, storagePath, size, artifact.ContentType); err != nil {
			// Detached from ctx: a client disconnecting must not abort
			// cleanup of a genuinely blocked artifact and leave its bytes
			// orphaned in storage with no DB row pointing at them.
			if delErr := p.Storage.Delete(context.WithoutCancel(ctx), storagePath); delErr != nil {
				p.Logger.Warn("failed to delete blocked artifact from storage",
					"path", storagePath, "error", delErr)
			}
			return artifacts.Artifact{}, "", err
		}
	}

	sharedArtifact := artifacts.Artifact{
		PURL:      versionPURL,
		Digest:    digest.Digest("sha256:" + hash),
		Size:      size,
		Filename:  filename,
		MediaType: artifact.ContentType,
	}

	// Update database
	if err := p.updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath, sharedArtifact, declared); err != nil {
		p.Logger.Warn("failed to update cache database", "error", err)
		// Continue anyway - we have the file. Queue it for deletion in case
		// no record points at it; the queue skips it while one does.
		if qErr := p.DB.QueuePendingDelete(storagePath); qErr != nil {
			p.Logger.Warn("failed to queue unrecorded artifact for deletion", "path", storagePath, "error", qErr)
		}
	}

	return sharedArtifact, storagePath, nil
}

// openStoredArtifact gives one caller its own reader over just-committed
// bytes. Callers sharing a fetch cannot share a handle: it has one read
// position, so they would consume each other's bytes and the first Close would
// break the rest.
func (p *Proxy) openStoredArtifact(ctx context.Context, artifact artifacts.Artifact, storagePath string) (*CacheResult, error) {
	readStart := time.Now()
	reader, err := p.Storage.Open(ctx, storagePath)
	metrics.RecordStorageOperation("read", time.Since(readStart))

	if err != nil {
		metrics.RecordStorageError("read", metrics.FailureDetail{
			PURL: artifact.PURL, Filename: artifact.Filename, Path: storagePath, Err: err})
		return nil, fmt.Errorf("opening cached artifact: %w", err)
	}

	return &CacheResult{
		Reader:   reader,
		Artifact: artifact,
		Cached:   false,
	}, nil
}

// artifactCoalesceKey identifies one artifact fetch. downloadURL, upstreamHash
// and declared are included so callers expecting different bytes (multiple
// upstreams, or a re-published version) never share a fetch. The hash is
// lowercased because artifactHashMatches compares case-insensitively, so one
// digest in two casings describes one artifact and must not split the fetch.
func artifactCoalesceKey(versionPURL, filename, downloadURL, upstreamHash, declared string) string {
	return strings.Join([]string{versionPURL, filename, downloadURL, strings.ToLower(upstreamHash), declared}, "\x00")
}

// cachedArtifactRecord reports an artifact already committed to the cache,
// without opening it. A caller checks the cache before it gets here, so a
// concurrent fetch can commit the same artifact in between; rechecking the
// record keeps that caller from fetching it a second time. A lookup error is
// reported as a miss, which costs a redundant fetch rather than a failure.
func (p *Proxy) cachedArtifactRecord(pkgPURL, versionPURL, filename, upstreamHash string) (artifacts.Artifact, string, bool) {
	record, err := p.DB.GetCachedArtifact(pkgPURL, versionPURL, filename)
	if err != nil || record == nil {
		return artifacts.Artifact{}, "", false
	}
	if !artifactHashMatches(record.Artifact.Digest.Encoded(), upstreamHash) {
		return artifacts.Artifact{}, "", false
	}
	return record.Artifact, record.StoragePath, true
}

// errSharedFetchAbandoned is what waiters see if the caller running a shared
// fetch panicked out of it.
var errSharedFetchAbandoned = errors.New("shared upstream fetch did not complete")

// inflightFetch is one upstream fetch that concurrent callers share. val and
// err are written before done closes and read only after, so the close is the
// handoff.
type inflightFetch struct {
	done chan struct{}
	val  fetchedArtifact
	err  error
}

// coalesceFetch runs commit at most once for concurrent callers sharing key,
// then gives each its own reader over the stored bytes.
//
// The first caller in runs the fetch and the rest wait on it. Roles are
// decided under fetchMu rather than inferred afterwards, because the two need
// different cancellation behaviour: a waiter may leave when its own client goes
// away, while the caller running the fetch must see it through so
// storeArtifact's scan-on-disconnect handling still decides the outcome.
//
// Before fetching, that caller rechecks the cache through recheck: its own
// lookup happened before it took the key, so a fetch that committed in
// between would otherwise be repeated. A hit fills the shared value as a
// fetch would.
//
// commit runs on that caller's context, so cancellation behaves as it did
// uncoalesced and mirroring still relies on it aborting the fetch. If that
// caller goes away, everyone sharing the fetch gets its error and the key is
// released for a later retry. A context marked with DetachUpstreamFetch is the
// exception: there commit keeps running after the caller goes away, so the
// artifact still lands in the cache.
//
// Every sharing caller still records a cache miss, so the gap between
// proxy_cache_misses_total and upstream fetch observations is what coalescing
// saved.
func (p *Proxy) coalesceFetch(ctx context.Context, key string, recheck func() (artifacts.Artifact, string, bool), commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) {
	p.fetchMu.Lock()
	if p.inFlight == nil {
		p.inFlight = make(map[string]*inflightFetch)
	}
	f, joined := p.inFlight[key]
	if !joined {
		f = &inflightFetch{done: make(chan struct{})}
		p.inFlight[key] = f
	}
	p.fetchMu.Unlock()

	if !joined {
		return p.runSharedFetch(ctx, key, f, recheck, commit)
	}

	select {
	case <-ctx.Done():
		// This caller gave up; the fetch continues for everyone else.
		return nil, ctx.Err()
	case <-f.done:
	}
	if f.err != nil {
		return nil, f.err
	}
	return p.openStoredArtifact(ctx, f.val.artifact, f.val.storagePath)
}

// runSharedFetch performs the fetch that joined callers are waiting on. It is
// never abandoned early, and always releases the key and wakes the waiters.
func (p *Proxy) runSharedFetch(ctx context.Context, key string, f *inflightFetch, recheck func() (artifacts.Artifact, string, bool), commit func(context.Context) (artifacts.Artifact, string, error)) (*CacheResult, error) {
	// Set before running so a panicking commit leaves waiters with an error
	// rather than a zero-valued artifact.
	f.err = errSharedFetchAbandoned
	defer func() {
		p.fetchMu.Lock()
		delete(p.inFlight, key)
		p.fetchMu.Unlock()
		close(f.done)
	}()

	// A caller checks the cache before reaching here, so a fetch that finished
	// in between would otherwise be repeated. Serve that record only if its
	// bytes are still present: a record can outlive them, and refetching is
	// the same recovery the cache lookup makes.
	if stored, path, ok := recheck(); ok {
		if res, err := p.openStoredArtifact(ctx, stored, path); err == nil {
			f.val, f.err = fetchedArtifact{artifact: stored, storagePath: path}, nil
			return res, nil
		}
	}

	commitCtx := ctx
	if upstreamFetchDetached(ctx) && p.acquireDetachSlot() {
		defer p.releaseDetachSlot()
		commitCtx = context.WithoutCancel(ctx)
	}
	stored, path, err := commit(commitCtx)
	f.val, f.err = fetchedArtifact{artifact: stored, storagePath: path}, err
	if err != nil {
		return nil, err
	}
	return p.openStoredArtifact(ctx, stored, path)
}

type detachUpstreamFetchKey struct{}

// DetachUpstreamFetch marks ctx so that an artifact fetch started under it
// runs to completion even when ctx is cancelled. The server sets it on client
// requests: a client that gives up on a slow cache miss (pacman aborts after
// 10 s without data) would otherwise cancel the fetch, and every retry would
// start again from zero, so a large artifact never reaches the cache. The
// fetch stays bounded by the artifact HTTP client's timeout, and at most
// maxDetachedFetches run detached at once. Mirror jobs leave it unset so
// cancelling a job still aborts its fetches.
func DetachUpstreamFetch(ctx context.Context) context.Context {
	return context.WithValue(ctx, detachUpstreamFetchKey{}, true)
}

func upstreamFetchDetached(ctx context.Context) bool {
	detached, _ := ctx.Value(detachUpstreamFetchKey{}).(bool)
	return detached
}

// maxDetachedFetches caps the artifact fetches that may outlive their client.
// Without a cap a client could start any number of large uncached downloads
// and hang up straight away, leaving each to run to the artifact client's
// timeout. Fetches beyond the cap stay tied to their client, as before
// detaching existed.
const maxDetachedFetches = 64

func (p *Proxy) acquireDetachSlot() bool {
	p.fetchMu.Lock()
	if p.detachSlots == nil {
		p.detachSlots = make(chan struct{}, maxDetachedFetches)
	}
	slots := p.detachSlots
	p.fetchMu.Unlock()

	select {
	case slots <- struct{}{}:
		return true
	default:
		return false
	}
}

func (p *Proxy) releaseDetachSlot() {
	<-p.detachSlots
}

// fetchedArtifact is what a shared fetch hands its callers: metadata and a
// storage path, neither holding reader state.
type fetchedArtifact struct {
	artifact    artifacts.Artifact
	storagePath string
}

// runScan generates a signed fetch URL for the just-staged artifact and
// asks the configured scanners for a verdict. Returns a wrapped
// ErrArtifactBlocked if any scanner blocks, or a scan-infrastructure error.
//
// The scan call runs on a context detached from ctx's cancellation
// (context.WithoutCancel): ctx is the original client request's context, and
// a client disconnecting mid-scan must not be indistinguishable from a real
// scanner verdict. Group.Scan still bounds the call with its own configured
// timeout, so a detached context cannot hang forever.
func (p *Proxy) runScan(ctx context.Context, ecosystem, name, version, filename, purlStr, storagePath string, size int64, contentType string) error {
	fetchURL := p.scanFetchURL(storagePath, p.Scanners.Timeout())
	result := p.Scanners.Scan(context.WithoutCancel(ctx), scanner.Request{
		Ecosystem:   ecosystem,
		Name:        name,
		Version:     version,
		Filename:    filename,
		PURL:        purlStr,
		FetchURL:    fetchURL,
		Size:        size,
		ContentType: contentType,
	})
	if !result.Allowed {
		p.Logger.Warn("artifact blocked by security scan",
			"ecosystem", ecosystem, "name", name, "version", version, "filename", filename,
			"scanner", result.ScannerName, "reason", result.Reason, "infra_error", result.InfraError)
		reason := result.Reason
		if result.InfraError {
			// result.Reason may contain raw scanner-infrastructure details
			// (internal hostnames, ports, connection errors) that must not
			// reach an untrusted client via the 403 response body.
			reason = "scan could not be completed"
		}
		return fmt.Errorf("%w: %s", ErrArtifactBlocked, reason)
	}
	return nil
}

func (p *Proxy) updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath string, artifact artifacts.Artifact, declared string) error {
	now := time.Now()

	// Upsert package
	pkg := &database.Package{
		PURL:        pkgPURL,
		Ecosystem:   ecosystem,
		Name:        name,
		RegistryURL: sql.NullString{String: upstreamURL, Valid: true},
		EnrichedAt:  sql.NullTime{Time: now, Valid: true},
	}
	if err := p.DB.UpsertPackage(pkg); err != nil {
		return fmt.Errorf("upserting package: %w", err)
	}

	// Upsert version
	ver := &database.Version{
		PURL:        artifact.PURL,
		PackagePURL: pkgPURL,
		Integrity:   sql.NullString{String: declared, Valid: declared != ""},
		EnrichedAt:  sql.NullTime{Time: now, Valid: true},
	}
	if err := p.DB.UpsertVersion(ver); err != nil {
		return fmt.Errorf("upserting version: %w", err)
	}

	// Upsert artifact
	art := &database.Artifact{
		VersionPURL: artifact.PURL,
		Filename:    artifact.Filename,
		UpstreamURL: upstreamURL,
		StoragePath: sql.NullString{String: storagePath, Valid: true},
		ContentHash: sql.NullString{String: artifact.Digest.Encoded(), Valid: true},
		Size:        sql.NullInt64{Int64: artifact.Size, Valid: true},
		ContentType: sql.NullString{String: artifact.MediaType, Valid: true},
		FetchedAt:   sql.NullTime{Time: now, Valid: true},
	}
	if err := p.DB.UpsertArtifact(art); err != nil {
		return fmt.Errorf("upserting artifact: %w", err)
	}

	return nil
}

// ServeArtifact writes a CacheResult to an HTTP response.
func ServeArtifact(w http.ResponseWriter, result *CacheResult) {
	serveArtifact(w, http.MethodGet, result)
}

// ServeArtifactRequest writes a CacheResult to an HTTP response using request
// headers such as Range and If-Range.
func ServeArtifactRequest(w http.ResponseWriter, r *http.Request, result *CacheResult) {
	if r == nil {
		ServeArtifact(w, result)
		return
	}
	serveArtifactResponse(w, r.Method, r, result)
}

func serveArtifact(w http.ResponseWriter, method string, result *CacheResult) {
	serveArtifactResponse(w, method, nil, result)
}

func serveArtifactResponse(w http.ResponseWriter, method string, request *http.Request, result *CacheResult) {
	contentHash := ""
	if result.Artifact.Digest != "" {
		contentHash = result.Artifact.Digest.Encoded()
	}
	if result.RedirectURL != "" {
		if contentHash != "" {
			w.Header().Set(headerETag, `"`+contentHash+`"`)
		}
		w.Header().Set("Location", result.RedirectURL)
		w.WriteHeader(http.StatusFound)
		return
	}

	if result.Reader != nil {
		defer func() { _ = result.Reader.Close() }()
	}

	if result.Artifact.MediaType != "" {
		w.Header().Set(headerContentType, result.Artifact.MediaType)
	}
	if contentHash != "" {
		w.Header().Set(headerETag, `"`+contentHash+`"`)
	}

	var seeker io.Seeker
	if result.Reader != nil {
		seeker, _ = result.Reader.(io.Seeker)
	}
	if seeker != nil && result.Artifact.Size >= 0 {
		w.Header().Set("Accept-Ranges", "bytes")
		if request != nil && serveArtifactRange(w, request, result, seeker) {
			return
		}
	}

	if result.Artifact.Size > 0 || (method == http.MethodHead && result.Artifact.Size == 0) {
		w.Header().Set(headerContentLength, strconv.FormatInt(result.Artifact.Size, 10))
	}

	w.WriteHeader(http.StatusOK)
	if method != http.MethodHead && result.Reader != nil {
		copyArtifactBody(w, result.Reader, result.Artifact.Size, false)
	}
}

type parsedByteRange struct {
	start int64
	end   int64
}

func serveArtifactRange(w http.ResponseWriter, request *http.Request, result *CacheResult, seeker io.Seeker) bool {
	if request.Method != http.MethodGet {
		return false
	}
	ranges := request.Header.Values("Range")
	if len(ranges) != 1 || !ifRangeMatches(request, w.Header().Get(headerETag)) {
		return false
	}

	byteRange, valid, satisfiable := parseByteRange(ranges[0], result.Artifact.Size)
	if !valid {
		return false
	}
	if !satisfiable {
		w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", result.Artifact.Size))
		w.Header().Set(headerContentLength, "0")
		w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
		return true
	}

	if _, err := seeker.Seek(byteRange.start, io.SeekStart); err != nil {
		w.Header().Del(headerContentType)
		w.Header().Del(headerContentLength)
		w.Header().Del(headerETag)
		w.Header().Del("Accept-Ranges")
		http.Error(w, "failed to seek cached artifact", http.StatusInternalServerError)
		return true
	}

	length := byteRange.end - byteRange.start + 1
	w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", byteRange.start, byteRange.end, result.Artifact.Size))
	w.Header().Set(headerContentLength, strconv.FormatInt(length, 10))
	w.WriteHeader(http.StatusPartialContent)
	copyArtifactBody(w, result.Reader, length, true)
	return true
}

func ifRangeMatches(request *http.Request, etag string) bool {
	values := request.Header.Values("If-Range")
	if len(values) == 0 {
		return true
	}
	return len(values) == 1 && etag != "" && strings.TrimSpace(values[0]) == etag
}

func parseByteRange(value string, size int64) (parsedByteRange, bool, bool) {
	unit, spec, ok := strings.Cut(strings.TrimSpace(value), "=")
	if !ok || !strings.EqualFold(strings.TrimSpace(unit), "bytes") {
		return parsedByteRange{}, false, false
	}
	spec = strings.TrimSpace(spec)
	if spec == "" || strings.Contains(spec, ",") {
		return parsedByteRange{}, false, false
	}
	first, last, ok := strings.Cut(spec, "-")
	if !ok {
		return parsedByteRange{}, false, false
	}

	if first == "" {
		suffixLength, ok := parseRangeNumber(last)
		if !ok {
			return parsedByteRange{}, false, false
		}
		if size == 0 || suffixLength == 0 {
			return parsedByteRange{}, true, false
		}
		if suffixLength >= size {
			return parsedByteRange{start: 0, end: size - 1}, true, true
		}
		return parsedByteRange{start: size - suffixLength, end: size - 1}, true, true
	}

	start, ok := parseRangeNumber(first)
	if !ok {
		return parsedByteRange{}, false, false
	}
	if last == "" {
		if size == 0 || start >= size {
			return parsedByteRange{}, true, false
		}
		return parsedByteRange{start: start, end: size - 1}, true, true
	}
	end, ok := parseRangeNumber(last)
	if !ok || start > end {
		return parsedByteRange{}, false, false
	}
	if size == 0 || start >= size {
		return parsedByteRange{}, true, false
	}
	if end >= size {
		end = size - 1
	}
	return parsedByteRange{start: start, end: end}, true, true
}

func parseRangeNumber(value string) (int64, bool) {
	if value == "" {
		return 0, false
	}
	for _, digit := range value {
		if digit < '0' || digit > '9' {
			return 0, false
		}
	}
	number, err := strconv.ParseInt(value, 10, 64)
	return number, err == nil
}

func copyArtifactBody(w http.ResponseWriter, reader io.Reader, expectedSize int64, bounded bool) {
	buffer := artifactCopyBufferPool.Get().(*[]byte)
	defer artifactCopyBufferPool.Put(buffer)

	source := reader
	if bounded {
		source = io.LimitReader(reader, expectedSize)
	}
	// Hide optional ReaderFrom methods so io.CopyBuffer uses the pooled buffer.
	written, err := io.CopyBuffer(struct{ io.Writer }{w}, source, *buffer)
	if err != nil || (bounded && written != expectedSize) || (!bounded && expectedSize > 0 && written != expectedSize) {
		// Headers are already committed, so an error status is no longer
		// possible. Aborting leaves the response unterminated so clients discard
		// a truncated or unverified artifact.
		// net/http recovers ErrAbortHandler for this request and keeps the server
		// running; on HTTP/1.x it may close this connection as well.
		panic(http.ErrAbortHandler)
	}
}

// ProxyUpstream forwards a request to an upstream URL without caching.
// It copies the request, forwards specified headers, and streams the response back.
// forwardHeaders controls the request headers sent upstream. End-to-end response
// headers and trailers are relayed independently of that list.
func (p *Proxy) ProxyUpstream(w http.ResponseWriter, r *http.Request, upstreamURL string, forwardHeaders []string) {
	p.Logger.Debug("proxying to upstream", "url", upstreamURL)

	req, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, nil)
	if err != nil {
		http.Error(w, "failed to create request", http.StatusInternalServerError)
		return
	}

	// Copy request headers that affect content negotiation / caching
	for _, header := range forwardHeaders {
		if v := r.Header.Get(header); v != "" {
			req.Header.Set(header, v)
		}
	}

	resp, err := p.HTTPClient.Do(req)
	if err != nil {
		if writeUpstreamHeld(w, err) {
			return
		}
		p.Logger.Error("upstream request failed", "error", err)
		http.Error(w, "upstream request failed", http.StatusBadGateway)
		return
	}
	defer func() { _ = resp.Body.Close() }()

	p.relayResponse(w, r, resp, nil)
}

// ProxyFile forwards a file request, relaying end-to-end headers and trailers.
func (p *Proxy) ProxyFile(w http.ResponseWriter, r *http.Request, upstreamURL string) {
	req, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, nil)
	if err != nil {
		http.Error(w, "failed to create request", http.StatusInternalServerError)
		return
	}

	resp, err := p.HTTPClient.Do(req)
	if err != nil {
		if writeUpstreamHeld(w, err) {
			return
		}
		http.Error(w, "failed to fetch from upstream", http.StatusBadGateway)
		return
	}
	defer func() { _ = resp.Body.Close() }()

	p.relayResponse(w, r, resp, nil)
}

// JSONError writes a JSON error response.
func JSONError(w http.ResponseWriter, status int, message string) {
	w.Header().Set(headerContentType, contentTypeJSON)
	w.WriteHeader(status)
	_, _ = fmt.Fprintf(w, `{"error":%q}`, message)
}

// ErrUpstreamNotFound indicates the upstream returned 404.
var ErrUpstreamNotFound = fmt.Errorf("upstream: %w", fetch.ErrNotFound)

// upstreamNotFoundBodyLimit caps the part of an upstream 404 body kept in
// upstreamNotFoundError.
const upstreamNotFoundBodyLimit = 4 << 10

// upstreamNotFoundError is ErrUpstreamNotFound with the start of the
// upstream's 404 body, for handlers that relay the reason to clients.
type upstreamNotFoundError struct {
	body []byte
}

func (e *upstreamNotFoundError) Error() string { return ErrUpstreamNotFound.Error() }

func (e *upstreamNotFoundError) Unwrap() error { return ErrUpstreamNotFound }

// ErrArtifactBlocked indicates a pre-cache security scan blocked the artifact.
var ErrArtifactBlocked = errors.New("artifact blocked by security scan")

// ErrVersionDenied indicates an operator explicitly denied this package version.
var ErrVersionDenied = errors.New("package version blocked by denylist")

// ErrVersionNotFound indicates the requested version is not available through
// the proxy, including when cooldown or other filtering removed it.
var ErrVersionNotFound = errors.New("version not found")

// ErrPackageNotFound indicates the requested package metadata is not available
// through the proxy or upstream registry.
var ErrPackageNotFound = errors.New("package not found")

// serveArtifactError writes response for a failed fetch:
// 404 when upstream reports artifact missing, 403 when a security scan or
// denylist blocked the artifact, 502 otherwise.
func (p *Proxy) serveArtifactError(w http.ResponseWriter, err error, clientMsg string) {
	if errors.Is(err, ErrUpstreamNotFound) {
		http.Error(w, "not found", http.StatusNotFound)
		return
	}
	if errors.Is(err, ErrArtifactBlocked) || errors.Is(err, ErrVersionDenied) {
		JSONError(w, http.StatusForbidden, err.Error())
		return
	}
	if errors.Is(err, ErrHostedArtifactMissing) {
		http.Error(w, "not found", http.StatusNotFound)
		return
	}
	if errors.Is(err, ErrHostedArtifactUnavailable) {
		p.Logger.Error("failed to read pushed artifact", "error", err)
		http.Error(w, "pushed package is temporarily unavailable", http.StatusServiceUnavailable)
		return
	}
	if writeUpstreamHeld(w, err) {
		return
	}
	p.Logger.Error("failed to get artifact", "error", err)
	http.Error(w, clientMsg, http.StatusBadGateway)
}

// errStale304 is returned when upstream sends 304 but the cached file is missing.
var errStale304 = fmt.Errorf("upstream returned 304 but cached file is missing")

// metadataStoragePath builds a storage path for cached metadata.
func metadataStoragePath(ecosystem, cacheKey string) string {
	return "_metadata/" + ecosystem + "/" + cacheKey + "/metadata"
}

// FetchOrCacheMetadata fetches metadata from upstream with caching.
// On success it returns the raw response bytes and content type.
// If upstream fails and a cached copy exists, the cached version is returned.
// cacheKey is typically the package name but can include subpath components.
// Optional acceptHeaders specify the Accept header(s) to send; defaults to application/json.
func (p *Proxy) FetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL string, acceptHeaders ...string) ([]byte, string, error) {
	body, contentType, _, err := p.fetchOrCacheMetadata(ctx, ecosystem, cacheKey, upstreamURL, "", nil, acceptHeaders...)
	return body, contentType, err
}

// fetchOrCacheMetadata implements FetchOrCacheMetadata. acceptEncoding controls
// the upstream Accept-Encoding: an empty string leaves it unset so Go
// transparently decompresses (for direct callers that parse or rewrite the
// body); any non-empty value is sent verbatim, which disables Go's
// decompression so the wire bytes and their Content-Encoding are stored and
// replayed as sent. The ProxyCached path uses "identity" for signed indexes and
// "gzip" where both hops should stay compressed.
// validate, when supplied, runs before caching or serving a document. Validation
// failures follow the same stale-cache fallback path as upstream failures. It
// runs for every caller, including one that shares another caller's fetch, so
// it can also decode the document into request-local state; a caller that
// joined a fetch gets its own validation error directly. It must not modify the
// body, which joined callers share.
func (p *Proxy) fetchOrCacheMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, acceptEncoding string, validate func([]byte) error, acceptHeaders ...string) ([]byte, string, string, error) {
	accept := contentTypeJSON
	if len(acceptHeaders) > 0 && acceptHeaders[0] != "" {
		accept = acceptHeaders[0]
	}
	res := p.lookupMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate)
	return res.body, res.contentType, res.contentEncoding, res.err
}

// fetchMetadataDocument is FetchOrCacheMetadata for callers that rewrite the
// document and need to know which one they got: the result carries its
// digest along with the body.
func (p *Proxy) fetchMetadataDocument(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string) metadataResult {
	return p.lookupMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, "", nil)
}

// lookupMetadata implements fetchOrCacheMetadata and fetchMetadataDocument.
func (p *Proxy) lookupMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error) metadataResult {
	if containsPathTraversal(cacheKey) {
		return metadataResult{err: fmt.Errorf("invalid cache key: %q", cacheKey)}
	}

	// Serve from cache if within TTL (skip upstream entirely)
	if _, hit := p.cachedMetadataState(ctx, ecosystem, cacheKey, validate); hit != nil {
		metrics.RecordCacheHit(ecosystem)
		return *hit
	}
	p.recordMetadataCacheMiss(ecosystem)

	return p.coalescedMetadataMiss(ctx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate)
}

// coalescedMetadataMiss handles a metadata cache miss, sharing one upstream
// fetch among concurrent callers with the same key.
func (p *Proxy) coalescedMetadataMiss(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error) metadataResult {
	key := metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate != nil)
	res, shared := p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult {
		// The caller's lookup ran before it took the key, so a fetch that
		// finished in between has already refreshed the row. Recheck it
		// rather than fetching again, and revalidate against the row as it
		// is now.
		entry, hit := p.cachedMetadataState(fetchCtx, ecosystem, cacheKey, validate)
		if hit != nil {
			return *hit
		}
		return p.fetchMetadataFromUpstream(fetchCtx, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, validate, entry)
	})
	// The fetch ran the first caller's validate. A caller that joined it runs
	// its own on the shared bytes, since validate may also decode them for it.
	if shared && res.err == nil && validate != nil {
		if err := validate(res.body); err != nil {
			return metadataResult{err: err}
		}
	}
	return res
}

// cachedMetadataState reads the cache row for a metadata lookup. hit is set
// when the row is within TTL and its bytes are usable. Otherwise entry is the
// row to revalidate against with its ETag: nil when there is none, or when
// validate rejected the cached body, since revalidating an unusable body would
// keep it.
func (p *Proxy) cachedMetadataState(ctx context.Context, ecosystem, cacheKey string, validate func([]byte) error) (*database.MetadataCacheEntry, *metadataResult) {
	if !p.CacheMetadata || p.DB == nil {
		return nil, nil
	}
	entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey)
	if entry != nil && p.MetadataTTL > 0 && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) < p.MetadataTTL {
		data, ct, err := p.readCachedMetadata(ctx, entry, validate)
		if err == nil {
			hit := metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String}.describedBy(entry)
			return entry, &hit
		}
		if validate != nil {
			return nil, nil
		}
		// Cache file missing/unreadable, fall through to upstream
	}
	return entry, nil
}

// fetchMetadataFromUpstream fetches metadata after a cache miss, caches it,
// and falls back to the cached copy if upstream fails. entry is the row to
// revalidate against, or nil.
func (p *Proxy) fetchMetadataFromUpstream(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validate func([]byte) error, entry *database.MetadataCacheEntry) metadataResult {
	meta, err := p.fetchUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding)
	if errors.Is(err, errStale304) {
		// 304 but cached file is gone; retry without ETag
		meta, err = p.fetchUpstreamMetadata(ctx, upstreamURL, nil, accept, acceptEncoding)
	}
	if err == nil && validate != nil {
		err = validate(meta.body)
	}
	if err == nil {
		var digest string
		if p.CacheMetadata {
			digest = p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta)
		}
		return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding, digest: digest}
	}

	// Upstream failed -- fall back to cache if available
	if !p.CacheMetadata || entry == nil {
		return metadataResult{err: fmt.Errorf("upstream failed and no cached metadata: %w", err)}
	}

	p.Logger.Warn("upstream metadata fetch failed, checking cache",
		"ecosystem", ecosystem, "key", cacheKey, "error", err)

	// Re-read the row so the encoding describes the blob as it is now: a
	// concurrent refetch may have replaced both since entry was read above
	// (an identity blob swapped for a gzip one during rollout).
	entry = p.currentMetadataEntry(ecosystem, cacheKey, entry)

	data, ct, readErr := p.readCachedMetadata(ctx, entry, validate)
	if readErr != nil {
		return metadataResult{err: fmt.Errorf("upstream failed and cached metadata unusable (%v): %w", readErr, err)}
	}

	p.Logger.Info("serving metadata from cache",
		"ecosystem", ecosystem, "key", cacheKey)
	return metadataResult{body: data, contentType: ct, contentEncoding: entry.ContentEncoding.String, stale: true}.describedBy(entry)
}

// metadataResult is what a metadata lookup hands back. A shared fetch gives
// every waiter the same body, so callers must treat it as read-only.
type metadataResult struct {
	body            []byte
	contentType     string
	contentEncoding string
	// digest identifies body when the metadata cache stored it, and is
	// empty otherwise.
	digest string
	err    error
	// stale is set when the body is the cached copy served because upstream
	// failed.
	stale bool
}

// describedBy returns res with the digest that entry, the cache row its body
// was read for, records.
func (res metadataResult) describedBy(entry *database.MetadataCacheEntry) metadataResult {
	if entry.ContentDigest.Valid {
		res.digest = entry.ContentDigest.String
	}
	return res
}

// inflightMetadata is one metadata fetch that concurrent callers share. res is
// written before done closes and read only after, so the close is the handoff.
type inflightMetadata struct {
	done chan struct{}
	res  metadataResult

	// waiters counts callers that joined the fetch, guarded by metaMu. Tests
	// read it to know every caller has joined before the fetch finishes.
	waiters int
}

// metadataCoalesceKey identifies metadata requests that can share one fetch.
// Accept and Accept-Encoding are part of it because they change the bytes
// upstream returns: npm serves an abbreviated or a full document for the same
// package, and the cached-proxy paths ask for identity or gzip. Whether the
// caller validates is part of it so an unvalidated caller is never handed a
// result that skipped validation, or the reverse.
func metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, acceptEncoding string, validated bool) string {
	return strings.Join([]string{ecosystem, cacheKey, upstreamURL, accept, acceptEncoding, strconv.FormatBool(validated)}, "\x00")
}

// coalesceMetadata runs fetch at most once for concurrent callers sharing key
// and gives each the result, reporting whether this caller joined another's
// fetch rather than running it. It follows coalesceFetch with one difference:
// fetch runs on a context detached from the first caller's cancellation. A
// metadata fetch has no scan or mirror that depends on the caller aborting it,
// and CI jobs asking for the same Composer or npm metadata would otherwise all
// fail when the first of them disconnects. It stays bounded by the HTTP
// client's timeout. Waiters still leave when their own context ends.
func (p *Proxy) coalesceMetadata(ctx context.Context, key string, fetch func(context.Context) metadataResult) (metadataResult, bool) {
	p.metaMu.Lock()
	if p.inFlightMeta == nil {
		p.inFlightMeta = make(map[string]*inflightMetadata)
	}
	f, joined := p.inFlightMeta[key]
	if joined {
		f.waiters++
	} else {
		f = &inflightMetadata{done: make(chan struct{})}
		p.inFlightMeta[key] = f
	}
	p.metaMu.Unlock()

	if joined {
		select {
		case <-ctx.Done():
			return metadataResult{err: ctx.Err()}, true
		case <-f.done:
			return f.res, true
		}
	}

	// Set before running so a panicking fetch leaves waiters with an error
	// rather than an empty body.
	f.res = metadataResult{err: errSharedFetchAbandoned}
	defer func() {
		p.metaMu.Lock()
		delete(p.inFlightMeta, key)
		p.metaMu.Unlock()
		close(f.done)
	}()
	f.res = fetch(context.WithoutCancel(ctx))
	return f.res, false
}

func (p *Proxy) readCachedMetadata(ctx context.Context, entry *database.MetadataCacheEntry, validate func([]byte) error) ([]byte, string, error) {
	cached, err := p.Storage.Open(ctx, entry.StoragePath)
	if err != nil {
		return nil, "", err
	}
	defer func() { _ = cached.Close() }()
	data, err := p.ReadMetadata(cached)
	if err == nil && validate != nil {
		err = validate(data)
	}
	if err != nil {
		return nil, "", err
	}
	ct := contentTypeJSON
	if entry.ContentType.Valid {
		ct = entry.ContentType.String
	}
	return data, ct, nil
}

func (p *Proxy) recordMetadataCacheMiss(ecosystem string) {
	if p.CacheMetadata {
		metrics.RecordCacheMiss(ecosystem)
	}
}

// upstreamMetadata holds a fetched metadata response in upstream byte form.
type upstreamMetadata struct {
	body            []byte
	contentType     string
	contentEncoding string
	etag            string
	lastModified    time.Time
}

// fetchUpstreamMetadata fetches metadata from upstream, using ETag for conditional revalidation.
// When acceptEncoding is non-empty it is sent as the Accept-Encoding header, which disables Go's
// transparent decompression (it only applies when the transport adds the header itself), so the
// returned bytes are exactly what the upstream sent and any Content-Encoding it applied is reported
// alongside for the caller to store and replay. An empty acceptEncoding leaves Go to negotiate and
// decompress transparently.
func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, entry *database.MetadataCacheEntry, accept, acceptEncoding string) (*upstreamMetadata, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil)
	if err != nil {
		return nil, fmt.Errorf("creating request: %w", err)
	}
	req.Header.Set("Accept", accept)
	if acceptEncoding != "" {
		req.Header.Set(headerAcceptEncoding, acceptEncoding)
	}

	if entry != nil && entry.ETag.Valid {
		req.Header.Set("If-None-Match", entry.ETag.String)
	} else if entry != nil && entry.LastModified.Valid {
		req.Header.Set("If-Modified-Since", entry.LastModified.Time.UTC().Format(http.TimeFormat))
	}

	resp, err := p.HTTPClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("fetching metadata: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()

	// 304 Not Modified -- our cached copy is still good
	if resp.StatusCode == http.StatusNotModified && entry != nil {
		cached, readErr := p.Storage.Open(ctx, entry.StoragePath)
		if readErr != nil {
			return nil, errStale304
		}
		defer func() { _ = cached.Close() }()
		data, readErr := p.ReadMetadata(cached)
		if readErr != nil {
			return nil, errStale304
		}
		meta := &upstreamMetadata{body: data, contentType: contentTypeJSON, etag: entry.ETag.String}
		if entry.ContentType.Valid {
			meta.contentType = entry.ContentType.String
		}
		if entry.ContentEncoding.Valid {
			meta.contentEncoding = entry.ContentEncoding.String
		}
		if entry.LastModified.Valid {
			meta.lastModified = entry.LastModified.Time
		}
		return meta, nil
	}

	if resp.StatusCode == http.StatusNotFound {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, upstreamNotFoundBodyLimit))
		return nil, &upstreamNotFoundError{body: body}
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("upstream returned %d", resp.StatusCode)
	}

	body, err := p.ReadMetadata(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("reading response: %w", err)
	}

	meta := &upstreamMetadata{
		body:            body,
		contentType:     resp.Header.Get(headerContentType),
		contentEncoding: resp.Header.Get(headerContentEncoding),
		etag:            resp.Header.Get(headerETag),
	}
	if meta.contentType == "" {
		meta.contentType = contentTypeJSON
	}
	if lm := resp.Header.Get(headerLastModified); lm != "" {
		meta.lastModified, _ = http.ParseTime(lm)
	}
	return meta, nil
}

// cacheMetadataBlob stores metadata bytes in storage and updates the
// database. It returns the digest it recorded for the bytes, or "" when they
// were not cached.
func (p *Proxy) cacheMetadataBlob(ctx context.Context, ecosystem, cacheKey, storagePath string, meta *upstreamMetadata) string {
	if p.DB == nil || p.Storage == nil {
		return ""
	}

	size, hash, err := p.Storage.Store(ctx, storagePath, bytes.NewReader(meta.body))
	if err != nil {
		p.Logger.Warn("failed to cache metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err)
		return ""
	}
	var digest string
	if hash != "" {
		digest = "sha256:" + hash
	}

	err = p.DB.UpsertMetadataCache(&database.MetadataCacheEntry{
		Ecosystem:       ecosystem,
		Name:            cacheKey,
		StoragePath:     storagePath,
		ETag:            sql.NullString{String: meta.etag, Valid: meta.etag != ""},
		ContentType:     sql.NullString{String: meta.contentType, Valid: meta.contentType != ""},
		ContentEncoding: sql.NullString{String: meta.contentEncoding, Valid: meta.contentEncoding != ""},
		// The digest identifies the stored bytes, so a rewrite cached for them
		// can be found without reading them back (see storedRewrite).
		ContentDigest: sql.NullString{String: digest, Valid: digest != ""},
		Size:          sql.NullInt64{Int64: size, Valid: true},
		LastModified:  sql.NullTime{Time: meta.lastModified, Valid: !meta.lastModified.IsZero()},
		FetchedAt:     sql.NullTime{Time: time.Now(), Valid: true},
	})
	if err != nil {
		// The blob is written but the row describing it is not, so a later
		// TTL hit or stale fallback would serve these bytes with the previous
		// row's encoding. Drop the blob so row and bytes can never disagree;
		// the next request refetches instead.
		p.Logger.Warn("failed to record cached metadata, discarding blob", "ecosystem", ecosystem, "key", cacheKey, "error", err)
		if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil {
			p.Logger.Warn("failed to discard metadata blob", "ecosystem", ecosystem, "key", cacheKey, "error", delErr)
		}
		return ""
	}
	return digest
}

// currentMetadataEntry re-reads the metadata cache row and returns it, or
// fallback when the row cannot be read. Used before serving a stored blob so
// its encoding comes from the row as it is now rather than from a snapshot
// taken before the upstream fetch.
func (p *Proxy) currentMetadataEntry(ecosystem, cacheKey string, fallback *database.MetadataCacheEntry) *database.MetadataCacheEntry {
	if fresh, err := p.DB.GetMetadataCache(ecosystem, cacheKey); err == nil && fresh != nil {
		return fresh
	}
	return fallback
}

// cachedMeta holds cache validators and freshness state from a metadata cache entry.
type cachedMeta struct {
	etag         string
	lastModified time.Time
	stale        bool
}

// lookupCachedMeta retrieves cache validators for a metadata entry.
func (p *Proxy) lookupCachedMeta(ecosystem, cacheKey string) cachedMeta {
	if p.DB == nil {
		return cachedMeta{}
	}
	entry, err := p.DB.GetMetadataCache(ecosystem, cacheKey)
	if err != nil || entry == nil {
		return cachedMeta{}
	}
	var cm cachedMeta
	if entry.ETag.Valid {
		cm.etag = entry.ETag.String
	}
	if entry.LastModified.Valid {
		cm.lastModified = entry.LastModified.Time
	}
	// If FetchedAt is older than TTL, upstream must have failed and
	// we served from stale cache (successful fetches update FetchedAt).
	if p.MetadataTTL > 0 && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) > p.MetadataTTL {
		cm.stale = true
	}
	return cm
}

// ProxyCached fetches metadata from upstream (with optional caching for offline fallback)
// and writes it to the response. Optional acceptHeaders specify the Accept header to send.
// When metadata caching is disabled, the response is streamed directly to avoid buffering
// large metadata responses (e.g. npm packages with many versions) in memory.
func (p *Proxy) ProxyCached(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey string, acceptHeaders ...string) {
	p.proxyCachedWithEncoding(w, r, upstreamURL, ecosystem, cacheKey, "identity", acceptHeaders...)
}

// proxyCachedWithEncoding is ProxyCached with an explicit upstream Accept-Encoding.
// "identity" preserves signed index bytes (the default); "gzip" keeps both hops
// compressed for large, non-hash-pinned metadata whose clients decode gzip
// (Homebrew API and Conda repodata). The stored bytes and Content-Encoding are
// replayed verbatim either way.
func (p *Proxy) proxyCachedWithEncoding(w http.ResponseWriter, r *http.Request, upstreamURL, ecosystem, cacheKey, acceptEncoding string, acceptHeaders ...string) {
	if !p.CacheMetadata {
		// Stream directly without buffering when caching is off.
		p.proxyMetadataStream(w, r, upstreamURL, acceptEncoding, acceptHeaders...)
		return
	}

	body, contentType, contentEncoding, err := p.fetchOrCacheMetadata(r.Context(), ecosystem, cacheKey, upstreamURL, acceptEncoding, nil, acceptHeaders...)
	if err != nil {
		if errors.Is(err, ErrUpstreamNotFound) {
			http.Error(w, "not found", http.StatusNotFound)
			return
		}
		if writeUpstreamHeld(w, err) {
			return
		}
		p.Logger.Error("metadata fetch failed", "error", err)
		http.Error(w, "failed to fetch from upstream", http.StatusBadGateway)
		return
	}

	p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, contentEncoding)
}

// writeMetadataCachedResponse writes a cached metadata response and handles
// conditional request headers using metadata cache validators.
func (p *Proxy) writeMetadataCachedResponse(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType string) {
	p.writeMetadataCachedResponseWithEncoding(w, r, ecosystem, cacheKey, body, contentType, "")
}

// writeMetadataCachedResponseWithEncoding is writeMetadataCachedResponse with
// an explicit Content-Encoding. contentEncoding must describe the body being
// written; it is passed in rather than re-read from the cache row, which is
// missing or stale when the metadata cache write failed and would otherwise
// mislabel the bytes.
func (p *Proxy) writeMetadataCachedResponseWithEncoding(w http.ResponseWriter, r *http.Request, ecosystem, cacheKey string, body []byte, contentType, contentEncoding string) {
	cm := p.lookupCachedMeta(ecosystem, cacheKey)

	if cm.etag != "" {
		w.Header().Set(headerETag, cm.etag)
	}
	if !cm.lastModified.IsZero() {
		w.Header().Set(headerLastModified, cm.lastModified.UTC().Format(http.TimeFormat))
	}
	if ifNoneMatchHits(r.Header.Get("If-None-Match"), cm.etag) {
		w.WriteHeader(http.StatusNotModified)
		return
	}
	if !cm.lastModified.IsZero() {
		if ims := r.Header.Get("If-Modified-Since"); ims != "" {
			if t, err := http.ParseTime(ims); err == nil && !cm.lastModified.After(t) {
				w.WriteHeader(http.StatusNotModified)
				return
			}
		}
	}

	w.Header().Set(headerContentType, contentType)
	w.Header().Set(headerContentLength, strconv.Itoa(len(body)))
	if contentEncoding != "" {
		w.Header().Set(headerContentEncoding, contentEncoding)
	}
	if cm.stale {
		w.Header().Set("Warning", `110 - "Response is Stale"`)
	}
	w.WriteHeader(http.StatusOK)
	if r.Method != http.MethodHead {
		_, _ = w.Write(body)
	}
}

// proxyMetadataStream forwards an upstream metadata response by streaming it to the client
// without buffering the full body in memory.
func (p *Proxy) proxyMetadataStream(w http.ResponseWriter, r *http.Request, upstreamURL, acceptEncoding string, acceptHeaders ...string) {
	req, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, nil)
	if err != nil {
		http.Error(w, "failed to create request", http.StatusInternalServerError)
		return
	}

	accept := contentTypeJSON
	if len(acceptHeaders) > 0 && acceptHeaders[0] != "" {
		accept = acceptHeaders[0]
	}
	req.Header.Set("Accept", accept)
	// Set Accept-Encoding explicitly (identity, or gzip for compressible
	// verbatim metadata) so Go does not transparently decompress and strip the
	// Content-Encoding of the bytes we forward, regardless of what the client
	// negotiated. An empty value leaves the header unset, as in
	// fetchUpstreamMetadata.
	if acceptEncoding != "" {
		req.Header.Set(headerAcceptEncoding, acceptEncoding)
	}

	for _, header := range []string{"If-Modified-Since", "If-None-Match"} {
		if v := r.Header.Get(header); v != "" {
			req.Header.Set(header, v)
		}
	}

	resp, err := p.HTTPClient.Do(req)
	if err != nil {
		if writeUpstreamHeld(w, err) {
			return
		}
		http.Error(w, "failed to fetch from upstream", http.StatusBadGateway)
		return
	}
	defer func() { _ = resp.Body.Close() }()

	p.relayResponse(w, r, resp, func(dst, src http.Header) {
		for _, header := range []string{headerContentType, headerContentLength, headerContentEncoding, headerLastModified, headerETag} {
			if v := src.Get(header); v != "" {
				dst.Set(header, v)
			}
		}
	})
}

// GetOrFetchArtifactFromURL retrieves an artifact from cache or fetches from a specific URL.
// This is useful for registries where download URLs are determined from metadata.
func (p *Proxy) GetOrFetchArtifactFromURL(ctx context.Context, ecosystem, name, version, filename, downloadURL string) (*CacheResult, error) {
	return p.getOrFetchArtifactFromURL(ctx, ecosystem, name, version, filename, downloadURL, nil, "", "")
}

// GetOrFetchArtifactFromURLWithHeaders retrieves an artifact from cache or fetches from a URL
// with additional request-specific HTTP headers.
func (p *Proxy) GetOrFetchArtifactFromURLWithHeaders(ctx context.Context, ecosystem, name, version, filename, downloadURL string, headers http.Header) (*CacheResult, error) {
	return p.getOrFetchArtifactFromURL(ctx, ecosystem, name, version, filename, downloadURL, headers, "", "")
}

// GetOrFetchArtifactFromURLWithDigest retrieves an artifact and verifies its
// SHA-256 digest before adding a newly fetched response to the cache.
// Non-sha256 digests are proxied without verification.
func (p *Proxy) GetOrFetchArtifactFromURLWithDigest(ctx context.Context, ecosystem, name, version, filename, downloadURL, digest string) (*CacheResult, error) {
	upstreamHash, _ := strings.CutPrefix(digest, "sha256:")
	if upstreamHash == digest {
		upstreamHash = ""
	}
	return p.getOrFetchArtifactFromURL(ctx, ecosystem, name, version, filename, downloadURL, nil, upstreamHash, "")
}

// GetOrFetchArtifactFromURLWithIntegrity retrieves an artifact and verifies a
// newly fetched response against declared, the integrity the registry's
// metadata declares, before serving or caching it: a Subresource Integrity
// string, or a single algorithm-prefixed hex checksum such as "sha1-<hex>".
// declared is recorded with the cached version, so cache hits are verified
// against it as well. An empty declared fetches without verification.
func (p *Proxy) GetOrFetchArtifactFromURLWithIntegrity(ctx context.Context, ecosystem, name, version, filename, downloadURL, declared string) (*CacheResult, error) {
	return p.getOrFetchArtifactFromURL(ctx, ecosystem, name, version, filename, downloadURL, nil, "", declared)
}

func (p *Proxy) getOrFetchArtifactFromURL(ctx context.Context, ecosystem, name, version, filename, downloadURL string, headers http.Header, upstreamHash, declared string) (*CacheResult, error) {
	pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version)
	if err != nil {
		return nil, err
	}
	return p.getOrFetchArtifactFromURLWithCachePURLs(
		ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash, declared,
	)
}

func (p *Proxy) getOrFetchArtifactFromURLWithCachePURLs(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash, declared string) (*CacheResult, error) {
	// Some protocols use qualified cache keys. Check their package identity too.
	if p.versionDenied(ecosystem, name, version) {
		return nil, fmt.Errorf("%w: %s", ErrVersionDenied, canonicalVersionPURL(ecosystem, name, version))
	}
	if p.StreamArtifacts {
		return p.streamFromUpstream(ctx, ecosystem, name, version, filename, versionPURL, downloadURL, upstreamHash, declared,
			func(fetchCtx context.Context) (*fetch.Artifact, error) {
				return p.Fetcher.FetchWithHeaders(fetchCtx, downloadURL, headers)
			})
	}
	if cached, err := p.getCachedArtifactWithUpstreamHash(ctx, pkgPURL, versionPURL, filename, upstreamHash); err != nil {
		return nil, err
	} else if cached != nil {
		return cached, nil
	}
	metrics.RecordCacheMiss(ecosystem)
	return p.coalescedFetchFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash, declared)
}

// coalescedFetchFromURL fetches an artifact the cache could not serve, sharing
// the fetch with concurrent callers. The caller running it discards a stale
// entry under the key, where it cannot delete a fetch that just replaced it.
func (p *Proxy) coalescedFetchFromURL(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash, declared string) (*CacheResult, error) {
	key := artifactCoalesceKey(versionPURL, filename, downloadURL, upstreamHash, declared)
	recheck := func() (artifacts.Artifact, string, bool) {
		return p.cachedArtifactRecord(pkgPURL, versionPURL, filename, upstreamHash)
	}
	return p.coalesceFetch(ctx, key, recheck, func(fetchCtx context.Context) (artifacts.Artifact, string, error) {
		p.discardStaleArtifact(pkgPURL, versionPURL, filename, upstreamHash)
		return p.fetchAndCacheFromURL(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers, upstreamHash, declared)
	})
}

// getCachedArtifactWithUpstreamHash returns a cached artifact whose recorded
// content hash matches the checksum the upstream currently declares for it.
// This detects an upstream re-publishing under the same version, which the
// stream integrity check in checkCache cannot: that check only verifies the
// stored blob against the hash recorded when it was cached. A stale entry is
// a miss and is left in place: the fetch that replaces it discards it under
// the coalescing key.
func (p *Proxy) getCachedArtifactWithUpstreamHash(ctx context.Context, pkgPURL, versionPURL, filename, upstreamHash string) (*CacheResult, error) {
	cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename)
	if err != nil || cached == nil {
		return cached, err
	}
	if artifactHashMatches(cached.Artifact.Digest.Encoded(), upstreamHash) {
		return cached, nil
	}
	if cached.Reader != nil {
		_ = cached.Reader.Close()
	}
	return nil, nil
}

// discardStaleArtifact clears the cached entry when its digest disagrees with
// upstreamHash, queueing its object for deletion. It runs under the coalescing
// key, after the recheck, so an entry a previous fetch refreshed is kept.
func (p *Proxy) discardStaleArtifact(pkgPURL, versionPURL, filename, upstreamHash string) {
	record, err := p.DB.GetCachedArtifact(pkgPURL, versionPURL, filename)
	if err != nil {
		p.Logger.Warn("failed to read cache record before refetch",
			"purl", versionPURL, "filename", filename, "error", err)
		return
	}
	if record == nil || artifactHashMatches(record.Artifact.Digest.Encoded(), upstreamHash) {
		return
	}
	p.Logger.Warn("cached artifact hash disagrees with upstream metadata, discarding",
		"purl", versionPURL, "filename", filename, "cached", record.Artifact.Digest.Encoded(), "upstream", upstreamHash)
	if err := p.DB.DiscardArtifact(versionPURL, filename, record.StoragePath); err != nil {
		p.Logger.Warn("failed to clear artifact cache record", "purl", versionPURL, "filename", filename, "error", err)
	}
}

func (p *Proxy) fetchAndCacheFromURL(ctx context.Context, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL string, headers http.Header, upstreamHash, declared string) (artifacts.Artifact, string, error) {
	if err := p.RateLimits.Held(downloadURL); err != nil {
		return artifacts.Artifact{}, "", err
	}
	p.Logger.Info("fetching from upstream",
		"ecosystem", ecosystem, "name", name, "version", version, "url", downloadURL)

	fetchStart := time.Now()
	artifact, err := p.Fetcher.FetchWithHeaders(ctx, downloadURL, headers)
	metrics.RecordUpstreamFetch(ecosystem, time.Since(fetchStart))
	if err != nil {
		metrics.RecordUpstreamError(ecosystem, "fetch_failed", metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: downloadURL, Err: err})
		if errors.Is(err, fetch.ErrNotFound) {
			return artifacts.Artifact{}, "", ErrUpstreamNotFound
		}
		return artifacts.Artifact{}, "", fmt.Errorf("fetching from upstream: %w", err)
	}

	return p.storeArtifact(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, upstreamHash, declared, artifact)
}

// streamFromUpstream fetches an artifact when StreamArtifacts is set and hands its
// body to the caller without storing it.
//
// With upstreamHash or the declared integrity set the body is verified as it
// streams. The size is left
// unknown so the response goes out chunked: a mismatch only shows at EOF,
// after the bytes have been sent, and an unterminated chunked response is the
// only way left to make the client reject them (see serveArtifact).
func (p *Proxy) streamFromUpstream(ctx context.Context, ecosystem, name, version, filename, versionPURL, upstreamURL, upstreamHash, declared string, fetchArtifact func(context.Context) (*fetch.Artifact, error)) (*CacheResult, error) {
	if err := p.RateLimits.Held(upstreamURL); err != nil {
		return nil, err
	}
	p.Logger.Info("streaming from upstream",
		"ecosystem", ecosystem, "name", name, "version", version, "url", upstreamURL)

	fetchStart := time.Now()
	artifact, err := fetchArtifact(ctx)
	metrics.RecordUpstreamFetch(ecosystem, time.Since(fetchStart))
	if err != nil {
		metrics.RecordUpstreamError(ecosystem, "fetch_failed", metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: upstreamURL, Err: err})
		if errors.Is(err, fetch.ErrNotFound) {
			return nil, ErrUpstreamNotFound
		}
		return nil, fmt.Errorf("fetching from upstream: %w", err)
	}

	body := &streamErrorLogger{
		ReadCloser: artifact.Body,
		onError: func(read int64, err error) {
			p.Logger.Warn("streaming artifact from upstream failed",
				"purl", versionPURL, "filename", filename, "url", upstreamURL, "bytes", read, "error", err)
			metrics.RecordUpstreamError(ecosystem, "stream_failed", metrics.FailureDetail{
				PURL: versionPURL, Filename: filename, URL: upstreamURL, Err: err})
		},
	}
	result := &CacheResult{
		Reader: body,
		Artifact: artifacts.Artifact{
			PURL:      versionPURL,
			Size:      artifact.Size,
			Filename:  filename,
			MediaType: artifact.ContentType,
		},
	}
	if upstreamHash == "" && declared == "" {
		return result, nil
	}

	hash := strings.ToLower(upstreamHash)
	checks, err := newIntegrityChecks(hash, declared)
	if err != nil {
		_ = artifact.Body.Close()
		return nil, fmt.Errorf("parsing upstream digest: %w", err)
	}
	result.Reader, err = checks.wrapFailOnMismatch(body, func(reason string) {
		p.Logger.Error("streamed artifact failed integrity check",
			"purl", versionPURL, "filename", filename, "url", upstreamURL, "reason", reason)
		metrics.RecordIntegrityFailure(purl.NormalizeEcosystem(ecosystem), metrics.FailureDetail{
			PURL: versionPURL, Filename: filename, URL: upstreamURL, Err: errors.New(reason)})
	})
	if err != nil {
		_ = artifact.Body.Close()
		return nil, err
	}
	if hash != "" {
		result.Artifact.Digest = digest.Digest("sha256:" + hash)
	}
	result.Artifact.Size = -1
	return result, nil
}

// streamErrorLogger reports the first read error of a streamed upstream body.
// The error itself still reaches serveArtifact, which aborts the response.
type streamErrorLogger struct {
	io.ReadCloser
	onError func(read int64, err error)
	read    int64
	logged  bool
}

func (r *streamErrorLogger) Read(p []byte) (int, error) {
	n, err := r.ReadCloser.Read(p)
	r.read += int64(n)
	if err != nil && err != io.EOF && !r.logged {
		r.logged = true
		r.onError(r.read, err)
	}
	return n, err
}

// ErrArtifactDigestMismatch indicates that fetched bytes did not match the
// checksum the upstream declared and were not recorded in the cache database.
var ErrArtifactDigestMismatch = errors.New("artifact digest mismatch")

func artifactHashMatches(got, expected string) bool {
	return expected == "" || strings.EqualFold(got, expected)
}