graphql-monitoring-proxy/proxy.go

package main

import (
	"bytes"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"go.opentelemetry.io/otel/trace"

	"github.com/avast/retry-go/v4"
	"github.com/gofiber/fiber/v3"
	libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
	libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
	libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
	libpack_tracing "github.com/lukaszraczylo/graphql-monitoring-proxy/tracing"
	"github.com/sony/gobreaker"
	"github.com/valyala/fasthttp"
)

// Errors related to circuit breaker
var (
	ErrCircuitOpen = errors.New("circuit breaker is open")
)

// Sentinel errors for the proxy request retry path. Grouped here so callers
// can use errors.Is for comparison instead of brittle string matching.
// Message text MUST match the historical fmt.Errorf strings — tests and
// callers may assert on .Error().
var (
	// errFiberCtxNilDuringRetry — fiber context dropped while retrying.
	errFiberCtxNilDuringRetry = errors.New("fiber context became nil during retry")
	// errFiberRespNil — fiber response object became nil mid-request.
	errFiberRespNil = errors.New("fiber response became nil")
	// errFiberCtxNil — fiber context was nil before the request started.
	errFiberCtxNil = errors.New("fiber context is nil")
)

// Default values for circuit breaker
const (
	defaultMaxRequestsInHalfOpen = 10 // Default maximum requests in half-open state
)

// circuitBreakerCountWindow bounds how long the circuit breaker accumulates
// Counts while CLOSED before clearing them (gobreaker Settings.Interval).
// gobreaker treats Interval<=0 as "never clear Counts during the closed
// state" (see sony/gobreaker Settings docs), which makes
// CIRCUIT_FAILURE_RATIO lifetime-cumulative instead of a recent-window
// ratio: after long uptime a handful of new failures can no longer move a
// ratio diluted by millions of past successes. A positive Interval makes
// gobreaker periodically clear Counts while closed, so the ratio measures a
// rolling window instead.
//
// The window must comfortably exceed the worst-case time a single logical
// request can take, not just typical request latency. gobreaker checks the
// interval lazily, both when a request starts (beforeRequest) and when it
// finishes (afterRequest); if the interval elapses *while* a request is
// in flight, afterRequest rolls Counts over to a new generation before
// recording that request's outcome, and then silently drops it (its
// generation no longer matches the one captured at the start). Each
// logical proxied request here can itself retry up to 7x with exponential
// backoff capped at 10s (performProxyRequestWithEnhancedRetries), i.e.
// ~25.5s worst case (0.5+1+2+4+8+10s) before it finally fails -- so with
// the library default MaxFailures=10, a run of consecutive failing
// requests can legitimately take ~4-5 minutes. A short window (e.g. the
// more typical 60s) reproduces exactly that: it was caught by
// TestCachingAndCircuitBreakerInteraction, where 3 sequential
// retry-exhausting failures (~76s total) spuriously left the circuit
// closed because the last failure's generation had already rolled over
// out from under it. 10 minutes keeps the ratio meaningfully "recent"
// (instead of lifetime-cumulative) while leaving multiples of headroom
// over that worst case. With that much headroom, a single request is very
// unlikely to straddle the window boundary; it is not impossible, since
// gobreaker checks the interval lazily against wall-clock time rather than
// reserving a slot up front. If one still does straddle it, gobreaker drops
// only that one outcome, which delays a trip by at most one request rather
// than breaking correctness.
//
// It is a package var (not const) purely so tests can shrink it to observe
// window-clearing behaviour without sleeping for the real duration.
var circuitBreakerCountWindow = 10 * time.Minute

// Global circuit breaker
var (
	cb      *gobreaker.CircuitBreaker
	cbMutex sync.RWMutex
)

// circuitBackoffState tracks the progressive open-state backoff (finding
// C7) for the global circuit breaker cb. gobreaker v1.0.0's own open-state
// Timeout (cb.timeout) is fixed at construction and has no setter, so it
// cannot be varied per-trip from outside the library. This state instead
// backs a decorator layered IN FRONT of gobreaker -- see
// circuitBackoffGateBlocks and its use in performProxyRequestCore -- that
// never reaches into gobreaker's internals; gobreaker's own Timeout keeps
// governing its internal half-open transition unchanged.
//
// consecutiveTrips counts transitions to StateOpen since the last
// successful recovery (a transition to StateClosed); lastTripUnixNano is
// the UnixNano() of the most recent such transition (0 = never tripped).
// Both are written from createStateChangeFunc's callback, which gobreaker
// invokes synchronously while still holding its own internal mutex (see
// sony/gobreaker CircuitBreaker.setState) -- so that callback must never
// call back into the breaker (e.g. cb.State()), which would deadlock on
// gobreaker's own non-reentrant lock. circuitBackoffGateBlocks reads this
// state on every proxied request. Atomics (not a mutex) keep both
// directions lock-free and sidestep any lock-ordering dependency on
// gobreaker's internal lock.
type circuitBackoffState struct {
	consecutiveTrips atomic.Int64
	lastTripUnixNano atomic.Int64
}

// recordTrip increments the consecutive-trip counter and stamps the current
// time. Called from createStateChangeFunc on transition to StateOpen.
func (s *circuitBackoffState) recordTrip(now time.Time) {
	s.consecutiveTrips.Add(1)
	s.lastTripUnixNano.Store(now.UnixNano())
}

// recordRecovery resets the consecutive-trip counter. Called from
// createStateChangeFunc on transition to StateClosed (a successful
// recovery), so the next trip after a clean close starts progressive
// backoff over again at the base timeout.
func (s *circuitBackoffState) recordRecovery() {
	s.consecutiveTrips.Store(0)
}

