package yandex import ( "context" "encoding/json" "fmt" "math/rand" "net/http" "net/url" "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 } // Network-level errors are retryable return true } // 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() }