package com.battery.business.service.impl;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Clock;
import java.time.Instant;
import java.util.Date;
import java.util.HexFormat;
import java.util.Map;
import java.util.TreeMap;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.StringUtils;
import com.battery.device.api.credentials.LocalCredentials;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.mapper.TDeviceIngestRecordMapper;
import com.battery.business.service.DeviceAdapterErrorCodes;
import com.battery.business.service.DeviceIngestErrorCodes;
import com.battery.business.service.IDeviceIngestService;

/**
 * 推式来源的接收端点实现（ADR-0065）：校验来源凭据，把厂商原始行落进落地表。
 *
 * <p>这里也是**唯一**解释厂商列名的地方：原始行里有 22 所 IV 表的列（{@code record_time}、
 * {@code Data_Point}、{@code Voltage}、{@code eCurrent} 等），本类把它们映射成落地表的身份、时间、
 * 协议序号与测量列，同一份原始行原样存进 {@code payload_json}（键值不变，格式由 JSON 列规范化）。适配器因此只做投影、不认识厂商列名。
 *
 * <p>三条判定口径：
 * <ul>
 *   <li><b>凭据不入库</b>：端点令牌与厂商库口令共用同一套取值顺序（本地凭据文件 → 环境变量 →
 *       启动参数），实例配置里只有变量名。令牌缺失或不匹配时拒绝，且不落任何数据。</li>
 *   <li><b>时间换算</b>：22 所的 {@code record_time} 是 Windows FILETIME（100 ns 刻度，自 1601-01-01 UTC），
 *       直接换算出 UTC 时刻；原始字符串同时保留。厂商给的是绝对时间，不需要按时区再解释一次。</li>
 *   <li><b>幂等与冲突</b>：来源行身份是「库/表 + Data_Point」，重复推送被唯一键幂等忽略；
 *       同身份但内容不同只记警告、不覆盖既有事实。</li>
 * </ul>
 */
@Service
public class DeviceIngestServiceImpl implements IDeviceIngestService
{
    private static final Logger log = LoggerFactory.getLogger(DeviceIngestServiceImpl.class);
    /** 端点令牌的变量名默认值；实例配置用 {@code tokenEnv} 覆盖它。 */
    private static final String TOKEN_ENV_OPTION = "tokenEnv";
    private static final String DEFAULT_TOKEN_ENV = "BATTERY_INGEST_TOKEN";
    /** 心跳行按来源唯一，键名固定。 */
    private static final String HEARTBEAT_KEY = "heartbeat";
    /** 一批的上限：程序按批推，一次推来整表说明配置有问题，直接拒绝而不是把它读进内存。 */
    private static final int MAX_ROWS_PER_REQUEST = 2000;
    /** Windows FILETIME 的起点（1601-01-01 UTC）与 Unix 纪元之间的秒数差。 */
    private static final long FILETIME_EPOCH_DIFFERENCE_SECONDS = 11644473600L;

    private final TDeviceAdapterMapper adapters;
    private final TDeviceIngestRecordMapper records;
    private final Clock clock;
    /** 端点令牌的取值入口；测试注入固定值，生产走本地凭据文件 → 环境变量 → 启动参数。 */
    private final Function<String, String> credentials;

    @Autowired
    public DeviceIngestServiceImpl(TDeviceAdapterMapper adapters, TDeviceIngestRecordMapper records)
    {
        this(adapters, records, Clock.systemUTC(), LocalCredentials::resolve);
    }

    /** 允许测试固定接收时间与凭据取值，生产运行仍使用 UTC 系统时钟与本地凭据文件。 */
    public DeviceIngestServiceImpl(TDeviceAdapterMapper adapters, TDeviceIngestRecordMapper records, Clock clock,
            Function<String, String> credentials)
    {
        this.adapters = adapters;
        this.records = records;
        this.clock = clock;
        this.credentials = credentials;
    }

