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

using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Admin.NET.Plugin.AiDOP.Entity.S0.Sales;
using Admin.NET.Plugin.AiDOP.Entity.S0.Supply;
using Admin.NET.Plugin.AiDOP.Entity.S0.Warehouse;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Microsoft.Extensions.Logging;

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

/// <summary>主数据 stg 批 COMMITTED 后回灌 S0 镜像。失败不阻断接收。</summary>
public sealed class MdmMirrorUpsertService : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly AdoS0ReferenceChecker _refs;
    private readonly ILogger<MdmMirrorUpsertService> _logger;

    public MdmMirrorUpsertService(
        ISqlSugarClient db,
        AdoS0ReferenceChecker refs,
        ILogger<MdmMirrorUpsertService> logger)
    {
        _db = db;
        _refs = refs;
        _logger = logger;
    }

    public async Task MirrorCommittedAsync(
        string entityCode,
        long tenantId,
        long? factoryId,
        string sourceCode,
        IReadOnlyList<MdpInboundPreparedRow> rows,
        CancellationToken ct)
    {
        if (rows == null || rows.Count == 0)
            return;
        var factory = factoryId is > 0 ? factoryId.Value : 1L;
        foreach (var row in rows)
        {
            try
            {
                var code = (entityCode ?? string.Empty).Trim().ToUpperInvariant();
                if (code is "MDM_ITEM")
                    await MirrorItemAsync(tenantId, factory, sourceCode, row, ct);
                else if (code is "MDM_CUSTOMER")
                    await MirrorCustomerAsync(tenantId, factory, sourceCode, row, ct);
                else if (code is "MDM_SUPPLIER")
                    await MirrorSupplierAsync(tenantId, factory, sourceCode, row, ct);
                else if (code is "MDM_LOCATION")
                    await MirrorLocationAsync(tenantId, factory, sourceCode, row, ct);
                else if (code is "MDM_EMPLOYEE_HEADCOUNT")
                    await MirrorEmployeeAsync(tenantId, factory, sourceCode, row, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "inbound mirror failed entity={Entity} biz={Biz}", entityCode, row.BizKey);
            }
        }
    }

    private async Task MirrorItemAsync(
        long tenantId, long factoryId, string sourceCode, MdpInboundPreparedRow row, CancellationToken ct)
    {
        var itemNum = Str(row.Dict, "ItemNum") ?? Str(row.Dict, "number");
        if (string.IsNullOrWhiteSpace(itemNum))
            return;

        var incomingAt = ParseTime(row.SourceUpdatedAt);
        var existing = await _db.Queryable<AdoS0ItemMaster>()
            .Where(x => x.TenantId == tenantId && x.ItemNum == itemNum)
            .FirstAsync(ct);

        if (existing != null && string.IsNullOrWhiteSpace(existing.SourceSystem))
        {
            await InsertConflictAsync(tenantId, "MDM_ITEM", row.BizKey, sourceCode, row.RawJson, "ItemMaster", ct);
            return;
        }
        if (existing?.SourceUpdatedAt is { } oldAt && incomingAt is { } neu && oldAt > neu)
        {
            _logger.LogInformation("inbound mirror skipped_stale item={Item}", itemNum);
            return;
        }

        if (!string.IsNullOrWhiteSpace(Str(row.Dict, "Location")))
            await _refs.LocationExistsAsync(tenantId, Str(row.Dict, "Location"));

        var now = DateTime.Now;
        if (existing == null)
        {
            await _db.Insertable(new AdoS0ItemMaster
            {
                TenantId = tenantId,
                FactoryRefId = factoryId,
                DomainCode = Str(row.Dict, "Domain"),
                ItemNum = itemNum,
                Descr = Str(row.Dict, "Descr") ?? Str(row.Dict, "name") ?? itemNum,
                Drawing = Str(row.Dict, "Drawing") ?? Str(row.Dict, "model"),
                UM = Str(row.Dict, "UM") ?? Str(row.Dict, "unit"),
                ItemType = Str(row.Dict, "ItemType"),
                Status = Str(row.Dict, "Status") ?? "normal",
                IsActive = !IsInactive(Str(row.Dict, "Status") ?? Str(row.Dict, "is_active")),
                SourceSystem = sourceCode,
                SourceUpdatedAt = incomingAt,
                CreateTime = now,
                UpdateTime = now,
                UpdateUser = "API_INBOUND"
            }).ExecuteCommandAsync(ct);
            return;
        }

        existing.Descr = Str(row.Dict, "Descr") ?? Str(row.Dict, "name") ?? existing.Descr;
        existing.Drawing = Str(row.Dict, "Drawing") ?? Str(row.Dict, "model") ?? existing.Drawing;
        existing.UM = Str(row.Dict, "UM") ?? Str(row.Dict, "unit") ?? existing.UM;
        existing.ItemType = Str(row.Dict, "ItemType") ?? existing.ItemType;
        existing.DomainCode = Str(row.Dict, "Domain") ?? existing.DomainCode;
        if (!string.IsNullOrWhiteSpace(Str(row.Dict, "Status")))
        {
            existing.Status = Str(row.Dict, "Status");
            existing.IsActive = !IsInactive(existing.Status);
        }
        existing.SourceSystem = sourceCode;
        existing.SourceUpdatedAt = incomingAt ?? existing.SourceUpdatedAt;
        existing.UpdateTime = now;
        existing.UpdateUser = "API_INBOUND";
        await _db.Updateable(existing)
            .IgnoreColumns(x => new { x.Location, x.DefaultShelf, x.SafetyStk, x.LotSerialControl, x.AllocateSingleLot })
            .ExecuteCommandAsync(ct);
    }

    private async Task MirrorCustomerAsync(
        long tenantId, long factoryId, string sourceCode, MdpInboundPreparedRow row, CancellationToken ct)
    {
        var cust = Str(row.Dict, "Cust") ?? Str(row.Dict, "custom_no");
        if (string.IsNullOrWhiteSpace(cust))
            return;
        var incomingAt = ParseTime(row.SourceUpdatedAt);
        var existing = await _db.Queryable<AdoS0CustMaster>()
            .Where(x => x.TenantId == tenantId && x.Cust == cust)
            .FirstAsync(ct);
        if (existing != null && string.IsNullOrWhiteSpace(existing.SourceSystem))
        {
            await InsertConflictAsync(tenantId, "MDM_CUSTOMER", row.BizKey, sourceCode, row.RawJson, "CustMaster", ct);
            return;
        }
        if (existing?.SourceUpdatedAt is { } oldAt && incomingAt is { } neu && oldAt > neu)
            return;

        var now = DateTime.Now;
        if (existing == null)
        {
            await _db.Insertable(new AdoS0CustMaster
            {
                TenantId = tenantId,
                FactoryRefId = factoryId,
                Cust = cust,
                SortName = Str(row.Dict, "SortName") ?? Str(row.Dict, "custom_name"),
                SourceSystem = sourceCode,
                SourceUpdatedAt = incomingAt,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
            return;
        }
        existing.SortName = Str(row.Dict, "SortName") ?? existing.SortName;
        existing.SourceSystem = sourceCode;
        existing.SourceUpdatedAt = incomingAt ?? existing.SourceUpdatedAt;
        existing.UpdateTime = now;
        await _db.Updateable(existing).ExecuteCommandAsync(ct);
    }

    private async Task MirrorSupplierAsync(
        long tenantId, long factoryId, string sourceCode, MdpInboundPreparedRow row, CancellationToken ct)
    {
        var supp = Str(row.Dict, "Supp") ?? Str(row.Dict, "supplier_number");
        if (string.IsNullOrWhiteSpace(supp))
            return;
        var incomingAt = ParseTime(row.SourceUpdatedAt);
        var existing = await _db.Queryable<AdoS0SuppMaster>()
            .Where(x => x.TenantId == tenantId && x.Supp == supp)
            .FirstAsync(ct);
        if (existing != null && string.IsNullOrWhiteSpace(existing.SourceSystem))
        {
            await InsertConflictAsync(tenantId, "MDM_SUPPLIER", row.BizKey, sourceCode, row.RawJson, "SuppMaster", ct);
            return;
        }
        if (existing?.SourceUpdatedAt is { } oldAt && incomingAt is { } neu && oldAt > neu)
            return;
        var now = DateTime.Now;
        if (existing == null)
        {
            await _db.Insertable(new AdoS0SuppMaster
            {
                TenantId = tenantId,
                FactoryRefId = factoryId,
                Supp = supp,
                SortName = Str(row.Dict, "SortName"),
                SourceSystem = sourceCode,
                SourceUpdatedAt = incomingAt,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
            return;
        }
        existing.SortName = Str(row.Dict, "SortName") ?? existing.SortName;
        existing.SourceSystem = sourceCode;
        existing.SourceUpdatedAt = incomingAt ?? existing.SourceUpdatedAt;
        existing.UpdateTime = now;
        await _db.Updateable(existing).ExecuteCommandAsync(ct);
    }

    private async Task MirrorLocationAsync(
        long tenantId, long factoryId, string sourceCode, MdpInboundPreparedRow row, CancellationToken ct)
    {
        var loc = Str(row.Dict, "location") ?? Str(row.Dict, "Location");
        if (string.IsNullOrWhiteSpace(loc))
            return;
        var incomingAt = ParseTime(row.SourceUpdatedAt);
        var existing = await _db.Queryable<AdoS0LocationMaster>()
            .Where(x => x.TenantId == tenantId && x.Location == loc)
            .FirstAsync(ct);
        if (existing != null && string.IsNullOrWhiteSpace(existing.SourceSystem))
        {
            await InsertConflictAsync(tenantId, "MDM_LOCATION", row.BizKey, sourceCode, row.RawJson, "LocationMaster", ct);
            return;
        }
        if (existing?.SourceUpdatedAt is { } oldAt && incomingAt is { } neu && oldAt > neu)
            return;
        var now = DateTime.Now;
        if (existing == null)
        {
            await _db.Insertable(new AdoS0LocationMaster
            {
                TenantId = tenantId,
                FactoryRefId = factoryId,
                DomainCode = Str(row.Dict, "Domain") ?? "",
                Location = loc,
                Descr = Str(row.Dict, "descr") ?? Str(row.Dict, "Descr"),
                SourceSystem = sourceCode,
                SourceUpdatedAt = incomingAt,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
            return;
        }
        existing.Descr = Str(row.Dict, "descr") ?? existing.Descr;
        existing.SourceSystem = sourceCode;
        existing.SourceUpdatedAt = incomingAt ?? existing.SourceUpdatedAt;
        existing.UpdateTime = now;
        await _db.Updateable(existing).ExecuteCommandAsync(ct);
    }

    private async Task MirrorEmployeeAsync(
        long tenantId, long factoryId, string sourceCode, MdpInboundPreparedRow row, CancellationToken ct)
    {
        var emp = Str(row.Dict, "Employee") ?? Str(row.Dict, "employee");
        if (string.IsNullOrWhiteSpace(emp))
            return;
        if (!string.IsNullOrWhiteSpace(Str(row.Dict, "Department")))
            await _refs.DepartmentExistsAsync(tenantId, Str(row.Dict, "Department"));

        var incomingAt = ParseTime(row.SourceUpdatedAt);
        var existing = await _db.Queryable<AdoS0EmployeeMaster>()
            .Where(x => x.TenantId == tenantId && x.Employee == emp)
            .FirstAsync(ct);
        if (existing != null && string.IsNullOrWhiteSpace(existing.SourceSystem))
        {
            await InsertConflictAsync(tenantId, "MDM_EMPLOYEE_HEADCOUNT", row.BizKey, sourceCode, row.RawJson, "EmployeeMaster", ct);
            return;
        }
        if (existing?.SourceUpdatedAt is { } oldAt && incomingAt is { } neu && oldAt > neu)
            return;
        var now = DateTime.Now;
        if (existing == null)
        {
            await _db.Insertable(new AdoS0EmployeeMaster
            {
                TenantId = tenantId,
                FactoryRefId = factoryId,
                DomainCode = Str(row.Dict, "Domain") ?? "",
                Employee = emp,
                Name = Str(row.Dict, "Name"),
                SourceSystem = sourceCode,
                SourceUpdatedAt = incomingAt,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
            return;
        }
        existing.Name = Str(row.Dict, "Name") ?? existing.Name;
        existing.SourceSystem = sourceCode;
        existing.SourceUpdatedAt = incomingAt ?? existing.SourceUpdatedAt;
        existing.UpdateTime = now;
        await _db.Updateable(existing).ExecuteCommandAsync(ct);
    }

    private async Task InsertConflictAsync(
        long tenantId, string entityCode, string bizKey, string sourceCode, string raw, string table, CancellationToken ct)
    {
        var now = DateTime.Now;
        await _db.Insertable(new MdpInboundConflict
        {
            TenantId = tenantId,
            EntityCode = entityCode,
            BizKey = bizKey ?? string.Empty,
            SourceSystem = sourceCode,
            IncomingRaw = raw ?? string.Empty,
            MirrorTable = table,
            ConflictType = "MANUAL_ROW_EXISTS",
            Status = "PENDING",
            CreateTime = now,
            UpdateTime = now
        }).ExecuteCommandAsync(ct);
    }

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

    private static DateTime? ParseTime(string raw) =>
        DateTimeOffset.TryParse(raw, out var dto) ? dto.LocalDateTime : null;

    private static bool IsInactive(string status) =>
        string.Equals(status, "INACTIVE", StringComparison.OrdinalIgnoreCase)
        || status == "0"
        || string.Equals(status, "false", StringComparison.OrdinalIgnoreCase);
}