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

internal/handlers/handlers.go

@@ -4,19 +4,15 @@ package handlers
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"log/slog"
 	"net/http"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/fx"
 	"sneak.berlin/go/pixa/internal/allowlist"
 	"sneak.berlin/go/pixa/internal/config"
-	"sneak.berlin/go/pixa/internal/database"
 	"sneak.berlin/go/pixa/internal/encurl"
 	"sneak.berlin/go/pixa/internal/healthcheck"
-	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imgcache"
 	"sneak.berlin/go/pixa/internal/logger"
 	"sneak.berlin/go/pixa/internal/session"
@@ -28,29 +24,19 @@ type Params struct {
 
 	Logger      *logger.Logger
 	Healthcheck *healthcheck.Healthcheck
-	Database    *database.Database
 	Config      *config.Config
 
-	// MetricsRegistry is the registry the image cache registers its
-	// counters on. pixad provides Prometheus' default registry, the one
-	// /metrics serves.
-	MetricsRegistry prometheus.Registerer
-
-	// Fetcher, when provided, fetches upstream images in place of the
-	// fetcher the handlers build from the config. Only tests provide one;
-	// pixad does not.
-	Fetcher httpfetcher.Fetcher `optional:"true"`
+	ImageService          *imgcache.Service
+	SessionManager        *session.Manager
+	EncryptedURLGenerator *encurl.Generator
 }
 
 // Handlers provides HTTP request handlers.
 type Handlers struct {
 	log         *slog.Logger
 	hc          *healthcheck.Healthcheck
-	db          *database.Database
 	config      *config.Config
-	fetcher     httpfetcher.Fetcher
 	imgSvc      *imgcache.Service
-	imgCache    *imgcache.Cache
 	sessMgr     *session.Manager
 	encGen      *encurl.Generator
 	csrfProtect func(http.Handler) http.Handler
@@ -58,50 +44,25 @@ type Handlers struct {
 	// refererBlocklist matches the hosts of referer_blocklist; its IsAllowed
 	// reports whether a URL's host is on that list.
 	refererBlocklist *allowlist.HostAllowList
-
-	// metricsRegistry is the registry the image cache registers its
-	// counters on.
-	metricsRegistry prometheus.Registerer
 }
 
 // New creates a new Handlers instance.
-func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
+func New(_ fx.Lifecycle, params Params) (*Handlers, error) {
 	csrfProtect, err := newCSRFProtect(params.Config.SigningKey, params.Config.Debug)
 	if err != nil {
 		return nil, err
 	}
 
-	s := &Handlers{
+	return &Handlers{
 		log:              params.Logger.Get(),
 		hc:               params.Healthcheck,
-		db:               params.Database,
 		config:           params.Config,
-		fetcher:          params.Fetcher,
+		imgSvc:           params.ImageService,
+		sessMgr:          params.SessionManager,
+		encGen:           params.EncryptedURLGenerator,
 		csrfProtect:      csrfProtect,
 		refererBlocklist: allowlist.New(params.Config.RefererBlocklist),
-		metricsRegistry:  params.MetricsRegistry,
-	}
-
-	lc.Append(fx.Hook{
-		//nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them
-		OnStart: func(_ context.Context) error {
-			return s.initImageService()
-		},
-		// The pending access times are written here, before the database's
-		// stop hook closes it.
-		OnStop: func(ctx context.Context) error {
-			if s.imgCache == nil {
-				return nil
-			}
-
-			return errors.Join(
-				s.imgCache.StopEviction(ctx),
-				s.imgCache.StopPendingAccessTimeWrites(ctx),
-			)
-		},
-	})
-
-	return s, nil
+	}, nil
 }
 
 // WaitForProcessing waits until no image is being processed, or until ctx
@@ -110,96 +71,6 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
 	return s.imgSvc.WaitForProcessing(ctx)
 }
 
-// newCacheConfig builds the image cache's configuration from cfg.
-// cache_max_bytes: 0 disables the disk cache entirely; any other value
-// is the eviction limit in bytes; when it is omitted, the cache works
-// out the default limit itself.
-func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
-	return imgcache.CacheConfig{
-		StateDir:           cfg.StateDir,
-		NegativeTTL:        imgcache.DefaultNegativeTTL,
-		MaxBytes:           cfg.CacheMaxBytes,
-		UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit,
-		DisableDiskCache:   cfg.CacheMaxBytesExplicit && cfg.CacheMaxBytes == 0,
-		Logger:             log,
-	}
-}
-
-// initImageService initializes the image cache and service.
-func (s *Handlers) initImageService() error {
-	cache, err := imgcache.NewCache(
-		s.db.DB(), newCacheConfig(s.config, s.log), s.metricsRegistry)
-	if err != nil {
-		return err
-	}
-
-	s.imgCache = cache
-
-	// 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 disabled.
-	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()
-
-	// Create the fetcher config
-	fetcherCfg := httpfetcher.DefaultConfig()
-	fetcherCfg.AllowHTTP = s.config.AllowHTTP
-	fetcherCfg.Timeout = s.config.UpstreamFetchTimeout
-	fetcherCfg.MaxResponseSize = s.config.UpstreamMaxResponseSize
-
-	if s.config.UpstreamConnectionsPerHost > 0 {
-		fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
-	}
-
-	fetcherCfg.MaxConnections = s.config.UpstreamConnections
-	fetcherCfg.BlockedNetworks = s.config.BlockedNetworks
-
-	// Create the service. With no fetcher provided, it builds its own from
-	// fetcherCfg.
-	svc, err := imgcache.NewService(&imgcache.ServiceConfig{
-		Cache:                   cache,
-		FetcherConfig:           fetcherCfg,
-		Fetcher:                 s.fetcher,
-		SigningKey:              s.config.SigningKey,
-		Allowlist:               s.config.AllowlistHosts,
-		MaxConcurrentProcessing: s.config.MaxConcurrentProcessing,
-		Logger:                  s.log,
-	})
-	if err != nil {
-		return err
-	}
-
-	s.imgSvc = svc
-	s.log.Info("image service initialized")
-
-	// Initialize session manager (signing key is validated at config load
-	// time). Session cookies are always Secure/HttpOnly/SameSite=Strict.
-	// The random value that each logout replaces is saved in the database,
-	// so it outlasts a restart.
-	sessMgr, err := session.NewManagerWithDatabase(
-		context.Background(), s.config.SigningKey, s.db.DB())
-	if err != nil {
-		return err
-	}
-
-	s.sessMgr = sessMgr
-
-	// Initialize encrypted URL generator
-	encGen, err := encurl.NewGenerator(s.config.SigningKey)
-	if err != nil {
-		return err
-	}
-
-	s.encGen = encGen
-
-	s.log.Info("session manager and URL generator initialized")
-
-	return nil
-}
-
 func (s *Handlers) respondJSON(w http.ResponseWriter, data any, status int) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)