package study.iot.command.support;

import study.iot.command.Shadow;

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.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import com.hivemq.embedded.EmbeddedExtension;
import com.hivemq.embedded.EmbeddedHiveMQ;
import com.hivemq.embedded.EmbeddedHiveMQBuilder;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.ServerSocket;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Comparator;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 真实嵌入式 HiveMQ CE 2026.5 broker，带设备身份/ACL 扩展，127.0.0.1 随机端口、file 持久化。
 * 与 U88 的 EmbeddedBroker 同一装配方式；本模块独立一份，避免跨实验耦合。
 */
public final class ShadowBroker implements AutoCloseable {

    private final Path root;
    private final EmbeddedHiveMQ broker;
    public final int port;

    private ShadowBroker(Path root, EmbeddedHiveMQ broker, int port) {
        this.root = root;
        this.broker = broker;
        this.port = port;
    }

    public static ShadowBroker start() {
        try {
            return start(Files.createTempDirectory("iot-command-lab-"));
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    public static ShadowBroker start(Path root) {
        try {
            Files.createDirectories(root);
            Path conf = Files.createDirectories(root.resolve("conf"));
            Path data = Files.createDirectories(root.resolve("data"));
            Path ext = Files.createDirectories(root.resolve("ext"));
            int port = freePort();
            Files.writeString(conf.resolve("config.xml"), """
                    <hivemq xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
                        <listeners>
                            <tcp-listener>
                                <port>%d</port>
                                <bind-address>127.0.0.1</bind-address>
                            </tcp-listener>
                        </listeners>
                        <mqtt>
                            <session-expiry>
                                <max-interval>3600</max-interval>
                            </session-expiry>
                        </mqtt>
                        <persistence>
                            <mode>file</mode>
                        </persistence>
                        <anonymous-usage-statistics>
                            <enabled>false</enabled>
                        </anonymous-usage-statistics>
                    </hivemq>
                    """.formatted(port), StandardCharsets.UTF_8);
            EmbeddedHiveMQBuilder builder = EmbeddedHiveMQ.builder()
                    .withConfigurationFolder(conf)
                    .withDataFolder(data)
                    .withExtensionsFolder(ext)
                    .withoutLoggingBootstrap()
                    .withEmbeddedExtension(EmbeddedExtension.builder()
                            .withId("iot-command-lab-security")
                            .withName("IoT command lab security")
                            .withVersion("1")
                            .withAuthor("study")
                            .withPriority(0)
                            .withStartPriority(1000)
                            .withExtensionMain(new DeviceSecurityMain())
                            .build());
            EmbeddedHiveMQ broker = builder.build();
            broker.start().join();
            return new ShadowBroker(root, broker, port);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    public static int freePort() {
        try (ServerSocket socket = new ServerSocket(0)) {
            socket.setReuseAddress(true);
            return socket.getLocalPort();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    public Mqtt5BlockingClient device(String deviceId) {
        Mqtt5BlockingClient client = MqttClient.builder().useMqttVersion5().identifier(deviceId)
                .serverHost("127.0.0.1").serverPort(port).buildBlocking();
        return client;
    }

    public static Mqtt5ConnAck connect(Mqtt5BlockingClient client, boolean cleanStart, long sessionExpiry) {
        String id = client.getConfig().getClientIdentifier().orElseThrow().toString();
        return client.connectWith().cleanStart(cleanStart).sessionExpiryInterval(sessionExpiry)
                .simpleAuth().username("device-" + id)
                .password(("secret-" + id).getBytes(StandardCharsets.UTF_8)).applySimpleAuth().send();
    }

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

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

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

    @Override
    public void close() {
        try {
            broker.stop().join();
        } finally {
            try {
                broker.close();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException(e);
            } catch (java.util.concurrent.ExecutionException e) {
                throw new IllegalStateException(e);
            }
        }
    }

    public void closeAndDelete() {
        close();
        deleteTree(root);
    }

    public Path root() {
        return root;
    }

    public static void deleteTree(Path root) {
        if (root == null || !Files.exists(root)) {
            return;
        }
        try (Stream<Path> walk = Files.walk(root)) {
            walk.sorted(Comparator.reverseOrder()).forEach(path -> {
                try {
                    Files.deleteIfExists(path);
                } catch (IOException ignored) {
                    // 清理尽力而为，残留由证据 JSON 的 cleanup 字段注明。
                }
            });
        } catch (IOException ignored) {
            // 同上。
        }
    }
}