package com.sonymusic.hierdispatcher.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;

@EnableKafka
@Configuration
public class KafkaConfig {

    /** Kafka bootstrap server list. */
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    /** Kafka consumer group id. */
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    /** Kafka auto offset reset policy. */
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    /** Maximum number of records per poll. */
    @Value("${spring.kafka.consumer.max-poll-records}")
    private int maxPollRecords;

    /** Maximum bytes to fetch per request. */
    @Value("${spring.kafka.consumer.properties.fetch.max.bytes}")
    private int fetchMaxBytes;

    /** Maximum bytes to fetch per partition. */
    @Value("${spring.kafka.consumer.properties.max.partition.fetch.bytes}")
    private int maxPartitionFetchBytes;

    /** Kafka listener concurrency value. */
    @Value("${app.kafka.concurrency}")
    private int concurrency;

    /** Security protocol used to connect to the Kafka brokers. */
    @Value("${spring.kafka.properties.security.protocol}")
    private String securityProtocol;

    /** SASL mechanism used for broker authentication. */
    @Value("${spring.kafka.properties.sasl.mechanism}")
    private String saslMechanism;

    /** JAAS login module configuration for SASL authentication. */
    @Value("${spring.kafka.properties.sasl.jaas.config}")
    private String saslJaasConfig;

    /** Callback handler class used for SASL authentication. */
    @Value("${spring.kafka.properties.sasl.client.callback.handler.class}")
    private String saslClientCallbackHandlerClass;

    /** Confluent Schema Registry URL used to resolve Avro schemas. */
    @Value("${spring.kafka.properties.schema.registry.url}")
    private String schemaRegistryUrl;

    /**
     * Creates the Kafka consumer factory used by the listener container.
     * All consumer properties are read from application.yml /
     * application-{profile}.yml via @Value bindings.
     * The value type is a raw Avro {@code GenericRecord}, produced by the
     * Confluent Avro deserializer against the schema registry; the listener
     * maps the fields it needs onto {@link KafkaMessage}.
     *
     * @return configured consumer factory
     */
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);
        props.put(
                ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
                maxPartitionFetchBytes);
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
        props.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
        props.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);
        props.put(
                SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS,
                saslClientCallbackHandlerClass);
        props.put(
                AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
                schemaRegistryUrl);
        props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, false);

        return new DefaultKafkaConsumerFactory<>(
                props,
                new ErrorHandlingDeserializer<>(new StringDeserializer()),
                new ErrorHandlingDeserializer<>(new KafkaAvroDeserializer()));
    }

    /**
     * Creates the Kafka listener container factory with manual acknowledgment.
     *
     * @return configured listener container factory
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object>
            kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setAckMode(AckMode.MANUAL);
        return factory;
    }
}