jobtrackingapp/JobTrackerApi.MariaDb.Tests/UserOperationHistoryPersistenceTests.cs

using JobTrackerApi.Models;
using JobTrackerApi.Controllers;
using JobTrackerApi.Services;
using Microsoft.AspNetCore.Mvc;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Xunit;

namespace JobTrackerApi.MariaDb.Tests;

public sealed class UserOperationHistoryPersistenceTests
{
    [Fact]
    public async Task Administrator_metadata_grouping_paging_associations_and_string_cursors_use_real_MariaDB_SQL()
    {
        await using var database = await MariaDbDatabase.CreateAsync();
        await using var db = database.Context();
        db.Users.Add(new() { Id = "rd21-owner", UserName = "synthetic-history", Email = "history@example.test" });
        var artifact = new CvUploadArtifact { OwnerUserId = "rd21-owner", OriginalFileName = "synthetic.pdf", MimeType = "application/pdf", ByteSize = 400 };
        db.CvUploadArtifacts.Add(artifact);
        await db.SaveChangesAsync();
        var run = new CvExtractionRun { OwnerUserId = "rd21-owner", ArtifactId = artifact.Id, Trigger = "document_import", ParserVersion = "docling-v1", DocumentJson = "{\"PageCount\":2,\"Text\":\"SECRET_SOURCE\"}" };
        db.CvExtractionRuns.Add(run);
        await db.SaveChangesAsync();
        var operation = (await new UserOperationStore(db, TimeProvider.System).CreateAsync(Request("admin-query") with { SubjectType = CvProcessingQueue.SubjectType, SubjectId = run.Id.ToString() }, default)).Operation;
        var controller = new AdminProcessingController(db, new ConfigurationBuilder().Build(), TimeProvider.System);
        var listing = Assert.IsType<AdminProcessingController.PageDto>(Assert.IsType<OkObjectResult>((await controller.List(owner: "rd21-owner", pageSize: 1)).Result).Value);
        Assert.Equal(1, listing.Total);
        Assert.Equal(1, listing.Counts[OperationStatuses.Queued]);
        Assert.Equal("synthetic.pdf", Assert.Single(listing.Items).Document!.Name);
        var detail = Assert.IsType<AdminProcessingController.ItemDto>(Assert.IsType<OkObjectResult>((await controller.Detail(operation.Id, default)).Result).Value);
        Assert.Equal(2, detail.Document!.Pages);
        Assert.DoesNotContain("SECRET", System.Text.Json.JsonSerializer.Serialize(detail));
        var events = Assert.IsType<AdminProcessingController.EventPageDto>(Assert.IsType<OkObjectResult>((await controller.Events(operation.Id)).Result).Value);
        var item = Assert.Single(events.Items);
        Assert.True(long.TryParse(item.Id, out _));
        Assert.Equal(DateTimeKind.Utc, item.AtUtc.Kind);
    }

    [Fact]
    public async Task Admin_diagnostic_filters_translate_and_preserve_owner_joins_in_MariaDB()
    {
        await using var database = await MariaDbDatabase.CreateAsync();
        await using var db = database.Context();
        db.Users.Add(new() { Id = "rd21-owner", UserName = "diagnostic-owner", Email = "diagnostic@example.test" });
        db.Users.Add(new() { Id = "foreign", UserName = "diagnostic-foreign", Email = "foreign@example.test" });
        var company = new Company { OwnerUserId = "rd21-owner", Name = "Diagnostic company" };
        var job = new JobApplication { OwnerUserId = "rd21-owner", Company = company, JobTitle = "Diagnostic engineer", Status = "Applied" };
        var artifact = new CvUploadArtifact { OwnerUserId = "rd21-owner", OriginalFileName = "synthetic.pdf", MimeType = "application/pdf", ByteSize = 400 };
        db.JobApplications.Add(job); db.CvUploadArtifacts.Add(artifact);
        await db.SaveChangesAsync();
        var run = new CvExtractionRun { OwnerUserId = "rd21-owner", ArtifactId = artifact.Id, Trigger = "document_import", ParserVersion = ProfileCvController.DocumentParserVersion };
        db.CvExtractionRuns.Add(run); await db.SaveChangesAsync();
        var store = new UserOperationStore(db, TimeProvider.System);
        await store.CreateAsync(Request("diagnostic-document") with { SubjectType = CvProcessingQueue.SubjectType, SubjectId = run.Id.ToString() }, default);
        var strategy = (await store.CreateAsync(Request("diagnostic-strategy") with { TaskType = StrategySnapshotService.TaskType, SubjectType = "job_strategy", SubjectId = StrategySnapshotService.EncodeSubject(job.Id, []) }, default)).Operation;
        strategy.CreatedAtUtc = new DateTime(2026, 10, 15, 12, 0, 0, DateTimeKind.Utc);
        strategy.FailureCategory = "timeout";
        var failedAt = DateTime.UtcNow.AddMinutes(-5);
        db.UserOperationEvents.Add(new() { OperationId = strategy.Id, OwnerUserId = strategy.OwnerUserId, Kind = "RetryScheduled", Status = OperationStatuses.WaitingForRetry, AtUtc = failedAt, Stage = "cv_extracting", FailureCode = "timeout" });
        db.UserOperations.Add(new() { Id = Guid.NewGuid(), OwnerUserId = "foreign", IdempotencyKey = "foreign-diagnostic", TaskType = StrategySnapshotService.TaskType, SubjectType = "job_strategy", SubjectId = StrategySnapshotService.EncodeSubject(job.Id, []), Status = OperationStatuses.Queued, CreatedAtUtc = DateTime.UtcNow });
        await db.SaveChangesAsync();
        var controller = new AdminProcessingController(db, new ConfigurationBuilder().Build(), TimeProvider.System);
        var results = new[]
        {
            await controller.List(jobId: job.Id), await controller.List(q: "Diagnostic engineer"),
            await controller.List(worker: "document"), await controller.List(failureCode: "timeout"),
            await controller.List(createdBefore: new DateTimeOffset(2026, 10, 16, 0, 0, 0, TimeSpan.Zero)),
        };
        foreach (var result in results)
        {
            var page = Assert.IsType<AdminProcessingController.PageDto>(Assert.IsType<OkObjectResult>(result.Result).Value);
            Assert.Equal(1, page.Total);
            Assert.Equal("rd21-owner", Assert.Single(page.Items).Owner.Id);
        }
        var detail = Assert.IsType<AdminProcessingController.ItemDto>(Assert.IsType<OkObjectResult>((await controller.Detail(strategy.Id, default)).Result).Value);
        Assert.Equal(failedAt, Assert.IsType<DateTime>(detail.LastFailureAtUtc), TimeSpan.FromMilliseconds(1));
        Assert.Equal("cv_extracting", detail.LastFailureStage);
    }

