Kiến Trúc Thuật Toán Surge Pricing & Không Gian Uber H3

-- 1. Create Source Table consuming raw ride-hailing pings from Kafka
CREATE TABLE telemetry_events (
    event_id STRING,
    user_id STRING,
    ping_type STRING, -- 'RIDER_DEMAND' or 'DRIVER_ONLINE'
    h3_index STRING,  -- Uber H3 Resolution 9 Hex String
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'telemetry.spatial.pings',
    'properties.bootstrap.servers' = 'kafka-cluster:9092',
    'properties.group.id' = 'flink-surge-engine',
    'format' = 'json'
);

-- 2. Create Redis Sink Table using Hash mode
CREATE TABLE redis_surge_sink (
    h3_index STRING,
    surge_multiplier STRING,
    PRIMARY KEY (h3_index) NOT ENFORCED
) WITH (
    'connector' = 'redis',
    'mode' = 'single',
    'endpoint' = 'redis-cluster:6379',
    'redis-mode' = 'hash',
    'key' = 'surge:h3:res9'
);

-- 3. Execute Continuous Sliding Window Aggregation (5-Min Window, 30-Sec Slide)
INSERT INTO redis_surge_sink
SELECT 
    h3_index,
    CAST(
        CASE 
            -- Rule 1: Zero online drivers -> Maximum Surge Cap
            WHEN COUNT(CASE WHEN ping_type = 'DRIVER_ONLINE' THEN 1 END) = 0 
                 AND COUNT(CASE WHEN ping_type = 'RIDER_DEMAND' THEN 1 END) > 0 THEN 2.5
            
            -- Rule 2: Extreme Demand-to-Supply Ratio (> 3.0x) -> 2.5x Multiplier
            WHEN (CAST(COUNT(CASE WHEN ping_type = 'RIDER_DEMAND' THEN 1 END) AS DOUBLE) / 
                  NULLIF(CAST(COUNT(CASE WHEN ping_type = 'DRIVER_ONLINE' THEN 1 END) AS DOUBLE), 0)) >= 3.0 THEN 2.5

            -- Rule 3: High Demand (> 2.0x) -> 1.8x Multiplier
            WHEN (CAST(COUNT(CASE WHEN ping_type = 'RIDER_DEMAND' THEN 1 END) AS DOUBLE) / 
                  NULLIF(CAST(COUNT(CASE WHEN ping_type = 'DRIVER_ONLINE' THEN 1 END) AS DOUBLE), 0)) >= 2.0 THEN 1.8

            -- Rule 4: Moderate Demand (> 1.5x) -> 1.4x Multiplier
            WHEN (CAST(COUNT(CASE WHEN ping_type = 'RIDER_DEMAND' THEN 1 END) AS DOUBLE) / 
                  NULLIF(CAST(COUNT(CASE WHEN ping_type = 'DRIVER_ONLINE' THEN 1 END) AS DOUBLE), 0)) >= 1.5 THEN 1.4

            -- Baseline: Equilibrium
            ELSE 1.0
        END AS STRING
    ) AS surge_multiplier
FROM TABLE(
    HOP(TABLE telemetry_events, DESCRIPTOR(event_time), INTERVAL '30' SECOND, INTERVAL '5' MINUTE)
)
GROUP BY h3_index, window_start, window_end;

package spatial

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/uber/h3-go/v3"
)

type SpatialSurgeCache struct {
	redisClient *redis.Client
	resolution  int // Standard resolution for urban area: 9 (~0.10 km²)
}

func NewSpatialSurgeCache(redisClient *redis.Client, resolution int) *SpatialSurgeCache {
	if resolution == 0 {
		resolution = 9
	}
	return &SpatialSurgeCache{
		redisClient: redisClient,
		resolution:  resolution,
	}
}

// LatLngToH3Index converts latitude/longitude coordinates to Uber H3 Index string
func (s *SpatialSurgeCache) LatLngToH3Index(lat, lng float64) string {
	coord := h3.GeoCoord{Latitude: lat, Longitude: lng}
	h3Index := h3.FromGeo(coord, s.resolution)
	return h3.ToString(h3Index)
}

// GetSurgeMultiplier retrieves the surge multiplier for a given lat/lng from Redis Hash
func (s *SpatialSurgeCache) GetSurgeMultiplier(ctx context.Context, lat, lng float64) (float64, string, error) {
	h3IndexStr := s.LatLngToH3Index(lat, lng)
	hashKey := fmt.Sprintf("surge:h3:res%d", s.resolution)

	// HGET surge:h3:res9 <h3_index>
	valStr, err := s.redisClient.HGet(ctx, hashKey, h3IndexStr).Result()
	if err == redis.Nil {
		// Primary Cell miss: fallback to checking k-ring (neighboring cells) average
		return s.getNeighborRingFallback(ctx, lat, lng, hashKey)
	} else if err != nil {
		return 1.0, h3IndexStr, fmt.Errorf("redis hget error: %w", err)
	}

	multiplier, err := strconv.ParseFloat(valStr, 64)
	if err != nil {
		return 1.0, h3IndexStr, fmt.Errorf("invalid surge numeric value '%s': %w", valStr, err)
	}

	return multiplier, h3IndexStr, nil
}

