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

TODO.md

@@ -34,11 +34,11 @@ P2: security: per-origin rate limiting
   `database.New` now opens the database, so the parts below can read it while
   they are built; its stop hook still runs after the cache's. The cache's
   `MaxBytes` is the one value for the disk cache limit, and 0 turns the disk
-  cache off. For an omitted `cache_max_bytes`, `internal/providers` works out
-  `config.DefaultCacheMaxBytes` before it builds the cache. `internal/app` and
-  the server tests build the parts below with `providers.Option`; the image
-  proxy test replaces only the fetcher's dial function.
-    - Built by the providers in `internal/providers`, and given to
+  cache off. For an omitted `cache_max_bytes`, `internal/app` works out
+  `config.DefaultCacheMaxBytes` before it builds the cache. The server tests
+  build the pipeline with `newImagePipeline`, as they cannot import
+  `internal/app`.
+    - Built by the providers in `internal/app/providers.go`, and given to
       `handlers.New`:
         - the cache
         - the fetcher

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 (
@@ -17,7 +18,6 @@ import (
 	"sneak.berlin/go/pixa/internal/healthcheck"
 	"sneak.berlin/go/pixa/internal/logger"
 	"sneak.berlin/go/pixa/internal/middleware"
-	"sneak.berlin/go/pixa/internal/providers"
 	"sneak.berlin/go/pixa/internal/server"
 )
 
@@ -63,10 +63,15 @@ 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 },
 		),
-		providers.Option,
 		fx.Invoke(
 			func(log *logger.Logger) { log.Identify() },
 			func(*server.Server) {},

internal/app/cache_max_bytes_internal_test.go

@@ -1,4 +1,4 @@
-package providers
+package app
 
 import (
 	"context"
@@ -19,9 +19,6 @@ import (
 	"sneak.berlin/go/pixa/internal/logger"
 )
 
-// testSigningKey is a throwaway signing key.
-const testSigningKey = "test-signing-key-0123456789abcdef"
-
 // Static errors returned by the stub free-space probes below.
 var (
 	errTestStatfsFailed     = errors.New("statfs failed")
@@ -152,12 +149,7 @@ func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
 				t.Fatalf("database.New() error = %v", err)
 			}
 
-			_, err = newCache(lc, cacheParams{
-				Config:          cfg,
-				Database:        db,
-				Logger:          log,
-				MetricsRegistry: prometheus.NewRegistry(),
-			})
+			_, err = newCache(lc, cfg, db, log, prometheus.NewRegistry())
 			if err != nil {
 				t.Fatalf("newCache() error = %v", err)
 			}

internal/app/fetcher_internal_test.go

@@ -20,7 +20,6 @@ import (
 	"sneak.berlin/go/pixa/internal/handlers"
 	"sneak.berlin/go/pixa/internal/healthcheck"
 	"sneak.berlin/go/pixa/internal/logger"
-	"sneak.berlin/go/pixa/internal/providers"
 )
 
 // testSigningKey is a throwaway signing key.
@@ -31,9 +30,9 @@ const testSigningKey = "test-signing-key-0123456789abcdef"
 // 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 providers.Option 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.
+// 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()
 
@@ -59,9 +58,9 @@ func TestFetcherFromConfigRefusesBlockedNetworks(t *testing.T) {
 	app := fxtest.New(t,
 		fx.Supply(cfg),
 		fx.Provide(globals.New, logger.New, database.New, healthcheck.New,
-			handlers.New,
+			handlers.New, newCache, newFetcher, newProcessor, newImageService,
+			newSessionManager, newEncryptedURLGenerator,
 			func() prometheus.Registerer { return prometheus.NewRegistry() }),
-		providers.Option,
 		fx.Populate(&h),
 	)
 	app.RequireStart()

internal/app/providers.go

@@ -1,12 +1,4 @@
-// Package providers builds from the config the parts pixad serves images
-// and logins with, and provides them to the fx graph:
-//   - the image cache, with its default limit
-//   - the fetcher
-//   - the processor
-//   - the image service
-//   - the session manager
-//   - the encrypted URL generator
-package providers
+package app
 
 import (
 	"context"
@@ -30,51 +22,30 @@ import (
 	"sneak.berlin/go/pixa/internal/session"
 )
 
-// Option provides the parts this package builds. The fx graph must also
-// provide the prometheus.Registerer the cache registers its counters on.
-//
-//nolint:gochecknoglobals // an fx option, built once and never changed
-var Option = fx.Provide(
-	newCache,
-	newFetcherConfig,
-	httpfetcher.New,
-	newProcessor,
-	newImageService,
-	newSessionManager,
-	newEncryptedURLGenerator,
-)
-
-// cacheParams defines dependencies for newCache.
-type cacheParams struct {
-	fx.In
-
-	Config          *config.Config
-	Database        *database.Database
-	Logger          *logger.Logger
-	MetricsRegistry prometheus.Registerer
-}
-
 // 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, params cacheParams) (*imgcache.Cache, error) {
-	cfg := params.Config
-	db := params.Database.DB()
-
+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, params.Logger.Get(), defaultFreeSpaceProbe)
+		context.Background(), cfg, db.DB(), log.Get(), defaultFreeSpaceProbe)
 	if err != nil {
 		return nil, err
 	}
 
-	cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
+	cache, err := imgcache.NewCache(db.DB(), imgcache.CacheConfig{
 		StateDir:    cfg.StateDir,
 		NegativeTTL: imgcache.DefaultNegativeTTL,
 		MaxBytes:    maxBytes,
-		Logger:      params.Logger.Get(),
-	}, params.MetricsRegistry)
+		Logger:      log.Get(),
+	}, metricsRegistry)
 	if err != nil {
 		return nil, err
 	}
