package org.mte.numecoeval.donneesentrees;

import io.zonky.test.db.AutoConfigureEmbeddedDatabase;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mte.numecoeval.donneesentrees.test.jdbc.ScriptUtils;
import org.mte.numecoeval.donneesentrees.test.kafka.KafkaConsumerMessagerie;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.ActiveProfiles;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@ActiveProfiles(profiles = {"test"})
@AutoConfigureEmbeddedDatabase
@EmbeddedKafka(
        bootstrapServersProperty = "spring.kafka.bootstrap-servers",
        partitions = 1
)
class IntegrationMessagerieTest {

    @Autowired
    JdbcTemplate jdbcTemplate;

    @Autowired
    KafkaConsumerMessagerie kafkaConsumer;

    @BeforeEach
    void setup() {
        ScriptUtils.loadScript(jdbcTemplate.getDataSource(), "sql/schema.sql");
        jdbcTemplate.batchUpdate("DELETE FROM en_messagerie");
    }

    @Test
    void whenDataAvailable_shouldUpdateStatusAndSendMessage() {
        ScriptUtils.loadScript(jdbcTemplate.getDataSource(), "sql/messagerie.sql");

        // Given
        var queryResultCountEntrees = jdbcTemplate.query(
                "SELECT count(*) as nbr from en_messagerie",
                (rs, rowNum) -> rs.getInt("nbr")
        );
        var nbrEntrees = queryResultCountEntrees.get(0);

        // When
        jdbcTemplate.batchUpdate("UPDATE en_messagerie SET statut_traitement = 'A_INGERER'");

        // Then
        assertEquals(1, nbrEntrees);
        await().atMost(5, TimeUnit.SECONDS)
                .until(() -> {
                    var queryResultCountEntreesIngerees = jdbcTemplate.query(
                            "SELECT count(*) as nbr from en_messagerie WHERE statut_traitement = 'INGERE'",
                            (rs, rowNum) -> rs.getInt("nbr")
                    );

                    return !queryResultCountEntreesIngerees.isEmpty() && Objects.equals(nbrEntrees, queryResultCountEntreesIngerees.get(0));
                });

        await().atMost(5, TimeUnit.SECONDS)
                .until(() -> nbrEntrees == kafkaConsumer.getPayloads().size());
    }

}
