using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Xunit;
namespace Admin.NET.Plugin.AiDOP.Tests.DataPlatform;
///
/// 采购交付 Authority（mdp_std_purchase_order → dwd_supplier_delivery）的生产契约。
///
/// 这条链是 S8 Rule01 的唯一数据来源。它曾同时坏在三个地方，且三处根因都落在同一段
/// STD 构建 SQL 上：丢了租户谓词（跨租户串值）、丢了批次作用域（孤儿只增不减）、
/// 以及有两个业务 API 绕过它直写下游 DWD。本文件守住修好之后的形状。
///
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;
}
/// 取 mdp_std_purchase_order 那一条 INSERT 的 SQL 区间。
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];
}
///
/// STD 构建的两个 enrichment JOIN 必须自带租户谓词。
///
/// 为什么不能靠作用域注入兜底：MdpSqlScope.InjectTenantFactory 的实现是
/// Regex.Replace(sql, @"\bWHERE\b", ...)，它只改写 WHERE，
/// 对 JOIN ... ON 零覆盖。这不是配置问题，是机制上够不到。
///
/// 后果不是行数膨胀：扇出会被 uk_po_line(tenant_id, po_no, po_line) 的
/// ON DUPLICATE KEY UPDATE 吸收，最终行数不变，但 item_name / supplier_code / order_date / buyer
/// 这些 enrichment 值由 MySQL 任取一行决定 —— 症状是取值随机化，不是报错。
/// 实测单租户扇出 870/220 = 3.95 倍，item_code 跨租户重名 5529 组。
///
[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);
}
///
/// DWD 构建的三个 JOIN 本来就是租户安全的，本批不得把它们改坏。
/// 它们已带 po.tenant_id=X.tenant_id AND po.factory_id=X.factory_id，
/// 泄漏点在 STD 而不在这里 —— 这条断言是防止后续改动误伤。
///
[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);
}
///
/// STD 构建必须按本轮批次收窄驱动行。
/// 少了它，INSERT 会把整张贴源历史重新物化，并给死行盖上最新批次戳 ——
/// 实测租户 797 因此在 STD 沉淀了 336 行源侧已不存在的采购订单行（518 行中的 64.9%），
/// 且全部 518 行共享同一个批次号，连时间戳都认不出哪些是死的。
///
[Fact]
public void StdBuild_DrivingRows_ScopedToCurrentBatch()
{
var block = StdInsertBlock();
Assert.Contains("AND d.source_table='PurOrdDetail' AND d.sync_batch_id=@BatchId", block);
}
///
/// enrichment 必须按 (租户, 键) 去重取最新一行，且**不得**用 GROUP_CONCAT 实现。
/// 贴源层同时存在两代 source_biz_key（旧 Domain:PurOrd:Line 与新的裸 RecID），
/// 两代永远不会互相 ODKU 覆盖，因此同一个 PurOrd/ItemNum 在同租户内有多行 —— 不去重就仍会扇出。
/// 禁用 GROUP_CONCAT 的理由是硬的：raw_data 是 JSON 大字段，
/// 而 group_concat_max_len 实测仅 1024，拼接必然截断、首元素本身就可能超限。
///
[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);
}
///
/// 每轮淘汰必须存在，且只在全量路径执行。
/// 只加批次过滤只能让死行「可识别」，不能「退役」—— 运行期链路里没有任何 DELETE。
/// 而增量批次只含变更行，按它淘汰会把未变更的存量行全删掉，所以必须按路径分流。
///
[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);
}
/// 取一条语句的 SQL 区间（从锚点到原始字符串字面量的收尾 """）。
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];
}
///
/// 本文件新增/改写的每条 SQL，经 处理后
/// 都不得出现「注入到不存在的列上」。
///
/// 为什么必须用真注入器跑一遍，而不是比对源码文本：这个坑有两个入口，
/// 光看源码任何一处都不显眼，只有跑一遍注入才暴露。二者本地沙箱都实测炸过：
///
///
/// - 空前缀 —— ResolveSourcePrefix 反向找最近的同层 FROM 表 别名。
/// 派生表外层是 FROM (，别名正则匹配不到括号，于是回退成空前缀，生成裸
/// factory_id；派生表 SELECT 列表里没有这一列 → 整轮转换失败。
/// - 别名指向没有该列的表 —— 淘汰语句的 EXISTS 子查询若写成
/// l.tenant_id = s.tenant_id（关联外层、不含 @TenantId），会被注入
/// COALESCE(NULLIF(l.factory_id,0),1)=@FactoryId，而 mdp_sync_log 没有
/// factory_id 列 → 同样整轮失败。
///
///
/// 失败形态是整轮跑批 FAILED，不是少几行数据 —— 而 Authority 一轮不产出，
/// S8 的恢复闸门就判 DEGRADED、整个租户的恢复停摆。所以这条守卫值得用真注入器跑。
///
[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);
}
///
/// 驱动行必须**显式**带上租户与工厂谓词，不能依赖注入器补。
///
/// 历史上它确实是靠注入的：ResolveSourcePrefix 反向找到 FROM ... d
/// 就能补出 d. 前缀。但两个 enrichment 派生表插进 FROM ... d 与本
/// WHERE 之间以后，反向扫描会先撞上派生表的 FROM (，别名正则匹配不到括号
/// → 回退成空前缀，生成裸 tenant_id。而派生表 m / i 也各自暴露 tenant_id，
/// 该列立刻变成歧义列 —— 整轮转换 FAILED，不是少几行。
///
/// 所以这不是风格偏好：只要这条 INSERT 还带派生表，作用域就必须写死。
///
[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"));
}
///
/// Rule01 Authority 链上的三条语句必须**完全自带**作用域 —— 注入对它们应当是恒等变换。
///
/// 这是本文件与注入器之间的边界声明：这三条 SQL 的作用域由字面 SQL 唯一决定，
/// 不交给一个够不到 JOIN...ON、认不出派生表别名、且对 SQL 注释同样生效的正则去猜。
/// 恒等变换是这里唯一可接受的结果；任何非恒等都意味着有一个 WHERE 漏了
/// @TenantId，而它下一步会以整轮跑批失败的形式暴露。
///
[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));
}
///
/// DWD 必须做同日淘汰，且**放在 INSERT 之后**。
/// 放之前会让快照短暂为空；此处无事务，并发的 Rule01 取数会读到空快照，
/// 把全部在办异常判成已恢复 —— 那正是 Authority 健康门禁要防的形态。
/// 该淘汰同时保证 MAX(stat_date) 取到的是单一批次：实测同一天混入两个批次
/// 已发生 3 次（如 2026-06-07 的 178 行 + 7 行）。
///
[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);
}
///
/// dwd_supplier_delivery 与 mdp_std_purchase_order 在运行期只能有一个写入方。
///
/// 它们是 Rule01 的 Authority。此前另有两个业务 API 直写：
/// ProcessOutsourceOrderService 把 delivery_status 硬编码为 'OPEN' 并放进
/// ON DUPLICATE KEY UPDATE，一次建单就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN；
/// DeliveryScheduleService 则写出 CANCELLED / CLOSED / PARTIAL 三个没有任何读方认识的值。
///
/// 更严重的是它们能凭空造出一个「最新快照」：Rule01 取 MAX(stat_date) 那一天的全部行，
/// 实测租户 797403760988229 的 2026-05-16 快照只有 2 行、且全部出自旁路写入 ——
/// 中台停摆时，一次业务 API 调用就让 Rule01 在 2 行残缺数据上评估，而快照看上去还很新鲜。
///
/// 用「非注释行里出现写入语句」判定，避免被解释性注释里的表名误判。
///
[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} 处");
}
}
///
/// 旁路写入方不得把 delivery_status 硬编码后写进 Authority。
/// 状态只能由正式 MDP 转换从源侧派生（COMPLETED / DELAYED / OPEN 三值）。
///
[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);
}
}
///
/// 租户是唯一安全边界，不得被工厂替代。
/// 该链的源侧根本没有可用的数字工厂号（源表只有 varchar 的 Site/POSite，
/// 三个写入方都写字面量 1），所以 enrichment JOIN 只加租户、不加工厂 ——
/// 加了反而会在 factory_id 为 NULL 与 1 混存时把行丢掉。
///
[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);
}
}