    /**
     * 接收推式来源程序的一批载荷：校验来源与令牌，按载荷类别写入落地表，并刷新该来源的心跳。
     *
     * <p>前置条件：来源已登记为启用中的推式实例，端点令牌与本地凭据一致，载荷为合法 JSON 且类别已知。
     * 整批在同一事务内落库，任一行失败则整批回滚；重复推送按来源行身份幂等忽略，内容不一致的行记为冲突
     * 而不覆盖既有事实。心跳、拓扑与采样都会刷新心跳时间。</p>
     *
     * @param adapterCode 来源实例编码，取自接收端点路径
     * @param token       端点令牌，取自请求头
     * @param body        载荷正文，含 kind 与对应的行数据
     * @return 本批的接收计数：成功写入、幂等忽略、内容冲突的行数
     * @throws ServiceException 来源不是启用的推式实例、令牌不匹配、载荷格式非法或行数超出单批上限
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public IngestResult ingest(String adapterCode, String token, String body)
    {
        // STEP 1: 校验来源是启用中的推式实例，并校验端点令牌。
        TDeviceAdapter adapter = requirePushSource(adapterCode);
        requireToken(adapter, token);
        // STEP 2: 解析载荷并按 kind 分流；未知类别直接拒绝，不按猜测处理。
        JSONObject payload = parse(body);
        String kind = payload.getString("kind");
        Instant receivedAt = clock.instant();
        Counters counters = new Counters();
        if (TDeviceIngestRecord.KIND_HEARTBEAT.equals(kind))
        {
            // 心跳没有行：它只刷新"取数程序还在推"这一条事实。
        }
        else if (TDeviceIngestRecord.KIND_TOPOLOGY.equals(kind))
        {
            storeTopology(adapterCode, adapter.getTimezoneId(), payload, receivedAt, counters);
        }
        else if (TDeviceIngestRecord.KIND_SAMPLE.equals(kind))
        {
            storeSamples(adapterCode, adapter.getTimezoneId(), payload, receivedAt, counters);
        }
        else
        {
            throw payloadInvalid("未知的载荷类别 kind：" + kind);
        }
        // STEP 3: 刷新心跳时间；每一批成功接收都是"程序还在推"的证据。
        // 每一批成功接收都刷新心跳：收到采样行同样是"程序还在推"的证据，缺一次心跳不会误判离线。
        touchHeartbeat(adapterCode, receivedAt);
        return new IngestResult(counters.accepted, counters.ignored, counters.conflicts);
    }

    private TDeviceAdapter requirePushSource(String adapterCode)
    {
        TDeviceAdapter adapter = StringUtils.isBlank(adapterCode) ? null
                : adapters.selectDeviceAdapterByCode(adapterCode);
        // 来源不存在、类型不对与未启用都明确拒绝：这几种情况一律不落数据。
        if (adapter == null) throw new ServiceException("来源不存在：" + adapterCode, DeviceAdapterErrorCodes.NOT_FOUND);
        if (!PushIngestAdapter.ADAPTER_TYPE.equalsIgnoreCase(String.valueOf(adapter.getAdapterType())))
            throw new ServiceException("该来源不是推式来源：" + adapterCode, DeviceAdapterErrorCodes.UNSUPPORTED_TYPE);
        if (!"0".equals(adapter.getStatus()))
            throw new ServiceException("来源未启用：" + adapterCode, DeviceIngestErrorCodes.SOURCE_DISABLED);
        return adapter;
    }

    /** 端点令牌与厂商库口令共用同一套取值顺序；配置缺失同样按凭据错误拒绝（失败关闭）。 */
    private void requireToken(TDeviceAdapter adapter, String token)
    {
        String key = option(adapter.getSourceConfig(), TOKEN_ENV_OPTION, DEFAULT_TOKEN_ENV);
        String expected;
        try
        {
            expected = credentials.apply(key);
        }
        catch (RuntimeException failure)
        {
            throw new ServiceException("来源凭据不可用：" + failure.getMessage(), DeviceIngestErrorCodes.TOKEN_INVALID);
        }
        if (StringUtils.isBlank(token) || !MessageDigest.isEqual(
                expected.getBytes(StandardCharsets.UTF_8), token.trim().getBytes(StandardCharsets.UTF_8)))
            throw new ServiceException("来源凭据不匹配：" + adapter.getAdapterCode(),
                    DeviceIngestErrorCodes.TOKEN_INVALID);
    }

    /**
     * 拓扑上报：声明该来源当前有哪些通道。行里只有厂商自己的通道号，身份映射在本方法里完成。
     * 拓扑是"这个通道还在"的长期声明，重复上报被幂等忽略（不多占空间，也不改写既有行）。
     */
    private void storeTopology(String adapterCode, String timezoneId, JSONObject payload, Instant receivedAt,
            Counters counters)
    {
        JSONArray rows = rows(payload);
        for (int index = 0; index < rows.size(); index++)
        {
            JSONObject row = rowAt(rows, index);
            Integer channel = row.getInteger("channel");
            if (channel == null || channel < 1) throw payloadInvalid("第 " + index + " 行的 channel 不合法：" + row);
            TDeviceIngestRecord value = base(adapterCode, timezoneId, TDeviceIngestRecord.KIND_TOPOLOGY, channel,
                    receivedAt);
            value.setRecordKey("topology:" + channel);
            value.setPayloadJson(JSON.toJSONString(row));
            store(value, counters);
        }
    }

