package com.battery.device.bts;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Instant;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DeviceControlCommand;
import com.battery.device.api.model.DeviceWorkStatus;

/**
 * 直连新威接口的适配器契约：命令分工、水位与增量逐点、以及三处"不猜"的边界。
 *
 * <p>传输用脚本化的假管道替代（真实管道只在现场那台机器上存在），
 * 但应答原文取自现场实测，所以这里验的是"适配器怎么用这些命令"，不是"命令能不能通"。
 */
class NewareBtsAdapterContractTest
{
    private static final String ADAPTER = "neware-f";
    private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER, 25, 81, 1, 2);

    @Test
    void discoversChannelsAndObservesRealtimeThroughThePipe()
    {
        ScriptedPipe pipe = new ScriptedPipe()
                .on("getdevinfo", "<?xml version=\"1.0\" encoding=\"UTF-8\"?><bts version=\"1.0\">"
                        + "<cmd>getdevinfo_resp</cmd><middle count=\"2\">"
                        + "<channel ip=\"127.0.0.1\" devtype=\"25\" devid=\"81\" subdevid=\"1\" Channelid=\"1\">true</channel>"
                        + "<channel ip=\"127.0.0.1\" devtype=\"25\" devid=\"81\" subdevid=\"1\" Channelid=\"2\">true</channel>"
                        + "</middle></bts>")
                .on("inquire", "<bts><cmd>inquire_resp</cmd><list count=\"1\">"
                        + "<inquire dev=\"25-81-1-2-0\" cycle_id=\"1\" step_id=\"6\" step_type=\"cc\""
                        + " workstatus=\"working\" current=\"0.6853\" voltage=\"1.7337\" capacity=\"5.3894\""
                        + " energy=\"9.3408\" totaltime=\"37737.6\" relativetime=\"28311.8\" /></list></bts>");
        try (DeviceAdapter adapter = adapter(pipe))
        {
            assertEquals(2, adapter.discoverChannels().size());

            var observation = adapter.observe(Set.of(CHANNEL)).get(CHANNEL);

            assertTrue(observation.online());
            // 读库路线拿不到通道工作状态，这正是本路线的主要收获。
            assertEquals(DeviceWorkStatus.WORKING, observation.workStatus());
            assertEquals(List.of("getdevinfo", "inquire"), pipe.commands());
        }
    }

    @Test
    void latestCabinetCarriesRealtimeValuesAndTheServerReceiptTime()
    {
        ScriptedPipe pipe = new ScriptedPipe().on("inquire",
                "<bts><cmd>inquire_resp</cmd><list count=\"1\"><inquire dev=\"25-81-1-2-0\" cycle_id=\"1\""
                        + " step_id=\"6\" step_type=\"cc\" workstatus=\"working\" current=\"0.6853\""
                        + " voltage=\"1.7337\" capacity=\"5.3894\" energy=\"9.3408\" relativetime=\"28311.8\" />"
                        + "</list></bts>");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            var latest = adapter.latestCabinet(CHANNEL);

            assertEquals(0.6853, latest.currentA(), 1e-9);
            assertEquals(1.7337, latest.voltageV(), 1e-9);
            assertEquals(1, latest.cycleNo());
            assertEquals("cc", latest.stepType());
            assertEquals("充电", latest.stepName());
            // inquire 不带绝对时间：这里的时间口径是服务器接收时间，序号留 0。
            assertNull(latest.deviceTime());
            assertEquals(0L, latest.sequenceId());
            assertEquals(Long.valueOf(28312L), latest.relativeTimeSeconds());
        }
    }

    @Test
    void watermarkIsTheLastPersistedRowAndSamplesAreStrictlyAfterIt()
    {
        ScriptedPipe pipe = new ScriptedPipe().rows(1, 5).on("inquire", CAPABLE_CLIENT);
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            SampleWatermark watermark = adapter.currentWatermark(CHANNEL);

            // 水位 = 已落盘行数那一行本身；行数就是最后一行的 seqid（实测一致）。
            assertEquals(5L, watermark.sequenceId());
            assertEquals(rowTime(5), watermark.deviceTime());
            // 先探测一次客户端能力（inquire 带 cycle_id → 支持数据文件命令），再取水位。
            assertEquals(List.of("inquire", "inquiredf", "download:5:1"), pipe.commands());

            pipe.reset();
            var rows = adapter.samplesSince(CHANNEL, new SampleWatermark(rowTime(3), 3L));

            assertEquals(2, rows.size());
            assertEquals(4L, rows.get(0).sequenceId());
            assertEquals("316:4", rows.get(0).sourceRecordId());
            // 分页从「水位 + 1」开始，只取到已落盘行数为止。
            assertEquals(List.of("inquiredf", "download:4:2"), pipe.commands());
        }
    }

    @Test
    void samplesSinceReadsNothingWithoutAWatermark()
    {
        ScriptedPipe pipe = new ScriptedPipe().rows(1, 5);
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            assertTrue(adapter.samplesSince(CHANNEL, null).isEmpty());
            // 没有水位就不该去读来源：来源里的设备历史不得并入本次接收。
            assertTrue(pipe.commands().isEmpty());
        }
    }

    @Test
    void treatsAnInquiredfReplyWithoutARecordAsNotRowBased()
    {
        // 应答里没有 chl 记录（命令看似支持却没有该通道的数据）→ 不按数据行接收，回落轮询槽位。
        ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT).on("inquiredf",
                "<bts><cmd>inquiredf_resp</cmd><list count=\"0\" /></bts>");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            assertNull(adapter.currentWatermark(CHANNEL));
        }
    }

    @Test
    void doesNotSendDataFileCommandsToAClientWithoutCycleId()
    {
        // 实测 7.6.0.1475：`inquire` 应答不带 cycle_id，而它没有 inquiredf/download/downloadStepLayer。
        // 向它发 inquiredf 的代价不只是超时——实测会让该接口的 `inquire` 连续 8 分钟无应答
        // （getdevinfo 仍正常），等于干扰了那台上位机的对外接口。所以一条都不发。
        ScriptedPipe pipe = new ScriptedPipe().on("inquire", "<bts><cmd>inquire_resp</cmd><list count=\"1\">"
                + "<inquire dev=\"21-103-1-1-0\" step_id=\"5\" step_type=\"rest\" workstatus=\"working\""
                + " voltage=\"1.2531\" current=\"0.0000\" auxtemp=\"0\" auxvol=\"0\" /></list></bts>");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            assertNull(adapter.currentWatermark(CHANNEL));
            assertTrue(adapter.stepRecords(CHANNEL).isEmpty());
            assertTrue(adapter.cycleRecords(CHANNEL).isEmpty());

            assertFalse(pipe.commands().contains("inquiredf"));
            assertFalse(pipe.commands().contains("download"));
            assertFalse(pipe.commands().contains("downloadsteplayer"));
            // 探测只做一次，之后一直按"不支持"处理。
            assertEquals(List.of("inquire"), pipe.commands());
        }
    }

    @Test
    void aChannelWithNoTestHasAZeroWatermarkAndNoSamples()
    {
        ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT).on("inquiredf",
                "<bts><cmd>inquiredf_resp</cmd><list count=\"1\"><chl devtype=\"25\" devid=\"81\" subdevid=\"1\""
                        + " chlid=\"2\" testid=\"316\" count=\"0\">true</chl></list></bts>");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            SampleWatermark watermark = adapter.currentWatermark(CHANNEL);

            // 空时间表示"这一刻来源里没有任何可用行"：之后出现的行全部属于本次接收。
            assertNull(watermark.deviceTime());
            assertEquals(0L, watermark.sequenceId());
            assertTrue(adapter.samplesSince(CHANNEL, watermark).isEmpty());
        }
    }

    @Test
    void doesNotInventCollectorOrPlanFacts()
    {
        ScriptedPipe pipe = new ScriptedPipe();
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            // 采集器测量 ADR-0068 已复核不可得；协议里没有读计划的命令。
            assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
            assertNull(adapter.testPlan(CHANNEL));
            assertEquals(NewareBtsAdapter.SAMPLE_INTERVAL_MS, adapter.persistenceIntervalMillis(CHANNEL));
            assertTrue(pipe.commands().isEmpty());
        }
    }

    @Test
    void readOnlySurfaceStaysClosedForEveryControlCommand()
    {
        ScriptedPipe pipe = new ScriptedPipe();
        try (DeviceAdapter adapter = adapter(pipe))
        {
            assertTrue(adapter.capabilities().discovery());
            assertTrue(adapter.capabilities().realtimeObservation());
            assertTrue(adapter.capabilities().historyDownload());
            assertFalse(adapter.capabilities().control());

            for (DeviceControlCommand command : DeviceControlCommand.values())
                assertThrows(UnsupportedOperationException.class, () -> adapter.control(CHANNEL, command));
            assertTrue(pipe.commands().isEmpty());
        }
    }

    @Test
    void handsOutEveryStepAndCycleWhenTheChannelHasEnded()
    {
        ScriptedPipe pipe = finishedTestPipe(DeviceWorkStatus.FINISHED);
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            var steps = adapter.stepRecords(CHANNEL);
            var cycles = adapter.cycleRecords(CHANNEL);

            assertEquals(4, steps.size());
            assertEquals("充电", steps.get(1).stepName());
            // 循环号已按 0 起归一到与厂商库、逐点同口径（应答里的 cycleid 是 1 起）。
            assertEquals(0, steps.get(0).cycleNo());

            assertEquals(1, cycles.size());
            var cycle = cycles.get(0);
            assertEquals(0, cycle.cycleNo());
            assertEquals(2.81630277777778, cycle.chargeCapacityAh(), 1e-9);
            assertEquals(0.00431017035590278, cycle.dischargeCapacityAh(), 1e-12);
            assertEquals(5.58447055555556, cycle.chargeEnergyWh(), 1e-9);
            assertEquals(0.00426197048611111, cycle.dischargeEnergyWh(), 1e-12);
            // 效率 = 放电容量 / 充电容量 × 100；搁置工步不进任何一侧。
            assertEquals(0.153, cycle.efficiency(), 0.0005);
            assertEquals("neware:316:cycle:0", cycle.sourceRecordId());
            assertEquals(CycleRecord.SOURCE_AGGREGATED, cycle.sourceKind());
        }
    }

    @Test
    void holdsBackTheRunningStepAndItsCycleWhileTheTestIsStillRunning()
    {
        // 实测这条命令会把正在进行的那一步也交出来，而它的 endseqid 恰好等于已落盘行数
        // （四个测试都相等），所以只能靠通道工作状态判断它有没有结束。
        ScriptedPipe pipe = finishedTestPipe(DeviceWorkStatus.WORKING);
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            var steps = adapter.stepRecords(CHANNEL);
            var cycles = adapter.cycleRecords(CHANNEL);

            // 最后一步（stepindex=4）与它所在的循环都不交出，等它结束再交。
            assertEquals(3, steps.size());
            assertEquals("neware:316:step:3", steps.get(2).sourceRecordId());
            assertTrue(cycles.isEmpty());
        }
    }

    @Test
    void readsTheStepLayerOnceForBothStepsAndCycles()
    {
        // 工步与循环必须来自同一次读取：否则一个工步边界落在两次读取之间时，
        // 循环合计会包含工步列表里没有的那一步，使循环记录每轮内容都不同。
        ScriptedPipe pipe = finishedTestPipe(DeviceWorkStatus.FINISHED);
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            adapter.stepRecords(CHANNEL);
            adapter.cycleRecords(CHANNEL);

            assertEquals(1, pipe.commands().stream().filter("downloadsteplayer"::equals).count());
        }
    }

    @Test
    void aChannelWithNoTestInTheBufferHasNoStepsOrCycles()
    {
        ScriptedPipe pipe = new ScriptedPipe().on("inquire", CAPABLE_CLIENT)
                .on("downloadStepLayer", "<bts><cmd>downloadStepLayer_resp</cmd><list count=\"0\" /></bts>");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            assertTrue(adapter.stepRecords(CHANNEL).isEmpty());
            assertTrue(adapter.cycleRecords(CHANNEL).isEmpty());
            // 缓冲里没有测试就直接是空，不必再去问工作状态。
            assertEquals(List.of("inquire", "downloadsteplayer"), pipe.commands());
        }
    }

    /** 一个"支持数据文件命令"的客户端应答（8.0 系：`inquire` 带 `cycle_id`）。 */
    private static final String CAPABLE_CLIENT = "<bts><cmd>inquire_resp</cmd><list count=\"1\">"
            + "<inquire dev=\"25-81-1-2-0\" cycle_id=\"1\" step_id=\"5\" step_type=\"rest\""
            + " workstatus=\"working\" voltage=\"1.25\" /></list></bts>";

    @Test
    void aFailedStateReadSkipsTheRoundInsteadOfHandingOutTheRunningStep()
    {
        // 读取失败不吞：调用方本来就有"来源读不到就本轮放弃、下一轮重试"的约定，
        // 所以一次传输抖动不会变成"把一个还在跑的工步交出去"，也不会被当成结论缓存下来。
        ScriptedPipe pipe = finishedTestPipe(DeviceWorkStatus.WORKING).failing("inquire");
        try (TestDataObservationAdapter adapter = (TestDataObservationAdapter) adapter(pipe))
        {
            assertThrows(RuntimeException.class, () -> adapter.stepRecords(CHANNEL));

            pipe.recover();
            // 恢复后重新读，拿到的是完整结论（在跑 → 交出 3 步、按住最后一步），不是上一次的空结果。
            assertEquals(3, adapter.stepRecords(CHANNEL).size());
        }
    }

    /** 一个已结束测试的工步层（现场实测原文，测试 316）：4 步都在循环 1（归一后为 0）。 */
    private static ScriptedPipe finishedTestPipe(DeviceWorkStatus workStatus)
    {
        return new ScriptedPipe()
                .on("inquire", "<bts><cmd>inquire_resp</cmd><list count=\"1\"><inquire dev=\"25-81-1-2-0\""
                        + " cycle_id=\"1\" step_id=\"4\" step_type=\"dc\" workstatus=\""
                        + workStatus.name().toLowerCase(java.util.Locale.ROOT)
                        + "\" current=\"0\" voltage=\"1.2\" /></list></bts>")
                .on("downloadStepLayer", "<bts><cmd>downloadStepLayer_resp</cmd>"
                        + "<downloadStepLayer devtype=\"25\" devid=\"81\" subdevid=\"1\" chlid=\"2\" testid=\"316\" />"
                        + "<list count=\"4\">"
                        + "<data startseqid=\"1\" endseqid=\"2\" stepindex=\"1\" stepid=\"1\" cycleid=\"1\""
                        + " steptype=\"rest\" steptime=\"23100\" endatime=\"2026-10-10 09:09:14\""
                        + " startvolt=\"0.806611083984375\" endvolt=\"0.80638037109375\" startcurr=\"0\""
                        + " endcurr=\"0\" cap=\"0\" eng=\"0\" dcir=\"0\" />"
                        + "<data startseqid=\"3\" endseqid=\"304\" stepindex=\"2\" stepid=\"2\" cycleid=\"1\""
                        + " steptype=\"cc\" steptime=\"18000000\" endatime=\"2026-10-10 14:09:14\""
                        + " startvolt=\"1.07542861328125\" endvolt=\"1.972108203125\""
                        + " startcurr=\"0.563903076171875\" endcurr=\"0.563332702636719\""
                        + " cap=\"2.81630277777778\" eng=\"5.58447055555556\" dcir=\"0\" />"
                        + "<data startseqid=\"305\" endseqid=\"307\" stepindex=\"3\" stepid=\"3\" cycleid=\"1\""
                        + " steptype=\"rest\" steptime=\"83600\" endatime=\"2026-10-10 14:10:38\""
                        + " startvolt=\"1.7850955078125\" endvolt=\"1.42794287109375\" startcurr=\"0\""
                        + " endcurr=\"0\" cap=\"0\" eng=\"0\" dcir=\"0\" />"
                        + "<data startseqid=\"308\" endseqid=\"309\" stepindex=\"4\" stepid=\"4\" cycleid=\"1\""
                        + " steptype=\"dc\" steptime=\"27600\" endatime=\"2026-10-10 14:11:05\""
                        + " startvolt=\"1.27452001953125\" endvolt=\"0.49604326171875\""
                        + " startcurr=\"-0.562971618652344\" endcurr=\"-0.5622275390625\""
                        + " cap=\"0.00431017035590278\" eng=\"0.00426197048611111\" dcir=\"0\" /></list></bts>");
    }

    private static DeviceAdapter adapter(ScriptedPipe pipe)
    {
        return new NewareBtsAdapter(ADAPTER, pipe, new BtsXmlCodec(ADAPTER, ZoneId.of(DEVICE_ZONE)));
    }

    /** 假管道里第 {@code sequenceId} 行对应的设备绝对时间（每行间隔 60 秒）。 */
    private static Instant rowTime(int sequenceId)
    {
        return Instant.parse("2026-10-10T15:00:00Z").plusSeconds(sequenceId * 60L);
    }

    /** 设备本地时间文本（厂商格式）：与适配器时区一致，编解码器按它解释回 UTC。 */
    private static final java.time.format.DateTimeFormatter DEVICE_TIME_TEXT =
            java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private static final String DEVICE_ZONE = "Asia/Hong_Kong";

    private static String rowTimeText(int sequenceId)
    {
        return DEVICE_TIME_TEXT.withZone(ZoneId.of(DEVICE_ZONE)).format(rowTime(sequenceId));
    }

    /**
     * 脚本化的假管道：按 {@code <cmd>} 返回固定应答，{@code download}/{@code inquiredf} 按内存里的行集动态生成。
     *
     * <p>它也把只读白名单当断言用：任何越权命令在到达"对端"之前就被拒绝，测试会因此失败。
     */
    private static final class ScriptedPipe implements BtsTransport
    {
        private static final Pattern COMMAND = Pattern.compile("<cmd>\\s*([A-Za-z_]+)\\s*</cmd>");
        private static final Pattern START_POSITION = Pattern.compile("startpos=\"(\\d+)\"");
        private static final Pattern COUNT = Pattern.compile("count=\"(\\d+)\"");
        private static final long TEST_ID = 316L;

        private final Map<String, String> responses = new java.util.LinkedHashMap<>();
        private final List<String> commands = new ArrayList<>();
        /** 让指定命令暂时失败，用来验"读取失败不吞、不当结论缓存"。 */
        private final java.util.Set<String> failing = new java.util.LinkedHashSet<>();
        private int firstRow = 0;
        private int lastRow = 0;

        ScriptedPipe on(String command, String response)
        {
            // 请求里的命令名按小写匹配（与生产传输的比对口径一致）。
            responses.put(command.toLowerCase(java.util.Locale.ROOT), response);
            return this;
        }

        ScriptedPipe failing(String command)
        {
            failing.add(command.toLowerCase(java.util.Locale.ROOT));
            return this;
        }

        void recover()
        {
            failing.clear();
        }

        /** 生成 {@code [first, last]} 的逐点行集，供 download/inquiredf 按窗口回答。 */
        ScriptedPipe rows(int first, int last)
        {
            firstRow = first;
            lastRow = last;
            return this;
        }

        void reset()
        {
            commands.clear();
        }

        List<String> commands()
        {
            return List.copyOf(commands);
        }

        @Override
        public String exchange(String request)
        {
            NewareBtsReadOnlyCommands.requireAllowed(request);
            Matcher matcher = COMMAND.matcher(request);
            if (!matcher.find()) throw new IllegalStateException("请求缺少命令：" + request);
            String command = matcher.group(1).toLowerCase(java.util.Locale.ROOT);
            if (failing.contains(command))
                throw new BtsTransportException(BtsTransportException.Reason.TIMEOUT, "测试脚本让 " + command + " 失败", null);
            if ("inquiredf".equals(command))
            {
                commands.add("inquiredf");
                if (responses.containsKey("inquiredf")) return responses.get("inquiredf");
                return "<bts><cmd>inquiredf_resp</cmd><list count=\"1\"><chl devtype=\"25\" devid=\"81\""
                        + " subdevid=\"1\" chlid=\"2\" testid=\"" + TEST_ID + "\" count=\"" + lastRow
                        + "\">false</chl></list></bts>";
            }
            if ("download".equals(command))
            {
                long from = Long.parseLong(value(START_POSITION, request, "1"));
                int count = Integer.parseInt(value(COUNT, request, "1"));
                commands.add("download:" + from + ":" + count);
                return downloadResponse(from, count);
            }
            commands.add(command);
            String response = responses.get(command);
            if (response == null) throw new IllegalStateException("测试脚本没有为命令准备应答：" + command);
            return response;
        }

        private String downloadResponse(long from, int count)
        {
            StringBuilder rows = new StringBuilder();
            int returned = 0;
            for (int sequenceId = (int) from; sequenceId <= lastRow && returned < count; sequenceId++, returned++)
                rows.append("<data seqid=\"").append(sequenceId).append("\" stepid=\"6\" cycleid=\"0\""
                        + " steptype=\"cc\" testtime=\"").append(sequenceId * 60000L).append("\""
                        + " atime=\"").append(rowTimeText(sequenceId)).append("\"")
                        .append(" volt=\"1.75\" curr=\"0.685\" cap=\"4.1\" eng=\"7.1\" />");
            return "<bts><cmd>donwload_resp</cmd><download devtype=\"25\" devid=\"81\" subdevid=\"1\" chlid=\"2\""
                    + " testid=\"" + TEST_ID + "\" startpos=\"" + from + "\" count=\"" + returned
                    + "\" auxid=\"0\" /><list count=\"" + returned + "\">" + rows + "</list></bts>";
        }

        private static String value(Pattern pattern, String request, String fallback)
        {
            Matcher matcher = pattern.matcher(request);
            return matcher.find() ? matcher.group(1) : fallback;
        }
    }
}