AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/NeutralRequiredColumnsMonitor.cs

using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// 运行期核对中立层必填列是否整列为空。
/// <para>
/// <see cref="NeutralRequiredColumns"/> 是「投影 SQL 有没有写这一列」的静态守卫，守不住
/// 「写了但取到的全是 NULL」和「这批存量是加列之前物化的」两种情况。而整列为空不会报错，
/// 只会让依赖它的 KPI 一起算出 NO_DATA —— 2026-09 S5/S7 全租户掉数就是这么发生的。
/// 故物化之后必须实测一次，把整列为空变成一条可查询的告警。
/// </para>
/// </summary>
public sealed class NeutralRequiredColumnsMonitor : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly ILogger _logger;

    public NeutralRequiredColumnsMonitor(ISqlSugarClient db, ILoggerFactory loggerFactory)
    {
        _db = db;
        _logger = loggerFactory.CreateLogger(nameof(NeutralRequiredColumnsMonitor));
    }

    /// <summary>返回该租户该表里「有行但整列为空」的必填列名。顺带落 mdp_source_gate_log 与错误日志。</summary>
    public async Task<IReadOnlyList<string>> AssertAsync(
        long tenantId, string stdObject, string table, string? batchId, CancellationToken cancellationToken = default)
    {
        if (tenantId <= 0) return [];
        var required = NeutralRequiredColumns.For(stdObject).Select(c => c.Name).ToList();
        if (required.Count == 0) return [];

        cancellationToken.ThrowIfCancellationRequested();
        var present = await _db.Ado.SqlQueryAsync<string>(
            """
            SELECT COLUMN_NAME FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@table
            """,
            new { table });
        var columns = required
            .Where(c => present.Contains(c, StringComparer.OrdinalIgnoreCase))
            .ToList();
        if (columns.Count == 0) return [];

        // 列名取自 NeutralRequiredColumns 常量，不来自外部输入，故可直接拼进 SQL。
        var counters = string.Join(", ", columns.Select(c => $"SUM(`{c}` IS NOT NULL) AS `{c}`"));
        var stats = await _db.Ado.GetDataTableAsync(
            $"SELECT COUNT(*) AS `__total`, {counters} FROM `{table}` WHERE tenant_id=@tenant",
            new { tenant = tenantId });
        if (stats.Rows.Count == 0) return [];

        var row = stats.Rows[0];
        if (Convert.ToInt64(row["__total"]) <= 0) return [];

        var empty = columns
            .Where(c => row[c] == DBNull.Value || Convert.ToInt64(row[c]) == 0)
            .ToList();
        if (empty.Count == 0) return empty;

        _logger.LogError(
            "中立层必填列整列为空 tenant={Tenant} object={Object} table={Table} columns={Columns}",
            tenantId, stdObject, table, string.Join(",", empty));
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_source_gate_log
              (tenant_id, std_object, source_system, gate_reason, row_count, sample_keys, sync_batch_id)
            VALUES (@tenant, @obj, '', 'REQUIRED_COLUMN_EMPTY', @rows, @sample, @batch)
            """,
            new
            {
                tenant = tenantId,
                obj = stdObject,
                rows = empty.Count,
                sample = string.Join(",", empty),
                batch = batchId ?? ""
            });
        return empty;
    }
}