Files
job-search/internal/server/server.go
T
bwilliams 70d62f280e Added the opportunities REST API, read-only apart from their marks, described in OpenAPI 3.1 with swag
GET /api/opportunities lists the stored opportunities, newest release first,
a page at a time (offset, and limit up to 100, default 50); has_more says
whether another page follows. GET /api/opportunities/{uuid} returns one with
its releases. PUT and DELETE /api/opportunities/{uuid}/marks/{mark} set and
clear one mark (hidden, skipped, interested, bid_submitted, won, lost) and
return them all; won and lost exclude each other (409). The API uses the
types in internal/types, and changing a mark is made atomic, as the store
replaces all of an opportunity's marks at once. The marks are found by
reflection: every bool field of types.OpportunityMarks, by its JSON name, so
a new mark needs no API change.

The API has one entry point, api.New, with its dependencies as options
(api.WithStore), and the server only mounts it under /api, keeping the
middleware. New resources are added inside internal/api. job-search serve
passes server.WithStore when database.file is set, opening the store before
the server; without one there are no opportunity routes.

The API is described by swag v2 annotations on the handlers. go generate
writes the OpenAPI 3.1 description (git-ignored, like the OCDS types), which
the server embeds and serves at /api/openapi.json, with Swagger UI at
/api/docs. A test checks the description and the routes match. swag is in
the README's dev tools, pinned to v2.0.0-rc6, and the build now needs it.
2026-10-16 13:13:44 +01:00

418 lines
12 KiB
Go

