package com.lavinmq.springbootlavinmq;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;

class LavinmqLongDelayMessageTest {

    private static final String HOST = System.getProperty("lavinmq.host", "127.0.0.1");
    private static final int PORT = Integer.getInteger("lavinmq.port", 5672);
    private static final String USERNAME = System.getProperty("lavinmq.username", "guest");
    private static final String PASSWORD = System.getProperty("lavinmq.password", "guest");

    private static final String EXCHANGE_NAME = "test-long-delay-exchange";
    private static final String QUEUE_NAME = "test-long-delay-queue";
    private static final String ROUTING_KEY = "test-long-delay-key";

    private CachingConnectionFactory connectionFactory;
    private RabbitAdmin rabbitAdmin;
    private RabbitTemplate rabbitTemplate;

    @BeforeEach
    void setUp() {
        connectionFactory = new CachingConnectionFactory(HOST, PORT);
        connectionFactory.setUsername(USERNAME);
        connectionFactory.setPassword(PASSWORD);
        rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitTemplate = new RabbitTemplate(connectionFactory);

        declareTopology();
    }

    @AfterEach
    void tearDown() {
        if (rabbitAdmin != null) {
            rabbitAdmin.deleteQueue(QUEUE_NAME);
            rabbitAdmin.deleteExchange(EXCHANGE_NAME);
        }
        if (connectionFactory != null) {
            connectionFactory.destroy();
        }
    }

    @Test
    void shouldNotConsumeImmediatelyWhenDelayExceedsInt32() throws InterruptedException {
        long delay = Duration.ofDays(365).toMillis();
        long observeWindowMs = 60_000L;

        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, "long-delay-message", message -> {
            message.getMessageProperties().setHeader("x-delay", delay);
            return message;
        });

        long start = System.currentTimeMillis();
        long deadline = start + observeWindowMs;
        Object firstReceived = null;
        Long firstReceivedAtMs = null;
        while (System.currentTimeMillis() < deadline) {
            Object message = rabbitTemplate.receiveAndConvert(QUEUE_NAME);
            if (message != null && firstReceived == null) {
                firstReceived = message;
                firstReceivedAtMs = System.currentTimeMillis() - start;
                System.out.printf(
                        "Consumed delayed message: delay=%dms, arrivedAt=%dms, payload=%s%n",
                        delay, firstReceivedAtMs, firstReceived
                );
            }
            Thread.sleep(100L);
        }

        assertThat(firstReceived)
                .as("Message with x-delay=%d should not be delivered within %d ms, but first arrived at %s ms",
                        delay, observeWindowMs, firstReceivedAtMs)
                .isNull();
    }

    private void declareTopology() {
        Map<String, Object> exchangeArgs = new HashMap<>();
        exchangeArgs.put("x-delayed-type", "direct");
        CustomExchange exchange = new CustomExchange(EXCHANGE_NAME, "x-delayed-message", true, false, exchangeArgs);
        Queue queue = QueueBuilder.durable(QUEUE_NAME).build();
        Binding binding = BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY).noargs();

        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareBinding(binding);
        rabbitAdmin.purgeQueue(QUEUE_NAME, true);
    }
}