pixa/internal/imageproxy/imageproxy.go

// Package imageproxy fetches, transforms, caches and serves images. Its
// Proxy serves an image from the cache when it can, and otherwise makes it
// from the cached or fetched source image.
package imageproxy

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"runtime/debug"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/getsentry/sentry-go"
	"github.com/go-chi/chi/v5/middleware"
	"github.com/prometheus/client_golang/prometheus"
	"golang.org/x/sync/singleflight"
	"sneak.berlin/go/pixa/internal/httpfetcher"
	"sneak.berlin/go/pixa/internal/imageprocessor"
	"sneak.berlin/go/pixa/internal/imagerequest"
	"sneak.berlin/go/pixa/internal/imgcache"
	"sneak.berlin/go/pixa/internal/magic"
)

// Proxy orchestrates the cache, fetcher, and processor.
type Proxy struct {
	cache           *imgcache.Cache
	fetcher         httpfetcher.Fetcher
	processor       *imageprocessor.ImageProcessor
	log             *slog.Logger
	maxResponseSize int64
	// variantsInProgress lets the requests that miss the same variant at the
	// same time share one fetch and one transcode.
	variantsInProgress singleflight.Group

	// The metrics of upstream fetches and transforms, made and registered by
	// New. They are kept in memory, so updating one waits for nothing.
	upstreamFetches        prometheus.Counter
	upstreamFetchBytes     prometheus.Counter
	upstreamFetchDurations prometheus.Histogram
	upstreamFetchErrors    prometheus.Counter
	transforms             prometheus.Counter
	transformDurations     prometheus.Histogram
	transformsInProgress   prometheus.Gauge
}

// Config holds the proxy's configuration.
type Config struct {
	// Cache is the cache instance
	Cache *imgcache.Cache
	// Fetcher fetches source images from upstream hosts
	Fetcher httpfetcher.Fetcher
	// Processor converts and resizes the images
	Processor *imageprocessor.ImageProcessor
	// MaxResponseSize is the largest source accepted, in bytes; a cached
	// source larger than this is fetched again
	MaxResponseSize int64
	// Logger for logging
	Logger *slog.Logger
}

// errCacheRequired is returned when the proxy is built without a cache.
var errCacheRequired = errors.New("cache is required")

// New creates a new image proxy and registers its metrics on
// metricsRegistry. pixad passes Prometheus' default registry, the one
// /metrics serves. It fails when metricsRegistry already holds metrics of
// the same names.
func New(
	cfg *Config, metricsRegistry prometheus.Registerer,
) (*Proxy, error) {
	if cfg.Cache == nil {
		return nil, errCacheRequired
	}

	log := cfg.Logger
	if log == nil {
		log = slog.Default()
	}

	// JPEG XL is the default output format, so pixad does not start
	// without it.
	err := imageprocessor.CheckJPEGXLSupport()
	if err != nil {
		return nil, err
	}

	p := &Proxy{
		cache:           cfg.Cache,
		fetcher:         cfg.Fetcher,
		processor:       cfg.Processor,
		log:             log,
		maxResponseSize: cfg.MaxResponseSize,
		upstreamFetches: prometheus.NewCounter(prometheus.CounterOpts{
			Name: "pixa_upstream_fetches_total",
			Help: "Fetches of a source image from upstream that read any bytes.",
		}),
		upstreamFetchBytes: prometheus.NewCounter(prometheus.CounterOpts{
			Name: "pixa_upstream_fetch_bytes_total",
			Help: "Bytes read from upstream by fetches of source images.",
		}),
		upstreamFetchDurations: prometheus.NewHistogram(prometheus.HistogramOpts{
			Name: "pixa_upstream_fetch_duration_seconds",
			Help: "Duration of the fetches pixa_upstream_fetches_total counts, " +
				"until the response's last byte.",
		}),
		upstreamFetchErrors: prometheus.NewCounter(prometheus.CounterOpts{
			Name: "pixa_upstream_fetch_errors_total",
			Help: "Fetches of a source image from upstream that failed, " +
				"before or while reading the response.",
		}),
		transforms: prometheus.NewCounter(prometheus.CounterOpts{
			Name: "pixa_transforms_total",
			Help: "Images the image processor transformed.",
		}),
		transformDurations: prometheus.NewHistogram(prometheus.HistogramOpts{
			Name: "pixa_transform_duration_seconds",
			Help: "Duration of the transforms pixa_transforms_total counts, " +
				"waiting for a processing slot included.",
		}),
		transformsInProgress: prometheus.NewGauge(prometheus.GaugeOpts{
			Name: "pixa_transforms_in_progress",
			Help: "Images being transformed or waiting for a processing slot.",
		}),
	}

	for _, metric := range []prometheus.Collector{
		p.upstreamFetches, p.upstreamFetchBytes, p.upstreamFetchDurations,
		p.upstreamFetchErrors, p.transforms, p.transformDurations,
		p.transformsInProgress,
	} {
		err = metricsRegistry.Register(metric)
		if err != nil {
			return nil, fmt.Errorf("failed to register the proxy's metrics: %w", err)
		}
	}

	return p, nil
}

// ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
var ErrNegativeCached = errors.New("request is in negative cache (recently failed)")

// ErrProcessingTimedOut is returned when processing an image stopped because
// the downstream_timeout of the request doing it ended, to that request and to
// every request that waited for its image.
var ErrProcessingTimedOut = errors.New(
	"image processing stopped at the deadline of the request doing it")

// ImageResponse represents a processed image ready to serve
type ImageResponse struct {
	// Content is the image data reader
	Content io.ReadCloser
	// ContentLength is the size in bytes (-1 if unknown)
	ContentLength int64
	// ContentType is the MIME type of the response
	ContentType string
	// ETag is the entity tag for caching
	ETag string
	// CacheStatus indicates HIT or MISS
	CacheStatus CacheStatus
	// FetchedBytes is the number of bytes fetched from upstream (0 if cache hit)
	FetchedBytes int64
}

// CacheStatus indicates whether the response was served from cache.
type CacheStatus string

// Cache status values for response headers.
const (
	CacheHit  CacheStatus = "HIT"
	CacheMiss CacheStatus = "MISS"
)

// Get retrieves a processed image, fetching and processing if necessary.
// While its source URL is in the negative cache, a cached variant is still
// served and a variant not in the cache is still made from a cached source;
// only a request that needs the source fetched is refused.
//
// A cache hit opens the variant file and makes no database call: the cache
// hands its access time to the goroutine that writes access times, without
// waiting for it, and its hit is counted in memory. A variant whose source
// was last checked with upstream CacheTTL or more ago is a miss, which checks
// the source first (see processFromSourceOrFetch). A variant file that
// cannot be served, such as one whose first bytes are not those of an output
// format, is a miss, and the variant is made again.
func (p *Proxy) Get(
	ctx context.Context, req *imagerequest.Request,
) (*ImageResponse, error) {
	cacheKey := imagerequest.CacheKey(req)

	cached, err := p.cache.Variant(cacheKey)
	if err == nil {
		p.cache.IncrementStats(true)

		return &ImageResponse{
			Content:       cached.Content,
			ContentLength: cached.Size,
			ContentType:   cached.ContentType,
			CacheStatus:   CacheHit,
			ETag:          formatETag(cacheKey, cached.CheckedAt),
		}, nil
	}

	if !errors.Is(err, imgcache.ErrNotFound) && !errors.Is(err, imgcache.ErrStale) {
		p.log.Warn("cached variant not served, making it again",
			"cache_key", cacheKey, "error", err)
	}

	// Cache miss - get the variant, processed once for all the requests that
	// miss it at the same time, then count this request's miss, also when it
	// failed or the request context has ended meanwhile
	response, err := p.processOrWait(ctx, req)

	p.cache.IncrementStats(false)

	if err != nil {
		return nil, err
	}

	response.CacheStatus = CacheMiss

	return response, nil
}

