pixa/internal/imageproxy/cache_ttl_internal_test.go

package imageproxy

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"image/color"
	"io"
	"net/http"
	"os"
	"sync"
	"testing"
	"time"

	"sneak.berlin/go/pixa/internal/httpfetcher"
	"sneak.berlin/go/pixa/internal/imagerequest"
)

// Validators the test upstream sends with its image.
const (
	testETag         = `"v1"`
	testLastModified = "Mon, 02 Nov 2026 10:00:00 GMT"
)

// errConnectionRefused stands for an upstream that cannot be reached.
var errConnectionRefused = errors.New("connection refused")

// upstreamFailure is one way a fetch from upstream fails.
type upstreamFailure struct {
	name string
	err  error
}

// upstreamFailures returns the ways a check of a cached image with upstream
// can fail: an error status, a timeout and no connection.
func upstreamFailures() []upstreamFailure {
	return []upstreamFailure{
		{"error status", fmt.Errorf("%w: status 503", httpfetcher.ErrUpstreamError)},
		{"timeout", httpfetcher.ErrUpstreamTimeout},
		{"no connection", errConnectionRefused},
	}
}

// conditionalUpstream serves one JPEG image with an ETag or a Last-Modified
// header, or both. It answers 304 Not Modified to a fetch whose If-None-Match
// is its ETag, or, when it has no ETag, whose If-Modified-Since is its
// Last-Modified. While err is set, every fetch fails with it. It records the
// conditional headers of every fetch.
type conditionalUpstream struct {
	mu           sync.Mutex
	image        []byte
	etag         string
	lastModified string
	err          error

	ifNoneMatch     []string
	ifModifiedSince []string
}

func (u *conditionalUpstream) Fetch(
	_ context.Context, _ imagerequest.Source,
	ifNoneMatch, ifModifiedSince string,
) (*httpfetcher.FetchResult, error) {
	u.mu.Lock()
	defer u.mu.Unlock()

	u.ifNoneMatch = append(u.ifNoneMatch, ifNoneMatch)
	u.ifModifiedSince = append(u.ifModifiedSince, ifModifiedSince)

	if u.err != nil {
		return nil, u.err
	}

	unchanged := ifNoneMatch == u.etag
	if u.etag == "" {
		unchanged = ifModifiedSince == u.lastModified
	}

	if unchanged {
		return &httpfetcher.FetchResult{
			Content:    http.NoBody,
			StatusCode: http.StatusNotModified,
		}, nil
	}

	headers := http.Header{}
	if u.etag != "" {
		headers.Set("ETag", u.etag)
	}

	if u.lastModified != "" {
		headers.Set("Last-Modified", u.lastModified)
	}

	return &httpfetcher.FetchResult{
		Content:       io.NopCloser(bytes.NewReader(u.image)),
		ContentLength: int64(len(u.image)),
		ContentType:   testContentTypeJPEG,
		Headers:       headers,
		StatusCode:    http.StatusOK,
	}, nil
}

// change replaces the upstream image and its ETag.
func (u *conditionalUpstream) change(image []byte, etag string) {
	u.mu.Lock()
	defer u.mu.Unlock()

	u.image = image
	u.etag = etag
}

// fail makes every later fetch fail with err.
func (u *conditionalUpstream) fail(err error) {
	u.mu.Lock()
	defer u.mu.Unlock()

	u.err = err
}

// fetches returns how many fetches upstream was asked for.
func (u *conditionalUpstream) fetches() int {
	u.mu.Lock()
	defer u.mu.Unlock()

	return len(u.ifNoneMatch)
}

// lastConditionalHeaders returns the If-None-Match and If-Modified-Since of
// the last fetch.
func (u *conditionalUpstream) lastConditionalHeaders() (string, string) {
	u.mu.Lock()
	defer u.mu.Unlock()

	last := len(u.ifNoneMatch) - 1

	return u.ifNoneMatch[last], u.ifModifiedSince[last]
}

