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

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

namespace Tiku.IntegrationTests.Api;

public sealed class MonolithBackgroundProcessingTests
{
    private static readonly IReadOnlyDictionary<string, string?> EnabledConfiguration =
        new Dictionary<string, string?>
        {
            ["BackgroundProcessing:Enabled"] = "true",
            ["BackgroundProcessing:JobPollSeconds"] = "1",
            ["BackgroundProcessing:JobParallelism"] = "2",
            ["BackgroundProcessing:JobBatchSize"] = "2",
            ["TenantDomains:Enabled"] = "false",
            ["SaasSubscriptions:Enabled"] = "false",
            ["FeatureUsageReconciliation:Enabled"] = "false"
        };

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

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

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

    [Fact]
    public async Task Background_processing_registration_obeys_master_switch()
    {
        await using var disabledFactory = new ApiTestFactory();
        using var disabledClient = disabledFactory.CreateClient();
        var disabledNames = disabledFactory.Services.GetServices<IHostedService>()
            .Select(service => service.GetType().Name)
            .ToArray();

        Assert.False(disabledFactory.Services.GetRequiredService<IOptions<BackgroundProcessingOptions>>().Value.Enabled);
        Assert.Contains("TenantDomainBackgroundService", disabledNames);
        Assert.Contains("SaasSubscriptionBackgroundService", disabledNames);
        Assert.Contains("FeatureUsageBackgroundService", disabledNames);
        Assert.Contains("BackgroundJobsBackgroundService", disabledNames);

        await using var enabledFactory = new ApiTestFactory(configurationOverrides: EnabledConfiguration);
        using var enabledClient = enabledFactory.CreateClient();
        var enabledNames = enabledFactory.Services.GetServices<IHostedService>()
            .Select(service => service.GetType().Name)
            .ToArray();

        Assert.True(enabledFactory.Services.GetRequiredService<IOptions<BackgroundProcessingOptions>>().Value.Enabled);
        Assert.Contains("TenantDomainBackgroundService", enabledNames);
        Assert.Contains("SaasSubscriptionBackgroundService", enabledNames);
        Assert.Contains("FeatureUsageBackgroundService", enabledNames);
        Assert.Contains("BackgroundJobsBackgroundService", enabledNames);
    }

    [Fact]
    public async Task Api_host_processes_immediate_and_due_postgres_jobs()
    {
        await using var factory = new ApiTestFactory(configurationOverrides: EnabledConfiguration);
        using var client = factory.CreateClient();
        var tenantId = Guid.NewGuid();
        await factory.SeedAsync(new Tenant
        {
            Id = tenantId,
            Slug = tenantId.ToString("N"),
            Name = "Monolith Background Processing"
        });

        BackgroundJobItem immediate;
        BackgroundJobItem delayed;
        using (var scope = factory.CreateSystemScope("Queue monolith background jobs"))
        {
            var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
            immediate = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
                tenantId,
                "statistics_aggregation",
                JsonSerializer.SerializeToElement(new { scope = "tenant" })));
            delayed = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
                tenantId,
                "statistics_aggregation",
                JsonSerializer.SerializeToElement(new { scope = "tenant" }),
                DateTimeOffset.UtcNow.AddMinutes(5)));
        }

        Assert.Equal(BackgroundJobStatus.Succeeded, await WaitForStatusAsync(factory, immediate.Id));
        Assert.Equal(BackgroundJobStatus.Pending, await ReadStatusAsync(factory, delayed.Id));

        using (var scope = factory.CreateSystemScope("Make delayed monolith job due"))
        {
            var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
            await dbContext.BackgroundJobs
                .Where(job => job.Id == delayed.Id)
                .ExecuteUpdateAsync(setters => setters.SetProperty(job => job.RunAfter, DateTimeOffset.UtcNow.AddSeconds(-1)));
        }

        Assert.Equal(BackgroundJobStatus.Succeeded, await WaitForStatusAsync(factory, delayed.Id));
    }

    [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));
    }

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

    private static async Task<BackgroundJobStatus> WaitForStatusAsync(ApiTestFactory factory, Guid jobId)
    {
        var timeout = DateTimeOffset.UtcNow.AddSeconds(10);
        while (DateTimeOffset.UtcNow < timeout)
        {
            var status = await ReadStatusAsync(factory, jobId);
            if (status is BackgroundJobStatus.Succeeded or BackgroundJobStatus.Failed)
            {
                return status;
            }

            await Task.Delay(100);
        }

        return await ReadStatusAsync(factory, jobId);
    }

    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();
    }
}