@RestController
@RequiredArgsConstructor
public class OrderController {

    private final MeterRegistry registry;
    private final OrderService service;

    @GetMapping("/orders/{id}")
    public Order get(@PathVariable Long id) {
        Timer.Sample sample = Timer.start(registry);
        try {
            return service.findById(id);
        } finally {
            // Tag by outcome so prod dashboards split fast vs slow paths
            sample.stop(registry.timer("orders.get", "id", String.valueOf(id % 10)));
        }
    }
}

@KafkaListener(topics = "payments", concurrency = "6")
public void consume(List<ConsumerRecord<String, Payment>> batch, Acknowledgment ack) {
    // Process batch in parallel but bounded — don't unbounded-fan-out to a thread pool
    List<CompletableFuture<Void>> futures = batch.stream()
        .map(rec -> CompletableFuture.runAsync(() -> handle(rec.value()), workerPool))
        .toList();

    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    ack.acknowledge(); // commit only after all succeed; otherwise replay
}

private void handle(Payment p) {
    // keep this idempotent — see duplicate-payment Q below
    paymentService.process(p);
}

@Configuration
public class DataSourceConfig {

    @Bean
    public HikariDataSource dataSource() {
        HikariConfig cfg = new HikariConfig();
        cfg.setJdbcUrl(System.getenv("DB_URL"));
        cfg.setMaximumPoolSize(30);          // tune from real metrics, not guess
        cfg.setMinimumIdle(10);
        cfg.setConnectionTimeout(2_000);     // fail fast instead of queuing forever
        cfg.setLeakDetectionThreshold(10_000); // log stack trace if held > 10s
        cfg.setMaxLifetime(1_800_000);
        return new HikariDataSource(cfg);
    }
}

// ❌ BAD: holds a DB connection for the entire HTTP round-trip
@Transactional
public void badCharge(Long id) {
    Order o = repo.findById(id).orElseThrow();
    paymentGateway.charge(o); // 2s remote call — connection held the whole time
    o.markPaid();
}

// ✅ GOOD: do remote work outside the transaction
public void goodCharge(Long id) {
    Order o = readOnly(() -> repo.findById(id).orElseThrow());
    var receipt = paymentGateway.charge(o);            // no DB connection held
    txTemplate.executeWithoutResult(s -> repo.markPaid(id, receipt));
}

// Resilience4j bulkhead — cap concurrent calls into a shared dependency,
// so adding pods cannot multiply pressure beyond what the DB can handle.
@Bean
public Bulkhead dbBulkhead() {
    return Bulkhead.of("db", BulkheadConfig.custom()
        .maxConcurrentCalls(20)
        .maxWaitDuration(Duration.ofMillis(50))
        .build());
}

@Service
@RequiredArgsConstructor
public class ReportService {
    private final Bulkhead dbBulkhead;
    private final ReportRepo repo;

    public Report build(Long id) {
        return Bulkhead.decorateSupplier(dbBulkhead, () -> repo.heavyQuery(id)).get();
    }
}

@Service
@RequiredArgsConstructor
public class PaymentService {

    private final PaymentRepo repo;
    private final StripeClient stripe;

    @Retryable(retryFor = IOException.class, maxAttempts = 3,
               backoff = @Backoff(delay = 200, multiplier = 2))
    public Receipt charge(ChargeRequest req) {
        // 1. Reserve a stable idempotency key tied to the business operation
        String key = "charge:" + req.orderId();
        PaymentRecord rec = repo.findByKey(key).orElseGet(() ->
            repo.save(new PaymentRecord(key, req.orderId(), Status.PENDING)));

        if (rec.getStatus() == Status.SUCCEEDED) return rec.toReceipt(); // short-circuit

        // 2. Provider sees the same key on retry and returns the original charge
        Receipt r = stripe.charge(req, key);

        repo.markSucceeded(rec.getId(), r.id());
        return r;
    }
}

// ShedLock: backed by JDBC / Redis / Mongo — only one pod runs the job per window
@Configuration
@EnableSchedulerLock(defaultLockAtMostFor = "PT5M")
public class SchedulerConfig {
    @Bean
    public LockProvider lockProvider(DataSource ds) {
        return new JdbcTemplateLockProvider(ds);
    }
}

@Component
public class ReportJob {
    @Scheduled(cron = "0 0 * * * *")
    @SchedulerLock(name = "hourlyReport", lockAtLeastFor = "PT1M", lockAtMostFor = "PT4M")
    public void run() {
        // guaranteed single execution cluster-wide
    }
}

@Service
@RequiredArgsConstructor
public class RecommendationService {

    private final RecommendationClient client;

    @CircuitBreaker(name = "reco", fallbackMethod = "fallback")
    @TimeLimiter(name = "reco")
    @Bulkhead(name = "reco", type = Bulkhead.Type.THREADPOOL)
    public CompletableFuture<List<Item>> recommend(Long userId) {
        return CompletableFuture.supplyAsync(() -> client.fetch(userId));
    }

    @SuppressWarnings("unused")
    private CompletableFuture<List<Item>> fallback(Long userId, Throwable t) {
        // degraded but available — better than a cascading outage
        return CompletableFuture.completedFuture(List.of());
    }
}

// Global handler that logs *everything* needed to correlate later
@RestControllerAdvice
@Slf4j
public class GlobalExceptionHandler {

    @ExceptionHandler(Exception.class)
    public ResponseEntity<ErrorBody> handle(Exception ex, HttpServletRequest req) {
        String traceId = MDC.get("traceId"); // set by your tracing filter
        log.error("Unhandled: method={} uri={} pod={} traceId={}",
            req.getMethod(), req.getRequestURI(), System.getenv("HOSTNAME"), traceId, ex);
        return ResponseEntity.status(500)
            .body(new ErrorBody("INTERNAL", traceId)); // give clients the trace id
    }

