< Summary

Information
Class: Services.BitrixStockServices.BitrixStockSyncPipelineOrchestrator
Assembly: Services
File(s): /home/runner/work/Studio2Prod/Studio2Prod/AppRepository/BitrixStockServices/BitrixStockSyncPipelineOrchestrator.cs
Line coverage
85%
Covered lines: 187
Uncovered lines: 31
Coverable lines: 218
Total lines: 342
Line coverage: 85.7%
Branch coverage
57%
Covered branches: 66
Total branches: 114
Branch coverage: 57.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
RunPipelineAsync()82%505098.91%
EmitStarted()100%22100%
EmitStage()100%22100%
EmitFinished()100%22100%
FileProcessAsync()75%44100%
TryNotifyFailedAsync()100%1180%
TryNotifyCompletedAsync()0%27833.33%
NotifyBellAsync()0%702260%
PlanStages(...)100%1212100%
BuildApplyRequest(...)50%88100%
BuildCleanupRequest(...)100%11100%
BuildPricePushRequest(...)100%11100%
BuildOrphanRequest(...)100%11100%

File(s)

/home/runner/work/Studio2Prod/Studio2Prod/AppRepository/BitrixStockServices/BitrixStockSyncPipelineOrchestrator.cs

#LineLine coverage
 1using Hangfire;
 2using Microsoft.AspNetCore.SignalR;
 3using Microsoft.Extensions.Logging;
 4using Services.HangfireService;
 5using Services.SignalRServices;
 6
 7namespace Services.BitrixStockServices;
 8
 9/// <summary>
 10/// Единый серверный оркестратор цепочки этапов BitrixStock. См. <see cref="IBitrixStockSyncPipelineOrchestrator"/>.
 11/// Модель — как у <c>RunAllPresetsJob</c>: последовательные await в одном задании. UI получает прогресс через
 12/// единую SignalR-подписку на группу прогона (события PipelineStarted / PipelineStage / PipelineFinished).
 13/// </summary>
 14public class BitrixStockSyncPipelineOrchestrator : IBitrixStockSyncPipelineOrchestrator
 15{
 16    private readonly IBitrixStockSyncReportService _reportService;
 17    private readonly IBitrixStockSyncReportApplyService _applyService;
 18    private readonly IBitrixStockSyncReportCleanupService _cleanupService;
 19    private readonly IBitrixStockSyncReportPricePushService _pricePushService;
 20    private readonly IBitrixStockOzonOrphanCleanupService _orphanCleanupService;
 21    private readonly IHubContext<JobProgressHub> _jobProgressHub;
 22    private readonly BitrixStockActiveStageStore _stageStore;
 23    private readonly BitrixStockRolloutTelegramReporter _telegramReporter;
 24    private readonly ILogger<BitrixStockSyncPipelineOrchestrator> _logger;
 25    private readonly IBitrixStockRunFileLogger? _runFileLogger;
 26
 1927    public BitrixStockSyncPipelineOrchestrator(
 1928        IBitrixStockSyncReportService reportService,
 1929        IBitrixStockSyncReportApplyService applyService,
 1930        IBitrixStockSyncReportCleanupService cleanupService,
 1931        IBitrixStockSyncReportPricePushService pricePushService,
 1932        IBitrixStockOzonOrphanCleanupService orphanCleanupService,
 1933        IHubContext<JobProgressHub> jobProgressHub,
 1934        BitrixStockActiveStageStore stageStore,
 1935        BitrixStockRolloutTelegramReporter telegramReporter,
 1936        ILogger<BitrixStockSyncPipelineOrchestrator> logger,
 1937        IBitrixStockRunFileLogger? runFileLogger = null)
 38    {
 1939        _reportService = reportService;
 1940        _applyService = applyService;
 1941        _cleanupService = cleanupService;
 1942        _pricePushService = pricePushService;
 1943        _orphanCleanupService = orphanCleanupService;
 1944        _jobProgressHub = jobProgressHub;
 1945        _stageStore = stageStore;
 1946        _telegramReporter = telegramReporter;
 1947        _logger = logger;
 1948        _runFileLogger = runFileLogger;
 1949    }
 50
 51    [Queue(HangfireQueueNames.Main)]
 52    public async Task<BitrixStockPipelineResult> RunPipelineAsync(
 53        string reportId,
 54        BitrixStockSyncReportStartRequest request,
 55        string? rootJobId,
 56        CancellationToken cancellationToken = default)
 57    {
 1958        var hubId = rootJobId?.Trim() ?? string.Empty;
 1959        var hasHub = !string.IsNullOrWhiteSpace(hubId);
 1960        var ss = request.StageSettings;
 1961        var result = new BitrixStockPipelineResult();
 1962        using var fileRun = _runFileLogger?.BeginRun(new BitrixStockRunContext(
 1963            reportId, rootJobId == null ? "PresetPipeline" : "UI", reportId, rootJobId,
 1964            ClientName: string.Join(",", request.OzonClientIds)));
 65
 66        async Task EmitStarted(IReadOnlyList<int> stages)
 67        {
 1768            if (hasHub)
 369                await _jobProgressHub.Clients.Group(hubId)
 370                    .SendAsync("PipelineStarted", new { reportId, stages }, cancellationToken);
 1771        }
 72
 73        async Task EmitStage(int stage, string state, object? stats = null)
 74        {
 8075            if (hasHub)
 976                await _jobProgressHub.Clients.Group(hubId)
 977                    .SendAsync("PipelineStage", new { reportId, stage, state, stats }, cancellationToken);
 8078        }
 79
 80        async Task EmitFinished(bool success, string? error)
 81        {
 1782            if (hasHub)
 383                await _jobProgressHub.Clients.Group(hubId)
 384                    .SendAsync("PipelineFinished", new { reportId, success, error }, cancellationToken);
 1785        }
 86
 1987        if (_stageStore.Get(reportId) != 0)
 88        {
 189            _logger.LogInformation("Pipeline {ReportId} уже выполняется — повторный запуск отменён", reportId);
 190            result.Error = "Цепочка уже выполняется";
 191            return result;
 92        }
 93
 1894        var otherActive = _stageStore.GetAnyActiveReportId();
 1895        if (otherActive != null && !string.Equals(otherActive, reportId, StringComparison.Ordinal))
 96        {
 197            _logger.LogInformation("Pipeline {ReportId} не запущен: уже активна цепочка {Other}", reportId, otherActive)
 198            result.Error = $"Уже выполняется цепочка {otherActive}";
 199            return result;
 100        }
 101
 102        try
 103        {
 17104            await EmitStarted(PlanStages(ss));
 17105            await FileProcessAsync("Запуск pipeline этапов 1–5");
 106
 17107            _stageStore.Set(reportId, 1);
 17108            await FileProcessAsync("Этап 1: формирование отчёта");
 17109            await EmitStage(1, "running");
 17110            request.Chained = true;
 17111            await _reportService.BuildReportAsync(reportId, request, hasHub ? hubId : null, cancellationToken);
 13112            await EmitStage(1, "done");
 13113            await FileProcessAsync("Этап 1 завершён");
 13114            result.LastStage = 1;
 115
 13116            if (ss?.AutoRunApply == true)
 117            {
 11118                _stageStore.Set(reportId, 2);
 11119                await FileProcessAsync("Этап 2: накат остатков");
 11120                await EmitStage(2, "running");
 11121                var applyResult = await _applyService.ApplyReportAsync(
 11122                    reportId, BuildApplyRequest(reportId, ss, hubId), hasHub ? hubId : null, cancellationToken);
 11123                await EmitStage(2, "done", new { applyResult.SuccessCount, applyResult.FailedCount, applyResult.Total })
 11124                result.LastStage = 2;
 11125                await FileProcessAsync($"Этап 2 завершён: успешно {applyResult.SuccessCount}, ошибок {applyResult.Failed
 126
 11127                if (applyResult.SuccessCount > 0)
 128                {
 8129                    if (ss.AutoRunCleanup)
 130                    {
 4131                        _stageStore.Set(reportId, 3);
 4132                        await FileProcessAsync("Этап 3: очистка FBS");
 4133                        await EmitStage(3, "running");
 4134                        await _cleanupService.RunCleanupAsync(
 4135                            reportId, BuildCleanupRequest(reportId, ss, hubId), hasHub ? hubId : null, cancellationToken
 4136                        await EmitStage(3, "done");
 4137                        result.LastStage = 3;
 4138                        await FileProcessAsync("Этап 3 завершён");
 139                    }
 140
 8141                    if (ss.AutoRunPricePush)
 142                    {
 3143                        _stageStore.Set(reportId, 4);
 3144                        await FileProcessAsync("Этап 4: накат цен");
 3145                        await EmitStage(4, "running");
 3146                        await _pricePushService.PushPricesAsync(
 3147                            reportId, BuildPricePushRequest(reportId, ss, hubId), hasHub ? hubId : null, cancellationTok
 3148                        await EmitStage(4, "done");
 3149                        result.LastStage = 4;
 3150                        await FileProcessAsync("Этап 4 завершён");
 151                    }
 152                }
 153                else
 154                {
 6155                    if (ss.AutoRunCleanup)  await EmitStage(3, "skipped", new { reason = "0 успешных применений" });
 4156                    if (ss.AutoRunPricePush) await EmitStage(4, "skipped", new { reason = "0 успешных применений" });
 3157                    _logger.LogInformation("Pipeline {ReportId}: этап 2 без успешных применений — этапы 3/4 пропущены", 
 158                }
 11159            }
 160
 13161            if (ss?.AutoRunOzonOrphanCleanup == true)
 162            {
 4163                var catalogSource = ss.Stage5CatalogSource ?? "OzonProductsDb";
 4164                _logger.LogInformation(
 4165                    "Pipeline {ReportId}: этап 5 — источник каталога {CatalogSource}",
 4166                    reportId, catalogSource);
 4167                _stageStore.Set(reportId, 5);
 4168                await FileProcessAsync("Этап 5: обработка сирот Ozon");
 4169                await EmitStage(5, "running", new { catalogSource });
 4170                await _orphanCleanupService.RunOrphanCleanupAsync(
 4171                    reportId, BuildOrphanRequest(reportId, ss, hubId), hasHub ? hubId : null, cancellationToken);
 4172                await EmitStage(5, "done");
 4173                result.LastStage = 5;
 4174                await FileProcessAsync("Этап 5 завершён");
 4175            }
 176
 13177            result.Success = true;
 13178            if (_runFileLogger != null) await _runFileLogger.CompleteAsync(true, cancellationToken: cancellationToken);
 13179            await TryNotifyCompletedAsync(reportId, result.LastStage, cancellationToken);
 13180            await EmitFinished(true, null);
 13181            return result;
 182        }
 2183        catch (OperationCanceledException)
 184        {
 2185            if (_runFileLogger != null) await _runFileLogger.CompleteAsync(false, "Отменено", CancellationToken.None);
 2186            await EmitFinished(false, "Отменено");
 2187            throw;
 0188        }
 2189        catch (Exception ex)
 190        {
 2191            if (_runFileLogger != null) await _runFileLogger.CompleteAsync(false, ex.Message, CancellationToken.None);
 2192            _logger.LogError(ex, "Pipeline {ReportId} прерван на этапе {Stage}", reportId, result.LastStage + 1);
 2193            result.Error = ex.Message;
 2194            await EmitStage(result.LastStage + 1, "failed", new { error = ex.Message });
 2195            await TryNotifyFailedAsync(reportId, result.LastStage + 1, ex.Message);
 2196            await EmitFinished(false, ex.Message);
 2197            return result;
 198        }
 199        finally
 200        {
 17201            _stageStore.Clear(reportId);
 202        }
 203
 91204        Task FileProcessAsync(string message) => _runFileLogger?.ProcessAsync(message, cancellationToken: cancellationTo
 17205    }
 206
 207    /// <summary>Сообщение в Telegram о том, что конвейер прерван: этап, ошибка и то, что успело примениться.</summary>
 208    private async Task TryNotifyFailedAsync(string reportId, int failedStage, string error)
 209    {
 210        try
 211        {
 2212            BitrixStockSyncReportRunDto? run = null;
 213            try
 214            {
 2215                var page = await _reportService.ListReportsAsync(new BitrixStockSyncReportRunsQuery { PageSize = 5 }, Ca
 0216                run = page.Items.FirstOrDefault(r => r.ReportId == reportId);
 0217            }
 2218            catch
 219            {
 220                // Отчёта может не быть (сбой на этапе 1) — сообщим без частичных результатов.
 2221            }
 222
 2223            await _telegramReporter.NotifyFailedAsync(reportId, run, failedStage, error);
 0224        }
 2225        catch
 226        {
 227            // Уведомление не должно менять исход пайплайна.
 2228        }
 2229    }
 230
 231    private async Task TryNotifyCompletedAsync(string reportId, int lastStage, CancellationToken ct)
 232    {
 233        BitrixStockSyncReportRunDto? run;
 234        try
 235        {
 13236            var page = await _reportService.ListReportsAsync(
 13237                new BitrixStockSyncReportRunsQuery { PageSize = 5 }, ct);
 0238            run = page.Items.FirstOrDefault(r => r.ReportId == reportId);
 0239        }
 26240        catch { return; }
 241
 0242        if (run == null) return;
 243
 244        // Необнулённая сирота — такая же ошибка, как отказ этапа 2: остаток продолжает продаваться
 245        // при нулях в 1С, пока кто-нибудь не заметит расхождение вручную.
 0246        var hasErrors = run.OzonApplyFailedCount > 0 || run.PricePushErrorCount > 0
 0247            || run.OzonCleanupFailedCount > 0 || run.OrphansZeroed < run.OrphansFound;
 248
 0249        try { await NotifyBellAsync(run, hasErrors); }
 0250        catch { /* не роняем пайплайн */ }
 251
 0252        try { await _telegramReporter.NotifyAsync(run, ct); }
 0253        catch { /* не роняем пайплайн */ }
 13254    }
 255
 256    /// <summary>Короткое уведомление-колокольчик в UI (шапка Studio2).</summary>
 257    private static async Task NotifyBellAsync(BitrixStockSyncReportRunDto run, bool hasErrors)
 258    {
 0259        var parts = new List<string>();
 260
 0261        if (run.PushCount > 0 || run.DecreaseCount > 0 || run.ClearCount > 0)
 0262            parts.Add($"↑{run.PushCount} добавить · ↓{run.DecreaseCount} убавить · ⊘{run.ClearCount} обнулить");
 263
 0264        if (run.OzonAppliedSuccessCount > 0 || run.OzonApplyFailedCount > 0)
 0265            parts.Add($"применено {run.OzonAppliedSuccessCount}" +
 0266                      (run.OzonApplyFailedCount > 0 ? $", ошибок {run.OzonApplyFailedCount}" : ""));
 267
 0268        if (run.PricePushCount > 0 || run.PricePushUnchangedCount > 0)
 0269            parts.Add($"₽{run.PricePushCount} цен" +
 0270                      (run.PricePushUnchangedCount > 0 ? $" · ≈{run.PricePushUnchangedCount} без изм" : ""));
 271
 0272        if (run.OrphansFound > 0 || run.OrphansZeroed > 0)
 0273            parts.Add($"{run.OrphansFound}/{run.OrphansZeroed} сирот");
 274
 0275        if (parts.Count == 0) return;
 276
 0277        await NotificationService.NotifyAsync(
 0278            "Bitrix Остатки",
 0279            hasErrors ? "#D97706" : "#6366F1",
 0280            string.Join(" | ", parts),
 0281            "fa-arrows-rotate");
 0282    }
 283
 284    private static List<int> PlanStages(PresetStageSettings? ss)
 285    {
 17286        var stages = new List<int> { 1 };
 17287        if (ss?.AutoRunApply == true)
 288        {
 15289            stages.Add(2);
 22290            if (ss.AutoRunCleanup) stages.Add(3);
 19291            if (ss.AutoRunPricePush) stages.Add(4);
 292        }
 21293        if (ss?.AutoRunOzonOrphanCleanup == true) stages.Add(5);
 17294        return stages;
 295    }
 296
 11297    private static BitrixStockSyncReportApplyStartRequest BuildApplyRequest(string reportId, PresetStageSettings ss, str
 11298    {
 11299        JobId       = rootJobId,
 11300        ReportId    = reportId,
 11301        OnlyPending = true,
 11302        Actions     = (ss.ApplyActions ?? []).Count > 0
 11303            ? ss.ApplyActions!
 11304            : new List<string> { "Push", "Decrease", "Clear" },
 11305        ArchivedMode    = (ArchivedProductMode)ss.ApplyArchivedMode,
 11306        UnarchiveFilter = (UnarchiveFilterMode)ss.ApplyUnarchiveFilter,
 11307        MaxLines        = ss.ApplyMaxLines,
 11308        DisabledWarehouseMappingIds = (ss.ApplyDisabledWarehouseMappingIds ?? []).Count > 0
 11309            ? ss.ApplyDisabledWarehouseMappingIds
 11310            : null,
 11311        StageSettings = ss,
 11312        Chained = true
 11313    };
 314
 4315    private static BitrixStockSyncCleanupStartRequest BuildCleanupRequest(string reportId, PresetStageSettings ss, strin
 4316    {
 4317        JobId       = rootJobId,
 4318        ReportId    = reportId,
 4319        OnlyPending = ss.CleanupOnlyPending,
 4320        MaxProducts = null,
 4321        StageSettings = ss,
 4322        Chained = true
 4323    };
 324
 3325    private static BitrixStockSyncReportPricePushStartRequest BuildPricePushRequest(string reportId, PresetStageSettings
 3326    {
 3327        JobId       = rootJobId,
 3328        ReportId    = reportId,
 3329        OnlyPending = true,
 3330        StageSettings = ss,
 3331        Chained = true
 3332    };
 333
 4334    private static BitrixStockOzonOrphanCleanupRequest BuildOrphanRequest(string reportId, PresetStageSettings ss, strin
 4335    {
 4336        ReportId      = reportId,
 4337        JobId         = rootJobId,
 4338        DryRun        = false,
 4339        StageSettings = ss,
 4340        Chained = true
 4341    };
 342}
Generated by: ReportGenerator 5.5.11.0
10/25/2026 - 07:43:48
GitHub | reportgenerator.io

Methods/Properties

.ctor(Services.BitrixStockServices.IBitrixStockSyncReportService,Services.BitrixStockServices.IBitrixStockSyncReportApplyService,Services.BitrixStockServices.IBitrixStockSyncReportCleanupService,Services.BitrixStockServices.IBitrixStockSyncReportPricePushService,Services.BitrixStockServices.IBitrixStockOzonOrphanCleanupService,Microsoft.AspNetCore.SignalR.IHubContext`1<Services.SignalRServices.JobProgressHub>,Services.BitrixStockServices.BitrixStockActiveStageStore,Services.BitrixStockServices.BitrixStockRolloutTelegramReporter,Microsoft.Extensions.Logging.ILogger`1<Services.BitrixStockServices.BitrixStockSyncPipelineOrchestrator>,Services.BitrixStockServices.IBitrixStockRunFileLogger)
RunPipelineAsync()
EmitStarted()
EmitStage()
EmitFinished()
FileProcessAsync()
TryNotifyFailedAsync()
TryNotifyCompletedAsync()
NotifyBellAsync()
PlanStages(Services.BitrixStockServices.PresetStageSettings)
BuildApplyRequest(System.String,Services.BitrixStockServices.PresetStageSettings,System.String)
BuildCleanupRequest(System.String,Services.BitrixStockServices.PresetStageSettings,System.String)
BuildPricePushRequest(System.String,Services.BitrixStockServices.PresetStageSettings,System.String)
BuildOrphanRequest(System.String,Services.BitrixStockServices.PresetStageSettings,System.String)