package com.purui.prims.headless.server.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.purui.prims.auth.api.authentication.pojo.User;
import com.purui.prims.common.jsqlparser.SqlSelectHelper;
import com.purui.prims.common.pojo.enums.AggOperatorEnum;
import com.purui.prims.common.pojo.enums.AuthType;
import com.purui.prims.common.pojo.enums.QueryType;
import com.purui.prims.common.pojo.enums.StatusEnum;
import com.purui.prims.common.pojo.enums.TypeEnums;
import com.purui.prims.common.pojo.exception.InvalidArgumentException;
import com.purui.prims.common.util.BeanMapper;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.headless.api.pojo.DataSetDetail;
import com.purui.prims.headless.api.pojo.QueryConfig;
import com.purui.prims.headless.api.pojo.request.DataSetReq;
import com.purui.prims.headless.api.pojo.request.QueryDataSetReq;
import com.purui.prims.headless.api.pojo.request.QueryMultiSqlReq;
import com.purui.prims.headless.api.pojo.request.QuerySqlReq;
import com.purui.prims.headless.api.pojo.request.QueryStructReq;
import com.purui.prims.headless.api.pojo.request.SchemaFilterReq;
import com.purui.prims.headless.api.pojo.request.SemanticQueryReq;
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.api.pojo.response.DimensionResp;
import com.purui.prims.headless.api.pojo.response.DomainResp;
import com.purui.prims.headless.api.pojo.response.MetricResp;
import com.purui.prims.headless.api.pojo.response.ModelResp;
import com.purui.prims.headless.api.pojo.response.RoleAuthResp;
import com.purui.prims.headless.api.pojo.response.SemanticQueryResp;
import com.purui.prims.headless.api.pojo.response.SemanticSchemaResp;
import com.purui.prims.headless.server.persistence.dataobject.DataSetDO;
import com.purui.prims.headless.server.persistence.mapper.DataSetDOMapper;
import com.purui.prims.headless.server.pojo.MetaFilter;
import com.purui.prims.headless.server.pojo.MetricFilter;
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.DimensionService;
import com.purui.prims.headless.server.service.DomainService;
import com.purui.prims.headless.server.service.MetricService;
import com.purui.prims.headless.server.service.ModelRelaService;
import com.purui.prims.headless.server.service.ModelService;
import com.purui.prims.headless.server.service.RoleAuthService;
import com.purui.prims.headless.server.service.SchemaService;
import com.purui.prims.headless.server.service.SemanticLayerService;
import com.purui.prims.headless.server.utils.DorisUtils;
import com.purui.prims.headless.server.utils.QueryReqConverter;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.statement.select.WithItem;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.Comparator;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DataSetServiceImpl
    extends ServiceImpl<DataSetDOMapper, DataSetDO> implements DataSetService {

    @Autowired
    private DomainService domainService;

    @Lazy
    @Autowired
    private DimensionService dimensionService;

    @Lazy
    @Autowired
    private MetricService metricService;

    @Lazy
    @Autowired
    private ModelRelaService modelRelaService;

    @Override
    public DataSetResp save(DataSetReq dataSetReq, User user) {
        if (StringUtils.isEmpty(dataSetReq.getBizName())) {
            throw new RuntimeException("业务名称不能为空");
        }
        if (dataSetReq.getDataSetDetail() == null || CollectionUtils.isEmpty(dataSetReq.getDataSetDetail().getDataSetModelConfigs())) {
            throw new RuntimeException("数据集配置不能为空");
        }
        if (dataSetReq.getDataSetDetail().getDataSetModelConfigs().stream().anyMatch(cfg -> cfg.getId() == null)) {
            throw new RuntimeException("数据集配置中模型ID不能为空");
        }
        if (dataSetReq.getModelRela() != null) {
            dataSetReq.getModelRela().forEach(modelRela -> {
                if (modelRela.getId() == null) {
                    modelRelaService.save(modelRela, user);
                } else {
                    modelRelaService.update(modelRela, user);
                }
            });
        }
        dataSetReq.createdBy(user.getUserName());
        DataSetDO dataSetDO = convert(dataSetReq);
        dataSetDO.setStatus(StatusEnum.ONLINE.getCode());
        DataSetResp dataSetResp = convert(dataSetDO);
        conflictCheck(dataSetResp);
        if (dataSetDO.getId() == null) {
            save(dataSetDO);
        } else {
            updateById(dataSetDO);
        }
        dataSetResp.setId(dataSetDO.getId());
        return dataSetResp;
    }

    @Override
    public DataSetResp update(DataSetReq dataSetReq, User user) {
        dataSetReq.updatedBy(user.getUserName());
        DataSetDO dataSetDO = convert(dataSetReq);
        DataSetResp dataSetResp = convert(dataSetDO);
        conflictCheck(dataSetResp);
        updateById(dataSetDO);
        return dataSetResp;
    }

    @Override
    public DataSetResp getDataSet(Long id) {
        DataSetDO dataSetDO = getById(id);
        return convert(dataSetDO);
    }

    @Override
    public List<DataSetResp> getDataSetList(MetaFilter metaFilter, User user) {
        QueryWrapper<DataSetDO> wrapper = new QueryWrapper<>();
        if (metaFilter.getDomainId() != null) {
            wrapper.lambda().eq(DataSetDO::getDomainId, metaFilter.getDomainId());
        }
        if (!CollectionUtils.isEmpty(metaFilter.getIds())) {
            wrapper.lambda().in(DataSetDO::getId, metaFilter.getIds());
        }
        if (metaFilter.getStatus() != null) {
            wrapper.lambda().eq(DataSetDO::getStatus, metaFilter.getStatus());
        }
        if (metaFilter.getName() != null) {
            wrapper.lambda().eq(DataSetDO::getName, metaFilter.getName());
        }
        if (!CollectionUtils.isEmpty(metaFilter.getNames())) {
            wrapper.lambda().in(DataSetDO::getName, metaFilter.getNames());
        }
        wrapper.lambda().ne(DataSetDO::getStatus, StatusEnum.DELETED.getCode());

        // dataset控制权限
        RoleAuthService roleAuthService = ContextUtils.getBean(RoleAuthService.class);
        List<RoleAuthResp> roleAuthDOList = roleAuthService.getDatasetRoleAuthList(user.getRoleId());
        if (!CollectionUtils.isEmpty(roleAuthDOList)) {
            List<Long> datasetIds = roleAuthDOList.stream().map(RoleAuthResp::getDataId).collect(Collectors.toList());
            wrapper.lambda().in(DataSetDO::getId, datasetIds);
        }

        return list(wrapper).stream().map(this::convert).collect(Collectors.toList());
    }

    @Override
    public List<DataSetResp> getDataSetList(MetaFilter metaFilter) {
        return getDataSetList(metaFilter, User.getFakeUser());
    }

    @Override
    public void delete(Long id, User user) {
        DataSetDO dataSetDO = getById(id);
        dataSetDO.setStatus(StatusEnum.DELETED.getCode());
        dataSetDO.setUpdatedBy(user.getUserName());
        dataSetDO.setUpdatedAt(new Date());
        updateById(dataSetDO);

        // 删除持久化配置
        DatasetPersistenceConfigService datasetPersistenceConfigService = ContextUtils.getBean(DatasetPersistenceConfigService.class);
        DatasetPersistenceConfigResp configResp = datasetPersistenceConfigService.getByDatasetId(id);
        if (configResp != null) {
            datasetPersistenceConfigService.delete(configResp.getId());
        }
    }

    @Override
    public List<DataSetResp> getDataSets(User user) {
        MetaFilter metaFilter = new MetaFilter();
        return getDataSetsByAuth(user, metaFilter);
    }

    @Override
    public List<DataSetResp> getDataSets(String dataSetName, User user) {
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setName(dataSetName);
        return getDataSetsByAuth(user, metaFilter);
    }

    @Override
    public List<DataSetResp> getDataSets(List<String> dataSetNames, User user) {
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setNames(dataSetNames);
        return getDataSetsByAuth(user, metaFilter);
    }

    @Override
    public boolean publish(Long id, User user) {
        DataSetDO dataSetDO = getById(id);
        if (dataSetDO != null) {
            // 已发布
            if (StatusEnum.ONLINE.getCode().equals(dataSetDO.getStatus())) {
                return true;
            }
            dataSetDO.setStatus(StatusEnum.ONLINE.getCode());
            dataSetDO.setUpdatedAt(new Date());
            dataSetDO.setUpdatedBy(user.getUserName());
            accelerate(dataSetDO);
            return updateById(dataSetDO);
        }
        return false;
    }

    // 物化加速
    @Async
    public void accelerate(DataSetDO dataSetDO) {
        if (!dataSetDO.getStatus().equals(StatusEnum.ONLINE.getCode())) {
            return;
        }
        // todo: 根据配置的调度周期不同，设置物化视图更新周期 [persistenceConf]
        String viewName = DorisUtils.getDataSetViewName(dataSetDO.getBizName());
        String sql = buildSql(dataSetDO.getId());
        // 仅针对子查询进行加速
        sql = extractWithSql(sql);
        DatabaseService databaseService = ContextUtils.getBean(DatabaseService.class);
        ModelService modelService = ContextUtils.getBean(ModelService.class);
        List<ModelResp> modelList = modelService.getAllModelByDomainIds(Arrays.asList(dataSetDO.getDomainId()));
        if (!CollectionUtils.isEmpty(modelList)) {
            DatabaseResp databaseResp = databaseService.getDatabase(modelList.stream().findFirst().get().getDatabaseId());
            DorisUtils.accelerateModelOrDataset(viewName, null, sql, databaseResp);
        }
    }

    // 解析with语句中的子查询
    private String extractWithSql(String sql) {
        List<WithItem> withItems = SqlSelectHelper.getWithItem(sql);
        if (!CollectionUtils.isEmpty(withItems)) {
            return withItems.stream().findFirst().get().getSelect().getPlainSelect().toString();
        }
        return null;
    }

    private SemanticSchemaResp getSemanticSchema(DataSetResp dataSetResp) {
        SchemaFilterReq schemaFilterReq = new SchemaFilterReq();
        schemaFilterReq.setDataSetId(dataSetResp.getId());
        schemaFilterReq.setModelIds(dataSetResp.getAllModels());
        SchemaService schemaService = ContextUtils.getBean(SchemaService.class);
        return schemaService.fetchSemanticSchema(schemaFilterReq);
    }

    // 构建用于直接查询视图的sql(仅用于BI数据集)
    @Override
    public String buildAsSelectSql(Long id) {
        DataSetResp dataSetResp = getDataSet(id);
        String sql = DorisUtils.buildQuerySql(dataSetResp);
        if (StringUtils.isNotEmpty(sql)) {
            SemanticSchemaResp semanticSchemaResp = getSemanticSchema(dataSetResp);
            semanticSchemaResp.setDataSetResp(dataSetResp);
            sql = DorisUtils.buildAsSelectSql(semanticSchemaResp);
        }
        return sql;
    }

    @Override
    public String buildSql(Long id) {
        DataSetResp dataSet = getDataSet(id);
        SemanticQueryReq queryReq = buildSqlQuery(dataSet);
        SemanticLayerService queryService = ContextUtils.getBean(SemanticLayerService.class);
        String sql = "";
        if (queryService != null) {
            try {
                sql = queryService.buildSql(queryReq, User.getFakeUser());
            } catch (Exception e) {
                log.error("build query statement error", e);
            }
        }
        return sql;
    }

    private SemanticQueryReq buildSqlQuery(DataSetResp dataSetResp) {
        QuerySqlReq queryReq = new QuerySqlReq();
        queryReq.setSql(DorisUtils.buildQuerySql(dataSetResp));
        queryReq.setDataSetId(dataSetResp.getId());
        queryReq.setDataSetName(dataSetResp.getName());
        queryReq.setLimit(-1);
        queryReq.setModelIds(dataSetResp.getAllModels().stream().collect(Collectors.toSet()));
        return queryReq;
    }

    @Override
    public boolean unPublish(Long id, User user) {
        DataSetDO dataSetDO = getById(id);
        if (dataSetDO != null) {
            dataSetDO.setStatus(StatusEnum.OFFLINE.getCode());
            dataSetDO.setUpdatedAt(new Date());
            dataSetDO.setUpdatedBy(user.getUserName());

            // 取消加速
            DataSetResp dataSetResp = convert(dataSetDO);
            if (!CollectionUtils.isEmpty(dataSetResp.getAllModels())) {
                String viewName = DorisUtils.getDataSetViewName(dataSetDO.getBizName());
                ModelService modelService = ContextUtils.getBean(ModelService.class);
                DatabaseResp databaseResp = modelService.getDatabaseByModelId(dataSetResp.getAllModels().get(0));
                DorisUtils.deleteView(viewName, databaseResp);
            }
            return updateById(dataSetDO);
        }
        return false;
    }

    private List<DataSetResp> getDataSetsByAuth(User user, MetaFilter metaFilter) {
        List<DataSetResp> dataSetResps = getDataSetList(metaFilter);
        return getDataSetFilterByAuth(dataSetResps, user);
    }

    public List<DataSetResp> getDataSetsInheritAuth(User user, Long domainId) {
        List<DataSetResp> dataSetResps = getDataSetList(new MetaFilter());
        List<DataSetResp> inheritAuthFormDomain = getDataSetFilterByDomainAuth(dataSetResps, user);
        Set<DataSetResp> dataSetRespSet = new HashSet<>(inheritAuthFormDomain);
        List<DataSetResp> dataSetFilterByAuth = getDataSetFilterByAuth(dataSetResps, user);
        dataSetRespSet.addAll(dataSetFilterByAuth);
        if (domainId != null && domainId > 0) {
            dataSetRespSet = dataSetRespSet.stream().filter(modelResp ->
                modelResp.getDomainId().equals(domainId)).collect(Collectors.toSet());
        }
        return dataSetRespSet.stream().sorted(Comparator.comparingLong(DataSetResp::getId))
                             .collect(Collectors.toList());
    }

    private List<DataSetResp> getDataSetFilterByAuth(List<DataSetResp> dataSetResps, User user) {
        return dataSetResps.stream()
                           .filter(dataSetResp -> checkAdminPermission(user, dataSetResp))
                           .collect(Collectors.toList());
    }

    private List<DataSetResp> getDataSetFilterByDomainAuth(List<DataSetResp> dataSetResps, User user) {
        Set<DomainResp> domainResps = domainService.getDomainAuthSet(user, AuthType.ADMIN);
        if (CollectionUtils.isEmpty(domainResps)) {
            return Lists.newArrayList();
        }
        Set<Long> domainIds = domainResps.stream().map(DomainResp::getId).collect(Collectors.toSet());
        return dataSetResps.stream().filter(dataSetResp ->
            domainIds.contains(dataSetResp.getDomainId())).collect(Collectors.toList());
    }

    private DataSetResp convert(DataSetDO dataSetDO) {
        DataSetResp dataSetResp = new DataSetResp();
        BeanMapper.mapper(dataSetDO, dataSetResp);
        DataSetDetail dataSetDetail = JSON.parseObject(dataSetDO.getDataSetDetail(), DataSetDetail.class);
        if (dataSetDetail != null) {
            dataSetResp.setDataSetDetail(dataSetDetail);
            // 维度
            List<Long> dimensionIds = dataSetDetail.getDataSetModelConfigs().stream().flatMap(cfg ->
                cfg.getDimensions().stream()
            ).collect(Collectors.toList());
            MetaFilter metaFilter = new MetaFilter();
            metaFilter.setIds(dimensionIds);
            dataSetResp.setDimensionList(dimensionService.getDimensions(metaFilter));

            // 指标
            List<Long> metricIds = dataSetDetail.getDataSetModelConfigs().stream().flatMap(
                cfg -> cfg.getMetrics().stream()
            ).collect(Collectors.toList());
            metaFilter.setIds(metricIds);
            dataSetResp.setMetricList(metricService.getMetrics(metaFilter));

            dataSetResp.setModelRelas(modelRelaService.getModelRela(dataSetResp.getAllModels()));

        }
        if (dataSetDO.getQueryConfig() != null) {
            dataSetResp.setQueryConfig(JSON.parseObject(dataSetDO.getQueryConfig(), QueryConfig.class));
        }
        dataSetResp.setTypeEnum(TypeEnums.DATASET);
        return dataSetResp;
    }

    private DataSetDO convert(DataSetReq dataSetReq) {
        DataSetDO dataSetDO = new DataSetDO();
        BeanMapper.mapper(dataSetReq, dataSetDO);
        dataSetDO.setDataSetDetail(JSON.toJSONString(dataSetReq.getDataSetDetail()));
        dataSetDO.setQueryConfig(JSON.toJSONString(dataSetReq.getQueryConfig()));
        return dataSetDO;
    }

    public SemanticQueryReq convert(QueryDataSetReq queryDataSetReq) throws Exception {
        QueryStructReq queryStructReq = new QueryStructReq();
        BeanUtils.copyProperties(queryDataSetReq, queryStructReq);
        if (Objects.nonNull(queryDataSetReq.getQueryType())
            && QueryType.DETAIL.equals(queryDataSetReq.getQueryType())) {
            queryStructReq.setInnerLayerNative(true);
        }
        // 采用MultiSqlReq
        return this.convert(queryStructReq);
    }

    private QueryMultiSqlReq convert(QueryStructReq queryReq) throws Exception {
        // 修正聚合函数
        queryReq.getAggregators().stream().forEach(x -> {
            MetricFilter metricFilter = new MetricFilter();
            metricFilter.setBizName(x.getColumn());
            metricFilter.setStatus(1);
            MetricResp metric = metricService.getMetrics(metricFilter).stream().findFirst().orElse(null);
            if (metric != null && StringUtils.isNotEmpty(metric.getDefaultAgg())) {
                x.setFunc(AggOperatorEnum.valueOf(metric.getDefaultAgg()));
            }
        });
        QuerySqlReq querySqlReq = queryReq.convert();
        QueryReqConverter queryReqConverter = ContextUtils.getBean(QueryReqConverter.class);
        return queryReqConverter.buildMultiStructReq(querySqlReq.getSql(), null);
    }

    public static boolean checkAdminPermission(User user, DataSetResp dataSetResp) {
        if (user.isSuperAdmin()) {
            return true;
        }
        return false;
    }

    @Override
    public Map<Long, List<Long>> getModelIdToDataSetIds(List<Long> dataSetIds, User user) {
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setStatus(StatusEnum.ONLINE.getCode());
        metaFilter.setIds(dataSetIds);
        List<DataSetResp> dataSetList = getDataSetList(metaFilter);
        return dataSetList.stream()
                          .flatMap(
                              dataSetResp -> dataSetResp.getAllModels().stream().map(modelId ->
                                  Pair.of(modelId, dataSetResp.getId())))
                          .collect(Collectors.groupingBy(Pair::getLeft,
                              Collectors.mapping(Pair::getRight, Collectors.toList())));
    }

    @Override
    public Map<Long, List<Long>> getModelIdToDataSetIds() {
        return getModelIdToDataSetIds(Lists.newArrayList(), User.getFakeUser());
    }

    private void conflictCheck(DataSetResp dataSetResp) {
        List<Long> allDimensionIds = dataSetResp.dimensionIds();
        List<Long> allMetricIds = dataSetResp.metricIds();
        MetaFilter metaFilter = new MetaFilter();
        if (!CollectionUtils.isEmpty(allDimensionIds)) {
            metaFilter.setIds(allDimensionIds);
            List<DimensionResp> dimensionResps = dimensionService.getDimensions(metaFilter);
            List<String> duplicateDimensionNames = findDuplicates(dimensionResps, DimensionResp::getName);
            if (!duplicateDimensionNames.isEmpty()) {
                throw new InvalidArgumentException("存在相同的维度名: " + duplicateDimensionNames);
            }
        }
        if (!CollectionUtils.isEmpty(allMetricIds)) {
            metaFilter.setIds(allMetricIds);
            List<MetricResp> metricResps = metricService.getMetrics(metaFilter);
            List<String> duplicateMetricNames = findDuplicates(metricResps, MetricResp::getName);
            if (!duplicateMetricNames.isEmpty()) {
                throw new InvalidArgumentException("存在相同的指标名: " + duplicateMetricNames);
            }
        }
    }

    private <T, R> List<String> findDuplicates(List<T> list, Function<T, R> keyExtractor) {
        return list.stream()
                   .collect(Collectors.groupingBy(keyExtractor, Collectors.counting()))
                   .entrySet().stream()
                   .filter(entry -> entry.getValue() > 1)
                   .map(Map.Entry::getKey)
                   .map(Object::toString)
                   .collect(Collectors.toList());
    }

    public Long getDataSetIdFromSql(String sql, User user) {
        List<DataSetResp> dataSets = null;
        try {
            String tableName = SqlSelectHelper.getTableName(sql);
            dataSets = getDataSets(tableName, user);
        } catch (Exception e) {
            log.error("getDataSetIdFromSql error:{}", e);
        }
        if (CollectionUtils.isEmpty(dataSets)) {
            throw new InvalidArgumentException("从Sql参数中无法获取到DataSetId");
        }
        Long dataSetId = dataSets.get(0).getId();
        log.info("getDataSetIdFromSql dataSetId:{}", dataSetId);
        return dataSetId;
    }

    @Override
    public SemanticQueryResp queryByDataSet(QueryDataSetReq queryDataSetReq, User user) {
        if (queryDataSetReq.getDataSetId() == null) {
            throw new InvalidArgumentException("数据集ID不能为空");
        }
        try {
            SemanticLayerService queryService = ContextUtils.getBean(SemanticLayerService.class);
            SemanticQueryReq queryReq = this.convert(queryDataSetReq);
            if (queryReq != null) {
                return queryService.queryByReq(queryReq, user);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return null;
    }
}
