AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Inbound/MdpInboundReceiveService.cs

using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.RegularExpressions;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Inbound;

/// <summary>入站接收主流水线。步骤顺序按任务书 §6.3 定死，勿换序。</summary>
public sealed class MdpInboundReceiveService : ITransient
{
    private static readonly Regex TableNameRe = new(@"^[A-Za-z0-9_]+$", RegexOptions.Compiled);
    private static readonly Regex IsoOffsetRe = new(@"[Zz]|[+\-]\d{2}:\d{2}", RegexOptions.Compiled);
    private const int MaxRowBytes = 64 * 1024;
    private static readonly TimeSpan InProgressWindow = TimeSpan.FromMinutes(10);

    private readonly ISqlSugarClient _db;
    private readonly MdpInboundAuthService _auth;
    private readonly MdpInboundFieldMapper _mapper;
    private readonly MdpStagingWriter _writer;
    private readonly MdpInboundModuleTrigger _trigger;
    private readonly MdpInboundSnapshotService _snapshots;
    private readonly MdmMirrorUpsertService _mirror;
    private readonly ILogger<MdpInboundReceiveService> _logger;

    public MdpInboundReceiveService(
        ISqlSugarClient db,
        MdpInboundAuthService auth,
        MdpInboundFieldMapper mapper,
        MdpStagingWriter writer,
        MdpInboundModuleTrigger trigger,
        MdpInboundSnapshotService snapshots,
        MdmMirrorUpsertService mirror,
        ILogger<MdpInboundReceiveService> logger)
    {
        _db = db;
        _auth = auth;
        _mapper = mapper;
        _writer = writer;
        _trigger = trigger;
        _snapshots = snapshots;
        _mirror = mirror;
        _logger = logger;
    }

