package com.purui.prims.headless.server.schedule;

import com.purui.prims.common.pojo.PersistenceConf;
import com.purui.prims.common.pojo.enums.DataTypeEnums;
import com.purui.prims.common.pojo.enums.TimeDimensionEnum;
import com.purui.prims.common.util.BeanMapper;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.dscheduler.model.DSTaskRequest;
import com.purui.prims.dscheduler.service.DatasourceService;
import com.purui.prims.dscheduler.service.ProjectService;
import com.purui.prims.headless.api.pojo.request.MetricPersistenceConfigReq;
import com.purui.prims.headless.api.pojo.response.DimensionResp;
import com.purui.prims.headless.api.pojo.response.MetricPersistenceConfigResp;
import com.purui.prims.headless.api.pojo.response.MetricResp;
import com.purui.prims.headless.api.pojo.response.ModelResp;
import com.purui.prims.headless.server.pojo.MetaFilter;
import com.purui.prims.headless.server.pojo.MetricVersion;
import com.purui.prims.headless.server.service.DatabaseService;
import com.purui.prims.headless.server.service.DimensionService;
import com.purui.prims.headless.server.service.MetricPersistenceConfigService;
import com.purui.prims.headless.server.service.MetricService;
import com.purui.prims.headless.server.service.MetricVersionService;
import com.purui.prims.headless.server.service.ModelService;
import com.purui.prims.headless.server.service.SchedulerService;
import com.purui.prims.headless.server.utils.DorisUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 指标持久化任务
 *
 * @author charlie
 * @date 2024/07/29
 */
@Component
@Slf4j
public class MetricPersistenceTask {

    MetricPersistenceConfigService metricPersistenceConfigService;
    MetricService metricService;
    ModelService modelService;
    ProjectService projectService;
    SchedulerService schedulerService;
    DatabaseService databaseService;
    DimensionService dimensionService;
    DatasourceService datasourceService;

    int sourceId;
    int targetId;
    String sourceType;
    String targetType;

    public MetricPersistenceTask(MetricPersistenceConfigService metricPersistenceConfigService, MetricService metricService,
                                 ModelService modelService, ProjectService projectService, SchedulerService schedulerService,
                                 DatabaseService databaseService, DimensionService dimensionService, DatasourceService datasourceService) {
        this.metricPersistenceConfigService = metricPersistenceConfigService;
        this.metricService = metricService;
        this.modelService = modelService;
        this.projectService = projectService;
        this.schedulerService = schedulerService;
        this.databaseService = databaseService;
        this.dimensionService = dimensionService;
        this.datasourceService = datasourceService;
    }

    /**
     * 指标持久化任务(10分钟运行一次)
     */
    @Scheduled(cron = "${persistence.metric.cron:0 0/10 * * * ?}")
    public void executeTask() {
        this.runPersistNow();
    }

    public void runPersistNow() {
        sourceId = datasourceService.getSourceId();
        targetId = datasourceService.getTargetId();
        sourceType = datasourceService.getSourceType();
        targetType = datasourceService.getTargetType();
        // 读取指标持久化配置
        Map<Long, Map<String, List<MetricPersistenceConfigResp>>> modelAndRunCycle = readPersistenceConfig();
        // 将模型中相同周期、相同维度【相同维度指标自动合并】的指标持久化到同一个表
        modelAndRunCycle.forEach((modelId, runCycleMap) -> runCycleMap.forEach((runCycleCode, configList) -> {
            persistenceShareMetric(modelId, runCycleCode, configList);
        }));
    }

