package study.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

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;

/**
 * U59 RocketMQ 5.5.1：真实 NameServer + 单 Broker 上的普通消息、tag 过滤、顺序与延时。
 *
 * <p>broker 由 run.py 启动（127.0.0.1:9876 / 10911，存储目录临时），测试只连接本机。</p>
 */
class RocketBasicsTest {

    static final String NAMESRV = System.getProperty("lab.namesrv", "127.0.0.1:9876");
    static final String PREFIX = System.getProperty("lab.topic.prefix", "u59test");
    static final String RUN = "g" + Long.toString(System.nanoTime(), 36);
    static DefaultMQProducer producer;

    @BeforeAll
    static void startProducer() throws Exception {
        producer = new DefaultMQProducer("u59-producer-group");
        producer.setNamesrvAddr(NAMESRV);
        producer.setSendMsgTimeout(5000);
        producer.start();
    }

    @AfterAll
    static void stopProducer() {
        if (producer != null) {
            producer.shutdown();
        }
    }

    static DefaultMQPushConsumer consumer(String group, String topic, String tag, MessageListenerConcurrently listener)
            throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(NAMESRV);
        consumer.subscribe(topic, tag);
        consumer.registerMessageListener(listener);
        consumer.start();
        return consumer;
    }

    @Test
    void normalMessageIsSentAndConsumedByClusteringConsumer() throws Exception {
        String topic = PREFIX + "-normal";
        List<String> received = new CopyOnWriteArrayList<>();
        CountDownLatch latch = new CountDownLatch(1);
        DefaultMQPushConsumer consumer = consumer(RUN + "-normal-group", topic, "*", (msgs, context) -> {
            for (MessageExt message : msgs) {
                received.add(message.getBody() == null ? "" : new String(message.getBody(), StandardCharsets.UTF_8));
            }
            latch.countDown();
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            Thread.sleep(1500); // 等待路由注册与队列分配
            SendResult result = producer.send(new Message(topic, "hello-u59".getBytes(StandardCharsets.UTF_8)));
            assertEquals(SendStatus.SEND_OK, result.getSendStatus());
            assertTrue(latch.await(60, TimeUnit.SECONDS), "消息应在 60 秒内被消费");
            Thread.sleep(500);
            // 同 topic 上跨运行的残留会让 received 多于一条；本轮消息必然是最后一条且 body 相同
            assertTrue(received.contains("hello-u59"), "本轮消息应被消费: " + received);
            assertEquals("hello-u59", received.get(received.size() - 1));
            System.out.println("U59 normal=PASS status=" + result.getSendStatus()
                    + " queue=" + result.getMessageQueue().getQueueId() + " body=" + received);
        } finally {
            consumer.shutdown();
        }
    }

    @Test
    void tagFilterOnlyDeliversMatchingMessages() throws Exception {
        String topic = PREFIX + "-tags";
        List<String> received = new CopyOnWriteArrayList<>();
        CountDownLatch latch = new CountDownLatch(1);
        DefaultMQPushConsumer consumer = consumer(RUN + "-tag-group", topic, "*", (msgs, context) -> {
            for (MessageExt message : msgs) {
                String tag = message.getTags();
                if (("paid-" + RUN).equals(tag)) {
                    received.add(tag);
                    latch.countDown();
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            Thread.sleep(1500);
            producer.send(new Message(topic, "paid-" + RUN, "p1".getBytes(StandardCharsets.UTF_8)));
            producer.send(new Message(topic, "created-" + RUN, "c1".getBytes(StandardCharsets.UTF_8)));
            assertTrue(latch.await(60, TimeUnit.SECONDS));
            Thread.sleep(1500);
            assertTrue(received.stream().allMatch(t -> t.equals("paid-" + RUN)), "只应收到本运行 tag 的消息: " + received);
            System.out.println("U59 tagFilter=PASS receivedTags=" + received);
        } finally {
            consumer.shutdown();
        }
    }

    @Test
    void delayMessageIsInvisibleBeforeDeliverTime() throws Exception {
        String topic = PREFIX + "-delay";
        List<Long> arrivedAt = new CopyOnWriteArrayList<>();
        String delayBody = "delayed-" + RUN;
        CountDownLatch latch = new CountDownLatch(1);
        long sentAt = System.currentTimeMillis();
        DefaultMQPushConsumer consumer = consumer(RUN + "-delay-group", topic, "*", (msgs, context) -> {
            for (MessageExt message : msgs) {
                String body = new String(message.getBody(), StandardCharsets.UTF_8);
                if (body.equals(delayBody)) {
                    arrivedAt.add(System.currentTimeMillis());
                    latch.countDown();
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            Thread.sleep(1500);
            Message message = new Message(topic, delayBody.getBytes(StandardCharsets.UTF_8));
            // RocketMQ 5.x：delayTimeLevel 仍是可靠语义（等级 3 = 10s），deliverTimeMs 需 broker 支持定时消息
            message.setDelayTimeLevel(3);
            producer.send(message);
            Thread.sleep(3000);
            // 只统计本轮 body 的到达；残留是旧运行的消息，body 不同不会进入 arrivedAt。
            assertTrue(arrivedAt.isEmpty(), "本轮延时消息不应在 3s 内到达");
            assertTrue(latch.await(30, TimeUnit.SECONDS), "延时结束后应被消费");
            assertTrue(!arrivedAt.isEmpty(), "本轮延时消息应有一次到达记录");
            long delay = arrivedAt.get(arrivedAt.size() - 1) - sentAt;
            assertTrue(delay >= 9000, "实际延迟应接近 10s，实际 " + delay + "ms");
            System.out.println("U59 delay=PASS level=3 actualDelayMs=" + delay);
        } finally {
            consumer.shutdown();
        }
    }

    @Test
    void orderedMessagesStayInQueueOrder() throws Exception {
        String topic = PREFIX + "-ordered";
        List<String> received = new CopyOnWriteArrayList<>();
        CountDownLatch latch = new CountDownLatch(6);
        org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly listener = (msgs, context) -> {
            for (MessageExt message : msgs) {
                received.add(new String(message.getBody(), StandardCharsets.UTF_8));
                latch.countDown();
            }
            return org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus.SUCCESS;
        };
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RUN + "-order-group");
        consumer.setNamesrvAddr(NAMESRV);
        consumer.subscribe(topic, "*");
        consumer.registerMessageListener(listener);
        consumer.start();
        try {
            Thread.sleep(1500);
            // 同一业务键（shardKey）固定到同一 MessageQueue：同队列 FIFO
            for (int i = 0; i < 6; i++) {
                Message message = new Message(topic, ("shard-" + i).getBytes(StandardCharsets.UTF_8));
                producer.send(message, (mqs, msg, arg) -> mqs.get(0), "shard");
            }
            assertTrue(latch.await(30, TimeUnit.SECONDS), "6 条顺序消息应在 30 秒内消费");
            Thread.sleep(500);
            // 队列内 FIFO：本轮 6 条是发送的最后 6 条，必须是 received 的尾部子序列且严格递增
            assertTrue(received.size() >= 6, "至少收到本轮 6 条: " + received);
            List<String> tail = received.subList(received.size() - 6, received.size());
            assertEquals(List.of("shard-0", "shard-1", "shard-2", "shard-3", "shard-4", "shard-5"),
                    tail, "同一队列内应严格 FIFO（本轮尾部）");
            System.out.println("U59 ordered=PASS fifo=" + received);
        } finally {
            consumer.shutdown();
        }
    }

    @Test
    void failedConsumptionGoesToRetryQueue() throws Exception {
        String topic = PREFIX + "-retry";
        AtomicInteger attempts = new AtomicInteger();
        CountDownLatch retried = new CountDownLatch(2);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RUN + "-retry-group");
        consumer.setNamesrvAddr(NAMESRV);
        consumer.subscribe(topic, "*");
        consumer.setMaxReconsumeTimes(3);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            attempts.incrementAndGet();
            retried.countDown();
            // 第一次失败返回 RECONSUME_LATER，应进 %RETRY% 队列稍后重投
            return attempts.get() == 1
                    ? org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus.RECONSUME_LATER
                    : org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        try {
            Thread.sleep(1500);
            SendResult result = producer.send(new Message(topic, "retry-me".getBytes(StandardCharsets.UTF_8)));
            assertEquals(SendStatus.SEND_OK, result.getSendStatus());
            assertTrue(retried.await(60, TimeUnit.SECONDS), "失败后应至少重投一次");
            assertTrue(attempts.get() >= 2, "失败后应有第二次尝试，实际 " + attempts.get());
            System.out.println("U59 retry=PASS attempts=" + attempts.get()
                    + " retryQueue=%RETRY%" + RUN + "-retry-group");
        } finally {
            consumer.shutdown();
        }
    }

    @Test
    void transactionMessageCommitsAfterLocalTransaction() throws Exception {
        String topic = PREFIX + "-tx";
        List<String> received = new CopyOnWriteArrayList<>();
        CountDownLatch consumed = new CountDownLatch(1);
        String txBody = "tx-commit-" + RUN;
        DefaultMQPushConsumer consumer = consumer(RUN + "-tx-group", topic, "*", (msgs, context) -> {
            for (MessageExt message : msgs) {
                String body = new String(message.getBody(), StandardCharsets.UTF_8);
                if (body.equals(txBody)) {
                    received.add(body);
                    consumed.countDown();
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        org.apache.rocketmq.client.producer.TransactionMQProducer txProducer =
                new org.apache.rocketmq.client.producer.TransactionMQProducer(RUN + "-tx-producer");
        txProducer.setNamesrvAddr(NAMESRV);
        txProducer.setTransactionListener(new org.apache.rocketmq.client.producer.TransactionListener() {
            @Override
            public org.apache.rocketmq.client.producer.LocalTransactionState executeLocalTransaction(
                    org.apache.rocketmq.common.message.Message message, Object arg) {
                // 半消息已入 broker；本地事务"成功"→ 提交
                return org.apache.rocketmq.client.producer.LocalTransactionState.COMMIT_MESSAGE;
            }
            @Override
            public org.apache.rocketmq.client.producer.LocalTransactionState checkLocalTransaction(
                    org.apache.rocketmq.common.message.MessageExt message) {
                return org.apache.rocketmq.client.producer.LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        txProducer.start();
        try {
            Thread.sleep(1500);
            org.apache.rocketmq.client.producer.TransactionSendResult result =
                    txProducer.sendMessageInTransaction(
                            new Message(topic, txBody.getBytes(StandardCharsets.UTF_8)), null);
            assertEquals(org.apache.rocketmq.client.producer.LocalTransactionState.COMMIT_MESSAGE,
                    result.getLocalTransactionState());
            assertTrue(consumed.await(30, TimeUnit.SECONDS), "提交后的半消息应对消费者可见");
            assertEquals(List.of(txBody), received);
            System.out.println("U59 transaction=PASS half->commit->visible body=" + received);
        } finally {
            txProducer.shutdown();
            consumer.shutdown();
        }
    }
}