@@ -175,19 +146,16 @@ func cacheMaxBytes(
 	return limit, nil
 }
 
-// newFetcherConfig builds the upstream fetcher's settings from cfg, and
-// httpfetcher.New builds the fetcher from them. The settings are a value
-// of their own in the fx graph, so that a test can replace the dial
-// function alone.
-func newFetcherConfig(cfg *config.Config) *httpfetcher.Config {
-	return &httpfetcher.Config{
+// 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
@@ -199,53 +167,36 @@ func newProcessor(cfg *config.Config) *imageprocessor.ImageProcessor {
 	})
 }
 
-// imageServiceParams defines dependencies for newImageService.
-type imageServiceParams struct {
-	fx.In
-
-	Config    *config.Config
-	Cache     *imgcache.Cache
-	Fetcher   *httpfetcher.HTTPFetcher
-	Processor *imageprocessor.ImageProcessor
-	Logger    *logger.Logger
-}
-
 // newImageService builds the image service from the cache, the fetcher
 // and the processor, with the signing key, the allowlist and the upstream
-// settings of the config.
+// settings of cfg.
 func newImageService(
-	_ fx.Lifecycle, params imageServiceParams,
+	cfg *config.Config,
+	cache *imgcache.Cache,
+	fetcher *httpfetcher.HTTPFetcher,
+	processor *imageprocessor.ImageProcessor,
+	log *logger.Logger,
 ) (*imgcache.Service, error) {
-	cfg := params.Config
-
 	return imgcache.NewService(&imgcache.ServiceConfig{
-		Cache:           params.Cache,
-		Fetcher:         params.Fetcher,
-		Processor:       params.Processor,
+		Cache:           cache,
+		Fetcher:         fetcher,
+		Processor:       processor,
 		SigningKey:      cfg.SigningKey,
 		Allowlist:       cfg.AllowlistHosts,
 		AllowHTTP:       cfg.AllowHTTP,
 		MaxResponseSize: cfg.UpstreamMaxResponseSize,
-		Logger:          params.Logger.Get(),
+		Logger:          log.Get(),
 	})
 }
 
-// sessionManagerParams defines dependencies for newSessionManager.
-type sessionManagerParams struct {
-	fx.In
-
-	Config   *config.Config
-	Database *database.Database
-}
-
 // 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(
-	_ fx.Lifecycle, params sessionManagerParams,
+	cfg *config.Config, db *database.Database,
 ) (*session.Manager, error) {
 	return session.NewManagerWithDatabase(
-		context.Background(), params.Config.SigningKey, params.Database.DB())
+		context.Background(), cfg.SigningKey, db.DB())
 }
 
 // newEncryptedURLGenerator builds the encrypted URL generator, with a key

internal/imgcache/cache.go