    /**
     * 持久化指标（已发布的版本）
     *
     * @param modelId
     * @param runCycleCode
     * @param configList
     */
    private void persistenceShareMetric(Long modelId, String runCycleCode, List<MetricPersistenceConfigResp> configList) {
        List<MetricResp> metricRespList = configList.stream().map(config -> metricService.getLastVersionMetric(config.getMetricId()))
                                                    .filter(metricResp -> (metricResp.getRelateDimension() != null
                                                                           && metricResp.getRelateDimension().getDrillDownDimensions()
                                                                              != null)).collect(Collectors.toList());

        // 根据血缘关系，放入依赖DAG
        // metricService.buildDependencyDAG(metricRespList);

        // 相同维度的指标合并（目前存在和前端单独调度冲突）
        /**
         Map<Object, List<MetricResp>> metricsGroup =
         metricRespList.stream().filter(
         metricResp -> metricResp != null && metricResp.getRelateDimension() != null)
         .collect(Collectors.groupingBy(s -> {
         return s.getRelateDimension().getDrillDownDimensions();
         }));
         */
        String processName = getProcessName(runCycleCode);
        ModelResp modelResp = modelService.getModel(modelId);
        List<DSTaskRequest> taskRequests = new ArrayList<>();

        MetricPersistenceConfigResp resp = configList.stream().filter(x -> x.getConf() != null).findFirst().orElse(null);
        // 默认一组相同周期的持久化保持相同
        PersistenceConf conf = resp.getConf();
        for (MetricResp metricResp : metricRespList) {
            // 根据持久化配置中，是否增量更新，增量更新的条件（天，小时）sql中的where条件
            String sql = metricService.buildMetricSql(modelResp, Arrays.asList(metricResp), false, conf, true);
            MetricPersistenceConfigResp configResp =
                configList.stream().filter(config -> config.getMetricId().equals(metricResp.getId())).findFirst().orElse(null);
            if (configResp != null) {
                taskRequests.addAll(buildTaskList(configResp, runCycleCode, metricResp.getId(), sql));
            }
        }
        try {
            // 提交任务
            schedulerService.pushWorkflow(processName, taskRequests, conf);
            // 更新持久化版本
            configList.forEach(config -> {
                MetricPersistenceConfigReq configReq = new MetricPersistenceConfigReq();
                BeanMapper.mapper(config, configReq);
                metricPersistenceConfigService.save(configReq);
            });
        } catch (Exception ex) {
            log.error("指标持久化任务执行失败", ex);
        }
    }

    public List<DSTaskRequest> buildTaskList(MetricPersistenceConfigResp configResp, Long metricId) {
        MetricResp metricResp = metricService.getLastVersionMetric(metricId);
        ModelResp modelResp = modelService.getModel(metricResp.getModelId());
        String sql = metricService.buildMetricSql(modelResp, Arrays.asList(metricResp), false, configResp.getConf(), true);
        return buildTaskList(configResp, configResp.getRunCycleCode(), metricId, sql);
    }

    private List<DSTaskRequest> buildTaskList(MetricPersistenceConfigResp config, String runCycleCode,
                                              Long metricId, String sql) {
        String taskName = metricService.getMetricTableName(metricId);
        // 此时合并后的metrics表，表名和任务名一致
        List<String> preSql = checkTableExist(config, taskName);
        if (sourceId == 0) {
            sourceId = datasourceService.getSourceId();
        }
        if (targetId == 0) {
            targetId = datasourceService.getTargetId();
        }
        DSTaskRequest task = new DSTaskRequest().create(taskName, sql, preSql, null, sourceId, targetId, sourceType, targetType);
        task.setTaskId(metricId.toString());
        task.setScheduleCycle(runCycleCode);
        return Arrays.asList(task);
    }

    public String getProcessName(String cycleCode) {
        return String.format("prims_metrics_%s", cycleCode.toLowerCase(Locale.ROOT));
    }

