package study.iot.command;

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.connack.Mqtt5ConnAck;
import com.fasterxml.jackson.databind.JsonNode;
import org.junit.jupiter.api.Test;
import study.iot.command.support.ShadowBroker;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

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

/**
 * U89 命令状态机、影子与断线恢复：真实嵌入式 HiveMQ CE 2026.5 broker + 真实状态机。
 *
 * <p>服务端状态机是真实 Java 实现（不是 mock），设备用真实 MQTT 5 客户端或 Python
 * 跨语言客户端；时间用可控 Clock 精确制造"过期"，不 sleep 等真实时间。</p>
 */
class CommandShadowTest {

    static Command commandAt(String id, String device, Instant created, Instant expire) {
        return new Command(id, device, "set-speed", "{\"speed\":3}", created, expire);
    }

    static Clock at(Instant instant) {
        return Clock.fixed(instant, ZoneOffset.UTC);
    }

    static final Instant T0 = Instant.parse("2026-10-28T00:00:00Z");

    @Test
    void commandLifecycleEndsSuccessAndWritesEffectOnce() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            String root = broker.root().toString();
            EffectLog effects = new EffectLog(java.nio.file.Path.of(root, "effects.log"));
            CommandService service = new CommandService(at(T0), effects);
            Shadow shadow = new Shadow("device-a");
            Command command = commandAt("c-1", "device-a", T0, T0.plusSeconds(300));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(command));
            service.markPublished(command.commandId());
            service.markDelivered(command.commandId());
            assertEquals(CommandService.Outcome.ACKED, service.acknowledge(command.commandId(), true));
            assertEquals("SUCCEEDED", service.find(command.commandId()).orElseThrow().state().name());
            assertEquals(List.of("c-1"), effects.ids());
            long version = shadow.updateDesired(Map.of("speed", 3), 0);
            version = shadow.report(Map.of("speed", 3), version);
            assertEquals(0, shadow.delta().size(), "reported 追上 desired 后差集为空");
            System.out.println("U89 lifecycle=PASS state=SUCCEEDED effects=" + effects.ids()
                    + " version=" + version + " delta=" + shadow.delta().size());
        }
    }

    @Test
    void duplicateDeliveryProducesOneEffect() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            EffectLog effects = new EffectLog(java.nio.file.Path.of(broker.root().toString(), "effects.log"));
            CommandService service = new CommandService(at(T0), effects);
            Command command = commandAt("c-dup", "device-a", T0, T0.plusSeconds(300));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(command));
            service.markPublished(command.commandId());
            service.markDelivered(command.commandId());
            assertEquals(CommandService.Outcome.ACKED, service.acknowledge(command.commandId(), true));
            // 同一命令再次投递 arrive：幂等表拒绝，效果仍只有一条。
            assertEquals(CommandService.Outcome.DUPLICATE_SUCCEEDED, service.create(command));
            assertEquals(CommandService.Outcome.DUPLICATE_SUCCEEDED, service.acknowledge(command.commandId(), true));
            assertEquals(1, effects.ids().size());
            System.out.println("U89 duplicate=PASS duplicateOutcome=DUPLICATE_SUCCEEDED effects=" + effects.ids());
        }
    }

    /** 可推进的测试时钟：同一张状态表里制造"命令已下发、ACK 迟到"的时间线。 */
    static final class MutableClock extends Clock {
        private Instant now;

        MutableClock(Instant start) {
            this.now = start;
        }

        void advance(java.time.Duration duration) {
            now = now.plus(duration);
        }

        @Override
        public java.time.ZoneId getZone() {
            return ZoneOffset.UTC;
        }

        @Override
        public Clock withZone(java.time.ZoneId zone) {
            return this;
        }

        @Override
        public Instant instant() {
            return now;
        }
    }

    @Test
    void lateAckAfterExpiryIsRejectedWithoutEffect() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            EffectLog effects = new EffectLog(java.nio.file.Path.of(broker.root().toString(), "effects.log"));
            MutableClock clock = new MutableClock(T0);
            CommandService service = new CommandService(clock, effects);
            Command command = commandAt("c-late", "device-a", T0, T0.plusSeconds(60));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(command));
            service.markPublished(command.commandId());
            service.markDelivered(command.commandId());
            // 时间推进到过期之后，设备此时才回 ACK：必须拒绝且不写效果。
            clock.advance(java.time.Duration.ofSeconds(120));
            assertEquals(CommandService.Outcome.EXPIRED, service.acknowledge(command.commandId(), true));
            assertTrue(effects.ids().isEmpty(), "late ack must not write an effect");
            System.out.println("U89 lateAck=PASS state="
                    + service.find(command.commandId()).orElseThrow().state() + " effects=" + effects.ids());
        }
    }

    @Test
    void expiredCommandIsRejectedBeforePublish() {
        EffectLog effects = new EffectLog(java.nio.file.Path.of(System.getProperty("java.io.tmpdir"),
                "iot-command-expired-" + System.nanoTime() + ".log"));
        CommandService service = new CommandService(at(T0.plusSeconds(400)), effects);
        Command command = commandAt("c-expired", "device-a", T0, T0.plusSeconds(300));
        assertEquals(CommandService.Outcome.EXPIRED, service.create(command));
        assertEquals("EXPIRED", service.find(command.commandId()).orElseThrow().state().name());
        assertTrue(effects.ids().isEmpty());
        System.out.println("U89 expired=PASS state=EXPIRED effects=" + effects.ids());
    }

    @Test
    void shadowVersionConflictRejectsStaleUpdate() {
        Shadow shadow = new Shadow("device-a");
        long v1 = shadow.updateDesired(Map.of("speed", 3), 0);
        assertEquals(1, v1);
        Shadow.VersionConflictException error = org.junit.jupiter.api.Assertions.assertThrows(
                Shadow.VersionConflictException.class, () -> shadow.updateDesired(Map.of("speed", 5), 0));
        assertTrue(error.getMessage().contains("expected 0"));
        long v2 = shadow.updateDesired(Map.of("speed", 5), v1);
        assertEquals(2, v2);
        assertEquals(5, shadow.desired().get("speed"));
        System.out.println("U89 shadowConflict=PASS version=" + v2 + " desired=" + shadow.desired());
    }

    @Test
    void shadowSnapshotRoundTripsThroughJson() {
        Shadow shadow = new Shadow("device-a");
        long v = shadow.updateDesired(Map.of("speed", 3), 0);
        shadow.report(Map.of("speed", 1), v);
        JsonNode json = shadow.toJson();
        Shadow copy = Shadow.fromJson(json);
        assertEquals(shadow.version(), copy.version());
        assertEquals(shadow.desired(), copy.desired());
        assertEquals(shadow.reported(), copy.reported());
        assertEquals(1, copy.delta().size(), "speed 未追上，delta 仍有 1 项");
        System.out.println("U89 shadowJson=PASS version=" + copy.version() + " delta=" + copy.delta());
    }

    @Test
    void offlineCommandsAreQueuedAndExpiredOnReconnect() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            Mqtt5BlockingClient device = broker.device("dev-offline");
            ShadowBroker.connect(device, false, 600);
            device.subscribeWith().topicFilter("devices/dev-offline/cmd").qos(MqttQos.AT_LEAST_ONCE).send();
            device.disconnect();

            Mqtt5BlockingClient service = broker.device("svc-offline");
            ShadowBroker.connect(service, true, 0);
            ShadowBroker.publish(service, "devices/dev-offline/cmd",
                    "{\"commandId\":\"c-old\"}", false);
            ShadowBroker.publish(service, "devices/dev-offline/cmd",
                    "{\"commandId\":\"c-new\"}", false);
            service.disconnect();

            Mqtt5Publishes publishes = ShadowBroker.publishes(device);
            Mqtt5ConnAck second = ShadowBroker.connect(device, false, 600);
            assertTrue(second.isSessionPresent());
            Optional<String> firstMessage = ShadowBroker.receive(publishes, 5000);
            Optional<String> secondMessage = ShadowBroker.receive(publishes, 5000);
            assertTrue(firstMessage.isPresent() && secondMessage.isPresent());
            System.out.println("U89 offlineQueue=PASS messages=" + firstMessage.get() + "," + secondMessage.get());
            device.disconnect();
        }
    }

    @Test
    void crossLanguageContractDigestMatchesPython() throws Exception {
        Command command = new Command("c-x", "device-a", "set-speed", "{\"speed\":3}",
                T0, T0.plusSeconds(300));
        String javaJson = Contract.encodeCommand(command).toString();
        ProcessBuilder builder = new ProcessBuilder("python3",
                java.nio.file.Path.of("device_client.py").toAbsolutePath().toString(),
                "--contract",
                java.nio.file.Path.of("src/test/resources/contract-fixture.json").toAbsolutePath().toString(),
                "--mode", "decode", "--json", javaJson);
        builder.redirectErrorStream(true);
        Process process = builder.start();
        String output = new String(process.getInputStream().readAllBytes(), java.nio.charset.StandardCharsets.UTF_8);
        assertTrue(process.waitFor(30, TimeUnit.SECONDS));
        assertEquals(0, process.exitValue(), output);
        JsonNode node = Shadow.MAPPER.readTree(output);
        assertTrue(node.path("ok").asBoolean(), output);
        assertEquals(CommandService.digest(command), node.get("digest").asText(),
                "Java 与 Python 必须对同一命令得出相同契约摘要");
        System.out.println("U89 crossLanguage=PASS digest=" + node.get("digest").asText().substring(0, 16)
                + "... createdAtUtc=" + node.get("createdAtUtc").asText());
    }

    @Test
    void shadowDeleteIsTerminalAndRejectsFurtherUpdates() {
        Shadow shadow = new Shadow("device-a");
        long v1 = shadow.updateDesired(Map.of("speed", 3), 0);
        long v2 = shadow.delete(v1);
        assertTrue(shadow.deleted());
        Shadow.VersionConflictException error = org.junit.jupiter.api.Assertions.assertThrows(
                Shadow.VersionConflictException.class, () -> shadow.updateDesired(Map.of("speed", 9), v2));
        assertTrue(error.getMessage().contains("shadow deleted"));
        assertEquals(3, shadow.desired().get("speed"), "删除不清空历史 desired，只是进入终态");
        System.out.println("U89 shadowDelete=PASS version=" + shadow.version() + " deleted=" + shadow.deleted());
    }

    @Test
    void partialPatchMergesKeysWithoutDroppingOthers() {
        Shadow shadow = new Shadow("device-a");
        long v = shadow.updateDesired(Map.of("speed", 3, "mode", "auto"), 0);
        v = shadow.updateDesired(Map.of("speed", 5), v);
        assertEquals(Map.of("speed", 5, "mode", "auto"), shadow.desired(),
                "部分更新是合并，不是整份替换");
        assertEquals(2, v);
        System.out.println("U89 shadowMerge=PASS desired=" + shadow.desired() + " version=" + v);
    }

    @Test
    void duplicateWhileInFlightIsReportedNotReexecuted() {
        EffectLog effects = new EffectLog(java.nio.file.Path.of(System.getProperty("java.io.tmpdir"),
                "iot-command-inflight-" + System.nanoTime() + ".log"));
        CommandService service = new CommandService(at(T0), effects);
        Command command = commandAt("c-flight", "device-a", T0, T0.plusSeconds(300));
        assertEquals(CommandService.Outcome.ACCEPTED, service.create(command));
        service.markPublished(command.commandId());
        assertEquals(CommandService.Outcome.DUPLICATE_IN_FLIGHT, service.create(command),
                "命令还在途中时重复下发，不能当成新命令");
        assertTrue(effects.ids().isEmpty());
        System.out.println("U89 duplicateInFlight=PASS outcome=DUPLICATE_IN_FLIGHT effects=" + effects.ids());
    }

    @Test
    void pythonAckIsDecodedByJavaContract() throws Exception {
        Command command = new Command("c-ack", "device-a", "set-speed", "{\"speed\":3}",
                T0, T0.plusSeconds(300));
        ProcessBuilder builder = new ProcessBuilder("python3",
                java.nio.file.Path.of("device_client.py").toAbsolutePath().toString(),
                "--contract",
                java.nio.file.Path.of("src/test/resources/contract-fixture.json").toAbsolutePath().toString(),
                "--mode", "ack", "--command", Contract.encodeCommand(command).toString(),
                "--success", "true");
        builder.redirectErrorStream(true);
        Process process = builder.start();
        String output = new String(process.getInputStream().readAllBytes(), java.nio.charset.StandardCharsets.UTF_8);
        assertTrue(process.waitFor(30, TimeUnit.SECONDS));
        assertEquals(0, process.exitValue(), output);
        JsonNode node = Shadow.MAPPER.readTree(output);
        Contract.Ack ack = Contract.decodeAck(node.get("ack").toString());
        assertEquals("c-ack", ack.commandId());
        assertEquals("device-a", ack.deviceId());
        assertTrue(ack.success());
        assertEquals(0, ack.reported().path("speed").asInt());
        System.out.println("U89 pythonAck=PASS commandId=" + ack.commandId() + " success=" + ack.success());
    }

    @Test
    void malformedContractIsRejectedOnBothSides() throws Exception {
        String missingField = "{\"schemaVersion\":1,\"commandId\":\"c-1\",\"deviceId\":\"device-a\","
                + "\"type\":\"set-speed\",\"payload\":\"{}\",\"createdAt\":\"2026-10-28T00:00:00Z\"}";
        IllegalArgumentException error = org.junit.jupiter.api.Assertions.assertThrows(
                IllegalArgumentException.class, () -> Contract.decodeCommand(missingField));
        assertTrue(error.getMessage().contains("expireAt"));
        ProcessBuilder builder = new ProcessBuilder("python3",
                java.nio.file.Path.of("device_client.py").toAbsolutePath().toString(),
                "--contract",
                java.nio.file.Path.of("src/test/resources/contract-fixture.json").toAbsolutePath().toString(),
                "--mode", "decode", "--json", missingField);
        builder.redirectErrorStream(true);
        Process process = builder.start();
        String output = new String(process.getInputStream().readAllBytes(), java.nio.charset.StandardCharsets.UTF_8);
        assertTrue(process.waitFor(30, TimeUnit.SECONDS));
        assertEquals(1, process.exitValue(), output);
        assertTrue(output.contains("\"ok\": false"), output);
        System.out.println("U89 malformedRejected=PASS java=missing expireAt python=" + output.strip());
    }
}