// WaitForProcessing waits until no image is being processed, or until ctx
// ends, and returns how many images were still being processed then.
func (p *Proxy) WaitForProcessing(ctx context.Context) int {
	return p.processor.WaitForProcessing(ctx)
}

// errPanicked is returned when processing a variant panicked.
var errPanicked = errors.New("panic while processing image")

// processOrWait returns the variant req asks for. The first of the requests
// that miss a variant at the same time processes it, and singleflight hands
// its result, or its error, to the others: they fetch nothing, read no source
// and take no upstream connection or processing slot. The processing ignores
// the first request's cancellation, so the others are still served if that
// client goes away, but keeps its deadline.
func (p *Proxy) processOrWait(
	ctx context.Context, req *imagerequest.Request,
) (*ImageResponse, error) {
	// A request that has already ended starts no processing
	if ctx.Err() != nil {
		return nil, ctx.Err()
	}

	cacheKey := imagerequest.CacheKey(req)

	// Closed when this request's own function runs, which singleflight does
	// only when no other request is processing the variant
	processing := make(chan struct{})

	results := p.variantsInProgress.DoChan(string(cacheKey),
		func() (_ any, err error) {
			close(processing)

			// singleflight would raise a panic again in a goroutine of its
			// own, where no handler recovers it, and stop pixad. It is
			// reported through the Sentry hub that the Sentry middleware puts
			// on the request's context when sentry_dsn is set.
			defer func() {
				recovered := recover()
				if recovered != nil {
					p.log.Error("panic while processing image",
						"host", req.Source.Host, "path", req.Source.Path,
						"panic", recovered, "stack", string(debug.Stack()))

					if hub := sentry.GetHubFromContext(ctx); hub != nil {
						hub.RecoverWithContext(ctx, recovered)
					}

					err = fmt.Errorf("%w: %v", errPanicked, recovered)
				}
			}()

			processingCtx, cancel := withoutCancelKeepingDeadline(ctx)
			defer cancel()

			return p.processFromSourceOrFetch(processingCtx, req, cacheKey)
		})

	var result singleflight.Result

	select {
	case result = <-results:
	case <-ctx.Done():
		select {
		case <-processing:
			// This request is processing the variant: it waits for the
			// result, as every request did before misses were shared
			result = <-results
		default:
			// Another request is processing the variant, or this request's
			// function has not started yet; the processing goes on without it
			return nil, ctx.Err()
		}
	}

	if result.Err != nil {
		return nil, result.Err
	}

	variant, _ := result.Val.(*processedVariant)

	return &ImageResponse{
		Content:       io.NopCloser(bytes.NewReader(variant.data)),
		ContentLength: int64(len(variant.data)),
		ContentType:   variant.contentType,
		FetchedBytes:  variant.fetchedBytes,
		ETag:          formatETag(cacheKey, variant.checkedAt),
	}, nil
}

// withoutCancelKeepingDeadline returns ctx without its cancellation but with
// its deadline, if it has one, and the func that releases the returned
// context.
func withoutCancelKeepingDeadline(
	ctx context.Context,
) (context.Context, context.CancelFunc) {
	deadline, hasDeadline := ctx.Deadline()
	if !hasDeadline {
		return context.WithoutCancel(ctx), func() {}
	}

	return context.WithDeadline(context.WithoutCancel(ctx), deadline)
}

// loadCachedSource opens src's cached source, reading only the first bytes,
// which give its format, and returns it with its size and its format. It
// returns nil if the source is not in the cache, cannot be read, is in no
// source format or exceeds maxResponseSize.
func (p *Proxy) loadCachedSource(
	ctx context.Context, src imagerequest.Source,
) (io.ReadCloser, int64, imagerequest.Format) {
	reader, size, format, err := p.cache.Source(ctx, src)
	if errors.Is(err, imgcache.ErrNotFound) {
		return nil, 0, ""
	}

	if err != nil {
		p.log.Warn("failed to load cached source, fetching", "error", err)

		return nil, 0, ""
	}

	p.log.Debug("using cached source", "source_key", src.Key())

	if size > p.maxResponseSize {
		_ = reader.Close()

		p.log.Warn("cached source exceeds max response size, discarding",
			"source_key", src.Key(),
			"max_bytes", p.maxResponseSize,
		)

		return nil, 0, ""
	}

	return reader, size, format
}

