using System.Data.Common; using System.Security.Claims; using JobTrackerApi.Controllers; using JobTrackerApi.Data; using JobTrackerApi.Models; using JobTrackerApi.Services; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Identity; using Microsoft.AspNetCore.Mvc; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Moq; using Xunit; namespace JobTrackerApi.MariaDb.Tests; public class CvImportLifecycleConcurrencyTests { private const string Owner = "rd21-owner"; private static readonly ProfileCvController.ApplyReviewedRequest Review = new(0, new[] { new CvImportDecision("skills.0", "use") }); [Fact] public async Task Document_only_claim_is_owner_bound_and_translates_on_MariaDb_without_restricting_general_claims() { await using var database = await MariaDbDatabase.CreateAsync(); Guid documentId, legacyId, strategyId, mismatchedId, wrongTriggerId; await using (var seed = database.Context()) { seed.Users.AddRange(new ApplicationUser { Id = Owner, UserName = "claim-owner" }, new ApplicationUser { Id = "other-owner", UserName = "claim-other" }); var document = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "document_import", Status = "queued" }; var foreign = new CvExtractionRun { OwnerUserId = "other-owner", ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "document_import", Status = "queued" }; var legacy = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = "legacy-text", Trigger = "document_import", Status = "queued" }; var wrongTrigger = new CvExtractionRun { OwnerUserId = Owner, ParserVersion = ProfileCvController.DocumentParserVersion, Trigger = "text_extraction", Status = "queued" }; seed.CvExtractionRuns.AddRange(document, foreign, legacy, wrongTrigger); await seed.SaveChangesAsync(); UserOperation Queued(CvExtractionRun run, int priority) { var operation = Operation(run.Id, OperationStatuses.Queued); operation.IdempotencyKey = operation.Id.ToString("N"); operation.LeaseToken = null; operation.Priority = priority; return operation; } var eligible = Queued(document, 100); var legacyOperation = Queued(legacy, 1000); var mismatched = Queued(foreign, 900); var wrongTriggerOperation = Queued(wrongTrigger, 800); var strategy = Queued(document, 1100); strategy.TaskType = StrategySnapshotService.TaskType; strategy.SubjectType = "job_application"; seed.UserOperations.AddRange(eligible, legacyOperation, mismatched, wrongTriggerOperation, strategy); await seed.SaveChangesAsync(); (documentId, legacyId, strategyId, mismatchedId, wrongTriggerId) = (eligible.Id, legacyOperation.Id, strategy.Id, mismatched.Id, wrongTriggerOperation.Id); } await using var neutral = database.Context(null!); var store = new UserOperationStore(neutral, TimeProvider.System); var claimed = Assert.IsType(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, documentImportsOnly: true)); Assert.Equal(documentId, claimed.OperationId); Assert.Equal(Owner, claimed.OwnerUserId); Assert.Null(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, documentImportsOnly: true)); var generalCv = Assert.IsType(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, new[] { CvProcessingQueue.TaskType })); Assert.Equal(legacyId, generalCv.OperationId); var generalOther = Assert.IsType(await store.ClaimNextAsync(TimeSpan.FromMinutes(1), default, new[] { StrategySnapshotService.TaskType })); Assert.Equal(strategyId, generalOther.OperationId); var excluded = await neutral.UserOperations.IgnoreQueryFilters().AsNoTracking() .Where(operation => operation.Id == mismatchedId || operation.Id == wrongTriggerId).ToListAsync(); Assert.Equal(2, excluded.Count); Assert.All(excluded, operation => { Assert.Equal(OperationStatuses.Queued, operation.Status); Assert.Equal(0, operation.AttemptCount); Assert.Null(operation.LeaseToken); }); } [Fact] public async Task Acknowledged_discard_prevents_an_apply_that_already_read_pending() { await using var database = await MariaDbDatabase.CreateAsync(); var id = await Seed(database); var gate = new RunReadGate(); await using var applying = database.Context(Owner, gate); var apply = Controller(applying).ApplyReviewed(id, Review); try { await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); await using var discarding = database.Context(); Assert.IsType(await Controller(discarding).DiscardRun(id)); } finally { gate.Release.TrySetResult(); } Assert.IsType(await apply.WaitAsync(TimeSpan.FromSeconds(15))); await applying.SaveChangesAsync(); await using var verify = database.Context(); Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status); Assert.Empty(await verify.CareerProfileVersions.ToListAsync()); Assert.Empty(await verify.CareerSkills.ToListAsync()); Assert.Null((await verify.Users.SingleAsync()).CurrentCvExtractionRunId); } [Fact] public async Task Applied_review_prevents_a_discard_that_already_read_pending() { await using var database = await MariaDbDatabase.CreateAsync(); var id = await Seed(database); var gate = new RunReadGate(); await using var discarding = database.Context(Owner, gate); var discard = Controller(discarding).DiscardRun(id); try { await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); await using var applying = database.Context(); Assert.IsType(await Controller(applying).ApplyReviewed(id, Review)); } finally { gate.Release.TrySetResult(); } Assert.IsType(await discard.WaitAsync(TimeSpan.FromSeconds(15))); await using var verify = database.Context(); Assert.Equal("applied", (await verify.CvExtractionRuns.SingleAsync()).Status); Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name); Assert.Single(await verify.CareerProfileVersions.ToListAsync()); } [Fact] public async Task Retention_with_a_stale_pending_snapshot_preserves_applied_result_and_replay() { await using var database = await MariaDbDatabase.CreateAsync(); var paths = database.App.Services.GetRequiredService(); var source = Path.Combine(paths.CvArtifactsRoot, "lifecycle-synthetic.pdf"); await File.WriteAllTextAsync(source, "Synthetic source"); var id = await Seed(database, source); var gate = new RunReadGate(); await using var sweeping = database.Context(Owner, gate); var sweep = CvImportRetentionHostedService.SweepAsync(sweeping, paths, DateTimeOffset.UtcNow, default); string? candidate = null; string? receipt = null; try { await gate.Entered.Task.WaitAsync(TimeSpan.FromSeconds(15)); await using var applying = database.Context(); Assert.IsType(await Controller(applying).ApplyReviewed(id, Review)); var saved = await applying.CvExtractionRuns.AsNoTracking().SingleAsync(); candidate = saved.StructuredProfileJson; receipt = saved.ReviewJson; } finally { gate.Release.TrySetResult(); } await sweep.WaitAsync(TimeSpan.FromSeconds(15)); await using var verify = database.Context(); var run = await verify.CvExtractionRuns.SingleAsync(); Assert.Equal("applied", run.Status); Assert.Equal(candidate, run.StructuredProfileJson); Assert.Equal(receipt, run.ReviewJson); Assert.False(File.Exists(source)); Assert.Equal("", (await verify.CvUploadArtifacts.SingleAsync()).StoragePath); Assert.IsType(await Controller(verify).ApplyReviewed(id, Review)); Assert.Single(await verify.CareerProfileVersions.ToListAsync()); Assert.Equal("C#", (await verify.CareerSkills.SingleAsync()).Name); } [Fact] public async Task Cancellation_acknowledgement_preserves_a_discarded_review() { await using var database = await MariaDbDatabase.CreateAsync(); var id = await Seed(database); await using var db = database.Context(); var operation = Operation(id, OperationStatuses.Running); operation.CancellationRequestedAtUtc = DateTime.UtcNow; db.UserOperations.Add(operation); await db.SaveChangesAsync(); Assert.IsType(await Controller(db).DiscardRun(id)); var store = new UserOperationStore(db, TimeProvider.System); Assert.Equal(1, await store.AcknowledgeCancellationAsync(operation.Id, "lease", default)); await using var verify = database.Context(); Assert.Equal("discarded", (await verify.CvExtractionRuns.SingleAsync()).Status); Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status); } [Theory] [InlineData("discarded")] [InlineData("expired")] [InlineData("applied")] public async Task Retry_cannot_reopen_a_terminal_review(string status) { await using var database = await MariaDbDatabase.CreateAsync(); var id = await Seed(database); await using var db = database.Context(); var run = await db.CvExtractionRuns.SingleAsync(); run.Status = status; var operation = Operation(id, OperationStatuses.Cancelled); db.UserOperations.Add(operation); await db.SaveChangesAsync(); Assert.False(await new UserOperationStore(db, TimeProvider.System).RetryAsync(operation.Id, default)); await using var verify = database.Context(); Assert.Equal(status, (await verify.CvExtractionRuns.SingleAsync()).Status); Assert.Equal(OperationStatuses.Cancelled, (await verify.UserOperations.SingleAsync()).Status); } private static UserOperation Operation(int runId, string status) => new() { Id = Guid.NewGuid(), OwnerUserId = Owner, TaskType = CvProcessingQueue.TaskType, SubjectType = CvProcessingQueue.SubjectType, SubjectId = runId.ToString(System.Globalization.CultureInfo.InvariantCulture), IdempotencyKey = "lifecycle", Status = status, LeaseToken = "lease", CreatedAtUtc = DateTime.UtcNow, AvailableAtUtc = DateTime.UtcNow, }; private static async Task Seed(MariaDbDatabase database, string? source = null) { await using var db = database.Context(); db.Users.Add(new() { Id = Owner, UserName = "lifecycle-synthetic" }); var run = new CvExtractionRun { OwnerUserId = Owner, Trigger = "document_import", ParserVersion = ProfileCvController.DocumentParserVersion, Status = "pending_review", StartedAtUtc = DateTimeOffset.UtcNow.AddDays(-8), RawExtractedText = "C#", DocumentJson = "{}", StructuredProfileJson = StructuredCvProfileJson.SerializePersisted(new() { Skills = { "C#" } }), Artifact = source is null ? null : new() { OwnerUserId = Owner, StoragePath = source, OriginalFileName = "synthetic.pdf" }, }; db.CvExtractionRuns.Add(run); await db.SaveChangesAsync(); return run.Id; } private static ProfileCvController Controller(JobTrackerContext db) { var user = db.Users.Single(u => u.Id == Owner); var users = TestHostFactory.CreateUserManager(); users.Setup(u => u.GetUserAsync(It.IsAny())).ReturnsAsync(user); users.Setup(u => u.UpdateAsync(user)).Returns(async () => { await db.SaveChangesAsync(); return IdentityResult.Success; }); return new(users.Object, Mock.Of(), db, null!) { ControllerContext = new() { HttpContext = new DefaultHttpContext() } }; } // Pause after the server has executed the SELECT, so the result contains the old row. private sealed class RunReadGate : DbCommandInterceptor { public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); private int _entered; public override async ValueTask ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default) { if (command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase) && command.CommandText.Contains("`CvExtractionRuns`", StringComparison.Ordinal) && Interlocked.Exchange(ref _entered, 1) == 0) { Entered.TrySetResult(); await Release.Task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken); } return result; } } }