using JobTrackerApi.Controllers; using JobTrackerApi.Data; using JobTrackerApi.Models; using Microsoft.EntityFrameworkCore; namespace JobTrackerApi.Services; public sealed record CvProcessingOutcome( bool Succeeded, string? FailureCategory = null, string? FailureMessage = null, bool Retryable = false, string? Provider = null, string? Model = null, string? RouteReason = null); public interface ICvProcessingQueue { Task EnqueueAsync(int runId, CancellationToken cancellationToken = default); } /// /// Compatibility name for the CV producer bridge. Durable scheduling and execution are owned by /// the shared AI operation queue; this type does not keep an in-memory CV queue. /// public sealed class CvProcessingQueue(AiOperationAdmission admission, JobTrackerContext db, IConfiguration configuration) : ICvProcessingQueue { public const string TaskType = "cv.process"; public const string SubjectType = "cv_extraction_run"; public async Task EnqueueAsync(int runId, CancellationToken cancellationToken = default) { var document = await db.CvExtractionRuns.AnyAsync(run => run.Id == runId && run.ParserVersion == ProfileCvController.DocumentParserVersion && run.Trigger == "document_import", cancellationToken); if (document && !configuration.GetValue("Workers:AiOperationsEnabled", false) && !configuration.GetValue("Workers:CvDocumentImportsEnabled", false)) throw new AiOperationAdmissionException("cv_worker_disabled", "CV processing is disabled on this server. Ask the administrator to enable document processing, then retry the upload.", StatusCodes.Status503ServiceUnavailable); return await admission.EnqueueAsync( TaskType, $"run:{runId}", SubjectType, runId.ToString(System.Globalization.CultureInfo.InvariantCulture), AiOperationPriorities.UserVisible, cancellationToken); } } public sealed class NoOpCvProcessingQueue : ICvProcessingQueue { public static readonly NoOpCvProcessingQueue Instance = new(); public Task EnqueueAsync(int runId, CancellationToken cancellationToken = default) => Task.FromResult(null); } public sealed class CvProcessingOperationHandler : IAiOperationHandler { public string TaskType => CvProcessingQueue.TaskType; public async Task ExecuteAsync( AiOperationExecutionContext context, IServiceProvider services, CancellationToken cancellationToken) { if (!string.Equals(context.Lease.SubjectType, CvProcessingQueue.SubjectType, StringComparison.Ordinal) || !int.TryParse(context.Lease.SubjectId, out var runId) || runId <= 0) { throw new AiOperationFailure("invalid_cv_run", "The CV processing operation has an invalid run reference.", retryable: false); } CvProcessingOutcome? outcome; try { outcome = await services.GetRequiredService() .ProcessQueuedRunAsync(runId, cancellationToken, context); } catch (OperationCanceledException) { var operation = await services.GetRequiredService() .GetAsync(context.Lease.OperationId, CancellationToken.None); await SetRunStatusAsync( services, runId, operation?.CancellationRequestedAtUtc is null ? "queued" : "cancelled", operation?.CancellationRequestedAtUtc is null ? "CV processing timed out and may be retried." : "CV processing was cancelled.", context.Lease); throw; } if (outcome is null) throw new AiOperationFailure("cv_run_not_found", "The CV processing run is no longer available.", retryable: false); if (!outcome.Succeeded) { if (outcome.Retryable) { var operation = await services.GetRequiredService() .GetAsync(context.Lease.OperationId, cancellationToken); var canRetry = operation is not null && operation.AttemptCount < operation.MaxAttempts && (operation.DeadlineAtUtc is null || operation.DeadlineAtUtc > DateTime.UtcNow); if (!canRetry) await SetRunStatusAsync(services, runId, "failed", outcome.FailureMessage ?? "CV processing failed.", context.Lease); } if (outcome.Provider is not null || outcome.Model is not null || outcome.RouteReason is not null) { throw new AiGenerationException( outcome.FailureCategory ?? "cv_processing_failed", outcome.FailureMessage ?? "CV processing failed.", outcome.Retryable, outcome.Provider, outcome.Model, outcome.RouteReason); } throw new AiOperationFailure( outcome.FailureCategory ?? "cv_processing_failed", outcome.FailureMessage ?? "CV processing failed.", outcome.Retryable); } var isDocumentImport = await services.GetRequiredService().CvExtractionRuns.Where(r => r.Id == runId).Select(r => r.ParserVersion).FirstAsync(cancellationToken) == ProfileCvController.DocumentParserVersion; return new AiOperationExecutionResult( $"/api/profile-cv/runs/{runId}/{(isDocumentImport ? "review" : "diff")}", outcome.Provider, outcome.Model, outcome.RouteReason ?? (isDocumentImport ? "cv_review_ready" : "cv_pipeline")); } private static Task SetRunStatusAsync(IServiceProvider services, int runId, string status, string message, UserOperationLease lease) { var completedAtUtc = status == "failed" || status == "cancelled" ? DateTimeOffset.UtcNow : (DateTimeOffset?)null; var db = services.GetRequiredService(); return db.CvExtractionRuns .Where(run => run.Id == runId && (run.Status == "running" || run.Status == "queued") && db.UserOperations.Any(o => o.Id == lease.OperationId && o.LeaseToken == lease.LeaseToken && o.Status == OperationStatuses.Running)) .ExecuteUpdateAsync(setters => setters .SetProperty(run => run.Status, status) .SetProperty(run => run.ErrorMessage, message) .SetProperty(run => run.CompletedAtUtc, completedAtUtc), CancellationToken.None); } }