| | | 1 | | using Microsoft.EntityFrameworkCore; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using Services.SystemMonitoringServices; |
| | | 4 | | using OzonDomains.Models.SupplierStatuses; |
| | | 5 | | using OzonRepositories.Context; |
| | | 6 | | |
| | | 7 | | namespace Services.SupplierStatusServices; |
| | | 8 | | |
| | | 9 | | public interface IOrderSupplierStatusSyncService |
| | | 10 | | { |
| | | 11 | | /// <summary>Опрашивает поставщиков по активным строкам состояния и применяет результат.</summary> |
| | | 12 | | Task<int> SyncAsync(CancellationToken ct = default); |
| | | 13 | | |
| | | 14 | | /// <summary>Синхронизация по конкретным заказам — для кнопки «обновить» в интерфейсе.</summary> |
| | | 15 | | Task<int> SyncOrdersAsync(IReadOnlyCollection<int> orderIds, CancellationToken ct = default); |
| | | 16 | | } |
| | | 17 | | |
| | | 18 | | /// <summary> |
| | | 19 | | /// Разбирает очередь активных состояний и раздаёт её провайдерам. Стадии обрабатываются |
| | | 20 | | /// раздельно: Cart идёт в FetchCartAsync, Ordered — в FetchOrderAsync. |
| | | 21 | | /// </summary> |
| | | 22 | | public sealed class OrderSupplierStatusSyncService : IOrderSupplierStatusSyncService |
| | | 23 | | { |
| | | 24 | | /// <summary>Верхняя граница батча: очередь дренируется за несколько прогонов, БД не залипает.</summary> |
| | | 25 | | private const int BatchSize = 500; |
| | | 26 | | |
| | | 27 | | /// <summary>Строку, опрошенную недавно, пропускаем — иначе один поставщик съест весь батч.</summary> |
| | | 28 | | private const int MinResyncMinutes = 15; |
| | | 29 | | |
| | | 30 | | /// <summary>За пределами окна статусы у поставщиков уже не меняются.</summary> |
| | | 31 | | private const int LookbackDays = 60; |
| | | 32 | | |
| | | 33 | | private readonly ApplicationDbContext _db; |
| | | 34 | | private readonly IOrderSupplierStateService _stateService; |
| | | 35 | | private readonly IReadOnlyDictionary<string, ISupplierStatusProvider> _providers; |
| | | 36 | | private readonly ILogger<OrderSupplierStatusSyncService> _logger; |
| | | 37 | | private readonly ISystemMonitoringSignalWriter? _signals; |
| | | 38 | | |
| | 24 | 39 | | public OrderSupplierStatusSyncService( |
| | 24 | 40 | | ApplicationDbContext db, |
| | 24 | 41 | | IOrderSupplierStateService stateService, |
| | 24 | 42 | | IEnumerable<ISupplierStatusProvider> providers, |
| | 24 | 43 | | ILogger<OrderSupplierStatusSyncService> logger, |
| | 24 | 44 | | ISystemMonitoringSignalWriter? signals = null) |
| | | 45 | | { |
| | 24 | 46 | | _db = db; |
| | 24 | 47 | | _stateService = stateService; |
| | 49 | 48 | | _providers = providers.ToDictionary(x => x.ProviderKey, StringComparer.Ordinal); |
| | 24 | 49 | | _logger = logger; |
| | 24 | 50 | | _signals = signals; |
| | 24 | 51 | | } |
| | | 52 | | |
| | 20 | 53 | | public Task<int> SyncAsync(CancellationToken ct = default) => SyncCoreAsync(null, ct); |
| | | 54 | | |
| | | 55 | | public Task<int> SyncOrdersAsync(IReadOnlyCollection<int> orderIds, CancellationToken ct = default) => |
| | 4 | 56 | | orderIds.Count == 0 ? Task.FromResult(0) : SyncCoreAsync(orderIds.ToList(), ct); |
| | | 57 | | |
| | | 58 | | private async Task<int> SyncCoreAsync(List<int>? orderIds, CancellationToken ct) |
| | | 59 | | { |
| | 23 | 60 | | var query = _db.OrderSupplierStates |
| | 23 | 61 | | .AsNoTracking() |
| | 23 | 62 | | .Where(x => x.ActiveOrderId != null |
| | 23 | 63 | | && (x.Stage == OrderSupplierStage.Cart || x.Stage == OrderSupplierStage.Ordered)); |
| | | 64 | | |
| | 23 | 65 | | if (orderIds is not null) |
| | | 66 | | { |
| | 3 | 67 | | query = query.Where(x => orderIds.Contains(x.OrderId)); |
| | | 68 | | } |
| | | 69 | | else |
| | | 70 | | { |
| | | 71 | | // Ручной запрос обновляет что просили; фоновой прогон дренирует очередь по свежести. |
| | 20 | 72 | | var resyncCutoff = DateTime.UtcNow.AddMinutes(-MinResyncMinutes); |
| | 20 | 73 | | var createdCutoff = DateTime.UtcNow.AddDays(-LookbackDays); |
| | 20 | 74 | | query = query |
| | 20 | 75 | | .Where(x => x.CreatedAt >= createdCutoff) |
| | 20 | 76 | | .Where(x => x.LastSyncedAt == null || x.LastSyncedAt < resyncCutoff) |
| | 20 | 77 | | .OrderBy(x => x.LastSyncedAt ?? DateTime.MinValue) |
| | 20 | 78 | | .Take(BatchSize); |
| | | 79 | | } |
| | | 80 | | |
| | 23 | 81 | | var states = await query |
| | 23 | 82 | | .Select(x => new OrderSupplierStateRef( |
| | 23 | 83 | | x.Id, x.OrderId, x.ProviderKey, x.Stage, x.CartItemId, x.ExternalBasketItemId, |
| | 23 | 84 | | x.ExternalOrderNumber, x.ExternalOrderId, x.ExternalLineId, x.SupplierStatusId)) |
| | 23 | 85 | | .ToListAsync(ct); |
| | | 86 | | |
| | 28 | 87 | | if (states.Count == 0) return 0; |
| | | 88 | | |
| | 16 | 89 | | var changed = 0; |
| | 594 | 90 | | foreach (var group in states.GroupBy(x => new { x.ProviderKey, x.Stage })) |
| | | 91 | | { |
| | 19 | 92 | | if (!_providers.TryGetValue(group.Key.ProviderKey, out var provider)) |
| | | 93 | | { |
| | 2 | 94 | | _logger.LogWarning( |
| | 2 | 95 | | "Нет провайдера статусов для '{ProviderKey}' — {Count} строк пропущено", |
| | 2 | 96 | | group.Key.ProviderKey, group.Count()); |
| | 5 | 97 | | var stateIds = group.Select(x => x.StateId).ToList(); |
| | 2 | 98 | | await _stateService.MarkSyncedAsync(stateIds, $"Нет провайдера статусов для '{group.Key.ProviderKey}'", |
| | 2 | 99 | | continue; |
| | | 100 | | } |
| | | 101 | | |
| | 17 | 102 | | var batch = group.ToList(); |
| | | 103 | | try |
| | | 104 | | { |
| | | 105 | | // Ошибка одного поставщика не должна ронять синхронизацию остальных. |
| | 17 | 106 | | var snapshots = group.Key.Stage == OrderSupplierStage.Cart |
| | 17 | 107 | | ? await provider.FetchCartAsync(batch, ct) |
| | 17 | 108 | | : await provider.FetchOrderAsync(batch, ct); |
| | | 109 | | |
| | 16 | 110 | | changed += await _stateService.ApplySnapshotsAsync(group.Key.ProviderKey, snapshots, ct); |
| | 16 | 111 | | await (_signals?.RecordAsync("supplier-status", "success", group.Key.ProviderKey, cancellationToken: ct) |
| | | 112 | | |
| | | 113 | | // Строки, по которым поставщик ничего не сказал, тоже отмечаем опрошенными, |
| | | 114 | | // иначе они займут весь следующий батч и заблокируют очередь. |
| | 535 | 115 | | var answered = snapshots.Select(x => x.OrderId).ToHashSet(); |
| | 537 | 116 | | var silent = batch.Where(x => !answered.Contains(x.OrderId)).Select(x => x.StateId).ToList(); |
| | 16 | 117 | | await _stateService.MarkSyncedAsync(silent, null, ct); |
| | 16 | 118 | | } |
| | 1 | 119 | | catch (Exception ex) |
| | | 120 | | { |
| | 1 | 121 | | _logger.LogError(ex, |
| | 1 | 122 | | "Синхронизация статусов {ProviderKey} ({Stage}) не удалась", |
| | 1 | 123 | | group.Key.ProviderKey, group.Key.Stage); |
| | 1 | 124 | | await _stateService.MarkSyncedAsync( |
| | 2 | 125 | | batch.Select(x => x.StateId).ToList(), ex.Message, ct); |
| | 1 | 126 | | await (_signals?.RecordAsync("supplier-status", "failure", group.Key.ProviderKey, "Ошибка синхронизации |
| | | 127 | | } |
| | 17 | 128 | | } |
| | | 129 | | |
| | 16 | 130 | | return changed; |
| | 22 | 131 | | } |
| | | 132 | | } |