// setupConditionalUpstreamProxy returns a proxy whose upstream is a
// conditionalUpstream serving a red JPEG with an ETag.
func setupConditionalUpstreamProxy(
	t *testing.T,
) (*testProxy, *TestFixtures, *conditionalUpstream) {
	t.Helper()

	svc, fixtures := setupTestProxy(t)
	upstream := &conditionalUpstream{
		image: generateTestJPEG(t, 100, 100, color.RGBA{255, 0, 0, 255}),
		etag:  testETag,
	}
	svc.fetcher = upstream

	return svc, fixtures, upstream
}

// TestProxy_Get_FreshVariantAsksUpstreamNothing checks that a variant whose
// source was checked with upstream less than the cache TTL ago is served from
// the cache without asking upstream.
func TestProxy_Get_FreshVariantAsksUpstreamNothing(t *testing.T) {
	t.Parallel()

	svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
	variant := photoVariant(fixtures, 85, imagerequest.FitCover)

	first, _ := getImage(t, svc, variant)
	second, status := getImage(t, svc, variant)

	t.Logf("cache status = %s, upstream fetches = %d", status, upstream.fetches())

	if status != CacheHit {
		t.Errorf("CacheStatus = %q, want %q", status, CacheHit)
	}

	if !bytes.Equal(first, second) {
		t.Error("the cached variant differs from the one first served")
	}

	if n := upstream.fetches(); n != 1 {
		t.Errorf("upstream fetches = %d, want 1", n)
	}
}

// TestProxy_Get_UnchangedSourceAfterTTLIsNotDownloadedAgain caches a variant,
// sets its source's check time and the variant's file time two days back,
// and requests it again. Upstream must be asked with the validator it sent,
// answer 304, and send no bytes; the variant must be served again without
// being made again, and then be served from the cache until the TTL passes.
func TestProxy_Get_UnchangedSourceAfterTTLIsNotDownloadedAgain(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name                string
		etag                string
		lastModified        string
		wantIfNoneMatch     string
		wantIfModifiedSince string
	}{
		{"ETag", testETag, "", testETag, ""},
		{"Last-Modified only", "", testLastModified, "", testLastModified},
		{"both", testETag, testLastModified, testETag, testLastModified},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
			upstream.etag = tt.etag
			upstream.lastModified = tt.lastModified
			variant := photoVariant(fixtures, 85, imagerequest.FitCover)

			first, _ := getImage(t, svc, variant)
			setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2))

			second, status := getImage(t, svc, variant)

			ifNoneMatch, ifModifiedSince := upstream.lastConditionalHeaders()
			counters := readCounters(t, svc)
			t.Logf("cache status = %s, If-None-Match = %q, If-Modified-Since = %q, "+
				"counters = %+v", status, ifNoneMatch, ifModifiedSince, counters)

			if ifNoneMatch != tt.wantIfNoneMatch {
				t.Errorf("If-None-Match = %q, want %q", ifNoneMatch, tt.wantIfNoneMatch)
			}

			if ifModifiedSince != tt.wantIfModifiedSince {
				t.Errorf("If-Modified-Since = %q, want %q",
					ifModifiedSince, tt.wantIfModifiedSince)
			}

			if !bytes.Equal(first, second) {
				t.Error("the variant served after the check differs from the cached one")
			}

			if counters.upstreamFetchCount != 1 || counters.transformCount != 1 {
				t.Errorf("upstream downloads = %d, transforms = %d; want 1 and 1",
					counters.upstreamFetchCount, counters.transformCount)
			}

			_, status = getImage(t, svc, variant)
			if status != CacheHit {
				t.Errorf("CacheStatus after the check = %q, want %q", status, CacheHit)
			}

			if n := upstream.fetches(); n != 2 {
				t.Errorf("upstream fetches = %d, want 2", n)
			}
		})
	}
}

