En/rate limiter

docs/advanced-guide/http-communication/page.md

@@ -95,10 +95,24 @@ GoFr provides its user with additional configurational options while registering
 - **DefaultHeaders** - This option allows user to set some default headers that will be propagated to the downstream HTTP Service every time it is being called.
 - **HealthConfig** - This option allows user to add the `HealthEndpoint` along with `Timeout` to enable and perform the timely health checks for downstream HTTP Service.
 - **RetryConfig** - This option allows user to add the maximum number of retry count if before returning error if any downstream HTTP Service fails.
+- **RateLimiterConfig** -  This option allows user to configure rate limiting for downstream service calls using token bucket algorithm. It controls the request rate to prevent overwhelming dependent services and supports both in-memory and Redis-based implementations.
+
+**Rate Limiter Store: Customization**
+GoFr allows you to use a custom rate limiter store by implementing the RateLimiterStore interface.This enables integration with any backend (e.g., Redis, database, or custom logic)
+Interface:
+```go
+type RateLimiterStore interface {
+Allow(ctx context.Context, key string, config RateLimiterConfig) (allowed bool, retryAfter time.Duration, err error)
+StartCleanup(ctx context.Context)
+StopCleanup()
+}
+```
 
 #### Usage:
 
 ```go
+rc := redis.NewClient(a.Config, a.Logger(), a.Metrics())
+
 a.AddHTTPService("cat-facts", "https://catfact.ninja",
 	service.NewAPIKeyConfig("some-random-key"),
 	service.NewBasicAuthConfig("username", "password"),
@@ -119,5 +133,12 @@ a.AddHTTPService("cat-facts", "https://catfact.ninja",
   &service.RetryConfig{
       MaxRetries: 5
   },
+
+  &service.RateLimiterConfig{
+	  Requests: 5,
+	  Window:   time.Minute,
+      Burst:     10,
+	  Store:    service.NewRedisRateLimiterStore(rc)}, // Skip this field to use in-memory store
+    },
 )
 ```

pkg/gofr/service/oauth.go

@@ -79,7 +79,7 @@ func validateTokenURL(tokenURL string) error {
 		return AuthErr{nil, "invalid host pattern, contains `..`"}
 	case strings.HasSuffix(u.Host, "."):
 		return AuthErr{nil, "invalid host pattern, ends with `.`"}
-	case u.Scheme != "http" && u.Scheme != "https":
+	case u.Scheme != methodHTTP && u.Scheme != methodHTTPS:
 		return AuthErr{nil, "invalid scheme, allowed http and https only"}
 	default:
 		return nil

pkg/gofr/service/rate_limiter.go

@@ -0,0 +1,198 @@
+package service
+
+import (
+	"context"
+	"net/http"
+	"strings"
+)
+
+// rateLimiter provides unified rate limiting for HTTP clients.
+type rateLimiter struct {
+	config RateLimiterConfig
+	store  RateLimiterStore
+	HTTP   // Embedded HTTP service
+}
+
+// NewRateLimiter creates a new unified rate limiter.
+func NewRateLimiter(config RateLimiterConfig, h HTTP) HTTP {
+	rl := &rateLimiter{
+		config: config,
+		store:  config.Store,
+		HTTP:   h,
+	}
+
+	// Start cleanup routine
+	ctx := context.Background()
+	rl.store.StartCleanup(ctx)
+
+	return rl
+}
+
+// buildFullURL constructs an absolute URL by combining the base service URL with the given path.
+func (rl *rateLimiter) buildFullURL(path string) string {
+	if strings.HasPrefix(path, "http://") || strings.HasPrefix(path, "https://") {
+		return path
+	}
+
+	// Get base URL from embedded HTTP service
+	httpSvcImpl, ok := rl.HTTP.(*httpService)
+	if !ok {
+		return path
+	}
+
+	base := strings.TrimRight(httpSvcImpl.url, "/")
+	if base == "" {
+		return path
+	}
+
+	// Ensure path starts with /
+	if !strings.HasPrefix(path, "/") {
+		path = "/" + path
+	}
+
+	return base + path
+}
+
+// checkRateLimit performs rate limit check using the configured store.
+func (rl *rateLimiter) checkRateLimit(req *http.Request) error {
+	serviceKey := rl.config.KeyFunc(req)
+
+	allowed, retryAfter, err := rl.store.Allow(req.Context(), serviceKey, rl.config)
+	if err != nil {
+		return nil // Fail open
+	}
+
+	if !allowed {
+		return &RateLimitError{ServiceKey: serviceKey, RetryAfter: retryAfter}
+	}
+
+	return nil
+}
+
+// Get performs rate-limited HTTP GET request.
+func (rl *rateLimiter) Get(ctx context.Context, path string, queryParams map[string]any) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.Get(ctx, path, queryParams)
+}
+
+// GetWithHeaders performs rate-limited HTTP GET request with custom headers.
+func (rl *rateLimiter) GetWithHeaders(ctx context.Context, path string, queryParams map[string]any,
+	headers map[string]string) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.GetWithHeaders(ctx, path, queryParams, headers)
+}
+
+// Post performs rate-limited HTTP POST request.
+func (rl *rateLimiter) Post(ctx context.Context, path string, queryParams map[string]any,
+	body []byte) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.Post(ctx, path, queryParams, body)
+}
+
+// PostWithHeaders performs rate-limited HTTP POST request with custom headers.
+func (rl *rateLimiter) PostWithHeaders(ctx context.Context, path string, queryParams map[string]any,
+	body []byte, headers map[string]string) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.PostWithHeaders(ctx, path, queryParams, body, headers)
+}
+
+// Put performs rate-limited HTTP PUT request.
+func (rl *rateLimiter) Put(ctx context.Context, path string, queryParams map[string]any,
+	body []byte) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPut, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.Put(ctx, path, queryParams, body)
+}
+
+// PutWithHeaders performs rate-limited HTTP PUT request with custom headers.
+func (rl *rateLimiter) PutWithHeaders(ctx context.Context, path string, queryParams map[string]any, body []byte,
+	headers map[string]string) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPut, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.PutWithHeaders(ctx, path, queryParams, body, headers)
+}
+
+// Patch performs rate-limited HTTP PATCH request.
+func (rl *rateLimiter) Patch(ctx context.Context, path string, queryParams map[string]any,
+	body []byte) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPatch, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.Patch(ctx, path, queryParams, body)
+}
+
+// PatchWithHeaders performs rate-limited HTTP PATCH request with custom headers.
+func (rl *rateLimiter) PatchWithHeaders(ctx context.Context, path string, queryParams map[string]any,
+	body []byte, headers map[string]string) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPatch, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.PatchWithHeaders(ctx, path, queryParams, body, headers)
+}
+
+// Delete performs rate-limited HTTP DELETE request.
+func (rl *rateLimiter) Delete(ctx context.Context, path string, body []byte) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodDelete, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.Delete(ctx, path, body)
+}
+
+// DeleteWithHeaders performs rate-limited HTTP DELETE request with custom headers.
+func (rl *rateLimiter) DeleteWithHeaders(ctx context.Context, path string, body []byte,
+	headers map[string]string) (*http.Response, error) {
+	fullURL := rl.buildFullURL(path)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodDelete, fullURL, http.NoBody)
+
+	if err := rl.checkRateLimit(req); err != nil {
+		return nil, err
+	}
+
+	return rl.HTTP.DeleteWithHeaders(ctx, path, body, headers)
+}