    /**
     * 逐点数据行：一批来自同一张厂商表（同一测试同一通道），因此身份与来源定位取批次头上的声明。
     * 行本身原样保存，只额外解析出水位与投影需要的字段。
     */
    private void storeSamples(String adapterCode, String timezoneId, JSONObject payload, Instant receivedAt,
            Counters counters)
    {
        Integer channel = payload.getInteger("channel");
        if (channel == null || channel < 1) throw payloadInvalid("批次缺少合法的 channel：" + payload);
        String sourceRef = sourceRef(payload);
        JSONArray rows = rows(payload);
        for (int index = 0; index < rows.size(); index++)
        {
            JSONObject row = rowAt(rows, index);
            TDeviceIngestRecord value = base(adapterCode, timezoneId, TDeviceIngestRecord.KIND_SAMPLE, channel,
                    receivedAt);
            value.setSourceRef(sourceRef);
            value.setRecordSequence(requiredSequence(row));
            value.setDeviceTimeRaw(requiredText(row, "record_time"));
            value.setDeviceTimeUtc(Date.from(filetime(value.getDeviceTimeRaw())));
            value.setCycleNo(row.getInteger("Cycle_Index"));
            value.setStepNo(row.getInteger("Step_Index"));
            value.setRelativeTimeSeconds(row.getBigDecimal("Step_Time"));
            value.setVoltageV(row.getBigDecimal("Voltage"));
            value.setCurrentA(row.getBigDecimal("eCurrent"));
            value.setChargeCapacityAh(row.getBigDecimal("ChargeCapacity"));
            value.setDischargeCapacityAh(row.getBigDecimal("DischargeCapacity"));
            value.setChargeEnergyWh(row.getBigDecimal("ChargeEnergy"));
            value.setDischargeEnergyWh(row.getBigDecimal("DischargeEnergy"));
            // 幂等键含来源定位：22 所每个测试新建一个库，Data_Point 在每个测试内重新编号，只看它会让两次测试撞键。
            value.setRecordKey(sourceRef + ":" + value.getRecordSequence());
            value.setPayloadJson(JSON.toJSONString(row));
            store(value, counters);
        }
    }

    // ------------------------------------------------------------------ 落地行