// reset clears all state. Called when a new circuit breaker is constructed
// (initCircuitBreaker) so a fresh breaker never inherits a stale trip count
// or timestamp left over from a previous one (e.g. across tests, or a
// config reload).
func (s *circuitBackoffState) reset() {
	s.consecutiveTrips.Store(0)
	s.lastTripUnixNano.Store(0)
}

// trips returns the current consecutive-trip count (0 = never tripped, or
// recovered since the last trip).
func (s *circuitBackoffState) trips() int64 {
	return s.consecutiveTrips.Load()
}

// lastTrip returns the time of the most recent transition to StateOpen, or
// the zero time.Time if the breaker has never tripped.
func (s *circuitBackoffState) lastTrip() time.Time {
	ns := s.lastTripUnixNano.Load()
	if ns == 0 {
		return time.Time{}
	}
	return time.Unix(0, ns)
}

// cbBackoff is the progressive-backoff tracker for the global circuit
// breaker cb -- one breaker, one tracker, mirroring the existing cb/cbMutex
// global pattern.
var cbBackoff circuitBackoffState

// Package-level substring tables used by isConnectionError / isTimeoutError.
// Hoisted to avoid per-call slice allocations on the hot path. All entries
// must be lower-case; callers lower-case the error string once before matching.
var (
	connectionErrorSubstrings = []string{
		"connection refused",
		"connection reset",
		"no route to host",
		"network is unreachable",
		"broken pipe",
		"connection closed",
		"eof",
		"no such host",
		"dial tcp",
		"dial udp",
	}

	timeoutErrorSubstrings = []string{
		"timeout",
		"deadline exceeded",
		"context deadline exceeded",
	}
)

// safeUint32 converts an int to uint32 safely, handling negative values and values exceeding uint32 max
func safeUint32(value int) uint32 {
	// Handle negative values
	if value < 0 {
		return 0
	}

	// Handle values exceeding uint32 max
	if value > math.MaxUint32 {
		return math.MaxUint32
	}

	return uint32(value)
}

// initCircuitBreaker initializes the circuit breaker with configured settings
func initCircuitBreaker(config *config) {
	// Only initialize if enabled
	if !config.CircuitBreaker.Enable {
		config.Logger.Info(&libpack_logger.LogMessage{
			Message: "Circuit breaker is disabled",
		})
		return
	}

	cbMutex.Lock()
	defer cbMutex.Unlock()

	// A freshly constructed breaker must never inherit a stale
	// consecutive-trip count or timestamp from a previous one (see
	// circuitBackoffState.reset doc) -- e.g. across tests, or a config
	// reload that reinitializes cb.
	cbBackoff.reset()

	// Initialize circuit breaker metrics
	InitializeCircuitBreakerMetrics(config.Monitoring)

	// Create circuit breaker settings
	cbSettings := gobreaker.Settings{
		Name:        "graphql-proxy-circuit",
		MaxRequests: safeMaxRequests(config.CircuitBreaker.MaxRequestsInHalfOpen),
		// Periodic reset while closed keeps CIRCUIT_FAILURE_RATIO windowed
		// instead of lifetime-cumulative; see circuitBreakerCountWindow doc.
		Interval:      circuitBreakerCountWindow,
		Timeout:       time.Duration(config.CircuitBreaker.Timeout) * time.Second,
		ReadyToTrip:   createTripFunc(config),
		OnStateChange: createStateChangeFunc(config),
		IsSuccessful:  circuitBreakerIsSuccessful,
	}

	// Initialize the circuit breaker
	cb = gobreaker.NewCircuitBreaker(cbSettings)

	config.Logger.Info(&libpack_logger.LogMessage{
		Message: "Circuit breaker initialized",
		Pairs: map[string]any{
			"max_failures":       config.CircuitBreaker.MaxFailures,
			"timeout_seconds":    config.CircuitBreaker.Timeout,
			"max_half_open_reqs": config.CircuitBreaker.MaxRequestsInHalfOpen,
		},
	})
}

// createTripFunc returns a function that determines when to trip the circuit
func createTripFunc(config *config) func(counts gobreaker.Counts) bool {
	return func(counts gobreaker.Counts) bool {
		// Check consecutive failures first
		if counts.ConsecutiveFailures >= safeUint32(config.CircuitBreaker.MaxFailures) {
			config.Logger.Warning(&libpack_logger.LogMessage{
				Message: "Circuit breaker tripped due to consecutive failures",
				Pairs: map[string]any{
					"consecutive_failures": counts.ConsecutiveFailures,
					"max_failures":         config.CircuitBreaker.MaxFailures,
					"total_requests":       counts.Requests,
				},
			})
			return true
		}

		// Check failure ratio if configured and enough samples
		if config.CircuitBreaker.FailureRatio > 0 &&
			config.CircuitBreaker.SampleSize > 0 &&
			counts.Requests >= safeUint32(config.CircuitBreaker.SampleSize) {
			failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
			if failureRatio >= config.CircuitBreaker.FailureRatio {
				config.Logger.Warning(&libpack_logger.LogMessage{
					Message: "Circuit breaker tripped due to failure ratio",
					Pairs: map[string]any{
						"failure_ratio":  failureRatio,
						"threshold":      config.CircuitBreaker.FailureRatio,
						"total_failures": counts.TotalFailures,
						"total_requests": counts.Requests,
					},
				})
				return true
			}
		}

		return false
	}
}

// circuitBreakerOutcome wraps a proxy failure with a pre-computed verdict on
// whether it should count against the circuit breaker's failure statistics.
// gobreaker.Settings.IsSuccessful only receives the error returned from the
// Execute closure and cannot see the per-request fiber context, so the
// classification (which needs the response status code) is made once, at
// the point where both are available -- see performProxyRequestCore -- and
// travels with the error. Error() and Unwrap() forward to the wrapped
// error, so retry policy, StatusCodeForError, and logging all see the
// original, unwrapped error.
type circuitBreakerOutcome struct {
	err             error
	countsAsFailure bool
}

