package study.iot;

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient.Mqtt5Publishes;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5PubAckException;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5SubAckException;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCode;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCode;
import org.junit.jupiter.api.Test;
import study.iot.support.EmbeddedBroker;
import study.iot.support.RawMqttProbe;
import study.iot.support.SimulatedDevice;

import java.nio.charset.StandardCharsets;
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.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * U88 设备身份、ACL 与连接会话：全部针对真实嵌入式 HiveMQ CE 2026.5 broker 的 TCP 监听器。
 *
 * <p>身份/ACL/优雅断开用 HiveMQ MQTT Client 1.4.1 真实客户端；突然掉线的遗嘱用原始
 * MQTT 5 报文探针完成，因为高层客户端不公开“不发 DISCONNECT 直接关 socket”的入口。</p>
 */
class DeviceIdentityTest {

    @Test
    void validCredentialsConnectAndForeignTopicIsDenied() {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-a", broker.port)) {
            Mqtt5ConnAck connAck = device.connect(false, 3600);
            assertEquals(Mqtt5ConnAckReasonCode.SUCCESS, connAck.getReasonCode());
            assertFalse(connAck.isSessionPresent());

            Mqtt5SubAck own = device.client().subscribeWith()
                    .topicFilter("devices/dev-a/cmd").qos(MqttQos.AT_LEAST_ONCE).send();
            assertEquals(Mqtt5SubAckReasonCode.GRANTED_QOS_1, own.getReasonCodes().get(0));

            // 越权订阅时 broker 的 SUBACK 只含错误码，HiveMQ 客户端会抛 Mqtt5SubAckException。
            Mqtt5SubAckException foreign = assertThrows(Mqtt5SubAckException.class,
                    () -> device.client().subscribeWith()
                            .topicFilter("devices/dev-b/cmd").qos(MqttQos.AT_LEAST_ONCE).send());
            assertEquals(Mqtt5SubAckReasonCode.NOT_AUTHORIZED,
                    foreign.getMqttMessage().getReasonCodes().get(0));
            System.out.println("U88 identity=PASS ownSub=" + own.getReasonCodes()
                    + " foreignSub=" + foreign.getMqttMessage().getReasonCodes()
                    + " sessionPresent=" + connAck.isSessionPresent());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void wrongSecretIsRejectedWithNotAuthorized() {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice device = SimulatedDevice.build("dev-a", broker.port)) {
            Mqtt5ConnAckException error = assertThrows(Mqtt5ConnAckException.class,
                    device::connectWithWrongSecret);
            assertEquals(Mqtt5ConnAckReasonCode.NOT_AUTHORIZED, error.getMqttMessage().getReasonCode());
            System.out.println("U88 identityReject=PASS reason=" + error.getMqttMessage().getReasonCode());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void anonymousClientCannotConnectWhenAuthenticatorRequiresCredentials() {
        EmbeddedBroker broker = EmbeddedBroker.start();
        Mqtt5BlockingClient anonymous = MqttClient.builder()
                .useMqttVersion5().identifier("anon-device")
                .serverHost("127.0.0.1").serverPort(broker.port).buildBlocking();
        try {
            Mqtt5ConnAckException error = assertThrows(Mqtt5ConnAckException.class,
                    () -> anonymous.connectWith().cleanStart(true).send());
            assertEquals(Mqtt5ConnAckReasonCode.NOT_AUTHORIZED, error.getMqttMessage().getReasonCode());
            System.out.println("U88 anonymousReject=PASS reason=" + error.getMqttMessage().getReasonCode());
        } finally {
            if (anonymous.getState().isConnected()) {
                anonymous.disconnect();
            }
            broker.closeAndDelete();
        }
    }

    @Test
    void publishOutsideOwnPrefixIsRejectedWithErrorPubAck() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (RawMqttProbe probe = new RawMqttProbe("127.0.0.1", broker.port)) {
            assertEquals(0, probe.connect("dev-a", "device-dev-a", "secret-dev-a", true, 0),
                    "raw CONNECT must succeed");
            // QoS1 越权发布：broker 必须回错误 PUBACK（0x87 NOT_AUTHORIZED），随后断开连接。
            probe.publish("devices/dev-b/cmd", "spoof".getBytes(StandardCharsets.UTF_8), 1, false, 3, false);
            RawMqttProbe.Packet puback = probe.read();
            assertEquals(RawMqttProbe.PUBACK, puback.type(), "expected PUBACK for unauthorized publish");
            int reason = puback.body().length >= 3 ? puback.body()[2] & 0xff : 0;
            assertEquals(0x87, reason, "MQTT5 PUBACK reason must be NOT_AUTHORIZED");
            RawMqttProbe.Packet disconnect = probe.read();
            assertEquals(14, disconnect.type(), "broker must disconnect after the error PUBACK");
            System.out.println("U88 aclPublish=PASS pubackReason=0x" + Integer.toHexString(reason)
                    + " disconnectReason=0x" + Integer.toHexString(disconnect.body()[0] & 0xff));
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void embeddedBrokerWithoutAuthenticatorAllowsAnonymousConnections() {
        // 嵌入式 CE 默认把“拒绝未认证连接”关掉；没有安装认证扩展时，匿名连接会成功。
        // 这解释了为什么生产必须确认扩展真的加载，而不能只看 broker 启动成功。
        EmbeddedBroker broker = EmbeddedBroker.startWithoutExtension();
        try (SimulatedDevice anonymous = SimulatedDevice.build("anon-raw", broker.port)) {
            Mqtt5ConnAck connAck = anonymous.client().connectWith().cleanStart(true).send();
            assertEquals(Mqtt5ConnAckReasonCode.SUCCESS, connAck.getReasonCode());
            System.out.println("U88 embeddedDefaultAllowAll=PASS anonymous=" + connAck.getReasonCode());
        } finally {
            broker.closeAndDelete();
        }
    }

    @Test
    void willFiresOnAbruptSocketCloseButNotOnGracefulDisconnect() throws Exception {
        EmbeddedBroker broker = EmbeddedBroker.start();
        try (SimulatedDevice service = SimulatedDevice.build("svc-will", broker.port)) {
            service.connect(true, 0);
            service.subscribe(service.client(), "devices/#", MqttQos.AT_LEAST_ONCE);
            Mqtt5Publishes publishes = service.publishes();

            // 1) 突然断开：原始报文建连后直接关闭 socket，不发送 DISCONNECT。
            try (RawMqttProbe probe = new RawMqttProbe("127.0.0.1", broker.port)) {
                int reason = probe.connect("dev-will", "device-dev-will", "secret-dev-will",
                        true, 0, "devices/dev-will/status", "offline", 1, true);
                assertEquals(0, reason, "raw CONNECT must succeed");
                Optional<String> beforeKill = SimulatedDevice.payload(publishes, 600);
                assertTrue(beforeKill.isEmpty(), "will must not fire while connected");
            } // try-with-resources 关闭 TCP socket，不发 DISCONNECT
            Optional<String> afterKill = SimulatedDevice.payload(publishes, 5000);
            assertTrue(afterKill.isPresent(), "abrupt socket close must publish the will");
            assertEquals("offline", afterKill.get());

            // 2) 优雅断开：同一会话规格走 HiveMQ 客户端 DISCONNECT，遗嘱不得触发。
            String gracefulWillTopic = "devices/dev-will/status";
            try (SimulatedDevice graceful = SimulatedDevice.build("dev-will", broker.port)) {
                Mqtt5Publish will = Mqtt5Publish.builder()
                        .topic(gracefulWillTopic)
                        .payload("graceful-offline".getBytes(StandardCharsets.UTF_8))
                        .qos(MqttQos.AT_LEAST_ONCE)
                        .build();
                graceful.client().connectWith().cleanStart(true).sessionExpiryInterval(0)
                        .willPublish(will)
                        .simpleAuth().username("device-dev-will")
                        .password("secret-dev-will".getBytes(StandardCharsets.UTF_8)).applySimpleAuth()
                        .send();
                graceful.client().disconnect();
                Optional<String> afterGraceful = SimulatedDevice.payload(publishes, 1500);
                assertTrue(afterGraceful.isEmpty(), "graceful disconnect must not fire will: " + afterGraceful);
            }
            System.out.println("U88 will=PASS abruptWill=offline gracefulWill=false");
        } finally {
            broker.closeAndDelete();
        }
    }
}