AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP.Tests/DataPlatform/S3PurchaseAuthorityContractTests.cs

using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Xunit;

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

/// <summary>
/// 采购交付 Authority（<c>mdp_std_purchase_order</c> → <c>dwd_supplier_delivery</c>）的生产契约。
///
/// <para>这条链是 S8 Rule01 的唯一数据来源。它曾同时坏在三个地方，且三处根因都落在同一段
/// STD 构建 SQL 上：丢了租户谓词（跨租户串值）、丢了批次作用域（孤儿只增不减）、
/// 以及有两个业务 API 绕过它直写下游 DWD。本文件守住修好之后的形状。</para>
/// </summary>
public class S3PurchaseAuthorityContractTests
{
    private static string Source(params string[] parts)
    {
        var dir = new DirectoryInfo(AppContext.BaseDirectory);
        while (dir != null && !Directory.Exists(Path.Combine(dir.FullName, "Admin.NET.Plugin.AiDOP")))
            dir = dir.Parent;
        Assert.NotNull(dir);

        var full = Path.Combine(new[] { dir!.FullName, "Admin.NET.Plugin.AiDOP" }.Concat(parts).ToArray());
        Assert.True(File.Exists(full), $"源码文件不存在，路径需同步更新：{full}");
        return File.ReadAllText(full);
    }

    private static string S3Service() => Source("Supply", "S3MdpSyncTransformService.cs");

    private static int At(string code, string anchor)
    {
        var i = code.IndexOf(anchor, StringComparison.Ordinal);
        Assert.True(i >= 0, $"未找到锚点：{anchor}");
        return i;
    }

    /// <summary>取 <c>mdp_std_purchase_order</c> 那一条 INSERT 的 SQL 区间。</summary>
    private static string StdInsertBlock()
    {
        var code = S3Service();
        var start = At(code, "INSERT INTO mdp_std_purchase_order");
        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
        Assert.True(end > start, "mdp_std_purchase_order 的 INSERT 未正常收尾");
        return code[start..end];
    }

    /// <summary>
    /// STD 构建的两个 enrichment JOIN 必须自带租户谓词。
    ///
    /// <para><b>为什么不能靠作用域注入兜底</b>：<c>MdpSqlScope.InjectTenantFactory</c> 的实现是
    /// <c>Regex.Replace(sql, @"\bWHERE\b", ...)</c>，它只改写 <c>WHERE</c>，
    /// 对 <c>JOIN ... ON</c> 零覆盖。这不是配置问题，是机制上够不到。</para>
    ///
    /// <para><b>后果不是行数膨胀</b>：扇出会被 <c>uk_po_line(tenant_id, po_no, po_line)</c> 的
    /// ON DUPLICATE KEY UPDATE 吸收，最终行数不变，但 item_name / supplier_code / order_date / buyer
    /// 这些 enrichment 值由 MySQL 任取一行决定 —— 症状是取值随机化，不是报错。
    /// 实测单租户扇出 870/220 = 3.95 倍，item_code 跨租户重名 5529 组。</para>
    /// </summary>
    [Fact]
    public void StdBuild_EnrichmentJoins_CarryTenantPredicate()
    {
        var block = StdInsertBlock();

        // 表头 enrichment：按 PurOrd 字符串匹配，必须限定同租户。
        Assert.Contains("m ON m.tenant_id = d.tenant_id", block);
        Assert.Contains("m.pur_ord = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))", block);

        // 物料 enrichment：按 ItemNum 字符串匹配，同样必须限定同租户。
        Assert.Contains("i ON i.tenant_id = d.tenant_id", block);
        Assert.Contains("i.item_num = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))", block);
    }

    /// <summary>
    /// DWD 构建的三个 JOIN 本来就是租户安全的，本批不得把它们改坏。
    /// <para>它们已带 <c>po.tenant_id=X.tenant_id AND po.factory_id=X.factory_id</c>，
    /// 泄漏点在 STD 而不在这里 —— 这条断言是防止后续改动误伤。</para>
    /// </summary>
    [Fact]
    public void DwdBuild_JoinsRemainTenantAndFactoryScoped()
    {
        var code = S3Service();
        var start = At(code, "INSERT INTO dwd_supplier_delivery");
        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
        var block = code[start..end];

        Assert.Contains("ds ON po.tenant_id=ds.tenant_id AND po.factory_id=ds.factory_id", block);
        Assert.Contains("dr ON po.tenant_id=dr.tenant_id AND po.factory_id=dr.factory_id", block);
        Assert.Contains("s ON po.tenant_id=s.tenant_id AND po.factory_id=s.factory_id", block);
    }

