package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.Date;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.TTest;
import com.battery.business.domain.TTestBattery;
import com.battery.business.domain.TTestChannelBinding;
import com.battery.business.domain.TTestRealtimeRecord;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.mapper.TTestMapper;
import com.battery.business.service.ITTestService;

/**
 * 推式来源接进业务接收路径端到端：取数程序推来的厂商原始行 → 落地表 → 适配器读回统一观测 →
 * 接收中测试按水位逐行保存。
 *
 * <p>用的是同一个观测契约，所以这里不需要任何推送专属的接收代码：落地行按事件落库、
 * 轮询槽位为空（同秒多行因此各自成立），来源行身份承担幂等与冲突判定。
 */
class PushIngestReceptionContractTest
{
    private static final String ADAPTER = "j22";
    private static final Instant START = Instant.parse("2026-09-27T08:00:00Z");
    /** 现场实测的一批：两行落在同一个绝对秒（08:03:20 UTC），协议序号不同。 */
    private static final String BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"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":4,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":0.5,"ChargeCapacity":0.122,"ChargeEnergy":0.32}]}
            """;
    /** 同一批里带上已声明的环境量列：辅助测量的值取自这一行自己，不做任何缩放。 */
    private static final String AUXILIARY_BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"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,
               "Tempbox":26.5},
              {"record_time":"134349704000000000","Data_Point":422,"Step_Time":660.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":0.5,"ChargeCapacity":0.122,"ChargeEnergy":0.32,
               "Tempbox":26.5}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private final TestClock clock = new TestClock(START);

    @Test
    void pushedRawRowsBecomeTestFactsWithoutPollingSlots()
    {
        PushIngestAdapter adapter = ingestPushSource();
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        TTest test = receivingTest();
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(tests.selectTestById(1L)).thenReturn(test);
        TestDataReceptionServiceImpl service = new TestDataReceptionServiceImpl(mapper, tests, adapter, clock);

        // 第一次接收只取水位：开始接收之前的行即使躺在落地表里也不进本测试。
        landing.insertIngestRecord(boundaryRow("DB_20260927_155209/Ws26_IV_0:420", 420L));
        service.captureReceivedData();
        verify(mapper, never()).insertRealtimeRecord(any());

        // 之后推来的行逐行落库：同秒两行各自成立，因为槽位为空（推送记录按事件落库，不占轮询槽位）。
        ingest(BATCH);
        clock.advance(Duration.ofSeconds(60));
        service.captureReceivedData();

        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(mapper, times(2)).insertRealtimeRecord(rows.capture());
        assertEquals(List.of("push:DB_20260927_155209/Ws26_IV_0:421",
                "push:DB_20260927_155209/Ws26_IV_0:422"),
                rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
        assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
        assertEquals("j22:0:0:1:26", rows.getAllValues().get(0).getSourceIdentity());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"),
                rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
        assertEquals("Asia/Shanghai", rows.getAllValues().get(0).getCabinetAdapterTimezoneId());
        // 测量值按物理单位直接使用，不做缩放。
        assertEquals(0, new java.math.BigDecimal("1.2701")
                .compareTo(rows.getAllValues().get(0).getVoltageV()));
        assertEquals(0, new java.math.BigDecimal("0.121")
                .compareTo(rows.getAllValues().get(0).getCapacityAh()));

        // 同一批再推一次不会重复落库：来源行身份让接收幂等。
        ingest(BATCH);
        clock.advance(Duration.ofSeconds(60));
        service.captureReceivedData();
        verify(mapper, times(2)).insertRealtimeRecord(any());
    }

    /**
     * 采集器辅助测量按 ADR-0064 由同一来源提供：它作为扩展测量记录进入实时记录，
     * **不跟检测柜记录伪造配对**——检测柜走各自的设备数据行（槽位为空），采集器走轮询槽位行（柜端字段为空），
     * 两边各自保留自己的设备时间，绑定角色决定它进哪个指标。
     */
    @Test
    void collectorAuxiliaryMeasurementBecomesItsOwnRealtimeRowWithoutFakingAPairing()
    {
        PushIngestAdapter adapter = ingestPushSource("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\",\"auxiliaryMetrics\":"
                + "[{\"vendorChannel\":26,\"channelCode\":\"S-01\",\"column\":\"Tempbox\","
                + "\"metricCode\":\"TEMPERATURE\",\"unit\":\"℃\"}]}");
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        TTest test = receivingTest();
        TTestBattery battery = test.getBatteries().get(0);
        battery.setCollectorId(7L);
        com.battery.business.domain.TTestCollectorChannel collectorChannel =
                new com.battery.business.domain.TTestCollectorChannel();
        collectorChannel.setChannelCode("S-01");
        collectorChannel.setMeasurementCode("PRIMARY_TEMPERATURE");
        battery.setCollectorChannels(List.of(collectorChannel));
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(tests.selectTestById(1L)).thenReturn(test);
        when(mapper.selectCollectorCode(7L)).thenReturn("j22-collector");
        TestDataReceptionServiceImpl service = new TestDataReceptionServiceImpl(mapper, tests, adapter, clock);

        landing.insertIngestRecord(boundaryRow("DB_20260927_155209/Ws26_IV_0:420", 420L));
        service.captureReceivedData();
        ingest(AUXILIARY_BATCH);
        clock.advance(Duration.ofSeconds(60));
        service.captureReceivedData();

        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(mapper, times(3)).insertRealtimeRecord(rows.capture());
        List<TTestRealtimeRecord> cabinetRows = rows.getAllValues().subList(0, 2);
        TTestRealtimeRecord collectorRow = rows.getAllValues().get(2);
        // 检测柜事实不占轮询槽位，采集器事实占槽位：两条路径互不覆盖。
        assertTrue(cabinetRows.stream().allMatch(row -> row.getCaptureSlotUtc() == null));
        assertTrue(collectorRow.getCaptureSlotUtc() != null);
        // 采集器行不携带检测柜设备时间与检测柜身份，来源行身份里也没有 cabinet 段。
        assertEquals(null, collectorRow.getCabinetDeviceTimeUtc());
        assertEquals(null, collectorRow.getCabinetChlid());
        assertTrue(collectorRow.getSourceRecordId().indexOf("cabinet:") < 0);
        // 测量值取自这一行自己的设备时间（该通道落地表里最新一行），指标编码与单位按配置声明，
        // 固定指标落到自己的列。
        assertEquals(Instant.parse("2026-09-27T08:13:20Z"), collectorRow.getCollectorDeviceTimeUtc().toInstant());
        assertEquals("S-01", collectorRow.getCollectorChannelCode());
        assertEquals(0, new java.math.BigDecimal("26.5").compareTo(collectorRow.getTemperature()));
        assertEquals(Long.valueOf(7L), collectorRow.getCollectorId());
    }

    private PushIngestAdapter ingestPushSource()
    {
        return ingestPushSource("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
    }

    private PushIngestAdapter ingestPushSource(String sourceConfig)
    {
        TDeviceAdapter source = new TDeviceAdapter();
        source.setAdapterCode(ADAPTER);
        source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
        source.setStatus("0");
        source.setTimezoneId("Asia/Shanghai");
        source.setSourceConfig(sourceConfig);
        when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        return new PushIngestAdapter(new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE,
                "127.0.0.1", 8080, 60000, "Asia/Shanghai", true, AdapterCapabilities.observeOnly(),
                com.alibaba.fastjson2.JSON.parseObject(sourceConfig)
                        .toJavaObject(new com.alibaba.fastjson2.TypeReference<java.util.Map<String, String>>() { })),
                landing, clock);
    }

    private void ingest(String body)
    {
        new DeviceIngestServiceImpl(adapters, landing, clock, key -> "token").ingest(ADAPTER, "token", body);
    }

    /** 开始接收之前就躺在来源里的那一行：它只用来把水位定在边界时刻。 */
    private TDeviceIngestRecord boundaryRow(String recordKey, long sequence)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(26);
        value.setRecordSequence(sequence);
        value.setRecordKey(recordKey);
        value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
        value.setDeviceTimeUtc(Date.from(START.minusSeconds(60)));
        value.setDeviceTimeRaw("134349691400000000");
        value.setTimezoneId("Asia/Shanghai");
        value.setPayloadJson("{}");
        value.setContentSha256(new byte[32]);
        value.setReceivedAtUtc(Date.from(START.minusSeconds(60)));
        return value;
    }

    private static TTest receivingTest()
    {
        TTestChannelBinding binding = new TTestChannelBinding();
        binding.setBindingId(2L);
        binding.setChannelId(3L);
        binding.setAdapterCode(ADAPTER);
        binding.setAdapterTimezoneId("Asia/Shanghai");
        binding.setDevtype(0);
        binding.setDevid(0);
        binding.setSubdevid(1);
        binding.setChlid(26);
        TTestBattery battery = new TTestBattery();
        battery.setTestBatteryId(4L);
        battery.setBinding(binding);
        battery.setCollectorChannels(List.of());
        TTest test = new TTest();
        test.setTestId(1L);
        test.setTestStatus("RUNNING");
        test.setReceptionStatus("RECEIVING");
        test.setStartedAtUtc(Date.from(START));
        test.setReceptionBoundaryAtUtc(Date.from(START));
        test.setBatteries(List.of(battery));
        return test;
    }

    /** 可推进的测试时钟：接收节奏与推送节奏都按它推进。 */
    private static final class TestClock extends Clock
    {
        private Instant now;

        private TestClock(Instant now) { this.now = now; }

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

        @Override public ZoneId getZone() { return ZoneOffset.UTC; }
        @Override public Clock withZone(ZoneId zone) { return this; }
        @Override public Instant instant() { return now; }
    }
}