pixa/internal/imgcache/service.go

package imgcache

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/url"
	"time"

	"github.com/dustin/go-humanize"
	"sneak.berlin/go/pixa/internal/allowlist"
	"sneak.berlin/go/pixa/internal/httpfetcher"
	"sneak.berlin/go/pixa/internal/imageprocessor"
	"sneak.berlin/go/pixa/internal/magic"
	"sneak.berlin/go/pixa/internal/signature"
)

// Service implements the ImageCache interface, orchestrating cache,
// fetcher, and processor.
type Service struct {
	cache           *Cache
	fetcher         httpfetcher.Fetcher
	processor       *imageprocessor.ImageProcessor
	signer          *signature.Signer
	allowlist       *allowlist.HostAllowList
	log             *slog.Logger
	allowHTTP       bool
	maxResponseSize int64
}

// ServiceConfig holds configuration for the image service.
type ServiceConfig struct {
	// Cache is the cache instance
	Cache *Cache
	// FetcherConfig configures the upstream fetcher (ignored if Fetcher is set)
	FetcherConfig *httpfetcher.Config
	// Fetcher is an optional custom fetcher (for testing)
	Fetcher httpfetcher.Fetcher
	// SigningKey is the HMAC signing key (empty disables signing)
	SigningKey string
	// Allowlist is the list of hosts that don't require signatures
	Allowlist []string
	// MaxConcurrentProcessing is the most images processed at once; zero
	// uses the image processor's default, one per CPU
	MaxConcurrentProcessing int
	// Logger for logging
	Logger *slog.Logger
}

// Static errors for service construction and unimplemented operations.
var (
	errCacheRequired       = errors.New("cache is required")
	errSigningKeyRequired  = errors.New("signing key is required")
	errPurgeNotImplemented = errors.New("purge not implemented")
)

// NewService creates a new image service.
func NewService(cfg *ServiceConfig) (*Service, error) {
	if cfg.Cache == nil {
		return nil, errCacheRequired
	}

	if cfg.SigningKey == "" {
		return nil, errSigningKeyRequired
	}

	// Resolve fetcher config for defaults
	fetcherCfg := cfg.FetcherConfig
	if fetcherCfg == nil {
		fetcherCfg = httpfetcher.DefaultConfig()
	}

	// Use custom fetcher if provided, otherwise create HTTP fetcher
	var fetcher httpfetcher.Fetcher
	if cfg.Fetcher != nil {
		fetcher = cfg.Fetcher
	} else {
		fetcher = httpfetcher.New(fetcherCfg)
	}

	signer := signature.New(cfg.SigningKey)

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

	allowHTTP := false
	if cfg.FetcherConfig != nil {
		allowHTTP = cfg.FetcherConfig.AllowHTTP
	}

	maxResponseSize := fetcherCfg.MaxResponseSize
	processor := imageprocessor.New(imageprocessor.Params{
		MaxInputBytes:           maxResponseSize,
		MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
	})

	return &Service{
		cache:           cfg.Cache,
		fetcher:         fetcher,
		processor:       processor,
		signer:          signer,
		allowlist:       allowlist.New(cfg.Allowlist),
		log:             log,
		allowHTTP:       allowHTTP,
		maxResponseSize: maxResponseSize,
	}, 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)")

// Get retrieves a processed image, fetching and processing if necessary.
func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, error) {
	// Propagate AllowHTTP setting to the request
	req.AllowHTTP = s.allowHTTP

	// Check negative cache first - skip fetching for recently-failed URLs
	negHit, err := s.cache.checkNegativeCache(ctx, req)
	if err != nil {
		s.log.Warn("negative cache check failed", "error", err)
	}

	if negHit {
		s.log.Debug("negative cache hit",
			"host", req.SourceHost,
			"path", req.SourcePath,
		)

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

	// Check variant cache first (disk only, no DB)
	result, err := s.cache.Lookup(ctx, req)
	if err != nil {
		s.log.Warn("cache lookup failed", "error", err)
	}

	// Cache hit - serve directly from disk
	if result != nil && result.Hit {
		reader, size, contentType, err := s.cache.GetVariant(result.CacheKey)
		if err != nil {
			s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
			// Fall through to re-process
		} else {
			// Counted also when the request context has ended meanwhile
			s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0)

			return &ImageResponse{
				Content:       reader,
				ContentLength: size,
				ContentType:   contentType,
				CacheStatus:   CacheHit,
				ETag:          formatETag(result.CacheKey),
			}, nil
		}
	}

	// Cache miss - process the cached source or fetch it, then count the
	// miss with the bytes it fetched from upstream, also when it failed or
	// the request context has ended meanwhile
	cacheKey := CacheKey(req)

	response, fetchedBytes, err := s.processFromSourceOrFetch(ctx, req, cacheKey)

	s.cache.IncrementStats(context.WithoutCancel(ctx), false, fetchedBytes)

	if err != nil {
		return nil, err
	}

	response.CacheStatus = CacheMiss

	return response, nil
}

