scrabble-game/gateway/internal/connectsrv/server.go

// Package connectsrv implements the public Connect edge service over h2c. Execute
// rate-limits, authenticates (resolving the Authorization bearer token to a user
// id for non-auth operations), and dispatches to the transcode registry; the
// domain outcome is carried back in the ExecuteResponse result_code. Subscribe
// bridges the gateway push hub to a client server-stream with a keep-alive
// heartbeat.
package connectsrv

import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"io"
	"net"
	"net/http"
	"strconv"
	"strings"
	"time"

	"connectrpc.com/connect"
	"go.opentelemetry.io/otel/metric"
	"go.uber.org/zap"
	"golang.org/x/net/http2"
	"golang.org/x/net/http2/h2c"

	"scrabble/gateway/internal/backendclient"
	"scrabble/gateway/internal/config"
	"scrabble/gateway/internal/push"
	"scrabble/gateway/internal/ratelimit"
	"scrabble/gateway/internal/session"
	"scrabble/gateway/internal/transcode"
	"scrabble/gateway/internal/webui"
	edgev1 "scrabble/gateway/proto/edge/v1"
	"scrabble/gateway/proto/edge/v1/edgev1connect"
)

// heartbeatKind is the live-stream keep-alive event kind.
const heartbeatKind = "heartbeat"

// honeypotHeader marks a request the edge proxy routed from a honeypot decoy path;
// any request carrying it is treated as a scanner hit. The proxy strips any
// client-supplied value before setting its own, and a spoofed value only bans the
// spoofer, so trusting it is safe.
const honeypotHeader = "X-Scrabble-Honeypot"

// Limiter classes, the `class` attribute of gateway_rate_limited_total and the
// class field of the periodic rejection report.
const (
	classUser   = "user"
	classPublic = "public"
	classEmail  = "email"
	classAdmin  = "admin"
)

// Explicit h2c server sizing, made explicit rather than relying on the
// implicit defaults.
const (
	// h2cMaxConcurrentStreams bounds the open streams per client connection — the
	// x/net default made explicit. A real client holds one Subscribe stream plus a
	// few unary calls; only a synthetic load multiplexing many players over one
	// transport approaches it.
	h2cMaxConcurrentStreams = 250
	// h2cIdleTimeout closes a connection with no open streams. A live Subscribe
	// stream keeps its connection active, so long-lived clients are unaffected;
	// only abandoned connections are reaped.
	h2cIdleTimeout = 3 * time.Minute
)

// Server implements edgev1connect.GatewayHandler.
type Server struct {
	registry   *transcode.Registry
	sessions   *session.Cache
	backend    *backendclient.Client
	limiter    *ratelimit.Limiter
	tracker    *ratelimit.Tracker
	banlist    *ratelimit.Banlist
	honeytoken string
	hub        *push.Hub
	heartbeat  time.Duration
	log        *zap.Logger
	adminProxy http.Handler
	metrics    *serverMetrics

	maxBodyBytes int

	publicPolicy ratelimit.Policy
	userPolicy   ratelimit.Policy
	emailPolicy  ratelimit.Policy
	adminPolicy  ratelimit.Policy
}

// Deps carries the Server's dependencies. A nil Limiter, nil Tracker, zero
// RateLimit and non-positive MaxBodyBytes each select a safe default.
type Deps struct {
	Registry *transcode.Registry
	Sessions *session.Cache
	// Backend is the REST client backing the session-gated /dict blob route; a nil
	// value disables that route (it 404s).
	Backend *backendclient.Client
	Limiter *ratelimit.Limiter
	// Tracker accumulates limiter rejections for the periodic report; nil
	// selects a private tracker (rejections are then only counted, never
	// reported).
	Tracker *ratelimit.Tracker
	// Banlist enforces temporary IP bans on the hot path; nil selects a disabled
	// (inert) banlist.
	Banlist *ratelimit.Banlist
	// Honeytoken, when non-empty, is the planted bearer value whose presentation
	// bans the caller and raises a high-severity alarm.
	Honeytoken string
	Hub        *push.Hub
	RateLimit  config.RateLimitConfig
	Heartbeat  time.Duration
	Logger     *zap.Logger
	AdminProxy http.Handler
	Meter      metric.Meter
	// MaxBodyBytes caps one inbound request body and one Connect message read;
	// zero or negative selects config.DefaultMaxBodyBytes.
	MaxBodyBytes int
}

