pixa/internal/imgcache/eviction_test.go

package imgcache

import (
	"bytes"
	"context"
	"database/sql"
	"io/fs"
	"os"
	"path/filepath"
	"testing"
	"time"

	_ "modernc.org/sqlite"
	"sneak.berlin/go/pixa/internal/database"
	"sneak.berlin/go/pixa/internal/httpfetcher"
)

// sqliteTimestampFormat matches the format SQLite's CURRENT_TIMESTAMP
// produces, so injected timestamps compare correctly against ones the
// implementation writes.
const sqliteTimestampFormat = "2006-01-02 15:04:05"

// evictionTestDB creates an in-memory SQLite database with the real
// production schema, limited to a single connection so the background
// eviction goroutine shares the same in-memory database as the test.
func evictionTestDB(t *testing.T) *sql.DB {
	t.Helper()

	db, err := sql.Open("sqlite", ":memory:")
	if err != nil {
		t.Fatalf("failed to open test db: %v", err)
	}

	db.SetMaxOpenConns(1)

	if err := database.ApplyMigrations(context.Background(), db, nil); err != nil {
		t.Fatalf("failed to apply migrations: %v", err)
	}

	t.Cleanup(func() { _ = db.Close() })

	return db
}

// newEvictionTestCache creates a Cache backed by a temp directory and
// an in-memory database, with the given size limit.
func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
	t.Helper()

	tmpDir := t.TempDir()
	db := evictionTestDB(t)

	// maxBytes zero mirrors the production mapping of
	// cache_max_bytes: 0 (handlers sets DisableDiskCache); at the
	// CacheConfig layer itself a zero MaxBytes means "no limit" for
	// backwards compatibility with existing fixtures.
	cache, err := NewCache(db, CacheConfig{
		StateDir:         tmpDir,
		CacheTTL:         time.Hour,
		NegativeTTL:      5 * time.Minute,
		MaxBytes:         maxBytes,
		DisableDiskCache: maxBytes == 0,
	})
	if err != nil {
		t.Fatalf("failed to create cache: %v", err)
	}

	return cache, tmpDir
}

// storeEvictionTestSource stores content as a fetched source for
// host/path and returns the resulting content hash.
func storeEvictionTestSource(
	t *testing.T, cache *Cache, host, path string, content []byte,
) ContentHash {
	t.Helper()

	req := &ImageRequest{
		SourceHost: host,
		SourcePath: path,
		Format:     FormatJPEG,
		Quality:    85,
		FitMode:    FitCover,
	}

	result := &httpfetcher.FetchResult{
		StatusCode:    200,
		ContentType:   "image/jpeg",
		ContentLength: int64(len(content)),
		Headers:       map[string][]string{"Content-Type": {"image/jpeg"}},
	}

	hash, err := cache.StoreSource(context.Background(), req, bytes.NewReader(content), result)
	if err != nil {
		t.Fatalf("StoreSource(%s%s) failed: %v", host, path, err)
	}

	return hash
}

// storeEvictionTestVariant stores content as a processed variant under
// the given cache key.
func storeEvictionTestVariant(t *testing.T, cache *Cache, key VariantKey, content []byte) {
	t.Helper()

	if err := cache.StoreVariant(key, bytes.NewReader(content), "image/webp"); err != nil {
		t.Fatalf("StoreVariant(%s) failed: %v", key, err)
	}
}

// setVariantLastAccessed backdates the last access time of a tracked
// variant, to make LRU ordering deterministic in tests.
func setVariantLastAccessed(t *testing.T, cache *Cache, key VariantKey, when time.Time) {
	t.Helper()

	res, err := cache.db.Exec(
		`UPDATE variant_content SET last_accessed_at = ? WHERE cache_key = ?`,
		when.UTC().Format(sqliteTimestampFormat), string(key),
	)
	if err != nil {
		t.Fatalf("failed to set variant last_accessed_at: %v", err)
	}

	affected, err := res.RowsAffected()
	if err != nil {
		t.Fatalf("failed to read affected rows: %v", err)
	}

	if affected != 1 {
		t.Fatalf("variant %s has no accounting row (affected=%d); "+
			"stores must track variants in the database", key, affected)
	}
}

