AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/DataAccess/Providers/S8PurchaseDeliveryDataProvider.cs

using Microsoft.Extensions.Logging;
using SqlSugar;

namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.DataAccess.Providers;

/// <summary>
/// S8-RULE01-PURCHASE-DELIVERY-A：<c>PURCHASE_DELIVERY</c> 数据集的取数 Provider。
/// S8 第一个真实 STANDARD_DATASET Provider。
///
/// <b>职责边界</b>：只做「租户作用域 → 最新快照收敛 → DWD 行 → canonical 行」。
/// 不判断是否异常、不判 severity、不建 Exception、不写 Detection、不决定 Recovery、
/// 不读 <c>rule.Expression</c>、不读 HttpContext / UserManager、不回落 Legacy SQL、
/// 不硬编码租户 / 工厂 / PO 号，也不修任何数据平台数据。
/// 判定完全交给 <see cref="S8TimeoutRuleEvaluator"/>。
///
/// <b>为什么不在 SQL 里过滤 status / remaining_qty</b>：
/// 若 Provider 先滤掉「已完成」行，那么「逾期但已完成不命中」这条契约就退化成
/// 「Provider 把它藏起来了」，而不是「evaluator 判它不命中」——真正的判定逻辑测不到。
/// 更要命的是：被滤掉的行不在 hits 里，<c>ReconcileRecoveriesForRuleAsync</c> 会把它当成
/// 「不再命中」而正常触发 recovery，结果看似正确但恢复原因不可解释。
/// 因此 Provider 返回当前快照的<b>全部</b> PO Line 事实，一行不滤。
///
/// <b>remaining_qty 仅作诊断</b>：字段级审计已证其不可信（781/784 行恒等于 order_qty），
/// 绝不参与 Trigger，也绝不出现在 WHERE 中。
/// </summary>
/// <remarks>
/// <b>刻意不实现 <c>ITransient</c></b>：Furion 的自动扫描会按接口再注册一次，而
/// <see cref="S8MonitoringDataProviderRegistry"/> 是以 <c>IEnumerable&lt;IS8MonitoringDataProvider&gt;</c>
/// 注入的——同一个类型会出现两份，注册表按"同 DatasetCode 重复注册"抛
/// <c>InvalidOperationException</c>。且因两者均为 Transient，该异常在<b>首次解析时</b>才发生：
/// 应用启动完全正常，直到第一个碰它的请求返回 500。
/// （<c>S8DatasetCatalog</c> / <c>S8LegacySqlDataProvider</c> 同时带 ITransient 与显式注册却不出问题，
/// 是因为它们按单一接口解析，后注册覆盖先注册，不走集合。）
/// 本类只保留 <c>Startup.cs</c> 中的显式注册。
/// </remarks>
public sealed class S8PurchaseDeliveryDataProvider : IS8MonitoringDataProvider
{
    /// <summary>
    /// 本 Provider 产出行对应的业务对象类型。
    /// 仅用于 DedupKey 长度预算估算；运行期真正写入 hit 的是
    /// <c>rule.SourceObjectType ?? rule.WatchObjectType</c>（见 <see cref="S8TimeoutRuleEvaluator"/>），
    /// 由规则配置决定，本类不参与。两者应保持一致（规则配置为 PURCHASE_ORDER_LINE）。
    /// </summary>
    internal const string ExpectedSourceObjectType = "PURCHASE_ORDER_LINE";

    /// <summary>
    /// <c>ado_s8_exception.dedup_key</c> 的列长上限（<c>AdoS8Exception</c> 中 <c>Length = 128</c>）。
    /// DedupKey 由 <c>S8TimeoutRuleEvaluator.BuildDedupKey</c> 拼成
    /// <c>T{tenant}:F{factory}:R{ruleCode}:{objectType}:{objectId}</c>。
    /// </summary>
    internal const int DedupKeyMaxLength = 128;

