package eta
import (
"context"
"encoding/json"
"fmt"
"math"
"net/http"
"sync"
"time"
"github.com/redis/go-redis/v9"
)
// TravelTimeFetcher fetches travel time from routing API with ML fallback.
type TravelTimeFetcher struct {
httpClient *http.Client
rdb *redis.Client
mlClient *MLPredictionClient
osrmBaseURL string
apiKey string
circuitBreaker *CircuitBreaker
cacheTTL time.Duration
}
// NewTravelTimeFetcher creates the fetcher.
func NewTravelTimeFetcher(
httpClient *http.Client,
rdb *redis.Client,
mlClient *MLPredictionClient,
osrmBaseURL string,
apiKey string,
cacheTTL time.Duration,
) *TravelTimeFetcher {
return &TravelTimeFetcher{
httpClient: httpClient,
rdb: rdb,
mlClient: mlClient,
osrmBaseURL: osrmBaseURL,
apiKey: apiKey,
circuitBreaker: NewCircuitBreaker(3, 30*time.Second),
cacheTTL: cacheTTL,
}
}
// Fetch combines routing API + ML prediction for travel time.
func (tf *TravelTimeFetcher) Fetch(ctx context.Context, hubLat, hubLng, custLat, custLng float64, hour, weekday int) (float64, error) {
// Generate cache key
cacheKey := fmt.Sprintf("eta:travel:%0.4f:%0.4f:%0.4f:%0.4f:%d:%d",
hubLat, hubLng, custLat, custLng, hour, weekday)
// Try cache first
if cached, err := tf.rdb.Get(ctx, cacheKey).Float64(); err == nil {
return cached, nil
}
// Try OSRM/Google Maps routing API (with circuit breaker)
var routeDuration float64
if tf.circuitBreaker.Allow() {
var err error
routeDuration, err = tf.callRoutingAPI(ctx, hubLat, hubLng, custLat, custLng)
if err != nil {
tf.circuitBreaker.Failure()
} else {
tf.circuitBreaker.Success()
}
}
// Get ML prediction
mlDuration, err := tf.mlClient.Predict(ctx, hubLat, hubLng, custLat, custLng, hour, weekday)
if err != nil {
// If ML also fails and we have route, use route directly
if routeDuration > 0 {
mlDuration = routeDuration
} else {
// Last resort: haversine * speed factor
dist := haversine(hubLat, hubLng, custLat, custLng)
mlDuration = dist / 8.33 // ~30 km/h
}
}
// Final: weighted average (more weight to ML if available)
var finalDuration float64
if routeDuration > 0 && mlDuration > 0 {
// Route from API but ML adjusts for conditions
finalDuration = routeDuration*0.6 + mlDuration*0.4
} else if routeDuration > 0 {
finalDuration = routeDuration
} else {
finalDuration = mlDuration
}
// Cache async
go func() {
cctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
tf.rdb.Set(cctx, cacheKey, finalDuration, tf.cacheTTL)
cancel()
}()
return math.Round(finalDuration), nil
}
// callRoutingAPI calls OSRM or Google Maps Directions API.
func (tf *TravelTimeFetcher) callRoutingAPI(ctx context.Context, hubLat, hubLng, custLat, custLng float64) (float64, error) {
url := fmt.Sprintf("%s/route/v1/driving/%f,%f;%f,%f?overview=false",
tf.osrmBaseURL, hubLng, hubLat, custLng, custLat)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return 0, fmt.Errorf("create request: %w", err)
}
if tf.apiKey != "" {
req.Header.Set("Authorization", "Bearer "+tf.apiKey)
}
resp, err := tf.httpClient.Do(req)
if err != nil {
return 0, fmt.Errorf("routing api call: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("routing api status: %d", resp.StatusCode)
}
var result struct {
Routes []struct {
Duration float64 `json:"duration"`
} `json:"routes"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return 0, fmt.Errorf("decode routing response: %w", err)
}
if len(result.Routes) == 0 {
return 0, fmt.Errorf("no routes found")
}
return result.Routes[0].Duration, nil
}
// CircuitBreaker protects the routing API from cascading failures.
type CircuitBreaker struct {
mu sync.Mutex
failCount int
maxFailures int
resetTimeout time.Duration
lastFailure time.Time
state string // closed, open, half-open
}
// NewCircuitBreaker creates a circuit breaker.
func NewCircuitBreaker(maxFailures int, resetTimeout time.Duration) *CircuitBreaker {
return &CircuitBreaker{
maxFailures: maxFailures,
resetTimeout: resetTimeout,
state: "closed",
}
}
// Allow returns true if the request should be sent.
func (cb *CircuitBreaker) Allow() bool {
cb.mu.Lock()
defer cb.mu.Unlock()
switch cb.state {
case "closed":
return true
case "open":
if time.Since(cb.lastFailure) >= cb.resetTimeout {
cb.state = "half-open"
return true
}
return false
case "half-open":
return true
default:
return true
}
}
// Success records a successful call.
func (cb *CircuitBreaker) Success() {
cb.mu.Lock()
defer cb.mu.Unlock()
cb.failCount = 0
cb.state = "closed"
}
// Failure records a failed call.
func (cb *CircuitBreaker) Failure() {
cb.mu.Lock()
defer cb.mu.Unlock()
cb.failCount++
cb.lastFailure = time.Now()
if cb.failCount >= cb.maxFailures {
cb.state = "open"
}
}
