package com.battery.business.runtime;

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

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.UUID;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ScriptUtils;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.vo.TDeviceIngestChannel;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

/**
 * 推式来源落地表的迁移与落库语义（生产 Mapper + 迁移脚本 + H2 内存库）：
 * 来源行身份让重复推送幂等、同身份不同内容不覆盖；心跳行按来源覆盖接收时间；
 * 拓扑上报的通道清单带上自己的最近数据时间；水位按 (设备时间, 协议序号) 取严格更晚的行。
 */
public class PushIngestLandingMigrationContractTest
{
    private static final String PATH = "db/migration-final/V24__push_ingest_landing_table.sql";
    private static final String ADAPTER = "j22";
    private static final Instant RECEIVED_AT = Instant.parse("2026-09-27T09:20:00Z");

    private java.sql.Connection keepAlive;
    private JdbcTemplate jdbc;
    private TDeviceIngestRecordMapper mapper;

    @BeforeEach void setUp() throws Exception
    {
        var source = new UnpooledDataSource("org.h2.Driver",
                "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
        keepAlive = source.getConnection();
        jdbc = new JdbcTemplate(source);
        // H2 无 MySQL 的 HEX()；测试中补一个等价别名，生产 SQL 仍在 MySQL 上验证。
        jdbc.execute("create alias HEX for 'com.battery.business.runtime.PushIngestLandingMigrationContractTest.hexOf'");
        ScriptUtils.executeSqlScript(keepAlive, new ClassPathResource(PATH));

        SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
        factory.setDataSource(source);
        factory.setTypeAliasesPackage("com.battery.business.domain");
        factory.setMapperLocations(new ClassPathResource("mapper/battery/TDeviceIngestRecordMapper.xml"));
        mapper = new SqlSessionTemplate(factory.getObject()).getMapper(TDeviceIngestRecordMapper.class);
    }

    @AfterEach void closeDatabase() throws Exception { if (keepAlive != null) keepAlive.close(); }

    @Test
    void migrationCreatesTheLandingTableWithoutWritingAnyData()
    {
        String migration = resource(PATH);
        // 迁移只建表：不导入演示数据、不改写任何既有表。
        assertFalse(migration.matches("(?s).*(INSERT|UPDATE|DELETE|DROP)\\b.*"));
        assertTrue(migration.contains("UNIQUE KEY uk_ingest_record (adapter_code, payload_kind, record_key)"));

        assertEquals(0, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
        // 落地表保留厂商原始载荷，且测量列可以缺列（设备没给就是空，不写零代替缺失）。
        jdbc.update("insert into t_device_ingest_record(adapter_code, payload_kind, record_key, payload_json,"
                + " content_sha256, received_at_utc) values(?,?,?,?,?,?)", ADAPTER, "SAMPLE", "k",
                "{\"Voltage\":1.2}", new byte[32], Timestamp.from(RECEIVED_AT));
        assertEquals(1, jdbc.queryForObject(
                "select count(*) from t_device_ingest_record where voltage_v is null", Integer.class));
    }

    @Test
    void repeatedPushIsIgnoredBySourceRowIdentityAndDifferentContentIsKept()
    {
        TDeviceIngestRecord first = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
        assertEquals(1, mapper.insertIngestRecord(first));

        // 同身份同内容：幂等忽略，库里仍是一行。
        TDeviceIngestRecord repeat = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
        assertEquals(0, mapper.insertIngestRecord(repeat));
        assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));

        // 同身份不同内容：内容摘要不同，调用方据此报冲突；既有那一行不被覆盖。
        TDeviceIngestRecord conflict = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}");
        assertEquals(0, mapper.insertIngestRecord(conflict));
        assertEquals(HexFormat.of().formatHex(first.getContentSha256()),
                mapper.selectIngestRecordHashByKey(ADAPTER, "SAMPLE", first.getRecordKey()));
        // H2 的 JSON 列会再引一层，所以只断内容；既有那一行没有被冲突推送覆盖。
        assertTrue(jdbc.queryForObject("select payload_json from t_device_ingest_record", String.class)
                .contains("Voltage"));
    }

    @Test
    void heartbeatRowIsUpsertedPerSourceSoAHeartbeatNeverGrows()
    {
        mapper.upsertHeartbeat(heartbeat(RECEIVED_AT));
        assertEquals(Timestamp.from(RECEIVED_AT).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());

        // 下一轮心跳只推进接收时间：它表示"取数程序还在推"，不是设备事实，没有保留历史的意义。
        Instant later = RECEIVED_AT.plusSeconds(60);
        mapper.upsertHeartbeat(heartbeat(later));
        assertEquals(Timestamp.from(later).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());
        assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
        // 从未收到推送的来源没有心跳事实。
        assertNull(mapper.selectHeartbeatAt("other"));
    }

    @Test
    void topologyChannelsCarryTheirOwnLatestSampleTimeAndNeverExpire()
    {
        mapper.insertIngestRecord(topology(26));
        mapper.insertIngestRecord(topology(18));
        // 18 号只有更早的采样行；26 号有两行。
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
                Instant.parse("2026-09-27T07:52:20Z")));
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z")));
        mapper.insertIngestRecord(sample(18, "DB_20260923_161840/Ws18_IV_0:5", 5L, "{\"Voltage\":1.1}",
                Instant.parse("2026-09-23T08:18:40Z")));

        List<TDeviceIngestChannel> channels = mapper.selectTopologyChannels(ADAPTER);
        assertEquals(List.of(18, 26), channels.stream().map(TDeviceIngestChannel::getChlid).toList());
        assertEquals(Instant.parse("2026-09-23T08:18:40Z"), channels.get(0).getLastDataAt().toInstant());
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), channels.get(1).getLastDataAt().toInstant());
        // 没有采样行的通道不给"最近数据时间"：这个事实是观测到的，不按上报时间代替。
        mapper.insertIngestRecord(topology(21));
        assertNull(mapper.selectTopologyChannels(ADAPTER).stream()
                .filter(channel -> channel.getChlid() == 21).findFirst().orElseThrow().getLastDataAt());
    }

    @Test
    void samplesSinceReturnsOnlyRowsStrictlyAfterTheWatermarkIncludingSameSecondRows()
    {
        TDeviceIngestRecord before = sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
                Instant.parse("2026-09-27T07:52:20Z"));
        TDeviceIngestRecord sameSecond = sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z"));
        TDeviceIngestRecord sameSecondLaterSequence = sample(26, "DB_20260927_155209/Ws26_IV_0:13", 13L,
                "{\"Voltage\":1.4}", Instant.parse("2026-09-27T07:53:20Z"));
        TDeviceIngestRecord after = sample(26, "DB_20260927_155209/Ws26_IV_0:14", 14L, "{\"Voltage\":1.5}",
                Instant.parse("2026-09-27T07:54:20Z"));
        for (TDeviceIngestRecord row : List.of(before, sameSecond, sameSecondLaterSequence, after))
            mapper.insertIngestRecord(row);

        // 水位落在一行上：那一行自己不算，严格更晚的才进本次接收。
        List<TDeviceIngestRecord> rows = mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26,
                Date.from(sameSecond.getDeviceTimeUtc().toInstant()), 12L, 500);
        assertEquals(List.of(13L, 14L), rows.stream().map(TDeviceIngestRecord::getRecordSequence).toList());

        // 同一绝对秒内的多行各自成立：水位为空表示边界时刻什么也没有，之后的行全部属于本次接收。
        assertEquals(4, mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26, null, 0L, 500).size());
        // 其他通道与来源的行不混进来。
        assertTrue(mapper.selectSamplesSince("other", 0, 0, 1, 26, null, 0L, 500).isEmpty());
    }

    @Test
    void latestSampleDrivesBothRealtimeObservationAndWatermark()
    {
        assertNull(mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26));
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z")));
        TDeviceIngestRecord latest = mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26);
        assertNotNull(latest);
        assertEquals(12L, latest.getRecordSequence());
        assertEquals("2026-09-27 15:53:20", latest.getDeviceTimeRaw());
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), latest.getDeviceTimeUtc().toInstant());
        // 原始行逐字留在落地表里，回放与核对解释逻辑都靠它（H2 的 JSON 列会再引一层，所以只断内容）。
        assertTrue(latest.getPayloadJson().contains("Voltage"));
    }

    /** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
    public static String hexOf(byte[] value)
    {
        return value == null ? null : HexFormat.of().formatHex(value);
    }

    /** 内容摘要固定 32 字节：列是 BINARY(32)，短值会被补齐，比较时对不上。 */
    private static byte[] digest(String content)
    {
        try
        {
            return java.security.MessageDigest.getInstance("SHA-256")
                    .digest(content.getBytes(StandardCharsets.UTF_8));
        }
        catch (Exception exception)
        {
            throw new IllegalStateException(exception);
        }
    }

    private TDeviceIngestRecord sample(String recordKey, long sequence, String payload)
    {
        return sample(26, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
    }

    private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload)
    {
        return sample(channel, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
    }

    private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload,
            Instant deviceTime)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
        value.setRecordSequence(sequence);
        value.setRecordKey(recordKey);
        value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
        value.setDeviceTimeUtc(Timestamp.from(deviceTime));
        value.setDeviceTimeRaw("2026-09-27 15:53:20");
        value.setTimezoneId("Asia/Shanghai");
        value.setPayloadJson(payload);
        value.setContentSha256(digest(payload));
        value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
        return value;
    }

    private TDeviceIngestRecord topology(int channel)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_TOPOLOGY);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
        value.setRecordSequence(0L);
        value.setRecordKey("topology:" + channel);
        value.setPayloadJson("{\"channel\":" + channel + "}");
        value.setContentSha256(digest("topology:" + channel));
        value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
        return value;
    }

    private TDeviceIngestRecord heartbeat(Instant receivedAt)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_HEARTBEAT);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(0);
        value.setRecordSequence(0L);
        value.setRecordKey("heartbeat");
        value.setPayloadJson("{}");
        value.setContentSha256(new byte[32]);
        value.setReceivedAtUtc(Timestamp.from(receivedAt));
        return value;
    }

    private static String resource(String path)
    {
        try (InputStream stream = PushIngestLandingMigrationContractTest.class.getClassLoader()
                .getResourceAsStream(path))
        {
            if (stream == null) throw new IllegalStateException("缺少迁移文件：" + path);
            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (java.io.IOException exception)
        {
            throw new IllegalStateException("无法读取迁移文件：" + path, exception);
        }
    }
}