using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using SqlSugar;
namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;
///
/// 165 LocationDetail ↔ mdp_std_inventory 日终对账 + 冷链滞后检测。
///
/// **对账口径不变量**:LIVE 与 STD 必须是同一集合。标准层是「租户 + 该租户 LocationMaster
/// 合法库位(Typed <> 'Supp')」维度的快照,因此 LIVE 侧也必须按同一租户的同一库位白名单收窄,
/// 且逐租户各对一次。历史缺陷:LIVE 取整个 Domain、STD 取单一租户,
/// 差集里全是「本租户根本不该有」的库位,产生系统性假差异(详见 CHANGELOG / 报告)。
///
///
/// 库位白名单一律由 提供,与查询层、标准层写入层同源;
/// 白名单为空 → 该租户跳过(fail closed),**绝不退回整个 Domain**。
///
///
public sealed class InventoryReconService : ITransient
{
private readonly ISqlSugarClient _db;
private readonly MdpSourceScopeFactory _scopeFactory;
private readonly SourceDomainTenantResolver _domainTenant;
private readonly AidopInventoryOptions _opt;
private readonly ILogger _logger;
public InventoryReconService(
ISqlSugarClient db,
MdpSourceScopeFactory scopeFactory,
SourceDomainTenantResolver domainTenant,
IOptions opt,
ILoggerFactory loggerFactory)
{
_db = db;
_scopeFactory = scopeFactory;
_domainTenant = domainTenant;
_opt = opt.Value;
_logger = loggerFactory.CreateLogger(nameof(InventoryReconService));
}
public async Task RunDailyReconAsync(CancellationToken cancellationToken = default)
{
var sourceCode = string.IsNullOrWhiteSpace(_opt.SourceCode) ? "DOPDEMORQ_SQLSERVER" : _opt.SourceCode.Trim();
var domain = string.IsNullOrWhiteSpace(_opt.DefaultDomain) ? "8010" : _opt.DefaultDomain.Trim();
// 源归属租户:只校验 (source, domain) 已登记贴源落地区(未登记 → fail closed)。
// 它是 stg 落地区归属,**不是**业务归属,绝不能当成对账的租户口径。
var sourceOwnerTenantId = await _domainTenant.ResolveTenantIdAsync(sourceCode, domain, cancellationToken);
var asOf = DateTime.Now;
var batchId = $"S5_INV_RECON_{asOf:yyyyMMddHHmmss}";
await CheckSyncLagAsync(cancellationToken);
var tenants = await ListInventoryScopedTenantsAsync(domain, cancellationToken);
if (tenants.Count == 0)
{
_logger.LogWarning(
"[InventoryRecon] domain={Domain} 无任何配置了合法库位的租户,本轮不对账 batch={Batch}", domain, batchId);
return 0;
}
var remote = await _scopeFactory.GetScopeAsync(sourceCode, cancellationToken);
var diffCount = 0;
foreach (var tenantId in tenants)
{
cancellationToken.ThrowIfCancellationRequested();
diffCount += await ReconcileTenantAsync(
remote, tenantId, domain, sourceCode, batchId, asOf, cancellationToken);
}
if (diffCount > 0)
_logger.LogWarning(
"[InventoryRecon] batch={Batch} tenants={Tenants} sourceOwner={Owner} diffs={Diffs}",
batchId, tenants.Count, sourceOwnerTenantId, diffCount);
else
_logger.LogInformation(
"[InventoryRecon] batch={Batch} tenants={Tenants} clean", batchId, tenants.Count);
return diffCount;
}
///
/// 单租户对账:LIVE 与 STD 用**同一份库位白名单**收窄,保证差集只反映真实数量偏差。
///
private async Task ReconcileTenantAsync(
ISqlSugarClient remote,
long tenantId,
string domain,
string sourceCode,
string batchId,
DateTime asOf,
CancellationToken cancellationToken)
{
var scope = TenantLocationScope.FromWhitelist(
await TenantLocationScopeLoader.LoadAsync(_db, tenantId, domain, cancellationToken));
// fail closed:空白名单不等于全 Domain;不查源库、不写差异。
if (scope.IsEmpty)
{
_logger.LogWarning(
"[InventoryRecon] tenant={Tenant} domain={Domain} 库位白名单为空,跳过对账(fail closed)",
tenantId, domain);
return 0;
}
// 165 = SQL Server:库位值一律参数化下发,禁止拼进 SQL 文本。
var (liveLocationClause, liveLocationPars) = scope.BuildInClause("Location", "loc");
var livePars = new List { new("@Domain", domain) };
livePars.AddRange(liveLocationPars);
var live = await remote.Ado.SqlQueryAsync(
$"""
SELECT Domain, ItemNum, ISNULL(LotSerial,'') AS LotSerial, Location,
ISNULL(AvailStatusQty,0) AS Qty
FROM LocationDetail
WHERE Domain = @Domain
AND {liveLocationClause}
AND (ISNULL(AvailStatusQty,0)+ISNULL(Assay,0)+ISNULL(FreezeQty,0)) <> 0
""",
livePars);
// STD 侧同样按白名单收窄:标准层历史脏快照(修复前写入、越界库位)不得被当成真实差异。
var (stdLocationClause, stdLocationPars) = scope.BuildInClause("location", "loc");
var stdPars = new List
{
new("@TenantId", tenantId),
new("@Domain", domain),
new("@SourceSystem", sourceCode)
};
stdPars.AddRange(stdLocationPars);
var std = await _db.Ado.SqlQueryAsync(
$"""
SELECT domain AS Domain, item_num AS ItemNum, IFNULL(lot_serial,'') AS LotSerial,
location AS Location, IFNULL(qty_unrestricted,0) AS Qty
FROM mdp_std_inventory
WHERE tenant_id=@TenantId AND domain=@Domain AND source_system=@SourceSystem
AND {stdLocationClause}
""",
stdPars);
var liveMap = live.ToDictionary(
x => Key(x), x => x.Qty, StringComparer.OrdinalIgnoreCase);
var stdMap = std.ToDictionary(
x => Key(x), x => x.Qty, StringComparer.OrdinalIgnoreCase);
var keys = liveMap.Keys.Union(stdMap.Keys, StringComparer.OrdinalIgnoreCase).ToList();
var diffCount = 0;
foreach (var key in keys)
{
cancellationToken.ThrowIfCancellationRequested();
liveMap.TryGetValue(key, out var qLive);
stdMap.TryGetValue(key, out var qStd);
var diff = qLive - qStd;
if (diff == 0) continue;
diffCount++;
var parts = key.Split('\u001f');
await _db.Ado.ExecuteCommandAsync(
"""
INSERT INTO ado_inventory_recon_diff
(tenant_id, domain, item_num, lot_serial, location,
qty_live, qty_std, qty_diff, recon_batch_id, as_of, remark)
VALUES
(@TenantId, @Domain, @ItemNum, @Lot, @Loc,
@QtyLive, @QtyStd, @QtyDiff, @BatchId, @AsOf, NULL)
""",
new List
{
new("@TenantId", tenantId),
new("@Domain", parts.ElementAtOrDefault(0) ?? domain),
new("@ItemNum", parts.ElementAtOrDefault(1) ?? ""),
new("@Lot", parts.ElementAtOrDefault(2) ?? ""),
new("@Loc", parts.ElementAtOrDefault(3) ?? ""),
new("@QtyLive", qLive),
new("@QtyStd", qStd),
new("@QtyDiff", diff),
new("@BatchId", batchId),
new("@AsOf", asOf)
});
}
_logger.LogInformation(
"[InventoryRecon] tenant={Tenant} domain={Domain} allowedLoc={Loc} live={Live} std={Std} diffs={Diffs} batch={Batch}",
tenantId, domain, scope.Count, liveMap.Count, stdMap.Count, diffCount, batchId);
return diffCount;
}
///
/// 枚举该 domain 下**拥有合法库存范围**的租户 —— 必须与标准层物化时的租户枚举口径一致,
/// 否则会出现「物化了却从不对账」或「对账一个没有快照的租户」的盲区。
/// (口径同 InventoryMdpSyncService.ListInventoryScopedTenantsAsync,改动须两处同步。)
///
private async Task> ListInventoryScopedTenantsAsync(string domain, CancellationToken ct)
{
return await _db.Ado.SqlQueryAsync(
"""
SELECT DISTINCT lm.tenant_id
FROM LocationMaster lm
JOIN SysTenant t ON t.Id = lm.tenant_id AND t.Status = 1
WHERE lm.Domain = @Domain
AND IFNULL(lm.typed,'') <> 'Supp'
AND TRIM(lm.location) <> ''
ORDER BY lm.tenant_id
""",
new List { new("@Domain", domain) });
}
private async Task CheckSyncLagAsync(CancellationToken ct)
{
var maxAge = _opt.StdMaxAgeMinutes <= 0 ? 90 : _opt.StdMaxAgeMinutes;
var entities = await _db.Queryable()
.Where(x => x.EntityCode == "S5_LOCATION_DETAIL_SQLSERVER"
|| x.EntityCode == "S5_INV_TRANS_HIST_SQLSERVER")
.ToListAsync(ct);
foreach (var e in entities)
{
if (!e.LastSyncTo.HasValue)
{
_logger.LogWarning("[InventoryRecon] 冷链无 last_sync_to entity={Code}", e.EntityCode);
continue;
}
var age = (DateTime.Now - e.LastSyncTo.Value).TotalMinutes;
if (age > maxAge)
_logger.LogWarning(
"[InventoryRecon] 冷链滞后 entity={Code} lastSync={Last} ageMin={Age} threshold={Th}",
e.EntityCode, e.LastSyncTo, (int)age, maxAge);
}
}
private static string Key(QtyKeyRow r) =>
$"{r.Domain}\u001f{r.ItemNum}\u001f{r.LotSerial}\u001f{r.Location}";
private sealed class QtyKeyRow
{
public string? Domain { get; set; }
public string? ItemNum { get; set; }
public string? LotSerial { get; set; }
public string? Location { get; set; }
public decimal Qty { get; set; }
}
}