pixa/internal/imgcache/eviction.go

package imgcache

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"strings"
	"time"

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

// DefaultEvictionInterval is how often the background evictor checks
// cache usage against the configured limit, in addition to the
// write-pressure wakeups triggered by stores.
const DefaultEvictionInterval = 5 * time.Minute

// reconciliationWaitFactor is how many times as long as a
// reconciliation pass took the eviction loop waits at least, after that
// pass ends, before it starts the next one. A pass reads a row and stats a
// file for every cached file, so its time grows with the cache; waiting
// ten times as long keeps reconciliation to at most a tenth of the time,
// however large the cache is.
const reconciliationWaitFactor = 10

// evictionBatchSize is how many LRU candidates of each class (variants
// and source blobs) one eviction pass fetches from the database.
const evictionBatchSize = 100

// defaultReconciliationPageSize is the most rows one read of the
// reconciliation pass returns, unless a test sets
// Cache.reconciliationPageSize smaller. Each read is a query of its own,
// so a request waits for one page at most, however large the cache is.
const defaultReconciliationPageSize = 1000

// staleTempFileAge is how old an orphaned temp file (left behind by a
// crashed write) must be before reconciliation removes it. Fresh temp
// files may still belong to an in-flight store.
const staleTempFileAge = time.Hour

// sqliteTimestampLayout matches SQLite's CURRENT_TIMESTAMP format, so
// timestamps written by reconciliation order correctly against ones
// written by the hot path.
const sqliteTimestampLayout = "2006-01-02 15:04:05"

// tempFilePrefix is the prefix os.CreateTemp uses for in-flight cache
// writes (".tmp-*" patterns in the storage layer).
const tempFilePrefix = ".tmp-"

// fallbackContentType is the content type reconciliation records for a
// variant file it adopts whose first bytes are not those of an output
// format. Such a file is never served.
const fallbackContentType = "application/octet-stream"

// UsageBytes returns the total number of bytes of cache content
// tracked in the database (source content blobs plus processed
// variants). It reads the total the database keeps up to date as rows
// are added and removed, so it neither scans the cache directories nor
// sums the tables.
func (c *Cache) UsageBytes(ctx context.Context) (int64, error) {
	if c.disabled {
		return 0, nil
	}

	var total int64

	err := c.db.QueryRowContext(ctx,
		`SELECT total_size_bytes FROM cache_usage WHERE id = 1`,
	).Scan(&total)
	if err != nil {
		return 0, fmt.Errorf("failed to read cache usage: %w", err)
	}

	return total, nil
}

// evictionCandidate is one LRU eviction victim candidate: either a
// processed variant (isVariant true, identified by cacheKey) or a
// source content blob (identified by sourceKey).
type evictionCandidate struct {
	isVariant      bool
	cacheKey       imagerequest.VariantKey
	sourceKey      imagerequest.SourceKey
	sizeBytes      int64
	lastAccessedAt string
}

// EvictToLimit evicts least-recently-used cache entries until total
// tracked usage is at or below the configured MaxBytes limit. It is a
// no-op when the cache is disabled or no limit is configured.
func (c *Cache) EvictToLimit(ctx context.Context) error {
	if c.disabled || c.config.MaxBytes <= 0 {
		return nil
	}

	for {
		usage, err := c.UsageBytes(ctx)
		if err != nil {
			return err
		}

		if usage <= c.config.MaxBytes {
			return nil
		}

		freed, err := c.evictBatch(ctx, usage-c.config.MaxBytes)
		if err != nil {
			return err
		}

		if freed == 0 {
			c.log.Warn("cache eviction made no progress",
				"usage_bytes", usage,
				"cache_max_bytes", c.config.MaxBytes,
			)

			return nil
		}

		c.log.Info("evicted cache content",
			"freed_bytes", freed,
			"usage_bytes", usage-freed,
			"cache_max_bytes", c.config.MaxBytes,
		)
	}
}