    /// <summary>
    /// 取数 SQL。
    ///
    /// <b>作用域只按 tenant_id</b>（Phase 1 已拍板：Security Boundary = Tenant，Factory = Business Context）。
    /// 刻意<b>不</b>加 <c>factory_id = @FactoryId</c>：<c>dwd_supplier_delivery.factory_id</c> 是写死的常量 1
    /// （<c>S3MdpSyncTransformService</c> 的 <c>COALESCE(NULLIF(po.factory_id,0),1)</c> 与
    /// <c>DeliveryScheduleService</c> 的字面量 1），与 S8 规则侧的真实 <c>SysOrg.Id</c> 语义不同，
    /// 强行相等匹配会让每条规则恒返 0 行。客户运营模式为 1 Tenant = 1 Factory，
    /// 因此租户过滤已构成完整的数据隔离边界。
    ///
    /// <b>当前快照由发布证据指认，不再按时间挑</b>。这里曾经写的是
    /// <c>stat_date = (SELECT stat_date … ORDER BY stat_date DESC LIMIT 1)</c>，
    /// 即「取最新的那一天」。它在**合法的空**面前会静默失效：
    /// 源侧被清空的那一轮产出 0 行 DWD，当天根本不存在 <c>stat_date</c> 分区，
    /// 中台的同日淘汰也无行可删（它只能删「今天」），于是「最新的那一天」回落到<b>上一天</b>，
    /// 把一批早已作废的行当成当前态返回。实测：租户 838257212780613 源侧清零、
    /// STD 归 0、S3 跑批 SUCCESS，而本 Provider 仍返回前一日的 10 行。
    ///
    /// 改为与 <c>mdp_transform_run_log</c> 的发布证据 JOIN，读者与生产者对「哪一批是当前」
    /// 达成同一个事实。三种情形都由这一条 SQL 直接给出正确答案，<b>没有兜底分支</b>：
    /// <list type="bullet">
    ///   <item><c>currentRows &gt; 0</c> → JOIN 命中该批次，返回该批次的行；</item>
    ///   <item><c>currentRows = 0</c>（合法空）→ 没有行带这个批次号，自然 0 行；</item>
    ///   <item>没有发布证据 → CTE 为空，JOIN 无从命中，同样 0 行（fail-safe）。</item>
    /// </list>
    /// 关键在于：<b>任何一种情形都不会去看别的 stat_date</b> —— 回落这条路径在结构上不存在。
    ///
    /// 同时按 <c>stat_date</c> 与 <c>sync_batch_id</c> 两个条件收窄：前者让查询仍能走
    /// <c>idx_stat_date(tenant_id, stat_date)</c>，后者才是身份判据。只有 batch 相等时，
    /// 单靠 stat_date 无法排除同日残留的其它批次。
    ///
    /// <c>po_no</c> / <c>po_line</c> 均为 NOT NULL，故只判空串；两者共同构成对象身份，
    /// 任一为空会产生退化的 "#" 身份，必须排除。
    /// </summary>
    /// <remarks>
    /// <b>结果列别名必须是 PascalCase，与 <see cref="PurchaseDeliveryRow"/> 属性逐字同名。</b>
    ///
    /// 本仓 <c>DbSettings.EnableUnderLine = false</c>（Database.json），SqlSugar 不做
    /// snake_case → PascalCase 的自动换算；<c>Ado.SqlQueryAsync&lt;T&gt;</c> 按列名与属性名匹配。
    /// 曾经这里写的是 <c>AS source_object_id</c> / <c>AS due_at</c> 这类 snake_case 别名，
    /// 结果是**每个属性都映射不上、全部为 null**：行数照常返回（36 行），但 <c>DueAt</c> 为 null，
    /// evaluator 在 <c>due == null</c> 处逐行跳过 → 命中恒为 0，且不报任何错。
    ///
    /// 这类缺陷单元测试抓不到：契约测试是直接 new <c>PurchaseDeliveryRow</c> 再喂 evaluator，
    /// 整条 SqlSugar 映射被绕开。守护改由 <c>S8PurchaseDeliverySqlMappingTests</c> 承担
    /// （用反射比对 SQL 别名与属性名）。
    ///
    /// 仓内既有同款写法可参照 <c>S8WatchSchedulerService.ListEnabledScopesAsync</c>：
    /// <c>SELECT r.tenant_id AS TenantId</c>。
    ///
    /// WHERE / FROM 中引用的仍是数据库真实列名（snake_case），只有输出别名改为 PascalCase。
    /// </remarks>
    internal const string Sql =
        """
        WITH pub AS (
            SELECT JSON_UNQUOTE(JSON_EXTRACT(r.summary_json, '$.publish.batchId'))  AS batch_id,
                   JSON_UNQUOTE(JSON_EXTRACT(r.summary_json, '$.publish.statDate')) AS stat_date
            FROM mdp_transform_run_log r
            WHERE r.job_code = @ProducerJobCode
              AND r.tenant_id = @TenantId
              AND r.status = 'SUCCESS'
              AND r.end_time IS NOT NULL
              AND JSON_EXTRACT(r.summary_json, '$.publish.batchId') IS NOT NULL
            ORDER BY r.start_time DESC, r.id DESC
            LIMIT 1
        )
        SELECT
            CONCAT(d.po_no, '#', d.po_line)                        AS SourceObjectId,
            d.po_no                                                AS RelatedObjectCode,
            COALESCE(NULLIF(d.supplier_name, ''), d.supplier_code) AS RelatedObjectName,
            COALESCE(d.need_date, d.due_date)                      AS DueAt,
            d.delivery_status                                      AS Status,
            CASE WHEN d.need_date IS NULL THEN 'due_date' ELSE 'need_date' END AS DueAtSource,
            d.po_no          AS PoNo,
            d.po_line        AS PoLine,
            d.po_type        AS PoType,
            d.supplier_code  AS SupplierCode,
            d.supplier_name  AS SupplierName,
            d.item_code      AS ItemCode,
            d.item_name      AS ItemName,
            d.order_qty      AS OrderQty,
            d.remaining_qty  AS RemainingQty,
            d.need_date      AS NeedDate,
            d.due_date       AS DueDate,
            d.risk_level     AS RiskLevel,
            d.stat_date      AS StatDate
        FROM dwd_supplier_delivery d
        JOIN pub p
          ON d.stat_date = p.stat_date
         AND d.sync_batch_id = p.batch_id
        WHERE d.tenant_id = @TenantId
          AND d.po_no <> ''
          AND d.po_line <> ''
        """;