// processFromSourceOrFetch makes the variant req asks for, from the cached
// source when there is one and from a fetched source otherwise. It counts
// whether the source was in the cache.
//
// A cached source last checked with upstream CacheTTL or more ago is checked
// first: upstream is asked to send it only if it changed since. When it
// changed, the new bytes replace the cached source and the variant is made
// from them. When it did not change, or the check fails, the cached source
// is used: the variant's cached file is served again if it was made from the
// cached source's bytes, and otherwise the variant is made from them. When
// the source is not cached but the variant is, the variant is checked with
// upstream the same way (see fetchAndProcess).
func (p *Proxy) processFromSourceOrFetch(
	ctx context.Context,
	req *imagerequest.Request,
	cacheKey imagerequest.VariantKey,
) (*processedVariant, error) {
	cached, err := p.cache.LookupSource(ctx, req.Source)
	if err != nil {
		if !errors.Is(err, imgcache.ErrNotFound) {
			p.log.Warn("failed to look up cached source, fetching", "error", err)
		}

		p.cache.IncrementSourceStats(false)

		return p.fetchAndProcess(ctx, req, cacheKey)
	}

	p.cache.IncrementSourceStats(true)

	if p.cache.Stale(cached.CheckedAt) {
		fetched, err := p.fetchSource(ctx, req, cached.ETag, cached.LastModified)

		switch {
		case errors.Is(err, errSourceNotModified):
			cached.CheckedAt = time.Now().UTC()
			p.renewSource(ctx, req.Source, cached.CheckedAt)
		case err != nil:
			p.recordFailedCheck(ctx, req, err)
		default:
			return p.processFetched(ctx, req, cacheKey, fetched)
		}
	}

	variant := p.renewVariant(ctx, cacheKey, cached)
	if variant != nil {
		return variant, nil
	}

	source, sourceSize, sourceFormat := p.loadCachedSource(ctx, req.Source)

	// Fetch from upstream if we don't have source data or it's empty
	if source == nil {
		return p.fetchAndProcess(ctx, req, cacheKey)
	}

	defer func() { _ = source.Close() }()

	// Process using cached source; nothing was fetched from upstream. The
	// image processor reads the source only once it has a processing slot,
	// so a request waiting for one holds none of it in memory.
	return p.processAndStore(ctx, req, cacheKey, cached, source, sourceFormat,
		sourceSize)
}

// recordFailedCheck logs a failed check of req's source with upstream, after
// which the image is still served from the cache. It puts the source URL in
// the negative cache, whatever the failure, a timeout included, so upstream
// is asked at most once per NegativeTTL while it fails, however often the
// image is served. A failure that was a negative cache hit records nothing.
func (p *Proxy) recordFailedCheck(
	ctx context.Context, req *imagerequest.Request, err error,
) {
	if errors.Is(err, ErrNegativeCached) {
		return
	}

	p.log.Warn("checking source with upstream failed, serving from the cache",
		"request_id", middleware.GetReqID(ctx),
		"host", req.Source.Host, "path", req.Source.Path, "error", err)
	p.storeNegative(ctx, req)
}

// renewSource records that upstream answered at checkedAt that src's cached
// source had not changed. A failure to record it is only logged: the source
// is then checked again on its next use.
func (p *Proxy) renewSource(
	ctx context.Context, src imagerequest.Source, checkedAt time.Time,
) {
	err := p.cache.RenewSource(ctx, src, checkedAt)
	if err != nil {
		p.log.Warn("failed to renew cached source", "source_key", src.Key(),
			"error", err)
	}
}

