trip-planner/internal/yandex/client.go

package yandex

import (
	"context"
	"encoding/json"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"trip-planner/internal/metrics"
)

// Client represents a Yandex Schedules API client with rate limiting,
// circuit breaking, and retry capabilities.
type Client struct {
	apiKey         string
	httpClient     *http.Client
	rateLimiter    *tokenBucket
	circuitBreaker *circuitBreaker
	retryConfig    *retryConfig
	metrics        *metrics.Metrics
}

// tokenBucket implements a token bucket rate limiter.
type tokenBucket struct {
	mu           sync.Mutex
	capacity     int
	tokens       int
	refillPerSec int // tokens to add per second
	lastRefill   time.Time
}

// circuitBreaker implements the circuit breaker pattern with states:
// closed (normal operation), open (failing), half-open (testing).
type circuitBreaker struct {
	mu            sync.Mutex
	state         state
	failures      int
	successes     int
	openSince     time.Time
	timeout       time.Duration
	failThreshold int // number of failures to open the circuit
}

type state int

const (
	closed state = iota
	open
	halfOpen
)

// retryConfig holds configuration for retry behavior.
type retryConfig struct {
	maxRetries  int
	baseBackoff time.Duration
	maxBackoff  time.Duration
	jitter      bool
}

// NewClient creates a new Yandex API client with the given API key and options.
func NewClient(apiKey string, options ...Option) *Client {
	c := &Client{
		apiKey: apiKey,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
		},
		rateLimiter:    newTokenBucket(10, 1), // default: 1 TPS, capacity 10
		circuitBreaker: newCircuitBreaker(),
		retryConfig: &retryConfig{
			maxRetries:  3,
			baseBackoff: 100 * time.Millisecond,
			maxBackoff:  5 * time.Second,
			jitter:      true,
		},
	}

	for _, opt := range options {
		opt(c)
	}

	return c
}

// Option configures a Yandex Client.
type Option func(*Client)

// WithRateLimiter sets a custom rate limiter (tokens per period).
func WithRateLimiter(capacity, perSeconds int) Option {
	return func(c *Client) {
		c.rateLimiter = newTokenBucket(capacity, perSeconds)
	}
}

// WithCircuitBreakerTimeout sets the circuit breaker open timeout.
func WithCircuitBreakerTimeout(timeout time.Duration) Option {
	return func(c *Client) {
		c.circuitBreaker.timeout = timeout
	}
}

// WithRetryConfig sets custom retry configuration.
func WithRetryConfig(maxRetries int, baseBackoff, maxBackoff time.Duration, jitter bool) Option {
	return func(c *Client) {
		c.retryConfig = &retryConfig{
			maxRetries:  maxRetries,
			baseBackoff: baseBackoff,
			maxBackoff:  maxBackoff,
			jitter:      jitter,
		}
	}
}

// WithMetrics sets the metrics recorder for the client.
func WithMetrics(m *metrics.Metrics) Option {
	return func(c *Client) {
		c.metrics = m
	}
}

// Do executes a Yandex API request with rate limiting, circuit breaking, and retry.
func (c *Client) Do(ctx context.Context, method, path string, query map[string]string) (*Response, error) {
	// Apply rate limiting
	if err := c.rateLimiter.acquire(); err != nil {
		return nil, fmt.Errorf("rate limit exceeded: %w", err)
	}

	// Build request URL
	url := buildURL(path, query)

	var resp *Response
	var err error

	// Execute with retry, checking circuit breaker on each attempt
	for attempt := 0; attempt <= c.retryConfig.maxRetries; attempt++ {
		// Check circuit breaker on each retry attempt
		if !c.circuitBreaker.allow() {
			c.metrics.RecordCircuitBreakerTrip()
			return nil, fmt.Errorf("circuit breaker is open")
		}

		resp, err = c.executeRequest(ctx, url)
		if err == nil {
			c.circuitBreaker.recordSuccess()
			return resp, nil
		}

		// Check if error is retryable
		if !isRetryableError(err) {
			c.circuitBreaker.recordFailure()
			return nil, err
		}

		c.circuitBreaker.recordFailure()

		if attempt < c.retryConfig.maxRetries {
			backoff := c.retryConfig.baseBackoff
			if c.retryConfig.jitter {
				backoff = applyJitter(backoff)
			}
			time.Sleep(backoff)
		}
	}

	c.circuitBreaker.recordFailure() // final failure
	c.metrics.RecordCircuitBreakerTrip()
	return nil, err
}

// executeRequest performs a single HTTP request to the Yandex API.
func (c *Client) executeRequest(ctx context.Context, url string) (*Response, error) {
	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return nil, fmt.Errorf("failed to create request: %w", err)
	}

	req.Header.Set("Accept", "application/json")

	// Add API key
	if c.apiKey != "" {
		req.Header.Set("apikey", c.apiKey)
	}

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("request failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 400 {
		return nil, newAPIError(resp.StatusCode, resp.Status)
	}

	var body Response
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return nil, fmt.Errorf("failed to decode response: %w", err)
	}

	return &body, nil
}

