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

import com.purui.prims.common.pojo.DateConf;
import com.purui.prims.common.pojo.enums.DataTypeEnums;
import com.purui.prims.common.pojo.enums.TimeDimensionEnum;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.common.util.DateModeUtils;
import com.purui.prims.headless.api.pojo.MeasureParam;
import com.purui.prims.headless.api.pojo.MetricParam;
import com.purui.prims.headless.api.pojo.TimeDefaultConfig;
import com.purui.prims.headless.api.pojo.enums.EngineType;
import com.purui.prims.headless.api.pojo.enums.MetricType;
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.DimSchemaResp;
import com.purui.prims.headless.api.pojo.response.DimensionResp;
import com.purui.prims.headless.api.pojo.response.MetricResp;
import com.purui.prims.headless.api.pojo.response.MetricSchemaResp;
import com.purui.prims.headless.api.pojo.response.SemanticSchemaResp;
import com.purui.prims.headless.core.translator.calcite.schema.SchemaBuilder;
import com.purui.prims.headless.core.translator.calcite.schema.SemanticSchema;
import com.purui.prims.headless.core.translator.calcite.sql.TableView;
import com.purui.prims.headless.core.translator.calcite.sql.node.DimensionNode;
import com.purui.prims.headless.core.translator.calcite.sql.node.MetricNode;
import com.purui.prims.headless.server.manager.SemanticSchemaManager;
import com.purui.prims.headless.server.pojo.MetaFilter;
import com.purui.prims.headless.server.service.DatabaseService;
import com.purui.prims.headless.server.service.DimensionService;
import com.purui.prims.headless.server.service.MetricService;
import lombok.extern.slf4j.Slf4j;
import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.sql.validate.SqlValidatorScope;
import org.apache.commons.lang3.StringUtils;
import org.assertj.core.util.Lists;

import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;

/**
 * Doris工具类
 *
 * @author charlie
 * @date 2024-07-20
 */
@Slf4j
public class DorisUtils {

    /**
     * 新建物化视图
     *
     * @param name     名称
     * @param sql      sql语句
     * @param database 数据库
     * @return
     */
    public static boolean createView(String name, String sql, DatabaseResp database) {
        if (database.isSupportMaterializedView() && StringUtils.isNotEmpty(sql)) {
            sql = String.format("create materialized view `%s` AS %s", name, sql);
            DatabaseService datasourceService = ContextUtils.getBean(DatabaseService.class);
            datasourceService.runOnly(sql, database);
        }
        return false;
    }

    /**
     * 删除物化视图
     *
     * @param name      视图名称
     * @param tableName 表名称
     * @param database  数据库
     */
    public static void deleteView(String name, String tableName, DatabaseResp database) {
        if (database.isSupportMaterializedView()) {
            // count(distinct x) 会自动转换为 bitmap_union(to_bitmap(x))
            String sql = null;
            if (StringUtils.isEmpty(tableName)) {
                sql = String.format("drop materialized view IF EXISTS `%s`", name);
            } else {
                sql = String.format("drop materialized view IF EXISTS `%s` on %s", name, tableName);
            }
            DatabaseService databaseService = ContextUtils.getBean(DatabaseService.class);
            databaseService.runOnly(sql, database);
        }
    }

    public static void deleteView(String name, DatabaseResp database) {
        deleteView(name, null, database);
    }

    /**
     * 创建数据集对应的持久化表，已交调度平台完成，实现跨库持久化
     *
     * @param dataset  数据集
     * @param database 数据库
     */
    @Deprecated
    public static void createDatasetTable(DataSetResp dataset, DatabaseResp database) {
        if (dataset.getDataSetDetail() != null) {
            String tableName = dataset.getBizName();
            List<DimensionResp> dimensions = getDimensions(dataset);
            List<MetricResp> metrics = getMetrics(dataset);
            createTable(tableName, dimensions, metrics, database);
        }
    }

    /**
     * 加速模型或数据集关联查询(多表关联加速)
     *
     * @param name         视图名称
     * @param partitionKey 分区键
     * @param sql          sql语句
     * @param database     数据库
     */
    public static void accelerateModelOrDataset(String name, String partitionKey, String sql, DatabaseResp database) {
        // 默认按照每4小时更新一次
        String template = "CREATE MATERIALIZED VIEW IF NOT EXISTS %s REFRESH ON SCHEDULE EVERY 4 HOUR \n";
        if (StringUtils.isNotEmpty(partitionKey)) {
            template += "        partition by(%s)\n";
        }
        template += "        DISTRIBUTED BY RANDOM BUCKETS 20\n"
                    + "        PROPERTIES ('replication_num' = '3') AS %s";
        if (StringUtils.isNotEmpty(partitionKey)) {
            sql = String.format(template, name, partitionKey, sql);
        } else {
            sql = String.format(template, name, sql);
        }
        DatabaseService databaseService = ContextUtils.getBean(DatabaseService.class);
        databaseService.runOnly(sql, database);
    }

