package com.ruoyi.web.report.service.impl;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.web.report.domain.bo.ReportRebuildBo;
import com.ruoyi.web.report.domain.design.ReportAggregatePolicyDesign;
import com.ruoyi.web.report.domain.meta.ReportFieldMeta;
import com.ruoyi.web.report.domain.meta.ReportJoinMeta;
import com.ruoyi.web.report.mapper.ErpReportDefinitionMapper;
import com.ruoyi.web.report.mapper.ReportDesignMapper;
import com.ruoyi.web.report.service.IReportAggregateService;
import com.ruoyi.web.report.service.support.ReportSqlSecurityValidator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.DigestUtils;

import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.StringJoiner;

/**
 * 报表离线聚合服务实现。
 *
 * <p>存储策略：每个报表一个聚合结果表，命名为 {@code rpt_agg_report_{reportId}}。
 * 该方案便于分表隔离、冷热分离和后续单表归档。</p>
 */
@Service
public class ReportAggregateServiceImpl implements IReportAggregateService {
    /** 允许的聚合函数。 */
    private static final Set<String> SAFE_AGG_FN = new HashSet<>(Arrays.asList("SUM", "AVG", "MAX", "MIN", "COUNT"));
    /** 允许的关联类型。 */
    private static final Set<String> SAFE_JOIN_TYPE = new HashSet<>(Arrays.asList("INNER", "LEFT", "RIGHT"));

    @Autowired
    private ErpReportDefinitionMapper erpReportDefinitionMapper;

    @Autowired
    private ReportDesignMapper reportDesignMapper;

    @Autowired
    private NamedParameterJdbcTemplate namedParameterJdbcTemplate;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private ReportSqlSecurityValidator sqlSecurityValidator;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public int rebuild(Long reportId, ReportRebuildBo rebuildBo) {
        if (reportId == null) {
            throw new IllegalArgumentException("reportId 不能为空。");
        }
        if (rebuildBo == null) {
            throw new IllegalArgumentException("补算参数不能为空。");
        }
        ReportAggregatePolicyDesign policy = reportDesignMapper.selectAggregatePolicyByReportId(reportId);
        if (policy == null) {
            throw new IllegalArgumentException("未配置聚合策略，无法执行补算。");
        }
        String granularity = safeUpper(policy.getAggregateGranularity());
        if ("REALTIME".equals(granularity) || granularity.isEmpty()) {
            throw new IllegalArgumentException("当前策略为实时模式，不支持补算。");
        }
        String timeField = Objects.toString(policy.getTimeDimensionField(), "").trim();
        if (timeField.isEmpty()) {
            throw new IllegalArgumentException("聚合策略缺少 timeDimensionField。");
        }
        sqlSecurityValidator.validateIdentifier(timeField);

        LocalDateTime start = parseTime(rebuildBo.getStartTime(), true);
        LocalDateTime end = parseTime(rebuildBo.getEndTime(), false);
        if (!start.isBefore(end)) {
            throw new IllegalArgumentException("开始时间必须早于结束时间。");
        }

        String tableName = aggregateTableName(reportId);
        ensureAggregateTable(tableName);

        int totalAffected = 0;
        for (Window window : splitWindows(start, end, granularity)) {
            List<Map<String, Object>> rows = queryAggregateRows(reportId, window, timeField);
            totalAffected += upsertAggregateRows(tableName, rows, window);
        }
        return totalAffected <= 0 ? 1 : totalAffected;
    }

    /**
     * 查询某个时间窗的聚合结果行。
     */
    private List<Map<String, Object>> queryAggregateRows(Long reportId, Window window, String timeField) {
        List<ReportFieldMeta> fieldMetaList = erpReportDefinitionMapper.selectFieldMetaByReportId(reportId);
        if (fieldMetaList == null || fieldMetaList.isEmpty()) {
            throw new IllegalArgumentException("缺少字段元数据，无法聚合。");
        }
        List<ReportJoinMeta> joinMetaList = erpReportDefinitionMapper.selectJoinMetaByReportId(reportId);

        Map<String, String> fieldExprMap = new HashMap<String, String>();
        List<String> dimItems = new ArrayList<String>();
        List<String> metricItems = new ArrayList<String>();
        List<String> groupByItems = new ArrayList<String>();
        for (ReportFieldMeta meta : fieldMetaList) {
            if (meta == null || meta.getIsEnabled() == null || meta.getIsEnabled() != 1) {
                continue;
            }
            String fieldKey = meta.getFieldKey();
            sqlSecurityValidator.validateIdentifier(fieldKey);
            String sourceExpr = resolveSourceExpr(meta);
            fieldExprMap.put(fieldKey, sourceExpr);
            String aggFn = safeUpper(meta.getAggregateFn());
            if (!aggFn.isEmpty()) {
                if (!SAFE_AGG_FN.contains(aggFn)) {
                    throw new IllegalArgumentException("不支持的聚合函数: " + aggFn);
                }
                metricItems.add(aggFn + "(" + sourceExpr + ") AS " + fieldKey);
            } else if (Objects.equals(meta.getIsGroupDim(), 1)) {
                dimItems.add(sourceExpr + " AS " + fieldKey);
                groupByItems.add(sourceExpr);
            }
        }
        if (metricItems.isEmpty()) {
            throw new IllegalArgumentException("聚合字段为空，至少需要一个 aggregateFn 字段。");
        }

        StringBuilder sql = new StringBuilder();
        List<String> selectItems = new ArrayList<String>();
        selectItems.addAll(dimItems);
        selectItems.addAll(metricItems);
        sql.append("select ").append(String.join(", ", selectItems)).append(" ");
        sql.append(buildFromAndJoinSql(fieldMetaList, joinMetaList));
        sql.append(" where ").append(timeField).append(" >= :startTime and ").append(timeField).append(" <= :endTime ");
        if (!groupByItems.isEmpty()) {
            sql.append(" group by ").append(String.join(", ", groupByItems));
        }

        MapSqlParameterSource params = new MapSqlParameterSource();
        params.addValue("startTime", window.start);
        params.addValue("endTime", window.end);
        return namedParameterJdbcTemplate.queryForList(sql.toString(), params);
    }