    public async Task<MdpInboundOutcome> ReceiveAsync(MdpInboundReceiveArgs args, CancellationToken ct)
    {
        var sw = System.Diagnostics.Stopwatch.StartNew();
        var entityCode = (args.EntityCode ?? string.Empty).Trim().ToUpperInvariant();
        var hash = ToSha256Hex(args.RawBody);

        MdpEntity entity;
        MdpInboundGrant grant;
        try
        {
            entity = await LoadInboundEntityAsync(entityCode, ct)
                     ?? throw new MdpInboundBatchException(403, "entity not authorized");

            var denied = await _auth.CheckAsync(args.AccessKey, args.TenantId, entityCode, args.ClientIp, ct);
            if (denied != null)
                return Fail(denied.Value.Status, denied.Value.Message);

            grant = await _auth.GetGrantAsync(args.AccessKey, entityCode, ct)
                    ?? throw new MdpInboundBatchException(403, "entity not authorized");

            var version = ContractVersion(args);
            if (version == "v1")
                _logger.LogInformation(
                    "inbound receive v1 entity={Entity} accessKey={AccessKey}",
                    entity.EntityCode, args.AccessKey);
            var minimum = await MinContractVersionAsync(grant.Id, ct);
            if (string.Equals(minimum, "v2", StringComparison.OrdinalIgnoreCase) && version != "v2")
                return Fail(400, "contract version v2 required");
        }
        catch (MdpInboundBatchException ex)
        {
            return Fail(ex.HttpStatus, ex.Message);
        }

        MdpInboundParsedBatch batch;
        try
        {
            batch = ParseBatch(args.RawBody, grant, entity);
        }
        catch (MdpInboundBatchException ex)
        {
            return Fail(ex.HttpStatus, ex.Message);
        }
        catch (JsonException)
        {
            return Fail(400, "invalid json");
        }

        var source = await _db.Queryable<MdpSource>()
            .Where(s => s.SourceCode == grant.SourceCode && s.Status == 1)
            .FirstAsync(ct);
        if (source == null)
            return Fail(500, "inbound source not configured");

        var sourceTable = string.IsNullOrWhiteSpace(entity.SourceTableName)
            ? entity.EntityCode
            : entity.SourceTableName!;

        try
        {
            await _snapshots.EnsureAcceptAsync(
                batch.SnapshotId, args.AccessKey, args.TenantId, entity.EntityCode, batch.Seq, ct);
        }
        catch (MdpInboundBatchException ex)
        {
            return Fail(ex.HttpStatus, ex.Message);
        }

        MdpInboundRequest request;
        try
        {
            request = await BeginRequestAsync(args, entity, hash, batch.SnapshotId, ct);
        }
        catch (MdpInboundBatchException ex)
        {
            return Fail(ex.HttpStatus, ex.Message);
        }

        if (request.Status == "COMMITTED" &&
            string.Equals(request.RequestHash, hash, StringComparison.OrdinalIgnoreCase) &&
            !string.IsNullOrWhiteSpace(request.ReceiptJson))
        {
            return Receipt(request.ReceiptJson, replay: true);
        }

        await TryArchiveEnvelopeAsync(args, request, hash, ct);

        try
        {
            var maps = await _mapper.LoadAsync(args.TenantId, entity.EntityCode, args.AccessKey, ct);
            var accepted = new List<MdpInboundPreparedRow>();
            var rejected = new List<MdpInboundRejectedRow>();
            var staleRejected = 0;
            var equalTsOverride = 0;

            foreach (var el in batch.Rows)
            {
                var prepared = PrepareRow(el, entity, maps, grant, args.TenantId, rejected, out var batchDeny);
                if (batchDeny != null)
                {
                    await MarkFailedAsync(request.Id, batchDeny.Message, sw.ElapsedMilliseconds, ct);
                    return Fail(batchDeny.HttpStatus, batchDeny.Message);
                }
                if (prepared == null)
                    continue;

                if (entity.StaleGuardEnabled == 1)
                {
                    var decision = await GuardStaleAsync(
                        entity, args.TenantId, sourceTable, prepared, grant.SourceCode, ct);
                    if (decision.RejectStale)
                    {
                        staleRejected++;
                        rejected.Add(new MdpInboundRejectedRow { BizKey = prepared.BizKey, Reason = "stale" });
                        continue;
                    }
                    if (decision.EqualTsOverride)
                        equalTsOverride++;
                    prepared = decision.Row ?? prepared;
                }

                accepted.Add(prepared);
            }

            var ctx = new MdpPullContext { TenantId = args.TenantId, BatchId = request.SyncBatchId };
            var written = new List<MdpInboundPreparedRow>();
            var tran = await _db.Ado.UseTranAsync(async () =>
            {
                foreach (var row in accepted)
                {
                    await TransferIfNeededAsync(
                        entity, args.TenantId, sourceTable, row.BizKey, grant.SourceCode, ct);
                    var n = await _writer.UpsertAsync(
                        source, entity, sourceTable, row.Dict, row.RawJson, row.BizKey, ctx);
                    if (n == 0)
                    {
                        rejected.Add(new MdpInboundRejectedRow
                        {
                            BizKey = row.BizKey,
                            Reason = "tenant_unresolved"
                        });
                    }
                    else
                    {
                        written.Add(row);
                    }
                }

                var data = new MdpInboundReceiptData
                {
                    EntityCode = entity.EntityCode,
                    Accepted = written.Count,
                    Rejected = rejected.Count,
                    RejectedRows = rejected,
                    IdempotentReplay = false,
                    SyncBatchId = request.SyncBatchId,
                    RequestId = request.Id,
                    TransformEnqueued = false,
                    StaleRejected = staleRejected,
                    EqualTsOverride = equalTsOverride
                };
                var resultCode = rejected.Count > 0 ? 2 : 0;
                var body = Envelope(resultCode, resultCode == 2 ? "partial" : "ok", data);
                var receiptJson = JsonSerializer.Serialize(body, MdpInboundJson.Options);
                var now = DateTime.Now;
                await _db.Updateable<MdpInboundRequest>()
                    .SetColumns(r => new MdpInboundRequest
                    {
                        Status = "COMMITTED",
                        HttpStatus = 202,
                        ResultCode = resultCode,
                        Accepted = written.Count,
                        Rejected = rejected.Count,
                        StaleRejected = staleRejected,
                        EqualTsOverride = equalTsOverride,
                        ReceiptJson = receiptJson,
                        ErrorMsg = null,
                        ElapsedMs = (int)sw.ElapsedMilliseconds,
                        UpdateTime = now
                    })
                    .Where(r => r.Id == request.Id)
                    .ExecuteCommandAsync(ct);
                if (!string.IsNullOrWhiteSpace(batch.SnapshotId) && batch.Seq is > 0)
                    await _snapshots.AdvanceAsync(batch.SnapshotId, batch.Seq.Value, written.Count, ct);
            });

            if (!tran.IsSuccess)
            {
                var err = tran.ErrorException?.Message ?? "transaction failed";
                await MarkFailedAsync(request.Id, Truncate(err, 1000), sw.ElapsedMilliseconds, ct);
                _logger.LogError(tran.ErrorException, "inbound write transaction failed requestId={Id}", request.Id);
                return Fail(500, "inbound write failed");
            }

            var committed = await _db.Queryable<MdpInboundRequest>()
                .Where(r => r.Id == request.Id)
                .FirstAsync(ct);
            var receiptJson = committed?.ReceiptJson;
            try
            {
                await _mirror.MirrorCommittedAsync(
                    entity.EntityCode, args.TenantId, grant.FactoryId, grant.SourceCode, written, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "inbound mirror after commit failed requestId={Id}", request.Id);
            }

            var enqueued = false;
            try
            {
                enqueued = await _trigger.EnqueueForEntityAsync(
                    entity.EntityCode, args.TenantId, grant.FactoryId, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "inbound trigger after commit failed requestId={Id}", request.Id);
            }

            if (enqueued)
                receiptJson = await PatchTransformEnqueuedAsync(request.Id, receiptJson, ct);
            return Receipt(receiptJson, replay: false);
        }
        catch (Exception ex)
        {
            await MarkFailedAsync(request.Id, Truncate(ex.Message, 1000), sw.ElapsedMilliseconds, ct);
            _logger.LogError(ex, "inbound receive failed requestId={Id}", request.Id);
            return Fail(500, "inbound receive failed");
        }
    }