// NewServer constructs the edge service.
func NewServer(d Deps) *Server {
	log := d.Logger
	if log == nil {
		log = zap.NewNop()
	}
	maxBody := d.MaxBodyBytes
	if maxBody <= 0 {
		maxBody = config.DefaultMaxBodyBytes
	}
	tracker := d.Tracker
	if tracker == nil {
		tracker = ratelimit.NewTracker()
	}
	limiter := d.Limiter
	if limiter == nil {
		limiter = ratelimit.New()
	}
	banlist := d.Banlist
	if banlist == nil {
		banlist = ratelimit.NewBanlist(ratelimit.BanConfig{})
	}
	rl := d.RateLimit
	if rl == (config.RateLimitConfig{}) {
		rl = config.DefaultRateLimit()
	}
	return &Server{
		registry:     d.Registry,
		sessions:     d.Sessions,
		backend:      d.Backend,
		limiter:      limiter,
		tracker:      tracker,
		banlist:      banlist,
		honeytoken:   d.Honeytoken,
		hub:          d.Hub,
		heartbeat:    d.Heartbeat,
		log:          log,
		adminProxy:   d.AdminProxy,
		metrics:      newServerMetrics(d.Meter),
		maxBodyBytes: maxBody,
		publicPolicy: ratelimit.PerMinute(rl.PublicPerMinute, rl.PublicBurst),
		userPolicy:   ratelimit.PerMinute(rl.UserPerMinute, rl.UserBurst),
		emailPolicy:  ratelimit.Per(rl.EmailPer10Min, 10*time.Minute, rl.EmailBurst),
		adminPolicy:  ratelimit.PerMinute(rl.AdminPerMinute, rl.AdminBurst),
	}
}

// HTTPHandler returns the h2c-wrapped Connect handler ready to serve.
func (s *Server) HTTPHandler() http.Handler {
	mux := http.NewServeMux()
	// The Connect read cap mirrors the HTTP-level body cap below; an oversized
	// Execute message is refused (resource_exhausted) instead of buffered.
	path, h := edgev1connect.NewGatewayHandler(s, connect.WithReadMaxBytes(s.maxBodyBytes))
	mux.Handle(path, h)
	if s.adminProxy != nil {
		// The admin console (backend /_gm) is served on the public listener behind
		// the proxy's Basic-Auth, mounted below the h2c wrap so the Connect edge keeps
		// working over h2c (docs/ARCHITECTURE.md §12). In the deployed contour the
		// front caddy owns the /_gm Basic-Auth and Grafana routing and proxies /_gm to
		// the backend directly (bypassing this mount); this mount serves a non-caddy
		// (local) setup. The per-IP admin limiter class guards it — notably a Basic-Auth
		// brute force. Note: the shared maxBodyHandler cap (1 MiB by default) also covers
		// this mount, so a dictionary-archive upload through the gateway-fronted console
		// needs a larger GATEWAY_MAX_BODY_BYTES; the contour path (caddy → backend) is not
		// affected and the backend self-caps the upload.
		mux.Handle("/_gm/", s.limitAdmin(s.adminProxy))
	} else {
		// With the console disabled here, keep /_gm a 404 so the SPA catch-all below
		// does not serve the app shell at the operator path.
		mux.Handle("/_gm/", http.NotFoundHandler())
	}
	// The client-side local move preview pulls each game's pinned dictionary blob
	// through this session-gated route (not public); see dictBytesHandler.
	mux.Handle("/dict/", s.dictBytesHandler())
	mux.Handle("/dl/", s.exportDownloadHandler())
	// The client posts its local-move-preview adoption telemetry here (session-gated).
	mux.Handle("/metrics/local-eval", s.localEvalMetricsHandler())
	// The embedded UI: the game SPA under /app/ (web), /telegram/ (the Telegram Mini
	// App) and /vk/ (the VK Mini App) — the single-origin model (docs/ARCHITECTURE.md
	// §13). All sit below the h2c wrap so the Connect edge (a more specific prefix) keeps
	// priority, and each mount falls back to the app shell (index.html) for the hash
	// router. The public landing lives in its own static container behind the contour
	// caddy, so the catch-all redirects a stray root hit to the app shell — which keeps a
	// local no-caddy run usable.
	mux.Handle("/telegram/", webui.Handler("/telegram/", "index.html"))
	mux.Handle("/vk/", webui.Handler("/vk/", "index.html"))
	mux.Handle("/app/", webui.Handler("/app/", "index.html"))
	mux.Handle("/", http.RedirectHandler("/app/", http.StatusPermanentRedirect))
	// abuseGuard is the outermost wrap (right under h2c) so a banned IP or a
	// honeypot hit is turned away before the body cap and the mux. Every request
	// body on the public listener is then capped (the admin proxy POSTs included);
	// the h2c server carries explicit stream/idle sizing.
	return h2c.NewHandler(s.abuseGuard(maxBodyHandler(s.maxBodyBytes, mux)), &http2.Server{
		MaxConcurrentStreams: h2cMaxConcurrentStreams,
		IdleTimeout:          h2cIdleTimeout,
	})
}