@@ -82,7 +82,7 @@ type Cache struct {
 	transforms         prometheus.Counter
 
 	// reconciliationPageSize is the most rows one read of a content table
-	// returns in the reconciliation pass. NewCache sets it to
+	// returns in the reconciliation pass. newCache sets it to
 	// defaultReconciliationPageSize; tests set it smaller.
 	reconciliationPageSize int
 

internal/server/image_proxy_internal_test.go

@@ -19,20 +19,20 @@ import (
 	"testing"
 	"time"
 
-	"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/encurl"
 	"sneak.berlin/go/pixa/internal/globals"
 	"sneak.berlin/go/pixa/internal/handlers"
 	"sneak.berlin/go/pixa/internal/healthcheck"
-	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imagerequest"
+	"sneak.berlin/go/pixa/internal/imgcache"
 	"sneak.berlin/go/pixa/internal/logger"
 	"sneak.berlin/go/pixa/internal/middleware"
-	"sneak.berlin/go/pixa/internal/providers"
+	"sneak.berlin/go/pixa/internal/session"
 )
 
 // upstreamHost is the upstream host of the image URLs below. It is a
@@ -121,9 +121,10 @@ func TestImageProxyFlow(t *testing.T) {
 }
 
 // startImageProxy starts the components pixad's fx app builds, from a config
-// with a fresh state directory and upstreamHost on the allowlist, and with an
-// upstream fetcher that connects every upstream address to upstream. It
-// returns the server with its routes, the database and the state directory.
+// with a fresh state directory and upstreamHost on the allowlist, with the
+// image pipeline newImagePipeline builds and an upstream fetcher that
+// connects every upstream address to upstream. It returns the server with
+// its routes, the database and the state directory.
 func startImageProxy(
 	t *testing.T, upstream *httptest.Server,
 ) (*Server, *sql.DB, string) {
@@ -170,14 +171,12 @@ func startImageProxy(
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func() prometheus.Registerer { return prometheus.NewRegistry() },
+			func(db *database.Database) (
+				*imgcache.Service, *session.Manager, *encurl.Generator, error,
+			) {
+				return newImagePipeline(cfg, db, dialUpstream)
+			},
 		),
-		providers.Option,
-		fx.Decorate(func(fetcherConfig *httpfetcher.Config) *httpfetcher.Config {
-			fetcherConfig.DialContext = dialUpstream
-
-			return fetcherConfig
-		}),
 		fx.Populate(&h, &mw, &db),
 	)
 	app.RequireStart()
@@ -193,10 +192,11 @@ func startImageProxy(
 
 // startImageProxyWithBlockedNetworks starts the components pixad's fx app
 // builds, from a config with a fresh state directory and blockedNetworks as
-// blocked_networks. Unlike startImageProxy it keeps pixad's dial function,
-// which checks every address the fetcher connects to against the built-in
-// blocked ranges and blocked_networks. It returns the server with its
-// routes.
+// blocked_networks, with the image pipeline newImagePipeline builds. Unlike
+// startImageProxy it gives the fetcher no dial function, so the fetcher
+// connects with the dialer pixad uses, which checks every address it
+// connects to against the built-in blocked ranges and blocked_networks. It
+// returns the server with its routes.
 func startImageProxyWithBlockedNetworks(
 	t *testing.T, blockedNetworks []netip.Prefix,
 ) *Server {
@@ -231,9 +231,12 @@ func startImageProxyWithBlockedNetworks(
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func() prometheus.Registerer { return prometheus.NewRegistry() },
+			func(db *database.Database) (
+				*imgcache.Service, *session.Manager, *encurl.Generator, error,
+			) {
+				return newImagePipeline(cfg, db, nil)
+			},
 		),
-		providers.Option,
 		fx.Populate(&h, &mw),
 	)
 	app.RequireStart()

internal/server/login_rate_limit_internal_test.go

@@ -1,7 +1,9 @@
 package server
 
 import (
+	"context"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/netip"
@@ -12,7 +14,21 @@ import (
 	"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/encurl"
+	"sneak.berlin/go/pixa/internal/globals"
+	"sneak.berlin/go/pixa/internal/handlers"
+	"sneak.berlin/go/pixa/internal/healthcheck"
+	"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/middleware"
+	"sneak.berlin/go/pixa/internal/session"
 )
 
 // testSigningKey is a throwaway signing key; submitting it logs in.
@@ -52,7 +68,110 @@ func newTestServer(t *testing.T) *Server {
 		DownstreamTimeout: config.DefaultDownstreamTimeout,
 	}
 
-	return startPixa(t, cfg)
+	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)
+	}
+
+	hc, err := healthcheck.New(lc, healthcheck.Params{
+		Globals: &globals.Globals{}, Config: cfg, Logger: log, Database: db,
+	})
+	if err != nil {
+		t.Fatalf("healthcheck.New() error = %v", err)
+	}
+
+	svc, sessMgr, encGen, err := newImagePipeline(cfg, db, nil)
+	if err != nil {
+		t.Fatalf("newImagePipeline() error = %v", err)
+	}
+
+	h, err := handlers.New(lc, handlers.Params{
+		Logger: log, Healthcheck: hc, Config: cfg,
+		ImageService: svc, SessionManager: sessMgr, EncryptedURLGenerator: encGen,
+	})
+	if err != nil {
+		t.Fatalf("handlers.New() error = %v", err)
+	}
+
+	mw, err := middleware.New(lc, middleware.Params{Logger: log, Config: cfg})
+	if err != nil {
+		t.Fatalf("middleware.New() error = %v", err)
+	}
+
+	lc.RequireStart()
+	t.Cleanup(lc.RequireStop)
+
+	s := &Server{config: cfg, mw: mw, h: h}
+	s.SetupRoutes()
+
+	return s
+}
+
+// newImagePipeline builds from cfg and db the parts handlers.New takes
+// that pixad's fx app builds with the providers of internal/app: the image
+// service, the session manager and the encrypted URL generator. These
+// tests cannot import internal/app, as it imports this package. The
+// upstream fetcher connects with dialContext when that is not nil, as
+// httpfetcher.Config says. The disk cache has the limit cache_max_bytes
+// sets, and is off when it is omitted, so that no test sizes it from the
+// host's free disk space. The cache's background goroutines do not run.
+func newImagePipeline(
+	cfg *config.Config,
+	db *database.Database,
+	dialContext func(ctx context.Context, network, addr string) (net.Conn, error),
+) (*imgcache.Service, *session.Manager, *encurl.Generator, error) {
+	cache, err := imgcache.NewCache(db.DB(), imgcache.CacheConfig{
+		StateDir:    cfg.StateDir,
+		NegativeTTL: imgcache.DefaultNegativeTTL,
+		MaxBytes:    cfg.CacheMaxBytes,
+	}, prometheus.NewRegistry())
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	svc, err := imgcache.NewService(&imgcache.ServiceConfig{
+		Cache: cache,
+		Fetcher: httpfetcher.New(&httpfetcher.Config{
+			Timeout:               cfg.UpstreamFetchTimeout,
+			MaxResponseSize:       cfg.UpstreamMaxResponseSize,
+			AllowHTTP:             cfg.AllowHTTP,
+			MaxConnectionsPerHost: cfg.UpstreamConnectionsPerHost,
+			MaxConnections:        cfg.UpstreamConnections,
+			BlockedNetworks:       cfg.BlockedNetworks,
+			DialContext:           dialContext,
+		}),
+		Processor: imageprocessor.New(imageprocessor.Params{
+			MaxInputBytes:           cfg.UpstreamMaxResponseSize,
+			MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
+		}),
+		SigningKey:      cfg.SigningKey,
+		Allowlist:       cfg.AllowlistHosts,
+		AllowHTTP:       cfg.AllowHTTP,
+		MaxResponseSize: cfg.UpstreamMaxResponseSize,
+	})
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	sessMgr, err := session.NewManagerWithDatabase(
+		context.Background(), cfg.SigningKey, db.DB())
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	encGen, err := encurl.NewGenerator(cfg.SigningKey)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	return svc, sessMgr, encGen, nil
 }
 
 // clientRequest builds a request for / arriving from remoteAddr, carrying