// setSourceLastAccessed backdates the last access time of a tracked
// source content blob.
func setSourceLastAccessed(t *testing.T, cache *Cache, hash ContentHash, when time.Time) {
	t.Helper()

	res, err := cache.db.Exec(
		`UPDATE source_content SET last_accessed_at = ? WHERE content_hash = ?`,
		when.UTC().Format(sqliteTimestampFormat), string(hash),
	)
	if err != nil {
		t.Fatalf("failed to set source last_accessed_at: %v", err)
	}

	affected, err := res.RowsAffected()
	if err != nil {
		t.Fatalf("failed to read affected rows: %v", err)
	}

	if affected != 1 {
		t.Fatalf("source %s has no accounting row (affected=%d)", hash, affected)
	}
}

// countRows returns the number of rows the given query yields.
func countRows(t *testing.T, cache *Cache, query string, args ...interface{}) int {
	t.Helper()

	var n int
	if err := cache.db.QueryRow(query, args...).Scan(&n); err != nil {
		t.Fatalf("count query %q failed: %v", query, err)
	}

	return n
}

// assertNoDanglingReferences verifies the core eviction invariant:
// every database row that references cache content on disk points at a
// file that actually exists.
func assertNoDanglingReferences(t *testing.T, cache *Cache) {
	t.Helper()

	rows, err := cache.db.Query(
		`SELECT content_hash FROM source_metadata
		 WHERE content_hash IS NOT NULL AND content_hash != ''`,
	)
	if err != nil {
		t.Fatalf("failed to query source_metadata: %v", err)
	}

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

	for rows.Next() {
		var hash string
		if err := rows.Scan(&hash); err != nil {
			t.Fatalf("failed to scan content_hash: %v", err)
		}

		if !cache.srcContent.Exists(ContentHash(hash)) {
			t.Errorf("source_metadata references content %s but the file is missing", hash)
		}
	}

	if err := rows.Err(); err != nil {
		t.Fatalf("source_metadata iteration failed: %v", err)
	}

	variantRows, err := cache.db.Query(`SELECT cache_key FROM variant_content`)
	if err != nil {
		t.Fatalf("failed to query variant_content: %v", err)
	}

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

	for variantRows.Next() {
		var key string
		if err := variantRows.Scan(&key); err != nil {
			t.Fatalf("failed to scan cache_key: %v", err)
		}

		if !cache.variants.Exists(VariantKey(key)) {
			t.Errorf("variant_content references key %s but the file is missing", key)
		}
	}

	if err := variantRows.Err(); err != nil {
		t.Fatalf("variant_content iteration failed: %v", err)
	}
}

// waitForUsageAtOrBelow polls UsageBytes until it reaches limit or the
// timeout expires, returning the last observed usage.
func waitForUsageAtOrBelow(t *testing.T, cache *Cache, limit int64, timeout time.Duration) int64 {
	t.Helper()

	deadline := time.Now().Add(timeout)

	var usage int64

	for time.Now().Before(deadline) {
		var err error

		usage, err = cache.UsageBytes(context.Background())
		if err != nil {
			t.Fatalf("UsageBytes failed: %v", err)
		}

		if usage <= limit {
			return usage
		}

		time.Sleep(25 * time.Millisecond)
	}

	return usage
}

func TestUsageBytesAccountsSourceAndVariantBytes(t *testing.T) {
	cache, _ := newEvictionTestCache(t, 1<<30)

	storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg",
		bytes.Repeat([]byte{0xAA}, 1000))
	storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg",
		bytes.Repeat([]byte{0xAB}, 2000))
	storeEvictionTestVariant(t, cache, "aabbccdd0001", bytes.Repeat([]byte{0xAC}, 500))
	storeEvictionTestVariant(t, cache, "aabbccdd0002", bytes.Repeat([]byte{0xAD}, 250))

	usage, err := cache.UsageBytes(context.Background())
	if err != nil {
		t.Fatalf("UsageBytes failed: %v", err)
	}

	if usage != 3750 {
		t.Errorf("UsageBytes = %d, want 3750 (1000+2000+500+250)", usage)
	}
}