    private static List<DimensionResp> getDimensions(DataSetResp dataset) {
        DimensionService dimensionService = ContextUtils.getBean(DimensionService.class);
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setIds(dataset.dimensionIds());
        List<DimensionResp> dimensions = dimensionService.getDimensions(metaFilter);
        return dimensions;
    }

    private static List<MetricResp> getMetrics(DataSetResp dataset) {
        MetricService metricService = ContextUtils.getBean(MetricService.class);
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setIds(dataset.metricIds());
        List<MetricResp> metrics = metricService.getMetrics(metaFilter);
        return metrics;
    }

    public static String getIndexViewName(String bizName) {
        return String.format("mv_idx_%s", bizName);
    }

    public static String getDataSetViewName(String bizName) {
        return String.format("mv_ds_%s", bizName);
    }

    public static String buildAsSelectSql(SemanticSchemaResp semanticSchemaResp) {
        DataSetResp dataset = semanticSchemaResp.getDataSetResp();
        SemanticSchema semanticSchema = SemanticSchema.newBuilder(semanticSchemaResp.getSchemaKey()).build();
        List<DimSchemaResp> dimensions = semanticSchemaResp.getDimensions();
        List<MetricSchemaResp> metrics = semanticSchemaResp.getMetrics();
        String tableName = getDataSetViewName(dataset.getBizName());
        TableView tableView = new TableView();
        SqlValidatorScope scope = null;
        try {
            SemanticSchemaManager semanticSchemaManager = ContextUtils.getBean(SemanticSchemaManager.class);
            semanticSchema.setSemanticModel(semanticSchemaManager.getSemanticModel(semanticSchemaResp));
            scope = SchemaBuilder.getScope(semanticSchema);
        } catch (Exception e) {
            log.info("build scope error", e);
        }
        if (scope == null) {
            return null;
        }
        EngineType engineType = EngineType.DORIS;
        for (DimSchemaResp dimensionResp : dimensions) {
            SqlNode node = null;
            try {
                node = DimensionNode.parse(dimensionResp.getBizName(), scope, engineType);
                node = DimensionNode.buildAs(dimensionResp.getName(), node);
                // 此时是明细查询，不需要按维度进行group，统一按指标处理
                tableView.getMeasure().add(node);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        Set<String> metricNames = new TreeSet<>();
        for (MetricSchemaResp metricResp : metrics) {
            if (MetricType.COMPOSITE.name().equals(metricResp.getType())) {
                for (MetricParam metric : metricResp.getMetricDefineByMetricParams().getMetrics()) {
                    MetricResp subMetricResp =
                        semanticSchemaResp.getAllMetrics().stream().filter(m -> m.getBizName().equals(metric.getBizName())).findFirst()
                                          .orElse(null);
                    if (subMetricResp != null) {
                        if (!metricNames.contains(subMetricResp.getBizName())) {
                            SqlNode subNode = buildAsMetricNode(subMetricResp, scope, engineType);
                            tableView.getMeasure().add(subNode);
                            metricNames.add(subMetricResp.getBizName());
                        }
                    }
                }
            } else {
                if (!metricNames.contains(metricResp.getBizName())) {
                    SqlNode node = buildAsMetricNode(metricResp, scope, engineType);
                    tableView.getMeasure().add(node);
                    metricNames.add(metricResp.getBizName());
                }
            }
        }
        tableView.setTable(new SqlIdentifier(tableName, SqlParserPos.ZERO));
        return tableView.build().toString();
    }

    private static SqlNode buildAsMetricNode(MetricResp metricResp, SqlValidatorScope scope, EngineType engineType) {
        SqlNode node = null;
        try {
            String bizName = metricResp.getBizName();
            if (metricResp.getMetricDefineByMeasureParams() != null) {
                // 派生指标
                MeasureParam measureParam = metricResp.getMetricDefineByMeasureParams().getMeasures().stream().findFirst().orElse(null);
                if (measureParam != null) {
                    bizName = measureParam.getBizName();
                }
            }
            node = MetricNode.parse(bizName, scope, engineType);
            node = MetricNode.buildAs(metricResp.getName(), node);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return node;
    }

    // 根据维度和指标构建SQL语句，用于持久化
    public static String buildQuerySql(DataSetResp dataset) {
        List<DimensionResp> dimensions = getDimensions(dataset);
        List<MetricResp> metrics = getMetrics(dataset);
        List<String> columns = dimensions.stream().map(dimension -> dimension.getName()).collect(Collectors.toList());
        columns.addAll(metrics.stream().map(metric -> metric.getName()).collect(Collectors.toList()));
        String sql = String.format("select %s from %s", String.join(",", columns), dataset.getName());
        // TODO 根据不用的数据库类型，采用不同的日期方法
        TimeDefaultConfig defaultConfig = dataset.getQueryConfig().getMetricTypeDefaultConfig().getTimeDefaultConfig();
        String startDate = "";
        String endDate = "";
        switch (defaultConfig.getTimeMode()) {
            case LAST:
                // between and mode
                break;
            case RECENT:
                startDate =
                    String.format("DATE_SUB(NOW(),interval %s %s)", defaultConfig.getUnit(),
                        defaultConfig.getPeriod().toUpperCase(Locale.ROOT));
                endDate = "NOW()";
                break;
            default:
                break;
        }
        DateModeUtils dateModeUtils = ContextUtils.getBean(DateModeUtils.class);
        DateConf dateInfo = new DateConf();
        dateInfo.setPeriod(defaultConfig.getPeriod());
        String dateColumn = dateModeUtils.getSysDateCol(dateInfo);
        if (StringUtils.isNotEmpty(startDate) && StringUtils.isNotEmpty(endDate)) {
            sql = String.format("%s where %s between %s and %s ", sql, dateColumn, startDate,
                endDate);
        }
        return sql;
    }

    /**
     * 根据维度和指标创建持久化表
     *
     * @param tableName  表名
     * @param dimensions 维度
     * @param metrics    指标
     * @param database   数据库
     */
    public static void createTable(String tableName, List<DimensionResp> dimensions, List<MetricResp> metrics,
                                   DatabaseResp database) {
        String sql = buildCreateTableSql(tableName, dimensions, metrics, null);
        DatabaseService databaseService = ContextUtils.getBean(DatabaseService.class);
        databaseService.runOnly(sql, database);
    }

    /**
     * 根据指标创建持久化表
     *
     * @param tableName 表名
     * @param metric    指标
     * @param database  数据库
     */
    public static void createMetricTable(String tableName, MetricResp metric, DatabaseResp database) {
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setIds(
            metric.getRelateDimension().getDrillDownDimensions().stream().map(d -> d.getDimensionId()).collect(
                Collectors.toList()));
        DimensionService dimensionService = ContextUtils.getBean(DimensionService.class);
        List<DimensionResp> dimensions = dimensionService.getDimensions(metaFilter);
        // 根据相同模型中的指标统一合并到一个持久化表中
        String sql = buildCreateTableSql(tableName, dimensions, Lists.list(metric), null);
        DatabaseService databaseService = ContextUtils.getBean(DatabaseService.class);
        databaseService.runOnly(sql, database);
    }

    public static String buildClearSql(String tableName, String runCycleCode) {
        TimeDimensionEnum timeDimensionEnum = TimeDimensionEnum.valueOf(runCycleCode.toUpperCase(Locale.ROOT));
        if (timeDimensionEnum != null) {
            return String.format("delete from %s where %s is not null;", tableName, timeDimensionEnum.getName());
        } else {
            return "";
        }
    }

    /**
     * Doris中根据维度数组和指标数组创建对应的物理表
     *
     * @param tableName  表名
     * @param dimensions 维度
     * @param metrics    指标
     * @return
     */
    public static String buildCreateTableSql(String tableName, List<DimensionResp> dimensions,
                                             List<MetricResp> metrics, String runCycleCode) {
        List<String> columns = dimensions.stream().map(dimension -> String.format("%s %s", dimension.getBizName(),
            convertColumnType(dimension.getDataType()))).collect(Collectors.toList());

        String hashKeys = dimensions.stream().map(dimension -> dimension.getBizName())
                                    .collect(Collectors.joining(","));

        // 统一采用明细模型
        columns.addAll(metrics.stream().map(metric -> String.format("%s %s", metric.getBizName(),
            convertColumnType(DataTypeEnums.DOUBLE))).collect(Collectors.toList()));

        // 增加时间戳(datax可能出现不传此字段时任务失败?)
        columns.add(String.format("%s %s", "update_time", "datetime DEFAULT CURRENT_TIMESTAMP"));
        return String.format(
            "CREATE TABLE IF NOT EXISTS %s (%s) DUPLICATE KEY (%s) DISTRIBUTED BY HASH(%s) BUCKETS 32 PROPERTIES ('replication_num' = '3');",
            tableName, String.join(",", columns), hashKeys, hashKeys);
    }

    // 仅返回创建表的SQL语句
    public static String buildCreateTableSql(String tableName, DataSetResp dataset, DatabaseResp database) {
        if (dataset.getDataSetDetail() != null) {
            List<DimensionResp> dimensions = getDimensions(dataset);
            List<MetricResp> metrics = getMetrics(dataset);
            return buildCreateTableSql(tableName, dimensions, metrics, null);
        }
        return null;
    }

    private static int getLength(DataTypeEnums typeEnums) {
        if (typeEnums.equals(DataTypeEnums.VARCHAR)) {
            return 255;
        }
        return 0;
    }

    private static String convertColumnType(DataTypeEnums type) {
        if (type == null) {
            return "VARCHAR(200)";
        }
        boolean isString = type.equals(DataTypeEnums.VARCHAR);
        if (isString) {
            int length = getLength(type);
            return String.format("VARCHAR(%d)", length);
        }
        return type.getType();
    }

}