// getNeighborRingFallback queries k-ring (distance 1 láng giềng) if primary cell is cold
func (s *SpatialSurgeCache) getNeighborRingFallback(ctx context.Context, lat, lng float64, hashKey string) (float64, string, error) {
	coord := h3.GeoCoord{Latitude: lat, Longitude: lng}
	centerCell := h3.FromGeo(coord, s.resolution)
	
	// KRing returns center cell + 6 immediate hexagonal neighbors
	neighbors := h3.KRing(centerCell, 1)
	fields := make([]string, len(neighbors))
	for i, n := range neighbors {
		fields[i] = h3.ToString(n)
	}

	vals, err := s.redisClient.HMGet(ctx, hashKey, fields...).Result()
	if err != nil {
		return 1.0, h3.ToString(centerCell), err
	}

	var sum float64
	var count int
	for _, v := range vals {
		if v != nil {
			if strVal, ok := v.(string); ok {
				if parsed, err := strconv.ParseFloat(strVal, 64); err == nil {
					sum += parsed
					count++
				}
			}
		}
	}

	if count == 0 {
		return 1.0, h3.ToString(centerCell), nil // Standard baseline multiplier fallback
	}

	return sum / float64(count), h3.ToString(centerCell), nil
}

package handler

import (
	"context"
	"encoding/json"
	"log/slog"
	"net/http"
	"time"

	"D:/myproject/learn/content/posts/spatial" // conceptual import
)

type FareRequest struct {
	RiderID    string  `json:"rider_id"`
	PickupLat  float64 `json:"pickup_lat"`
	PickupLng  float64 `json:"pickup_lng"`
	DropoffLat float64 `json:"dropoff_lat"`
	DropoffLng float64 `json:"dropoff_lng"`
	BasePrice  float64 `json:"base_price"`
}

type FareResponse struct {
	OrderID         string  `json:"order_id"`
	BasePrice       float64 `json:"base_price"`
	SurgeMultiplier float64 `json:"surge_multiplier"`
	FinalFare       float64 `json:"final_fare"`
	H3Index         string  `json:"h3_index"`
	IsFallback      bool    `json:"is_fallback"`
	ExecutionMs     int64   `json:"execution_ms"`
}

type FareHandler struct {
	surgeCache *spatial.SpatialSurgeCache
	logger     *slog.Logger
}

func NewFareHandler(surgeCache *spatial.SpatialSurgeCache, logger *slog.Logger) *FareHandler {
	return &FareHandler{
		surgeCache: surgeCache,
		logger:     logger,
	}
}

func (h *FareHandler) CalculateFare(w http.ResponseWriter, r *http.Request) {
	startTime := time.Now()

	var req FareRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "Invalid JSON request body", http.StatusBadRequest)
		return
	}

	// 1. Strict Fail-Safe Timeout Context (50ms SLA boundary)
	ctx, cancel := context.WithTimeout(r.Context(), 50*time.Millisecond)
	defer cancel()

	// 2. Fetch Surge Multiplier with Fail-Safe Catch
	surgeMultiplier, h3Index, err := h.surgeCache.GetSurgeMultiplier(ctx, req.PickupLat, req.PickupLng)
	
	isFallback := false
	if err != nil {
		// Log structured warning alert; trigger metrics alarm without interrupting rider purchase flow
		h.logger.WarnContext(ctx, "Surge Redis lookup failed, activating 1.0x baseline fail-safe",
			slog.String("error", err.Error()),
			slog.String("h3_index", h3Index),
			slog.Float64("pickup_lat", req.PickupLat),
			slog.Float64("pickup_lng", req.PickupLng),
		)
		surgeMultiplier = 1.0 // Mandatory baseline default
		isFallback = true
	}

	// 3. Compute final price
	finalFare := req.BasePrice * surgeMultiplier

	resp := FareResponse{
		BasePrice:       req.BasePrice,
		SurgeMultiplier: surgeMultiplier,
		FinalFare:       finalFare,
		H3Index:         h3Index,
		IsFallback:      isFallback,
		ExecutionMs:     time.Since(startTime).Milliseconds(),
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	_ = json.NewEncoder(w).Encode(resp)
}