package com.ruoyi.energymeter.service.impl;

import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.energymeter.config.EnergyMeterProperties;
import com.ruoyi.energymeter.constant.EnergyMeterConstants;
import com.ruoyi.energymeter.dto.MeterReadingPoint;
import com.ruoyi.energymeter.dto.MeterUsagePoint;
import com.ruoyi.energymeter.dto.TaskStatusCount;
import com.ruoyi.energymeter.mapper.EnergyMeterDeviceMapper;
import com.ruoyi.energymeter.mapper.EnergyMeterMeasurementMapper;
import com.ruoyi.energymeter.mapper.EnergyMeterSyncCheckpointMapper;
import com.ruoyi.energymeter.mapper.EnergyMeterSyncRunMapper;
import com.ruoyi.energymeter.mapper.EnergyMeterSyncTaskMapper;
import com.ruoyi.energymeter.pojo.EnergyMeterDevice;
import com.ruoyi.energymeter.pojo.EnergyMeterMeasurement;
import com.ruoyi.energymeter.pojo.EnergyMeterSyncCheckpoint;
import com.ruoyi.energymeter.pojo.EnergyMeterSyncRun;
import com.ruoyi.energymeter.pojo.EnergyMeterSyncTask;
import com.ruoyi.energymeter.provider.MeterProviderClient;
import com.ruoyi.energymeter.provider.MeterProviderException;
import com.ruoyi.energymeter.service.EnergyMeterSyncService;
import com.ruoyi.energymeter.util.EnergyMeterSupport;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * 同步编排：批次创建、任务执行、幂等落库、进度推进与失败重试。
 *
 * <p>外部 HTTP 调用不放在数据库事务中；写入采集表、更新任务和进度在短事务内提交。</p>
 */
@Slf4j
@Service
public class EnergyMeterSyncServiceImpl implements EnergyMeterSyncService {

    private static final int UPSERT_CHUNK = 500;

    private final EnergyMeterDeviceMapper deviceMapper;
    private final EnergyMeterMeasurementMapper measurementMapper;
    private final EnergyMeterSyncRunMapper runMapper;
    private final EnergyMeterSyncTaskMapper taskMapper;
    private final EnergyMeterSyncCheckpointMapper checkpointMapper;
    private final MeterProviderClient provider;
    private final EnergyMeterProperties properties;
    private final TransactionTemplate transactionTemplate;
    private final ThreadPoolTaskExecutor taskExecutor;
    private final ThreadPoolTaskExecutor runExecutor;

    public EnergyMeterSyncServiceImpl(EnergyMeterDeviceMapper deviceMapper,
                                      EnergyMeterMeasurementMapper measurementMapper,
                                      EnergyMeterSyncRunMapper runMapper,
                                      EnergyMeterSyncTaskMapper taskMapper,
                                      EnergyMeterSyncCheckpointMapper checkpointMapper,
                                      MeterProviderClient provider,
                                      EnergyMeterProperties properties,
                                      PlatformTransactionManager transactionManager,
                                      @Qualifier("energyMeterSyncExecutor") ThreadPoolTaskExecutor taskExecutor,
                                      @Qualifier("energyMeterRunExecutor") ThreadPoolTaskExecutor runExecutor) {
        this.deviceMapper = deviceMapper;
        this.measurementMapper = measurementMapper;
        this.runMapper = runMapper;
        this.taskMapper = taskMapper;
        this.checkpointMapper = checkpointMapper;
        this.provider = provider;
        this.properties = properties;
        this.transactionTemplate = new TransactionTemplate(transactionManager);
        this.taskExecutor = taskExecutor;
        this.runExecutor = runExecutor;
    }

    /* ------------------------------------------------------------------ */
    /* 批次创建                                                             */
    /* ------------------------------------------------------------------ */

