AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/InventoryMdpSyncService.cs

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;

/// <summary>
/// S5 库存冷链：165 LocationDetail / InvTransHist → stg → std。
/// 余额事实源仅 LocationDetail；InvMaster 不入标准层。
/// </summary>
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<AidopInventoryOptions> opt,
        ILoggerFactory loggerFactory)
    {
        _db = db;
        _pullDispatcher = pullDispatcher;
        _scopeFactory = scopeFactory;
        _domainTenant = domainTenant;
        _opt = opt.Value;
        _logger = loggerFactory.CreateLogger(nameof(InventoryMdpSyncService));
    }

    public Task<InventorySyncResult> RunBootstrapAsync(CancellationToken cancellationToken = default)
        => RunAsync(bootstrap: true, reconcile: false, cancellationToken);

    public Task<InventorySyncResult> RunIncrementalAsync(CancellationToken cancellationToken = default)
        => RunAsync(bootstrap: false, reconcile: false, cancellationToken);

    public Task<InventorySyncResult> RunReconcileFullAsync(CancellationToken cancellationToken = default)
        => RunAsync(bootstrap: false, reconcile: true, cancellationToken);

    /// <summary>
    /// 仅 stg→std：不拉源。用于首刷被中断后补落标准层（覆盖该租户全部已贴源批次）。
    /// </summary>
    public async Task<InventorySyncResult> 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();
        }
    }

    /// <summary>
    /// 仅 stg→std（库存余额）：**不访问源库**，从一个已完整落地的贴源批次重新物化标准层。
    /// <para>
    /// 用途：同步链路修复后，复用既有完整 stg 批次让各租户按新的库位范围口径重新物化，
    /// 避免为此重新全量拉取源库。
    /// </para>
    /// <para>
    /// 语义为 <b>UPSERT，不做 FULL REPLACE</b>：只按业务键写入/更新本批次覆盖到的行，
    /// 不删除任何既有标准层数据 —— 单个增量批次不代表全量，replace 会造成数据丢失。
    /// 因此本入口<b>不负责</b>清理历史脏快照，那属于全量校准（reconcile）的职责。
    /// </para>
    /// </summary>
    /// <param name="stgBatchId">贴源批次号（sync_batch_id），必须是已完整落地的批次。</param>
    public async Task<InventorySyncResult> 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;

        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<SugarParameter>
            {
                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<InventorySyncResult> 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);

        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);
            }

            // —— 标准层按租户物化：一次贴源，逐租户按各自库位范围投影 ——
            // 贴源层是「源+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"
            };
        }
        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<UpperRow>(
            $"""
             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);
    }

    /// <summary>
    /// stg → std 物化：**按目标租户的合法库位范围投影**。
    /// <para>
    /// 贴源层（stg）是「源 + domain」维度的全量落地区，归属 <paramref name="sourceTenantId"/>；
    /// 标准层（std）是「租户」维度的可见快照，因此这里必须内联 LocationMaster 做投影：
    /// 只有落在目标租户自己 LocationMaster（同 Domain 且 Typed &lt;&gt; 'Supp'）内的库位才写入。
    /// </para>
    /// <para>
    /// 写入不变量：∀ 写入行 → tenant_id = targetTenantId
    /// ∧ location ∈ AllowedLocations(targetTenantId) ∧ domain = 该租户 LocationMaster 的 Domain。
    /// 租户白名单为空 → JOIN 命中 0 行 → 写 0 条（fail closed，绝不退回整个 Domain）。
    /// </para>
    /// <para>
    /// tenant_id 直接取 <paramref name="targetTenantId"/> 而非 MdpJsonSql.TenantFromStg：
    /// 贴源行的 tenant 是「源落地区归属」，不是业务归属，不能顺着传下来。
    /// </para>
    /// </summary>
    private async Task<int> 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));
    }

    /// <summary>
    /// 租户库位范围内联投影：贴源行只有落在目标租户自己的合法库位（同 Domain、Typed &lt;&gt; 'Supp'）
    /// 才允许进入标准层。这是标准层写入侧的租户安全边界。
    /// </summary>
    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) <> ''
         """;

    /// <summary>
    /// 枚举该 domain 下**拥有合法库存范围**的租户：即在 LocationMaster 里配了非 Supp 库位的启用租户。
    /// 没有库位范围的租户（如默认租户）不会被物化，标准层里不会出现它的快照。
    /// </summary>
    private async Task<List<long>> ListInventoryScopedTenantsAsync(string domain, CancellationToken ct)
    {
        return await _db.Ado.SqlQueryAsync<long>(
            """
            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<SugarParameter> { new("@Domain", domain) });
    }

    private async Task<int> 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<SugarParameter>
        {
            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<bool> TryAcquireLockAsync()
    {
        var ok = await _db.Ado.GetIntAsync(
            "SELECT GET_LOCK(@k, 0)",
            new List<SugarParameter> { new("@k", LockKey) });
        return ok == 1;
    }

    private Task ReleaseLockAsync()
        => _db.Ado.ExecuteCommandAsync(
            "SELECT RELEASE_LOCK(@k)",
            new List<SugarParameter> { 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; }
}