// maxBodyHandler caps every inbound request body at limit bytes: a read past the
// cap fails with *http.MaxBytesError and the connection is marked to close.
func maxBodyHandler(limit int, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		r.Body = http.MaxBytesReader(w, r.Body, int64(limit))
		next.ServeHTTP(w, r)
	})
}

// abuseGuard refuses a banned client IP with 429 before any work, and turns a
// honeypot decoy hit (the proxy-set honeypotHeader) into an instant ban plus a
// bland 404 that is indistinguishable from an ordinary miss. The ban check and the
// tripwire ban are inert on a disabled banlist (the prod-only gate); the tripwire
// hit is logged either way as scanner telemetry.
func (s *Server) abuseGuard(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ip := peerIP(r.RemoteAddr, r.Header)
		if s.banlist.Banned(ip) {
			http.Error(w, "banned", http.StatusTooManyRequests)
			return
		}
		if r.Header.Get(honeypotHeader) != "" {
			s.log.Warn("honeypot tripwire",
				zap.String("path", r.URL.Path),
				zap.String("client_ip", ip))
			if s.banlist.BanNow(ip, ratelimit.ReasonTripwire) {
				s.metrics.recordBan(r.Context(), string(ratelimit.ReasonTripwire))
			}
			http.NotFound(w, r)
			return
		}
		next.ServeHTTP(w, r)
	})
}

// Execute runs one unary operation. Domain failures are returned in the envelope
// (result_code != "ok", HTTP 200); only edge failures (rate limit, missing
// session, unknown type, internal) become Connect errors.
func (s *Server) Execute(ctx context.Context, req *connect.Request[edgev1.ExecuteRequest]) (*connect.Response[edgev1.ExecuteResponse], error) {
	start := time.Now()
	msgType := req.Msg.GetMessageType()
	result := "internal"
	defer func() { s.metrics.recordEdge(ctx, msgType, result, start) }()

	op, ok := s.registry.Lookup(msgType)
	if !ok {
		result = "unknown_type"
		return nil, connect.NewError(connect.CodeNotFound, errUnknownMessageType(msgType))
	}
	clientIP := peerIP(req.Peer().Addr, req.Header())

	tr := transcode.Request{Payload: req.Msg.GetPayload(), ClientIP: clientIP}
	if op.Auth {
		uid, isGuest, err := s.resolve(ctx, req.Header(), clientIP)
		if err != nil {
			result = "unauthenticated"
			return nil, err
		}
		// A guest may not perform a non-guest operation: reject it here as a domain
		// outcome, before any backend call.
		if op.NonGuest && isGuest {
			result = "domain"
			return connect.NewResponse(&edgev1.ExecuteResponse{
				RequestId:  req.Msg.GetRequestId(),
				ResultCode: "guest_forbidden",
			}), nil
		}
		// A valid session proving an authenticated request is an "action" for the
		// active_users gauge, counted before the rate-limit/domain outcome.
		s.metrics.recordActive(uid)
		if !s.limiter.Allow("user:"+uid, s.userPolicy) {
			result = "rate_limited"
			return nil, s.rejectRateLimited(ctx, classUser, uid, msgType)
		}
		tr.UserID = uid
	} else {
		if !s.limiter.Allow("ip:"+clientIP, s.publicPolicy) {
			result = "rate_limited"
			return nil, s.rejectRateLimited(ctx, classPublic, clientIP, msgType)
		}
		if op.Email && !s.limiter.Allow("email:"+clientIP, s.emailPolicy) {
			result = "rate_limited"
			return nil, s.rejectRateLimited(ctx, classEmail, clientIP, msgType)
		}
	}

	payload, err := op.Handler(ctx, tr)
	if err != nil {
		if code, domain := transcode.DomainCode(err); domain {
			result = "domain"
			return connect.NewResponse(&edgev1.ExecuteResponse{
				RequestId:  req.Msg.GetRequestId(),
				ResultCode: code,
			}), nil
		}
		s.log.Error("execute failed", zap.String("message_type", msgType), zap.Error(err))
		return nil, connect.NewError(connect.CodeInternal, errInternal)
	}
	result = "ok"
	return connect.NewResponse(&edgev1.ExecuteResponse{
		RequestId:  req.Msg.GetRequestId(),
		ResultCode: "ok",
		Payload:    payload,
	}), nil
}

