src/PdfEncoder.Infrastructure/Persistence/DocumentJobService.cs - master - afanasevn/PdfEncoder

src/PdfEncoder.Infrastructure/Persistence/DocumentJobService.cs
378 строк12 KB
IBS\NAfanasev
Field validation, integration tests, UI polish
06 июл 2026, 17:18
06 июл 2026, 17:180e936b0
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using PdfEncoder.Application.Abstractions;
using PdfEncoder.Application.Errors;
using PdfEncoder.Application.Models;
using PdfEncoder.Domain.Enums;
using PdfEncoder.Domain.Errors;
using PdfEncoder.Domain.Models;
using PdfEncoder.Infrastructure.Persistence.Entities;

namespace PdfEncoder.Infrastructure.Persistence;

/// <summary>
/// CRUD document jobs с idempotency по ContentHash.
/// </summary>
public sealed class DocumentJobService(
    PdfEncoderDbContext dbContext,
    IFileStorage fileStorage,
    IContentHashService contentHashService,
    IPdfInputValidator inputValidator,
    IDocumentParsePipeline parsePipeline,
    IDocumentJobQueue jobQueue,
    ILogger<DocumentJobService> logger) : IDocumentJobService
{
    /// <inheritdoc />
    public async Task<DocumentJobSubmitResult> SubmitAsync(
        SubmitDocumentJobRequest request,
        CancellationToken ct)
    {
        _ = await inputValidator.ValidateAsync(request.Content, request.FileName, ct);
        var contentHash = contentHashService.ComputeHash(request.Content.Span);

        var existing = await dbContext.DocumentJobs
            .Include(job => job.ParseResult)
            .FirstOrDefaultAsync(job => job.ContentHash == contentHash, ct);

        if (existing is not null)
        {
            return MapExistingJob(existing);
        }

        var documentId = Guid.NewGuid();
        await fileStorage.SaveAsync(documentId, request.Content, ct);

        var job = new DocumentJobEntity
        {
            Id = documentId,
            ContentHash = contentHash,
            OriginalFileName = SanitizeFileName(request.FileName),
            FileSizeBytes = request.Content.Length,
            Status = DocumentJobStatus.Pending,
            DocumentType = request.Options.DocumentType,
            Language = request.Options.Language,
            CreatedAt = DateTime.UtcNow
        };

        dbContext.DocumentJobs.Add(job);
        await dbContext.SaveChangesAsync(ct);

        if (request.Options.Async)
        {
            await EnqueueJobAsync(job, ct);
            return new DocumentJobSubmitResult
            {
                Response = MapJob(job),
                StatusCode = 202,
                FromCache = false
            };
        }

        return await RunSynchronouslyAsync(job, request, ct);
    }

    /// <inheritdoc />
    public async Task<DocumentJobResponse?> GetAsync(
        Guid documentId,
        CancellationToken ct)
    {
        var job = await dbContext.DocumentJobs
            .AsNoTracking()
            .Include(j => j.ParseResult)
            .FirstOrDefaultAsync(j => j.Id == documentId, ct);

        return job is null ? null : MapJob(job);
    }

    /// <inheritdoc />
    public async Task<DocumentListResponse> ListAsync(
        ListDocumentsQuery query,
        CancellationToken ct)
    {
        var limit = Math.Clamp(query.Limit, 1, 100);
        var offset = Math.Max(0, query.Offset);

        var jobsQuery = dbContext.DocumentJobs.AsNoTracking();
        if (query.Status.HasValue)
        {
            jobsQuery = jobsQuery.Where(job => job.Status == query.Status.Value);
        }

        var total = await jobsQuery.CountAsync(ct);
        var jobs = await jobsQuery
            .Include(job => job.ParseResult)
            .OrderByDescending(job => job.CreatedAt)
            .Skip(offset)
            .Take(limit)
            .ToListAsync(ct);

        var items = jobs
            .Select(job => new DocumentListItem
            {
                DocumentId = job.Id,
                OriginalFileName = job.OriginalFileName,
                DetectedDocumentType = job.ParseResult?.DetectedDocumentType ?? "unknown",
                OverallConfidence = job.ParseResult?.OverallConfidence ?? 0,
                Status = job.Status,
                CreatedAt = new DateTimeOffset(job.CreatedAt, TimeSpan.Zero),
                ErrorCode = job.Status == DocumentJobStatus.Failed ? job.ErrorCode : null,
                ErrorSummary = DocumentJobErrorSummary.Resolve(job.ErrorCode, job.Status)
            })
            .ToList();

        return new DocumentListResponse
        {
            Items = items,
            Total = total
        };
    }

    /// <inheritdoc />
    public async Task<DocumentContext?> GetRawContextAsync(
        Guid documentId,
        CancellationToken ct)
    {
        return await dbContext.ParseResults
            .AsNoTracking()
            .Where(result => result.DocumentJobId == documentId)
            .Select(result => result.RawContextJson)
            .FirstOrDefaultAsync(ct);
    }

    /// <inheritdoc />
    public async Task MarkProcessingAsync(
        Guid documentId,
        CancellationToken ct)
    {
        var job = await dbContext.DocumentJobs
            .FirstOrDefaultAsync(j => j.Id == documentId, ct)
            ?? throw new InvalidOperationException($"Document job {documentId} not found.");

        job.Status = DocumentJobStatus.Processing;
        job.StartedAt ??= DateTime.UtcNow;
        await dbContext.SaveChangesAsync(ct);
    }

    /// <inheritdoc />
    public async Task CompleteAsync(
        Guid documentId,
        DocumentParseResult parseResult,
        CancellationToken ct)
    {
        var job = await dbContext.DocumentJobs
            .Include(j => j.ParseResult)
            .FirstOrDefaultAsync(j => j.Id == documentId, ct)
            ?? throw new InvalidOperationException($"Document job {documentId} not found.");

        job.Status = DocumentJobStatus.Completed;
        job.CompletedAt = DateTime.UtcNow;
        job.PdfKind = parseResult.Result.PdfKind;
        job.ErrorCode = null;
        job.ErrorMessage = null;

        if (job.ParseResult is null)
        {
            job.ParseResult = new ParseResultEntity
            {
                Id = Guid.NewGuid(),
                DocumentJobId = documentId,
                CreatedAt = DateTime.UtcNow
            };
            dbContext.ParseResults.Add(job.ParseResult);
        }

        job.ParseResult.ResultJson = parseResult.Result;
        job.ParseResult.RawContextJson = parseResult.RawContext;
        job.ParseResult.DetectedDocumentType = parseResult.Result.DetectedDocumentType;
        job.ParseResult.OverallConfidence = parseResult.Result.OverallConfidence;

        await dbContext.SaveChangesAsync(ct);
    }

    /// <inheritdoc />
    public async Task FailAsync(
        Guid documentId,
        string errorCode,
        string message,
        CancellationToken ct)
    {
        var job = await dbContext.DocumentJobs
            .FirstOrDefaultAsync(j => j.Id == documentId, ct)
            ?? throw new InvalidOperationException($"Document job {documentId} not found.");

        job.Status = DocumentJobStatus.Failed;
        job.CompletedAt = DateTime.UtcNow;
        job.ErrorCode = errorCode;
        job.ErrorMessage = message;
        await dbContext.SaveChangesAsync(ct);
    }

    /// <inheritdoc />
    public async Task RecoverPendingJobsAsync(CancellationToken ct)
    {
        var stuck = await dbContext.DocumentJobs
            .Where(job => job.Status == DocumentJobStatus.Processing)
            .ToListAsync(ct);

        foreach (var job in stuck)
        {
            job.Status = DocumentJobStatus.Pending;
            job.StartedAt = null;
        }

        if (stuck.Count > 0)
        {
            await dbContext.SaveChangesAsync(ct);
            logger.LogInformation("Recovered {Count} stuck Processing jobs to Pending.", stuck.Count);
        }

        var pending = await dbContext.DocumentJobs
            .Where(job => job.Status == DocumentJobStatus.Pending)
            .ToListAsync(ct);

        foreach (var job in pending)
        {
            if (!fileStorage.Exists(job.Id))
            {
                logger.LogWarning("Skip re-enqueue for job {DocumentId}: PDF file missing.", job.Id);
                continue;
            }

            await EnqueueJobAsync(job, ct);
        }

        logger.LogInformation("Re-enqueued {Count} Pending jobs.", pending.Count);
    }

    /// <inheritdoc />
    public async Task<ParseOptions?> GetParseOptionsAsync(
        Guid documentId,
        CancellationToken ct)
    {
        var job = await dbContext.DocumentJobs
            .AsNoTracking()
            .FirstOrDefaultAsync(j => j.Id == documentId, ct);

        if (job is null)
        {
            return null;
        }

        return new ParseOptions
        {
            DocumentType = job.DocumentType,
            Language = job.Language,
            Async = true
        };
    }

    private async Task<DocumentJobSubmitResult> RunSynchronouslyAsync(
        DocumentJobEntity job,
        SubmitDocumentJobRequest request,
        CancellationToken ct)
    {
        await MarkProcessingAsync(job.Id, ct);

        try
        {
            var parseResult = await parsePipeline.RunAsync(new DocumentParseRequest
            {
                DocumentId = job.Id,
                Content = request.Content,
                FileName = request.FileName,
                Options = request.Options
            }, ct);

            await CompleteAsync(job.Id, parseResult, ct);
            var completed = await GetAsync(job.Id, ct)
                ?? throw new InvalidOperationException("Completed job not found after sync parse.");

            return new DocumentJobSubmitResult
            {
                Response = completed,
                StatusCode = 200,
                FromCache = false
            };
        }
        catch (LlmException ex)
        {
            await FailAsync(job.Id, ex.ErrorCode, ex.Message, ct);
            throw;
        }
        catch (PdfValidationException ex)
        {
            await FailAsync(job.Id, ex.ErrorCode, ex.Message, ct);
            throw;
        }
        catch (Exception ex)
        {
            await FailAsync(job.Id, "pipeline-failed", ex.Message, ct);
            throw;
        }
    }

    private DocumentJobSubmitResult MapExistingJob(DocumentJobEntity job)
    {
        var response = MapJob(job);
        return job.Status switch
        {
            DocumentJobStatus.Completed => new DocumentJobSubmitResult
            {
                Response = response,
                StatusCode = 200,
                FromCache = true
            },
            DocumentJobStatus.Failed => new DocumentJobSubmitResult
            {
                Response = response,
                StatusCode = 200,
                FromCache = true
            },
            _ => new DocumentJobSubmitResult
            {
                Response = response,
                StatusCode = 202,
                FromCache = true
            }
        };
    }

    private async Task EnqueueJobAsync(
        DocumentJobEntity job,
        CancellationToken ct)
    {
        await jobQueue.EnqueueAsync(new DocumentJobWorkItem
        {
            DocumentId = job.Id,
            FilePath = fileStorage.GetAbsolutePath(job.Id)
        }, ct);
    }

    private static DocumentJobResponse MapJob(DocumentJobEntity job)
    {
        return new DocumentJobResponse
        {
            DocumentId = job.Id,
            Status = job.Status,
            CreatedAt = new DateTimeOffset(job.CreatedAt, TimeSpan.Zero),
            CompletedAt = job.CompletedAt.HasValue
                ? new DateTimeOffset(job.CompletedAt.Value, TimeSpan.Zero)
                : null,
            Result = job.ParseResult?.ResultJson,
            Error = job.Status == DocumentJobStatus.Failed
                ? new JobError
                {
                    Code = job.ErrorCode ?? "pipeline-failed",
                    Message = DocumentJobErrorSummary.Resolve(job.ErrorCode, job.Status)
                        ?? "Не удалось обработать документ"
                }
                : null
        };
    }

    private static string SanitizeFileName(string fileName)
    {
        var safeName = Path.GetFileName(fileName);
        return string.IsNullOrWhiteSpace(safeName) ? "document.pdf" : safeName;
    }
}