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 EnabledConfiguration = new Dictionary { ["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() .Select(service => service.GetType().Name) .ToArray(); Assert.False(disabledFactory.Services.GetRequiredService>().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() .Select(service => service.GetType().Name) .ToArray(); Assert.True(enabledFactory.Services.GetRequiredService>().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(); 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(); 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(); 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(); 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 ProcessPendingAsync(ApiTestFactory factory, string workerId) { using var scope = factory.CreateSystemScope($"Process jobs with {workerId}"); return await scope.ServiceProvider.GetRequiredService() .ProcessPendingAsync(workerId, 1); } private static async Task 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 ReadStatusAsync(ApiTestFactory factory, Guid jobId) { using var scope = factory.CreateSystemScope("Read monolith background job status"); return await scope.ServiceProvider.GetRequiredService() .BackgroundJobs.AsNoTracking() .Where(job => job.Id == jobId) .Select(job => job.Status) .SingleAsync(); } }