    // 构建合并后的指标表SQL
    private List<String> checkTableExist(MetricPersistenceConfigResp config, String tableName) {
        // 所有指标
        List<Long> metricsIds = Arrays.asList(config.getMetricId());
        List<String> preSqlList = new ArrayList<>();
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setIds(metricsIds);
        List<MetricResp> metricList = metricService.getMetrics(metaFilter);

        if (metricList.isEmpty()) {
            throw new RuntimeException("指标不存在!");
        }
        // 仅合并指标
        MetricResp metricResp = metricList.stream().filter(x -> x.getRelateDimension() != null).findFirst().get();
        List<Long> dimensionIds =
            metricResp.getRelateDimension().getDrillDownDimensions().stream().map(dd -> dd.getDimensionId()).collect(Collectors.toList());
        metricList.stream()
                  .filter(metric -> (metric.getRelateDimension() != null && metric.getRelateDimension().getDrillDownDimensions() != null))
                  .forEach(metric -> {
                      List<Long> currentDimensionIds =
                          metric.getRelateDimension().getDrillDownDimensions().stream().map(dd -> dd.getDimensionId())
                                .collect(Collectors.toList());
                      dimensionIds.retainAll(currentDimensionIds);
                  });
        // 主要日期维度
        if (metricResp.getDateDimId() != null) {
            // 根据日期维度的类型，调整runCycleCode
        }
        MetaFilter filter = new MetaFilter();
        filter.setIds(dimensionIds);
        if (dimensionIds.size() > 0) {
            // 如何是全量，则自动清除历史数据
            PersistenceConf conf = config.getConf();
            // 合并后的维度(取消日期维度，自动识别）
            List<DimensionResp> dimensionRespList =
                dimensionService.getDimensions(filter).stream().filter(d -> !d.isTimeDimension()).collect(Collectors.toList());

            if (StringUtils.isNotEmpty(metricResp.getStatisticsPeriod())) {
                TimeDimensionEnum timeDimensionEnum = TimeDimensionEnum.valueOf(metricResp.getStatisticsPeriod().toUpperCase(Locale.ROOT));
                if (timeDimensionEnum != null) {
                    DimensionResp timeDimension = new DimensionResp();
                    timeDimension.setDataType(DataTypeEnums.VARCHAR);
                    timeDimension.setBizName(timeDimensionEnum.getName());
                    dimensionRespList.add(timeDimension);
                }
            }

            String sql = DorisUtils.buildCreateTableSql(tableName, dimensionRespList, metricList, metricResp.getStatisticsPeriod());
            if (StringUtils.isNotEmpty(sql)) {
                preSqlList.add(sql);
            }

            if (conf != null && "all".equalsIgnoreCase(conf.getWay()) && StringUtils.isNotEmpty(metricResp.getStatisticsPeriod())) {
                preSqlList.add(DorisUtils.buildClearSql(tableName, metricResp.getStatisticsPeriod()));
            }
        }
        return preSqlList;
    }

    private Map<Long, Map<String, List<MetricPersistenceConfigResp>>> readPersistenceConfig() {
        List<MetricPersistenceConfigResp> configList = metricPersistenceConfigService.getConfigList();
        MetricVersionService metricVersionService = ContextUtils.getBean(MetricVersionService.class);
        configList = configList.stream().filter(x -> {
                                   MetricVersion version = metricVersionService.getLastVersion(x.getMetricId());
                                   // 未发布版本或已持久化过
                                   if (version == null || version.getVersionNo().equalsIgnoreCase(x.getLastVersion())) {
                                       return false;
                                   }
                                   return true;
                               }).map(config -> {
                                   // 记录需要持久化的版本
                                   MetricVersion version = metricVersionService.getLastVersion(config.getMetricId());
                                   config.setLastVersion(version.getVersionNo());
                                   return config;
                               })
                               .collect(Collectors.toList());
        // 对configList按模型和相同计算周期进行分组
        Map<Long, Map<String, List<MetricPersistenceConfigResp>>> collect =
            configList.stream().filter(config -> config.getModelId() != null && StringUtils.isNotEmpty(config.getRunCycleCode())).collect(
                Collectors.groupingBy(MetricPersistenceConfigResp::getModelId,
                    Collectors.groupingBy(MetricPersistenceConfigResp::getRunCycleCode)));
        return collect;
    }

    public MetricPersistenceConfigResp getConfig(Long metricId) {
        return metricPersistenceConfigService.getByMetricId(metricId);
    }

}
