package com.lions.dev.health;

import jakarta.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.eclipse.microprofile.health.HealthCheck;
import org.eclipse.microprofile.health.HealthCheckResponse;
import org.eclipse.microprofile.health.HealthCheckResponseBuilder;
import org.eclipse.microprofile.health.Readiness;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Optional;

/**
 * Health check pour la connexion Kafka.
 * Vérifie que le broker Kafka est accessible.
 * 
 * @since 2.0 - Production-ready
 */
@Readiness
@ApplicationScoped
public class KafkaHealthCheck implements HealthCheck {

    @ConfigProperty(name = "kafka.bootstrap.servers", defaultValue = "localhost:9092")
    String bootstrapServers;

    @Override
    public HealthCheckResponse call() {
        HealthCheckResponseBuilder responseBuilder = HealthCheckResponse.named("Kafka Connection");
        
        try {
            // Parse the first bootstrap server
            String[] servers = bootstrapServers.split(",");
            String[] hostPort = servers[0].trim().split(":");
            String host = hostPort[0];
            int port = hostPort.length > 1 ? Integer.parseInt(hostPort[1]) : 9092;
            
            // Try to connect
            try (Socket socket = new Socket()) {
                socket.connect(new InetSocketAddress(host, port), 2000);
                responseBuilder.up()
                        .withData("bootstrapServers", bootstrapServers)
                        .withData("status", "Kafka broker is reachable");
            }
        } catch (Exception e) {
            responseBuilder.down()
                    .withData("bootstrapServers", bootstrapServers)
                    .withData("error", e.getMessage())
                    .withData("status", "Kafka broker is NOT reachable");
        }
        
        return responseBuilder.build();
    }
}