package com.njcn.gather.tool.adddata.component;

import com.njcn.gather.tool.adddata.pojo.bo.AddDataBatchWriteResult;
import com.njcn.gather.tool.adddata.pojo.bo.AddDataTableDefinition;
import com.njcn.gather.tool.adddata.pojo.bo.AddDataTaskCommand;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;

/**
 * InfluxDB 补数异步任务执行器。
 */
@Slf4j
@Component
public class AddDataInfluxTaskExecutor {

    /** 多个时间点组成一个处理窗口，兼顾批量效率和内存占用。 */
    private static final int TIME_WINDOW_SIZE = 30;

    /** 任务线程池。 */
    private final ExecutorService addDataTaskExecutorService;

    /** 表定义注册器。 */
    private final AddDataTableRegistry addDataTableRegistry;

    /** 时间槽计算器。 */
    private final AddDataTimeSlotCalculator addDataTimeSlotCalculator;

    /** 数据生成器。 */
    private final AddDataValueGenerator addDataValueGenerator;

    /** InfluxDB 写入器。 */
    private final AddDataInfluxWriter addDataInfluxWriter;

    /** 状态持有器。 */
    private final AddDataTaskStatusHolder addDataTaskStatusHolder;

    public AddDataInfluxTaskExecutor(@Qualifier("addDataTaskExecutorService") ExecutorService addDataTaskExecutorService,
                                     AddDataTableRegistry addDataTableRegistry,
                                     AddDataTimeSlotCalculator addDataTimeSlotCalculator,
                                     AddDataValueGenerator addDataValueGenerator,
                                     AddDataInfluxWriter addDataInfluxWriter,
                                     AddDataTaskStatusHolder addDataTaskStatusHolder) {
        this.addDataTaskExecutorService = addDataTaskExecutorService;
        this.addDataTableRegistry = addDataTableRegistry;
        this.addDataTimeSlotCalculator = addDataTimeSlotCalculator;
        this.addDataValueGenerator = addDataValueGenerator;
        this.addDataInfluxWriter = addDataInfluxWriter;
        this.addDataTaskStatusHolder = addDataTaskStatusHolder;
    }

    /**
     * 提交后台任务。
     *
     * @param taskId 任务编号
     * @param command 任务命令
     */
    public void submit(String taskId, AddDataTaskCommand command) {
        addDataTaskExecutorService.submit(() -> execute(taskId, command));
    }

    private void execute(String taskId, AddDataTaskCommand command) {
        try {
            addDataTaskStatusHolder.markRunning(taskId);
            Map<String, List<LocalDateTime>> timeSlotsByTable = buildTimeSlotsByTable(command);
            Map<String, Set<LocalDateTime>> timeSlotLookupByTable = buildTimeSlotLookupByTable(timeSlotsByTable);
            List<LocalDateTime> mergedTimeSlots = mergeTimeSlots(timeSlotsByTable);
            Map<String, Integer> batchNoMap = new HashMap<String, Integer>();
            Map<String, List<List<Object>>> pendingRowsByTable = buildPendingRowsByTable();
            for (int startIndex = 0; startIndex < mergedTimeSlots.size(); startIndex += TIME_WINDOW_SIZE) {
                int endIndex = Math.min(startIndex + TIME_WINDOW_SIZE, mergedTimeSlots.size());
                List<LocalDateTime> timeWindow = mergedTimeSlots.subList(startIndex, endIndex);
                GeneratedBatchData generatedBatchData = generateBatchData(command, timeWindow, timeSlotLookupByTable);
                if (!writeBatchData(taskId, generatedBatchData, pendingRowsByTable, batchNoMap)) {
                    return;
                }
            }
            if (!flushRemainingBatchData(taskId, pendingRowsByTable, batchNoMap)) {
                return;
            }
            addDataTaskStatusHolder.markSuccess(taskId);
        } catch (Exception ex) {
            log.error("执行 InfluxDB 补数任务失败，taskId={}", taskId, ex);
            addDataTaskStatusHolder.markFailed(taskId, ex.getMessage());
        }
    }