    @Override
    public Long triggerScheduledSync() {
        if (!provider.mockMode() && !provider.hasApiKey()) {
            log.warn("未配置第三方授权码（NH_ENERGY_API_KEY），跳过每日自动同步");
            return null;
        }
        LocalDate today = EnergyMeterSupport.today();
        LocalDate target = today.minusDays(1);
        String runKey = EnergyMeterConstants.AUTO_RUN_KEY_PREFIX
                + today.format(DateTimeFormatter.BASIC_ISO_DATE);
        EnergyMeterSyncRun existing = runMapper.selectByRunKey(runKey);
        if (existing != null) {
            log.info("当天自动同步批次已存在，跳过重复触发 runKey={}", runKey);
            return existing.getId();
        }

        EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_SCHEDULED, runKey, today, target, target);
        try {
            runMapper.insert(run);
        } catch (DuplicateKeyException e) {
            EnergyMeterSyncRun duplicated = runMapper.selectByRunKey(runKey);
            log.info("并发触发自动同步，复用已有批次 runKey={}", runKey);
            return duplicated == null ? null : duplicated.getId();
        }

        List<EnergyMeterDevice> devices = deviceMapper.selectScannableDevices();
        LocalDate minStart = target;
        int taskCount = 0;
        for (EnergyMeterDevice device : devices) {
            LocalDate start = resolveAutoStart(device, target);
            if (start == null) {
                continue;
            }
            if (start.isBefore(minStart)) {
                minStart = start;
            }
            taskCount += createDeviceTasks(run.getId(), device, start, target, properties.isRawReadingEnabled());
        }
        applyRunRange(run, minStart, target, taskCount);
        log.info("自动同步批次创建完成 runKey={} 设备数={} 任务数={}", runKey, devices.size(), taskCount);

