package study.iot.support;

import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
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.message.connect.Mqtt5Connect;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;

import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * 一台“模拟设备”：使用 HiveMQ MQTT Client 1.4.1 的真实 MQTT 5 客户端，通过真实 TCP 连接
 * 嵌入式 broker。设备 id 同时决定 clientId、用户名（device-<id>）和密码（secret-<id>）。
 */
public final class SimulatedDevice implements AutoCloseable {

    private final Mqtt5BlockingClient client;
    private final String deviceId;

    private SimulatedDevice(String deviceId, Mqtt5BlockingClient client) {
        this.deviceId = deviceId;
        this.client = client;
    }

    public static SimulatedDevice build(String deviceId, int port) {
        Mqtt5BlockingClient client = MqttClient.builder()
                .useMqttVersion5()
                .identifier(deviceId)
                .serverHost("127.0.0.1")
                .serverPort(port)
                .buildBlocking();
        return new SimulatedDevice(deviceId, client);
    }

    public String deviceId() {
        return deviceId;
    }

    public Mqtt5BlockingClient client() {
        return client;
    }

    /** 使用设备自己的凭证建立或恢复会话。 */
    public Mqtt5ConnAck connect(boolean cleanStart, long sessionExpirySeconds) {
        return client.connectWith()
                .cleanStart(cleanStart)
                .sessionExpiryInterval(sessionExpirySeconds)
                .simpleAuth()
                .username(DeviceSecurityMain.DEVICE_PREFIX + deviceId)
                .password((DeviceSecurityMain.SECRET_PREFIX + deviceId).getBytes(StandardCharsets.UTF_8))
                .applySimpleAuth()
                .send();
    }

    public Mqtt5ConnAck connect() {
        return connect(true, 0);
    }

    /** 用错误密码连接，返回 broker 的拒绝原因。 */
    public Mqtt5ConnAck connectWithWrongSecret() {
        return client.connectWith()
                .cleanStart(true)
                .simpleAuth()
                .username(DeviceSecurityMain.DEVICE_PREFIX + deviceId)
                .password("wrong".getBytes(StandardCharsets.UTF_8))
                .applySimpleAuth()
                .send();
    }

    public Mqtt5ConnAck connectWith(Mqtt5Connect connect) {
        return client.connect(connect);
    }

    public Mqtt5Publishes publishes() {
        return client.publishes(MqttGlobalPublishFilter.ALL);
    }

    public static Optional<String> payload(Mqtt5Publishes publishes, long timeoutMillis) {
        try {
            Optional<Mqtt5Publish> message = publishes.receive(timeoutMillis, TimeUnit.MILLISECONDS);
            return message.map(publish -> new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return Optional.empty();
        }
    }

    public static void subscribe(Mqtt5BlockingClient client, String topicFilter, MqttQos qos) {
        client.subscribeWith().topicFilter(topicFilter).qos(qos).send();
    }

    public static void publish(Mqtt5BlockingClient client, String topic, String payload, MqttQos qos, boolean retain) {
        client.publishWith()
                .topic(topic)
                .payload(payload.getBytes(StandardCharsets.UTF_8))
                .qos(qos)
                .retain(retain)
                .send();
    }

    @Override
    public void close() {
        if (client.getState().isConnected()) {
            client.disconnect();
        }
    }
}