791 lines
33 KiB
C#
791 lines
33 KiB
C#
|
using Microsoft.Extensions.Configuration;
|
|||
|
|
using System.IO;
|
|||
|
|
using System.Net.Http;
|
|||
|
|
using System.Text;
|
|||
|
|
using System.Text.Json;
|
|||
|
|
using System.Text.Json.Serialization;
|
|||
|
|
using System.Web;
|
|||
|
|
using Prism.Events;
|
|||
|
|
using YY.Admin.Core;
|
|||
|
|
using YY.Admin.Core.Entity;
|
|||
|
|
using YY.Admin.Core.Events;
|
|||
|
|
using YY.Admin.Core.Services;
|
|||
|
|
|
|||
|
|
namespace YY.Admin.Services.Service.Customer;
|
|||
|
|
|
|||
|
|
public class CustomerService : ICustomerService, ISingletonDependency
|
|||
|
|
{
|
|||
|
|
private readonly IHttpClientFactory _httpClientFactory;
|
|||
|
|
private readonly IConfiguration _configuration;
|
|||
|
|
private readonly INetworkMonitor _networkMonitor;
|
|||
|
|
private readonly IEventAggregator _eventAggregator;
|
|||
|
|
private readonly ILoggerService _logger;
|
|||
|
|
private readonly SemaphoreSlim _syncLock = new(1, 1);
|
|||
|
|
private readonly object _cacheLock = new();
|
|||
|
|
private readonly string _pendingOpsFilePath;
|
|||
|
|
private readonly string _cacheFilePath;
|
|||
|
|
private List<CustomerPendingOperation> _pendingOps = new();
|
|||
|
|
private List<MesXslCustomer> _localCache = new();
|
|||
|
|
|
|||
|
|
private static readonly JsonSerializerOptions _jsonOpts = new()
|
|||
|
|
{
|
|||
|
|
PropertyNameCaseInsensitive = true,
|
|||
|
|
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
|||
|
|
Converters = { new NullableDateTimeJsonConverter() }
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
public CustomerService(
|
|||
|
|
IHttpClientFactory httpClientFactory,
|
|||
|
|
IConfiguration configuration,
|
|||
|
|
INetworkMonitor networkMonitor,
|
|||
|
|
IEventAggregator eventAggregator,
|
|||
|
|
ILoggerService logger)
|
|||
|
|
{
|
|||
|
|
_httpClientFactory = httpClientFactory;
|
|||
|
|
_configuration = configuration;
|
|||
|
|
_networkMonitor = networkMonitor;
|
|||
|
|
_eventAggregator = eventAggregator;
|
|||
|
|
_logger = logger;
|
|||
|
|
|
|||
|
|
var appDataDir = Path.Combine(
|
|||
|
|
Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData),
|
|||
|
|
"YY.Admin", "sync-cache");
|
|||
|
|
Directory.CreateDirectory(appDataDir);
|
|||
|
|
_pendingOpsFilePath = Path.Combine(appDataDir, "mes-xsl-customer-pending-ops.json");
|
|||
|
|
_cacheFilePath = Path.Combine(appDataDir, "mes-xsl-customer-cache.json");
|
|||
|
|
|
|||
|
|
LoadPendingOpsFromDisk();
|
|||
|
|
LoadCacheFromDisk();
|
|||
|
|
_logger.Information($"[客户同步] 服务初始化，缓存={_localCache.Count}，待上传={_pendingOps.Count}，在线={_networkMonitor.IsOnline}");
|
|||
|
|
|
|||
|
|
_networkMonitor.StatusChanged += OnNetworkStatusChanged;
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
_ = Task.Run(() => SyncAfterReconnectAsync(CancellationToken.None));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private const int MaxPendingRetries = 5;
|
|||
|
|
|
|||
|
|
private string BaseUrl => (_configuration.GetValue<string>("JeecgIntegration:BaseUrl") ?? "http://localhost:8080/jeecg-boot").TrimEnd('/');
|
|||
|
|
private int DefaultTenantId => (int?)_configuration.GetValue<long?>("JeecgIntegration:DefaultTenantId") ?? 1002;
|
|||
|
|
|
|||
|
|
private HttpClient CreateClient() => _httpClientFactory.CreateClient("JeecgApi");
|
|||
|
|
|
|||
|
|
// ── 分页 ──────────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<CustomerPageResult> PageAsync(int pageNo, int pageSize,
|
|||
|
|
string? customerCode = null, string? customerName = null,
|
|||
|
|
string? status = null, string? customerRegion = null,
|
|||
|
|
CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
List<MesXslCustomer>? source = null;
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
source = await FetchRemoteListAsync(ct).ConfigureAwait(false);
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_localCache = source.Select(Clone).ToList();
|
|||
|
|
SaveCacheToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
_logger.Information($"[客户列表] 远端拉取成功 count={source.Count}");
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
source = null;
|
|||
|
|
_logger.Warning($"[客户列表] 远端拉取失败，回退缓存：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
source ??= _localCache.Select(Clone).ToList();
|
|||
|
|
source = ApplyPendingOpsSnapshotUnsafe(source);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var filtered = ApplyFilters(source, customerCode, customerName, status, customerRegion);
|
|||
|
|
var total = filtered.Count;
|
|||
|
|
var records = filtered.Skip(Math.Max(0, (pageNo - 1) * pageSize)).Take(pageSize).ToList();
|
|||
|
|
return new CustomerPageResult(records, total, pageNo, pageSize);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 详情 ──────────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<MesXslCustomer?> GetByIdAsync(string id, CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/queryById?id={Uri.EscapeDataString(id)}&tenantId={DefaultTenantId}";
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.GetAsync(url, ct).ConfigureAwait(false);
|
|||
|
|
if (!resp.IsSuccessStatusCode) return null;
|
|||
|
|
var json = await resp.Content.ReadAsStringAsync(ct).ConfigureAwait(false);
|
|||
|
|
using var doc = JsonDocument.Parse(json);
|
|||
|
|
if (!doc.RootElement.TryGetProperty("result", out var resultEl)) return null;
|
|||
|
|
return resultEl.Deserialize<MesXslCustomer>(_jsonOpts);
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户详情] 远端查询失败 id={id}，回退缓存：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
return _localCache.FirstOrDefault(c => string.Equals(c.Id, id, StringComparison.OrdinalIgnoreCase)) is { } found
|
|||
|
|
? Clone(found) : null;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 新增 ──────────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<bool> AddAsync(MesXslCustomer customer, CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
if (!customer.TenantId.HasValue || customer.TenantId.Value <= 0)
|
|||
|
|
customer.TenantId = DefaultTenantId;
|
|||
|
|
|
|||
|
|
var local = Clone(customer);
|
|||
|
|
if (string.IsNullOrWhiteSpace(local.Id))
|
|||
|
|
local.Id = $"local-{Guid.NewGuid():N}";
|
|||
|
|
if (string.IsNullOrWhiteSpace(local.Status))
|
|||
|
|
local.Status = "0";
|
|||
|
|
SyncIzEnable(local);
|
|||
|
|
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var ok = await RemoteAddAsync(local, ct).ConfigureAwait(false);
|
|||
|
|
if (ok) { UpsertLocalCache(local); return true; }
|
|||
|
|
return false;
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户新增] 远端失败，转离线入队：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
EnqueuePendingOperation(new CustomerPendingOperation
|
|||
|
|
{ OpType = CustomerOperationType.Add, CustomerId = local.Id, Customer = local });
|
|||
|
|
UpsertLocalCache(local);
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 编辑 ──────────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<bool> EditAsync(MesXslCustomer customer, CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
if (!customer.TenantId.HasValue || customer.TenantId.Value <= 0)
|
|||
|
|
customer.TenantId = DefaultTenantId;
|
|||
|
|
|
|||
|
|
var local = Clone(customer);
|
|||
|
|
SyncIzEnable(local);
|
|||
|
|
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var (ok, _) = await RemoteEditAsync(local, ct).ConfigureAwait(false);
|
|||
|
|
if (ok) { UpsertLocalCache(local); return true; }
|
|||
|
|
return false;
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户编辑] 远端失败，转离线入队：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
EnqueuePendingOperation(new CustomerPendingOperation
|
|||
|
|
{
|
|||
|
|
OpType = CustomerOperationType.Edit,
|
|||
|
|
CustomerId = local.Id,
|
|||
|
|
Customer = local,
|
|||
|
|
AnchorUpdateTime = local.UpdateTime
|
|||
|
|
});
|
|||
|
|
UpsertLocalCache(local);
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 删除 ──────────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<bool> DeleteAsync(string id, CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var ok = await RemoteDeleteAsync(id, ct).ConfigureAwait(false);
|
|||
|
|
if (ok) { RemoveFromLocalCache(id); return true; }
|
|||
|
|
return false;
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户删除] 远端失败，转离线入队：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
DateTime? anchor;
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
anchor = _localCache.FirstOrDefault(c => string.Equals(c.Id, id, StringComparison.OrdinalIgnoreCase))?.UpdateTime;
|
|||
|
|
}
|
|||
|
|
EnqueuePendingOperation(new CustomerPendingOperation
|
|||
|
|
{
|
|||
|
|
OpType = CustomerOperationType.Delete,
|
|||
|
|
CustomerId = id,
|
|||
|
|
AnchorUpdateTime = anchor
|
|||
|
|
});
|
|||
|
|
RemoveFromLocalCache(id);
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 状态切换 ──────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
public async Task<bool> UpdateStatusAsync(string id, string status, CancellationToken ct = default)
|
|||
|
|
{
|
|||
|
|
if (_networkMonitor.IsOnline)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var ok = await RemoteUpdateStatusAsync(id, status, ct).ConfigureAwait(false);
|
|||
|
|
if (ok) { UpdateLocalStatus(id, status); return true; }
|
|||
|
|
return false;
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户状态] 远端失败，转离线入队：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
DateTime? anchor;
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
anchor = _localCache.FirstOrDefault(c => string.Equals(c.Id, id, StringComparison.OrdinalIgnoreCase))?.UpdateTime;
|
|||
|
|
}
|
|||
|
|
EnqueuePendingOperation(new CustomerPendingOperation
|
|||
|
|
{
|
|||
|
|
OpType = CustomerOperationType.UpdateStatus,
|
|||
|
|
CustomerId = id,
|
|||
|
|
Status = status,
|
|||
|
|
AnchorUpdateTime = anchor
|
|||
|
|
});
|
|||
|
|
UpdateLocalStatus(id, status);
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 远端调用 ──────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
private async Task<List<MesXslCustomer>> FetchRemoteListAsync(CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var query = HttpUtility.ParseQueryString(string.Empty);
|
|||
|
|
query["pageNo"] = "1";
|
|||
|
|
query["pageSize"] = "10000";
|
|||
|
|
query["tenantId"] = DefaultTenantId.ToString();
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/list?{query}";
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.GetAsync(url, ct).ConfigureAwait(false);
|
|||
|
|
resp.EnsureSuccessStatusCode();
|
|||
|
|
var json = await resp.Content.ReadAsStringAsync(ct).ConfigureAwait(false);
|
|||
|
|
using var doc = JsonDocument.Parse(json);
|
|||
|
|
return doc.RootElement.GetProperty("result").GetProperty("records")
|
|||
|
|
.Deserialize<List<MesXslCustomer>>(_jsonOpts) ?? new();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<bool> RemoteAddAsync(MesXslCustomer customer, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/add?tenantId={DefaultTenantId}";
|
|||
|
|
var payload = Clone(customer);
|
|||
|
|
if (IsLocalTempId(payload.Id)) payload.Id = null;
|
|||
|
|
return await PostJsonAsync(url, payload, ct).ConfigureAwait(false);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<(bool Ok, bool IsVersionConflict)> RemoteEditAsync(MesXslCustomer customer, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/edit?tenantId={DefaultTenantId}";
|
|||
|
|
return await PostJsonCheckVersionAsync(url, customer, ct).ConfigureAwait(false);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<bool> RemoteDeleteAsync(string id, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/delete?id={Uri.EscapeDataString(id)}&tenantId={DefaultTenantId}";
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.DeleteAsync(url, ct).ConfigureAwait(false);
|
|||
|
|
return resp.IsSuccessStatusCode && await IsSuccessResultAsync(resp, ct).ConfigureAwait(false);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<bool> RemoteUpdateStatusAsync(string id, string status, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/updateStatus?id={Uri.EscapeDataString(id)}&status={Uri.EscapeDataString(status)}&tenantId={DefaultTenantId}";
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.PostAsync(url, null, ct).ConfigureAwait(false);
|
|||
|
|
return resp.IsSuccessStatusCode && await IsSuccessResultAsync(resp, ct).ConfigureAwait(false);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<bool> PostJsonAsync(string url, object body, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var content = new StringContent(JsonSerializer.Serialize(body, _jsonOpts), Encoding.UTF8, "application/json");
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.PostAsync(url, content, ct).ConfigureAwait(false);
|
|||
|
|
return resp.IsSuccessStatusCode && await IsSuccessResultAsync(resp, ct).ConfigureAwait(false);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<(bool Ok, bool IsVersionConflict)> PostJsonCheckVersionAsync(string url, object body, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var content = new StringContent(JsonSerializer.Serialize(body, _jsonOpts), Encoding.UTF8, "application/json");
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.PostAsync(url, content, ct).ConfigureAwait(false);
|
|||
|
|
if (!resp.IsSuccessStatusCode) return (false, false);
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var json = await resp.Content.ReadAsStringAsync(ct).ConfigureAwait(false);
|
|||
|
|
using var doc = JsonDocument.Parse(json);
|
|||
|
|
int code = 200;
|
|||
|
|
if (doc.RootElement.TryGetProperty("code", out var codeEl)) code = codeEl.GetInt32();
|
|||
|
|
if (code == 200) return (true, false);
|
|||
|
|
if (doc.RootElement.TryGetProperty("message", out var msgEl))
|
|||
|
|
{
|
|||
|
|
var msg = msgEl.GetString() ?? "";
|
|||
|
|
if (msg.Contains("已被他人修改")) return (false, true);
|
|||
|
|
}
|
|||
|
|
return (false, false);
|
|||
|
|
}
|
|||
|
|
catch { return (true, false); }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private static async Task<bool> IsSuccessResultAsync(HttpResponseMessage resp, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var json = await resp.Content.ReadAsStringAsync(ct).ConfigureAwait(false);
|
|||
|
|
using var doc = JsonDocument.Parse(json);
|
|||
|
|
if (doc.RootElement.TryGetProperty("code", out var code)) return code.GetInt32() == 200;
|
|||
|
|
if (doc.RootElement.TryGetProperty("success", out var success)) return success.GetBoolean();
|
|||
|
|
return true;
|
|||
|
|
}
|
|||
|
|
catch { return true; }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 断线续连 ──────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
private void OnNetworkStatusChanged(bool isOnline)
|
|||
|
|
{
|
|||
|
|
if (!isOnline) return;
|
|||
|
|
_ = Task.Run(() => SyncAfterReconnectAsync(CancellationToken.None));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task SyncAfterReconnectAsync(CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var pushResult = await PushPendingOnReconnectAsync(ct).ConfigureAwait(false);
|
|||
|
|
if (!_networkMonitor.IsOnline) return;
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var remote = await FetchRemoteListAsync(ct).ConfigureAwait(false);
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_localCache = remote.Select(Clone).ToList();
|
|||
|
|
SaveCacheToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
_eventAggregator.GetEvent<CustomerChangedEvent>()
|
|||
|
|
.Publish(new CustomerChangedPayload { Action = "pull" });
|
|||
|
|
_logger.Information($"[客户重连] 全量回拉成功 count={remote.Count}");
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户重连] 全量回拉失败：{ex.Message}");
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var hasActivity = pushResult.PushedCount > 0
|
|||
|
|
|| pushResult.ConflictCount > 0
|
|||
|
|
|| pushResult.NewRecordsPushed > 0;
|
|||
|
|
if (hasActivity)
|
|||
|
|
{
|
|||
|
|
_eventAggregator.GetEvent<SyncConflictEvent>()
|
|||
|
|
.Publish(new SyncConflictPayload
|
|||
|
|
{
|
|||
|
|
EntityName = "客户",
|
|||
|
|
PushedCount = pushResult.PushedCount,
|
|||
|
|
ConflictCount = pushResult.ConflictCount,
|
|||
|
|
NewRecordsPushed = pushResult.NewRecordsPushed
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<PushPendingResult> PushPendingOnReconnectAsync(CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
if (!await _syncLock.WaitAsync(0, ct).ConfigureAwait(false))
|
|||
|
|
return new PushPendingResult(0, 0, 0);
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
List<CustomerPendingOperation> snapshot;
|
|||
|
|
lock (_cacheLock) { snapshot = _pendingOps.OrderBy(x => x.CreatedAt).ToList(); }
|
|||
|
|
_logger.Information($"[客户回放] pending={snapshot.Count}");
|
|||
|
|
|
|||
|
|
int pushed = 0, conflicts = 0, newPushed = 0;
|
|||
|
|
|
|||
|
|
foreach (var op in snapshot)
|
|||
|
|
{
|
|||
|
|
if (!_networkMonitor.IsOnline) break;
|
|||
|
|
|
|||
|
|
// 如果该 op 已在上一条冲突处理中被清理，则跳过
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
if (!_pendingOps.Any(x => x.Id == op.Id)) continue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var result = await ExecutePendingOpWithConflictAsync(op, ct).ConfigureAwait(false);
|
|||
|
|
if (!result.Ok)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
op.RetryCount++;
|
|||
|
|
if (op.RetryCount >= MaxPendingRetries)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户回放] op={op.OpType} 超过最大重试次数({MaxPendingRetries})，放弃 customerId={op.CustomerId}");
|
|||
|
|
_pendingOps.RemoveAll(x => x.Id == op.Id);
|
|||
|
|
SavePendingOpsToDiskUnsafe();
|
|||
|
|
continue;
|
|||
|
|
}
|
|||
|
|
SavePendingOpsToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (result.IsConflict)
|
|||
|
|
{
|
|||
|
|
conflicts++;
|
|||
|
|
// 冲突时：以服务器版本为准，直接丢弃同一条记录的所有 pending
|
|||
|
|
if (!string.IsNullOrWhiteSpace(result.EntityId))
|
|||
|
|
RemovePendingOpsByCustomerId(result.EntityId);
|
|||
|
|
}
|
|||
|
|
else if (op.OpType == CustomerOperationType.Add)
|
|||
|
|
{
|
|||
|
|
newPushed++;
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_pendingOps.RemoveAll(x => x.Id == op.Id);
|
|||
|
|
SavePendingOpsToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
else
|
|||
|
|
{
|
|||
|
|
pushed++;
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_pendingOps.RemoveAll(x => x.Id == op.Id);
|
|||
|
|
SavePendingOpsToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return new PushPendingResult(pushed, conflicts, newPushed);
|
|||
|
|
}
|
|||
|
|
finally { _syncLock.Release(); }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private sealed record PendingReplayResult(bool Ok, bool IsConflict, string? EntityId);
|
|||
|
|
|
|||
|
|
private async Task<PendingReplayResult> ExecutePendingOpWithConflictAsync(CustomerPendingOperation op, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
return op.OpType switch
|
|||
|
|
{
|
|||
|
|
CustomerOperationType.Add => await ExecuteAddAsync(op, ct).ConfigureAwait(false),
|
|||
|
|
CustomerOperationType.Edit => await ExecuteEditWithConflictAsync(op, ct).ConfigureAwait(false),
|
|||
|
|
CustomerOperationType.Delete => await ExecuteDeleteWithConflictAsync(op, ct).ConfigureAwait(false),
|
|||
|
|
CustomerOperationType.UpdateStatus => await ExecuteUpdateStatusWithConflictAsync(op, ct).ConfigureAwait(false),
|
|||
|
|
_ => new PendingReplayResult(true, false, null)
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
catch (Exception ex)
|
|||
|
|
{
|
|||
|
|
_logger.Warning($"[客户回放] 执行失败 op={op.OpType}：{ex.Message}");
|
|||
|
|
return new PendingReplayResult(false, false, null);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<PendingReplayResult> ExecuteAddAsync(CustomerPendingOperation op, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var ok = op.Customer != null && await RemoteAddAsync(op.Customer, ct).ConfigureAwait(false);
|
|||
|
|
return ok
|
|||
|
|
? new PendingReplayResult(true, false, op.CustomerId)
|
|||
|
|
: new PendingReplayResult(false, false, null);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<PendingReplayResult> ExecuteEditWithConflictAsync(CustomerPendingOperation op, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
var id = op.Customer?.Id;
|
|||
|
|
if (string.IsNullOrWhiteSpace(id))
|
|||
|
|
return new PendingReplayResult(false, false, null);
|
|||
|
|
|
|||
|
|
// 冲突检测：服务器 UpdateTime != 本地 AnchorUpdateTime
|
|||
|
|
var remote = await FetchRemoteSingleAsync(id!, ct).ConfigureAwait(false);
|
|||
|
|
if (remote != null && op.AnchorUpdateTime != null && remote.UpdateTime != op.AnchorUpdateTime)
|
|||
|
|
{
|
|||
|
|
UpsertLocalCache(remote);
|
|||
|
|
return new PendingReplayResult(true, true, id);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var (ok, isVersionConflict) = await RemoteEditAsync(op.Customer!, ct).ConfigureAwait(false);
|
|||
|
|
if (isVersionConflict)
|
|||
|
|
{
|
|||
|
|
var fresh = await FetchRemoteSingleAsync(id!, ct).ConfigureAwait(false);
|
|||
|
|
if (fresh != null) UpsertLocalCache(fresh);
|
|||
|
|
return new PendingReplayResult(true, true, id);
|
|||
|
|
}
|
|||
|
|
return ok
|
|||
|
|
? new PendingReplayResult(true, false, id)
|
|||
|
|
: new PendingReplayResult(false, false, null);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<PendingReplayResult> ExecuteDeleteWithConflictAsync(CustomerPendingOperation op, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
if (string.IsNullOrWhiteSpace(op.CustomerId))
|
|||
|
|
return new PendingReplayResult(false, false, null);
|
|||
|
|
|
|||
|
|
var id = op.CustomerId!;
|
|||
|
|
var remote = await FetchRemoteSingleAsync(id, ct).ConfigureAwait(false);
|
|||
|
|
if (remote == null)
|
|||
|
|
{
|
|||
|
|
// 后端已不存在：删除无需操作，也视为“成功清理 pending”
|
|||
|
|
return new PendingReplayResult(true, false, id);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (op.AnchorUpdateTime != null && remote.UpdateTime != op.AnchorUpdateTime)
|
|||
|
|
{
|
|||
|
|
// 冲突：服务器版本获胜，恢复到服务器版本
|
|||
|
|
UpsertLocalCache(remote);
|
|||
|
|
return new PendingReplayResult(true, true, id);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var ok = await RemoteDeleteAsync(id, ct).ConfigureAwait(false);
|
|||
|
|
return ok
|
|||
|
|
? new PendingReplayResult(true, false, id)
|
|||
|
|
: new PendingReplayResult(false, false, null);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<PendingReplayResult> ExecuteUpdateStatusWithConflictAsync(CustomerPendingOperation op, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
if (string.IsNullOrWhiteSpace(op.CustomerId) || string.IsNullOrWhiteSpace(op.Status))
|
|||
|
|
return new PendingReplayResult(false, false, null);
|
|||
|
|
|
|||
|
|
var id = op.CustomerId!;
|
|||
|
|
var remote = await FetchRemoteSingleAsync(id, ct).ConfigureAwait(false);
|
|||
|
|
if (remote != null && op.AnchorUpdateTime != null && remote.UpdateTime != op.AnchorUpdateTime)
|
|||
|
|
{
|
|||
|
|
UpsertLocalCache(remote);
|
|||
|
|
return new PendingReplayResult(true, true, id);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var ok = await RemoteUpdateStatusAsync(id, op.Status!, ct).ConfigureAwait(false);
|
|||
|
|
return ok
|
|||
|
|
? new PendingReplayResult(true, false, id)
|
|||
|
|
: new PendingReplayResult(false, false, null);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void RemovePendingOpsByCustomerId(string customerId)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_pendingOps.RemoveAll(x =>
|
|||
|
|
(x.CustomerId != null && string.Equals(x.CustomerId, customerId, StringComparison.OrdinalIgnoreCase)) ||
|
|||
|
|
(x.Customer?.Id != null && string.Equals(x.Customer.Id, customerId, StringComparison.OrdinalIgnoreCase)));
|
|||
|
|
SavePendingOpsToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private async Task<MesXslCustomer?> FetchRemoteSingleAsync(string id, CancellationToken ct)
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
var url = $"{BaseUrl}/xslmes/mesXslCustomer/anon/queryById?id={Uri.EscapeDataString(id)}&tenantId={DefaultTenantId}";
|
|||
|
|
using var client = CreateClient();
|
|||
|
|
var resp = await client.GetAsync(url, ct).ConfigureAwait(false);
|
|||
|
|
if (!resp.IsSuccessStatusCode) return null;
|
|||
|
|
|
|||
|
|
var json = await resp.Content.ReadAsStringAsync(ct).ConfigureAwait(false);
|
|||
|
|
using var doc = JsonDocument.Parse(json);
|
|||
|
|
if (doc.RootElement.TryGetProperty("result", out var resultEl))
|
|||
|
|
return resultEl.Deserialize<MesXslCustomer>(_jsonOpts);
|
|||
|
|
return null;
|
|||
|
|
}
|
|||
|
|
catch
|
|||
|
|
{
|
|||
|
|
return null;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ── 本地缓存操作 ──────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
private static List<MesXslCustomer> ApplyFilters(List<MesXslCustomer> source,
|
|||
|
|
string? customerCode, string? customerName, string? status, string? customerRegion)
|
|||
|
|
{
|
|||
|
|
IEnumerable<MesXslCustomer> q = source;
|
|||
|
|
if (!string.IsNullOrWhiteSpace(customerCode))
|
|||
|
|
q = q.Where(c => (c.CustomerCode ?? "").Contains(customerCode, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
if (!string.IsNullOrWhiteSpace(customerName))
|
|||
|
|
q = q.Where(c => (c.CustomerName ?? "").Contains(customerName, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
if (!string.IsNullOrWhiteSpace(status))
|
|||
|
|
q = q.Where(c => string.Equals(c.Status, status, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
if (!string.IsNullOrWhiteSpace(customerRegion))
|
|||
|
|
q = q.Where(c => string.Equals(c.CustomerRegion, customerRegion, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
return q.OrderByDescending(c => c.CreateTime ?? DateTime.MinValue).ToList();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private List<MesXslCustomer> ApplyPendingOpsSnapshotUnsafe(List<MesXslCustomer> source)
|
|||
|
|
{
|
|||
|
|
var map = source.Where(c => !string.IsNullOrWhiteSpace(c.Id))
|
|||
|
|
.ToDictionary(c => c.Id!, Clone, StringComparer.OrdinalIgnoreCase);
|
|||
|
|
foreach (var op in _pendingOps.OrderBy(x => x.CreatedAt))
|
|||
|
|
{
|
|||
|
|
switch (op.OpType)
|
|||
|
|
{
|
|||
|
|
case CustomerOperationType.Add:
|
|||
|
|
case CustomerOperationType.Edit:
|
|||
|
|
if (op.Customer?.Id != null) map[op.Customer.Id] = Clone(op.Customer);
|
|||
|
|
break;
|
|||
|
|
case CustomerOperationType.Delete:
|
|||
|
|
if (op.CustomerId != null) map.Remove(op.CustomerId);
|
|||
|
|
break;
|
|||
|
|
case CustomerOperationType.UpdateStatus:
|
|||
|
|
if (op.CustomerId != null && op.Status != null && map.TryGetValue(op.CustomerId, out var c))
|
|||
|
|
{
|
|||
|
|
c.Status = op.Status;
|
|||
|
|
SyncIzEnable(c);
|
|||
|
|
}
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return map.Values.ToList();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void EnqueuePendingOperation(CustomerPendingOperation op)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock) { _pendingOps.Add(op); SavePendingOpsToDiskUnsafe(); }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void UpsertLocalCache(MesXslCustomer customer)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
var idx = _localCache.FindIndex(c => string.Equals(c.Id, customer.Id, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
if (idx >= 0) _localCache[idx] = Clone(customer);
|
|||
|
|
else _localCache.Insert(0, Clone(customer));
|
|||
|
|
SaveCacheToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void RemoveFromLocalCache(string id)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
_localCache.RemoveAll(c => string.Equals(c.Id, id, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
SaveCacheToDiskUnsafe();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void UpdateLocalStatus(string id, string status)
|
|||
|
|
{
|
|||
|
|
lock (_cacheLock)
|
|||
|
|
{
|
|||
|
|
var item = _localCache.FirstOrDefault(c => string.Equals(c.Id, id, StringComparison.OrdinalIgnoreCase));
|
|||
|
|
if (item != null) { item.Status = status; SyncIzEnable(item); SaveCacheToDiskUnsafe(); }
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void LoadPendingOpsFromDisk()
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
if (!File.Exists(_pendingOpsFilePath)) return;
|
|||
|
|
_pendingOps = JsonSerializer.Deserialize<List<CustomerPendingOperation>>(
|
|||
|
|
File.ReadAllText(_pendingOpsFilePath), _jsonOpts) ?? new();
|
|||
|
|
}
|
|||
|
|
catch { _pendingOps = new(); }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void LoadCacheFromDisk()
|
|||
|
|
{
|
|||
|
|
try
|
|||
|
|
{
|
|||
|
|
if (!File.Exists(_cacheFilePath)) return;
|
|||
|
|
_localCache = JsonSerializer.Deserialize<List<MesXslCustomer>>(
|
|||
|
|
File.ReadAllText(_cacheFilePath), _jsonOpts) ?? new();
|
|||
|
|
}
|
|||
|
|
catch { _localCache = new(); }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private void SavePendingOpsToDiskUnsafe() =>
|
|||
|
|
File.WriteAllText(_pendingOpsFilePath, JsonSerializer.Serialize(_pendingOps, _jsonOpts));
|
|||
|
|
|
|||
|
|
private void SaveCacheToDiskUnsafe() =>
|
|||
|
|
File.WriteAllText(_cacheFilePath, JsonSerializer.Serialize(_localCache, _jsonOpts));
|
|||
|
|
|
|||
|
|
// ── 辅助方法 ──────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
// status "0"启用 → izEnable=1；status "1"停用 → izEnable=0
|
|||
|
|
private static void SyncIzEnable(MesXslCustomer c) =>
|
|||
|
|
c.IzEnable = c.Status == "1" ? 0 : 1;
|
|||
|
|
|
|||
|
|
private static bool IsLocalTempId(string? id) =>
|
|||
|
|
!string.IsNullOrWhiteSpace(id) && id.StartsWith("local-", StringComparison.OrdinalIgnoreCase);
|
|||
|
|
|
|||
|
|
private static MesXslCustomer Clone(MesXslCustomer c) => new()
|
|||
|
|
{
|
|||
|
|
Id = c.Id, CustomerCode = c.CustomerCode, CustomerName = c.CustomerName,
|
|||
|
|
CustomerShortName = c.CustomerShortName, CustomerRegion = c.CustomerRegion,
|
|||
|
|
ErpCode = c.ErpCode, Status = c.Status, IzEnable = c.IzEnable,
|
|||
|
|
CustomerDesc = c.CustomerDesc, TenantId = c.TenantId,
|
|||
|
|
CreateBy = c.CreateBy, CreateTime = c.CreateTime,
|
|||
|
|
UpdateBy = c.UpdateBy, UpdateTime = c.UpdateTime, SysOrgCode = c.SysOrgCode
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
// ── 内部类型 ──────────────────────────────────────────────────────────
|
|||
|
|
|
|||
|
|
private sealed class CustomerPendingOperation
|
|||
|
|
{
|
|||
|
|
public string Id { get; set; } = Guid.NewGuid().ToString("N");
|
|||
|
|
public CustomerOperationType OpType { get; set; }
|
|||
|
|
public string? CustomerId { get; set; }
|
|||
|
|
public string? Status { get; set; }
|
|||
|
|
public MesXslCustomer? Customer { get; set; }
|
|||
|
|
// 冲突检测用的版本锚点：当本地首次针对该记录产生修改时，记录当时的服务器 UpdateTime
|
|||
|
|
public DateTime? AnchorUpdateTime { get; set; }
|
|||
|
|
public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
|
|||
|
|
public int RetryCount { get; set; } = 0;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
private enum CustomerOperationType { Add = 1, Edit = 2, Delete = 3, UpdateStatus = 4 }
|
|||
|
|
|
|||
|
|
private sealed class NullableDateTimeJsonConverter : JsonConverter<DateTime?>
|
|||
|
|
{
|
|||
|
|
private static readonly string[] Formats =
|
|||
|
|
[
|
|||
|
|
"yyyy-MM-dd HH:mm:ss", "yyyy-MM-dd HH:mm:ss.fff",
|
|||
|
|
"yyyy-MM-ddTHH:mm:ss", "yyyy-MM-ddTHH:mm:ss.fff",
|
|||
|
|
"yyyy-MM-ddTHH:mm:ssZ", "yyyy-MM-ddTHH:mm:ss.fffZ"
|
|||
|
|
];
|
|||
|
|
|
|||
|
|
public override DateTime? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
|
|||
|
|
{
|
|||
|
|
if (reader.TokenType == JsonTokenType.Null) return null;
|
|||
|
|
if (reader.TokenType == JsonTokenType.String)
|
|||
|
|
{
|
|||
|
|
var raw = reader.GetString();
|
|||
|
|
if (string.IsNullOrWhiteSpace(raw)) return null;
|
|||
|
|
if (DateTime.TryParseExact(raw, Formats, System.Globalization.CultureInfo.InvariantCulture,
|
|||
|
|
System.Globalization.DateTimeStyles.AssumeLocal, out var dt)) return dt;
|
|||
|
|
if (DateTime.TryParse(raw, out var fb)) return fb;
|
|||
|
|
}
|
|||
|
|
throw new JsonException($"无法转换为 DateTime?，token={reader.TokenType}");
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
public override void Write(Utf8JsonWriter writer, DateTime? value, JsonSerializerOptions options)
|
|||
|
|
{
|
|||
|
|
if (value.HasValue) writer.WriteStringValue(value.Value.ToString("yyyy-MM-dd HH:mm:ss"));
|
|||
|
|
else writer.WriteNullValue();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|