package study.iot;

import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient.Mqtt5Publishes;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import org.junit.jupiter.api.Test;
import study.iot.support.EmbeddedBroker;
import study.iot.support.RawMqttProbe;
import study.iot.support.SimulatedDevice;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

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

/**
 * U88 MQTT 会话、QoS、订阅与保留：真实嵌入式 HiveMQ CE 2026.5 broker。
 */
class MqttSessionTest {

    @Test
    void persistentSessionQueuesQoS1WhileOfflineAndReconnects() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-q", broker.port);
             SimulatedDevice service = SimulatedDevice.build("svc-q", broker.port)) {
            Mqtt5ConnAck first = device.connect(false, 600);
            assertFalse(first.isSessionPresent());
            device.subscribe(device.client(), "devices/dev-q/cmd", MqttQos.AT_LEAST_ONCE);
            device.client().disconnect();

            service.connect(true, 0);
            for (int i = 1; i <= 3; i++) {
                SimulatedDevice.publish(service.client(), "devices/dev-q/cmd", "cmd-" + i,
                        MqttQos.AT_LEAST_ONCE, false);
            }

            Mqtt5Publishes publishes = device.publishes();
            Mqtt5ConnAck second = device.connect(false, 600);
            assertTrue(second.isSessionPresent(), "broker must restore the existing session");
            List<String> received = new ArrayList<>();
            for (int i = 0; i < 3; i++) {
                Optional<Mqtt5Publish> publish = publishes.receive(5, java.util.concurrent.TimeUnit.SECONDS);
                assertTrue(publish.isPresent(), "offline QoS1 message " + (i + 1) + " must be delivered");
                received.add(new String(publish.get().getPayloadAsBytes(), java.nio.charset.StandardCharsets.UTF_8));
            }
            assertEquals(List.of("cmd-1", "cmd-2", "cmd-3"), received);
            System.out.println("U88 session=PASS restored=" + second.isSessionPresent() + " offline=" + received);
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void cleanStartDropsPreviousSubscription() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-clean", broker.port);
             SimulatedDevice service = SimulatedDevice.build("svc-clean", broker.port)) {
            device.connect(false, 600);
            device.subscribe(device.client(), "devices/dev-clean/cmd", MqttQos.AT_LEAST_ONCE);
            device.client().disconnect();
            // Clean start：broker 必须丢弃旧订阅与会话。
            Mqtt5ConnAck clean = device.connect(true, 600);
            assertFalse(clean.isSessionPresent());
            service.connect(true, 0);
            SimulatedDevice.publish(service.client(), "devices/dev-clean/cmd", "after-clean",
                    MqttQos.AT_LEAST_ONCE, false);
            Mqtt5Publishes publishes = device.publishes();
            Optional<Mqtt5Publish> message = publishes.receive(1500, java.util.concurrent.TimeUnit.MILLISECONDS);
            assertTrue(message.isEmpty(), "cleanStart=true must not retain old subscription");
            System.out.println("U88 cleanStart=PASS sessionPresent=" + clean.isSessionPresent()
                    + " leak=" + message.isPresent());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void qos0LosesMessagesWhileOfflineQos1KeepsThem() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-qos", broker.port);
             SimulatedDevice service = SimulatedDevice.build("svc-qos", broker.port)) {
            device.connect(false, 600);
            device.subscribe(device.client(), "devices/dev-qos/q0", MqttQos.AT_MOST_ONCE);
            device.subscribe(device.client(), "devices/dev-qos/q1", MqttQos.AT_LEAST_ONCE);
            device.client().disconnect();
            service.connect(true, 0);
            SimulatedDevice.publish(service.client(), "devices/dev-qos/q0", "lost", MqttQos.AT_MOST_ONCE, false);
            SimulatedDevice.publish(service.client(), "devices/dev-qos/q1", "kept", MqttQos.AT_LEAST_ONCE, false);
            Mqtt5Publishes publishes = device.publishes();
            device.connect(false, 600);
            List<String> received = new ArrayList<>();
            Optional<Mqtt5Publish> first = publishes.receive(2500, java.util.concurrent.TimeUnit.MILLISECONDS);
            first.ifPresent(publish -> received.add(new String(publish.getPayloadAsBytes(),
                    java.nio.charset.StandardCharsets.UTF_8)));
            Optional<Mqtt5Publish> second = publishes.receive(800, java.util.concurrent.TimeUnit.MILLISECONDS);
            second.ifPresent(publish -> received.add(new String(publish.getPayloadAsBytes(),
                    java.nio.charset.StandardCharsets.UTF_8)));
            assertEquals(List.of("kept"), received, "QoS0 offline message must be dropped, QoS1 kept");
            System.out.println("U88 qosOffline=PASS qos0=lost qos1=" + received);
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void qos2HandshakeDeliversExactlyOnceToApplicationAcrossRedelivery() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice service = SimulatedDevice.build("svc-q2", broker.port)) {
            service.connect(true, 0);
            service.subscribe(service.client(), "devices/dev-q2/in", MqttQos.EXACTLY_ONCE);
            Mqtt5Publishes publishes = service.publishes();

            try (RawMqttProbe probe = new RawMqttProbe("127.0.0.1", broker.port)) {
                assertEquals(0, probe.connect("dev-q2", "device-dev-q2", "secret-dev-q2", true, 0));
                // QoS2 发布：等 PUBREC → 发 PUBREL → 等 PUBCOMP。
                probe.publish("devices/dev-q2/in", "exactly-once".getBytes(java.nio.charset.StandardCharsets.UTF_8),
                        2, false, 11, false);
                RawMqttProbe.Packet pubrec = probe.read();
                assertEquals(RawMqttProbe.PUBREC, pubrec.type(), "expected PUBREC");
                probe.ack(0x62, 11); // PUBREL
                RawMqttProbe.Packet pubcomp = probe.read();
                assertEquals(RawMqttProbe.PUBCOMP, pubcomp.type(), "expected PUBCOMP");
            }
            Optional<Mqtt5Publish> message = publishes.receive(5, java.util.concurrent.TimeUnit.SECONDS);
            assertTrue(message.isPresent());
            assertEquals("exactly-once", new String(message.get().getPayloadAsBytes(),
                    java.nio.charset.StandardCharsets.UTF_8));
            Optional<Mqtt5Publish> duplicate = publishes.receive(1200, java.util.concurrent.TimeUnit.MILLISECONDS);
            assertTrue(duplicate.isEmpty(), "QoS2 handshake must deliver once to the application");
            System.out.println("U88 qos2=PASS deliveredOnce=true");
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void redeliveryAfterUnackedQos1CarriesDupFlag() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (RawMqttProbe subscriber = new RawMqttProbe("127.0.0.1", broker.port)) {
            assertEquals(0, subscriber.connect("dev-dup", "device-dev-dup", "secret-dev-dup", false, 600));
            List<Integer> codes = subscriber.subscribe("devices/dev-dup/cmd", 1, 1);
            assertEquals(1, codes.get(codes.size() - 1), "subscription must be granted QoS1");

            try (RawMqttProbe publisher = new RawMqttProbe("127.0.0.1", broker.port)) {
                assertEquals(0, publisher.connect("svc-dup", "device-svc-dup", "secret-svc-dup", true, 0));
                publisher.publish("devices/dev-dup/cmd", "dup-check".getBytes(java.nio.charset.StandardCharsets.UTF_8),
                        1, false, 7, false);
                RawMqttProbe.Packet puback = publisher.read();
                assertEquals(RawMqttProbe.PUBACK, puback.type(), "broker must PUBACK the inbound QoS1 publish");
            }

            RawMqttProbe.Packet first = subscriber.readPublish();
            assertFalse(first.dup(), "first delivery must not set DUP");
            assertEquals("dup-check", RawMqttProbe.publishPayload(first));
            // 不发送 PUBACK 直接关闭 socket，再以 cleanStart=false 重连。
        }
        try (RawMqttProbe reconnect = new RawMqttProbe("127.0.0.1", broker.port)) {
            assertEquals(0, reconnect.connect("dev-dup", "device-dev-dup", "secret-dev-dup", false, 600));
            RawMqttProbe.Packet redeliver = reconnect.readPublish();
            assertTrue(redeliver.dup(), "unacknowledged QoS1 redelivery must set DUP=1");
            assertEquals("dup-check", RawMqttProbe.publishPayload(redeliver));
            reconnect.ack(0x40, redeliver.packetId());
            System.out.println("U88 dupRedelivery=PASS dup=" + redeliver.dup());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void retainedMessageSurvivesSubscribeAndClearsWithEmptyPayload() {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice service = SimulatedDevice.build("svc-ret", broker.port);
             SimulatedDevice late = SimulatedDevice.build("dev-ret", broker.port)) {
            service.connect(true, 0);
            SimulatedDevice.publish(service.client(), "devices/dev-ret/state", "online",
                    MqttQos.AT_LEAST_ONCE, true);

            late.connect(true, 0);
            Mqtt5Publishes publishes = late.publishes();
            late.subscribe(late.client(), "devices/dev-ret/state", MqttQos.AT_LEAST_ONCE);
            Optional<String> retained = SimulatedDevice.payload(publishes, 3000);
            assertTrue(retained.isPresent(), "retained message must be delivered to a new subscriber");
            assertEquals("online", retained.get());

            // 空 payload + retain=true 清除保留消息。
            SimulatedDevice.publish(service.client(), "devices/dev-ret/state", "",
                    MqttQos.AT_LEAST_ONCE, true);
            SimulatedDevice.payload(publishes, 500); // 消费掉清除广播本身

            try (SimulatedDevice after = SimulatedDevice.build("svc-ret2", broker.port)) {
                after.connect(true, 0);
                Mqtt5Publishes afterPublishes = after.publishes();
                after.subscribe(after.client(), "devices/dev-ret/state", MqttQos.AT_LEAST_ONCE);
                Optional<String> cleared = SimulatedDevice.payload(afterPublishes, 1500);
                assertTrue(cleared.isEmpty(), "cleared retained topic must not deliver to new subscriber");
            }
            System.out.println("U88 retained=PASS before=online afterClear=none");
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void deliveryQosIsDowngradedToSubscriptionMaximum() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-down", broker.port);
             SimulatedDevice service = SimulatedDevice.build("svc-down", broker.port)) {
            device.connect(true, 0);
            device.subscribe(device.client(), "devices/dev-down/t", MqttQos.AT_MOST_ONCE);
            Mqtt5Publishes publishes = device.publishes();
            service.connect(true, 0);
            SimulatedDevice.publish(service.client(), "devices/dev-down/t", "downgraded",
                    MqttQos.EXACTLY_ONCE, false);
            Optional<Mqtt5Publish> message = publishes.receive(5, java.util.concurrent.TimeUnit.SECONDS);
            assertTrue(message.isPresent());
            assertEquals(MqttQos.AT_MOST_ONCE, message.get().getQos(),
                    "delivered QoS must be min(publish QoS, subscription QoS)");
            System.out.println("U88 qosDowngrade=PASS published=2 subscribed=0 delivered="
                    + message.get().getQos().getCode());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void sameClientIdTakeoverKicksPreviousConnection() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice first = SimulatedDevice.build("dev-take", broker.port)) {
            first.connect(false, 600);
            first.subscribe(first.client(), "devices/dev-take/cmd", MqttQos.AT_LEAST_ONCE);
            Mqtt5Publishes firstPublishes = first.publishes();

            try (SimulatedDevice second = SimulatedDevice.build("dev-take", broker.port)) {
                Mqtt5ConnAck connAck = second.connect(false, 600);
                assertTrue(connAck.isSessionPresent(), "takeover must reuse the persistent session");
                // publishes 全局流先建立，再让第三方发布，避免把“未注册流”误判为未投递。
                Mqtt5Publishes secondPublishes = second.publishes();
                try (SimulatedDevice service = SimulatedDevice.build("svc-take", broker.port)) {
                    service.connect(true, 0);
                    SimulatedDevice.publish(service.client(), "devices/dev-take/cmd", "to-new-connection",
                            MqttQos.AT_LEAST_ONCE, false);
                }
                Optional<String> received = SimulatedDevice.payload(secondPublishes, 5000);
                assertTrue(received.isPresent(), "new connection must receive the queued command");
                assertEquals("to-new-connection", received.get());
                assertTrue(SimulatedDevice.payload(firstPublishes, 800).isEmpty(),
                        "the kicked connection must not receive new messages");
                System.out.println("U88 takeover=PASS sessionPresent=true oldKicked=true deliveredToNew="
                        + received.get());
            }
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void retainedMessageSurvivesBrokerRestartWithFilePersistence() {
        java.nio.file.Path data = null;
        try {
            data = java.nio.file.Files.createTempDirectory("iot-device-persist-");
        } catch (java.io.IOException e) {
            throw new java.io.UncheckedIOException(e);
        }
        try {
            try (EmbeddedBroker first = EmbeddedBroker.start(data)) {
                try (SimulatedDevice service = SimulatedDevice.build("svc-persist", first.port)) {
                    service.connect(true, 0);
                    SimulatedDevice.publish(service.client(), "devices/svc-persist/state", "persisted",
                            MqttQos.AT_LEAST_ONCE, true);
                }
            } // 关闭 broker（正常 stop，不是 kill -9）
            try (EmbeddedBroker second = EmbeddedBroker.start(data)) {
                try (SimulatedDevice late = SimulatedDevice.build("svc-persist2", second.port)) {
                    late.connect(true, 0);
                    Mqtt5Publishes publishes = late.publishes();
                    late.subscribe(late.client(), "devices/svc-persist/state", MqttQos.AT_LEAST_ONCE);
                    Optional<String> retained = SimulatedDevice.payload(publishes, 3000);
                    assertTrue(retained.isPresent(), "retained message must survive a normal broker restart");
                    assertEquals("persisted", retained.get());
                    System.out.println("U88 retainedRestart=PASS value=" + retained.get());
                }
            }
        } finally {
            EmbeddedBroker.deleteTree(data);
        }
    }
}