// Subscribe streams the authenticated user's live events with a keep-alive
// heartbeat until the client disconnects.
func (s *Server) Subscribe(ctx context.Context, req *connect.Request[edgev1.SubscribeRequest], stream *connect.ServerStream[edgev1.Event]) error {
	uid, _, err := s.resolve(ctx, req.Header(), peerIP(req.Peer().Addr, req.Header()))
	if err != nil {
		return err
	}
	if !s.limiter.Allow("user:"+uid, s.userPolicy) {
		return s.rejectRateLimited(ctx, classUser, uid, "subscribe")
	}

	events, cancel := s.hub.Subscribe(uid)
	defer cancel()

	// Send an immediate heartbeat so the stream's first byte flushes through the proxy chain
	// right away and resets edge/client idle timers, instead of the connection sitting silent
	// until the first tick — which otherwise raced a ~15 s idle timeout and forced a reconnect
	// every interval.
	if err := stream.Send(&edgev1.Event{Kind: heartbeatKind}); err != nil {
		return err
	}

	ticker := time.NewTicker(s.heartbeat)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			if err := stream.Send(&edgev1.Event{Kind: heartbeatKind}); err != nil {
				return err
			}
		case e, ok := <-events:
			if !ok {
				return nil
			}
			if err := stream.Send(&edgev1.Event{Kind: e.Kind, Payload: e.Payload, EventId: e.EventID}); err != nil {
				return err
			}
		}
	}
}

// noteRateLimited accounts one limiter rejection: the aggregate counter, the
// per-rejection Debug line and the periodic-report tracker. The operational
// signal is the reporter's Warn summary; per-rejection logging stays at Debug so
// a rejection flood cannot flood the log.
func (s *Server) noteRateLimited(ctx context.Context, class, key, msgType string) {
	s.metrics.recordRateLimited(ctx, class)
	s.tracker.Add(class, key)
	// IP-keyed rejections (public, email, admin — the key is the client IP) feed
	// the ban; the user class is keyed by account id and is the backend soft-flag's
	// concern, not the IP ban's.
	if class != classUser && s.banlist.Strike(key) {
		s.metrics.recordBan(ctx, string(ratelimit.ReasonRejections))
	}
	s.log.Debug("rate limited",
		zap.String("class", class),
		zap.String("key", key),
		zap.String("message_type", msgType))
}

// rejectRateLimited accounts one limiter rejection and returns the Connect error
// for the caller.
func (s *Server) rejectRateLimited(ctx context.Context, class, key, msgType string) error {
	s.noteRateLimited(ctx, class, key, msgType)
	return connect.NewError(connect.CodeResourceExhausted, errRateLimited)
}

// limitAdmin guards the admin proxy with the per-IP admin limiter class, ahead
// of its Basic-Auth check (a credential brute force is exactly what it bounds).
// It covers the gateway-fronted /_gm mount; in the deployed contour /_gm reaches
// the backend through caddy, whose Basic-Auth has no limiter (stock caddy) — see
// docs/ARCHITECTURE.md §12.
func (s *Server) limitAdmin(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ip := peerIP(r.RemoteAddr, r.Header)
		if !s.limiter.Allow("admin:"+ip, s.adminPolicy) {
			s.noteRateLimited(r.Context(), classAdmin, ip, "admin")
			http.Error(w, "rate limited", http.StatusTooManyRequests)
			return
		}
		next.ServeHTTP(w, r)
	})
}