// renewVariant returns the cached variant of cacheKey when it was made from
// the bytes of the cached source that source describes, and renews it to
// source.CheckedAt (see imgcache.Cache.RenewVariant). It returns nil when
// there is no such variant or it cannot be read.
func (p *Proxy) renewVariant(
	ctx context.Context, cacheKey imagerequest.VariantKey,
	source imgcache.SourceInfo,
) *processedVariant {
	cached, err := p.cache.RenewVariant(ctx, cacheKey, source)
	if err != nil {
		if !errors.Is(err, imgcache.ErrNotFound) {
			p.log.Warn("cached variant not renewed, making it again",
				"cache_key", cacheKey, "error", err)
		}

		return nil
	}

	return p.readCachedVariant(cacheKey, cached)
}

// cachedVariantIgnoringTTL returns the cached variant of cacheKey whatever
// its age, with what its row keeps about the source bytes it was made from
// (see imgcache.Cache.LookupVariant). It returns nil when the variant is not
// cached or cannot be read. The SourceInfo is empty when the variant is not
// returned, or its row is missing or cannot be read.
func (p *Proxy) cachedVariantIgnoringTTL(
	ctx context.Context, cacheKey imagerequest.VariantKey,
) (*processedVariant, imgcache.SourceInfo) {
	cached, err := p.cache.VariantIgnoringTTL(cacheKey)
	if err != nil {
		if !errors.Is(err, imgcache.ErrNotFound) {
			p.log.Warn("cached variant not served", "cache_key", cacheKey,
				"error", err)
		}

		return nil, imgcache.SourceInfo{}
	}

	variant := p.readCachedVariant(cacheKey, cached)
	if variant == nil {
		return nil, imgcache.SourceInfo{}
	}

	madeFrom, err := p.cache.LookupVariant(ctx, cacheKey)
	if err != nil && !errors.Is(err, imgcache.ErrNotFound) {
		p.log.Warn("failed to look up cached variant", "cache_key", cacheKey,
			"error", err)
	}

	return variant, madeFrom
}

// readCachedVariant reads the cached variant of cacheKey from its opened file,
// and closes the file. It returns nil when the variant cannot be read.
func (p *Proxy) readCachedVariant(
	cacheKey imagerequest.VariantKey, file imgcache.VariantFile,
) *processedVariant {
	defer func() { _ = file.Content.Close() }()

	data, err := io.ReadAll(file.Content)
	if err != nil {
		p.log.Warn("failed to read cached variant", "cache_key", cacheKey,
			"error", err)

		return nil
	}

	return &processedVariant{
		data:        data,
		contentType: file.ContentType,
		checkedAt:   file.CheckedAt,
	}
}

// fetchAndProcess fetches the source from upstream, stores it, and makes
// and stores the variant from it.
//
// The variant can still be cached after eviction removed its source. The
// fetch then sends the ETag and Last-Modified of the source bytes the variant
// was made from, so upstream sends the source only if it changed since. When
// it did not change, the cached variant is renewed and served, with no
// download and no conversion. When the fetch fails, the cached variant is
// served whatever its age, and the failure is recorded as a failed check (see
// recordFailedCheck).
func (p *Proxy) fetchAndProcess(
	ctx context.Context,
	req *imagerequest.Request,
	cacheKey imagerequest.VariantKey,
) (*processedVariant, error) {
	variant, madeFrom := p.cachedVariantIgnoringTTL(ctx, cacheKey)

	fetched, err := p.fetchSource(ctx, req, madeFrom.ETag, madeFrom.LastModified)

	switch {
	case err == nil:
		return p.processFetched(ctx, req, cacheKey, fetched)
	case variant == nil:
		return nil, err
	case errors.Is(err, errSourceNotModified):
		variant.checkedAt = time.Now().UTC()

		err = p.cache.SetVariantCheckedAt(ctx, cacheKey, variant.checkedAt)
		if err != nil {
			p.log.Warn("failed to renew cached variant", "cache_key", cacheKey,
				"error", err)
		}
	default:
		p.recordFailedCheck(ctx, req, err)
	}

	return variant, nil
}