    private TDeviceIngestRecord base(String adapterCode, String timezoneId, String kind, int channel,
            Instant receivedAt)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(adapterCode);
        value.setPayloadKind(kind);
        // 身份映射与适配器读落地行时用的是同一组常量：这个形态只在 PushIngestAdapter 里定义一次。
        value.setDevtype(PushIngestAdapter.DEVICE_TYPE);
        value.setDevid(PushIngestAdapter.DEVICE_ID);
        value.setSubdevid(PushIngestAdapter.SUB_DEVICE_ID);
        value.setChlid(channel);
        value.setRecordSequence(0L);
        // 来源时区是"当时怎么解释这条来源"的一部分，跟着行落库：厂商给的是绝对时间，
        // 换算出的是 UTC，时区只用于展示；实例时区以后改了，历史行仍保留当时的口径。
        value.setTimezoneId(timezoneId);
        value.setReceivedAtUtc(Date.from(receivedAt));
        return value;
    }

    /**
     * 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
     * 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
     */
    /**
     * 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
     * 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
     *
     * <p>落库没有新增、又查不到这一行时**明确失败**，而不是当作重复忽略：`insert ignore` 把
     * 约束冲突与"转换失败被跳过"都变成"没插入"，后者若不报错，取数程序会拿到 2xx 推进水位，
     * 那条设备行就此消失。宁可让这一批重试，也不静默丢数据。
     */
    private void store(TDeviceIngestRecord value, Counters counters)
    {
        value.setContentSha256(canonicalHash(value.getPayloadJson()));
        int inserted = records.insertIngestRecord(value);
        if (inserted > 0)
        {
            counters.accepted++;
            return;
        }
        String existing = records.selectIngestRecordHashByKey(value.getAdapterCode(), value.getPayloadKind(),
                value.getRecordKey());
        if (existing == null)
            throw payloadInvalid("落地行未能写入且查不到既有行，原因不明：" + value.getRecordKey());
        if (!existing.equalsIgnoreCase(HexFormat.of().formatHex(value.getContentSha256())))
        {
            counters.conflicts++;
            log.warn("落地记录内容冲突，拒绝覆盖 adapterCode={}, payloadKind={}, recordKey={}",
                    value.getAdapterCode(), value.getPayloadKind(), value.getRecordKey());
            return;
        }
        counters.ignored++;
    }

    private void touchHeartbeat(String adapterCode, Instant receivedAt)
    {
        TDeviceIngestRecord heartbeat = base(adapterCode, null, TDeviceIngestRecord.KIND_HEARTBEAT, 0, receivedAt);
        heartbeat.setRecordKey(HEARTBEAT_KEY);
        heartbeat.setPayloadJson("{}");
        heartbeat.setContentSha256(canonicalHash(heartbeat.getPayloadJson()));
        records.upsertHeartbeat(heartbeat);
    }

    /**
     * 内容摘要按规范化后的键值算：程序重复推同一行时字段顺序不同也算同一条，
     * 但任何一个值的改动都会被识别为冲突。
     */
    private static byte[] canonicalHash(String payloadJson)
    {
        Map<String, Object> content = new TreeMap<>(JSON.parseObject(payloadJson));
        try
        {
            return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content));
        }
        catch (Exception exception)
        {
            throw new IllegalStateException("无法计算落地记录内容摘要", exception);
        }
    }

    // ------------------------------------------------------------------ 请求解析

    private JSONObject parse(String body)
    {
        if (StringUtils.isBlank(body)) throw payloadInvalid("请求体为空");
        JSONObject payload;
        try
        {
            payload = JSON.parseObject(body);
        }
        catch (RuntimeException failure)
        {
            throw payloadInvalid("请求体不是合法的 JSON：" + failure.getMessage());
        }
        if (payload == null) throw payloadInvalid("请求体不是 JSON 对象");
        return payload;
    }

    private JSONArray rows(JSONObject payload)
    {
        JSONArray rows = payload.getJSONArray("rows");
        if (rows == null) throw payloadInvalid("请求体缺少 rows 数组");
        if (rows.size() > MAX_ROWS_PER_REQUEST)
            throw payloadInvalid("一批最多 " + MAX_ROWS_PER_REQUEST + " 行，本批 " + rows.size() + " 行");
        return rows;
    }

    private JSONObject rowAt(JSONArray rows, int index)
    {
        JSONObject row = rows.getJSONObject(index);
        if (row == null) throw payloadInvalid("第 " + index + " 行不是 JSON 对象");
        return row;
    }

    /** 来源定位（库/表）取批次头：幂等键需要它，回放与排查也需要它。 */
    private String sourceRef(JSONObject payload)
    {
        String database = payload.getString("database");
        String table = payload.getString("table");
        if (StringUtils.isBlank(database) || StringUtils.isBlank(table))
            throw payloadInvalid("逐点数据批缺少 database 或 table：无法确定来源行身份");
        return database.trim() + "/" + table.trim();
    }

    private long requiredSequence(JSONObject row)
    {
        Long sequence = row.getLong("Data_Point");
        if (sequence == null) throw payloadInvalid("行缺少 Data_Point：" + row);
        return sequence;
    }

    private String requiredText(JSONObject row, String key)
    {
        Object value = row.get(key);
        if (value == null) throw payloadInvalid("行缺少 " + key + "：" + row);
        String text = String.valueOf(value).trim();
        if (text.isEmpty()) throw payloadInvalid("行的 " + key + " 为空：" + row);
        return text;
    }

    /**
     * Windows FILETIME（100 ns 刻度，自 1601-01-01 UTC）换算为 UTC 时刻。
     * 用毫秒精度换算：落地表与实时记录的时间列都是 {@code DATETIME(3)}，多的精度存不下。
     */
    private static Instant filetime(String ticks)
    {
        BigDecimal value = new BigDecimal(ticks);
        long epochMillis = value.divide(BigDecimal.valueOf(10000), 0, RoundingMode.DOWN).longValue()
                - FILETIME_EPOCH_DIFFERENCE_SECONDS * 1000L;
        return Instant.ofEpochMilli(epochMillis);
    }

    private static String option(String sourceConfig, String key, String fallback)
    {
        if (StringUtils.isBlank(sourceConfig)) return fallback;
        try
        {
            JSONObject config = JSON.parseObject(sourceConfig);
            String value = config == null ? null : config.getString(key);
            return StringUtils.isBlank(value) ? fallback : value.trim();
        }
        catch (RuntimeException ignored)
        {
            return fallback;
        }
    }

    private static ServiceException payloadInvalid(String message)
    {
        return new ServiceException(message, DeviceIngestErrorCodes.PAYLOAD_INVALID);
    }

    private static final class Counters
    {
        private int accepted;
        private int ignored;
        private int conflicts;
    }
}