// dictBytesHandler serves the raw dictionary blob for a (variant, version) pair to
// a signed-in client (the local move preview), proxying the backend's authed
// endpoint. It is session-gated — not public — bounds the download with the
// user-class limiter, and forwards the backend's immutable Cache-Control so the
// browser caches the blob hard. Only GET is allowed; the path is
// /dict/{variant}/{version}.
// exportDownloadHandler serves the signed finished-game export downloads (/dl/*).
// It is the gateway's only unauthenticated data route: the platforms' native
// download calls (Telegram downloadFile, VKWebAppDownloadFile, a plain anchor)
// carry no session, so the URL's HMAC — verified by the backend — is the whole
// grant. The gateway only rate-limits by IP and forwards, passing the public Host
// along for the image footer.
func (s *Server) exportDownloadHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if s.backend == nil {
			http.NotFound(w, r)
			return
		}
		if r.Method != http.MethodGet {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}
		ip := peerIP(r.RemoteAddr, r.Header)
		if !s.limiter.Allow("public:"+ip, s.publicPolicy) {
			s.noteRateLimited(r.Context(), classPublic, ip, "export-dl")
			http.Error(w, "rate limited", http.StatusTooManyRequests)
			return
		}
		rest := strings.TrimPrefix(r.URL.Path, "/dl")
		if rest == "" || rest == "/" {
			http.NotFound(w, r)
			return
		}
		if r.URL.RawQuery != "" {
			rest += "?" + r.URL.RawQuery
		}
		data, contentType, disposition, err := s.backend.ExportDownload(r.Context(), rest, r.Host)
		if err != nil {
			var apiErr *backendclient.APIError
			if errors.As(err, &apiErr) && apiErr.Status < http.StatusInternalServerError {
				http.NotFound(w, r) // invalid, expired or unknown link
				return
			}
			s.log.Warn("export download failed", zap.Error(err))
			http.Error(w, "bad gateway", http.StatusBadGateway)
			return
		}
		if contentType != "" {
			w.Header().Set("Content-Type", contentType)
		}
		if disposition != "" {
			w.Header().Set("Content-Disposition", disposition)
		}
		w.Header().Set("X-Content-Type-Options", "nosniff")
		w.Header().Set("Cache-Control", "private, max-age=0")
		// Explicit length, or the body goes out chunked and Android's system
		// DownloadManager (the VKWebAppDownloadFile executor) hangs on it.
		w.Header().Set("Content-Length", strconv.Itoa(len(data)))
		_, _ = w.Write(data)
	})
}

func (s *Server) dictBytesHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if s.backend == nil {
			http.NotFound(w, r)
			return
		}
		if r.Method != http.MethodGet {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}
		ip := peerIP(r.RemoteAddr, r.Header)
		if !s.limiter.Allow("user:"+ip, s.userPolicy) {
			s.noteRateLimited(r.Context(), classUser, ip, "dict")
			http.Error(w, "rate limited", http.StatusTooManyRequests)
			return
		}
		uid, _, err := s.resolve(r.Context(), r.Header, ip)
		if err != nil {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		variant, version, ok := parseDictPath(r.URL.Path)
		if !ok {
			http.NotFound(w, r)
			return
		}
		data, cacheControl, err := s.backend.DictBytes(r.Context(), uid, variant, version)
		if err != nil {
			var apiErr *backendclient.APIError
			if errors.As(err, &apiErr) && apiErr.Status < http.StatusInternalServerError {
				http.NotFound(w, r) // unknown variant or version
				return
			}
			s.log.Warn("dict fetch failed", zap.Error(err))
			http.Error(w, "bad gateway", http.StatusBadGateway)
			return
		}
		if cacheControl != "" {
			w.Header().Set("Cache-Control", cacheControl)
		}
		w.Header().Set("Content-Type", "application/octet-stream")
		_, _ = w.Write(data)
	})
}

// parseDictPath splits /dict/{variant}/{version}; both segments must be present
// and non-empty, and the version must not contain a further slash.
func parseDictPath(p string) (variant, version string, ok bool) {
	rest, found := strings.CutPrefix(p, "/dict/")
	if !found {
		return "", "", false
	}
	i := strings.IndexByte(rest, '/')
	if i <= 0 || i == len(rest)-1 {
		return "", "", false
	}
	variant, version = rest[:i], rest[i+1:]
	if strings.Contains(version, "/") {
		return "", "", false
	}
	return variant, version, true
}

