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.MessagerieDTO;
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.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;

@Slf4j
@Configuration
@ConditionalOnProperty(
        value = "numecoeval.features.mes"
)
public class MessagerieIntegrationConfig {

    @Value("${numecoeval.topic.entree.messagerie}")
    String topicEntreeMessagerie;

    @Bean
    public NewTopic topicEntreeMessagerie() {
        return new NewTopic(topicEntreeMessagerie, 1, (short) 1);
    }

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

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

    @Bean
    @ServiceActivator(inputChannel = "entreeMessagerie")
    public MessageHandler messagerieHandler(KafkaTemplate<String, MessagerieDTO> kafkaTemplate) {
        KafkaProducerMessageHandler<String, MessagerieDTO> handler =
                new KafkaProducerMessageHandler<>(kafkaTemplate);
        handler.setMessageKeyExpression(new LiteralExpression(UUID.randomUUID().toString()));
        handler.setTopicExpression(new ValueExpression<>(topicEntreeMessagerie));
        return handler;
    }

    @Bean
    @InboundChannelAdapter(
            value = "entreeMessagerieSplitter",
            poller = @Poller(fixedDelay = "5000")
    )
    @SuppressWarnings("java:S1452") // La classe JdbcPollingChannelAdapter n'est pas compatible avec une classe fixe
    public MessageSource<?> getMessagerieToProcess(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource,
                """
                        SELECT *
                        FROM en_messagerie mes
                        WHERE mes.statut_traitement = 'A_INGERER'
                        ORDER BY nom_lot ASC, date_lot ASC, nom_organisation ASC
                        LIMIT 1000
                        """
        );
        adapter.setUpdateSql("UPDATE en_messagerie SET statut_traitement = 'INGERE', date_update = now() WHERE id in (:id)");
        adapter.setRowMapper((rs, index) ->
                MessagerieDTO.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"))
                        .nomSourceDonnee(rs.getString("nom_source_donnee"))
                        .moisAnnee(rs.getInt("mois_annee"))
                        .nombreMailEmis(ResultSetUtils.getInteger(rs, "nombre_mail_emis"))
                        .nombreMailEmisXDestinataires(ResultSetUtils.getInteger(rs, "nombre_mail_emisxdestinataires"))
                        .volumeTotalMailEmis(ResultSetUtils.getInteger(rs, "volume_total_mail_emis"))
                        .build()
        );
        return adapter;
    }

    @Splitter(
            inputChannel = "entreeMessagerieSplitter",
            outputChannel = "entreeMessagerie"
    )
    public List<Message<MessagerieDTO>> splitListMessagerie(Message<List<MessagerieDTO>> messageList) {
        if (messageList == null) return List.of();

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

}
