// PaymentServiceClient.java — full circuit breaker + retry + fallback
@Service
@Slf4j
@RequiredArgsConstructor
public class PaymentServiceClient {

    private final WebClient webClient;
    private final CircuitBreakerRegistry circuitBreakerRegistry;
    private final RetryRegistry retryRegistry;

    /**
     * Process payment with circuit breaker, retry, and fallback.
     * Decorators applied in order: Retry → CircuitBreaker
     */
    public Mono<PaymentResult> processPayment(PaymentRequest request) {
        CircuitBreaker cb = circuitBreakerRegistry.circuitBreaker("payment-service");
        Retry retry = retryRegistry.retry("payment-service");

        return Mono.fromCallable(() -> request)
            .flatMap(this::callPaymentApi)
            .transformDeferred(CircuitBreakerOperator.of(cb))
            .transformDeferred(RetryOperator.of(retry))
            .onErrorResume(CallNotPermittedException.class, ex -> {
                log.warn("Payment circuit breaker OPEN — using fallback for request {}",
                    request.getOrderId());
                return processFallbackPayment(request);
            })
            .onErrorResume(ex -> {
                log.error("Payment service failed after retries for order {}: {}",
                    request.getOrderId(), ex.getMessage());
                return Mono.error(new PaymentServiceUnavailableException(
                    "Payment service temporarily unavailable", ex));
            });
    }

    private Mono<PaymentResult> callPaymentApi(PaymentRequest request) {
        return webClient.post()
            .uri("/payments")
            .bodyValue(request)
            .retrieve()
            .onStatus(status -> status.is5xxServerError(),
                resp -> resp.bodyToMono(String.class)
                    .flatMap(body -> Mono.error(
                        new PaymentServerException("Payment service error: " + body))))
            .bodyToMono(PaymentResult.class)
            .timeout(Duration.ofMillis(2000));  // Hard timeout
    }

    /**
     * Fallback: queue the payment for async processing when circuit is open.
     */
    private Mono<PaymentResult> processFallbackPayment(PaymentRequest request) {
        return paymentQueueService.enqueue(request)
            .map(queueId -> PaymentResult.pending(request.getOrderId(), queueId));
    }

    @EventListener
    public void onCircuitBreakerStateChange(CircuitBreakerOnStateTransitionEvent event) {
        log.warn("Payment circuit breaker state: {} → {}",
            event.getStateTransition().getFromState(),
            event.getStateTransition().getToState());

        // Alert on-call if circuit opens
        if (event.getStateTransition().getToState() == CircuitBreaker.State.OPEN) {
            alertingService.sendAlert(Alert.builder()
                .severity(Severity.HIGH)
                .message("Payment service circuit breaker opened")
                .build());
        }
    }
}

@Service
@Slf4j
@RequiredArgsConstructor
public class RecommendationService {

    private final RecommendationsApiClient apiClient;
    private final RedisTemplate<String, List<String>> redisTemplate;
    private final RecommendationProperties properties;
    private final MeterRegistry meterRegistry;

    @CircuitBreaker(name = "recommendations-api", fallbackMethod = "getRecommendationsFallback")
    public List<String> getRecommendations(String userId) {
        return apiClient.fetchRecommendations(userId);
    }

    private List<String> getRecommendationsFallback(String userId, Throwable ex) {
        // Try Redis cache first
        String cacheKey = "recs:" + userId;
        List<String> cached = redisTemplate.opsForValue().get(cacheKey);

        if (cached != null && !cached.isEmpty()) {
            log.info("recommendations fallback: using Redis cache for user={}", userId);
            meterRegistry.counter("recommendations.fallback", "source", "cache").increment();
            return cached;
        }

        // Fall back to static defaults
        log.warn("recommendations fallback: Redis miss, using static defaults for user={}", userId);
        meterRegistry.counter("recommendations.fallback", "source", "static").increment();
        return properties.getDefaultProductIds();
    }
}