using Admin.NET.Plugin.AiDOP.DataPlatform; using Admin.NET.Plugin.AiDOP.DataPlatform.Schema; using Admin.NET.Plugin.AiDOP.DataPlatform.Executors; using Admin.NET.Plugin.AiDOP.Infrastructure; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using SqlSugar; namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse; /// /// S5 库存冷链:165 LocationDetail / InvTransHist → stg → std。 /// 余额事实源仅 LocationDetail;InvMaster 不入标准层。 /// public sealed class InventoryMdpSyncService : ITransient { /// 跨实例互斥锁键;真正的取/放锁生命周期见 。 public const string LockKey = InventoryInboundLockGuard.LockKey; private const string LocationEntity = "S5_LOCATION_DETAIL_SQLSERVER"; private const string TransEntity = "S5_INV_TRANS_HIST_SQLSERVER"; private const string SourceCodeDefault = "DOPDEMORQ_SQLSERVER"; private readonly ISqlSugarClient _db; private readonly MdpSourcePullDispatcher _pullDispatcher; private readonly MdpSourceScopeFactory _scopeFactory; private readonly SourceDomainTenantResolver _domainTenant; private readonly AidopInventoryOptions _opt; private readonly ILogger _logger; public InventoryMdpSyncService( ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher, MdpSourceScopeFactory scopeFactory, SourceDomainTenantResolver domainTenant, IOptions opt, ILoggerFactory loggerFactory) { _db = db; _pullDispatcher = pullDispatcher; _scopeFactory = scopeFactory; _domainTenant = domainTenant; _opt = opt.Value; _logger = loggerFactory.CreateLogger(nameof(InventoryMdpSyncService)); } public Task RunBootstrapAsync(CancellationToken cancellationToken = default) => RunAsync(bootstrap: true, reconcile: false, cancellationToken); public Task RunIncrementalAsync(CancellationToken cancellationToken = default) => RunAsync(bootstrap: false, reconcile: false, cancellationToken); public Task RunReconcileFullAsync(CancellationToken cancellationToken = default) => RunAsync(bootstrap: false, reconcile: true, cancellationToken); /// /// 仅 stg→std:**不访问源库**。用于链路修复后,复用已完整落地的贴源层让各业务租户 /// 按当前库位范围口径重新物化标准层。 /// /// 语义为 FULL REPLACE(逐租户、逐正式切片):贴源层是完整历史, /// 只跑 UPSERT 既清不掉旧口径残留、也补不齐从未物化过的租户。 /// 删除范围严格限定「该租户 + 当前正式 source_system + 当前 domain」, /// 不触碰 UAT_GENERATOR 等非正式切片。 /// /// public async Task TransformTransStdFromStgAsync(CancellationToken cancellationToken = default) { cancellationToken.ThrowIfCancellationRequested(); var sourceCode = string.IsNullOrWhiteSpace(_opt.SourceCode) ? SourceCodeDefault : _opt.SourceCode.Trim(); var domain = string.IsNullOrWhiteSpace(_opt.DefaultDomain) ? "8010" : _opt.DefaultDomain.Trim(); var tenantId = await _domainTenant.ResolveTenantIdAsync(sourceCode, domain, cancellationToken); var asOf = DateTime.Now; var months = _opt.TransBootstrapMonths <= 0 ? 12 : _opt.TransBootstrapMonths; var historyFrom = asOf.Date.AddMonths(-months); var batchId = $"S5_INV_XFORM_{asOf:yyyyMMddHHmmss}"; // await using:异常路径也一定走到释放(DisposeAsync 内先 RELEASE_LOCK 再关专用连接) await using var guard = await InventoryInboundLockGuard.TryAcquireAsync(_db, _logger, cancellationToken: cancellationToken); if (!guard.Acquired) { _logger.LogWarning("[InventoryMdpSync] 互斥锁占用,transform-std 本轮跳过 batch={Batch} reason={Reason}", batchId, guard.BusyReason); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, AsOf = asOf, HistoryFrom = historyFrom, Skipped = true, Message = "lock busy" }; } var targetTenants = await ListInventoryScopedTenantsAsync(domain, cancellationToken); if (targetTenants.Count == 0) { // fail closed:没有任何配了合法库位的租户时不写标准层,绝不回落成「按源归属租户写一份」 _logger.LogWarning( "[InventoryMdpSync] domain={Domain} 无任何配置了合法库位的租户,transform-std 本轮不写入", domain); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, TransStdRows = 0, AsOf = asOf, HistoryFrom = historyFrom, Skipped = true, Message = "no scoped tenant" }; } using var xformTimeout = WithLongCommandTimeout(); var rows = 0; foreach (var targetTenantId in targetTenants) { cancellationToken.ThrowIfCancellationRequested(); var n = await MdpStdFullReplace.ReplaceAsync( _db, "mdp_std_inv_trans", targetTenantId, FormalSliceWhere(sourceCode, domain), () => MaterializeInvTransStdAsync( tenantId, targetTenantId, batchId: null, asOf, historyFrom, sourceCode), cancellationToken); rows += n; _logger.LogInformation( "[InventoryMdpSync] transform-std materialized tenant={Tenant} domain={Domain} rows={Rows}", targetTenantId, domain, n); } _logger.LogInformation( "[InventoryMdpSync] transform-std done source={Source} tenants={Cnt} rows={Rows}", tenantId, targetTenants.Count, rows); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, TransStdRows = rows, AsOf = asOf, HistoryFrom = historyFrom, Message = "OK transform-std" }; } /// /// 仅 stg→std(库存余额):**不访问源库**,从一个已完整落地的贴源批次重新物化标准层。 /// /// 用途:同步链路修复后,复用既有完整 stg 批次让各租户按新的库位范围口径重新物化, /// 避免为此重新全量拉取源库。 /// /// /// 语义为 UPSERT,不做 FULL REPLACE:只按业务键写入/更新本批次覆盖到的行, /// 不删除任何既有标准层数据 —— 单个增量批次不代表全量,replace 会造成数据丢失。 /// 因此本入口不负责清理历史脏快照,那属于全量校准(reconcile)的职责。 /// /// /// 贴源批次号(sync_batch_id),必须是已完整落地的批次。 public async Task TransformInventoryStdFromStgAsync( string stgBatchId, CancellationToken cancellationToken = default) { cancellationToken.ThrowIfCancellationRequested(); if (string.IsNullOrWhiteSpace(stgBatchId) || !System.Text.RegularExpressions.Regex.IsMatch(stgBatchId, @"^[A-Za-z0-9_]+$")) throw new InvalidOperationException("非法贴源批次号"); var sourceCode = string.IsNullOrWhiteSpace(_opt.SourceCode) ? SourceCodeDefault : _opt.SourceCode.Trim(); var domain = string.IsNullOrWhiteSpace(_opt.DefaultDomain) ? "8010" : _opt.DefaultDomain.Trim(); var sourceTenantId = await _domainTenant.ResolveTenantIdAsync(sourceCode, domain, cancellationToken); var asOf = DateTime.Now; // 本入口同样写 mdp_std_inventory,必须与 bootstrap/reconcile/incremental 互斥(原实现漏取锁) await using var guard = await InventoryInboundLockGuard.TryAcquireAsync(_db, _logger, cancellationToken: cancellationToken); if (!guard.Acquired) { _logger.LogWarning("[InventoryMdpSync] 互斥锁占用,transform-inventory-std 本轮跳过 batch={Batch} reason={Reason}", stgBatchId, guard.BusyReason); return new InventorySyncResult { BatchId = stgBatchId, TenantId = sourceTenantId, Domain = domain, AsOf = asOf, Skipped = true, Message = "lock busy" }; } var stgRows = await _db.Ado.GetIntAsync( """ SELECT COUNT(1) FROM mdp_stg_inventory WHERE tenant_id=@TenantId AND source_system=@SourceSystem AND source_table='LocationDetail' AND sync_batch_id=@BatchId """, new List { new("@TenantId", sourceTenantId), new("@SourceSystem", sourceCode), new("@BatchId", stgBatchId) }); if (stgRows == 0) throw new InvalidOperationException( $"贴源批次为空或不属于源归属租户:batch={stgBatchId}, sourceTenant={sourceTenantId}"); var targetTenants = await ListInventoryScopedTenantsAsync(domain, cancellationToken); var total = 0; foreach (var targetTenantId in targetTenants) { cancellationToken.ThrowIfCancellationRequested(); var rows = await InsertInventoryStdAsync( targetTenantId, sourceTenantId, stgBatchId, asOf, sourceCode, replaceMode: false); total += rows; _logger.LogInformation( "[InventoryMdpSync] std re-materialized from stg tenant={Tenant} domain={Domain} batch={Batch} rows={Rows}", targetTenantId, domain, stgBatchId, rows); } return new InventorySyncResult { BatchId = stgBatchId, TenantId = sourceTenantId, Domain = domain, InventoryStdRows = total, AsOf = asOf, Message = $"OK transform-inventory-std from stg (stgRows={stgRows}, tenants={targetTenants.Count})" }; } /// /// 将指定租户已存在的全部库存交易贴源批次转换到标准层。 /// 供租户级模块重算使用,不拉外部源,也不依赖默认 Domain→Tenant 映射。 /// public async Task TransformTransStdFromStgAsync( long tenantId, CancellationToken cancellationToken = default) { if (tenantId <= 0) throw new ArgumentOutOfRangeException(nameof(tenantId)); cancellationToken.ThrowIfCancellationRequested(); var sources = await _db.Ado.SqlQueryAsync( """ SELECT DISTINCT source_system AS SourceSystem FROM mdp_stg_inv_trans WHERE tenant_id=@TenantId AND source_table='InvTransHist' AND NULLIF(TRIM(source_system),'') IS NOT NULL """, new SugarParameter("@TenantId", tenantId)); var asOf = DateTime.Now; var months = _opt.TransBootstrapMonths <= 0 ? 12 : _opt.TransBootstrapMonths; var historyFrom = asOf.Date.AddMonths(-months); var affected = 0; foreach (var source in sources) { cancellationToken.ThrowIfCancellationRequested(); if (string.IsNullOrWhiteSpace(source.SourceSystem)) continue; affected += await MaterializeInvTransStdAsync( tenantId, tenantId, batchId: null, asOf, historyFrom, source.SourceSystem); } return affected; } private async Task RunAsync(bool bootstrap, bool reconcile, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); var sourceCode = string.IsNullOrWhiteSpace(_opt.SourceCode) ? SourceCodeDefault : _opt.SourceCode.Trim(); var domain = string.IsNullOrWhiteSpace(_opt.DefaultDomain) ? "8010" : _opt.DefaultDomain.Trim(); // 源归属租户:只决定贴源层(stg)落在谁名下,**不代表业务归属**; // 业务归属在标准层物化时按各租户 LocationMaster 库位范围投影决定。 var tenantId = await _domainTenant.ResolveTenantIdAsync(sourceCode, domain, cancellationToken); var asOf = DateTime.Now; var batchId = $"S5_INV_{(bootstrap ? "BOOT" : reconcile ? "RECON" : "INCR")}_{asOf:yyyyMMddHHmmss}"; var months = _opt.TransBootstrapMonths <= 0 ? 12 : _opt.TransBootstrapMonths; var historyFrom = asOf.Date.AddMonths(-months); // await using:异常路径也一定走到释放(DisposeAsync 内先 RELEASE_LOCK 再关专用连接) await using var guard = await InventoryInboundLockGuard.TryAcquireAsync(_db, _logger, cancellationToken: cancellationToken); if (!guard.Acquired) { _logger.LogWarning("[InventoryMdpSync] 跨实例锁占用,本轮跳过 batch={Batch} reason={Reason}", batchId, guard.BusyReason); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, Bootstrap = bootstrap, AsOf = asOf, HistoryFrom = historyFrom, Skipped = true, Message = "lock busy" }; } var upperLoc = await CaptureUpperBoundAsync(sourceCode, "LocationDetail", "UpdateTime", cancellationToken); var upperTrans = await CaptureUpperBoundAsync(sourceCode, "InvTransHist", "CreateTime", cancellationToken); MdpPullResult locPull; if (bootstrap || reconcile) { var locBatch = $"{batchId}_LOC"; // NULL UpdateTime 段与非 NULL 段共用同一 batchId,保证 Replace 不漏 NULL 行 var nullCtx = BuildKeysetCtx(tenantId, locBatch, asOf, historyFrom, upperLoc, cursorColumn: "UpdateTime", nullPhase: true, bootstrapFull: true); nullCtx.CursorValue = null; nullCtx.TieBreakerValue = null; await _pullDispatcher.PullAllByEntityCodeAsync(LocationEntity, nullCtx, cancellationToken, maxPages: 50); var fullCtx = BuildKeysetCtx(tenantId, locBatch, asOf, historyFrom, upperLoc, cursorColumn: "UpdateTime", nullPhase: false, bootstrapFull: true); fullCtx.CursorValue = null; fullCtx.TieBreakerValue = null; fullCtx.BootstrapFrom = null; // LocationDetail 首刷不截时间窗 locPull = await _pullDispatcher.PullAllByEntityCodeAsync(LocationEntity, fullCtx, cancellationToken, maxPages: 200); } else { var incrCtx = BuildKeysetCtx(tenantId, $"{batchId}_LOC", asOf, historyFrom, upperLoc, cursorColumn: "UpdateTime", nullPhase: false, bootstrapFull: false); // 重叠窗口:从上次游标时间向前回退 LocationOverlapMinutes if (!string.IsNullOrWhiteSpace(incrCtx.CursorValue) && DateTime.TryParse(incrCtx.CursorValue, out var lastDt)) { var overlap = Math.Max(0, _opt.LocationOverlapMinutes); incrCtx.CursorValue = lastDt.AddMinutes(-overlap) .ToString("yyyy-MM-dd HH:mm:ss.fff"); incrCtx.TieBreakerValue = "0"; } locPull = await _pullDispatcher.PullAllByEntityCodeAsync(LocationEntity, incrCtx, cancellationToken, maxPages: 200); } // —— 标准层按租户物化:一次贴源,逐租户按各自库位范围投影 —— // 贴源层是「源+domain」维度(归属 sourceTenantId),标准层是「租户」维度。 // 拉取游标持久化在 mdp_entity 上、跨租户共享,故绝不能为每个租户各拉一次。 var targetTenants = await ListInventoryScopedTenantsAsync(domain, cancellationToken); if (targetTenants.Count == 0) _logger.LogWarning( "[InventoryMdpSync] domain={Domain} 无任何配置了合法库位的租户,标准层本轮不写入", domain); var inventoryStdRows = 0; foreach (var targetTenantId in targetTenants) { cancellationToken.ThrowIfCancellationRequested(); int rows; if (bootstrap || reconcile) { rows = await MdpStdFullReplace.ReplaceAsync( _db, "mdp_std_inventory", targetTenantId, "source_system='DOPDEMORQ_SQLSERVER'", () => InsertInventoryStdAsync(targetTenantId, tenantId, $"{batchId}_LOC", asOf, sourceCode, replaceMode: true), cancellationToken); } else { rows = await InsertInventoryStdAsync(targetTenantId, tenantId, $"{batchId}_LOC", asOf, sourceCode, replaceMode: false); } inventoryStdRows += rows; _logger.LogInformation( "[InventoryMdpSync] std materialized tenant={Tenant} domain={Domain} rows={Rows}", targetTenantId, domain, rows); } var transCtx = BuildKeysetCtx(tenantId, $"{batchId}_TRN", asOf, historyFrom, upperTrans, cursorColumn: "CreateTime", nullPhase: false, bootstrapFull: bootstrap || reconcile); if (bootstrap || reconcile) { transCtx.CursorValue = null; transCtx.TieBreakerValue = null; transCtx.BootstrapFrom = historyFrom; } var transPull = await _pullDispatcher.PullAllByEntityCodeAsync(TransEntity, transCtx, cancellationToken, maxPages: 500); // —— 流水腿与余额腿同构:一次贴源,逐业务租户按各自库位范围投影 —— using var transTimeout = WithLongCommandTimeout(); var transStdRows = 0; foreach (var targetTenantId in targetTenants) { cancellationToken.ThrowIfCancellationRequested(); int rows; if (bootstrap || reconcile) { rows = await MdpStdFullReplace.ReplaceAsync( _db, "mdp_std_inv_trans", targetTenantId, FormalSliceWhere(sourceCode, domain), () => MaterializeInvTransStdAsync( tenantId, targetTenantId, batchId: null, asOf, historyFrom, sourceCode), cancellationToken); } else { rows = await MaterializeInvTransStdAsync( tenantId, targetTenantId, $"{batchId}_TRN", asOf, historyFrom, sourceCode); } transStdRows += rows; _logger.LogInformation( "[InventoryMdpSync] trans std materialized tenant={Tenant} domain={Domain} rows={Rows}", targetTenantId, domain, rows); } _logger.LogInformation( "[InventoryMdpSync] done batch={Batch} boot={Boot} recon={Recon} locPulled={LocP} invStd={Inv} trnPulled={TrnP} trnStd={Trn}", batchId, bootstrap, reconcile, locPull.RowsPulled, inventoryStdRows, transPull.RowsPulled, transStdRows); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, Bootstrap = bootstrap, LocationPulled = locPull.RowsPulled, LocationWritten = locPull.RowsWritten, InventoryStdRows = inventoryStdRows, TransPulled = transPull.RowsPulled, TransWritten = transPull.RowsWritten, TransStdRows = transStdRows, AsOf = asOf, HistoryFrom = historyFrom, Message = "OK" }; } private MdpPullContext BuildKeysetCtx( long tenantId, string batchId, DateTime asOf, DateTime historyFrom, (string? Cursor, string? Tie) upper, string cursorColumn, bool nullPhase, bool bootstrapFull) { return new MdpPullContext { TenantId = tenantId, BatchId = batchId, FullRefresh = false, UseKeysetCursor = true, CursorColumn = cursorColumn, TieBreakerColumn = "RecID", UpperCursorValue = upper.Cursor, UpperTieBreakerValue = upper.Tie, BootstrapFrom = bootstrapFull && cursorColumn == "CreateTime" ? historyFrom : null, DeferCursorPersist = false, NullTimePhase = nullPhase, // 首刷/校准不得继承实体脏水位,否则只会抽到「游标之后」的尾巴 SkipPersistedKeysetCursor = bootstrapFull || nullPhase }; } private async Task<(string? Cursor, string? Tie)> CaptureUpperBoundAsync( string sourceCode, string table, string cursorColumn, CancellationToken ct) { if (!System.Text.RegularExpressions.Regex.IsMatch(table, @"^[A-Za-z0-9_]+$") || !System.Text.RegularExpressions.Regex.IsMatch(cursorColumn, @"^[A-Za-z0-9_]+$")) throw new InvalidOperationException("非法上界查询标识符"); var remote = await _scopeFactory.GetScopeAsync(sourceCode, ct); var rows = await remote.Ado.SqlQueryAsync( $""" SELECT TOP 1 CONVERT(varchar(30), {cursorColumn}, 121) AS CursorText, CAST(RecID AS varchar(30)) AS TieText FROM {table} WHERE {cursorColumn} IS NOT NULL ORDER BY {cursorColumn} DESC, RecID DESC """); var hit = rows.FirstOrDefault(); return (hit?.CursorText, hit?.TieText); } /// /// stg → std 物化:**按目标租户的合法库位范围投影**。 /// /// 贴源层(stg)是「源 + domain」维度的全量落地区,归属 ; /// 标准层(std)是「租户」维度的可见快照,因此这里必须内联 LocationMaster 做投影: /// 只有落在目标租户自己 LocationMaster(同 Domain 且 Typed <> 'Supp')内的库位才写入。 /// /// /// 写入不变量:∀ 写入行 → tenant_id = targetTenantId /// ∧ location ∈ AllowedLocations(targetTenantId) ∧ domain = 该租户 LocationMaster 的 Domain。 /// 租户白名单为空 → JOIN 命中 0 行 → 写 0 条(fail closed,绝不退回整个 Domain)。 /// /// /// tenant_id 直接取 而非 MdpJsonSql.TenantFromStg: /// 贴源行的 tenant 是「源落地区归属」,不是业务归属,不能顺着传下来。 /// /// private async Task InsertInventoryStdAsync( long targetTenantId, long sourceTenantId, string batchId, DateTime asOf, string sourceSystem, bool replaceMode) { // replaceMode:MdpStdFullReplace 已 DELETE,直接 INSERT; // 增量:UPSERT 本批变化。 var sTenant = "@TargetTenantId"; var sql = replaceMode ? $""" INSERT INTO mdp_std_inventory (tenant_id, source_system, domain, location, lot_serial, item_num, dimension1, dimension2, refs, site, inv_status, qty_on_hand, qty_unrestricted, qty_inspection, qty_frozen, qty_available, src_rec_id, source_update_time, as_of, sync_batch_id) SELECT {sTenant}, @SourceSystem, IFNULL({MdpJsonSql.Str("s", "Domain")}, ''), IFNULL({MdpJsonSql.Str("s", "Location")}, ''), IFNULL({MdpJsonSql.Str("s", "LotSerial")}, ''), IFNULL({MdpJsonSql.Str("s", "ItemNum")}, ''), IFNULL({MdpJsonSql.Str("s", "Dimension1")}, ''), IFNULL({MdpJsonSql.Str("s", "Dimension2")}, ''), IFNULL({MdpJsonSql.Str("s", "Refs")}, ''), IFNULL({MdpJsonSql.Str("s", "Site")}, ''), {MdpJsonSql.Str("s", "InvStatus")}, IFNULL({MdpJsonSql.Dec("s", "QtyOnHand", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "AvailStatusQty", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "Assay", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "FreezeQty", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "AvailStatusQty", 18, 5)}, 0), s.source_row_id, {MdpJsonSql.DateTimeSec("s", "UpdateTime")}, @AsOf, @BatchId FROM mdp_stg_inventory s {TenantScopeJoin} WHERE s.tenant_id=@SourceTenantId AND s.source_system=@SourceSystem AND s.source_table='LocationDetail' AND s.sync_batch_id=@BatchId """ : $""" INSERT INTO mdp_std_inventory (tenant_id, source_system, domain, location, lot_serial, item_num, dimension1, dimension2, refs, site, inv_status, qty_on_hand, qty_unrestricted, qty_inspection, qty_frozen, qty_available, src_rec_id, source_update_time, as_of, sync_batch_id) SELECT {sTenant}, @SourceSystem, IFNULL({MdpJsonSql.Str("s", "Domain")}, ''), IFNULL({MdpJsonSql.Str("s", "Location")}, ''), IFNULL({MdpJsonSql.Str("s", "LotSerial")}, ''), IFNULL({MdpJsonSql.Str("s", "ItemNum")}, ''), IFNULL({MdpJsonSql.Str("s", "Dimension1")}, ''), IFNULL({MdpJsonSql.Str("s", "Dimension2")}, ''), IFNULL({MdpJsonSql.Str("s", "Refs")}, ''), IFNULL({MdpJsonSql.Str("s", "Site")}, ''), {MdpJsonSql.Str("s", "InvStatus")}, IFNULL({MdpJsonSql.Dec("s", "QtyOnHand", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "AvailStatusQty", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "Assay", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "FreezeQty", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "AvailStatusQty", 18, 5)}, 0), s.source_row_id, {MdpJsonSql.DateTimeSec("s", "UpdateTime")}, @AsOf, @BatchId FROM mdp_stg_inventory s {TenantScopeJoin} WHERE s.tenant_id=@SourceTenantId AND s.source_system=@SourceSystem AND s.source_table='LocationDetail' AND s.sync_batch_id=@BatchId ON DUPLICATE KEY UPDATE inv_status=VALUES(inv_status), qty_on_hand=VALUES(qty_on_hand), qty_unrestricted=VALUES(qty_unrestricted), qty_inspection=VALUES(qty_inspection), qty_frozen=VALUES(qty_frozen), qty_available=VALUES(qty_available), src_rec_id=VALUES(src_rec_id), source_update_time=VALUES(source_update_time), as_of=VALUES(as_of), sync_batch_id=VALUES(sync_batch_id), update_time=CURRENT_TIMESTAMP """; return await _db.Ado.ExecuteCommandAsync(sql, new SugarParameter("@TargetTenantId", targetTenantId), new SugarParameter("@SourceTenantId", sourceTenantId), new SugarParameter("@SourceSystem", sourceSystem), new SugarParameter("@BatchId", batchId), new SugarParameter("@AsOf", asOf)); } /// /// 租户库位范围内联投影:贴源行只有落在目标租户自己的合法库位(同 Domain、Typed <> 'Supp') /// 才允许进入标准层。这是标准层写入侧的租户安全边界。 /// /// **余额腿与流水腿共用本实现**,只有贴源 JSON 里的库位字段名不同 /// (LocationDetail 用 Location,InvTransHist 用 Loc)。 /// 禁止再手写第二套语义略有差异的 LocationMaster JOIN。 /// /// /// 贴源 raw_data 中的库位字段名。 private static string TenantScopeJoinSql(string locationJsonKey) => $""" INNER JOIN LocationMaster lm ON lm.tenant_id = @TargetTenantId AND lm.Domain = IFNULL({MdpJsonSql.Str("s", "Domain")}, '') AND lm.location = IFNULL({MdpJsonSql.Str("s", locationJsonKey)}, '') AND IFNULL(lm.typed, '') <> 'Supp' AND TRIM(lm.location) <> '' """; /// 库存余额贴源的库位字段名。 private static readonly string TenantScopeJoin = TenantScopeJoinSql("Location"); /// 进出存流水贴源的库位字段名(InvTransHist 用 Loc)。 private static readonly string TransTenantScopeJoin = TenantScopeJoinSql("Loc"); /// /// 库位角色左联:165 的事务码(iss-tr / rct-tr / rct-wo)单看码判不出阶段, /// 必须结合库位角色与数量方向,见 。 /// 角色配置缺失时取 UNKNOWN —— 规则一律不成立,流水按「库位角色未配置」隔离,不猜。 /// private static readonly string LocationRoleJoin = $""" LEFT JOIN mdp_location_role lr ON lr.tenant_id = @TargetTenantId AND lr.domain = IFNULL({MdpJsonSql.Str("s", "Domain")}, '') AND lr.location = IFNULL({MdpJsonSql.Str("s", "Loc")}, '') """; /// 库位角色表达式(未配置 → UNKNOWN)。 private static readonly string LocationRoleExpr = $"IFNULL(lr.location_role, '{NeutralTransTypeCodes.UnknownRole}')"; /// /// FULL Replace 的删除范围:**只替换「当前租户 + 当前正式源 + 当前 domain」这一份物化投影**。 /// 绝不能只按 tenant_id 删 —— 那会连带删掉其它来源切片。 /// UAT 名下的 UAT_GENERATOR 演示行由 1.0.565 清理,不在这次物化删除范围里。 /// sourceSystem/domain 均来自服务端配置(AidopInventoryOptions),非用户输入; /// 仍做标识符白名单校验,杜绝任何拼接注入。 /// /// /// 标准层物化的命令超时(秒)。默认 30s 不够:正式切片 FULL REPLACE 单次要 /// DELETE 40 万+ 行再 INSERT ... SELECT 40 万+ 行,实测 DELETE 一步就超时, /// 且超时后连接已断、连 ROLLBACK 都会抛 "Connection must be Open" 掩盖真正的超时异常。 /// 用 在物化区间内临时放宽、退出时还原。 /// private const int MaterializeCommandTimeoutSeconds = 900; /// 物化区间内临时放宽命令超时,Dispose 时还原原值(异常路径同样还原)。 private sealed class LongCommandTimeoutScope : IDisposable { private readonly ISqlSugarClient _db; private readonly int _original; public LongCommandTimeoutScope(ISqlSugarClient db, int seconds) { _db = db; _original = db.Ado.CommandTimeOut; db.Ado.CommandTimeOut = seconds; } public void Dispose() => _db.Ado.CommandTimeOut = _original; } private LongCommandTimeoutScope WithLongCommandTimeout() => new(_db, MaterializeCommandTimeoutSeconds); private static string FormalSliceWhere(string sourceSystem, string domain) { if (!System.Text.RegularExpressions.Regex.IsMatch(sourceSystem, "^[A-Za-z0-9_]+$")) throw new InvalidOperationException($"非法 source_system:{sourceSystem}"); if (!System.Text.RegularExpressions.Regex.IsMatch(domain, "^[A-Za-z0-9_]+$")) throw new InvalidOperationException($"非法 domain:{domain}"); return $"source_system='{sourceSystem}' AND domain='{domain}'"; } /// /// 枚举该 domain 下**拥有合法库存范围**的租户:即在 LocationMaster 里配了非 Supp 库位的启用租户。 /// 没有库位范围的租户(如默认租户)不会被物化,标准层里不会出现它的快照。 /// 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) }); } /// /// 进出存流水 stg→std **按业务租户物化**:一次贴源、逐目标租户按各自库位范围投影。 /// /// 修复前本方法只有一个 tenantId 参数,把「源归属租户」当成了「业务租户」: /// ado_source_domain_tenant_map 把 DOPDEMORQ_SQLSERVER/8010 登记在 797 名下, /// 于是全部流水标准层都写 797,UAT(838257186181189) 名下恒为 0 —— 而正式贴源层里 /// 落在 UAT 18 个合法库位上的流水实测有 188,419 行。余额腿(InsertInventoryStdAsync) /// 早已是「逐租户 + TenantScopeJoin」,本方法此前漏了这一步,本次对齐。 /// /// /// 贴源层归属租户(决定读哪批 stg),非业务归属。 /// 业务租户(决定 std.tenant_id 与库位投影范围)。 /// true=已由 MdpStdFullReplace 删除正式切片,直接 INSERT;false=增量 UPSERT。 private async Task MaterializeInvTransStdAsync( long sourceTenantId, long targetTenantId, string? batchId, DateTime asOf, DateTime historyFrom, string sourceSystem) { // 物化 SQL 要 LEFT JOIN 库位角色;未跑到 1.0.564 的环境也不能因缺表而整批失败 await EnsureLocationRoleTableAsync(); await MdpSchemaAligner.EnsureWrittenByColumnAsync(_db, "mdp_std_inv_trans"); // batchId 为空:转换该源租户下全部已贴源 InvTransHist(中断恢复 / 补物化用) var batchPred = string.IsNullOrWhiteSpace(batchId) ? "1=1" : "s.sync_batch_id=@BatchId"; var syncBatchExpr = string.IsNullOrWhiteSpace(batchId) ? "s.sync_batch_id" : "@BatchId"; // 业务归属 = 目标租户,不再从 stg 反推 var sTenant = "@TargetTenantId"; var sql = $""" INSERT INTO mdp_std_inv_trans (tenant_id, source_system, written_by, domain, src_rec_id, trans_type, src_trans_type_raw, biz_doc_type, approved_flag, void_flag, summary_flag, approved_time, item_num, lot_serial, location, dimension1, dimension2, refs, site, qty_change, begin_balance, end_balance, eff_date, trans_time, ord_nbr, work_ord, ref_task_no, doc_qty, shipper_num, ship_type, reason, remark, create_user, history_from, as_of, sync_batch_id) SELECT {sTenant}, @SourceSystem, 'DB_SYNC', IFNULL({MdpJsonSql.Str("s", "Domain")}, ''), s.source_row_id, {NeutralTransTypeCodes.TransTypeCase( MdpJsonSql.Str("s", "TransType"), LocationRoleExpr, $"IFNULL({MdpJsonSql.Dec("s", "QtyChange", 18, 5)}, 0)")}, {MdpJsonSql.Str("s", "TransType")}, {NeutralTransTypeCodes.BizDocCase(MdpJsonSql.Str("s", "TransType"), "'OTHER'")}, 1, 0, 0, {MdpJsonSql.DateTimeSec("s", "CreateTime")}, {MdpJsonSql.Str("s", "ItemNum")}, {MdpJsonSql.Str("s", "LotSerial")}, {MdpJsonSql.Str("s", "Loc")}, IFNULL({MdpJsonSql.Str("s", "Dimension1")}, ''), IFNULL({MdpJsonSql.Str("s", "Dimension2")}, ''), {MdpJsonSql.Str("s", "Refs")}, {MdpJsonSql.Str("s", "Site")}, IFNULL({MdpJsonSql.Dec("s", "QtyChange", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "BeginBalance", 18, 5)}, 0), IFNULL({MdpJsonSql.Dec("s", "BeginBalance", 18, 5)}, 0) + IFNULL({MdpJsonSql.Dec("s", "QtyChange", 18, 5)}, 0), {MdpJsonSql.DateTimeSec("s", "EffDate")}, {MdpJsonSql.DateTimeSec("s", "CreateTime")}, {MdpJsonSql.Str("s", "OrdNbr")}, {MdpJsonSql.Str("s", "WorkOrd")}, {MdpJsonSql.Str("s", "WorkOrd")}, NULLIF({MdpJsonSql.Dec("s", "QtyRequired", 18, 5)}, 0), {MdpJsonSql.Str("s", "ShipperNum")}, {MdpJsonSql.Str("s", "ShipType")}, {MdpJsonSql.Str("s", "Reason")}, {MdpJsonSql.Str("s", "Remark")}, {MdpJsonSql.Str("s", "CreateUser")}, @HistoryFrom, @AsOf, {syncBatchExpr} FROM mdp_stg_inv_trans s {TransTenantScopeJoin} {LocationRoleJoin} WHERE s.tenant_id=@SourceTenantId AND s.source_system=@SourceSystem AND s.source_table='InvTransHist' AND {batchPred} AND {ProjectionGateSql($"IFNULL({MdpJsonSql.Str("s", "Domain")}, '')")} ON DUPLICATE KEY UPDATE trans_type=VALUES(trans_type), src_trans_type_raw=VALUES(src_trans_type_raw), biz_doc_type=VALUES(biz_doc_type), approved_flag=VALUES(approved_flag), approved_time=VALUES(approved_time), item_num=VALUES(item_num), lot_serial=VALUES(lot_serial), location=VALUES(location), qty_change=VALUES(qty_change), begin_balance=VALUES(begin_balance), end_balance=VALUES(end_balance), eff_date=VALUES(eff_date), trans_time=VALUES(trans_time), ord_nbr=VALUES(ord_nbr), work_ord=VALUES(work_ord), ref_task_no=VALUES(ref_task_no), doc_qty=VALUES(doc_qty), shipper_num=VALUES(shipper_num), ship_type=VALUES(ship_type), reason=VALUES(reason), remark=VALUES(remark), create_user=VALUES(create_user), as_of=VALUES(as_of), sync_batch_id=VALUES(sync_batch_id), update_time=CURRENT_TIMESTAMP """; var ps = new List { new("@SourceTenantId", sourceTenantId), new("@TargetTenantId", targetTenantId), new("@SourceSystem", sourceSystem), new("@AsOf", asOf), new("@HistoryFrom", historyFrom) }; if (!string.IsNullOrWhiteSpace(batchId)) ps.Add(new SugarParameter("@BatchId", batchId)); var affected = await _db.Ado.ExecuteCommandAsync(sql, ps); await RegisterUnmappedAsync(targetTenantId, sourceSystem, batchId); return affected; } /// /// 库位角色配置表兜底建表。权威定义与初值生成在 1.0.564.sql; /// 这里只保证「表存在」,不写初值——角色缺失时阶段码留空并进隔离,不猜。 /// private async Task EnsureLocationRoleTableAsync() => await MdpSchemaAligner.ExecuteAsync(_db, """ CREATE TABLE IF NOT EXISTS mdp_location_role ( id BIGINT AUTO_INCREMENT PRIMARY KEY, tenant_id BIGINT NOT NULL DEFAULT 0, domain VARCHAR(50) NOT NULL DEFAULT '', location VARCHAR(100) NOT NULL, location_role VARCHAR(24) NOT NULL DEFAULT 'UNKNOWN', role_source VARCHAR(24) NOT NULL DEFAULT 'AUTO_DESCR', remark VARCHAR(255) NULL, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NULL ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_location_role (tenant_id, domain, location), KEY idx_location_role_role (tenant_id, location_role) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='库位中立角色(阶段码判定用)' """); /// /// 未能映射到中立阶段码的流水登记隔离表,按原因分级记日志。 /// /// 三类语义不同,禁止一律 Warning: /// /// 非阶段事件(,如冻结/解冻): /// 不登记、不记日志 —— 它们永远不该出现在待办里。 /// 待核验(PENDING_VERIFY):登记,只记 Information。 /// 库位角色未配置(LOCATION_ROLE_UNCONFIGURED)/ 真未知(UNKNOWN_CODE): /// 登记 + Warning,这两类是真有待办动作的。 /// /// /// 隔离表尚未建出时只记 Warning,不打断库存物化。 /// /// /// 业务投影收口。映射表 uk(source_code, domain) 只登记源归属租户,不能再插一行代表 UAT, /// 否则会覆盖 797 并让 SourceDomainTenantResolver 报「不唯一」。 /// 允许投影的目标:映射表上的源归属租户,或该 domain 下自有非 Supp 库位的启用租户。 /// 默认租户 1300000000001 不是源归属、也没有库位主数据,两条都不成立,历史残留不再增长。 /// private static string ProjectionGateSql(string domainExpr) => $""" ( EXISTS ( SELECT 1 FROM ado_source_domain_tenant_map m WHERE m.source_code = @SourceSystem AND m.domain = {domainExpr} AND m.tenant_id = @TargetTenantId AND m.status = 1 ) OR EXISTS ( SELECT 1 FROM LocationMaster lm JOIN SysTenant st ON st.Id = lm.tenant_id AND st.Status = 1 WHERE lm.tenant_id = @TargetTenantId AND IFNULL(lm.Domain,'') = {domainExpr} AND IFNULL(lm.typed,'') <> 'Supp' AND TRIM(lm.location) <> '' ) ) """; private async Task RegisterUnmappedAsync(long tenantId, string sourceSystem, string? batchId) { try { var batchPred = string.IsNullOrWhiteSpace(batchId) ? "1=1" : "t.sync_batch_id=@BatchId"; var ps = new List { new("@TenantId", tenantId), new("@SourceSystem", sourceSystem) }; if (!string.IsNullOrWhiteSpace(batchId)) ps.Add(new SugarParameter("@BatchId", batchId)); var pendingScope = $""" t.trans_type IS NULL AND t.src_trans_type_raw IS NOT NULL AND t.src_trans_type_raw NOT IN ({NeutralTransTypeCodes.NonStageInList}) AND t.src_trans_type_raw NOT IN ({NeutralTransTypeCodes.ExcludedInList}) """; var roleExpr = $"IFNULL(lr.location_role, '{NeutralTransTypeCodes.UnknownRole}')"; var reasonCase = NeutralTransTypeCodes.UnmappedReasonCase("t.src_trans_type_raw", roleExpr); await _db.Ado.ExecuteCommandAsync( $""" INSERT INTO mdp_std_inv_trans_unmapped (tenant_id, source_system, src_rec_id, src_trans_type_raw, unmapped_reason, location, row_count, first_seen, last_seen) SELECT t.tenant_id, t.source_system, t.src_rec_id, t.src_trans_type_raw, {reasonCase}, IFNULL(t.location,''), 1, NOW(), NOW() FROM mdp_std_inv_trans t LEFT JOIN mdp_location_role lr ON lr.tenant_id = t.tenant_id AND lr.domain = IFNULL(t.domain,'') AND lr.location = IFNULL(t.location,'') WHERE t.tenant_id=@TenantId AND t.source_system=@SourceSystem AND {pendingScope} AND {batchPred} AND NOT (t.tenant_id = 1300000000001) ON DUPLICATE KEY UPDATE last_seen=NOW(), row_count=row_count+1, unmapped_reason=VALUES(unmapped_reason), location=VALUES(location) """, ps); var groups = await _db.Ado.SqlQueryAsync( $""" SELECT t.src_trans_type_raw AS Code, {reasonCase} AS Reason, COUNT(*) AS Cnt FROM mdp_std_inv_trans t LEFT JOIN mdp_location_role lr ON lr.tenant_id = t.tenant_id AND lr.domain = IFNULL(t.domain,'') AND lr.location = IFNULL(t.location,'') WHERE t.tenant_id=@TenantId AND t.source_system=@SourceSystem AND {pendingScope} AND {batchPred} AND NOT (t.tenant_id = 1300000000001) GROUP BY t.src_trans_type_raw, {reasonCase} """, ps); foreach (var g in groups) { if (g.Reason == "PENDING_VERIFY") { _logger.LogInformation( "库存流水有 {Count} 行源事务码 {Code} 语义待核验,已进隔离表暂不计入指标。tenant={TenantId} source={Source}", g.Cnt, g.Code, tenantId, sourceSystem); continue; } var action = g.Reason switch { "LOCATION_ROLE_UNCONFIGURED" => "请在 mdp_location_role 配置该库位角色", "OUT_OF_SCOPE_ROLE" => "该库位角色不在此事务码的阶段规则内,不计入指标", _ => "请确认该码语义后补入 NeutralTransTypeCodes" }; _logger.LogWarning( "库存流水有 {Count} 行源事务码 {Code} 未映射到中立阶段码({Reason}),已进隔离表,指标不计入。{Action}。tenant={TenantId} source={Source}", g.Cnt, g.Code, g.Reason, action, tenantId, sourceSystem); } } catch (Exception ex) { _logger.LogWarning(ex, "未映射流水隔离登记跳过(隔离表未就绪)。tenant={TenantId} source={Source}", tenantId, sourceSystem); } } private sealed class UnmappedTransGroup { public string Code { get; set; } = ""; public string Reason { get; set; } = ""; public int Cnt { get; set; } } // 取/放锁已迁至 InventoryInboundLockGuard: // 原实现在共享 _db(IsAutoCloseConnection=true)上跑 GET_LOCK/RELEASE_LOCK, // 命令执行完连接即回池并被驱动 reset,MySQL 当场释放咨询锁 → 跨实例互斥失效。 private sealed class UpperRow { public string? CursorText { get; set; } public string? TieText { get; set; } } private sealed class InventoryStageSourceRow { public string? SourceSystem { get; set; } } } public sealed class InventorySyncResult { public string BatchId { get; init; } = ""; public long TenantId { get; init; } public string Domain { get; init; } = ""; public bool Bootstrap { get; init; } public int LocationPulled { get; init; } public int LocationWritten { get; init; } public int InventoryStdRows { get; init; } public int TransPulled { get; init; } public int TransWritten { get; init; } public int TransStdRows { get; init; } public DateTime AsOf { get; init; } public DateTime? HistoryFrom { get; init; } public bool Skipped { get; init; } public string? Message { get; init; } }