    /**
     * 将聚合结果写入报表专属聚合表。
     */
    private int upsertAggregateRows(String tableName, List<Map<String, Object>> rows, Window window) {
        if (rows == null || rows.isEmpty()) {
            return 0;
        }
        int affected = 0;
        String upsertSql =
                "insert into " + tableName + " (stat_bucket, bucket_type, dimension_json, metric_json, dim_hash, row_count) " +
                "values (?, ?, ?, ?, ?, ?) " +
                "on duplicate key update dimension_json=values(dimension_json), metric_json=values(metric_json), row_count=values(row_count), updated_at=now()";
        for (Map<String, Object> row : rows) {
            Map<String, Object> dim = new HashMap<String, Object>();
            Map<String, Object> metric = new HashMap<String, Object>();
            for (Map.Entry<String, Object> e : row.entrySet()) {
                if (e.getValue() instanceof Number) {
                    metric.put(e.getKey(), e.getValue());
                } else {
                    dim.put(e.getKey(), e.getValue());
                }
            }
            String dimJson = JSON.toJSONString(dim);
            String metricJson = JSON.toJSONString(metric);
            String hashSource = window.bucket.toString() + "|" + dimJson;
            String dimHash = DigestUtils.md5DigestAsHex(hashSource.getBytes(StandardCharsets.UTF_8));
            affected += jdbcTemplate.update(
                    upsertSql,
                    window.bucket,
                    window.bucketType,
                    dimJson,
                    metricJson,
                    dimHash,
                    1L
            );
        }
        return affected;
    }

    /**
     * 创建报表专属聚合结果表（不存在时）。
     */
    private void ensureAggregateTable(String tableName) {
        String ddl = "create table if not exists " + tableName + " (" +
                "id bigint primary key auto_increment," +
                "stat_bucket datetime not null comment '统计桶开始时间'," +
                "bucket_type varchar(16) not null comment '桶类型: DAILY/MONTHLY'," +
                "dimension_json json null comment '维度数据'," +
                "metric_json json not null comment '指标数据'," +
                "dim_hash varchar(64) not null comment '维度哈希'," +
                "row_count bigint not null default 0 comment '聚合行计数'," +
                "created_at datetime not null default current_timestamp," +
                "updated_at datetime not null default current_timestamp on update current_timestamp," +
                "unique key uk_bucket_dim (stat_bucket, dim_hash)" +
                ") engine=InnoDB default charset=utf8mb4 comment='报表聚合结果表'";
        jdbcTemplate.execute(ddl);
    }

    /**
     * 将时间区间按策略拆分为窗口。
     */
    private List<Window> splitWindows(LocalDateTime start, LocalDateTime end, String granularity) {
        List<Window> windows = new ArrayList<Window>();
        if ("DAILY".equals(granularity)) {
            LocalDate cursor = start.toLocalDate();
            LocalDate endDate = end.toLocalDate();
            while (!cursor.isAfter(endDate)) {
                LocalDateTime ws = cursor.atStartOfDay();
                LocalDateTime we = cursor.atTime(LocalTime.MAX);
                windows.add(new Window(ws, we, ws, "DAILY"));
                cursor = cursor.plusDays(1);
            }
            return windows;
        }
        if ("MONTHLY".equals(granularity)) {
            LocalDate cursor = start.toLocalDate().withDayOfMonth(1);
            LocalDate endMonth = end.toLocalDate().withDayOfMonth(1);
            while (!cursor.isAfter(endMonth)) {
                LocalDateTime ws = cursor.atStartOfDay();
                LocalDate monthEnd = cursor.withDayOfMonth(cursor.lengthOfMonth());
                LocalDateTime we = monthEnd.atTime(LocalTime.MAX);
                windows.add(new Window(ws, we, ws, "MONTHLY"));
                cursor = cursor.plusMonths(1);
            }
            return windows;
        }
        throw new IllegalArgumentException("不支持的聚合粒度: " + granularity);
    }

