#Uber 架构

// Cassandra 存储司机位置轨迹（高吞吐写入）
@Service
public class DriverLocationEventStore {
    @Autowired private CassandraTemplate cassandraTemplate;

    private static final String TABLE = "driver_location_events";

    // 写入位置事件：每天数十亿条
    public void recordLocation(Long driverId, double lat, double lng,
                               long timestampMs) {
        // 按天分表：driver_location_events_2024_03_15
        String table = getTableForDate(new Date(timestampMs));

        String cql = "INSERT INTO " + table +
            " (driver_id, ts, lat, lng, geohash) VALUES (?, ?, ?, ?, ?)";

        cassandraTemplate.execute(cql,
            driverId, timestampMs, lat, lng, Geohash.encode(lat, lng));

        // 这条数据会被用于：行程分析、司机审核、地图优化
    }
}

// Kafka 事件处理：行程状态变更
@Service
public class TripEventProducer {
    @Autowired private KafkaTemplate kafkaTemplate;

    // 发布事件：任何服务都可以订阅这个 topic
    public void publishTripCreated(Trip trip) {
        TripEvent event = new TripEvent();
        event.setEventType("TRIP_CREATED");
        event.setTripId(trip.getId());
        event.setTimestamp(System.currentTimeMillis());
        event.setPayload(trip.toJson());

        kafkaTemplate.send("trip-events", trip.getId().toString(), event);
    }

    public void publishTripCompleted(Trip trip) {
        TripEvent event = new TripEvent();
        event.setEventType("TRIP_COMPLETED");
        event.setTripId(trip.getId());
        event.setTimestamp(System.currentTimeMillis());
        event.setAmount(trip.getFare().getAmount());
        event.setDriverPayout(trip.getDriverPayout());

        kafkaTemplate.send("trip-events", trip.getId().toString(), event);
        // 消费方：支付系统、财务系统、数据分析系统
    }
}

// 消费事件：支付服务
@KafkaListener(topics = "trip-events", groupId = "payment-service")
public class PaymentEventConsumer {
    public void onEvent(TripEvent event) {
        switch (event.getEventType()) {
            case "TRIP_COMPLETED":
                // 触发支付扣款
                paymentService.chargeForTrip(event.getTripId());
                break;
            case "TRIP_CANCELLED":
                // 处理退款（如有预授权）
                paymentService.handleCancellation(event.getTripId());
                break;
        }
    }
}

// 调度服务：核心决策流程（伪代码）
@Service
public class DispatchService {
    @Autowired private DriverService driverService;
    @Autowired private PricingService pricingService;
    @Autowired private LocationService locationService;
    @Autowired private MetricsService metrics;

    // 调度决策：SLA = 100ms
    public DispatchDecision dispatch(Long passengerId, Location pickup) {
        long startTime = System.nanoTime();

        // 1. 查询附近可用司机（Redis GEO 查询，< 5ms）
        //    这个查询本身要快，因为是整个链路的第一步
        List<Driver> nearbyDrivers = driverService.findNearby(
            pickup.getLat(), pickup.getLng(), 5000); // 5km 半径

        if (nearbyDrivers.isEmpty()) {
            metrics.record("dispatch.no_drivers", 1);
            return DispatchDecision.noDriversAvailable();
        }

        // 2. 过滤不在线的司机（Redis 缓存，< 2ms）
        //    司机 App 定期上报心跳，心跳超时视为离线
        List<Driver> onlineDrivers = nearbyDrivers.stream()
            .filter(d -> driverService.isOnline(d.getId()))
            .collect(Collectors.toList());

        // 3. 并行计算每个司机的匹配分数
        //    充分利用多核，不要串行
        List<DriverScore> scoredDrivers = onlineDrivers.parallelStream()
            .map(driver -> calculateScore(driver, pickup))
            .collect(Collectors.toList());

        // 4. 选择最优司机
        DriverScore bestMatch = scoredDrivers.stream()
            .max(Comparator.comparingDouble(DriverScore::getTotalScore))
            .orElseThrow();

        // 5. 尝试派单（乐观锁，失败则重试最多 3 次）
        DispatchResult result = attemptDispatchWithRetry(
            bestMatch.getDriver(), passengerId, 3);

        long latencyMs = (System.nanoTime() - startTime) / 1_000_000;
        metrics.record("dispatch.latency_ms", latencyMs);
        metrics.record("dispatch.candidates_count", scoredDrivers.size());

        return result;
    }

    // 计算匹配分数
    private DriverScore calculateScore(Driver driver, Location pickup) {
        // 距离分数：越近越高（指数衰减，5km 外接近 0）
        double distanceScore = calculateDistanceScore(
            driver.getCurrentLocation(), pickup);

        // 评分分数：越高越好（线性归一化）
        double ratingScore = driver.getRating() / 5.0;

        // 供需系数：高峰期加成
        double surgeMultiplier = pricingService.getSurgeMultiplier(pickup);

        // 综合分数：距离权重最高（60%），因为等待时间是用户最敏感的指标
        double totalScore =
            distanceScore * 0.6 +
            ratingScore * 0.3 +
            surgeMultiplier * 0.1;

        return new DriverScore(driver, totalScore,
            distanceScore, ratingScore, surgeMultiplier);
    }
}

