package com.battery.device.neware;

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.assertTrue;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;

/**
 * 直连新威库的适配器契约测试：用可控的假厂商库验证发现清单与最近数据时间、
 * 最新测试按起始时间取、分表定位用测试记录给出的表名、活跃只看新采样行、单位换算、身份与水位。
 */
class NewareMysqlAdapterContractTest
{
    private static final String ADAPTER_CODE = "neware76";
    private static final long DEV_UID = 210036L;
    private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);
    private static final Clock CLOCK = Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC);

    private static final String ACTIVE_TABLE = "data_210036_2026_09_18";

    private NewareMysqlAdapter adapter(FakeVendorDatabase database)
    {
        AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE,
                NewareMysqlAdapter.ADAPTER_TYPE, "26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true,
                AdapterCapabilities.observeOnly(), Map.of("database", "bts63", "user", "visitor"));
        // 缓存关闭，让每个断言都读到假厂商库的当前内容。
        return new NewareMysqlAdapter(config, database.dataSource(), () -> { }, CLOCK, 0);
    }

    /** 一台在用设备（dev_uid 210036）、一台只出现在历史数据里但不在登记表中的设备（240098）。 */
    private FakeVendorDatabase activeDeviceDatabase()
    {
        return new FakeVendorDatabase()
                .registeredChannel(DEV_UID, 1, 1)
                .registeredChannel(DEV_UID, 1, 2)
                .registeredChannel(DEV_UID, 2, 1)
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 440L, "2026-09-18 09:00:00",
                        ACTIVE_TABLE, null))
                // 已退役设备的历史数据：登记表里没有它，不得成为候选。
                .test(new FakeVendorDatabase.TestRow(240098L, 1, 1, 900L, "2021-09-30 09:00:00",
                        "data_240098_2021_09_30", null))
                .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7435L, 3, 12, "2", 300000L,
                        13733, -11889, 50350024, 1103331, 50350024, 1103331, "2026-09-28 09:54:56"))
                .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7436L, 3, 9, "4", 0L,
                        13700, 0, 0, 0, 0, 0, "2026-09-28 09:54:56"));
    }

    @Test
    void discoveryListsRegisteredChannelsWithLastDataTimeOnlyFromSampleRows()
    {
        List<DetectedChannel> channels = adapter(activeDeviceDatabase()).discoverChannels();
        assertEquals(3, channels.size());
        // 候选集合来自通道登记表：只出现在历史数据里的退役设备不出现。
        assertTrue(channels.stream().allMatch(item -> item.identity().devid() == 36));
        DetectedChannel active = channels.stream()
                .filter(item -> item.identity().chlid() == 2).findFirst().orElseThrow();
        assertEquals(new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2), active.identity());
        assertTrue(active.available());
        // 最近数据时间取该通道最新采样行的设备时间（按适配器时区解释）：Asia/Hong_Kong 09:54:56 = 01:54:56Z。
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), active.lastDataAt());
        // 没有采样行的通道不报最近数据时间：活跃只看有没有新采样行写入。
        DetectedChannel idle = channels.stream()
                .filter(item -> item.identity().chlid() == 1 && item.identity().subdevid() == 1)
                .findFirst().orElseThrow();
        assertNull(idle.lastDataAt());
    }

    @Test
    void latestTestIsPickedByStartTimeNotByTestId()
    {
        FakeVendorDatabase database = new FakeVendorDatabase()
                .registeredChannel(DEV_UID, 1, 2)
                // 测试编号更大的那条起始时间更早：按 test_id 排序会取错。
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575907L, "2026-02-20 09:00:00",
                        "data_210036_2026_02_20", null))
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575540L, "2026-09-28 09:00:00",
                        "data_210036_2026_09_28", null))
                .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_02_20", 2818575907L, 1, 2, 10L, 1, 1, "1",
                        1000L, 5000, 1000, 10, 10, 10, 10, "2026-02-20 09:00:01"))
                .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_28", 2818575540L, 1, 2, 20L, 4, 7, "2",
                        60000L, 31541, -12000, 700000, 900000, 800000, 1000000, "2026-09-28 09:01:00"));
        CabinetRealtime latest = adapter(database).latestCabinet(CHANNEL);
        assertEquals(20L, latest.sequenceId());
        assertEquals(Instant.parse("2026-09-28T01:01:00Z"), latest.deviceTime());
    }

    @Test
    void shardTableComesFromTheTestRecordInsteadOfTheDeviceTimeDate()
    {
        // 实测表名按建表日：9-28 启动的测试写进 9-18 的表。适配器必须用测试记录给出的表名。
        CabinetRealtime latest = adapter(activeDeviceDatabase()).latestCabinet(CHANNEL);
        assertEquals("2026-09-28 09:54:56", latest.deviceTimeRaw());
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), latest.deviceTime());
        assertEquals("Asia/Hong_Kong", latest.timezoneId());
        assertEquals(CLOCK.instant(), latest.serverReceivedAt());
    }

    @Test
    void sampledValuesAreConvertedWithTheFieldMeasuredFormula()
    {
        // 同秒两行的第一行承载本次实测的换算样本。
        List<CabinetRealtime> rows = adapter(activeDeviceDatabase()).samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L));
        CabinetRealtime latest = rows.get(0);
        // test_vol 13733 → 1.3733 V；test_cur -11889 → -1.1889 A；放电容量与能量除以 36000000。
        assertEquals(1.3733, latest.voltageV(), 1e-9);
        assertEquals(-1.1889, latest.currentA(), 1e-9);
        assertEquals(50350024d / 36000000d, latest.capacityAh(), 1e-9);
        assertEquals(1103331d / 36000000d, latest.energyWh(), 1e-9);
        // test_time 单位为毫秒，作为工步内相对时长输出秒。
        assertEquals(300L, latest.relativeTimeSeconds());
        assertEquals(3, latest.cycleNo());
        assertEquals(12, latest.stepNo());
        assertEquals("2", latest.stepType());
        assertEquals("neware:440:7435", latest.sourceRecordId());
        // 工步类型保留设备给出的数字，不按电流符号猜方向；同一秒的另一行靠工步与序号区分。
        assertEquals(9, rows.get(1).stepNo());
    }

    @Test
    void onlineFactIsSourceReachabilityAndChannelWorkStatusStaysUnobservable()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(CHANNEL));
        RealtimeSnapshot snapshot = snapshots.get(CHANNEL);
        assertEquals(DeviceConnectionStatus.ONLINE, snapshot.connectionStatus());
        // 读库路线下通道工作状态不可观测，不按数据行推断。
        assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshot.workStatus());
        assertFalse(adapter.capabilities().control());
        // 辅助测量来自导入，不来自新威辅助表族。
        assertNull(adapter.latestCollector(ADAPTER_CODE, "anything"));
        assertEquals(NewareMysqlAdapter.SAMPLE_INTERVAL_MS, adapter.persistenceIntervalMillis(CHANNEL));
    }

    @Test
    void samplesSinceKeepsEveryRowInTheSameSecondAndDropsRowsAtOrBeforeTheWatermark()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        SampleWatermark watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L);
        List<CabinetRealtime> rows = adapter.samplesSince(CHANNEL, watermark);
        // 同一绝对秒内的两行都保留，并且靠协议序号区分。
        assertEquals(2, rows.size());
        assertEquals(List.of(7435L, 7436L), rows.stream().map(CabinetRealtime::sequenceId).toList());
        assertEquals(List.of("neware:440:7435", "neware:440:7436"),
                rows.stream().map(CabinetRealtime::sourceRecordId).toList());
        // 严格大于水位：等于水位的行不返回。
        assertEquals(1, adapter.samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7435L)).size());
        // 水位之后没有新行时返回空。
        assertTrue(adapter.samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)).isEmpty());
    }

    @Test
    void samplesWithoutWatermarkReturnNothingSoDeviceHistoryNeverEntersTheTest()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        assertTrue(adapter.samplesSince(CHANNEL, null).isEmpty());
    }

    @Test
    void emptyWatermarkMeansTheChannelHadNoRowsSoEveryLaterRowBelongsToTheReception()
    {
        // 边界时该通道在来源里没有任何行（例如通道尚未开始测试），之后出现的行全部属于本次接收。
        List<CabinetRealtime> rows = adapter(activeDeviceDatabase())
                .samplesSince(CHANNEL, new SampleWatermark(null, 0L));
        assertEquals(2, rows.size());
    }

    @Test
    void watermarkIsTheLatestRowOfTheLatestTest()
    {
        SampleWatermark watermark = adapter(activeDeviceDatabase()).currentWatermark(CHANNEL);
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), watermark.deviceTime());
        assertEquals(7436L, watermark.sequenceId());
        // 通道没有测试记录时水位为空，接收路径不把设备历史并入本测试。
        NewareMysqlAdapter empty = adapter(new FakeVendorDatabase().registeredChannel(DEV_UID, 1, 2));
        assertNull(empty.currentWatermark(CHANNEL).deviceTime());
    }

    @Test
    void rejectsChannelIdentitiesThatBelongToAnotherInstance()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        ChannelIdentity foreign = new ChannelIdentity("other-source", 21, 36, 1, 2);
        org.junit.jupiter.api.Assertions.assertThrows(RuntimeException.class, () -> adapter.latestCabinet(foreign));
        assertTrue(adapter.discoverChannels().stream()
                .allMatch(item -> ADAPTER_CODE.equals(item.identity().adapterCode())));
    }
}