    /// <summary>
    /// STD 构建必须按本轮批次收窄驱动行。
    /// <para>少了它，INSERT 会把整张贴源历史重新物化，并给死行盖上最新批次戳 ——
    /// 实测租户 797 因此在 STD 沉淀了 336 行源侧已不存在的采购订单行（518 行中的 64.9%），
    /// 且全部 518 行共享同一个批次号，连时间戳都认不出哪些是死的。</para>
    /// </summary>
    [Fact]
    public void StdBuild_DrivingRows_ScopedToCurrentBatch()
    {
        var block = StdInsertBlock();
        Assert.Contains("AND d.source_table='PurOrdDetail' AND d.sync_batch_id=@BatchId", block);
    }

    /// <summary>
    /// enrichment 必须按 (租户, 键) 去重取最新一行，且**不得**用 GROUP_CONCAT 实现。
    /// <para>贴源层同时存在两代 <c>source_biz_key</c>（旧 <c>Domain:PurOrd:Line</c> 与新的裸 RecID），
    /// 两代永远不会互相 ODKU 覆盖，因此同一个 PurOrd/ItemNum 在同租户内有多行 —— 不去重就仍会扇出。</para>
    /// <para>禁用 GROUP_CONCAT 的理由是硬的：<c>raw_data</c> 是 JSON 大字段，
    /// 而 <c>group_concat_max_len</c> 实测仅 1024，拼接必然截断、首元素本身就可能超限。</para>
    /// </summary>
    [Fact]
    public void StdBuild_EnrichmentDeduplicated_WithRowNumberNotGroupConcat()
    {
        var block = StdInsertBlock();

        Assert.Contains("ROW_NUMBER() OVER (", block);
        Assert.Contains("PARTITION BY x0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd'))", block);
        Assert.Contains("PARTITION BY y0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum'))", block);

        // 可执行 SQL 里不得出现 GROUP_CONCAT（注释里提到它是允许的）。
        var sqlOnly = string.Join('\n', block.Split('\n').Where(l => !l.TrimStart().StartsWith("--", StringComparison.Ordinal)));
        Assert.DoesNotContain("GROUP_CONCAT", sqlOnly);
    }

    /// <summary>
    /// 每轮淘汰必须存在，且只在全量路径执行。
    /// <para>只加批次过滤只能让死行「可识别」，不能「退役」—— 运行期链路里没有任何 DELETE。
    /// 而增量批次只含变更行，按它淘汰会把未变更的存量行全删掉，所以必须按路径分流。</para>
    /// </summary>
    [Fact]
    public void StdBuild_RetiresStaleRows_OnlyOnFullPath()
    {
        var code = S3Service();

        Assert.Contains("DELETE s FROM mdp_std_purchase_order s", code);
        Assert.Contains("IFNULL(s.sync_batch_id,'') <> @BatchId", code);

        // 安全闸门必须用 mdp_sync_log 的成功记录，而不是「贴源层有没有行」——
        // 后者会把源侧合法清空的租户永久锁在幽灵行上。
        Assert.Contains("FROM mdp_sync_log l", code);
        Assert.Contains("l.status='SUCCESS'", code);

        // 路径分流
        Assert.Contains("BuildStandardCommands(string batchId, DateTime now, bool retireStale)", code);
        Assert.Contains("TransformStandardAsync(batchId, now, retireStale: true, cancellationToken)", code);
        Assert.Contains("TransformStandardAsync(batchId, now, retireStale: false, cancellationToken)", code);
    }

    /// <summary>取一条语句的 SQL 区间（从锚点到原始字符串字面量的收尾 <c>"""</c>）。</summary>
    private static string StatementBlock(string startAnchor)
    {
        var code = S3Service();
        var start = At(code, startAnchor);
        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
        Assert.True(end > start, $"语句未正常收尾：{startAnchor}");
        return code[start..end];
    }

