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.ArrayList;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
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.DeviceObservation;
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.TTestMapper;
import com.battery.business.service.ITTestService;

/**
 * 逐行接收的归属边界：水位由"开始接收"与"继续接收"这一刻来源里的最新一行决定，
 * 只取严格晚于水位的数据行；同秒多行都保留；重复同内容幂等；同身份不同内容不覆盖。
 */
class VendorSampleReceptionServiceTest
{
    private static final Instant START = Instant.parse("2026-09-28T02:00:00Z");
    private static final ChannelIdentity CHANNEL = new ChannelIdentity("neware76", 21, 36, 1, 2);
    private static final String IDENTITY = "neware76:21:36:1:2";

    @Test
    void firstPollReadsTheWatermarkWithoutWritingDeviceHistory()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);

        fixture.service.captureReceivedData();

        assertEquals(1, fixture.adapter.watermarkReads);
        assertTrue(fixture.adapter.sampleReads.isEmpty());
        verify(fixture.mapper, never()).insertRealtimeRecord(any());
    }

    @Test
    void everyRowStrictlyAfterTheWatermarkBecomesATestFactIncludingSameSecondRows()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        // 同秒两行都落库：槽位留空，来源行身份把它们区分开。
        fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"),
                sample(7438L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();

        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper, times(2)).insertRealtimeRecord(rows.capture());
        assertEquals(List.of("neware:440:7437", "neware:440:7438"),
                rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
        assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
        assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
                rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
        assertEquals("2026-09-28 09:55:56", rows.getAllValues().get(0).getCabinetDeviceTimeRaw());
        assertEquals(IDENTITY, rows.getAllValues().get(0).getSourceIdentity());
    }

    @Test
    void resumingReceptionReadsANewWatermarkSoThePauseGapIsNotBackfilled()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)),
                fixture.adapter.sampleReads);

        // 暂停后继续：接收边界推进，水位重新读取；暂停期间产生的行不再进本测试。
        fixture.test.setReceptionBoundaryAtUtc(Date.from(START.plusSeconds(600)));
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L);
        fixture.adapter.samples = List.of();
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();

        assertEquals(2, fixture.adapter.watermarkReads);
        // 取到新水位后，只追加新水位之后的行；暂停期间的空档不会被补回来。
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L),
                new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L)), fixture.adapter.sampleReads);
    }

    @Test
    void repeatedIdenticalRowIsIgnoredAndConflictingContentIsNotOverwritten()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        CabinetRealtime sample = sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56");
        fixture.adapter.samples = List.of(sample);
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        ArgumentCaptor<TTestRealtimeRecord> first = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper).insertRealtimeRecord(first.capture());

        // 同身份同内容：幂等忽略，不再插入。
        when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
                .thenReturn(HexFormat.of().formatHex(first.getValue().getContentSha256()));
        fixture.adapter.watermark = new SampleWatermark(sample.deviceTime(), sample.sequenceId());
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        verify(fixture.mapper, times(1)).insertRealtimeRecord(any());

        // 同身份不同内容：保留原事实，拒绝覆盖。
        when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
                .thenReturn("0000000000000000000000000000000000000000000000000000000000000000");
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
    }

    @Test
    void sourceWithoutWatermarkKeepsThePollingSlotPath()
    {
        Fixture fixture = fixture();
        // 不提供水位的来源（例如开发模拟器）继续按轮询槽位保存柜端事实。
        fixture.adapter.watermark = null;
        fixture.adapter.latest = sample(9L, "2026-09-28T02:00:30Z", "2026-09-28 10:00:30");

        fixture.service.captureReceivedData();

        assertEquals(1, fixture.adapter.watermarkReads);
        assertTrue(fixture.adapter.sampleReads.isEmpty());
        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper).insertRealtimeRecord(rows.capture());
        assertTrue(rows.getValue().getCaptureSlotUtc() != null);
    }

    private static CabinetRealtime sample(long sequenceId, String deviceTime, String raw)
    {
        return new CabinetRealtime(sequenceId, Instant.parse(deviceTime), -1.1889, 1.3733, 1.398611, 0.030648,
                3, 12, null, "2", Instant.parse("2026-09-28T02:01:00Z"), raw, "Asia/Hong_Kong", 300L,
                "neware:440:" + sequenceId);
    }

    private static Fixture fixture()
    {
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        AdaptiveAdapter adapter = new AdaptiveAdapter();
        TTest test = receivingTest();
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(mapper.selectCollectorCode(9L)).thenReturn("collector-9");
        when(tests.selectTestById(1L)).thenReturn(test);
        TestClock clock = new TestClock(START);
        return new Fixture(mapper, tests, adapter, test, clock,
                new TestDataReceptionServiceImpl(mapper, tests, adapter, clock));
    }

    private static TTest receivingTest()
    {
        TTestChannelBinding binding = new TTestChannelBinding();
        binding.setBindingId(2L);
        binding.setChannelId(3L);
        binding.setAdapterCode("neware76");
        binding.setAdapterTimezoneId("Asia/Hong_Kong");
        binding.setDevtype(21);
        binding.setDevid(36);
        binding.setSubdevid(1);
        binding.setChlid(2);
        TTestBattery battery = new TTestBattery();
        battery.setTestBatteryId(4L);
        battery.setCollectorId(9L);
        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 record Fixture(TTestMapper mapper, ITTestService tests, AdaptiveAdapter adapter, TTest test,
            TestClock clock, TestDataReceptionServiceImpl service) { }

    /** 只暴露水位与逐行读取的测试替身，并记录被问到的水位与逐行读取次数。 */
    private static final class AdaptiveAdapter implements TestDataObservationAdapter
    {
        private SampleWatermark watermark;
        private List<CabinetRealtime> samples = new ArrayList<>();
        private CabinetRealtime latest;
        private int watermarkReads;
        private final List<SampleWatermark> sampleReads = new ArrayList<>();

        @Override public String adapterCode() { return "neware76"; }
        @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
        @Override public List<DetectedChannel> discoverChannels() { return List.of(); }
        @Override public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
        { return Map.of(); }
        @Override public CabinetRealtime latestCabinet(ChannelIdentity channel) { return latest; }
        @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode) { return null; }
        @Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
        @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
        @Override public long persistenceIntervalMillis(ChannelIdentity channel) { return 60000L; }

        @Override public SampleWatermark currentWatermark(ChannelIdentity channel)
        {
            watermarkReads++;
            return watermark;
        }

        @Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark since)
        {
            sampleReads.add(since);
            return List.copyOf(samples);
        }
    }

    /** 可推进的测试时钟：接收节奏与厂商写库节奏都按它推进。 */
    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; }
    }
}