using Admin.NET.Plugin.AiDOP.DataPlatform; 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 = "aidop:s5:inventory-inbound"; 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:不拉源。用于首刷被中断后补落标准层(覆盖该租户全部已贴源批次)。 /// 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}"; var locked = await TryAcquireLockAsync(); if (!locked) { return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, AsOf = asOf, HistoryFrom = historyFrom, Skipped = true, Message = "lock busy" }; } try { var rows = await UpsertInvTransStdAsync(tenantId, batchId: null, asOf, historyFrom, sourceCode); _logger.LogInformation("[InventoryMdpSync] transform-std done tenant={Tenant} rows={Rows}", tenantId, rows); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, TransStdRows = rows, AsOf = asOf, HistoryFrom = historyFrom, Message = "OK transform-std" }; } finally { await ReleaseLockAsync(); } } 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(); 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); var locked = await TryAcquireLockAsync(); if (!locked) { _logger.LogWarning("[InventoryMdpSync] 跨实例锁占用,本轮跳过 batch={Batch}", batchId); return new InventorySyncResult { BatchId = batchId, TenantId = tenantId, Domain = domain, Bootstrap = bootstrap, AsOf = asOf, HistoryFrom = historyFrom, Skipped = true, Message = "lock busy" }; } try { 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); } int inventoryStdRows; if (bootstrap || reconcile) { inventoryStdRows = await MdpStdFullReplace.ReplaceAsync( _db, "mdp_std_inventory", tenantId, "source_system='DOPDEMORQ_SQLSERVER'", () => InsertInventoryStdAsync(tenantId, $"{batchId}_LOC", asOf, sourceCode, replaceMode: true), cancellationToken); } else { inventoryStdRows = await InsertInventoryStdAsync(tenantId, $"{batchId}_LOC", asOf, sourceCode, replaceMode: false); } 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); var transStdRows = await UpsertInvTransStdAsync(tenantId, $"{batchId}_TRN", asOf, historyFrom, sourceCode); _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" }; } finally { await ReleaseLockAsync(); } } 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); } private async Task InsertInventoryStdAsync( long tenantId, string batchId, DateTime asOf, string sourceSystem, bool replaceMode) { // replaceMode:MdpStdFullReplace 已 DELETE,直接 INSERT; // 增量:UPSERT 本批变化。 var sTenant = MdpJsonSql.TenantFromStg("s", "@TenantId"); 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 WHERE s.tenant_id=@TenantId AND s.source_system=@SourceSystem AND s.source_table='LocationDetail' AND s.sync_batch_id=@BatchId AND {MdpJsonSql.TenantGuard(sTenant)} """ : $""" 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 WHERE s.tenant_id=@TenantId AND s.source_system=@SourceSystem AND s.source_table='LocationDetail' AND s.sync_batch_id=@BatchId AND {MdpJsonSql.TenantGuard(sTenant)} 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("@TenantId", tenantId), new SugarParameter("@SourceSystem", sourceSystem), new SugarParameter("@BatchId", batchId), new SugarParameter("@AsOf", asOf)); } private async Task UpsertInvTransStdAsync( long tenantId, string? batchId, DateTime asOf, DateTime historyFrom, string sourceSystem) { // batchId 为空:转换该租户下全部已贴源 InvTransHist(中断恢复用) var batchPred = string.IsNullOrWhiteSpace(batchId) ? "1=1" : "s.sync_batch_id=@BatchId"; var syncBatchExpr = string.IsNullOrWhiteSpace(batchId) ? "s.sync_batch_id" : "@BatchId"; var sTenant = MdpJsonSql.TenantFromStg("s", "@TenantId"); var sql = $""" INSERT INTO mdp_std_inv_trans (tenant_id, source_system, domain, src_rec_id, trans_type, item_num, lot_serial, location, dimension1, dimension2, refs, site, qty_change, begin_balance, end_balance, eff_date, trans_time, ord_nbr, work_ord, shipper_num, ship_type, reason, remark, create_user, history_from, as_of, sync_batch_id) SELECT {sTenant}, @SourceSystem, IFNULL({MdpJsonSql.Str("s", "Domain")}, ''), s.source_row_id, {MdpJsonSql.Str("s", "TransType")}, {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", "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 WHERE s.tenant_id=@TenantId AND s.source_system=@SourceSystem AND s.source_table='InvTransHist' AND {batchPred} AND {MdpJsonSql.TenantGuard(sTenant)} ON DUPLICATE KEY UPDATE trans_type=VALUES(trans_type), 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), 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("@TenantId", tenantId), new("@SourceSystem", sourceSystem), new("@AsOf", asOf), new("@HistoryFrom", historyFrom) }; if (!string.IsNullOrWhiteSpace(batchId)) ps.Add(new SugarParameter("@BatchId", batchId)); return await _db.Ado.ExecuteCommandAsync(sql, ps); } private async Task TryAcquireLockAsync() { var ok = await _db.Ado.GetIntAsync( "SELECT GET_LOCK(@k, 0)", new List { new("@k", LockKey) }); return ok == 1; } private Task ReleaseLockAsync() => _db.Ado.ExecuteCommandAsync( "SELECT RELEASE_LOCK(@k)", new List { new("@k", LockKey) }); private sealed class UpperRow { public string? CursorText { get; set; } public string? TieText { 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; } }