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; /// 主数据 stg 批 COMMITTED 后回灌 S0 镜像。失败不阻断接收。 public sealed class MdmMirrorUpsertService : ITransient { private readonly ISqlSugarClient _db; private readonly AdoS0ReferenceChecker _refs; private readonly ILogger _logger; public MdmMirrorUpsertService( ISqlSugarClient db, AdoS0ReferenceChecker refs, ILogger logger) { _db = db; _refs = refs; _logger = logger; } public async Task MirrorCommittedAsync( string entityCode, long tenantId, long? factoryId, string sourceCode, IReadOnlyList 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() .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() .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() .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() .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() .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 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); }