AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimSqlBuilder.cs

using System.Text;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.S0Dim;

/// <summary>
/// 由 <see cref="S0DimDefinition"/> 生成 SQL 的**唯一**位置（纯函数，不接触 DB，可单测）。
///
/// 生成的所有语句都必须能逐条证明租户范围：
/// staging 侧固定四段谓词 <c>tenant_id / source_system / source_table / sync_batch_id</c>；
/// dim 侧固定 <c>tenant_id</c>；source 侧固定 <c>tenant_id</c>。
/// </summary>
public static class S0DimSqlBuilder
{
    /// <summary>staging 别名。</summary>
    public const string StgAlias = "s";

    /// <summary>dim 别名。</summary>
    public const string DimAlias = "d";

    /// <summary>source 别名。</summary>
    public const string SrcAlias = "a";

    /// <summary>业务键各段之间的分隔符，与 <c>MdpStagingWriter.BuildBizKey</c> 的 <c>string.Join("#")</c> 一致。</summary>
    public const string BizKeySeparator = "#";

    /// <summary>checksum 行内字段分隔（US，业务值不可能包含）。</summary>
    private const string FieldSep = "CHAR(31 USING utf8mb4)";

    /// <summary>checksum 的 NULL 哨兵（RS），与空串可区分。</summary>
    private const string NullSentinel = "CHAR(30 USING utf8mb4)";

    private static string Q(string ident)
    {
        if (!S0DimDefinition.IdentifierRe.IsMatch(ident))
            throw new InvalidOperationException($"非法标识符：{ident}");
        return $"`{ident}`";
    }

    /// <summary>列取值表达式（作用于 staging 的 raw_data；TenantIdColumn 例外）。</summary>
    private static string ValueExpr(S0DimColumn c) => c.Kind switch
    {
        S0DimValueKind.TenantIdColumn => $"{StgAlias}.{Q(S0DimDefinition.TenantColumn)}",
        S0DimValueKind.Str => MdpJsonSql.Str(StgAlias, c.JsonPath!),
        S0DimValueKind.Int => MdpJsonSql.Int(StgAlias, c.JsonPath!),
        S0DimValueKind.Dec => MdpJsonSql.Dec(StgAlias, c.JsonPath!, c.Precision, c.Scale),
        S0DimValueKind.DateTimeSec => MdpJsonSql.DateTimeSec(StgAlias, c.JsonPath!),
        S0DimValueKind.BoolTrue => MdpJsonSql.BoolTrue(StgAlias, c.JsonPath!),
        _ => throw new InvalidOperationException($"未支持的取值方式：{c.Kind}")
    };

    /// <summary>
    /// staging 四段固定谓词。<paramref name="withBatch"/> 为 false 时用于「不限本批」的只读对账。
    /// </summary>
    public static string StagingFilter(bool withBatch = true)
    {
        var sb = new StringBuilder();
        sb.Append($"{StgAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId");
        sb.Append($" AND {StgAlias}.`source_system` = @SourceSystem");
        sb.Append($" AND {StgAlias}.`source_table` = @SourceTable");
        if (withBatch) sb.Append($" AND {StgAlias}.`sync_batch_id` = @BatchId");
        return sb.ToString();
    }

    /// <summary>行级过滤：Required 列非空 + 布尔列可辨识（不满足的行不进 dim，随后被行数对账捕获）。</summary>
    private static IReadOnlyList<string> RowGuards(S0DimDefinition def)
    {
        var guards = new List<string>();
        foreach (var c in def.Columns)
        {
            if (c.Kind == S0DimValueKind.TenantIdColumn) continue;

            if (c.Required)
            {
                guards.Add($"({ValueExpr(c)}) IS NOT NULL");

                // IS NOT NULL 挡不住空串。业务键走 CONCAT_WS 拼接，空段会被**保留**，
                // 于是 Domain='' 会产出 '#LOC01' 这种「看起来合法」的键，而
                // AssertNoSourceDuplicateAsync 的 blank_cnt 只捕获整串全空的情况。
                // 当前 Catalog 中 Required 的 Str 列恰好就是全部业务键分量
                // （domain_code / work_center_code / department_code / location_code / shelf_code），
                // 故此守卫命中面 = 业务键，无附带影响。
                if (c.Kind == S0DimValueKind.Str)
                    guards.Add($"({ValueExpr(c)}) <> ''");
            }

            if (c.Kind == S0DimValueKind.BoolTrue)
            {
                // MdpJsonSql.BoolTrue 对 null / 不可识别值静默返回 0；此处强制要求原始值可辨识，
                // 否则整行被过滤 → dim_count < stg_count → 对账 FAIL（暴露而非吞掉）
                var raw = MdpJsonSql.Raw(StgAlias, c.JsonPath!);
                guards.Add($"LOWER({raw}) IN ('1','0','true','false')");
            }
        }
        return guards;
    }