// TestProxy_Get_ChangedSourceAfterTTLReplacesSourceAndVariants caches two
// variants of one source, changes the image upstream, and sets the source's
// check time and both variants' file times two days back. Each variant must
// then be made again from the new image, which is downloaded once.
func TestProxy_Get_ChangedSourceAfterTTLReplacesSourceAndVariants(t *testing.T) {
	t.Parallel()

	svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
	variants := []*imagerequest.Request{
		photoVariant(fixtures, 85, imagerequest.FitCover),
		photoVariant(fixtures, 40, imagerequest.FitCover),
	}

	oldImages := make([][]byte, len(variants))
	for i, variant := range variants {
		oldImages[i], _ = getImage(t, svc, variant)
	}

	newSource := generateTestJPEG(t, 120, 90, color.RGBA{0, 0, 255, 255})
	upstream.change(newSource, `"v2"`)

	twoDaysAgo := time.Now().AddDate(0, 0, -2)
	for _, variant := range variants {
		setBackSourceAndVariant(t, svc, variant, twoDaysAgo)
	}

	for i, variant := range variants {
		image, _ := getImage(t, svc, variant)
		if bytes.Equal(image, oldImages[i]) {
			t.Errorf("variant %d is still made from the old image", i)
		}
	}

	counters := readCounters(t, svc)
	t.Logf("counters = %+v, upstream fetches = %d", counters, upstream.fetches())

	if counters.upstreamFetchCount != 2 {
		t.Errorf("upstream downloads = %d, want 2", counters.upstreamFetchCount)
	}

	cached, err := os.ReadFile(sourceFilePath(svc, variants[0].Source.Key()))
	if err != nil {
		t.Fatalf("reading the cached source: %v", err)
	}

	if !bytes.Equal(cached, newSource) {
		t.Error("the cached source is not the new image")
	}
}

// TestProxy_Get_UpstreamFailureAfterTTLServesCachedImage caches a variant,
// sets its source's check time and the variant's file time two days back, and
// makes upstream fail. The cached variant must still be served, and a variant
// not yet cached must be made from the cached source.
func TestProxy_Get_UpstreamFailureAfterTTLServesCachedImage(t *testing.T) {
	t.Parallel()

	for _, failure := range upstreamFailures() {
		t.Run(failure.name, func(t *testing.T) {
			t.Parallel()

			svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
			variant := photoVariant(fixtures, 85, imagerequest.FitCover)

			first, _ := getImage(t, svc, variant)
			setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2))
			upstream.fail(failure.err)

			second, _ := getImage(t, svc, variant)
			t.Logf("upstream fetches = %d", upstream.fetches())

			if upstream.fetches() != 2 {
				t.Errorf("upstream fetches = %d, want 2", upstream.fetches())
			}

			if !bytes.Equal(first, second) {
				t.Error("the variant served while upstream fails differs from the cached one")
			}

			getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
		})
	}
}

// TestProxy_Get_FailingUpstreamIsAskedOnceWhileImageIsServedOften caches a
// variant, sets its source's check time and the variant's file time two days
// back, and makes every fetch time out. Serving the variant ten times, and a
// variant not yet cached, must ask upstream once: the failed check puts the
// source URL in the negative cache.
func TestProxy_Get_FailingUpstreamIsAskedOnceWhileImageIsServedOften(t *testing.T) {
	t.Parallel()

	svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
	variant := photoVariant(fixtures, 85, imagerequest.FitCover)

	getImage(t, svc, variant)
	setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2))
	upstream.fail(httpfetcher.ErrUpstreamTimeout)

	for range 10 {
		getImage(t, svc, variant)
	}

	getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))

	t.Logf("upstream fetches = %d", upstream.fetches())

	if n := upstream.fetches(); n != 2 {
		t.Errorf("upstream fetches = %d, want 2: the first fetch and one check", n)
	}
}

