Build the image pipeline in pixad's fx graph (closes #313)

internal/app now provides the cache, the fetcher, the processor, the
image service, the session manager and the encrypted URL generator, and
handlers.New receives them built. database.New opens the database, so
these parts can read it while fx builds them; the cache's stop hook
still runs before the database's.

Each setting's default is declared in internal/config, the default
cache_max_bytes included, which internal/app works out before it builds
the cache. The cache's MaxBytes is the one value for the disk cache
limit, and 0 turns the disk cache off. The fetcher and the processor
take explicit values. The server tests build the pipeline themselves,
as they cannot import internal/app.

Model: opus-5-5

internal/app/app.go

@@ -1,4 +1,5 @@
-// Package app reads the pixad command line and runs the server.
+// Package app reads the pixad command line, builds pixad's parts from the
+// config and runs the server.
 package app
 
 import (
@@ -62,6 +63,12 @@ func run(_ *cobra.Command, _ []string) {
 			server.New,
 			middleware.New,
 			healthcheck.New,
+			newCache,
+			newFetcher,
+			newProcessor,
+			newImageService,
+			newSessionManager,
+			newEncryptedURLGenerator,
 			// The registry /metrics serves
 			func() prometheus.Registerer { return prometheus.DefaultRegisterer },
 		),

internal/app/cache_max_bytes_internal_test.go

@@ -0,0 +1,307 @@
+package app
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/fx/fxtest"
+
+	"sneak.berlin/go/pixa/internal/config"
+	"sneak.berlin/go/pixa/internal/database"
+	"sneak.berlin/go/pixa/internal/globals"
+	"sneak.berlin/go/pixa/internal/logger"
+)
+
+// Static errors returned by the stub free-space probes below.
+var (
+	errTestStatfsFailed     = errors.New("statfs failed")
+	errTestProbeNotExpected = errors.New("probe must not be called")
+)
+
+// newTestDB opens an in-memory database with pixa's schema, on one
+// connection, as pixad's database has.
+func newTestDB(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)
+
+	err = database.ApplyMigrations(context.Background(), db, nil)
+	if err != nil {
+		t.Fatalf("failed to apply migrations: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	return db
+}
+
+// TestCacheMaxBytesFromConfig checks the disk cache limit worked out from
+// cache_max_bytes: omitted, it is the default from the free space; 0 turns
+// the disk cache off; a positive value is the limit, unchanged.
+func TestCacheMaxBytesFromConfig(t *testing.T) {
+	t.Parallel()
+
+	const oneGiB = 1 << 30
+
+	cases := []struct {
+		name                  string
+		cacheMaxBytes         int64
+		cacheMaxBytesExplicit bool
+		wantMaxBytes          int64
+	}{
+		{
+			name:                  "cache_max_bytes omitted",
+			cacheMaxBytes:         0,
+			cacheMaxBytesExplicit: false,
+			// 75% of the 4 GiB the probe reports free
+			wantMaxBytes: 3221225472,
+		},
+		{
+			name:                  "cache_max_bytes: 0",
+			cacheMaxBytes:         0,
+			cacheMaxBytesExplicit: true,
+			wantMaxBytes:          0,
+		},
+		{
+			name:                  "cache_max_bytes: 1 GiB",
+			cacheMaxBytes:         oneGiB,
+			cacheMaxBytesExplicit: true,
+			wantMaxBytes:          oneGiB,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			cfg := &config.Config{
+				StateDir:              t.TempDir(),
+				CacheMaxBytes:         tc.cacheMaxBytes,
+				CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
+			}
+
+			got, err := cacheMaxBytes(t.Context(), cfg, newTestDB(t),
+				slog.New(slog.DiscardHandler),
+				func(string) (uint64, error) { return 4294967296, nil })
+			if err != nil {
+				t.Fatalf("cacheMaxBytes() error = %v", err)
+			}
+
+			t.Logf("MaxBytes = %d", got)
+
+			if got != tc.wantMaxBytes {
+				t.Errorf("MaxBytes = %d, want %d", got, tc.wantMaxBytes)
+			}
+		})
+	}
+}
+
+// TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes builds and starts the
+// cache once with cache_max_bytes omitted and once with cache_max_bytes:
+// 0, and checks by whether the cache directories were created that the
+// disk cache is on in the first case and off in the second.
+func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name                  string
+		cacheMaxBytesExplicit bool
+		wantDiskCache         bool
+	}{
+		{name: "cache_max_bytes omitted", cacheMaxBytesExplicit: false, wantDiskCache: true},
+		{name: "cache_max_bytes: 0", cacheMaxBytesExplicit: true, wantDiskCache: false},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			stateDir := t.TempDir()
+			cfg := &config.Config{
+				SigningKey:            testSigningKey,
+				StateDir:              stateDir,
+				DBURL:                 "file:" + filepath.Join(stateDir, "state.sqlite3"),
+				CacheMaxBytes:         0,
+				CacheMaxBytesExplicit: tc.cacheMaxBytesExplicit,
+			}
+
+			lc := fxtest.NewLifecycle(t)
+
+			log, err := logger.New(lc, logger.Params{Globals: &globals.Globals{}})
+			if err != nil {
+				t.Fatalf("logger.New() error = %v", err)
+			}
+
+			db, err := database.New(lc, database.Params{Logger: log, Config: cfg})
+			if err != nil {
+				t.Fatalf("database.New() error = %v", err)
+			}
+
+			_, err = newCache(lc, cfg, db, log, prometheus.NewRegistry())
+			if err != nil {
+				t.Fatalf("newCache() error = %v", err)
+			}
+
+			lc.RequireStart()
+			t.Cleanup(lc.RequireStop)
+
+			_, err = os.Stat(filepath.Join(stateDir, "cache", "variants"))
+			gotDiskCache := err == nil
+
+			if gotDiskCache != tc.wantDiskCache {
+				t.Errorf("cache directories created = %v, want %v",
+					gotDiskCache, tc.wantDiskCache)
+			}
+		})
+	}
+}
+
+// TestDefaultCacheMaxBytesCountsWhatTheCacheHolds verifies that the
+// default limit is 75% of the free space plus what the cache already
+// holds, so a cache filled to its limit keeps that limit across a
+// restart instead of shrinking to 75% of the space left free.
+func TestDefaultCacheMaxBytesCountsWhatTheCacheHolds(t *testing.T) {
+	t.Parallel()
+
+	cfg := &config.Config{StateDir: t.TempDir()}
+	db := newTestDB(t)
+	log := slog.New(slog.DiscardHandler)
+
+	// Empty cache, 4 GiB free -> 3 GiB default.
+	got, err := cacheMaxBytes(t.Context(), cfg, db, log,
+		func(string) (uint64, error) { return 4294967296, nil })
+	if err != nil {
+		t.Fatalf("cacheMaxBytes returned error: %v", err)
+	}
+
+	t.Logf("default for an empty cache with 4 GiB free: %d", got)
+
+	if got != 3221225472 {
+		t.Errorf("default for an empty cache = %d, want 3221225472 (75%% of 4 GiB)",
+			got)
+	}
+
+	// The cache now holds those 3 GiB, which leaves 1 GiB free.
+	_, err = db.ExecContext(t.Context(),
+		`INSERT INTO variant_content (cache_key, size_bytes) VALUES (?, ?)`,
+		"aabbccdd0001", 3221225472,
+	)
+	if err != nil {
+		t.Fatalf("failed to insert variant accounting row: %v", err)
+	}
+
+	got, err = cacheMaxBytes(t.Context(), cfg, db, log,
+		func(string) (uint64, error) { return 1073741824, nil })
+	if err != nil {
+		t.Fatalf("cacheMaxBytes returned error: %v", err)
+	}
+
+	t.Logf("default for a cache holding 3 GiB with 1 GiB free: %d", got)
+
+	if got != 3221225472 {
+		t.Errorf("default for a cache holding 3 GiB with 1 GiB free = %d, "+
+			"want 3221225472 (75%% of 1 GiB + 3 GiB)", got)
+	}
+}
+
+// TestDefaultCacheMaxBytesPropagatesProbeError verifies that a failing
+// free-space probe produces an error naming cache_max_bytes, instead of
+// a silently wrong default.
+func TestDefaultCacheMaxBytesPropagatesProbeError(t *testing.T) {
+	t.Parallel()
+
+	_, err := cacheMaxBytes(t.Context(), &config.Config{StateDir: t.TempDir()},
+		newTestDB(t), slog.New(slog.DiscardHandler),
+		func(string) (uint64, error) { return 0, errTestStatfsFailed })
+	if err == nil {
+		t.Fatal("probe failure must produce an error, got nil")
+	}
+
+	t.Logf("got expected error: %v", err)
+
+	if !strings.Contains(err.Error(), "cache_max_bytes") {
+		t.Errorf("error %q does not name cache_max_bytes", err.Error())
+	}
+}
+
+// TestDefaultCacheMaxBytesProbesTheCacheDirectory verifies that with
+// cache_max_bytes omitted the limit is the computed default, and that the
+// probe is pointed at <state_dir>/cache/, which must be created first so
+// statfs measures the right filesystem.
+func TestDefaultCacheMaxBytesProbesTheCacheDirectory(t *testing.T) {
+	t.Parallel()
+
+	stateDir := t.TempDir()
+	wantCacheDir := filepath.Join(stateDir, "cache")
+
+	var probedPath string
+
+	// 4 GiB free -> 3 GiB default.
+	probe := func(path string) (uint64, error) {
+		probedPath = path
+
+		info, err := os.Stat(path)
+		if err != nil || !info.IsDir() {
+			t.Errorf("cache directory %q was not created before probing: info=%v err=%v",
+				path, info, err)
+		}
+
+		return 4294967296, nil
+	}
+
+	got, err := cacheMaxBytes(t.Context(), &config.Config{StateDir: stateDir},
+		newTestDB(t), slog.New(slog.DiscardHandler), probe)
+	if err != nil {
+		t.Fatalf("cacheMaxBytes returned error: %v", err)
+	}
+
+	if got != 3221225472 {
+		t.Errorf("MaxBytes = %d, want computed default 3221225472", got)
+	}
+
+	if probedPath != wantCacheDir {
+		t.Errorf("free space probed at %q, want cache directory %q",
+			probedPath, wantCacheDir)
+	}
+}
+
+// TestExplicitCacheMaxBytesIsKept verifies that an explicit
+// cache_max_bytes is kept exactly as given and the free-space probe is
+// never consulted.
+func TestExplicitCacheMaxBytesIsKept(t *testing.T) {
+	t.Parallel()
+
+	probe := func(string) (uint64, error) {
+		t.Error("free-space probe must not be consulted for explicit values")
+
+		return 0, errTestProbeNotExpected
+	}
+
+	cfg := &config.Config{
+		StateDir:              t.TempDir(),
+		CacheMaxBytes:         1024,
+		CacheMaxBytesExplicit: true,
+	}
+
+	got, err := cacheMaxBytes(t.Context(), cfg, newTestDB(t),
+		slog.New(slog.DiscardHandler), probe)
+	if err != nil {
+		t.Fatalf("cacheMaxBytes returned error: %v", err)
+	}
+
+	if got != 1024 {
+		t.Errorf("MaxBytes = %d, want explicit 1024 (no floor, no recompute)", got)
+	}
+}

