max_concurrent_processing (default: the number of CPUs Go uses) bounds the images processed at once, and upstream_connections (default 64) the fetches from all upstream hosts together, beside the per-host limit. A request that finds either full waits up to 10 seconds, then gets 503 "server busy, try again later". The processor holds its slot from before it reads the input until it returns, and takes a free slot even after the request context has ended; a fetch holds its connection until the response body is closed, after its image is processed. libvips now starts with one worker thread per image and no operation cache. Both settings have PIXA_ variables and are in README.md and config.example.yml. Model: opus-5-5
174 lines
4.7 KiB
Go
174 lines
4.7 KiB
Go
// Package handlers provides HTTP request handlers.
|
|
package handlers
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"net/http"
|
|
"time"
|
|
|
|
"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/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"
|
|
)
|
|
|
|
// Params defines dependencies for Handlers.
|
|
type Params struct {
|
|
fx.In
|
|
|
|
Logger *logger.Logger
|
|
Healthcheck *healthcheck.Healthcheck
|
|
Database *database.Database
|
|
Config *config.Config
|
|
}
|
|
|
|
// Handlers provides HTTP request handlers.
|
|
type Handlers struct {
|
|
log *slog.Logger
|
|
hc *healthcheck.Healthcheck
|
|
db *database.Database
|
|
config *config.Config
|
|
imgSvc *imgcache.Service
|
|
imgCache *imgcache.Cache
|
|
sessMgr *session.Manager
|
|
encGen *encurl.Generator
|
|
csrfProtect func(http.Handler) http.Handler
|
|
}
|
|
|
|
// New creates a new Handlers instance.
|
|
func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
|
csrfProtect, err := newCSRFProtect(params.Config.SigningKey, params.Config.Debug)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
s := &Handlers{
|
|
log: params.Logger.Get(),
|
|
hc: params.Healthcheck,
|
|
db: params.Database,
|
|
config: params.Config,
|
|
csrfProtect: csrfProtect,
|
|
}
|
|
|
|
lc.Append(fx.Hook{
|
|
// The eviction goroutine must outlive OnStart, so it cannot
|
|
// inherit this hook's context. It makes its own instead, which
|
|
// leaves it uncancellable: an in-flight pass runs to completion
|
|
// during OnStop regardless of the shutdown deadline. Making the
|
|
// loop cancellable changes shutdown semantics and is tracked
|
|
// separately in issue #102, rather than being folded into the
|
|
// lint-conformance change that surfaced it.
|
|
//nolint:contextcheck // see issue #102
|
|
OnStart: func(_ context.Context) error {
|
|
return s.initImageService()
|
|
},
|
|
OnStop: func(_ context.Context) error {
|
|
if s.imgCache != nil {
|
|
s.imgCache.StopEviction()
|
|
}
|
|
|
|
return nil
|
|
},
|
|
})
|
|
|
|
return s, nil
|
|
}
|
|
|
|
// initImageService initializes the image cache and service.
|
|
func (s *Handlers) initImageService() error {
|
|
// Create the cache. cache_max_bytes: 0 disables the disk cache
|
|
// entirely; any other value is the eviction limit in bytes.
|
|
cache, err := imgcache.NewCache(s.db.DB(), imgcache.CacheConfig{
|
|
StateDir: s.config.StateDir,
|
|
CacheTTL: imgcache.DefaultCacheTTL,
|
|
NegativeTTL: imgcache.DefaultNegativeTTL,
|
|
MaxBytes: s.config.CacheMaxBytes,
|
|
DisableDiskCache: s.config.CacheMaxBytes == 0,
|
|
Logger: s.log,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
s.imgCache = cache
|
|
|
|
// Background eviction: startup reconciliation, then periodic and
|
|
// write-pressure passes. No-op when the disk cache is disabled.
|
|
cache.StartEviction(imgcache.DefaultEvictionInterval)
|
|
|
|
// Create the fetcher config
|
|
fetcherCfg := httpfetcher.DefaultConfig()
|
|
fetcherCfg.AllowHTTP = s.config.AllowHTTP
|
|
|
|
if s.config.UpstreamConnectionsPerHost > 0 {
|
|
fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
|
|
}
|
|
|
|
fetcherCfg.MaxConnections = s.config.UpstreamConnections
|
|
fetcherCfg.BlockedNetworks = s.config.BlockedNetworks
|
|
|
|
// Create the service
|
|
svc, err := imgcache.NewService(&imgcache.ServiceConfig{
|
|
Cache: cache,
|
|
FetcherConfig: fetcherCfg,
|
|
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.
|
|
sessMgr, err := session.NewManager(s.config.SigningKey)
|
|
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)
|
|
|
|
if data != nil {
|
|
err := json.NewEncoder(w).Encode(data)
|
|
if err != nil {
|
|
s.log.Error("json encode error", "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Handlers) respondError(w http.ResponseWriter, message string, status int) {
|
|
s.respondJSON(w, map[string]any{
|
|
"error": message,
|
|
"status": status,
|
|
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
|
}, status)
|
|
}
|