    record ErrorBody(String code, String traceId) {}
}

@Component
public class DbReadinessIndicator implements HealthIndicator {
    private final JdbcTemplate jdbc;
    public DbReadinessIndicator(JdbcTemplate jdbc) { this.jdbc = jdbc; }

    @Override
    public Health health() {
        try {
            jdbc.queryForObject("SELECT 1", Integer.class);
            return Health.up().build();
        } catch (Exception e) {
            // Pod leaves load balancer rotation; users stop seeing failures from this pod
            return Health.down(e).build();
        }
    }
}

// application.yml
// management.endpoint.health.probes.enabled: true
// management.health.livenessstate.enabled: true
// management.health.readinessstate.enabled: true

@Service
@RequiredArgsConstructor
public class ProductService {

    private final ProductRepo repo;

    @Cacheable(value = "product", key = "#id")
    public Product get(Long id) {
        return repo.findById(id).orElseThrow();
    }

    // Every mutation path MUST evict — including admin tools and batch jobs
    @CacheEvict(value = "product", key = "#p.id")
    @Transactional
    public Product update(Product p) {
        return repo.save(p);
    }

    @CacheEvict(value = "product", allEntries = true)
    public void onBulkImport() { /* called from batch job */ }
}

// Filter that ensures every request has a traceId in MDC and logs
@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class TraceIdFilter extends OncePerRequestFilter {

    @Override
    protected void doFilterInternal(HttpServletRequest req, HttpServletResponse res,
                                    FilterChain chain) throws IOException, ServletException {
        String traceId = Optional.ofNullable(req.getHeader("X-Trace-Id"))
            .orElse(UUID.randomUUID().toString());
        MDC.put("traceId", traceId);
        res.setHeader("X-Trace-Id", traceId);
        try {
            chain.doFilter(req, res);
        } finally {
            MDC.clear(); // critical with thread pools — avoid leaking into next request
        }
    }
}

// logback pattern: %d %-5level [%X{traceId}] %logger - %msg%n

// ❌ unbounded static cache — heap grows forever
public class BadCache {
    private static final Map<String, byte[]> CACHE = new HashMap<>();
    public static void put(String k, byte[] v) { CACHE.put(k, v); }
}

// ✅ bounded with Caffeine — evicts by size and time
public class GoodCache {
    private static final Cache<String, byte[]> CACHE = Caffeine.newBuilder()
        .maximumWeight(64L * 1024 * 1024)              // 64MB cap
        .weigher((String k, byte[] v) -> v.length)
        .expireAfterWrite(Duration.ofMinutes(10))
        .recordStats()
        .build();
}

// ThreadLocal in a pooled thread — always clear in finally
ThreadLocal<User> CURRENT = new ThreadLocal<>();
try {
    CURRENT.set(user);
    handle();
} finally {
    CURRENT.remove(); // otherwise the next request inherits this user
}

// Log the exact request the gateway delivers, once, to diff against staging
@Component
public class GatewayDiagnosticsFilter extends OncePerRequestFilter {

    @Override
    protected void doFilterInternal(HttpServletRequest req, HttpServletResponse res,
                                    FilterChain chain) throws IOException, ServletException {
        if ("true".equalsIgnoreCase(req.getHeader("X-Diag"))) {
            Collections.list(req.getHeaderNames()).forEach(h ->
                log.info("hdr {}={}", h, req.getHeader(h)));
            log.info("uri={} query={} proto={} forwarded={}",
                req.getRequestURI(), req.getQueryString(),
                req.getHeader("X-Forwarded-Proto"),
                req.getHeader("Forwarded"));
        }
        chain.doFilter(req, res);
    }
}

// Dedicated, bounded pool per downstream — slow service can't drain the shared pool
@Configuration
public class Pools {

    @Bean("paymentsPool")
    public ExecutorService paymentsPool() {
        return new ThreadPoolExecutor(
            10, 10,
            60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(100),                  // bounded queue
            new ThreadFactoryBuilder().setNameFormat("payments-%d").build(),
            new ThreadPoolExecutor.AbortPolicy()            // fail fast — don't queue forever
        );
    }
}

@Service
@RequiredArgsConstructor
public class PaymentsFacade {
    @Qualifier("paymentsPool") private final ExecutorService pool;
    private final PaymentsClient client;

    public CompletableFuture<Receipt> charge(ChargeRequest r) {
        return CompletableFuture.supplyAsync(() -> client.charge(r), pool)
            .orTimeout(2, TimeUnit.SECONDS); // every blocking call gets a timeout
    }
}

// Full resiliency stack — order matters: bulkhead → timeLimiter → circuitBreaker → retry
@Service
@RequiredArgsConstructor
public class InventoryFacade {

    private final InventoryClient client;

    @Bulkhead(name = "inventory", type = Bulkhead.Type.THREADPOOL)
    @TimeLimiter(name = "inventory")          // hard deadline so breaker can see failures
    @CircuitBreaker(name = "inventory", fallbackMethod = "fallback")
    @Retry(name = "inventory")                // retries only when breaker is closed
    public CompletableFuture<Stock> check(Long sku) {
        return CompletableFuture.supplyAsync(() -> client.stock(sku));
    }

    @SuppressWarnings("unused")
    private CompletableFuture<Stock> fallback(Long sku, Throwable t) {
        return CompletableFuture.completedFuture(Stock.unknown(sku)); // degraded, never throw
    }
}

// application.yml
// resilience4j.circuitbreaker.instances.inventory:
//   slidingWindowSize: 50
//   failureRateThreshold: 30
//   waitDurationInOpenState: 10s
//   permittedNumberOfCallsInHalfOpenState: 5