// processFetched makes and stores the variant from the source fetchSource
// fetched, then closes the fetch's response.
func (p *Proxy) processFetched(
	ctx context.Context,
	req *imagerequest.Request,
	cacheKey imagerequest.VariantKey,
	fetched *fetchedSource,
) (*processedVariant, error) {
	// Closing the response frees the upstream connection. It is closed only
	// after processing, so the fetcher's connection limit also bounds the
	// fetched sources held in memory while their requests wait for a
	// processing slot.
	defer func() { _ = fetched.response.Close() }()

	return p.processAndStore(ctx, req, cacheKey, fetched.info,
		bytes.NewReader(fetched.data), fetched.format, int64(len(fetched.data)))
}

// errSourceNotModified is returned by fetchSource when upstream answers 304
// Not Modified: the cached source has not changed.
var errSourceNotModified = errors.New("upstream source not modified")

// fetchedSource is a source fetchSource fetched and stored in the cache.
type fetchedSource struct {
	data   []byte
	format imagerequest.Format
	// info is what the cache keeps about the source
	info imgcache.SourceInfo
	// response is the fetch's response, still open; closing it frees the
	// upstream connection
	response io.Closer
}

// fetchSource fetches req's source from upstream and stores it in the cache.
// A non-empty ifNoneMatch or ifModifiedSince asks upstream to send the source
// only if it changed since; errSourceNotModified says it did not. It counts
// the fetch with the bytes read from upstream and its duration, including
// when reading the response or a later step fails, and counts a fetch that
// fails or whose response cannot be read as an upstream fetch error. It
// fetches nothing while the source URL is in the negative cache. The caller
// closes the returned response.
func (p *Proxy) fetchSource(
	ctx context.Context,
	req *imagerequest.Request,
	ifNoneMatch, ifModifiedSince string,
) (*fetchedSource, error) {
	err := p.refuseNegativeCached(ctx, req)
	if err != nil {
		return nil, err
	}

	// Fetch from upstream
	p.log.Debug("fetching from upstream", "host", req.Source.Host,
		"path", req.Source.Path, "query", req.Source.Query)

	fetchStart := time.Now()

	fetchResult, err := p.fetcher.Fetch(ctx, req.Source,
		ifNoneMatch, ifModifiedSince)
	if err != nil {
		p.upstreamFetchErrors.Inc()

		// Store negative cache for certain errors
		if isNegativeCacheable(err) {
			p.storeNegative(ctx, req)
		}

		return nil, fmt.Errorf("upstream fetch failed: %w", err)
	}

	if fetchResult.StatusCode == http.StatusNotModified {
		_ = fetchResult.Content.Close()

		p.log.Info("upstream source not modified",
			"request_id", middleware.GetReqID(ctx), "host", req.Source.Host,
			"path", req.Source.Path, "fetch_ms", fetchResult.FetchDurationMs)

		return nil, errSourceNotModified
	}

	// Read and validate the source content
	sourceData, err := io.ReadAll(fetchResult.Content)
	fetchBytes := int64(len(sourceData))

	// Counted also when the request context has ended meanwhile
	p.incrementUpstreamFetch(fetchBytes, time.Since(fetchStart))

	if err != nil {
		p.upstreamFetchErrors.Inc()

		_ = fetchResult.Content.Close()

		return nil, fmt.Errorf("failed to read upstream response: %w", err)
	}

	p.logUpstreamFetch(ctx, req, fetchResult, fetchBytes)

	// The source's format, detected from its first bytes, which must match
	// its content type. A body that does not is the upstream's error, so it
	// gets a negative cache entry, as an upstream error status does.
	sourceFormat, err := magic.ValidateMagicBytes(sourceData,
		fetchResult.ContentType)
	if err != nil {
		p.log.Warn("source refused", "request_id", middleware.GetReqID(ctx),
			"host", req.Source.Host, "path", req.Source.Path,
			"content_type", fetchResult.ContentType, "error", err)
		p.storeNegative(ctx, req)

		_ = fetchResult.Content.Close()

		return nil, fmt.Errorf("content validation failed: %w", err)
	}

	info := fetchedSourceInfo(fetchResult)

	// Store source content
	_, err = p.cache.PutSource(ctx, req.Source, bytes.NewReader(sourceData), info)
	if err != nil {
		p.log.Warn("failed to store source content", "error", err)
		// Continue even if caching fails
	}

	return &fetchedSource{
		data:     sourceData,
		format:   sourceFormat,
		info:     info,
		response: fetchResult.Content,
	}, nil
}