// Warm pre-fetches and caches an image without returning it.
func (s *Service) Warm(ctx context.Context, req *ImageRequest) error {
	_, err := s.Get(ctx, req)

	return err
}

// Purge removes a cached image. Purging is not implemented yet.
func (s *Service) Purge(_ context.Context, _ *ImageRequest) error {
	return errPurgeNotImplemented
}

// Stats returns cache statistics.
func (s *Service) Stats(ctx context.Context) (*CacheStats, error) {
	return s.cache.Stats(ctx)
}

// ValidateRequest validates the request signature if required.
func (s *Service) ValidateRequest(req *ImageRequest) error {
	// Check if host is allowed (no signature required)
	sourceURL := req.SourceURL()

	parsedURL, err := url.Parse(sourceURL)
	if err != nil {
		return fmt.Errorf("invalid source URL: %w", err)
	}

	if s.allowlist.IsAllowed(parsedURL) {
		return nil
	}

	// Signature required for non-allowed hosts
	return s.signer.Verify(signatureRequest(req))
}

// GenerateSignedURL generates a signed URL for the given request. The URL
// carries q and fit next to sig and exp, so the image route verifies it for
// the quality and fit it was signed with. An unset quality or fit is first
// set to 85 or cover, the values the route uses when a URL has no q or fit.
func (s *Service) GenerateSignedURL(
	baseURL string,
	req *ImageRequest,
	ttl time.Duration,
) (string, error) {
	if req.Quality == 0 {
		req.Quality = 85
	}

	if req.FitMode == "" {
		req.FitMode = FitCover
	}

	sigReq := signatureRequest(req)
	path, sig, exp := s.signer.GenerateSignedURL(sigReq, ttl)

	// Propagate the generated signature and expiration back onto the request.
	req.Expires = sigReq.Expires
	req.Signature = sigReq.Signature

	return fmt.Sprintf("%s%s?sig=%s&exp=%d&q=%d&fit=%s",
		baseURL, path, sig, exp, req.Quality, req.FitMode), nil
}

// loadCachedSource opens source content from cache, without reading it, and
// returns it with its size; nil if the cached data is unavailable, empty or
// exceeds maxResponseSize.
func (s *Service) loadCachedSource(
	contentHash ContentHash,
) (io.ReadCloser, int64) {
	reader, size, err := s.cache.GetSourceContent(contentHash)
	if err != nil {
		s.log.Warn("failed to load cached source, fetching", "error", err)

		return nil, 0
	}

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

		s.log.Warn("cached source exceeds max response size, discarding",
			"hash", contentHash,
			"max_bytes", s.maxResponseSize,
		)

		return nil, 0
	}

	if size == 0 {
		_ = reader.Close()

		return nil, 0
	}

	return reader, size
}

// processFromSourceOrFetch processes an image, using cached source content
// if available. It also returns the number of bytes fetched from upstream,
// as fetchAndProcess does, or 0 when the cached source was used.
func (s *Service) processFromSourceOrFetch(
	ctx context.Context,
	req *ImageRequest,
	cacheKey VariantKey,
) (*ImageResponse, int64, error) {
	// Check if we have cached source content
	contentHash, _, err := s.cache.LookupSource(ctx, req)
	if err != nil {
		s.log.Warn("source lookup failed", "error", err)
	}

	var (
		source     io.ReadCloser
		sourceSize int64
	)

	if contentHash != "" {
		s.log.Debug("using cached source", "hash", contentHash)
		source, sourceSize = s.loadCachedSource(contentHash)
	}

	// Fetch from upstream if we don't have source data or it's empty
	if source == nil {
		return s.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.
	resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)

	return resp, 0, err
}