func (e *circuitBreakerOutcome) Error() string { return e.err.Error() }
func (e *circuitBreakerOutcome) Unwrap() error { return e.err }

// circuitBreakerIsSuccessful is gobreaker's IsSuccessful hook (see
// Settings.IsSuccessful). It treats nil errors as success and defers to the
// pre-computed circuitBreakerOutcome.countsAsFailure verdict for classified
// proxy failures. Any other non-nil error -- one that never went through
// performProxyRequestCore's classification -- is treated as a failure,
// matching gobreaker's own default (defaultIsSuccessful) and pre-fix
// behaviour.
func circuitBreakerIsSuccessful(err error) bool {
	if err == nil {
		return true
	}
	var outcome *circuitBreakerOutcome
	if errors.As(err, &outcome) {
		return !outcome.countsAsFailure
	}
	return false
}

// classifyCircuitFailure decides whether a failed proxy attempt should count
// toward the circuit breaker's failure statistics, gated by
// CIRCUIT_TRIP_ON_TIMEOUTS / CIRCUIT_TRIP_ON_4XX / CIRCUIT_TRIP_ON_5XX.
// statusCode is the last response status observed for this request (0 when
// no HTTP response was ever received, e.g. a connection error).
//
// Every failure that isn't a classified timeout/4xx/5xx (connection errors,
// nil fiber context, etc.) always counts -- those outcomes have no
// dedicated flag and always tripped the breaker before this fix, so that
// baseline is preserved regardless of flag values.
func classifyCircuitFailure(config *config, err error, statusCode int) bool {
	if isTimeoutError(err) {
		return config.CircuitBreaker.TripOnTimeouts
	}

	// doProxyRequestWithTimeout's non-200 error always carries the real,
	// freshly-copied status in c.Response() at the moment it was produced
	// (see the comment in executeProxyAttempt). Guarding on the message
	// first means a stale statusCode left over from an earlier attempt
	// (e.g. this attempt was a connection error, which never copies a
	// response) can't be misread as an HTTP status classification.
	if strings.Contains(err.Error(), "non-200 response") {
		switch {
		case statusCode >= 500 && statusCode < 600:
			return config.CircuitBreaker.TripOn5xx
		case statusCode >= 400 && statusCode < 500:
			return config.CircuitBreaker.TripOn4xx
		}
	}

	return true
}

// createStateChangeFunc returns a function that handles circuit state changes
func createStateChangeFunc(config *config) func(name string, from gobreaker.State, to gobreaker.State) {
	return func(name string, from gobreaker.State, to gobreaker.State) {
		// Progressive backoff bookkeeping (C7). gobreaker invokes
		// OnStateChange synchronously while still holding its own internal
		// mutex, so this must only touch cbBackoff's atomics -- never call
		// back into the breaker (e.g. cb.State()), which would deadlock.
		switch to {
		case gobreaker.StateOpen:
			cbBackoff.recordTrip(time.Now())
		case gobreaker.StateClosed:
			cbBackoff.recordRecovery()
		}

		var stateValue float64
		var stateName string

		switch to {
		case gobreaker.StateOpen:
			stateValue = float64(libpack_monitoring.CircuitOpen)
			stateName = "open"
		case gobreaker.StateHalfOpen:
			stateValue = float64(libpack_monitoring.CircuitHalfOpen)
			stateName = "half-open"
		case gobreaker.StateClosed:
			stateValue = float64(libpack_monitoring.CircuitClosed)
			stateName = "closed"
		}

		// Update metrics using atomic operations to prevent race conditions
		// Use a separate atomic variable to track state instead of recreating gauges
		updateCircuitBreakerState(config, stateValue)

		// Log state change
		config.Logger.Info(&libpack_logger.LogMessage{
			Message: "Circuit breaker state changed",
			Pairs: map[string]any{
				"from": from.String(),
				"to":   to.String(),
				"name": name,
			},
		})

		// Use the new metrics system
		if cbMetrics != nil {
			// Replace hyphens with underscores to avoid validation errors
			safeStateName := strings.ReplaceAll(stateName, "-", "_")
			stateKey := fmt.Sprintf("circuit_state_%s", safeStateName)
			counter := cbMetrics.GetOrCreateFailCounter(config.Monitoring, stateKey)
			counter.Inc()
		}
	}
}

// gobreakerDefaultTimeout mirrors sony/gobreaker's own fallback for
// Settings.Timeout<=0 ("If Timeout is less than or equal to 0, the timeout
// value of the CircuitBreaker is set to 60 seconds" -- gobreaker.go
// NewCircuitBreaker). Duplicated here, since *gobreaker.CircuitBreaker
// exposes no getter for its internal timeout, so circuitBreakerBaseTimeout's
// result always matches what gobreaker itself actually uses as its
// open-state Timeout.
const gobreakerDefaultTimeout = 60 * time.Second

// circuitBreakerBaseTimeout returns the open-state Timeout gobreaker was
// constructed with (see initCircuitBreaker), applying the same <=0
// fallback gobreaker applies internally.
func circuitBreakerBaseTimeout(config *config) time.Duration {
	base := time.Duration(config.CircuitBreaker.Timeout) * time.Second
	if base <= 0 {
		return gobreakerDefaultTimeout
	}
	return base
}

// maxSaneCircuitBackoff hard-caps effectiveCircuitBackoff's result
// independently of the configured maxBackoff. When an operator sets
// CIRCUIT_MAX_BACKOFF_TIMEOUT=0 (uncapped) together with a
// BackoffMultiplier>1, a high enough consecutive-trip count makes
// base*multiplier^(trips-1) overflow math.Pow's float64 arithmetic (or the
// subsequent float64->time.Duration conversion) into a garbage or absurdly
// large Duration. This ceiling makes sure the breaker can never block
// requests for longer than a sane maximum, no matter what maxBackoff is
// configured to.
const maxSaneCircuitBackoff = 24 * time.Hour