// evictBatch fetches one batch of LRU candidates across variants and
// source blobs and evicts them oldest-first until excessBytes are
// freed or the batch is exhausted. It returns the bytes freed. A
// candidate that fails once ctx is cancelled (every one started after
// that fails at its first database call) ends the batch with ctx's
// error, without a warning.
func (c *Cache) evictBatch(ctx context.Context, excessBytes int64) (int64, error) {
	candidates, err := c.evictionCandidates(ctx)
	if err != nil {
		return 0, err
	}

	var freed int64

	for _, candidate := range candidates {
		if freed >= excessBytes {
			break
		}

		err := c.evictCandidate(ctx, candidate)
		if err != nil {
			if ctx.Err() != nil {
				return freed, ctx.Err()
			}

			c.log.Warn("failed to evict cache entry",
				"cache_key", candidate.cacheKey,
				"source_key", candidate.sourceKey,
				"error", err,
			)

			continue
		}

		freed += candidate.sizeBytes
	}

	return freed, nil
}

// evictCandidate removes a single eviction victim.
func (c *Cache) evictCandidate(ctx context.Context, candidate evictionCandidate) error {
	if candidate.isVariant {
		return c.evictVariant(ctx, candidate.cacheKey)
	}

	return c.evictSourceBlob(ctx, candidate.sourceKey)
}

// evictionCandidates returns up to evictionBatchSize variants and
// evictionBatchSize source blobs, merged into a single list ordered by
// last access time (oldest first).
func (c *Cache) evictionCandidates(ctx context.Context) ([]evictionCandidate, error) {
	variants, err := c.variantCandidates(ctx)
	if err != nil {
		return nil, err
	}

	sources, err := c.sourceCandidates(ctx)
	if err != nil {
		return nil, err
	}

	// Merge the two lists, each already sorted oldest-first. SQLite
	// CURRENT_TIMESTAMP strings compare correctly lexicographically.
	merged := make([]evictionCandidate, 0, len(variants)+len(sources))

	for len(variants) > 0 && len(sources) > 0 {
		if variants[0].lastAccessedAt <= sources[0].lastAccessedAt {
			merged = append(merged, variants[0])
			variants = variants[1:]
		} else {
			merged = append(merged, sources[0])
			sources = sources[1:]
		}
	}

	merged = append(merged, variants...)
	merged = append(merged, sources...)

	return merged, nil
}

// variantCandidates returns the least recently used variants.
func (c *Cache) variantCandidates(ctx context.Context) ([]evictionCandidate, error) {
	rows, err := c.db.QueryContext(ctx, `
		SELECT cache_key, size_bytes, last_accessed_at
		FROM variant_content
		ORDER BY last_accessed_at ASC, cache_key ASC
		LIMIT ?
	`, evictionBatchSize)
	if err != nil {
		return nil, fmt.Errorf("failed to query variant eviction candidates: %w", err)
	}

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

	var candidates []evictionCandidate

	for rows.Next() {
		candidate := evictionCandidate{isVariant: true}

		var key string

		err := rows.Scan(&key, &candidate.sizeBytes, &candidate.lastAccessedAt)
		if err != nil {
			return nil, fmt.Errorf("failed to scan variant candidate: %w", err)
		}

		candidate.cacheKey = imagerequest.VariantKey(key)
		candidates = append(candidates, candidate)
	}

	err = rows.Err()
	if err != nil {
		return nil, fmt.Errorf("variant candidate iteration failed: %w", err)
	}

	return candidates, nil
}

// sourceCandidatesQuery selects the least recently used source blobs. It
// orders by the last_accessed_at column itself, not by an expression, so
// SQLite reads the rows in order from that column's index instead of
// sorting the whole table.
const sourceCandidatesQuery = `
	SELECT source_key, size_bytes, last_accessed_at
	FROM source_content
	ORDER BY last_accessed_at ASC, source_key ASC
	LIMIT ?`