pkg/gofr/service/rate_limiter_config.go

@@ -0,0 +1,127 @@
+package service
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+)
+
+var (
+	errInvalidRequestRate     = errors.New("requests must be greater than 0 per configured time window")
+	errBurstLessThanRequests  = errors.New("burst must be greater than requests per window")
+	errInvalidRedisResultType = errors.New("unexpected Redis result type")
+)
+
+const (
+	unknownServiceKey = "unknown"
+	methodHTTP        = "http"
+	methodHTTPS       = "https"
+)
+
+// RateLimiterConfig with custom keying support.
+type RateLimiterConfig struct {
+	Requests float64                    // Number of requests allowed
+	Window   time.Duration              // Time window (e.g., time.Minute, time.Hour)
+	Burst    int                        // Maximum burst capacity (must be > 0)
+	KeyFunc  func(*http.Request) string // Optional custom key extraction
+	Store    RateLimiterStore
+}
+
+// defaultKeyFunc extracts a normalized service key from an HTTP request.
+func defaultKeyFunc(req *http.Request) string {
+	if req == nil || req.URL == nil {
+		return unknownServiceKey
+	}
+
+	scheme := req.URL.Scheme
+	host := req.URL.Host
+
+	if scheme == "" {
+		if req.TLS != nil {
+			scheme = methodHTTPS
+		} else {
+			scheme = methodHTTP
+		}
+	}
+
+	if host == "" {
+		host = req.Host
+	}
+
+	if host == "" {
+		host = unknownServiceKey
+	}
+
+	return scheme + "://" + host
+}
+
+// Validate checks if the configuration is valid.
+func (config *RateLimiterConfig) Validate() error {
+	if config.Requests <= 0 {
+		return fmt.Errorf("%w: %f", errInvalidRequestRate, config.Requests)
+	}
+
+	if config.Window <= 0 {
+		config.Window = time.Minute // Default: per-minute rate limiting
+	}
+
+	if config.Burst <= 0 {
+		config.Burst = int(config.Requests)
+	}
+
+	if float64(config.Burst) < config.Requests {
+		return fmt.Errorf("%w: burst=%d, requests=%f", errBurstLessThanRequests, config.Burst, config.Requests)
+	}
+
+	// Set default key function if not provided.
+	if config.KeyFunc == nil {
+		config.KeyFunc = defaultKeyFunc
+	}
+
+	return nil
+}
+
+// AddOption implements the Options interface.
+func (config *RateLimiterConfig) AddOption(h HTTP) HTTP {
+	if err := config.Validate(); err != nil {
+		if httpSvc, ok := h.(*httpService); ok {
+			httpSvc.Logger.Log("Invalid rate limiter config, disabling rate limiting", "error", err)
+		}
+
+		return h
+	}
+
+	// Default to local store if not set
+	if config.Store == nil {
+		config.Store = NewLocalRateLimiterStore()
+
+		// Log warning for local rate limiting.
+		if httpSvc, ok := h.(*httpService); ok {
+			httpSvc.Logger.Log("Using local rate limiting - not suitable for multi-instance deployments")
+		}
+	}
+
+	return NewRateLimiter(*config, h)
+}
+
+// RequestsPerSecond converts the configured rate to requests per second.
+func (config *RateLimiterConfig) RequestsPerSecond() float64 {
+	// Convert any time window to "requests per second" for internal math
+	return float64(config.Requests) / config.Window.Seconds()
+}
+
+// RateLimitError represents a rate limiting error.
+type RateLimitError struct {
+	ServiceKey string
+	RetryAfter time.Duration
+}
+
+func (e *RateLimitError) Error() string {
+	return fmt.Sprintf("rate limit exceeded for service: %s, retry after: %v", e.ServiceKey, e.RetryAfter)
+}
+
+// StatusCode Implement StatusCodeResponder so Responder picks correct HTTP code.
+func (*RateLimitError) StatusCode() int {
+	return http.StatusTooManyRequests // 429
+}