    /// <summary>
    /// dim 物化语句：<c>INSERT INTO dim (...) SELECT ... FROM staging</c>。
    /// **刻意不带 ON DUPLICATE KEY UPDATE** —— 源侧业务键重复时直接撞 dim 唯一键报错，
    /// 由外层事务回滚（transform fail），绝不 last-wins。
    /// 参数：@TenantId @SourceSystem @SourceTable @BatchId @Now
    /// </summary>
    public static string BuildInsertSql(S0DimDefinition def)
    {
        def.Validate();

        var cols = new List<string>();
        var vals = new List<string>();
        foreach (var c in def.Columns)
        {
            cols.Add(Q(c.TargetColumn));
            vals.Add(ValueExpr(c));
        }

        cols.AddRange(["`source_system`", "`source_biz_key`", "`sync_batch_id`", "`sync_time`"]);
        vals.AddRange([$"{StgAlias}.`source_system`", $"{StgAlias}.`source_biz_key`", "@BatchId", "@Now"]);

        var where = new List<string> { StagingFilter() };
        where.AddRange(RowGuards(def));

        return $"""
                INSERT INTO {Q(def.DimTable)}
                  ({string.Join(", ", cols)})
                SELECT
                  {string.Join(",\n  ", vals)}
                FROM {Q(def.StagingTable)} {StgAlias}
                WHERE {string.Join("\n  AND ", where)}
                """;
    }

    /// <summary>staging 分区清理（purge）。参数：@TenantId @SourceSystem @SourceTable</summary>
    public static string BuildPurgeStagingSql(S0DimDefinition def) =>
        $"""
         DELETE {StgAlias} FROM {Q(def.StagingTable)} {StgAlias}
         WHERE {StagingFilter(withBatch: false)}
         """;

