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

import com.purui.prims.common.pojo.DateConf;
import com.purui.prims.common.pojo.PersistenceConf;
import com.purui.prims.common.pojo.enums.QueryType;
import com.purui.prims.dscheduler.model.DSTaskRequest;
import com.purui.prims.dscheduler.service.ProjectService;
import com.purui.prims.headless.api.pojo.TimeDefaultConfig;
import com.purui.prims.headless.api.pojo.request.QueryStructReq;
import com.purui.prims.headless.api.pojo.response.DataSetResp;
import com.purui.prims.headless.api.pojo.response.DatabaseResp;
import com.purui.prims.headless.api.pojo.response.DatasetPersistenceConfigResp;
import com.purui.prims.headless.server.service.DataSetService;
import com.purui.prims.headless.server.service.DatabaseService;
import com.purui.prims.headless.server.service.DatasetPersistenceConfigService;
import com.purui.prims.headless.server.service.MetricService;
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 org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 数据集持久化任务
 *
 * @author charlie
 * @date 2024/07/29
 */
@Component
@Slf4j
public class DatasetPersistenceTask {

    DatasetPersistenceConfigService datasetPersistenceConfigService;
    MetricService metricService;
    ModelService modelService;
    ProjectService projectService;
    SchedulerService schedulerService;
    DataSetService dataSetService;
    DatabaseService databaseService;

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

    public DatasetPersistenceTask(DatasetPersistenceConfigService datasetPersistenceConfigService, MetricService metricService,
                                  ModelService modelService, ProjectService projectService, SchedulerService schedulerService,
                                  DataSetService dataSetService, DatabaseService databaseService) {
        this.datasetPersistenceConfigService = datasetPersistenceConfigService;
        this.metricService = metricService;
        this.modelService = modelService;
        this.projectService = projectService;
        this.schedulerService = schedulerService;
        this.dataSetService = dataSetService;
        this.databaseService = databaseService;
    }

    /**
     * 数据集持久化任务
     */
    @Scheduled(cron = "${persistence.dataset.cron:0 10 * * * ?}")
    public void executeTask() {
        sourceId = schedulerService.getSourceId();
        targetId = schedulerService.getTargetId();
        sourceType = schedulerService.getSourceType();
        targetType = schedulerService.getTargetType();
        // 读取持久化配置
        Map<String, List<DatasetPersistenceConfigResp>> configMap = readPersistenceConfig();
        // 将模型中相同周期、共同维度的指标持久化到同一个表
        configMap.forEach((runCycleCode, configList) -> {
            // 根据不同周期构建不同的任务
            String processName = String.format("prims_dataset_%s", runCycleCode);
            List<DSTaskRequest> taskRequests = buildTaskList(configList);
            DatasetPersistenceConfigResp resp = configList.stream().filter(x -> x.getConf() != null).findFirst().orElse(null);
            // 默认一组相同周期的持久化保持相同
            PersistenceConf conf = resp.getConf();
            schedulerService.pushWorkflow(processName, taskRequests, conf);
        });
    }

    // 检查持久化表是否存在，如果不存在则创建
    private List<String> checkTableExist(DatasetPersistenceConfigResp config) {
        if (config.getDatasetId() == null || config.getDatabaseId() == null) {
            return null;
        }
        DataSetResp dataset = dataSetService.getDataSet(config.getDatasetId());
        if (dataset == null) {
            return null;
        }
        DatabaseResp database = databaseService.getDatabase(config.getDatabaseId());
        if (database == null) {
            return null;
        }
        List<String> preSqlList = new ArrayList<>();
        String tableName = getTableName(config.getDatasetId());
        // 直接将创建的负载放到工作流中，实现跨数据源任务
        String sql = DorisUtils.buildCreateTableSql(tableName, dataset, database);
        if (StringUtils.isNotEmpty(sql)) {
            preSqlList.add(sql);
        }
        // 如何是全量，则自动清除历史数据
        PersistenceConf conf = config.getConf();
        if (conf != null && "all".equalsIgnoreCase(conf.getWay())) {
            preSqlList.add(DorisUtils.buildClearSql(tableName, conf.getPeriod()));
        }
        return preSqlList;
    }

    private String getTableName(Long datasetId) {
        return String.format("index_dataset_%s", datasetId);
    }

    private List<DSTaskRequest> buildTaskList(List<DatasetPersistenceConfigResp> configList) {
        List<DSTaskRequest> taskList =
            configList.stream().map(config -> {
                String taskName = getTableName(config.getDatasetId());
                String sql = config.getTaskScripts();
                List<String> preSql = checkTableExist(config);
                DSTaskRequest task = new DSTaskRequest().create(taskName, sql, preSql, null, sourceId, targetId, sourceType, targetType);
                task.setTaskId(config.getId().toString());
                task.setScheduleCycle(config.getRunCycleCode());
                return task;
            }).collect(Collectors.toList());
        return taskList;
    }

    private Map<String, List<DatasetPersistenceConfigResp>> readPersistenceConfig() {
        List<DatasetPersistenceConfigResp> configList = datasetPersistenceConfigService.getConfigList();
        if (CollectionUtils.isEmpty(configList)) {
            return new HashMap<>();
        }
        return configList.stream().filter(x -> StringUtils.isNotEmpty(x.getRunCycleCode())).map(this::buildTaskScript)
                         .filter(x -> StringUtils.isNotEmpty(x.getTaskScripts()))
                         .collect(Collectors.groupingBy(DatasetPersistenceConfigResp::getRunCycleCode));
    }

    private QueryStructReq buildS2Query(DatasetPersistenceConfigResp config) {
        QueryStructReq queryStructReq = new QueryStructReq();
        queryStructReq.setDataSetId(config.getDatasetId());
        queryStructReq.setDataSetName(config.getDatasetName());
        queryStructReq.setQueryType(QueryType.METRIC);
        // 日期条件
        DataSetResp dataSet = dataSetService.getDataSet(config.getDatasetId());
        TimeDefaultConfig timeDefaultConfig = dataSet.getQueryConfig().getMetricTypeDefaultConfig().getTimeDefaultConfig();
        DateConf dateConf = new DateConf();
        dateConf.setDateMode(DateConf.DateMode.valueOf(timeDefaultConfig.getTimeMode().name()));
        dateConf.setUnit(timeDefaultConfig.getUnit());
        dateConf.setPeriod(timeDefaultConfig.getPeriod());
        queryStructReq.setDateInfo(dateConf);
        return queryStructReq;
    }

    private DatasetPersistenceConfigResp buildTaskScript(DatasetPersistenceConfigResp config) {
        String sql = dataSetService.buildSql(config.getDatasetId());
        if (StringUtils.isNotEmpty(sql)) {
            config.setTaskScripts(sql);
        }
        return config;
    }

}
