using JobTrackerApi.Data; using JobTrackerApi.Models; using JobTrackerApi.Services; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Xunit; namespace JobTrackerApi.Tests; public sealed class UserOperationStoreTests { [Fact] public async Task Create_is_idempotent_within_owner_and_isolated_between_owners() { await using var fixture = await Fixture.CreateAsync(); var request = Request("same-key"); await using var context = fixture.Context("user-1"); var store = fixture.Store(context); var first = await store.CreateAsync(request, default); var repeated = await store.CreateAsync(request, default); Assert.True(first.Created); Assert.False(repeated.Created); Assert.Equal(first.Operation.Id, repeated.Operation.Id); await using var otherContext = fixture.Context("user-2"); var other = await fixture.Store(otherContext).CreateAsync(request, default); Assert.True(other.Created); Assert.NotEqual(first.Operation.Id, other.Operation.Id); Assert.Single(await context.UserOperations.AsNoTracking().ToListAsync()); Assert.Single(await otherContext.UserOperations.AsNoTracking().ToListAsync()); } [Fact] public async Task Concurrent_workers_cannot_claim_the_same_operation() { await using var fixture = await Fixture.CreateAsync(); await using (var ownerContext = fixture.Context("user-1")) await fixture.Store(ownerContext).CreateAsync(Request("claim-once"), default); await using var context1 = fixture.Context(null); await using var context2 = fixture.Context(null); var claims = await Task.WhenAll( fixture.Store(context1).ClaimNextAsync(TimeSpan.FromSeconds(10), default), fixture.Store(context2).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); Assert.Single(claims, claim => claim is not null); Assert.Single(claims, claim => claim is null); } [Fact] public async Task Claim_filters_unknown_tasks_and_prioritises_interactive_work() { await using var fixture = await Fixture.CreateAsync(); await using (var ownerContext = fixture.Context("user-1")) { var store = fixture.Store(ownerContext); await store.CreateAsync(Request("scheduled") with { Priority = AiOperationPriorities.Scheduled }, default); await store.CreateAsync(Request("interactive") with { Priority = AiOperationPriorities.Interactive }, default); await store.CreateAsync(Request("unknown") with { TaskType = "unknown.ai", Priority = 999 }, default); } await using var neutralContext = fixture.Context(null); var lease = Assert.IsType(await fixture.Store(neutralContext) .ClaimNextAsync(TimeSpan.FromSeconds(10), default, new[] { "synthetic-test" })); Assert.Equal("interactive", (await neutralContext.UserOperations.IgnoreQueryFilters().AsNoTracking() .SingleAsync(operation => operation.Id == lease.OperationId)).IdempotencyKey); Assert.Equal("synthetic-test", lease.TaskType); } [Fact] public async Task Concurrent_duplicate_creation_returns_one_operation() { await using var fixture = await Fixture.CreateAsync(); await using var context1 = fixture.Context("user-1"); await using var context2 = fixture.Context("user-1"); var results = await Task.WhenAll( fixture.Store(context1).CreateAsync(Request("double-click"), default), fixture.Store(context2).CreateAsync(Request("double-click"), default)); Assert.Single(results, result => result.Created); Assert.Single(results, result => !result.Created); Assert.Equal(results[0].Operation.Id, results[1].Operation.Id); await using var neutralContext = fixture.Context(null); Assert.Single(await neutralContext.UserOperations.IgnoreQueryFilters().AsNoTracking().ToListAsync()); } [Fact] public async Task Expired_lease_retries_then_fails_after_bounded_attempts() { await using var fixture = await Fixture.CreateAsync(); Guid operationId; await using (var ownerContext = fixture.Context("user-1")) operationId = (await fixture.Store(ownerContext).CreateAsync(Request("lease", maxAttempts: 2), default)).Operation.Id; await using var neutralContext = fixture.Context(null); var store = fixture.Store(neutralContext); var first = Assert.IsType(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default)); Assert.Equal(operationId, first.OperationId); fixture.Time.Advance(TimeSpan.FromSeconds(6)); var second = Assert.IsType(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default)); Assert.Equal(operationId, second.OperationId); Assert.Equal(2, second.AttemptCount); fixture.Time.Advance(TimeSpan.FromSeconds(6)); Assert.Null(await store.ClaimNextAsync(TimeSpan.FromSeconds(5), default)); neutralContext.ChangeTracker.Clear(); var failed = await neutralContext.UserOperations.IgnoreQueryFilters().AsNoTracking().SingleAsync(); Assert.Equal(OperationStatuses.Failed, failed.Status); Assert.Equal("lease_expired", failed.FailureCategory); Assert.Equal("operation_failed", (await neutralContext.UserNotifications.IgnoreQueryFilters().AsNoTracking().SingleAsync()).Kind); } [Fact] public async Task Running_cancellation_is_acknowledged_or_recovered_after_expiry() { await using var fixture = await Fixture.CreateAsync(); Guid operationId; await using (var ownerContext = fixture.Context("user-1")) operationId = (await fixture.Store(ownerContext).CreateAsync(Request("cancel"), default)).Operation.Id; await using var neutralContext = fixture.Context(null); var lease = Assert.IsType(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(5), default)); await using (var ownerContext = fixture.Context("user-1")) Assert.True(await fixture.Store(ownerContext).RequestCancellationAsync(operationId, default)); await using (var otherOwner = fixture.Context("user-2")) Assert.Equal(0, await fixture.Store(otherOwner).AcknowledgeCancellationAsync(operationId, lease.LeaseToken, default)); fixture.Time.Advance(TimeSpan.FromSeconds(6)); Assert.Null(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(5), default)); neutralContext.ChangeTracker.Clear(); Assert.Equal(OperationStatuses.Cancelled, (await neutralContext.UserOperations.IgnoreQueryFilters().AsNoTracking().SingleAsync()).Status); Assert.Equal("operation_cancelled", (await neutralContext.UserNotifications.IgnoreQueryFilters().AsNoTracking().SingleAsync()).Kind); } [Fact] public async Task Failure_retry_completion_and_owner_guards_follow_the_state_machine() { await using var fixture = await Fixture.CreateAsync(); Guid operationId; await using (var ownerContext = fixture.Context("user-1")) operationId = (await fixture.Store(ownerContext).CreateAsync(Request("retry"), default)).Operation.Id; await using var neutralContext = fixture.Context(null); var lease = Assert.IsType(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); await Assert.ThrowsAsync(() => fixture.Store(neutralContext).CompleteAsync(operationId, lease.LeaseToken, null, default)); await using (var ownerContext = fixture.Context("user-1")) Assert.True(await fixture.Store(ownerContext).FailAsync(operationId, lease.LeaseToken, true, "temporary", "Please retry.", TimeSpan.FromSeconds(10), default)); Assert.Null(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); fixture.Time.Advance(TimeSpan.FromSeconds(11)); var retried = Assert.IsType(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); await using (var wrongOwner = fixture.Context("user-2")) Assert.Equal(0, await fixture.Store(wrongOwner).CompleteAsync(operationId, retried.LeaseToken, "result:wrong", default)); await using (var ownerContext = fixture.Context("user-1")) Assert.Equal(1, await fixture.Store(ownerContext).CompleteAsync(operationId, retried.LeaseToken, "result:ok", default)); await using (var ownerContext = fixture.Context("user-1")) Assert.Equal(0, await fixture.Store(ownerContext).CompleteAsync(operationId, retried.LeaseToken, "result:duplicate", default)); await using (var ownerContext = fixture.Context("user-1")) Assert.False(await fixture.Store(ownerContext).RequestCancellationAsync(operationId, default)); Assert.Equal("operation_succeeded", (await neutralContext.UserNotifications.IgnoreQueryFilters().AsNoTracking().SingleAsync()).Kind); } [Fact] public async Task Deadlines_and_input_bounds_fail_closed() { await using var fixture = await Fixture.CreateAsync(); await using var ownerContext = fixture.Context("user-1"); var store = fixture.Store(ownerContext); await Assert.ThrowsAsync(() => store.CreateAsync(Request("local-deadline") with { DeadlineAtUtc = DateTime.Now.AddMinutes(1) }, default)); await Assert.ThrowsAsync(() => store.CreateAsync(Request(new string('x', 129)), default)); var expired = await store.CreateAsync(Request("expired") with { DeadlineAtUtc = fixture.Time.GetUtcNow().UtcDateTime.AddSeconds(1) }, default); fixture.Time.Advance(TimeSpan.FromSeconds(2)); await using var neutralContext = fixture.Context(null); Assert.Null(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); var row = await neutralContext.UserOperations.IgnoreQueryFilters().AsNoTracking().SingleAsync(operation => operation.Id == expired.Operation.Id); Assert.Equal(OperationStatuses.Failed, row.Status); Assert.Equal("deadline_exceeded", row.FailureCategory); Assert.Equal("operation_failed", (await neutralContext.UserNotifications.IgnoreQueryFilters().AsNoTracking().SingleAsync()).Kind); } [Fact] public async Task Terminal_notification_is_single_and_owner_scoped_with_read_and_dismiss_state() { await using var fixture = await Fixture.CreateAsync(); Guid operationId; await using (var ownerContext = fixture.Context("user-1")) { var store = fixture.Store(ownerContext); operationId = (await store.CreateAsync(Request("notification"), default)).Operation.Id; Assert.True(await store.RequestCancellationAsync(operationId, default)); Assert.False(await store.RequestCancellationAsync(operationId, default)); var notifications = fixture.Notifications(ownerContext); var notification = Assert.Single(await notifications.ListAsync(10, default)); Assert.Equal(1, await notifications.UnreadCountAsync(default)); Assert.Equal(1, await notifications.MarkReadAsync(notification.Id, default)); Assert.Equal(0, await notifications.UnreadCountAsync(default)); Assert.Equal(1, await notifications.DismissAsync(notification.Id, default)); Assert.Empty(await notifications.ListAsync(10, default)); } await using var otherOwnerContext = fixture.Context("user-2"); var otherNotifications = fixture.Notifications(otherOwnerContext); Assert.Empty(await otherNotifications.ListAsync(10, default)); Assert.Equal(0, await otherNotifications.MarkReadAsync(Guid.NewGuid(), default)); } [Fact] public async Task Notification_write_failure_rolls_back_terminal_operation_state() { await using var fixture = await Fixture.CreateAsync(); Guid operationId; await using (var ownerContext = fixture.Context("user-1")) operationId = (await fixture.Store(ownerContext).CreateAsync(Request("rollback"), default)).Operation.Id; await using var neutralContext = fixture.Context(null); var lease = Assert.IsType(await fixture.Store(neutralContext).ClaimNextAsync(TimeSpan.FromSeconds(10), default)); var interceptor = new SaveFailureInterceptor { Fail = true }; await using (var ownerContext = fixture.Context("user-1", interceptor)) await Assert.ThrowsAsync(() => fixture.Store(ownerContext).CompleteAsync(operationId, lease.LeaseToken, null, default)); await using var verificationContext = fixture.Context(null); var operation = await verificationContext.UserOperations.IgnoreQueryFilters().AsNoTracking().SingleAsync(item => item.Id == operationId); Assert.Equal(OperationStatuses.Running, operation.Status); Assert.Empty(await verificationContext.UserNotifications.IgnoreQueryFilters().AsNoTracking().ToListAsync()); } private static CreateUserOperation Request(string key, int maxAttempts = 3) => new( "synthetic-test", key, "authorized", "local-only", "job", "42", MaxAttempts: maxAttempts); private sealed class MutableUser(string? userId) : ICurrentUserService { public string? UserId { get; set; } = userId; } private sealed class ManualTimeProvider(DateTimeOffset now) : TimeProvider { private DateTimeOffset _now = now; public override DateTimeOffset GetUtcNow() => _now; public void Advance(TimeSpan value) => _now = _now.Add(value); } private sealed class Fixture : IAsyncDisposable { private readonly string _root; private readonly string _connectionString; public ManualTimeProvider Time { get; } = new(new DateTimeOffset(2026, 8, 2, 12, 0, 0, TimeSpan.Zero)); private Fixture(string root, string connectionString) { _root = root; _connectionString = connectionString; } public static async Task CreateAsync() { var root = Path.Combine(Path.GetTempPath(), $"jobtracker-operation-store-{Guid.NewGuid():N}"); Directory.CreateDirectory(root); var fixture = new Fixture(root, $"Data Source={Path.Combine(root, "operations.db")};Default Timeout=5;Pooling=False"); await using var context = fixture.Context(null); await context.Database.EnsureCreatedAsync(); return fixture; } public JobTrackerContext Context(string? owner, SaveChangesInterceptor? interceptor = null) { var options = new DbContextOptionsBuilder().UseSqlite(_connectionString); if (interceptor is not null) options.AddInterceptors(interceptor); return new JobTrackerContext(options.Options, new MutableUser(owner)); } public UserOperationStore Store(JobTrackerContext context) => new(context, Time); public UserNotificationStore Notifications(JobTrackerContext context) => new(context, Time); public ValueTask DisposeAsync() { if (Directory.Exists(_root)) Directory.Delete(_root, true); return ValueTask.CompletedTask; } } private sealed class SaveFailureInterceptor : SaveChangesInterceptor { public bool Fail { get; init; } public override ValueTask> SavingChangesAsync( DbContextEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { if (Fail) throw new InvalidOperationException("Synthetic notification write failure."); return base.SavingChangesAsync(eventData, result, cancellationToken); } } }