    private Map<String, List<LocalDateTime>> buildTimeSlotsByTable(AddDataTaskCommand command) {
        Map<String, List<LocalDateTime>> result = new LinkedHashMap<String, List<LocalDateTime>>();
        for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
            int intervalMinutes = definition.resolveIntervalMinutes(command.getIntervalMinutes());
            List<LocalDateTime> timeSlots = addDataTimeSlotCalculator.buildTimeSlots(
                    command.getStartTime(), command.getEndTime(), intervalMinutes);
            result.put(definition.getTableName(), timeSlots);
        }
        return result;
    }

    private Map<String, Set<LocalDateTime>> buildTimeSlotLookupByTable(Map<String, List<LocalDateTime>> timeSlotsByTable) {
        Map<String, Set<LocalDateTime>> result = new LinkedHashMap<String, Set<LocalDateTime>>();
        for (Map.Entry<String, List<LocalDateTime>> entry : timeSlotsByTable.entrySet()) {
            result.put(entry.getKey(), new HashSet<LocalDateTime>(entry.getValue()));
        }
        return result;
    }

    private Map<String, List<List<Object>>> buildPendingRowsByTable() {
        Map<String, List<List<Object>>> result = new LinkedHashMap<String, List<List<Object>>>();
        for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
            result.put(definition.getTableName(), new ArrayList<List<Object>>());
        }
        return result;
    }

    private List<LocalDateTime> mergeTimeSlots(Map<String, List<LocalDateTime>> timeSlotsByTable) {
        TreeSet<LocalDateTime> merged = new TreeSet<LocalDateTime>();
        for (List<LocalDateTime> timeSlots : timeSlotsByTable.values()) {
            merged.addAll(timeSlots);
        }
        return new ArrayList<LocalDateTime>(merged);
    }

    private GeneratedBatchData generateBatchData(AddDataTaskCommand command, List<LocalDateTime> timeWindow,
                                                 Map<String, Set<LocalDateTime>> timeSlotLookupByTable) {
        Map<String, List<List<Object>>> rowsByTable = new LinkedHashMap<String, List<List<Object>>>();
        for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
            rowsByTable.put(definition.getTableName(), new ArrayList<List<Object>>());
        }
        for (LocalDateTime timeSlot : timeWindow) {
            for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
                Set<LocalDateTime> tableTimeSlots = timeSlotLookupByTable.get(definition.getTableName());
                if (!tableTimeSlots.contains(timeSlot)) {
                    continue;
                }
                List<List<Object>> rows = rowsByTable.get(definition.getTableName());
                for (String lineId : command.getLineIds()) {
                    for (String phaseCode : definition.getPhaseCodes()) {
                        rows.add(addDataValueGenerator.generateRow(definition, lineId, timeSlot, phaseCode));
                    }
                }
            }
        }
        return new GeneratedBatchData(rowsByTable);
    }

    private boolean writeBatchData(String taskId, GeneratedBatchData generatedBatchData,
                                   Map<String, List<List<Object>>> pendingRowsByTable, Map<String, Integer> batchNoMap) {
        for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
            List<List<Object>> rows = generatedBatchData.getRows(definition.getTableName());
            if (rows.isEmpty()) {
                continue;
            }
            List<List<Object>> pendingRows = pendingRowsByTable.get(definition.getTableName());
            pendingRows.addAll(rows);
            int batchSize = definition.getBatchSize();
            while (pendingRows.size() >= batchSize) {
                List<List<Object>> batchRows = new ArrayList<List<Object>>(pendingRows.subList(0, batchSize));
                int batchNo = nextBatchNo(batchNoMap, definition.getTableName());
                if (!flushBatch(taskId, definition, batchRows, batchNo)) {
                    return false;
                }
                pendingRows.subList(0, batchSize).clear();
            }
        }
        return true;
    }

    private boolean flushRemainingBatchData(String taskId, Map<String, List<List<Object>>> pendingRowsByTable,
                                            Map<String, Integer> batchNoMap) {
        for (AddDataTableDefinition definition : addDataTableRegistry.getTableDefinitions()) {
            List<List<Object>> pendingRows = pendingRowsByTable.get(definition.getTableName());
            if (pendingRows == null || pendingRows.isEmpty()) {
                continue;
            }
            int batchNo = nextBatchNo(batchNoMap, definition.getTableName());
            List<List<Object>> batchRows = new ArrayList<List<Object>>(pendingRows);
            if (!flushBatch(taskId, definition, batchRows, batchNo)) {
                return false;
            }
            pendingRows.clear();
        }
        return true;
    }

    private int nextBatchNo(Map<String, Integer> batchNoMap, String tableName) {
        Integer currentBatchNo = batchNoMap.get(tableName);
        int nextBatchNo = currentBatchNo == null ? 1 : currentBatchNo + 1;
        batchNoMap.put(tableName, nextBatchNo);
        return nextBatchNo;
    }

    private boolean flushBatch(String taskId, AddDataTableDefinition definition, List<List<Object>> batchRows, int batchNo) {
        addDataTaskStatusHolder.updateCurrentBatch(taskId, definition.getTableName(), batchNo, batchRows.size());
        AddDataBatchWriteResult writeResult = addDataInfluxWriter.writeBatch(definition, batchRows);
        addDataTaskStatusHolder.addProgress(taskId, writeResult.getInsertedCount(), writeResult.getSkippedCount(), writeResult.getFailedCount());
        if (writeResult.hasFailure()) {
            addDataTaskStatusHolder.markFailed(taskId,
                    "InfluxDB 表 " + definition.getTableName() + " 第 " + batchNo + " 批执行失败：" + writeResult.getFirstFailureMessage());
            return false;
        }
        return true;
    }

    /**
     * 当前窗口生成结果。
     */
    private static final class GeneratedBatchData {

        /** 按表名分组的待写入行数据。 */
        private final Map<String, List<List<Object>>> rowsByTable;

        private GeneratedBatchData(Map<String, List<List<Object>>> rowsByTable) {
            this.rowsByTable = rowsByTable;
        }

        private List<List<Object>> getRows(String tableName) {
            List<List<Object>> rows = rowsByTable.get(tableName);
            return rows == null ? Collections.emptyList() : rows;
        }
    }
}