internal/server/logout_internal_test.go

@@ -9,18 +9,18 @@ import (
 	"strings"
 	"testing"
 
-	"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/encurl"
 	"sneak.berlin/go/pixa/internal/globals"
 	"sneak.berlin/go/pixa/internal/handlers"
 	"sneak.berlin/go/pixa/internal/healthcheck"
+	"sneak.berlin/go/pixa/internal/imgcache"
 	"sneak.berlin/go/pixa/internal/logger"
 	"sneak.berlin/go/pixa/internal/middleware"
-	"sneak.berlin/go/pixa/internal/providers"
 	"sneak.berlin/go/pixa/internal/session"
 )
 
@@ -29,8 +29,9 @@ import (
 var logoutFormPattern = regexp.MustCompile(
 	`(?s)action="/logout".*?name="gorilla\.csrf\.Token" value="([^"]+)"`)
 
-// startPixa starts the components pixad's fx app builds, from cfg, and
-// returns the server with its routes.
+// startPixa starts the components pixad's fx app builds, from cfg, with the
+// image pipeline newImagePipeline builds, and returns the server with its
+// routes, as newTestServer does.
 func startPixa(t *testing.T, cfg *config.Config) *Server {
 	t.Helper()
 
@@ -48,9 +49,12 @@ func startPixa(t *testing.T, cfg *config.Config) *Server {
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func() prometheus.Registerer { return prometheus.NewRegistry() },
+			func(db *database.Database) (
+				*imgcache.Service, *session.Manager, *encurl.Generator, error,
+			) {
+				return newImagePipeline(cfg, db, nil)
+			},
 		),
-		providers.Option,
 		fx.Populate(&h, &mw),
 	)
 	app.RequireStart()