// Response represents a Yandex API response.
type Response struct {
	Pagination Pagination `json:"pagination"`
	Search     Search     `json:"search"`
	Intervals  []Segment  `json:"interval_segments"`
	Segments   []Segment  `json:"segments"`
}

// Pagination represents API pagination metadata.
type Pagination struct {
	Total  int `json:"total"`
	Limit  int `json:"limit"`
	Offset int `json:"offset"`
}

// Search represents search metadata.
type Search struct {
	Date string `json:"date"`
	From City   `json:"from"`
	To   City   `json:"to"`
}

// City represents a city or station in the API response.
type City struct {
	Code         string `json:"code"`
	Type         string `json:"type"`
	Title        string `json:"title"`
	ShortTitle   string `json:"short_title"`
	PopularTitle string `json:"popular_title"`
}

// Segment represents a single route segment.
type Segment struct {
	Departure    string  `json:"departure"`
	Arrival      string  `json:"arrival"`
	Duration     int     `json:"duration"`
	HasTransfers bool    `json:"has_transfers"`
	From         Station `json:"from"`
	To           Station `json:"to"`
}

// Station represents a station in the API response.
type Station struct {
	Code  string `json:"code"`
	Title string `json:"title"`
	// Other fields can be added as needed
}

// APIError represents a Yandex API error.
type APIError struct {
	Code    int
	Message string
}

func (e *APIError) Error() string {
	return fmt.Sprintf("API error %d: %s", e.Code, e.Message)
}

func newAPIError(code int, message string) *APIError {
	return &APIError{Code: code, Message: message}
}

// isRetryableError checks if an error is retryable (transient/network error).
func isRetryableError(err error) bool {
	if err == nil {
		return false
	}
	// Check for HTTP status codes that are retryable (5xx errors)
	apiErr, ok := err.(*APIError)
	if ok {
		return apiErr.Code >= 500 && apiErr.Code < 600
	}
	// Check for network errors
	errStr := err.Error()
	return strings.Contains(errStr, "timeout") ||
		strings.Contains(errStr, "connection refused") ||
		strings.Contains(errStr, "dial tcp") ||
		strings.Contains(errStr, "context deadline exceeded")
}

// buildURL constructs a Yandex API URL with query parameters.
func buildURL(path string, query map[string]string) string {
	u := fmt.Sprintf("https://api.rasp.yandex.net%s", path)
	params := url.Values{}
	for k, v := range query {
		params.Set(k, v)
	}
	u += "?" + params.Encode()
	return u
}

// --- Token Bucket Rate Limitter ---

func newTokenBucket(capacity, perSeconds int) *tokenBucket {
	return &tokenBucket{
		capacity:     capacity,
		tokens:       capacity,
		refillPerSec: perSeconds,
		lastRefill:   time.Now(),
	}
}

func (tb *tokenBucket) acquire() error {
	tb.mu.Lock()
	defer tb.mu.Unlock()

	now := time.Now()
	tb.refill(now)

	if tb.tokens > 0 {
		tb.tokens--
		return nil
	}

	return fmt.Errorf("rate limit: rate exceeded (%.1f TPS configured)", float64(tb.refillPerSec)/float64(time.Second))
}

func (tb *tokenBucket) refill(now time.Time) {
	elapsed := now.Sub(tb.lastRefill)
	if elapsed >= time.Second {
		// Refill tokens based on elapsed time and rate
		tb.tokens = tb.capacity
		tb.lastRefill = now
	}
	// else: keep current tokens, will fully refill on next second boundary
}

// --- Circuit Breaker ---

func newCircuitBreaker() *circuitBreaker {
	return &circuitBreaker{
		state:         closed,
		timeout:       30 * time.Second,
		failThreshold: 3,
	}
}


func (cb *circuitBreaker) ResetCircuitBreaker() {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.state = closed
	cb.failures = 0
	cb.successes = 0
	cb.openSince = time.Time{}
}

func (cb *circuitBreaker) allow() bool {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case closed:
		return true
	case open:
		// Check if timeout has elapsed
		if time.Since(cb.openSince) >= cb.timeout {
			cb.state = halfOpen
			cb.successes = 0
			return true
		}
		return false
	case halfOpen:
		return true
	}
	return false
}

func (cb *circuitBreaker) recordSuccess() {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case closed:
		// Nothing to do
	case halfOpen:
		cb.successes++
		if cb.successes >= 3 {
			cb.state = closed
			cb.failures = 0
		}
	case open:
		// Should not happen (allow would have transitioned)
	}
}

func (cb *circuitBreaker) recordFailure() {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case closed:
		cb.failures++
		if cb.failures >= cb.failThreshold {
			cb.state = open
			cb.openSince = time.Now()
		}
	case halfOpen:
		cb.state = open
		cb.openSince = time.Now()
	case open:
		// Stay open
	}
}

// --- Retry helpers ---

func applyJitter(backoff time.Duration) time.Duration {
	jitter := time.Duration(float64(backoff) * 0.1 * (randFloat64()*2 - 1))
	if jitter < 0 {
		jitter = -jitter
	}
	return backoff + jitter
}

func randFloat64() float64 {
	// Use math/rand with a seed based on function call index for variability
	return rand.Float64()
}