    /// <summary>源表行数（按租户）。参数：@TenantId</summary>
    public static string BuildSourceCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>源表中无法归属租户的行数（tenant_id IS NULL 或 &lt;= 0）—— 这些行整条链路都看不见。</summary>
    public static string BuildSourceUntenantedCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} IS NULL " +
        $"OR {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} <= 0";

    /// <summary>staging 行数。参数：@TenantId @SourceSystem @SourceTable [@BatchId]</summary>
    public static string BuildStagingCountSql(S0DimDefinition def, bool withBatch = true) =>
        $"SELECT COUNT(*) FROM {Q(def.StagingTable)} {StgAlias} WHERE {StagingFilter(withBatch)}";

    /// <summary>dim 行数（按租户）。参数：@TenantId</summary>
    public static string BuildDimCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>dim 中不属于本批的残留行数（FULL Replace 之后必须为 0）。参数：@TenantId @BatchId</summary>
    public static string BuildDimBatchImpuritySql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId " +
        $"AND {DimAlias}.`sync_batch_id` <> @BatchId";

    /// <summary>源侧业务键表达式（物理列，与 mdp_entity.biz_key_expr 同序同分隔）。</summary>
    public static string SourceBizKeyExpr(S0DimDefinition def) =>
        $"CONCAT_WS('{BizKeySeparator}', {string.Join(", ", def.SourceBizKeyColumns.Select(c => $"{SrcAlias}.{Q(c)}"))})";

    /// <summary>dim 侧业务键表达式（去掉 tenant_id 后的业务键列）。</summary>
    public static string DimBizKeyExpr(S0DimDefinition def) =>
        $"CONCAT_WS('{BizKeySeparator}', {string.Join(", ", def.BusinessKeyColumns.Skip(1).Select(c => $"{DimAlias}.{Q(c)}"))})";

    /// <summary>三层业务键集合来源子查询。</summary>
    public static string SourceBizKeySetSql(S0DimDefinition def) =>
        $"SELECT {SourceBizKeyExpr(def)} AS bk FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <inheritdoc cref="SourceBizKeySetSql"/>
    public static string StagingBizKeySetSql(S0DimDefinition def, bool withBatch = true) =>
        $"SELECT {StgAlias}.`source_biz_key` AS bk FROM {Q(def.StagingTable)} {StgAlias} WHERE {StagingFilter(withBatch)}";

    /// <inheritdoc cref="SourceBizKeySetSql"/>
    public static string DimBizKeySetSql(S0DimDefinition def) =>
        $"SELECT {DimBizKeyExpr(def)} AS bk FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>集合差集计数：<paramref name="leftSql"/> \ <paramref name="rightSql"/>。</summary>
    public static string BuildAntiJoinCountSql(string leftSql, string rightSql) =>
        $"""
         SELECT COUNT(*) FROM ({leftSql}) x
         LEFT JOIN ({rightSql}) y ON y.bk <=> x.bk
         WHERE y.bk IS NULL
         """;

    /// <summary>集合内是否有重复 / 空业务键：返回 (总数, 去重数, 空值数)。</summary>
    public static string BuildBizKeyQualitySql(string setSql) =>
        $"""
         SELECT COUNT(*) AS total, COUNT(DISTINCT x.bk) AS distinct_cnt,
                SUM(CASE WHEN x.bk IS NULL OR x.bk = '' THEN 1 ELSE 0 END) AS blank_cnt
         FROM ({setSql}) x
         """;

    /// <summary>NULL 归一（checksum 用）。</summary>
    private static string Norm(string expr, S0DimValueKind kind) => kind switch
    {
        S0DimValueKind.DateTimeSec => $"IFNULL(DATE_FORMAT({expr}, '%Y-%m-%d %H:%i:%s'), {NullSentinel})",
        S0DimValueKind.BoolTrue => $"IFNULL(CAST(CAST({expr} AS UNSIGNED) AS CHAR), {NullSentinel})",
        _ => $"IFNULL(CAST({expr} AS CHAR), {NullSentinel})"
    };

    private static IReadOnlyList<S0DimColumn> ChecksumColumns(S0DimDefinition def) =>
        def.Columns.Where(c => c.Kind != S0DimValueKind.TenantIdColumn).ToList();

    /// <summary>
    /// 单行哈希：MD5 前 15 个十六进制位 → 60 bit 无符号整数。
    ///
    /// 🔴 <c>CAST(... AS UNSIGNED)</c> **不可省略**：<c>CONV()</c> 返回的是**字符串**，
    /// 而 MySQL 的 <c>SUM(字符串)</c> 会按 DOUBLE 累加 —— 只有约 16 位有效数字，
    /// 19 位的和末几位不可靠，且**两侧扫描顺序不同会产生不同舍入**，
    /// 于是完全相同的数据也会算出不同校验和（2026-09-07 Batch 2 实测：
    /// 同一租户 source=…980000 / dim=…980700，count 相同却误报不一致）。
    /// 转成整数后 <c>SUM</c> 走 DECIMAL 精确累加，与顺序无关。
    /// </summary>
    private static string RowHash(string rowStringExpr) =>
        $"CAST(CONV(SUBSTRING(MD5({rowStringExpr}),1,15),16,10) AS UNSIGNED)";

    /// <summary>
    /// 属性校验和：<c>SUM(<see cref="RowHash"/>)</c>。
    /// 用 SUM 而非 GROUP_CONCAT/BIT_XOR：交换律使**排序与结果无关**（无需 ORDER BY，也不受 group_concat_max_len 截断）；
    /// XOR 会让成对重复相互抵消从而掩盖重复。必须与 COUNT 成对使用。
    /// </summary>
    public static string BuildSourceChecksumSql(S0DimDefinition def)
    {
        var parts = ChecksumColumns(def).Select(c => Norm($"{SrcAlias}.{Q(c.JsonPath!)}", c.Kind));
        var rowStr = $"CONCAT_WS({FieldSep}, {Norm(SourceBizKeyExpr(def), S0DimValueKind.Str)}, {string.Join(", ", parts)})";
        return $"SELECT COUNT(*) AS cnt, CAST(IFNULL(SUM({RowHash(rowStr)}),0) AS DECIMAL(40,0)) AS chk " +
               $"FROM {Q(def.SourceTable)} {SrcAlias} WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";
    }

    /// <inheritdoc cref="BuildSourceChecksumSql"/>
    public static string BuildDimChecksumSql(S0DimDefinition def)
    {
        var parts = ChecksumColumns(def).Select(c => Norm($"{DimAlias}.{Q(c.TargetColumn)}", c.Kind));
        var rowStr = $"CONCAT_WS({FieldSep}, {Norm(DimBizKeyExpr(def), S0DimValueKind.Str)}, {string.Join(", ", parts)})";
        return $"SELECT COUNT(*) AS cnt, CAST(IFNULL(SUM({RowHash(rowStr)}),0) AS DECIMAL(40,0)) AS chk " +
               $"FROM {Q(def.DimTable)} {DimAlias} WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";
    }

    /// <summary>dim 业务键重复计数（唯一键之外的兜底断言）。参数：@TenantId</summary>
    public static string BuildDimDuplicateBizKeySql(S0DimDefinition def) =>
        $"""
         SELECT COUNT(*) FROM (
           SELECT {DimBizKeyExpr(def)} AS bk
           FROM {Q(def.DimTable)} {DimAlias}
           WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId
           GROUP BY bk HAVING COUNT(*) > 1
         ) z
         """;

    /// <summary>镜像唯一性断言（如 dim_location 的 (tenant_id, location_code)）。参数：@TenantId</summary>
    public static string BuildMirrorUniqueViolationSql(S0DimDefinition def)
    {
        if (def.MirrorUniqueColumns is not { Count: > 0 }) throw new InvalidOperationException($"[{def.Key}] 未声明 MirrorUniqueColumns");
        var cols = string.Join(", ", def.MirrorUniqueColumns.Select(c => $"{DimAlias}.{Q(c)}"));
        return $"""
                SELECT COUNT(*) FROM (
                  SELECT {cols}
                  FROM {Q(def.DimTable)} {DimAlias}
                  WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId
                  GROUP BY {cols} HAVING COUNT(*) > 1
                ) z
                """;
    }

    /// <summary>孤儿子行（父维度中找不到父行）。参数：@TenantId</summary>
    /// <summary>
    /// 孤儿判定的公共 FROM/WHERE —— 样本查询与计数查询**共用同一份**，避免两处口径漂移。
    ///
    /// 「未建立关系」不是孤儿：父键列为 NULL 表示该行压根没声明父（如员工未分配部门），
    /// 与「声明了父但父不存在」（真悬挂）是两种语义。<c>&lt;=&gt;</c> 是 NULL 安全等值，
    /// 而父维度的业务键列均为 NOT NULL，故不加 <c>IS NOT NULL</c> 过滤会把全部未分配行
    /// 误报成孤儿，淹没真正的悬挂引用。对父键全部 Required 的维度（如 LocationShelf）本条件恒真，行为不变。
    /// </summary>
    private static string OrphanFromWhere(S0DimDefinition def)
    {
        var on = string.Join(" AND ", def.ParentKeyColumns!.Select(c => $"p.{Q(c)} <=> {DimAlias}.{Q(c)}"));
        var declared = string.Join(" AND ", def.ParentKeyColumns!.Select(c => $"{DimAlias}.{Q(c)} IS NOT NULL"));
        return $"""
                FROM {Q(def.DimTable)} {DimAlias}
                LEFT JOIN {Q(def.ParentDimTable!)} p
                  ON p.{Q(S0DimDefinition.TenantColumn)} = {DimAlias}.{Q(S0DimDefinition.TenantColumn)} AND {on}
                WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId AND {declared} AND p.`id` IS NULL
                """;
    }

    /// <summary>孤儿子行**样本**（受 limit 截断）。参数：@TenantId</summary>
    public static string BuildOrphanChildSql(S0DimDefinition def, int limit)
    {
        AssertParentDeclared(def);
        var sel = string.Join(", ", def.BusinessKeyColumns.Skip(1).Select(c => $"{DimAlias}.{Q(c)}"));
        return $"SELECT {sel} {OrphanFromWhere(def)} LIMIT {limit}";
    }

    /// <summary>孤儿子行**真实总数**（不截断）。参数：@TenantId</summary>
    public static string BuildOrphanChildCountSql(S0DimDefinition def)
    {
        AssertParentDeclared(def);
        return $"SELECT COUNT(*) {OrphanFromWhere(def)}";
    }

    private static void AssertParentDeclared(S0DimDefinition def)
    {
        if (def.ParentDimTable is null || def.ParentKeyColumns is not { Count: > 0 })
            throw new InvalidOperationException($"[{def.Key}] 未声明父维度");
    }
}