package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.Date;
import org.junit.jupiter.api.Test;
import com.ruoyi.common.exception.ServiceException;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.service.DeviceAdapterErrorCodes;
import com.battery.business.service.DeviceIngestErrorCodes;
import org.mockito.Mockito;

/**
 * 推式来源接收端点的判定口径：凭据、来源状态与载荷不合法时**整批不落库**（取数程序因此不推进水位），
 * 合法批次里厂商原始行原样落地、身份与时间由后端解析、重复推送幂等、同身份不同内容不覆盖。
 */
class DeviceIngestServiceTest
{
    private static final String ADAPTER = "j22";
    private static final String TOKEN = "j22-token";
    /** 现场实测的一行：FILETIME 134349692000000000 等于 2026-09-27 07:53:20 UTC（北京 15:53）。 */
    private static final String SAMPLE_BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Schedule_Index":2818575540,
               "Test_Time":600.0,"Step_Time":540.0,"Step_Index":4,"Cycle_Index":1,
               "Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"DischargeCapacity":0.0,
               "ChargeEnergy":0.3,"DischargeEnergy":0.0,"IR":0.01,"Power":0.63,"Tempbox":25.4}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private final Clock clock = Clock.fixed(Instant.parse("2026-09-27T09:00:00Z"), ZoneOffset.UTC);
    private final DeviceIngestServiceImpl service =
            new DeviceIngestServiceImpl(adapters, landing, clock, key -> TOKEN);

    @Test
    void unknownSourceIsRejectedWithoutWritingAnything()
    {
        var failure = assertThrows(ServiceException.class, () -> service.ingest("mystery", TOKEN, "{}"));

        assertEquals(DeviceAdapterErrorCodes.NOT_FOUND, failure.getCode());
        assertTrue(landing.rows().isEmpty());
    }

    @Test
    void sourceThatIsNotAPushSourceOrIsDisabledIsRejectedWithoutWritingAnything()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("NEWARE_MYSQL", "0"));
        var wrongType = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
        assertEquals(DeviceAdapterErrorCodes.UNSUPPORTED_TYPE, wrongType.getCode());

        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "1"));
        var disabled = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
        assertEquals(DeviceIngestErrorCodes.SOURCE_DISABLED, disabled.getCode());

        assertTrue(landing.rows().isEmpty());
    }

    /** 凭据不入库：实例配置里只有变量名，取值缺失同样按凭据错误拒绝（失败关闭）。 */
    @Test
    void wrongMissingOrUnresolvableTokenIsRejectedWithoutWritingAnything()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, "wrong", SAMPLE_BATCH)).getCode());
        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, null, SAMPLE_BATCH)).getCode());

        DeviceIngestServiceImpl withoutCredential = new DeviceIngestServiceImpl(adapters, landing, clock,
                key -> { throw new IllegalStateException("缺少来源凭据：" + key); });
        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> withoutCredential.ingest(ADAPTER, TOKEN, SAMPLE_BATCH))
                        .getCode());

        assertTrue(landing.rows().isEmpty());
    }

    /** 请求体承载厂商原始行：不含已换算的数值，单位换算、时间换算、身份映射与幂等键都在后端完成。 */
    @Test
    void rawVendorRowLandsWithResolvedIdentityTimeAndUnscaledMeasurements()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        var result = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);

        assertEquals(1, result.accepted());
        TDeviceIngestRecord row = landing.rows().get(0);
        // 身份映射：22 所没有设备类型与单元维度，一台电脑登记为一个柜，通道号就是厂商通道号。
        assertEquals(0, row.getDevtype());
        assertEquals(0, row.getDevid());
        assertEquals(1, row.getSubdevid());
        assertEquals(26, row.getChlid());
        // 时间换算：FILETIME → UTC，同时保留设备给出的原始时间文本。
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), row.getDeviceTimeUtc().toInstant());
        assertEquals("134349692000000000", row.getDeviceTimeRaw());
        // 幂等键：库/表 + Data_Point（22 所每个测试新建一个库，Data_Point 在每个测试内重新编号）。
        assertEquals(420L, row.getRecordSequence());
        assertEquals("DB_20260927_155209/Ws26_IV_0:420", row.getRecordKey());
        assertEquals("DB_20260927_155209/Ws26_IV_0", row.getSourceRef());
        // 测量值按物理单位直接落库，不做缩放：1.26778 V 就是 1.26778 V。
        assertEquals(0, new java.math.BigDecimal("1.26778").compareTo(row.getVoltageV()));
        assertEquals(0, new java.math.BigDecimal("0.5").compareTo(row.getCurrentA()));
        assertEquals(0, new java.math.BigDecimal("0.12").compareTo(row.getChargeCapacityAh()));
        assertEquals(1, row.getCycleNo());
        assertEquals(4, row.getStepNo());
        assertEquals(0, new java.math.BigDecimal("540.0").compareTo(row.getRelativeTimeSeconds()));
        // 原始行的键值原样保留，可供回放与核对。
        assertTrue(row.getPayloadJson().contains("Data_Point"));
        assertTrue(row.getPayloadJson().contains("\"Schedule_Index\":2818575540"));
    }

    @Test
    void repeatedPushIsIdempotentAndConflictingContentIsReportedWithoutOverwriting()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
        // 同一批再推一次：同一条记录被幂等忽略，采样行仍是一行。
        var repeat = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);
        assertEquals(0, repeat.accepted());
        assertEquals(1, repeat.ignored());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());

        // 同身份但内容不同：报冲突且不覆盖已有事实。
        var conflict = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH.replace("1.26778", "1.99"));
        assertEquals(1, conflict.conflicts());
        assertEquals(0, conflict.accepted());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
        assertEquals(0, new java.math.BigDecimal("1.26778")
                .compareTo(landing.samples(ADAPTER, 0, 0, 1, 26).get(0).getVoltageV()));
    }

    @Test
    void topologyHeartbeatAndSamplesShareTheSameLandingTable()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        service.ingest(ADAPTER, TOKEN, "{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26},{\"channel\":18}]}");
        // 心跳只刷新存活事实，不产生数据行。
        var heartbeat = service.ingest(ADAPTER, TOKEN, "{\"kind\":\"HEARTBEAT\"}");
        assertEquals(0, heartbeat.accepted());
        assertEquals(3, landing.rows().size());
        assertEquals(clock.instant(), landing.selectHeartbeatAt(ADAPTER).toInstant());

        // 收到采样行同样是"程序还在推"的证据：接受请求就会刷新心跳，不依赖心跳消息本身。
        Date before = landing.selectHeartbeatAt(ADAPTER);
        assertNotNull(before);
        assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
        assertEquals(4, landing.rows().size());
        assertEquals(before, landing.selectHeartbeatAt(ADAPTER));
        // 拓扑上报的通道清单是长期声明，重复上报幂等；采样行按事件逐行落库。
        assertEquals(2, landing.selectTopologyChannels(ADAPTER).size());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
    }

    @Test
    void malformedBatchIsRejectedWithTheOffendingRowNamed()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        for (String body : new String[] {
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"rows\":[]}",
                "{\"kind\":\"SAMPLE\",\"channel\":26,\"rows\":[]}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
                        + "\"rows\":[{\"Voltage\":1.2}]}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
                        + "\"rows\":[{\"Data_Point\":1}]}",
                "{\"kind\":\"SOMETHING_ELSE\"}",
                "not-json" })
        {
            var failure = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, body), body);
            assertEquals(DeviceIngestErrorCodes.PAYLOAD_INVALID, failure.getCode(), body);
        }
        assertTrue(landing.rows().isEmpty());
    }

    private static TDeviceAdapter adapter(String type, String status)
    {
        TDeviceAdapter adapter = new TDeviceAdapter();
        adapter.setAdapterId(7L);
        adapter.setAdapterCode(ADAPTER);
        adapter.setAdapterName("22 所那台");
        adapter.setAdapterType(type);
        adapter.setStatus(status);
        adapter.setHost("127.0.0.1");
        adapter.setPort(8080);
        adapter.setTimezoneId("Asia/Shanghai");
        adapter.setSourceConfig("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
        return adapter;
    }
}