// refuseNegativeCached returns an error wrapping ErrNegativeCached while
// req's source URL is in the negative cache, and nil otherwise. A failure to
// read the negative cache is only logged, and returns nil.
func (p *Proxy) refuseNegativeCached(
	ctx context.Context, req *imagerequest.Request,
) error {
	negHit, err := p.cache.Failure(ctx, req.Source)
	if err != nil {
		p.log.Warn("negative cache check failed", "error", err)
	}

	if !negHit {
		return nil
	}

	p.log.Debug("negative cache hit",
		"host", req.Source.Host,
		"path", req.Source.Path,
	)

	return fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
}

// fetchedSourceInfo returns what the cache keeps about a source upstream
// just sent: fetched and checked now, with the ETag and Last-Modified
// headers of fetchResult.
func fetchedSourceInfo(fetchResult *httpfetcher.FetchResult) imgcache.SourceInfo {
	now := time.Now().UTC()
	upstreamHeaders := http.Header(fetchResult.Headers)

	return imgcache.SourceInfo{
		FetchedAt:    now,
		CheckedAt:    now,
		ETag:         upstreamHeaders.Get("ETag"),
		LastModified: upstreamHeaders.Get("Last-Modified"),
	}
}

// logUpstreamFetch logs a fetch of req's source that read fetchBytes bytes,
// with its download rate and connection details.
func (p *Proxy) logUpstreamFetch(
	ctx context.Context,
	req *imagerequest.Request,
	fetchResult *httpfetcher.FetchResult,
	fetchBytes int64,
) {
	var downloadRate string

	if fetchResult.FetchDurationMs > 0 {
		seconds := float64(fetchResult.FetchDurationMs) / 1000.0 //nolint:mnd // ms to seconds
		bitsPerSecond := float64(fetchBytes*8) / seconds         //nolint:mnd // bytes to bits
		downloadRate = humanize.SI(bitsPerSecond, "bps")
	}

	p.log.Info("upstream fetched",
		"request_id", middleware.GetReqID(ctx),
		"host", req.Source.Host,
		"path", req.Source.Path,
		"bytes", fetchBytes,
		"fetch_ms", fetchResult.FetchDurationMs,
		"rate", downloadRate,
		"remote_addr", fetchResult.RemoteAddr,
		"http", fetchResult.HTTPVersion,
		"tls", fetchResult.TLSVersion,
		"cipher", fetchResult.TLSCipherSuite,
	)
}

// incrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes
// and took duration. A fetch that read no bytes is not counted.
func (p *Proxy) incrementUpstreamFetch(fetchBytes int64, duration time.Duration) {
	if fetchBytes <= 0 {
		return
	}

	p.upstreamFetches.Inc()
	p.upstreamFetchBytes.Add(float64(fetchBytes))
	p.upstreamFetchDurations.Observe(duration.Seconds())
}

// storeNegative stores a negative cache entry for req's source, whose fetch
// failed, so the source is not fetched again until the entry expires. A
// failure to store the entry is only logged.
func (p *Proxy) storeNegative(ctx context.Context, req *imagerequest.Request) {
	err := p.cache.PutFailure(ctx, req.Source)
	if err != nil {
		p.log.Warn("failed to store negative cache entry",
			"host", req.Source.Host, "path", req.Source.Path, "error", err)
	}
}

// processedVariant is a variant as processAndStore made it. Each request that
// shared its processing serves it through a reader of its own.
type processedVariant struct {
	data         []byte
	contentType  string
	fetchedBytes int64
	// checkedAt is its source's check time when the variant was made or last
	// renewed, which its cached file has as its modification time
	checkedAt time.Time
}

