package com.battery.business.service.impl;

import java.math.BigDecimal;
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.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.alibaba.fastjson2.JSON;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.CollectorMeasurement;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.business.domain.TTest;
import com.battery.business.domain.TTestBattery;
import com.battery.business.domain.TTestChannelBinding;
import com.battery.business.domain.TTestCycleRecord;
import com.battery.business.domain.TTestRealtimeRecord;
import com.battery.business.domain.TTestStepRecord;
import com.battery.business.mapper.TTestMapper;
import com.battery.business.service.ITTestService;
import com.battery.business.service.ITestDataReceptionService;

/** 接收期间将成功观测直接归属测试，并写入实时、工步和循环事实表。 */
@Service
public class TestDataReceptionServiceImpl implements ITestDataReceptionService
{
    private static final Logger log = LoggerFactory.getLogger(TestDataReceptionServiceImpl.class);
    private final TTestMapper mapper;
    private final ITTestService tests;
    private final TestDataObservationAdapter adapter;
    private final TestWarningExceptionService warningExceptions;
    private final Clock clock;
    private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
    private final Map<Long, Instant> lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>();

    @Autowired
    public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
            TestWarningExceptionService warningExceptions)
    { this(mapper, tests, adapter, warningExceptions, Clock.systemUTC()); }

    /** 允许测试固定接收时间，生产运行仍使用 UTC 系统时钟。 */
    public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter, Clock clock)
    { this(mapper, tests, adapter, new TestWarningExceptionService(mapper), clock); }

    public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter)
    { this(mapper, tests, adapter, new TestWarningExceptionService(mapper), Clock.systemUTC()); }

    private TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter,
            TestWarningExceptionService warningExceptions, Clock clock)
    { this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; }

    /** 轮询接收中测试，判定实时超限和来源离线异常，并按独立周期保存成功观测。 */
    @Override @Transactional(rollbackFor = Exception.class)
    public void captureReceivedData()
    {
        for (TTest test : receivingTests()) {
            Instant now = clock.instant();
            Instant previous = lastRealtimeCheckAt.get(test.getTestId());
            // 实时预警独立于持久化槽位，并与暂停、停止和删除共享测试事务锁。
            if (previous == null || !now.isBefore(previous.plusSeconds(5))) {
                warningExceptions.recordOfflineSources(test, Date.from(now));
                for (TTestBattery battery : batteries(test)) saveRealtime(test, battery);
                lastRealtimeCheckAt.put(test.getTestId(), now);
            }
            for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery);
        }
    }

    private List<TTest> receivingTests()
    {
        List<Long> ids = mapper.selectReceivingTestIds(); if (ids == null) return List.of();
        return ids.stream().map(id -> {
            TTest locked = mapper.selectTestByIdForUpdate(id);
            if (locked == null || !"RUNNING".equals(locked.getTestStatus()) || !"RECEIVING".equals(locked.getReceptionStatus())) return null;
            TTest current = tests.selectTestById(id);
            return current != null && "RUNNING".equals(current.getTestStatus()) && "RECEIVING".equals(current.getReceptionStatus()) ? current : null;
        }).filter(java.util.Objects::nonNull).toList();
    }
    private List<TTestBattery> batteries(TTest test) { return test.getBatteries() == null ? List.of() : test.getBatteries(); }

    private void saveRealtime(TTest test, TTestBattery battery)
    {
        // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
        if (adapter == null) return;
        TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
        Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId();
        CabinetRealtime cabinet = latestCabinet(binding);
        if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null;
        CollectorRealtime collector = latestCollector(battery);
        Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector);
        if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return;
        // 当前成功测量先判定预警；槽位尚未到期或历史行被删除，都不能阻断实时告警。
        if (cabinet != null) {
            Date observedAt = Date.from(cabinet.serverReceivedAt());
            evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(cabinet.voltageV()), "V", observedAt);
            evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(cabinet.currentA()), "A", observedAt);
            evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(cabinet.capacityAh()), "Ah", observedAt);
            evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(cabinet.energyWh()), "Wh", observedAt);
        }
        for (CollectorMeasurement measurement : mapped.values().stream().distinct().toList()) {
            warningExceptions.evaluate(test, "COLLECTOR", battery.getCollectorId() + ":" + measurement.channelCode(),
                    measurement.metricCode(), measurement.value(), measurement.unit(), Date.from(measurement.serverReceivedAt()));
        }
        Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return;
        Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null);
        Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval);
        // 删除修订随数据库保留；重启后同一槽位也不能重新出现。
        if (mapper.wasRealtimeSlotDeleted(test.getTestId(), battery.getTestBatteryId(), slot.toEpochMilli())) return;
        TTestRealtimeRecord row = new TTestRealtimeRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId());
        row.setAdapterCode(binding.getAdapterCode()); row.setCaptureSlotUtc(Date.from(slot)); row.setServerReceivedAtUtc(Date.from(now));
        if (cabinet != null) { row.setCabinetDevtype(binding.getDevtype()); row.setCabinetDevid(binding.getDevid()); row.setCabinetSubdevid(binding.getSubdevid()); row.setCabinetChlid(binding.getChlid()); row.setCabinetAdapterTimezoneId(cabinet.timezoneId()); }
        if (collector != null && collector.online() && !mapped.isEmpty()) { row.setCollectorId(battery.getCollectorId()); row.setCollectorAdapterTimezoneId(collector.timezoneId()); }
        row.setCabinetSequenceId(cabinet == null ? null : cabinet.sequenceId()); row.setCabinetDeviceTimeUtc(cabinet == null ? null : Date.from(cabinet.deviceTime())); row.setCabinetDeviceTimeRaw(cabinet == null ? null : cabinet.deviceTimeRaw());
        row.setCollectorSequenceId(mapped.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo).orElse(null)); row.setCollectorDeviceTimeUtc(collectorTime == null ? null : Date.from(collectorTime)); row.setCollectorDeviceTimeRaw(mapped.values().stream().filter(value -> value.deviceTime().equals(collectorTime)).map(CollectorMeasurement::deviceTimeRaw).findFirst().orElse(null));
        if (cabinet != null) { row.setCycleNo(cabinet.cycleNo()); row.setStepNo(cabinet.stepNo()); row.setStepType(cabinet.stepType()); row.setStepName(cabinet.stepName()); row.setVoltageV(BigDecimal.valueOf(cabinet.voltageV())); row.setCurrentA(BigDecimal.valueOf(cabinet.currentA())); row.setCapacityAh(BigDecimal.valueOf(cabinet.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(cabinet.energyWh())); }
        row.setTemperature(metricValue(mapped, "TEMPERATURE")); row.setPositivePotential(metricValue(mapped, "POSITIVE_POTENTIAL")); row.setNegativePotential(metricValue(mapped, "NEGATIVE_POTENTIAL")); row.setCollectorChannelCode(mapped.values().stream().map(CollectorMeasurement::channelCode).findFirst().orElse(null));
        Map<String, Object> extension = new LinkedHashMap<>();
        // 工步内相对时长由成功柜端观测提供，不以轮询槽位或服务器时间推算。
        if (cabinet != null && cabinet.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", cabinet.relativeTimeSeconds());
        for (Map.Entry<String, CollectorMeasurement> e : mapped.entrySet()) if (!isFixedMetric(e.getKey()))
        {
            Map<String, Object> value = new LinkedHashMap<>(); value.put("value", e.getValue().value()); value.put("unit", e.getValue().unit()); value.put("channelCode", e.getValue().channelCode()); extension.put(e.getKey(), value);
        }
        row.setExtensionMeasurementsJson(JSON.toJSONString(extension)); row.setSourceIdentity(sourceIdentity(binding));
        row.setSourceRecordId(sourceRecordId(cabinet, mapped));
        row.setContentSha256(hash(row));
        String old = mapper.selectRealtimeRecordContentHash(test.getTestId(), battery.getTestBatteryId(), row.getCaptureSlotUtc());
        if (old != null) { if (!old.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256()))) log.warn("实时采样内容冲突，拒绝覆盖 testId={}, testBatteryId={}, captureSlotUtc={}", test.getTestId(), battery.getTestBatteryId(), slot); return; }
        int inserted = mapper.insertRealtimeRecord(row);
        if (inserted == 0)
        {
            String actual = mapper.selectRealtimeRecordContentHash(test.getTestId(), battery.getTestBatteryId(), row.getCaptureSlotUtc());
            if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
                log.warn("实时采样并发内容冲突，拒绝覆盖 testId={}, testBatteryId={}, captureSlotUtc={}", test.getTestId(), battery.getTestBatteryId(), slot);
        }
        lastSnapshotAt.put(key, now);
    }

    private void evaluateCabinet(TTest test, TTestChannelBinding binding, String metricCode, BigDecimal value, String unit, Date occurredAt)
    {
        if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt);
    }

    private void saveDeviceAggregates(TTest test, TTestBattery battery)
    {
        // 工步、循环只接受设备直接返回的统计记录，并以来源编号或明确的时间组合执行幂等。
        if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
        ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid());
        List<StepRecord> steps; List<CycleRecord> cycles; try { steps = adapter.stepRecords(channel); cycles = adapter.cycleRecords(channel); } catch (RuntimeException ex) { return; }
        Date receivedAt = Date.from(clock.instant()); String identity = sourceIdentity(binding);
        for (StepRecord source : steps == null ? List.<StepRecord>of() : steps)
        {
            if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
            TTestStepRecord row = new TTestStepRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setAdapterTimezoneId(source.timezoneId()); row.setCycleNo(source.cycleNo()); row.setStepNo(source.stepNo()); row.setStepType(source.stepType()); row.setStepName(source.stepName()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.deviceTimeRaw()); row.setServerReceivedAtUtc(Date.from(source.serverReceivedAt())); row.setVoltageV(BigDecimal.valueOf(source.endVoltageV())); row.setCapacityAh(BigDecimal.valueOf(source.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(source.energyWh())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() + ":" + source.stepNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
            // 物理删除依靠已有修订追溯识别来源，重复观测不能把已删除汇总重新写回。
            row.setExtensionMeasurementsJson(JSON.toJSONString(source.measurements()));
            row.setContentSha256(hash(row));
            if (mapper.wasAggregateDeleted(test.getTestId(), "STEP", battery.getTestBatteryId(), identity, row.getDedupeKey())) continue;
            if (mapper.insertStepRecord(row) == 0)
            {
                String actual = mapper.selectStepRecordContentHash(row.getTestBatteryId(), row.getSourceIdentity(), row.getDedupeKey());
                if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
                    log.warn("工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
            }
        }
        for (CycleRecord source : cycles == null ? List.<CycleRecord>of() : cycles)
        {
            if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
            TTestCycleRecord row = new TTestCycleRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setAdapterTimezoneId(source.timezoneId()); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setCycleNo(source.cycleNo()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.deviceTimeRaw()); row.setServerReceivedAtUtc(Date.from(source.serverReceivedAt())); row.setChargeCapacityAh(BigDecimal.valueOf(source.chargeCapacityAh())); row.setDischargeCapacityAh(BigDecimal.valueOf(source.dischargeCapacityAh())); row.setChargeEnergyWh(BigDecimal.valueOf(source.chargeEnergyWh())); row.setDischargeEnergyWh(BigDecimal.valueOf(source.dischargeEnergyWh())); row.setEfficiency(BigDecimal.valueOf(source.efficiency())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
            if (mapper.wasAggregateDeleted(test.getTestId(), "CYCLE", battery.getTestBatteryId(), identity, row.getDedupeKey())) continue;
            if (mapper.insertCycleRecord(row) == 0)
            {
                String actual = mapper.selectCycleRecordContentHash(row.getTestBatteryId(), row.getSourceIdentity(), row.getDedupeKey());
                if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
                    log.warn("循环记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
            }
        }
    }

    private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid())); } catch (RuntimeException ex) { return null; } }
    private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null) return null; try { return adapter.latestCollector(mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
    private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source)
    {
        Map<String, CollectorMeasurement> result = new LinkedHashMap<>();
        if (source == null || !source.online() || battery.getCollectorChannels() == null) return result;
        for (var binding : battery.getCollectorChannels()) for (var measurement : source.measurements()) {
            if (accepted(test, measurement.deviceTime(), measurement.serverReceivedAt())
                    && java.util.Objects.equals(binding.getChannelCode(), measurement.channelCode())
                    && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), measurement.metricCode())) {
                // 绑定角色区分主、辅助测量；不能用设备通用指标名覆盖其他物理通道。
                result.put(binding.getMeasurementCode().toUpperCase(Locale.ROOT), measurement);
            }
        }
        return result;
    }
    private static boolean isFixedMetric(String code) { return List.of("TEMPERATURE", "POSITIVE_POTENTIAL", "NEGATIVE_POTENTIAL", "PRIMARY_TEMPERATURE", "PRIMARY_POSITIVE_POTENTIAL", "PRIMARY_NEGATIVE_POTENTIAL").contains(code); }
    private static BigDecimal metricValue(Map<String, CollectorMeasurement> values, String code) {
        var measurement = values.get("PRIMARY_" + code);
        if (measurement == null) measurement = values.get(code);
        return measurement == null ? null : measurement.value();
    }
    private static String sourceRecordId(CabinetRealtime cabinet, Map<String, CollectorMeasurement> measurements)
    {
        List<String> parts = new java.util.ArrayList<>();
        if (cabinet != null) parts.add("cabinet:" + cabinet.sequenceId());
        measurements.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo)
                .ifPresent(sequence -> parts.add("collector:" + sequence));
        return String.join(":", parts);
    }
    private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()))); } catch (RuntimeException ex) { return 60000L; } }
    private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
    private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
    private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }
    // 实时提醒与持久化共用归属边界，避免同一旧观测只在首页触发异常。
    static boolean accepted(TTest test, Instant deviceTime, Instant serverReceivedAt)
    {
        if (deviceTime != null && test.getStartedAtUtc() != null && deviceTime.isBefore(test.getStartedAtUtc().toInstant())) return false;
        return serverReceivedAt != null && (test.getReceptionBoundaryAtUtc() == null || !serverReceivedAt.isBefore(test.getReceptionBoundaryAtUtc().toInstant()));
    }

    private byte[] hash(Object value)
    {
        // 重复观测比较设备内容，不把服务器轮询时间作为内容变化。
        Map<String, Object> content = new java.util.TreeMap<>(JSON.parseObject(JSON.toJSONString(value)));
        content.remove("serverReceivedAtUtc");
        try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content)); }
        catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); }
    }
}