// fetchAndProcess fetches from upstream, processes, and caches the result.
// It also returns the number of bytes read from upstream, including when
// reading the response or a later step fails.
func (s *Service) fetchAndProcess(
	ctx context.Context,
	req *ImageRequest,
	cacheKey VariantKey,
) (*ImageResponse, int64, error) {
	// Fetch from upstream
	sourceURL := req.SourceURL()

	s.log.Debug("fetching from upstream", "url", sourceURL)

	fetchResult, err := s.fetcher.Fetch(ctx, sourceURL)
	if err != nil {
		// Store negative cache for certain errors
		if isNegativeCacheable(err) {
			statusCode := extractStatusCode(err)

			storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
			if storeErr != nil {
				s.log.Warn("failed to store negative cache entry",
					"host", req.SourceHost, "path", req.SourcePath, "error", storeErr)
			}
		}

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

	// Closing the body 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() { _ = fetchResult.Content.Close() }()

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

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

	// Calculate download bitrate
	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")
	}

	// Log upstream fetch details
	s.log.Info("upstream fetched",
		"host", req.SourceHost,
		"path", req.SourcePath,
		"bytes", fetchBytes,
		"fetch_ms", fetchResult.FetchDurationMs,
		"rate", downloadRate,
		"remote_addr", fetchResult.RemoteAddr,
		"http", fetchResult.HTTPVersion,
		"tls", fetchResult.TLSVersion,
		"cipher", fetchResult.TLSCipherSuite,
	)

	// Validate magic bytes match content type
	err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
	if err != nil {
		return nil, fetchBytes, fmt.Errorf("content validation failed: %w", err)
	}

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

	resp, err := s.processAndStore(
		ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
	)

	return resp, fetchBytes, err
}

// processAndStore processes the image read from source and stores the
// result.
func (s *Service) processAndStore(
	ctx context.Context,
	req *ImageRequest,
	cacheKey VariantKey,
	source io.Reader,
	fetchBytes int64,
) (*ImageResponse, error) {
	// Process the image
	processStart := time.Now()

	processReq := &imageprocessor.Request{
		Size:    imageprocessor.Size{Width: req.Size.Width, Height: req.Size.Height},
		Format:  imageprocessor.Format(req.Format),
		Quality: req.Quality,
		FitMode: imageprocessor.FitMode(req.FitMode),
	}

	processResult, err := s.processor.Process(ctx, source, processReq)
	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
	s.cache.IncrementTransformCount(context.WithoutCancel(ctx))

	// 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
	}

	s.log.Info("image converted",
		"host", req.SourceHost,
		"path", req.SourcePath,
		"src_format", processResult.InputFormat,
		"dst_format", req.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.Quality,
		"fit", req.FitMode,
	)

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

	return &ImageResponse{
		Content:       io.NopCloser(bytes.NewReader(processedData)),
		ContentLength: outputSize,
		ContentType:   processResult.ContentType,
		FetchedBytes:  fetchBytes,
		ETag:          formatETag(cacheKey),
	}, nil
}

// signatureRequest projects an ImageRequest onto the standalone
// signature.Request type used by the signature package. This keeps the
// import edge one-way: imgcache depends on signature, never the reverse.
func signatureRequest(req *ImageRequest) *signature.Request {
	return &signature.Request{
		SourceHost:  req.SourceHost,
		SourcePath:  req.SourcePath,
		SourceQuery: req.SourceQuery,
		Width:       req.Size.Width,
		Height:      req.Size.Height,
		Format:      string(req.Format),
		Quality:     req.Quality,
		FitMode:     string(req.FitMode),
		Signature:   req.Signature,
		Expires:     req.Expires,
	}
}

// HTTP status codes for error responses.
const (
	httpStatusBadGateway    = 502
	httpStatusInternalError = 500
)

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

// extractStatusCode extracts HTTP status code from error message.
func extractStatusCode(err error) int {
	// Default to 502 Bad Gateway for upstream errors
	if errors.Is(err, httpfetcher.ErrUpstreamError) {
		return httpStatusBadGateway
	}

	return httpStatusInternalError
}

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

// formatETag formats a VariantKey as a quoted ETag value.
func formatETag(key VariantKey) 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 `"` + hash + `"`
}