package study.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.GetResponse;
import com.rabbitmq.client.Return;
import com.rabbitmq.client.ReturnCallback;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * U58 RabbitMQ 4.4.0：exchange/路由、publisher confirm、mandatory return、
 * consumer ack/nack、prefetch、重投与死信。全部对真实 broker 执行。
 */
class RabbitTopologyTest {

    Connection connection;
    Channel channel;

    @BeforeEach
    void setUp() throws Exception {
        connection = Broker.connect();
        channel = connection.createChannel();
    }

    @AfterEach
    void tearDown() throws Exception {
        Broker.closeQuietly(channel);
        Broker.closeQuietly(connection);
    }

    @Test
    void directExchangeRoutesByExactBindingKey() throws Exception {
        String exchange = "u58.direct";
        String qA = "u58.direct.a";
        String qB = "u58.direct.b";
        Broker.purge(channel, exchange, qA, qB);
        channel.exchangeDeclare(exchange, "direct", false, true, null);
        channel.queueDeclare(qA, false, true, true, null);
        channel.queueDeclare(qB, false, true, true, null);
        channel.queueBind(qA, exchange, "order.created");
        channel.queueBind(qB, exchange, "order.paid");
        channel.basicPublish(exchange, "order.created", null, "e1".getBytes(StandardCharsets.UTF_8));
        Thread.sleep(300);
        assertEquals(1, channel.messageCount(qA));
        assertEquals(0, channel.messageCount(qB));
        GetResponse received = channel.basicGet(qA, true);
        assertNotNull(received);
        assertEquals("e1", new String(received.getBody(), StandardCharsets.UTF_8));
        System.out.println("U58 direct=PASS qA=1 qB=0 routed=e1");
        Broker.purge(channel, exchange, qA, qB);
    }

    @Test
    void fanoutDeliversToEveryBoundQueue() throws Exception {
        String exchange = "u58.fanout";
        String q1 = "u58.fanout.1";
        String q2 = "u58.fanout.2";
        Broker.purge(channel, exchange, q1, q2);
        channel.exchangeDeclare(exchange, "fanout", false, true, null);
        channel.queueDeclare(q1, false, true, true, null);
        channel.queueDeclare(q2, false, true, true, null);
        channel.queueBind(q1, exchange, "");
        channel.queueBind(q2, exchange, "");
        channel.basicPublish(exchange, "ignored", null, "broadcast".getBytes(StandardCharsets.UTF_8));
        Thread.sleep(300);
        assertEquals(1, channel.messageCount(q1));
        assertEquals(1, channel.messageCount(q2));
        System.out.println("U58 fanout=PASS q1=1 q2=1");
        Broker.purge(channel, exchange, q1, q2);
    }

    @Test
    void topicExchangeMatchesWildcards() throws Exception {
        String exchange = "u58.topic";
        String qStar = "u58.topic.star";
        String qHash = "u58.topic.hash";
        Broker.purge(channel, exchange, qStar, qHash);
        channel.exchangeDeclare(exchange, "topic", false, true, null);
        channel.queueDeclare(qStar, false, true, true, null);
        channel.queueDeclare(qHash, false, true, true, null);
        channel.queueBind(qStar, exchange, "order.*.created");
        channel.queueBind(qHash, exchange, "order.#");
        channel.basicPublish(exchange, "order.cn.created", null, "x".getBytes(StandardCharsets.UTF_8));
        channel.basicPublish(exchange, "order.cn.paid.retry", null, "y".getBytes(StandardCharsets.UTF_8));
        Thread.sleep(300);
        assertEquals(1, channel.messageCount(qStar), "order.*.created 只匹配三段");
        assertEquals(2, channel.messageCount(qHash), "order.# 匹配全部");
        System.out.println("U58 topic=PASS star=1 hash=2");
        Broker.purge(channel, exchange, qStar, qHash);
    }