// effectiveCircuitBackoff computes the progressive open-state backoff
// duration for the given consecutive-trip count (finding C7):
//
//	effective = base * multiplier^(trips-1), capped at maxBackoff.
//
// trips-1, not trips, so the FIRST trip (trips<=1) always backs off by
// exactly base -- identical to gobreaker's own fixed Timeout for that first
// open period. The multiplier only compounds starting from the SECOND
// consecutive trip. trips<=0 is defensive (a breaker that never tripped
// never reaches this path) and also returns base.
//
// With the default BackoffMultiplier=1.0, effective==base unconditionally
// for every trip count -- an exact fast path below skips the float
// round-trip entirely, so this is byte-identical to pre-C7 behaviour, not
// merely float-equal. A misconfigured multiplier<1 is clamped to 1, and the
// result is never allowed to fall below base or above a maxBackoff that is
// itself (misconfigured) smaller than base -- effective is always >= base,
// so this can only ever ADD closed-door time on top of gobreaker's own
// Timeout, never remove it. See maxSaneCircuitBackoff for the absolute
// ceiling applied regardless of maxBackoff.
func effectiveCircuitBackoff(base time.Duration, multiplier float64, maxBackoff time.Duration, trips int64) time.Duration {
	if base <= 0 || trips <= 1 {
		return base
	}

	if multiplier < 1 {
		multiplier = 1
	}
	if multiplier == 1 {
		return base
	}

	exponent := float64(trips - 1)
	effective := time.Duration(float64(base) * math.Pow(multiplier, exponent))
	if effective < base {
		// Overflow or float rounding pushed us below base; never shorten.
		return base
	}

	// Hard sanity ceiling, applied before the configured cap below and
	// independent of it -- see maxSaneCircuitBackoff.
	if effective > maxSaneCircuitBackoff {
		effective = maxSaneCircuitBackoff
	}

	if maxBackoff > 0 && effective > maxBackoff {
		if maxBackoff < base {
			return base // misconfigured cap below base; never shorten
		}
		return maxBackoff
	}

	return effective
}

// circuitBackoffGateBlocks reports whether the C7 progressive-backoff gate
// should reject a request IN FRONT of gobreaker, without calling
// breaker.Execute -- and therefore without calling the backend or touching
// gobreaker's own Counts/generation.
//
// gobreaker's own Timeout still fully governs its internal half-open
// transition (see sony/gobreaker CircuitBreaker.currentState); this gate
// only adds an extra rejection window on top of it. It blocks exactly when
// both hold:
//   - the progressively-extended effective backoff (computed from the
//     consecutive-trip count) has not yet elapsed since the last trip, and
//   - gobreaker's own (shorter-or-equal) Timeout HAS already elapsed, i.e.
//     breaker.State() has lazily moved off StateOpen and would otherwise
//     let a half-open probe through.
//
// When gobreaker itself is still StateOpen, calling Execute would reject
// with ErrOpenState anyway, so no gate is needed there. When there have
// been no trips yet (trips<=0), this always returns false without touching
// breaker.State() at all, keeping the common (never-tripped) case
// allocation- and lock-free on the hot path. now is a parameter (not
// time.Now() read internally) so tests can drive this deterministically
// without wall-clock sleeps.
func circuitBackoffGateBlocks(config *config, breaker *gobreaker.CircuitBreaker, now time.Time) bool {
	trips := cbBackoff.trips()
	if trips <= 0 {
		return false
	}

	lastTrip := cbBackoff.lastTrip()
	if lastTrip.IsZero() {
		return false
	}

	base := circuitBreakerBaseTimeout(config)
	maxBackoff := time.Duration(config.CircuitBreaker.MaxBackoffTimeout) * time.Second
	effective := effectiveCircuitBackoff(base, config.CircuitBreaker.BackoffMultiplier, maxBackoff, trips)

	if now.Sub(lastTrip) >= effective {
		return false // extended window elapsed; defer to gobreaker as normal
	}

	return breaker.State() != gobreaker.StateOpen
}

// createFasthttpClient creates and configures a fasthttp client with optimized settings.
// The client is configured based on the provided configuration settings, with careful
// attention to performance and security considerations.
func createFasthttpClient(clientConfig *config) *fasthttp.Client {
	tlsConfig := &tls.Config{
		InsecureSkipVerify: clientConfig.Client.DisableTLSVerify,
	}

	// Calculate timeout values, ensuring they're always positive
	clientTimeout := time.Duration(clientConfig.Client.ClientTimeout) * time.Second
	if clientTimeout <= 0 {
		clientTimeout = 30 * time.Second // Default timeout of 30 seconds
	}

	// Honour the per-side read/write timeouts (CLIENT_READ_TIMEOUT /
	// CLIENT_WRITE_TIMEOUT); they fall back to the client timeout when not
	// explicitly configured, so behaviour is unchanged for operators who only
	// set CLIENT_TIMEOUT.
	readTimeout := time.Duration(clientConfig.Client.ReadTimeout) * time.Second
	if readTimeout <= 0 {
		readTimeout = clientTimeout
	}
	writeTimeout := time.Duration(clientConfig.Client.WriteTimeout) * time.Second
	if writeTimeout <= 0 {
		writeTimeout = clientTimeout
	}

	// Create a custom dialer with timeout
	dialer := &fasthttp.TCPDialer{
		Concurrency:      1000,
		DNSCacheDuration: time.Hour,
	}

	client := &fasthttp.Client{
		Name:                     "graphql_proxy",
		NoDefaultUserAgentHeader: true,
		TLSConfig:                tlsConfig,
		// Control connection pool size to prevent overwhelming backend services
		MaxConnsPerHost: clientConfig.Client.MaxConnsPerHost,
		// Configure timeouts to handle different network scenarios
		// Setting all timeout-related parameters to ensure proper timeout behavior
		Dial: func(addr string) (net.Conn, error) {
			return dialer.DialTimeout(addr, clientTimeout)
		},
		ReadTimeout:                   readTimeout,
		WriteTimeout:                  writeTimeout,
		MaxIdleConnDuration:           time.Duration(clientConfig.Client.MaxIdleConnDuration) * time.Second,
		MaxConnDuration:               clientTimeout,
		DisableHeaderNamesNormalizing: false,
		// Performance tuning
		ReadBufferSize:         4096,
		WriteBufferSize:        4096,
		MaxResponseBodySize:    1024 * 1024 * 10, // 10MB max response size
		DisablePathNormalizing: false,
	}

	// Initialize connection pool manager
	InitializeConnectionPool(client)

	return client
}