// processAndStore processes the image read from source, which is in
// sourceFormat, and stores the result. sourceInfo describes the source's
// bytes, and is stored with the variant (see imgcache.Cache.PutVariant).
func (p *Proxy) processAndStore(
	ctx context.Context,
	req *imagerequest.Request,
	cacheKey imagerequest.VariantKey,
	sourceInfo imgcache.SourceInfo,
	source io.Reader,
	sourceFormat imagerequest.Format,
	fetchBytes int64,
) (*processedVariant, error) {
	// Process the image
	processStart := time.Now()

	// ctx has the deadline of the request doing the processing, which can
	// end while a request waiting for its image has time left
	processResult, err := p.transform(ctx, source, sourceFormat, &req.Transform)
	if errors.Is(err, context.DeadlineExceeded) {
		return nil, fmt.Errorf("%w: %w", ErrProcessingTimedOut, err)
	}

	if err != nil {
		return nil, fmt.Errorf("image processing failed: %w", err)
	}

	processDuration := time.Since(processStart)

	// Counted also when the request context has ended meanwhile
	p.transforms.Inc()
	p.transformDurations.Observe(processDuration.Seconds())

	// Read processed content
	processedData, err := io.ReadAll(processResult.Content)
	_ = processResult.Content.Close()

	if err != nil {
		return nil, fmt.Errorf("failed to read processed content: %w", err)
	}

	// Log conversion details
	outputSize := int64(len(processedData))

	var sizePercent float64
	if fetchBytes > 0 {
		//nolint:mnd // percentage calculation
		sizePercent = float64(outputSize) / float64(fetchBytes) * 100.0
	}

	p.log.Info("image converted",
		"request_id", middleware.GetReqID(ctx),
		"host", req.Source.Host,
		"path", req.Source.Path,
		"src_format", sourceFormat,
		"dst_format", req.Transform.Format,
		"src_bytes", fetchBytes,
		"dst_bytes", outputSize,
		"src_dimensions", fmt.Sprintf("%dx%d",
			processResult.InputWidth, processResult.InputHeight),
		"dst_dimensions", fmt.Sprintf("%dx%d",
			processResult.Width, processResult.Height),
		"size_ratio", fmt.Sprintf("%.1f%%", sizePercent),
		"convert_ms", processDuration.Milliseconds(),
		"quality", req.Transform.Quality,
		"fit", req.Transform.FitMode,
	)

	// Store variant to cache
	err = p.cache.PutVariant(ctx, cacheKey, sourceInfo,
		bytes.NewReader(processedData))
	if err != nil {
		p.log.Warn("failed to store variant", "error", err)
		// Continue even if caching fails
	}

	return &processedVariant{
		data:         processedData,
		contentType:  processResult.ContentType,
		fetchedBytes: fetchBytes,
		checkedAt:    sourceInfo.CheckedAt,
	}, nil
}

// transform has the image processor transform the image read from source
// as requested says, and counts it in pixa_transforms_in_progress until the
// processor returns or panics.
func (p *Proxy) transform(
	ctx context.Context,
	source io.Reader,
	sourceFormat imagerequest.Format,
	requested *imagerequest.Transform,
) (*imageprocessor.Result, error) {
	p.transformsInProgress.Inc()
	defer p.transformsInProgress.Dec()

	return p.processor.Process(ctx, source, sourceFormat, requested)
}

// isNegativeCacheable returns true if the error should be cached.
func isNegativeCacheable(err error) bool {
	return errors.Is(err, httpfetcher.ErrUpstreamError)
}

// etagHashLength is the number of hash characters to use for ETags.
const etagHashLength = 16

// formatETag returns the quoted ETag of the variant of key: the first
// etagHashLength characters of key, then checkedAt in Unix nanoseconds.
// checkedAt is its source's check time when the variant was made or last
// renewed, so a variant made again from new source bytes gets another ETag.
func formatETag(key imagerequest.VariantKey, checkedAt time.Time) string {
	hash := string(key)
	// Use first 16 characters of hash for a shorter but still unique ETag
	if len(hash) > etagHashLength {
		hash = hash[:etagHashLength]
	}

	return fmt.Sprintf(`"%s-%d"`, hash, checkedAt.UnixNano())
}