    /**
     * 解析时间字符串。
     */
    private LocalDateTime parseTime(String value, boolean isStart) {
        if (value == null || value.trim().isEmpty()) {
            throw new IllegalArgumentException((isStart ? "startTime" : "endTime") + " 不能为空。");
        }
        String text = value.trim();
        try {
            return LocalDateTime.parse(text, DateTimeFormatter.ISO_DATE_TIME);
        } catch (DateTimeParseException ignored) {
            try {
                LocalDate d = LocalDate.parse(text, DateTimeFormatter.ISO_DATE);
                return isStart ? d.atStartOfDay() : d.atTime(LocalTime.MAX);
            } catch (DateTimeParseException ex) {
                throw new IllegalArgumentException("时间格式非法: " + text + "，支持 yyyy-MM-dd 或 ISO_DATE_TIME");
            }
        }
    }

    /**
     * 构建报表聚合表名。
     */
    private String aggregateTableName(Long reportId) {
        return "rpt_agg_report_" + reportId;
    }

    /**
     * 构建 FROM + JOIN 子句。
     */
    private String buildFromAndJoinSql(List<ReportFieldMeta> fieldMetaList, List<ReportJoinMeta> joinMetaList) {
        String baseTable = null;
        for (ReportFieldMeta meta : fieldMetaList) {
            if (meta != null && meta.getSourceTable() != null && !meta.getSourceTable().trim().isEmpty()) {
                baseTable = meta.getSourceTable().trim();
                break;
            }
        }
        if (baseTable == null) {
            throw new IllegalArgumentException("低代码模式缺少基础表配置。");
        }
        sqlSecurityValidator.validateIdentifier(baseTable);

        StringBuilder sql = new StringBuilder("from ").append(baseTable).append(" ");
        if (joinMetaList == null || joinMetaList.isEmpty()) {
            return sql.toString();
        }
        for (ReportJoinMeta joinMeta : joinMetaList) {
            if (joinMeta == null) {
                continue;
            }
            String joinType = safeUpper(joinMeta.getJoinType());
            if (!SAFE_JOIN_TYPE.contains(joinType)) {
                throw new IllegalArgumentException("不支持的关联类型: " + joinType);
            }
            sqlSecurityValidator.validateIdentifier(joinMeta.getLeftTable());
            sqlSecurityValidator.validateIdentifier(joinMeta.getLeftColumn());
            sqlSecurityValidator.validateIdentifier(joinMeta.getRightTable());
            sqlSecurityValidator.validateIdentifier(joinMeta.getRightColumn());
            sql.append(joinType).append(" JOIN ").append(joinMeta.getRightTable())
                    .append(" ON ")
                    .append(joinMeta.getLeftTable()).append(".").append(joinMeta.getLeftColumn())
                    .append(" = ")
                    .append(joinMeta.getRightTable()).append(".").append(joinMeta.getRightColumn());
            if (joinMeta.getExtraCondition() != null && !joinMeta.getExtraCondition().trim().isEmpty()) {
                sqlSecurityValidator.validateExpression(joinMeta.getExtraCondition());
                sql.append(" AND ").append(joinMeta.getExtraCondition().trim());
            }
            sql.append(" ");
        }
        return sql.toString();
    }

    /**
     * 解析字段来源表达式。
     */
    private String resolveSourceExpr(ReportFieldMeta meta) {
        String expr = meta.getExprText();
        if (expr != null && !expr.trim().isEmpty()) {
            sqlSecurityValidator.validateExpression(expr);
            return expr.trim();
        }
        String table = Objects.toString(meta.getSourceTable(), "").trim();
        String column = Objects.toString(meta.getSourceColumn(), "").trim();
        if (table.isEmpty() || column.isEmpty()) {
            throw new IllegalArgumentException("字段缺少来源表或来源字段: " + meta.getFieldKey());
        }
        sqlSecurityValidator.validateIdentifier(table);
        sqlSecurityValidator.validateIdentifier(column);
        return table + "." + column;
    }

    /**
     * 安全转大写。
     */
    private String safeUpper(String text) {
        return text == null ? "" : text.trim().toUpperCase(Locale.ROOT);
    }

    /**
     * 聚合窗口。
     */
    private static class Window {
        private final LocalDateTime start;
        private final LocalDateTime end;
        private final LocalDateTime bucket;
        private final String bucketType;

        private Window(LocalDateTime start, LocalDateTime end, LocalDateTime bucket, String bucketType) {
            this.start = start;
            this.end = end;
            this.bucket = bucket;
            this.bucketType = bucketType;
        }
    }
}
