tiku-backend.net/Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs

using System.Net;
using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using Tiku.Application.Jobs;
using Tiku.Domain.Operations;
using Tiku.Domain.Tenancy;
using Tiku.Infrastructure.Jobs;
using Tiku.Infrastructure.Persistence;

namespace Tiku.IntegrationTests.Api;

public sealed class MonolithBackgroundProcessingTests
{
    [Fact]
    public async Task Anonymous_readiness_is_minimal_and_does_not_expose_dependency_topology()
    {
        await using var factory = new ApiTestFactory();
        using var client = factory.CreateClient();

        var response = await client.GetAsync("/api/system/health/ready");
        using var document = await JsonDocument.ParseAsync(await response.Content.ReadAsStreamAsync());

        Assert.Equal(HttpStatusCode.OK, response.StatusCode);
        Assert.Equal("ready", document.RootElement.GetProperty("status").GetString());
        Assert.False(document.RootElement.TryGetProperty("database", out _));
        Assert.False(document.RootElement.TryGetProperty("redis", out _));
        Assert.False(document.RootElement.TryGetProperty("rabbitMq", out _));
        Assert.False(document.RootElement.TryGetProperty("outbox", out _));
    }

    [Fact]
    public async Task Api_host_does_not_register_background_processors()
    {
        await using var factory = new ApiTestFactory();
        using var client = factory.CreateClient();
        var hostedServiceNames = factory.Services.GetServices<IHostedService>()
            .Select(service => service.GetType().Name)
            .ToArray();

        Assert.DoesNotContain("TenantDomainWorker", hostedServiceNames);
        Assert.DoesNotContain("SaasSubscriptionWorker", hostedServiceNames);
        Assert.DoesNotContain("FeatureUsageWorker", hostedServiceNames);
        Assert.DoesNotContain("BackgroundJobsWorker", hostedServiceNames);
    }