    /// <summary>
    /// 本文件新增/改写的每条 SQL，经 <see cref="MdpSqlScope.InjectTenantFactory"/> 处理后
    /// 都不得出现「注入到不存在的列上」。
    ///
    /// <para><b>为什么必须用真注入器跑一遍，而不是比对源码文本</b>：这个坑有两个入口，
    /// 光看源码任何一处都不显眼，只有跑一遍注入才暴露。二者本地沙箱都实测炸过：</para>
    ///
    /// <list type="number">
    /// <item><b>空前缀</b> —— <c>ResolveSourcePrefix</c> 反向找最近的同层 <c>FROM 表 别名</c>。
    /// 派生表外层是 <c>FROM (</c>，别名正则匹配不到括号，于是回退成空前缀，生成裸
    /// <c>factory_id</c>；派生表 SELECT 列表里没有这一列 → 整轮转换失败。</item>
    /// <item><b>别名指向没有该列的表</b> —— 淘汰语句的 EXISTS 子查询若写成
    /// <c>l.tenant_id = s.tenant_id</c>（关联外层、不含 <c>@TenantId</c>），会被注入
    /// <c>COALESCE(NULLIF(l.factory_id,0),1)=@FactoryId</c>，而 <c>mdp_sync_log</c> 没有
    /// <c>factory_id</c> 列 → 同样整轮失败。</item>
    /// </list>
    ///
    /// <para>失败形态是<b>整轮跑批 FAILED</b>，不是少几行数据 —— 而 Authority 一轮不产出，
    /// S8 的恢复闸门就判 DEGRADED、整个租户的恢复停摆。所以这条守卫值得用真注入器跑。</para>
    /// </summary>
    [Theory]
    [InlineData("INSERT INTO mdp_std_purchase_order")]
    [InlineData("DELETE s FROM mdp_std_purchase_order s")]
    [InlineData("DELETE w FROM dwd_supplier_delivery w")]
    public void ScopeInjection_NeverTargetsAColumnThatDoesNotExist(string anchor)
    {
        var injected = MdpSqlScope.InjectTenantFactory(StatementBlock(anchor));

        // ① 空前缀：注入器没能定位别名。在本文件里所有表都带别名，因此这必定是缺陷。
        Assert.DoesNotContain("WHERE tenant_id=@TenantId", injected);

        // ② 前缀指向的表没有 factory_id：派生表别名（x/y/m/i）与 mdp_sync_log（l）都没有这一列。
        foreach (var alias in new[] { "x", "y", "m", "i", "l" })
            Assert.DoesNotContain($"NULLIF({alias}.factory_id", injected);
    }

    /// <summary>
    /// 驱动行必须**显式**带上租户与工厂谓词，不能依赖注入器补。
    ///
    /// <para>历史上它确实是靠注入的：<c>ResolveSourcePrefix</c> 反向找到 <c>FROM ... d</c>
    /// 就能补出 <c>d.</c> 前缀。但两个 enrichment 派生表插进 <c>FROM ... d</c> 与本
    /// <c>WHERE</c> 之间以后，反向扫描会先撞上派生表的 <c>FROM (</c>，别名正则匹配不到括号
    /// → 回退成空前缀，生成裸 <c>tenant_id</c>。而派生表 m / i 也各自暴露 <c>tenant_id</c>，
    /// 该列立刻变成歧义列 —— 整轮转换 FAILED，不是少几行。</para>
    ///
    /// <para>所以这不是风格偏好：只要这条 INSERT 还带派生表，作用域就必须写死。</para>
    /// </summary>
    [Fact]
    public void DrivingRows_CarryTenantAndFactoryExplicitly()
    {
        Assert.Contains(
            "WHERE d.tenant_id=@TenantId AND COALESCE(NULLIF(d.factory_id,0),1)=@FactoryId",
            StatementBlock("INSERT INTO mdp_std_purchase_order"));
    }

    /// <summary>
    /// Rule01 Authority 链上的三条语句必须**完全自带**作用域 —— 注入对它们应当是恒等变换。
    ///
    /// <para>这是本文件与注入器之间的边界声明：这三条 SQL 的作用域由字面 SQL 唯一决定，
    /// 不交给一个够不到 <c>JOIN...ON</c>、认不出派生表别名、且对 SQL 注释同样生效的正则去猜。
    /// 恒等变换是这里唯一可接受的结果；任何非恒等都意味着有一个 <c>WHERE</c> 漏了
    /// <c>@TenantId</c>，而它下一步会以整轮跑批失败的形式暴露。</para>
    /// </summary>
    [Theory]
    [InlineData("INSERT INTO mdp_std_purchase_order")]
    [InlineData("DELETE s FROM mdp_std_purchase_order s")]
    [InlineData("DELETE w FROM dwd_supplier_delivery w")]
    public void AuthorityStatements_AreFullySelfScoped(string anchor)
    {
        var raw = StatementBlock(anchor);
        Assert.Equal(raw, MdpSqlScope.InjectTenantFactory(raw));
    }

    /// <summary>
    /// DWD 必须做同日淘汰，且**放在 INSERT 之后**。
    /// <para>放之前会让快照短暂为空；此处无事务，并发的 Rule01 取数会读到空快照，
    /// 把全部在办异常判成已恢复 —— 那正是 Authority 健康门禁要防的形态。</para>
    /// <para>该淘汰同时保证 <c>MAX(stat_date)</c> 取到的是单一批次：实测同一天混入两个批次
    /// 已发生 3 次（如 2026-06-07 的 178 行 + 7 行）。</para>
    /// </summary>
    [Fact]
    public void DwdBuild_RetiresSameDayStaleRows_AfterInsert()
    {
        var code = S3Service();
        var insertAt = At(code, "INSERT INTO dwd_supplier_delivery");
        var deleteAt = code.IndexOf("DELETE w FROM dwd_supplier_delivery w", insertAt, StringComparison.Ordinal);

        Assert.True(deleteAt > insertAt, "同日淘汰必须在 INSERT 之后，否则快照会短暂为空");
        Assert.Contains("w.stat_date=@StatDate", code);
        Assert.Contains("IFNULL(w.sync_batch_id,'') <> @BatchId", code);
        Assert.Contains("BuildDwdAsync(batchId, now, retireStale: true, result, cancellationToken)", code);
        Assert.Contains("BuildDwdAsync(batchId, now, retireStale: false, result, cancellationToken)", code);
    }

