jobtrackingapp/JobTrackerApi/Services/CvImportRetentionHostedService.cs

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)
    {
        await OperationEventLog.PruneAsync(db, now.UtcDateTime, 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;
                db.UserOperationEvents.Add(OperationEventLog.Create(operation, Models.OperationEventKind.Cancelled, 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 = "";
            for (var attempt = 0; ; attempt++)
            {
                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.";
                }
                try { await db.SaveChangesAsync(token); break; }
                catch (DbUpdateConcurrencyException ex) when (attempt < 2 && ex.Entries.All(e => ReferenceEquals(e.Entity, run)))
                {
                    // Retry source cleanup against the new status; never replay canonical profile writes.
                    await db.Entry(run).ReloadAsync(token);
                    if (db.Entry(run).State == EntityState.Detached) break;
                }
            }
            removed++;
        }
        await db.SaveChangesAsync(token);
        var registered = (await db.CvUploadArtifacts.IgnoreQueryFilters().AsNoTracking().Select(a => a.StoragePath).ToListAsync(token))
            .Where(p => !string.IsNullOrWhiteSpace(p)).Select(Path.GetFullPath).ToHashSet(OperatingSystem.IsWindows() ? StringComparer.OrdinalIgnoreCase : StringComparer.Ordinal);
        foreach (var directory in Directory.EnumerateDirectories(paths.CvArtifactsRoot).Take(1000))
        {
            if ((File.GetAttributes(directory) & FileAttributes.ReparsePoint) != 0) continue;
            foreach (var file in Directory.EnumerateFiles(directory).Take(1000))
            {
                if (registered.Contains(Path.GetFullPath(file)) || !System.Text.RegularExpressions.Regex.IsMatch(Path.GetFileName(file), @"^\d{14}-[a-f0-9]{32}\.(pdf|docx)$", System.Text.RegularExpressions.RegexOptions.IgnoreCase)
                    || File.GetLastWriteTimeUtc(file) > now.AddDays(-7).UtcDateTime || (File.GetAttributes(file) & FileAttributes.ReparsePoint) != 0) continue;
                try { File.Delete(file); removed++; }
                catch (IOException) { }
                catch (UnauthorizedAccessException) { }
            }
        }
        return removed + await AvatarStorage.SweepOrphansAsync(db, paths.DataRoot, now, token);
    }
}