cesnimda
72 lines
3.7 KiB
C#
72 lines
3.7 KiB
C#
using JobTrackerApi.Data;
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
namespace JobTrackerApi.Services;
|
|
|
|
public sealed class CvImportRetentionHostedService(IServiceScopeFactory scopes, AppPaths paths, ILogger<CvImportRetentionHostedService> logger) : BackgroundService
|
|
{
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
using var timer = new PeriodicTimer(TimeSpan.FromHours(1));
|
|
do
|
|
{
|
|
try
|
|
{
|
|
await using var scope = scopes.CreateAsyncScope();
|
|
await SweepAsync(scope.ServiceProvider.GetRequiredService<JobTrackerContext>(), paths, DateTimeOffset.UtcNow, stoppingToken);
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; }
|
|
catch (Exception) { logger.LogWarning("CV import retention could not complete; it will retry on the next sweep."); }
|
|
} while (await timer.WaitForNextTickAsync(stoppingToken));
|
|
}
|
|
|
|
public static async Task<int> SweepAsync(JobTrackerContext db, AppPaths paths, DateTimeOffset now, CancellationToken token)
|
|
{
|
|
var runs = await db.CvExtractionRuns.IgnoreQueryFilters().Include(r => r.Artifact)
|
|
.Where(r => r.ParserVersion == "docling-v1" && r.Artifact != null && r.Artifact.StoragePath != "")
|
|
.OrderBy(r => r.Id).Take(1000).ToListAsync(token);
|
|
var root = Path.GetFullPath(paths.CvArtifactsRoot) + Path.DirectorySeparatorChar;
|
|
var removed = 0;
|
|
foreach (var run in runs)
|
|
{
|
|
token.ThrowIfCancellationRequested();
|
|
if (run.StartedAtUtc > now.AddDays(-7) && (!(run.Status is "cancelled" or "failed" or "discarded") || run.StartedAtUtc > now.AddDays(-1))) continue;
|
|
var artifact = run.Artifact!;
|
|
if (artifact.OwnerUserId != run.OwnerUserId) continue;
|
|
var path = Path.GetFullPath(artifact.StoragePath);
|
|
if (!path.StartsWith(root, OperatingSystem.IsWindows() ? StringComparison.OrdinalIgnoreCase : StringComparison.Ordinal)) continue;
|
|
// Revoke the operation before deleting its source; a late worker cannot publish a result.
|
|
var subject = run.Id.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
|
var operations = await db.UserOperations.IgnoreQueryFilters().Where(o => o.OwnerUserId == run.OwnerUserId && o.SubjectType == CvProcessingQueue.SubjectType && o.SubjectId == subject).ToListAsync(token);
|
|
foreach (var operation in operations.Where(o => !Models.OperationStatuses.IsTerminal(o.Status)))
|
|
{
|
|
operation.Status = Models.OperationStatuses.Cancelled;
|
|
operation.CancellationRequestedAtUtc = now.UtcDateTime;
|
|
operation.CompletedAtUtc = now.UtcDateTime;
|
|
}
|
|
await db.SaveChangesAsync(token);
|
|
try
|
|
{
|
|
if (File.Exists(path) && (File.GetAttributes(path) & FileAttributes.ReparsePoint) != 0) continue;
|
|
File.Delete(path);
|
|
}
|
|
catch (IOException) { continue; }
|
|
catch (UnauthorizedAccessException) { continue; }
|
|
artifact.StoragePath = "";
|
|
run.DocumentJson = null;
|
|
run.RawExtractedText = null;
|
|
run.NormalizedText = null;
|
|
if (run.Status != "applied")
|
|
{
|
|
run.StructuredProfileJson = null;
|
|
if (run.Status is "pending_review" or "queued" or "running") run.Status = "expired";
|
|
run.CompletedAtUtc ??= now;
|
|
run.ErrorMessage = "The source document expired. Upload it again to start a new review.";
|
|
}
|
|
removed++;
|
|
}
|
|
await db.SaveChangesAsync(token);
|
|
return removed;
|
|
}
|
|
}
|