func TestUsageBytesCountsMultiReferencedBlobOnce(t *testing.T) {
	cache, _ := newEvictionTestCache(t, 1<<30)

	content := bytes.Repeat([]byte{0xCC}, 1200)

	hashOne := storeEvictionTestSource(t, cache, "src.example.com", "/one.jpg", content)
	hashTwo := storeEvictionTestSource(t, cache, "src.example.com", "/two.jpg", content)

	if hashOne != hashTwo {
		t.Fatalf("identical content produced different hashes: %s vs %s", hashOne, hashTwo)
	}

	usage, err := cache.UsageBytes(context.Background())
	if err != nil {
		t.Fatalf("UsageBytes failed: %v", err)
	}

	if usage != 1200 {
		t.Errorf("UsageBytes = %d, want 1200 (deduplicated blob counted once)", usage)
	}
}

func TestEvictToLimitEvictsLeastRecentlyUsedFirst(t *testing.T) {
	const limit = 3000

	cache, _ := newEvictionTestCache(t, limit)

	now := time.Now()
	keys := []VariantKey{"aabbccdd0001", "aabbccdd0002", "aabbccdd0003", "aabbccdd0004"}
	fills := []byte{0x01, 0x02, 0x03, 0x04}
	ages := []time.Duration{4 * time.Hour, 3 * time.Hour, 2 * time.Hour, 1 * time.Hour}

	for i, key := range keys {
		storeEvictionTestVariant(t, cache, key, bytes.Repeat([]byte{fills[i]}, 1000))
		setVariantLastAccessed(t, cache, key, now.Add(-ages[i]))
	}

	if err := cache.EvictToLimit(context.Background()); err != nil {
		t.Fatalf("EvictToLimit failed: %v", err)
	}

	usage, err := cache.UsageBytes(context.Background())
	if err != nil {
		t.Fatalf("UsageBytes failed: %v", err)
	}

	if usage > limit {
		t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
	}

	if cache.variants.Exists(keys[0]) {
		t.Errorf("least recently used variant %s must be evicted", keys[0])
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(keys[0]),
	); n != 0 {
		t.Errorf("evicted variant %s still has %d accounting rows", keys[0], n)
	}

	for _, key := range keys[1:] {
		if !cache.variants.Exists(key) {
			t.Errorf("more recently used variant %s must survive eviction", key)
		}
	}

	assertNoDanglingReferences(t, cache)
}

func TestEvictionRemovesMultiReferencedBlobTogetherWithAllReferences(t *testing.T) {
	const limit = 1000

	cache, _ := newEvictionTestCache(t, limit)

	now := time.Now()

	// One 800-byte blob referenced by two source paths.
	sharedContent := bytes.Repeat([]byte{0xDD}, 800)
	sharedHash := storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", sharedContent)

	if h := storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", sharedContent); h != sharedHash {
		t.Fatalf("identical content produced different hashes: %s vs %s", h, sharedHash)
	}

	// A newer 600-byte blob referenced by one source path.
	recentHash := storeEvictionTestSource(t, cache, "src.example.com", "/c.jpg",
		bytes.Repeat([]byte{0xEE}, 600))

	setSourceLastAccessed(t, cache, sharedHash, now.Add(-2*time.Hour))
	setSourceLastAccessed(t, cache, recentHash, now.Add(-time.Minute))

	if err := cache.EvictToLimit(context.Background()); err != nil {
		t.Fatalf("EvictToLimit failed: %v", err)
	}

	usage, err := cache.UsageBytes(context.Background())
	if err != nil {
		t.Fatalf("UsageBytes failed: %v", err)
	}

	if usage > limit {
		t.Errorf("usage after eviction = %d, want <= %d", usage, limit)
	}

	// The multi-referenced blob must be gone from disk, from
	// source_content, and from BOTH source_metadata rows: references
	// are removed together with the blob, never left dangling.
	if cache.srcContent.Exists(sharedHash) {
		t.Errorf("evicted blob %s still exists on disk", sharedHash)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(sharedHash),
	); n != 0 {
		t.Errorf("evicted blob %s still has %d source_content rows", sharedHash, n)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(sharedHash),
	); n != 0 {
		t.Errorf("evicted blob %s still has %d source_metadata references", sharedHash, n)
	}

	// The JSON metadata sidecars for both referencing paths must be
	// removed along with the rows.
	for _, path := range []string{"/a.jpg", "/b.jpg"} {
		pathHash := HashPath(path + "?")
		if cache.srcMetadata.Exists("src.example.com", pathHash) {
			t.Errorf("metadata sidecar for %s must be removed with its row", path)
		}
	}

	// The more recently used blob survives fully intact.
	if !cache.srcContent.Exists(recentHash) {
		t.Errorf("recently used blob %s must survive eviction", recentHash)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(recentHash),
	); n != 1 {
		t.Errorf("recently used blob %s has %d source_metadata rows, want 1", recentHash, n)
	}

	assertNoDanglingReferences(t, cache)
}

