package com.battery.business.service.impl;

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.Clock;
import java.time.Duration;
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.DeviceAdapterException;
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.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.mapper.TDeviceAdapterMapper;
import org.mockito.Mockito;

/**
 * 推式来源适配器的观测契约：落地行投影成统一观测（身份、时间、单位、幂等身份），
 * 发现候选带最近数据时间且心跳缺失即明确失败，水位按 (设备时间, 协议序号) 取严格更晚的行。
 * 原始行由接收端点解析，所以这里只喂落地行，适配器里不应出现任何厂商列名。
 */
class PushIngestAdapterContractTest
{
    private static final String ADAPTER = "j22";
    private static final String TOKEN = "j22-token";
    /** 前两行同一个绝对秒（08:03:20 UTC）但协议序号不同；第三行晚一分钟。 */
    private static final String BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Step_Time":540.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"ChargeEnergy":0.3},
              {"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31},
              {"record_time":"134349698000000000","Data_Point":422,"Step_Time":600.0,"Step_Index":5,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":-0.4,"DischargeCapacity":0.05,"DischargeEnergy":0.1}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private static final Instant NOW = Instant.parse("2026-09-27T09:00:00Z");

    @Test
    void topologyPushMakesChannelsCandidatesWithTheirOwnLatestDataTime()
    {
        // 拓扑行里可以带 channelNo：来源自己软件里显示的通道号（现场核对：22 所界面号 = 表名号 + 1）
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26,\"channelNo\":27},{\"channel\":18}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        List<DetectedChannel> channels = adapter.discoverChannels();

        assertEquals(List.of(18, 26), channels.stream().map(channel -> channel.identity().chlid()).toList());
        DetectedChannel active = channels.stream().filter(channel -> channel.identity().chlid() == 26).findFirst()
                .orElseThrow();
        // 最近数据时间来自该通道自己的采样行，不是拓扑上报时间。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), active.lastDataAt());
        assertTrue(active.available());
        // 候选的机器地址取实例配置：登记时会被带进检测柜的 IP 字段。
        assertEquals("127.0.0.1", active.serverIp());
        // 来源自己的通道叫法随候选一起交出（界面与登记用它显示），身份仍是表名里的号。
        assertEquals("27", active.sourceDisplayNo());
        assertEquals(26, active.identity().chlid());
        // 登记范围由操作员决定：没有采样行的通道照样出现在候选里，只是没有最近数据时间。
        assertNull(channels.stream().filter(channel -> channel.identity().chlid() == 18).findFirst()
                .orElseThrow().lastDataAt());
        // 来源没报叫法的通道保持为空，界面回退到「单元-通道」。
        assertNull(channels.stream().filter(channel -> channel.identity().chlid() == 18).findFirst()
                .orElseThrow().sourceDisplayNo());
    }

    /** 心跳缺失或过期即来源不可达：发现路径明确失败，业务侧据此把已登记设备置为离线并撤下旧快照。 */
    @Test
    void missingOrStaleHeartbeatFailsDiscoveryInsteadOfReturningStaleValues()
    {
        // 还没收到过任何推送：这是"取数程序没起来"这一条可见的失败，不是沉默的空清单。
        DeviceAdapterException missing = assertThrows(DeviceAdapterException.class, () -> adapter(NOW).discoverChannels());
        assertEquals("SOURCE_OFFLINE", missing.failureCode());

        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        // 收到推送即刷新心跳；容忍窗口是三个轮询周期，一轮没推不算离线。
        assertEquals(1, adapter(NOW.plus(Duration.ofSeconds(179))).discoverChannels().size());
        // 连丢三轮才算离线，且不必等候选过期。
        DeviceAdapterException stale = assertThrows(DeviceAdapterException.class,
                () -> adapter(NOW.plus(Duration.ofSeconds(181))).discoverChannels());
        assertEquals("SOURCE_OFFLINE", stale.failureCode());
    }

    @Test
    void samplesAreProjectedWithRawTimeUnscaledUnitsAndSourceRowIdentity()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        CabinetRealtime latest = adapter.latestCabinet(channel());

        // 时间按厂商格式换算为 UTC，同时保留原始值与来源时区；单位是物理单位，不做缩放。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), latest.deviceTime());
        assertEquals("134349698000000000", latest.deviceTimeRaw());
        assertEquals("Asia/Shanghai", latest.timezoneId());
        assertEquals(1.2702, latest.voltageV(), 1e-9);
        assertEquals(-0.4, latest.currentA(), 1e-9);
        assertEquals(600L, latest.relativeTimeSeconds());
        assertEquals(1, latest.cycleNo());
        assertEquals(5, latest.stepNo());
        // 放电容量的取法按电流方向，与直连来源同一口径；来源行身份供幂等与冲突判定。
        assertEquals(0.05, latest.capacityAh(), 1e-9);
        assertEquals("push:DB_20260927_155209/Ws26_IV_0:422", latest.sourceRecordId());

        // 实时快照：通道工作状态在读落地表路线下不可观测（ADR-0063）。
        Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(channel()));
        assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshots.get(channel()).workStatus());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), snapshots.get(channel()).deviceTime());
    }

    @Test
    void watermarkAndSamplesSinceKeepSameSecondRowsDistinctAndSkipNothingBeforeTheBoundary()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        // 水位是落地表里该通道最新一行：设备绝对时间与协议序号。
        assertEquals(new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L),
                adapter.currentWatermark(channel()));

        // 只取严格晚于水位的行：同一绝对秒内的多行靠协议序号各自成立。
        List<CabinetRealtime> rows = adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T07:53:20Z"), 420L));
        assertEquals(List.of(421L, 422L), rows.stream().map(CabinetRealtime::sequenceId).toList());
        assertEquals(0, adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L)).size());
        // 同一绝对秒内序号更大的行仍然成立：水位是二元组而不是秒。
        assertEquals(List.of(422L), adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 421L)).stream()
                .map(CabinetRealtime::sequenceId).toList());
        // 调用方尚未取得水位时不把来源里的历史并入本次接收。
        assertTrue(adapter.samplesSince(channel(), null).isEmpty());
        // 水位为空的二元组表示边界时刻来源里什么也没有：之后出现的行全部属于本次接收。
        assertEquals(3, adapter.samplesSince(channel(), new SampleWatermark(null, 0L)).size());
    }

    @Test
    void emptySourceHasNoWatermarkAndNoRealtimeObservation()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        PushIngestAdapter adapter = adapter(NOW);

        assertEquals(new SampleWatermark(null, 0L), adapter.currentWatermark(channel()));
        assertNull(adapter.latestCabinet(channel()));
        assertTrue(adapter.inquire(Set.of(channel())).isEmpty());
    }

    /** 采集器辅助测量按 ADR-0064 逐来源确认；22 所环境量的列语义未确认，因此不交出测量值。 */
    @Test
    void collectorAndStepCycleObservationsStayEmptyForThisStorageForm()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
        assertTrue(adapter.stepRecords(channel()).isEmpty());
        assertTrue(adapter.cycleRecords(channel()).isEmpty());
        // 落地表在本地，取数不产生厂商负载，所以按展示节奏取数；厂商写库节奏仍是 60 秒。
        assertEquals(5000L, adapter.persistenceIntervalMillis(channel()));
        AdapterCapabilities capabilities = adapter.capabilities();
        assertTrue(capabilities.discovery() && capabilities.realtimeObservation() && !capabilities.control());
    }

    @Test
    void channelIdentityFromAnotherInstanceFailsInsteadOfFallingBack()
    {
        PushIngestAdapter adapter = adapter(NOW);

        DeviceAdapterException failure = assertThrows(DeviceAdapterException.class,
                () -> adapter.currentWatermark(new ChannelIdentity("neware76", 21, 36, 1, 2)));
        assertTrue(failure.getMessage().contains("neware76"));
        assertFalse(adapter.adapterCode().isEmpty());
    }

    private ChannelIdentity channel()
    {
        return new ChannelIdentity(ADAPTER, 0, 0, 1, 26);
    }

    private PushIngestAdapter adapter(Instant now)
    {
        return new PushIngestAdapter(config(), landing, Clock.fixed(now, ZoneOffset.UTC));
    }

    private static AdapterInstanceConfig config()
    {
        return new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE, "127.0.0.1", 8080, 60000,
                "Asia/Shanghai", true, AdapterCapabilities.observeOnly(),
                Map.of("tokenEnv", "BATTERY_INGEST_TOKEN"));
    }

    private void ingest(String body)
    {
        TDeviceAdapter source = new TDeviceAdapter();
        source.setAdapterCode(ADAPTER);
        source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
        source.setStatus("0");
        source.setTimezoneId("Asia/Shanghai");
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
        new DeviceIngestServiceImpl(adapters, landing, Clock.fixed(NOW, ZoneOffset.UTC), key -> TOKEN)
                .ingest(ADAPTER, TOKEN, body);
    }
}