package org.mte.numecoeval.donneesentrees.infrastructure.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.mte.numecoeval.common.utils.ResultSetUtils;
import org.mte.numecoeval.donneesentrees.infrastructure.utils.Constants;
import org.mte.numecoeval.topic.data.OperationNonITDTO;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.jdbc.JdbcPollingChannelAdapter;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.function.Function;

@Slf4j
@Configuration
@ConditionalOnProperty(
        value = "numecoeval.features.opnit"
)
public class OperationNonITIntegrationConfig {

    private static final String COLUMN_NAME_NOM_COURT_DATACENTER = "nom_court_datacenter";

    @Value("${numecoeval.topic.entree.operationNonIT}")
    String topicEntreeOperationNonIT;

    @Value("${numecoeval.topic.partition}")
    Integer topicPartition;

    /**
     * Topic Kafka pour les opérationsNonIT
     *
     * @return Topic à créer dans Kafka
     */
    @Bean
    public NewTopic topicEntreeOperationNonIT() {
        return new NewTopic(topicEntreeOperationNonIT, topicPartition, (short) 1);
    }

    @Bean
    public MessageChannel entreeOperationNonIT() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel entreeOperationNonITSplitter() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "entreeOperationNonIT")
    public MessageHandler operationNonITHandler(KafkaTemplate<String, OperationNonITDTO> kafkaTemplate) {
        KafkaProducerMessageHandler<String, OperationNonITDTO> handler =
                new KafkaProducerMessageHandler<>(kafkaTemplate);
        handler.setMessageKeyExpression(new LiteralExpression(UUID.randomUUID().toString()));
        handler.setTopicExpression(new ValueExpression<>(topicEntreeOperationNonIT));
        Function<Message<?>, Long> partitionIdRandomFn = (m) -> (long) (Math.random() * topicPartition);
        handler.setPartitionIdExpression(new FunctionExpression<>(partitionIdRandomFn));
        return handler;
    }

    @Bean
    @InboundChannelAdapter(
            value = "entreeOperationNonITSplitter",
            poller = @Poller(fixedDelay = "1000")
    )
    @SuppressWarnings("java:S1452") // La classe JdbcPollingChannelAdapter n'est pas compatible avec une classe fixe
    public MessageSource<?> getOperationNonITToProcess(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource,
                """
                        SELECT *
                        FROM en_operation_non_it
                        WHERE statut_traitement = 'A_INGERER'
                        ORDER BY nom_lot ASC, date_lot ASC, nom_organisation ASC
                        LIMIT 1000
                        """
        );
        adapter.setUpdateSql("UPDATE en_operation_non_it SET statut_traitement = 'INGERE', date_update = now() WHERE id in (:id)");
        adapter.setRowMapper((rs, index) ->
                OperationNonITDTO.builder()
                        .id(rs.getLong("id"))
                        .nomLot(rs.getString("nom_lot"))
                        .dateLot(ResultSetUtils.getLocalDate(rs, "date_lot"))
                        .nomOrganisation(rs.getString("nom_organisation"))
                        .nomItemNonIT(rs.getString("nom_item_non_it"))
                        .quantite(ResultSetUtils.getDouble(rs, "quantite"))
                        .type(rs.getString("type"))
                        .dureeDeVie(ResultSetUtils.getDouble(rs, "duree_de_vie"))
                        .localisation(rs.getString("localisation"))
                        .nomEntite(rs.getString("nom_entite"))
                        .nomSourceDonnee(rs.getString("nom_source_donnee"))
                        .nomCourtDatacenter(rs.getString(COLUMN_NAME_NOM_COURT_DATACENTER))
                        .description(rs.getString("description"))
                        .consoElecAnnuelle(ResultSetUtils.getDouble(rs, "conso_elec_annuelle"))
                        .qualite(rs.getString("qualite"))
                        .build()
        );
        return adapter;
    }

    @Splitter(
            inputChannel = "entreeOperationNonITSplitter",
            outputChannel = "entreeOperationNonIT"
    )
    public List<Message<OperationNonITDTO>> splitListOperationNonIT(Message<List<OperationNonITDTO>> messageList) {
        if (messageList == null) return List.of();

        return messageList.getPayload().stream()
                .map((OperationNonITDTO operationNonITDTO) -> {
                    var headers = new HashMap<String, Object>();
                    headers.put(Constants.NOM_LOT, operationNonITDTO.getNomLot());
                    headers.put(Constants.DATELOT, Objects.toString(operationNonITDTO.getDateLot(), ""));
                    headers.put(Constants.NOM_ORGANISATION, operationNonITDTO.getNomOrganisation());
                    return (Message<OperationNonITDTO>) new GenericMessage<>(operationNonITDTO, headers);
                })
                .toList();
    }
}