    @Test
    void mandatoryUnroutableMessageIsReturned() throws Exception {
        String exchange = "u58.return";
        Broker.purge(channel, exchange);
        channel.exchangeDeclare(exchange, "direct", false, true, null);
        List<String> returned = new CopyOnWriteArrayList<>();
        CountDownLatch latch = new CountDownLatch(1);
        channel.addReturnListener((replyCode, replyText, retExchange, routingKey, properties, body) -> {
            returned.add(replyCode + ":" + routingKey);
            latch.countDown();
        });
        channel.basicPublish(exchange, "no.such.key", true, null, "lost".getBytes(StandardCharsets.UTF_8));
        assertTrue(latch.await(5, TimeUnit.SECONDS), "mandatory 不可路由消息应被 return");
        assertEquals(312, Integer.parseInt(returned.get(0).split(":")[0]), "NO_ROUTE 的 replyCode 是 312");
        System.out.println("U58 mandatoryReturn=PASS " + returned.get(0));
        Broker.purge(channel, exchange);
    }

    @Test
    void publisherConfirmAcksPersistentMessage() throws Exception {
        String queue = "u58.confirm";
        Broker.purge(channel, "", queue);
        channel.queueDeclare(queue, true, false, false, null);
        channel.confirmSelect();
        channel.basicPublish("", queue,
                new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
                "durable".getBytes(StandardCharsets.UTF_8));
        assertTrue(channel.waitForConfirms(5000), "persistent 消息在 waitForConfirms 内应被 ack");
        Thread.sleep(300);
        assertEquals(1, channel.messageCount(queue));
        System.out.println("U58 publisherConfirm=PASS acked=true durable=1");
        channel.queueDelete(queue);
    }

    @Test
    void basicNackRequeuesAndRedeliveredFlagIsSet() throws Exception {
        String queue = "u58.nack";
        Broker.purge(channel, "", queue);
        channel.queueDeclare(queue, false, true, true, null);
        channel.basicPublish("", queue, null, "retry-me".getBytes(StandardCharsets.UTF_8));
        GetResponse first = channel.basicGet(queue, false);
        assertNotNull(first);
        assertEquals(Boolean.FALSE, first.getEnvelope().isRedeliver(), "第一次投递 redelivered=false");
        channel.basicNack(first.getEnvelope().getDeliveryTag(), false, true);
        GetResponse second = channel.basicGet(queue, false);
        assertNotNull(second, "nack+requeue 后消息应再次可投递");
        assertEquals(Boolean.TRUE, second.getEnvelope().isRedeliver(), "重投 redelivered=true");
        channel.basicAck(second.getEnvelope().getDeliveryTag(), false);
        System.out.println("U58 nackRequeue=PASS firstDelivered=false secondDelivered=true");
        channel.queueDelete(queue);
    }

    @Test
    void prefetchLimitsUnackedDeliveries() throws Exception {
        String queue = "u58.prefetch";
        Broker.purge(channel, "", queue);
        channel.queueDeclare(queue, false, true, true, null);
        for (int i = 0; i < 5; i++) {
            channel.basicPublish("", queue, null, ("m" + i).getBytes(StandardCharsets.UTF_8));
        }
        Channel consumerChannel = connection.createChannel();
        consumerChannel.basicQos(2);
        List<String> delivered = new CopyOnWriteArrayList<>();
        CountDownLatch twoDelivered = new CountDownLatch(2);
        consumerChannel.basicConsume(queue, false, new DefaultConsumer(consumerChannel) {
            @Override
            public void handleDelivery(String tag, Envelope envelope, AMQP.BasicProperties props, byte[] body)
                    throws IOException {
                delivered.add(new String(body, StandardCharsets.UTF_8));
                twoDelivered.countDown();
            }
        });
        assertTrue(twoDelivered.await(5, TimeUnit.SECONDS));
        Thread.sleep(800);
        assertEquals(2, delivered.size(), "prefetch=2 时未确认前不应投递第 3 条");
        // ack 一条 → 才会补投下一条
        consumerChannel.basicAck(1L, false);
        Thread.sleep(500);
        assertEquals(3, delivered.size(), "ack 一条后应补投一条");
        System.out.println("U58 prefetch=PASS beforeAck=2 afterAck=" + delivered.size());
        consumerChannel.close();
        channel.queueDelete(queue);
    }