    private readonly ISqlSugarClient _db;
    private readonly ILogger<S8PurchaseDeliveryDataProvider> _logger;

    public S8PurchaseDeliveryDataProvider(
        ISqlSugarClient db,
        ILogger<S8PurchaseDeliveryDataProvider> logger)
    {
        _db = db;
        _logger = logger;
    }

    public string DatasetCode => S8BusinessDatasetDefinitions.PurchaseDeliveryCode;

    /// <summary>与 Catalog 引用同一实例，杜绝目录与实现的能力声明漂移。</summary>
    public S8DatasetCapabilities Capabilities => S8BusinessDatasetDefinitions.PurchaseDeliveryCapabilities;

    public async Task<S8MonitoringRowSet> LoadAsync(
        S8MonitoringDataRequest request,
        CancellationToken cancellationToken = default)
    {
        // 作用域必须由调用方显式给出；缺失时显式失败，绝不用默认值兜底
        // （兜底会让跨租户读取以"看起来正常"的方式发生）。
        request.EnsureValid();

        var rows = await _db.Ado.SqlQueryAsync<PurchaseDeliveryRow>(
            Sql,
            new SugarParameter[]
            {
                new("@TenantId", request.TenantId),
                // 生产者作业码取自数据集定义，不在这里另写一份字面量 —— 目录与取数各写一份必然漂移。
                new("@ProducerJobCode", S8BusinessDatasetDefinitions.PurchaseDeliveryProducerJobCode)
            });

        var budget = ResolveSourceObjectIdBudget(request);
        var mapped = new List<S8MonitoringRow>(rows.Count);
        var skipped = 0;

        foreach (var row in rows)
        {
            var sourceObjectId = row.SourceObjectId ?? string.Empty;

            // DedupKey 越界必须"响"而不是静默消失：数据库层的截断会让两个不同 PO Line
            // 折叠成同一个异常，且没有任何痕迹。这里显式跳过并计数，收尾统一 warn。
            if (sourceObjectId.Length > budget)
            {
                skipped++;
                _logger.LogWarning(
                    "s8_purchase_delivery_soid_over_dedup_budget rule={RuleCode} tenant={Tenant} " +
                    "soid={Soid} length={Length} budget={Budget}",
                    request.RuleCode, request.TenantId, sourceObjectId, sourceObjectId.Length, budget);
                continue;
            }

            mapped.Add(S8MonitoringRow.FromValues(ToCanonicalValues(row)));
        }

        _logger.LogDebug(
            "s8_purchase_delivery_loaded rule={RuleCode} tenant={Tenant} rows={Rows} skipped={Skipped}",
            request.RuleCode, request.TenantId, mapped.Count, skipped);

        return S8MonitoringRowSet.FromRows(mapped);
    }

