using ItWorx.KnowledgeHub.Application.Identity;
using ItWorx.KnowledgeHub.Application.Jobs;
using ItWorx.KnowledgeHub.Application.Jobs.UseCases;
using ItWorx.KnowledgeHub.Domain.Jobs;
using ItWorx.KnowledgeHub.Infrastructure.Persistence;
using ItWorx.KnowledgeHub.IntegrationTests.Infrastructure;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;

namespace ItWorx.KnowledgeHub.IntegrationTests.Jobs;

[Collection(PostgresTestGroup.Name)]
public sealed class BackgroundJobConsistencyTests(PostgresContainerFixture postgres)
    : OwnerSetupTestBase(postgres, new Dictionary<string, string?> { ["KnowledgeHub:Jobs:Enabled"] = "false" })
{
    [Fact]
    public async Task Queue_and_cancel_audit_failures_roll_back_job_mutations_and_retries_are_idempotent()
    {
        var (workspaceId, actorId) = await GetOwnerContextAsync();
        const string key = "audit-atomic-background-job";

        using (var scope = Factory.Services.CreateScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
            await InstallAuditFailureTriggerAsync(db, "job.queued");
        }

        try
        {
            await Assert.ThrowsAsync<DbUpdateException>(async () =>
            {
                using var scope = Factory.Services.CreateScope();
                await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
                    .QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
            });
        }
        finally
        {
            await RemoveAuditFailureTriggerAsync();
        }

        using (var scope = Factory.Services.CreateScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
            Assert.Empty(await db.BackgroundJobs.AsNoTracking().ToListAsync());
            Assert.False(await db.AuditEvents.AsNoTracking().AnyAsync(item => item.Action == "job.queued"));
        }

        EnqueueJobResult created;
        using (var scope = Factory.Services.CreateScope())
        {
            created = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
                .QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
        }
        Assert.Equal(EnqueueJobStatus.Created, created.Status);
        Assert.NotNull(created.Job);

        using (var scope = Factory.Services.CreateScope())
        {
            var replay = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
                .QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, key, "audit-atomic"), CancellationToken.None);
            Assert.Equal(EnqueueJobStatus.Existing, replay.Status);
            Assert.Equal(created.Job.Id, replay.Job?.Id);
        }

        var queuedVersion = created.Job.Version;
        using (var scope = Factory.Services.CreateScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
            await InstallAuditFailureTriggerAsync(db, "job.cancel_requested");
        }

        try
        {
            await Assert.ThrowsAsync<DbUpdateException>(async () =>
            {
                using var scope = Factory.Services.CreateScope();
                await scope.ServiceProvider.GetRequiredService<CancelBackgroundJobUseCase>()
                    .ExecuteAsync(workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);
            });
        }
        finally
        {
            await RemoveAuditFailureTriggerAsync();
        }

        using (var scope = Factory.Services.CreateScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
            var persisted = await db.BackgroundJobs.AsNoTracking().SingleAsync(job => job.Id == created.Job.Id);
            Assert.Equal(BackgroundJobStatus.Queued, persisted.Status);
            Assert.Null(persisted.CancelRequestedAt);
            Assert.Equal(queuedVersion, persisted.Version);
            Assert.False(await db.AuditEvents.AsNoTracking().AnyAsync(item => item.Action == "job.cancel_requested"));
        }

        var lease = Guid.NewGuid();
        var now = DateTimeOffset.UtcNow;
        using (var scope = Factory.Services.CreateScope())
        {
            var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
            Assert.Equal(created.Job.Id, await repository.TryClaimNextAsync(
                lease, now, now.AddMinutes(2), CancellationToken.None));
        }

        using (var scope = Factory.Services.CreateScope())
        {
            var cancellation = scope.ServiceProvider.GetRequiredService<CancelBackgroundJobUseCase>();
            var first = await cancellation.ExecuteAsync(
                workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);
            var second = await cancellation.ExecuteAsync(
                workspaceId, created.Job.Id, actorId, "audit-atomic", CancellationToken.None);

            Assert.Equal(CancelBackgroundJobStatus.Requested, first.Status);
            Assert.Equal(CancelBackgroundJobStatus.Requested, second.Status);
            Assert.NotNull(first.Job);
            Assert.NotNull(second.Job);
            Assert.Equal(first.Job.Version, second.Job.Version);
            Assert.Equal(first.Job.Version, queuedVersion + 2);
        }

        using (var scope = Factory.Services.CreateScope())
        {
            var db = scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
            Assert.Equal(1, await db.AuditEvents.AsNoTracking().CountAsync(item =>
                item.Action == "job.queued" && item.TargetId == created.Job.Id.ToString("D")));
            Assert.Equal(1, await db.AuditEvents.AsNoTracking().CountAsync(item =>
                item.Action == "job.cancel_requested" && item.TargetId == created.Job.Id.ToString("D")));
        }
    }

    [Fact]
    public async Task Worker_rebases_its_version_when_a_concurrent_cancellation_wins_the_row_lock()
    {
        var (workspaceId, actorId) = await GetOwnerContextAsync();
        EnqueueJobResult queued;
        using (var scope = Factory.Services.CreateScope())
        {
            queued = await scope.ServiceProvider.GetRequiredService<QueueBackgroundJobUseCase>()
                .QueueBackupAsync(new QueueJobCommand(workspaceId, actorId, "version-race", "version-race"), CancellationToken.None);
        }
        Assert.Equal(EnqueueJobStatus.Created, queued.Status);
        Assert.NotNull(queued.Job);

        var leaseToken = Guid.NewGuid();
        var now = DateTimeOffset.UtcNow;
        using (var scope = Factory.Services.CreateScope())
        {
            var repository = scope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
            Assert.Equal(queued.Job.Id, await repository.TryClaimNextAsync(
                leaseToken, now, now.AddMinutes(2), CancellationToken.None));
        }

        using var lockScope = Factory.Services.CreateScope();
        var lockDb = lockScope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>();
        await lockDb.Database.OpenConnectionAsync();
        await using var transaction = await lockDb.Database.BeginTransactionAsync();
        var cancellationAt = now.AddSeconds(1);
        Assert.Equal(1, await lockDb.Database.ExecuteSqlInterpolatedAsync($"""
            UPDATE background_jobs
            SET cancel_requested_at = {cancellationAt}, version = version + 1
            WHERE id = {queued.Job.Id} AND status = 'Running' AND lease_token = {leaseToken}
            """, CancellationToken.None));
        var cancellationVersion = await lockDb.BackgroundJobs.AsNoTracking()
            .Where(job => job.Id == queued.Job.Id)
            .Select(job => job.Version)
            .SingleAsync();

        using var workerScope = Factory.Services.CreateScope();
        var workerRepository = workerScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>();
        var workerWrite = workerRepository.RecordJobOutcomeAsync(
            queued.Job.Id, leaseToken, BackgroundJobItemState.Succeeded, null, null, CancellationToken.None);

        var updateBlockedByCancellation = false;
        var waitUntil = DateTimeOffset.UtcNow.AddSeconds(5);
        while (!updateBlockedByCancellation && DateTimeOffset.UtcNow < waitUntil)
        {
            // PostgreSQL caches statistics snapshots inside a transaction unless
            // explicitly cleared, so refresh before each activity check.
            await lockDb.Database.SqlQueryRaw<bool>("SELECT pg_stat_clear_snapshot() IS NULL AS \"Value\"")
                .SingleAsync();
            updateBlockedByCancellation = await lockDb.Database.SqlQueryRaw<bool>("""
                SELECT EXISTS (
                    SELECT 1 FROM pg_stat_activity
                    WHERE datname = current_database()
                      AND pid <> pg_backend_pid()
                      AND wait_event_type = 'Lock'
                      AND query ILIKE '%background_jobs%'
                ) AS "Value"
                """).SingleAsync();
            if (!updateBlockedByCancellation)
                await Task.Delay(25);
        }

        await transaction.CommitAsync();
        await workerWrite.WaitAsync(TimeSpan.FromSeconds(5));
        Assert.True(updateBlockedByCancellation, "The worker update should wait behind the concurrent cancellation row lock.");

        using var verifyScope = Factory.Services.CreateScope();
        var current = await verifyScope.ServiceProvider.GetRequiredService<IBackgroundJobRepository>()
            .FindAsync(workspaceId, queued.Job.Id, CancellationToken.None);
        Assert.NotNull(current);
        Assert.Equal(BackgroundJobStatus.Running, current.Status);
        Assert.Equal(leaseToken, current.LeaseToken);
        Assert.NotNull(current.CancelRequestedAt);
        Assert.InRange(Math.Abs((current.CancelRequestedAt.Value - cancellationAt).Ticks), 0, 10);
        Assert.Equal(1, current.Processed);
        Assert.Equal(1, current.Succeeded);
        Assert.True(current.Version > cancellationVersion);
    }

    private async Task<(Guid WorkspaceId, Guid ActorId)> GetOwnerContextAsync()
    {
        using var scope = Factory.Services.CreateScope();
        var provider = scope.ServiceProvider;
        var workspaceId = await provider.GetRequiredService<IWorkspaceRepository>()
            .GetDefaultWorkspaceIdAsync(CancellationToken.None);
        var actorId = await provider.GetRequiredService<KnowledgeHubDbContext>().Users.AsNoTracking()
            .Where(user => user.Email == OwnerEmail)
            .Select(user => user.Id)
            .SingleAsync();
        return (workspaceId ?? throw new InvalidOperationException("The owner workspace is missing."), actorId);
    }

    private static async Task InstallAuditFailureTriggerAsync(KnowledgeHubDbContext db, string action)
    {
        await db.Database.ExecuteSqlRawAsync("""
            CREATE TABLE test_job_audit_failure_control (action text PRIMARY KEY);
            CREATE OR REPLACE FUNCTION test_reject_job_audit() RETURNS trigger
            LANGUAGE plpgsql AS $$
            BEGIN
                IF EXISTS (SELECT 1 FROM test_job_audit_failure_control WHERE action = NEW.action) THEN
                    RAISE EXCEPTION 'intentional job audit failure';
                END IF;
                RETURN NEW;
            END;
            $$;
            CREATE TRIGGER test_reject_job_audit
            BEFORE INSERT ON audit_events
            FOR EACH ROW EXECUTE FUNCTION test_reject_job_audit();
            """);
        await db.Database.ExecuteSqlInterpolatedAsync($"""
            INSERT INTO test_job_audit_failure_control(action) VALUES ({action})
            """);
    }

    private async Task RemoveAuditFailureTriggerAsync()
    {
        using var scope = Factory.Services.CreateScope();
        await scope.ServiceProvider.GetRequiredService<KnowledgeHubDbContext>().Database.ExecuteSqlRawAsync("""
            DROP TRIGGER IF EXISTS test_reject_job_audit ON audit_events;
            DROP FUNCTION IF EXISTS test_reject_job_audit();
            DROP TABLE IF EXISTS test_job_audit_failure_control;
            """);
    }
}