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

The new package internal/providers builds the cache, the fetcher, the
processor, the image service, the session manager and the encrypted URL
generator from the config. providers.Option gives them to pixad's fx
graph, and handlers.New receives them built. The server tests build the
same graph, with only the fetcher's dial function replaced. 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, and
internal/providers works out the default cache_max_bytes. 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.

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/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
+  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
       `handlers.New`:
         - the cache
         - the fetcher

internal/app/app.go

@@ -1,5 +1,4 @@
-// Package app reads the pixad command line, builds pixad's parts from the
-// config and runs the server.
+// Package app reads the pixad command line and runs the server.
 package app
 
 import (
@@ -18,6 +17,7 @@ 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,15 +63,10 @@ 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/fetcher_internal_test.go

@@ -20,6 +20,7 @@ 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.
@@ -30,9 +31,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 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.
+// 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.
 func TestFetcherFromConfigRefusesBlockedNetworks(t *testing.T) {
 	t.Parallel()
 
@@ -58,9 +59,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, newCache, newFetcher, newProcessor, newImageService,
-			newSessionManager, newEncryptedURLGenerator,
+			handlers.New,
 			func() prometheus.Registerer { return prometheus.NewRegistry() }),
+		providers.Option,
 		fx.Populate(&h),
 	)
 	app.RequireStart()

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/providers/cache_max_bytes_internal_test.go

@@ -1,4 +1,4 @@
-package app
+package providers
 
 import (
 	"context"
@@ -19,6 +19,9 @@ 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")
@@ -149,7 +152,12 @@ func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
 				t.Fatalf("database.New() error = %v", err)
 			}
 
-			_, err = newCache(lc, cfg, db, log, prometheus.NewRegistry())
+			_, err = newCache(lc, cacheParams{
+				Config:          cfg,
+				Database:        db,
+				Logger:          log,
+				MetricsRegistry: prometheus.NewRegistry(),
+			})
 			if err != nil {
 				t.Fatalf("newCache() error = %v", err)
 			}

internal/providers/providers.go

@@ -1,4 +1,12 @@
-package app
+// 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
 
 import (
 	"context"
@@ -22,30 +30,51 @@ 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,
-	cfg *config.Config,
-	db *database.Database,
-	log *logger.Logger,
-	metricsRegistry prometheus.Registerer,
-) (*imgcache.Cache, error) {
+func newCache(lc fx.Lifecycle, params cacheParams) (*imgcache.Cache, error) {
+	cfg := params.Config
+	db := params.Database.DB()
+
 	maxBytes, err := cacheMaxBytes(
-		context.Background(), cfg, db.DB(), log.Get(), defaultFreeSpaceProbe)
+		context.Background(), cfg, db, params.Logger.Get(), defaultFreeSpaceProbe)
 	if err != nil {
 		return nil, err
 	}
 
-	cache, err := imgcache.NewCache(db.DB(), imgcache.CacheConfig{
+	cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
 		StateDir:    cfg.StateDir,
 		NegativeTTL: imgcache.DefaultNegativeTTL,
 		MaxBytes:    maxBytes,
-		Logger:      log.Get(),
-	}, metricsRegistry)
+		Logger:      params.Logger.Get(),
+	}, params.MetricsRegistry)
 	if err != nil {
 		return nil, err
 	}
@@ -146,16 +175,19 @@ func cacheMaxBytes(
 	return limit, nil
 }
 