    private static CreateUserOperation Request(string key) => new("cv.process", key, "authorized", "local_only");

    [Fact]
    public async Task Concurrent_claim_and_repeated_stage_have_one_durable_event_each()
    {
        await using var database = await MariaDbDatabase.CreateAsync();
        await using (var owner = database.Context())
            await new UserOperationStore(owner, TimeProvider.System).CreateAsync(Request("history-concurrency"), default);
        await using var first = database.Context(null!);
        await using var second = database.Context(null!);
        var claims = await Task.WhenAll(new UserOperationStore(first, TimeProvider.System).ClaimNextAsync(TimeSpan.FromSeconds(30), default),
            new UserOperationStore(second, TimeProvider.System).ClaimNextAsync(TimeSpan.FromSeconds(30), default));
        var lease = Assert.Single(claims.OfType<UserOperationLease>());
        await using var ownerOne = database.Context();
        await using var ownerTwo = database.Context();
        var stages = await Task.WhenAll(new UserOperationStore(ownerOne, TimeProvider.System).ReportStageAsync(lease.OperationId, lease.LeaseToken, "cv_extracting", default),
            new UserOperationStore(ownerTwo, TimeProvider.System).ReportStageAsync(lease.OperationId, lease.LeaseToken, "cv_extracting", default));
        Assert.All(stages, count => Assert.Equal(1, count));
        await using var readback = database.Context();
        var events = await readback.UserOperationEvents.OrderBy(item => item.Id).ToListAsync();
        Assert.Equal(new[] { "Created", "Claimed", "StageChanged" }, events.Select(item => item.Kind));
        Assert.All(events, item => Assert.Equal("rd21-owner", item.OwnerUserId));
        await using var other = database.Context("other");
        Assert.Empty(await other.UserOperationEvents.ToListAsync());
    }

    [Fact]
    public async Task Event_insert_failure_rolls_back_completion_and_notifications_in_MariaDB()
    {
        await using var database = await MariaDbDatabase.CreateAsync();
        await using var owner = database.Context();
        var store = new UserOperationStore(owner, TimeProvider.System);
        var id = (await store.CreateAsync(Request("history-rollback"), default)).Operation.Id;
        await using var worker = database.Context(null!);
        var lease = Assert.IsType<UserOperationLease>(await new UserOperationStore(worker, TimeProvider.System).ClaimNextAsync(TimeSpan.FromSeconds(30), default));
        await owner.Database.ExecuteSqlRawAsync("CREATE TRIGGER fail_history BEFORE INSERT ON UserOperationEvents FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'synthetic event insert failure';");
        await Assert.ThrowsAnyAsync<Exception>(() => store.CompleteAsync(id, lease.LeaseToken, null, default));
        await using var readback = database.Context();
        Assert.Equal(OperationStatuses.Running, (await readback.UserOperations.SingleAsync()).Status);
        Assert.Equal(2, await readback.UserOperationEvents.CountAsync());
        Assert.Empty(await readback.UserNotifications.ToListAsync());
    }
}