        if (taskCount == 0) {
            EnergyMeterSyncRun finished = new EnergyMeterSyncRun();
            finished.setId(run.getId());
            finished.setStatus(EnergyMeterConstants.RUN_SUCCESS);
            finished.setFinishedAt(EnergyMeterSupport.now());
            runMapper.updateById(finished);
            return run.getId();
        }
        executeRunAfterCommit(run.getId());
        return run.getId();
    }

    @Override
    public void triggerBackfill(EnergyMeterDevice device, LocalDate start, LocalDate end) {
        if (device == null || start == null || end == null || start.isAfter(end)) {
            return;
        }
        if (EnergyMeterConstants.isManual(device.getSourceType())) {
            log.info("手工抄表设备不创建补采批次 meterId={}", device.getId());
            return;
        }
        if (!provider.mockMode() && !provider.hasApiKey()) {
            log.warn("未配置第三方授权码，跳过补采 meterId={}", device.getId());
            return;
        }
        EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_BACKFILL,
                "BACKFILL:" + UUID.randomUUID(), null, start, end);
        runMapper.insert(run);
        int taskCount = createDeviceTasks(run.getId(), device, start, end, properties.isRawReadingEnabled());
        applyRunRange(run, start, end, taskCount);
        log.info("历史补采批次创建完成 meterId={} range={}~{} tasks={}", device.getId(), start, end, taskCount);
        if (taskCount > 0) {
            executeRunAfterCommit(run.getId());
        }
    }

    @Override
    public Long triggerBackfillGap(EnergyMeterDevice device) {
        if (device == null) {
            return null;
        }
        if (EnergyMeterConstants.isManual(device.getSourceType())) {
            log.info("手工抄表设备不参与缺口补采 meterId={}", device.getId());
            return null;
        }
        if (!provider.mockMode() && !provider.hasApiKey()) {
            log.warn("未配置第三方授权码，跳过缺口补采 meterId={}", device.getId());
            return null;
        }
        LocalDate target = EnergyMeterSupport.today().minusDays(1);
        LocalDate base = device.getSyncStartDate() != null
                ? device.getSyncStartDate()
                : LocalDate.of(target.getYear() - 1, 1, 1);
        EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
                EnergyMeterConstants.usageSource(device.getMeterType()), EnergyMeterConstants.GRANULARITY_DAY);
        LocalDate start = base;
        if (checkpoint != null && checkpoint.getLastCompleteDate() != null) {
            start = checkpoint.getLastCompleteDate().plusDays(1);
        }
        if (start.isBefore(base)) {
            start = base;
        }
        if (start.isAfter(target)) {
            log.info("设备 {} 无缺口需要补采", device.getId());
            return null;
        }
        EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_BACKFILL,
                "BACKFILL:" + UUID.randomUUID(), null, start, target);
        runMapper.insert(run);
        int taskCount = createDeviceTasks(run.getId(), device, start, target, properties.isRawReadingEnabled());
        applyRunRange(run, start, target, taskCount);
        if (taskCount > 0) {
            executeRunAfterCommit(run.getId());
        }
        return run.getId();
    }

    @Override
    public Long triggerManual(List<Long> meterIds, LocalDate start, LocalDate end, Boolean includeReadings) {
        if (!provider.mockMode() && !provider.hasApiKey()) {
            throw new ServiceException("未配置第三方授权码（环境变量 NH_ENERGY_API_KEY），无法查表");
        }
        if (meterIds == null || meterIds.isEmpty()) {
            throw new ServiceException("请选择至少一台设备");
        }
        // 手动查表允许查到今天（当天用量为截至此刻的累计值），只挡未来日期
        LocalDate today = EnergyMeterSupport.today();
        LocalDate rangeEnd = end != null ? end : today;
        if (rangeEnd.isAfter(today)) {
            rangeEnd = today;
        }
        LocalDate rangeStart = start != null ? start : rangeEnd.minusDays(2);
        if (rangeStart.isAfter(rangeEnd)) {
            throw new ServiceException("开始日期不能晚于结束日期");
        }
        if (ChronoUnit.DAYS.between(rangeStart, rangeEnd) > EnergyMeterConstants.MAX_RANGE_DAYS) {
            throw new ServiceException("查询跨度超出上限，请缩小日期范围");
        }

        List<EnergyMeterDevice> devices = new ArrayList<>();
        for (Long meterId : new LinkedHashSet<>(meterIds)) {
            EnergyMeterDevice device = deviceMapper.selectById(meterId);
            if (device == null) {
                throw new ServiceException("设备不存在：" + meterId);
            }
            Long tenantId = EnergyMeterSupport.currentTenantId();
            if (tenantId != null && device.getTenantId() != null && !tenantId.equals(device.getTenantId())) {
                throw new ServiceException("无权操作设备：" + meterId);
            }
            if (!EnergyMeterConstants.BIND_VERIFIED.equals(device.getBindStatus())) {
                throw new ServiceException("设备未通过第三方档案验证，不能查表：" + meterId);
            }
            if (EnergyMeterConstants.isManual(device.getSourceType())) {
                throw new ServiceException("手工抄表设备无需查表，请在表具台账中录入表码："
                        + StringUtils.defaultIfBlank(device.getDisplayName(), device.getMeterAddress()));
            }
            devices.add(device);
        }

        EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_MANUAL,
                "MANUAL:" + UUID.randomUUID(), null, rangeStart, rangeEnd);
        runMapper.insert(run);
        boolean withReadings = includeReadings == null ? properties.isRawReadingEnabled() : includeReadings;
        int taskCount = 0;
        for (EnergyMeterDevice device : devices) {
            taskCount += createDeviceTasks(run.getId(), device, rangeStart, rangeEnd, withReadings);
        }
        applyRunRange(run, rangeStart, rangeEnd, taskCount);
        log.info("手动查表批次创建完成 runId={} 设备数={} 任务数={} range={}~{}",
                run.getId(), devices.size(), taskCount, rangeStart, rangeEnd);
        if (taskCount > 0) {
            executeRunAfterCommit(run.getId());
        }
        return run.getId();
    }

    /* ------------------------------------------------------------------ */
    /* 批次执行                                                             */
    /* ------------------------------------------------------------------ */

    @Override
    public void executeRunAsync(Long runId) {
        if (runId == null) {
            return;
        }
        runExecutor.execute(() -> {
            try {
                processRun(runId);
            } catch (Throwable e) {
                log.error("同步批次执行异常 runId=" + runId, e);
            }
        });
    }

    /**
     * 在事务提交后再开始执行，避免异步线程读不到尚未提交的任务记录。
     */
    private void executeRunAfterCommit(Long runId) {
        if (runId == null) {
            return;
        }
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    executeRunAsync(runId);
                }
            });
        } else {
            executeRunAsync(runId);
        }
    }

    private void processRun(Long runId) {
        EnergyMeterSyncRun run = runMapper.selectById(runId);
        if (run == null) {
            return;
        }
        if (EnergyMeterConstants.RUN_PENDING.equals(run.getStatus())) {
            EnergyMeterSyncRun running = new EnergyMeterSyncRun();
            running.setId(runId);
            running.setStatus(EnergyMeterConstants.RUN_RUNNING);
            running.setStartedAt(EnergyMeterSupport.now());
            runMapper.updateById(running);
        }

        LocalDateTime now = EnergyMeterSupport.now();
        // 同一设备、同一接口、同一粒度的分片按日期顺序串行执行，保证连续进度不会被乱序推进；
        // 不同设备/接口之间仍然并行，受线程池并发数限制。
        Map<String, List<EnergyMeterSyncTask>> groups = new LinkedHashMap<>();
        for (EnergyMeterSyncTask task : taskMapper.selectByRunId(runId)) {
            if (!isRunnable(task, now)) {
                continue;
            }
            String key = task.getMeterId() + "|" + task.getSourceApi() + "|" + task.getGranularity();
            groups.computeIfAbsent(key, k -> new ArrayList<>()).add(task);
        }
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (List<EnergyMeterSyncTask> group : groups.values()) {
            group.sort(Comparator.comparing(EnergyMeterSyncTask::getRangeStart,
                    Comparator.nullsFirst(LocalDate::compareTo)));
            futures.add(CompletableFuture.runAsync(() -> {
                for (EnergyMeterSyncTask task : group) {
                    executeTask(task.getId());
                }
            }, taskExecutor));
        }
        if (!futures.isEmpty()) {
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        }
        finalizeRun(runId);
    }

    private boolean isRunnable(EnergyMeterSyncTask task, LocalDateTime now) {
        if (EnergyMeterConstants.TASK_PENDING.equals(task.getStatus())) {
            return true;
        }
        if (EnergyMeterConstants.TASK_RETRY.equals(task.getStatus())) {
            return task.getNextRetryAt() == null || !task.getNextRetryAt().isAfter(now);
        }
        return false;
    }

    private void executeTask(Long taskId) {
        LocalDateTime now = EnergyMeterSupport.now();
        if (taskMapper.claim(taskId, now) == 0) {
            return;
        }
        EnergyMeterSyncTask task = taskMapper.selectById(taskId);
        EnergyMeterDevice device = deviceMapper.selectById(task.getMeterId());
        if (device == null) {
            markTaskFailed(task, "DEVICE_MISSING", "本地设备不存在，无法采集");
            return;
        }
        try {
            FetchResult result = fetch(device, task);
            persist(device, task, result);
        } catch (MeterProviderException e) {
            markTaskFailed(task, e.getCode(), e.getMessage());
        } catch (Exception e) {
            log.error("同步任务异常 taskId=" + taskId, e);
            markTaskFailed(task, EnergyMeterConstants.ERROR_UPSTREAM, e.getMessage());
        }
    }

    private FetchResult fetch(EnergyMeterDevice device, EnergyMeterSyncTask task) {
        String meterType = device.getMeterType();
        String address = device.getMeterAddress();
        String providerMeterId = device.getProviderMeterId();
        LocalDate start = task.getRangeStart();
        LocalDate end = task.getRangeEnd();
        LocalDateTime collectedAt = EnergyMeterSupport.now();
        List<EnergyMeterMeasurement> measurements = new ArrayList<>();
        int fetched = 0;

        if (EnergyMeterConstants.GRANULARITY_DAY.equals(task.getGranularity())) {
            List<MeterUsagePoint> points = provider.queryDailyUsage(meterType, address, providerMeterId, start, end);
            fetched = points.size();
            String metric = EnergyMeterConstants.usageMetric(meterType);
            String unit = EnergyMeterConstants.unit(meterType);
            for (MeterUsagePoint point : points) {
                validateAddress(address, point.getAddress());
                if (point.getTime() == null || point.getValue() == null) {
                    throw new MeterProviderException(EnergyMeterConstants.ERROR_PARSE, "用量数据缺少时间或数值");
                }
                LocalDate day = point.getTime().toLocalDate();
                if (day.isBefore(start) || day.isAfter(end)) {
                    continue;
                }
                EnergyMeterMeasurement measurement = new EnergyMeterMeasurement();
                measurement.setTenantId(device.getTenantId());
                measurement.setMeterId(device.getId());
                measurement.setMetricCode(metric);
                measurement.setGranularity(EnergyMeterConstants.GRANULARITY_DAY);
                measurement.setMeasureTime(day.atStartOfDay());
                measurement.setMetricValue(point.getValue());
                measurement.setUnit(unit);
                measurement.setSourceApi(task.getSourceApi());
                measurement.setSourceRecordTime(point.getTime());
                measurement.setCollectedAt(collectedAt);
                measurements.add(measurement);
            }
        } else {
            List<MeterReadingPoint> points = provider.queryReadings(meterType, address, providerMeterId, start, end);
            fetched = points.size();
            String metric = EnergyMeterConstants.readingMetric(meterType);
            String unit = EnergyMeterConstants.unit(meterType);
            for (MeterReadingPoint point : points) {
                validateAddress(address, point.getAddress());
                if (point.getTime() == null || point.getReading() == null) {
                    throw new MeterProviderException(EnergyMeterConstants.ERROR_PARSE, "示值数据缺少时间或数值");
                }
                LocalDate day = point.getTime().toLocalDate();
                if (day.isBefore(start) || day.isAfter(end)) {
                    continue;
                }
                EnergyMeterMeasurement measurement = new EnergyMeterMeasurement();
                measurement.setTenantId(device.getTenantId());
                measurement.setMeterId(device.getId());
                measurement.setMetricCode(metric);
                measurement.setGranularity(EnergyMeterConstants.GRANULARITY_POINT);
                measurement.setMeasureTime(point.getTime());
                measurement.setMetricValue(point.getReading());
                measurement.setUnit(unit);
                measurement.setSourceApi(task.getSourceApi());
                measurement.setSourceRecordTime(point.getTime());
                measurement.setCollectedAt(collectedAt);
                measurements.add(measurement);
            }
        }
        FetchResult result = new FetchResult();
        result.fetched = fetched;
        result.measurements = measurements;
        result.metricCode = EnergyMeterConstants.GRANULARITY_DAY.equals(task.getGranularity())
                ? EnergyMeterConstants.usageMetric(meterType)
                : EnergyMeterConstants.readingMetric(meterType);
        return result;
    }

    private void validateAddress(String expected, String actual) {
        if (StringUtils.isNotBlank(actual) && !expected.equals(actual.trim())) {
            throw new MeterProviderException(EnergyMeterConstants.ERROR_ADDRESS_MISMATCH,
                    "第三方返回的通信地址与本地绑定不一致");
        }
    }

    private void persist(EnergyMeterDevice device, EnergyMeterSyncTask task, FetchResult result) {
        LocalDateTime now = EnergyMeterSupport.now();
        LocalDateTime rangeStart = task.getRangeStart().atStartOfDay();
        LocalDateTime rangeEnd = task.getRangeEnd().plusDays(1).atStartOfDay();

        int[] counters = transactionTemplate.execute(txStatus -> {
            int inserted = 0;
            int updated = 0;
            List<EnergyMeterMeasurement> writes = new ArrayList<>();
            if (!result.measurements.isEmpty()) {
                Map<LocalDateTime, BigDecimal> existing = new HashMap<>();
                for (EnergyMeterMeasurement row : measurementMapper.selectExistingPoints(device.getId(),
                        result.metricCode, task.getGranularity(), rangeStart, rangeEnd)) {
                    existing.put(row.getMeasureTime(), row.getMetricValue());
                }
                for (EnergyMeterMeasurement measurement : result.measurements) {
                    BigDecimal oldValue = existing.get(measurement.getMeasureTime());
                    if (oldValue == null) {
                        inserted++;
                        writes.add(measurement);
                    } else if (oldValue.compareTo(measurement.getMetricValue()) != 0) {
                        updated++;
                        writes.add(measurement);
                    }
                }
            }
            for (int from = 0; from < writes.size(); from += UPSERT_CHUNK) {
                int to = Math.min(from + UPSERT_CHUNK, writes.size());
                measurementMapper.upsertBatch(writes.subList(from, to));
            }

            String status = result.measurements.isEmpty()
                    ? EnergyMeterConstants.TASK_EMPTY : EnergyMeterConstants.TASK_SUCCESS;
            taskMapper.markFinished(task.getId(), status, result.fetched, inserted, updated, now);
            deviceMapper.markCollectSuccess(device.getId(), now);
            advanceCheckpoint(device, task, now);
            return new int[]{inserted, updated};
        });
        int inserted = counters == null ? 0 : counters[0];
        int updated = counters == null ? 0 : counters[1];
        log.info("同步任务完成 taskId={} meterId={} 拉取={} 新增={} 修正={}",
                task.getId(), device.getId(), result.fetched, inserted, updated);
    }

    private void advanceCheckpoint(EnergyMeterDevice device, EnergyMeterSyncTask task, LocalDateTime now) {
        LocalDate start = task.getRangeStart();
        LocalDate end = task.getRangeEnd();
        // 当天数据仍在滚动累计（手动查表或当天补采会不断覆盖），不能把今天记为“已完成”，
        // 否则补采缺口判断会误以为今天已经完整。完成进度只推进到昨天及更早。
        if (!end.isBefore(EnergyMeterSupport.today())) {
            log.debug("跳过当天进度推进 meterId={} source={} granularity={} end={}",
                    device.getId(), task.getSourceApi(), task.getGranularity(), end);
            return;
        }
        EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
                task.getSourceApi(), task.getGranularity());
        LocalDate existingDate = checkpoint == null ? null : checkpoint.getLastCompleteDate();
        boolean contiguous = existingDate == null || !start.isAfter(existingDate.plusDays(1));
        LocalDate lastComplete;
        if (contiguous) {
            lastComplete = existingDate == null || end.isAfter(existingDate) ? end : existingDate;
        } else {
            lastComplete = existingDate != null ? existingDate : start.minusDays(1);
        }
        EnergyMeterSyncCheckpoint upsert = new EnergyMeterSyncCheckpoint();
        upsert.setTenantId(device.getTenantId());
        upsert.setMeterId(device.getId());
        upsert.setSourceApi(task.getSourceApi());
        upsert.setGranularity(task.getGranularity());
        upsert.setLastCompleteDate(lastComplete);
        upsert.setLastSuccessAt(now);
        upsert.setLastRunId(task.getRunId());
        checkpointMapper.upsert(upsert);
    }

    private void markTaskFailed(EnergyMeterSyncTask task, String code, String message) {
        String safeCode = StringUtils.defaultIfBlank(code, EnergyMeterConstants.ERROR_UPSTREAM);
        String safeMessage = EnergyMeterSupport.truncate(
                StringUtils.defaultIfBlank(message, "未知错误"), 450);
        int attempts = task.getAttemptCount() == null ? 1 : task.getAttemptCount();
        // max-attempts 表示“最多重试次数”，首次执行之外还可重试 N 次
        int maxTotalAttempts = Math.max(1, properties.getMaxAttempts()) + 1;
        boolean retry = attempts < maxTotalAttempts;
        LocalDateTime now = EnergyMeterSupport.now();
        LocalDateTime nextRetryAt = null;
        if (retry) {
            int index = Math.min(Math.max(attempts - 1, 0), EnergyMeterConstants.RETRY_DELAY_MINUTES.length - 1);
            nextRetryAt = now.plusMinutes(EnergyMeterConstants.RETRY_DELAY_MINUTES[index]);
        }
        String status = retry ? EnergyMeterConstants.TASK_RETRY : EnergyMeterConstants.TASK_FAILED;
        LocalDateTime finalNextRetryAt = nextRetryAt;
        transactionTemplate.executeWithoutResult(state -> {
            taskMapper.markFailed(task.getId(), status, safeCode, safeMessage, finalNextRetryAt, now);
            deviceMapper.markCollectFailure(task.getMeterId(), safeCode, safeMessage);
        });
        log.warn("同步任务失败 taskId={} meterId={} code={} attempts={} 将重试={}",
                task.getId(), task.getMeterId(), safeCode, attempts, retry);
    }

    private void finalizeRun(Long runId) {
        TaskStatusCount count = taskMapper.countByRunId(runId);
        runMapper.refreshTotals(runId);
        int total = count == null || count.getTotal() == null ? 0 : count.getTotal();
        int success = count == null || count.getSuccess() == null ? 0 : count.getSuccess();
        int empty = count == null || count.getEmpty() == null ? 0 : count.getEmpty();
        int failed = count == null || count.getFailed() == null ? 0 : count.getFailed();
        int retry = count == null || count.getRetry() == null ? 0 : count.getRetry();
        int pending = count == null || count.getPending() == null ? 0 : count.getPending();
        int running = count == null || count.getRunning() == null ? 0 : count.getRunning();

        boolean finished = pending == 0 && running == 0 && retry == 0;
        String status;
        if (!finished) {
            status = EnergyMeterConstants.RUN_PARTIAL;
        } else if (failed == 0) {
            status = EnergyMeterConstants.RUN_SUCCESS;
        } else if (success + empty > 0) {
            status = EnergyMeterConstants.RUN_PARTIAL;
        } else {
            status = EnergyMeterConstants.RUN_FAILED;
        }
        EnergyMeterSyncRun update = new EnergyMeterSyncRun();
        update.setId(runId);
        update.setStatus(status);
        if (finished) {
            update.setFinishedAt(EnergyMeterSupport.now());
        }
        if (failed > 0) {
            update.setErrorSummary(EnergyMeterSupport.truncate(
                    failed + "/" + total + " 个任务失败，可在批次详情中重试", 500));
        }
        runMapper.updateById(update);
        log.info("同步批次汇总 runId={} status={} total={} success={} empty={} failed={} retry={}",
                runId, status, total, success, empty, failed, retry);
    }

    /* ------------------------------------------------------------------ */
    /* 重试与恢复                                                           */
    /* ------------------------------------------------------------------ */

    @Override
    public int recoverDueTasks() {
        LocalDateTime now = EnergyMeterSupport.now();
        int timeoutMinutes = Math.max(1, properties.getRunningTimeoutMinutes());
        LocalDateTime deadline = now.minusMinutes(timeoutMinutes);
        int handled = 0;
        int maxTotalAttempts = Math.max(1, properties.getMaxAttempts()) + 1;
        for (EnergyMeterSyncTask task : taskMapper.selectStaleRunning(deadline, 200)) {
            int attempts = task.getAttemptCount() == null ? 0 : task.getAttemptCount();
            if (attempts >= maxTotalAttempts) {
                transactionTemplate.executeWithoutResult(state -> taskMapper.markFailed(task.getId(),
                        EnergyMeterConstants.TASK_FAILED, EnergyMeterConstants.ERROR_TIMEOUT,
                        "任务执行超时（超过 " + timeoutMinutes + " 分钟）", null, now));
            } else {
                taskMapper.resetToRetry(task.getId(), now, now, "任务执行超时，自动转入重试");
            }
            handled++;
        }

        List<EnergyMeterSyncTask> dueTasks = taskMapper.selectDueRetry(now, 200);
        Set<Long> runIds = dueTasks.stream().map(EnergyMeterSyncTask::getRunId)
                .collect(Collectors.toCollection(LinkedHashSet::new));
        for (Long runId : runIds) {
            try {
                processRun(runId);
            } catch (Exception e) {
                log.error("恢复任务处理批次失败 runId=" + runId, e);
            }
        }
        handled += dueTasks.size();

        // 崩溃兜底：批次停在 RUNNING 但任务都已结束（例如汇总阶段异常退出）时重新汇总
        for (EnergyMeterSyncRun stuck : runMapper.selectStuckRuns(now.minusMinutes(5), 50)) {
            try {
                finalizeRun(stuck.getId());
                handled++;
                log.info("兜底汇总滞留批次 runId={}", stuck.getId());
            } catch (Exception e) {
                log.error("兜底汇总批次失败 runId=" + stuck.getId(), e);
            }
        }
        if (handled > 0) {
            log.info("同步恢复作业处理 {} 个任务/批次", handled);
        }
        return handled;
    }

    /* ------------------------------------------------------------------ */
    /* 内部工具                                                             */
    /* ------------------------------------------------------------------ */

    private EnergyMeterSyncRun newRun(String triggerType, String runKey, LocalDate scheduleDate,
                                      LocalDate start, LocalDate end) {
        EnergyMeterSyncRun run = new EnergyMeterSyncRun();
        run.setRunKey(runKey);
        run.setTriggerType(triggerType);
        run.setScheduleDate(scheduleDate);
        run.setRangeStart(start);
        run.setRangeEnd(end);
        run.setStatus(EnergyMeterConstants.RUN_PENDING);
        run.setTotalTasks(0);
        run.setSuccessTasks(0);
        run.setEmptyTasks(0);
        run.setFailedTasks(0);
        run.setCreateUser(EnergyMeterSupport.currentUserId());
        return run;
    }

    private void applyRunRange(EnergyMeterSyncRun run, LocalDate start, LocalDate end, int taskCount) {
        EnergyMeterSyncRun update = new EnergyMeterSyncRun();
        update.setId(run.getId());
        update.setRangeStart(start);
        update.setRangeEnd(end);
        update.setTotalTasks(taskCount);
        runMapper.updateById(update);
    }

    private int createDeviceTasks(Long runId, EnergyMeterDevice device, LocalDate start, LocalDate end,
                                  boolean includeReadings) {
        if (EnergyMeterConstants.isManual(device.getSourceType())) {
            return 0;
        }
        int created = createTasksFor(runId, device,
                EnergyMeterConstants.usageSource(device.getMeterType()),
                EnergyMeterConstants.GRANULARITY_DAY, start, end,
                Math.max(1, properties.getMaxShardDays()));
        if (includeReadings && !EnergyMeterConstants.isGas(device.getMeterType())) {
            // 示值接口的查询跨度限制比用量接口更严（集社云为 15 天），必须用更小的分片。
            // 气表示值接口只支持 mi15（15 分钟）粒度，96 点/天，数据量过大，暂不创建示值任务。
            created += createTasksFor(runId, device,
                    EnergyMeterConstants.readingSource(device.getMeterType()),
                    EnergyMeterConstants.GRANULARITY_POINT, start, end,
                    Math.max(1, properties.getMaxReadingShardDays()));
        }
        return created;
    }

    private int createTasksFor(Long runId, EnergyMeterDevice device, String sourceApi, String granularity,
                               LocalDate start, LocalDate end, int shardDays) {
        int created = 0;
        LocalDate cursor = start;
        while (!cursor.isAfter(end)) {
            LocalDate shardEnd = cursor.plusDays(shardDays - 1L);
            if (shardEnd.isAfter(end)) {
                shardEnd = end;
            }
            created += taskMapper.insertIgnore(buildTask(runId, device, sourceApi, granularity, cursor, shardEnd));
            cursor = shardEnd.plusDays(1);
        }
        return created;
    }

    private EnergyMeterSyncTask buildTask(Long runId, EnergyMeterDevice device, String sourceApi,
                                          String granularity, LocalDate start, LocalDate end) {
        EnergyMeterSyncTask task = new EnergyMeterSyncTask();
        task.setRunId(runId);
        task.setTenantId(device.getTenantId());
        task.setMeterId(device.getId());
        task.setSourceApi(sourceApi);
        task.setGranularity(granularity);
        task.setRangeStart(start);
        task.setRangeEnd(end);
        task.setStatus(EnergyMeterConstants.TASK_PENDING);
        return task;
    }

    /**
     * 自动同步起点：优先补齐进度缺口，同时始终回查最近 N 天（默认 3 天）。
     */
    private LocalDate resolveAutoStart(EnergyMeterDevice device, LocalDate target) {
        LocalDate base = device.getSyncStartDate() != null
                ? device.getSyncStartDate()
                : LocalDate.of(target.getYear() - 1, 1, 1);
        LocalDate start = base;
        EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
                EnergyMeterConstants.usageSource(device.getMeterType()), EnergyMeterConstants.GRANULARITY_DAY);
        if (checkpoint != null && checkpoint.getLastCompleteDate() != null) {
            LocalDate next = checkpoint.getLastCompleteDate().plusDays(1);
            if (next.isAfter(start)) {
                start = next;
            }
        }
        int lookback = Math.max(1, properties.getLookbackDays());
        LocalDate lookbackStart = target.minusDays(lookback - 1L);
        if (start.isAfter(lookbackStart)) {
            start = lookbackStart;
        }
        if (start.isBefore(base)) {
            start = base;
        }
        if (start.isAfter(target)) {
            return null;
        }
        return start;
    }

    /** 一次任务的拉取结果。 */
    private static class FetchResult {
        private int fetched;
        private String metricCode;
        private List<EnergyMeterMeasurement> measurements = new ArrayList<>();
    }
}