-// newFetcher builds the upstream fetcher from cfg.
-func newFetcher(cfg *config.Config) *httpfetcher.HTTPFetcher {
-	return httpfetcher.New(&httpfetcher.Config{
+// 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{
 		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
@@ -167,36 +199,53 @@ 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 cfg.
+// settings of the config.
 func newImageService(
-	cfg *config.Config,
-	cache *imgcache.Cache,
-	fetcher *httpfetcher.HTTPFetcher,
-	processor *imageprocessor.ImageProcessor,
-	log *logger.Logger,
+	_ fx.Lifecycle, params imageServiceParams,
 ) (*imgcache.Service, error) {
+	cfg := params.Config
+
 	return imgcache.NewService(&imgcache.ServiceConfig{
-		Cache:           cache,
-		Fetcher:         fetcher,
-		Processor:       processor,
+		Cache:           params.Cache,
+		Fetcher:         params.Fetcher,
+		Processor:       params.Processor,
 		SigningKey:      cfg.SigningKey,
 		Allowlist:       cfg.AllowlistHosts,
 		AllowHTTP:       cfg.AllowHTTP,
 		MaxResponseSize: cfg.UpstreamMaxResponseSize,
-		Logger:          log.Get(),
+		Logger:          params.Logger.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(
-	cfg *config.Config, db *database.Database,
+	_ fx.Lifecycle, params sessionManagerParams,
 ) (*session.Manager, error) {
 	return session.NewManagerWithDatabase(
-		context.Background(), cfg.SigningKey, db.DB())
+		context.Background(), params.Config.SigningKey, params.Database.DB())
 }
 
 // newEncryptedURLGenerator builds the encrypted URL generator, with a key

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/session"
+	"sneak.berlin/go/pixa/internal/providers"
 )
 
 // upstreamHost is the upstream host of the image URLs below. It is a
@@ -121,10 +121,9 @@ 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, 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.
+// 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.
 func startImageProxy(
 	t *testing.T, upstream *httptest.Server,
 ) (*Server, *sql.DB, string) {
@@ -171,12 +170,14 @@ func startImageProxy(
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func(db *database.Database) (
-				*imgcache.Service, *session.Manager, *encurl.Generator, error,
-			) {
-				return newImagePipeline(cfg, db, dialUpstream)
-			},
+			func() prometheus.Registerer { return prometheus.NewRegistry() },
 		),
+		providers.Option,
+		fx.Decorate(func(fetcherConfig *httpfetcher.Config) *httpfetcher.Config {
+			fetcherConfig.DialContext = dialUpstream
+
+			return fetcherConfig
+		}),
 		fx.Populate(&h, &mw, &db),
 	)
 	app.RequireStart()
@@ -192,11 +193,10 @@ func startImageProxy(
 
 // startImageProxyWithBlockedNetworks starts the components pixad's fx app
 // builds, from a config with a fresh state directory and blockedNetworks as
-// 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.
+// 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.
 func startImageProxyWithBlockedNetworks(
 	t *testing.T, blockedNetworks []netip.Prefix,
 ) *Server {
@@ -231,12 +231,9 @@ func startImageProxyWithBlockedNetworks(
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func(db *database.Database) (
-				*imgcache.Service, *session.Manager, *encurl.Generator, error,
-			) {
-				return newImagePipeline(cfg, db, nil)
-			},
+			func() prometheus.Registerer { return prometheus.NewRegistry() },
 		),
+		providers.Option,
 		fx.Populate(&h, &mw),
 	)
 	app.RequireStart()

internal/server/login_rate_limit_internal_test.go

@@ -1,9 +1,7 @@
 package server
 
 import (
-	"context"
 	"io"
-	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/netip"
@@ -14,21 +12,7 @@ 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.
@@ -68,110 +52,7 @@ func newTestServer(t *testing.T) *Server {
 		DownstreamTimeout: config.DefaultDownstreamTimeout,
 	}
 
-	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
+	return startPixa(t, cfg)
 }
 
 // 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,9 +29,8 @@ import (
 var logoutFormPattern = regexp.MustCompile(
 	`(?s)action="/logout".*?name="gorilla\.csrf\.Token" value="([^"]+)"`)
 
-// 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.
+// startPixa starts the components pixad's fx app builds, from cfg, and
+// returns the server with its routes.
 func startPixa(t *testing.T, cfg *config.Config) *Server {
 	t.Helper()
 
@@ -49,12 +48,9 @@ func startPixa(t *testing.T, cfg *config.Config) *Server {
 			healthcheck.New,
 			handlers.New,
 			middleware.New,
-			func(db *database.Database) (
-				*imgcache.Service, *session.Manager, *encurl.Generator, error,
-			) {
-				return newImagePipeline(cfg, db, nil)
-			},
+			func() prometheus.Registerer { return prometheus.NewRegistry() },
 		),
+		providers.Option,
 		fx.Populate(&h, &mw),
 	)
 	app.RequireStart()