    [Fact]
    public async Task Postgres_leases_recover_expired_work_without_duplicate_claims()
    {
        await using var factory = new ApiTestFactory();
        var tenantId = Guid.NewGuid();
        await factory.SeedAsync(new Tenant
        {
            Id = tenantId,
            Slug = tenantId.ToString("N"),
            Name = "Monolith Lease Recovery"
        });

        Guid expiredJobId;
        Guid pendingJobId;
        using (var scope = factory.CreateSystemScope("Seed lease recovery jobs"))
        {
            var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
            expiredJobId = (await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
                tenantId,
                "statistics_aggregation",
                JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id;
            pendingJobId = (await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
                tenantId,
                "statistics_aggregation",
                JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id;
            var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
            await dbContext.BackgroundJobs
                .Where(job => job.Id == expiredJobId)
                .ExecuteUpdateAsync(setters => setters
                    .SetProperty(job => job.Status, BackgroundJobStatus.Processing)
                    .SetProperty(job => job.LockedBy, "stopped-worker")
                    .SetProperty(job => job.LockExpiresAt, DateTimeOffset.UtcNow.AddSeconds(-1)));
        }

        var processed = await Task.WhenAll(
            ProcessPendingAsync(factory, "monolith-lease-a"),
            ProcessPendingAsync(factory, "monolith-lease-b"));

        Assert.Equal(2, processed.Sum());
        Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, expiredJobId));
        Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, pendingJobId));
    }

    [Fact]
    public async Task Active_job_lease_is_renewed_and_cannot_be_reclaimed()
    {
        var handler = new BlockingStatisticsHandler();
        await using var factory = CreateLeaseFactory(handler);
        var jobId = await SeedJobAsync(factory, "Lease Renewal");

        var firstWorker = ProcessPendingAsync(factory, "lease-renewal-a");
        await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5));
        await Task.Delay(TimeSpan.FromMilliseconds(700));

        Assert.Equal(0, await ProcessPendingAsync(factory, "lease-renewal-b"));
        handler.Release.TrySetResult();
        Assert.Equal(1, await firstWorker);
        Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, jobId));
    }

    [Fact]
    public async Task Lost_job_lease_cancels_old_worker_without_overwriting_new_owner()
    {
        var handler = new BlockingStatisticsHandler();
        await using var factory = CreateLeaseFactory(handler);
        var jobId = await SeedJobAsync(factory, "Lease Ownership Loss");

        var firstWorker = ProcessPendingAsync(factory, "lease-owner-a");
        await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5));
        using (var scope = factory.CreateSystemScope("Transfer job lease ownership"))
        {
            var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
            await dbContext.BackgroundJobs.Where(job => job.Id == jobId)
                .ExecuteUpdateAsync(setters => setters
                    .SetProperty(job => job.LockedBy, "lease-owner-b")
                    .SetProperty(job => job.LockExpiresAt, DateTimeOffset.UtcNow.AddMinutes(1)));
        }

        Assert.Equal(0, await firstWorker.WaitAsync(TimeSpan.FromSeconds(5)));
        using var verificationScope = factory.CreateSystemScope("Verify transferred job lease ownership");
        var stored = await verificationScope.ServiceProvider.GetRequiredService<TikuDbContext>()
            .BackgroundJobs.AsNoTracking()
            .SingleAsync(job => job.Id == jobId);
        Assert.Equal(BackgroundJobStatus.Processing, stored.Status);
        Assert.Equal("lease-owner-b", stored.LockedBy);
    }

    [Fact]
    public async Task Cancelled_worker_leaves_processing_lease_for_expiry_recovery()
    {
        var handler = new BlockingStatisticsHandler();
        await using var factory = CreateLeaseFactory(handler);
        var jobId = await SeedJobAsync(factory, "Lease Cancellation Recovery");
        using var cancellation = new CancellationTokenSource();

        var firstWorker = ProcessPendingAsync(factory, "lease-cancelled-a", cancellation.Token);
        await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5));
        cancellation.Cancel();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => firstWorker);

        using (var scope = factory.CreateSystemScope("Verify cancelled worker lease"))
        {
            var stored = await scope.ServiceProvider.GetRequiredService<TikuDbContext>()
                .BackgroundJobs.AsNoTracking()
                .SingleAsync(job => job.Id == jobId);
            Assert.Equal(BackgroundJobStatus.Processing, stored.Status);
            Assert.Equal("lease-cancelled-a", stored.LockedBy);
            Assert.NotNull(stored.LockExpiresAt);
        }

        await Task.Delay(TimeSpan.FromMilliseconds(700));
        Assert.Equal(1, await ProcessPendingAsync(factory, "lease-recovery-b"));
        Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, jobId));
    }

    private static async Task<int> ProcessPendingAsync(
        ApiTestFactory factory,
        string workerId,
        CancellationToken cancellationToken = default)
    {
        using var scope = factory.CreateSystemScope($"Process jobs with {workerId}");
        return await scope.ServiceProvider.GetRequiredService<IBackgroundJobService>()
            .ProcessPendingAsync(workerId, 1, cancellationToken: cancellationToken);
    }

    private static ApiTestFactory CreateLeaseFactory(BlockingStatisticsHandler handler)
    {
        return new ApiTestFactory(configureTestServices: services =>
        {
            services.RemoveAll<IBackgroundJobHandler>();
            services.AddSingleton<IBackgroundJobHandler>(handler);
            services.RemoveAll<BackgroundJobLeaseTiming>();
            services.AddSingleton(new BackgroundJobLeaseTiming(
                TimeSpan.FromMilliseconds(400),
                TimeSpan.FromMilliseconds(50),
                TimeSpan.FromMilliseconds(20)));
        });
    }

    private static async Task<Guid> SeedJobAsync(ApiTestFactory factory, string tenantName)
    {
        var tenantId = Guid.NewGuid();
        await factory.SeedAsync(new Tenant
        {
            Id = tenantId,
            Slug = tenantId.ToString("N"),
            Name = tenantName
        });
        using var scope = factory.CreateSystemScope($"Seed {tenantName} job");
        return (await scope.ServiceProvider.GetRequiredService<IBackgroundJobService>()
            .EnqueueAsync(new CreateBackgroundJobCommand(
                tenantId,
                "statistics_aggregation",
                JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id;
    }

    private static async Task<BackgroundJobStatus> ReadStatusAsync(ApiTestFactory factory, Guid jobId)
    {
        using var scope = factory.CreateSystemScope("Read monolith background job status");
        return await scope.ServiceProvider.GetRequiredService<TikuDbContext>()
            .BackgroundJobs.AsNoTracking()
            .Where(job => job.Id == jobId)
            .Select(job => job.Status)
            .SingleAsync();
    }

    private sealed class BlockingStatisticsHandler : IBackgroundJobHandler
    {
        private int calls;

        public string JobType => "statistics_aggregation";
        public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
        public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);

        public async Task<BackgroundJobHandlerResult> HandleAsync(
            BackgroundJobExecutionContext context,
            CancellationToken cancellationToken = default)
        {
            if (Interlocked.Increment(ref calls) == 1)
            {
                Started.TrySetResult();
                await Release.Task.WaitAsync(cancellationToken);
            }

            return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new { completed = true }));
        }
    }
}