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 = 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:不拉源。用于首刷被中断后补落标准层(覆盖该租户全部已贴源批次)。
///
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 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"
};
}
///
/// 仅 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})"
};
}
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);
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"
};
}
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')
/// 才允许进入标准层。这是标准层写入侧的租户安全边界。
///
private static readonly string TenantScopeJoin =
$"""
INNER JOIN LocationMaster lm
ON lm.tenant_id = @TargetTenantId
AND lm.Domain = IFNULL({MdpJsonSql.Str("s", "Domain")}, '')
AND lm.location = IFNULL({MdpJsonSql.Str("s", "Location")}, '')
AND IFNULL(lm.typed, '') <> 'Supp'
AND TRIM(lm.location) <> ''
""";
///
/// 枚举该 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) });
}
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);
}
// 取/放锁已迁至 InventoryInboundLockGuard:
// 原实现在共享 _db(IsAutoCloseConnection=true)上跑 GET_LOCK/RELEASE_LOCK,
// 命令执行完连接即回池并被驱动 reset,MySQL 当场释放咨询锁 → 跨实例互斥失效。
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; }
}