// TestProxy_Get_UpstreamFailureAfterTTLServesVariantOfEvictedSource caches a
// variant, sets its file time two days back, removes its source's row and
// file, as eviction does, and makes upstream fail. The cached variant must
// still be served.
func TestProxy_Get_UpstreamFailureAfterTTLServesVariantOfEvictedSource(t *testing.T) {
	t.Parallel()

	for _, failure := range upstreamFailures() {
		t.Run(failure.name, func(t *testing.T) {
			t.Parallel()

			svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
			variant := photoVariant(fixtures, 85, imagerequest.FitCover)

			first, _ := getImage(t, svc, variant)
			setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2))
			evictSource(t, svc, variant.Source)
			upstream.fail(failure.err)

			second, _ := getImage(t, svc, variant)
			t.Logf("upstream fetches = %d", upstream.fetches())

			if !bytes.Equal(first, second) {
				t.Error("the variant served while upstream fails differs from the cached one")
			}

			if n := upstream.fetches(); n != 2 {
				t.Errorf("upstream fetches = %d, want 2", n)
			}
		})
	}
}

// TestProxy_Get_FailingUpstreamIsAskedOnceForVariantOfEvictedSource caches a
// variant, sets its file time two days back, removes its source's row and
// file, as eviction does, and makes every fetch time out. Serving the variant
// ten times must ask upstream once: the failed check puts the source URL in
// the negative cache.
func TestProxy_Get_FailingUpstreamIsAskedOnceForVariantOfEvictedSource(t *testing.T) {
	t.Parallel()

	svc, fixtures, upstream := setupConditionalUpstreamProxy(t)
	variant := photoVariant(fixtures, 85, imagerequest.FitCover)

	first, _ := getImage(t, svc, variant)
	setBackSourceAndVariant(t, svc, variant, time.Now().AddDate(0, 0, -2))
	evictSource(t, svc, variant.Source)
	upstream.fail(httpfetcher.ErrUpstreamTimeout)

	for range 10 {
		image, _ := getImage(t, svc, variant)
		if !bytes.Equal(image, first) {
			t.Fatal("the variant served while upstream fails differs from the cached one")
		}
	}

	t.Logf("upstream fetches = %d", upstream.fetches())

	if n := upstream.fetches(); n != 2 {
		t.Errorf("upstream fetches = %d, want 2: the first fetch and one check", n)
	}
}

// evictSource removes the row and then the file of src's cached source, as
// the cache's eviction does, and leaves its variants cached.
func evictSource(t *testing.T, svc *testProxy, src imagerequest.Source) {
	t.Helper()

	_, err := svc.db.ExecContext(t.Context(),
		`DELETE FROM source_content WHERE source_key = ?`, string(src.Key()))
	if err != nil {
		t.Fatalf("deleting the source's row: %v", err)
	}

	err = os.Remove(sourceFilePath(svc, src.Key()))
	if err != nil {
		t.Fatalf("removing the source's file: %v", err)
	}
}

// setBackSourceAndVariant sets back to when the times the cache keeps for
// req's source and for the variant req asks for: the source's fetch and check
// times, both last access times, the variant's check time, and the variant
// file's modification time.
func setBackSourceAndVariant(
	t *testing.T, svc *testProxy, req *imagerequest.Request, when time.Time,
) {
	t.Helper()

	key := req.Source.Key()
	setSourceLastAccessed(t, svc.db, key, when)

	_, err := svc.db.ExecContext(t.Context(),
		`UPDATE source_content SET fetched_at = ?, checked_at = ?
		WHERE source_key = ?`, when.UTC(), when.UTC(), string(key))
	if err != nil {
		t.Fatalf("setting back the source's fetch and check times: %v", err)
	}

	variantKey := imagerequest.CacheKey(req)
	setVariantLastAccessed(t, svc.db, variantKey, when)

	_, err = svc.db.ExecContext(t.Context(),
		`UPDATE variant_content SET checked_at = ? WHERE cache_key = ?`,
		when.UTC(), string(variantKey))
	if err != nil {
		t.Fatalf("setting back the variant's check time: %v", err)
	}

	path := variantFilePath(svc, variantKey)

	err = os.Chtimes(path, when, when)
	if err != nil {
		t.Fatalf("setting back the times of %s: %v", path, err)
	}
}