func TestEvictionKeepsEverythingWhenUnderLimit(t *testing.T) {
	cache, _ := newEvictionTestCache(t, 1<<30)

	content := bytes.Repeat([]byte{0xDF}, 800)
	hash := storeEvictionTestSource(t, cache, "src.example.com", "/a.jpg", content)

	if h := storeEvictionTestSource(t, cache, "src.example.com", "/b.jpg", content); h != hash {
		t.Fatalf("identical content produced different hashes: %s vs %s", h, hash)
	}

	storeEvictionTestVariant(t, cache, "aabbccdd0001", bytes.Repeat([]byte{0xE0}, 500))

	if err := cache.EvictToLimit(context.Background()); err != nil {
		t.Fatalf("EvictToLimit failed: %v", err)
	}

	if !cache.srcContent.Exists(hash) {
		t.Errorf("blob %s must not be evicted while usage is under the limit", hash)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM source_metadata WHERE content_hash = ?`, string(hash),
	); n != 2 {
		t.Errorf("blob %s has %d source_metadata rows, want 2", hash, n)
	}

	if !cache.variants.Exists("aabbccdd0001") {
		t.Error("variant must not be evicted while usage is under the limit")
	}

	assertNoDanglingReferences(t, cache)
}

func TestZeroMaxBytesDisablesDiskCache(t *testing.T) {
	cache, tmpDir := newEvictionTestCache(t, 0)
	ctx := context.Background()

	req := &ImageRequest{
		SourceHost: "src.example.com",
		SourcePath: "/a.jpg",
		Format:     FormatJPEG,
		Quality:    85,
		FitMode:    FitCover,
	}

	// Writes are no-ops that report success.
	if err := cache.StoreVariant(CacheKey(req), bytes.NewReader([]byte("data")), "image/webp"); err != nil {
		t.Fatalf("StoreVariant on disabled cache must be a no-op, got error: %v", err)
	}

	result := &httpfetcher.FetchResult{
		StatusCode:    200,
		ContentType:   "image/jpeg",
		ContentLength: 4,
		Headers:       map[string][]string{},
	}

	hash, err := cache.StoreSource(ctx, req, bytes.NewReader([]byte("data")), result)
	if err != nil {
		t.Fatalf("StoreSource on disabled cache must be a no-op, got error: %v", err)
	}

	if hash != "" {
		t.Errorf("StoreSource on disabled cache returned hash %q, want empty", hash)
	}

	// Reads always miss.
	lookup, err := cache.Lookup(ctx, req)
	if err != nil {
		t.Fatalf("Lookup on disabled cache failed: %v", err)
	}

	if lookup.Hit {
		t.Error("Lookup on disabled cache must always miss")
	}

	srcHash, srcType, err := cache.LookupSource(ctx, req)
	if err != nil {
		t.Fatalf("LookupSource on disabled cache failed: %v", err)
	}

	if srcHash != "" || srcType != "" {
		t.Errorf("LookupSource on disabled cache = (%q, %q), want empty", srcHash, srcType)
	}

	// Nothing is tracked and nothing is written to disk.
	usage, err := cache.UsageBytes(ctx)
	if err != nil {
		t.Fatalf("UsageBytes failed: %v", err)
	}

	if usage != 0 {
		t.Errorf("UsageBytes on disabled cache = %d, want 0", usage)
	}

	if n := countRows(t, cache, `SELECT COUNT(*) FROM source_content`); n != 0 {
		t.Errorf("disabled cache wrote %d source_content rows, want 0", n)
	}

	if n := countRows(t, cache, `SELECT COUNT(*) FROM source_metadata`); n != 0 {
		t.Errorf("disabled cache wrote %d source_metadata rows, want 0", n)
	}

	if _, err := os.Stat(filepath.Join(tmpDir, "cache")); !os.IsNotExist(err) {
		t.Errorf("disabled cache must not create the cache directory tree (stat err=%v)", err)
	}

	var foundFiles []string

	walkErr := filepath.WalkDir(tmpDir, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}

		if !d.IsDir() {
			foundFiles = append(foundFiles, path)
		}

		return nil
	})
	if walkErr != nil {
		t.Fatalf("failed to walk state dir: %v", walkErr)
	}

	if len(foundFiles) != 0 {
		t.Errorf("disabled cache wrote files to disk: %v", foundFiles)
	}
}

func TestEvictionRunsUnderWritePressure(t *testing.T) {
	const limit = 1500

	cache, _ := newEvictionTestCache(t, limit)

	// An interval far longer than the test ensures only write
	// pressure can trigger eviction here.
	cache.StartEviction(time.Hour)
	defer cache.StopEviction()

	keys := []VariantKey{"aabbccdd0001", "aabbccdd0002", "aabbccdd0003"}
	fills := []byte{0x11, 0x12, 0x13}

	for i, key := range keys {
		storeEvictionTestVariant(t, cache, key, bytes.Repeat([]byte{fills[i]}, 1000))
	}

	usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
	if usage > limit {
		t.Errorf("write pressure did not trigger eviction: usage = %d, want <= %d",
			usage, limit)
	}

	assertNoDanglingReferences(t, cache)
}

func TestEvictionRunsOnPeriodicSchedule(t *testing.T) {
	const limit = 1500

	cache, _ := newEvictionTestCache(t, limit)

	// Start the evictor while the cache is empty, then create tracked
	// over-limit state WITHOUT going through the store methods, so no
	// write-pressure notification fires and only the periodic ticker
	// can trigger eviction.
	cache.StartEviction(100 * time.Millisecond)
	defer cache.StopEviction()

	keys := []VariantKey{"aabbccdd0001", "aabbccdd0002", "aabbccdd0003"}
	fills := []byte{0x21, 0x22, 0x23}

	for i, key := range keys {
		content := bytes.Repeat([]byte{fills[i]}, 1000)

		if _, err := cache.variants.Store(key, bytes.NewReader(content), "image/webp"); err != nil {
			t.Fatalf("failed to store variant file: %v", err)
		}

		if _, err := cache.db.Exec(
			`INSERT INTO variant_content (cache_key, size_bytes, content_type)
			 VALUES (?, ?, ?)`,
			string(key), len(content), "image/webp",
		); err != nil {
			t.Fatalf("failed to insert variant accounting row: %v", err)
		}
	}

	usage := waitForUsageAtOrBelow(t, cache, limit, 5*time.Second)
	if usage > limit {
		t.Errorf("periodic schedule did not trigger eviction: usage = %d, want <= %d",
			usage, limit)
	}

	assertNoDanglingReferences(t, cache)
}

func TestStartEvictionReconcilesAccountingWithDisk(t *testing.T) {
	cache, _ := newEvictionTestCache(t, 1<<30)

	// An untracked variant file on disk (e.g. written before this
	// feature existed) must be adopted into the accounting.
	untracked := bytes.Repeat([]byte{0x31}, 1000)
	if _, err := cache.variants.Store("aabbccdd0001", bytes.NewReader(untracked), "image/webp"); err != nil {
		t.Fatalf("failed to store untracked variant file: %v", err)
	}

	// An accounting row whose file is missing must be dropped.
	if _, err := cache.db.Exec(
		`INSERT INTO variant_content (cache_key, size_bytes, content_type)
		 VALUES (?, ?, ?)`,
		"deadbeef0001", 700, "image/webp",
	); err != nil {
		t.Fatalf("failed to insert stale variant accounting row: %v", err)
	}

	cache.StartEviction(time.Hour)
	defer cache.StopEviction()

	deadline := time.Now().Add(5 * time.Second)

	var usage int64

	for time.Now().Before(deadline) {
		var err error

		usage, err = cache.UsageBytes(context.Background())
		if err != nil {
			t.Fatalf("UsageBytes failed: %v", err)
		}

		if usage == 1000 {
			break
		}

		time.Sleep(25 * time.Millisecond)
	}

	if usage != 1000 {
		t.Errorf("usage after reconciliation = %d, want 1000 "+
			"(untracked file adopted, stale row dropped)", usage)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "aabbccdd0001",
	); n != 1 {
		t.Errorf("untracked variant file was not adopted into accounting (rows=%d)", n)
	}

	if n := countRows(t, cache,
		`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, "deadbeef0001",
	); n != 0 {
		t.Errorf("stale accounting row without a file was not dropped (rows=%d)", n)
	}
}