internal/app/fetcher_internal_test.go

@@ -0,0 +1,106 @@
+package app
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/fx"
+	"go.uber.org/fx/fxtest"
+
+	"sneak.berlin/go/pixa/internal/config"
+	"sneak.berlin/go/pixa/internal/database"
+	"sneak.berlin/go/pixa/internal/globals"
+	"sneak.berlin/go/pixa/internal/handlers"
+	"sneak.berlin/go/pixa/internal/healthcheck"
+	"sneak.berlin/go/pixa/internal/logger"
+)
+
+// testSigningKey is a throwaway signing key.
+const testSigningKey = "test-signing-key-0123456789abcdef"
+
+// TestFetcherFromConfigRefusesBlockedNetworks builds the handlers as pixad
+// does, in an fx app with the providers of pixad's fx app, and requests an
+// image from 192.0.2.10, which is on the allowlist and in blocked_networks.
+// The URL check accepts that address; only the dialer that refuses internal
+// addresses checks blocked_networks, so the answer is 403 only if the
+// fetcher newFetcher builds from the config connects with that dialer. Any
+// other dialer would try to connect until the upstream fetch timeout, which
+// is short so that the test then fails quickly.
+func TestFetcherFromConfigRefusesBlockedNetworks(t *testing.T) {
+	t.Parallel()
+
+	const host = "192.0.2.10"
+
+	stateDir := t.TempDir()
+	cfg := &config.Config{
+		SigningKey:           testSigningKey,
+		StateDir:             stateDir,
+		DBURL:                "file:" + filepath.Join(stateDir, "state.sqlite3"),
+		AllowlistHosts:       []string{host},
+		BlockedNetworks:      []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
+		UpstreamFetchTimeout: 2 * time.Second,
+		// With no connection slots, the fetch would fail before dialing.
+		UpstreamConnectionsPerHost: config.DefaultUpstreamConnectionsPerHost,
+		UpstreamConnections:        config.DefaultUpstreamConnections,
+		UpstreamMaxResponseSize:    config.DefaultUpstreamMaxResponseSize,
+		MaxConcurrentProcessing:    1,
+	}
+
+	var h *handlers.Handlers
+
+	app := fxtest.New(t,
+		fx.Supply(cfg),
+		fx.Provide(globals.New, logger.New, database.New, healthcheck.New,
+			handlers.New, newCache, newFetcher, newProcessor, newImageService,
+			newSessionManager, newEncryptedURLGenerator,
+			func() prometheus.Registerer { return prometheus.NewRegistry() }),
+		fx.Populate(&h),
+	)
+	app.RequireStart()
+	t.Cleanup(app.RequireStop)
+
+	r := chi.NewRouter()
+	r.Get("/v1/image/*", h.HandleImage())
+
+	target := "/v1/image/" + host + "/images/photo.jpg/50x50.jpeg"
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, target, nil))
+	t.Logf("GET %s: %d %s", target, rec.Code, rec.Body)
+
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("status = %d, want %d", rec.Code, http.StatusForbidden)
+	}
+
+	if ct := rec.Header().Get("Content-Type"); ct != "application/json" {
+		t.Errorf("Content-Type = %q, want application/json", ct)
+	}
+
+	var body struct {
+		Error     string `json:"error"`
+		Status    int    `json:"status"`
+		Timestamp string `json:"timestamp"`
+	}
+
+	err := json.NewDecoder(rec.Body).Decode(&body)
+	if err != nil {
+		t.Fatalf("decoding response body: %v", err)
+	}
+
+	if body.Error != "forbidden" || body.Status != http.StatusForbidden {
+		t.Errorf("body error = %q, status = %d; want %q, %d",
+			body.Error, body.Status, "forbidden", http.StatusForbidden)
+	}
+
+	_, err = time.Parse(time.RFC3339, body.Timestamp)
+	if err != nil {
+		t.Errorf("body timestamp %q is not RFC 3339: %v", body.Timestamp, err)
+	}
+}