    @Test
    void deadLetterExchangeReceivesRejectedMessage() throws Exception {
        String work = "u58.work";
        String dlx = "u58.dlx";
        String dead = "u58.dead";
        Broker.purge(channel, dlx, work, dead);
        channel.exchangeDeclare(dlx, "direct", false, true, null);
        channel.queueDeclare(dead, false, true, true, null);
        channel.queueBind(dead, dlx, "dead");
        channel.queueDeclare(work, false, true, true,
                java.util.Map.of("x-dead-letter-exchange", dlx, "x-dead-letter-routing-key", "dead"));
        channel.basicPublish("", work, null, "poison".getBytes(StandardCharsets.UTF_8));
        GetResponse message = channel.basicGet(work, false);
        assertNotNull(message);
        channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
        Thread.sleep(500);
        assertEquals(1, channel.messageCount(dead), "reject 不重回队列时应进入死信队列");
        GetResponse deadMessage = channel.basicGet(dead, true);
        assertNotNull(deadMessage);
        assertEquals("poison", new String(deadMessage.getBody(), StandardCharsets.UTF_8));
        System.out.println("U58 deadLetter=PASS work=0 dead=1");
        Broker.purge(channel, dlx, work, dead);
    }

    @Test
    void unackedMessageReturnsToQueueAfterChannelClose() throws Exception {
        String queue = "u58.recover";
        Broker.purge(channel, "", queue);
        channel.queueDeclare(queue, false, true, true, null);
        channel.basicPublish("", queue, null, "unacked".getBytes(StandardCharsets.UTF_8));
        Channel first = connection.createChannel();
        GetResponse got = first.basicGet(queue, false);
        assertNotNull(got);
        first.close(); // 未 ack 就关闭 channel：消息应回到队列
        Thread.sleep(500);
        Channel second = connection.createChannel();
        GetResponse again = second.basicGet(queue, false);
        assertNotNull(again, "未 ack 的 channel 关闭后消息应重回队列");
        assertEquals(Boolean.TRUE, again.getEnvelope().isRedeliver());
        second.basicAck(again.getEnvelope().getDeliveryTag(), false);
        second.close();
        System.out.println("U58 unackedRecovery=PASS redelivered=true");
        channel.queueDelete(queue);
    }

    @Test
    void alternateExchangeReceivesUnroutableMessage() throws Exception {
        String main = "u58.main";
        String alt = "u58.alt";
        String altQueue = "u58.alt.q";
        Broker.purge(channel, main, altQueue);
        channel.exchangeDelete(alt);
        channel.exchangeDeclare(alt, "fanout", false, true, null);
        channel.queueDeclare(altQueue, false, true, true, null);
        channel.queueBind(altQueue, alt, "");
        channel.exchangeDeclare(main, "direct", false, true,
                java.util.Map.of("alternate-exchange", alt));
        channel.basicPublish(main, "unmatched", null, "to-alt".getBytes(StandardCharsets.UTF_8));
        Thread.sleep(300);
        assertEquals(1, channel.messageCount(altQueue), "无匹配绑定时应由 alternate exchange 接收");
        System.out.println("U58 alternateExchange=PASS altQueue=1");
        Broker.purge(channel, main, altQueue);
        channel.exchangeDelete(alt);
    }

    @Test
    void quorumQueueDeliversOnSingleNode() throws Exception {
        String queue = "u58.quorum";
        Broker.purge(channel, null, queue);
        channel.queueDeclare(queue, true, false, false,
                java.util.Map.of("x-queue-type", "quorum"));
        channel.confirmSelect();
        channel.basicPublish("", queue,
                new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
                "quorum".getBytes(StandardCharsets.UTF_8));
        assertTrue(channel.waitForConfirms(5000));
        Thread.sleep(300);
        GetResponse received = channel.basicGet(queue, true);
        assertNotNull(received);
        assertEquals("quorum", new String(received.getBody(), StandardCharsets.UTF_8));
        // 单节点 quorum 队列：能声明、能投递；副本数与 Raft 多数派恢复未在本实验运行
        System.out.println("U58 quorumQueue=PASS type=quorum delivered=true replicas=1-of-1");
        channel.queueDelete(queue);
    }
}