// proxyTheRequest handles the request proxying logic.
func proxyTheRequest(c fiber.Ctx, currentEndpoint string) error {
	// Record request for RPS tracking
	if rpsTracker := GetRPSTracker(); rpsTracker != nil {
		rpsTracker.RecordRequest()
	}

	// Setup tracing if enabled
	var span trace.Span
	var ctx context.Context

	if cfg.Tracing.Enable && tracer != nil {
		ctx = setupTracing(c)
		span, _ = tracer.StartSpan(ctx, "proxy_request")
		defer span.End()
	}

	// Check if URL is allowed
	if !checkAllowedURLs(c) {
		if ifNotInTest() {
			cfg.Monitoring.Increment(libpack_monitoring.MetricsSkipped, nil)
		}
		return fmt.Errorf("request blocked - not allowed URL: %s", c.Path())
	}

	// Construct and validate proxy URL
	proxyURL := currentEndpoint + c.OriginalURL()
	if _, err := url.Parse(proxyURL); err != nil {
		return fmt.Errorf("invalid URL: %v", err)
	}

	// Log request details in debug mode
	if cfg.LogLevel == "DEBUG" {
		logDebugRequest(c)
	}

	// Perform the proxy request with retries
	if err := performProxyRequest(c, proxyURL); err != nil {
		if ifNotInTest() {
			cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
		}
		return err
	}

	// Log response details in debug mode
	if cfg.LogLevel == "DEBUG" {
		logDebugResponse(c)
	}

	// Handle gzipped responses
	if err := handleGzippedResponse(c); err != nil {
		return err
	}

	// Final status check
	if c.Response().StatusCode() != fiber.StatusOK {
		if ifNotInTest() {
			cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
		}
		return fmt.Errorf("received non-200 response from the GraphQL server: %d", c.Response().StatusCode())
	}

	// Remove server header for security
	c.Response().Header.Del(fiber.HeaderServer)
	return nil
}

// setupTracing extracts and sets up tracing context from request headers
func setupTracing(c fiber.Ctx) context.Context {
	ctx := context.Background()

	if !cfg.Tracing.Enable || tracer == nil {
		return ctx
	}

	// Extract trace information from header
	if traceHeader := c.Get("X-Trace-Span"); traceHeader != "" {
		spanInfo, err := libpack_tracing.ParseTraceHeader(traceHeader)
		if err != nil {
			cfg.Logger.Warning(&libpack_logger.LogMessage{
				Message: "Failed to parse trace header",
				Pairs:   map[string]any{"error": err.Error()},
			})
		} else if spanCtx, err := tracer.ExtractSpanContext(spanInfo); err == nil {
			ctx = trace.ContextWithSpanContext(ctx, spanCtx)
		}
	}

	return ctx
}

// performProxyRequest executes the proxy request with retries, circuit breaker, and request coalescing
func performProxyRequest(c fiber.Ctx, proxyURL string) error {
	// Reuse the cache key already computed by handleCaching (stored on the
	// request context) to avoid a second MD5 of the full request body per
	// proxied request. Fall back to computing it when this function is
	// reached without going through handleCaching (direct callers).
	cacheKey, _ := c.Locals("query_cache_hash").(string)
	if cacheKey == "" {
		userID, userRole := extractUserInfo(c)
		cacheKey = libpack_cache.CalculateHash(c, userID, userRole)
	}

	// Check if request coalescing is enabled
	rc := GetRequestCoalescer()
	if rc != nil && cfg.RequestCoalescing.Enable {
		// Use request coalescing to deduplicate identical concurrent requests
		response, err := rc.Do(cacheKey, func() (*CoalescedResponse, error) {
			// Execute the actual proxy request
			proxyErr := performProxyRequestCore(c, proxyURL, cacheKey)

			// Capture the response for coalescing
			if proxyErr != nil {
				return &CoalescedResponse{
					Err:        proxyErr,
					StatusCode: c.Response().StatusCode(),
				}, proxyErr
			}

			return &CoalescedResponse{
				Body:       c.Response().Body(),
				StatusCode: c.Response().StatusCode(),
				// Headers intentionally left nil; not populated or read anywhere.
			}, nil
		})

		// Check for error from rc.Do (though it typically returns nil)
		if err != nil {
			return err
		}

		// Check for error stored in the response (for coalesced requests)
		if response != nil && response.Err != nil {
			return response.Err
		}

		// For coalesced requests (not the primary), we need to copy the response
		if response != nil && response.Body != nil && len(response.Body) > 0 {
			// Only set response if this is a coalesced request (body would be empty otherwise)
			if len(c.Response().Body()) == 0 {
				c.Response().SetStatusCode(response.StatusCode)
				c.Response().SetBody(response.Body)
			}
		}

		return nil
	}

	// No coalescing - execute directly
	return performProxyRequestCore(c, proxyURL, cacheKey)
}