    private static string ContractVersion(MdpInboundReceiveArgs args)
    {
        if (args.Headers != null
            && args.Headers.TryGetValue(MdpInboundFieldMapper.ContractVersionHeader, out var raw))
            return MdpInboundFieldMapper.NormalizeVersion(raw);
        return "v1";
    }

    /// <summary>授权上的最低契约版本。列尚未迁移时按 v1，不打断推数。</summary>
    private async Task<string> MinContractVersionAsync(long grantId, CancellationToken ct)
    {
        ct.ThrowIfCancellationRequested();
        try
        {
            var value = await _db.Ado.GetStringAsync(
                "SELECT min_contract_version FROM mdp_inbound_grant WHERE id=@id",
                new SugarParameter("@id", grantId));
            return string.IsNullOrWhiteSpace(value) ? "v1" : value.Trim();
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex, "入站最低契约版本列未就绪，按 v1 继续");
            return "v1";
        }
    }

    public async Task<MdpInboundOutcome> SchemaAsync(
        string entityCode, string accessKey, long tenantId, string clientIp, CancellationToken ct,
        string? contractVersion = null)
    {
        var code = (entityCode ?? string.Empty).Trim().ToUpperInvariant();
        var denied = await _auth.CheckAsync(accessKey, tenantId, code, clientIp, ct);
        if (denied != null)
            return Fail(denied.Value.Status, denied.Value.Message);

        var entity = await LoadInboundEntityAsync(code, ct);
        if (entity == null)
            return Fail(403, "entity not authorized");

        var maps = await _mapper.LoadAsync(tenantId, entity.EntityCode, accessKey, ct);
        var version = MdpInboundFieldMapper.NormalizeVersion(contractVersion);
        if (version == "v1")
            _logger.LogInformation("inbound schema v1 entity={Entity} accessKey={AccessKey}", entity.EntityCode, accessKey);
        return Ok(_mapper.BuildSchema(entity, maps, version));
    }

    public async Task<MdpInboundOutcome> ReceiptAsync(
        string syncBatchId, string accessKey, CancellationToken ct)
    {
        var row = await _db.Queryable<MdpInboundRequest>()
            .Where(r => r.SyncBatchId == syncBatchId && r.AccessKey == accessKey)
            .FirstAsync(ct);
        if (row == null)
            return Fail(404, "receipt not found");

        object receipt = null;
        if (!string.IsNullOrWhiteSpace(row.ReceiptJson))
        {
            try
            {
                receipt = JsonSerializer.Deserialize<object>(row.ReceiptJson);
            }
            catch (JsonException)
            {
                receipt = row.ReceiptJson;
            }
        }

        return Ok(new
        {
            status = row.Status,
            accepted = row.Accepted,
            rejected = row.Rejected,
            staleRejected = row.StaleRejected,
            receiptJson = receipt
        });
    }

    /// <summary>NDJSON 分块：每行须为合法接收报文，并带根级 idempotencyKey。</summary>
    public async Task<MdpInboundOutcome> BulkAsync(MdpInboundReceiveArgs args, CancellationToken ct)
    {
        var text = Encoding.UTF8.GetString(args.RawBody ?? []);
        var lines = text.Replace("\r\n", "\n").Split('\n');
        var nonempty = lines.Count(l => !string.IsNullOrWhiteSpace(l));
        if (nonempty == 0)
            return Fail(400, "empty bulk");
        if (nonempty > 100)
            return Fail(413, "bulk line limit exceeded");

        var chunks = new List<object>();
        var lineNo = 0;
        foreach (var rawLine in lines)
        {
            var line = rawLine.Trim();
            if (string.IsNullOrEmpty(line))
                continue;
            lineNo++;
            string idem;
            try
            {
                using var doc = JsonDocument.Parse(line);
                if (doc.RootElement.ValueKind != JsonValueKind.Object
                    || !TryGetPropertyIgnoreCase(doc.RootElement, "idempotencyKey", out var keyEl)
                    || keyEl.ValueKind != JsonValueKind.String
                    || string.IsNullOrWhiteSpace(keyEl.GetString()))
                {
                    chunks.Add(new { line = lineNo, httpStatus = 400, message = "missing line idempotencyKey" });
                    continue;
                }
                idem = keyEl.GetString();
            }
            catch (JsonException)
            {
                chunks.Add(new { line = lineNo, httpStatus = 400, message = "invalid json" });
                continue;
            }

            var outcome = await ReceiveAsync(new MdpInboundReceiveArgs
            {
                EntityCode = args.EntityCode,
                RawBody = Encoding.UTF8.GetBytes(line),
                AccessKey = args.AccessKey,
                TenantId = args.TenantId,
                IdempotencyKey = idem,
                ClientIp = args.ClientIp,
                Path = args.Path,
                Headers = args.Headers
            }, ct);
            chunks.Add(new { line = lineNo, httpStatus = outcome.HttpStatus, body = outcome.Body });
        }

        return new MdpInboundOutcome
        {
            HttpStatus = 202,
            Body = Envelope(0, "ok", new { chunks })
        };
    }

    /// <summary>日终对账：扫当日请求 + 按 sync_batch_id 取 stg 键，字典序拼接后 sha256。首期不加新索引。</summary>
    public async Task<MdpInboundOutcome> DigestAsync(
        string entityCode, string accessKey, long tenantId, string clientIp, string dateRaw, CancellationToken ct)
    {
        var code = (entityCode ?? string.Empty).Trim().ToUpperInvariant();
        var denied = await _auth.CheckAsync(accessKey, tenantId, code, clientIp, ct);
        if (denied != null)
            return Fail(denied.Value.Status, denied.Value.Message);

        if (!DateTime.TryParse(dateRaw, out var parsed))
            return Fail(400, "date required (yyyy-MM-dd)");
        var from = parsed.Date;
        var to = from.AddDays(1);

        var entity = await LoadInboundEntityAsync(code, ct);
        if (entity == null)
            return Fail(403, "entity not authorized");

        var reqs = await _db.Queryable<MdpInboundRequest>()
            .Where(r => r.AccessKey == accessKey && r.CreateTime >= from && r.CreateTime < to)
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", code))
            .ToListAsync(ct);

        var accepted = reqs.Where(r => r.Status == "COMMITTED").Sum(r => r.Accepted);
        var rejected = reqs.Sum(r => r.Rejected);
        var batchIds = reqs
            .Where(r => r.Status == "COMMITTED" && !string.IsNullOrWhiteSpace(r.SyncBatchId))
            .Select(r => r.SyncBatchId)
            .Distinct()
            .ToList();

        var keys = new List<string>();
        if (batchIds.Count > 0
            && !string.IsNullOrWhiteSpace(entity.TargetTableName)
            && TableNameRe.IsMatch(entity.TargetTableName))
        {
            keys = await _db.Ado.SqlQueryAsync<string>(
                $"""
                SELECT source_biz_key FROM `{entity.TargetTableName}`
                 WHERE tenant_id=@t AND sync_batch_id IN ({string.Join(",", batchIds.Select((_, i) => "@b" + i))})
                """,
                new[] { new SugarParameter("@t", tenantId) }
                    .Concat(batchIds.Select((b, i) => new SugarParameter("@b" + i, b)))
                    .ToArray());
        }

        var sorted = keys
            .Where(k => !string.IsNullOrWhiteSpace(k))
            .Distinct(StringComparer.Ordinal)
            .OrderBy(k => k, StringComparer.Ordinal)
            .ToList();
        var digest = ToSha256Hex(Encoding.UTF8.GetBytes(string.Join("\n", sorted)));

        return Ok(new
        {
            entityCode = code,
            date = from.ToString("yyyy-MM-dd"),
            requestCount = reqs.Count,
            accepted,
            rejected,
            committedBatches = batchIds.Count,
            bizKeyCount = sorted.Count,
            bizKeyDigest = digest,
            sort = "ordinal-asc join \\n then sha256-hex"
        });
    }

    private async Task<MdpEntity> LoadInboundEntityAsync(string entityCode, CancellationToken ct)
    {
        return await _db.Queryable<MdpEntity>()
            .Where(e => e.InboundEnabled == 1)
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", entityCode))
            .FirstAsync(ct);
    }

    private static MdpInboundParsedBatch ParseBatch(byte[] rawBody, MdpInboundGrant grant, MdpEntity entity)
    {
        using var doc = JsonDocument.Parse(Encoding.UTF8.GetString(rawBody));
        var root = doc.RootElement;
        JsonElement listEl;
        string snapshotId = null;
        int? parsedSeq = null;

        if (root.ValueKind == JsonValueKind.Array)
        {
            listEl = root;
        }
        else if (root.ValueKind == JsonValueKind.Object
                 && TryGetPropertyIgnoreCase(root, "data", out var data)
                 && data.ValueKind == JsonValueKind.Object
                 && TryGetPropertyIgnoreCase(data, "list", out var list)
                 && list.ValueKind == JsonValueKind.Array)
        {
            listEl = list;
            if (TryGetPropertyIgnoreCase(data, "snapshotId", out var snap) && snap.ValueKind == JsonValueKind.String)
                snapshotId = snap.GetString();
            if (TryGetPropertyIgnoreCase(data, "seq", out var seqEl)
                && seqEl.ValueKind == JsonValueKind.Number
                && seqEl.TryGetInt32(out var seqVal))
                parsedSeq = seqVal;
        }
        else
        {
            throw new MdpInboundBatchException(400, "invalid payload: expect data.list or root array");
        }

        var rows = new List<JsonElement>();
        foreach (var item in listEl.EnumerateArray())
        {
            if (item.ValueKind != JsonValueKind.Object)
                throw new MdpInboundBatchException(400, "invalid payload: row must be object");
            AssertNoCaseDuplicateKeys(item);
            var raw = item.GetRawText();
            if (Encoding.UTF8.GetByteCount(raw) > MaxRowBytes)
                throw new MdpInboundBatchException(413, "row too large");
            rows.Add(item.Clone());
        }

        var limit = Math.Min(
            grant.BatchRowLimit > 0 ? grant.BatchRowLimit : 500,
            entity.InboundMaxRows > 0 ? entity.InboundMaxRows : 500);
        if (rows.Count > limit)
            throw new MdpInboundBatchException(413, "row limit exceeded");

        return new MdpInboundParsedBatch { Rows = rows, SnapshotId = snapshotId, Seq = parsedSeq };
    }

    private static void AssertNoCaseDuplicateKeys(JsonElement obj)
    {
        var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
        foreach (var prop in obj.EnumerateObject())
        {
            if (!seen.Add(prop.Name))
                throw new MdpInboundBatchException(400, "duplicate key");
        }
    }

    private async Task<MdpInboundRequest> BeginRequestAsync(
        MdpInboundReceiveArgs args, MdpEntity entity, string hash, string snapshotId, CancellationToken ct)
    {
        var now = DateTime.Now;
        var draft = new MdpInboundRequest
        {
            TenantId = args.TenantId,
            AccessKey = args.AccessKey,
            EntityCode = entity.EntityCode,
            IdempotencyKey = args.IdempotencyKey,
            RequestHash = hash,
            SyncBatchId = "PENDING",
            SnapshotId = snapshotId,
            Status = "RECEIVED",
            ClientIp = args.ClientIp,
            CreateTime = now,
            UpdateTime = now
        };

        try
        {
            var id = await _db.Insertable(draft).ExecuteReturnBigIdentityAsync();
            draft.Id = id;
            draft.SyncBatchId = $"INB_{now:yyyyMMdd}_{id:D6}";
            await _db.Updateable<MdpInboundRequest>()
                .SetColumns(r => new MdpInboundRequest { SyncBatchId = draft.SyncBatchId, UpdateTime = now })
                .Where(r => r.Id == id)
                .ExecuteCommandAsync(ct);
            return draft;
        }
        catch (Exception ex) when (IsDuplicateKey(ex))
        {
            var existing = await _db.Queryable<MdpInboundRequest>()
                .Where(r => r.TenantId == args.TenantId
                            && r.AccessKey == args.AccessKey
                            && r.IdempotencyKey == args.IdempotencyKey)
                .Where("UPPER(entity_code) = @code", new SugarParameter("@code", entity.EntityCode.ToUpperInvariant()))
                .FirstAsync(ct)
                        ?? throw new MdpInboundBatchException(409, "in-progress");

            if (string.Equals(existing.Status, "COMMITTED", StringComparison.OrdinalIgnoreCase))
            {
                if (!string.Equals(existing.RequestHash, hash, StringComparison.OrdinalIgnoreCase))
                    throw new MdpInboundBatchException(409, "idempotency key conflict");
                return existing;
            }

            if (string.Equals(existing.Status, "RECEIVED", StringComparison.OrdinalIgnoreCase)
                && existing.UpdateTime > now - InProgressWindow)
                throw new MdpInboundBatchException(409, "in-progress");

            var n = await _db.Ado.ExecuteCommandAsync(
                """
                UPDATE mdp_inbound_request
                   SET status='RECEIVED', request_hash=@h, snapshot_id=@snap,
                       error_msg=NULL, update_time=@now
                 WHERE id=@id AND status=@old
                """,
                new SugarParameter("@h", hash),
                new SugarParameter("@snap", (object)snapshotId ?? DBNull.Value),
                new SugarParameter("@now", now),
                new SugarParameter("@id", existing.Id),
                new SugarParameter("@old", existing.Status));
            if (n != 1)
                throw new MdpInboundBatchException(409, "in-progress");

            existing.Status = "RECEIVED";
            existing.RequestHash = hash;
            existing.SnapshotId = snapshotId;
            existing.UpdateTime = now;
            if (string.IsNullOrWhiteSpace(existing.SyncBatchId) || existing.SyncBatchId == "PENDING")
            {
                existing.SyncBatchId = $"INB_{now:yyyyMMdd}_{existing.Id:D6}";
                await _db.Updateable<MdpInboundRequest>()
                    .SetColumns(r => new MdpInboundRequest { SyncBatchId = existing.SyncBatchId })
                    .Where(r => r.Id == existing.Id)
                    .ExecuteCommandAsync(ct);
            }
            return existing;
        }
    }

    private async Task TryArchiveEnvelopeAsync(
        MdpInboundReceiveArgs args, MdpInboundRequest request, string hash, CancellationToken ct)
    {
        try
        {
            var exists = await _db.Queryable<MdpInboundEnvelope>()
                .Where(e => e.RequestId == request.Id)
                .AnyAsync(ct);
            if (exists)
                return;

            var headers = args.Headers
                .Where(kv => !IsSensitiveHeader(kv.Key))
                .ToDictionary(kv => kv.Key, kv => kv.Value, StringComparer.OrdinalIgnoreCase);
            await _db.Insertable(new MdpInboundEnvelope
            {
                RequestId = request.Id,
                TenantId = args.TenantId,
                AccessKey = args.AccessKey,
                EntityCode = request.EntityCode,
                Path = args.Path,
                HeadersJson = JsonSerializer.Serialize(headers, MdpInboundJson.Options),
                Body = Encoding.UTF8.GetString(args.RawBody),
                BodySha256 = hash,
                CreateTime = DateTime.Now
            }).ExecuteCommandAsync(ct);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "inbound envelope archive failed requestId={Id}", request.Id);
        }
    }

    private MdpInboundPreparedRow PrepareRow(
        JsonElement el,
        MdpEntity entity,
        IReadOnlyList<MdpFieldMap> maps,
        MdpInboundGrant grant,
        long boundTenantId,
        List<MdpInboundRejectedRow> rejected,
        out MdpInboundBatchException batchDeny)
    {
        batchDeny = null;
        var incoming = MdpInboundFieldMapper.ElementToDict(el);
        var mapped = _mapper.MapRow(incoming, maps);
        var missing = _mapper.MissingRequired(mapped, maps);
        var previewKey = TryGetString(mapped, "bizKey");

        if (missing.Count > 0)
        {
            rejected.Add(new MdpInboundRejectedRow
            {
                BizKey = previewKey,
                Reason = "missing required field " + string.Join(",", missing)
            });
            return null;
        }

        var sourceUpdatedAt = TryGetString(mapped, "sourceUpdatedAt");
        if (!HasIsoOffset(sourceUpdatedAt))
        {
            rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "sourceUpdatedAt required with timezone offset" });
            return null;
        }

        var sourceVersion = TryGetString(mapped, "sourceVersion");
        if (!string.IsNullOrWhiteSpace(sourceVersion) && !long.TryParse(sourceVersion, out _))
        {
            rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "sourceVersion must be long" });
            return null;
        }

        if (TryGetLong(mapped, "tenant_id", out var rowTenant) || TryGetLong(mapped, "TenantId", out rowTenant))
        {
            if (rowTenant != boundTenantId)
            {
                batchDeny = new MdpInboundBatchException(403, "tenant mismatch");
                return null;
            }
        }
        mapped["tenant_id"] = boundTenantId;

        if (grant.FactoryId is > 0)
        {
            if (!TryGetLong(mapped, "factory_id", out var rowFactory) && !TryGetLong(mapped, "FactoryId", out rowFactory))
            {
                rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "factory mismatch" });
                return null;
            }
            if (rowFactory != grant.FactoryId.Value)
            {
                rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "factory mismatch" });
                return null;
            }
        }

        if (string.Equals(entity.EntityCode, "S5_INVENTORY_TXN", StringComparison.OrdinalIgnoreCase))
        {
            var transType = FirstMapped(mapped, "TransType", "trans_type");
            if (!string.IsNullOrWhiteSpace(transType)
                && !NeutralTransTypeCodes.AllStageCodes.Contains(transType, StringComparer.Ordinal))
            {
                batchDeny = new MdpInboundBatchException(400,
                    "TransType must be one of " + string.Join(",", NeutralTransTypeCodes.AllStageCodes));
                return null;
            }

            var bizDoc = FirstMapped(mapped, "BizDocType", "biz_doc_type");
            if (!string.IsNullOrWhiteSpace(bizDoc)
                && !NeutralTransTypeCodes.KnownBizDocTypes.Contains(bizDoc, StringComparer.Ordinal))
            {
                batchDeny = new MdpInboundBatchException(400,
                    "BizDocType must be one of " + string.Join(",", NeutralTransTypeCodes.KnownBizDocTypes));
                return null;
            }
        }

        var bizKey = MdpStagingWriter.BuildBizKey(entity.BizKeyExpr, mapped);
        if (string.IsNullOrWhiteSpace(bizKey))
        {
            rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "missing biz key fields" });
            return null;
        }
        if (!string.IsNullOrWhiteSpace(previewKey) && !string.Equals(previewKey, bizKey, StringComparison.Ordinal))
        {
            rejected.Add(new MdpInboundRejectedRow { BizKey = previewKey, Reason = "bizKey mismatch" });
            return null;
        }

        var op = TryGetString(mapped, "op");
        if (string.IsNullOrWhiteSpace(op))
            op = "upsert";
        if (string.Equals(op, "delete", StringComparison.OrdinalIgnoreCase))
        {
            if (!DeleteSupported(entity))
            {
                rejected.Add(new MdpInboundRejectedRow { BizKey = bizKey, Reason = "op=delete not supported" });
                return null;
            }
            mapped["is_deleted"] = 1;
        }

        var rawJson = JsonSerializer.Serialize(mapped, MdpInboundJson.Options);
        return new MdpInboundPreparedRow
        {
            Dict = mapped,
            RawJson = rawJson,
            BizKey = bizKey,
            SourceUpdatedAt = sourceUpdatedAt,
            SourceVersion = sourceVersion
        };
    }

    private async Task<(bool RejectStale, bool EqualTsOverride, MdpInboundPreparedRow Row)> GuardStaleAsync(
        MdpEntity entity,
        long tenantId,
        string sourceTable,
        MdpInboundPreparedRow incoming,
        string inboundSource,
        CancellationToken ct)
    {
        var existing = await LoadGuardRowAsync(entity.TargetTableName, tenantId, sourceTable, incoming.BizKey, inboundSource, ct);
        if (existing == null)
            return (false, false, incoming);

        var oldHasVer = long.TryParse(existing.SourceVersion, out var oldVer);
        var newHasVer = long.TryParse(incoming.SourceVersion, out var newVer);
        if (oldHasVer && newHasVer)
        {
            if (newVer < oldVer)
                return (true, false, incoming);
            if (newVer > oldVer)
                return (false, false, incoming);
        }
        else
        {
            var oldHasTs = DateTimeOffset.TryParse(existing.SourceUpdatedAt, out var oldTs);
            if (!oldHasTs)
                return (false, false, incoming);
            if (!DateTimeOffset.TryParse(incoming.SourceUpdatedAt, out var newTs))
                return (true, false, incoming);
            if (newTs < oldTs)
                return (true, false, incoming);
            if (newTs > oldTs)
                return (false, false, incoming);
        }

        var same = string.Equals(incoming.RawJson, existing.RawData, StringComparison.Ordinal);
        return (false, !same, incoming);
    }

    private async Task<MdpInboundStgGuardRow> LoadGuardRowAsync(
        string table, long tenantId, string sourceTable, string bizKey, string inboundSource, CancellationToken ct)
    {
        if (string.IsNullOrWhiteSpace(table) || !TableNameRe.IsMatch(table))
            throw new InvalidOperationException("illegal target_table_name");

        var rows = await _db.Ado.SqlQueryAsync<MdpInboundStgGuardRow>(
            $"""
            SELECT id AS Id,
                   source_system AS SourceSystem,
                   source_row_id AS SourceRowId,
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sourceUpdatedAt')) AS SourceUpdatedAt,
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sourceVersion')) AS SourceVersion,
                   raw_data AS RawData
              FROM `{table}`
             WHERE tenant_id=@t AND source_table=@st AND source_biz_key=@bk
             ORDER BY CASE WHEN source_system=@sys THEN 0 ELSE 1 END
             LIMIT 1
            """,
            new SugarParameter("@t", tenantId),
            new SugarParameter("@st", sourceTable),
            new SugarParameter("@bk", bizKey),
            new SugarParameter("@sys", inboundSource));
        return rows?.FirstOrDefault();
    }

    private async Task TransferIfNeededAsync(
        MdpEntity entity, long tenantId, string sourceTable, string bizKey, string inboundSource, CancellationToken ct)
    {
        var existing = await LoadGuardRowAsync(entity.TargetTableName, tenantId, sourceTable, bizKey, inboundSource, ct);
        if (existing == null)
            return;
        if (string.Equals(existing.SourceSystem, inboundSource, StringComparison.OrdinalIgnoreCase))
            return;
        if (string.IsNullOrWhiteSpace(entity.TargetTableName) || !TableNameRe.IsMatch(entity.TargetTableName))
            throw new InvalidOperationException("illegal target_table_name");

        await _db.Ado.ExecuteCommandAsync(
            $"""
            UPDATE `{entity.TargetTableName}`
               SET source_system=@sys, source_row_id=@bk
             WHERE id=@id
            """,
            new SugarParameter("@sys", inboundSource),
            new SugarParameter("@bk", bizKey),
            new SugarParameter("@id", existing.Id));
    }

    private async Task MarkFailedAsync(long requestId, string error, long elapsedMs, CancellationToken ct)
    {
        try
        {
            await _db.Updateable<MdpInboundRequest>()
                .SetColumns(r => new MdpInboundRequest
                {
                    Status = "FAILED",
                    HttpStatus = 500,
                    ResultCode = 500,
                    ErrorMsg = error,
                    ElapsedMs = (int)elapsedMs,
                    UpdateTime = DateTime.Now
                })
                .Where(r => r.Id == requestId)
                .ExecuteCommandAsync(ct);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "inbound mark FAILED failed requestId={Id}", requestId);
        }
    }

    private static bool DeleteSupported(MdpEntity entity) =>
        (entity.Remark ?? string.Empty).Contains("inbound_delete=supported", StringComparison.OrdinalIgnoreCase);

    private static bool HasIsoOffset(string value) =>
        !string.IsNullOrWhiteSpace(value)
        && IsoOffsetRe.IsMatch(value)
        && DateTimeOffset.TryParse(value, out _);

    private static bool TryGetPropertyIgnoreCase(JsonElement obj, string name, out JsonElement value)
    {
        foreach (var prop in obj.EnumerateObject())
        {
            if (string.Equals(prop.Name, name, StringComparison.OrdinalIgnoreCase))
            {
                value = prop.Value;
                return true;
            }
        }
        value = default;
        return false;
    }

    private static string TryGetString(IDictionary<string, object?> row, string key)
    {
        foreach (var kv in row)
        {
            if (!string.Equals(kv.Key, key, StringComparison.OrdinalIgnoreCase))
                continue;
            return kv.Value?.ToString();
        }
        return null;
    }

    private static string FirstMapped(IDictionary<string, object?> row, params string[] keys)
    {
        foreach (var key in keys)
        {
            var value = TryGetString(row, key);
            if (!string.IsNullOrWhiteSpace(value))
                return value;
        }
        return null;
    }

    private static bool TryGetLong(IDictionary<string, object?> row, string key, out long value)
    {
        value = 0;
        var s = TryGetString(row, key);
        return !string.IsNullOrWhiteSpace(s) && long.TryParse(s, out value);
    }

    private static bool IsSensitiveHeader(string name) =>
        string.Equals(name, "X-Signature", StringComparison.OrdinalIgnoreCase)
        || string.Equals(name, "Authorization", StringComparison.OrdinalIgnoreCase)
        || string.Equals(name, "Cookie", StringComparison.OrdinalIgnoreCase);

    private static bool IsDuplicateKey(Exception ex)
    {
        for (var e = ex; e != null; e = e.InnerException)
        {
            var m = e.Message ?? string.Empty;
            if (m.Contains("Duplicate", StringComparison.OrdinalIgnoreCase)
                || m.Contains("1062")
                || m.Contains("uk_inbound_idem", StringComparison.OrdinalIgnoreCase))
                return true;
        }
        return false;
    }

    private static string ToSha256Hex(byte[] raw) =>
        Convert.ToHexString(SHA256.HashData(raw ?? [])).ToLowerInvariant();

    private static string Truncate(string s, int max) =>
        string.IsNullOrEmpty(s) ? s : (s.Length <= max ? s : s[..max]);

    private static object Envelope(int code, string message, object data) =>
        new { code, message, data };

    private static MdpInboundOutcome Fail(int status, string message) =>
        new() { HttpStatus = status, Body = Envelope(status, message, null) };

    private static MdpInboundOutcome Ok(object data) =>
        new() { HttpStatus = 200, Body = Envelope(0, "ok", data) };

    private async Task<string> PatchTransformEnqueuedAsync(long requestId, string receiptJson, CancellationToken ct)
    {
        if (string.IsNullOrWhiteSpace(receiptJson))
            return receiptJson;
        try
        {
            var node = JsonNode.Parse(receiptJson);
            if (node is JsonObject obj && obj["data"] is JsonObject data)
                data["transformEnqueued"] = true;
            var updated = node.ToJsonString(MdpInboundJson.Options);
            await _db.Updateable<MdpInboundRequest>()
                .SetColumns(r => new MdpInboundRequest
                {
                    ReceiptJson = updated,
                    UpdateTime = DateTime.Now
                })
                .Where(r => r.Id == requestId)
                .ExecuteCommandAsync(ct);
            return updated;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "inbound patch transformEnqueued failed requestId={Id}", requestId);
            return receiptJson;
        }
    }

    private static MdpInboundOutcome Receipt(string receiptJson, bool replay)
    {
        if (string.IsNullOrWhiteSpace(receiptJson))
            return new MdpInboundOutcome { HttpStatus = 202, Body = Envelope(0, "ok", null) };
        try
        {
            var node = JsonNode.Parse(receiptJson);
            if (replay && node is JsonObject obj && obj["data"] is JsonObject data)
                data["idempotentReplay"] = true;
            return new MdpInboundOutcome { HttpStatus = 202, Body = node };
        }
        catch (JsonException)
        {
            return new MdpInboundOutcome { HttpStatus = 202, Body = Envelope(0, "ok", receiptJson) };
        }
    }
}