// sourceCandidates returns the least recently used source blobs.
func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, error) {
	rows, err := c.db.QueryContext(ctx, sourceCandidatesQuery, evictionBatchSize)
	if err != nil {
		return nil, fmt.Errorf("failed to query source eviction candidates: %w", err)
	}

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

	var candidates []evictionCandidate

	for rows.Next() {
		var candidate evictionCandidate

		var key string

		err := rows.Scan(&key, &candidate.sizeBytes, &candidate.lastAccessedAt)
		if err != nil {
			return nil, fmt.Errorf("failed to scan source candidate: %w", err)
		}

		candidate.sourceKey = imagerequest.SourceKey(key)
		candidates = append(candidates, candidate)
	}

	err = rows.Err()
	if err != nil {
		return nil, fmt.Errorf("source candidate iteration failed: %w", err)
	}

	return candidates, nil
}

// evictVariant removes one variant: accounting row first, then the
// file, so the database never references a deleted file.
func (c *Cache) evictVariant(
	ctx context.Context, cacheKey imagerequest.VariantKey,
) error {
	_, err := c.db.ExecContext(ctx,
		`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
	if err != nil {
		return fmt.Errorf("failed to delete variant accounting row: %w", err)
	}

	err = c.variants.Delete(cacheKey)
	if err != nil {
		return err
	}

	return nil
}

// evictSourceBlob removes one source: its row first, then its file, as
// evictVariant does for a variant. It holds no lock: a store of the same
// source in between can leave the source's row with no file, which
// LookupSource treats as a miss.
func (c *Cache) evictSourceBlob(ctx context.Context, key imagerequest.SourceKey) error {
	_, err := c.db.ExecContext(ctx,
		`DELETE FROM source_content WHERE source_key = ?`, string(key))
	if err != nil {
		return fmt.Errorf("failed to delete source content row: %w", err)
	}

	return c.srcContent.Delete(key)
}

// notifyWritePressure wakes the background evictor after a store, so
// eviction under write pressure happens promptly without blocking the
// storing request. The notification channel has capacity one and drops
// when a wakeup is already pending.
func (c *Cache) notifyWritePressure() {
	if c.disabled || c.config.MaxBytes <= 0 {
		return
	}

	select {
	case c.evictionPressure <- struct{}{}:
	default:
	}
}

// StartEviction launches the background eviction goroutine, which
// removes the expired negative cache rows at startup and again on every
// periodic tick thereafter, reconciles the database accounting with the
// cache directories at startup and again on the first periodic tick
// that comes at least ten times as long after a reconciliation pass
// ended as that pass took, and evicts to the configured limit on the given
// periodic interval and on write-pressure notifications. On a
// disabled cache it only removes the expired negative cache rows, as
// the negative cache is used either way. It is a no-op when already
// started. The goroutine outlives the caller, so it runs with its own
// context, which StopEviction cancels.
func (c *Cache) StartEviction(interval time.Duration) {
	if c.evictionCancel != nil {
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	c.evictionCancel = cancel

	go c.evictionLoop(ctx, interval)
}

// StopEviction cancels the background eviction goroutine, which
// interrupts a pass in progress, and waits for it to exit or for ctx to
// end, whichever comes first. In the second case it returns an error
// wrapping ctx's error. It is safe to call when eviction was never
// started, and safe to call more than once.
func (c *Cache) StopEviction(ctx context.Context) error {
	if c.evictionCancel == nil {
		return nil
	}

	c.evictionCancel()

	select {
	case <-c.evictionDone:
		return nil
	case <-ctx.Done():
		return fmt.Errorf("cache eviction still running: %w", ctx.Err())
	}
}

// evictionLoop is the body of the background eviction goroutine. It
// returns when ctx is cancelled, and starts no pass after that.
func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
	defer close(c.evictionDone)

	nextReconciliation := c.runReconciliationPass(ctx)
	c.runCleanExpired(ctx)
	c.runEvictionPass(ctx)

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			// Reconciliation walks the cache directories and reads a
			// row for every file, so it runs only on the periodic
			// ticker rather than on every write-pressure wakeup, and
			// only on the first tick at or after nextReconciliation,
			// which is ten times as long after the last pass ended as
			// that pass took. Passes then never run back to back,
			// however long one takes. A file whose best-effort
			// accounting insert failed in a store stays unaccounted for
			// until the next pass, up to about one eviction interval
			// plus ten times the last pass's time.
			if !time.Now().Before(nextReconciliation) {
				nextReconciliation = c.runReconciliationPass(ctx)
			}

			c.runCleanExpired(ctx)
		case <-c.evictionPressure:
		}

		c.runEvictionPass(ctx)
	}
}

// runEvictionPass runs one eviction pass, logging failures instead of
// propagating them (the loop must keep running). It does nothing once
// ctx is cancelled.
func (c *Cache) runEvictionPass(ctx context.Context) {
	if ctx.Err() != nil {
		return
	}

	err := c.EvictToLimit(ctx)
	if err != nil {
		c.log.Warn("cache eviction pass failed", "error", err)
	}
}

// runReconciliationPass runs one reconciliation pass, logging failures
// instead of propagating them (the loop must keep running), and returns
// the earliest time the next pass may start: reconciliationWaitFactor
// times as long after this pass ended as it took. It does nothing once
// ctx is cancelled.
func (c *Cache) runReconciliationPass(ctx context.Context) time.Time {
	if ctx.Err() != nil {
		return time.Now()
	}

	started := time.Now()

	err := c.reconcileAccounting(ctx)
	if err != nil {
		c.log.Warn("cache accounting reconciliation failed", "error", err)
	}

	ended := time.Now()

	return ended.Add(reconciliationWaitFactor * ended.Sub(started))
}

// runCleanExpired runs CleanExpired, logging a failure instead of
// propagating it (the loop must keep running). It does nothing once ctx
// is cancelled.
func (c *Cache) runCleanExpired(ctx context.Context) {
	if ctx.Err() != nil {
		return
	}

	err := c.CleanExpired(ctx)
	if err != nil {
		c.log.Warn("negative cache cleanup failed", "error", err)
	}
}

// reconcileAccounting synchronizes the database size accounting with
// the actual contents of the cache directories. It runs at startup and
// then on the first periodic eviction tick that comes at least ten times
// as long after the last pass ended as that pass took, off the request
// hot path: it adopts variant files that predate the accounting table
// (or whose accounting insert failed, e.g. StoreVariant's best-effort
// insert under transient DB contention), drops variant accounting rows
// whose files are missing, removes source blob files the database does
// not know, removes source and variant files that are empty or not the
// size their rows record, sweeps stale temp files left behind by crashed
// writes, and last checks the total cache usage against the tables. It
// reads the tables a page at a time.
// Running it periodically, not just once, bounds how long such drift can
// accumulate unaccounted for on a long-running process to about one
// eviction interval plus ten times the last pass's time.
// Once ctx is cancelled, it stops at the next file or row and returns
// ctx's error.
func (c *Cache) reconcileAccounting(ctx context.Context) error {
	if c.disabled {
		return nil
	}

	err := c.reconcileVariantFiles(ctx)
	if err != nil {
		return err
	}

	err = c.reconcileVariantRows(ctx)
	if err != nil {
		return err
	}

	err = c.reconcileSourceFiles(ctx)
	if err != nil {
		return err
	}

	return c.reconcileUsageTotal(ctx)
}

// reconcileFiles walks baseDir, one of the cache's storage directories,
// sweeping stale temp files and passing each other file, with its name, to
// reconcileFile. Once ctx is cancelled it stops at the next file and
// returns ctx's error.
func (c *Cache) reconcileFiles(
	ctx context.Context, baseDir string,
	reconcileFile func(path string, entry fs.DirEntry, name string) error,
) error {
	return filepath.WalkDir(
		baseDir,
		func(path string, entry fs.DirEntry, err error) error {
			if ctx.Err() != nil {
				return ctx.Err()
			}

			if err != nil || entry.IsDir() {
				return err
			}

			name := entry.Name()

			if strings.HasPrefix(name, tempFilePrefix) {
				c.sweepStaleTempFile(path, entry)

				return nil
			}

			return reconcileFile(path, entry, name)
		},
	)
}

// reconcileVariantFiles walks the variant storage directory, checking
// each variant file against its accounting row (reconcileVariantFile)
// and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
	return c.reconcileFiles(ctx, c.variants.baseDir,
		func(path string, entry fs.DirEntry, name string) error {
			return c.reconcileVariantFile(ctx, path, entry, imagerequest.VariantKey(name))
		})
}

// reconcileVariantFile checks one variant file against its accounting
// row. A file without a row is adopted into the accounting, unless it is
// empty. A file that is empty, or not the size its row records, as a
// crash or a restored snapshot can leave it, is removed with its row and
// a warning naming it, so the variant is converted again.
func (c *Cache) reconcileVariantFile(
	ctx context.Context, path string, entry fs.DirEntry, cacheKey imagerequest.VariantKey,
) error {
	info, err := entry.Info()
	if err != nil {
		return fmt.Errorf("failed to stat variant file: %w", err)
	}

	var recordedSize int64

	err = c.db.QueryRowContext(ctx,
		`SELECT size_bytes FROM variant_content WHERE cache_key = ?`, string(cacheKey),
	).Scan(&recordedSize)

	switch {
	case errors.Is(err, sql.ErrNoRows) && info.Size() > 0:
		return c.adoptVariantFile(ctx, info, cacheKey)
	case err != nil && !errors.Is(err, sql.ErrNoRows):
		return fmt.Errorf("failed to check variant accounting row: %w", err)
	case info.Size() > 0 && info.Size() == recordedSize:
		return nil
	}

	// The file is empty, or not the size its row records
	c.log.Warn("removing damaged cached variant file", "path", path,
		"file_bytes", info.Size(), "recorded_bytes", recordedSize)

	return c.evictVariant(ctx, cacheKey)
}

// adoptVariantFile inserts an accounting row for a variant file that
// has none, using the file's size and modification time, and the content
// type its first bytes show, as a hit reads it.
func (c *Cache) adoptVariantFile(
	ctx context.Context, info fs.FileInfo, cacheKey imagerequest.VariantKey,
) error {
	modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
	contentType := fallbackContentType

	content, _, detectedType, err := c.variants.LoadWithContentType(cacheKey)
	if err == nil {
		_ = content.Close()
		contentType = detectedType
	}

	_, err = c.db.ExecContext(ctx, `
		INSERT INTO variant_content
			(cache_key, size_bytes, content_type, created_at, last_accessed_at)
		VALUES (?, ?, ?, ?, ?)
	`, string(cacheKey), info.Size(), contentType, modTime, modTime)
	if err != nil {
		return fmt.Errorf("failed to adopt variant file into accounting: %w", err)
	}

	c.log.Info("adopted untracked variant file into size accounting",
		"cache_key", cacheKey, "size_bytes", info.Size())

	return nil
}

// reconcileVariantRows drops accounting rows whose variant files are
// missing, so the database never references deleted content.
func (c *Cache) reconcileVariantRows(ctx context.Context) error {
	var after imagerequest.VariantKey

	for {
		keys, err := c.variantKeysAfter(ctx, after)
		if err != nil {
			return err
		}

		if c.reconciliationReadTestHook != nil {
			c.reconciliationReadTestHook(len(keys))
		}

		if len(keys) == 0 {
			return nil
		}

		for _, key := range keys {
			if ctx.Err() != nil {
				return ctx.Err()
			}

			if c.variants.Exists(key) {
				continue
			}

			_, err := c.db.ExecContext(ctx,
				`DELETE FROM variant_content WHERE cache_key = ?`, string(key))
			if err != nil {
				return fmt.Errorf("failed to drop stale variant accounting row: %w", err)
			}

			c.log.Info("dropped accounting row for missing variant file", "cache_key", key)
		}

		after = keys[len(keys)-1]
	}
}

// variantKeysAfter returns, in order, up to c.reconciliationPageSize
// tracked variant cache keys that sort after the given one.
func (c *Cache) variantKeysAfter(
	ctx context.Context, after imagerequest.VariantKey,
) ([]imagerequest.VariantKey, error) {
	return queryStringColumn[imagerequest.VariantKey](ctx, c.db, `
		SELECT cache_key FROM variant_content
		WHERE cache_key > ? ORDER BY cache_key LIMIT ?
	`, "variant keys", "variant key", string(after), c.reconciliationPageSize)
}

// queryStringColumn runs a single-column query with args and returns the
// column values as T. plural names the set for the query and scan
// failure messages; singular names one row for the scan and iteration
// failure messages.
func queryStringColumn[T ~string](
	ctx context.Context, db *sql.DB, query, plural, singular string, args ...any,
) ([]T, error) {
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("failed to query %s: %w", plural, err)
	}

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

	var values []T

	for rows.Next() {
		var value string

		err := rows.Scan(&value)
		if err != nil {
			return nil, fmt.Errorf("failed to scan %s: %w", singular, err)
		}

		values = append(values, T(value))
	}

	err = rows.Err()
	if err != nil {
		return nil, fmt.Errorf("%s iteration failed: %w", singular, err)
	}

	return values, nil
}

// reconcileSourceFiles walks the source content directory, checking
// each blob file against its source_content row (reconcileSourceFile)
// and sweeping stale temp files.
func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
	return c.reconcileFiles(ctx, c.srcContent.baseDir,
		func(path string, entry fs.DirEntry, name string) error {
			return c.reconcileSourceFile(ctx, path, entry, imagerequest.SourceKey(name))
		})
}

// reconcileSourceFile checks one source blob file against its
// source_content row. A file without a row is removed, as no lookup
// serves it and no eviction removes it. A file that is empty, or not the
// size its row records, as a crash or a restored snapshot can leave it,
// is removed with its row and a warning naming it, so the source is
// fetched again.
func (c *Cache) reconcileSourceFile(
	ctx context.Context, path string, entry fs.DirEntry, key imagerequest.SourceKey,
) error {
	info, err := entry.Info()
	if err != nil {
		return fmt.Errorf("failed to stat source content file: %w", err)
	}

	var recordedSize int64

	err = c.db.QueryRowContext(ctx,
		`SELECT size_bytes FROM source_content WHERE source_key = ?`,
		string(key),
	).Scan(&recordedSize)

	switch {
	case errors.Is(err, sql.ErrNoRows):
		return c.removeUntrackedSourceFile(path, key)
	case err != nil:
		return fmt.Errorf("failed to check source content row: %w", err)
	case info.Size() > 0 && info.Size() == recordedSize:
		return nil
	}

	// The file is empty, or not the size its row records
	c.log.Warn("removing damaged cached source file", "path", path,
		"file_bytes", info.Size(), "recorded_bytes", recordedSize)

	return c.evictSourceBlob(ctx, key)
}

// removeUntrackedSourceFile deletes a source blob file that has no
// source_content row.
func (c *Cache) removeUntrackedSourceFile(
	path string, key imagerequest.SourceKey,
) error {
	err := os.Remove(path)
	if err != nil && !os.IsNotExist(err) {
		return fmt.Errorf("failed to remove untracked source file: %w", err)
	}

	c.log.Info("removed untracked source content file", "source_key", key)

	return nil
}

// Each of these queries sums size_bytes over the next page of rows of
// one content table, the rows that sort after a key, and returns the
// page's last key, the sum and the number of rows in the page. Past the
// last row the page has no rows and the key is NULL.
const (
	sourceSizePageQuery = `
		SELECT MAX(source_key), COALESCE(SUM(size_bytes), 0), COUNT(*)
		FROM (
			SELECT source_key, size_bytes FROM source_content
			WHERE source_key > ? ORDER BY source_key LIMIT ?
		)`
	variantSizePageQuery = `
		SELECT MAX(cache_key), COALESCE(SUM(size_bytes), 0), COUNT(*)
		FROM (
			SELECT cache_key, size_bytes FROM variant_content
			WHERE cache_key > ? ORDER BY cache_key LIMIT ?
		)`
)

// sumContentSizeBytes sums size_bytes over both content tables, a page
// of rows per query.
func (c *Cache) sumContentSizeBytes(ctx context.Context) (int64, error) {
	sourceBytes, err := c.sumSizeBytesInPages(ctx, sourceSizePageQuery)
	if err != nil {
		return 0, err
	}

	variantBytes, err := c.sumSizeBytesInPages(ctx, variantSizePageQuery)
	if err != nil {
		return 0, err
	}

	return sourceBytes + variantBytes, nil
}

// sumSizeBytesInPages runs pageQuery, one of the size page queries
// above, from the first page to the last and adds up the page sums.
func (c *Cache) sumSizeBytesInPages(
	ctx context.Context, pageQuery string,
) (int64, error) {
	var total int64

	after := ""

	for {
		var lastKey sql.NullString

		var pageBytes int64

		var pageRows int

		err := c.db.QueryRowContext(ctx, pageQuery, after, c.reconciliationPageSize).
			Scan(&lastKey, &pageBytes, &pageRows)
		if err != nil {
			return 0, fmt.Errorf("failed to sum cache content sizes: %w", err)
		}

		if c.reconciliationReadTestHook != nil {
			c.reconciliationReadTestHook(pageRows)
		}

		if pageRows == 0 {
			return total, nil
		}

		total += pageBytes
		after = lastKey.String
	}
}

// reconcileUsageTotal checks the total cache usage the database keeps
// against size_bytes summed over both content tables, and corrects the
// total when they differ.
func (c *Cache) reconcileUsageTotal(ctx context.Context) error {
	var totalBytes, changeCount int64

	err := c.db.QueryRowContext(ctx,
		`SELECT total_size_bytes, change_count FROM cache_usage WHERE id = 1`,
	).Scan(&totalBytes, &changeCount)
	if err != nil {
		return fmt.Errorf("failed to read cache usage: %w", err)
	}

	sumBytes, err := c.sumContentSizeBytes(ctx)
	if err != nil {
		return err
	}

	if sumBytes == totalBytes {
		return nil
	}

	corrected, err := c.correctUsageTotal(ctx, sumBytes, changeCount)
	if err != nil || !corrected {
		return err
	}

	c.log.Warn("corrected total cache usage to the sum of the content tables",
		"previous_usage_bytes", totalBytes, "usage_bytes", sumBytes)

	return nil
}

// correctUsageTotal sets the total cache usage to sumBytes, a sum of the
// content tables taken when the change count was changeCount, and
// reports whether it did. If a row was added, removed or resized since,
// the count has moved and the total is left alone: the sum may have
// missed that change, and the next pass checks again.
func (c *Cache) correctUsageTotal(
	ctx context.Context, sumBytes, changeCount int64,
) (bool, error) {
	result, err := c.db.ExecContext(ctx, `
		UPDATE cache_usage SET total_size_bytes = ?
		WHERE id = 1 AND change_count = ?
	`, sumBytes, changeCount)
	if err != nil {
		return false, fmt.Errorf("failed to correct cache usage: %w", err)
	}

	affected, err := result.RowsAffected()
	if err != nil {
		return false, fmt.Errorf("failed to read cache usage correction: %w", err)
	}

	return affected > 0, nil
}

// sweepStaleTempFile removes a temp file left behind by a crashed
// write once it is old enough that no in-flight store can own it.
func (c *Cache) sweepStaleTempFile(path string, entry fs.DirEntry) {
	info, err := entry.Info()
	if err != nil {
		return
	}

	if time.Since(info.ModTime()) < staleTempFileAge {
		return
	}

	err = os.Remove(path)
	if err != nil && !os.IsNotExist(err) {
		c.log.Warn("failed to remove stale temp file", "path", path, "error", err)

		return
	}

	c.log.Info("removed stale temp file", "path", path)
}