src/PdfEncoder.Infrastructure/Jobs/DocumentProcessingBackgroundService.cs - master - afanasevn/PdfEncoder

src/PdfEncoder.Infrastructure/Jobs/DocumentProcessingBackgroundService.cs
139 строк5 KB
IBS\NAfanasev
Full async product flow
01 июл 2026, 17:49
01 июл 2026, 17:49d25833e
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using PdfEncoder.Application.Abstractions;
using PdfEncoder.Application.Models;
using PdfEncoder.Application.Options;
using PdfEncoder.Domain.Errors;

namespace PdfEncoder.Infrastructure.Jobs;

/// <summary>
/// Фоновый worker in-process очереди с persistence (Фаза 3).
/// </summary>
public sealed class DocumentProcessingBackgroundService(
    IDocumentJobQueue jobQueue,
    IServiceScopeFactory scopeFactory,
    IOptions<PipelineOptions> pipelineOptions,
    ILogger<DocumentProcessingBackgroundService> logger) : BackgroundService
{
    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        logger.LogInformation(
            "Document processing background service started. MaxConcurrentJobs={MaxConcurrentJobs}",
            pipelineOptions.Value.MaxConcurrentJobs);

        var workers = Enumerable.Range(0, pipelineOptions.Value.MaxConcurrentJobs)
            .Select(workerId => ProcessQueueAsync(workerId, stoppingToken))
            .ToArray();

        await Task.WhenAll(workers);
    }

    private async Task ProcessQueueAsync(
        int workerId,
        CancellationToken stoppingToken)
    {
        await foreach (var workItem in ReadItemsAsync(stoppingToken))
        {
            await ProcessWorkItemAsync(workerId, workItem, stoppingToken);
        }
    }

    private async IAsyncEnumerable<DocumentJobWorkItem> ReadItemsAsync(
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            DocumentJobWorkItem workItem;
            try
            {
                workItem = await jobQueue.DequeueAsync(stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                yield break;
            }

            yield return workItem;
        }
    }

    private async Task ProcessWorkItemAsync(
        int workerId,
        DocumentJobWorkItem workItem,
        CancellationToken stoppingToken)
    {
        logger.LogInformation(
            "Worker {WorkerId} picked job {DocumentId}",
            workerId,
            workItem.DocumentId);

        if (!File.Exists(workItem.FilePath))
        {
            logger.LogWarning(
                "Worker {WorkerId}: file not found for job {DocumentId}: {FilePath}",
                workerId,
                workItem.DocumentId,
                workItem.FilePath);

            await using var failScope = scopeFactory.CreateAsyncScope();
            var failService = failScope.ServiceProvider.GetRequiredService<IDocumentJobService>();
            await failService.FailAsync(
                workItem.DocumentId,
                "invalid-file",
                "Stored PDF file is missing.",
                stoppingToken);
            return;
        }

        await using var scope = scopeFactory.CreateAsyncScope();
        var jobService = scope.ServiceProvider.GetRequiredService<IDocumentJobService>();
        var pipeline = scope.ServiceProvider.GetRequiredService<IDocumentParsePipeline>();

        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
        timeoutCts.CancelAfter(TimeSpan.FromSeconds(pipelineOptions.Value.TimeoutSeconds));

        try
        {
            await jobService.MarkProcessingAsync(workItem.DocumentId, timeoutCts.Token);
            var parseOptions = await jobService.GetParseOptionsAsync(workItem.DocumentId, timeoutCts.Token)
                ?? new ParseOptions();

            var content = await File.ReadAllBytesAsync(workItem.FilePath, timeoutCts.Token);
            var result = await pipeline.RunAsync(new DocumentParseRequest
            {
                DocumentId = workItem.DocumentId,
                Content = content,
                FileName = Path.GetFileName(workItem.FilePath),
                Options = parseOptions
            }, timeoutCts.Token);

            await jobService.CompleteAsync(workItem.DocumentId, result, timeoutCts.Token);

            logger.LogInformation(
                "Worker {WorkerId}: job {DocumentId} completed. Type={DocumentType}, Confidence={Confidence}",
                workerId,
                workItem.DocumentId,
                result.Result.DetectedDocumentType,
                result.Result.OverallConfidence);
        }
        catch (LlmException ex)
        {
            await jobService.FailAsync(workItem.DocumentId, ex.ErrorCode, ex.Message, CancellationToken.None);
            logger.LogError(ex, "Worker {WorkerId}: job {DocumentId} failed (LLM).", workerId, workItem.DocumentId);
        }
        catch (PdfValidationException ex)
        {
            await jobService.FailAsync(workItem.DocumentId, ex.ErrorCode, ex.Message, CancellationToken.None);
            logger.LogError(ex, "Worker {WorkerId}: job {DocumentId} failed (validation).", workerId, workItem.DocumentId);
        }
        catch (Exception ex)
        {
            await jobService.FailAsync(workItem.DocumentId, "pipeline-failed", ex.Message, CancellationToken.None);
            logger.LogError(ex, "Worker {WorkerId}: job {DocumentId} failed.", workerId, workItem.DocumentId);
        }
    }
}