// performProxyRequestCore executes the proxy request with retries and circuit breaker
// This is the core implementation used by both direct calls and coalesced requests
func performProxyRequestCore(c fiber.Ctx, proxyURL string, cacheKey string) error {
	// If circuit breaker is not enabled, use the original method
	if !cfg.CircuitBreaker.Enable || cb == nil {
		return performProxyRequestWithRetries(c, proxyURL)
	}

	var err error
	if circuitBackoffGateBlocks(cfg, cb, time.Now()) {
		// Progressive backoff (C7): gobreaker's own Timeout has already
		// elapsed -- it would otherwise let a half-open probe through -- but
		// the extended per-trip backoff window has not. Reject exactly like
		// an open breaker, without calling cb.Execute: no backend call, and
		// gobreaker's own Counts/generation are left untouched. gobreaker
		// stays idle in half-open until the extended window also elapses,
		// at which point the Execute call below performs the real probe.
		err = gobreaker.ErrOpenState
	} else {
		// Execute request through circuit breaker
		_, err = cb.Execute(func() (any, error) {
			// Execute the request with retries
			err := performProxyRequestWithRetries(c, proxyURL)
			if err != nil {
				// Log error that could potentially trip the circuit
				cfg.Logger.Warning(&libpack_logger.LogMessage{
					Message: "Error in circuit-protected request",
					Pairs: map[string]any{
						"path":  c.Path(),
						"error": err.Error(),
					},
				})
				// Classify the failure so CIRCUIT_TRIP_ON_TIMEOUTS/_4XX/_5XX gate
				// whether it counts toward the breaker's failure statistics
				// (see circuitBreakerIsSuccessful). The error returned to the
				// caller is unchanged either way -- only the breaker bookkeeping
				// is affected.
				countsAsFailure := classifyCircuitFailure(cfg, err, c.Response().StatusCode())
				if countsAsFailure {
					// Mirrors MetricsCircuitSuccessful below: incremented once per
					// Execute call, only for outcomes that actually count against
					// the breaker's failure statistics, so a single request is
					// never double-counted.
					cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitFailed, nil)
				}
				return nil, &circuitBreakerOutcome{err: err, countsAsFailure: countsAsFailure}
			}

			// Request was successful
			cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitSuccessful, nil)
			return nil, nil
		})
	}

	// If the circuit is open, implement graceful degradation
	if err == gobreaker.ErrOpenState {
		cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitRejected, nil)
		// If cache fallback is disabled, return the original circuit breaker error
		if !cfg.CircuitBreaker.ReturnCachedOnOpen {
			return gobreaker.ErrOpenState
		}
		return handleCircuitOpenGracefulDegradation(c, cacheKey)
	}

	return err
}

// performProxyRequestWithRetries executes the proxy request with retries
// This is the original implementation extracted for reuse
func performProxyRequestWithRetries(c fiber.Ctx, proxyURL string) error {
	// Check backend health first if available
	healthMgr := GetBackendHealthManager()
	if healthMgr != nil && !healthMgr.IsHealthy() {
		// If backend is unhealthy, use more aggressive retry strategy
		return performProxyRequestWithEnhancedRetries(c, proxyURL, true)
	}

	return performProxyRequestWithEnhancedRetries(c, proxyURL, false)
}

// executeProxyAttempt performs a single proxy attempt with error handling
func executeProxyAttempt(c fiber.Ctx, req *fasthttp.Request) error {
	// Additional safety check inside retry loop
	if c == nil {
		return retry.Unrecoverable(errFiberCtxNilDuringRetry)
	}

	// Get connection pool manager for stats tracking
	poolMgr := GetConnectionPoolManager()

	// Execute the proxy request. req was built once by
	// performProxyRequestWithEnhancedRetries and is reused unmodified across
	// every attempt (see the comment there for why that's safe).
	proxyErr := doProxyRequestWithTimeout(c, req, cfg.Client.FastProxyClient)
	if proxyErr != nil {
		// Check if this is a connection error
		if isConnectionError(proxyErr) {
			notifyHealthManager(false)
			// Track connection failure
			if poolMgr != nil {
				poolMgr.RecordConnectionFailure()
			}
			return proxyErr // Connection errors are retryable
		}

		// Check if this is a timeout error - don't retry timeouts
		if isTimeoutError(proxyErr) {
			return retry.Unrecoverable(proxyErr)
		}

		// The server responded with an HTTP status code. Decide retry policy
		// from the status: 5xx and 429 are retryable; other 4xx client
		// errors fail fast instead of burning backoff retries.
		// doProxyRequestWithTimeout already copied the status into c.Response().
		if strings.Contains(proxyErr.Error(), "non-200 response") {
			statusCode := c.Response().StatusCode()
			shouldRetry, _ := isRetryableStatusCode(statusCode)
			if !shouldRetry {
				return retry.Unrecoverable(proxyErr)
			}
			// Track as a failure for retryable HTTP errors (5xx, 429)
			if poolMgr != nil {
				poolMgr.RecordConnectionFailure()
			}
		}
		return proxyErr
	}

	// Safety check before accessing response (c is already validated at function entry)
	if c.Response() == nil {
		return retry.Unrecoverable(errFiberRespNil)
	}

	// Check status code and determine retry strategy
	statusCode := c.Response().StatusCode()
	shouldRetry, err := isRetryableStatusCode(statusCode)

	if err == nil {
		// Success case
		notifyHealthManager(true)
		// Track successful connection
		if poolMgr != nil {
			poolMgr.RecordConnectionSuccess()
		}
		return nil
	}

	if shouldRetry {
		// Track connection failure for retryable errors (5xx, etc)
		if poolMgr != nil {
			poolMgr.RecordConnectionFailure()
		}
		return err // Retryable error
	}

	return err // Non-retryable error (already wrapped with retry.Unrecoverable)
}