// Package server assembles the HTTP server for the serve command.
package server
import (
"context"
"crypto/tls"
"errors"
"fmt"
"log"
"log/slog"
"net"
"net/http"
"net/netip"
"strconv"
"strings"
"time"
"github.com/pires/go-proxyproto"
"git.avsdev.uk/Go/job-search/internal/api"
"git.avsdev.uk/Go/job-search/internal/config"
"git.avsdev.uk/Go/job-search/internal/logging"
"git.avsdev.uk/Go/job-search/internal/middleware"
"git.avsdev.uk/Go/job-search/internal/store"
"git.avsdev.uk/Go/job-search/internal/util"
"git.avsdev.uk/Go/job-search/internal/websocket"
)
// Group identifies where hook middleware is inserted.
type Group int
// The route groups hook middleware can be added to (see WithMiddleware).
const (
// GroupGlobal is every request, after real-IP, request-ID and logging.
GroupGlobal Group = iota
// GroupAPI is /api/*, before CSRF protection.
GroupAPI
// GroupWebSocket is /ws upgrade requests.
GroupWebSocket
)
// Middleware is the standard net/http middleware shape.
type Middleware = func(http.Handler) http.Handler
// Option customises a Server.
type Option func(*Server)
// WithLogger makes the server, its middleware and its WebSocket hub log to
// logger instead of slog.Default().
func WithLogger(logger *slog.Logger) Option {
return func(s *Server) {
s.log = logger
}
}
// WithAccessLog makes the server write a line to logger for every request, in
// the Combined Log Format (see logging.AccessLog), as well as the request
// lines it logs with slog. The line is written when the request ends, and
// covers requests that middleware refuses.
func WithAccessLog(logger *log.Logger) Option {
return func(s *Server) {
s.accessLog = logger
}
}
// WithStore makes the server's API serve the opportunities in `store` under
// /api/opportunities (see api.WithStore). Without it there are no
// opportunity routes.
func WithStore(store *store.Store) Option {
return func(s *Server) {
s.store = store
}
}
// WithMiddleware appends middleware to a group. This is the generic hook
// (auth, metrics, tracing, custom header rules, ...).
func WithMiddleware(group Group, middlewares ...Middleware) Option {
return func(server *Server) {
server.hooks[group] = append(server.hooks[group], middlewares...)
}
}
// WithRateLimiter replaces the configured rate limiter of a group (for
// example with a Redis-backed implementation). Pass a nil Limiter to
// disable limiting for the group. A nil key defaults to the client IP.
func WithRateLimiter(
group Group, limiter middleware.Limiter, key middleware.KeyFunc,
) Option {
return func(server *Server) {
server.limiters[group] = limiterHook{
limiter: limiter,
key: key,
set: true,
}
}
}
// limiterHook is a group's rate limiter set by WithRateLimiter, replacing the
// configured one. set distinguishes a nil limiter (no limiting) from none.
type limiterHook struct {
limiter middleware.Limiter
key middleware.KeyFunc
set bool
}
// Server is the application server.
type Server struct {
conf *config.JSCServer
log *slog.Logger
// accessLog is where the access log is written, or nil for none.
accessLog *log.Logger
hooks map[Group][]Middleware
limiters map[Group]limiterHook
// trusted are the proxies whose forwarding headers are believed.
trusted []netip.Prefix
// proxyProtocolTrusted are the sources allowed to send a PROXY header.
proxyProtocolTrusted []netip.Prefix
csrf *middleware.CSRF
hub *websocket.Hub
// store is the opportunities' database, or nil for none.
store *store.Store
// api is the REST API mounted under /api, built from the options.
api *api.API
started time.Time
handler http.Handler
// backgroundCtx is for background work, such as the rate limiters'
// janitors; backgroundCancel stops it.
backgroundCtx context.Context
backgroundCancel context.CancelFunc
}
// New builds the server and its router. It logs to slog.Default() unless
// given WithLogger.
func New(conf *config.JSCServer, opts ...Option) (*Server, error) {
server := &Server{
conf: conf,
hooks: make(map[Group][]Middleware),
limiters: make(map[Group]limiterHook),
started: time.Now(),
}
for _, option := range opts {
option(server)
}
server.api = api.New(api.WithStore(server.store))
var err error
server.trusted, err = util.ParseNetPrefixes(conf.Proxy.TrustedProxies)
if err != nil {
return nil, fmt.Errorf("parsing proxy.trusted: %w", err)
}
proxyProtocolSources := conf.Proxy.ProxyProtocol.TrustedSources
if len(proxyProtocolSources) == 0 {
proxyProtocolSources = conf.Proxy.TrustedProxies
}
server.proxyProtocolTrusted, err = util.ParseNetPrefixes(
proxyProtocolSources,
)
if err != nil {
return nil, fmt.Errorf(
"parsing proxy.proxy_protocol.trusted_sources: %w", err,
)
}
server.csrf, err = middleware.NewCSRF(
conf.CSRF, middleware.WithLogger(server.untaggedLogger()),
)
if err != nil {
return nil, err
}
server.hub = websocket.NewHub(
conf.WebSocket,
websocket.WithLogger(server.untaggedLogger()),
websocket.WithTrustedOrigins(
conf.CSRF.CrossOriginProtection.TrustedOrigins,
),
)
server.backgroundCtx, server.backgroundCancel = context.WithCancel(
context.Background(),
)
server.handler, err = server.routes()
if err != nil {
server.backgroundCancel()
return nil, err
}
if server.accessLog != nil {
server.handler = logging.AccessLog(server.accessLog, true, server.handler)
}
return server, nil
}
// untaggedLogger returns the logger given WithLogger, or slog.Default(), for
// the server's middleware and hub to tag with their own components.
func (s *Server) untaggedLogger() *slog.Logger {
if s.log == nil {
return slog.Default()
}
return s.log
}
// logger returns the logger the server logs to, tagged with its component.
func (s *Server) logger() *slog.Logger {
return s.untaggedLogger().With("component", "server")
}
// Handler exposes the root handler (useful for httptest).
func (s *Server) Handler() http.Handler { return s.handler }
// PushEvent sends an event to every client of the server's WebSocket hub
// (see websocket.Hub.PushEvent).
func (s *Server) PushEvent(eventType string, payload any) {
s.hub.PushEvent(eventType, payload)
}
// Run listens and serves until ctx is cancelled, then shuts down gracefully.
// With TLS, it also serves an HTTP to HTTPS redirect if tls.redirect_http_port
// is set.
func (s *Server) Run(ctx context.Context) error {
defer s.backgroundCancel()
conf := s.conf
addr := net.JoinHostPort(conf.Host, strconv.Itoa(conf.Port))
listen, err := net.Listen("tcp", addr)
if err != nil {
return fmt.Errorf("listen %s: %w", addr, err)
}
if conf.Proxy.ProxyProtocol.Enabled {
listen = &proxyproto.Listener{
Listener: listen,
ConnPolicy: s.proxyPolicy,
ReadHeaderTimeout: noLimitIfUnset(
conf.Proxy.ProxyProtocol.ReadHeaderTimeout,
),
}
}
httpServer := &http.Server{
Handler: s.handler,
ReadHeaderTimeout: noLimitIfUnset(conf.Timeouts.ReadHeaderTimeout),
ReadTimeout: conf.Timeouts.ReadTimeout,
WriteTimeout: conf.Timeouts.WriteTimeout,
IdleTimeout: noLimitIfUnset(conf.Timeouts.IdleTimeout),
MaxHeaderBytes: maxHeaderBytes(conf),
ErrorLog: slog.NewLogLogger(
s.logger().Handler(), slog.LevelWarn,
),
}
if conf.TLS.Enabled {
minVersion := uint16(tls.VersionTLS12)
if conf.TLS.MinVersion == "1.3" {
minVersion = tls.VersionTLS13
}
httpServer.TLSConfig = &tls.Config{
MinVersion: minVersion,
}
}
serveErrors := make(chan error, 2)
go func() {
s.logger().Info(
"listening",
"addr", listen.Addr().String(),
"tls", conf.TLS.Enabled,
"proxy_protocol", conf.Proxy.ProxyProtocol.Enabled,
)
var err error
if conf.TLS.Enabled {
err = httpServer.ServeTLS(listen, conf.TLS.CertFile, conf.TLS.KeyFile)
} else {
err = httpServer.Serve(listen)
}
serveErrors <- err
}()
var redirect *http.Server
if conf.TLS.Enabled && conf.TLS.RedirectHTTPPort != 0 {
httpAddr := net.JoinHostPort(
conf.Host, strconv.Itoa(conf.TLS.RedirectHTTPPort),
)
redirect = &http.Server{
Addr: httpAddr,
Handler: middleware.RequestChecks(*conf)(httpsRedirect(addr)),
ReadHeaderTimeout: noLimitIfUnset(conf.Timeouts.ReadHeaderTimeout),
IdleTimeout: noLimitIfUnset(conf.Timeouts.IdleTimeout),
}
go func() {
s.logger().Info("http->https redirect listening", "addr", httpAddr)
serveErrors <- redirect.ListenAndServe()
}()
}
select {
case err := <-serveErrors:
if !errors.Is(err, http.ErrServerClosed) {
return err
}
case <-ctx.Done():
}
s.logger().Info("shutting down")
shutdownCtx, cancel := util.WithTimeout(
context.Background(), conf.Timeouts.ShutdownTimeout,
)
defer cancel()
// Stop accepting first, then close WebSockets (hijacked connections are
// not tracked by http.Server.Shutdown), then drain in-flight requests.
var errs []error
shutdownErr := make(chan error, 1)
go func() {
shutdownErr <- httpServer.Shutdown(shutdownCtx)
}()
if err := s.hub.Shutdown(shutdownCtx); err != nil {
errs = append(errs, fmt.Errorf("websocket shutdown: %w", err))
}
if err := <-shutdownErr; err != nil {
errs = append(errs, fmt.Errorf("http shutdown: %w", err))
}
if redirect != nil {
redirect.Shutdown(shutdownCtx) //nolint:errcheck,gosec // only redirects; nothing to drain
}
return errors.Join(errs...)
}
// noLimitIfUnset returns timeout for a field that reads 0 as some other
// timeout, such as http.Server's ReadHeaderTimeout and IdleTimeout (which
// use ReadTimeout) or the PROXY listener's ReadHeaderTimeout (10s): -1,
// which those fields read as no limit, if timeout is 0 or less, as the
// configuration's timeouts mean.
func noLimitIfUnset(timeout time.Duration) time.Duration {
if timeout <= 0 {
return -1
}
return timeout
}
// defaultMaxHeaderBytes is the largest request header block when
// max_header_bytes is 0 or less: 8KB, as the configuration's default. The
// configuration rejects -1 (no limit), as http.Server can't have one.
const defaultMaxHeaderBytes = 8_000
// maxHeaderBytes is the largest request header block: max_header_bytes, or
// defaultMaxHeaderBytes if that is 0 or less.
func maxHeaderBytes(conf *config.JSCServer) int {
if conf.MaxHeaderBytes <= 0 {
return defaultMaxHeaderBytes
}
return int(conf.MaxHeaderBytes)
}
// proxyPolicy decides per connection how to treat a PROXY header. Only
// trusted sources may supply one. Untrusted peers are SKIPped (the header is
// not parsed; a spoofed one becomes garbage to the HTTP parser and fails).
// Never return a non-ErrInvalidUpstream error here: it would stop Accept.
func (s *Server) proxyPolicy(
options proxyproto.ConnPolicyOptions,
) (proxyproto.Policy, error) {
ip := addrOf(options.Upstream)
for _, prefix := range s.proxyProtocolTrusted {
if ip.IsValid() && prefix.Contains(ip) {
if s.conf.Proxy.ProxyProtocol.Required {
return proxyproto.REQUIRE, nil
}
return proxyproto.USE, nil
}
}
return proxyproto.SKIP, nil
}
// addrOf returns the IP address of a connection's address (unmapped to IPv4
// if it is an IPv4-mapped IPv6 address), or the zero Addr if it has none.
func addrOf(netAddr net.Addr) netip.Addr {
switch tcpAddr := netAddr.(type) {
case *net.TCPAddr:
ip, _ := netip.AddrFromSlice(tcpAddr.IP)
return ip.Unmap()
default:
addrPort, err := netip.ParseAddrPort(netAddr.String())
if err != nil {
return netip.Addr{}
}
return addrPort.Addr().Unmap()
}
}
// httpsRedirect redirects every request permanently to the same host and
// path over HTTPS, on the port of tlsAddr (the TLS listener's address),
// which is left out if it is 443.
func httpsRedirect(tlsAddr string) http.Handler {
_, port, _ := net.SplitHostPort(tlsAddr)
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
host := r.Host
if hostname, _, err := net.SplitHostPort(host); err == nil {
host = hostname
}
if strings.Contains(host, ":") { // bare IPv6
host = "[" + host + "]"
}
if port != "" && port != "443" {
host = net.JoinHostPort(strings.Trim(host, "[]"), port)
}
// The redirect only switches the scheme (and port) of the URL the
// client asked for, on the host it asked for, so it can't send the
// client anywhere else; RequestChecks has checked that host.
http.Redirect( //nolint:gosec // G710: see above
w, r, "https://"+host+r.URL.RequestURI(), http.StatusPermanentRedirect,
)
})
}