internal/app/providers.go

@@ -0,0 +1,206 @@
+package app
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"syscall"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/fx"
+	"sneak.berlin/go/pixa/internal/config"
+	"sneak.berlin/go/pixa/internal/database"
+	"sneak.berlin/go/pixa/internal/encurl"
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
+	"sneak.berlin/go/pixa/internal/imgcache"
+	"sneak.berlin/go/pixa/internal/logger"
+	"sneak.berlin/go/pixa/internal/session"
+)
+
+// newCache builds the image cache with the disk cache limit cacheMaxBytes
+// returns. Its start hook starts the cache's background goroutines. Its
+// stop hook stops them, which writes the pending access times. The
+// database is open for both: it opens before the cache is built and
+// closes after this stop hook.
+func newCache(
+	lc fx.Lifecycle,
+	cfg *config.Config,
+	db *database.Database,
+	log *logger.Logger,
+	metricsRegistry prometheus.Registerer,
+) (*imgcache.Cache, error) {
+	maxBytes, err := cacheMaxBytes(
+		context.Background(), cfg, db.DB(), log.Get(), defaultFreeSpaceProbe)
+	if err != nil {
+		return nil, err
+	}
+
+	cache, err := imgcache.NewCache(db.DB(), imgcache.CacheConfig{
+		StateDir:    cfg.StateDir,
+		NegativeTTL: imgcache.DefaultNegativeTTL,
+		MaxBytes:    maxBytes,
+		Logger:      log.Get(),
+	}, metricsRegistry)
+	if err != nil {
+		return nil, err
+	}
+
+	// The start hook takes no context: the goroutines it starts outlive it,
+	// and the stop hook stops them.
+	lc.Append(fx.StartStopHook(
+		func() {
+			// Background eviction: startup reconciliation, then periodic
+			// and write-pressure passes, and the removal of expired
+			// negative cache rows, which is all it does when the disk
+			// cache is off.
+			cache.StartEviction(imgcache.DefaultEvictionInterval)
+
+			// Writes the access times that requests hand over, so that no
+			// request waits for the database to record them
+			cache.StartPendingAccessTimeWrites()
+		},
+		func(ctx context.Context) error {
+			return errors.Join(
+				cache.StopEviction(ctx),
+				cache.StopPendingAccessTimeWrites(ctx),
+			)
+		},
+	))
+
+	return cache, nil
+}
+
+// freeSpaceProbeFunc reports the number of free bytes available on the
+// filesystem containing path. It is a function type so tests can pass a
+// fake probe instead of depending on the host disk.
+type freeSpaceProbeFunc func(path string) (uint64, error)
+
+var errNegativeBlockSize = errors.New("statfs reported negative block size")
+
+// defaultFreeSpaceProbe reports free filesystem bytes via statfs on the
+// given path, as available to unprivileged processes.
+func defaultFreeSpaceProbe(path string) (uint64, error) {
+	var stat syscall.Statfs_t
+
+	err := syscall.Statfs(path, &stat)
+	if err != nil {
+		return 0, err
+	}
+
+	if stat.Bsize < 0 {
+		return 0, fmt.Errorf("%w %d for %q", errNegativeBlockSize, stat.Bsize, path)
+	}
+
+	blockSize := uint64(stat.Bsize)
+
+	return stat.Bavail * blockSize, nil
+}
+
+// cacheMaxBytes returns cache_max_bytes as set. When the key is omitted,
+// it returns config.DefaultCacheMaxBytes of the free bytes probe reports
+// for <state_dir>/cache/ and the bytes the cache holds, as db records
+// them. It creates that directory first, so the probe measures the
+// filesystem that holds the cache.
+func cacheMaxBytes(
+	ctx context.Context,
+	cfg *config.Config,
+	db *sql.DB,
+	log *slog.Logger,
+	probe freeSpaceProbeFunc,
+) (int64, error) {
+	if cfg.CacheMaxBytesExplicit {
+		return cfg.CacheMaxBytes, nil
+	}
+
+	cacheDir := filepath.Join(cfg.StateDir, "cache")
+
+	err := os.MkdirAll(cacheDir, imgcache.StorageDirPerm)
+	if err != nil {
+		return 0, fmt.Errorf("failed to create cache directory: %w", err)
+	}
+
+	freeBytes, err := probe(cacheDir)
+	if err != nil {
+		return 0, fmt.Errorf(
+			"default cache_max_bytes: cannot determine free space for %q: %w",
+			cacheDir, err)
+	}
+
+	heldBytes, err := imgcache.UsageBytes(ctx, db)
+	if err != nil {
+		return 0, err
+	}
+
+	limit := config.DefaultCacheMaxBytes(freeBytes, heldBytes)
+
+	log.Info("computed default cache size limit from free space and cache contents",
+		"cache_max_bytes", limit,
+		"cache_dir", cacheDir,
+	)
+
+	return limit, nil
+}
+
+// newFetcher builds the upstream fetcher from cfg.
+func newFetcher(cfg *config.Config) *httpfetcher.HTTPFetcher {
+	return httpfetcher.New(&httpfetcher.Config{
+		Timeout:               cfg.UpstreamFetchTimeout,
+		MaxResponseSize:       cfg.UpstreamMaxResponseSize,
+		AllowHTTP:             cfg.AllowHTTP,
+		MaxConnectionsPerHost: cfg.UpstreamConnectionsPerHost,
+		MaxConnections:        cfg.UpstreamConnections,
+		BlockedNetworks:       cfg.BlockedNetworks,
+	})
+}
+
+// newProcessor builds the image processor from cfg. Its input limit is
+// the largest upstream response the fetcher accepts.
+func newProcessor(cfg *config.Config) *imageprocessor.ImageProcessor {
+	return imageprocessor.New(imageprocessor.Params{
+		MaxInputBytes:           cfg.UpstreamMaxResponseSize,
+		MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
+	})
+}
+
+// newImageService builds the image service from the cache, the fetcher
+// and the processor, with the signing key, the allowlist and the upstream
+// settings of cfg.
+func newImageService(
+	cfg *config.Config,
+	cache *imgcache.Cache,
+	fetcher *httpfetcher.HTTPFetcher,
+	processor *imageprocessor.ImageProcessor,
+	log *logger.Logger,
+) (*imgcache.Service, error) {
+	return imgcache.NewService(&imgcache.ServiceConfig{
+		Cache:           cache,
+		Fetcher:         fetcher,
+		Processor:       processor,
+		SigningKey:      cfg.SigningKey,
+		Allowlist:       cfg.AllowlistHosts,
+		AllowHTTP:       cfg.AllowHTTP,
+		MaxResponseSize: cfg.UpstreamMaxResponseSize,
+		Logger:          log.Get(),
+	})
+}
+
+// newSessionManager builds the session manager. Session cookies are
+// always Secure, HttpOnly and SameSite=Strict. The random value that each
+// logout replaces is saved in the database, so it outlasts a restart.
+func newSessionManager(
+	cfg *config.Config, db *database.Database,
+) (*session.Manager, error) {
+	return session.NewManagerWithDatabase(
+		context.Background(), cfg.SigningKey, db.DB())
+}
+
+// newEncryptedURLGenerator builds the encrypted URL generator, with a key
+// derived from the signing key.
+func newEncryptedURLGenerator(cfg *config.Config) (*encurl.Generator, error) {
+	return encurl.NewGenerator(cfg.SigningKey)
+}