diff --git a/cache/ttllru.go b/cache/ttllru.go new file mode 100644 index 0000000..462b8aa --- /dev/null +++ b/cache/ttllru.go @@ -0,0 +1,87 @@ +// Package cache provides a small, concurrency-safe TTL + LRU cache. It's a +// generic building block for sharing an expensive computation across requests +// (e.g. the reader shares one relevance match between /api/stream and +// /api/facets): a hit returns the stored value, a miss recomputes. The TTL +// bounds staleness; the capacity bounds memory. +package cache + +import ( + "container/list" + "sync" + "time" +) + +type entry[K comparable, V any] struct { + key K + val V + at time.Time +} + +// TTLLRU is a fixed-capacity cache that evicts the least-recently-used entry on +// overflow and treats entries older than ttl as misses. The zero value is not +// usable — construct with New. All methods are safe for concurrent use. +type TTLLRU[K comparable, V any] struct { + mu sync.Mutex + ll *list.List // front = most-recently-used + m map[K]*list.Element + cap int + ttl time.Duration +} + +// New returns a cache holding up to capacity entries, each valid for ttl. +func New[K comparable, V any](capacity int, ttl time.Duration) *TTLLRU[K, V] { + if capacity < 1 { + capacity = 1 + } + return &TTLLRU[K, V]{ + ll: list.New(), + m: make(map[K]*list.Element, capacity), + cap: capacity, + ttl: ttl, + } +} + +// Get returns the cached value for key. The boolean is false on a miss or when +// the entry has expired (in which case it is also evicted). A hit refreshes the +// entry's recency. +func (c *TTLLRU[K, V]) Get(key K) (V, bool) { + c.mu.Lock() + defer c.mu.Unlock() + el, ok := c.m[key] + if !ok { + var zero V + return zero, false + } + ent := el.Value.(*entry[K, V]) + if time.Since(ent.at) > c.ttl { + c.ll.Remove(el) + delete(c.m, key) + var zero V + return zero, false + } + c.ll.MoveToFront(el) + return ent.val, true +} + +// Put stores val under key, refreshing its timestamp and recency, evicting the +// least-recently-used entry if that exceeds the capacity. +func (c *TTLLRU[K, V]) Put(key K, val V) { + c.mu.Lock() + defer c.mu.Unlock() + if el, ok := c.m[key]; ok { + ent := el.Value.(*entry[K, V]) + ent.val = val + ent.at = time.Now() + c.ll.MoveToFront(el) + return + } + c.m[key] = c.ll.PushFront(&entry[K, V]{key: key, val: val, at: time.Now()}) + for c.ll.Len() > c.cap { + back := c.ll.Back() + if back == nil { + break + } + c.ll.Remove(back) + delete(c.m, back.Value.(*entry[K, V]).key) + } +} diff --git a/cache/ttllru_test.go b/cache/ttllru_test.go new file mode 100644 index 0000000..31cbe70 --- /dev/null +++ b/cache/ttllru_test.go @@ -0,0 +1,47 @@ +package cache + +import ( + "testing" + "time" +) + +func TestGetPut(t *testing.T) { + c := New[string, int](2, time.Minute) + if _, ok := c.Get("a"); ok { + t.Fatal("empty cache returned a hit") + } + c.Put("a", 1) + got, ok := c.Get("a") + if !ok || got != 1 { + t.Fatalf("get after put: ok=%v val=%v", ok, got) + } +} + +func TestTTL(t *testing.T) { + c := New[string, int](8, 10*time.Millisecond) + c.Put("a", 1) + if _, ok := c.Get("a"); !ok { + t.Fatal("fresh entry missing") + } + time.Sleep(20 * time.Millisecond) + if _, ok := c.Get("a"); ok { + t.Fatal("expired entry still served") + } +} + +func TestLRUEviction(t *testing.T) { + c := New[string, int](2, time.Minute) + c.Put("a", 1) + c.Put("b", 2) + _, _ = c.Get("a") // make "a" most-recently-used so "b" is the eviction victim + c.Put("c", 3) // exceeds cap → evict LRU ("b") + if _, ok := c.Get("b"); ok { + t.Fatal("LRU victim 'b' not evicted") + } + if _, ok := c.Get("a"); !ok { + t.Fatal("recently-used 'a' was evicted") + } + if _, ok := c.Get("c"); !ok { + t.Fatal("freshly-put 'c' missing") + } +} diff --git a/conv/conv.go b/conv/conv.go new file mode 100644 index 0000000..b75d469 --- /dev/null +++ b/conv/conv.go @@ -0,0 +1,140 @@ +package conv + +import ( + "encoding/json" + "fmt" + "strconv" + "strings" +) + +// Uint32 coerces an interface value to a uint32. +// It returns the uint32 value and true on success, or 0 and false if coercion fails. +func Uint32(v any) (uint32, bool) { + if v == nil { + return 0, false + } + switch n := v.(type) { + case uint32: + return n, true + case float64: + if n < 0 || n > 4294967295 { + return 0, false + } + return uint32(n), true + case float32: + if n < 0 || n > 4294967295 { + return 0, false + } + return uint32(n), true + case int: + if n < 0 || n > 4294967295 { + return 0, false + } + return uint32(n), true + case int64: + if n < 0 || n > 4294967295 { + return 0, false + } + return uint32(n), true + case uint64: + if n > 4294967295 { + return 0, false + } + return uint32(n), true + case json.Number: + u, err := strconv.ParseUint(n.String(), 10, 32) + if err != nil { + return 0, false + } + return uint32(u), true + case string: + s := strings.TrimSpace(n) + if s == "" { + return 0, false + } + u, err := strconv.ParseUint(s, 10, 32) + if err != nil { + return 0, false + } + return uint32(u), true + } + return 0, false +} + +// String coerces an interface value to a string. +// It handles common numeric types and falls back to fmt.Sprintf. +func String(v any) string { + if v == nil { + return "" + } + switch n := v.(type) { + case string: + return n + case int: + return strconv.Itoa(n) + case int64: + return strconv.FormatInt(n, 10) + case uint32: + return strconv.FormatUint(uint64(n), 10) + case uint64: + return strconv.FormatUint(n, 10) + case float64: + // If it's a whole number, represent it as an integer to avoid scientific notation + if n == float64(int64(n)) { + return strconv.FormatInt(int64(n), 10) + } + return strconv.FormatFloat(n, 'f', -1, 64) + case float32: + if n == float32(int32(n)) { + return strconv.Itoa(int(n)) + } + return strconv.FormatFloat(float64(n), 'f', -1, 32) + case json.Number: + return n.String() + case bool: + if n { + return "true" + } + return "false" + default: + return fmt.Sprintf("%v", v) + } +} + +// Float64 coerces an interface value to a float64. +func Float64(v any) (float64, bool) { + if v == nil { + return 0, false + } + switch n := v.(type) { + case float64: + return n, true + case float32: + return float64(n), true + case int: + return float64(n), true + case int64: + return float64(n), true + case uint32: + return float64(n), true + case uint64: + return float64(n), true + case string: + s := strings.TrimSpace(n) + if s == "" { + return 0, false + } + f, err := strconv.ParseFloat(s, 64) + if err != nil { + return 0, false + } + return f, true + case json.Number: + f, err := n.Float64() + if err != nil { + return 0, false + } + return f, true + } + return 0, false +} diff --git a/conv/conv_test.go b/conv/conv_test.go new file mode 100644 index 0000000..dfd7ce1 --- /dev/null +++ b/conv/conv_test.go @@ -0,0 +1,90 @@ +package conv + +import ( + "encoding/json" + "testing" +) + +func TestUint32(t *testing.T) { + tests := []struct { + input any + expected uint32 + ok bool + }{ + {uint32(123), 123, true}, + {float64(456.0), 456, true}, + {float64(-1.0), 0, false}, + {float32(789.0), 789, true}, + {100, 100, true}, + {int64(200), 200, true}, + {uint64(300), 300, true}, + {uint64(0xFFFFFFFF + 1), 0, false}, + {json.Number("42"), 42, true}, + {json.Number("invalid"), 0, false}, + {"500", 500, true}, + {" 600 ", 600, true}, + {"invalid", 0, false}, + {nil, 0, false}, + } + + for _, tt := range tests { + val, ok := Uint32(tt.input) + if val != tt.expected || ok != tt.ok { + t.Errorf("Uint32(%v) = (%d, %t), expected (%d, %t)", tt.input, val, ok, tt.expected, tt.ok) + } + } +} + +func TestString(t *testing.T) { + tests := []struct { + input any + expected string + }{ + {"hello", "hello"}, + {123, "123"}, + {int64(456), "456"}, + {uint32(789), "789"}, + {uint64(100), "100"}, + {float64(12.34), "12.34"}, + {float64(56.0), "56"}, + {float32(78.0), "78"}, + {json.Number("42.2"), "42.2"}, + {true, "true"}, + {false, "false"}, + {nil, ""}, + } + + for _, tt := range tests { + val := String(tt.input) + if val != tt.expected { + t.Errorf("String(%v) = %q, expected %q", tt.input, val, tt.expected) + } + } +} + +func TestFloat64(t *testing.T) { + tests := []struct { + input any + expected float64 + ok bool + }{ + {float64(12.34), 12.34, true}, + {float32(12.5), 12.5, true}, + {100, 100.0, true}, + {int64(200), 200.0, true}, + {uint32(300), 300.0, true}, + {uint64(400), 400.0, true}, + {"12.34", 12.34, true}, + {" 56.78 ", 56.78, true}, + {"invalid", 0.0, false}, + {json.Number("42.2"), 42.2, true}, + {nil, 0.0, false}, + } + + for _, tt := range tests { + val, ok := Float64(tt.input) + if val != tt.expected || ok != tt.ok { + t.Errorf("Float64(%v) = (%f, %t), expected (%f, %t)", tt.input, val, ok, tt.expected, tt.ok) + } + } +} diff --git a/httpclient/clientip.go b/httpclient/clientip.go new file mode 100644 index 0000000..677729e --- /dev/null +++ b/httpclient/clientip.go @@ -0,0 +1,38 @@ +// Package httpclient provides shared HTTP client helpers used across the +// monorepo: client IP extraction, CORS middleware, and related utilities. +package httpclient + +import ( + "net" + "net/http" + "strings" +) + +// ClientIP extracts the best-effort client IP from a request by checking +// common reverse-proxy and CDN headers in priority order, falling back to +// the TCP remote address when no header is present. It does NOT respect +// query-parameter overrides (those belong at the handler level — see +// slask-finder/cmd/stores for the ?ip= debug pattern). +// +// Priority order: CF-Connecting-IP → X-Real-IP → X-Forwarded-For (first +// entry) → RemoteAddr (with port stripped when present). +func ClientIP(r *http.Request) string { + for _, h := range []string{"CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"} { + v := r.Header.Get(h) + if v == "" { + continue + } + // X-Forwarded-For may be a comma-separated list; take the first. + if h == "X-Forwarded-For" { + if i := strings.IndexByte(v, ','); i >= 0 { + v = v[:i] + } + } + return strings.TrimSpace(v) + } + // Fall back to the TCP remote address, stripping the port when present. + if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil { + return host + } + return r.RemoteAddr +} diff --git a/httpclient/cors.go b/httpclient/cors.go new file mode 100644 index 0000000..d0a4923 --- /dev/null +++ b/httpclient/cors.go @@ -0,0 +1,101 @@ +package httpclient + +import ( + "net/http" +) + +type corsConfig struct { + origin string // "" = mirror request Origin; "*" = wildcard (no credentials) + methods string + headers string + exposeHeaders string +} + +// CORSOption customises the CORS middleware. +type CORSOption func(*corsConfig) + +// WithOrigin sets the Access-Control-Allow-Origin value. Use "*" for public +// APIs that don't need credentials. The default (empty string) mirrors the +// request's Origin header and enables credentials. +func WithOrigin(origin string) CORSOption { + return func(c *corsConfig) { c.origin = origin } +} + +// WithMethods sets Access-Control-Allow-Methods. Default: "GET, POST, PUT, OPTIONS". +func WithMethods(methods string) CORSOption { + return func(c *corsConfig) { c.methods = methods } +} + +// WithHeaders sets Access-Control-Allow-Headers. Default: "*". +func WithHeaders(headers string) CORSOption { + return func(c *corsConfig) { c.headers = headers } +} + +// WithExposeHeaders sets Access-Control-Expose-Headers. Default: empty. +func WithExposeHeaders(headers string) CORSOption { + return func(c *corsConfig) { c.exposeHeaders = headers } +} + +// CORS returns middleware that sets CORS response headers and handles +// preflight (OPTIONS) requests. By default it mirrors the request Origin, +// enables credentials, and allows GET/POST/PUT/OPTIONS with any request +// headers. +// +// Use WithOrigin("*") for public APIs that don't need credentials (per the +// CORS spec, credentials and wildcard origin are mutually exclusive). +// +// Common patterns: +// +// // Public API (wildcard origin, no credentials): +// mux := http.NewServeMux() +// handler := httpclient.CORS( +// httpclient.WithOrigin("*"), +// httpclient.WithMethods("GET, POST, OPTIONS"), +// )(mux) +// +// // Authenticated API (origin reflection, credentials): +// handler := httpclient.CORS()(mux) +func CORS(opts ...CORSOption) func(http.Handler) http.Handler { + cfg := corsConfig{ + methods: "GET, POST, PUT, OPTIONS", + headers: "*", + } + for _, o := range opts { + o(&cfg) + } + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + setCORSHeaders(w, r, cfg) + if r.Method == http.MethodOptions { + w.Header().Set("Access-Control-Max-Age", "86400") + w.WriteHeader(http.StatusNoContent) + return + } + next.ServeHTTP(w, r) + }) + } +} + +func setCORSHeaders(w http.ResponseWriter, r *http.Request, cfg corsConfig) { + origin := cfg.origin + if origin == "" { + origin = r.Header.Get("Origin") + } + if origin == "" { + return + } + w.Header().Set("Access-Control-Allow-Origin", origin) + // Per spec, credentials can't be used with wildcard origin. + if origin != "*" { + w.Header().Set("Access-Control-Allow-Credentials", "true") + } + if cfg.exposeHeaders != "" { + w.Header().Set("Access-Control-Expose-Headers", cfg.exposeHeaders) + } + if cfg.methods != "" { + w.Header().Set("Access-Control-Allow-Methods", cfg.methods) + } + if cfg.headers != "" { + w.Header().Set("Access-Control-Allow-Headers", cfg.headers) + } +} diff --git a/httpserver/server.go b/httpserver/server.go new file mode 100644 index 0000000..110cc8e --- /dev/null +++ b/httpserver/server.go @@ -0,0 +1,253 @@ +// Package httpserver holds the shared HTTP server lifecycle helpers used by +// every service in the monorepo: graceful shutdown with ordered hooks, +// timeout configuration, and standard probe/debug mux registration. +// +// This package consolidates the ~40-line signal+server+shutdown block that +// was hand-rolled in 12+ service main.go files (go-redirector, +// go-form-submissions, go-order-manager, go-cart-actor cmds, go-graph-cms-v2, +// slask-finder cmds, backoffice). It is promoted from +// slask-finder/pkg/common/graceful.go, which was the first implementation. +// +// # Usage +// +// Simplest (no custom readiness, no shutdown hooks): +// +// mux := http.NewServeMux() +// // … register handlers … +// httpserver.RegisterProbes(mux, nil) // /healthz + /readyz (always ready) +// +// server := &http.Server{Addr: ":8080", Handler: mux, ReadHeaderTimeout: 5*time.Second} +// if err := httpserver.RunServerWithShutdown(server, "my-service", 15*time.Second, 5*time.Second); err != nil { +// log.Fatalf("server: %v", err) +// } +// +// With custom readiness and shutdown hooks: +// +// httpserver.RegisterProbes(mux, func() bool { return store.Loaded() }) +// +// if err := httpserver.RunServerWithShutdown(server, "my-service", 15*time.Second, 5*time.Second, +// func(ctx context.Context) error { return store.Flush() }, +// func(ctx context.Context) error { return cache.Close() }, +// ); err != nil { +// log.Fatalf("server: %v", err) +// } +// +// With a separate debug mux (metrics + pprof on a second port): +// +// debugMux := http.NewServeMux() +// httpserver.RegisterDebug(debugMux, promhttp.Handler()) +// go http.ListenAndServe(":8081", debugMux) +package httpserver + +import ( + "context" + "fmt" + "log" + "net/http" + "net/http/pprof" + "os" + "os/signal" + "strconv" + "syscall" + "time" +) + +// ShutdownHook is a function executed after a termination signal is received +// but before the HTTP server begins its graceful shutdown. If a hook returns +// an error it will be logged; shutdown continues regardless. +type ShutdownHook func(ctx context.Context) error + +// RunServerWithShutdown starts the provided *http.Server and blocks until either +// a listener error occurs or a termination signal (SIGINT or SIGTERM) is +// received. It then runs any provided hooks (in order) with a context that +// shares the overall shutdown deadline, and finally gracefully shuts down the +// server. A per-hook timeout can be specified; if zero a default is used. +// +// Returns nil if the server shut down cleanly after a signal. Returns a +// wrapped listener error if ListenAndServe failed (the composition root then +// decides retry vs exit). +// +// Parameters: +// +// server: configured *http.Server (Addr, Handler, timeouts etc.) +// startupLog: message to log when the server starts listening +// shutdownTimeout: total time allowed for all hooks + server shutdown +// hookTimeout: individual timeout for each hook (if <=0 defaults to 5s) +// hooks: optional ordered list of shutdown hooks +func RunServerWithShutdown(server *http.Server, startupLog string, shutdownTimeout, hookTimeout time.Duration, hooks ...ShutdownHook) error { + if hookTimeout <= 0 { + hookTimeout = 5 * time.Second + } + + // serveErr is buffered so the goroutine never blocks. nil means clean stop. + serveErr := make(chan error, 1) + + // Start server in goroutine + go func() { + log.Printf("starting %s on %s", startupLog, server.Addr) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("%s listen error: %v", startupLog, err) + serveErr <- err + return + } + serveErr <- nil + }() + + // Wait for either a signal or a listener error. + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) + + defer signal.Stop(stop) + + var listenerErr error + select { + case <-stop: + log.Printf("shutdown signal received for %s", startupLog) + case err := <-serveErr: + // listener died before any signal — return wrapped error to the caller + // instead of crashing the process from inside this helper. + if err != nil { + listenerErr = fmt.Errorf("%s listen error: %w", startupLog, err) + } + } + + ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) + defer cancel() + + // Run hooks sequentially with individual timeouts + for i, h := range hooks { + if h == nil { + continue + } + hCtx, hCancel := context.WithTimeout(ctx, hookTimeout) + if err := h(hCtx); err != nil { + log.Printf("shutdown hook %d failed: %v", i, err) + } + hCancel() + if err := hCtx.Err(); err == context.DeadlineExceeded { + log.Printf("shutdown hook %d timed out", i) + } + } + + // Finally shut down the HTTP server (idempotent if listener already died). + if err := server.Shutdown(ctx); err != nil { + log.Printf("graceful shutdown failed: %v", err) + } else if listenerErr == nil { + log.Printf("%s shutdown complete", startupLog) + } + + // Drain the serve goroutine result if it raced. + select { + case <-serveErr: + default: + } + + if listenerErr != nil { + return listenerErr + } + return nil +} + +// TimeoutConfig holds server and shutdown related timeouts (all durations). +type TimeoutConfig struct { + ReadHeader time.Duration + Read time.Duration + Write time.Duration + Idle time.Duration + Shutdown time.Duration + Hook time.Duration +} + +// LoadTimeoutConfig reads environment variables (if present) to override defaults. +// Each env var is parsed as an integer number of seconds. If parsing fails or value <=0, +// the provided default is retained. +// Env variables: +// +// READ_HEADER_TIMEOUT +// READ_TIMEOUT +// WRITE_TIMEOUT +// IDLE_TIMEOUT +// SHUTDOWN_TIMEOUT +// HOOK_TIMEOUT +func LoadTimeoutConfig(defaults TimeoutConfig) TimeoutConfig { + apply := func(curr *time.Duration, env string) { + if v := os.Getenv(env); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + *curr = time.Duration(n) * time.Second + } + } + } + apply(&defaults.ReadHeader, "READ_HEADER_TIMEOUT") + apply(&defaults.Read, "READ_TIMEOUT") + apply(&defaults.Write, "WRITE_TIMEOUT") + apply(&defaults.Idle, "IDLE_TIMEOUT") + apply(&defaults.Shutdown, "SHUTDOWN_TIMEOUT") + apply(&defaults.Hook, "HOOK_TIMEOUT") + return defaults +} + +// NewServerWithTimeouts attaches timeout settings to an existing *http.Server or creates a new one if nil. +func NewServerWithTimeouts(base *http.Server, cfg TimeoutConfig) *http.Server { + if base == nil { + return &http.Server{ + ReadHeaderTimeout: cfg.ReadHeader, + ReadTimeout: cfg.Read, + WriteTimeout: cfg.Write, + IdleTimeout: cfg.Idle, + } + } + base.ReadHeaderTimeout = cfg.ReadHeader + base.ReadTimeout = cfg.Read + base.WriteTimeout = cfg.Write + base.IdleTimeout = cfg.Idle + return base +} + +// RegisterProbes registers /healthz and /readyz on the given mux. +// +// /healthz always returns 200 OK (liveness — the process is running). +// /readyz returns 200 if ready() returns true (or if ready is nil), 503 +// otherwise (readiness — the service can handle requests). +// +// Both endpoints are registered with the "GET " method prefix so they work +// with Go 1.22+ pattern-based mux routing. The paths are pod-local ops +// endpoints (never exposed at the edge); register them first so they take +// precedence over catch-all handlers. +func RegisterProbes(mux *http.ServeMux, ready func() bool) { + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, _ *http.Request) { + if ready != nil && !ready() { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("not ready")) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ready")) + }) +} + +// RegisterDebug registers /metrics and /debug/pprof/* on the given mux for a +// separate debug/admin port. The metricsHandler is typically +// prometheus/promhttp.Handler(); it is passed as a parameter so this package +// does not depend on the prometheus client library. The pprof endpoints are +// registered from net/http/pprof. +// +// Use this on a separate mux served on a different port (e.g. :8081) so the +// debug surface is not exposed at the edge: +// +// debugMux := http.NewServeMux() +// httpserver.RegisterDebug(debugMux, promhttp.Handler()) +// go http.ListenAndServe(":8081", debugMux) +func RegisterDebug(mux *http.ServeMux, metricsHandler http.Handler) { + if metricsHandler != nil { + mux.Handle("/metrics", metricsHandler) + } + mux.HandleFunc("/debug/pprof/", pprof.Index) + mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("/debug/pprof/profile", pprof.Profile) + mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("/debug/pprof/trace", pprof.Trace) +}