    /// <summary>
    /// <c>dwd_supplier_delivery</c> 与 <c>mdp_std_purchase_order</c> 在运行期<b>只能有一个写入方</b>。
    ///
    /// <para>它们是 Rule01 的 Authority。此前另有两个业务 API 直写：
    /// <c>ProcessOutsourceOrderService</c> 把 <c>delivery_status</c> 硬编码为 <c>'OPEN'</c> 并放进
    /// ON DUPLICATE KEY UPDATE，一次建单就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN；
    /// <c>DeliveryScheduleService</c> 则写出 CANCELLED / CLOSED / PARTIAL 三个没有任何读方认识的值。</para>
    ///
    /// <para>更严重的是它们能凭空造出一个「最新快照」：Rule01 取 <c>MAX(stat_date)</c> 那一天的全部行，
    /// 实测租户 797403760988229 的 2026-05-16 快照只有 2 行、且全部出自旁路写入 ——
    /// 中台停摆时，一次业务 API 调用就让 Rule01 在 2 行残缺数据上评估，而快照看上去还很新鲜。</para>
    ///
    /// <para>用「非注释行里出现写入语句」判定，避免被解释性注释里的表名误判。</para>
    /// </summary>
    [Fact]
    public void AuthorityTables_HaveExactlyOneRuntimeWriter()
    {
        // (文件, 表, 期望写入处数)
        var expectations = new[]
        {
            ("S3MdpSyncTransformService.cs",   "dwd_supplier_delivery",   1),
            ("DeliveryScheduleService.cs",     "dwd_supplier_delivery",   0),
            ("ProcessOutsourceOrderService.cs","dwd_supplier_delivery",   0),
            ("S3MdpSyncTransformService.cs",   "mdp_std_purchase_order",  1),
            ("DeliveryScheduleService.cs",     "mdp_std_purchase_order",  0),
            ("ProcessOutsourceOrderService.cs","mdp_std_purchase_order",  0)
        };

        foreach (var (file, table, expected) in expectations)
        {
            var writes = Source("Supply", file).Split('\n')
                .Where(l => !l.TrimStart().StartsWith("//", StringComparison.Ordinal))
                .Count(l => l.Contains($"INSERT INTO {table}", StringComparison.Ordinal)
                         || l.Contains($"UPDATE {table}", StringComparison.Ordinal));

            Assert.True(writes == expected,
                $"{file} 对 {table} 的运行期写入应为 {expected} 处，实测 {writes} 处");
        }
    }

    /// <summary>
    /// 旁路写入方不得把 <c>delivery_status</c> 硬编码后写进 Authority。
    /// <para>状态只能由正式 MDP 转换从源侧派生（COMPLETED / DELAYED / OPEN 三值）。</para>
    /// </summary>
    [Fact]
    public void BypassServices_NoLongerHardcodeDeliveryStatus()
    {
        foreach (var f in new[] { "DeliveryScheduleService.cs", "ProcessOutsourceOrderService.cs" })
        {
            var code = Source("Supply", f);
            var sqlOnly = string.Join('\n', code.Split('\n')
                .Where(l => !l.TrimStart().StartsWith("//", StringComparison.Ordinal)));

            Assert.DoesNotContain("delivery_status=VALUES(delivery_status)", sqlOnly);
            Assert.DoesNotContain("'CANCELLED'", sqlOnly);
            Assert.DoesNotContain("'PARTIAL'", sqlOnly);
        }
    }

    /// <summary>
    /// 租户是唯一安全边界，不得被工厂替代。
    /// <para>该链的源侧根本没有可用的数字工厂号（源表只有 varchar 的 Site/POSite，
    /// 三个写入方都写字面量 1），所以 enrichment JOIN 只加租户、不加工厂 ——
    /// 加了反而会在 factory_id 为 NULL 与 1 混存时把行丢掉。</para>
    /// </summary>
    [Fact]
    public void StdBuild_EnrichmentJoins_DoNotSubstituteFactoryForTenant()
    {
        var block = StdInsertBlock();

        // 不得出现"只按工厂不按租户"的写法。
        Assert.DoesNotContain("m.factory_id = d.factory_id", block);
        Assert.DoesNotContain("i.factory_id = d.factory_id", block);
    }
}