    /// <summary>
    /// 计算 <c>source_object_id</c> 在 DedupKey 中可用的字符预算。
    ///
    /// 预算<b>由 request 现算</b>而非写死常量：DedupKey 前缀长度随租户 / 工厂 / 规则编码变化，
    /// 写死一个数字会在换规则或换租户时悄悄失效。
    ///
    /// 注意这个预算通常比 <c>source_object_id</c> 的列宽（64）更紧：
    /// S8-TENANT-ONLY-BATCH5 去掉 <c>:F{factory}</c> 段后，
    /// 15 位租户 + 36 字符规则编码 + PURCHASE_ORDER_LINE 时前缀约占 76，余量比之前宽松了 16 字符。
    /// 落在 37–64 之间的键会过得了列校验、卡在 DedupKey 上，因此必须按本预算判定。
    /// </summary>
    internal static int ResolveSourceObjectIdBudget(S8MonitoringDataRequest request)
    {
        var prefixLength =
            $"T{request.TenantId}:R{request.RuleCode}:{ExpectedSourceObjectType}:".Length;
        return Math.Max(0, DedupKeyMaxLength - prefixLength);
    }

    /// <summary>
    /// DWD 行 → canonical 值。canonical 四列驱动 TIMEOUT 判定，其余列随 payload 落盘供追溯。
    ///
    /// <c>due_at_source</c> 必须保留：<c>due_at</c> 是 need_date / due_date 的混合映射
    /// （实测该租户 18/36 行两者不同），若不逐行记录实际用了哪个基准，将来复盘无从下手。
    /// </summary>
    internal static Dictionary<string, object> ToCanonicalValues(PurchaseDeliveryRow row) => new(StringComparer.OrdinalIgnoreCase)
    {
        // canonical：evaluator 判定与身份
        [S8CanonicalColumns.SourceObjectId] = row.SourceObjectId,
        [S8CanonicalColumns.RelatedObjectCode] = row.RelatedObjectCode,
        [S8CanonicalColumns.RelatedObjectName] = row.RelatedObjectName,
        [S8CanonicalColumns.DueAt] = row.DueAt,
        [S8CanonicalColumns.Status] = row.Status,

        // 诊断 payload：不参与任何判定
        ["due_at_source"] = row.DueAtSource,
        ["po_no"] = row.PoNo,
        ["po_line"] = row.PoLine,
        ["po_type"] = row.PoType,
        ["supplier_code"] = row.SupplierCode,
        ["supplier_name"] = row.SupplierName,
        ["item_code"] = row.ItemCode,
        ["item_name"] = row.ItemName,
        ["order_qty"] = row.OrderQty,
        ["remaining_qty"] = row.RemainingQty,
        ["need_date"] = row.NeedDate,
        ["due_date"] = row.DueDate,
        ["risk_level"] = row.RiskLevel,
        ["stat_date"] = row.StatDate
    };

    /// <summary>
    /// DWD 只读投影。刻意<b>不建 CodeFirst Entity</b>：<c>dwd_supplier_delivery</c> 由数据平台维护，
    /// S8 侧建 Entity 会把它拉进结构同步范围，造成归属混乱。
    /// </summary>
    internal sealed class PurchaseDeliveryRow
    {
        public string SourceObjectId { get; set; }
        public string RelatedObjectCode { get; set; }
        public string RelatedObjectName { get; set; }
        public DateTime? DueAt { get; set; }
        public string Status { get; set; }
        public string DueAtSource { get; set; }

        public string PoNo { get; set; }
        public string PoLine { get; set; }
        public string PoType { get; set; }
        public string SupplierCode { get; set; }
        public string SupplierName { get; set; }
        public string ItemCode { get; set; }
        public string ItemName { get; set; }
        public decimal? OrderQty { get; set; }
        public decimal? RemainingQty { get; set; }
        public DateTime? NeedDate { get; set; }
        public DateTime? DueDate { get; set; }
        public string RiskLevel { get; set; }
        public DateTime? StatDate { get; set; }
    }
}