// Redis GEO 实现司机位置服务
@Service
public class DriverLocationService {
    @Autowired private RedisTemplate<String, String> redisTemplate;

    private static final String GEO_KEY = "driver:locations";

    // 更新司机位置：每 5 秒调用一次
    public void updateLocation(Long driverId, double longitude, double latitude) {
        // GEOADD：原子操作，同时设置经纬度和 Geohash
        redisTemplate.opsForGeo().add(
            GEO_KEY,
            new Point(longitude, latitude),
            driverId.toString()
        );

        // 同时更新司机在线状态（带过期时间）
        String availabilityKey = "driver:" + driverId + ":available";
        redisTemplate.opsForValue().set(availabilityKey, "1",
            Duration.ofMinutes(2)); // 2 分钟内无心跳视为离线
    }

    // 查询附近可用司机
    public List<Driver> findNearby(double longitude, double latitude,
                                   double radiusMeters) {
        // 构建圆形查询范围
        Circle circle = new Circle(
            new Point(longitude, latitude),
            new Distance(radiusMeters, MetricsUnit.METERS)
        );

        // GEORADIUS：返回指定半径内的所有成员及其距离
        GeoResults<GeoOperations.GeoLocation<String>> results =
            redisTemplate.opsForGeo().radius(GEO_KEY, circle);

        if (results == null) return Collections.emptyList();

        return results.getContent().stream()
            .filter(r -> {
                // 检查司机是否在线（心跳未超时）
                String key = "driver:" + r.getContent().getName() + ":available";
                return Boolean.TRUE.equals(
                    redisTemplate.hasKey(key));
            })
            .map(r -> {
                // 转换为完整的 Driver 对象
                Long driverId = Long.parseLong(r.getContent().getName());
                return driverService.findById(driverId);
            })
            .collect(Collectors.toList());
    }
}

// 支付服务：使用事务保证一致性
@Service
public class PaymentService {
    @Autowired private PaymentRepository paymentRepository;
    @Autowired private TripRepository tripRepository;
    @Autowired private LedgerService ledgerService;
    @Autowired private PaymentGateway paymentGateway;

    @Transactional(isolation = Isolation.SERIALIZABLE)
    public PaymentResult chargeRide(Trip trip) {
        // 1. 幂等检查：防止重复扣款
        if (paymentRepository.existsByTripId(trip.getId())) {
            return PaymentResult.alreadyCharged(trip.getId());
        }

        // 2. 验证行程状态：只有完成的行程才能扣款
        if (trip.getStatus() != TripStatus.COMPLETED) {
            throw new IllegalStateException(
                "Trip not completed: " + trip.getStatus());
        }

        // 3. 计算费用（使用 BigDecimal，精确到分）
        Money fare = pricingService.calculateFare(trip);
        // platformFee = fare * 25%（Uber 佣金）
        Money platformFee = fare.multiply(PLATFORM_COMMISSION_RATE);
        // driverPay = fare - platformFee（司机实收）
        Money driverPay = fare.subtract(platformFee);

        // 4. 创建支付记录（先插入，状态为 PENDING）
        Payment payment = new Payment();
        payment.setId(UUID.randomUUID().toString());
        payment.setTripId(trip.getId());
        payment.setPassengerId(trip.getPassengerId());
        payment.setAmount(fare.getAmount()); // BigDecimal 类型
        payment.setCurrency(fare.getCurrency()); // "USD", "CNY" 等
        payment.setStatus(PaymentStatus.PENDING);
        paymentRepository.save(payment);

        try {
            // 5. 调用支付网关（Stripe、Braintree 等）
            PaymentGatewayResult gatewayResult =
                paymentGateway.charge(trip.getPassenger().getPaymentMethod(), fare);

            if (gatewayResult.isSuccess()) {
                // 6. 支付成功：更新状态 + 写入交易流水
                payment.setStatus(PaymentStatus.COMPLETED);
                payment.setGatewayTransactionId(
                    gatewayResult.getTransactionId());
                paymentRepository.save(payment);

                // 7. 更新司机账户（异步，通过 Kafka 消息触发）
                ledgerService.creditDriver(trip.getDriverId(), driverPay);

                metrics.record("payment.success", 1);
                return PaymentResult.success(payment);

            } else {
                // 8. 支付失败：记录原因，等待重试
                payment.setStatus(PaymentStatus.FAILED);
                payment.setFailureReason(gatewayResult.getErrorMessage());
                payment.setRetryCount(payment.getRetryCount() + 1);
                paymentRepository.save(payment);

                metrics.record("payment.failed", 1);
                return PaymentResult.failed(gatewayResult.getErrorMessage());
            }

        } catch (PaymentGatewayException e) {
            // 9. 网关超时：标记为待重试
            //    定期重试任务会处理这些待重试的支付
            payment.setStatus(PaymentStatus.RETRY);
            payment.setRetryCount(payment.getRetryRetry() + 1);
            paymentRepository.save(payment);

            metrics.record("payment.retry", 1);
            return PaymentResult.retry();
        }
    }
}