package study.iot.command;

import com.fasterxml.jackson.databind.JsonNode;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient.Mqtt5Publishes;
import org.junit.jupiter.api.Test;
import study.iot.command.support.ShadowBroker;

import java.nio.file.Path;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.Map;
import java.util.Optional;

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

/**
 * U89 贯通用例：真实 broker 上跑完整闭环——
 * 云端下发命令 → 设备（独立 MQTT 客户端）收到并执行 → 回执 → 服务端状态机与影子更新。
 *
 * <p>设备端用真实 TCP 连接与真实 MQTT 5 客户端；服务端状态机、影子、幂等表都是真实
 * Java 实现。业务效果写本地文件，不是生产数据库，这一点在未运行清单中说明。</p>
 */
class EndToEndCommandTest {

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

    static final class MutableClock extends Clock {
        private Instant now;

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

        void advance(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 normalPathCommandReachesDeviceAndShadowConverges() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            EffectLog effects = new EffectLog(Path.of(broker.root().toString(), "effects.log"));
            MutableClock clock = new MutableClock(T0);
            CommandService service = new CommandService(clock, effects);
            Shadow shadow = new Shadow("e2e-a");

            Mqtt5BlockingClient cloud = broker.device("svc-e2e");
            ShadowBroker.connect(cloud, true, 0);
            Mqtt5BlockingClient device = broker.device("e2e-a");
            ShadowBroker.connect(device, false, 600);
            Mqtt5Publishes deviceInbox = ShadowBroker.publishes(device);
            device.subscribeWith().topicFilter("devices/e2e-a/cmd").qos(MqttQos.AT_LEAST_ONCE).send();

            Command command = new Command("cmd-e2e-1", "e2e-a", "set-speed", "{\"speed\":4}",
                    T0, T0.plusSeconds(300));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(command));
            long shadowVersion = shadow.updateDesired(Map.of("speed", 4), 0);
            service.markPublished(command.commandId());
            ShadowBroker.publish(cloud, "devices/e2e-a/cmd", Contract.encodeCommand(command).toString(), false);

            Optional<String> received = ShadowBroker.receive(deviceInbox, 5000);
            assertTrue(received.isPresent(), "device must receive the command over real MQTT");
            Command decoded = Contract.decodeCommand(received.get());
            assertEquals("cmd-e2e-1", decoded.commandId());
            service.markDelivered(decoded.commandId());

            // 设备执行后回执：reported.speed=4；云端处理 ack 并更新影子。
            shadowVersion = shadow.report(Map.of("speed", 4), shadowVersion);
            assertEquals(CommandService.Outcome.ACKED, service.acknowledge(decoded.commandId(), true));
            assertTrue(shadow.delta().isEmpty(), "shadow delta must converge after reported");

            JsonNode shadowJson = shadow.toJson();
            assertEquals(4, shadowJson.path("reported").path("speed").asInt());
            assertEquals(0, shadowJson.path("delta").size());
            assertEquals(1, effects.ids().size());

            // 重复投递同一命令：设备与云端各收到一次重复，effect 仍 1 条。
            ShadowBroker.publish(cloud, "devices/e2e-a/cmd", Contract.encodeCommand(command).toString(), false);
            Optional<String> duplicate = ShadowBroker.receive(deviceInbox, 5000);
            assertTrue(duplicate.isPresent());
            assertEquals(CommandService.Outcome.DUPLICATE_SUCCEEDED, service.create(command));
            assertEquals(1, effects.ids().size());
            System.out.println("U89 e2e=PASS commandId=cmd-e2e-1 shadowVersion=" + shadow.version()
                    + " delta=0 effects=" + effects.ids());
            device.disconnect();
            cloud.disconnect();
        }
    }

    @Test
    void offlineThenReconnectExecutesFreshAndDropsExpired() throws Exception {
        try (ShadowBroker broker = ShadowBroker.start()) {
            EffectLog effects = new EffectLog(Path.of(broker.root().toString(), "effects.log"));
            MutableClock clock = new MutableClock(T0);
            CommandService service = new CommandService(clock, effects);

            Mqtt5BlockingClient cloud = broker.device("svc-e2e2");
            ShadowBroker.connect(cloud, true, 0);
            // 设备必须先建立持久会话与订阅，之后离线期间的消息才会进入会话队列。
            Mqtt5BlockingClient before = broker.device("e2e-b");
            ShadowBroker.connect(before, false, 600);
            before.subscribeWith().topicFilter("devices/e2e-b/cmd").qos(MqttQos.AT_LEAST_ONCE).send();
            before.disconnect();
            // 第一条命令在设备离线时下发，并在重连前过期；第二条仍然有效。
            Command expired = new Command("cmd-old", "e2e-b", "set-speed", "{\"speed\":1}",
                    T0, T0.plusSeconds(60));
            Command fresh = new Command("cmd-new", "e2e-b", "set-speed", "{\"speed\":2}",
                    T0, T0.plusSeconds(600));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(expired));
            assertEquals(CommandService.Outcome.ACCEPTED, service.create(fresh));
            ShadowBroker.publish(cloud, "devices/e2e-b/cmd", Contract.encodeCommand(expired).toString(), false);
            ShadowBroker.publish(cloud, "devices/e2e-b/cmd", Contract.encodeCommand(fresh).toString(), false);
            clock.advance(Duration.ofSeconds(120));

            Mqtt5BlockingClient device = broker.device("e2e-b");
            ShadowBroker.connect(device, false, 600);
            Mqtt5Publishes inbox = ShadowBroker.publishes(device);
            device.subscribeWith().topicFilter("devices/e2e-b/cmd").qos(MqttQos.AT_LEAST_ONCE).send();
            Optional<String> first = ShadowBroker.receive(inbox, 5000);
            Optional<String> second = ShadowBroker.receive(inbox, 5000);
            assertTrue(first.isPresent() && second.isPresent(), "offline queue must deliver both commands");

            // 设备按 expireAt 过滤：旧命令拒绝执行并回 EXPIRED，新命令正常执行。
            for (Optional<String> message : java.util.List.of(first, second)) {
                Command command = Contract.decodeCommand(message.orElseThrow());
                boolean expiredNow = command.expiredAt(clock.instant());
                CommandService.Outcome outcome;
                if (expiredNow) {
                    service.markPublished(command.commandId());
                    service.markDelivered(command.commandId());
                    outcome = service.acknowledge(command.commandId(), true);
                    assertEquals(CommandService.Outcome.EXPIRED, outcome);
                } else {
                    service.markPublished(command.commandId());
                    service.markDelivered(command.commandId());
                    outcome = service.acknowledge(command.commandId(), true);
                    assertEquals(CommandService.Outcome.ACKED, outcome);
                }
            }
            assertEquals("EXPIRED", service.find("cmd-old").orElseThrow().state().name());
            assertEquals("SUCCEEDED", service.find("cmd-new").orElseThrow().state().name());
            assertEquals(java.util.List.of("cmd-new"), effects.ids(),
                    "only the fresh command may produce an effect");
            System.out.println("U89 e2eOffline=PASS old=EXPIRED new=SUCCEEDED effects=" + effects.ids());
            device.disconnect();
            cloud.disconnect();
        }
    }
}