package study.iot.support;

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.stream.Stream;

/**
 * 真实的嵌入式 HiveMQ Community Edition 2026.5 broker。
 *
 * <p>每个实例使用独立的临时配置目录与数据目录，监听 127.0.0.1 上的动态空闲端口，
 * 以 file 持久化模式运行，并安装本实验的认证/ACL 扩展。开始/停止都会真正启动和关闭
 * Netty 监听器，测试结束后删除自己的临时目录。</p>
 */
public final class EmbeddedBroker implements AutoCloseable {

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

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

    public static EmbeddedBroker start() {
        return start(true);
    }

    /**
     * 启动时不安装安全扩展。嵌入式 CE 会把 AUTH_DENY_UNAUTHENTICATED_CONNECTIONS 置为 false，
     * 因此没有认证扩展时任何人都能匿名连接——这是必须显式验证的默认值，而不是猜测。
     */
    public static EmbeddedBroker startWithoutExtension() {
        return start(false);
    }

    private static EmbeddedBroker start(boolean withSecurityExtension) {
        try {
            return start(Files.createTempDirectory("iot-device-lab-"), withSecurityExtension);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    public static EmbeddedBroker start(Path root) {
        return start(root, true);
    }

    /**
     * 在给定目录上启动。使用同一个目录重启实例，可观察文件持久化后会话与保留消息是否还在。
     */
    public static EmbeddedBroker start(Path root, boolean withSecurityExtension) {
        try {
            Files.createDirectories(root);
            Path conf = root.resolve("conf");
            Path data = root.resolve("data");
            Path ext = root.resolve("ext");
            Files.createDirectories(conf);
            Files.createDirectories(data);
            Files.createDirectories(ext);
            int port = freePort();
            String config = """
                    <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);
            Files.writeString(conf.resolve("config.xml"), config, StandardCharsets.UTF_8);
            EmbeddedHiveMQBuilder builder = EmbeddedHiveMQ.builder()
                    .withConfigurationFolder(conf)
                    .withDataFolder(data)
                    .withExtensionsFolder(ext)
                    .withoutLoggingBootstrap();
            if (withSecurityExtension) {
                builder = builder.withEmbeddedExtension(EmbeddedExtension.builder()
                        .withId("iot-device-lab-security")
                        .withName("IoT device lab security")
                        .withVersion("1")
                        .withAuthor("study")
                        .withPriority(0)
                        .withStartPriority(1000)
                        .withExtensionMain(new DeviceSecurityMain())
                        .build());
            }
            EmbeddedHiveMQ broker = builder.build();
            broker.start().join();
            return new EmbeddedBroker(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 Path root() {
        return root;
    }

    @Override
    public void close() {
        try {
            broker.stop().join();
        } finally {
            try {
                broker.close();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("关闭嵌入式 broker 时线程被中断", e);
            } catch (java.util.concurrent.ExecutionException e) {
                throw new IllegalStateException("关闭嵌入式 broker 失败", e);
            }
        }
    }

    /** 关闭 broker 并删除本实例的全部临时目录；用于不希望保留数据的用例。 */
    public void closeAndDelete() {
        close();
        deleteTree(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) {
            // 同上。
        }
    }
}