// localEvalMetricsHandler ingests a client's local move-preview telemetry batch into the
// edge's adoption counters (docs/ARCHITECTURE.md §11). Session-gated so only real clients
// report; the values are aggregate (no per-user attributes) and clamped against a spoofed
// inflation. Only POST; the body is a small JSON of per-metric deltas.
func (s *Server) localEvalMetricsHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}
		ip := peerIP(r.RemoteAddr, r.Header)
		if _, _, err := s.resolve(r.Context(), r.Header, ip); err != nil {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		var rep localEvalReport
		if err := json.NewDecoder(io.LimitReader(r.Body, 512)).Decode(&rep); err != nil {
			http.Error(w, "bad request", http.StatusBadRequest)
			return
		}
		clampReport(&rep)
		s.metrics.recordLocalEval(r.Context(), rep)
		w.WriteHeader(http.StatusNoContent)
	})
}

// clampReport bounds each counter of a client report against a spoofed inflation — one batch
// reflects at most a few minutes of a single client's activity.
func clampReport(r *localEvalReport) {
	const maxPerField = 1000
	clamp := func(n *int) {
		if *n < 0 {
			*n = 0
		} else if *n > maxPerField {
			*n = maxPerField
		}
	}
	clamp(&r.ColdStart)
	clamp(&r.DictFetched)
	clamp(&r.DictCacheHit)
	clamp(&r.DictMiss)
	clamp(&r.PreviewLocal)
	clamp(&r.PreviewNetwork)
}

// resolve extracts and resolves the Authorization bearer token to an account id
// and its guest flag, returning a Connect Unauthenticated error when it is missing
// or unknown.
func (s *Server) resolve(ctx context.Context, h http.Header, clientIP string) (string, bool, error) {
	token := bearerToken(h.Get("Authorization"))
	if token == "" {
		return "", false, connect.NewError(connect.CodeUnauthenticated, errMissingToken)
	}
	// The honeytoken is a planted value no real client holds: presenting it is a
	// high-confidence intrusion signal, so ban the caller and raise the alarm, then
	// return the ordinary invalid-session error so the trap stays indistinguishable.
	if s.honeytoken != "" && subtle.ConstantTimeCompare([]byte(token), []byte(s.honeytoken)) == 1 {
		s.log.Warn("honeytoken presented", zap.String("client_ip", clientIP))
		if s.banlist.BanNow(clientIP, ratelimit.ReasonHoneytoken) {
			s.metrics.recordBan(ctx, string(ratelimit.ReasonHoneytoken))
		}
		return "", false, connect.NewError(connect.CodeUnauthenticated, errInvalidSession)
	}
	uid, isGuest, err := s.sessions.Resolve(ctx, token)
	if err != nil {
		// An unknown or expired token (a backend 4xx) is the client's problem and
		// stays silent; anything else — a resolve timeout, a refused connection, a
		// backend 5xx — is an infra failure misread as "unauthenticated" by the
		// client, so surface the cause (the transient resolves seen under load).
		// The token itself is never logged.
		var apiErr *backendclient.APIError
		if !errors.As(err, &apiErr) || apiErr.Status >= http.StatusInternalServerError {
			s.log.Warn("session resolve failed", zap.Error(err))
		}
		return "", false, connect.NewError(connect.CodeUnauthenticated, errInvalidSession)
	}
	return uid, isGuest, nil
}

// bearerToken extracts the token from an "Authorization: Bearer <token>" header,
// tolerating a bare token for convenience.
func bearerToken(header string) string {
	header = strings.TrimSpace(header)
	if header == "" {
		return ""
	}
	if rest, ok := strings.CutPrefix(header, "Bearer "); ok {
		return strings.TrimSpace(rest)
	}
	return header
}

// peerIP prefers the X-Forwarded-For client hop, falling back to the connection
// peer address (host part).
func peerIP(peerAddr string, h http.Header) string {
	if xff := h.Get("X-Forwarded-For"); xff != "" {
		first, _, _ := strings.Cut(xff, ",")
		return strings.TrimSpace(first)
	}
	if host, _, err := net.SplitHostPort(peerAddr); err == nil {
		return host
	}
	return peerAddr
}