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.DataCenterDTO;
import org.mte.numecoeval.topic.data.EquipementPhysiqueDTO;
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.eqp"
)
public class EquipementPhysiqueIntegrationConfig {

    private static final String COLUMN_NAME_NOM_COURT_DATACENTER = "nom_court_datacenter";

    @Value("${numecoeval.topic.entree.equipementPhysique}")
    String topicEntreeEquipementPhysique;

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

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

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

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

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

    @Bean
    @InboundChannelAdapter(
            value = "entreeEquipementPhysiqueSplitter",
            poller = @Poller(fixedDelay = "5000")
    )
    @SuppressWarnings("java:S1452") // La classe JdbcPollingChannelAdapter n'est pas compatible avec une classe fixe
    public MessageSource<?> getEquipementPhysiqueToProcess(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource,
                """
                        SELECT eqp.*,
                          dc.id as dc_id,
                          dc.date_creation as dc_date_creation,
                          dc.date_update as dc_date_update,
                          dc.localisation as dc_localisation,
                          dc.nom_long_datacenter as dc_nom_long_datacenter,
                          dc.pue as dc_pue,
                          dc.nom_entite as dc_nom_entite
                        FROM en_equipement_physique eqp
                        LEFT JOIN en_data_center dc ON dc.nom_lot = eqp.nom_lot and dc.nom_court_datacenter = eqp.nom_court_datacenter
                        WHERE eqp.statut_traitement = 'A_INGERER'
                        ORDER BY nom_lot ASC, date_lot ASC, nom_organisation ASC
                        LIMIT 1000
                        """
        );
        adapter.setUpdateSql("UPDATE en_equipement_physique SET statut_traitement = 'INGERE', date_update = now() WHERE id in (:id)");
        adapter.setRowMapper((rs, index) ->
                EquipementPhysiqueDTO.builder()
                        .id(rs.getLong("id"))
                        .nomLot(rs.getString("nom_lot"))
                        .dateLot(ResultSetUtils.getLocalDate(rs, "date_lot"))
                        .nomOrganisation(rs.getString("nom_organisation"))
                        .nomEntite(rs.getString("nom_entite"))
                        .nomEquipementPhysique(rs.getString("nom_equipement_physique"))
                        .consoElecAnnuelle(ResultSetUtils.getDouble(rs, "conso_elec_annuelle"))
                        .dateAchat(ResultSetUtils.getLocalDate(rs, "date_achat"))
                        .dateRetrait(ResultSetUtils.getLocalDate(rs, "date_retrait"))
                        .goTelecharge(ResultSetUtils.getFloat(rs, "go_telecharge"))
                        .modele(rs.getString("modele"))
                        .nbCoeur(rs.getString("nb_coeur"))
                        .modeUtilisation(rs.getString("mode_utilisation"))
                        .tauxUtilisation(rs.getDouble("taux_utilisation"))
                        .paysDUtilisation(rs.getString("pays_utilisation"))
                        .quantite(ResultSetUtils.getDouble(rs, "quantite"))
                        .serveur(rs.getBoolean("serveur"))
                        .statut(rs.getString("statut"))
                        .type(rs.getString("type"))
                        .utilisateur(rs.getString("utilisateur"))
                        .nomCourtDatacenter(rs.getString(COLUMN_NAME_NOM_COURT_DATACENTER))
                        .nomSourceDonnee(rs.getString("nom_source_donnee"))
                        .dataCenter(
                                DataCenterDTO.builder()
                                        .id(rs.getLong("dc_id"))
                                        .localisation(rs.getString("dc_localisation"))
                                        .nomLongDatacenter(rs.getString("dc_nom_long_datacenter"))
                                        .pue(ResultSetUtils.getDouble(rs, "dc_pue"))
                                        .nomCourtDatacenter(rs.getString(COLUMN_NAME_NOM_COURT_DATACENTER))
                                        .nomEntite(rs.getString("dc_nom_entite"))
                                        .nomOrganisation(rs.getString("nom_organisation"))
                                        .dateLot(ResultSetUtils.getLocalDate(rs, "date_lot"))
                                        .build()
                        )
                        .build()
        );
        return adapter;
    }

    @Splitter(
            inputChannel = "entreeEquipementPhysiqueSplitter",
            outputChannel = "entreeEquipementPhysique"
    )
    public List<Message<EquipementPhysiqueDTO>> splitListEquipementPhysique(Message<List<EquipementPhysiqueDTO>> messageList) {
        if (messageList == null) return List.of();

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