KArtSell.Aegis/tests/KArtSell.Integration.Tests/Scheduling/ModelScheduleCasTests.cs

using Dapper;
using KArtSell.BuildingBlocks.Data;
using KArtSell.Modules.ModelOperations.Infrastructure;
using Npgsql;
using Xunit;

namespace KArtSell.Integration.Tests.Scheduling;

[Collection("Database")]
public sealed class ModelScheduleCasTests : IAsyncLifetime
{
    private readonly NpgsqlDataSource _dataSource = new NpgsqlDataSourceBuilder(TestDatabaseConnection.GetConnectionString()).Build();

    public async Task InitializeAsync()
    {
        await using var connection = await _dataSource.OpenConnectionAsync();
        await connection.ExecuteAsync("select 1");
    }

    public Task DisposeAsync() => _dataSource.DisposeAsync().AsTask();

    [Fact]
    public async Task Stale_dispatch_revision_cannot_advance_a_reacquired_schedule()
    {
        var scheduleId = Guid.NewGuid();
        var scheduledFor = new DateTimeOffset(2026, 8, 9, 0, 0, 0, TimeSpan.Zero);
        const string firstOwner = "cas-test:first";
        const string secondOwner = "cas-test:second";
        var repository = new DapperModelScheduleRepository(new NpgsqlConnectionFactory(_dataSource));

        try
        {
            await using (var connection = await _dataSource.OpenConnectionAsync())
            {
                await connection.ExecuteAsync("""
                    insert into evaluation.model_operation_schedule
                    (schedule_id, operation_code, operation_name, scope_key, cadence, automation_mode,
                     queue_name, schedule_version, enabled, next_due_at, max_lag, primary_owner, secondary_owner)
                    values
                    (@ScheduleId, 'J31', 'CasTest', @ScopeKey, 'DAILY', 'EVALUATION_ONLY',
                     'q-control', 1, true, @ScheduledFor, interval '1 hour', 'BE', 'QA');
                    """, new { ScheduleId = scheduleId, ScopeKey = $"CAS-{scheduleId:N}", ScheduledFor = scheduledFor });
            }

            var first = Assert.Single(await repository.AcquireDueAsync(scheduledFor.AddMinutes(1), firstOwner, TimeSpan.FromMinutes(5), 1, CancellationToken.None));

            await using (var connection = await _dataSource.OpenConnectionAsync())
            {
                await connection.ExecuteAsync("""
                    update evaluation.model_operation_schedule
                    set lease_until = @ExpiredAt
                    where schedule_id = @ScheduleId;
                    """, new { ScheduleId = scheduleId, ExpiredAt = scheduledFor });
            }

            var second = Assert.Single(await repository.AcquireDueAsync(scheduledFor.AddMinutes(2), secondOwner, TimeSpan.FromMinutes(5), 1, CancellationToken.None));
            Assert.True(second.DispatchRevision > first.DispatchRevision);

            await Assert.ThrowsAsync<InvalidOperationException>(() => repository.MarkDispatchedAsync(
                scheduleId, firstOwner, first.DispatchRevision, "stale-job", scheduledFor.AddMinutes(3), scheduledFor.AddDays(1), CancellationToken.None));

            await using var verify = await _dataSource.OpenConnectionAsync();
            var row = await verify.QuerySingleAsync<(DateTimeOffset NextDueAt, string LeaseOwner, int DispatchRevision)>("""
                select next_due_at as NextDueAt, lease_owner as LeaseOwner, dispatch_revision as DispatchRevision
                from evaluation.model_operation_schedule
                where schedule_id = @ScheduleId;
                """, new { ScheduleId = scheduleId });
            Assert.Equal(scheduledFor, row.NextDueAt);
            Assert.Equal(secondOwner, row.LeaseOwner);
            Assert.Equal(second.DispatchRevision, row.DispatchRevision);
        }
        finally
        {
            await using var cleanup = await _dataSource.OpenConnectionAsync();
            await cleanup.ExecuteAsync("delete from evaluation.model_operation_schedule where schedule_id = @ScheduleId", new { ScheduleId = scheduleId });
        }
    }
}