// performProxyRequestWithEnhancedRetries executes the proxy request with intelligent retry strategy
func performProxyRequestWithEnhancedRetries(c fiber.Ctx, proxyURL string, backendUnhealthy bool) error {
	// Safety check for nil context
	if c == nil {
		return errFiberCtxNil
	}

	// Build the outbound request once, before the retry loop, instead of
	// re-copying c.Request() (headers + body) on every attempt. A retry
	// replays the exact same request, so the copy-in is invariant across
	// attempts; only the response is fresh per attempt (doProxyRequestWithTimeout
	// acquires/releases its own *fasthttp.Response each call). This previously
	// ran c.Request().CopyTo(req) up to 7x per proxied request.
	req := fasthttp.AcquireRequest()
	defer fasthttp.ReleaseRequest(req)
	c.Request().CopyTo(req)
	req.SetRequestURI(proxyURL)

	var attempts uint
	var initialDelay time.Duration
	var maxDelayTime time.Duration

	if backendUnhealthy {
		// Backend is known to be unhealthy, fail fast
		// Circuit breaker should handle this, so reduce retries
		attempts = 3
		initialDelay = 500 * time.Millisecond
		maxDelayTime = 5 * time.Second
	} else {
		// Normal retry strategy
		attempts = 7
		initialDelay = 500 * time.Millisecond
		maxDelayTime = 10 * time.Second
	}

	return retry.Do(
		func() error {
			return executeProxyAttempt(c, req)
		},
		retry.Attempts(attempts),
		retry.DelayType(retry.BackOffDelay),
		retry.Delay(initialDelay),
		retry.MaxDelay(maxDelayTime),
		retry.OnRetry(func(n uint, err error) {
			cfg.Logger.Warning(&libpack_logger.LogMessage{
				Message: "Retrying the request",
				Pairs: map[string]any{
					"path":              c.Path(),
					"attempt":           n + 1,
					"max_attempts":      attempts,
					"error":             err.Error(),
					"error_type":        fmt.Sprintf("%T", err),
					"is_timeout":        strings.Contains(strings.ToLower(err.Error()), "timeout"),
					"is_connection":     isConnectionError(err),
					"backend_unhealthy": backendUnhealthy,
				},
			})
		}),
		retry.LastErrorOnly(true),
		retry.RetryIf(func(err error) bool {
			// Unrecoverable errors (non-retryable 4xx, timeouts) must not be
			// retried. retry-go's finite-attempt loop only consults RetryIf,
			// so without this the default IsRecoverable gate is lost.
			if !retry.IsRecoverable(err) {
				return false
			}
			// Don't retry if context is cancelled or context is nil
			if c == nil {
				return false
			}

			// Safely check if context is done/cancelled
			// Note: fasthttp.RequestCtx.Done() can panic if not properly initialized
			// If we panic, don't retry (maintains backward compatibility with test behavior)
			shouldRetry := true
			func() {
				defer func() {
					if r := recover(); r != nil {
						// If we panic accessing context, don't retry
						// This typically happens in test scenarios with mock contexts
						shouldRetry = false
					}
				}()
				ctx := c.Context()
				if ctx == nil {
					return
				}
				select {
				case <-ctx.Done():
					shouldRetry = false
				default:
				}
			}()

			if !shouldRetry {
				return false
			}

			// Check retry budget before allowing retry
			if rb := GetRetryBudget(); rb != nil {
				if !rb.AllowRetry() {
					cfg.Logger.Warning(&libpack_logger.LogMessage{
						Message: "Retry denied by budget",
						Pairs: map[string]any{
							"path":  c.Path(),
							"error": err.Error(),
						},
					})
					return false
				}
			}
			return true
		}),
	)
}

// isConnectionError checks if the error is a connection-related error
func isConnectionError(err error) bool {
	if err == nil {
		return false
	}

	errStr := strings.ToLower(err.Error())
	for _, connErr := range connectionErrorSubstrings {
		if strings.Contains(errStr, connErr) {
			return true
		}
	}

	return false
}

// isTimeoutError checks if the error is a timeout-related error
func isTimeoutError(err error) bool {
	if err == nil {
		return false
	}
	errStr := strings.ToLower(err.Error())
	for _, tErr := range timeoutErrorSubstrings {
		if strings.Contains(errStr, tErr) {
			return true
		}
	}
	return false
}

// isRetryableStatusCode determines if an HTTP status code should trigger a retry
func isRetryableStatusCode(statusCode int) (bool, error) {
	// Don't retry client errors (4xx) except for specific cases
	if statusCode >= 400 && statusCode < 500 {
		// Retry on 429 (rate limit) and 503 (service unavailable - misclassified as 4xx)
		if statusCode == 429 || statusCode == 503 {
			return true, fmt.Errorf("retryable status code: %d", statusCode)
		}
		// Other 4xx errors are not retryable
		return false, retry.Unrecoverable(fmt.Errorf("client error: %d", statusCode))
	}

	// Retry on 5xx errors
	if statusCode >= 500 {
		return true, fmt.Errorf("server error: %d", statusCode)
	}

	// Success for 2xx and 3xx
	if statusCode >= 200 && statusCode < 400 {
		return false, nil // No error, no retry needed
	}

	return true, fmt.Errorf("unexpected status code: %d", statusCode)
}

// notifyHealthManager notifies the backend health manager of request success or failure
func notifyHealthManager(success bool) {
	if healthMgr := GetBackendHealthManager(); healthMgr != nil {
		healthMgr.updateHealthStatus(success)
	}
}

