jobtrackingapp/JobTrackerApi.MariaDb.Tests/CvImportLifecycleConcurrencyTests.cs

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 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<NoContentResult>(await Controller(discarding).DiscardRun(id));
        }
        finally { gate.Release.TrySetResult(); }
        Assert.IsType<ConflictObjectResult>(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<OkObjectResult>(await Controller(applying).ApplyReviewed(id, Review));
        }
        finally { gate.Release.TrySetResult(); }
        Assert.IsType<ConflictObjectResult>(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<AppPaths>();
        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<OkObjectResult>(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<OkObjectResult>(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<NoContentResult>(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<int> 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<ClaimsPrincipal>())).ReturnsAsync(user);
        users.Setup(u => u.UpdateAsync(user)).Returns(async () =>
        {
            await db.SaveChangesAsync();
            return IdentityResult.Success;
        });
        return new(users.Object, Mock.Of<ISummarizerService>(), 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<DbDataReader> 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;
        }
    }
}