// handleCircuitOpenGracefulDegradation handles requests when the circuit breaker is open
func handleCircuitOpenGracefulDegradation(c fiber.Ctx, cacheKey string) error {
	// Try to serve from cache if configured and available
	if cfg.CircuitBreaker.ReturnCachedOnOpen {
		if cachedResponse := libpack_cache.CacheLookup(cacheKey); cachedResponse != nil {
			cfg.Logger.Info(&libpack_logger.LogMessage{
				Message: "Circuit open - serving from cache",
				Pairs: map[string]any{
					"path": c.Path(),
				},
			})

			// Set response from cache
			c.Response().SetBody(cachedResponse)
			c.Response().SetStatusCode(fiber.StatusOK)

			// Mark as cache hit since we're serving from cache
			cfg.Monitoring.Increment(libpack_monitoring.MetricsCacheHit, nil)
			cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitFallbackSuccess, nil)

			return nil
		}
	}

	// No cached response available - provide helpful error response
	cfg.Logger.Warning(&libpack_logger.LogMessage{
		Message: "Circuit open - no cached response available",
		Pairs: map[string]any{
			"path": c.Path(),
		},
	})

	cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitFallbackFailed, nil)

	return ErrCircuitOpen
}

// doProxyRequestWithTimeout performs a proxy request with proper timeout
// handling. req is the already-built outbound request (see
// performProxyRequestWithEnhancedRetries); only the response is
// acquired/released per call, so each retry attempt gets fresh
// per-attempt response state while reusing the same request.
func doProxyRequestWithTimeout(c fiber.Ctx, req *fasthttp.Request, client *fasthttp.Client) error {
	// Calculate timeout from client configuration
	clientTimeout := time.Duration(cfg.Client.ClientTimeout) * time.Second
	if clientTimeout <= 0 {
		clientTimeout = 30 * time.Second
	}

	// Acquire a fresh response object for this attempt
	resp := fasthttp.AcquireResponse()
	defer fasthttp.ReleaseResponse(resp)

	// Perform the request with timeout
	err := client.DoTimeout(req, resp, clientTimeout)
	if err != nil {
		return err
	}

	// Copy response back to fiber context
	resp.CopyTo(c.Response())

	// Check for non-200 responses and return error for tests
	if c.Response().StatusCode() != fiber.StatusOK {
		return fmt.Errorf("received non-200 response: %d", c.Response().StatusCode())
	}

	return nil
}

// handleGzippedResponse decompresses gzipped responses
func handleGzippedResponse(c fiber.Ctx) error {
	if !bytes.EqualFold(c.Response().Header.Peek("Content-Encoding"), []byte("gzip")) {
		return nil
	}

	// Use pooled gzip reader
	reader, err := GetGzipReader(bytes.NewReader(c.Response().Body()))
	if err != nil {
		cfg.Logger.Error(&libpack_logger.LogMessage{
			Message: "Failed to create gzip reader",
			Pairs:   map[string]any{"error": err.Error()},
		})
		return err
	}
	defer func() {
		// Return reader to pool
		PutGzipReader(reader)
	}()

	// Use pooled buffer for reading
	buf := GetHTTPBuffer()
	defer PutHTTPBuffer(buf)

	// Read decompressed data into pooled buffer
	_, err = io.Copy(buf, reader)
	if err != nil {
		cfg.Logger.Error(&libpack_logger.LogMessage{
			Message: "Failed to decompress response",
			Pairs:   map[string]any{"error": err.Error()},
		})
		return err
	}

	// Get decompressed data
	decompressed := buf.Bytes()

	// Update response
	c.Response().SetBody(decompressed)
	c.Response().Header.Del("Content-Encoding")
	return nil
}

// logDebugRequest logs the request details when in debug mode with sanitization.
func logDebugRequest(c fiber.Ctx) {
	contentType := string(c.Request().Header.ContentType())
	sanitizedBody := sanitizeForLogging(c.Body(), contentType)
	sanitizedHeaders := sanitizeHeaders(convertHeaders(c.GetReqHeaders()))

	cfg.Logger.Debug(&libpack_logger.LogMessage{
		Message: "Proxying the request",
		Pairs: map[string]any{
			"path":         c.Path(),
			"body":         sanitizedBody,
			"headers":      sanitizedHeaders,
			"request_uuid": c.Locals("request_uuid"),
		},
	})
}

// logDebugResponse logs the response details when in debug mode with sanitization.
func logDebugResponse(c fiber.Ctx) {
	contentType := string(c.Response().Header.ContentType())
	sanitizedBody := sanitizeForLogging(c.Response().Body(), contentType)
	sanitizedHeaders := sanitizeHeaders(convertHeaders(c.GetRespHeaders()))

	cfg.Logger.Debug(&libpack_logger.LogMessage{
		Message: "Received proxied response",
		Pairs: map[string]any{
			"path":          c.Path(),
			"response_body": sanitizedBody,
			"response_code": c.Response().StatusCode(),
			"headers":       sanitizedHeaders,
			"request_uuid":  c.Locals("request_uuid"),
		},
	})
}

// safeMaxRequests converts MaxRequestsInHalfOpen safely to uint32, providing a fallback value if out of bounds
func safeMaxRequests(maxRequestsInHalfOpen int) uint32 {
	// Check if value is invalid (negative or too large)
	if maxRequestsInHalfOpen < 0 || maxRequestsInHalfOpen > math.MaxUint32 {
		// Log warning and return a default value
		if cfg != nil && cfg.Logger != nil {
			cfg.Logger.Warning(&libpack_logger.LogMessage{
				Message: "Invalid MaxRequestsInHalfOpen value, using default",
				Pairs: map[string]any{
					"requested_value": maxRequestsInHalfOpen,
					"default_value":   defaultMaxRequestsInHalfOpen,
				},
			})
		}
		return uint32(defaultMaxRequestsInHalfOpen)
	}

	return uint32(maxRequestsInHalfOpen)
}

// updateCircuitBreakerState safely updates the circuit breaker state using atomic operations
func updateCircuitBreakerState(config *config, stateValue float64) {
	// Update the state atomically using the new metrics system
	if cbMetrics != nil {
		cbMetrics.UpdateState(stateValue)
	}
}